diff --git a/.github/workflows/sdk-java.yml b/.github/workflows/sdk-java.yml
index a9c232260bc..bb84ad46ead 100644
--- a/.github/workflows/sdk-java.yml
+++ b/.github/workflows/sdk-java.yml
@@ -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
diff --git a/docs/design/2026-10-20-hosted-hook-stream-gap-gates-e.md b/docs/design/2026-10-20-hosted-hook-stream-gap-gates-e.md
new file mode 100644
index 00000000000..55dc3810db5
--- /dev/null
+++ b/docs/design/2026-10-20-hosted-hook-stream-gap-gates-e.md
@@ -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.
diff --git a/docs/design/2026-10-20-hosted-hook-stream-gap-gates-e.zh-CN.md b/docs/design/2026-10-20-hosted-hook-stream-gap-gates-e.zh-CN.md
new file mode 100644
index 00000000000..2ead12fbe60
--- /dev/null
+++ b/docs/design/2026-10-20-hosted-hook-stream-gap-gates-e.zh-CN.md
@@ -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 分钟。
diff --git a/packages/managed-agent-server/pom.xml b/packages/managed-agent-server/pom.xml
index 59f74c6db49..3425e0fac99 100644
--- a/packages/managed-agent-server/pom.xml
+++ b/packages/managed-agent-server/pom.xml
@@ -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. -->
- 10312
+ 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. -->
+ 10462
diff --git a/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedHookStreamGapIT.java b/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedHookStreamGapIT.java
new file mode 100644
index 00000000000..00ee9be5fd0
--- /dev/null
+++ b/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedHookStreamGapIT.java
@@ -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 modelRequests = new CopyOnWriteArrayList<>();
+ private final AtomicReference modelFailure =
+ new AtomicReference<>();
+ private final List 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 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 publicView = new Observer("public", session,
+ publicCutAt).continueFrom(publicCut, session);
+ List 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 union, long cutAt) {
+ assertThat(union.size())
+ .as("frames landed after the cut and the resume served them")
+ .isGreaterThan((int) cutAt);
+ List 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 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 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 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 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 events(String kind) throws Exception {
+ if (!Files.exists(evidence)) {
+ return List.of();
+ }
+ List 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> 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 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 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 frames = new CopyOnWriteArrayList<>();
+ private final AtomicReference 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 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 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 view = new ArrayList<>(cut.frames);
+ view.addAll(frames);
+ List 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 expected = new ArrayList<>();
+ for (Map 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.
+ }
+ }
+}
diff --git a/packages/managed-agent-server/src/test/resources/hosted-hook-fault-process.mjs b/packages/managed-agent-server/src/test/resources/hosted-hook-fault-process.mjs
index 193a4f42558..a01dbb8e7ee 100644
--- a/packages/managed-agent-server/src/test/resources/hosted-hook-fault-process.mjs
+++ b/packages/managed-agent-server/src/test/resources/hosted-hook-fault-process.mjs
@@ -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') {
diff --git a/scripts/tests/hosted-process-ci.test.js b/scripts/tests/hosted-process-ci.test.js
index 852089fb083..0f6908814bc 100644
--- a/scripts/tests/hosted-process-ci.test.js
+++ b/scripts/tests/hosted-process-ci.test.js
@@ -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',