test(managed-agent): Run the EventTransport fault matrix against two nodes (MQ3) (#14242)

MQ3 of the cross-node EventTransport design: the §8 deployment fault matrix runs against two control-plane nodes — each the whole application in its own JVM with its own packaged Hosted Harness — sharing one MySQL database and one RocketMQ 5.5 LiteTopic cluster, with the faults injected into the processes themselves. EventTransportMultiNodeGate starts the nodes through EventTransportNodeMain, which swaps test-only subclasses in for the wake activator and both wires. A fault is armed for one Session by a file and fires once across the nodes: halt on receiving a wake, halt after claiming it, hold a load until released, halt after a wake's record or a public batch commits and before it is handed to the broker. Each node appends every wake decision, load and hand-off to an evidence file; the gate asserts on that, on SQL and on the public API, and reads the lanes through its own oracle group. The broker outage freezes the gRPC proxy with SIGSTOP after checking its pid holds the endpoint; the clock-skew row moves one node's JVM wall clock with libfaketime and checks that it moved, and is skipped without it. Ten rows: duplicate/reorder/redelivery, consumer crash before processing, consumer crash after the claim, producer crash on the journal and on the public wire, short and long broker outage, clock skew, a stale wake after a re-route, and flag rollback. The rows assert the as-built behavior; the MQ3 design records where it differs from §8: a crash with a wake in flight stalls the shared bootstrap lane for about two minutes, there is no producer retry buffer, a node cannot start while the broker is down, and the Turn dispatch lease is on the JVM clock. A new profile, event-transport-multi-node-gates, and a new SDK Java job run the gate on ubuntu-latest with a MySQL 8.4 service, the bundled CLI, the broker and libfaketime, daily and on demand like the rest of that workflow; check-failsafe-reports gains the event-transport-multi-node family, and a CI pin test keeps the job, the profile and the gate in step. No production code changes.
2026-10-21 22:39:45 +00:00 · 2026-10-15 04:34:45 +00:00 · 2026-10-15 04:34:45 +00:00 · 404845a17a
commit 404845a17a
parent 859a337500
15 changed files with 2398 additions and 9 deletions
--- a/.github/workflows/.size-baseline
+++ b/.github/workflows/.size-baseline
@ -57,7 +57,7 @@
 1079 scorecard-monthly.yml
 9751 sdk-java-contract-open-prs.yml
 9555 sdk-java-flyway-open-prs.yml
-69416 sdk-java.yml
+69263 sdk-java.yml
 3886 sdk-python.yml
 5595 security-checks.yml
 6777 serve-ab-publish.yml
--- a/.github/workflows/sdk-java.yml
+++ b/.github/workflows/sdk-java.yml
@ -938,6 +938,115 @@ jobs:
          RMQ_WORK_DIR: '${{ runner.temp }}/rocketmq'
        run: 'scripts/rocketmq-lite-broker.sh stop'

+  event-transport-multi-node:
+    name: 'EventTransport multi-node gates / RocketMQ 5.5.1 + MySQL 8.4 / Java 21'
+    # Hosted: the gate boots its own broker on fixed ports and its own
+    # control-plane nodes.
+    runs-on: 'ubuntu-latest'
+    # 10 + 60 capped minutes plus ~15 for the uncapped setup (the CLI
+    # bundle, the module installs, libfaketime).
+    timeout-minutes: 90
+    services:
+      mysql:
+        image: 'mysql:8.4.6'
+        env:
+          MYSQL_DATABASE: 'event_transport_test'
+          MYSQL_ROOT_PASSWORD: 'multi-node-fixture'
+        ports:
+          - '3306/tcp'
+        options: >-
+          --health-cmd "mysqladmin ping -h 127.0.0.1 -pmulti-node-fixture"
+          --health-interval 5s
+          --health-timeout 5s
+          --health-retries 30
+    steps:
+      - name: 'Checkout'
+        uses: 'actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10' # v6.0.3
+      - name: 'Set up Java'
+        uses: 'actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654' # v5.2.0
+        with:
+          distribution: 'temurin'
+          # 17 runs the broker; 21, last, is the default.
+          java-version: |-
+            17
+            21
+          cache: 'maven'
+          cache-dependency-path: 'packages/sdk-java/**/pom.xml'
+          settings-path: '${{ runner.temp }}/setup-java-m2'
+      - name: 'Set up Node.js'
+        uses: 'actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e' # v6.4.0
+        with:
+          node-version: '22'
+          package-manager-cache: false
+      - name: 'Enable Corepack'
+        run: 'corepack enable'
+      - name: 'Install, build and bundle CLI'
+        run: |-
+          corepack pnpm install --frozen-lockfile --prefer-offline --reporter=append-only
+      - name: 'Install Managed Agent dependencies'
+        env:
+          MAVEN_ARGS: '--settings ${{ runner.temp }}/setup-java-m2/settings.xml --toolchains ${{ runner.temp }}/setup-java-m2/toolchains.xml'
+        run: |-
+          mvn --batch-mode --no-transfer-progress -f packages/sdk-java/qwencode/pom.xml -DskipTests -Dgpg.skip=true -Dmaven.javadoc.skip=true install
+          mvn --batch-mode --no-transfer-progress -f packages/sdk-java/runtime-broker/pom.xml -DskipTests -Dspotbugs.skip=true install
+      - name: 'Install libfaketime'
+        # The clock-skew row moves one node's whole wall clock.
+        run: |-
+          sudo apt-get update -q
+          sudo apt-get install -y -q faketime
+          test -f /usr/lib/x86_64-linux-gnu/faketime/libfaketimeMT.so.1
+      - name: 'Cache the RocketMQ distribution'
+        uses: 'actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9' # v6.1.0
+        with:
+          # The script checks the zip against a pinned SHA-512.
+          path: '${{ runner.temp }}/rocketmq-dist'
+          key: 'rocketmq-5.5.1-bin-release'
+      - name: 'Start the RocketMQ LiteTopic cluster'
+        id: 'broker'
+        # Runs the broker on setup-java's JAVA_HOME_17_X64.
+        timeout-minutes: 10
+        env:
+          RMQ_WORK_DIR: '${{ runner.temp }}/rocketmq'
+          RMQ_CACHE_DIR: '${{ runner.temp }}/rocketmq-dist'
+          RMQ_EXTRA_GROUPS: 'qwen-transport-gate-oracle'
+        run: 'scripts/rocketmq-lite-broker.sh start'
+      - name: 'Verify the EventTransport multi-node gates'
+        # Outlives the 3000 s failsafe fork plus the compile and SpotBugs
+        # around it (#13503).
+        timeout-minutes: 60
+        env:
+          MAVEN_ARGS: '--settings ${{ runner.temp }}/setup-java-m2/settings.xml --toolchains ${{ runner.temp }}/setup-java-m2/toolchains.xml'
+          MYSQL_PORT: "${{ job.services.mysql.ports['3306'] }}"
+        run: |-
+          mvn --batch-mode --no-transfer-progress -f packages/sdk-java/managed-agent-server/pom.xml \
+            -Pevent-transport-multi-node-gates \
+            -Dnode.executable="$(command -v node)" \
+            -Dqwen.cli.entry="${GITHUB_WORKSPACE}/dist/cli.js" \
+            -Dmysql.url="jdbc:mysql://127.0.0.1:${MYSQL_PORT}/event_transport_test?allowPublicKeyRetrieval=true&useSSL=false" \
+            -Dmysql.user=root -Dmysql.password=multi-node-fixture \
+            -Dqwen.gate.rocketmq.endpoints=127.0.0.1:18081 \
+            -Dqwen.gate.rocketmq.topic=qwen-managed-events \
+            -Dqwen.gate.rocketmq.work-dir="${RUNNER_TEMP}/rocketmq" \
+            -Dqwen.gate.faketime.library=/usr/lib/x86_64-linux-gnu/faketime/libfaketimeMT.so.1 \
+            clean verify && \
+            node scripts/check-failsafe-reports.js event-transport-multi-node packages/sdk-java/managed-agent-server
+      - name: 'Upload multi-node gate reports'
+        if: "${{ always() && steps.broker.outcome != 'skipped' }}"
+        uses: 'actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a' # v7.0.1
+        with:
+          name: 'event-transport-multi-node-gate-reports'
+          path: |-
+            packages/sdk-java/managed-agent-server/target/failsafe-reports/
+            packages/sdk-java/managed-agent-server/target/mq3-multi-node/
+            ${{ runner.temp }}/rocketmq/logs/
+            ${{ runner.temp }}/rocketmq/home/logs/
+          retention-days: 7
+      - name: 'Stop the RocketMQ cluster'
+        if: "${{ always() && steps.broker.outcome != 'skipped' }}"
+        env:
+          RMQ_WORK_DIR: '${{ runner.temp }}/rocketmq'
+        run: 'scripts/rocketmq-lite-broker.sh stop'
+
  daemon-e2e:
    name: 'Real daemon E2E / Java 11'
    runs-on: '${{ (github.repository == ''QwenLM/qwen-code'' && vars.MAINTAINER_ECS_RUNNER_DISABLED != ''true'' && (github.event_name != ''pull_request'' || github.event.pull_request.head.repo.full_name == github.repository || contains(fromJSON(''["OWNER","MEMBER","COLLABORATOR"]''), github.event.pull_request.author_association))) && fromJSON(''["self-hosted", "linux", "x64", "ecs-qwen"]'') || ''ubuntu-latest'' }}'
--- a/docs/design/2026-10-04-managed-agent-event-transport.md
+++ b/docs/design/2026-10-04-managed-agent-event-transport.md
@ -474,6 +474,16 @@ the source that committed its input, where one exists — there is no generic
 wake scan. The MQ2 design's "Where a missed or burned wake recovers" names
 the ledgers and the gaps they already have.

+As run (MQ3): the rows run against two control-plane nodes, a MySQL
+database and a RocketMQ cluster in
+[2026-10-14-managed-agent-event-transport-mq3](2026-10-14-managed-agent-event-transport-mq3.md)
+(the public lane's duplicate and reorder stay with MQ2d's gate, and no epoch
+travels with a wake for the stale-wake row to bump). That design also records
+where the as-built rows differ from the required behavior above: a crash
+with a wake in flight holds the shared bootstrap lane for about two minutes; there is no producer retry buffer, so an envelope sent while the
+broker stays away is dropped; a node restarted during an outage needs the flag
+off to start; the Turn dispatch lease is on the JVM clock.
+
 ## 9. Constraints and risks

 - **Nothing here may appear on the synchronous frontend path.** SQL failure
--- a/docs/design/2026-10-04-managed-agent-event-transport.zh-CN.md
+++ b/docs/design/2026-10-04-managed-agent-event-transport.zh-CN.md
@ -359,6 +359,13 @@ MQ2d 落地情况：上表各行中的「SQL 侧扫描」「SQL 扫描」，对
 扫描；对唤醒而言是提交该输入的那个来源的账本（如果有）——不存在通用的唤醒扫描。
 MQ2 设计的「丢失或被烧掉的唤醒靠什么恢复」列出了这些账本及其早已存在的缺口。

+MQ3 实测情况：各行在两个控制面节点、一个 MySQL 数据库和一个 RocketMQ 集群上运行，
+见 [2026-10-14-managed-agent-event-transport-mq3](2026-10-14-managed-agent-event-transport-mq3.zh-CN.md)
+（公开事件通道的重复与乱序仍归 MQ2d 的门禁；唤醒不携带 epoch，过期唤醒行也就没有代际可变）；
+该文档同时记录了现有实现与上表「要求的行为」的出入：节点在处理唤醒时崩溃，会让共享的
+bootstrap 通道停顿约两分钟；没有生产者重试缓冲，broker 离开期间发出的信封会被丢弃；
+故障期间重启的节点要关掉开关才能启动；Turn 调度租约用的是 JVM 时钟。
+
 ## 9. 约束与风险

 - **此处任何内容不得出现在同步前端路径上。** SQL 故障以有界背压呈现（#12380）；
--- a/docs/design/2026-10-12-managed-agent-event-transport-mq2.md
+++ b/docs/design/2026-10-12-managed-agent-event-transport-mq2.md
@ -289,7 +289,8 @@ the transport.

 - No enablement by default anywhere; the flag is off in every shipped
  configuration.
- No multi-node fault matrix (that is MQ3).
+- No multi-node fault matrix (that is MQ3:
+  [2026-10-14-managed-agent-event-transport-mq3](2026-10-14-managed-agent-event-transport-mq3.md)).
 - No change to the commit path's transaction shape; the publication point is
  after-commit only, and publication failure never blocks the local
  post-commit SSE (#12380 constraint).
--- a/docs/design/2026-10-12-managed-agent-event-transport-mq2.zh-CN.md
+++ b/docs/design/2026-10-12-managed-agent-event-transport-mq2.zh-CN.md
@ -77,7 +77,7 @@ MQ2c 不包含:真实 broker 故障行(MQ2d)、接收侧对 journal 事实的 di
 ## 范围边界

 - 任何交付配置都不默认启用;flag 在所有交付配置中关闭。
- 不含多节点故障矩阵(那是 MQ3)。
+- 不含多节点故障矩阵(那是 MQ3:[2026-10-14-managed-agent-event-transport-mq3](2026-10-14-managed-agent-event-transport-mq3.zh-CN.md))。
 - 不改提交路径的事务形态;发布点仅 after-commit,且发布失败永不阻塞本地 post-commit SSE(#12380 约束)。
 - 不重新标称:不主张 MQ1 hub seam 覆盖 journal 线。

--- a/docs/design/2026-10-14-managed-agent-event-transport-mq3.md
+++ b/docs/design/2026-10-14-managed-agent-event-transport-mq3.md
@ -0,0 +1,191 @@
+# EventTransport MQ3: the deployment fault matrix
+
+[English](2026-10-14-managed-agent-event-transport-mq3.md) | [简体中文](2026-10-14-managed-agent-event-transport-mq3.zh-CN.md)
+
+> **Status:** Implemented as a gate. This is phase **MQ3** of the cross-node
+> EventTransport design
+> ([2026-10-04-managed-agent-event-transport](2026-10-04-managed-agent-event-transport.md),
+> §7), tracked under [#12380](https://github.com/QwenLM/qwen-code/issues/12380).
+> Its prerequisite, MQ2, landed in four slices
+> ([2026-10-12-managed-agent-event-transport-mq2](2026-10-12-managed-agent-event-transport-mq2.md)):
+> MQ2a [#13940](https://github.com/QwenLM/qwen-code/pull/13940), MQ2b
+> [#14029](https://github.com/QwenLM/qwen-code/pull/14029), MQ2c
+> [#14080](https://github.com/QwenLM/qwen-code/pull/14080) and MQ2d
+> [#14109](https://github.com/QwenLM/qwen-code/pull/14109). MQ3 changes no
+> production code. Its rows assert the as-built behavior; where that differs
+> from §8's required behavior, the findings below record it.
+
+## Problem
+
+MQ2 ran every row it named against one application on H2 with a stand-in
+Harness. The parent design's §8 matrix is about deployments: more than one
+control-plane node, processes that die at awkward points, a broker that stops
+answering, clocks that disagree, and a flag that has to come off. Those rows
+only mean something against real processes. MQ3 runs them against two
+control-plane nodes, each with its own packaged Hosted Harness, sharing one
+MySQL database and one RocketMQ 5.5 LiteTopic cluster. It records, row by row,
+what the as-built transport does — including where that differs from §8's
+"required behavior" text.
+
+## Current state
+
+Read on `a6af39188a` (2026-10-14).
+
+- **Topology.** Every node joins the shared wake-bootstrap group on one
+  bootstrap lite topic, and the shared owner group on the lite topic of each
+  Session its own Harness writes. The listener returns — and the client
+  acknowledges — after the SQL checks and the claim, with the load queued on a
+  small pool.
+- **Producers.** Both wires send asynchronously and catch every failure. The
+  only retries are the client's own (three attempts in all, a three-second
+  request timeout); there is no application retry and no outage buffer. The
+  journal hand-off runs on the committing thread; the public hand-off copies
+  the batch onto a 1024-batch queue, also on the committing thread.
+- **Startup.** The capability probe, the send probe, the owner consumer and
+  the bootstrap subscription all need a reachable broker; any failure fails
+  the Spring context.
+- **Clocks.** The wake decisions and the journal writer lease read the
+  database clock. The Turn dispatch lease and the relay ledgers' leases are
+  stamped with the JVM clock.
+- **Fence.** No epoch travels with a wake. Past the checks every wake meets
+  (a known Session, the journal's workspace, a committed sequence, a Session
+  and journal not settled), a stale wake is refused only by a live writer or
+  by the claim watermark.
+
+## Design: the gate
+
+`EventTransportMultiNodeGate` (profile `event-transport-multi-node-gates`,
+SDK Java job "EventTransport multi-node gates / RocketMQ 5.5.1 + MySQL 8.4 /
+Java 21", failsafe family `event-transport-multi-node`).
+
+- **A node** is a packaged Hosted Harness (the CLI's hosted-harness serve
+  profile, with a fake model) plus the whole application in its own JVM,
+  started through `EventTransportNodeMain`. Each row starts its own nodes,
+  writer lease 6 s and Turn dispatch lease 6 s, so a dead node's holds lapse
+  within the row. All nodes of the class share one isolated MySQL database;
+  each row uses a fresh tenant. A crash kills the node's JVM and its Harness
+  with SIGKILL; a restart keeps the node's address.
+- **Fault points** are test-only subclasses the node JVM swaps in for the
+  production beans: the wake activator (halt on receiving a wake; halt after
+  claiming it, before the listener returns), the journal wire (halt after the
+  commit of a record that carries a wake, before it is handed to the broker)
+  and the public wire (halt after committing the batch that carries
+  `turn.completed`, before the hand-off). A fault is armed for one Session by
+  a file and fires once across both nodes; a fired fault calls
+  `Runtime.halt`, so nothing is acknowledged, flushed or shut down. A
+  hold-load fault instead parks every load of its Session, on every node,
+  until the gate releases it. The broker outage freezes the gRPC proxy with
+  SIGSTOP, after checking that the recorded pid is the one listening on the
+  endpoint. The clock-skew row moves the application JVM's wall clock with
+  libfaketime (Linux, CI) and checks that it moved; the node's Harness keeps
+  the host clock. Without libfaketime that row is skipped: the wake path reads
+  no Clock bean, so nothing short of the JVM's own clock can test it.
+- **Evidence.** Each node appends every wake decision, load outcome and
+  hand-off it makes to its own evidence file. The gate asserts on that, on
+  SQL (`ma_session_head` writer, generation and lease, `ma_event_stream`
+  watermarks, the committed journal) and on the public API, and reads what
+  reached the bootstrap lane — and, where a row watches it, the Session's own
+  lane — through its own oracle group. Per-row logs and evidence are uploaded
+  with the reports. Every wait fails fast once its row's timeout fires, and
+  the rows' timeouts fit the fork's. The broker must be the run's alone.
+
+## Rows
+
+| §8 fault                                            | Injection                                                                                                                                                                              | What the gate asserts                                                                                                                                                                                                                                                                                                           |
+| --------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| Duplicate, reorder, redelivery after a lost ACK     | With both nodes in the bootstrap group, a Session's committed wakes re-sent after its writer lapsed — the newer twice, then the older — with the load held until all three are decided | `SCHEDULED`, `DUPLICATE`, `DUPLICATE`, wherever they land; watermark = the newer; then one `ACTIVATED` through a real Harness load; writer generation +1, the writer is the activating node's Harness                                                                                                                           |
+| Consumer crash before processing                    | The receiving node halts on the wake before reading anything                                                                                                                           | Nothing claimed; the broker hands the wake to the other node, which activates the Session once                                                                                                                                                                                                                                  |
+| Consumer crash after a local effect, before the ACK | The receiving node claims the wake and halts before its listener returns                                                                                                               | The watermark holds the claim; the redelivery reaches the other node and is `DUPLICATE`; no load there; new input still runs there                                                                                                                                                                                              |
+| Producer crash between commit and publish (journal) | The node whose Harness commits the Turn's input halts after the commit, before the wake is sent                                                                                        | The wake is committed, reaches neither the bootstrap lane nor the Session's and is never claimed; the Turn dispatcher settles the Turn on the other node, once                                                                                                                                                                  |
+| Producer crash between commit and publish (public)  | The node recording the Turn's last batch halts after committing it, before the hand-off                                                                                                | No later notice follows, so only the other node's scheduled pass projects it: the committed `turn.completed` is served, the log has no gap, `covered_sequence = last_sequence`, and the items carry the input and the reply                                                                                                     |
+| Broker short outage                                 | The proxy is frozen while a Turn runs                                                                                                                                                  | The Turn completes; both hand-offs on the committing thread stay under 250 ms; the next Turn's wake reaches the broker once the proxy is back                                                                                                                                                                                   |
+| Broker long outage                                  | The proxy is frozen across a node restart                                                                                                                                              | Once sends are seen failing, a running node still serves Turns — a held Session's and a new Session's first — with both hand-offs under 250 ms; a node restarted with the flag on refuses to start; the same node, now alone, restarted flag-off starts and serves; after the outage, restarted flag-on, it carries wakes again |
+| Clock skew between Java nodes                       | One node's JVM wall clock 10 minutes ahead, then another's 10 minutes behind                                                                                                           | Ahead: another node's live writer is still `OWNED`, nothing claimed; behind: a lapsed writer is still lapsed, the Session activates there, and the lease it is granted is live by database time                                                                                                                                 |
+| Stale wake after a re-route                         | The Session moves to the other node's Harness; its old wake is re-sent                                                                                                                 | `OWNED` wherever it lands; the watermark stays empty, no load, the writer unchanged                                                                                                                                                                                                                                             |
+| Flag rollback                                       | Both nodes restarted flag-off, then flag-on                                                                                                                                            | Flag off: Turns run as before, nothing reaches either lane, and a wake sent meanwhile stays unconsumed; flag on again: that wake is consumed and the next wake is published                                                                                                                                                     |
+
+## Findings
+
+Findings 1–3, 5 and 6 are where the as-built behavior differs from §8's
+text; 4 and 7 are as-built properties the rows rest on.
+
+1. **A crash with a wake in flight stalls the bootstrap lane for about two
+   minutes.** LiteTopic gives a crashed member's unacknowledged message to
+   another member only once the broker expires the dead client (its channel
+   expiry, two minutes by default), and the lite topic's later messages wait
+   behind it in order. Measured locally with a probe of two members of one
+   group: the held message reached the other member after 124 s, and a
+   message sent five seconds after the crash arrived with it; across the
+   local runs of the gate's two crash rows the wake moved after 121–129 s.
+   On the shared bootstrap lane that is every Session's wake, cluster-wide.
+   Nothing is lost — every wake arrives late — and no input ledger depends on
+   the lane. A node killed while holding no unacknowledged message did not
+   hold the lane: the next wake reached the other node within the row's
+   window. (§8: "broker redelivers; the Session is woken late" — true, but
+   late is minutes, for every Session.)
+2. **No producer retry buffer.** Against a hung proxy (frozen with SIGSTOP),
+   the client's own attempts, three seconds each, ride out a pause of a few
+   seconds — in the local short-outage runs, a Turn's worth of freeze,
+   nothing was dropped — but an envelope sent while the proxy stays away
+   longer is dropped with a `WARN` naming the event and is never sent later
+   (13 journal and 4 public envelopes in one local run of the long-outage
+   row). The commit path is untouched: both hand-offs on the committing
+   thread stayed in single-digit milliseconds during the freezes. (§8:
+   "producer retries with a bounded buffer" — not built; a wake lost in an
+   outage recovers through its ledger.)
+3. **Startup needs the broker.** A node restarted while the broker is down
+   refuses to start on the capability probe, whose message leads with the
+   LITE group and topic requirements and names the endpoints property only
+   last; the way back is the flag. The long-outage row asserts this refusal,
+   so a fix inverts it. (§8: "transport consumers idle" holds only for nodes
+   already running.)
+4. **A claim is never rolled back.** A crash after the claim burns that
+   notice; the Session recovers through the ledger of its input's source,
+   with the gaps recorded in
+   [#14116](https://github.com/QwenLM/qwen-code/issues/14116).
+5. **No epoch fence.** A stale wake is refused by a live writer or by the
+   watermark; on a free Session with a higher sequence it loads it — harmless,
+   because the wake pump runs only what the journal holds.
+6. **The Turn dispatch lease is on the JVM clock.** A node whose clock runs
+   ahead by more than the dispatch lease reads live claims as lapsed and takes
+   Turns over; in a local run with a node ten minutes ahead it took a live
+   Turn, hit the holding Harness's writer lease on every attempt and failed
+   the Turn when its retries ran out. This is outside the transport — the
+   clock-skew row keeps Turns away from the skewed node — and is recorded here
+   because §8 claims the house lease pattern for every lease.
+7. **A takeover of a Turn whose Harness stopped at the model call is
+   refused.** In the runs of the journal producer-crash row the Turn was
+   settled on the other node as failed (`managed_runtime_recovery_blocked`,
+   reason `model_start`): a decline the Harness makes from the journal, which
+   the transport plays no part in. The row asserts only that the Turn
+   settles, once.
+
+## Scope boundaries
+
+- No production change. Findings 1–3 and 6 are recorded here for follow-up;
+  finding 4's source gaps stay with #14116.
+- No sharding of the bootstrap lane, no shorter broker channel expiry and no
+  producer buffer in this phase.
+- Workspace-bound Sessions, tools and the runtime broker are not exercised:
+  the rows use plain public Sessions, which keeps the gate portable. The
+  public lane's duplicate and reorder rows stay with MQ2d's single-node gate,
+  and the concurrent claim race with `EventTransportConsumerSqlTest`.
+
+## Validation
+
+- `EventTransportMultiNodeGate` runs locally (macOS, MySQL 8.4.7, RocketMQ
+  5.5.1), where the clock-skew row is skipped for want of libfaketime, and in
+  a new job of the SDK Java workflow (ubuntu-latest, a MySQL 8.4.6 service,
+  libfaketime), which runs every row and fails on a skipped one. Like the
+  rest of that workflow the job runs daily and on demand, not per pull
+  request (#14205). Before this change was proposed the job's definition ran
+  on a GitHub-hosted ubuntu-latest runner from a fork: all ten rows passed,
+  the clock-skew row included.
+- Mutation checks: each row is run against a deliberately broken build and
+  must fail — the clock-skew row's on that Linux runner; the pull request
+  lists the mutants and their results.
+- `scripts/tests/event-transport-multi-node-ci.test.js` pins the job, the
+  profile, the broker provisioning, the gate's required properties and its
+  timeout budget to one another; the failsafe report check's
+  `event-transport-multi-node` family fails the job unless the gate class ran
+  in full.
--- a/docs/design/2026-10-14-managed-agent-event-transport-mq3.zh-CN.md
+++ b/docs/design/2026-10-14-managed-agent-event-transport-mq3.zh-CN.md
@ -0,0 +1,75 @@
+# EventTransport MQ3：部署故障矩阵
+
+[English](2026-10-14-managed-agent-event-transport-mq3.md) | [简体中文](2026-10-14-managed-agent-event-transport-mq3.zh-CN.md)
+
+> **状态：** 已作为门禁实现。这是跨节点 EventTransport 设计
+> （[2026-10-04-managed-agent-event-transport](2026-10-04-managed-agent-event-transport.zh-CN.md)
+> §7）的 **MQ3** 阶段，由 [#12380](https://github.com/QwenLM/qwen-code/issues/12380) 跟踪。
+> 它的前置阶段 MQ2 已分四个切片落地
+> （[2026-10-12-managed-agent-event-transport-mq2](2026-10-12-managed-agent-event-transport-mq2.zh-CN.md)）：
+> MQ2a [#13940](https://github.com/QwenLM/qwen-code/pull/13940)、MQ2b
+> [#14029](https://github.com/QwenLM/qwen-code/pull/14029)、MQ2c
+> [#14080](https://github.com/QwenLM/qwen-code/pull/14080) 与 MQ2d
+> [#14109](https://github.com/QwenLM/qwen-code/pull/14109)。MQ3 不改任何生产代码。
+> 它的各行断言的是现有实现的行为；与 §8「要求的行为」不一致的地方，由下文的「发现」记录。
+
+## 问题
+
+MQ2 列出的每一行都是对着单个应用跑的：H2 数据库，Harness 用替身。父设计 §8 的矩阵讲的是部署：不止一个控制面节点、进程在尴尬的时刻死掉、broker 不再应答、时钟彼此不一致、开关需要关掉。这些行只有对着真实进程跑才有意义。MQ3 用两个控制面节点来跑它们：每个节点各带一个打包好的 Hosted Harness，共用一个 MySQL 数据库和一个 RocketMQ 5.5 LiteTopic 集群。它逐行记录现有 transport 的实际行为，包括与 §8「要求的行为」文字不一致的地方。
+
+## 现状
+
+2026-10-14 在 `a6af39188a` 上核对。
+
+- **拓扑。** 每个节点都加入共享的唤醒 bootstrap 组，订阅唯一的 bootstrap lite topic；也都加入共享的 owner 组，订阅自己 Harness 写入的每个 Session 的 lite topic。监听器在完成 SQL 检查与认领之后返回（客户端随即确认），加载则排进一个小线程池。
+- **生产者。** 两条线都异步发送，并捕获一切失败。唯一的重试是客户端自带的（总共三次尝试，请求超时三秒）；没有应用层重试，也没有故障期缓冲。journal 的交接在提交线程上执行；公开事件的交接把批次复制进一个 1024 批的队列，同样在提交线程上。
+- **启动。** 能力探针、发送探针、owner 消费者和 bootstrap 订阅都需要可达的 broker；任何一项失败都会让 Spring 上下文启动失败。
+- **时钟。** 唤醒判断与 journal writer 租约读数据库时钟。Turn 调度租约与各 relay 账本的租约用 JVM 时钟打时间戳。
+- **围栏。** 唤醒不携带 epoch。在每个唤醒都要过的检查（Session 存在、workspace 与 journal 一致、序号已提交、Session 与 journal 未终结）之后，过期唤醒只会被活着的 writer 或认领水位拒绝。
+
+## 设计：门禁
+
+`EventTransportMultiNodeGate`（profile `event-transport-multi-node-gates`，SDK Java 任务「EventTransport multi-node gates / RocketMQ 5.5.1 + MySQL 8.4 / Java 21」，failsafe 家族 `event-transport-multi-node`）。
+
+- **节点**：一个打包好的 Hosted Harness（CLI 的 hosted-harness serve profile，模型用假实现），加上在独立 JVM 里运行的完整应用，经 `EventTransportNodeMain` 启动。每一行都起自己的节点，writer 租约 6 秒、Turn 调度租约 6 秒，让死掉节点的占有在本行之内就过期。同一个类的所有节点共用一个隔离的 MySQL 数据库；每一行用全新的 tenant。崩溃即用 SIGKILL 杀掉该节点的 JVM 与 Harness；重启时节点保留原地址。
+- **故障点**：节点 JVM 用仅供测试的子类替换生产 bean：唤醒激活器（收到唤醒即停机；认领之后、监听器返回之前停机）、journal 线（带唤醒的记录提交之后、交给 broker 之前停机）、公开事件线（携带 `turn.completed` 的那一批提交之后、交接之前停机）。故障按 Session 由一个文件布防，在两个节点之间只触发一次；触发即调用 `Runtime.halt`，因此不会确认、不会刷盘、也不会走关闭流程。hold-load 故障则不同：它在每个节点上压住该 Session 的所有加载，直到门禁放行。broker 故障用 SIGSTOP 冻结 gRPC proxy，冻结前先核实记录的 pid 正是监听该端点的进程。时钟偏差这一行用 libfaketime（Linux，CI）平移应用 JVM 的墙钟，并检查它确实平移了；节点的 Harness 仍用宿主时钟。没有 libfaketime 时该行跳过：唤醒路径不读任何 Clock bean，除了 JVM 自己的时钟，没有别的办法测它。
+- **取证**：每个节点把自己做出的每个唤醒判断、加载结果和交接追加写入各自的证据文件。门禁对它、对 SQL（`ma_session_head` 的 writer、代数与租约，`ma_event_stream` 的水位，已提交的 journal）和对公开 API 做断言，并通过自己的 oracle 组读取到达 bootstrap 通道的内容——某一行需要时，也读该 Session 自己的通道。每一行的日志与证据随测试报告上传。每次等待都会在本行超时触发时立即失败，各行的超时之和也在 fork 的超时之内。broker 必须由这一次运行独占。
+
+## 各行
+
+| §8 故障                                          | 注入方式                                                                                                                                   | 门禁断言                                                                                                                                                                                                                                                   |
+| ------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| 重复投递、乱序、ACK 丢失后的再投递               | 两个节点都在 bootstrap 组里；某 Session 的 writer 过期后，把它已提交的唤醒重新发出：较新的两次，然后较旧的一次；加载被压住，直到三份都判完 | 无论落在哪个节点，都是 `SCHEDULED`、`DUPLICATE`、`DUPLICATE`；水位 = 较新的序号；随后经真实 Harness 加载一次 `ACTIVATED`；writer 代数 +1，writer 是激活节点的 Harness                                                                                      |
+| 消费者处理前崩溃                                 | 收到唤醒的节点在读任何东西之前停机                                                                                                         | 没有认领；broker 把唤醒交给另一个节点，由它把 Session 激活一次                                                                                                                                                                                             |
+| 消费者本地生效后、ACK 前崩溃                     | 收到唤醒的节点完成认领，并在监听器返回之前停机                                                                                             | 水位保留这次认领；再投递到达另一个节点，结果为 `DUPLICATE`；那里不加载；新的输入仍在那里正常执行                                                                                                                                                           |
+| SQL commit 与 publish 之间生产者崩溃（journal）  | Harness 提交 Turn 输入的那个节点在提交之后、发出唤醒之前停机                                                                               | 唤醒已提交，却既没到达 bootstrap 通道也没到达 Session 的通道，也从未被认领；Turn 调度器在另一个节点上把 Turn 推进到终态，只一次                                                                                                                            |
+| SQL commit 与 publish 之间生产者崩溃（公开事件） | 记录 Turn 最后一批事件的节点在提交它之后、交接之前停机                                                                                     | 之后不会再有通知，只有另一个节点的定时扫描能投影它：已提交的 `turn.completed` 照常可读，日志没有缺口，`covered_sequence = last_sequence`，items 里有输入与回复                                                                                             |
+| broker 短时故障                                  | 在一个 Turn 运行期间冻结 proxy                                                                                                             | Turn 正常完成；提交线程上的两种交接都保持在 250 ms 以内；proxy 恢复后，下一个 Turn 的唤醒到达 broker                                                                                                                                                       |
+| broker 长时故障                                  | 在冻结 proxy 期间重启一个节点                                                                                                              | 确认发送已经开始失败之后，运行中的节点仍在服务 Turn——既有已持有 Session 的，也有新 Session 的第一个 Turn——两种交接都在 250 ms 以内；开关开着重启的节点拒绝启动；同一节点（此时只剩它）关着开关重启后能启动并服务；故障结束后，开着开关重启，它重新承载唤醒 |
+| Java 节点间时钟偏移                              | 一个节点的 JVM 墙钟快 10 分钟，随后另一个节点慢 10 分钟                                                                                    | 快的节点仍把另一个节点的活 writer 判为 `OWNED`，不认领；慢的节点仍把过期的 writer 判为过期，在本节点激活 Session，且它得到的租约按数据库时间仍然有效                                                                                                       |
+| Session 改路由后的过期唤醒                       | Session 转到另一个节点的 Harness；重发它的旧唤醒                                                                                           | 无论落在哪个节点都是 `OWNED`；水位始终为空，不加载，writer 不变                                                                                                                                                                                            |
+| 开关回滚                                         | 两个节点先以开关关闭重启，再以开关打开重启                                                                                                 | 关闭时：Turn 照常执行，两条通道都收不到任何内容，期间发出的唤醒也无人消费；重新打开后：那个唤醒被消费，下一个唤醒照常发布                                                                                                                                  |
+
+## 发现
+
+发现 1–3、5 与 6 是现有实现与 §8 文字不一致的地方；4 与 7 是各行所依赖的现有特性。
+
+1. **节点在处理唤醒时崩溃，会让 bootstrap 通道停顿约两分钟。** LiteTopic 只有在 broker 让死掉的客户端过期（它的连接过期时限，默认两分钟）之后，才把崩溃成员未确认的消息交给另一个成员，而该 lite topic 后续的消息都按顺序排在它后面。本机用同组两个成员的探针实测：被占住的消息 124 秒后才到达另一个成员，崩溃 5 秒后发出的消息与它一起到达；门禁两条崩溃行的各次本机运行中，唤醒在 121–129 秒后转移。在共享的 bootstrap 通道上，这意味着全集群所有 Session 的唤醒。没有消息丢失——每个唤醒只是晚到——也没有任何输入账本依赖这条通道。被杀时手上没有未确认消息的节点不会占住通道：下一个唤醒在该行的等待时限内就到达了另一个节点。（§8：「broker 再投递；Session 被晚唤醒」——属实，但晚是以分钟计，而且是所有 Session。）
+2. **没有生产者重试缓冲。** 面对挂住的 proxy（用 SIGSTOP 冻结），客户端自带的尝试（每次三秒）能扛过几秒钟的停顿——本机的短时故障运行中（冻结持续一个 Turn）一条都没丢——但 proxy 离开更久时发出的信封会被丢弃，并打一条注明事件的 `WARN`，之后也不会补发（本机一次长时故障行的运行中丢了 13 条 journal 信封和 4 条公开信封）。提交路径不受影响：冻结期间，提交线程上的两种交接都只有个位数毫秒。（§8：「生产者以有界缓冲重试」——未实现；故障期间丢失的唤醒靠它的账本恢复。）
+3. **启动依赖 broker。** broker 不可用时重启的节点在能力探针上拒绝启动，而报错开头讲的是 LITE group 与 topic 的要求，端点配置项只在最后才提到；回退的办法是开关。长时故障行断言了这次拒绝，修好之后要反过来改。（§8：「传输消费者空转」只对已经在运行的节点成立。）
+4. **认领从不回滚。** 认领之后崩溃会烧掉这条通知；Session 靠其输入来源的账本恢复，各来源的缺口记录在 [#14116](https://github.com/QwenLM/qwen-code/issues/14116)。
+5. **没有 epoch 围栏。** 过期唤醒被活 writer 或水位拒绝；落在空闲 Session 上且序号更高时会加载它——无害，因为唤醒泵只执行 journal 里已有的内容。
+6. **Turn 调度租约用的是 JVM 时钟。** 时钟快出超过一个调度租约时长的节点会把活着的占有读成已过期并接管 Turn；本机一次节点快 10 分钟的运行中，它接管了一个活着的 Turn，每次尝试都撞上持有 Harness 的 writer 租约，重试耗尽后把这个 Turn 判为失败。这不在 transport 范围内——时钟偏差行让 Turn 不经过偏差节点——之所以记在这里，是因为 §8 声称所有租约都沿用本仓的租约惯例。
+7. **Harness 停在模型调用处的 Turn 不会被接管。** journal 生产者崩溃行的各次运行中，Turn 在另一个节点上被判为失败（`managed_runtime_recovery_blocked`，原因 `model_start`）：这是 Harness 根据 journal 做出的拒绝，transport 不参与。该行只断言 Turn 被推进到终态且只一次。
+
+## 范围边界
+
+- 不改生产代码。发现 1–3 与 6 记录在此，留待后续；发现 4 的各来源缺口仍归 #14116。
+- 本阶段不对 bootstrap 通道分片，不缩短 broker 的连接过期时限，也不加生产者缓冲。
+- 不覆盖绑定 Workspace 的 Session、工具与 runtime broker：各行用普通公开 Session，让门禁保持可移植。公开事件通道的重复与乱序行仍归 MQ2d 的单节点门禁，并发认领竞争仍归 `EventTransportConsumerSqlTest`。
+
+## 验证
+
+- `EventTransportMultiNodeGate` 在本机（macOS、MySQL 8.4.7、RocketMQ 5.5.1）运行，其中时钟偏差行因缺少 libfaketime 而跳过；SDK Java 工作流中的新任务（ubuntu-latest、MySQL 8.4.6 服务、libfaketime）运行每一行，任何一行被跳过都会失败。与该工作流的其余任务一样，它每天定时运行，也可手动触发，不随每个 PR 运行（#14205）。提出本变更之前，该任务的定义已在 fork 上借 GitHub 托管的 ubuntu-latest runner 跑过一次：十行全部通过，时钟偏差行也在内。
+- 变异检查：每一行都要对着一个故意改坏的构建运行并失败——时钟偏差行的变异在那台 Linux runner 上跑；PR 列出各变异体及其结果。
+- `scripts/tests/event-transport-multi-node-ci.test.js` 把任务、profile、broker 准备工作、门禁所需的属性及其超时预算彼此钉住；failsafe 报告检查的 `event-transport-multi-node` 家族在门禁类没有完整运行时让任务失败。
--- a/packages/sdk-java/managed-agent-server/pom.xml
+++ b/packages/sdk-java/managed-agent-server/pom.xml
@ -410,5 +410,36 @@
                </plugin>
            </plugins></build>
        </profile>
+        <profile>
+            <!-- The MQ3 deployment fault matrix
+                 (docs/design/2026-10-14-managed-agent-event-transport-mq3.md):
+                 two control-plane nodes, each with its own packaged Harness,
+                 against one MySQL server and a RocketMQ 5.5 LiteTopic cluster
+                 (scripts/rocketmq-lite-broker.sh start). The gate starts the
+                 nodes itself; the MySQL server (mysql.url / user / password),
+                 the bundled CLI (qwen.cli.entry, node.executable) and the
+                 broker coordinates (qwen.gate.rocketmq.endpoints / topic /
+                 work-dir) are required system properties. -->
+            <id>event-transport-multi-node-gates</id>
+            <build><plugins>
+                <plugin>
+                    <groupId>org.apache.maven.plugins</groupId>
+                    <artifactId>maven-surefire-plugin</artifactId>
+                    <configuration><skipTests>true</skipTests></configuration>
+                </plugin>
+                <plugin>
+                    <groupId>org.apache.maven.plugins</groupId>
+                    <artifactId>maven-failsafe-plugin</artifactId>
+                    <version>${maven-failsafe-plugin.version}</version>
+                    <configuration>
+                        <includes><include>**/EventTransportMultiNodeGate.java</include></includes>
+                        <failIfNoTests>true</failIfNoTests>
+                        <failIfNoSpecifiedTests>true</failIfNoSpecifiedTests>
+                        <forkedProcessTimeoutInSeconds>3000</forkedProcessTimeoutInSeconds>
+                    </configuration>
+                    <executions><execution><goals><goal>integration-test</goal><goal>verify</goal></goals></execution></executions>
+                </plugin>
+            </plugins></build>
+        </profile>
    </profiles>
 </project>
--- a/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/EventTransportMultiNodeGate.java
+++ b/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/EventTransportMultiNodeGate.java
--- a/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/service/EventTransportNodeMain.java
+++ b/packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/service/EventTransportNodeMain.java
@ -0,0 +1,342 @@
+package com.alibaba.qwen.code.managedagent.service;
+
+import com.alibaba.qwen.code.managedagent.ManagedAgentServerApplication;
+import com.alibaba.qwen.code.managedagent.config.ManagedAgentProperties;
+import com.alibaba.qwen.code.managedagent.harness.HarnessConnector;
+import com.alibaba.qwen.code.managedagent.store.ManagedEventEnvelope;
+import com.alibaba.qwen.code.managedagent.store.ManagedEventEnvelopes;
+import com.alibaba.qwen.code.managedagent.store.StoreModels.EventRecord;
+import com.alibaba.qwen.code.managedagent.store.TransportConsumerStore;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.client.apis.ClientServiceProvider;
+import org.apache.rocketmq.client.apis.producer.Producer;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext;
+
+/**
+ * One control-plane node of the MQ3 multi-node gate
+ * (EventTransportMultiNodeGate): the whole application in its own JVM, so a
+ * fault can end the process the way a crash does. Test-only hooks stand at
+ * the points the §8 rows name — a wake received, a wake claimed, a committed
+ * record handed to either wire — and fire only when the gate arms them with a
+ * file under the shared directory, at most once across every node; a
+ * hold-load fault instead parks every load of its Session, on every node,
+ * until the gate releases it. Each wake
+ * decision, load and hand-off this node makes is appended to its evidence
+ * file, so the gate reads what each node did instead of inferring it.
+ *
+ * <p>The single argument is a JSON file: {@code node}, {@code root},
+ * {@code shared}, {@code ready} and the Spring {@code properties}. A clock
+ * skew is applied to the whole JVM from outside (libfaketime), never here:
+ * the wake path reads the database clock and no Clock bean.
+ */
+public final class EventTransportNodeMain {
+    /** The exit status of a fault-injected crash, distinct from a kill. */
+    public static final int HALTED = 86;
+
+    private EventTransportNodeMain() {
+    }
+
+    public static void main(String[] args) throws Exception {
+        ObjectMapper json = new ObjectMapper();
+        JsonNode config = json.readTree(Files.readString(Path.of(args[0])));
+        String node = config.path("node").asText();
+        Path root = Path.of(config.path("root").asText());
+        Faults faults = new Faults(node, Path.of(config.path("shared").asText()),
+                root.resolve("evidence.jsonl"));
+        List<String> arguments = new ArrayList<>();
+        config.path("properties").fields().forEachRemaining(entry ->
+                arguments.add("--" + entry.getKey() + "=" + entry.getValue().asText()));
+        var spring = (ServletWebServerApplicationContext) new SpringApplicationBuilder(
+                ManagedAgentServerApplication.class)
+                .initializers(context -> context.getBeanFactory().addBeanPostProcessor(
+                        new Hooks(context.getBeanFactory(), faults)))
+                .run(arguments.toArray(String[]::new));
+        faults.record(Map.of("event", "ready"));
+        Path ready = Path.of(config.path("ready").asText());
+        Path writing = ready.resolveSibling(ready.getFileName() + ".tmp");
+        json.writeValue(writing.toFile(), Map.of("pid", ProcessHandle.current().pid(),
+                "port", spring.getWebServer().getPort()));
+        Files.move(writing, ready, StandardCopyOption.ATOMIC_MOVE,
+                StandardCopyOption.REPLACE_EXISTING);
+    }
+
+    /**
+     * The armed faults and this node's evidence. A fault is armed for a
+     * Session by the file {@code shared/faults/<rule>/<sessionId>}; it fires on
+     * the first node to create {@code shared/fired/<rule>-<sessionId>}.
+     */
+    static final class Faults {
+        private static final ObjectMapper JSON = new ObjectMapper();
+        private final String node;
+        private final Path shared;
+        private final Path evidence;
+
+        Faults(String node, Path shared, Path evidence) {
+            this.node = node;
+            this.shared = shared;
+            this.evidence = evidence;
+        }
+
+        boolean armed(String rule, String sessionId) {
+            return sessionId != null
+                    && Files.exists(shared.resolve("faults").resolve(rule)
+                            .resolve(sessionId));
+        }
+
+        /** Whether any node has already fired the rule for the Session. */
+        boolean fired(String rule, String sessionId) {
+            return Files.exists(shared.resolve("fired")
+                    .resolve(rule + "-" + sessionId));
+        }
+
+        /** Blocks until the gate releases the Session's held loads. */
+        void awaitRelease(String sessionId) {
+            Path release = shared.resolve("release").resolve(sessionId);
+            while (!Files.exists(release)) {
+                try {
+                    Thread.sleep(50);
+                } catch (InterruptedException interrupted) {
+                    Thread.currentThread().interrupt();
+                    return;
+                }
+            }
+        }
+
+        /** Halts this JVM if the rule is armed and no node has fired it yet. */
+        void haltIfArmed(String rule, String sessionId) {
+            haltIfArmed(rule, sessionId, Map.of());
+        }
+
+        void haltIfArmed(String rule, String sessionId, Map<String, Object> detail) {
+            if (!armed(rule, sessionId)) {
+                return;
+            }
+            try {
+                Files.createFile(shared.resolve("fired")
+                        .resolve(rule + "-" + sessionId));
+            } catch (FileAlreadyExistsException firedElsewhere) {
+                return;
+            } catch (IOException unwritable) {
+                throw new UncheckedIOException(unwritable);
+            }
+            Map<String, Object> line = new LinkedHashMap<>(detail);
+            line.put("event", "halt");
+            line.put("rule", rule);
+            line.put("session", sessionId);
+            record(line);
+            // No shutdown hook, no acknowledgement, no flush: a crash.
+            Runtime.getRuntime().halt(HALTED);
+        }
+
+        synchronized void record(Map<String, Object> fields) {
+            Map<String, Object> line = new LinkedHashMap<>();
+            line.put("node", node);
+            line.put("at", System.currentTimeMillis());
+            line.putAll(fields);
+            try {
+                Files.writeString(evidence, JSON.writeValueAsString(line) + "\n",
+                        StandardCharsets.UTF_8, StandardOpenOption.CREATE,
+                        StandardOpenOption.APPEND, StandardOpenOption.SYNC);
+            } catch (IOException unwritable) {
+                throw new UncheckedIOException(unwritable);
+            }
+        }
+    }
+
+    /** Swaps the hooked subclasses in for the production transport beans. */
+    private static final class Hooks implements BeanPostProcessor {
+        private final ConfigurableListableBeanFactory beans;
+        private final Faults faults;
+
+        Hooks(ConfigurableListableBeanFactory beans, Faults faults) {
+            this.beans = beans;
+            this.faults = faults;
+        }
+
+        @Override
+        public Object postProcessAfterInitialization(Object bean, String name) {
+            if (bean instanceof TransportWakeActivator original
+                    && !(bean instanceof HookedWakes)) {
+                original.close();
+                OwnerSubscriptions owners = beans.getBean(OwnerSubscriptions.class);
+                return new HookedWakes(beans.getBean(TransportConsumerStore.class),
+                        beans.getBean(HarnessConnector.class), owners::subscribe, faults);
+            }
+            if (bean instanceof LiteTopicEventTransport
+                    && !(bean instanceof HookedJournalWire)) {
+                var config = beans.getBean(ManagedAgentProperties.class)
+                        .getEventTransport();
+                return new HookedJournalWire(beans.getBean(Producer.class),
+                        beans.getBean(ClientServiceProvider.class), config.getTopic(),
+                        config.getWakeBootstrapLiteTopic(), faults);
+            }
+            if (bean instanceof LiteTopicPublicEventTransport original
+                    && !(bean instanceof HookedPublicWire)) {
+                original.close();
+                TransportConsumerStore store = beans.getBean(TransportConsumerStore.class);
+                return new HookedPublicWire(beans.getBean(Producer.class),
+                        beans.getBean(ClientServiceProvider.class),
+                        beans.getBean(ManagedAgentProperties.class).getEventTransport()
+                                .getTopic(),
+                        beans.getBean(ObjectMapper.class), store, faults);
+            }
+            return bean;
+        }
+    }
+
+    private static final class HookedWakes extends TransportWakeActivator {
+        private final Faults faults;
+        // Sessions whose load this JVM must not start: its listener is about
+        // to halt it after the claim. Local, so another node's load of the
+        // same Session — after the fault fired — runs and is recorded.
+        private final java.util.Set<String> halting =
+                java.util.concurrent.ConcurrentHashMap.newKeySet();
+
+        HookedWakes(TransportConsumerStore store, HarnessConnector harness,
+                java.util.function.Consumer<String> activated, Faults faults) {
+            super(store, harness, activated);
+            this.faults = faults;
+        }
+
+        @Override
+        public Outcome onWake(ManagedEventEnvelope envelope) {
+            String session = envelope.sessionId();
+            boolean wake = ManagedEventEnvelopes.STREAM_AUTHORITATIVE_JOURNAL.equals(
+                    envelope.stream()) && "wake.requested".equals(envelope.kind());
+            if (wake) {
+                // §8 consumer crash before processing: nothing read, nothing
+                // claimed, nothing acknowledged.
+                faults.haltIfArmed("halt-on-receive", session);
+            }
+            boolean haltAfterClaim = wake && faults.armed("halt-after-claim", session)
+                    && !faults.fired("halt-after-claim", session);
+            if (haltAfterClaim) {
+                halting.add(session);
+            }
+            Outcome outcome = super.onWake(envelope);
+            if (wake) {
+                faults.record(Map.of("event", "wake", "session", session,
+                        "sequence", envelope.sequence(), "outcome", outcome.name()));
+            }
+            if (haltAfterClaim && outcome == Outcome.SCHEDULED) {
+                // §8 consumer crash after a local effect: the claim committed,
+                // the listener has not returned, so nothing is acknowledged.
+                faults.haltIfArmed("halt-after-claim", session);
+            }
+            halting.remove(session);
+            return outcome;
+        }
+
+        @Override
+        LoadOutcome load(ManagedEventEnvelope envelope) {
+            String session = envelope.sessionId();
+            if (halting.contains(session)) {
+                // This JVM's listener is halting it; the load must not race
+                // it to the Harness.
+                try {
+                    Thread.sleep(Long.MAX_VALUE);
+                } catch (InterruptedException interrupted) {
+                    Thread.currentThread().interrupt();
+                }
+                return LoadOutcome.FAILED;
+            }
+            if (faults.armed("hold-load", session)) {
+                // Every copy the gate sends is decided before the Session
+                // activates, so a live writer cannot stand in for the claim.
+                faults.awaitRelease(session);
+            }
+            LoadOutcome outcome = super.load(envelope);
+            faults.record(Map.of("event", "load", "session", envelope.sessionId(),
+                    "sequence", envelope.sequence(), "outcome", outcome.name()));
+            return outcome;
+        }
+    }
+
+    private static final class HookedJournalWire extends LiteTopicEventTransport {
+        private static final ObjectMapper JSON = new ObjectMapper();
+        private final Faults faults;
+
+        HookedJournalWire(Producer producer, ClientServiceProvider provider,
+                String topic, String bootstrapLiteTopic, Faults faults) {
+            super(producer, provider, topic, bootstrapLiteTopic);
+            this.faults = faults;
+        }
+
+        @Override
+        public void publishCommitted(byte[] recordBytes) {
+            String session = null;
+            List<Long> wakes = new ArrayList<>();
+            for (String line : new String(recordBytes, StandardCharsets.UTF_8)
+                    .split("\n")) {
+                try {
+                    JsonNode event = JSON.readTree(line).path("managedSession");
+                    if (event.path("sessionKey").path("sessionId").isTextual()) {
+                        session = event.path("sessionKey").path("sessionId").asText();
+                    }
+                    if ("wake.requested".equals(event.path("kind").asText())) {
+                        wakes.add(event.path("sequence").asLong());
+                    }
+                } catch (IOException notJson) {
+                    // The wire itself reports a record it cannot read.
+                }
+            }
+            if (!wakes.isEmpty()) {
+                // §8 producer crash between the SQL commit and the publish:
+                // the record carrying the wake is committed and never sent.
+                faults.haltIfArmed("halt-on-journal-wake", session,
+                        Map.of("wakes", wakes));
+            }
+            long started = System.nanoTime();
+            super.publishCommitted(recordBytes);
+            faults.record(Map.of("event", "journal-handoff",
+                    "session", String.valueOf(session), "wakes", wakes,
+                    "micros", (System.nanoTime() - started) / 1000));
+        }
+    }
+
+    private static final class HookedPublicWire extends LiteTopicPublicEventTransport {
+        private final Faults faults;
+
+        HookedPublicWire(Producer producer, ClientServiceProvider provider, String topic,
+                ObjectMapper objectMapper, TransportConsumerStore store, Faults faults) {
+            super(producer, provider, topic, objectMapper, store::workspaceOf);
+            this.faults = faults;
+        }
+
+        @Override
+        public void publish(List<EventRecord> events) {
+            String session = events.isEmpty() ? null : events.getFirst().sessionId();
+            for (EventRecord event : events) {
+                if ("turn.completed".equals(event.type())) {
+                    // §8 producer crash between the SQL commit and the
+                    // publish: the Turn's last batch is committed and never
+                    // handed on, so no later notice of the Session follows.
+                    faults.haltIfArmed("halt-on-public-terminal", session,
+                            Map.of("eventId", event.eventId(),
+                                    "sequence", event.sequence()));
+                }
+            }
+            long started = System.nanoTime();
+            super.publish(events);
+            faults.record(Map.of("event", "public-handoff",
+                    "session", String.valueOf(session),
+                    "micros", (System.nanoTime() - started) / 1000));
+        }
+    }
+}
--- a/scripts/check-failsafe-reports.js
+++ b/scripts/check-failsafe-reports.js
@ -23,7 +23,10 @@ const isHosted = (className) => /^Hosted.*IT$/.test(className);
 const isGate = (className) => /^Hosted.*Gate$/.test(className);
 const [family, ...modules] = process.argv.slice(2);
 const optIn =
-  family === 'o4-mysql' || family === 'o4-oss' || family === 'event-transport';
+  family === 'o4-mysql' ||
+  family === 'o4-oss' ||
+  family === 'event-transport' ||
+  family === 'event-transport-multi-node';
 const inFamily = new Map([
  ['hosted', isHosted],
  ['non-hosted', (className) => /IT$/.test(className) && !isHosted(className)],
@ -31,10 +34,14 @@ const inFamily = new Map([
  ['o4-mysql', (className) => className === 'O4MySqlGate'],
  ['o4-oss', (className) => className === 'O4OssGate'],
  ['event-transport', (className) => className === 'EventTransportBrokerGate'],
+  [
+    'event-transport-multi-node',
+    (className) => className === 'EventTransportMultiNodeGate',
+  ],
 ]).get(family);
 if (!inFamily || modules.length === 0) {
  console.error(
-    'usage: node scripts/check-failsafe-reports.js hosted|non-hosted|hosted-gates|o4-mysql|o4-oss|event-transport <maven-module-dir>...',
+    'usage: node scripts/check-failsafe-reports.js hosted|non-hosted|hosted-gates|o4-mysql|o4-oss|event-transport|event-transport-multi-node <maven-module-dir>...',
  );
  process.exit(2);
 }
--- a/scripts/tests/check-failsafe-reports.test.js
+++ b/scripts/tests/check-failsafe-reports.test.js
@ -69,6 +69,8 @@ describe('check-failsafe-reports', () => {
    ['o4-oss', 'O4OssGate', 'error'],
    ['event-transport', 'EventTransportBrokerGate', 'failure'],
    ['event-transport', 'EventTransportBrokerGate', 'error'],
+    ['event-transport-multi-node', 'EventTransportMultiNodeGate', 'failure'],
+    ['event-transport-multi-node', 'EventTransportMultiNodeGate', 'error'],
  ])('rejects %s %s %s even with Maven failure-ignore', (family, gate, tag) => {
    const className = `com.example.${gate}`;
    const dir = module('failed', [className]);
@ -85,6 +87,7 @@ describe('check-failsafe-reports', () => {
    ['o4-mysql', 'O4MySqlGate'],
    ['o4-oss', 'O4OssGate'],
    ['event-transport', 'EventTransportBrokerGate'],
+    ['event-transport-multi-node', 'EventTransportMultiNodeGate'],
  ])(
    'requires complete opt-in %s evidence without changing the ordinary family',
    (family, gate) => {
@ -93,6 +96,7 @@ describe('check-failsafe-reports', () => {
        'com.example.O4MySqlGate',
        'com.example.O4OssGate',
        'com.example.EventTransportBrokerGate',
+        'com.example.EventTransportMultiNodeGate',
      ];
      const complete = module('complete', sources, {
        [`com.example.${gate}`]: [3, 0],
--- a/scripts/tests/event-transport-multi-node-ci.test.js
+++ b/scripts/tests/event-transport-multi-node-ci.test.js
@ -0,0 +1,171 @@
+/**
+ * @license
+ * Copyright 2026 Qwen Team
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+// The MQ3 multi-node gates span the SDK Java workflow job, the
+// event-transport-multi-node-gates Maven profile, the broker script and the
+// gate class. A name, a path or a budget that drifts in one of them fails the
+// gate only on CI, after a broker boot and four process starts, or not at all.
+
+import { readFileSync } from 'node:fs';
+import { parse } from 'yaml';
+import { describe, expect, it } from 'vitest';
+
+const workflow = parse(readFileSync('.github/workflows/sdk-java.yml', 'utf8'));
+const job = workflow.jobs['event-transport-multi-node'];
+const step = (name) => {
+  const found = job.steps.find((candidate) => candidate.name === name);
+  if (!found) throw new Error(`Missing workflow step: ${name}`);
+  return found;
+};
+const verify = () => step('Verify the EventTransport multi-node gates');
+const pom = readFileSync(
+  'packages/sdk-java/managed-agent-server/pom.xml',
+  'utf8',
+);
+const profile = pom
+  .split('<id>event-transport-multi-node-gates</id>')[1]
+  .split('</profile>')[0];
+const server = 'packages/sdk-java/managed-agent-server/src';
+const properties = readFileSync(
+  `${server}/main/java/com/alibaba/qwen/code/managedagent/config/ManagedAgentProperties.java`,
+  'utf8',
+);
+const gate = readFileSync(
+  `${server}/test/java/com/alibaba/qwen/code/managedagent/EventTransportMultiNodeGate.java`,
+  'utf8',
+);
+const script = readFileSync('scripts/rocketmq-lite-broker.sh', 'utf8');
+
+const gateConstant = (name) =>
+  new RegExp(`static final String ${name} = "([^"]+)";`).exec(gate)?.[1];
+const scriptDefault = (name) =>
+  new RegExp(`^${name}="\\$\\{${name}:-([^}]*)\\}"$`, 'm').exec(script)?.[1];
+
+describe('EventTransport multi-node gates CI', () => {
+  it('runs the gate profile with every coordinate the gate requires', () => {
+    const run = verify().run;
+    expect(run).toContain('-Pevent-transport-multi-node-gates');
+    const required = /for \(String name : List\.of\(([^)]*)\)\)/
+      .exec(gate)[1]
+      .match(/"[^"]+"/g)
+      .map((quoted) => quoted.slice(1, -1));
+    expect(required.length).toBeGreaterThan(0);
+    for (const property of required) {
+      expect(run).toContain(`-D${property}=`);
+    }
+    for (const property of [
+      'qwen.gate.rocketmq.work-dir',
+      'qwen.gate.faketime.library',
+      'node.executable',
+      'mysql.password',
+    ]) {
+      expect(run).toContain(`-D${property}=`);
+    }
+    expect(run).toContain(
+      'node scripts/check-failsafe-reports.js event-transport-multi-node packages/sdk-java/managed-agent-server',
+    );
+  });
+
+  it('runs only the gate class, never the unit suite, and fails on no tests', () => {
+    expect(profile).toContain(
+      '<include>**/EventTransportMultiNodeGate.java</include>',
+    );
+    expect(profile.match(/<include>/g)).toHaveLength(1);
+    expect(profile).toContain('<failIfNoTests>true</failIfNoTests>');
+    expect(profile).toMatch(
+      /<artifactId>maven-surefire-plugin<\/artifactId>\s*<configuration><skipTests>true<\/skipTests>/,
+    );
+  });
+
+  it('lets the gate step outlive its failsafe fork, and the job its steps', () => {
+    const fork = Number(
+      /<forkedProcessTimeoutInSeconds>(\d+)</.exec(profile)[1],
+    );
+    expect(verify()['timeout-minutes'] * 60).toBeGreaterThan(fork);
+    const capped = job.steps
+      .map((candidate) => candidate['timeout-minutes'] ?? 0)
+      .reduce((sum, minutes) => sum + minutes, 0);
+    expect(job['timeout-minutes']).toBeGreaterThan(capped);
+  });
+
+  // Awaitility ignores the interrupt a row's @Timeout sends, so the gate's
+  // waits fail fast on it; the rows' ceilings must then fit the fork, or a
+  // stalled class ends as a fork timeout with no per-row result (#13503).
+  it("keeps the rows' timeouts within the failsafe fork", () => {
+    const fork = Number(
+      /<forkedProcessTimeoutInSeconds>(\d+)</.exec(profile)[1],
+    );
+    const rows = [...gate.matchAll(/@Timeout\((\d+)\)/g)].map((match) =>
+      Number(match[1]),
+    );
+    expect(rows.length).toBe((gate.match(/@Test\b/g) ?? []).length);
+    expect(rows.reduce((sum, seconds) => sum + seconds, 0)).toBeLessThan(fork);
+    expect(gate).toContain('failFast(');
+  });
+
+  it('always stops the broker it started and keeps its evidence', () => {
+    for (const name of [
+      'Upload multi-node gate reports',
+      'Stop the RocketMQ cluster',
+    ]) {
+      expect(step(name).if).toContain('always()');
+      expect(step(name).if).toContain("steps.broker.outcome != 'skipped'");
+    }
+    expect(step('Stop the RocketMQ cluster').run).toBe(
+      'scripts/rocketmq-lite-broker.sh stop',
+    );
+    const workDir = step('Start the RocketMQ LiteTopic cluster').env
+      .RMQ_WORK_DIR;
+    expect(step('Stop the RocketMQ cluster').env.RMQ_WORK_DIR).toBe(workDir);
+    // The gate freezes the proxy whose pid file lives in that directory.
+    expect(workDir).toBe('${{ runner.temp }}/rocketmq');
+    expect(verify().run).toContain(
+      '-Dqwen.gate.rocketmq.work-dir="${RUNNER_TEMP}/rocketmq"',
+    );
+    expect(step('Upload multi-node gate reports').with.path).toContain(
+      'packages/sdk-java/managed-agent-server/target/mq3-multi-node/',
+    );
+  });
+
+  it('runs on a GitHub-hosted runner with the JDK 17 the broker needs', () => {
+    expect(job['runs-on']).toBe('ubuntu-latest');
+    const java = step('Set up Java').with['java-version'];
+    expect(String(java).split('\n')).toEqual(['17', '21']);
+  });
+
+  it('provisions the oracle group and points the gate at the bootstrap lane', () => {
+    expect(
+      step('Start the RocketMQ LiteTopic cluster').env.RMQ_EXTRA_GROUPS,
+    ).toBe(gateConstant('ORACLE_GROUP'));
+    expect(gateConstant('BOOTSTRAP_LITE_TOPIC')).toBe(
+      /private String wakeBootstrapLiteTopic =\s*"([^"]+)";/.exec(
+        properties,
+      )[1],
+    );
+    const run = verify().run;
+    expect(run).toContain(
+      `-Dqwen.gate.rocketmq.topic=${scriptDefault('RMQ_TOPIC')}`,
+    );
+    expect(run).toContain(
+      `-Dqwen.gate.rocketmq.endpoints=127.0.0.1:${scriptDefault('RMQ_PROXY_GRPC_PORT')}`,
+    );
+  });
+
+  it('installs the libfaketime the clock-skew row preloads', () => {
+    const library = /-Dqwen\.gate\.faketime\.library=(\S+)/.exec(
+      verify().run,
+    )[1];
+    expect(step('Install libfaketime').run).toContain(`test -f ${library}`);
+  });
+
+  it('points the gate at the database the MySQL service creates', () => {
+    const database = job.services.mysql.env.MYSQL_DATABASE;
+    expect(verify().run).toContain(`/${database}?`);
+    expect(verify().run).toContain(
+      `-Dmysql.password=${job.services.mysql.env.MYSQL_ROOT_PASSWORD}`,
+    );
+  });
+});
--- a/scripts/tests/spotbugs-gate.test.js
+++ b/scripts/tests/spotbugs-gate.test.js
@ -48,10 +48,10 @@ describe('SpotBugs gate witness', () => {
    );
    const byName = (name) => steps.filter((step) => step.name === name);

-    // Hosted MySQL, MariaDB and the EventTransport broker gates each stage
-    // the sibling modules before their own verify.
+    // Hosted MySQL, MariaDB and both EventTransport gate jobs each stage the
+    // sibling modules before their own verify.
    const staging = byName('Install Managed Agent dependencies');
-    expect(staging.length).toBe(3);
+    expect(staging.length).toBe(4);
    for (const step of staging) {
      const brokerInstall = step.run
        .split('\n')
@ -64,8 +64,9 @@ describe('SpotBugs gate witness', () => {
      ...byName('Run Managed Agent tests, Checkstyle, and MySQL integration'),
      ...byName('Verify Hosted Java, Spring and MySQL processes'),
      ...byName('Verify the EventTransport broker gates'),
+      ...byName('Verify the EventTransport multi-node gates'),
    ];
-    expect(enforcing.length).toBe(4);
+    expect(enforcing.length).toBe(5);
    for (const step of enforcing) {
      expect(step.run).toContain('verify');
      expect(step.run).not.toContain('spotbugs.skip');
@ -89,6 +90,7 @@ describe('SpotBugs gate witness', () => {
      'Install Managed Agent dependencies',
      'Install Managed Agent dependencies',
      'Install Managed Agent dependencies',
+      'Install Managed Agent dependencies',
    ]);
  });
 });