mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-11-07 06:19:43 -10:00
test(managed-agent): add the FG8c process-crash gates for monitor_run and background shells (#14163) (#14407)
* fix(core): end the managed command launcher by its command's signal A background Shell's exit evidence is its cgroup launcher's own exit, and the launcher replaced a signal death with exit(1). A background process killed by SIGKILL was therefore recorded as `exited` with `exitCode: 1, exitSignal: null`, on the child_run record and on the Broker's `:process` row alike: a status the process never returned. The launcher now ends itself by the signal its command died of, so the waiter sees that signal. Only the signals that cannot be re-raised keep the nonzero fallback: a core-dumping one would dump the launcher too, and Node.js owns SIGUSR1 and ignores SIGPIPE. Command hooks share the launcher and already read a null exit code as a signal kill, a branch the old launcher left unreachable whenever only the command was killed. Found by the FG8c process-kill gate (#14163). * test(managed-agent): add the FG8c process-crash gates for monitor_run and background shells (#14163) FG8c kills (or stops) a real process under a live background run and pins what the platform does next against JDBC ground truth, in seven cases: the packaged Harness after the dispatch committed and after the capture started (then a cold reload on a fresh Harness), the managed worker (SIGKILL, and SIGSTOP across the broker's 30 s worker bound), the background process itself, the worker behind later prompts that reach for the Runtime again, and the worker under a live Monitor watch whose idle stop then meets the loss. Every case asserts one :start and one :process row across every boot and prompt, and exactly one Runtime binding at generation 1: a lost generation is never re-minted. Records may stay open, but never settle on evidence nobody produced; the reloaded close drains the run its process never saw (#13986); a dead worker's release is refused without stop proof; a stopped worker reads as neither stopped nor lost; and a watch whose stop nobody could prove parks as recovery_blocked / runtime_lost (#13896). The probe's state reads and the IT's record and output reads now lead with ma_resource's session_scope_key: filtered by the unindexed tenant and session columns alone, every poll scanned all the inline capture bytes the lane had written, so each later letter polled slower. The shared broker proxy also treats a client gone mid-request as no request instead of a sticky proxy failure. The hosted fork's healthy allowance moves 2640 -> 2930 s for the letter (twice its local 144 s, the FG9 convention); the 3600 s fork and the 600 s per-letter ceiling are unchanged. Seven mutation witnesses, one per case, were each observed red. The bilingual design document records the cases, two expectations of the round design that the merged recovery surface does not meet, and the follow-ups. * fix(core): keep only reachable signal names in the launcher's kept list Node.js on Linux reports signal 6 as SIGABRT, never SIGIOT, and Linux defines no SIGEMT, so neither name could ever reach the launcher's exit handler. Dropping them leaves the list as exactly the set it can match. --------- Co-authored-by: 易良 <[email protected]>
This commit is contained in:
Shaojin Wenand易良 authored and GitHub committed
1 parent
f12e848f87
commit
d8fcdf1e36
9 files changed
+1705
-53
No files matched your search
@@ -0,0 +1,302 @@
|
||||
# Monitor run fault gates FG8c
|
||||
|
||||
[English](2026-10-16-monitor-run-fault-gates-c.md) | [简体中文](2026-10-16-monitor-run-fault-gates-c.zh-CN.md)
|
||||
|
||||
## Problem and scope
|
||||
|
||||
Issue #14163 tracks FG8's fault gates for `monitor_run` and the
|
||||
`child_run` `shell` kind. FG8e/FG8f landed as #14120
|
||||
(`docs/design/2026-10-13-monitor-run-fault-gates.md`, the round design)
|
||||
and FG8a/FG8b/FG8d landed as #14334
|
||||
(`docs/design/2026-10-15-monitor-run-fault-gates-abd.md`). This document is
|
||||
the design for the last letter, **FG8c — process crashes**: the Harness,
|
||||
the managed worker, and the background process itself are killed (or
|
||||
stopped) for real under a live background run, and the gate pins what the
|
||||
platform does next against JDBC ground truth.
|
||||
|
||||
FG8c was parked behind #13986 (restart-proof the held-Runtime drain
|
||||
obligation), which rewrote the restart behaviour this letter pins. #13986
|
||||
merged as `c24e99e9cb`, so the letter is written against the merged
|
||||
surface.
|
||||
|
||||
The rig is #14120's and #14334's: one IT class
|
||||
(`HostedMonitorRunFaultGatesIT`), one probe (`HostedMonitorRunProbe`), one
|
||||
per-case TypeScript driver
|
||||
(`integration-tests/helpers/hosted-monitor-run-fault-driver.ts`), the real
|
||||
packaged Harness, the in-JVM Spring with its embedded broker, a real
|
||||
MySQL/MariaDB, and a named mutation witness per case. Unlike the earlier
|
||||
slices this one carries **one production fix**: the gate found that the
|
||||
cgroup launcher reported a SIGKILLed background process as an exit with
|
||||
status 1 (see "Finding: a signal death read as exit 1").
|
||||
|
||||
## Current state (verified)
|
||||
|
||||
Read at `main` = `b4614845f0` (which carries #14334 as `5af87595f3` and
|
||||
#14303's never-started probe fan-out), and confirmed case by case on a real
|
||||
lane
|
||||
(Linux, cgroup v2, MySQL 8.4.6, JDK 21, the packaged CLI):
|
||||
|
||||
- **Topology.** The embedded broker spawns one managed worker per Harness
|
||||
Session binding as a direct child of the Spring JVM
|
||||
(`node dist/cli.js managed-runtime-worker`). The worker supervises each
|
||||
background run through a cgroup v2 unit (`qwen-bg-<callId>` /
|
||||
`qwen-mon-<callId>`): worker → launcher → shell → command. Nothing
|
||||
re-attaches a replacement worker to a running unit, and with
|
||||
`durable-local-process=false` the provisioner never produces
|
||||
stopped-writer evidence, so a lost binding is never released.
|
||||
- **Harness restart.** The Harness observes a background run only through
|
||||
the worker's pushes into its in-process publisher; there is no
|
||||
cross-boot re-attachment. A reload therefore never re-learns a run. What
|
||||
#13986 restores is the obligation: at load the Harness reads the Session's
|
||||
live hold rows (`POST /tool-sessions:active-holds`) into its held set, and
|
||||
the close drains each with `{ drain: true }`
|
||||
(`packages/cli/src/serve/hosted-harness-session.ts:5849-5867`,
|
||||
`:9971-10013`).
|
||||
- **Cold-load refusal.** A reload whose launch turn never settled is
|
||||
refused at the `unsettled_input` gate with 409
|
||||
`hosted_turn_recovery_required`, reason `turn_unsettled`
|
||||
(`hosted-harness-session.ts:5750-5776`). The never-started reconcile only
|
||||
reads the `:process` row and skips any row the broker still has — here a
|
||||
PREPARED one — as ambiguous (`hosted-never-started-reconcile.ts:76-90`).
|
||||
- **Worker death.** The broker notices lazily (`isUsable` is
|
||||
`process.isAlive()`). A release over a dead worker is routed to
|
||||
`releaseUnusableSession` (`RuntimeBrokerService.java:2889-2894`), which
|
||||
turns the binding LOST without stop evidence and answers 503
|
||||
`runtime_reconciliation_required` (`:3015-3019`); the Harness close
|
||||
tolerates that answer, marks the drain unproven, and retires the Session
|
||||
(`hosted-harness-session.ts:9996-10013`). The background process survives
|
||||
the worker as an orphan.
|
||||
- **Generation.** After the loss, the first warm answers 503
|
||||
`runtime_provision_failed`; the next one answers 503
|
||||
`runtime_broker_runtime_lost` (`RuntimeBrokerService.java:3864-3886`),
|
||||
after the reclaim's cleanup abandoned the generation's unsettled rows
|
||||
against `<bindingId>:1:journal-lost` (`:3888-3993`). The slot keeps the
|
||||
LOST binding
|
||||
active, so no generation 2 is minted: releasing a LOST binding requires
|
||||
loss **and** stop evidence (`RuntimeBindingRecord.java:202-204`).
|
||||
- **Worker stop.** A stopped worker is still alive to the broker. A
|
||||
release sweep's status read runs into the broker's 30 s worker-request
|
||||
bound, the failure is swallowed (`RuntimeBrokerService.java:2769`), and a
|
||||
routine release still answers 409 `runtime_session_held`
|
||||
(`:2923-2953`). Nothing marks the Runtime lost; after `SIGCONT` the run
|
||||
continues and settles from its own evidence.
|
||||
- **Monitor auto-stop.** An idle or quota stop asks the worker through the
|
||||
broker's `monitor-stop` (`hosted-monitor-remote-executor.ts:50-66`);
|
||||
anything but exit evidence parks the record
|
||||
(`hosted-monitor-loop.ts:459-472`) as `recovery_blocked` /
|
||||
`runtime_lost` / `outcome_unknown` (`hosted-monitor-session.ts:298-308`).
|
||||
|
||||
## Scope decisions
|
||||
|
||||
1. **The gates pin physical, ledger and record truth, not branch logic**,
|
||||
as #14334's scope decision 2 recorded. Every assertion reads JDBC
|
||||
(`ma_tool_execution`, `ma_runtime_binding`, `ma_runtime_slot`,
|
||||
`ma_runtime_session`, `ma_session_record` bodies, the session journal,
|
||||
the output resources) or the driver's physical evidence (signals sent,
|
||||
pids alive or gone, broker answers by declared code).
|
||||
2. **No Spring kill** (round design decision 3). FG8c kills the Harness,
|
||||
the worker and the background process; the control plane stays up, so
|
||||
the in-JVM Spring and its probe keep serving ground truth throughout.
|
||||
3. **The cases follow the code, not the planned table.** The round
|
||||
design's FG8c table predates #13986 and the merged recovery surface.
|
||||
Two of its expectations do not hold, and the gate pins what does:
|
||||
- a reload after the launch turn settled is **not** refused — it loads
|
||||
(200), restores the held Runtime, and its close drains the run it
|
||||
never saw;
|
||||
- an unsettled `child_run` does not move to a blocked state — the
|
||||
record stays `running` (only a monitor parks, and only through its own
|
||||
in-process loop). The gate pins that the record never settles on
|
||||
evidence nobody produced, and that the ledger tells the truth.
|
||||
4. **A seventh case, `monitor-worker-kill`.** The planned table had no
|
||||
monitor case. A watch whose worker died is the #13896 rule under a real
|
||||
fault — the idle stop's `monitor-stop` meets a refusal, not a dropped
|
||||
reply as in FG8a's `monitor-stop-reply` — so it is gated here.
|
||||
5. **One production fix, not a test workaround.** `process-kill` cannot
|
||||
pass honestly against the old launcher (see the finding below); the fix
|
||||
is small, sits on the evidence path both consumers already expect, and
|
||||
lands with unit tests in the same PR.
|
||||
6. **Observed but not asserted: the orphan.** After a worker death the
|
||||
background process outlives the worker and the Session close in this
|
||||
topology; the Workspace-level cascade owns it (#13986 R2-1). Asserting
|
||||
the orphan would pin a leak; the gate only kills it in teardown.
|
||||
|
||||
## Finding: a signal death read as exit 1
|
||||
|
||||
The worker's exit evidence for a background run is the cgroup launcher's
|
||||
own exit, and the launcher's exit handler ended with
|
||||
`process.exit(code ?? 1)` (`packages/core/src/hooks/hook-command-cgroup.ts`).
|
||||
A command a
|
||||
signal killed has a null code, so the launcher exited 1: on the lane, a
|
||||
SIGKILLed background process settled the `child_run` record `exited` with
|
||||
`exitCode: 1, exitSignal: null`, and the `:process` row carried the same
|
||||
evidence. Status 1 is a value the process never returned.
|
||||
|
||||
The fix re-raises the command's signal on the launcher itself, so its
|
||||
waiter sees the same signal. Only the signals that cannot be re-raised
|
||||
keep the old fallback: a core-dumping signal would dump the launcher too,
|
||||
Node.js owns SIGUSR1 and ignores SIGPIPE. The hook runner reads the same
|
||||
launcher and already treats a null exit code as "killed by signal"
|
||||
(`hookRunner.ts`), a branch the old launcher left unreachable whenever only
|
||||
the command was killed. Unit tests (SIGKILL, SIGTERM, SIGUSR2) pin the
|
||||
behaviour and turn red on the old launcher.
|
||||
|
||||
The fix covers a killed process that is the launcher's direct child, which
|
||||
is what a background Shell runs: `bash -c` execs a single simple command.
|
||||
A compound command keeps bash in between, and bash's own status for a
|
||||
signalled child (128 + N) is then the honest evidence of the process the
|
||||
launcher ran.
|
||||
|
||||
## The gate ledger
|
||||
|
||||
Every case runs one Session (per-case workspace mount and storage) and a
|
||||
background run launched by a turn whose first model call is the tool call,
|
||||
under a 10 s writer lease: a reload outlives the killed writer without a
|
||||
minute's wait, the capture a crashed or closed Session leaves open expires
|
||||
within seconds — the fixture tenant admits four active captures, and seven
|
||||
cases in a row under a 60 s lease exhausted it on the seventh — and a
|
||||
renewal, due at half the lease, keeps 5 s of slack on a loaded lane. Common assertions for every case: the delivered signal is
|
||||
the case's fault (the fault fired); exactly one broker `:start` was
|
||||
answered across every boot, reload and later prompt, and exactly one
|
||||
`:process` row exists (no duplicate dispatch); the Session holds exactly
|
||||
one binding, at generation 1, and its slot's `last_generation` is 1 with
|
||||
that binding active (no re-mint, in any case).
|
||||
|
||||
| Case | Injection | Expect (JDBC / evidence) |
|
||||
| ----------------------- | -------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `harness-kill-dispatch` | Park the committed `:start` reply at the broker proxy (the process already runs), SIGKILL the Harness, reload on a fresh Harness | The reload answers 409 `hosted_turn_recovery_required` / `turn_unsettled`; its broker traffic is reads only (the never-started probe's `status` read) and the model is not asked again; the run stays alive and unknown — record `running` / `dispatch_started`, `stopReason` null, `:process` row PREPARED, no `turn.settled` event in the journal |
|
||||
| `harness-kill-capture` | After the launch turn settled and ≥ 3 full segments committed, SIGKILL the Harness; reload; close | ≥ 3 full segments stand at the reload, which answers 200 and its only broker call is `active-holds`; the close sends exactly one release, `drain: true`, naming the launch turn's Runtime Session, answered 200; the drain is what ends the process (alive after the reload, gone after the close); the `:process` row settles from the stop's evidence (`error`, signal, never `success`) and the Runtime Session is RELEASED; the record stays `running`; the durable prefix is byte-identical and nothing lands once the dead writer is fenced |
|
||||
| `worker-kill` | SIGKILL the case's worker under a live run; close | The close's drain answers 503 `runtime_reconciliation_required` and the close still answers 204; the binding is LOST with no stop evidence, the `:process` row stays PREPARED, the Runtime Session is not released; the record stays `running`, `stopReason` null |
|
||||
| `worker-stop` | SIGSTOP the worker; a routine release from the driver waits out the broker's 30 s worker bound; SIGCONT; open the run's gate | The release answers 409 `runtime_session_held` after ≥ 25 s; during the stop the record is `running`, the `:process` row PREPARED, the binding READY; after the resume the run settles from its own evidence — record `exited`, exit 0; `:process` row SETTLED `success`, exit 0; the binding stays READY |
|
||||
| `process-kill` | SIGKILL the background process itself (pid from the producer) | The record settles `exited` with `exitSignal: SIGKILL`, `exitCode` null, `stopRequested` false; the `:process` row is SETTLED `error` with the same evidence; the binding stays READY |
|
||||
| `generation-remint` | SIGKILL the worker; two more prompts each ask for a foreground Shell; close | The first prompt's warm answers 503 `runtime_provision_failed`, the second's 503 `runtime_broker_runtime_lost` (each prompt's only refusal); no other execution row exists (nothing ran on the lost generation); the binding is LOST with loss evidence and no stop evidence; the `:process` row is ABANDONED against `<bindingId>:1:journal-lost` with no result; the close's drain answers 503 `runtime_reconciliation_required`; the record stays `running` |
|
||||
| `monitor-worker-kill` | A watch with a 60 s idle window emits one observation and goes silent; SIGKILL the worker; the idle stop fires; close | Every `monitor-stop` answer is a refusal (≥ 400); the record parks `recovery_blocked` / `runtime_lost` / `outcome_unknown` with `stopReason` null and `notifiedThrough == observationSequence ≥ 1`; the binding is LOST without stop evidence, the `:process` row PREPARED; the close's drain answers 503 `runtime_reconciliation_required` |
|
||||
|
||||
## Fixture design
|
||||
|
||||
**Probe.** Two read-only phases join `state` and `ledger`: `workers`
|
||||
answers the JVM's live children carrying the `managed-runtime-worker`
|
||||
argument (the in-JVM broker spawns workers as direct children), and
|
||||
`runtime` answers the Session's bindings (isolation is per Harness
|
||||
Session), their slots and Runtime Sessions, and the full loss fields of its
|
||||
execution rows. The driver diffs `workers` around a case's first dispatch
|
||||
to name the one worker it signals.
|
||||
|
||||
**Driver.** One `runCrashCase` serves the letter. The broker proxy gains an
|
||||
answer log (operation, status, declared refusal code, the control kind,
|
||||
and a release's Runtime Session and arm) and one arm, `holdCommittedStart`,
|
||||
which parks the reply to a `:start` the broker already committed and
|
||||
destroys it after the Harness kill. The Session body is shared between
|
||||
create and `/load` so the reload carries the new Harness's writer identity;
|
||||
crash Sessions talk to the store directly, so a Harness killed mid-upload
|
||||
cannot leave the store proxy holding an aborted request (and the broker
|
||||
proxy treats a client gone mid-request as no request). The reload retries
|
||||
the writer conflict until the killed writer's lease lapses — its last
|
||||
renewal can commit just after the kill — and the segment count is taken
|
||||
after the reload, when the dead writer is fenced and any commit it had in
|
||||
flight has landed or been refused.
|
||||
The model can answer a post-crash prompt with a foreground Shell
|
||||
(`foregroundNext`), and the watch's idle window is 60 s for
|
||||
`monitor-worker-kill` only: the window runs from the watch's last flush —
|
||||
here its one observation — and on a loaded lane ~25 s passed between that
|
||||
and the kill (a 10 s window expired first and the stop met a live worker),
|
||||
so the case asserts the watch is still `running` at the kill. The worker is signalled by pid; the
|
||||
background
|
||||
process by the pid its producer wrote; the Harness by its child handle.
|
||||
Every wait is a state, never a sleep, except the writer-lease lapse
|
||||
before a reload and the pacing of its conflict retries. Teardown resumes a stopped worker and kills an orphaned
|
||||
background process. `FG8_CRASH_STATE` lines trace JDBC snapshots;
|
||||
`FG8_PROCESS_CRASH_OK` is the letter's terminal marker.
|
||||
|
||||
**IT.** `fg8cProcessCrashesNeverDuplicateOrFabricateOnMySql` runs the
|
||||
seven cases through `runLetter` (`@Timeout(600)`, selector
|
||||
`-Dqwen.fg8c.case`), and `assertCrashCase` holds every assertion in the
|
||||
ledger, reading the session journal for the terminal-event check.
|
||||
|
||||
**Indexed ground-truth reads.** `ma_resource` keys on the Session's
|
||||
`session_scope_key` and holds the inline capture bytes; the probe's `state`
|
||||
reads and the IT's `extensionRecord` / `loadOutputRows` filtered it by the
|
||||
unindexed `tenant_id` / `session_id` columns alone, so every poll scanned
|
||||
all the inline capture bytes the lane had written. Within one lane run that data
|
||||
only grows (FG8f inlines tens of MiB a case), and every later letter polled
|
||||
slower: on a local table of 664 MB a case's "run live" wait took 12–15 s
|
||||
instead of ~2 s, enough to push timing-sensitive waits past their bounds.
|
||||
The reads now lead with `session_scope_key`
|
||||
(`ManagedSessionStore.sessionScopeKey`), keeping the tenant and session
|
||||
predicates: the record-body read moves from a full scan to the
|
||||
`uq_ma_resource_id` unique index and the resource listing to a
|
||||
`session_scope_key`-led index; the `ma_session_record` read leads with the
|
||||
same key for consistency.
|
||||
|
||||
## Mutation checks
|
||||
|
||||
Each was applied, shown to turn its case red, then reverted (backup-copy
|
||||
mutations; TypeScript ones patched into the packaged bundle so the gate
|
||||
runs the mutated binary):
|
||||
|
||||
1. `harness-kill-dispatch` — the cold load's `unsettled_input` refusal
|
||||
skipped: red at the reload status (expected 409, was 200).
|
||||
2. `harness-kill-capture` — the #13986 restore dropped (the active-holds
|
||||
answer never seeds the held set): red at the close's releases (expected
|
||||
one, was none).
|
||||
3. `worker-kill` — a release over a dead worker sent down the live-worker
|
||||
path instead of refused: the drain fails and so does the close (expected
|
||||
204, was 503).
|
||||
4. `worker-stop` — the release sweep reading a failed status as a lost
|
||||
Runtime: red at "a stopped worker is not a lost Runtime" (expected
|
||||
READY, was LOST).
|
||||
5. `process-kill` — the launcher fix reverted: red at "the kill is the
|
||||
evidence" (expected `SIGKILL`, was null with exit 1).
|
||||
6. `generation-remint` — a LOST binding released on loss evidence alone:
|
||||
the broker re-mints, and the second prompt's warm answers 409
|
||||
`workspace_unavailable` instead of `runtime_broker_runtime_lost`.
|
||||
7. `monitor-worker-kill` — the remote terminate answering a proven stop
|
||||
without asking the worker (the class of shortcut #13929 removed): the
|
||||
watch stays `running` with nothing observing it, and the case goes red
|
||||
at its park wait.
|
||||
|
||||
## CI wiring and budgets
|
||||
|
||||
The letter runs in the `hosted-harness-mysql` lane with no workflow
|
||||
change: the class already belongs to it, the lane already delegates the
|
||||
cgroup v2 root the background runs need, and `check-failsafe-reports.js
|
||||
hosted` already requires the class to report. The letter measured 144 s
|
||||
locally for all seven cases on a fresh MySQL 8.4.6 (`monitor-worker-kill`
|
||||
spends ~65 s waiting out its idle window, `worker-stop` ~30 s waiting out
|
||||
the broker's worker bound). Following the FG9 convention, the addition
|
||||
moves `hostedForkHealthySeconds` 2640 → 2930 at twice the local
|
||||
measurement; the `@Timeout(600)` ceiling keeps the per-ceiling relation
|
||||
(600 + 2930 ≤ 3600) inside the 3600 s fork, to be confirmed by a lane run
|
||||
on this branch.
|
||||
|
||||
## Validation and acceptance
|
||||
|
||||
- The launcher unit tests green, the signal cases red on the old launcher; the core
|
||||
hook and supervisor suites green.
|
||||
- The letter green on a local Linux lane (cgroup v2 root delegated,
|
||||
MySQL 8.4.6, JDK 21, the packaged CLI), every case's fault positively
|
||||
observed. The whole `HostedMonitorRunFaultGatesIT` class run with the new
|
||||
letter beside the others on a fresh database: FG8c, FG8e and FG8f green;
|
||||
three of #14334's cases (`dispatch-reply`, `page-envelope-commit`,
|
||||
`cancel-reply-lost`) red, and red the same way on #14334's head without
|
||||
this change — they are that PR's open review items, not this letter's.
|
||||
- All seven mutations individually red; sources and bundle restored.
|
||||
- `hosted-process-ci.test.js` green; ESLint, Prettier and typecheck clean
|
||||
on the driver; Checkstyle clean on the Java sources.
|
||||
- Two consecutive clean open-ended and reverse audit passes per AGENTS.md.
|
||||
|
||||
## Open questions and follow-ups
|
||||
|
||||
1. **The cross-boot record.** After a Harness restart a `child_run`
|
||||
record stays `running` forever, even when the restored drain stopped the
|
||||
process and the ledger settled it. The design for the rebuild driver and
|
||||
the cross-boot case is still open
|
||||
(`docs/design/2026-10-03-managed-shell-monitor-runtime.md`); when it
|
||||
lands, `harness-kill-capture`'s record assertion moves to the settled
|
||||
form it produces.
|
||||
2. **The orphan after a worker death.** Nothing in this topology reaps the
|
||||
background process a dead worker supervised; whether the Workspace
|
||||
cascade should be gated for it belongs with that cascade's own gates.
|
||||
3. **Signals the launcher does not re-raise.** A core-dumping signal,
|
||||
SIGUSR1 and SIGPIPE still read as exit 1; reporting the leaf's evidence
|
||||
over the launcher's fd 3 would cover them, at the cost of a supervisor
|
||||
protocol change.
|
||||
@@ -0,0 +1,243 @@
|
||||
# Monitor run 故障门禁 FG8c
|
||||
|
||||
[English](2026-10-16-monitor-run-fault-gates-c.md) | [简体中文](2026-10-16-monitor-run-fault-gates-c.zh-CN.md)
|
||||
|
||||
## 问题与范围
|
||||
|
||||
Issue #14163 跟踪 FG8 为 `monitor_run` 与 `child_run` 的 `shell` kind
|
||||
补的故障门禁。FG8e/FG8f 已由 #14120 合入（轮次设计
|
||||
`docs/design/2026-10-13-monitor-run-fault-gates.md`），FG8a/FG8b/FG8d 已由
|
||||
#14334 合入（`docs/design/2026-10-15-monitor-run-fault-gates-abd.md`）。本文是最后
|
||||
一个字母 **FG8c —— 进程崩溃** 的设计：在一个存活的后台 run 之下，真实地杀掉
|
||||
（或停住）Harness、managed worker 以及后台进程本身，并用 JDBC 地面真值钉住平台
|
||||
接下来的行为。
|
||||
|
||||
FG8c 曾排在 #13986（让 held-Runtime 的 drain 义务重启可证）之后，因为后者改写
|
||||
了本字母要钉住的重启行为。#13986 已作为 `c24e99e9cb` 合入，本字母针对合入后的
|
||||
实现编写。
|
||||
|
||||
rig 沿用 #14120 与 #14334：一个 IT 类（`HostedMonitorRunFaultGatesIT`）、一个
|
||||
probe（`HostedMonitorRunProbe`）、一个按用例运行的 TypeScript driver
|
||||
（`integration-tests/helpers/hosted-monitor-run-fault-driver.ts`）、真实的打包
|
||||
Harness、带内嵌 broker 的进程内 Spring、真实的 MySQL/MariaDB，以及每个用例一个
|
||||
命名的变异见证。与前几片不同，本片带 **一处生产修复**：门禁发现 cgroup launcher
|
||||
把一个被 SIGKILL 的后台进程报告成了退出码为 1 的正常退出（见「发现：信号死亡被读
|
||||
成 exit 1」）。
|
||||
|
||||
## 现状（已核实）
|
||||
|
||||
在 `main` = `b4614845f0`（其中包含以 `5af87595f3` 合入的 #14334，以及 #14303 对
|
||||
never-started 探测的并发化）上阅读，并在真实车道（Linux、cgroup v2、MySQL 8.4.6、
|
||||
JDK 21、打包 CLI）上逐个用例确认：
|
||||
|
||||
- **拓扑。** 内嵌 broker 为每个 Harness Session 的 binding 拉起一个 managed
|
||||
worker，它是 Spring JVM 的直接子进程（`node dist/cli.js
|
||||
managed-runtime-worker`）。worker 通过 cgroup v2 单元（`qwen-bg-<callId>` /
|
||||
`qwen-mon-<callId>`）监管每个后台 run：worker → launcher → shell → 命令。没有
|
||||
任何东西会让替补 worker 重新附着到正在运行的单元；且在
|
||||
`durable-local-process=false` 下 provisioner 从不产出 stopped-writer 证据，所以
|
||||
一个丢失的 binding 永远不会被释放。
|
||||
- **Harness 重启。** Harness 只能通过 worker 推送到其进程内 publisher 的数据来
|
||||
观测后台 run，没有跨启动的重新附着。因此重新加载永远不会重新认识一个 run。
|
||||
#13986 恢复的是义务：加载时 Harness 把 Session 存活的 hold 行
|
||||
（`POST /tool-sessions:active-holds`）读进它的持有集合，close 再以
|
||||
`{ drain: true }` 逐个 drain
|
||||
（`packages/cli/src/serve/hosted-harness-session.ts:5849-5867`、
|
||||
`:9971-10013`）。
|
||||
- **冷加载拒绝。** 启动轮次从未结算的重新加载会在 `unsettled_input` 门处被拒，
|
||||
答 409 `hosted_turn_recovery_required`，reason 为 `turn_unsettled`
|
||||
（`hosted-harness-session.ts:5750-5776`）。never-started reconcile 只读取
|
||||
`:process` 行，并把 broker 仍持有的任何行——这里是 PREPARED 的行——当作含糊而
|
||||
跳过（`hosted-never-started-reconcile.ts:76-90`）。
|
||||
- **worker 死亡。** broker 惰性地发现它（`isUsable` 即 `process.isAlive()`）。
|
||||
对死 worker 的 release 被分流到 `releaseUnusableSession`
|
||||
（`RuntimeBrokerService.java:2889-2894`），后者把 binding 置为 LOST 且没有停止
|
||||
证据，并答 503 `runtime_reconciliation_required`（`:3015-3019`）；Harness 的
|
||||
close 容忍这个回答，把 drain 记为未证明，并让 Session 退役
|
||||
（`hosted-harness-session.ts:9996-10013`）。后台进程作为孤儿活过 worker。
|
||||
- **generation。** 丢失之后，第一次 warm 答 503 `runtime_provision_failed`；下一
|
||||
次答 503 `runtime_broker_runtime_lost`（`RuntimeBrokerService.java:3864-3886`），
|
||||
此前 reclaim 的清理已以 `<bindingId>:1:journal-lost` 放弃了该 generation 未结算
|
||||
的行（`:3888-3993`）。slot 仍把
|
||||
LOST 的 binding 视为活跃，所以不会铸出 generation 2：释放一个 LOST binding 需要
|
||||
丢失 **且** 停止的证据（`RuntimeBindingRecord.java:202-204`）。
|
||||
- **worker 被停住。** 被停住的 worker 对 broker 而言仍然存活。release sweep 的
|
||||
status 读会撞上 broker 对 worker 请求的 30 s 上限，失败被吞掉
|
||||
（`RuntimeBrokerService.java:2769`），例行 release 仍答 409
|
||||
`runtime_session_held`（`:2923-2953`）。没有任何东西把 Runtime 标为丢失；
|
||||
`SIGCONT` 之后 run 继续运行，并由自己的证据结算。
|
||||
- **Monitor 自动停止。** idle 或配额停止会通过 broker 的 `monitor-stop` 去问
|
||||
worker（`hosted-monitor-remote-executor.ts:50-66`）；除退出证据之外的任何结果
|
||||
都会让记录 park（`hosted-monitor-loop.ts:459-472`），成为 `recovery_blocked` /
|
||||
`runtime_lost` / `outcome_unknown`（`hosted-monitor-session.ts:298-308`）。
|
||||
|
||||
## 范围决策
|
||||
|
||||
1. **门禁钉住物理、账本与记录层面的真相，而不是分支逻辑**，与 #14334 的范围决策
|
||||
2 一致。每条断言都读取 JDBC（`ma_tool_execution`、`ma_runtime_binding`、
|
||||
`ma_runtime_slot`、`ma_runtime_session`、`ma_session_record` 的记录体、
|
||||
session journal、输出资源），或 driver 的物理证据（发出的信号、pid 存活或消
|
||||
失、按声明错误码记录的 broker 回答）。
|
||||
2. **不杀 Spring**（轮次设计决策 3）。FG8c 杀的是 Harness、worker 与后台进程；
|
||||
控制面保持在线，因此进程内 Spring 与它的 probe 全程持续提供地面真值。
|
||||
3. **用例跟随代码，而不是原计划的表。** 轮次设计里的 FG8c 表早于 #13986 及合入
|
||||
后的恢复实现。其中两条期望并不成立，门禁钉住的是实际成立的部分：
|
||||
- 启动轮次已结算后的重新加载 **不会** 被拒——它加载成功（200）、恢复被持有的
|
||||
Runtime，而它的 close 会 drain 那个它从未见过的 run；
|
||||
- 未结算的 `child_run` 不会进入阻塞状态——记录停留在 `running`（只有 monitor
|
||||
会 park，而且只能经由它自己的进程内 loop）。门禁钉住的是：记录永远不会在无人
|
||||
产出的证据上结算，而账本说的是真话。
|
||||
4. **第七个用例 `monitor-worker-kill`。** 原计划的表里没有 monitor 用例。worker
|
||||
已死的 watch 正是 #13896 规则在真实故障下的情形——idle 停止的 `monitor-stop`
|
||||
碰到的是拒绝，而不是 FG8a `monitor-stop-reply` 里的丢失应答——因此在这里设门。
|
||||
5. **一处生产修复，而不是测试绕行。** 在旧 launcher 上 `process-kill` 无法诚实
|
||||
地通过（见下文发现）；修复很小，落在两个使用方都已预期的证据路径上，并在同一
|
||||
PR 中附带单元测试。
|
||||
6. **观察到但不断言：孤儿进程。** worker 死后，在这种拓扑下后台进程会活过 worker
|
||||
与 Session 的 close；它归 Workspace 级联所有（#13986 R2-1）。断言这个孤儿等于
|
||||
钉住一个泄漏；门禁只在清理阶段杀掉它。
|
||||
|
||||
## 发现：信号死亡被读成 exit 1
|
||||
|
||||
worker 对一个后台 run 的退出证据，是 cgroup launcher 自身的退出；而 launcher 的
|
||||
退出处理以 `process.exit(code ?? 1)` 结束
|
||||
（`packages/core/src/hooks/hook-command-cgroup.ts`）。被信号杀死的命令其
|
||||
code 为 null，于是 launcher 以 1 退出：在车道上，一个被 SIGKILL 的后台进程把
|
||||
`child_run` 记录结算成了 `exited`，带 `exitCode: 1, exitSignal: null`，
|
||||
`:process` 行也带着同样的证据。状态 1 是这个进程从未返回过的值。
|
||||
|
||||
修复让 launcher 对自身重新发出命令的信号，使其等待方看到同一个信号。只有无法重
|
||||
新发出的信号保留原来的兜底：会产生 core dump 的信号会让 launcher 也 dump，Node.js
|
||||
自己占用 SIGUSR1 并忽略 SIGPIPE。hook runner 读取的是同一个 launcher，并且早已把
|
||||
null 退出码当作「被信号杀死」（`hookRunner.ts`）——而只要被杀的仅是命令本身，旧
|
||||
launcher 就让这个分支无法到达。单元测试（SIGKILL、SIGTERM、SIGUSR2）钉住这一行为，
|
||||
在旧 launcher 上变红。
|
||||
|
||||
修复覆盖的是作为 launcher 直接子进程的被杀进程，这正是后台 Shell 所运行的：
|
||||
`bash -c` 会 exec 一条简单命令。复合命令会让 bash 留在中间，此时 bash 对被信号杀死
|
||||
的子进程给出的自身状态（128 + N）就是 launcher 所运行进程的诚实证据。
|
||||
|
||||
## 门禁账本
|
||||
|
||||
每个用例运行一个 Session（每用例独立的 workspace mount 与 storage），以及一个由
|
||||
轮次启动的后台 run，该轮次的第一次模型调用就是工具调用，并使用 10 s 的 writer 租
|
||||
约：重新加载不用等一分钟就能越过被杀 writer 的租约；崩溃或已关闭的 Session 留下的
|
||||
OPEN capture 会在几秒内过期——夹具租户只容许四个活跃 capture，在 60 s 租约下连续
|
||||
七个用例会在第七个用例时将其耗尽；而在租约一半时到期的续约，在高负载车道上仍有
|
||||
5 s 的余量。所有用例共有的断言：发出的
|
||||
信号就是该用例的故障（故障确实发生）；在所有启动、重新加载与后续 prompt 中，broker
|
||||
恰好只应答过一次 `:start`，且恰好存在一个 `:process` 行（没有重复派发）；Session
|
||||
恰好持有一个 binding，generation 为 1，其 slot 的 `last_generation` 为 1 且该
|
||||
binding 处于活跃（任何用例中都没有重铸）。
|
||||
|
||||
| 用例 | 注入 | 期望（JDBC / 证据） |
|
||||
| ----------------------- | ------------------------------------------------------------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `harness-kill-dispatch` | 在 broker 代理处扣住已提交的 `:start` 应答（进程已在运行），SIGKILL Harness，在新 Harness 上重新加载 | 重新加载答 409 `hosted_turn_recovery_required` / `turn_unsettled`；它的 broker 流量只有读操作（never-started probe 的 `status` 读），且不再询问模型；run 保持存活且结局未知——记录为 `running` / `dispatch_started`，`stopReason` 为 null，`:process` 行为 PREPARED，journal 中没有 `turn.settled` 事件 |
|
||||
| `harness-kill-capture` | 启动轮次结算且至少 3 个完整段提交之后，SIGKILL Harness；重新加载；close | 重新加载时已有 ≥ 3 个完整段，重新加载答 200，唯一的 broker 调用是 `active-holds`；close 恰好发出一次 release，`drain: true`，指名启动轮次的 Runtime Session，答 200；正是这次 drain 结束了进程（重新加载后仍存活，close 后消失）；`:process` 行由停止的证据结算（`error`、信号，绝不是 `success`），Runtime Session 为 RELEASED；记录保持 `running`；持久前缀逐字节相同，被杀 writer 被 fence 之后没有任何段落盘 |
|
||||
| `worker-kill` | 在存活 run 下 SIGKILL 该用例的 worker；close | close 的 drain 答 503 `runtime_reconciliation_required`，close 仍答 204；binding 为 LOST 且没有停止证据，`:process` 行保持 PREPARED，Runtime Session 没有被释放；记录保持 `running`，`stopReason` 为 null |
|
||||
| `worker-stop` | SIGSTOP worker；driver 发出的例行 release 等满 broker 对 worker 的 30 s 上限；SIGCONT；打开 run 的闸门 | release 在 ≥ 25 s 后答 409 `runtime_session_held`；停住期间记录为 `running`、`:process` 行为 PREPARED、binding 为 READY；恢复后 run 由自身证据结算——记录 `exited`、exit 0；`:process` 行 SETTLED `success`、exit 0；binding 保持 READY |
|
||||
| `process-kill` | SIGKILL 后台进程本身（pid 由 producer 写出） | 记录结算为 `exited`，`exitSignal: SIGKILL`，`exitCode` 为 null，`stopRequested` 为 false；`:process` 行为 SETTLED `error`，带同样的证据；binding 保持 READY |
|
||||
| `generation-remint` | SIGKILL worker；之后两个 prompt 各自要求一次前台 Shell；close | 第一个 prompt 的 warm 答 503 `runtime_provision_failed`，第二个答 503 `runtime_broker_runtime_lost`（各自是该 prompt 唯一的拒绝）；不存在其他执行行（丢失的 generation 上什么都没跑）；binding 为 LOST，有丢失证据而无停止证据；`:process` 行以 `<bindingId>:1:journal-lost` 被 ABANDONED 且没有结果；close 的 drain 答 503 `runtime_reconciliation_required`；记录保持 `running` |
|
||||
| `monitor-worker-kill` | 一个 idle 窗口为 60 s 的 watch 产出一次观测后沉默；SIGKILL worker；idle 停止触发；close | 每个 `monitor-stop` 的回答都是拒绝（≥ 400）；记录 park 为 `recovery_blocked` / `runtime_lost` / `outcome_unknown`，`stopReason` 为 null，且 `notifiedThrough == observationSequence ≥ 1`；binding 为 LOST 且没有停止证据，`:process` 行为 PREPARED；close 的 drain 答 503 `runtime_reconciliation_required` |
|
||||
|
||||
## 夹具设计
|
||||
|
||||
**Probe。** 在 `state` 与 `ledger` 之外新增两个只读阶段：`workers` 返回 JVM 中带
|
||||
`managed-runtime-worker` 参数的存活子进程（进程内 broker 以直接子进程的方式拉起
|
||||
worker）；`runtime` 返回该 Session 的 binding（隔离粒度是 Harness Session）、它们
|
||||
的 slot 与 Runtime Session，以及其执行行完整的丢失字段。driver 在用例第一次派发
|
||||
前后对 `workers` 做差集，以确定它要发信号的那一个 worker。
|
||||
|
||||
**Driver。** 一个 `runCrashCase` 服务整个字母。broker 代理新增一份应答日志（操
|
||||
作、状态、声明的拒绝码、control kind，以及 release 的 Runtime Session 与臂），以
|
||||
及一个臂 `holdCommittedStart`：扣住 broker 已提交的 `:start` 的应答，并在 Harness
|
||||
被杀之后销毁它。create 与 `/load` 共用同一个 Session 请求体，因此重新加载带上的是
|
||||
新 Harness 的 writer 身份；崩溃用例的 Session 直连 store，因此在上传中途被杀的
|
||||
Harness 不会让 store 代理留下一个被中止的请求（broker 代理也把请求中途离开的客户
|
||||
端视为没有请求）。重新加载会对 writer 冲突重试，直到被杀 writer 的租约过期——它最
|
||||
后一次续约可能恰好在 kill 之后才提交——段计数则在重新加载之后才取，那时死去的
|
||||
writer 已被 fence，它在途的提交要么已落盘、要么已被拒。模型可以用一次前台 Shell 回答崩溃后的 prompt
|
||||
（`foregroundNext`），且只有 `monitor-worker-kill` 的 watch 使用 60 s 的 idle 窗口：
|
||||
窗口从 watch 最后一次 flush——这里就是它唯一的那次观测——开始计时，而在高负载的
|
||||
车道上，从那时到 kill 之间过去了约 25 s（10 s 的窗口先到期，停止碰到的是存活的
|
||||
worker），因此该用例会断言 kill 时 watch 仍为 `running`。
|
||||
worker 按 pid 发信号；后台进程按其 producer 写出的 pid；Harness 按其子进程句柄。
|
||||
除了重新加载前等 writer 租约过期以及冲突重试之间的间隔之外，每次等待都是等一个
|
||||
状态，而不是 sleep。清理阶段会恢复被停住的 worker，并杀掉孤儿后台进程。`FG8_CRASH_STATE` 行
|
||||
记录 JDBC 快照；`FG8_PROCESS_CRASH_OK` 是本字母的终态标记。
|
||||
|
||||
**IT。** `fg8cProcessCrashesNeverDuplicateOrFabricateOnMySql` 通过 `runLetter`
|
||||
运行这七个用例（`@Timeout(600)`，选择器 `-Dqwen.fg8c.case`），`assertCrashCase`
|
||||
持有账本中的全部断言，并读取 session journal 做终态事件检查。
|
||||
|
||||
**走索引的地面真值读取。** `ma_resource` 以 Session 的 `session_scope_key` 为键，
|
||||
并存放 inline capture 字节；probe 的 `state` 读取以及 IT 的 `extensionRecord` /
|
||||
`loadOutputRows` 只按未建索引的 `tenant_id` / `session_id` 列过滤它，于是每次轮询
|
||||
都会扫描车道迄今写入的全部 inline capture 字节。在一次车道运行内这些数据只增不减
|
||||
（FG8f 每个用例 inline 数十 MiB），每个靠后的字母轮询得都更慢：在本地一张 664 MB
|
||||
的表上，一个用例的「run live」等待花了 12–15 s 而不是约 2 s，足以把时序敏感的等待
|
||||
推出其上限。现在这些读取以 `session_scope_key`
|
||||
（`ManagedSessionStore.sessionScopeKey`）开头，并保留 tenant 与 session 条件：记录
|
||||
体的读取从全表扫描变为走 `uq_ma_resource_id` 唯一索引，资源列表变为走以
|
||||
`session_scope_key` 开头的索引；`ma_session_record` 的读取为保持一致也以同一个键
|
||||
开头。
|
||||
|
||||
## 变异检查
|
||||
|
||||
每一个都经过施加、证明让其用例变红、再还原（备份拷贝式变异；TypeScript 的变异
|
||||
打进打包产物，让门禁运行的就是被变异的二进制）：
|
||||
|
||||
1. `harness-kill-dispatch` —— 跳过冷加载的 `unsettled_input` 拒绝：在重新加载的
|
||||
状态处变红（期望 409，实际 200）。
|
||||
2. `harness-kill-capture` —— 去掉 #13986 的恢复（active-holds 的回答从不回填持有
|
||||
集合）：在 close 的 release 处变红（期望一次，实际零次）。
|
||||
3. `worker-kill` —— 对死 worker 的 release 不再拒绝，而是走存活 worker 的路径：
|
||||
drain 失败，close 也随之失败（期望 204，实际 503）。
|
||||
4. `worker-stop` —— release sweep 把失败的 status 读成 Runtime 丢失：在「被停住
|
||||
的 worker 不是丢失的 Runtime」处变红（期望 READY，实际 LOST）。
|
||||
5. `process-kill` —— 还原 launcher 修复：在「kill 就是证据」处变红（期望
|
||||
`SIGKILL`，实际为 null 且 exit 1）。
|
||||
6. `generation-remint` —— 仅凭丢失证据就释放 LOST binding：broker 去重铸，第二个
|
||||
prompt 的 warm 答 409 `workspace_unavailable` 而不是
|
||||
`runtime_broker_runtime_lost`。
|
||||
7. `monitor-worker-kill` —— 远端 terminate 不问 worker 就回答「已证明停止」
|
||||
（#13929 去掉的那一类捷径）：watch 停留在 `running` 且无人再观测它，用例在等
|
||||
待 park 处变红。
|
||||
|
||||
## CI 接线与预算
|
||||
|
||||
本字母在 `hosted-harness-mysql` 车道运行，无需改 workflow：该类本就属于这条车道，
|
||||
车道已经为后台 run 委托了所需的 cgroup v2 根，`check-failsafe-reports.js hosted`
|
||||
也已要求该类必须产出报告。在全新的 MySQL 8.4.6 上，本地七个用例共耗时 144 s（其
|
||||
中 `monitor-worker-kill` 约 65 s 用于等满它的 idle 窗口，`worker-stop` 约 30 s 用于
|
||||
等满 broker 对 worker 的上限）。按 FG9 的惯例，这次新增按本地实测的两倍把
|
||||
`hostedForkHealthySeconds` 从 2640 调到 2930；`@Timeout(600)` 的上限让单上限关系
|
||||
（600 + 2930 ≤ 3600）保持在 3600 s 的 fork 之内，并由本分支上的一次车道运行来确
|
||||
认。
|
||||
|
||||
## 验证与验收
|
||||
|
||||
- launcher 单元测试为绿，其中的信号用例在旧 launcher 上为红；core 的 hook 与 supervisor 套件
|
||||
为绿。
|
||||
- 本字母在本地 Linux 车道（已委托 cgroup v2 根、MySQL 8.4.6、JDK 21、打包 CLI）上
|
||||
为绿，每个用例的故障都被正向观测到。在全新数据库上，整个
|
||||
`HostedMonitorRunFaultGatesIT` 类与新字母并存运行：FG8c、FG8e 与 FG8f 为绿；
|
||||
#14334 的三个用例（`dispatch-reply`、`page-envelope-commit`、
|
||||
`cancel-reply-lost`）为红，且在不含本改动的 #14334 head 上以同样方式变红——它们
|
||||
是那个 PR 尚未关闭的评审项，而不是本字母的问题。
|
||||
- 七个变异各自为红；源码与打包产物均已还原。
|
||||
- `hosted-process-ci.test.js` 为绿；driver 的 ESLint、Prettier 与类型检查干净；
|
||||
Java 源码的 Checkstyle 干净。
|
||||
- 按 AGENTS.md 连续两轮干净的无方向审计与反向审计。
|
||||
|
||||
## 未决问题与后续
|
||||
|
||||
1. **跨启动的记录。** Harness 重启之后，`child_run` 记录会永远停留在 `running`，
|
||||
即便恢复出来的 drain 已经停掉了进程、账本也已结算。rebuild driver 与跨启动情形
|
||||
的设计仍未完成（`docs/design/2026-10-03-managed-shell-monitor-runtime.md`）；
|
||||
它落地后，`harness-kill-capture` 的记录断言改为它产出的结算形态。
|
||||
2. **worker 死亡后的孤儿。** 在这种拓扑下，没有任何东西回收死 worker 曾监管的后台
|
||||
进程；Workspace 级联是否应为此设门，归属于该级联自己的门禁。
|
||||
3. **launcher 不重发的信号。** 会 core dump 的信号、SIGUSR1 与 SIGPIPE 仍会被读
|
||||
成 exit 1；通过 launcher 的 fd 3 报告叶子进程的证据可以覆盖它们，代价是改动
|
||||
supervisor 的协议。
|
||||
@@ -195,26 +195,38 @@ async function probe<T>(
|
||||
}
|
||||
|
||||
// One Session per case on the background-capable v3 lane: the captureBytes
|
||||
// budget is what admits is_background and monitor at all.
|
||||
// budget is what admits is_background and monitor at all. FG8c's reload
|
||||
// sends the same body to /load under the new Harness's writer identity.
|
||||
function sessionBody(
|
||||
cli: HostedHarnessProcess,
|
||||
storeBaseUrl: string,
|
||||
leaseDurationMs: number,
|
||||
) {
|
||||
return {
|
||||
sessionScope: 'thread',
|
||||
toolProfile: 'hosted-workspace-shell/1',
|
||||
captureBytes: 64 * 1024 * 1024,
|
||||
managedSessionStore: {
|
||||
baseUrl: storeBaseUrl,
|
||||
tenantId: config.tenantId,
|
||||
workspaceId: current.workspaceId,
|
||||
writerId: cli.bootId,
|
||||
leaseDurationMs,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function createSession(
|
||||
cli: HostedHarnessProcess,
|
||||
storeBaseUrl: string,
|
||||
leaseDurationMs = 60_000,
|
||||
): Promise<{ clientId: string; created: Record<string, unknown> }> {
|
||||
const created = await cli.request('/session', {
|
||||
method: 'POST',
|
||||
headers: { ...cli.headers(), 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({
|
||||
sessionId: current.sessionId,
|
||||
sessionScope: 'thread',
|
||||
toolProfile: 'hosted-workspace-shell/1',
|
||||
captureBytes: 64 * 1024 * 1024,
|
||||
managedSessionStore: {
|
||||
baseUrl: storeBaseUrl,
|
||||
tenantId: config.tenantId,
|
||||
workspaceId: current.workspaceId,
|
||||
writerId: cli.bootId,
|
||||
leaseDurationMs: 60_000,
|
||||
},
|
||||
...sessionBody(cli, storeBaseUrl, leaseDurationMs),
|
||||
}),
|
||||
});
|
||||
const createdBody = await created.text();
|
||||
@@ -404,12 +416,28 @@ function pacedProducer(): string {
|
||||
].join('\n');
|
||||
}
|
||||
|
||||
// FG8c's monitor producer: announce its pid, emit one observation, then
|
||||
// hold on a FIFO the driver never opens — the watch is live and silent, so
|
||||
// only its idle window can end it.
|
||||
function crashMonitorProducer(): string {
|
||||
return [
|
||||
"const fs = require('fs');",
|
||||
"fs.writeFileSync('bg-started.txt', String(process.pid));",
|
||||
"process.stdout.write('FG8_MON_0\\n');",
|
||||
"fs.readFileSync('monitor-gate');",
|
||||
].join('\n');
|
||||
}
|
||||
|
||||
function producerFor(name: string): { monitor: boolean; script: string } {
|
||||
if (name === 'monitor-worker-kill') {
|
||||
return { monitor: true, script: crashMonitorProducer() };
|
||||
}
|
||||
if (name.includes('monitor') || name.startsWith('observation')) {
|
||||
return { monitor: true, script: monitorProducer() };
|
||||
}
|
||||
switch (name) {
|
||||
case 'publisher-kill-mid-capture':
|
||||
case 'harness-kill-capture':
|
||||
return { monitor: false, script: pacedProducer() };
|
||||
case 'in-flight-cap':
|
||||
case 'receipt-drop-after-watermark':
|
||||
@@ -491,6 +519,24 @@ let heldStartArrived = false;
|
||||
let pausedBrokerControls = false;
|
||||
const heldStartReplies: Array<(destroy: boolean) => void> = [];
|
||||
const heldControlReplies: Array<() => void> = [];
|
||||
// FG8c: every broker answer (op, status, declared code; a release's
|
||||
// Runtime Session and arm), and the after-commit :start hold.
|
||||
const brokerAnswers: Array<{
|
||||
op: string;
|
||||
status: number;
|
||||
code: string;
|
||||
kind?: string;
|
||||
runtimeSessionId?: string;
|
||||
drain?: boolean;
|
||||
}> = [];
|
||||
let holdCommittedStart = false;
|
||||
let committedStartHeld = false;
|
||||
const heldCommittedStarts: Array<() => void> = [];
|
||||
|
||||
function releaseCommittedStarts() {
|
||||
holdCommittedStart = false;
|
||||
for (const resolve of heldCommittedStarts.splice(0)) resolve();
|
||||
}
|
||||
|
||||
function releaseHeldStart(destroy: boolean) {
|
||||
holdNextStart = false;
|
||||
@@ -509,7 +555,15 @@ function releaseBrokerControls() {
|
||||
const brokerProxy = createServer(async (req, res) => {
|
||||
try {
|
||||
const chunks: Buffer[] = [];
|
||||
for await (const chunk of req) chunks.push(Buffer.from(chunk));
|
||||
try {
|
||||
for await (const chunk of req) chunks.push(Buffer.from(chunk));
|
||||
} catch (cause) {
|
||||
// A request body fails only with its connection — a client killed
|
||||
// mid-request, as FG8c kills the Harness: the request was never
|
||||
// forwarded, and the proxy itself has not failed.
|
||||
console.log(`FG8_PROXY client went away mid-request: ${cause}`);
|
||||
return;
|
||||
}
|
||||
const body = Buffer.concat(chunks);
|
||||
const url = new URL(req.url!, config.brokerUrl);
|
||||
const operation =
|
||||
@@ -585,6 +639,50 @@ const brokerProxy = createServer(async (req, res) => {
|
||||
}
|
||||
const bytes = Buffer.from(await upstream.arrayBuffer());
|
||||
console.log(`FG8_PROXY ${req.method} ${operation} -> ${upstream.status}`);
|
||||
// FG8c reads every broker answer by its declared code — a crash case's
|
||||
// refusals are its evidence — and every release by the Runtime Session
|
||||
// it names and the arm it asked for.
|
||||
let answerCode = '';
|
||||
if (upstream.status >= 400) {
|
||||
try {
|
||||
answerCode = String(
|
||||
(JSON.parse(bytes.toString('utf8')) as Record<string, unknown>)[
|
||||
'code'
|
||||
] ?? '',
|
||||
);
|
||||
} catch {
|
||||
// an unparsable refusal keeps the empty code
|
||||
}
|
||||
}
|
||||
const released = /\/tool-sessions\/([^/]+):release$/.exec(url.pathname);
|
||||
brokerAnswers.push({
|
||||
op: operation,
|
||||
status: upstream.status,
|
||||
code: answerCode,
|
||||
...(controlKind ? { kind: controlKind } : {}),
|
||||
...(released
|
||||
? {
|
||||
runtimeSessionId: decodeURIComponent(released[1]),
|
||||
drain: parsed?.['drain'] === true,
|
||||
}
|
||||
: {}),
|
||||
});
|
||||
if (
|
||||
holdCommittedStart &&
|
||||
operation === 'start' &&
|
||||
upstream.status === 200
|
||||
) {
|
||||
// FG8c harness-kill-dispatch: the broker committed the admission and
|
||||
// started the process; the reply never reaches the Harness, which the
|
||||
// case kills while this socket is parked.
|
||||
holdCommittedStart = false;
|
||||
committedStartHeld = true;
|
||||
console.log('FG8_PROXY holding the committed :start reply');
|
||||
await new Promise<void>((resolve) => heldCommittedStarts.push(resolve));
|
||||
console.log('FG8_DROPPED broker committed :start reply');
|
||||
req.socket.destroy();
|
||||
return;
|
||||
}
|
||||
const matchDrop =
|
||||
(dropNextStart && operation === 'start') ||
|
||||
(dropNextMonitorStop && isMonitorStop);
|
||||
@@ -772,9 +870,29 @@ function checkStoreProxy() {
|
||||
|
||||
let current = config.cases[0];
|
||||
let modelCalls = 0;
|
||||
// FG8c generation-remint: the post-crash prompt's first model call asks for
|
||||
// a foreground Shell, so the turn awaits the Runtime warm every turn issues
|
||||
// and its refusal lands among that prompt's broker answers (the Shell
|
||||
// itself never runs).
|
||||
let foregroundNext = false;
|
||||
const model = await startFakeOpenAIServer(() => {
|
||||
modelCalls++;
|
||||
const producer = producerFor(current.name);
|
||||
if (foregroundNext) {
|
||||
foregroundNext = false;
|
||||
return {
|
||||
toolCalls: [
|
||||
fakeToolCall(
|
||||
'run_shell_command',
|
||||
{
|
||||
command: 'echo FG8_FOREGROUND',
|
||||
description: `fg8 ${current.name} foreground`,
|
||||
},
|
||||
'foreground-shell',
|
||||
),
|
||||
],
|
||||
};
|
||||
}
|
||||
if (modelCalls === 1) {
|
||||
return {
|
||||
toolCalls: [
|
||||
@@ -788,7 +906,13 @@ const model = await startFakeOpenAIServer(() => {
|
||||
// WebShell resume for a full poll cycle; the watch's idle
|
||||
// window must outlive that, or it idle-kills the gated
|
||||
// producer mid-gap and the gate never reopens.
|
||||
idle_timeout_ms: 180_000,
|
||||
// FG8c's monitor-worker-kill wants the idle stop inside
|
||||
// the case, after the worker is gone: the window runs from
|
||||
// the watch's last flush (its one observation), and a loaded
|
||||
// lane was seen to take ~25 s from there to the kill, so
|
||||
// 60 s keeps the order.
|
||||
idle_timeout_ms:
|
||||
current.name === 'monitor-worker-kill' ? 60_000 : 180_000,
|
||||
description: `fg8 ${current.name}`,
|
||||
},
|
||||
'monitor-watch',
|
||||
@@ -1740,6 +1864,7 @@ async function bootHarness() {
|
||||
async function brokerRequest(
|
||||
pathSuffix: string,
|
||||
fields: Record<string, unknown>,
|
||||
timeoutMs = 30_000,
|
||||
): Promise<{ status: number; code: string; body: Record<string, unknown> }> {
|
||||
const runtimeSessionId = brokerIdentity.runtimeSessionId;
|
||||
assert(
|
||||
@@ -1761,7 +1886,7 @@ async function brokerRequest(
|
||||
runtimeSessionId,
|
||||
...fields,
|
||||
}),
|
||||
signal: AbortSignal.timeout(30_000),
|
||||
signal: AbortSignal.timeout(timeoutMs),
|
||||
},
|
||||
);
|
||||
const text = await response.text();
|
||||
@@ -1779,10 +1904,11 @@ async function brokerRequest(
|
||||
};
|
||||
}
|
||||
|
||||
async function brokerRelease(drain: boolean) {
|
||||
async function brokerRelease(drain: boolean, timeoutMs?: number) {
|
||||
return brokerRequest(
|
||||
`/tool-sessions/${encodeURIComponent(brokerIdentity.runtimeSessionId!)}:release`,
|
||||
drain ? { drain: true } : {},
|
||||
timeoutMs,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1795,6 +1921,498 @@ function bgStartedFile(): boolean {
|
||||
}
|
||||
}
|
||||
|
||||
// FG8c — process crashes. The Harness, the managed worker, and the
|
||||
// background process itself are killed (or stopped) for real, one at a
|
||||
// time, under a live background run; a Harness restart reloads the Session
|
||||
// cold from the store under a new writer identity. Crash Sessions run a
|
||||
// 10 s writer lease: a reload outlives the killed writer without a minute's
|
||||
// wait, the capture a crashed or closed Session leaves OPEN expires within
|
||||
// seconds — the fixture tenant admits four active captures, and seven cases
|
||||
// in a row under a 60 s lease exhausted it on the seventh — and a renewal,
|
||||
// due at half the lease, still has 5 s of slack on a loaded lane. Crash
|
||||
// Sessions talk to the store directly: a Harness killed mid-upload must not
|
||||
// leave the store proxy holding an aborted request.
|
||||
const CRASH_LEASE_MS = 10_000;
|
||||
|
||||
type RuntimeProbe = {
|
||||
bindings: Array<{
|
||||
binding_id: string;
|
||||
binding_state: string;
|
||||
runtime_generation: number;
|
||||
loss_evidence: number | boolean;
|
||||
stop_evidence: number | boolean;
|
||||
}>;
|
||||
slots: Array<{ last_generation: number; active_binding_id: string | null }>;
|
||||
runtimeSessions: Array<{ runtime_session_id: string; session_state: string }>;
|
||||
executions: Array<{
|
||||
execution_call_id: string;
|
||||
execution_state: string;
|
||||
execution_status: string | null;
|
||||
runtime_generation: number;
|
||||
runtime_session_id: string;
|
||||
loss_evidence_id: string | null;
|
||||
abandoned: number | boolean;
|
||||
result_json: string | null;
|
||||
}>;
|
||||
};
|
||||
|
||||
function procState(pid: number): string | undefined {
|
||||
try {
|
||||
const stat = readFileSync(`/proc/${pid}/stat`, 'utf8');
|
||||
return stat[stat.lastIndexOf(')') + 2];
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
const pidAlive = (pid: number) => {
|
||||
const state = procState(pid);
|
||||
return state !== undefined && state !== 'Z' && state !== 'X';
|
||||
};
|
||||
|
||||
function bgPid(): number {
|
||||
const pid = Number(
|
||||
readFileSync(path.join(current.directory, 'bg-started.txt'), 'utf8'),
|
||||
);
|
||||
assert(Number.isSafeInteger(pid) && pid > 0, `bg pid ${pid}`);
|
||||
return pid;
|
||||
}
|
||||
|
||||
async function crashSnapshot(label: string) {
|
||||
const state = await probe<ProbeState>(current.sessionId, 'state');
|
||||
const runtime = await probe<RuntimeProbe>(current.sessionId, 'runtime');
|
||||
const snapshot = {
|
||||
label,
|
||||
record: childRunOf(state) ?? null,
|
||||
monitor: state.monitors[0] ?? null,
|
||||
bindings: runtime.bindings,
|
||||
slots: runtime.slots,
|
||||
runtimeSessions: runtime.runtimeSessions,
|
||||
executions: runtime.executions,
|
||||
fullSegments: fullSegments(state).length,
|
||||
};
|
||||
console.log(`FG8_CRASH_STATE ${current.name} ${JSON.stringify(snapshot)}`);
|
||||
return snapshot;
|
||||
}
|
||||
|
||||
async function harnessJson(
|
||||
cli: HostedHarnessProcess,
|
||||
method: string,
|
||||
route: string,
|
||||
clientId?: string,
|
||||
body?: unknown,
|
||||
): Promise<{ status: number; body: Record<string, unknown> }> {
|
||||
const response = await cli.request(route, {
|
||||
method,
|
||||
headers: {
|
||||
...cli.headers(clientId),
|
||||
...(body === undefined ? {} : { 'Content-Type': 'application/json' }),
|
||||
},
|
||||
...(body === undefined ? {} : { body: JSON.stringify(body) }),
|
||||
signal: AbortSignal.timeout(90_000),
|
||||
});
|
||||
const text = await response.text();
|
||||
let parsed: Record<string, unknown> = {};
|
||||
try {
|
||||
parsed = text ? (JSON.parse(text) as Record<string, unknown>) : {};
|
||||
} catch {
|
||||
parsed = { text };
|
||||
}
|
||||
return { status: response.status, body: parsed };
|
||||
}
|
||||
|
||||
async function sigkillHarness(cli: HostedHarnessProcess) {
|
||||
const child = cli.child!;
|
||||
const pid = child.pid!;
|
||||
const exited = new Promise<[number | null, string | null]>((resolve) =>
|
||||
child.once('exit', (code, signal) => resolve([code, signal])),
|
||||
);
|
||||
assert(child.kill('SIGKILL'));
|
||||
const [code, signal] = await exited;
|
||||
assert.equal(code, null);
|
||||
assert.equal(signal, 'SIGKILL');
|
||||
step(`harness ${pid} SIGKILLed`);
|
||||
return { process: 'harness', pid, signal };
|
||||
}
|
||||
|
||||
// The case's own worker: the JVM's worker children diffed around the
|
||||
// case's first dispatch — exactly one is new.
|
||||
async function caseWorker(before: number[]): Promise<number> {
|
||||
const after = await probe<number[]>(current.sessionId, 'workers');
|
||||
const fresh = after.filter((pid) => !before.includes(pid));
|
||||
assert.equal(fresh.length, 1, `one new worker: ${before} -> ${after}`);
|
||||
return fresh[0];
|
||||
}
|
||||
|
||||
// The case's Session status as the Harness reports it.
|
||||
async function sessionStatus(cli: HostedHarnessProcess, clientId: string) {
|
||||
const status = await harnessJson(
|
||||
cli,
|
||||
'GET',
|
||||
`/session/${current.sessionId}/status`,
|
||||
clientId,
|
||||
);
|
||||
assert.equal(status.status, 200, JSON.stringify(status.body));
|
||||
return status.body;
|
||||
}
|
||||
|
||||
// Create the Session on a booted Harness and launch the case's background
|
||||
// run through a launch turn that settles while the run keeps its hold
|
||||
// (unless the case parks the :start reply, when `settle` is false). The run
|
||||
// is live once its supervised process announced itself and the `:process`
|
||||
// row is PREPARED; the turn has settled once the Session reports no active
|
||||
// prompt.
|
||||
async function launchCrashRun(cli: HostedHarnessProcess, settle: boolean) {
|
||||
const workersBefore = await probe<number[]>(current.sessionId, 'workers');
|
||||
const monitor = current.name === 'monitor-worker-kill';
|
||||
if (current.name !== 'harness-kill-capture') {
|
||||
execFileSync('mkfifo', [
|
||||
path.join(current.directory, monitor ? 'monitor-gate' : 'run-gate'),
|
||||
]);
|
||||
}
|
||||
const { clientId } = await createSession(
|
||||
cli,
|
||||
config.storeUrl,
|
||||
CRASH_LEASE_MS,
|
||||
);
|
||||
const promptId = await submitPrompt(
|
||||
cli,
|
||||
clientId,
|
||||
`fg8c ${current.name} launch`,
|
||||
);
|
||||
step('prompt admitted');
|
||||
await waitUntil(async () => {
|
||||
checkProxy();
|
||||
checkStoreProxy();
|
||||
if (!bgStartedFile()) return false;
|
||||
const state = await probe<ProbeState>(current.sessionId, 'state');
|
||||
return state.executions.some(
|
||||
(row) =>
|
||||
row.execution_call_id.endsWith(':process') &&
|
||||
row.execution_state === 'PREPARED',
|
||||
);
|
||||
}, 120_000);
|
||||
step('run live');
|
||||
if (settle) {
|
||||
await waitUntil(async () => {
|
||||
checkProxy();
|
||||
if ((await sessionStatus(cli, clientId))['hasActivePrompt'] !== false)
|
||||
return false;
|
||||
if (!monitor) return true;
|
||||
// The watch committed its observation before the case faults it.
|
||||
const watch = (await probe<ProbeState>(current.sessionId, 'state'))
|
||||
.monitors[0] as { observationSequence?: number } | undefined;
|
||||
return (watch?.observationSequence ?? 0) >= 1;
|
||||
}, 120_000);
|
||||
step('launch turn settled, run still live');
|
||||
}
|
||||
const worker = await caseWorker(workersBefore);
|
||||
return { clientId, promptId, worker };
|
||||
}
|
||||
|
||||
// The cold reload on a fresh Harness, under its own writer identity. The
|
||||
// killed writer's lease runs from its last renewal, which the store can
|
||||
// commit just after the kill, so the one 409 that heals on its own — the
|
||||
// writer conflict — is retried until the lease lapses; any other answer is
|
||||
// the reload's.
|
||||
async function reloadSession(cli: HostedHarnessProcess) {
|
||||
const deadline = Date.now() + 30_000;
|
||||
for (;;) {
|
||||
const loaded = await harnessJson(
|
||||
cli,
|
||||
'POST',
|
||||
`/session/${current.sessionId}/load`,
|
||||
undefined,
|
||||
sessionBody(cli, config.storeUrl, CRASH_LEASE_MS),
|
||||
);
|
||||
step(`reload -> ${loaded.status} ${JSON.stringify(loaded.body)}`);
|
||||
if (
|
||||
loaded.status !== 409 ||
|
||||
loaded.body['code'] !== 'managed_session_writer_conflict' ||
|
||||
Date.now() >= deadline
|
||||
) {
|
||||
return loaded;
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 500));
|
||||
}
|
||||
}
|
||||
|
||||
// DELETE /session — the close whose drain arm releases every hold the
|
||||
// Session knows of, live or restored.
|
||||
async function closeCrashSession(cli: HostedHarnessProcess, clientId: string) {
|
||||
const at = brokerAnswers.length;
|
||||
const closed = await harnessJson(
|
||||
cli,
|
||||
'DELETE',
|
||||
`/session/${current.sessionId}`,
|
||||
clientId,
|
||||
);
|
||||
step(`close -> ${closed.status}`);
|
||||
return {
|
||||
status: closed.status,
|
||||
releases: brokerAnswers
|
||||
.slice(at)
|
||||
.filter((answer) => answer.op === 'release')
|
||||
.map(({ runtimeSessionId, drain, status, code }) => ({
|
||||
runtimeSessionId,
|
||||
drain,
|
||||
status,
|
||||
code,
|
||||
})),
|
||||
};
|
||||
}
|
||||
|
||||
async function sigkillWorker(worker: number, signals: unknown[]) {
|
||||
process.kill(worker, 'SIGKILL');
|
||||
await waitUntil(() => !pidAlive(worker), 10_000);
|
||||
signals.push({ process: 'worker', pid: worker, signal: 'SIGKILL' });
|
||||
step(`worker ${worker} SIGKILLed`);
|
||||
}
|
||||
|
||||
async function runCrashCase() {
|
||||
const name = current.name;
|
||||
const report: Record<string, unknown> = { case: name };
|
||||
const signals: Array<Record<string, unknown>> = [];
|
||||
report['signals'] = signals;
|
||||
const harnesses: HostedHarnessProcess[] = [];
|
||||
let bg: number | undefined;
|
||||
let stoppedWorker: number | undefined;
|
||||
try {
|
||||
// harness-kill-dispatch parks the committed :start reply, so its launch
|
||||
// turn never learns the run started and never settles.
|
||||
if (name === 'harness-kill-dispatch') holdCommittedStart = true;
|
||||
const first = await bootHarness();
|
||||
harnesses.push(first);
|
||||
const launched = {
|
||||
cli: first,
|
||||
...(await launchCrashRun(first, name !== 'harness-kill-dispatch')),
|
||||
};
|
||||
bg = bgPid();
|
||||
report['promptId'] = launched.promptId;
|
||||
if (name === 'harness-kill-dispatch' || name === 'harness-kill-capture') {
|
||||
if (name === 'harness-kill-dispatch') {
|
||||
// The broker writes the row and starts the process before it
|
||||
// answers, so the parked reply can trail the live run.
|
||||
await waitUntil(() => committedStartHeld, 30_000);
|
||||
} else {
|
||||
// Mid-capture is a state: three full segments committed, the paced
|
||||
// producer still appending.
|
||||
await waitUntil(async () => {
|
||||
const state = await probe<ProbeState>(current.sessionId, 'state');
|
||||
return fullSegments(state).length >= 3;
|
||||
}, 60_000);
|
||||
}
|
||||
signals.push(await sigkillHarness(launched.cli));
|
||||
releaseCommittedStarts();
|
||||
report['committedStartDropped'] = committedStartHeld;
|
||||
// The killed writer's lease must lapse before another writer may
|
||||
// open (the reload retries a writer conflict until it has).
|
||||
await new Promise((resolve) => setTimeout(resolve, CRASH_LEASE_MS + 200));
|
||||
const at = brokerAnswers.length;
|
||||
const cli = await bootHarness();
|
||||
harnesses.push(cli);
|
||||
const loaded = await reloadSession(cli);
|
||||
// Once another writer opened, the dead one is fenced: any commit it
|
||||
// still had in flight has landed or been refused, so this count is
|
||||
// the capture's last.
|
||||
report['segmentsAtReload'] = (
|
||||
await crashSnapshot('after-reload')
|
||||
).fullSegments;
|
||||
report['reload'] = {
|
||||
status: loaded.status,
|
||||
code: loaded.body['code'] ?? null,
|
||||
reason: loaded.body['reason'] ?? null,
|
||||
};
|
||||
report['reloadOps'] = brokerAnswers
|
||||
.slice(at)
|
||||
.map(({ op, status, code }) => ({ op, status, code }));
|
||||
report['bgAliveAfterReload'] = pidAlive(bg);
|
||||
if (loaded.status === 200) {
|
||||
report['close'] = await closeCrashSession(
|
||||
cli,
|
||||
loaded.body['clientId'] as string,
|
||||
);
|
||||
// The restored drain's stop is the only thing that may end the run.
|
||||
await waitUntil(() => !pidAlive(bg!), 30_000).catch(() => undefined);
|
||||
report['bgAliveAfterClose'] = pidAlive(bg);
|
||||
}
|
||||
} else if (name === 'worker-kill' || name === 'generation-remint') {
|
||||
await sigkillWorker(launched.worker, signals);
|
||||
await crashSnapshot('after-kill');
|
||||
if (name === 'generation-remint') {
|
||||
// Two ordinary prompts after the loss, each reaching for the
|
||||
// Runtime with a foreground Shell: the first finds the dead worker,
|
||||
// the second the lost generation.
|
||||
const attempts = [];
|
||||
for (const attempt of [1, 2]) {
|
||||
const at = brokerAnswers.length;
|
||||
foregroundNext = true;
|
||||
await submitPrompt(
|
||||
launched.cli,
|
||||
launched.clientId,
|
||||
`fg8c ${name} post-crash ${attempt}`,
|
||||
);
|
||||
await waitUntil(
|
||||
async () =>
|
||||
(await sessionStatus(launched.cli, launched.clientId))[
|
||||
'hasActivePrompt'
|
||||
] === false,
|
||||
120_000,
|
||||
);
|
||||
attempts.push(
|
||||
brokerAnswers
|
||||
.slice(at)
|
||||
.map(({ op, status, code }) => ({ op, status, code })),
|
||||
);
|
||||
}
|
||||
report['postCrash'] = attempts;
|
||||
}
|
||||
report['close'] = await closeCrashSession(
|
||||
launched.cli,
|
||||
launched.clientId,
|
||||
);
|
||||
} else if (name === 'worker-stop') {
|
||||
process.kill(launched.worker, 'SIGSTOP');
|
||||
stoppedWorker = launched.worker;
|
||||
await waitUntil(() => procState(launched.worker) === 'T', 5_000);
|
||||
signals.push({
|
||||
process: 'worker',
|
||||
pid: launched.worker,
|
||||
signal: 'SIGSTOP',
|
||||
});
|
||||
step('worker SIGSTOPped');
|
||||
// The uncertainty window: a release sweep's status read against the
|
||||
// stopped worker runs into the broker's 30 s worker-request bound.
|
||||
const started = Date.now();
|
||||
const held = await brokerRelease(false, 90_000);
|
||||
report['stoppedRelease'] = {
|
||||
status: held.status,
|
||||
code: held.code,
|
||||
elapsedMs: Date.now() - started,
|
||||
};
|
||||
step(`routine release while stopped -> ${held.status} ${held.code}`);
|
||||
const during = await crashSnapshot('while-stopped');
|
||||
report['whileStopped'] = {
|
||||
recordState: during.record?.run?.state ?? null,
|
||||
stopReason: during.record?.stopReason ?? null,
|
||||
processRow: during.executions.find((row) =>
|
||||
row.execution_call_id.endsWith(':process'),
|
||||
)?.execution_state,
|
||||
bindingStates: during.bindings.map((row) => row.binding_state),
|
||||
};
|
||||
process.kill(launched.worker, 'SIGCONT');
|
||||
await waitUntil(() => procState(launched.worker) !== 'T', 5_000);
|
||||
stoppedWorker = undefined;
|
||||
signals.push({
|
||||
process: 'worker',
|
||||
pid: launched.worker,
|
||||
signal: 'SIGCONT',
|
||||
});
|
||||
step('worker SIGCONTed');
|
||||
await releaseFifo(current.directory, 'run-gate');
|
||||
await waitUntil(async () => {
|
||||
checkStoreProxy();
|
||||
const record = childRunOf(
|
||||
await probe<ProbeState>(current.sessionId, 'state'),
|
||||
);
|
||||
return record?.run?.state === 'settled';
|
||||
}, 120_000);
|
||||
step('resumed run settled');
|
||||
report['close'] = await closeCrashSession(
|
||||
launched.cli,
|
||||
launched.clientId,
|
||||
);
|
||||
} else if (name === 'process-kill') {
|
||||
process.kill(bg, 'SIGKILL');
|
||||
await waitUntil(() => !pidAlive(bg!), 10_000);
|
||||
signals.push({ process: 'background', pid: bg, signal: 'SIGKILL' });
|
||||
step('background process SIGKILLed');
|
||||
await waitUntil(async () => {
|
||||
checkStoreProxy();
|
||||
const record = childRunOf(
|
||||
await probe<ProbeState>(current.sessionId, 'state'),
|
||||
);
|
||||
return record?.run?.state === 'settled';
|
||||
}, 120_000);
|
||||
step('killed run settled');
|
||||
report['close'] = await closeCrashSession(
|
||||
launched.cli,
|
||||
launched.clientId,
|
||||
);
|
||||
} else {
|
||||
assert.equal(name, 'monitor-worker-kill');
|
||||
// The idle window must still be open at the kill, or the stop would
|
||||
// meet a live worker and settle honestly — a case that never faulted.
|
||||
const watch = (await probe<ProbeState>(current.sessionId, 'state'))
|
||||
.monitors[0] as
|
||||
| { run?: { state?: string }; stopReason?: string | null }
|
||||
| undefined;
|
||||
assert.equal(
|
||||
watch?.run?.state,
|
||||
'running',
|
||||
'the watch is live at the kill',
|
||||
);
|
||||
assert.equal(watch?.stopReason, null);
|
||||
await sigkillWorker(launched.worker, signals);
|
||||
// The watch is silent past its observation, so its idle window ends
|
||||
// it — and that auto-stop must ask the worker the Runtime lost.
|
||||
await waitUntil(async () => {
|
||||
const watch = (await probe<ProbeState>(current.sessionId, 'state'))
|
||||
.monitors[0] as { run?: { state?: string } } | undefined;
|
||||
return watch?.run?.state === 'recovery_blocked';
|
||||
}, 120_000).catch((cause) => {
|
||||
throw new Error(
|
||||
'the watch never parked after its idle stop met the lost worker',
|
||||
{ cause },
|
||||
);
|
||||
});
|
||||
step('watch parked');
|
||||
report['close'] = await closeCrashSession(
|
||||
launched.cli,
|
||||
launched.clientId,
|
||||
);
|
||||
}
|
||||
report['monitorStops'] = brokerAnswers
|
||||
.filter((answer) => answer.kind === 'monitor-stop')
|
||||
.map(({ status, code }) => ({ status, code }));
|
||||
report['starts'] = brokerAnswers.filter(
|
||||
(answer) => answer.op === 'start',
|
||||
).length;
|
||||
report['modelCalls'] = modelCalls;
|
||||
checkProxy();
|
||||
checkStoreProxy();
|
||||
return report;
|
||||
} catch (cause) {
|
||||
await diagnose(
|
||||
harnesses.at(-1) ?? new HostedHarnessProcess(),
|
||||
`modelCalls=${modelCalls} signals=${JSON.stringify(signals)}\n` +
|
||||
`answers=${JSON.stringify(brokerAnswers)}`,
|
||||
);
|
||||
throw cause;
|
||||
} finally {
|
||||
releaseCommittedStarts();
|
||||
if (stoppedWorker !== undefined) {
|
||||
try {
|
||||
process.kill(stoppedWorker, 'SIGCONT');
|
||||
} catch {
|
||||
// already gone
|
||||
}
|
||||
}
|
||||
for (const cli of harnesses) await cli.close();
|
||||
// The case owns its physical leftovers: a refused reload or a killed
|
||||
// worker leaves the background process running, and nothing in
|
||||
// production reaps it in this topology.
|
||||
if (bg === undefined && bgStartedFile()) bg = bgPid();
|
||||
if (bg !== undefined && pidAlive(bg)) {
|
||||
try {
|
||||
process.kill(bg, 'SIGKILL');
|
||||
} catch {
|
||||
// already gone
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
const reports = [];
|
||||
for (current of config.cases) {
|
||||
@@ -1824,12 +2442,18 @@ try {
|
||||
dropNextPagePublish = false;
|
||||
droppedPagePublishes = 0;
|
||||
dropCommitOps.clear();
|
||||
brokerAnswers.length = 0;
|
||||
holdCommittedStart = false;
|
||||
committedStartHeld = false;
|
||||
foregroundNext = false;
|
||||
if (config.letter === 'fg8a') {
|
||||
reports.push(await runReplyLossCase());
|
||||
} else if (config.letter === 'fg8b') {
|
||||
reports.push(await runStoreFailureCase());
|
||||
} else if (config.letter === 'fg8d') {
|
||||
reports.push(await runCancelDrainCase());
|
||||
} else if (config.letter === 'fg8c') {
|
||||
reports.push(await runCrashCase());
|
||||
} else if (config.letter === 'fg8f') {
|
||||
reports.push(await runPublishCase());
|
||||
} else {
|
||||
@@ -1850,7 +2474,9 @@ try {
|
||||
? 'FG8_STORE_FAILURE_OK'
|
||||
: config.letter === 'fg8d'
|
||||
? 'FG8_CANCEL_DRAIN_OK'
|
||||
: 'FG8_STREAM_GAPS_OK',
|
||||
: config.letter === 'fg8c'
|
||||
? 'FG8_PROCESS_CRASH_OK'
|
||||
: 'FG8_STREAM_GAPS_OK',
|
||||
);
|
||||
} catch (cause) {
|
||||
console.error('FG8 failed at', current?.name, cause);
|
||||
|
||||
@@ -197,3 +197,39 @@ fs.appendFileSync(${JSON.stringify(marker)}, fs.readFileSync(${JSON.stringify(jo
|
||||
expect(result.output[3]).toBe('joined\n');
|
||||
});
|
||||
});
|
||||
|
||||
describe('exit evidence', () => {
|
||||
const run = (script: string) => {
|
||||
const unit: HookCommandCgroup = Reflect.construct(HookCommandCgroup, [
|
||||
directory,
|
||||
]);
|
||||
const launch = unit.launch(process.execPath, ['--eval', script], {});
|
||||
return spawnSync(launch.executable, launch.args, {
|
||||
env: launch.env,
|
||||
stdio: ['ignore', 'pipe', 'pipe', 'pipe'],
|
||||
encoding: 'utf8',
|
||||
timeout: 30_000,
|
||||
});
|
||||
};
|
||||
|
||||
it('keeps the exit status of a command that exited', () => {
|
||||
const result = run('process.exit(3)');
|
||||
expect(result.error).toBeUndefined();
|
||||
expect(result.status).toBe(3);
|
||||
expect(result.signal).toBeNull();
|
||||
});
|
||||
|
||||
// The launcher's own end is the background supervisor's exit evidence: a
|
||||
// command a signal killed must not read as one that exited nonzero.
|
||||
it
|
||||
.skipIf(process.platform === 'win32')
|
||||
.each(['SIGKILL', 'SIGTERM', 'SIGUSR2'])(
|
||||
'ends by the %s that ended its command',
|
||||
(signal) => {
|
||||
const result = run(`process.kill(process.pid, '${signal}')`);
|
||||
expect(result.error).toBeUndefined();
|
||||
expect(result.status).toBeNull();
|
||||
expect(result.signal).toBe(signal);
|
||||
},
|
||||
);
|
||||
});
|
||||
@@ -62,7 +62,24 @@ const child = spawn(executable, JSON.parse(args), {
|
||||
stdio: ['inherit', 'inherit', 'inherit'],
|
||||
});
|
||||
child.on('error', () => process.exit(1));
|
||||
child.on('exit', (code) => process.exit(code ?? 1));
|
||||
// A command a signal ended must end this launcher by the same signal: the
|
||||
// launcher's exit is its supervisor's only evidence, and the nonzero status a
|
||||
// plain exit substitutes is one the command never returned. A core-dumping
|
||||
// signal would dump this launcher too, and Node.js owns SIGUSR1 and ignores
|
||||
// SIGPIPE, so those keep the nonzero fallback. Delivery to self is
|
||||
// synchronous on POSIX; the timer only covers a signal that somehow left the
|
||||
// process running. Only names Node.js reports on Linux are listed: it calls
|
||||
// signal 6 SIGABRT, never SIGIOT, and Linux has no SIGEMT.
|
||||
const KEPT = ['SIGABRT', 'SIGBUS', 'SIGFPE', 'SIGILL', 'SIGQUIT', 'SIGSEGV',
|
||||
'SIGSYS', 'SIGTRAP', 'SIGXCPU', 'SIGXFSZ', 'SIGUSR1', 'SIGPIPE'];
|
||||
child.on('exit', (code, signal) => {
|
||||
if (signal !== null && !KEPT.includes(signal)) {
|
||||
process.kill(process.pid, signal);
|
||||
setTimeout(() => process.exit(1), 1000);
|
||||
return;
|
||||
}
|
||||
process.exit(code ?? 1);
|
||||
});
|
||||
`;
|
||||
|
||||
export class HookCommandCgroup {
|
||||
|
||||
@@ -358,7 +358,10 @@
|
||||
measuring 308–314 s healthy locally on
|
||||
JDK 21 + MySQL 8.4 (~534 s budgeted at
|
||||
the same 1.7×), which the same fork
|
||||
still covers. -->
|
||||
still covers. The FG8c process-crash
|
||||
letter (seven cases, 144 s healthy
|
||||
locally on a fresh MySQL 8.4.6) rides
|
||||
inside the same envelope. -->
|
||||
<forkedProcessTimeoutInSeconds>3600</forkedProcessTimeoutInSeconds>
|
||||
</configuration>
|
||||
<executions>
|
||||
@@ -441,8 +444,9 @@
|
||||
<configuration>
|
||||
<includes><include>**/HostedMonitorRun*IT.java</include></includes>
|
||||
<failIfNoTests>true</failIfNoTests>
|
||||
<!-- The class's five letters (fg8e/fg8f/fg8a/fg8b/fg8d)
|
||||
are independent fixtures sharing one fork: each
|
||||
<!-- The class's six letters (fg8e/fg8f/fg8a/fg8b/fg8d
|
||||
and fg8c) are independent fixtures sharing one
|
||||
fork: each
|
||||
letter's @Timeout ceiling (600 s) plus the healthy
|
||||
runtime must fit, the per-ceiling relation the CI
|
||||
profile pins. Fork measured on the lane (run
|
||||
|
||||
+368
-15
@@ -4,6 +4,7 @@ import com.alibaba.qwen.code.managedagent.api.WorkspaceSelection;
|
||||
import com.alibaba.qwen.code.managedagent.service.EmbeddedRuntimeBroker;
|
||||
import com.alibaba.qwen.code.managedagent.service.HarnessEventProjector;
|
||||
import com.alibaba.qwen.code.managedagent.store.ManagedAgentStore;
|
||||
import com.alibaba.qwen.code.managedagent.store.ManagedSessionStore;
|
||||
import com.alibaba.qwen.code.runtimebroker.JdbcRuntimeBindingRepository;
|
||||
import com.alibaba.qwen.code.runtimebroker.WorkspaceExecutionProfile;
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
@@ -42,9 +43,10 @@ import static org.assertj.core.api.Assertions.assertThat;
|
||||
// original identity); FG8b gates store failures (a rejected commit inside
|
||||
// a real transaction leaves zero OS effects); FG8d gates cancellation and
|
||||
// the routine/drain release split; FG8e gates SSE continuity across
|
||||
// abort/resume; FG8f gates output delivery under publisher loss. Real
|
||||
// packaged Harness, real MySQL, JDBC ground truth; no production code
|
||||
// changes. FG8c lands with #13986 per the issue's sequencing constraint.
|
||||
// abort/resume; FG8f gates output delivery under publisher loss; FG8c
|
||||
// gates process crashes — the Harness, the managed worker, or the
|
||||
// background process killed (or stopped) for real under a live run. Real
|
||||
// packaged Harness, real MySQL, JDBC ground truth.
|
||||
class HostedMonitorRunFaultGatesIT {
|
||||
private final ObjectMapper json = new ObjectMapper();
|
||||
|
||||
@@ -113,6 +115,15 @@ class HostedMonitorRunFaultGatesIT {
|
||||
"drain-release"));
|
||||
}
|
||||
|
||||
@Test
|
||||
@Timeout(600)
|
||||
void fg8cProcessCrashesNeverDuplicateOrFabricateOnMySql()
|
||||
throws Exception {
|
||||
runLetter("fg8c", List.of("harness-kill-dispatch",
|
||||
"harness-kill-capture", "worker-kill", "worker-stop",
|
||||
"process-kill", "generation-remint", "monitor-worker-kill"));
|
||||
}
|
||||
|
||||
/** Trigger rows this letter installed for FG8b (dropped after it). */
|
||||
private List<String> fg8bTriggers;
|
||||
|
||||
@@ -345,7 +356,9 @@ class HostedMonitorRunFaultGatesIT {
|
||||
? "FG8_STORE_FAILURE_OK"
|
||||
: letter.equals("fg8d")
|
||||
? "FG8_CANCEL_DRAIN_OK"
|
||||
: "FG8_STREAM_GAPS_OK");
|
||||
: letter.equals("fg8c")
|
||||
? "FG8_PROCESS_CRASH_OK"
|
||||
: "FG8_STREAM_GAPS_OK");
|
||||
JsonNode reports = json.readTree(
|
||||
Files.readString(resultFile));
|
||||
assertThat(reports).hasSize(cases.size());
|
||||
@@ -447,6 +460,334 @@ class HostedMonitorRunFaultGatesIT {
|
||||
jdbc.execute("DROP TABLE IF EXISTS fg8b_fire");
|
||||
}
|
||||
|
||||
// FG8c per-case ground truth: the signal the driver delivered (the fault
|
||||
// fired), the execution ledger's one `:process` row, the Session's
|
||||
// Runtime binding and slot (one generation, never re-minted), and the
|
||||
// capability record — which may stay open, but never settles on
|
||||
// evidence nobody produced.
|
||||
private void assertCrashCase(JdbcTemplate jdbc, String tenant,
|
||||
Map<String, Object> session, JsonNode report) throws Exception {
|
||||
String sessionId = session.get("sessionId").toString();
|
||||
String name = session.get("name").toString();
|
||||
String promptId = report.path("promptId").asText();
|
||||
assertThat(report.path("case").asText()).isEqualTo(name);
|
||||
JsonNode signal = report.path("signals").path(0);
|
||||
String fault = switch (name) {
|
||||
case "harness-kill-dispatch", "harness-kill-capture" ->
|
||||
"harness SIGKILL";
|
||||
case "process-kill" -> "background SIGKILL";
|
||||
case "worker-stop" -> "worker SIGSTOP";
|
||||
default -> "worker SIGKILL";
|
||||
};
|
||||
assertThat(signal.path("process").asText() + " "
|
||||
+ signal.path("signal").asText()).as(name + " fault fired")
|
||||
.isEqualTo(fault);
|
||||
// No duplicate dispatch across every boot, reload and later prompt.
|
||||
assertThat(report.path("starts").asInt()).as(name + " one :start")
|
||||
.isEqualTo(1);
|
||||
assertProcessRows(jdbc, sessionId, name, 1);
|
||||
Map<String, Object> process = jdbc.queryForMap(
|
||||
"SELECT execution_state, execution_status,"
|
||||
+ " CAST(result_json AS CHAR) AS result_json,"
|
||||
+ " CAST(loss_evidence_id AS CHAR) AS loss_evidence_id,"
|
||||
+ " abandoned_at FROM ma_tool_execution WHERE"
|
||||
+ " harness_session_id = ? AND execution_call_id"
|
||||
+ " LIKE '%:process'", sessionId);
|
||||
JsonNode evidence = process.get("result_json") == null
|
||||
? json.missingNode()
|
||||
: json.readTree(process.get("result_json").toString())
|
||||
.path("evidence");
|
||||
var bindings = jdbc.queryForList(
|
||||
"SELECT CAST(binding_id AS CHAR) AS binding_id, request_key,"
|
||||
+ " binding_state, runtime_generation,"
|
||||
+ " stop_evidence_json IS NOT NULL AS stopped,"
|
||||
+ " loss_evidence_json IS NOT NULL AS lost FROM"
|
||||
+ " ma_runtime_binding WHERE tenant_id = ? AND"
|
||||
+ " isolation_key = ?", tenant, sessionId);
|
||||
// A dead generation is never silently re-minted: whatever the case
|
||||
// killed, the Session holds exactly the one binding it began with.
|
||||
assertThat(bindings).as(name + " one binding").hasSize(1);
|
||||
var binding = bindings.getFirst();
|
||||
assertThat(((Number) binding.get("runtime_generation")).longValue())
|
||||
.as(name + " generation").isEqualTo(1);
|
||||
var slot = jdbc.queryForMap("SELECT last_generation,"
|
||||
+ " CAST(active_binding_id AS CHAR) AS active_binding_id"
|
||||
+ " FROM ma_runtime_slot WHERE request_key = ?",
|
||||
binding.get("request_key"));
|
||||
assertThat(((Number) slot.get("last_generation")).longValue())
|
||||
.as(name + " slot generation").isEqualTo(1);
|
||||
assertThat(slot.get("active_binding_id")).as(name + " slot binding")
|
||||
.isEqualTo(binding.get("binding_id"));
|
||||
String runtimeSessionState = jdbc.queryForObject(
|
||||
"SELECT session_state FROM ma_runtime_session WHERE"
|
||||
+ " tenant_id = ? AND harness_session_id = ? AND"
|
||||
+ " runtime_session_id = ?", String.class, tenant,
|
||||
sessionId, promptId);
|
||||
boolean monitor = name.equals("monitor-worker-kill");
|
||||
JsonNode record = extensionRecord(jdbc, tenant, sessionId,
|
||||
monitor ? "monitor_run" : "child_run");
|
||||
JsonNode close = report.path("close");
|
||||
switch (name) {
|
||||
case "harness-kill-dispatch" -> {
|
||||
assertThat(report.path("committedStartDropped").asBoolean())
|
||||
.as(name + " the committed :start reply was lost")
|
||||
.isTrue();
|
||||
// Cold load refuses the unsettled launch turn, by its tag.
|
||||
assertThat(report.path("reload").path("status").asInt())
|
||||
.isEqualTo(409);
|
||||
assertThat(report.path("reload").path("code").asText())
|
||||
.isEqualTo("hosted_turn_recovery_required");
|
||||
assertThat(report.path("reload").path("reason").asText())
|
||||
.isEqualTo("turn_unsettled");
|
||||
// The refusal dispatched nothing: its broker traffic is
|
||||
// reads only (the never-started probe's status read), and
|
||||
// the model was never asked to plan again.
|
||||
assertThat(report.path("reloadOps").findValuesAsText("op"))
|
||||
.as(name + " cold load broker traffic")
|
||||
.containsOnly("status");
|
||||
assertThat(report.path("modelCalls").asInt()).isEqualTo(1);
|
||||
assertThat(report.path("bgAliveAfterReload").asBoolean())
|
||||
.as(name + " the refusal left the run alone").isTrue();
|
||||
// The outcome stays unknown: no terminal event, no settle.
|
||||
assertThat(record.path("run").path("state").asText())
|
||||
.isEqualTo("running");
|
||||
assertThat(record.path("run").path("execution").asText())
|
||||
.isEqualTo("dispatch_started");
|
||||
assertThat(record.path("stopReason").isNull()).isTrue();
|
||||
assertThat(process.get("execution_state"))
|
||||
.isEqualTo("PREPARED");
|
||||
assertThat(journalKinds(jdbc, tenant, sessionId))
|
||||
.as(name + " no terminal event")
|
||||
.doesNotContain("turn.settled");
|
||||
}
|
||||
case "harness-kill-capture" -> {
|
||||
assertThat(report.path("reload").path("status").asInt())
|
||||
.as(name + " a settled launch reloads").isEqualTo(200);
|
||||
// The reload restored the held Runtime and dispatched nothing.
|
||||
assertThat(report.path("reloadOps").findValuesAsText("op"))
|
||||
.containsExactly("active-holds");
|
||||
assertThat(report.path("bgAliveAfterReload").asBoolean())
|
||||
.isTrue();
|
||||
// The restored obligation: the reloaded close drained the
|
||||
// run its process never saw, and that drain is what ended it.
|
||||
assertThat(close.path("status").asInt()).isEqualTo(204);
|
||||
assertThat(close.path("releases")).hasSize(1);
|
||||
JsonNode release = close.path("releases").path(0);
|
||||
assertThat(release.path("runtimeSessionId").asText())
|
||||
.isEqualTo(promptId);
|
||||
assertThat(release.path("drain").asBoolean()).isTrue();
|
||||
assertThat(release.path("status").asInt()).isEqualTo(200);
|
||||
assertThat(report.path("bgAliveAfterClose").asBoolean())
|
||||
.as(name + " the restored drain stopped the run")
|
||||
.isFalse();
|
||||
assertThat(process.get("execution_state"))
|
||||
.isEqualTo("SETTLED");
|
||||
assertThat(process.get("execution_status"))
|
||||
.isEqualTo("error");
|
||||
assertThat(evidence.path("exitCode").isNull()).isTrue();
|
||||
assertThat(evidence.path("exitSignal").asText())
|
||||
.as(name + " ended by the platform's stop")
|
||||
.isIn("SIGTERM", "SIGKILL");
|
||||
assertThat(runtimeSessionState).isEqualTo("RELEASED");
|
||||
// Nobody observed that end on the record's side: it stays
|
||||
// open rather than settling on evidence it never received.
|
||||
assertThat(record.path("run").path("state").asText())
|
||||
.isEqualTo("running");
|
||||
assertThat(record.path("stopReason").isNull()).isTrue();
|
||||
// The durable prefix stays byte-identical and readable, and
|
||||
// nothing lands once the dead writer is fenced: the count
|
||||
// taken after the reload opened a new writer is the last.
|
||||
assertThat(report.path("segmentsAtReload").asInt())
|
||||
.as(name + " killed mid-capture")
|
||||
.isGreaterThanOrEqualTo(3);
|
||||
verifyOutputChain(jdbc, tenant, sessionId, false);
|
||||
assertThat(loadOutputRows(jdbc, tenant, sessionId).byId()
|
||||
.values().stream().filter(row -> row.get("kind")
|
||||
.equals("managed-tool-result-content")
|
||||
&& ((Number) row.get("byte_length"))
|
||||
.intValue() == SEGMENT_BYTES)
|
||||
.count())
|
||||
.as(name + " no segment once the dead writer fenced")
|
||||
.isEqualTo(report.path("segmentsAtReload").asLong());
|
||||
}
|
||||
case "worker-kill", "generation-remint", "monitor-worker-kill" -> {
|
||||
// The close's drain cannot prove the release over a dead
|
||||
// worker: the broker defers it, the Session still retires.
|
||||
assertThat(close.path("status").asInt()).isEqualTo(204);
|
||||
assertThat(close.path("releases")).hasSize(1);
|
||||
JsonNode release = close.path("releases").path(0);
|
||||
assertThat(release.path("runtimeSessionId").asText())
|
||||
.isEqualTo(promptId);
|
||||
assertThat(release.path("drain").asBoolean()).isTrue();
|
||||
assertThat(release.path("status").asInt()).isEqualTo(503);
|
||||
assertThat(release.path("code").asText())
|
||||
.isEqualTo("runtime_reconciliation_required");
|
||||
// The hold is never released without evidence: the binding
|
||||
// is lost with no stop proof, its slot stays pinned, and the
|
||||
// Runtime Session is not released.
|
||||
assertThat(binding.get("binding_state")).isEqualTo("LOST");
|
||||
assertThat(((Number) binding.get("stopped")).intValue())
|
||||
.as(name + " no stop evidence").isZero();
|
||||
assertThat(runtimeSessionState).isNotEqualTo("RELEASED");
|
||||
if (name.equals("generation-remint")) {
|
||||
// Each later reach for the Runtime meets the loss, first
|
||||
// as the dead worker, then as the lost generation; the
|
||||
// foreground Shells never ran anywhere.
|
||||
assertThat(refusals(report.path("postCrash").path(0)))
|
||||
.containsExactly(
|
||||
"warm 503 runtime_provision_failed");
|
||||
assertThat(refusals(report.path("postCrash").path(1)))
|
||||
.containsExactly(
|
||||
"warm 503 runtime_broker_runtime_lost");
|
||||
assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM"
|
||||
+ " ma_tool_execution WHERE harness_session_id"
|
||||
+ " = ?", Integer.class, sessionId))
|
||||
.as(name + " nothing ran on the lost generation")
|
||||
.isEqualTo(2);
|
||||
assertThat(((Number) binding.get("lost")).intValue())
|
||||
.as(name + " loss evidence").isOne();
|
||||
// The lost generation's unsettled hold is abandoned
|
||||
// against the loss — never settled as an exit.
|
||||
assertThat(process.get("execution_state"))
|
||||
.isEqualTo("ABANDONED");
|
||||
assertThat(process.get("abandoned_at")).isNotNull();
|
||||
assertThat(process.get("loss_evidence_id"))
|
||||
.isEqualTo(binding.get("binding_id")
|
||||
+ ":1:journal-lost");
|
||||
assertThat(process.get("result_json")).isNull();
|
||||
} else {
|
||||
assertThat(process.get("execution_state"))
|
||||
.isEqualTo("PREPARED");
|
||||
}
|
||||
if (monitor) {
|
||||
// The idle stop asked the worker the Runtime lost; the
|
||||
// refusal parks the watch instead of settling a stop
|
||||
// the physical side never confirmed (#13896).
|
||||
assertThat(report.path("monitorStops")).isNotEmpty();
|
||||
for (JsonNode stop : report.path("monitorStops")) {
|
||||
assertThat(stop.path("status").asInt())
|
||||
.as(name + " monitor-stop refused")
|
||||
.isGreaterThanOrEqualTo(400);
|
||||
}
|
||||
assertThat(record.path("run").path("state").asText())
|
||||
.isEqualTo("recovery_blocked");
|
||||
assertThat(record.path("run").path("reason").asText())
|
||||
.isEqualTo("runtime_lost");
|
||||
assertThat(record.path("run").path("execution").asText())
|
||||
.isEqualTo("outcome_unknown");
|
||||
assertThat(record.path("stopReason").isNull()).isTrue();
|
||||
assertThat(record.path("observationSequence").asInt())
|
||||
.isPositive();
|
||||
assertThat(record.path("notifiedThrough").asInt())
|
||||
.isEqualTo(record.path("observationSequence")
|
||||
.asInt());
|
||||
} else {
|
||||
assertThat(record.path("run").path("state").asText())
|
||||
.as(name + " no exit nobody observed")
|
||||
.isEqualTo("running");
|
||||
assertThat(record.path("stopReason").isNull()).isTrue();
|
||||
}
|
||||
}
|
||||
case "worker-stop" -> {
|
||||
assertThat(report.path("signals").path(1).path("signal")
|
||||
.asText()).isEqualTo("SIGCONT");
|
||||
// The routine sweep's status read waited out the stopped
|
||||
// worker and still refused as held: an unreachable worker is
|
||||
// neither a stopped run nor a lost Runtime.
|
||||
JsonNode held = report.path("stoppedRelease");
|
||||
assertThat(held.path("status").asInt()).isEqualTo(409);
|
||||
assertThat(held.path("code").asText())
|
||||
.isEqualTo("runtime_session_held");
|
||||
assertThat(held.path("elapsedMs").asLong())
|
||||
.as(name + " the uncertainty window passed")
|
||||
.isGreaterThanOrEqualTo(25_000);
|
||||
JsonNode during = report.path("whileStopped");
|
||||
assertThat(during.path("recordState").asText())
|
||||
.isEqualTo("running");
|
||||
assertThat(during.path("stopReason").isNull()).isTrue();
|
||||
assertThat(during.path("processRow").asText())
|
||||
.isEqualTo("PREPARED");
|
||||
assertThat(during.path("bindingStates")).hasSize(1);
|
||||
assertThat(during.path("bindingStates").path(0).asText())
|
||||
.as(name + " a stopped worker is not a lost Runtime")
|
||||
.isEqualTo("READY");
|
||||
// The resumed worker's own evidence settled the run.
|
||||
assertThat(record.path("run").path("state").asText())
|
||||
.isEqualTo("settled");
|
||||
assertThat(record.path("stopReason").asText())
|
||||
.isEqualTo("exited");
|
||||
assertThat(record.path("exitCode").isInt()).isTrue();
|
||||
assertThat(record.path("exitCode").asInt()).isZero();
|
||||
assertThat(process.get("execution_state"))
|
||||
.isEqualTo("SETTLED");
|
||||
assertThat(process.get("execution_status"))
|
||||
.isEqualTo("success");
|
||||
assertThat(evidence.path("exitCode").isInt()).isTrue();
|
||||
assertThat(evidence.path("exitCode").asInt()).isZero();
|
||||
assertThat(binding.get("binding_state")).isEqualTo("READY");
|
||||
assertThat(close.path("status").asInt()).isEqualTo(204);
|
||||
}
|
||||
case "process-kill" -> {
|
||||
// The record carries the real end: the signal, not a status
|
||||
// the process never returned.
|
||||
assertThat(record.path("run").path("state").asText())
|
||||
.isEqualTo("settled");
|
||||
assertThat(record.path("stopReason").asText())
|
||||
.isEqualTo("exited");
|
||||
assertThat(record.path("exitSignal").asText())
|
||||
.as(name + " the kill is the evidence")
|
||||
.isEqualTo("SIGKILL");
|
||||
assertThat(record.path("exitCode").isNull()).isTrue();
|
||||
assertThat(record.path("stopRequested").asBoolean())
|
||||
.isFalse();
|
||||
assertThat(process.get("execution_state"))
|
||||
.isEqualTo("SETTLED");
|
||||
assertThat(process.get("execution_status"))
|
||||
.isEqualTo("error");
|
||||
assertThat(evidence.path("exitSignal").asText())
|
||||
.isEqualTo("SIGKILL");
|
||||
assertThat(evidence.path("exitCode").isNull()).isTrue();
|
||||
assertThat(binding.get("binding_state")).isEqualTo("READY");
|
||||
assertThat(close.path("status").asInt()).isEqualTo(204);
|
||||
}
|
||||
default -> throw new IllegalStateException(name);
|
||||
}
|
||||
}
|
||||
|
||||
// One prompt's refused broker answers, as "operation status code".
|
||||
private static List<String> refusals(JsonNode answers) {
|
||||
var refused = new ArrayList<String>();
|
||||
for (JsonNode answer : answers) {
|
||||
if (answer.path("status").asInt() >= 400) {
|
||||
refused.add(answer.path("op").asText() + " "
|
||||
+ answer.path("status").asInt() + " "
|
||||
+ answer.path("code").asText());
|
||||
}
|
||||
}
|
||||
return refused;
|
||||
}
|
||||
|
||||
// The managed-session event kinds the Session's journal committed.
|
||||
private List<String> journalKinds(JdbcTemplate jdbc, String tenant,
|
||||
String sessionId) throws Exception {
|
||||
var kinds = new ArrayList<String>();
|
||||
for (var row : jdbc.queryForList("SELECT record_bytes FROM"
|
||||
+ " ma_session_journal_tx WHERE tenant_id = ? AND session_id"
|
||||
+ " = ? ORDER BY journal_revision", tenant, sessionId)) {
|
||||
for (String line : new String((byte[]) row.get("record_bytes"),
|
||||
StandardCharsets.UTF_8).lines().toList()) {
|
||||
JsonNode entry = json.readTree(line);
|
||||
if (entry.path("subtype").asText()
|
||||
.equals("managed_session_event_v1")) {
|
||||
kinds.add(entry.path("managedSession").path("kind")
|
||||
.asText());
|
||||
}
|
||||
}
|
||||
}
|
||||
assertThat(kinds).as("journal events").isNotEmpty();
|
||||
return kinds;
|
||||
}
|
||||
|
||||
private void assertCase(String letter, JdbcTemplate jdbc, String tenant,
|
||||
Map<String, Object> session, JsonNode report) throws Exception {
|
||||
switch (letter) {
|
||||
@@ -455,6 +796,7 @@ class HostedMonitorRunFaultGatesIT {
|
||||
report);
|
||||
case "fg8d" -> assertCancelDrainCase(jdbc, tenant, session,
|
||||
report);
|
||||
case "fg8c" -> assertCrashCase(jdbc, tenant, session, report);
|
||||
default -> assertStreamPublishCase(letter, jdbc, tenant, session,
|
||||
report);
|
||||
}
|
||||
@@ -892,20 +1234,24 @@ class HostedMonitorRunFaultGatesIT {
|
||||
var byId = new HashMap<String, Map<String, Object>>();
|
||||
for (var row : jdbc.queryForList(
|
||||
"SELECT resource_id, kind, byte_length, sha256 FROM"
|
||||
+ " ma_resource WHERE tenant_id = ?"
|
||||
+ " AND session_id = ? AND kind LIKE"
|
||||
+ " 'managed-tool-result-%'", tenant, sessionId)) {
|
||||
+ " ma_resource WHERE session_scope_key = ? AND"
|
||||
+ " tenant_id = ? AND session_id = ? AND kind LIKE"
|
||||
+ " 'managed-tool-result-%'",
|
||||
ManagedSessionStore.sessionScopeKey(tenant, sessionId),
|
||||
tenant, sessionId)) {
|
||||
byId.put(row.get("resource_id").toString(), row);
|
||||
}
|
||||
var bodies = new HashMap<String, JsonNode>();
|
||||
for (var row : jdbc.queryForList(
|
||||
"SELECT resource_id, CONVERT(inline_bytes USING utf8mb4) AS"
|
||||
+ " body FROM ma_resource WHERE"
|
||||
+ " tenant_id = ? AND session_id = ? AND (kind IN"
|
||||
+ " body FROM ma_resource WHERE session_scope_key = ?"
|
||||
+ " AND tenant_id = ? AND session_id = ? AND (kind IN"
|
||||
+ " ('managed-tool-result-manifest',"
|
||||
+ " 'managed-tool-result-page') OR (kind ="
|
||||
+ " 'managed-tool-result-content' AND byte_length <"
|
||||
+ " 1024))", tenant, sessionId)) {
|
||||
+ " 1024))",
|
||||
ManagedSessionStore.sessionScopeKey(tenant, sessionId),
|
||||
tenant, sessionId)) {
|
||||
String body = row.get("body").toString();
|
||||
// The producers' short tail segment is raw bytes, not JSON;
|
||||
// only envelopes (manifest, page, seal) are parsed here — the
|
||||
@@ -1142,19 +1488,26 @@ class HostedMonitorRunFaultGatesIT {
|
||||
}
|
||||
}
|
||||
|
||||
// ma_resource keys on the Session's scope key and holds the inline
|
||||
// capture bytes: filtered by the unindexed tenant and session columns
|
||||
// alone, every read scans all the bytes the lane has written so far.
|
||||
// The record read leads with the same key.
|
||||
private JsonNode extensionRecord(JdbcTemplate jdbc, String tenant,
|
||||
String sessionId, String domain) throws Exception {
|
||||
String scope = ManagedSessionStore.sessionScopeKey(tenant, sessionId);
|
||||
var rows = jdbc.queryForList(
|
||||
"SELECT record_resource_id FROM"
|
||||
+ " ma_session_record WHERE"
|
||||
+ " ma_session_record WHERE session_scope_key = ? AND"
|
||||
+ " tenant_id = ? AND session_id = ? AND domain = ?",
|
||||
tenant, sessionId, domain);
|
||||
scope, tenant, sessionId, domain);
|
||||
assertThat(rows).as(domain + " records").hasSize(1);
|
||||
var resources = jdbc.queryForList(
|
||||
"SELECT CONVERT(inline_bytes USING utf8mb4) AS body FROM"
|
||||
+ " ma_resource WHERE tenant_id = ?"
|
||||
+ " AND session_id = ? AND resource_id = ?", tenant,
|
||||
sessionId, rows.getFirst().get("record_resource_id"));
|
||||
+ " ma_resource WHERE session_scope_key = ? AND"
|
||||
+ " resource_id = ? AND tenant_id = ? AND session_id"
|
||||
+ " = ?", scope,
|
||||
rows.getFirst().get("record_resource_id"), tenant,
|
||||
sessionId);
|
||||
assertThat(resources).hasSize(1);
|
||||
return json.readTree(resources.getFirst().get("body").toString());
|
||||
}
|
||||
|
||||
+82
-13
@@ -128,6 +128,8 @@ final class HostedMonitorRunProbe implements AutoCloseable {
|
||||
}
|
||||
case "state" -> state(sessionId);
|
||||
case "ledger" -> ledger(sessionId);
|
||||
case "workers" -> workers();
|
||||
case "runtime" -> runtime(sessionId);
|
||||
default -> throw new IllegalArgumentException(route[3]);
|
||||
};
|
||||
byte[] bytes = JSON.writeValueAsBytes(response);
|
||||
@@ -215,32 +217,43 @@ final class HostedMonitorRunProbe implements AutoCloseable {
|
||||
// arms the store-proxy faults.
|
||||
"resources", jdbc.queryForList(
|
||||
"SELECT resource_id, kind, byte_length, sha256 FROM"
|
||||
+ " ma_resource WHERE"
|
||||
+ " tenant_id = ? AND session_id = ? AND kind"
|
||||
+ " LIKE 'managed-tool-result-%' ORDER BY"
|
||||
+ " gmt_create, resource_id", tenant,
|
||||
sessionId),
|
||||
+ " ma_resource WHERE session_scope_key = ?"
|
||||
+ " AND tenant_id = ? AND session_id = ? AND"
|
||||
+ " kind LIKE 'managed-tool-result-%' ORDER"
|
||||
+ " BY gmt_create, resource_id",
|
||||
scopeKey(sessionId), tenant, sessionId),
|
||||
"relayAlive", relay != null && !relay.isDone(),
|
||||
"relayFailure", relayFailure == null ? ""
|
||||
: relayFailure.toString());
|
||||
}
|
||||
|
||||
// ma_resource keys on the Session's scope key and holds the inline
|
||||
// capture bytes, and the drivers poll these reads: filtered by the
|
||||
// unindexed tenant and session columns alone, each poll scans all the
|
||||
// bytes the lane has written so far (FG8f alone inlines tens of MiB a
|
||||
// case), so each later letter polls slower. The record read leads with
|
||||
// the same key.
|
||||
private String scopeKey(String sessionId) {
|
||||
return com.alibaba.qwen.code.managedagent.store.ManagedSessionStore
|
||||
.sessionScopeKey(tenant, sessionId);
|
||||
}
|
||||
|
||||
private List<JsonNode> extensionRecords(String sessionId, String domain) {
|
||||
try {
|
||||
var rows = jdbc.queryForList(
|
||||
"SELECT record_id, record_resource_id FROM"
|
||||
+ " ma_session_record WHERE"
|
||||
+ " tenant_id = ? AND session_id = ? AND domain"
|
||||
+ " = ? ORDER BY gmt_create, record_id", tenant,
|
||||
sessionId, domain);
|
||||
+ " ma_session_record WHERE session_scope_key = ?"
|
||||
+ " AND tenant_id = ? AND session_id = ? AND"
|
||||
+ " domain = ? ORDER BY gmt_create, record_id",
|
||||
scopeKey(sessionId), tenant, sessionId, domain);
|
||||
var records = new java.util.ArrayList<JsonNode>();
|
||||
for (var row : rows) {
|
||||
var resources = jdbc.queryForList(
|
||||
"SELECT CONVERT(inline_bytes USING utf8mb4) AS body"
|
||||
+ " FROM ma_resource WHERE"
|
||||
+ " tenant_id = ? AND session_id = ? AND"
|
||||
+ " resource_id = ?", tenant, sessionId,
|
||||
row.get("record_resource_id"));
|
||||
+ " FROM ma_resource WHERE session_scope_key"
|
||||
+ " = ? AND resource_id = ? AND tenant_id = ?"
|
||||
+ " AND session_id = ?", scopeKey(sessionId),
|
||||
row.get("record_resource_id"), tenant, sessionId);
|
||||
assertThat(resources).as(domain + " record body").hasSize(1);
|
||||
records.add(JSON.readTree(
|
||||
resources.getFirst().get("body").toString()));
|
||||
@@ -251,6 +264,62 @@ final class HostedMonitorRunProbe implements AutoCloseable {
|
||||
}
|
||||
}
|
||||
|
||||
// FG8c: the in-JVM embedded broker spawns each binding's managed worker
|
||||
// as a direct child of this JVM, so the live worker set is exactly the
|
||||
// JVM's children carrying the worker argument. The driver diffs it
|
||||
// around a case's first dispatch to name the one worker it signals.
|
||||
private List<Long> workers() {
|
||||
return ProcessHandle.current().children()
|
||||
.filter(ProcessHandle::isAlive)
|
||||
.filter(process -> List.of(process.info().arguments()
|
||||
.orElse(new String[0]))
|
||||
.contains("managed-runtime-worker"))
|
||||
.map(ProcessHandle::pid).sorted().toList();
|
||||
}
|
||||
|
||||
// FG8c's generation ground truth: the Session's bindings (isolation is
|
||||
// per Harness Session), their slots and Runtime Sessions, and the full
|
||||
// loss fields of its execution rows.
|
||||
private Map<String, Object> runtime(String sessionId) {
|
||||
var bindings = jdbc.queryForList(
|
||||
"SELECT CAST(binding_id AS CHAR) AS binding_id, request_key,"
|
||||
+ " binding_state, runtime_generation,"
|
||||
+ " loss_evidence_json IS NOT NULL AS loss_evidence,"
|
||||
+ " stop_evidence_json IS NOT NULL AS stop_evidence,"
|
||||
+ " drain_requested FROM ma_runtime_binding WHERE"
|
||||
+ " tenant_id = ? AND isolation_key = ? ORDER BY"
|
||||
+ " runtime_generation", tenant, sessionId);
|
||||
var slots = new java.util.ArrayList<Map<String, Object>>();
|
||||
for (var key : bindings.stream().map(row -> row.get("request_key"))
|
||||
.distinct().toList()) {
|
||||
slots.addAll(jdbc.queryForList(
|
||||
"SELECT request_key, last_generation,"
|
||||
+ " CAST(active_binding_id AS CHAR) AS"
|
||||
+ " active_binding_id FROM ma_runtime_slot"
|
||||
+ " WHERE request_key = ?", key));
|
||||
}
|
||||
return Map.of("bindings", bindings, "slots", slots,
|
||||
"runtimeSessions", jdbc.queryForList(
|
||||
"SELECT CAST(runtime_session_id AS CHAR) AS"
|
||||
+ " runtime_session_id, session_state,"
|
||||
+ " runtime_generation FROM"
|
||||
+ " ma_runtime_session WHERE tenant_id = ?"
|
||||
+ " AND harness_session_id = ? ORDER BY"
|
||||
+ " last_active_at", tenant, sessionId),
|
||||
"executions", jdbc.queryForList(
|
||||
"SELECT CAST(execution_call_id AS CHAR) AS"
|
||||
+ " execution_call_id, execution_state,"
|
||||
+ " execution_status, runtime_generation,"
|
||||
+ " CAST(runtime_session_id AS CHAR) AS"
|
||||
+ " runtime_session_id, CAST(loss_evidence_id"
|
||||
+ " AS CHAR) AS loss_evidence_id,"
|
||||
+ " abandoned_at IS NOT NULL AS abandoned,"
|
||||
+ " CAST(result_json AS CHAR) AS result_json"
|
||||
+ " FROM ma_tool_execution WHERE"
|
||||
+ " harness_session_id = ? ORDER BY"
|
||||
+ " execution_call_id", sessionId));
|
||||
}
|
||||
|
||||
private List<Map<String, Object>> ledger(String sessionId) {
|
||||
// The surfaces serve the SESSION stream only — a child task's
|
||||
// state_changed audit rows share the sequence space but are never
|
||||
|
||||
@@ -500,11 +500,13 @@ describe('Hosted real-process gates', () => {
|
||||
// headroom — and the FG9 session/relay letters, four since slice 2,
|
||||
// measure 308–314s for the class locally on JDK 21 + MySQL 8.4 across
|
||||
// three acceptance runs (~534s budgeted at >1.7x the heaviest measured
|
||||
// run, up from slice 1's ~300s) in the same envelope; with the fork
|
||||
// itself ~14s, 2640s covers them. A larger ceiling lets failsafe kill
|
||||
// the fork mid-run, and the classes left unrun never report, failing
|
||||
// the gate undiagnosably (#13780).
|
||||
const hostedForkHealthySeconds = 2640;
|
||||
// run, up from slice 1's ~300s) in the same envelope, and the FG8c
|
||||
// process-crash letter adds 144s measured locally on JDK 21 + MySQL
|
||||
// 8.4.6 (~290s budgeted at 2x); with the fork itself ~14s, 2930s covers
|
||||
// them. A larger ceiling lets failsafe kill the fork mid-run, and the
|
||||
// classes left unrun never report, failing the gate undiagnosably
|
||||
// (#13780).
|
||||
const hostedForkHealthySeconds = 2930;
|
||||
const hostedForkSeconds = () =>
|
||||
Number(
|
||||
read('packages/managed-agent-server/pom.xml')
|
||||
|
||||
Reference in new issue
Block a user