vrr/qwen-code
2
0
Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2026-10-21 05:40:14 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 30

test(managed-agent): add the FG13d Hook cancel and timeout fault gates (#14533)

Browse source 
* test(managed-agent): add the FG13d Hook cancel and timeout fault gates

Extend the packaged Hook rig with a descendant that ignores SIGTERM, and
pin two arms against the real stack: a cancel must escalate to cgroup.kill
after the grace and settle cancelled only once the unit is empty; a 2 s
timeout must kill the unit, keep the output unapplied, and block the tool.
The class adds 200 s to the shared fork budget and is registered in the
pinned Hosted class list.

* test(managed-agent): drop the unused FG13c helpers from the FG13d class

The new class was copied from the owner-restart rig, so it carried the
Harness-kill, worker-kill and status helpers those cells used. This slice
cancels through the Harness route and waits on JDBC, so those helpers are
dead; removing them keeps the class to what it exercises.

* docs(managed-agent): cite the whole Hook unit terminate() range in the FG13d design

* test(managed-agent): check the drained unit at the first terminal observation

The cells waited for the terminal record and only then waited out the
descendant, so a regression that settled early while the drain continued
still passed. Inspect the first terminal observation instead: the
descendant must be gone and the unit empty at that instant, and document
the early-terminal mutation witness.
This commit is contained in:
Shaojin Wen 2026-10-18 14:40:11 +08:00 • committed by GitHub
parent 193e365290
commit e5ceeb4bea
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 702 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

10
.github/workflows/sdk-java.yml vendored
View file

@ -679,10 +679,10 @@ jobs:
# The step ceilings today (95 + 10 + 20 + 50 = 175 — the FG8 letters
# grew the shared failsafe fork to 3600 s, FG9 slice 3 to 3990 s, the
# FG10b–f workflow relay letters to 4484 s, the FG12a MCP reply-loss
# letter to 4499 s, the FG12c restart matrix to 4548 s and the FG13c
# Hook owner-restart cells to 4799 s; with FG12c's 49 s the merged
# healthy sum is 4238 s and the fork 4848 s, so Verify wraps it at
# its fork + 600 s plus margin, the Broker fault gates keep 10, and
# letter to 4499 s, the FG12c restart matrix to 4548 s, the FG13c Hook
# owner-restart cells to 4848 s and the FG13d cancel/timeout cells to
# 5048 s, so Verify wraps it at its fork + 600 s plus margin, the
# Broker fault gates keep 10, and
# admission plus O4 keep per-fork ceilings) plus ~13 minutes for the
# uncapped setup steps (Checkout.. cgroup delegation, the report
# checks): 188 + 2. The failover arms run in hosted-failover-e2e.
@ -775,7 +775,7 @@ 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
# 4848 s, so the step wraps to fork + 600 s plus margin.
# 5048 s, so the step wraps to fork + 600 s plus margin.
timeout-minutes: 95
env:
MAVEN_ARGS: '--settings ${{ runner.temp }}/setup-java-m2/settings.xml --toolchains ${{ runner.temp }}/setup-java-m2/toolchains.xml'

70
docs/design/2026-10-18-hosted-hook-cancel-drain-gates-d.md Normal file
View file

@ -0,0 +1,70 @@
# Hosted Hook cancel and timeout gates FG13d
[English](2026-10-18-hosted-hook-cancel-drain-gates-d.md) | [简体中文](2026-10-18-hosted-hook-cancel-drain-gates-d.zh-CN.md)
## Problem and scope
Issue #14449 tracks FG13, the Stage F fault-gate round for the Hosted Hook
domains. FG13c covered the owner-restart matrix; this slice is **FG13d's
first half: cancellation and timeout**, the arms that close the rest of
"cgroup execution/drain" that `HostedExtensionsIT` did not.
A managed command Hook runs in the Runtime worker's delegated cgroup v2
unit, and a command may leave a descendant behind. Cancelling or timing it
out therefore has to signal the whole unit, escalate to `cgroup.kill` when a
descendant ignores the graceful signal, and only report a terminal outcome
once the unit is empty (`hook-command-cgroup.ts:227-248`,
`hookRunner.ts:1415-1439`).
## Decisions
1. **Reuse the FG13c rig.** The new class borrows the Hook-only definition,
the packaged Harness, the durable worker and the delegated root from
`HostedHookFaultGatesIT`, so the gate measures the same owner path.
2. **The fixture gets a TERM-ignoring descendant.** A cancel that only sends
SIGTERM cannot empty the unit while a descendant ignores it; the fixture
child installs a `SIGTERM` handler in the `hook-ignore-term` mode.
3. **Two independent cells, each its own fixture.** Cancellation and timeout
boot separate Spring/Harness/worker stacks, so a stall in one cannot mask
the other. Each keeps `@Timeout(300)`, and the class is budgeted at 200 s.
4. **The cancel and status probes use the Harness routes.** Both go through
the authenticated `/session/:id/hooks/operations/:operationId` routes
with the contract headers, the same entry the FG13c status probe uses.
## Cases
| Case | Inject | Expect |
| ---------------------- | ------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `cancel-held-command` | A `PreToolUse` command Hook holds with a descendant that ignores SIGTERM; cancel the operation through the Harness route. | The command gets SIGTERM; the descendant survives it, so the unit is only emptied by `cgroup.kill` after the grace. The record settles `cancelled` and only then reports; the unit reaches `populated 0`; exactly one physical call; `failClosed` turns the cancelled outcome into a block and the tool never runs. |
| `timeout-held-command` | The same held command with a 2 s Hook timeout and a descendant that ignores SIGTERM. | The command times out, its output is not applied, the record settles with `outcome=timeout`, the unit is killed and reaches `populated 0`, and the model sees the block instead of the file body. |
## Mutation witnesses
Each witness must turn its cell red before the slice merges.
- `cancel-held-command`: remove the `cgroup.kill` fallback from
`CommandCgroup.terminate` (leave only the SIGTERM pass). The descendant
keeps the unit populated, the `populated 0` assertion fails, and the
record cannot honestly report `cancelled`.
- `timeout-held-command`: skip termination on the timed-out branch so the
unit is left running. The cgroup assertion fails while the record still
settles `timeout`.
- `early-terminal` (both cells): publish the result before
`processUnit.terminate` returns, so the record reaches its terminal state
while the unit is still populated. `awaitTerminalDrained` inspects the
first terminal observation, so both cells fail on the populated unit.
## Non-goals
The rest of FG13d — cancel before dispatch, a dropped `hook-cancel` reply, a
cancel with the worker stopped, and the DELETE async-drain with
`SessionEnd`/`SessionDelete` — stays on #14449. Also out: FG13a, FG13b,
FG13e, FG13f, local non-managed Hooks, Windows/macOS and #13133's design
items.
## Validation
The full `hosted-harness-mysql` lane on Linux with MySQL 8.4 and a delegated
cgroup v2 root is the evidence. Checkstyle, the CI fork-budget guard and the
pinned class list move with the new class. A skipped cell is not a passing
gate.

60
docs/design/2026-10-18-hosted-hook-cancel-drain-gates-d.zh-CN.md Normal file
View file

@ -0,0 +1,60 @@
# Hosted Hook 取消与超时门禁 FG13d
[English](2026-10-18-hosted-hook-cancel-drain-gates-d.md) | [简体中文](2026-10-18-hosted-hook-cancel-drain-gates-d.zh-CN.md)
## 问题与范围
Issue #14449 跟踪 Hosted Hook 两个域的 Stage F 故障门禁轮次。FG13c 覆盖了
owner 重启矩阵；本切片是 **FG13d 的前半部分：取消与超时**，补齐
`HostedExtensionsIT` 没有覆盖的 “cgroup 执行 / drain” 其余部分。
托管命令 Hook 跑在 Runtime worker 的委派 cgroup v2 单元里，命令还可能留下
后代进程。因此取消或超时必须对整个单元发信号，在后代忽略优雅信号时升级到
`cgroup.kill`，并且只有单元清空后才报告终态（`hook-command-cgroup.ts:227-248`、
`hookRunner.ts:1415-1439`）。
## 决策
1. **复用 FG13c 的装置。** 新类沿用 `HostedHookFaultGatesIT` 的只含 Hook 定义、
打包 Harness、durable worker 与委派 cgroup 根，测的是同一条 owner 路径。
2. **夹具增加忽略 TERM 的后代。** 只发 SIGTERM 无法清空仍被后代占据的单元；
夹具子进程在 `hook-ignore-term` 模式下安装 `SIGTERM` handler。
3. **两个独立用例，各自一套夹具。** 取消与超时分别启动自己的 Spring、Harness
与 worker，一格的停顿不会掩盖另一格。两格均保留 `@Timeout(300)`，该类按
200 s 计入预算。
4. **取消与状态探针都走 Harness 路由。** 两者都通过认证的
`/session/:id/hooks/operations/:operationId` 路由并携带契约头，与 FG13c 的
状态探针同一入口。
## 用例
| 用例 | 注入 | 期望 |
| ---------------------- | --------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `cancel-held-command` | `PreToolUse` 命令 Hook 带一个忽略 SIGTERM 的后代保持运行；通过 Harness 路由取消该操作。 | 命令收到 SIGTERM；后代存活，因此只有宽限期后的 `cgroup.kill` 能清空单元。记录结算为 `cancelled` 且只在那之后报告；单元到达 `populated 0`；恰好一次物理调用；`failClosed` 把取消结果变成阻断，工具不运行。 |
| `timeout-held-command` | 同样的保持型命令，Hook 超时 2 s，后代忽略 SIGTERM。 | 命令超时，输出不被采用，记录结算为 `outcome=timeout`，单元被杀并到达 `populated 0`，模型看到阻断而不是文件正文。 |
## 变异见证
每个见证必须在合入前让自己的用例变红。
- `cancel-held-command`：删掉 `CommandCgroup.terminate` 里的 `cgroup.kill`
兜底（只保留 SIGTERM 一轮）。后代继续占据单元，`populated 0` 断言失败，
记录也无法如实报告 `cancelled`。
- `timeout-held-command`：超时分支跳过终止，单元被留在运行状态；cgroup 断言
失败，而记录仍会结算 `timeout`。
- `early-terminal`（两格）：在 `processUnit.terminate` 返回前先发布结果，使
记录在单元仍被占据时就进入终态。`awaitTerminalDrained` 检查首次终态观测，
两格都会因单元非空而失败。
## 非目标
FG13d 其余部分——派发前取消、`hook-cancel` 回复丢失、worker 停止时取消、以及
带 `SessionEnd`/`SessionDelete` 的 DELETE 异步 drain——仍留在 #14449。同样不
在范围：FG13a、FG13b、FG13e、FG13f、本地非托管 Hook、Windows/macOS，以及
#13133 的设计项。
## 验证
在 Linux、MySQL 8.4 与委派 cgroup v2 根上跑完整的 `hosted-harness-mysql`
车道即证据。Checkstyle、CI fork 预算守卫与 pin 名单随新类一起移动。被跳过的
用例不算通过。

8
packages/managed-agent-server/pom.xml
View file

@ -416,8 +416,12 @@
owner-restart cells (three cases,
issue #14449) are budgeted at 300 s:
the healthy sum becomes 4238 s, so the
fork moves to 4848 s. -->
<forkedProcessTimeoutInSeconds>4848</forkedProcessTimeoutInSeconds>
fork moves to 4848 s. The FG13d
cancel/timeout cells (two cases, issue
#14449) add 200 s: the healthy sum
becomes 4438 s, so the fork moves to
5048 s. -->
<forkedProcessTimeoutInSeconds>5048</forkedProcessTimeoutInSeconds>
</configuration>
<executions>
<execution>

551
packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedHookCancelFaultGatesIT.java Normal file
View file

@ -0,0 +1,551 @@
package com.alibaba.qwen.code.managedagent;
import com.alibaba.qwen.code.daemon.HarnessSessionRef;
import com.alibaba.qwen.code.managedagent.harness.HarnessConnector;
import com.alibaba.qwen.code.managedagent.store.ManagedSessionStore;
import com.alibaba.qwen.code.managedagent.store.WorkspaceStorageGuard;
import com.alibaba.qwen.code.runtimebroker.WorkspaceExecutionProfile;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
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.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.IOException;
import java.lang.reflect.Field;
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.NoSuchFileException;
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 static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertAll;
// FG13d (issues #14054/#14449): cancellation and timeout for the managed
// command Hook. The command runs in the Runtime worker's delegated cgroup
// unit and keeps a descendant of its own, so a cancel or a timeout must
// signal the whole unit, escalate to cgroup.kill when a descendant ignores
// SIGTERM, and only report the terminal outcome once the unit is empty.
@EnabledOnOs(OS.LINUX)
class HostedHookCancelFaultGatesIT {
private static final String PROFILE = "hosted-workspace-files/1";
private static final String TOKEN = "hook-fault-fixture-token";
private static final String CAPABILITY = "sha256:" + "a".repeat(64);
private static final String PROOF = "FG13_HOOK_TOOL_PROOF";
private final ObjectMapper json = new ObjectMapper();
private final HttpClient http = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(3)).build();
private final String tenant = "hook-fault-" + UUID.randomUUID();
private final List<JsonNode> modelRequests = new CopyOnWriteArrayList<>();
private final AtomicReference<Throwable> modelFailure = new AtomicReference<>();
@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;
@Test
@Timeout(300)
void fg13dHeldCancelEscalatesToCgroupKillAndSettlesCancelledOnMySql() throws Exception {
boot("60000", "60000", "hook-ignore-term");
JsonNode definition = api("POST", "/v1/agents", definition(), "definition", 202);
String session = createSession(definition.path("id").asText(), "initial");
submit(session, "FG13_CANCEL");
waitFor(() -> {
assertThat(events("hook.call")).hasSize(1);
assertThat(events("hook.child")).hasSize(1);
});
JsonNode hook = events("hook.call").getFirst();
JsonNode child = events("hook.child").getFirst();
assertThat(running(hook)).isTrue();
assertThat(running(child)).isTrue();
assertThat(Files.readString(cgroup(hook).resolve("cgroup.events"))).contains("populated 1");
String executionId = hookExecution(session).path("hookExecutionId").asText();
JsonNode cancelled = cancelHook(session, executionId);
assertThat(cancelled.path("cancelRequested").asBoolean()).isTrue();
// The command dies on the graceful SIGTERM; its descendant ignores
// it, so only the unit-wide cgroup.kill can empty the unit, and the
// record may only say cancelled once that happened.
awaitTerminalDrained(session, "cancelled", child);
assertThat(events("hook.call")).hasSize(1);
assertThat(events("hook.allow")).isEmpty();
assertThat(hookResult(session).path("outcome").asText()).isEqualTo("cancelled");
waitFor(() -> assertThat(modelRequests.toString()).contains("Hook evaluation failed."));
assertThat(modelRequests.toString()).doesNotContain(PROOF);
assertThat(modelFailure.get()).isNull();
}
@Test
@Timeout(300)
void fg13dTimeoutKillsTheUnitAndBlocksTheToolOnMySql() throws Exception {
boot("2000", "60000", "hook-ignore-term");
JsonNode definition = api("POST", "/v1/agents", definition(), "definition", 202);
String session = createSession(definition.path("id").asText(), "initial");
submit(session, "FG13_TIMEOUT");
waitFor(() -> {
assertThat(events("hook.call")).hasSize(1);
assertThat(events("hook.child")).hasSize(1);
});
JsonNode hook = events("hook.call").getFirst();
assertThat(running(hook)).isTrue();
assertThat(Files.readString(cgroup(hook).resolve("cgroup.events"))).contains("populated 1");
// The timed-out command and its TERM-ignoring descendant must be
// gone before the unit counts as drained, and the record may only
// report the terminal state once that happened.
JsonNode child = events("hook.child").getFirst();
awaitTerminalDrained(session, "settled", child);
assertThat(hookResult(session).path("outcome").asText()).isEqualTo("timeout");
assertThat(events("hook.call")).hasSize(1);
assertThat(events("hook.allow")).isEmpty();
waitFor(() -> assertThat(modelRequests.toString()).contains("Hook evaluation failed."));
assertThat(modelRequests.toString()).doesNotContain(PROOF);
assertThat(modelFailure.get()).isNull();
}
// The record may report a terminal state only once the unit is empty:
// inspect the first observation, not a later poll, so a regression that
// settles early while the drain continues still fails the cell.
private void awaitTerminalDrained(String session, String state, JsonNode child) throws Exception {
long deadline = System.nanoTime() + Duration.ofSeconds(60).toNanos();
while (System.nanoTime() < deadline) {
JsonNode execution = hookExecution(session);
if (state.equals(execution.at("/run/state").asText())) {
assertThat(execution.at("/run/execution").asText()).isEqualTo("settled");
assertThat(running(child))
.as("descendant alive when the record first reports %s", state).isFalse();
assertHookUnitsEmpty();
return;
}
Thread.sleep(100);
}
throw new AssertionError("The Hook record never reported " + state);
}
private void boot(String timeoutMs, String holdMs, String signalMode) throws Exception {
assertThat(System.getProperty("mysql.url")).as("Hook fault gates require disposable MySQL 8.4")
.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.writeString(workspace.resolve("proof.txt"), PROOF);
Files.setLastModifiedTime(workspace.getParent(), FileTime.fromMillis(1));
Files.createDirectory(temporary.resolve("decoy"));
Files.createDirectory(temporary.resolve("broker"),
PosixFilePermissions.asFileAttribute(PosixFilePermissions.fromString("rwx------")));
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-fault-" + UUID.randomUUID()));
Path program = temporary.resolve("hook-fault-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<>(Map.of("hookId", "pre", "eventName", "PreToolUse",
"sequential", false, "failClosed", true, "async", false, "config", Map.of("type", "command", "command",
quote(node) + " " + quote(program.toString()) + " hook " + quote(evidence.toString())
+ " " + quote(holdMs) + " " + quote(signalMode), "timeout", Integer.parseInt(timeoutMs))));
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", "fg13-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();
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=hook-fault",
"--qwen.managed-agent.runtime-broker.credential-key=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
"--qwen.managed-agent.runtime-broker.node-executable=" + node,
"--qwen.managed-agent.runtime-broker.worker-entry=" + worker,
"--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', 'Hooks', ?, ?, '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));
startHarness();
}
private String catalogDigest() {
return DigestUtils.sha256Hex(hookDigest + 1);
}
private Map<String, Object> definition() {
return Map.of("model", Map.of(), "instructions", "Exercise the pinned Hook gate.",
"tools", List.of(Map.of("type", "hosted_profile", "profile", PROFILE)),
"permission_policy", Map.of("approval_mode", "yolo"),
"hook_catalog", Map.of("catalog_id", "fg13-hooks", "catalog_revision", 1,
"definition_digest", catalogDigest()));
}
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 JsonNode hookExecution(String session) {
List<JsonNode> records = HostedGateTestSupport.extensionRecords(jdbc, tenant, session, "hook_execution").stream()
.filter(record -> !"__plan__".equals(record.path("hookId").asText()))
.toList();
assertThat(records).as("hook execution records").hasSize(1);
assertThat(records.getFirst().path("eventName").asText()).isEqualTo("PreToolUse");
return records.getFirst();
}
private JsonNode hookResult(String session) throws Exception {
JsonNode ref = hookExecution(session).path("resultRef");
assertThat(ref.isNull()).as("hook result ref").isFalse();
String body = jdbc.queryForObject("SELECT CONVERT(inline_bytes USING utf8mb4) FROM ma_resource"
+ " WHERE session_scope_key = ? AND resource_id = ? AND tenant_id = ? AND session_id = ?",
String.class, ManagedSessionStore.sessionScopeKey(tenant, session),
ref.path("resourceId").asText(), tenant, session);
return json.readTree(body);
}
// The Harness status route identifies the caller by the client id its
// attach minted and the process generation it attached to; the connector
// keeps both on the private attachment ref.
private record HarnessAttach(String clientId, String bootId) {}
private HarnessAttach attachment(HarnessConnector connector, String session) throws Exception {
Field field = connector.getClass().getDeclaredField("attachments");
field.setAccessible(true);
for (Object value : ((Map<?, ?>) field.get(connector)).values()) {
if (value instanceof HarnessSessionRef ref && session.equals(ref.getHarnessSessionId())) {
return new HarnessAttach(ref.getHarnessClientId(), ref.getHarnessBootId());
}
}
return null;
}
private JsonNode cancelHook(String session, String executionId) throws Exception {
HarnessAttach attach = attachment(spring.getBean(HarnessConnector.class), session);
assertThat(attach).as("Session attachment before cancel").isNotNull();
var request = HttpRequest.newBuilder(URI.create("http://127.0.0.1:" + harnessPort
+ "/session/" + session + "/hooks/operations/" + executionId + "/cancel"))
.timeout(Duration.ofSeconds(30)).header("Authorization", "Bearer " + TOKEN)
.header("X-Qwen-Client-Id", attach.clientId())
.header("X-Qwen-Harness-Protocol-Version", "1")
.header("X-Qwen-Harness-Boot-Id", attach.bootId())
.POST(HttpRequest.BodyPublishers.noBody()).build();
var response = http.send(request, HttpResponse.BodyHandlers.ofString());
assertThat(response.statusCode()).as("hook cancel: %s", response.body()).isEqualTo(200);
return json.readTree(response.body());
}
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", "hook-fault-fixture"),
"modelProviders", Map.of("openai", List.of(Map.of("id", "hook-fault-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();
assertThat(http.send(HttpRequest.newBuilder(URI.create("http://127.0.0.1:" + harnessPort + "/capabilities"))
.timeout(Duration.ofSeconds(2)).header("Authorization", "Bearer " + TOKEN).build(),
HttpResponse.BodyHandlers.discarding()).statusCode()).isEqualTo(200);
});
}
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("FG13_")) {
String text = message.path("content").toString();
var matcher = java.util.regex.Pattern.compile("FG13_[A-Z_]+").matcher(text);
prompt = "";
while (matcher.find()) {
prompt = matcher.group();
}
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()
&& List.of("FG13_CANCEL", "FG13_TIMEOUT").contains(prompt);
var chunk = json.createObjectNode().put("id", "hook-fault").put("object", "chat.completion.chunk")
.put("created", 0).put("model", "hook-fault-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", "FG13_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 boolean running(JsonNode process) throws Exception {
if (!Files.readString(Path.of("/proc/sys/kernel/random/boot_id")).strip().equals(process.path("bootId").asText())) {
return false;
}
Path stat = Path.of("/proc", process.path("pid").asText(), "stat");
String value;
try {
value = Files.readString(stat);
} catch (NoSuchFileException stopped) {
return false;
}
String[] fields = value.substring(value.lastIndexOf(')') + 2).split(" ");
return !"Z".equals(fields[0]) && fields[19].equals(process.path("startTicks").asText());
}
private Path cgroup(JsonNode process) {
Path path = Path.of("/sys/fs/cgroup" + process.path("cgroup").asText().substring(3));
assertThat(path).startsWith(hookRoot);
return path;
}
private void assertHookUnitsEmpty() throws Exception {
try (var units = Files.list(hookRoot)) {
for (Path unit : units.filter(Files::isDirectory).toList()) {
assertThat(Files.readString(unit.resolve("cgroup.events"))).contains("populated 0");
}
}
}
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;
}
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(60)).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);
}
private static String quote(String text) {
return "'" + text.replace("'", "'\"'\"'") + "'";
}
@AfterEach
void stop() {
assertAll(this::saveEvidence,
() -> { if (model != null) {
model.stop(0);
} },
() -> { if (modelExecutor != null) {
modelExecutor.close();
} },
() -> { if (harness != null) {
harness.destroyForcibly();
assertThat(harness.waitFor(10, TimeUnit.SECONDS)).isTrue();
} },
() -> { if (evidence != null && Files.exists(evidence)) {
for (String kind : List.of("hook.call", "hook.child")) {
for (JsonNode process : events(kind)) {
if (running(process)) {
ProcessHandle.of(process.path("pid").asLong()).orElseThrow().destroyForcibly();
await().atMost(Duration.ofSeconds(10)).until(() -> !running(process));
}
}
}
} },
() -> { if (spring != null) {
spring.close();
} },
() -> HostedDurableWorkers.stop(temporary.resolve("broker"), json),
() -> {
if (hookRoot == null) {
return;
}
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.delete(entry);
}
}
}, http::close);
}
private void saveEvidence() throws Exception {
if (temporary == null) {
return;
}
Path destination = Files.createDirectories(Path.of("target/failsafe-reports/hosted-hook-fault-evidence", tenant));
System.out.println("HOSTED_HOOK_FAULT_EVIDENCE " + destination.toAbsolutePath());
json.writeValue(destination.resolve("model-requests.json").toFile(), modelRequests);
for (String name : List.of("processes.jsonl", "harness-1.log", "harness-2.log")) {
Path source = temporary.resolve(name);
if (Files.exists(source)) {
Files.copy(source, destination.resolve(source.getFileName()));
}
}
if (jdbc != null) {
json.writeValue(destination.resolve("bindings.json").toFile(), jdbc.queryForList(
"SELECT binding_id, runtime_generation, binding_state, isolation_key FROM ma_runtime_binding WHERE tenant_id = ?", tenant));
json.writeValue(destination.resolve("hook-executions.json").toFile(),
jdbc.queryForList("SELECT record_id, domain, state, runtime_state FROM ma_session_record"
+ " WHERE tenant_id = ? AND domain = 'hook_execution'", tenant));
}
}
}

3
packages/managed-agent-server/src/test/resources/hosted-hook-fault-process.mjs
View file

@ -43,6 +43,9 @@ function record(event, fields = {}) {
const solo = signalMode === 'hook-solo';
if (mode === 'child') {
// FG13d: the graceful pass signals every member; this descendant has to
// survive it so only the unit-wide cgroup.kill can empty the unit.
if (signalMode === 'hook-ignore-term') process.on('SIGTERM', () => {});
record('hook.child');
setInterval(() => {}, 1000);
} else if (mode === 'hook') {

11
scripts/tests/hosted-process-ci.test.js
View file

@ -592,12 +592,13 @@ describe('Hosted real-process gates', () => {
// reply-loss letter adds ~15 s, so the fork moves to 4499; the FG12c
// restart matrix (five cases) adds ~49 s and the FG13c Hook
// owner-restart cells add 300 s, so the fork moves to 4848 and the
// Verify step stays at 95 minutes):
// Verify step stays at 95 minutes; the FG13d cancel/timeout cells add
// two cases budgeted at 200 s, so the fork moves to 5048):
// 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(4848);
expect(forkSeconds).toBe(5048);
}
},
);
@ -631,10 +632,11 @@ describe('Hosted real-process gates', () => {
// class's end. FG12c's five-case restart matrix measures ~49s healthy
// locally (~98s budgeted at 2x); the FG13c Hook owner-restart cells
// (issue #14449, three cases) are budgeted at 300s until the lane
// measures them. The healthy sum becomes 4238s.
// measures them. The healthy sum becomes 4238s; the FG13d cancel/timeout
// cells (issue #14449, two cases) add 200s, so it becomes 4438s.
// A larger ceiling lets failsafe kill the fork mid-run, and the classes
// left unrun never report, failing the gate undiagnosably (#13780).
const hostedForkHealthySeconds = 4238;
const hostedForkHealthySeconds = 4438;
const hostedForkSeconds = () =>
Number(
read('packages/managed-agent-server/pom.xml')
@ -699,6 +701,7 @@ describe('Hosted real-process gates', () => {
'HostedConcurrentTurnBurstMySqlIT',
'HostedExtensionsIT',
'HostedHarnessMySqlIT',
'HostedHookCancelFaultGatesIT',
'HostedHookFaultGatesIT',
'HostedMcpReplyLossIT',
'HostedMcpRestartFaultGatesIT',