test(managed-agent): resume both Session streams across a held Hook (FG13e) (#14917)

* test(managed-agent): resume both Session streams across a held Hook (FG13e) A PreToolUse command Hook holds inside the Runtime worker while the public and WebShell clients read the Session stream. Both clients are then cut, the deny lands while neither is connected, and each resumes with its own cursor (Last-Event-ID on the public route, afterSequence on the WebShell route). Each resumed view must equal the committed ledger byte for byte with no gap and no duplicate, carry exactly one completion, and the model must see exactly one tool result carrying the denial reason while the tool never runs. The tool-call start frame is only emitted after the call leaves the Hook, so the cut deliberately happens before it: the frames written during the hold are exactly the ones a cut client misses. Adds the bilingual design note. One cell adds 150 s to the shared fork: healthy 9630 s, fork 10230 s, Verify 181 minutes. * test(managed-agent): read the cut cursors after the readers stop The resume cursor was sampled before abort(), so a frame the reader thread appended in that window would be served again on resume and redden the union as a duplicate. Take the cursors after abort(), matching the FG12f class. * test(managed-agent): release the held Hook after both streams are cut The fixture now offers a hook-release mode that waits for <evidence>.release instead of sleeping for a fixed 20 s, and the test writes that file only after both readers abort. The deny therefore cannot land before the cut, so the gap the cell claims to test is the one it actually creates. assertResumedView now names what the Hook-held turn put on the ledger: one failed terminal for call_FG13E_HOLD, past the cut cursor, and one completion. Transport fidelity to a ledger that lost or duplicated the deny no longer passes. fg13-cases joins the Hosted artifact upload and the meta test pins it. Both design docs correct the CI trigger: pull_request selects the fast tier, and the full suite runs on the first maintainer approval of the head. * test(managed-agent): correct the FG13e witness and raise the stall bound Mutation witness #2 said a live-only resume reddens on the gap assertion. continueFrom runs after the turn completed, so that client never reaches the ledger tail: its 90 s wait exhausts and the case reddens on that timeout. Both design docs now say so, and the Non-goals point the duplicated-helper extraction at #14921. The method's stall bound is raised to 480 s, above the 420 s of inner await budgets, so a hung wait reports its own description instead of the outer timeout. The Hosted lane trigger sentence was already corrected in 1eb67df537. * docs(managed-agent): describe the FG13e hold as an explicit release The step table still said the Hook holds for 20 s, which the explicit-release fixture replaced. Both design docs now say it is held until the test releases it, matching the code and the PR body. * test(managed-agent): keep the digits in the FG13e fixture's call id The prompt capture used [^A-Z_], which stops at the 1 in FG13E_HOLD, so the fixture minted call_FG while both deny assertions look for call_FG13E_HOLD. Add 0-9 to the class; the capture now keeps the whole marker. The design docs also carry the merged ledger: this cell adds 150 s, and with main's FG12f (+20 s) and FG11d retire-at-intent (+18 s) the totals are healthy 9668 s, fork 10268 s, Verify 182 minutes. * test(managed-agent): close the FG13e review follow-ups - Raise the stall bound to 600 s and the Hook command timeout to 240 s so a slow stream cannot trip the outer timeout or fail-closed before the release. - Sweep the cgroup subtree in stop() and pin the rule over every Hosted IT that exports QWEN_MANAGED_HOOK_CGROUP_ROOT. - Restore the workflow ledger's FG11d cancel entry so the chain sums from 9915 to 10268. - Copy processes.jsonl and the harness log into the evidence bundle, and pin the writer side of the fg13-cases path. - Enumerate the deferred shared machinery in both design docs; #14921 now carries the measured 635/810-line scope. --------- Co-authored-by: 易良 <[email protected]>
2026-11-07 06:19:43 -10:00 · 2026-10-21 02:03:50 +00:00
1 parent 731c8d6d25
commit f8c3388470
@@ -745,16 +745,18 @@ jobs:
    # 10065 s and Verify to 178 minutes; the FG11d cancel cell adds 15 s,
    # moving the fork to 10080 s and Verify to 179 minutes; the FG12f delivery
    # entry moves from 30 s to 50 s, moving the fork to 10100 s; the FG11d
-    # retire-at-intent cell adds 18 s, moving the fork to 10118 s; Class-boundary
-    # cleanup and its regression add 60 s, moving the fork to 10178 s and Verify
-    # to 180 minutes; the FG11d close-queued cell adds 134 s, moving the fork to
-    # 10312 s and Verify to 182 minutes, wrapping fork + 600 s plus margin
-    # (10912 s of fork plus wrap);
+    # retire-at-intent cell adds 18 s, moving the fork to 10118 s;
+    # Class-boundary cleanup and its regression add 60 s, moving the fork to
+    # 10178 s and Verify to 180 minutes; the FG13e Hook stream-gap cell adds
+    # 150 s, moving the fork to 10328 s and Verify to 183 minutes; the FG11d
+    # close-queued cell adds 134 s, moving the fork to 10462 s and Verify to
+    # 185 minutes, wrapping fork + 600 s plus margin (11062 s of fork plus
+    # wrap);
    # the Broker fault gates use 20,
    # and admission plus O4 keep per-fork ceilings) plus ~13 minutes for the
    # uncapped setup steps (Checkout.. cgroup delegation, the report
-    # checks): 272 + 13. The failover arms run in hosted-failover-e2e.
-    timeout-minutes: 285
+    # checks): 275 + 13. The failover arms run in hosted-failover-e2e.
+    timeout-minutes: 288
    services:
      mysql:
        image: 'mysql:8.4.6'
@@ -854,8 +856,8 @@ jobs:
        # fork leaves no per-test failure lines, so the main-CI failure
        # analyzer can only file an undiagnosable per-commit issue (#13503).
        # hosted-process-ci.test.js pins this ordering; the fork budget is
-        # 10312 s, so the step wraps to fork + 600 s plus margin (182 min).
-        timeout-minutes: 182
+        # 10462 s, so the step wraps to fork + 600 s plus margin (185 min).
+        timeout-minutes: 185
        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'] }}"
@@ -897,6 +899,7 @@ jobs:
            packages/managed-agent-server/target/fg10-cases/
            packages/managed-agent-server/target/fg11-cases/
            packages/managed-agent-server/target/fg12-cases/
+            packages/managed-agent-server/target/fg13-cases/
            packages/managed-agent-server/target/l4-shell-lifecycle/
          retention-days: 7

@@ -0,0 +1,93 @@
+# Hosted Hook stream-gap gates FG13e
+
+[English](2026-10-20-hosted-hook-stream-gap-gates-e.md) | [简体中文](2026-10-20-hosted-hook-stream-gap-gates-e.zh-CN.md)
+
+> Tracks #14449. The FG12f stream-gap class covers an MCP tool call; this
+> layer adds the Hook arm the tracker lists under FG13e.
+
+## Problem and scope
+
+A `PreToolUse` Hook is a decision gate with a process behind it. While that
+process runs, the turn produces public events (the tool call is announced,
+then settles) that both the public and the WebShell streams must serve
+exactly once. A client cut during the Hook's hold reconnects with a cursor:
+`Last-Event-ID` on the public route, `afterSequence` on the WebShell route.
+
+FG13e asks that a disconnect during a held Hook lose nothing and duplicate
+nothing on either route, and that a denial is delivered once. The Hook state
+itself is not on the public stream, so the gate is about the ledger and the
+cut, not about a Hook-specific frame.
+
+## Case
+
+`bothStreamsResumeTheHeldHookTurnOnMySql` in
+`HostedHookStreamGapIT`, one cold boot on the `hosted-harness-mysql` lane:
+
+| Step | What happens                                                                                       |
+| ---- | -------------------------------------------------------------------------------------------------- |
+| 1    | A `PreToolUse` command Hook is held in the Runtime worker until the test releases it, then denies. |
+| 2    | Both clients read the Session stream from the start: public (SSE) and WebShell (JSON cursor).      |
+| 3    | Once the Hook is holding and both clients have read a frame, both clients are cut.                 |
+| 4    | The Hook's deny lands while neither client is connected; the turn completes.                       |
+| 5    | Each client reconnects with its own cursor and reads to the ledger's tail.                         |
+
+The tool-call start frame is emitted only when the call leaves the Hook, so
+the cut deliberately happens before it: the frames written during the hold
+are exactly the ones a cut client misses.
+
+Expectations:
+
+- Each resumed view is the ledger prefix byte-for-byte: sequence `1..n` with
+  no gap and no duplicate (`continueFrom`), on both surfaces.
+- The union carries at least one frame past the cut position — the frames
+  written while both clients were disconnected — and exactly one
+  `turn.completed` for the Turn.
+- One physical Hook call (`hook.call`), one deny (`hook.deny`), and the tool
+  never ran: the model sees exactly one tool result for the emitted call
+  carrying the denial reason, and never the proof file's contents.
+
+## Coverage
+
+Already on main:
+
+- `HostedMcpStreamGapIT` pins both cursors on a held MCP tool call.
+- `ManagedEventReplayTest` pins `Last-Event-ID`, `afterSequence`, catch-up to
+  live handover and store fallback with no gaps or duplicates.
+
+This cell adds the Hook-held turn: the gate is cut while a Runtime worker
+process is holding the decision, so the ledger rows that land during the cut
+are the tool call's terminal state and the Turn's completion, not an MCP
+result. The cursor machinery is shared; the Hook lane is the new subject.
+
+## Mutation witnesses
+
+- Ignore the cursor (serve from the start): each view replays the cut
+  client's frames and the union's sequence list reddens on the duplicate.
+- Serve only live frames to a resumed client: the turn is already complete,
+  so that client never reaches the ledger tail; `continueFrom`'s 90 s wait
+  exhausts and the case reddens on that timeout (the union's gap check is
+  never reached).
+- Let the tool run despite the deny: the model request contains the proof
+  file's contents and `doesNotContain(PROOF)` reddens.
+
+## Non-goals
+
+- Stop or `MessageDisplay` suppressed text in a replay: suppression is a
+  decision/output property, tracked as FG13f `message-display-reload`.
+- FG13a store acks and store-side faults stay on #14449.
+- Moving the shared machinery into `HostedGateTestSupport` is a follow-up
+  sweep tracked in #14921, not part of this cell: the stream and ledger
+  helpers (`Observer`, `ledger`, `served`, `camel`, `continueFrom`), the
+  Spring boot block (`startSpring`, `stop`), and the request/wait helpers
+  (`api`, `waitFor`, `quote`, `events`, `createSession`, `submit`,
+  `turnStatus`, `awaitCompleted`).
+
+## Validation
+
+The `hosted-harness-mysql` lane on Linux with MySQL 8.4 and a delegated
+cgroup v2 root is the evidence. `managed-agent-gate.yml` selects the `fast`
+tier on `pull_request`; the full suite runs on the first maintainer approval
+of the head commit, so a skipped lane is not a passing gate. This cell adds
+150 s; with main's FG12f delivery entry (30 s to 50 s, +20 s) and the FG11d
+retire-at-intent cell (+18 s), the merged ledger is healthy 9668 s, fork
+10268 s, Verify 182 minutes.
@@ -0,0 +1,78 @@
+# Hosted Hook 事件流断流门禁 FG13e
+
+[English](2026-10-20-hosted-hook-stream-gap-gates-e.md) | [简体中文](2026-10-20-hosted-hook-stream-gap-gates-e.zh-CN.md)
+
+> 跟踪 #14449。FG12f 的断流类覆盖的是一次 MCP 工具调用；本层补上跟踪项列在
+> FG13e 下的 Hook 臂。
+
+## 问题与范围
+
+`PreToolUse` Hook 是一个带进程的决策门。该进程运行期间，回合会产出公共事件
+（工具调用先宣告、后结算），公共流与 WebShell 流都必须恰好投递一次。被切断的
+客户端用游标重连：公共路由用 `Last-Event-ID`，WebShell 路由用 `afterSequence`。
+
+FG13e 要求：在 Hook 持有期间断流，两条路由都不丢帧、不重帧，拒绝只投递一次。
+Hook 状态本身不在公共流上，所以这一格钉的是台账与断流，而不是 Hook 专属帧。
+
+## 用例
+
+`HostedHookStreamGapIT` 的 `bothStreamsResumeTheHeldHookTurnOnMySql`，在
+`hosted-harness-mysql` 车道上只冷启一次：
+
+| 步骤 | 发生什么                                                                               |
+| ---- | -------------------------------------------------------------------------------------- |
+| 1    | 一个 `PreToolUse` 命令 Hook 在 Runtime worker 里保持持有，直到测试显式释放，随后拒绝。 |
+| 2    | 两个客户端从头读同一 Session 流：公共（SSE）与 WebShell（JSON 游标）。                 |
+| 3    | Hook 已进入持有、两个客户端都已读到帧后，同时切断两个客户端。                          |
+| 4    | Hook 的拒绝在两个客户端都不在线时落地；回合完成。                                      |
+| 5    | 两个客户端各自带游标重连，读到台账尾部。                                               |
+
+工具调用的起始帧只在调用离开 Hook 之后才发射，所以断流刻意发生在它之前：持有
+期间写入的帧，正是被切断客户端会漏掉的那些。
+
+期望：
+
+- 两个续读视图都与台账前缀逐字节一致：序号 `1..n` 无空洞、无重复
+  （`continueFrom`），两条路由各自成立。
+- 联合视图至少比断流位置多出一帧（即两个客户端离线期间写入的帧），且回合只有
+  一个 `turn.completed`。
+- 一次物理 Hook 调用（`hook.call`）、一次拒绝（`hook.deny`），工具从未执行：
+  模型只收到一条携带拒绝原因的工具结果，且从不包含证明文件的内容。
+
+## 覆盖
+
+已在 main 上：
+
+- `HostedMcpStreamGapIT` 用被持有的 MCP 工具调用钉住两个游标。
+- `ManagedEventReplayTest` 钉住 `Last-Event-ID`、`afterSequence`、追平到实时的
+  交接与 store 回退，均无空洞、无重复。
+
+本格补的是 Hook 持有中的回合：断流发生在 Runtime worker 进程持有决策之时，因此
+断流期间落台账的行是工具调用的终态与回合完成，而不是 MCP 结果。游标机制是共用
+的；Hook 车道是新的对象。
+
+## 变异见证
+
+- 忽略游标（从头重放）：每个视图都会重放被切客户端的帧，联合序号表在重复处变红。
+- 只给续读客户端投实时帧：回合已经完成，该客户端永远读不到台账尾部；`continueFrom`
+  的 90 秒等待耗尽，用例在该超时处变红（联合视图的空洞断言根本到不了）。
+- 拒绝后仍让工具执行：模型请求会带上证明文件的内容，`doesNotContain(PROOF)` 变红。
+
+## 非目标
+
+- Stop 或 `MessageDisplay` 被抑制文本的重放：抑制属于决策/输出性质，由 FG13f 的
+  `message-display-reload` 跟踪。
+- FG13a 的 store 确认与 store 侧故障仍留在 #14449。
+- 将共享机制移入 `HostedGateTestSupport` 是 #14921 跟踪的后续清扫，不属于本格：
+  流与台账辅助（`Observer`、`ledger`、`served`、`camel`、`continueFrom`）、
+  Spring 启动块（`startSpring`、`stop`），以及请求/等待辅助（`api`、`waitFor`、
+  `quote`、`events`、`createSession`、`submit`、`turnStatus`、`awaitCompleted`）。
+
+## 验证
+
+证据来自 Linux 上的 `hosted-harness-mysql` 车道，MySQL 8.4 与委派的 cgroup v2
+根。`managed-agent-gate.yml` 在 `pull_request` 上只选 `fast` 档；完整套件由维护者
+对 head 提交的首次批准触发，因此被跳过的车道不算通过。
+本格增加 150 s；叠加 main 的 FG12f 投递格（30→50 s，+20 s）与 FG11d
+retire-at-intent（+18 s）后，合并台账为 healthy 9668 s、fork 10268 s、
+Verify 182 分钟。
@@ -552,24 +552,23 @@
                                 measured 7.4 s), so the fork moves to 10080 s
                                 and the Verify step to 179 minutes: 10680 s of
                                 fork plus wrap is exactly 178, so 179 is what
-                                 keeps the margin. The FG12f delivery entry moves from 30 s to
-                                 50 s at 2x (four cases, 24.16 s measured): the
-                                 healthy sum becomes 9500 s, so the fork moves
-                                 to 10100 s and the 179-minute step still wraps
-                                 (10700 s of fork plus wrap). The FG11d
-                                 retire-at-intent cell adds 18 s (2x its measured
-                                 8.6 s): the healthy sum becomes 9518 s, so the
-                                 fork moves to 10118 s and the 179-minute step
-                                 still wraps (10718 s of fork plus wrap). Class-boundary
-                                 cleanup and its regression add 60 s:
-                                 healthy 9578 s, fork 10178 s, Verify 180
-                                 minutes with margin above the 10778-second
-                                 fork-plus-wrap sum. The FG11d close-queued
-                                 cell adds 134 s (2x its measured 67.2 s):
-                                 healthy 9712 s, fork 10312 s, Verify 182
-                                 minutes with margin above the 10912-second
-                                 fork-plus-wrap sum. -->
-                            <forkedProcessTimeoutInSeconds>10312</forkedProcessTimeoutInSeconds>
+                                 keeps the margin. The FG12f delivery entry
+                                 moves from 30 s to 50 s at 2x (four cases,
+                                 24.16 s measured): the healthy sum becomes
+                                 9500 s, so the fork moves to 10100 s and the
+                                 179-minute step still wraps (10700 s of fork
+                                 plus wrap). The FG11d retire-at-intent cell
+                                 adds 18 s (2x its measured 8.6 s): the
+                                 healthy sum becomes 9518 s, so the fork moves
+                                 to 10118 s. Class-boundary cleanup and its
+                                 regression add 60 s: healthy 9578 s, fork
+                                 10178 s, Verify 180 minutes. The FG13e Hook
+                                 stream-gap cell adds 150 s: healthy 9728 s,
+                                 fork 10328 s, Verify 183 minutes. The FG11d
+                                 close-queued cell adds 134 s (2x its measured
+                                 67.2 s): healthy 9862 s, fork 10462 s, Verify
+                                 185 minutes. -->
+                            <forkedProcessTimeoutInSeconds>10462</forkedProcessTimeoutInSeconds>
                        </configuration>
                        <executions>
                            <execution>
@@ -0,0 +1,841 @@
+package com.alibaba.qwen.code.managedagent;
+
+import com.alibaba.qwen.code.managedagent.persistence.WorkspaceStorageGuard;
+import com.alibaba.qwen.code.runtimebroker.WorkspaceExecutionProfile;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import org.apache.commons.codec.digest.DigestUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.condition.EnabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+import org.junit.jupiter.api.io.CleanupMode;
+import org.junit.jupiter.api.io.TempDir;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.attribute.FileTime;
+import java.nio.file.attribute.PosixFilePermissions;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.LongStream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+// FG13e (issue #14449): the Hosted Hook stream-gap gate. One tool call is
+// parked inside a PreToolUse command Hook while both clients read the Session
+// stream, both are cut mid-call, and each resumed client must continue the
+// same ledger exactly once. The public route resumes with Last-Event-ID, the
+// WebShell route with afterSequence; a cursor that is ignored replays and
+// reddens the union, it does not time out.
+@EnabledOnOs(OS.LINUX)
+class HostedHookStreamGapIT {
+    private static final String PROFILE = "hosted-workspace-files/1";
+    private static final String TOKEN = "hook-stream-fixture-token";
+    private static final String CAPABILITY = "sha256:" + "a".repeat(64);
+    private static final String PROOF = "FG13E_TOOL_PROOF";
+    // The Hook is released by the test only after both clients are cut, so
+    // the deny cannot land before the cut the way a fixed sleep could.
+    private static final String HOOK_HOLD_MS = "0";
+    private final ObjectMapper json = new ObjectMapper();
+    private final HttpClient http = HttpClient.newBuilder()
+            .connectTimeout(Duration.ofSeconds(3)).build();
+    private final String tenant = "hook-stream-" + UUID.randomUUID();
+    private final List<JsonNode> modelRequests = new CopyOnWriteArrayList<>();
+    private final AtomicReference<Throwable> modelFailure =
+            new AtomicReference<>();
+    private final List<Observer> observers = new CopyOnWriteArrayList<>();
+    @TempDir(cleanup = CleanupMode.ON_SUCCESS)
+    private Path temporary;
+    private Path cli;
+    private Path workspace;
+    private Path evidence;
+    private Path hookRoot;
+    private String node;
+    private String hookDigest;
+    private int port;
+    private int harnessPort;
+    private int brokerPort;
+    private int boots;
+    private HttpServer model;
+    private ExecutorService modelExecutor;
+    private ServletWebServerApplicationContext spring;
+    private JdbcTemplate jdbc;
+    private Process harness;
+
+    @BeforeEach
+    void boot() throws Exception {
+        assertThat(System.getProperty("mysql.url"))
+                .as("FG13e requires a disposable MySQL")
+                .startsWith("jdbc:mysql:");
+        node = System.getProperty("node.executable");
+        assertThat(node).isNotBlank();
+        cli = Path.of(System.getProperty("qwen.cli.entry",
+                "../../dist/cli.js")).toAbsolutePath().normalize();
+        assertThat(cli).isRegularFile();
+        temporary = temporary.toRealPath();
+        workspace = Files.createDirectories(
+                temporary.resolve("workspace/child"));
+        Files.setLastModifiedTime(workspace.getParent(),
+                FileTime.fromMillis(1));
+        Files.createDirectory(temporary.resolve("decoy"));
+        Files.createDirectory(temporary.resolve("broker"),
+                PosixFilePermissions.asFileAttribute(
+                        PosixFilePermissions.fromString("rwx------")));
+        Files.writeString(workspace.resolve("proof.txt"), PROOF);
+        evidence = temporary.resolve("processes.jsonl");
+        String delegated = System.getenv("QWEN_MANAGED_HOOK_CGROUP_ROOT");
+        assertThat(delegated)
+                .as("Run Maven inside a delegated Linux cgroup v2 subtree")
+                .isNotBlank();
+        hookRoot = Files.createDirectory(
+                Path.of(delegated).resolve("hook-stream-" + UUID.randomUUID()));
+        Path program = temporary.resolve("hook-stream-process.mjs");
+        try (var resource = getClass().getResourceAsStream(
+                "/hosted-hook-fault-process.mjs")) {
+            assertThat(resource).isNotNull();
+            Files.copy(resource, program);
+        }
+        hookDigest = DigestUtils.sha256Hex("hook:"
+                + Files.readString(program));
+        Map<String, Object> hook = new java.util.LinkedHashMap<>();
+        hook.put("hookId", "pre");
+        hook.put("eventName", "PreToolUse");
+        hook.put("sequential", false);
+        hook.put("failClosed", true);
+        hook.put("async", false);
+        hook.put("config", Map.of("type", "command", "command",
+                quote(node) + " " + quote(program.toString()) + " hook "
+                        + quote(evidence.toString()) + " "
+                        + quote(HOOK_HOLD_MS) + " " + quote("hook-release"),
+                // The release runs only after both readers stop, a path the
+                // awaits above budget past 60 s; a shorter timeout would
+                // fail-closed before the release and read as a lost deny.
+                "timeout", 240000));
+        hook.put("onceKey", null);
+        json.writeValue(temporary.resolve("hooks.json").toFile(),
+                Map.of("version", 1, "catalogs", List.of(Map.of("tenantId",
+                        tenant, "workspaceId", "workspace", "catalogId",
+                        "fg13e-hooks", "catalogRevision", 1,
+                        "definitionDigest", catalogDigest(), "hooks",
+                        List.of(hook)))));
+        Path worker = temporary.resolve("worker.mjs");
+        Files.writeString(worker, "process.env.QWEN_MANAGED_HOOK_CONFIG="
+                + json.writeValueAsString(
+                        temporary.resolve("hooks.json").toString())
+                + "; process.env.QWEN_MANAGED_HOOK_CGROUP_ROOT="
+                + json.writeValueAsString(hookRoot.toString())
+                + "; process.argv[1]="
+                + json.writeValueAsString(cli.toString())
+                + "; await import("
+                + json.writeValueAsString(cli.toUri().toString()) + ");\n");
+        try (var control = new ServerSocket(0, 1,
+                InetAddress.getByName("127.0.0.1"));
+                var hosted = new ServerSocket(0, 1,
+                        InetAddress.getByName("127.0.0.1"));
+                var runtime = new ServerSocket(0, 1,
+                        InetAddress.getByName("127.0.0.1"))) {
+            port = control.getLocalPort();
+            harnessPort = hosted.getLocalPort();
+            brokerPort = runtime.getLocalPort();
+        }
+        model = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+        modelExecutor = Executors.newVirtualThreadPerTaskExecutor();
+        model.setExecutor(modelExecutor);
+        model.createContext("/v1/chat/completions", this::modelReply);
+        model.start();
+        startSpring();
+        startHarness();
+    }
+
+    // The awaits total 480 s (5 x 60 for the Hook call, both first frames,
+    // the deny and the completion, plus 2 x 90 for the two resumes), so 600
+    // leaves 120 s for a hung wait to report its own description first.
+    @Test
+    @Timeout(600)
+    void bothStreamsResumeTheHeldHookTurnOnMySql() throws Exception {
+        String agent = api("POST", "/v1/agents", definition(), "definition",
+                202).path("id").asText();
+        String session = createSession(agent, "gap");
+        Observer publicCut = new Observer("public", session, 0);
+        Observer webShellCut = new Observer("webshell", session, 0);
+        String turn = submit(session, "FG13E_HOLD");
+        // The Hook is holding inside the worker. The tool-call frame is only
+        // emitted once the call leaves the Hook, so both clients are cut on
+        // the frames they already read, and the deny lands while neither is
+        // connected.
+        awaitHookCalled();
+        publicCut.awaitFirstFrame();
+        webShellCut.awaitFirstFrame();
+        publicCut.abort();
+        webShellCut.abort();
+        // Read the cursors only after the readers stopped: a frame appended
+        // between the read and the abort would otherwise be replayed on
+        // resume and redden the union as a duplicate.
+        long publicCutAt = publicCut.lastSequence();
+        long webShellCutAt = webShellCut.lastSequence();
+        // Release only now, so the deny cannot land before the cut.
+        releaseHook();
+        awaitDenied();
+        awaitCompleted(session, turn);
+        List<JsonNode> publicView = new Observer("public", session,
+                publicCutAt).continueFrom(publicCut, session);
+        List<JsonNode> webShellView = new Observer("webshell", session,
+                webShellCutAt).continueFrom(webShellCut, session);
+        assertResumedView(publicView, publicCutAt);
+        assertResumedView(webShellView, webShellCutAt);
+        assertThat(events("hook.call")).as("one physical Hook call")
+                .hasSize(1);
+        assertThat(events("hook.deny")).as("the Hook answered once")
+                .hasSize(1);
+        assertThat(denialResults()).as("the model sees the denial once")
+                .isEqualTo(1);
+        assertThat(modelRequests.toString()).doesNotContain(PROOF);
+        assertThat(modelFailure.get()).isNull();
+    }
+
+    /**
+     * The denial reaches the model as one tool result for the emitted call.
+     * A duplicated delivery, or a second physical call, would add another.
+     */
+    private long denialResults() {
+        long count = 0;
+        for (JsonNode message : modelRequests.getLast().path("messages")) {
+            if ("tool".equals(message.path("role").asText())
+                    && "call_FG13E_HOLD".equals(
+                            message.path("tool_call_id").asText())
+                    && message.toString().contains("FG13_DENIED")) {
+                count++;
+            }
+        }
+        return count;
+    }
+
+    private void awaitHookCalled() throws Exception {
+        waitFor(() -> assertThat(events("hook.call"))
+                .as("the Hook is holding inside the worker")
+                .hasSize(1));
+    }
+
+    private void awaitDenied() throws Exception {
+        waitFor(() -> assertThat(events("hook.deny"))
+                .as("the Hook's deny landed while both clients were cut")
+                .hasSize(1));
+    }
+
+    /**
+     * The resumed view closed the gap the cut opened and carries one
+     * completion. The byte-for-byte ledger check lives in {@code
+     * continueFrom}; this names what the Hook-held turn put there: the
+     * denied call's single failed terminal, after the cut, and one
+     * completion, so a ledger that lost the deny or duplicated it reddens
+     * even when transport matches it.
+     */
+    private void assertResumedView(List<JsonNode> union, long cutAt) {
+        assertThat(union.size())
+                .as("frames landed after the cut and the resume served them")
+                .isGreaterThan((int) cutAt);
+        List<JsonNode> denied = union.stream()
+                .filter(frame -> "item.tool_call.updated".equals(
+                        frame.path("type").asText()))
+                .filter(frame -> "call_FG13E_HOLD".equals(frame.path("data")
+                        .path("toolCallId").asText()))
+                .filter(frame -> "failed".equals(frame.path("data")
+                        .path("status").asText()))
+                .toList();
+        assertThat(denied).as("one failed terminal for the denied call: %s",
+                union).hasSize(1);
+        assertThat(denied.getFirst().path("sequence").asLong())
+                .as("the deny landed after the cut, not before it")
+                .isGreaterThan(cutAt);
+        assertThat(union.stream().filter(frame -> "turn.completed".equals(
+                frame.path("type").asText())).toList())
+                .as("one completion, not one per reconnect: %s", union)
+                .hasSize(1);
+    }
+
+    private void releaseHook() throws IOException {
+        Files.writeString(Path.of(evidence + ".release"), "release");
+    }
+
+    private void awaitCompleted(String session, String turn) {
+        waitFor(() -> assertThat(turnStatus(session, turn)).isEqualTo(
+                "COMPLETED"));
+    }
+
+    private static String quote(String text) {
+        return "'" + text.replace("'", "'\"'\"'") + "'";
+    }
+
+    private Map<String, Object> definition() {
+        return Map.of("model", Map.of(), "instructions",
+                "Read the workspace proof file.",
+                "tools", List.of(Map.of("type", "hosted_profile", "profile",
+                        PROFILE)),
+                "permission_policy", Map.of("approval_mode", "yolo"),
+                "hook_catalog", Map.of("catalog_id", "fg13e-hooks",
+                        "catalog_revision", 1, "definition_digest",
+                        catalogDigest()));
+    }
+
+    private String catalogDigest() {
+        return DigestUtils.sha256Hex(hookDigest + 1);
+    }
+
+    private String createSession(String agent, String key) throws Exception {
+        return api("POST", "/v1/agents/sessions", Map.of("agent_id", agent,
+                "input", List.of(), "workspace", Map.of("workspace_id",
+                        "workspace", "cwd_relative", "child")), key, 202)
+                .path("id").asText();
+    }
+
+    private String submit(String session, String prompt) throws Exception {
+        return api("POST", "/v1/agents/sessions/" + session + "/events",
+                Map.of("type", "agent.session.input.message", "input",
+                        List.of(Map.of("type", "input_text", "text", prompt))),
+                prompt, 202).path("turn_id").asText();
+    }
+
+    private String turnStatus(String session, String turn) {
+        return jdbc.queryForObject("SELECT status FROM ma_turn WHERE"
+                + " tenant_id = ? AND session_id = ? AND turn_id = ?",
+                String.class, tenant, session, turn);
+    }
+
+    private void modelReply(HttpExchange exchange) throws IOException {
+        try {
+            JsonNode request = json.readTree(exchange.getRequestBody());
+            modelRequests.add(request);
+            String prompt = "";
+            List<JsonNode> results = new ArrayList<>();
+            for (JsonNode message : request.path("messages")) {
+                if ("user".equals(message.path("role").asText())
+                        && message.path("content").toString().contains(
+                                "FG13E_")) {
+                    String text = message.path("content").toString();
+                    prompt = text.substring(text.lastIndexOf("FG13E_"))
+                            .replaceAll("[^A-Z0-9_].*", "");
+                    results.clear();
+                } else if ("tool".equals(message.path("role").asText())) {
+                    results.add(message);
+                }
+            }
+            assertThat(prompt).isNotBlank();
+            List<String> names = new ArrayList<>();
+            request.path("tools").forEach(tool -> names.add(
+                    tool.at("/function/name").asText()));
+            assertThat(names).contains("read_file");
+            boolean call = results.isEmpty();
+            var chunk = json.createObjectNode().put("id", "hook-stream-gap")
+                    .put("object", "chat.completion.chunk").put("created", 0)
+                    .put("model", "hook-stream-fixture");
+            var choice = chunk.putArray("choices").addObject().put("index", 0);
+            var delta = choice.putObject("delta").put("role", "assistant");
+            if (call) {
+                delta.putArray("tool_calls").addObject().put("index", 0)
+                        .put("id", "call_" + prompt).put("type", "function")
+                        .putObject("function").put("name", "read_file")
+                        .put("arguments", json.writeValueAsString(
+                                Map.of("file_path", "proof.txt")));
+            } else {
+                delta.put("content", "FG13E_DONE_" + prompt);
+            }
+            choice.putNull("finish_reason");
+            String first = "data: " + json.writeValueAsString(chunk) + "\n\n";
+            choice.putObject("delta");
+            choice.put("finish_reason", call ? "tool_calls" : "stop");
+            byte[] response = (first + "data: "
+                    + json.writeValueAsString(chunk)
+                    + "\n\ndata: [DONE]\n\n").getBytes(
+                            StandardCharsets.UTF_8);
+            exchange.getResponseHeaders().set("Content-Type",
+                    "text/event-stream");
+            exchange.sendResponseHeaders(200, response.length);
+            exchange.getResponseBody().write(response);
+        } catch (Exception | AssertionError failure) {
+            modelFailure.compareAndSet(null, failure);
+        } finally {
+            exchange.close();
+        }
+    }
+
+    private void startSpring() {
+        spring = (ServletWebServerApplicationContext) new SpringApplicationBuilder(
+                ManagedAgentServerApplication.class).run(
+                        "--server.address=127.0.0.1",
+                        "--server.port=" + port,
+                        "--spring.datasource.url="
+                                + System.getProperty("mysql.url"),
+                        "--spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver",
+                        "--spring.datasource.username="
+                                + System.getProperty("mysql.user"),
+                        "--spring.datasource.password="
+                                + System.getProperty("mysql.password", ""),
+                        "--qwen.managed-agent.trusted-actor-header=X-Qwen-Actor-Id",
+                        "--qwen.managed-agent.definition-tool-profiles="
+                                + PROFILE,
+                        "--qwen.managed-agent.session-store.enabled=true",
+                        "--qwen.managed-agent.session-store.base-url=http://127.0.0.1:"
+                                + port,
+                        "--qwen.managed-agent.session-store.workspace-id=unused-global-workspace",
+                        "--qwen.managed-agent.session-store.writer-lease-duration=3s",
+                        "--qwen.managed-agent.harness.enabled=true",
+                        "--qwen.managed-agent.harness.workspace-files-enabled=true",
+                        "--qwen.managed-agent.harness.base-url=http://127.0.0.1:"
+                                + harnessPort,
+                        "--qwen.managed-agent.harness.token=" + TOKEN,
+                        "--qwen.managed-agent.harness.capability-digest="
+                                + CAPABILITY,
+                        "--qwen.managed-agent.runtime-broker.enabled=true",
+                        "--qwen.managed-agent.runtime-broker.port="
+                                + brokerPort,
+                        "--qwen.managed-agent.runtime-broker.token=" + TOKEN,
+                        "--qwen.managed-agent.runtime-broker.durable-local-process=true",
+                        "--qwen.managed-agent.runtime-broker.trusted-local-reboot-recovery=false",
+                        "--qwen.managed-agent.runtime-broker.verified-workspace-recovery-enabled=true",
+                        "--qwen.managed-agent.runtime-broker.workspace-cwd="
+                                + temporary.resolve("decoy"),
+                        "--qwen.managed-agent.runtime-broker.state-directory="
+                                + temporary.resolve("broker"),
+                        "--qwen.managed-agent.runtime-broker.credential-key-id=extensions",
+                        "--qwen.managed-agent.runtime-broker.credential-key=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
+                        "--qwen.managed-agent.runtime-broker.node-executable="
+                                + node,
+                        "--qwen.managed-agent.runtime-broker.worker-entry="
+                                + temporary.resolve("worker.mjs"),
+                        "--qwen.managed-agent.runtime-broker.workspace-mounts[0].tenant-id="
+                                + tenant,
+                        "--qwen.managed-agent.runtime-broker.workspace-mounts[0].storage-id=storage",
+                        "--qwen.managed-agent.runtime-broker.workspace-mounts[0].root="
+                                + workspace.getParent());
+        jdbc = spring.getBean(JdbcTemplate.class);
+        spring.getBean(WorkspaceStorageGuard.class).register(tenant, "storage",
+                UUID.randomUUID().toString());
+        jdbc.update("INSERT INTO ma_workspace (tenant_id, workspace_id,"
+                + " workspace_generation, storage_id, display_name, config_ref,"
+                + " policy_ref, state) VALUES (?, 'workspace', 1, 'storage',"
+                + " 'Stream gap', ?, ?, 'ACTIVE')", tenant,
+                WorkspaceExecutionProfile.CONFIG_REF,
+                WorkspaceExecutionProfile.POLICY_REF);
+        jdbc.update("INSERT INTO ma_workspace_access (tenant_id,"
+                + " workspace_id, actor_id, role) VALUES (?, 'workspace', ?,"
+                + " 'OPERATOR')", tenant,
+                "actor".getBytes(StandardCharsets.UTF_8));
+    }
+
+    private void startHarness() throws Exception {
+        Path home = Files.createDirectory(temporary.resolve("home-" + ++boots));
+        Path config = Files.createDirectory(home.resolve(".qwen"));
+        String modelUrl = "http://127.0.0.1:"
+                + model.getAddress().getPort() + "/v1";
+        json.writeValue(config.resolve("settings.json").toFile(),
+                Map.of("security", Map.of("auth", Map.of("selectedType",
+                        "openai")), "telemetry", Map.of("enabled", false),
+                        "model", Map.of("name", "extension-fixture"),
+                        "modelProviders", Map.of("openai", List.of(Map.of(
+                                "id", "extension-fixture", "envKey",
+                                "OPENAI_API_KEY", "baseUrl", modelUrl)))));
+        ProcessBuilder builder = new ProcessBuilder(node, cli.toString(),
+                "serve", "--profile", "hosted-harness", "--http-bridge",
+                "--hostname", "127.0.0.1", "--port",
+                Integer.toString(harnessPort), "--require-auth", "--no-web",
+                "--workspace", temporary.resolve("decoy").toString(),
+                "--managed-runtime-broker-url", "http://127.0.0.1:"
+                        + brokerPort, "--managed-runtime-broker-token", TOKEN)
+                .directory(temporary.resolve("decoy").toFile())
+                .redirectErrorStream(true)
+                .redirectOutput(temporary.resolve("harness-" + boots + ".log")
+                        .toFile());
+        var env = builder.environment();
+        env.clear();
+        env.put("PATH", System.getenv("PATH"));
+        env.putAll(Map.of("HOME", home.toString(), "QWEN_HOME",
+                config.toString(), "QWEN_RUNTIME_DIR",
+                home.resolve("runtime").toString(), "OPENAI_API_KEY",
+                "fixture-only", "OPENAI_BASE_URL", modelUrl,
+                "QWEN_SERVER_TOKEN", TOKEN,
+                "QWEN_HOSTED_HARNESS_CAPABILITY_DIGEST", CAPABILITY,
+                "QWEN_CODE_MODELS_DEV_REFRESH", "off"));
+        for (String key : List.of("QWEN_CODE_SYSTEM_SETTINGS_PATH",
+                "QWEN_CODE_SYSTEM_DEFAULTS_PATH",
+                "QWEN_CODE_TRUSTED_FOLDERS_PATH")) {
+            env.put(key, home.resolve(key + ".json").toString());
+        }
+        harness = builder.start();
+        waitFor(() -> {
+            assertThat(harness.isAlive()).as("Harness log: %s",
+                    temporary.resolve("harness-" + boots + ".log")).isTrue();
+            HttpResponse<String> capabilities = http.send(
+                    HttpRequest.newBuilder(URI.create("http://127.0.0.1:"
+                            + harnessPort + "/capabilities"))
+                            .timeout(Duration.ofSeconds(2))
+                            .header("Authorization", "Bearer " + TOKEN)
+                            .build(), HttpResponse.BodyHandlers.ofString());
+            assertThat(capabilities.statusCode()).isEqualTo(200);
+        });
+    }
+
+    private List<JsonNode> events(String kind) throws Exception {
+        if (!Files.exists(evidence)) {
+            return List.of();
+        }
+        List<JsonNode> result = new ArrayList<>();
+        for (String line : Files.readAllLines(evidence)) {
+            JsonNode event = json.readTree(line);
+            if (kind.equals(event.path("event").asText())) {
+                result.add(event);
+            }
+        }
+        return result;
+    }
+
+    /** The Session stream's committed public events, in sequence. */
+    private List<Map<String, Object>> ledger(String session) {
+        return jdbc.queryForList("SELECT schema_version, projection_version,"
+                + " sequence_id, CAST(event_id AS CHAR) AS event_id,"
+                + " CAST(session_id AS CHAR) AS session_id,"
+                + " CAST(turn_id AS CHAR) AS turn_id,"
+                + " CAST(item_id AS CHAR) AS item_id,"
+                + " CAST(content_part_id AS CHAR) AS content_part_id,"
+                + " event_type, gmt_create, data_json, terminal"
+                + " FROM ma_public_event WHERE tenant_id = ? AND"
+                + " session_id = ? AND stream_kind = 'SESSION' AND"
+                + " stream_id = X'' ORDER BY sequence_id", tenant, session);
+    }
+
+    /** One ledger row exactly as the surface serves it. */
+    private JsonNode served(Map<String, Object> row, String surface)
+            throws IOException {
+        boolean web = "webshell".equals(surface);
+        Object created = row.get("gmt_create");
+        long millis = created instanceof java.sql.Timestamp stamp
+                ? stamp.getTime()
+                : ((Number) created).longValue();
+        Map<String, Object> fields = new java.util.LinkedHashMap<>();
+        fields.put("schema_version", row.get("schema_version"));
+        fields.put("projection_version", row.get("projection_version"));
+        fields.put("sequence", row.get("sequence_id"));
+        fields.put("event_id", row.get("event_id"));
+        fields.put("session_id", row.get("session_id"));
+        fields.put("turn_id", row.get("turn_id"));
+        fields.put("item_id", row.get("item_id"));
+        fields.put("content_part_id", row.get("content_part_id"));
+        fields.put("type", row.get("event_type"));
+        fields.put("created_at", web ? millis : millis / 1000);
+        fields.put("data", json.readTree(row.get("data_json").toString()));
+        Object terminal = row.get("terminal");
+        fields.put("terminal", terminal instanceof Boolean flag ? flag
+                : ((Number) terminal).intValue() != 0);
+        ObjectNode node = json.createObjectNode();
+        fields.forEach((key, value) -> {
+            if (value != null) {
+                node.set(web ? camel(key) : key, json.valueToTree(value));
+            }
+        });
+        return json.readTree(json.writeValueAsString(node));
+    }
+
+    private static String camel(String key) {
+        StringBuilder out = new StringBuilder();
+        for (int i = 0; i < key.length(); i++) {
+            char c = key.charAt(i);
+            if (c == '_' && i + 1 < key.length()) {
+                out.append(Character.toUpperCase(key.charAt(++i)));
+            } else {
+                out.append(c);
+            }
+        }
+        return out.toString();
+    }
+
+    private JsonNode api(String method, String path, Object body, String key,
+            int... expected) throws Exception {
+        var request = HttpRequest.newBuilder(URI.create(
+                "http://127.0.0.1:" + port + path))
+                .timeout(Duration.ofSeconds(10))
+                .header("X-Qwen-Tenant-Id", tenant)
+                .header("X-Qwen-Actor-Id", "actor")
+                .header("Content-Type", "application/json")
+                .method(method, body == null
+                        ? HttpRequest.BodyPublishers.noBody()
+                        : HttpRequest.BodyPublishers.ofString(
+                                json.writeValueAsString(body)));
+        if (key != null) {
+            request.header("Idempotency-Key", key);
+        }
+        var response = http.send(request.build(),
+                HttpResponse.BodyHandlers.ofString());
+        assertThat(response.statusCode())
+                .as("%s %s: %s", method, path, response.body())
+                .isIn(java.util.Arrays.stream(expected).boxed().toList());
+        return json.readTree(response.body());
+    }
+
+    private void waitFor(org.junit.jupiter.api.function.Executable assertion) {
+        await().atMost(Duration.ofSeconds(60))
+                .ignoreException(java.net.ConnectException.class)
+                .failFast(() -> {
+                    if (modelFailure.get() != null) {
+                        throw new AssertionError("Model fixture: "
+                                + temporary, modelFailure.get());
+                    }
+                }).untilAsserted(assertion::execute);
+    }
+
+    /**
+     * One client of a Session's public or WebShell event stream. It owns its
+     * HttpClient, so {@link #abort} cuts the connection mid-stream the way a
+     * lost client does; {@link #frames} is everything it read.
+     */
+    private final class Observer {
+        private final List<JsonNode> frames = new CopyOnWriteArrayList<>();
+        private final AtomicReference<Throwable> failure =
+                new AtomicReference<>();
+        private final HttpClient client = HttpClient.newBuilder()
+                .version(HttpClient.Version.HTTP_1_1).build();
+        private final String surface;
+        private final Thread reader;
+        private volatile boolean aborted;
+
+        Observer(String surface, String session, long after) throws Exception {
+            this.surface = surface;
+            HttpRequest.Builder request;
+            if ("public".equals(surface)) {
+                request = HttpRequest.newBuilder(URI.create("http://127.0.0.1:"
+                        + port + "/v1/agents/sessions/" + session
+                        + "/events?stream=true")).GET();
+                if (after > 0) {
+                    request.header("Last-Event-ID", Long.toString(after));
+                }
+            } else {
+                request = HttpRequest.newBuilder(URI.create("http://127.0.0.1:"
+                        + port + "/api/agent/web-shell/v1/events/stream"))
+                        .header("Content-Type", "application/json")
+                        .POST(HttpRequest.BodyPublishers.ofString(
+                                json.writeValueAsString(Map.of("sessionId",
+                                        session, "afterSequence", after))));
+            }
+            HttpResponse<InputStream> response = client.send(request
+                    .header("X-Qwen-Tenant-Id", tenant)
+                    .header("X-Qwen-Actor-Id", "actor")
+                    .header("Accept", "text/event-stream").build(),
+                    HttpResponse.BodyHandlers.ofInputStream());
+            assertThat(response.statusCode()).as("%s stream of %s", surface,
+                    session).isEqualTo(200);
+            assertThat(response.headers().firstValue("content-type")
+                    .orElse("")).contains("text/event-stream");
+            reader = new Thread(() -> read(response.body()),
+                    "fg13e-" + surface + "-" + session);
+            reader.setDaemon(true);
+            reader.start();
+            observers.add(this);
+        }
+
+        private void read(InputStream body) {
+            try (BufferedReader lines = new BufferedReader(
+                    new InputStreamReader(body, StandardCharsets.UTF_8))) {
+                String id = null;
+                String event = null;
+                StringBuilder data = new StringBuilder();
+                for (String line = lines.readLine(); line != null;
+                        line = lines.readLine()) {
+                    if (line.startsWith("id:")) {
+                        id = line.substring(3).trim();
+                    } else if (line.startsWith("event:")) {
+                        event = line.substring(6).trim();
+                    } else if (line.startsWith("data:")) {
+                        data.append(data.length() == 0 ? "" : "\n")
+                                .append(line.substring(5).stripLeading());
+                    } else if (line.isEmpty() && data.length() > 0) {
+                        JsonNode frame = json.readTree(data.toString());
+                        if (!frame.path("sequence").asText().equals(id)
+                                || !frame.path("type").asText()
+                                        .equals(event)) {
+                            throw new IllegalStateException("SSE id/event "
+                                    + id + "/" + event + " do not name "
+                                    + frame);
+                        }
+                        frames.add(frame);
+                        id = null;
+                        event = null;
+                        data.setLength(0);
+                    }
+                }
+            } catch (IOException | RuntimeException error) {
+                if (!aborted) {
+                    failure.compareAndSet(null, error);
+                }
+            }
+        }
+
+        long lastSequence() {
+            return frames.isEmpty() ? 0
+                    : frames.get(frames.size() - 1).path("sequence").asLong();
+        }
+
+        /** The server has served this client at least its first frame. */
+        void awaitFirstFrame() {
+            await().atMost(Duration.ofSeconds(60))
+                    .pollInterval(Duration.ofMillis(250)).untilAsserted(() -> {
+                        assertThat(failure.get()).isNull();
+                        assertThat(frames).isNotEmpty();
+                    });
+        }
+
+        void abort() throws InterruptedException {
+            aborted = true;
+            client.shutdownNow();
+            reader.join(10_000);
+            assertThat(reader.isAlive()).as("the cut stream stopped").isFalse();
+        }
+
+        /**
+         * The resumed client's frames after the cut one's, once it read the
+         * ledger's tail: one contiguous sequence from 1 with no gap and no
+         * duplicate, each frame exactly its committed row. A cursor the
+         * server ignores replays the cut client's frames and reddens this on
+         * the duplicate, not on a timeout.
+         */
+        List<JsonNode> continueFrom(Observer cut, String session)
+                throws Exception {
+            long tail = ledger(session).size();
+            await().atMost(Duration.ofSeconds(90))
+                    .pollInterval(Duration.ofMillis(250)).untilAsserted(() -> {
+                        assertThat(failure.get()).isNull();
+                        assertThat(lastSequence()).isGreaterThanOrEqualTo(tail);
+                    });
+            abort();
+            List<JsonNode> view = new ArrayList<>(cut.frames);
+            view.addAll(frames);
+            List<Long> sequences = view.stream()
+                    .map(frame -> frame.path("sequence").asLong()).toList();
+            assertThat(sequences).as("%s view of %s: no gap, no duplicate",
+                    surface, session).isEqualTo(LongStream
+                            .rangeClosed(1, sequences.size()).boxed().toList());
+            List<JsonNode> expected = new ArrayList<>();
+            for (Map<String, Object> row : ledger(session)
+                    .subList(0, view.size())) {
+                expected.add(served(row, surface));
+            }
+            assertThat(view).as("%s frames of %s are the ledger", surface,
+                    session).isEqualTo(expected);
+            return view;
+        }
+    }
+
+    @AfterEach
+    void stop() {
+        for (Observer observer : observers) {
+            try {
+                observer.client.shutdownNow();
+            } catch (RuntimeException ignored) {
+                // The client is already closed.
+            }
+        }
+        saveEvidence();
+        if (model != null) {
+            model.stop(0);
+        }
+        if (modelExecutor != null) {
+            modelExecutor.close();
+        }
+        if (harness != null) {
+            harness.destroyForcibly();
+            try {
+                assertThat(harness.waitFor(10, TimeUnit.SECONDS)).isTrue();
+            } catch (InterruptedException interrupted) {
+                Thread.currentThread().interrupt();
+            }
+        }
+        if (spring != null) {
+            spring.close();
+        }
+        try {
+            HostedDurableWorkers.stop(temporary.resolve("broker"), json);
+        } catch (Exception stopped) {
+            // The worker directory is already gone.
+        }
+        try {
+            if (hookRoot != null) {
+                try (var units = Files.list(hookRoot)) {
+                    for (Path unit : units.toList()) {
+                        if (Files.exists(unit.resolve("cgroup.kill"))) {
+                            Files.writeString(unit.resolve("cgroup.kill"), "1");
+                        }
+                    }
+                }
+                try (var entries = Files.walk(hookRoot)) {
+                    for (Path entry : entries.filter(Files::isDirectory)
+                            .sorted(Comparator.reverseOrder()).toList()) {
+                        Files.deleteIfExists(entry);
+                    }
+                }
+            }
+        } catch (IOException ignored) {
+            // Best effort: a still-populated unit keeps the subtree.
+        }
+        http.close();
+    }
+
+    private void saveEvidence() {
+        if (jdbc == null) {
+            return;
+        }
+        try {
+            Path destination = Files.createDirectories(
+                    Path.of("target/fg13-cases", "fg13e", tenant));
+            System.out.println("HOSTED_HOOK_STREAM_GAP_EVIDENCE "
+                    + destination.toAbsolutePath());
+            json.writeValue(destination.resolve("frames.json").toFile(),
+                    observers.stream().map(observer -> Map.of("surface",
+                            observer.surface, "frames", observer.frames))
+                            .toList());
+            json.writeValue(destination.resolve("events.json").toFile(),
+                    jdbc.queryForList("SELECT session_id, turn_id,"
+                            + " event_type, terminal FROM ma_public_event"
+                            + " WHERE tenant_id = ? ORDER BY sequence_id",
+                            tenant));
+            json.writeValue(destination.resolve("model-requests.json")
+                    .toFile(), modelRequests);
+            for (String name : List.of("processes.jsonl",
+                    "harness-" + boots + ".log")) {
+                Path source = temporary.resolve(name);
+                if (Files.exists(source)) {
+                    Files.copy(source, destination.resolve(source.getFileName()));
+                }
+            }
+        } catch (IOException ignored) {
+            // Evidence is best effort; the assertions already reported.
+        }
+    }
+}
@@ -5,6 +5,7 @@
 import { spawn } from 'node:child_process';
 import {
  closeSync,
+  existsSync,
  fsyncSync,
  openSync,
  readFileSync,
@@ -43,6 +44,7 @@ function record(event, fields = {}) {
 const solo =
  signalMode === 'hook-solo' ||
  signalMode === 'hook-deny' ||
+  signalMode === 'hook-release' ||
  signalMode === 'hook-fail';

 if (mode === 'child') {
@@ -78,7 +80,15 @@ if (mode === 'child') {
    );
    record('hook.hold', { childPid: child.pid });
  }
-  if (Number.isFinite(holdMs) && holdMs >= 0) {
+  if (signalMode === 'hook-release') {
+    // FG13e: the caller releases the Hook only after both clients are cut,
+    // so the deny is guaranteed to land while neither is connected. A fixed
+    // sleep cannot order the cut against the deny.
+    const release = `${evidence}.release`;
+    while (!existsSync(release)) {
+      await new Promise((resolve) => setTimeout(resolve, 50));
+    }
+  } else if (Number.isFinite(holdMs) && holdMs >= 0) {
    await new Promise((resolve) => setTimeout(resolve, holdMs));
  } else {
    // An indefinite hold must suspend this branch: setInterval alone keeps
@@ -91,7 +101,7 @@ if (mode === 'child') {
    child.kill('SIGKILL');
    await new Promise((resolve) => setTimeout(resolve, 500));
  }
-  const deny = signalMode === 'hook-deny';
+  const deny = signalMode === 'hook-deny' || signalMode === 'hook-release';
  // FG13f: a failed evaluation is what failClosed wraps. The process exits
  // without a decision so the receipt is a failure, not a deny.
  if (signalMode === 'hook-fail') {
@@ -458,6 +458,15 @@ describe('Hosted real-process gates', () => {
    expect(upload.with.path).toContain(
      'packages/managed-agent-server/target/fg12-cases/',
    );
+    // FG13e's Hook stream-gap evidence lives in target/fg13-cases/.
+    expect(upload.with.path).toContain(
+      'packages/managed-agent-server/target/fg13-cases/',
+    );
+    // The writer side of the same path: the Java literal must stay under
+    // the directory the step uploads, or the artifact is silently empty.
+    expect(read(hostedTestPath('HostedHookStreamGapIT'))).toContain(
+      '"target/fg13-cases"',
+    );
    // The runner writes the Step 3 gate evidence under .qwen/, a hidden
    // path that upload-artifact drops silently without the flag. The arms
    // that write it run in the failover job, so the upload lives there.
@@ -504,10 +513,24 @@ describe('Hosted real-process gates', () => {
    expect(delegate.run).toContain('/sys/fs/cgroup/fg8/smoke/cgroup.procs');
  });

+  it('sweeps every Hook cgroup subtree a Hosted IT allocates', () => {
+    const owners = hostedTestClasses().filter(({ file }) =>
+      read(file).includes('QWEN_MANAGED_HOOK_CGROUP_ROOT'),
+    );
+    expect(owners.map(({ name }) => name)).toEqual(
+      expect.arrayContaining(['HostedExtensionsIT', 'HostedHookStreamGapIT']),
+    );
+    for (const { name, file } of owners) {
+      const source = javaCode(read(file));
+      expect(source, name).toContain('Files.walk(');
+      expect(source, name).toContain('hookRoot');
+    }
+  });
+
  it.each([
-    // Step ceilings today (182 + 20 + 50 + 20): Verify, the Broker fault
+    // Step ceilings today (185 + 20 + 50 + 20): Verify, the Broker fault
    // gates, admission and O4. The failover arms moved to their own job.
-    ['hosted-harness-mysql', 272, 285],
+    ['hosted-harness-mysql', 275, 288],
    // Twenty failover arms at 10 each.
    ['hosted-failover-e2e', 200, 215],
  ])(
@@ -732,14 +755,15 @@ describe('Hosted real-process gates', () => {
      // the step to 179 minutes. The FG12f delivery allowance adds 20 s,
      // moving the fork to 10100. The FG11d retire-at-intent cell adds 18 s,
      // moving the fork to 10118. Cleanup adds 60 s, moving the fork to
-      // 10178 and the step to 180 minutes. The FG11d close-queued cell adds
-      // 134 s (2x its measured 67.2 s), so the fork moves to 10312 and the
-      // step to 182 minutes):
+      // 10178 and the step to 180 minutes. The FG13e Hook stream-gap cell
+      // adds 150 s, moving the fork to 10328 and the step to 183 minutes;
+      // the FG11d close-queued cell adds 134 s (2x its measured 67.2 s),
+      // moving the fork to 10462 and the step to 185 minutes):
      // the relational check above tolerates
      // anything up to step*60, so the value is pinned literally here —
      // an unnoticed raise has no other red to land on.
      if (profile === 'hosted-harness-mysql') {
-        expect(forkSeconds).toBe(10312);
+        expect(forkSeconds).toBe(10462);
      }
    },
  );
@@ -817,13 +841,14 @@ describe('Hosted real-process gates', () => {
  // FG12f delivery entry moves from 30s to 50s at 2x (four cases, 24.16s
  // measured), so it becomes 9500s; the FG11d retire-at-intent cell adds 18s
  // (2x its measured 8.6s), so it becomes 9518s.
-  // Class-boundary cleanup adds 60s for 9578s; the FG11d close-queued cell
-  // adds 134s (2x its measured 67.2s), so it becomes 9712s. The largest
+  // Class-boundary cleanup adds 60s for 9578s; the FG13e Hook stream-gap
+  // cell adds 150s, so it becomes 9728s; the FG11d close-queued cell adds
+  // 134s (2x its measured 67.2s), so it becomes 9862s. The largest
  // ceiling (600s) plus
  // that healthy sum still fits the fork. A larger ceiling lets failsafe
  // kill the fork mid-run, and the classes left unrun never report, failing
  // the gate undiagnosably (#13780).
-  const hostedForkHealthySeconds = 9712;
+  const hostedForkHealthySeconds = 9862;
  const hostedForkSeconds = () =>
    Number(
      read('packages/managed-agent-server/pom.xml')
@@ -895,6 +920,7 @@ describe('Hosted real-process gates', () => {
      'HostedHookCancelFaultGatesIT',
      'HostedHookFaultGatesIT',
      'HostedHookReplyLossFaultGatesIT',
+      'HostedHookStreamGapIT',
      'HostedMcpReplyLossIT',
      'HostedMcpRestartFaultGatesIT',
      'HostedMcpResultDeliveryIT',