vrr/qwen-code
2
0
Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2026-10-19 12:43:39 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 17

test(managed-agent): add the FG11a automation lost-reply fault gates (#14446) (#14534)

Browse source 
* 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.
This commit is contained in:
Shaojin Wen 2026-10-18 15:22:12 +08:00 • committed by GitHub
parent 3d9f4b7e89
commit 7a2ce9b99a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 1134 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -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

175
docs/design/2026-10-18-automation-fault-gates-a.md Normal file
View file

@ -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=<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=<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.

139
docs/design/2026-10-18-automation-fault-gates-a.zh-CN.md Normal file
View file

@ -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=<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=<case>` 逐个重跑，而不是把其他用例当作已通过。
- 本设计以两种语言发布，结构一致。
## 未决问题
- FG11c 要 kill Spring，扫描器及其 lease 都在 Spring 里，所以它需要像 `HostedProcessCrashIT`
那样的子 JVM 夹具，而不是本类这种测试 JVM 内的 Spring。
- FG11b 在 Harness 侧注入 store 故障时，能否原样复用这个 relay，还是需要 FG9/FG10 relay
门禁那种挡在 store 前面的代理。

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

@ -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. -->
<forkedProcessTimeoutInSeconds>5048</forkedProcessTimeoutInSeconds>
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. -->
<forkedProcessTimeoutInSeconds>5218</forkedProcessTimeoutInSeconds>
</configuration>
<executions>
<execution>
@ -583,6 +590,24 @@
<executions><execution><goals><goal>integration-test</goal><goal>verify</goal></goals></execution></executions>
</plugin></plugins></build>
</profile>
<profile>
<id>hosted-automation-gates</id>
<build><plugins><plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-failsafe-plugin</artifactId>
<version>${maven-failsafe-plugin.version}</version>
<configuration>
<includes><include>**/HostedAutomation*IT.java</include></includes>
<failIfNoTests>true</failIfNoTests>
<!-- Mirrors the hosted-harness-mysql lane's
per-ceiling rule for this class: one @Timeout
ceiling (600 s) plus the healthy runtime must
fit 1200 s. -->
<forkedProcessTimeoutInSeconds>1200</forkedProcessTimeoutInSeconds>
</configuration>
<executions><execution><goals><goal>integration-test</goal><goal>verify</goal></goals></execution></executions>
</plugin></plugins></build>
</profile>
<profile>
<id>o4-mysql-gates</id>
<build><plugins><plugin>

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

@ -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<String> modelPrompts = new CopyOnWriteArrayList<>();
private final AtomicReference<Throwable> modelFailure = new AtomicReference<>();
private final CountDownLatch heldModel = new CountDownLatch(1);
// Every automation operation Spring sent the Harness, in order.
private final List<Operation> operations = new CopyOnWriteArrayList<>();
private final Map<String, Predicate<JsonNode>> 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<String> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> definition = definition(session, "FG11A_PUBLIC", "0 0 1 1 *");
publicDrop.set("public-reply");
try {
HttpResponse<String> 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<String> 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<String, Object> 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<String, Object> 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<String> 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<JsonNode> 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<String> 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<String>();
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<String, Object> 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<String, Object> awaitOccurrence(String scheduleId, String outcome) {
AtomicReference<Map<String, Object>> 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<String> 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<String> events(String session, String kind) throws Exception {
var ids = new ArrayList<String>();
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<String, Object> 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<String> 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<String> 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<String> 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<Filter> authentication() {
FilterRegistrationBean<Filter> 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<String, String> 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<String> 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<String> 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<String>();
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<org.junit.jupiter.api.function.Executable> 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;
}
}
}

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

@ -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',