From 7a2ce9b99ad3bf614032393f4b40d430902fb229 Mon Sep 17 00:00:00 2001 From: Shaojin Wen Date: Sun, 18 Oct 2026 15:22:12 +0800 Subject: [PATCH] test(managed-agent): add the FG11a automation lost-reply fault gates (#14446) (#14534) * test(managed-agent): add the FG11a automation lost-reply fault gates (#14446) The first packaged lane that turns automation on: in-JVM Spring with the scanner running, a real Harness behind a relay that drops one committed answer, real MySQL and a model stub. Eight cases share one boot and one minute boundary: - create-reply, update-reply (a newer update lands meanwhile), noop-update-reply and retire-reply: the retry answers the committed revision and never commits again or over a newer one; - manual-run-reply and scheduled-fire-reply: the re-drive meets the same arun_ run, with one run chain and one model Turn; - reconcile-reply: a lost reconcile answer leaves the slot skipped under its own refusal, never unknown; - public-reply: a lost public 202 replays the receipt without a second Harness operation. Five mutation witnesses are red; one equivalent mutation is recorded in the design. The fixture runs one Turn per Session first: the automation verbs only load a Session on the Harness, so one created without input is refused 404 managed_session_not_found. The class measures 46.3-85.0 s locally; it is budgeted at ~170 s, so the shared fork's healthy sum moves to 4359 s and its timeout to 4969 s, inside the Verify step's 95 minutes. * test(managed-agent): keep FG11a's clocked cases off the scanner until armed (#14446) Review round 1: - The scanner ran from boot, so a minute boundary between a definition's create and its armed fault, or its held manual run, let a slot fire first: attempts 0 for scheduled-fire-reply, 409 automation_run_skipped for reconcile-reply. SchedulerHold now parks the scanner's only thread while the three clocked cases set up; manual runs fire on the request thread and are unaffected. - assertOneRun collected every fire and run command of the schedule, so a legitimate second slot on a slow runner read as a duplicate. The exactly-once checks are now scoped to the occurrence under test. - The design docs give the Verify step's 95 minutes, describe both fixes, and say a first red skips the later cases, which are re-run per case. --- .github/workflows/sdk-java.yml | 22 +- .../2026-10-18-automation-fault-gates-a.md | 175 ++++ ...26-10-18-automation-fault-gates-a.zh-CN.md | 139 ++++ packages/managed-agent-server/pom.xml | 29 +- .../HostedAutomationFaultGatesIT.java | 764 ++++++++++++++++++ scripts/tests/hosted-process-ci.test.js | 23 +- 6 files changed, 1134 insertions(+), 18 deletions(-) create mode 100644 docs/design/2026-10-18-automation-fault-gates-a.md create mode 100644 docs/design/2026-10-18-automation-fault-gates-a.zh-CN.md create mode 100644 packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedAutomationFaultGatesIT.java diff --git a/.github/workflows/sdk-java.yml b/.github/workflows/sdk-java.yml index d0aa021e8c..9cfe104dd4 100644 --- a/.github/workflows/sdk-java.yml +++ b/.github/workflows/sdk-java.yml @@ -676,17 +676,18 @@ jobs: name: 'Hosted process fault gates / MySQL 8.4 / Java 21' if: "${{ inputs.tier != 'fast' }}" runs-on: 'ubuntu-latest' - # The step ceilings today (95 + 10 + 20 + 50 = 175 — the FG8 letters + # The step ceilings today (100 + 10 + 20 + 50 = 180 — 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, 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. - timeout-minutes: 190 + # owner-restart cells to 4848 s, the FG13d cancel/timeout cells to + # 5048 s and the FG11a automation lost-reply letter to 5218 s, which + # moves Verify to 100 minutes so it 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): 193 + 2. The failover arms run + # in hosted-failover-e2e. + timeout-minutes: 195 services: mysql: image: 'mysql:8.4.6' @@ -775,8 +776,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 - # 5048 s, so the step wraps to fork + 600 s plus margin. - timeout-minutes: 95 + # 5218 s, so the step wraps to fork + 600 s plus margin. + timeout-minutes: 100 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'] }}" @@ -815,6 +816,7 @@ jobs: packages/managed-agent-server/target/fg8-cases/ packages/managed-agent-server/target/fg9-cases/ packages/managed-agent-server/target/fg10-cases/ + packages/managed-agent-server/target/fg11-cases/ packages/managed-agent-server/target/fg12-cases/ packages/managed-agent-server/target/l4-shell-lifecycle/ retention-days: 7 diff --git a/docs/design/2026-10-18-automation-fault-gates-a.md b/docs/design/2026-10-18-automation-fault-gates-a.md new file mode 100644 index 0000000000..25a9535171 --- /dev/null +++ b/docs/design/2026-10-18-automation-fault-gates-a.md @@ -0,0 +1,175 @@ +# Hosted automation fault gates FG11a: lost replies + +[English](2026-10-18-automation-fault-gates-a.md) | [简体中文](2026-10-18-automation-fault-gates-a.zh-CN.md) + +## Problem and scope + +Issue #14446 is Stage F round FG11 of #12380: fault gates for the +automation domains `schedule` and `automation_run`. The round has six +letters (FG11a–f). This slice implements **FG11a, lost replies**, and +nothing else. It also stands up the packaged automation stack that the +later letters build on. + +The property under gate: when an automation operation committed and its +answer was lost, the re-drive meets the **same identity** and answers what +committed. It never commits a second time, never runs a second model Turn, +and never records an occurrence that is not true. + +- A scheduled fire's re-drive meets the same derived `arun_` run. +- A manual run retried under its Idempotency-Key meets the same run. +- A retried definition mutation answers its own committed revision and + never commits over a newer one. +- A lost reconcile answer leaves the slot skipped under its own refusal, + never `unknown`. + +Eight cases ship, all in one boot: + +| Case | Injection | Expect | +| ---------------------- | --------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `create-reply` | The Harness `202` of a create (`define_schedule`) is dropped after it committed | The public answer is `503 automation_operation_unknown`. The retry under the same key re-drives the same operation id and answers revision 1. One `defineSchedule` commit | +| `update-reply` | An update's `202` is dropped; a newer update under another key then lands (revision 3) | The retry of the lost update answers its own revision 2. The definition stays at revision 3, with three `defineSchedule` commits in all | +| `noop-update-reply` | The `202` of an update that changes nothing is dropped | The retry answers revision 1. Only the create and the no-op's replay marker are committed | +| `retire-reply` | A retire's `202` is dropped | The retry answers the committed retire as its replay (`X-Qwen-Idempotent-Replay: true`) without asking the Harness again. One `retireSchedule` commit. The every-minute definition records no occurrence after the retire | +| `manual-run-reply` | A manual run's `fire_run` `202` is dropped after both run revisions committed and the Turn woke | `503 automation_operation_unknown`. The retry under the same key meets the same run. One run chain (`:1`, `:2`, `:3`), one accepted input, one `turn.settled`, one model Turn | +| `public-reply` | Spring settled a create's receipt; its own `202` is dropped on the way to the caller | The retry replays the receipt (`X-Qwen-Idempotent-Replay: true`). The Harness saw the operation once | +| `scheduled-fire-reply` | The scanner's fire of the next minute's slot commits, and its `202` is dropped | The scanner re-drives the occurrence once (`attempts` 1) and settles it `fired`. Every fire names the same occurrence. One run chain, one accepted input, one `turn.settled`, one model Turn | +| `reconcile-reply` | A held manual run keeps the skip policy's count closed; the next slot's `reconcile_run` answer is dropped | The slot is `skipped` under `overlap` or `count_limit`, never `unknown`, and the held run stays `running`. Released, it completes | + +Every case ends with the tenant holding no `unknown` occurrence. + +Deferred to later slices of #14446: FG11b–f, the Harness-side lost +answers of the Session Store (FG11b), and every process kill (FG11c). + +## Current state + +- No packaged lane turned automation on before this slice. + `qwen.managed-agent.automation.enabled` defaults to `false`. The only + automation tests were on H2, against `AutomationHarnessFake`. +- The Harness route is `POST /session/{id}/automations/operations` + (`hosted-harness-session.ts`). Spring reaches it through + `QwenHostedHarnessConnector.runAutomationOperation`, whose client turns + a lost answer into `MutationOutcomeUnknownException` with no retry of + its own. +- The scanner (`AutomationScanner`) defers such an occurrence with backoff + (2 s, 4 s, …) and re-drives it for up to 64 attempts. The public routes + answer `503 automation_operation_unknown`, and a retry under the same + key re-drives the claimed receipt. +- Definition operations carry an operation id derived from the tenant and + the Idempotency-Key, so the Harness's `mutationReplay` answers a re-drive + with the revision it committed. A run's id is derived from the schedule + and the occurrence key, so a re-driven fire meets the committed run. + +## Gate ledger + +Ground truth comes from three sources: + +- The relay's record of every automation operation Spring sent the + Harness, with its kind, schedule, occurrence and whether its reply was + dropped. +- The Session journal (`ma_session_journal_tx` commands, and event ids + decoded from `record_bytes`). +- The Java mirror (`ma_schedule_occurrence`, `ma_session_record`). + +The model stub records every prompt, so a second model Turn would show as +a second prompt naming the occurrence. + +Each case first establishes its injection: the relay must show the dropped +operation (`dropped=true`), and every armed rule must have fired by the +end of the run. + +## Fixture design + +`HostedAutomationFaultGatesIT` follows `HostedPublicWorkspaceIT`'s boot: + +- in-JVM Spring on real MySQL (absence fails rather than skips), with the + Harness connector and the scanner on (`automation.enabled=true`, + `scan-delay=1s`); +- one Workspace mount, registered on Linux; +- a real Harness, and a model stub that answers at once unless the prompt + carries `FG11A_HOLD`. + +Spring's `harness.base-url` points at a `HostedLifecycleRelay`. For a +matching automation operation, the relay lets the Harness answer and then +closes the connection without the reply, which is a connection-level loss +after commit. A second relay fronts Spring for `public-reply` and drops one +`202` by its Idempotency-Key. + +Each case uses its own public Workspace Session. The Session runs one +Turn first (see Findings). + +The three clock-bound cases (`scheduled-fire-reply`, `reconcile-reply`, +`retire-reply`) use an every-minute cron. They arm first and are settled +last, so they share one minute boundary while the other cases run. + +- While they set up, `SchedulerHold` parks the only thread of + `managedAutomationScheduler`. No slot can fire before its fault is + armed or before the held manual run is live; manual runs fire on the + request thread and are unaffected. A boundary that passes during the + hold is still within the late tolerance when the scanner resumes. +- Their exactly-once checks are scoped to the occurrence under test: + its fires, its run's commands, its input, its `turn.settled` and its + model prompt. A later slot of the same every-minute definition, due + while the other cases run on a slow runner, is a legitimate run of its + own. + +The class runs one case alone with `-Dqwen.fg11a.case=`. + +## Findings + +- **An automation on a Session that never ran a Turn is refused.** The + Turn and rename paths call `createOrLoad` with + `loadExisting = session.harnessBootId() != null`. The automation verbs + go through `attachment(…, true)`, which always loads. A Session created + with no input has never been created on the Harness, so the load answers + `404 managed_session_not_found` and creating an automation fails with + that code. The fixture runs one Turn per Session first. Fixing the + connector also means recording the Harness boot id the way the Turn path + does, so it is left out of this test-only slice and reported separately. + +## Mutation witnesses + +Each witness was run red against the full class and restored byte for +byte. + +| Witness | Mutation | First red | +| ------- | ----------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------ | +| m1 | `ManagedAutomationService.operationIdFor` mints a random id instead of deriving it from the key | `create-reply`: the retry re-drove under a second operation id | +| m2 | `AutomationScanner.MAX_FIRE_ATTEMPTS` is 1, so a lost answer is given up at once | `manual-run-reply` in the full class; `scheduled-fire-reply` alone (`-Dqwen.fg11a.case=scheduled-fire-reply`): the slot settled `unknown`, not `fired` | +| m3 | A reconcile that failed counts its occurrence as resolved | `reconcile-reply`: the slot did not settle `skipped` | +| m4 | `ManagedAutomationService.replay` re-drives a receipt that already holds its answer | `public-reply`: the Harness saw a second `define_schedule` | +| m5 | A manual run's occurrence key gains a random suffix, so a retry no longer meets its occurrence | `manual-run-reply`: the retry was refused `automation_run_skipped` (overlap) behind its own first run | + +One narrower mutation is equivalent and is not a witness: rethrowing a +failed reconcile only postpones the decision. Run against +`reconcile-reply` alone it stayed green, because the next scan, a second +later, asks again and records the same `skipped`. + +## Budget + +The class measured 46.3 s, 85.0 s, 57.1 s and 67.0 s locally on JDK 21 + +MySQL 8.4.6. Most of the spread is the wait for the next minute boundary. +It is budgeted at about 170 s (2× the heaviest run). The shared fork's +healthy sum and its timeout move by that amount, composed onto `main`'s +values at push time. On top of `main` after #14533 the fork reaches +5218 s, so the Verify step moves from 95 to 100 minutes (fork + 600 s = +5818 s) and the job ceiling from 190 to 195. + +## Acceptance + +- `mvn -DskipTests test-compile checkstyle:check` clean. +- The class green locally on JDK 21 + MySQL 8.4.6, and every mutation + witness red before merge. +- `scripts/tests/hosted-process-ci.test.js` green, with the class in the + pinned list and the fork budget composed. +- The eight cases share one `@Test`, so a first red names one case and + the cases after it never run. Read a red per case: re-run each with + `-Dqwen.fg11a.case=` rather than taking the others as passed. +- This design ships in both languages with matching structure. + +## Open questions + +- FG11c kills Spring, which owns the scanner and its lease, so it needs a + child-JVM fixture like `HostedProcessCrashIT`'s rather than this class's + in-JVM Spring. +- Whether FG11b's Harness-side store faults can reuse this relay unchanged + or need the store-fronting proxy of the FG9/FG10 relay gates. diff --git a/docs/design/2026-10-18-automation-fault-gates-a.zh-CN.md b/docs/design/2026-10-18-automation-fault-gates-a.zh-CN.md new file mode 100644 index 0000000000..d18dcf7659 --- /dev/null +++ b/docs/design/2026-10-18-automation-fault-gates-a.zh-CN.md @@ -0,0 +1,139 @@ +# Hosted automation 故障门禁 FG11a：应答丢失 + +[English](2026-10-18-automation-fault-gates-a.md) | [简体中文](2026-10-18-automation-fault-gates-a.zh-CN.md) + +## 问题与范围 + +Issue #14446 是 #12380 F 阶段的 FG11 轮：为 automation 两个域 `schedule` 和 +`automation_run` 补故障门禁。这一轮有六个字母（FG11a–f）。本切片只实现 +**FG11a，应答丢失**，不做其他字母。它同时搭起后续字母要复用的打包 automation 栈。 + +门禁要守的性质：一个 automation 操作已经提交、但应答丢了时，重驱要碰到**同一个身份**， +并回答已经提交的内容。它绝不会再提交一次，绝不会多跑一个模型 Turn，也绝不会记下一个 +不真实的 occurrence。 + +- 定时 fire 的重驱碰到同一个派生出的 `arun_` run。 +- 用同一个 Idempotency-Key 重试的手动 run 碰到同一个 run。 +- 重试的定义变更回答它自己提交的 revision，绝不会覆盖更新的 revision。 +- 丢失的 reconcile 应答让该时隙以它自己的拒绝原因记为 skipped，绝不会是 `unknown`。 + +共八个用例，在一次启动里跑完： + +| 用例 | 注入 | 预期 | +| ---------------------- | -------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `create-reply` | 创建（`define_schedule`）已提交后，丢掉 Harness 的 `202` | 公开应答为 `503 automation_operation_unknown`。用同一个 key 重试，重驱同一个 operation id，回答 revision 1。只有一次 `defineSchedule` 提交 | +| `update-reply` | 丢掉一次更新的 `202`；随后另一个 key 的更新落地（revision 3） | 丢失的那次更新的重试回答它自己的 revision 2。定义停在 revision 3，总共三次 `defineSchedule` 提交 | +| `noop-update-reply` | 丢掉一次什么都没改的更新的 `202` | 重试回答 revision 1。只提交了创建和这次 no-op 的 replay 标记 | +| `retire-reply` | 丢掉一次 retire 的 `202` | 重试把已提交的 retire 当作它的 replay 作答（`X-Qwen-Idempotent-Replay: true`），不再问 Harness。只有一次 `retireSchedule` 提交。这个每分钟触发的定义在 retire 之后没有任何 occurrence | +| `manual-run-reply` | 手动 run 的两个 run revision 已提交、Turn 已唤醒后，丢掉 `fire_run` 的 `202` | `503 automation_operation_unknown`。用同一个 key 重试碰到同一个 run。一条 run 链（`:1`、`:2`、`:3`）、一个被接受的输入、一个 `turn.settled`、一个模型 Turn | +| `public-reply` | Spring 已结清一次创建的回执，它自己发给调用方的 `202` 在路上丢了 | 重试 replay 该回执（`X-Qwen-Idempotent-Replay: true`）。Harness 只见过这个操作一次 | +| `scheduled-fire-reply` | 扫描器对下一分钟时隙的 fire 已提交，丢掉它的 `202` | 扫描器只重驱这个 occurrence 一次（`attempts` 为 1），并把它落定为 `fired`。每次 fire 都指向同一个 occurrence。一条 run 链、一个被接受的输入、一个 `turn.settled`、一个模型 Turn | +| `reconcile-reply` | 一个被挂住的手动 run 让 skip 策略的计数保持占满；丢掉下一个时隙 `reconcile_run` 的应答 | 该时隙以 `overlap` 或 `count_limit` 记为 `skipped`，绝不会是 `unknown`，被挂住的 run 保持 `running`。放开后它正常完成 | + +每个用例结束时，该租户都没有 `unknown` 的 occurrence。 + +留给 #14446 后续切片：FG11b–f、Harness 侧 Session Store 的应答丢失（FG11b），以及所有 +进程 kill（FG11c）。 + +## 当前状态 + +- 本切片之前，没有任何打包 lane 开启 automation。 + `qwen.managed-agent.automation.enabled` 默认是 `false`。仅有的 automation 测试都在 H2 上， + 对着 `AutomationHarnessFake` 跑。 +- Harness 路由是 `POST /session/{id}/automations/operations`（`hosted-harness-session.ts`）。 + Spring 通过 `QwenHostedHarnessConnector.runAutomationOperation` 调到它，其客户端把丢失的应答 + 变成 `MutationOutcomeUnknownException`，自身不重试。 +- 扫描器（`AutomationScanner`）对这样的 occurrence 按退避（2 s、4 s、……）延后，最多重驱 64 + 次。公开路由回答 `503 automation_operation_unknown`，用同一个 key 重试会重驱已占用的回执。 +- 定义操作带着由租户和 Idempotency-Key 派生的 operation id，所以 Harness 的 `mutationReplay` + 会用它提交过的 revision 回答重驱。run 的 id 由 schedule 和 occurrence key 派生，所以重驱的 + fire 会碰到已提交的 run。 + +## 门禁账本 + +事实依据来自三处： + +- relay 对 Spring 发给 Harness 的每个 automation 操作的记录，包括它的 kind、schedule、 + occurrence，以及回包是否被丢弃。 +- Session journal（`ma_session_journal_tx` 的命令，以及从 `record_bytes` 解出的事件 id）。 +- Java 镜像（`ma_schedule_occurrence`、`ma_session_record`）。 + +模型桩记录每一个 prompt，所以多出来的模型 Turn 会表现为第二个指向该 occurrence 的 prompt。 + +每个用例都先确认注入确实发生：relay 必须显示被丢掉的那次操作（`dropped=true`），而且运行 +结束时每条预设的规则都必须已经触发。 + +## 夹具设计 + +`HostedAutomationFaultGatesIT` 沿用 `HostedPublicWorkspaceIT` 的启动方式： + +- Spring 在测试 JVM 内启动，连真实 MySQL（缺失时失败而不是跳过），并开启 Harness 连接器 + 和扫描器（`automation.enabled=true`、`scan-delay=1s`）； +- 一个 Workspace 挂载，在 Linux 上注册； +- 一个真实 Harness，以及一个模型桩：除非 prompt 带有 `FG11A_HOLD`，否则立即作答。 + +Spring 的 `harness.base-url` 指向一个 `HostedLifecycleRelay`。对匹配的 automation 操作， +relay 先让 Harness 作答，再不带回包地关闭连接，也就是提交之后的连接级丢失。第二个 relay +挡在 Spring 前面，供 `public-reply` 使用，按 Idempotency-Key 丢掉一次 `202`。 + +每个用例使用自己的 public Workspace Session。Session 先跑一个 Turn（见"发现"）。 + +三个依赖时钟的用例（`scheduled-fire-reply`、`reconcile-reply`、`retire-reply`）使用每分钟 +触发的 cron。它们最先布置、最后落定，从而在其他用例运行期间共用同一个整分钟边界。 + +- 它们准备期间，`SchedulerHold` 挂住 `managedAutomationScheduler` 唯一的线程。任何时隙都 + 不可能在它的故障布置好之前、或被挂住的手动 run 上线之前触发；手动 run 在请求线程上 fire， + 不受影响。挂住期间经过的整分钟边界，在扫描器恢复时仍处在迟到容忍范围内。 +- 它们的"恰好一次"检查都限定在被测 occurrence 上：它的 fire、它的 run 的命令、它的输入、 + 它的 `turn.settled` 和它的模型 prompt。慢 runner 上其他用例运行期间到期的、同一个每分钟 + 定义的后续时隙，是它自己的合法 run。 + +用 `-Dqwen.fg11a.case=` 可以只跑其中一个用例。 + +## 发现 + +- **在从未跑过 Turn 的 Session 上建 automation 会被拒绝。** Turn 和 rename 路径调用 + `createOrLoad` 时用 `loadExisting = session.harnessBootId() != null`。automation 各动作 + 走的是 `attachment(…, true)`，永远只 load。没有输入就创建的 Session 从未在 Harness 上 + 创建过，于是 load 回答 `404 managed_session_not_found`，建 automation 也就以这个错误码 + 失败。夹具让每个 Session 先跑一个 Turn。修连接器还需要像 Turn 路径那样记录 Harness + boot id，所以不放进这个只含测试的切片，单独报告。 + +## 变异见证 + +每个见证都对整个类跑过、确认变红，并逐字节还原。 + +| 见证 | 变异 | 最先变红处 | +| ---- | ---------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------- | +| m1 | `ManagedAutomationService.operationIdFor` 改成随机 id，不再由 key 派生 | `create-reply`：重试用第二个 operation id 重驱 | +| m2 | `AutomationScanner.MAX_FIRE_ATTEMPTS` 改为 1，丢失的应答立刻被放弃 | 整个类里是 `manual-run-reply`；单跑 `scheduled-fire-reply`（`-Dqwen.fg11a.case=scheduled-fire-reply`）时：时隙落成 `unknown` 而不是 `fired` | +| m3 | 失败的 reconcile 把它的 occurrence 算作已解决 | `reconcile-reply`：时隙没有落成 `skipped` | +| m4 | `ManagedAutomationService.replay` 对已有应答的回执也重驱 | `public-reply`：Harness 见到了第二次 `define_schedule` | +| m5 | 手动 run 的 occurrence key 加上随机后缀，重试不再碰到它的 occurrence | `manual-run-reply`：重试被它自己的第一个 run 挡住，以 `automation_run_skipped`（overlap）拒绝 | + +另有一个更窄的变异是等价的，不作为见证：把失败的 reconcile 改为抛出，只会推迟这次决策。 +单跑 `reconcile-reply` 时它保持绿色，因为一秒后的下一次扫描会再问一次，并记下同样的 +`skipped`。 + +## 预算 + +本地在 JDK 21 + MySQL 8.4.6 上，这个类测得 46.3 s、85.0 s、57.1 s 和 67.0 s。差异主要来自等待 +下一个整分钟边界。预算约 170 s（最重一次的 2 倍）。共享 fork 的健康总和与超时都按这个量 +调整，在推送时叠加到 `main` 当时的数值上。叠加在 #14533 之后的 `main` 上，fork 达到 +5218 s，所以 Verify step 从 95 分钟调到 100 分钟（fork + 600 s = 5818 s），job 上限从 190 调到 195。 + +## 验收 + +- `mvn -DskipTests test-compile checkstyle:check` 无报错。 +- 这个类在本地 JDK 21 + MySQL 8.4.6 上通过，且合入前每个变异见证都已确认变红。 +- `scripts/tests/hosted-process-ci.test.js` 通过：该类已加入钉住的列表，fork 预算已叠加。 +- 八个用例共用一个 `@Test`，所以第一个红灯只指出一个用例，它之后的用例根本不会运行。要按用例 + 解读红灯：用 `-Dqwen.fg11a.case=` 逐个重跑，而不是把其他用例当作已通过。 +- 本设计以两种语言发布，结构一致。 + +## 未决问题 + +- FG11c 要 kill Spring，扫描器及其 lease 都在 Spring 里，所以它需要像 `HostedProcessCrashIT` + 那样的子 JVM 夹具，而不是本类这种测试 JVM 内的 Spring。 +- FG11b 在 Harness 侧注入 store 故障时，能否原样复用这个 relay，还是需要 FG9/FG10 relay + 门禁那种挡在 store 前面的代理。 diff --git a/packages/managed-agent-server/pom.xml b/packages/managed-agent-server/pom.xml index 6da01a54cd..038e300692 100644 --- a/packages/managed-agent-server/pom.xml +++ b/packages/managed-agent-server/pom.xml @@ -420,8 +420,15 @@ cancel/timeout cells (two cases, issue #14449) add 200 s: the healthy sum becomes 4438 s, so the fork moves to - 5048 s. --> - 5048 + 5048 s. The FG11a automation lost-reply + letter (HostedAutomationFaultGatesIT, + eight cases on one minute boundary) + measures 46.3-85.0 s locally on JDK 21 + + MySQL 8.4.6 (~170 s budgeted at 2x): + the healthy sum becomes 4608 s, so the + fork moves to 5218 s and the Verify + step to 100 minutes. --> + 5218 @@ -583,6 +590,24 @@ integration-testverify + + hosted-automation-gates + + org.apache.maven.plugins + maven-failsafe-plugin + ${maven-failsafe-plugin.version} + + **/HostedAutomation*IT.java + true + + 1200 + + integration-testverify + + o4-mysql-gates diff --git a/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedAutomationFaultGatesIT.java b/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedAutomationFaultGatesIT.java new file mode 100644 index 0000000000..d5f06befe2 --- /dev/null +++ b/packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedAutomationFaultGatesIT.java @@ -0,0 +1,764 @@ +package com.alibaba.qwen.code.managedagent; + +import com.alibaba.qwen.code.managedagent.api.AuthenticatedTenantActor; +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 jakarta.servlet.Filter; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletRequestWrapper; +import org.awaitility.core.ConditionFactory; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +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.FilterRegistrationBean; +import org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext; +import org.springframework.core.Ordered; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.attribute.FileTime; +import java.security.Principal; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Predicate; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +/** + * FG11 (#14446): the automation domains, {@code schedule} and + * {@code automation_run}, on the packaged stack: in-JVM Spring with the + * scanner on, a real Harness behind a relay, a real MySQL schema and a + * model stub. + */ +class HostedAutomationFaultGatesIT { + private static final String TOKEN = "fg11-local-fixture"; + private static final String DIGEST = "sha256:" + "a".repeat(64); + private static final String OPERATIONS = "/automations/operations"; + + private final ObjectMapper json = new ObjectMapper(); + private final HttpClient http = HttpClient.newHttpClient(); + private final String tenant = "fg11-a-" + UUID.randomUUID().toString().substring(0, 8); + private final List modelPrompts = new CopyOnWriteArrayList<>(); + private final AtomicReference modelFailure = new AtomicReference<>(); + private final CountDownLatch heldModel = new CountDownLatch(1); + // Every automation operation Spring sent the Harness, in order. + private final List operations = new CopyOnWriteArrayList<>(); + private final Map> armed = new ConcurrentHashMap<>(); + @TempDir(cleanup = CleanupMode.ON_SUCCESS) + private Path temporary; + private ServletWebServerApplicationContext spring; + private JdbcTemplate jdbc; + private HttpServer model; + private ExecutorService modelExecutor; + private HostedLifecycleRelay harnessRelay; + // Fronts Spring for the public-reply case: the Idempotency-Key whose 202 it drops once. + private HostedLifecycleRelay publicRelay; + private final AtomicReference publicDrop = new AtomicReference<>(); + // Parks the scanner while the clock-bound cases set up. + private CountDownLatch scannerHold; + private Process harness; + private Path decoy; + private String node; + private int port; + + record Operation(String kind, String scheduleId, String occurrenceKey, String body, int status, + boolean dropped) { } + + @Test + @Timeout(600) + void fg11aLostRepliesNeverFireTwiceOnMySql() throws Exception { + assertThat(System.getProperty("mysql.url")).as("FG11 requires real MySQL").startsWith("jdbc:mysql:"); + boot(); + // The clock-bound cases arm first and settle on the next minute boundary, while the + // others run against their own Sessions in the meantime. The scanner's only thread is + // parked until all three are armed, so no slot fires before its fault is in place or its + // held run is live; a boundary that passes meanwhile is still timely when it lets go. + scannerHold = SchedulerHold.hold(spring.getBean("managedAutomationScheduler", + ThreadPoolTaskScheduler.class), Duration.ofSeconds(30)); + Clocked scheduled = selected("scheduled-fire-reply") ? scheduledFireReply() : null; + Clocked reconcile = selected("reconcile-reply") ? reconcileReply() : null; + Clocked retire = selected("retire-reply") ? retireReply() : null; + scannerHold.countDown(); + if (selected("create-reply")) { + createReply(); + } + if (selected("update-reply")) { + updateReply(); + } + if (selected("noop-update-reply")) { + noopUpdateReply(); + } + if (selected("manual-run-reply")) { + manualRunReply(); + } + if (selected("public-reply")) { + publicReply(); + } + for (Clocked clocked : new Clocked[] {scheduled, reconcile, retire}) { + if (clocked != null) { + clocked.settle(); + } + } + assertThat(jdbc.queryForList("SELECT CONCAT(occurrence_key, ' ', reason) FROM ma_schedule_occurrence" + + " WHERE tenant_id = ? AND outcome = 'unknown'", String.class, tenant)).as("unknown occurrences") + .isEmpty(); + assertThat(armed).as("every armed reply was dropped").isEmpty(); + assertNoModelFailure(); + } + + interface Clocked { + void settle() throws Exception; + } + + // create-reply: the Harness committed the definition and its 202 is lost. The public + // answer is 503 automation_operation_unknown; the retry under the same key re-drives the + // same operation id and answers the committed revision 1 without committing again. + private void createReply() throws Exception { + String session = createSession("create"); + arm("create-reply", body -> kind(body, "define_schedule") && prompt(body).contains("FG11A_CREATE")); + Map definition = definition(session, "FG11A_CREATE", "0 0 1 1 *"); + assertRefused(send("POST", "/v1/agent-automations", definition, "create-reply")); + JsonNode created = request("POST", "/v1/agent-automations", definition, "create-reply", 202); + String id = created.path("id").asText(); + assertThat(created.path("definition_revision").asLong()).isEqualTo(1); + assertOperations(id, "define_schedule", 2, 1); + assertThat(sameOperationId(id, "define_schedule")).isTrue(); + assertThat(commands(session, "defineSchedule")).hasSize(1); + assertThat(recordRevision(session, "schedule", id)).isEqualTo(1); + } + + // update-reply: a lost update answer, then a newer update under another key lands. The + // retry answers the lost update's own committed revision and never commits over the newer. + private void updateReply() throws Exception { + String session = createSession("update"); + String id = request("POST", "/v1/agent-automations", definition(session, "FG11A_UPDATE", "0 0 1 1 *"), + "update-base", 202).path("id").asText(); + arm("update-reply", body -> kind(body, "define_schedule") && prompt(body).contains("FG11A_UPDATE_ONE")); + Map one = Map.of("prompt", "FG11A_UPDATE_ONE reply with FG11_DONE."); + assertRefused(send("POST", "/v1/agent-automations/" + id, one, "update-one")); + assertThat(request("POST", "/v1/agent-automations/" + id, Map.of("prompt", + "FG11A_UPDATE_TWO reply with FG11_DONE."), "update-two", 202).path("definition_revision").asLong()) + .isEqualTo(3); + assertThat(request("POST", "/v1/agent-automations/" + id, one, "update-one", 202) + .path("definition_revision").asLong()).isEqualTo(2); + assertThat(request("GET", "/v1/agent-automations/" + id, null, null, 200).path("definition_revision") + .asLong()).isEqualTo(3); + assertThat(commands(session, "defineSchedule")).hasSize(3); + assertThat(recordRevision(session, "schedule", id)).isEqualTo(3); + } + + // noop-update-reply: an update that changes nothing commits only its replay marker; with + // that answer lost, the retry answers revision 1 and the definition never moves. + private void noopUpdateReply() throws Exception { + String session = createSession("noop"); + String id = request("POST", "/v1/agent-automations", definition(session, "FG11A_NOOP", "0 0 1 1 *"), + "noop-base", 202).path("id").asText(); + arm("noop-update-reply", body -> kind(body, "define_schedule") && id.equals(body.path("scheduleId") + .asText())); + Map same = Map.of("prompt", "FG11A_NOOP reply with FG11_DONE."); + assertRefused(send("POST", "/v1/agent-automations/" + id, same, "noop-update")); + assertThat(request("POST", "/v1/agent-automations/" + id, same, "noop-update", 202) + .path("definition_revision").asLong()).isEqualTo(1); + assertOperations(id, "define_schedule", 3, 1); + assertThat(commands(session, "defineSchedule")).hasSize(2); + assertThat(recordRevision(session, "schedule", id)).isEqualTo(1); + } + + // manual-run-reply: the fire committed both run revisions and woke the Turn, and its 202 + // is lost. The public answer is 503; the retry under the same key meets the same run, and + // the occurrence has one run chain and one model Turn. + private void manualRunReply() throws Exception { + String session = createSession("manual"); + String id = request("POST", "/v1/agent-automations", definition(session, "FG11A_MANUAL", "0 0 1 1 *"), + "manual-base", 202).path("id").asText(); + arm("manual-run-reply", body -> kind(body, "fire_run") && id.equals(body.path("scheduleId").asText())); + assertRefused(send("POST", "/v1/agent-automations/" + id + "/runs", null, "manual-run")); + JsonNode run = request("POST", "/v1/agent-automations/" + id + "/runs", null, "manual-run", 202); + assertThat(run.path("occurrence_key").asText()).isEqualTo("manual:manual-run"); + assertOneRun(session, id, run.path("id").asText(), "manual:manual-run", 2); + } + + // public-reply: Spring settled the create's receipt and its own 202 is lost on the way to + // the caller. The retry replays the receipt; the Harness sees the operation once. + private void publicReply() throws Exception { + String session = createSession("public"); + Map definition = definition(session, "FG11A_PUBLIC", "0 0 1 1 *"); + publicDrop.set("public-reply"); + try { + HttpResponse answered = http.send(builder(publicRelay.uri() + "/v1/agent-automations", + "POST", definition, "public-reply").build(), HttpResponse.BodyHandlers.ofString()); + throw new AssertionError("The public 202 reached the caller: " + answered.statusCode() + " " + + answered.body()); + } catch (IOException dropped) { + // The relay closed the connection after Spring answered. + } + assertThat(publicDrop.get()).as("the public 202 was dropped").isNull(); + HttpResponse replay = send("POST", "/v1/agent-automations", definition, "public-reply"); + assertThat(replay.statusCode()).as(replay.body()).isEqualTo(202); + assertThat(replay.headers().firstValue("X-Qwen-Idempotent-Replay")).contains("true"); + String id = json.readTree(replay.body()).path("id").asText(); + assertOperations(id, "define_schedule", 1, 0); + assertThat(commands(session, "defineSchedule")).hasSize(1); + } + + // scheduled-fire-reply: the scanner's fire of the next minute's slot commits and its 202 + // is lost. The scanner re-drives the occurrence once and the Harness answers the same run. + private Clocked scheduledFireReply() throws Exception { + String session = createSession("scheduled"); + String id = request("POST", "/v1/agent-automations", definition(session, "FG11A_SCHEDULED", "* * * * *"), + "scheduled-base", 202).path("id").asText(); + arm("scheduled-fire-reply", body -> kind(body, "fire_run") && id.equals(body.path("scheduleId").asText())); + return () -> { + Map occurrence = awaitOccurrence(id, "fired"); + String key = (String) occurrence.get("occurrence_key"); + assertThat(key).startsWith("schedule:"); + assertThat(((Number) occurrence.get("attempts")).intValue()).as("re-drives").isEqualTo(1); + assertOneRun(session, id, (String) occurrence.get("run_id"), key, 2); + request("DELETE", "/v1/agent-automations/" + id, null, "scheduled-retire", 202); + }; + } + + // reconcile-reply: a held manual run keeps the skip policy's count closed when the next + // slot comes due, so the scanner asks reconcile_run, and that answer is lost. The slot is + // skipped under its own refusal, never unknown, and the held run is left running. + private Clocked reconcileReply() throws Exception { + String session = createSession("reconcile"); + String id = request("POST", "/v1/agent-automations", definition(session, "FG11A_HOLD", "* * * * *"), + "reconcile-base", 202).path("id").asText(); + String held = request("POST", "/v1/agent-automations/" + id + "/runs", null, "reconcile-hold", 202) + .path("id").asText(); + waiting().atMost(Duration.ofSeconds(60)).until(() -> modelPrompts.stream() + .anyMatch(prompt -> prompt.contains("manual:reconcile-hold"))); + arm("reconcile-reply", body -> kind(body, "reconcile_run") && id.equals(body.path("scheduleId").asText())); + return () -> { + Map occurrence = awaitOccurrence(id, "skipped"); + assertThat((String) occurrence.get("occurrence_key")).startsWith("schedule:"); + assertThat((String) occurrence.get("reason")).isIn("overlap", "count_limit"); + assertThat(operations).filteredOn(operation -> id.equals(operation.scheduleId()) + && "reconcile_run".equals(operation.kind())).first().extracting(Operation::dropped) + .isEqualTo(true); + assertThat(runState(held)).isEqualTo("running"); + request("DELETE", "/v1/agent-automations/" + id, null, "reconcile-retire", 202); + heldModel.countDown(); + waiting().atMost(Duration.ofSeconds(60)).until(() -> "completed".equals(runState(held))); + }; + } + + // retire-reply: the retire committed and its 202 is lost. The journal commit projected the + // retired mirror in the same transaction, so the retry answers that committed retire as its + // replay without asking the Harness again, and the every-minute definition never fires + // after it. + private Clocked retireReply() throws Exception { + String session = createSession("retire"); + String id = request("POST", "/v1/agent-automations", definition(session, "FG11A_RETIRE", "* * * * *"), + "retire-base", 202).path("id").asText(); + arm("retire-reply", body -> kind(body, "retire_schedule") && id.equals(body.path("scheduleId").asText())); + assertRefused(send("DELETE", "/v1/agent-automations/" + id, null, "retire-reply")); + HttpResponse retry = send("DELETE", "/v1/agent-automations/" + id, null, "retire-reply"); + assertThat(retry.statusCode()).as(retry.body()).isEqualTo(202); + assertThat(retry.headers().firstValue("X-Qwen-Idempotent-Replay")).contains("true"); + assertThat(json.readTree(retry.body()).path("state").asText()).isEqualTo("retired"); + long retiredAt = System.currentTimeMillis(); + assertOperations(id, "retire_schedule", 1, 1); + assertThat(commands(session, "retireSchedule")).hasSize(1); + return () -> { + // The boundary has passed once the other clock-bound cases settled; a slot before + // the retire may have fired, none after it. + waiting().atMost(Duration.ofSeconds(70)).until(() -> System.currentTimeMillis() + >= (retiredAt / 60_000 + 1) * 60_000 + 5_000); + assertThat(jdbc.queryForList("SELECT occurrence_key FROM ma_schedule_occurrence WHERE tenant_id = ?" + + " AND schedule_id = ? AND slot > ?", String.class, tenant, id, retiredAt)).isEmpty(); + }; + } + + private boolean selected(String name) { + String selected = System.getProperty("qwen.fg11a.case", ""); + return selected.isEmpty() || selected.equals(name); + } + + private void arm(String name, Predicate rule) { + assertThat(armed.putIfAbsent(name, rule)).as(name).isNull(); + } + + private static boolean kind(JsonNode body, String kind) { + return kind.equals(body.path("kind").asText()); + } + + private static String prompt(JsonNode body) { + return body.path("definition").path("prompt").asText(); + } + + // A lost Harness answer reaches the caller as the operation's unknown outcome. + private static void assertRefused(HttpResponse response) { + assertThat(response.statusCode()).as(response.body()).isEqualTo(503); + assertThat(response.body()).contains("automation_operation_unknown"); + } + + private void assertOperations(String scheduleId, String kind, int total, int dropped) { + var matching = operations.stream().filter(operation -> scheduleId.equals(operation.scheduleId()) + && kind.equals(operation.kind())).toList(); + assertThat(matching).as("%s operations for %s", kind, scheduleId).hasSize(total); + assertThat(matching).filteredOn(Operation::dropped).hasSize(dropped); + } + + private boolean sameOperationId(String scheduleId, String kind) throws Exception { + var ids = new java.util.HashSet(); + for (Operation operation : operations) { + if (scheduleId.equals(operation.scheduleId()) && kind.equals(operation.kind())) { + ids.add(json.readTree(operation.body()).path("operationId").asText()); + } + } + return ids.size() == 1; + } + + // One occurrence, one run chain: the fires the Harness saw for it (the dropped one and its + // re-drive), its two fire revisions and its settle committed once, one accepted input and + // one model Turn. Everything is scoped to this occurrence: a later slot of an every-minute + // definition, due while the other cases run, is a run of its own. + private void assertOneRun(String session, String scheduleId, String runId, String occurrenceKey, + int fires) throws Exception { + waiting().atMost(Duration.ofSeconds(60)).until(() -> "completed".equals(runState(runId))); + var fired = operations.stream().filter(operation -> scheduleId.equals(operation.scheduleId()) + && "fire_run".equals(operation.kind()) && occurrenceKey.equals(operation.occurrenceKey())).toList(); + assertThat(fired).as("fires of %s", occurrenceKey).hasSize(fires); + assertThat(fired).filteredOn(Operation::dropped).hasSize(1); + assertThat(commands(session, "fireRun")).filteredOn(command -> command.startsWith(runId + ":")) + .containsExactly(runId + ":1", runId + ":2"); + assertThat(commands(session, "settleRun")).filteredOn(command -> command.startsWith(runId + ":")) + .containsExactly(runId + ":3"); + assertThat(recordRevision(session, "automation_run", runId)).isEqualTo(3); + assertThat(events(session, "input.accepted")).filteredOn(event -> event.startsWith(runId + ":")) + .containsExactly(runId + ":input:accepted"); + assertThat(events(session, "turn.settled")).filteredOn(event -> event.contains(runId)) + .containsExactly("turn:" + runId + ":input"); + assertThat(modelPrompts).filteredOn(prompt -> prompt.contains("Occurrence: " + occurrenceKey)).hasSize(1); + Map occurrence = jdbc.queryForMap("SELECT outcome, run_id FROM ma_schedule_occurrence" + + " WHERE tenant_id = ? AND schedule_id = ? AND occurrence_key = ?", tenant, scheduleId, + occurrenceKey); + assertThat(occurrence).containsEntry("outcome", "fired").containsEntry("run_id", runId); + } + + private Map awaitOccurrence(String scheduleId, String outcome) { + AtomicReference> found = new AtomicReference<>(); + waiting().atMost(Duration.ofSeconds(150)).until(() -> { + var rows = jdbc.queryForList("SELECT occurrence_key, outcome, reason, attempts, run_id FROM" + + " ma_schedule_occurrence WHERE tenant_id = ? AND schedule_id = ? AND occurrence_key LIKE" + + " 'schedule:%' ORDER BY slot", tenant, scheduleId); + if (rows.isEmpty() || "firing".equals(rows.getFirst().get("outcome"))) { + return false; + } + found.set(rows.getFirst()); + return true; + }); + assertThat(found.get()).as("the first slot of %s", scheduleId).containsEntry("outcome", outcome); + return found.get(); + } + + private List commands(String session, String operation) { + return jdbc.queryForList("SELECT CAST(command_id AS CHAR) FROM ma_session_journal_tx WHERE tenant_id = ?" + + " AND session_id = ? AND operation = ? ORDER BY journal_revision", String.class, tenant, session, + operation); + } + + // The event ids of one kind the Session's journal committed. + private List events(String session, String kind) throws Exception { + var ids = new ArrayList(); + for (byte[] record : jdbc.queryForList("SELECT record_bytes FROM ma_session_journal_tx WHERE tenant_id" + + " = ? AND session_id = ? ORDER BY journal_revision", byte[].class, tenant, session)) { + for (String line : new String(record, StandardCharsets.UTF_8).lines().toList()) { + JsonNode entry = json.readTree(line); + if ("managed_session_event_v1".equals(entry.path("subtype").asText()) + && kind.equals(entry.path("managedSession").path("kind").asText())) { + ids.add(entry.path("managedSession").path("eventId").asText()); + } + } + } + return ids; + } + + private long recordRevision(String session, String domain, String recordId) { + return jdbc.queryForObject("SELECT revision FROM ma_session_record WHERE tenant_id = ? AND session_id = ?" + + " AND domain = ? AND record_id = ?", Long.class, tenant, session, domain, recordId); + } + + private Map definition(String session, String marker, String cron) { + return Map.of("session_id", session, "goal", marker, "cron", cron, "timezone", "UTC", + "prompt", marker + " reply with FG11_DONE."); + } + + private String runState(String runId) { + List states = jdbc.queryForList("SELECT task_state FROM ma_session_record WHERE tenant_id = ?" + + " AND domain = 'automation_run' AND record_id = ?", String.class, tenant, runId); + return states.isEmpty() ? null : states.getFirst(); + } + + private void boot() throws Exception { + Path cli = Path.of(System.getProperty("qwen.cli.entry", "../../dist/cli.js")).toAbsolutePath(); + assertThat(cli).as("Build and bundle the CLI first").isRegularFile(); + node = System.getProperty("node.executable"); + assertThat(node).as("Pass -Dnode.executable with an absolute Node.js 22+ path").isNotBlank(); + temporary = temporary.toRealPath(); + decoy = Files.createDirectory(temporary.resolve("harness-decoy")); + Files.createDirectory(temporary.resolve("broker")); + Path root = Files.createDirectory(temporary.resolve("workspace-0")); + Files.createDirectory(root.resolve("child")); + // Keep the initialized root's mtime distinct from its birth time without sleeping. + Files.setLastModifiedTime(root, FileTime.fromMillis(1)); + int harnessPort; + int brokerPort; + try (var springSocket = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1")); + var harnessSocket = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1")); + var brokerSocket = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) { + port = springSocket.getLocalPort(); + harnessPort = harnessSocket.getLocalPort(); + brokerPort = brokerSocket.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(); + harnessRelay = new HostedLifecycleRelay(); + harnessRelay.backend(URI.create("http://127.0.0.1:" + harnessPort)); + harnessRelay.fault(new HostedLifecycleRelay.Fault() { + @Override + public boolean after(HostedLifecycleRelay.Request request, int status) throws Exception { + return relayed(request, status); + } + }); + startSpring(cli, root, brokerPort); + startHarness(cli, harnessPort, brokerPort); + register(root); + publicRelay = new HostedLifecycleRelay(); + publicRelay.backend(URI.create("http://127.0.0.1:" + port)); + publicRelay.fault(new HostedLifecycleRelay.Fault() { + @Override + public boolean after(HostedLifecycleRelay.Request request, int status) { + // The relay copies the exchange's headers, whose names it normalizes. + List key = request.headers().entrySet().stream() + .filter(header -> header.getKey().equalsIgnoreCase("Idempotency-Key")) + .flatMap(header -> header.getValue().stream()).toList(); + String wanted = publicDrop.get(); + // compareAndSet compares identity, so it is handed the armed instance itself. + return status != 202 || wanted == null || !key.contains(wanted) + || !publicDrop.compareAndSet(wanted, null); + } + }); + } + + // Records each automation operation and drops the reply of the first one an armed rule + // matches: the Harness answered, the connection closes, Spring never reads the 202. + private boolean relayed(HostedLifecycleRelay.Request request, int status) throws Exception { + if (!"POST".equals(request.method()) || !request.path().endsWith(OPERATIONS)) { + return true; + } + JsonNode body = json.readTree(request.body()); + boolean drop = false; + if (status == 202) { + for (var rule : armed.entrySet()) { + if (rule.getValue().test(body) && armed.remove(rule.getKey()) != null) { + drop = true; + break; + } + } + } + operations.add(new Operation(body.path("kind").asText(), body.path("scheduleId").asText(), + body.path("occurrenceKey").asText(), body.toString(), status, drop)); + return !drop; + } + + private void startSpring(Path cli, Path root, int brokerPort) { + List arguments = new ArrayList<>(List.of("--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", "root"), + "--spring.datasource.password=" + System.getProperty("mysql.password", ""), + "--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.harness.enabled=true", + "--qwen.managed-agent.harness.workspace-files-enabled=true", + "--qwen.managed-agent.harness.base-url=" + harnessRelay.uri(), + "--qwen.managed-agent.harness.token=" + TOKEN, + "--qwen.managed-agent.harness.capability-digest=" + DIGEST, + "--qwen.managed-agent.automation.enabled=true", + "--qwen.managed-agent.automation.scan-delay=1s", + "--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.trusted-local-reboot-recovery=false", + "--qwen.managed-agent.runtime-broker.durable-local-process=false", + "--qwen.managed-agent.runtime-broker.verified-workspace-recovery-enabled=" + + "Linux".equals(System.getProperty("os.name")), + "--qwen.managed-agent.runtime-broker.workspace-cwd=" + decoy, + "--qwen.managed-agent.runtime-broker.state-directory=" + temporary.resolve("broker"), + "--qwen.managed-agent.runtime-broker.credential-key-id=fg11-fixture", + "--qwen.managed-agent.runtime-broker.credential-key=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=", + "--qwen.managed-agent.runtime-broker.node-executable=" + node, + "--qwen.managed-agent.runtime-broker.worker-entry=" + cli, + "--qwen.managed-agent.runtime-broker.workspace-mounts[0].tenant-id=" + tenant, + "--qwen.managed-agent.runtime-broker.workspace-mounts[0].storage-id=storage-0", + "--qwen.managed-agent.runtime-broker.workspace-mounts[0].root=" + root)); + spring = (ServletWebServerApplicationContext) new SpringApplicationBuilder(ManagedAgentServerApplication.class) + .initializers(context -> context.getBeanFactory().registerSingleton("fg11Authentication", + authentication())) + .run(arguments.toArray(String[]::new)); + jdbc = spring.getBean(JdbcTemplate.class); + } + + private void register(Path root) { + jdbc.update("INSERT INTO ma_workspace (tenant_id, workspace_id, workspace_generation," + + " storage_id, display_name, config_ref, policy_ref, state) VALUES (?, ?, 1, ?, 'FG11', ?, ?," + + " 'ACTIVE')", tenant, "workspace-0", "storage-0", WorkspaceExecutionProfile.CONFIG_REF, + WorkspaceExecutionProfile.POLICY_REF); + jdbc.update("INSERT INTO ma_workspace_access (tenant_id, workspace_id, actor_id, role)" + + " VALUES (?, ?, ?, 'OPERATOR')", tenant, "workspace-0", "actor".getBytes(StandardCharsets.UTF_8)); + if ("Linux".equals(System.getProperty("os.name"))) { + spring.getBean(com.alibaba.qwen.code.managedagent.store.WorkspaceStorageGuard.class) + .register(tenant, "storage-0", UUID.randomUUID().toString()); + } + } + + private String createSession(String name) throws Exception { + var body = Map.of("agent_id", "qwen-code", "input", List.of(), "workspace", + Map.of("workspace_id", "workspace-0", "cwd_relative", "child")); + String session = request("POST", "/v1/agents/sessions", body, "session-" + name, 202).path("id").asText(); + waiting().atMost(Duration.ofSeconds(60)).untilAsserted(() -> assertThat(jdbc.queryForObject( + "SELECT status FROM ma_session WHERE tenant_id = ? AND session_id = ?", String.class, tenant, + session)).isEqualTo("ACTIVE")); + // The automation verbs load the Session on the Harness, never create it, so it runs + // one Turn first. + request("POST", "/v1/agents/sessions/" + session + "/events", Map.of("type", "agent.session.input.message", + "input", List.of(Map.of("type", "input_text", "text", "FG11_WARM " + name))), "warm-" + name, 202); + waiting().atMost(Duration.ofSeconds(60)).untilAsserted(() -> assertThat(jdbc.queryForList( + "SELECT status FROM ma_turn WHERE tenant_id = ? AND session_id = ?", String.class, tenant, session)) + .containsExactly("COMPLETED")); + return session; + } + + private FilterRegistrationBean authentication() { + FilterRegistrationBean filter = new FilterRegistrationBean<>((request, response, chain) -> + chain.doFilter(new HttpServletRequestWrapper((HttpServletRequest) request) { + @Override + public Principal getUserPrincipal() { + return new AuthenticatedTenantActor() { + public String tenantId() { return tenant; } + public String actorId() { return "actor"; } + public String getName() { return "actor"; } + }; + } + }, response)); + filter.setOrder(Ordered.HIGHEST_PRECEDENCE); + filter.setAsyncSupported(true); + filter.addUrlPatterns("/v1/agents/*", "/v1/agent-automations", "/v1/agent-automations/*"); + return filter; + } + + private void startHarness(Path cli, int harnessPort, int brokerPort) throws Exception { + Path home = Files.createDirectory(temporary.resolve("home")); + Path config = Files.createDirectory(home.resolve(".qwen")); + String modelUrl = "http://127.0.0.1:" + model.getAddress().getPort() + "/v1"; + Files.writeString(config.resolve("settings.json"), json.writeValueAsString(Map.of( + "security", Map.of("auth", Map.of("selectedType", "openai")), "model", Map.of("name", "fg11-fixture"), + "telemetry", Map.of("enabled", false), "modelProviders", Map.of("openai", List.of(Map.of( + "id", "fg11-fixture", "envKey", "OPENAI_API_KEY", "baseUrl", modelUrl)))))); + Path log = temporary.resolve("harness.log"); + 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", decoy.toString(), + "--managed-runtime-broker-url", "http://127.0.0.1:" + brokerPort, + "--managed-runtime-broker-token", TOKEN) + .directory(decoy.toFile()).redirectErrorStream(true).redirectOutput(log.toFile()); + Map environment = builder.environment(); + environment.clear(); + for (String name : List.of("PATH", "SystemRoot", "WINDIR", "COMSPEC", "PATHEXT")) { + if (System.getenv(name) != null) { + environment.put(name, System.getenv(name)); + } + } + environment.putAll(Map.of("HOME", home.toString(), "USERPROFILE", home.toString(), "QWEN_HOME", + config.toString(), "QWEN_RUNTIME_DIR", temporary.resolve("runtime").toString(), "OPENAI_API_KEY", + "fake-local-key", "OPENAI_BASE_URL", modelUrl, "QWEN_SERVER_TOKEN", TOKEN, + "QWEN_HOSTED_HARNESS_CAPABILITY_DIGEST", DIGEST, "TMPDIR", temporary.toString())); + for (String name : List.of("QWEN_CODE_SYSTEM_SETTINGS_PATH", "QWEN_CODE_SYSTEM_DEFAULTS_PATH", + "QWEN_CODE_TRUSTED_FOLDERS_PATH")) { + environment.put(name, temporary.resolve(name + ".json").toString()); + } + harness = builder.start(); + waiting().atMost(Duration.ofSeconds(60)).ignoreExceptions().untilAsserted(() -> { + if (!harness.isAlive()) { + throw new AssertionError("Hosted Harness exited: " + Files.readString(log)); + } + 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); + }); + } + + // Answers every wake Turn at once with text, except a prompt carrying FG11A_HOLD, which + // waits for the case to release it. + private void modelReply(HttpExchange exchange) throws IOException { + try { + JsonNode body = json.readTree(exchange.getRequestBody()); + String prompt = ""; + for (JsonNode message : body.path("messages")) { + if ("user".equals(message.path("role").asText())) { + prompt = message.path("content").toString(); + } + } + modelPrompts.add(prompt); + if (prompt.contains("FG11A_HOLD")) { + heldModel.await(120, TimeUnit.SECONDS); + } + var chunk = json.createObjectNode().put("id", "fg11").put("object", "chat.completion.chunk") + .put("created", 1).put("model", "fg11-fixture"); + var choice = chunk.putArray("choices").addObject().put("index", 0); + choice.putObject("delta").put("role", "assistant").put("content", "FG11_DONE"); + choice.putNull("finish_reason"); + String first = "data: " + json.writeValueAsString(chunk) + "\n\n"; + choice.putObject("delta"); + choice.put("finish_reason", "stop"); + byte[] response = (first + "data: " + json.writeValueAsString(chunk) + "\n\ndata: [DONE]\n\n") + .getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "text/event-stream"); + exchange.sendResponseHeaders(200, response.length); + exchange.getResponseBody().write(response); + } catch (Exception | AssertionError failure) { + modelFailure.compareAndSet(null, failure); + } finally { + exchange.close(); + } + } + + private void assertNoModelFailure() { + assertThat(modelFailure.get()).as("model stub").isNull(); + assertThat(harnessRelay.failure()).as("Harness relay").isNull(); + assertThat(publicRelay.failure()).as("public relay").isNull(); + } + + private JsonNode request(String method, String path, Object body, String key, int expected) throws Exception { + HttpResponse response = send(method, path, body, key); + assertThat(response.statusCode()).as("%s %s: %s", method, path, response.body()).isEqualTo(expected); + return json.readTree(response.body()); + } + + private HttpResponse send(String method, String path, Object body, String key) throws Exception { + return http.send(builder("http://127.0.0.1:" + port + path, method, body, key).build(), + HttpResponse.BodyHandlers.ofString()); + } + + private HttpRequest.Builder builder(String url, String method, Object body, String key) throws Exception { + HttpRequest.Builder request = HttpRequest.newBuilder(URI.create(url)).timeout(Duration.ofSeconds(60)) + .header("X-Qwen-Tenant-Id", tenant).header("Accept", "application/json") + .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); + } + return request; + } + + // The relay's operation log and the Harness log, kept for the CI upload either way. + private void writeEvidence() throws IOException { + if (temporary == null || harnessRelay == null) { + return; + } + Path cases = Files.createDirectories(Path.of("target", "fg11-cases", "fg11a", tenant)); + var lines = new ArrayList(); + for (Operation operation : operations) { + lines.add(json.writeValueAsString(Map.of("kind", operation.kind(), "scheduleId", + operation.scheduleId(), "occurrenceKey", operation.occurrenceKey(), "status", + operation.status(), "dropped", operation.dropped()))); + } + Files.write(cases.resolve("operations.jsonl"), lines); + Path log = temporary.resolve("harness.log"); + if (Files.exists(log)) { + Files.copy(log, cases.resolve("harness.log"), java.nio.file.StandardCopyOption.REPLACE_EXISTING); + } + } + + // Awaitility ignores the interrupt a @Timeout sends, so every wait fails on it. + private static ConditionFactory waiting() { + return await().failFast("The case's @Timeout interrupted it", () -> Thread.currentThread().isInterrupted()) + .pollInterval(Duration.ofMillis(200)); + } + + @AfterEach + void stop() throws Exception { + heldModel.countDown(); + if (scannerHold != null) { + scannerHold.countDown(); + } + Throwable failure = null; + List cleanup = List.of( + this::writeEvidence, + () -> { + if (harness != null) { + harness.descendants().forEach(ProcessHandle::destroyForcibly); + harness.destroyForcibly(); + harness.waitFor(10, TimeUnit.SECONDS); + } + }, + () -> { if (spring != null) { + spring.close(); + } }, + () -> { if (harnessRelay != null) { + harnessRelay.close(); + } }, + () -> { if (publicRelay != null) { + publicRelay.close(); + } }, + () -> { if (model != null) { + model.stop(0); + } }, + () -> { if (modelExecutor != null) { + modelExecutor.close(); + } }); + for (var step : cleanup) { + try { + step.execute(); + } catch (Throwable error) { + if (failure == null) { + failure = error; + } else { + failure.addSuppressed(error); + } + } + } + if (failure instanceof Exception error) { + throw error; + } + if (failure instanceof Error error) { + throw error; + } + } +} diff --git a/scripts/tests/hosted-process-ci.test.js b/scripts/tests/hosted-process-ci.test.js index 27cee9d5c2..d75e93d183 100644 --- a/scripts/tests/hosted-process-ci.test.js +++ b/scripts/tests/hosted-process-ci.test.js @@ -360,6 +360,10 @@ describe('Hosted real-process gates', () => { expect(upload.with.path).toContain( 'packages/managed-agent-server/target/fg10-cases/', ); + // FG11a's automation lost-reply evidence lives in target/fg11-cases/. + expect(upload.with.path).toContain( + 'packages/managed-agent-server/target/fg11-cases/', + ); // FG12a's MCP reply-loss evidence lives in target/fg12-cases/. expect(upload.with.path).toContain( 'packages/managed-agent-server/target/fg12-cases/', @@ -411,9 +415,9 @@ describe('Hosted real-process gates', () => { }); it.each([ - // Step ceilings today (95 + 10 + 20 + 50): Verify, the Broker fault + // Step ceilings today (100 + 10 + 20 + 50): Verify, the Broker fault // gates, admission and O4. The failover arms moved to their own job. - ['hosted-harness-mysql', 175, 190], + ['hosted-harness-mysql', 180, 195], // Fifteen failover arms at 10 each. ['hosted-failover-e2e', 150, 165], ])( @@ -593,12 +597,14 @@ describe('Hosted real-process gates', () => { // 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; the FG13d cancel/timeout cells add - // two cases budgeted at 200 s, so the fork moves to 5048): + // two cases budgeted at 200 s, so the fork moves to 5048; the FG11a + // automation lost-reply letter adds ~170 s, so the fork moves to 5218 + // and the Verify step to 100 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(5048); + expect(forkSeconds).toBe(5218); } }, ); @@ -633,10 +639,14 @@ describe('Hosted real-process gates', () => { // 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; the FG13d cancel/timeout - // cells (issue #14449, two cases) add 200s, so it becomes 4438s. + // cells (issue #14449, two cases) add 200s, so it becomes 4438s. The FG11a + // automation lost-reply letter (HostedAutomationFaultGatesIT, eight cases + // sharing one minute boundary) measures 46.3-85.0s locally on JDK 21 + + // MySQL 8.4.6 across four acceptance runs (~170s budgeted at 2x), so it + // becomes 4608s. // A larger ceiling lets failsafe kill the fork mid-run, and the classes // left unrun never report, failing the gate undiagnosably (#13780). - const hostedForkHealthySeconds = 4438; + const hostedForkHealthySeconds = 4608; const hostedForkSeconds = () => Number( read('packages/managed-agent-server/pom.xml') @@ -695,6 +705,7 @@ describe('Hosted real-process gates', () => { // diff; pinning the list puts that on a red here, not on a fork kill // on main (#14148). expect(hostedTestClasses().map(({ name }) => name)).toEqual([ + 'HostedAutomationFaultGatesIT', 'HostedChannelFaultGatesIT', 'HostedChannelProcessCrashIT', 'HostedCommittedEventLineReplayIT',