diff --git a/.github/workflows/sdk-java.yml b/.github/workflows/sdk-java.yml index 12c650bb60..c0c1d297dc 100644 --- a/.github/workflows/sdk-java.yml +++ b/.github/workflows/sdk-java.yml @@ -688,13 +688,15 @@ jobs: # liveness gates to 6083 s, which moves Verify to 115 minutes, the # FG13d dropped-reply cancel cell to 6233 s, still inside those 115 # minutes, and the FG12b store-failure class to 6353 s; the FG12f - # stream-gap class adds 242 s, so the fork moves to 6585 s and Verify to - # 120 minutes, which wraps 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): 200 + 13. The failover arms run in + # stream-gap class adds 242 s, so the fork moves to 6585 s; the FG13a + # execute lost-reply cell adds 150 s and the FG13a status lost-reply + # cell another 150 s, moving the fork to 6885 s and Verify to 125 + # minutes to wrap 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): 205 + 13. The failover arms run in # hosted-failover-e2e. - timeout-minutes: 213 + timeout-minutes: 218 services: mysql: image: 'mysql:8.4.6' @@ -783,8 +785,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 - # 6585 s, so the step wraps to fork + 600 s plus margin. - timeout-minutes: 120 + # 6885 s, so the step wraps to fork + 600 s plus margin. + timeout-minutes: 125 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'] }}" diff --git a/docs/design/2026-10-18-hosted-hook-execute-reply-loss-gates-a.md b/docs/design/2026-10-18-hosted-hook-execute-reply-loss-gates-a.md new file mode 100644 index 0000000000..3e0e3f3201 --- /dev/null +++ b/docs/design/2026-10-18-hosted-hook-execute-reply-loss-gates-a.md @@ -0,0 +1,66 @@ +# Hosted Hook execute reply-loss gates FG13a + +[English](2026-10-18-hosted-hook-execute-reply-loss-gates-a.md) | [简体中文](2026-10-18-hosted-hook-execute-reply-loss-gates-a.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 and FG13d the cancellation +and timeout arms; this slice starts **FG13a, lost replies**, with the arm a +command Hook pays first: the reply to `hook-execute` is lost after the +worker admitted the operation. + +The dispatch reply is the one place where a lost answer could look like a +lost call. The worker already owns the receipt, so the retry must resolve +the same operation through the owner's status instead of dispatching a +second command. + +## Decisions + +1. **Reuse the FG13d rig and broker proxy.** The class borrows the Hook-only + definition, the packaged Harness, the durable worker and the forwarding + proxy from the FG13d stack base; only the `hook-execute` control reply is + dropped, and every other call passes through. +2. **The retry is a status poll, not a second execute.** The Harness falls + back to `hook-status` on the record's own `runtimeSessionId` when the + dispatch reply is lost, which is what keeps the physical call at one. +3. **Two cells, one fixture each.** Both hold a solo command for one second + and allow; a dropped dispatch reply must resolve through status, and a + dropped status reply must stay unknown until a later poll of the same + owner settles it. Neither may add a second physical call. + +## Case + +| Case | Inject | Expect | +| -------------------- | ----------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `execute-reply-lost` | A `PreToolUse` command Hook runs; the broker proxy drops the `hook-execute` control reply after the worker admitted it. | Exactly one physical call; the retry resolves the same operation through the owner's status; the record settles `settled` with `outcome=success`; the allow reaches the model once; the proxy log records the drop. | +| `status-reply-lost` | The same solo command; the broker proxy drops the reply to a `hook-status` poll after the dispatch. | The dropped poll cannot prove the end, so the record first reports `recovery_blocked` / `outcome_unknown`; a later poll of the same owner settles it `success`, with exactly one physical call and the drop in the proxy log. | + +## Mutation witness + +Each witness must turn the cell red before the slice merges. + +- `execute-reply-lost`: break the exercised fallback — make the dropped + dispatch reply answer `outcome_unknown` instead of looking up the receipt + (for example, remove the `hook-status` fallback in `fire`). The record + then never settles and the `settled` assertion fails. The execute + operation-id deduplication is a separate witness: explicitly replay the + `hook-execute` operation. +- `status-reply-lost`: drop the honest unknown commit in the `fire` loop + (the branch that records `recovery_blocked` / `outcome_unknown` when the + poll cannot prove the end). The record stays `running`, so the state + assertion fails. + +## Non-goals + +The rest of FG13a — `hook-catalog`, `hook-cancel`, the store acks of the +`dispatch_started` and result commits, and a lost HTTP Hook response — stays +on #14449. Also out: 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-execute-reply-loss-gates-a.zh-CN.md b/docs/design/2026-10-18-hosted-hook-execute-reply-loss-gates-a.zh-CN.md new file mode 100644 index 0000000000..90606a309d --- /dev/null +++ b/docs/design/2026-10-18-hosted-hook-execute-reply-loss-gates-a.zh-CN.md @@ -0,0 +1,53 @@ +# Hosted Hook execute 丢回复门禁 FG13a + +[English](2026-10-18-hosted-hook-execute-reply-loss-gates-a.md) | [简体中文](2026-10-18-hosted-hook-execute-reply-loss-gates-a.zh-CN.md) + +## 问题与范围 + +Issue #14449 跟踪 Hosted Hook 两个域的 Stage F 故障门禁轮次。FG13c 覆盖 owner +重启矩阵，FG13d 覆盖取消与超时；本切片开始 **FG13a（丢回复）**，先补命令 Hook +最先遇到的一臂：worker 已接收操作之后，`hook-execute` 的回复丢失。 + +派发回复是唯一会让“丢失的应答”看起来像“丢失的调用”的地方。回执已经在 worker +手里，因此重试必须通过 owner 的状态解析同一个操作，而不是再派发一条命令。 + +## 决策 + +1. **复用 FG13d 的装置与 broker 代理。** 新类沿用只含 Hook 的定义、打包 Harness、 + durable worker 与转发代理；只丢弃 `hook-execute` 控制调用的回复，其余请求直通。 +2. **重试走状态轮询，而不是第二次 execute。** 派发回复丢失时，Harness 会按记录 + 自身的 `runtimeSessionId` 回落到 `hook-status`，这正是物理调用保持一次的原因。 +3. **两格、各自一套夹具。** 两格都让 solo 命令保持 1 秒后放行；派发回复丢失必须 + 通过 status 解析，status 回复丢失必须先保持未知，直到同一 owner 的后续轮询 + 把它结算。两者都不得增加第二次物理调用。 + +## 用例 + +| 用例 | 注入 | 期望 | +| -------------------- | ---------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `execute-reply-lost` | `PreToolUse` 命令 Hook 运行中；broker 代理在 worker 已接收后丢弃 `hook-execute` 的回复。 | 恰好一次物理调用；重试通过 owner 状态解析同一操作；记录结算 `settled` 且 `outcome=success`；放行结果只到达模型一次；代理日志记录这次丢弃。 | +| `status-reply-lost` | 同样的 solo 命令；broker 代理在派发后丢弃一次 `hook-status` 轮询的回复。 | 丢失的轮询无法证明结束，因此记录先落 `recovery_blocked` / `outcome_unknown`；随后对同一 owner 的轮询把它结算为 `success`，物理调用仍恰好一次，代理日志记录这次丢弃。 | + +## 变异见证 + +每个见证必须在合入前让自己的用例变红。 + +- `execute-reply-lost`：改破坏这条格真正走到的兜底——让丢失的派发回复直接得到 + `outcome_unknown`，而不是回落到回执查询（例如去掉 `fire` 里的 `hook-status` + 兜底）。记录因此永不结算，`settled` 断言失败。execute 的 operation-id 幂等 + 是另一个见证：显式重放 `hook-execute`。 +- `status-reply-lost`：去掉 `fire` 循环里如实记录 unknown 的提交分支（轮询无法 + 证明结束时写 `recovery_blocked` / `outcome_unknown` 的那一支）。记录会停在 + `running`，状态断言失败。 + +## 非目标 + +FG13a 其余部分——`hook-catalog`、`hook-cancel`、`dispatch_started` 与结果提交的 +store ack、以及 HTTP Hook 响应丢失——仍留在 #14449。同样不在范围：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 aa1491c211..78a9db1b51 100644 --- a/packages/managed-agent-server/pom.xml +++ b/packages/managed-agent-server/pom.xml @@ -465,8 +465,16 @@ locally on the merged base (~242 s budgeted): the healthy sum becomes 5985 s, so the fork moves to 6585 s - and the Verify step to 120 minutes. --> - 6585 + and the Verify step to 120 minutes. + The FG13a execute lost-reply cell + (issue #14449) adds 150 s: the healthy + sum becomes 6135 s, so the fork moves + to 6735 s. The FG13a status lost-reply + cell (issue #14449) adds 150 s: the + healthy sum becomes 6285 s, so the + fork moves to 6885 s and the Verify + step to 125 minutes. --> + 6885 diff --git a/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedHookReplyLossFaultGatesIT.java b/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedHookReplyLossFaultGatesIT.java new file mode 100644 index 0000000000..0f5c2e64b3 --- /dev/null +++ b/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedHookReplyLossFaultGatesIT.java @@ -0,0 +1,541 @@ +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; + +// FG13a (issues #14054/#14449): lost replies for the managed command Hook. +// The dispatch reply is dropped after the worker admitted the operation, so +// the retry must resolve the same operation through the owner's status, keep +// exactly one physical call, and apply its single result. +@EnabledOnOs(OS.LINUX) +class HostedHookReplyLossFaultGatesIT { + 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; + private HostedSessionRelayForwardingProxy proxy; + + @Test + @Timeout(300) + void fg13aExecuteReplyLossKeepsOnePhysicalCallOnMySql() throws Exception { + boot("60000", "1000", "hook-solo"); + proxy.arm(HostedSessionRelayForwardingProxy.Rule.drop( + (method, path, body) -> path.contains("/control") && body.contains("\"kind\":\"hook-execute\""), + "hook-execute-reply")); + JsonNode definition = api("POST", "/v1/agents", definition(), "definition", 202); + String session = createSession(definition.path("id").asText(), "initial"); + submit(session, "FG13_EXECUTE_LOSS"); + waitFor(() -> assertThat(events("hook.call")).hasSize(1)); + // The dispatch reply is gone, but the retry resolves the same + // operation through the owner's status: one physical call, one allow. + waitFor(() -> { + JsonNode execution = hookExecution(session); + assertThat(execution.at("/run/state").asText()).isEqualTo("settled"); + assertThat(execution.at("/run/execution").asText()).isEqualTo("settled"); + }); + assertThat(events("hook.call")).hasSize(1); + assertThat(events("hook.allow")).hasSize(1); + assertThat(hookResult(session).path("outcome").asText()).isEqualTo("success"); + waitFor(() -> assertThat(modelRequests.toString()).contains(PROOF)); + // The completion request must carry exactly one result for the call + // the fixture emitted; a duplicate delivery would otherwise pass. + long results = 0; + for (JsonNode message : modelRequests.getLast().path("messages")) { + if ("tool".equals(message.path("role").asText()) + && "call_FG13_EXECUTE_LOSS".equals(message.path("tool_call_id").asText())) { + results++; + } + } + assertThat(results).as("tool results for the emitted call").isEqualTo(1); + assertThat(Files.readString(temporary.resolve("proxy.log"))).contains("PROXY dropped hook-execute-reply"); + assertThat(modelFailure.get()).isNull(); + } + + 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)); + proxy = HostedSessionRelayForwardingProxy.start("hook-broker", temporary.resolve("proxy.log")); + proxy.forwardTo(URI.create("http://127.0.0.1:" + brokerPort)); + 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); + } + + @Test + @Timeout(300) + void fg13aStatusReplyLossStaysUnknownUntilTheNextPollOnMySql() throws Exception { + boot("60000", "1000", "hook-solo"); + proxy.arm(HostedSessionRelayForwardingProxy.Rule.drop( + (method, path, body) -> path.contains("/control") && body.contains("\"kind\":\"hook-status\""), + "hook-status-reply")); + JsonNode definition = api("POST", "/v1/agents", definition(), "definition", 202); + String session = createSession(definition.path("id").asText(), "initial"); + submit(session, "FG13_STATUS_LOSS"); + waitFor(() -> assertThat(events("hook.call")).hasSize(1)); + // The dropped poll answer cannot prove the end, so the gate records + // the honest unknown instead of guessing. + waitFor(() -> { + JsonNode execution = hookExecution(session); + assertThat(execution.at("/run/state").asText()).isEqualTo("recovery_blocked"); + assertThat(execution.at("/run/execution").asText()).isEqualTo("outcome_unknown"); + }); + assertThat(events("hook.call")).hasSize(1); + // A later poll of the same owner resolves the receipt the worker + // still holds, without a second physical call. + String executionId = hookExecution(session).path("hookExecutionId").asText(); + HarnessAttach attach = attachment(spring.getBean(HarnessConnector.class), session); + assertThat(attach).as("Session attachment").isNotNull(); + waitFor(() -> { + var status = hookStatus(session, executionId, attach); + assertThat(status.statusCode()).as("hook status: %s", status.body()).isEqualTo(200); + assertThat(json.readTree(status.body()).path("state").asText()).isEqualTo("settled"); + }); + waitFor(() -> assertThat(hookExecution(session).at("/run/state").asText()).isEqualTo("settled")); + assertThat(hookResult(session).path("outcome").asText()).isEqualTo("success"); + // The allow lands when the one-second hold ends, after the first + // unknown observation, so it is asserted only past the settlement. + waitFor(() -> assertThat(events("hook.allow")).hasSize(1)); + assertThat(events("hook.call")).hasSize(1); + assertThat(Files.readString(temporary.resolve("proxy.log"))).contains("PROXY dropped hook-status-reply"); + assertThat(modelFailure.get()).isNull(); + } + + // 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 HttpResponse hookStatus(String session, String executionId, HarnessAttach attach) throws Exception { + var request = HttpRequest.newBuilder(URI.create("http://127.0.0.1:" + harnessPort + + "/session/" + session + "/hooks/operations/" + executionId)) + .timeout(Duration.ofSeconds(10)).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()).GET().build(); + return http.send(request, HttpResponse.BodyHandlers.ofString()); + } + + 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", + proxy.baseUrl(), "--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_EXECUTE_LOSS", "FG13_STATUS_LOSS").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 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); + } + } + }, () -> { + if (proxy != null) { + proxy.close(); + } + }, 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", "proxy.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/scripts/tests/hosted-process-ci.test.js b/scripts/tests/hosted-process-ci.test.js index ebdef398ad..b631e9cbab 100644 --- a/scripts/tests/hosted-process-ci.test.js +++ b/scripts/tests/hosted-process-ci.test.js @@ -415,9 +415,9 @@ describe('Hosted real-process gates', () => { }); it.each([ - // Step ceilings today (120 + 10 + 20 + 50): Verify, the Broker fault + // Step ceilings today (125 + 10 + 20 + 50): Verify, the Broker fault // gates, admission and O4. The failover arms moved to their own job. - ['hosted-harness-mysql', 200, 213], + ['hosted-harness-mysql', 205, 218], // Fifteen failover arms at 10 each. ['hosted-failover-e2e', 150, 165], ])( @@ -609,13 +609,15 @@ describe('Hosted real-process gates', () => { // 6233; the FG12b store-failure class adds ~120 s (49.6 s measured // locally on JDK 21 + MySQL 8.0.46), so the fork moves to 6353 and the // Verify step to 116 minutes; the FG12f stream-gap class adds ~242 s - // (121.1 s measured on the merged base), so the fork moves to 6585 and - // the Verify step to 120 minutes): + // (121.1 s measured on the merged base), so the fork moves to 6585; + // the FG13a execute lost-reply cell adds 150 s, so the fork moves to + // 6735; the FG13a status lost-reply cell adds 150 s, so the fork moves + // to 6885 and the Verify step to 125 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(6585); + expect(forkSeconds).toBe(6885); } }, ); @@ -669,11 +671,13 @@ describe('Hosted real-process gates', () => { // class adds ~120s (49.6s measured on JDK 21 + MySQL 8.0.46), so it // becomes 5743s; the FG12f stream-gap class (HostedMcpStreamGapIT, three // cases, three cold boots) measures 121.1s locally on the merged base - // (~242s budgeted), so it becomes 5985s. The largest ceiling (600s) plus + // (~242s budgeted), so it becomes 5985s; the FG13a execute lost-reply + // cell adds 150s, so it becomes 6135s; the FG13a status lost-reply cell + // adds 150s, so it becomes 6285s. 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 = 5985; + const hostedForkHealthySeconds = 6285; const hostedForkSeconds = () => Number( read('packages/managed-agent-server/pom.xml') @@ -742,6 +746,7 @@ describe('Hosted real-process gates', () => { 'HostedHarnessMySqlIT', 'HostedHookCancelFaultGatesIT', 'HostedHookFaultGatesIT', + 'HostedHookReplyLossFaultGatesIT', 'HostedMcpReplyLossIT', 'HostedMcpRestartFaultGatesIT', 'HostedMcpStoreFaultGatesIT',