test(managed-agent): pin the private CSI receipt guards and fix design decision 5 (#16011)

#16000 documented that Sessions created through the private CSI lane
record no tool receipts for projection. Pin the two places that hold:

- WorkspaceCsiSqlWorkerMySqlIT now wires a ManagedToolResultStore into
  the journal, as production does, and asserts that no
  TOOL_RESULT_PROJECTION row exists after its receipt-carrying CSI
  commits. Without the store the capture guard was never reached.
- ToolPublicationStoreTest marks the Session private (through
  tool_profile or runtime_request_key, since csi_guard is generated)
  before claiming, asserts the claim is empty and the row untouched,
  then clears the marker and claims the same row.

Decision 5 of the automation runtime design said a scanner whose lease
expired records and settles nothing. Occurrence writes check owner and
fence, never lease_until, so they are refused only once another claim
moves the fence. Both language versions now say so, matching the
AutomationLedgerStore Javadoc corrected in #16000.
This commit is contained in:
Shaojin Wen authored and GitHub committed 2026-10-31 20:28:07 +08:00
1 parent cca888dfcc
commit f99062525d
4 files changed
+38 -2

No files matched your search

@@ -34,7 +34,7 @@ The facts below are from this slice's base, `main` at `e92b60a282`.
2. **Identities are derived once and agree in both languages.** `automationRunId = arun_<sha256(scheduleId NUL occurrenceKey)>`. The authority and the Java store refuse a first `automation_run` revision whose id is not the derivation of its own `scheduleId` and `occurrenceKey`, so two claims of one occurrence collide on the record key and the second claimant reads the committed run — the structural dedup the H6a verification asked H6b for. The run's input is `<automationRunId>:input` (its turn id is the same), run command ids are `<automationRunId>:<revision>`, and the control plane derives `scheduleId` as `asch_<sha256(tenantId NUL Idempotency-Key)[0:32]>`, so a create retried after a crash between the Harness answer and the command row meets the same definition instead of minting a second one. Every definition mutation commits under the relayed operation id the control plane derives from the same `Idempotency-Key` (a name-based UUID), so a re-driven relay re-sends it: the authority replays the operation's originally committed revision from the journal — however the chain moved since — instead of committing the retried content over a newer revision, and the control plane records the replayed answer as the operation's result while its mirror never regresses past the revision it already tracks. A mutation honored without a revision (its content identical to the committed definition) commits an `operation.replayed` marker naming the honored revision, so its lost-answer retry answers the honored result instead of re-applying old content over the moved chain. Every replay verifies the operation's target and, for defines, its requested content; a retry under one identity naming another target or another content is refused `409 automation_operation_conflict` rather than answered with a different request's result.
3. **A run binds to the live definition at the current revision.** A first `automation_run` revision must name a committed, non-terminal `schedule` of the same Session at exactly its current `definitionRevision`, with the same `sessionMode` and `targetSessionId`. A scanner that derived a slot from an older revision is refused and re-reads; the per-definition watermark (decision 7) keeps the slot covered or not regardless of which revision reads it.
4. **The run's revision chain, persistent arm.** Revision 1 (`claim`): run `admitted`, execution `intent`, `dispatchId = automationRunId`, no runtime, no delivery. Revision 2 (`dispatch`): run `running`, execution `dispatch_started`, committed in one transaction with `input.accepted` (source `automation`, the envelope resource holding the turn text) and the generated `wake.requested`. Revision 3 (`settle`): run `settled` (turn completed), `failed` (turn error) or `cancelled` (turn cancelled), execution `settled`; an input the close path settled model-free ends `cancelled` with execution `not_started_proven`. Each step is one allowed move per state line (H0b), so a crash between any two leaves a chain the next step completes: a claim without its dispatch is re-driven by the scanner's `firing` row, and a settled turn without its settle revision is reconciled when the Session opens. A crash _inside_ the turn has no lawful proof either way: the run ends once, `failed` with execution `outcome_unknown` (the H0b pairing admits exactly that pair, widened here from recovery-blocked-only) — visible beside the definition instead of `running` forever, and the slot frees.
5. **One Harness operation fires; it is complete before any effect and replay-safe.** `POST /session/:id/automations/operations` with `kind: fire_run` commits revisions 1 and 2 in the funnel's serial chain and answers the committed run; a replay answers the same run and completes a missing revision 2. The scanner inserts the occurrence row as `firing` before the call and marks it `fired` only after the answer, so a crash anywhere before that re-drives the same operation and never a second run. Every occurrence write names the lease its writer holds (owner and fence), so a scanner whose lease expired records and settles nothing; a manual run takes the same lease (`409 automation_busy` when a tick keeps it). A lost answer is re-driven with backoff (2s doubling, the shift clamped at 256 seconds — the 64 attempts total close to a four-hour window, not a 5-minute cap per attempt) and after 64 attempts the row is recorded `unknown`: visible, never re-driven, not counted as active. A replay of an occurrence whose run is already dispatched answers that run even after the definition moved or retired; a claim whose definition ended before its dispatch settles `cancelled` with execution `not_started_proven`. A `revision_stale` answer refreshes the mirror from the committed record and re-pins the claim to the live revision once before the row is recorded `skipped(revision_stale)`. Any other definitive 4xx answer settles the occurrence `skipped` with the route's typed code recorded as its reason — a verdict the request alone decides ends the slot — while store and transport faults answer a retryable 503 the backoff re-drives; the reason recorded is the answer's own code, `hosted_session_blocked` included. A run settles from the journal's own `turn.settled` event of its input. `define_schedule` and `retire_schedule` commit definition revisions the same way, replaying by the relayed operation id (decision 2). The verb route's connector re-runs the Workspace admission `submit` performs before every relay, attachment cached or not, so a grant revoked after the Session attached stops fires and definition mutations alike. When the connector's cache misses but the Session was attached before (its record carries `harnessBootId`), the relay re-attaches through the same takeover load a Turn uses, so a control-plane redeploy never strands every automated Session on `hosted_session_already_attached`.
5. **One Harness operation fires; it is complete before any effect and replay-safe.** `POST /session/:id/automations/operations` with `kind: fire_run` commits revisions 1 and 2 in the funnel's serial chain and answers the committed run; a replay answers the same run and completes a missing revision 2. The scanner inserts the occurrence row as `firing` before the call and marks it `fired` only after the answer, so a crash anywhere before that re-drives the same operation and never a second run. Every occurrence write names the lease its writer holds (owner and fence) and never checks the lease's expiry: a scanner whose lease lapsed can still record and settle until another claim moves the fence, and from then on records and settles nothing; a manual run takes the same lease (`409 automation_busy` when a tick keeps it). A lost answer is re-driven with backoff (2s doubling, the shift clamped at 256 seconds — the 64 attempts total close to a four-hour window, not a 5-minute cap per attempt) and after 64 attempts the row is recorded `unknown`: visible, never re-driven, not counted as active. A replay of an occurrence whose run is already dispatched answers that run even after the definition moved or retired; a claim whose definition ended before its dispatch settles `cancelled` with execution `not_started_proven`. A `revision_stale` answer refreshes the mirror from the committed record and re-pins the claim to the live revision once before the row is recorded `skipped(revision_stale)`. Any other definitive 4xx answer settles the occurrence `skipped` with the route's typed code recorded as its reason — a verdict the request alone decides ends the slot — while store and transport faults answer a retryable 503 the backoff re-drives; the reason recorded is the answer's own code, `hosted_session_blocked` included. A run settles from the journal's own `turn.settled` event of its input. `define_schedule` and `retire_schedule` commit definition revisions the same way, replaying by the relayed operation id (decision 2). The verb route's connector re-runs the Workspace admission `submit` performs before every relay, attachment cached or not, so a grant revoked after the Session attached stops fires and definition mutations alike. When the connector's cache misses but the Session was attached before (its record carries `harnessBootId`), the relay re-attaches through the same takeover load a Turn uses, so a control-plane redeploy never strands every automated Session on `hosted_session_already_attached`.
6. **Exactly one scanner claims, and the fence guards the ledger.** Each definition has one ledger row (`qwen_managed_automation_schedule`) with a lease (`lease_owner`, `lease_until`) and a monotonic `fence`; a scanner claims a row only when the lease is free or expired, bumps the fence, and writes every later ledger step with `WHERE lease_owner = ? AND fence = ?`. A scanner that lost its lease therefore cannot move the watermark or record occurrences, and could at most re-drive a Harness operation that is idempotent by the derived run id — the effect is guarded by the journal, the ledger by the fence. The scanner runs on a dedicated `managedAutomationScheduler` pool, never the one-thread default scheduler that also carries the message materializer, the turn-retry dispatch and the recovery ticks, because its blocking Harness calls would otherwise delay unrelated work by a request timeout at a time.
7. **Slots, watermark, late slots and catch-up.** A tick derives the slots of a definition in the window `(from, now]`, where `from` is the greater of the covered watermark and the arming instant, and never older than the lookback horizon (`qwen.managed-agent.automation.lookback`, default 24 h). A slot is _timely_ when `now − slot ≤ late-tolerance` (default 5 min) and _late_ otherwise. Timely slots fire. Late slots are missed under `none`; under `latest` the newest late slot fires as a catch-up run and the rest are recorded missed; under `bounded: N` the newest N fire oldest first. Every slot in the window gets a ledger row, and the watermark advances to the newest, so a later revision — or a later tick — never re-arms a covered slot (H6 decision 2). The result set is bounded (`max-slots-per-tick`, default 1,000) — the minute walk itself follows the lookback horizon, so raising it widens per-tick work, never trims it; a window beyond the result bound is cut at the oldest end and the cut is logged — unbounded catch-up has no path here.
8. **Overlap is the Session's input-queue admission.** A `persistent` target runs one turn at a time, so an overlapping occurrence is one more pending input in the journal. `active` counts the definition's occurrences whose run is not terminal (`firing` rows included). `skip` fires only at `active = 0`, `queue_one` only at `active ≤ 1` (one running, one waiting in the journal), `allow` up to the concurrency quota (default 4); a dropped occurrence is recorded `skipped` with reason `overlap` or `count_limit`, never as a run. A target Session that is not `ACTIVE` records `skipped` with reason `session_not_active`. An open public Turn on the target suppresses every fire under any policy as `skipped` with reason `overlap`: the wake pump starts a queued automation input the moment the Session idles, while the dispatcher retries the Turn with backoff, so a recurring definition could otherwise starve the Turn indefinitely; the Turn goes first and the slot fires when it ends. The count joins the hosted Session store's record projection, so mutations answer `409 automation_unavailable` unless both `qwen.managed-agent.automation.enabled` and `qwen.managed-agent.session-store.enabled` are on.
@@ -34,7 +34,7 @@ H6a 钉住了*什么可以提交*:以 `scheduleId` 为键、只增修订的定
2. **身份只派生一次,两种语言一致。** `automationRunId = arun_<sha256(scheduleId NUL occurrenceKey)>`。authority 与 Java 存储拒绝 id 不是自身 `scheduleId` 与 `occurrenceKey` 派生值的首个 `automation_run` 修订,于是同一 occurrence 的两次 claim 在记录键上相撞,第二个 claim 方读取已提交的 run——这正是 H6a 验证要求 H6b 补上的结构性去重。run 的输入是 `<automationRunId>:input`(turn id 相同),run 的命令 id 为 `<automationRunId>:<revision>`,控制面以 `asch_<sha256(tenantId NUL Idempotency-Key)[0:32]>` 派生 `scheduleId`,因此在 Harness 答复与命令行之间崩溃后重试的 create 会遇到同一个定义,而不是铸造第二个。每个定义变更都在控制面从同一 `Idempotency-Key` 派生(基于名字的 UUID)并随 relay 传递的操作 id 下提交,于是被重新驱动的 relay 会重发同一个操作 id:authority 从日志重放该操作最初提交的修订——无论此后链如何前进——而不是把重试内容提交到更新的修订之上;控制面把重放答复记为该操作的结果,其镜像也绝不回退到已跟踪修订之前。内容与已提交定义完全一致、因而不追加修订的变更,会提交一条 `operation.replayed` 标记并记录所应答的修订,于是其答复丢失后的重试应答该已受理的结果,而不是把旧内容重新应用到已前进的链上。每次重放都校验操作的目标,以及 define 的请求内容;同一操作 id 指向另一个目标或另一份内容的重试以 `409 automation_operation_conflict` 拒绝,绝不拿别的请求的结果作答。
3. **run 以当前修订绑定活的定义。** 首个 `automation_run` 修订必须命名同一 Session 内已提交、非终态、且恰好处于其当前 `definitionRevision` 的 `schedule`,并带相同的 `sessionMode` 与 `targetSessionId`。从旧修订推导出 slot 的扫描者会被拒并重新读取;每定义水位(决策 7)使 slot 是否已覆盖与读它的修订无关。
4. **run 的修订链,persistent 分支。** 修订 1(`claim`):run `admitted`、execution `intent`、`dispatchId = automationRunId`,无 runtime、无 delivery。修订 2(`dispatch`):run `running`、execution `dispatch_started`,与 `input.accepted`(来源 `automation`,封装资源携带 turn 文本)及生成的 `wake.requested` 同一事务提交。修订 3(`settle`):run `settled`(turn 完成)、`failed`(turn 出错)或 `cancelled`(turn 取消),execution `settled`;被关闭路径无模型结算的输入以 `cancelled` 结束,execution 为 `not_started_proven`。每一步在每条状态线上至多移动一个允许的步(H0b),因此任意两步之间崩溃留下的链都能被下一步补全:没有 dispatch 的 claim 由扫描者的 `firing` 行重新驱动,没有结算修订的已结算 turn 在 Session 打开时对账。turn 内部*发生*的崩溃从日志侧无从取证:该 run 恰结算一次为 `failed` 且 execution `outcome_unknown`(H0b 配对在此恰放行这一对,自仅限 recovery_blocked 放宽)——可见地停在定义旁,而不是永远 `running`,名额立即释放。
5. **一次 Harness 操作点火;它在任何副作用前完整,且可安全重放。** `POST /session/:id/automations/operations` 的 `kind: fire_run` 在 funnel 的串行链中提交修订 1 与 2 并返回已提交的 run;重放返回同一个 run,并补上缺失的修订 2。扫描者在调用前把 occurrence 行写为 `firing`,只在收到答复后标为 `fired`,于是在此之前任何位置崩溃都会重新驱动同一操作,而绝不是第二个 run。每次 occurrence 写入都携带写入者持有的租约(owner 与 fence),租约过期的扫描者无法记录或结算任何行;手动运行持有同一租约(tick 一直持有时返回 `409 automation_busy`)。丢失的答复以退避方式重新驱动(2 秒起倍增、位移钳制在 256 秒——约 64 次的总窗口接近四小时,而不是每次尝试限 5 分钟),64 次后该行记为 `unknown`:可见、不再重驱、不计入 active。run 已 dispatch 的 occurrence 被重放时,即使定义已修订或退役也返回该 run;定义在 dispatch 前结束的 claim 以 `cancelled`、execution `not_started_proven` 结算。`revision_stale` 答复会先从已提交记录刷新镜像,并把 claim 重新钉到当前修订重试一次,仍失败才记为 `skipped(revision_stale)`。其他确定性 4xx 答复一律以软路由的类型化错误码把 occurrence 记为 `skipped`、原因即该码——单凭请求自身即可判决的拒答就此了结本档,存储与传输故障应答的可重试 503 则由退避重新驱动;记入的原因就是应答自己的码,`hosted_session_blocked` 也不例外。run 依据其输入在日志中的 `turn.settled` 事件结算。`define_schedule` 与 `retire_schedule` 以同样方式提交定义修订,按 relay 传递的操作 id 重放(决策 2)。动词路由的 connector 在每次 relay 前重新执行与 `submit` 相同的 Workspace 准入校验,无论 attachment 是否已缓存,因此 Session 挂载后才被撤销的授权会同时挡住点火与定义变更。缓存未命中而 Session 此前已挂接(其记录携带 `harnessBootId`)时,relay 经由与 Turn 相同的 takeover load 重新挂接,控制面的每次重发版不会让全部自动化 Session 卡在 `hosted_session_already_attached` 上。
5. **一次 Harness 操作点火;它在任何副作用前完整,且可安全重放。** `POST /session/:id/automations/operations` 的 `kind: fire_run` 在 funnel 的串行链中提交修订 1 与 2 并返回已提交的 run;重放返回同一个 run,并补上缺失的修订 2。扫描者在调用前把 occurrence 行写为 `firing`,只在收到答复后标为 `fired`,于是在此之前任何位置崩溃都会重新驱动同一操作,而绝不是第二个 run。每次 occurrence 写入都携带写入者持有的租约(owner 与 fence),但从不检查租约是否过期:租约已过期的扫描者在另一次 claim 推进 fence 之前仍能记录和结算,此后便无法记录或结算任何行;手动运行持有同一租约(tick 一直持有时返回 `409 automation_busy`)。丢失的答复以退避方式重新驱动(2 秒起倍增、位移钳制在 256 秒——约 64 次的总窗口接近四小时,而不是每次尝试限 5 分钟),64 次后该行记为 `unknown`:可见、不再重驱、不计入 active。run 已 dispatch 的 occurrence 被重放时,即使定义已修订或退役也返回该 run;定义在 dispatch 前结束的 claim 以 `cancelled`、execution `not_started_proven` 结算。`revision_stale` 答复会先从已提交记录刷新镜像,并把 claim 重新钉到当前修订重试一次,仍失败才记为 `skipped(revision_stale)`。其他确定性 4xx 答复一律以软路由的类型化错误码把 occurrence 记为 `skipped`、原因即该码——单凭请求自身即可判决的拒答就此了结本档,存储与传输故障应答的可重试 503 则由退避重新驱动;记入的原因就是应答自己的码,`hosted_session_blocked` 也不例外。run 依据其输入在日志中的 `turn.settled` 事件结算。`define_schedule` 与 `retire_schedule` 以同样方式提交定义修订,按 relay 传递的操作 id 重放(决策 2)。动词路由的 connector 在每次 relay 前重新执行与 `submit` 相同的 Workspace 准入校验,无论 attachment 是否已缓存,因此 Session 挂载后才被撤销的授权会同时挡住点火与定义变更。缓存未命中而 Session 此前已挂接(其记录携带 `harnessBootId`)时,relay 经由与 Turn 相同的 takeover load 重新挂接,控制面的每次重发版不会让全部自动化 Session 卡在 `hosted_session_already_attached` 上。
6. **恰有一个扫描者 claim,代数守护台账。** 每个定义有一行台账(`qwen_managed_automation_schedule`),带租约(`lease_owner`、`lease_until`)与单调的 `fence`;扫描者只在租约空闲或过期时 claim,递增 fence,之后每一步台账写入都带 `WHERE lease_owner = ? AND fence = ?`。因此失去租约的扫描者无法推进水位或记录 occurrence,至多重新驱动一次按派生 run id 幂等的 Harness 操作——副作用由日志守护,台账由代数守护。扫描者跑在独立的 `managedAutomationScheduler` 线程池上,不占用也承载消息物化、Turn 重试派发与恢复 tick 的单线程默认调度器,否则每次被阻塞的 Harness 调用都会按请求超时拖慢与之无关的工作。
7. **slot、水位、迟到 slot 与 catch-up。** 一次 tick 推导定义在窗口 `(from, now]` 内的 slot,其中 `from` 是已覆盖水位与布防时刻中较大者,且不早于回看地平线(`qwen.managed-agent.automation.lookback`,默认 24 小时)。`now − slot ≤ late-tolerance`(默认 5 分钟)的 slot 为*及时*,否则为*迟到*。及时 slot 点火。迟到 slot 在 `none` 下记为 missed;在 `latest` 下最新的一个迟到 slot 作为 catch-up run 点火、其余记为 missed;在 `bounded: N` 下最新的 N 个由旧到新点火。窗口内每个 slot 都有一行台账,水位推进到最新,因此后续修订——或后续 tick——绝不会重新布防已覆盖的 slot(H6 决策 2)。返回集有界(`max-slots-per-tick`,默认 1,000)——逐分钟的遍历本身跟随回看地平线,因此放大回看会扩大每 tick 的工作量而不是削减排程;窗口超出返回上界时从最旧一端截断并记录日志——这里不存在无限补跑的路径。
8. **overlap 是 Session 输入队列的准入。** `persistent` 目标一次运行一个 turn,因此重叠的 occurrence 只是日志里多一个待处理输入。`active` 统计该定义中 run 非终态的 occurrence(含 `firing` 行)。`skip` 仅在 `active = 0` 时点火,`queue_one` 仅在 `active ≤ 1` 时(一个在跑、一个在日志中等待),`allow` 最多到并发配额(默认 4);被丢弃的 occurrence 以原因 `overlap` 或 `count_limit` 记为 `skipped`,绝不记为 run。目标 Session 不是 `ACTIVE` 时记为 `skipped`,原因 `session_not_active`。目标上有未结束的公开 Turn 时,在任意 overlap 策略下一律记 `skipped`、原因 `overlap`:Session 一空闲,唤醒泵立即启动排队的自动化输入,而派发器以退避重试该 Turn——周期性定义可以因此把该 Turn 饿死;Turn 先行,结束后 slot 再点火。该计数关联托管 Session 存储的记录投影,因此只有 `qwen.managed-agent.automation.enabled` 与 `qwen.managed-agent.session-store.enabled` 同时开启时,变更类请求才不会返回 `409 automation_unavailable`。
@@ -3202,6 +3202,33 @@ class ToolPublicationStoreTest {
.isEqualTo(events);
}
@ParameterizedTest
@ValueSource(strings = {"tool_profile = 'csi-files-retirement/1'", "runtime_request_key = 'fixture-request'"})
void projectionClaimSkipsPrivateCsiSessions(String privateMarker) {
// csi_guard is generated from these two columns, so the seed sets one of them.
var fixture = apiFixture();
fixture.jdbc()
.update(
"UPDATE ma_operation SET state = 'PENDING', available_at = 0 WHERE"
+ " operation_kind = 'TOOL_RESULT_PROJECTION'");
var original = fixture.jdbc().queryForMap("SELECT tool_profile, runtime_request_key FROM ma_session");
fixture.jdbc().update("UPDATE ma_session SET " + privateMarker);
var before =
fixture.jdbc()
.queryForMap(
"SELECT " + ManagedSchemaProjection.OPERATION.columns() + " FROM ma_operation WHERE operation_kind = 'TOOL_RESULT_PROJECTION'");
assertThat(fixture.results().claim()).isEmpty();
assertThat(
fixture.jdbc()
.queryForMap(
"SELECT " + ManagedSchemaProjection.OPERATION.columns() + " FROM ma_operation WHERE operation_kind = 'TOOL_RESULT_PROJECTION'"))
.usingRecursiveComparison()
.isEqualTo(before);
fixture.jdbc().update("UPDATE ma_session SET tool_profile = ?, runtime_request_key = ?",
original.get("tool_profile"), original.get("runtime_request_key"));
assertThat(fixture.results().claim()).isPresent();
}
@Test
void retiredRootRejectsProjectionWhilePublicSessionIsStillReadable() {
var fixture = apiFixture();
@@ -177,6 +177,11 @@ class WorkspaceCsiSqlWorkerMySqlIT {
journal.setCsiBindings(bindings);
var properties = new ManagedAgentProperties();
properties.setAgentRevision("construction-fixture/1");
// Wired as in production, so the receipt commits below reach the capture guard.
journal.setToolResults(new ManagedToolResultStore(jdbc, manager,
com.alibaba.qwen.code.managedagent.VerificationFixtures.configure(new ManagedAgentStore(jdbc, JSON,
java.time.Clock.systemUTC(), ignored -> {}, new ManagedWorkspaceRegistry(jdbc), properties,
new com.alibaba.qwen.code.managedagent.harness.SdkHarnessRules()))));
var endpoint = new AtomicReference<URI>();
var actualClient = HttpClient.newBuilder().followRedirects(HttpClient.Redirect.NEVER).build();
var mappedClient = mock(HttpClient.class);
@@ -427,6 +432,10 @@ class WorkspaceCsiSqlWorkerMySqlIT {
}
assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_tool_execution WHERE native_authorization_json IS NOT NULL"
+ " AND execution_state = 'SETTLED'", Integer.class)).isEqualTo(4);
// The receipts of those executions were committed through the private CSI
// lane, which records none for projection.
assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_operation"
+ " WHERE operation_kind = 'TOOL_RESULT_PROJECTION'", Integer.class)).isZero();
assertThat(jdbc.queryForObject("SELECT writer_generation FROM ma_session_head", Long.class))
.isEqualTo(coldRecovery ? 3 : 1);
if (coldRecovery) {