diff --git a/.github/workflows/sdk-java.yml b/.github/workflows/sdk-java.yml index bb6d2b216d..d0aa021e8c 100644 --- a/.github/workflows/sdk-java.yml +++ b/.github/workflows/sdk-java.yml @@ -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' diff --git a/docs/design/2026-10-18-hosted-hook-cancel-drain-gates-d.md b/docs/design/2026-10-18-hosted-hook-cancel-drain-gates-d.md new file mode 100644 index 0000000000..da9ed0d93d --- /dev/null +++ b/docs/design/2026-10-18-hosted-hook-cancel-drain-gates-d.md @@ -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. diff --git a/docs/design/2026-10-18-hosted-hook-cancel-drain-gates-d.zh-CN.md b/docs/design/2026-10-18-hosted-hook-cancel-drain-gates-d.zh-CN.md new file mode 100644 index 0000000000..62a31bfec6 --- /dev/null +++ b/docs/design/2026-10-18-hosted-hook-cancel-drain-gates-d.zh-CN.md @@ -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 名单随新类一起移动。被跳过的 +用例不算通过。 diff --git a/packages/managed-agent-server/pom.xml b/packages/managed-agent-server/pom.xml index 0586d67650..6da01a54cd 100644 --- a/packages/managed-agent-server/pom.xml +++ b/packages/managed-agent-server/pom.xml @@ -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. --> - 4848 + 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. --> + 5048 diff --git a/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedHookCancelFaultGatesIT.java b/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedHookCancelFaultGatesIT.java new file mode 100644 index 0000000000..8a60c1a6be --- /dev/null +++ b/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedHookCancelFaultGatesIT.java @@ -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 modelRequests = new CopyOnWriteArrayList<>(); + private final AtomicReference 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 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 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 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 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 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 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; + } + + 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)); + } + } +} 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 9741a9d76f..8914923e65 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 @@ -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') { diff --git a/scripts/tests/hosted-process-ci.test.js b/scripts/tests/hosted-process-ci.test.js index 8997839ae2..27cee9d5c2 100644 --- a/scripts/tests/hosted-process-ci.test.js +++ b/scripts/tests/hosted-process-ci.test.js @@ -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',