mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-11-07 06:19:43 -10:00
feat(managed-agent): project channel deliveries as tasks and serve their cancel (#13746 F3) (#14414)
* feat(managed-agent): H4f-c public task cancel for background Shell and Monitor tasks The H3 kinds join the public task cancel path now that #13889 lifted their gates. A background_shell or monitor task in pending, running, waiting or degraded, in an active Session, advertises cancel. Its delivery stops the task's process through the embedded Runtime Broker (the drain sweep's status, stop, status for that one hold row, unfenced), and the operation completes from the committed record: a Shell's stop request, which the A8 publisher writes on the worker's stop-marked end, or a Monitor's stop_requested settlement, since monitor_run carries no request. A process that ended first fails the operation with task_already_settled; an unproven stop retries within the budget and then parks. Adds FG8g, a real-stack gate that cancels a live background Shell and a live Monitor watch through the public route. Contract 1.53.0 (provisional). Part of #13746. * feat(managed-agent): project channel deliveries as tasks and serve their cancel (#13746 F3) A channel delivery now projects as a task of its route Session (kind channel_delivery, in both languages and the shared projection fixtures), so the one public task cancel path serves it: no new route, operation type or failure code. The task is pending while the delivery is planned, running while an adapter sends it, waiting while its outcome is unknown, and completed, failed or cancelled at its end. Cancel is advertised and admitted only while the delivery is pending; the delivery arm sends the Hosted cancel_delivery verb, and the operation completes from the record's cancelRequested flag, so an adapter that claimed the delivery first keeps sending it to its natural outcome. Deliveries planned before this change keep projecting no task. FG7d gains two real-stack cases through the public route: a planned delivery cancelled before any send, and a claimed one refused with 409. Contract 1.55.0 (provisional). Part of #13746. * docs(managed-agent): note the delivery task contract number against main * docs(managed-agent): renumber the delivery task contract to 1.57.0
This commit is contained in:
Shaojin Wen authored and GitHub committed
1 parent
fce91ce444
commit
c909af765f
22 files changed
+669
-56
No files matched your search
@@ -219,7 +219,7 @@ The facts below are from `main` at `ba8615f4c4` (Flyway V60, contract `1.38.0`),
|
||||
| ----------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| H3 kinds | The Shell half was delivered by A8 (#13889): a platform stop settles `stop_requested` after its request. **Delivered in [H4f-c](2026-10-17-managed-h3-task-cancel.md): both kinds join `CANCELLABLE_TASK_KINDS`, their delivery stops the process through the Runtime Broker, and a Monitor's `stop_requested` settlement stands in for the stop request it does not carry.** |
|
||||
| Workflow, automation | With the workflow runtime (#13803) and H6 per_run: a stop request on their records, then the same delivery and stop arms. |
|
||||
| Channel delivery cancel | With H5's public mutation routes: reuse this slice's operation model (decision 10). |
|
||||
| Channel delivery cancel | **Delivered in [#13746 F3](2026-10-17-managed-delivery-task-cancel.md):** a delivery projects as a task of its route Session (`channel_delivery`) and takes the same task cancel while no adapter has claimed it; decision 10's separate route is not needed. |
|
||||
| Child output | **Delivered by the [child Session link](2026-10-17-managed-child-task-session-link.md) (#13746 F2): a child task names its child Session, whose routes serve its output.** Relaying it as task events behind `read_output` was not taken. |
|
||||
| WebShell | The task panel (#13936) and its client methods landed for `child_agent` tasks. **It now lists every task kind and offers cancel wherever the task advertises it ([panel note](2026-10-12-managed-child-task-panel.md)).** |
|
||||
| Stage F | The real-stack round's remaining ordering: a cancel after the owner is gone. |
|
||||
|
||||
@@ -230,7 +230,7 @@
|
||||
| ---------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| H3 类型 | Shell 那一半已由 A8（#13889）交付：平台停止在其请求之后以 `stop_requested` 结算。**已在 [H4f-c](2026-10-17-managed-h3-task-cancel.zh-CN.md) 交付：两类加入 `CANCELLABLE_TASK_KINDS`，其投递经 Runtime Broker 停止进程，Monitor 的 `stop_requested` 结算代替它并不携带的停止请求。** |
|
||||
| workflow、自动化 | 随 workflow 运行时（#13803）与 H6 per_run：先在各自记录上加停止请求，再复用相同的投递与停止分支。 |
|
||||
| channel 投递取消 | 随 H5 的公开变更路由：复用本切片的操作模型（决策 10）。 |
|
||||
| channel 投递取消 | **已在 [#13746 F3](2026-10-17-managed-delivery-task-cancel.zh-CN.md) 交付：** 投递投影为其路由 Session 的任务（`channel_delivery`），在尚无适配器认领时走同一条任务取消；不再需要决策 10 所说的独立路由。 |
|
||||
| child 输出 | **已由 [child Session 链接](2026-10-17-managed-child-task-session-link.zh-CN.md)（#13746 F2）交付：child 任务给出其 child Session，由该 Session 的路由提供输出。** 没有采用在 `read_output` 之后作为任务事件转发。 |
|
||||
| WebShell | 任务面板（#13936）及其客户端方法已为 `child_agent` 任务落地。**它现在列出所有任务类型，并在任务公布取消时提供取消（[面板文档](2026-10-12-managed-child-task-panel.zh-CN.md)）。** |
|
||||
| F 阶段 | 真实栈轮次剩下的时序：所有者消失后的取消。 |
|
||||
|
||||
@@ -59,7 +59,7 @@ Checked at `d28bc60d73` (Flyway V68, contract `1.44.0`):
|
||||
| ----------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| H3 kinds | **Delivered in [H4f-c](2026-10-17-managed-h3-task-cancel.md):** A8 (#13889) already settles a platform-stopped Shell `stop_requested`; both kinds join `CANCELLABLE_TASK_KINDS`, their delivery stops the process through the Runtime Broker, and a Monitor's `stop_requested` settlement is its recorded stop. |
|
||||
| `automation_run` | Carries no stop request; H6's per_run target owns it. |
|
||||
| Channel delivery cancel | With H5's public mutation routes (resend first), reusing this operation model — decision 10 of [the H4f note](2026-10-10-managed-task-cancel.md). |
|
||||
| Channel delivery cancel | **Delivered in [#13746 F3](2026-10-17-managed-delivery-task-cancel.md):** a delivery projects as a task of its route Session (`channel_delivery`) and takes the same task cancel while no adapter has claimed it; decision 10's separate route is not needed. |
|
||||
| Child output events | Delivered by the [child Session link](2026-10-17-managed-child-task-session-link.md) (#13746 F2): a workflow task names its child Session like a child agent task. |
|
||||
| WebShell panel | **Delivered:** the panel lists workflow tasks with every other kind and admits a cancel wherever the task advertises it ([panel note](2026-10-12-managed-child-task-panel.md)). |
|
||||
| Close stops the child | [#13875](https://github.com/QwenLM/qwen-code/issues/13875), unchanged: the relay's stop arm still skips a parent that is no longer `ACTIVE`. |
|
||||
@@ -59,7 +59,7 @@
|
||||
| --------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| H3 类型 | **已在 [H4f-c](2026-10-17-managed-h3-task-cancel.zh-CN.md) 交付：** A8（#13889）已把被平台停止的 Shell 结算为 `stop_requested`；两类加入 `CANCELLABLE_TASK_KINDS`，其投递经 Runtime Broker 停止进程，Monitor 的 `stop_requested` 结算即其已记录的停止。 |
|
||||
| `automation_run` | 不携带停止请求；归 H6 的 per_run 目标所有。 |
|
||||
| Channel delivery 取消 | 随 H5 的公开变更路由（resend 先行）一起做，复用本操作模型 —— 见 [H4f 设计](2026-10-10-managed-task-cancel.zh-CN.md)决策 10。 |
|
||||
| Channel delivery 取消 | **已在 [#13746 F3](2026-10-17-managed-delivery-task-cancel.zh-CN.md) 交付：** 投递投影为其路由 Session 的任务（`channel_delivery`），在尚无适配器认领时走同一条任务取消；不再需要决策 10 所说的独立路由。 |
|
||||
| Child 输出事件 | 已由 [child Session 链接](2026-10-17-managed-child-task-session-link.zh-CN.md)（#13746 F2）交付：workflow 任务像 child agent 任务一样给出其 child Session。 |
|
||||
| WebShell 面板 | **已交付：** 面板与其他所有类型一起列出 workflow 任务，并在任务公布取消时接受取消（[面板文档](2026-10-12-managed-child-task-panel.zh-CN.md)）。 |
|
||||
| Close 停止 child | [#13875](https://github.com/QwenLM/qwen-code/issues/13875)，不变：relay 的停止臂仍跳过不再 `ACTIVE` 的父 Session。 |
|
||||
@@ -0,0 +1,116 @@
|
||||
# Managed channel delivery as a task, and its cancel (#13746 F3)
|
||||
|
||||
[English](2026-10-17-managed-delivery-task-cancel.md) | [简体中文](2026-10-17-managed-delivery-task-cancel.zh-CN.md)
|
||||
|
||||
Status: implemented in this change. A channel delivery projects as a task of its route Session (kind `channel_delivery`), and the public task cancel serves it while no adapter has claimed it. The contract names the kind in `1.57.0` (main is at `1.56.0`: the child-Session-link slice took `1.55.0` and #14405 took `1.56.0`; renumber at merge if another pull request takes `1.57.0` first). There are no new routes, operation types or failure codes. The change builds on [H4f-c](2026-10-17-managed-h3-task-cancel.md) (merged), whose cancellable-kind rule and delivery switch it shares.
|
||||
|
||||
This is F3 of [#13746](https://github.com/QwenLM/qwen-code/issues/13746), "the H5 half": one cancellation path serves both a child task and a channel delivery, and neither grows its own vocabulary. It belongs to stage H of [#12380](https://github.com/QwenLM/qwen-code/issues/12380) under [#12827](https://github.com/QwenLM/qwen-code/issues/12827). It replaces decision 10 of the [H4f note](2026-10-10-managed-task-cancel.md), which kept deliveries out of the task model and left their cancel to H5's public mutation routes. The issue's own framing settles that question: channel deliveries are tasks with a delivery line.
|
||||
|
||||
## Problem
|
||||
|
||||
A channel reply waits in the outbox as a `planned` delivery until the adapter pulls it. A caller who sees the reply should not go out has no way to stop it:
|
||||
|
||||
- the public channel routes are read-only;
|
||||
- they read the delivery ledger, which has a row only once an adapter claimed the delivery, so a planned one is not even visible;
|
||||
- the Hosted `cancel_delivery` verb exists, but nothing outside the fault gates calls it.
|
||||
|
||||
## Current state
|
||||
|
||||
Checked at H4f-c on `b4614845f0` (contract `1.54.0`):
|
||||
|
||||
- **The record.**
|
||||
- `channel_delivery` is a record of the route Session. It carries a delivery line (`planned` → `sending` → `partial`, `delivered`, `unknown`, `rejected`, or `cancelled`) and a set-once `cancelRequested` flag.
|
||||
- `cancelled` needs the flag and zero receipts, so only a planned delivery can end cancelled (`ManagedChannelRecords.java:178`).
|
||||
- The record projects no task (`ManagedExtensionProjection.java:62`, and the TS mirror).
|
||||
- **The Hosted verb.**
|
||||
- `cancel_delivery` on `/session/:id/channels/operations` settles a planned delivery `cancelled` with the flag in one revision.
|
||||
- On a delivery an adapter already claimed it records the flag only, and the send proceeds.
|
||||
- A terminal delivery is answered unchanged (`hosted-channel-session.ts:707`).
|
||||
- Its only callers are the FG7d fault gates.
|
||||
- **The claim.** The adapter's pull discovers `planned` and `partial` records (`ChannelInstanceStore.java:263`, `:296`). A cancelled one is never claimed.
|
||||
- **Public reads.** Delivery list and detail read the ledger; a planned delivery answers `404 delivery_not_found` (`ManagedChannelService.java:502`).
|
||||
- **The task path.**
|
||||
- Task list, detail, events and cancel are served for every task-projecting record of a Session.
|
||||
- A record row gets its `task_kind` when it is first inserted, and later revisions update only the projected state (`ManagedExtensionRecordStore.java:1350`, `:1383`).
|
||||
|
||||
## Decisions
|
||||
|
||||
1. **A delivery projects as a task of its route Session.**
|
||||
- The `channel_delivery` body's task kind becomes `channel_delivery`, in both languages and in the shared projection fixtures, and the contract's `TaskKind` gains the value.
|
||||
- The states follow the run block like every other kind:
|
||||
|
||||
| Delivery line | Run | Task state |
|
||||
| ----------------------------- | ----------- | ----------- |
|
||||
| `planned` | `admitted` | `pending` |
|
||||
| `sending`, `partial` | `running` | `running` |
|
||||
| `unknown` (and its `partial`) | `waiting` | `waiting` |
|
||||
| `delivered` | `settled` | `completed` |
|
||||
| `rejected` | `failed` | `failed` |
|
||||
| `cancelled` | `cancelled` | `cancelled` |
|
||||
|
||||
- The run has no execution, so the task carries no `runtime_state`. It also has no Artifact reference.
|
||||
- Each change of the view is journaled on the H3 per-task event journal, as for every kind.
|
||||
- The task is visible to exactly the readers of the route Session, which is a Session of the channel's Workspace. That is the same set the channel reads admit.
|
||||
|
||||
2. **Rows written before this change stay non-tasks.** A row's kind is set at its insert, and every later revision follows the kind the row stores, not the one its body would project now: a delivery planned before the upgrade writes no task state and journals no task event on any later revision, and never projects a task. No backfill: such a delivery is past its cancel window by the time anyone could act on it, and the channel reads still show it once claimed.
|
||||
|
||||
3. **Cancel is admitted only while the delivery is `pending`.**
|
||||
- `taskActions` gives `channel_delivery` one state rule of its own: `cancel` is advertised, and a new cancel admitted, only for a `pending` (planned, unclaimed) delivery in an active Session.
|
||||
- A claimed delivery is on its way to the provider. Its view advertises nothing, and a new cancel answers `409 task_action_unavailable`.
|
||||
- Everything else is the H4f path unchanged: the A6 admission order, the Workspace OPERATOR rule, replay and the one-open-operation rule.
|
||||
|
||||
4. **The delivery arm and the verdict reuse the task cancel.**
|
||||
- `TaskCancelCoordinator` sends `cancel_delivery` through the route Session's channel operation route, keyed by the operation's attempt.
|
||||
- The verdict reads the committed record, and the `cancelRequested` flag is the stop request:
|
||||
|
||||
| Committed record | Operation outcome |
|
||||
| ------------------------------- | -------------------------------------------------- |
|
||||
| `cancelRequested` is set | `completed` with a receipt |
|
||||
| the delivery ended without it | `failed` `task_already_settled` |
|
||||
| the delivery is live without it | send `cancel_delivery`, then read the record again |
|
||||
|
||||
- An adapter that claims the delivery between the admission and the send wins the race. The verb records the flag, so the operation completes, and the delivery keeps sending to its natural outcome, as a completion that wins keeps its outcome in the task contract.
|
||||
- Unproven outcomes retry within the budget and then park, as H4f decision 5 says.
|
||||
|
||||
5. **No new vocabulary.** There is no delivery route, operation type, delivery id property or failure code. The F3 exit check holds literally: one cancellation path serves both kinds.
|
||||
|
||||
## Contract changes (1.57.0)
|
||||
|
||||
- `TaskKind` gains `channel_delivery`.
|
||||
- `TaskActionCapability`'s description names the kind and its `pending`-only rule.
|
||||
- `info.description` gains the v1.57 paragraph.
|
||||
- The generated WebShell types gain the enum value.
|
||||
|
||||
## Validation
|
||||
|
||||
- **Projection, both languages.** The shared projection fixtures map `channel_delivery` to its kind and list it among the task kinds. Both channel record contract tests pin that a delivery projects a task and a route does not. The TS authority's delivery chain and its reopen now carry the delivery's task view.
|
||||
- **Real records** (`ManagedExtensionRecordStoreTest`):
|
||||
- a planned delivery reads as a `pending` task with `cancel` and no runtime state;
|
||||
- a claimed one reads `running` with no action;
|
||||
- a cancelled one reads `cancelled`, with its request on the record the verdict reads;
|
||||
- a delivery row stored without a kind, as a server before this change wrote it, writes no task state and journals no task event on its next revision.
|
||||
- **Admission** (`ManagedTaskCancelOperationTest`): a `pending` delivery admits and replays. Every other state answers `task_action_unavailable`.
|
||||
- **Delivery** (`TaskCancelCoordinatorTest`):
|
||||
- a pending delivery takes `cancel_delivery` through the channel route, never the child operation route, and completes from the flag;
|
||||
- a delivery claimed meanwhile completes with the flag recorded;
|
||||
- a delivered one without the flag fails `task_already_settled` with no send.
|
||||
- **Real stack** (FG7d, `HostedChannelFaultGatesIT`, MySQL 8.4, packaged Harness, the email reference adapter):
|
||||
- `task-cancel-planned`: the route Session's task list shows the planned delivery as a `pending` task with `cancel`. The public cancel answers `202`, the operation completes and its key replays, and the task reads `cancelled`. The outbox pull reaches the control plane and sends nothing, and the ledger stays empty.
|
||||
- `task-cancel-claimed`: while the adapter's send is parked, the task reads `running` with no action, and a public cancel answers `409 task_action_unavailable`. Released, the delivery is sent once and the task reads `completed`.
|
||||
- The whole FG7 class and the channel crash IT pass beside the new cases.
|
||||
- **Not covered:** listing delivery tasks in the WebShell panel (the panel's own change); a real provider other than the email fixture.
|
||||
|
||||
## Constraints and risks
|
||||
|
||||
- **More rows in a route Session's task list.** Every reply of a channel Session is now one task there, and every revision is one task event. The journal's bounds and retention floor apply unchanged.
|
||||
- **The race is visible.** A cancel whose admission saw `pending` can complete while the reply still goes out, if an adapter claimed it first. The operation's `completed` is the recorded request, and the task's `completed` is the physical outcome, as the contract keeps them apart.
|
||||
- **Legacy rows.** Deliveries planned before the upgrade are not tasks (decision 2).
|
||||
- **Rollback floor.** A server older than this change writes a null task state on every revision of a row it does not know projects a task. Rolled back, it would keep listing the delivery tasks inserted meanwhile (their kind is stored), with a null state and a kind its contract does not name. Roll back below this change only after those Sessions are closed, or accept such rows reading wrong until they are.
|
||||
|
||||
## Follow-up work
|
||||
|
||||
| Item | Scope |
|
||||
| -------------- | ----------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| Public resend | A resend of an `unknown` or `rejected` delivery is still a trusted-adapter verb; a public one needs its own contract work (H5 open question 4). |
|
||||
| WebShell panel | List delivery tasks and admit their cancel. |
|
||||
| Partial stop | Stopping the remaining segments of a `partial` delivery needs a record contract change; no adapter produces one today. |
|
||||
@@ -0,0 +1,116 @@
|
||||
# Managed channel 投递作为任务及其取消（#13746 F3）
|
||||
|
||||
[English](2026-10-17-managed-delivery-task-cancel.md) | [简体中文](2026-10-17-managed-delivery-task-cancel.zh-CN.md)
|
||||
|
||||
状态：已在本变更中实现。channel 投递投影为其路由 Session 的一个任务（kind `channel_delivery`），并且在尚无适配器认领它时，公开任务取消可以服务它。契约在 `1.57.0` 中点名该 kind（main 现为 `1.56.0`：child Session 链接切片占用 `1.55.0`，#14405 占用 `1.56.0`；若合入前另有 PR 先占用 `1.57.0` 则再改号）。没有新路由、新操作类型或新失败码。本变更建立在已合入的 [H4f-c](2026-10-17-managed-h3-task-cancel.zh-CN.md) 之上，与它共用可取消类型规则与投递分支。
|
||||
|
||||
这是 [#13746](https://github.com/QwenLM/qwen-code/issues/13746) 的 F3，即「H5 的那一半」：一条取消路径同时服务 child 任务与 channel 投递，两者都不另长词汇。它属于 [#12380](https://github.com/QwenLM/qwen-code/issues/12380) 的阶段 H，在 [#12827](https://github.com/QwenLM/qwen-code/issues/12827) 下跟踪。它替代 [H4f 文档](2026-10-10-managed-task-cancel.zh-CN.md)的决策 10：那条决策把投递留在任务模型之外，并把它的取消留给 H5 的公开变更路由。issue 自己的表述已经回答了这个问题：channel 投递就是带投递行的任务。
|
||||
|
||||
## 问题
|
||||
|
||||
channel 的回复在适配器拉取之前，以 `planned` 投递的形式留在 outbox 里。看到这条回复、认为它不该发出的调用者没有任何办法阻止它：
|
||||
|
||||
- 公开 channel 路由只读；
|
||||
- 它们读的是投递台账，而台账只有在适配器认领投递后才有行，所以 planned 的投递甚至不可见；
|
||||
- Hosted 的 `cancel_delivery` 动词存在，但除故障门禁外没有任何调用方。
|
||||
|
||||
## 现状
|
||||
|
||||
核对于叠在 `b4614845f0` 上的 H4f-c（契约 `1.54.0`）：
|
||||
|
||||
- **记录。**
|
||||
- `channel_delivery` 是路由 Session 的一条记录。它带有一条投递行（`planned` → `sending` → `partial`、`delivered`、`unknown`、`rejected` 或 `cancelled`），以及一个只设一次的 `cancelRequested` 标志。
|
||||
- `cancelled` 需要该标志且零回执，所以只有 planned 的投递能以 cancelled 结束（`ManagedChannelRecords.java:178`）。
|
||||
- 该记录不投影任务（`ManagedExtensionProjection.java:62` 及其 TS 镜像）。
|
||||
- **Hosted 动词。**
|
||||
- `/session/:id/channels/operations` 上的 `cancel_delivery` 在一次修订中把 planned 投递连同标志结算为 `cancelled`。
|
||||
- 对已被适配器认领的投递，它只记录标志，发送照常进行。
|
||||
- 对终态投递原样应答（`hosted-channel-session.ts:707`）。
|
||||
- 它唯一的调用方是 FG7d 故障门禁。
|
||||
- **认领。** 适配器的拉取发现 `planned` 与 `partial` 记录（`ChannelInstanceStore.java:263`、`:296`）。已取消的投递永远不会被认领。
|
||||
- **公开读取。** 投递列表与详情读台账；planned 投递回 `404 delivery_not_found`（`ManagedChannelService.java:502`）。
|
||||
- **任务路径。**
|
||||
- 对 Session 中每条投影任务的记录，都提供任务列表、详情、事件与取消。
|
||||
- 记录行在首次插入时获得 `task_kind`，之后的修订只更新投影出的状态（`ManagedExtensionRecordStore.java:1350`、`:1383`）。
|
||||
|
||||
## 决策
|
||||
|
||||
1. **投递投影为其路由 Session 的一个任务。**
|
||||
- `channel_delivery` 记录体的任务 kind 变为 `channel_delivery`，两种语言与共享投影 fixtures 同步，契约的 `TaskKind` 增加该值。
|
||||
- 状态与其他类型一样跟随 run 块：
|
||||
|
||||
| 投递行 | run | 任务状态 |
|
||||
| ------------------------------- | ----------- | ----------- |
|
||||
| `planned` | `admitted` | `pending` |
|
||||
| `sending`、`partial` | `running` | `running` |
|
||||
| `unknown`（及其后的 `partial`） | `waiting` | `waiting` |
|
||||
| `delivered` | `settled` | `completed` |
|
||||
| `rejected` | `failed` | `failed` |
|
||||
| `cancelled` | `cancelled` | `cancelled` |
|
||||
|
||||
- run 没有 execution，所以任务不带 `runtime_state`，也没有 Artifact 引用。
|
||||
- 视图的每次变化都记入 H3 的每任务事件日志，与所有类型相同。
|
||||
- 能看到该任务的恰好是路由 Session 的读者，而路由 Session 属于该 channel 的 Workspace。这与 channel 读取所准入的集合相同。
|
||||
|
||||
2. **本变更之前写入的行保持非任务。** 行的 kind 在插入时确定，之后的每次修订都遵循该行存储的 kind，而不是其记录体现在会投影的 kind：升级前已 planned 的投递在之后任何修订上都不写任务状态、不记任务事件，永远不会投影任务。不做回填：等到有人能对它操作时，这样的投递已经过了可取消的窗口，而且它一旦被认领，channel 读取仍能看到它。
|
||||
|
||||
3. **只在投递为 `pending` 时准入取消。**
|
||||
- `taskActions` 为 `channel_delivery` 设一条自己的状态规则：只对处于活跃 Session 中、`pending`（planned、未被认领）的投递公布 `cancel` 并准入新的取消。
|
||||
- 已认领的投递正在发往提供方：其视图不公布任何动作，新的取消回 `409 task_action_unavailable`。
|
||||
- 其余都是未改动的 H4f 路径：A6 准入顺序、Workspace OPERATOR 规则、重放以及「每个 Session 只有一个 open 操作」规则。
|
||||
|
||||
4. **投递分支与 verdict 复用任务取消。**
|
||||
- `TaskCancelCoordinator` 经路由 Session 的 channel operation 路由发送 `cancel_delivery`，以操作的尝试为键。
|
||||
- verdict 读取已提交的记录，`cancelRequested` 标志就是停止请求：
|
||||
|
||||
| 已提交的记录 | 操作结果 |
|
||||
| ------------------------ | ---------------------------------------- |
|
||||
| `cancelRequested` 已设置 | `completed`，带 receipt |
|
||||
| 投递在没有它的情况下结束 | `failed` `task_already_settled` |
|
||||
| 投递存活且没有它 | 发送 `cancel_delivery`，然后再读一次记录 |
|
||||
|
||||
- 在准入与发送之间认领了投递的适配器赢得竞争：动词记录标志，于是操作完成，而投递继续发送直到其自然结果——正如任务契约中先完成者保留其结果。
|
||||
- 未获证明的结果在预算内重试，之后停住，与 H4f 决策 5 相同。
|
||||
|
||||
5. **没有新词汇。** 没有投递路由、操作类型、投递 id 属性或失败码。F3 的验收字面成立：一条取消路径服务两种类型。
|
||||
|
||||
## 契约变更（1.57.0）
|
||||
|
||||
- `TaskKind` 增加 `channel_delivery`。
|
||||
- `TaskActionCapability` 的描述点名该 kind 及其「仅 `pending`」规则。
|
||||
- `info.description` 增加 v1.57 段落。
|
||||
- 生成的 WebShell 类型增加该枚举值。
|
||||
|
||||
## 验证
|
||||
|
||||
- **投影，两种语言。** 共享投影 fixtures 把 `channel_delivery` 映射到其 kind，并把它列入任务 kind。两种语言的 channel 记录契约测试都钉住「投递投影任务、路由不投影」。TS authority 的投递链及其重开现在带有投递的任务视图。
|
||||
- **真实记录**（`ManagedExtensionRecordStoreTest`）：
|
||||
- planned 投递读为带 `cancel`、无 runtime state 的 `pending` 任务；
|
||||
- 已认领的读为无动作的 `running`；
|
||||
- 已取消的读为 `cancelled`，其请求在 verdict 读取的记录上；
|
||||
- 以无 kind 方式存储的投递行（即本变更之前的 server 写入的那样），在下一次修订时不写任务状态、不记任务事件。
|
||||
- **准入**（`ManagedTaskCancelOperationTest`）：`pending` 投递准入并重放；其他所有状态回 `task_action_unavailable`。
|
||||
- **投递**（`TaskCancelCoordinatorTest`）：
|
||||
- pending 投递经 channel 路由（从不经 child operation 路由）收到 `cancel_delivery`，并从标志完成；
|
||||
- 期间被认领的投递以已记录的标志完成；
|
||||
- 已 delivered 且无标志的投递以 `task_already_settled` 失败，不发送。
|
||||
- **真实栈**（FG7d，`HostedChannelFaultGatesIT`，MySQL 8.4，打包的 Harness，email 参考适配器）：
|
||||
- `task-cancel-planned`：路由 Session 的任务列表把 planned 投递显示为带 `cancel` 的 `pending` 任务。公开取消回 `202`，操作完成且其 key 可重放，任务读为 `cancelled`。outbox 拉取到达控制面但什么也不发送，台账保持为空。
|
||||
- `task-cancel-claimed`：适配器的发送被挂起期间，任务读为无动作的 `running`，公开取消回 `409 task_action_unavailable`。放行后投递发送一次，任务读为 `completed`。
|
||||
- 整个 FG7 类与 channel 崩溃 IT 和新用例一起通过。
|
||||
- **未覆盖：** 在 WebShell 面板中列出投递任务（面板自己的变更）；email fixture 之外的真实提供方。
|
||||
|
||||
## 约束与风险
|
||||
|
||||
- **路由 Session 的任务列表行数变多。** channel Session 的每条回复现在都是那里的一个任务，每次修订都是一条任务事件。日志的上限与保留下限照常适用。
|
||||
- **竞争可见。** 准入时看到 `pending` 的取消，可能在适配器先认领的情况下完成，而回复仍然发出。操作的 `completed` 是已记录的请求，任务的 `completed` 是物理结果，契约把二者分开。
|
||||
- **旧行。** 升级前已 planned 的投递不是任务（决策 2）。
|
||||
- **回滚下限。** 早于本变更的 server 对它不知道会投影任务的行，在每次修订时都写入空的任务状态。回滚后，它会继续列出期间插入的投递任务（它们的 kind 已存储），状态为空，kind 也是其契约不认识的值。只有在这些 Session 关闭之后才回滚到本变更之前，或者接受这些行在关闭前读起来不正确。
|
||||
|
||||
## 后续工作
|
||||
|
||||
| 项 | 范围 |
|
||||
| ------------- | ---------------------------------------------------------------------------------------------------------- |
|
||||
| 公开重发 | 对 `unknown` 或 `rejected` 投递的重发仍是受信适配器的动词；公开的重发需要自己的契约工作（H5 未决问题 4）。 |
|
||||
| WebShell 面板 | 列出投递任务并接受其取消。 |
|
||||
| 部分停止 | 停止 `partial` 投递的剩余分段需要改记录契约；目前没有适配器会产生 partial。 |
|
||||
@@ -112,10 +112,10 @@ Checked at `3bddf6e0bb` (contract `1.51.0`):
|
||||
|
||||
## Follow-up work
|
||||
|
||||
| Item | Scope |
|
||||
| ----------------------- | -------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `automation_run` | Carries no stop request. H6's per_run target owns it. |
|
||||
| Channel delivery cancel | #13746 F3: a delivery is not a task. Reuse the operation model ([H4f decision 10](2026-10-10-managed-task-cancel.md)). |
|
||||
| Child output | #13746 F2: link a child task to its child Session. |
|
||||
| WebShell panel | Show `workflow`, `background_shell` and `monitor` tasks and admit their cancel (`ManagedTaskPanel.tsx`, `use-managed-task-cancel.ts`). |
|
||||
| Stage F fault arms | FG8g real-stack arms for a lost cancel reply, a Harness down at finalize and a cancel after a Java restart. |
|
||||
| Item | Scope |
|
||||
| ----------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `automation_run` | Carries no stop request. H6's per_run target owns it. |
|
||||
| Channel delivery cancel | **Delivered in [#13746 F3](2026-10-17-managed-delivery-task-cancel.md):** a delivery projects as a task of its route Session (`channel_delivery`) and takes the same task cancel while no adapter has claimed it; decision 10's separate route is not needed. |
|
||||
| Child output | #13746 F2: link a child task to its child Session. |
|
||||
| WebShell panel | Show `workflow`, `background_shell` and `monitor` tasks and admit their cancel (`ManagedTaskPanel.tsx`, `use-managed-task-cancel.ts`). |
|
||||
| Stage F fault arms | FG8g real-stack arms for a lost cancel reply, a Harness down at finalize and a cancel after a Java restart. |
|
||||
@@ -112,10 +112,10 @@
|
||||
|
||||
## 后续工作
|
||||
|
||||
| 项 | 范围 |
|
||||
| ---------------- | --------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `automation_run` | 没有停止请求；归 H6 的 per_run 目标。 |
|
||||
| channel 投递取消 | #13746 F3：投递不是任务；复用操作模型（[H4f 决策 10](2026-10-10-managed-task-cancel.zh-CN.md)）。 |
|
||||
| child 输出 | #13746 F2：把 child 任务链接到其 child Session。 |
|
||||
| WebShell 面板 | 显示 `workflow`、`background_shell` 与 `monitor` 任务并接受其取消（`ManagedTaskPanel.tsx`、`use-managed-task-cancel.ts`）。 |
|
||||
| F 阶段故障分支 | FG8g 在真实栈上覆盖取消应答丢失、finalize 时 Harness 宕机、Java 重启后取消。 |
|
||||
| 项 | 范围 |
|
||||
| ---------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `automation_run` | 没有停止请求；归 H6 的 per_run 目标。 |
|
||||
| channel 投递取消 | **已在 [#13746 F3](2026-10-17-managed-delivery-task-cancel.zh-CN.md) 交付：** 投递投影为其路由 Session 的任务（`channel_delivery`），在尚无适配器认领时走同一条任务取消；不再需要决策 10 所说的独立路由。 |
|
||||
| child 输出 | #13746 F2：把 child 任务链接到其 child Session。 |
|
||||
| WebShell 面板 | 显示 `workflow`、`background_shell` 与 `monitor` 任务并接受其取消（`ManagedTaskPanel.tsx`、`use-managed-task-cancel.ts`）。 |
|
||||
| F 阶段故障分支 | FG8g 在真实栈上覆盖取消应答丢失、finalize 时 Harness 宕机、Java 重启后取消。 |
|
||||
+2
-1
@@ -13,7 +13,7 @@
|
||||
},
|
||||
"child_acceptance": null,
|
||||
"channel_route": null,
|
||||
"channel_delivery": null
|
||||
"channel_delivery": "channel_delivery"
|
||||
},
|
||||
"additionalRecordBodies": {
|
||||
"schedule": null,
|
||||
@@ -46,6 +46,7 @@
|
||||
"taskKinds": [
|
||||
"automation_run",
|
||||
"background_shell",
|
||||
"channel_delivery",
|
||||
"child_agent",
|
||||
"monitor",
|
||||
"workflow"
|
||||
|
||||
@@ -59,15 +59,19 @@ function merge(
|
||||
}
|
||||
|
||||
describe('managed-channel-record/1 shared contract', () => {
|
||||
it('registers both bodies as enabled, adapter-scoped domains that project no tasks', () => {
|
||||
it('registers both bodies as enabled, adapter-scoped domains, and projects only a delivery as a task', () => {
|
||||
// H5b/H5c: both domains are submittable, and the route policy's adapter
|
||||
// is the gate — only the email reference adapter is admitted.
|
||||
// is the gate — only the email reference adapter is admitted. A
|
||||
// delivery is a task of its route Session (#13746 F3); a route is not.
|
||||
for (const domain of ['channel_route', 'channel_delivery'] as const) {
|
||||
expect(MANAGED_SESSION_ENABLED_DOMAINS).toContain(domain);
|
||||
expect(
|
||||
MANAGED_EXTENSION_RECORD_BODIES[domain]!.taskKindOf({}),
|
||||
).toBeNull();
|
||||
}
|
||||
expect(
|
||||
MANAGED_EXTENSION_RECORD_BODIES.channel_route!.taskKindOf({}),
|
||||
).toBeNull();
|
||||
expect(
|
||||
MANAGED_EXTENSION_RECORD_BODIES.channel_delivery!.taskKindOf({}),
|
||||
).toBe('channel_delivery');
|
||||
expect(MANAGED_SESSION_ENABLED_CHANNEL_ADAPTERS).toEqual(['email']);
|
||||
});
|
||||
|
||||
|
||||
@@ -99,6 +99,7 @@ export const MANAGED_TASK_KINDS = [
|
||||
'background_shell',
|
||||
'monitor',
|
||||
'automation_run',
|
||||
'channel_delivery',
|
||||
] as const;
|
||||
export const MANAGED_TASK_STATES = [
|
||||
'pending',
|
||||
@@ -240,7 +241,7 @@ export const MANAGED_EXTENSION_RECORD_BODIES: Readonly<
|
||||
isSuccessor: isChannelRouteSuccessor,
|
||||
}),
|
||||
channel_delivery: Object.freeze({
|
||||
taskKindOf: () => null,
|
||||
taskKindOf: () => 'channel_delivery',
|
||||
parse: (value: unknown) => {
|
||||
const record = parseChannelDelivery(value);
|
||||
return { record, recordId: record.deliveryId, run: record.run };
|
||||
|
||||
@@ -404,7 +404,13 @@ describe('managed session authority channel records', () => {
|
||||
TRUSTED,
|
||||
);
|
||||
expect(last.revision).toBe(4);
|
||||
expect(authority.taskViews()).toEqual([]);
|
||||
// A delivery is a task of its route Session (#13746 F3).
|
||||
expect(authority.taskViews()).toEqual([
|
||||
expect.objectContaining({
|
||||
kind: 'channel_delivery',
|
||||
state: 'completed',
|
||||
}),
|
||||
]);
|
||||
expect(
|
||||
authority.extensionRecord('channel_delivery', 'delivery-1'),
|
||||
).toMatchObject({ revision: 4 });
|
||||
@@ -476,7 +482,12 @@ describe('managed session authority channel records', () => {
|
||||
expect(
|
||||
authority.extensionRecord('channel_delivery', 'delivery-1'),
|
||||
).toMatchObject({ revision: 1 });
|
||||
expect(authority.taskViews()).toEqual([]);
|
||||
expect(authority.taskViews()).toEqual([
|
||||
expect.objectContaining({
|
||||
kind: 'channel_delivery',
|
||||
state: 'pending',
|
||||
}),
|
||||
]);
|
||||
},
|
||||
{ create: false },
|
||||
);
|
||||
|
||||
+24
-6
@@ -55,7 +55,11 @@ import java.util.concurrent.atomic.AtomicBoolean;
|
||||
* Runtime Broker, whose worker answers only after the stop-marked finalize
|
||||
* settled the record: the record's own stop is the authority's statement,
|
||||
* a request revision for a Shell, the {@code stop_requested} settlement
|
||||
* for a Monitor, whose record carries no request.</li>
|
||||
* for a Monitor, whose record carries no request;</li>
|
||||
* <li>a planned channel delivery takes its cancel through the Hosted
|
||||
* Harness's channel operation route, whose {@code cancelRequested} flag is
|
||||
* the request: a planned one settles cancelled with it, and one an adapter
|
||||
* claimed meanwhile keeps sending with the flag recorded.</li>
|
||||
* </ul>
|
||||
* A delivery whose outcome stays unproven retries with the dispatch
|
||||
* backoff, bounded at {@link #ATTEMPT_BUDGET} claims; past it the
|
||||
@@ -227,6 +231,7 @@ public class TaskCancelCoordinator {
|
||||
"task_action_unavailable"), target);
|
||||
}
|
||||
if (target.body().path("stopRequested").asBoolean(false)
|
||||
|| target.body().path("cancelRequested").asBoolean(false)
|
||||
|| "stop_requested".equals(
|
||||
target.body().path("stopReason").textValue())) {
|
||||
return new Verdict(TaskCancelOutcome.completed(), target);
|
||||
@@ -386,10 +391,19 @@ public class TaskCancelCoordinator {
|
||||
}
|
||||
|
||||
/**
|
||||
* A child Session run's stop request onto the parent's journal, or a
|
||||
* background process's stop through the Broker; both replay-safe.
|
||||
* A child Session run's stop request onto the parent's journal, a
|
||||
* background process's stop through the Broker, or a delivery's cancel
|
||||
* onto its route Session's journal; all replay-safe.
|
||||
*/
|
||||
private void send(OperationRecord operation, TaskTarget target) {
|
||||
if ("channel_delivery".equals(target.kind())) {
|
||||
harness.runChannelOperation(operation.tenantId(),
|
||||
operation.sessionId(), Map.of(
|
||||
"operationId", attemptId(operation),
|
||||
"kind", "cancel_delivery",
|
||||
"deliveryId", target.recordId()));
|
||||
return;
|
||||
}
|
||||
if (BACKGROUND_TASK_KINDS.contains(target.kind())) {
|
||||
if (broker == null) {
|
||||
throw new IllegalStateException(
|
||||
@@ -401,15 +415,19 @@ public class TaskCancelCoordinator {
|
||||
return;
|
||||
}
|
||||
Map<String, Object> body = new LinkedHashMap<>();
|
||||
body.put("operationId", UUID.nameUUIDFromBytes(("qwen-task-cancel:\0"
|
||||
+ operation.operationId() + "\0" + operation.attemptCount())
|
||||
.getBytes(StandardCharsets.UTF_8)).toString());
|
||||
body.put("operationId", attemptId(operation));
|
||||
body.put("kind", "cancel");
|
||||
body.put("childRunId", target.recordId());
|
||||
harness.runChildOperation(operation.tenantId(),
|
||||
operation.sessionId(), body);
|
||||
}
|
||||
|
||||
private static String attemptId(OperationRecord operation) {
|
||||
return UUID.nameUUIDFromBytes(("qwen-task-cancel:\0"
|
||||
+ operation.operationId() + "\0" + operation.attemptCount())
|
||||
.getBytes(StandardCharsets.UTF_8)).toString();
|
||||
}
|
||||
|
||||
private static boolean runtimeNotHere(Throwable error) {
|
||||
for (Throwable cause = error; cause != null;
|
||||
cause = cause.getCause()) {
|
||||
|
||||
+14
-7
@@ -59,7 +59,7 @@ public final class ManagedExtensionProjection {
|
||||
body -> body.get("routeId").textValue(),
|
||||
ManagedChannelRecords::isRouteStart,
|
||||
ManagedChannelRecords::isRouteSuccessor)),
|
||||
entry("channel_delivery", new Body(record -> null,
|
||||
entry("channel_delivery", new Body(record -> "channel_delivery",
|
||||
ManagedChannelRecords::requireDelivery,
|
||||
body -> body.get("deliveryId").textValue(),
|
||||
ManagedChannelRecords::isDeliveryStart,
|
||||
@@ -113,7 +113,8 @@ public final class ManagedExtensionProjection {
|
||||
"running", "waiting", "completed", "failed", "cancelled",
|
||||
"degraded", "recovery_blocked");
|
||||
public static final List<String> TASK_KINDS = List.of("child_agent",
|
||||
"workflow", "background_shell", "monitor", "automation_run");
|
||||
"workflow", "background_shell", "monitor", "automation_run",
|
||||
"channel_delivery");
|
||||
public static final List<String> RUNTIME_STATES = List.of("unbound",
|
||||
"provisioning", "ready", "draining", "lost");
|
||||
/**
|
||||
@@ -124,11 +125,14 @@ public final class ManagedExtensionProjection {
|
||||
* made the close cascade's stop of a workflow row a shipped path. The
|
||||
* H3 kinds joined once their gates lifted (#13856): their cancel stops
|
||||
* the process through the Broker, and the worker's stop-marked end
|
||||
* settles the record {@code stop_requested} (A8). An automation run
|
||||
* has no stop path yet.
|
||||
* settles the record {@code stop_requested} (A8). A channel delivery
|
||||
* (#13746 F3) cancels only while no adapter has claimed it: its
|
||||
* cancelRequested flag is its stop request. An automation run has no
|
||||
* stop path yet.
|
||||
*/
|
||||
public static final Set<String> CANCELLABLE_TASK_KINDS = Set.of(
|
||||
"child_agent", "workflow", "background_shell", "monitor");
|
||||
"child_agent", "workflow", "background_shell", "monitor",
|
||||
"channel_delivery");
|
||||
/** Task states a cancel may still act on: the run has not ended and
|
||||
* is not waiting on recovery reconciliation. */
|
||||
private static final Set<String> CANCELLABLE_TASK_STATES = Set.of(
|
||||
@@ -231,13 +235,16 @@ public final class ManagedExtensionProjection {
|
||||
|
||||
/**
|
||||
* The actions a task supports now, the same for every caller:
|
||||
* {@code cancel} while a cancellable kind's run has not ended. The
|
||||
* {@code cancel} while a cancellable kind's run has not ended, and for
|
||||
* a channel delivery only while it is planned (pending): a claimed
|
||||
* delivery is already on its way to the provider. The
|
||||
* task view advertises this list and a new cancel admission rechecks
|
||||
* it under the Session lock.
|
||||
*/
|
||||
public static List<String> taskActions(String kind, String state) {
|
||||
return CANCELLABLE_TASK_KINDS.contains(kind)
|
||||
&& CANCELLABLE_TASK_STATES.contains(state)
|
||||
&& ("channel_delivery".equals(kind) ? "pending".equals(state)
|
||||
: CANCELLABLE_TASK_STATES.contains(state))
|
||||
? List.of("cancel") : List.of();
|
||||
}
|
||||
|
||||
|
||||
+12
-6
@@ -1364,6 +1364,11 @@ public class ManagedExtensionRecordStore {
|
||||
String deliveryState = delivery.isNull() ? null
|
||||
: delivery.get("state").textValue();
|
||||
long revision = previous == null ? 1 : previous.revision() + 1;
|
||||
// A row's task kind is fixed at its insert: a row written before
|
||||
// its domain projected tasks (a channel delivery planned before
|
||||
// #13746 F3) stays a non-task on every later revision.
|
||||
String taskKind = previous == null ? body.taskKindOf().apply(record)
|
||||
: previous.taskKind();
|
||||
if (previous == null) {
|
||||
try {
|
||||
jdbc.update("INSERT INTO ma_session_record"
|
||||
@@ -1379,8 +1384,8 @@ public class ManagedExtensionRecordStore {
|
||||
+ " ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||||
scopeKey, recordKey, tenantId, workspaceId, sessionId,
|
||||
domain, recordId, operationHash, revision, resourceId,
|
||||
body.taskKindOf().apply(record),
|
||||
body.taskKindOf().apply(record) == null ? null : projection.state(), projection.runtimeState(),
|
||||
taskKind, taskKind == null ? null : projection.state(),
|
||||
projection.runtimeState(),
|
||||
projection.definitionRevision(), deliveryTarget,
|
||||
deliveryState, projection.createdAt(),
|
||||
projection.startedAt(), projection.settledAt(), sequence,
|
||||
@@ -1403,7 +1408,7 @@ public class ManagedExtensionRecordStore {
|
||||
+ " = ?, runtime_state = ?, definition_revision = ?, delivery_target = ?,"
|
||||
+ " delivery_state = ?, started_at = ?, settled_at = ? WHERE record_kind ="
|
||||
+ " 'DOMAIN' AND session_scope_key = ? AND record_key = ?",
|
||||
revision, resourceId, body.taskKindOf().apply(record) == null ? null : projection.state(),
|
||||
revision, resourceId, taskKind == null ? null : projection.state(),
|
||||
projection.runtimeState(),
|
||||
projection.definitionRevision(), deliveryTarget,
|
||||
deliveryState, projection.startedAt(),
|
||||
@@ -1418,7 +1423,7 @@ public class ManagedExtensionRecordStore {
|
||||
jdbc.update(
|
||||
"INSERT INTO ma_resource_ref (session_scope_key, owner_kind, owner_key, owner_revision, resource_key, tenant_id, workspace_id, session_id, resource_id, gmt_create) VALUES (?, 'SESSION_RECORD', ?, ?, ?, ?, ?, ?, ?, " + ManagedSchemaTime.SQL_NOW + ")",
|
||||
scopeKey, recordKey, revision, ManagedSchemaKeys.key(scopeKey, resourceId), tenantId, workspaceId, sessionId, resourceId);
|
||||
if (body.taskKindOf().apply(record) != null && (previous == null
|
||||
if (taskKind != null && (previous == null
|
||||
|| !Objects.equals(previous.projection(), projection))) {
|
||||
String taskId = ManagedExtensionProjection.taskId(recordKey);
|
||||
if (isBeingDeleted(tenantId, sessionId)) {
|
||||
@@ -1461,7 +1466,8 @@ public class ManagedExtensionRecordStore {
|
||||
throws SQLException {
|
||||
return new StoredRow(result.getString("domain"),
|
||||
result.getString("record_id"), result.getLong("revision"),
|
||||
result.getString("record_resource_id"), projection(result));
|
||||
result.getString("record_resource_id"),
|
||||
result.getString("task_kind"), projection(result));
|
||||
}
|
||||
|
||||
private static TaskProjection projection(ResultSet result)
|
||||
@@ -1911,7 +1917,7 @@ public class ManagedExtensionRecordStore {
|
||||
}
|
||||
|
||||
private record StoredRow(String domain, String recordId, long revision,
|
||||
String resourceId, TaskProjection projection) {
|
||||
String resourceId, String taskKind, TaskProjection projection) {
|
||||
}
|
||||
|
||||
/** Whether the event id sits in the domain's own reserved {@code <domain>:<n>} namespace. */
|
||||
|
||||
+5
-4
@@ -2,8 +2,8 @@
|
||||
"openapi": "3.1.0",
|
||||
"info": {
|
||||
"title": "Qwen Managed Agent Public and WebShell API",
|
||||
"version": "1.56.0",
|
||||
"description": "Canonical contract for the public Managed Agent resources and the WebShell adapter. partial means that the route exists but generated types and full contract tests are pending; planned means that the route or field is not implemented. Workspace context and cwd operations are v1.9 target contracts, not current server capabilities. v1.10 adds planned command admission, Actions, lifecycle and actor authorization. v1.12 refines planned W0 creation discovery, actor-specific create hints and explicit default selection. Filter planned fields as well as routes from production SDKs. v1.13 moves this contract into the Qwen Code repository as its single source and records the shipped rename, archive, unarchive and delete routes as partial. v1.14 integrates W0d workspace discovery, the WebShell workspace lookup, and empty-session binding as partial; workspace_binding/workspaceBinding advertises this narrow flow without enabling workspace execution or context switching. v1.15 adds the archived, archiving and deleting Session statuses and the cancelling Turn status, requires request_id in error envelopes, declares the error responses that the server returns, and marks Session create, list and get on both surfaces implemented. v1.16 adds the planned Stage H task contract (SessionTaskView as PublicTask, task list, detail, events and cancel) and names the MCP catalog, hook catalog, automation and channel resources as planned; their shapes arrive with Stages H1 to H6. v1.17 implements event replay: events carry their schema and projection versions and a top-level Item and Part identity, JSON event pages return has_more and next_cursor and accept limits up to 1000, and a persisted replay floor answers expired cursors with 409 cursor_expired in JSON and one agent.session.resync_required frame over SSE; it marks the public event query and stream, the WebShell event stream and the WebShell transcript implemented. v1.18 implements the durable Session lifecycle (Stage D4): close, archive and delete answer 202 with a command operation on both surfaces, an archive requires a closed Session, a deleted Session leaves a tombstone whose operations stay readable, and the operation query serves them; unarchive restores a closed Session, and Session capabilities advertise session_lifecycle. v1.19 serves the task list and detail on both surfaces as partial (Stage H0c): the Session store projects them from the Stage H records the Session authority commits, a task.updated Session event announces each change of a task's view, and Session capabilities advertise tasks. Task events and cancel stay planned until a slice's tasks produce output and accept a cancel. v1.20 implements the Turn read model (Stage D5): the public Turn list and detail read a Session's Turns without their input, newest first by creation time in milliseconds and then Turn ID, with an opaque cursor; a deleted Session's Turns are not readable, and the durable admission fields of a Turn stay planned until D7. v1.21 declares on the task list, detail and event routes of both surfaces the 403 actor_scope_mismatch that the tenant filter answers on every /v1/agents/ and WebShell route, as the Session and Turn reads already do, and the shared Forbidden response names it; a caller that cannot read a task still gets 404. v1.22 adds deployment-opted-in creation with an initial Workspace file-tool Turn (G0) on the public and WebShell surfaces; the fixed server-owned profile does not enable later Workspace operations. v1.23 settles the planned task event and cancel semantics: a durable retention floor, committed-prefix publication, stable cursors, Artifact visibility before expiry, bounded backlog under archival failure, ordered idempotent cancellation and command outcomes. Clients tolerate unknown optional event fields across minor versions. Task events and cancel remain planned. v1.24 serves the read-only Session MCP catalog as partial (Stage H1), projecting display metadata and schemas from committed records without Runtime identities, credentials or production profile enablement. v1.25 implements permission Actions (Stage D6b) on both surfaces: list, get and respond, projected from the Session's action.changed records; a WebShell permission Action carries inputRevision, policyRevision, functionCallId, toolName and expiresAt, its option ids allow and deny are stable, a Turn has at most one requested approval at a time, and a response is a durable command operation carrying requestId. v1.26 declares the tenant filter's 403 actor_scope_mismatch on every covered public and WebShell route, including planned routes, and pins the refusal code in contract traffic. v1.27 adds O3 durable tool-result projection, artifact metadata, immutable range/streaming content and WebShell result reads. Artifact capability requires a configured O3 reader and a Workspace-bound Session; execution admission remains independent. v1.28 admits later Turns of a Workspace-bound Session for the SLine truncated
|
||||
"version": "1.57.0",
|
||||
"description": "Canonical contract for the public Managed Agent resources and the WebShell adapter. partial means that the route exists but generated types and full contract tests are pending; planned means that the route or field is not implemented. Workspace context and cwd operations are v1.9 target contracts, not current server capabilities. v1.10 adds planned command admission, Actions, lifecycle and actor authorization. v1.12 refines planned W0 creation discovery, actor-specific create hints and explicit default selection. Filter planned fields as well as routes from production SDKs. v1.13 moves this contract into the Qwen Code repository as its single source and records the shipped rename, archive, unarchive and delete routes as partial. v1.14 integrates W0d workspace discovery, the WebShell workspace lookup, and empty-session binding as partial; workspace_binding/workspaceBinding advertises this narrow flow without enabling workspace execution or context switching. v1.15 adds the archived, archiving and deleting Session statuses and the cancelling Turn status, requires request_id in error envelopes, declares the error responses that the server returns, and marks Session create, list and get on both surfaces implemented. v1.16 adds the planned Stage H task contract (SessionTaskView as PublicTask, task list, detail, events and cancel) and names the MCP catalog, hook catalog, automation and channel resources as planned; their shapes arrive with Stages H1 to H6. v1.17 implements event replay: events carry their schema and projection versions and a top-level Item and Part identity, JSON event pages return has_more and next_cursor and accept limits up to 1000, and a persisted replay floor answers expired cursors with 409 cursor_expired in JSON and one agent.session.resync_required frame over SSE; it marks the public event query and stream, the WebShell event stream and the WebShell transcript implemented. v1.18 implements the durable Session lifecycle (Stage D4): close, archive and delete answer 202 with a command operation on both surfaces, an archive requires a closed Session, a deleted Session leaves a tombstone whose operations stay readable, and the operation query serves them; unarchive restores a closed Session, and Session capabilities advertise session_lifecycle. v1.19 serves the task list and detail on both surfaces as partial (Stage H0c): the Session store projects them from the Stage H records the Session authority commits, a task.updated Session event announces each change of a task's view, and Session capabilities advertise tasks. Task events and cancel stay planned until a slice's tasks produce output and accept a cancel. v1.20 implements the Turn read model (Stage D5): the public Turn list and detail read a Session's Turns without their input, newest first by creation time in milliseconds and then Turn ID, with an opaque cursor; a deleted Session's Turns are not readable, and the durable admission fields of a Turn stay planned until D7. v1.21 declares on the task list, detail and event routes of both surfaces the 403 actor_scope_mismatch that the tenant filter answers on every /v1/agents/ and WebShell route, as the Session and Turn reads already do, and the shared Forbidden response names it; a caller that cannot read a task still gets 404. v1.22 adds deployment-opted-in creation with an initial Workspace file-tool Turn (G0) on the public and WebShell surfaces; the fixed server-owned profile does not enable later Workspace operations. v1.23 settles the planned task event and cancel semantics: a durable retention floor, committed-prefix publication, stable cursors, Artifact visibility before expiry, bounded backlog under archival failure, ordered idempotent cancellation and command outcomes. Clients tolerate unknown optional event fields across minor versions. Task events and cancel remain planned. v1.24 serves the read-only Session MCP catalog as partial (Stage H1), projecting display metadata and schemas from committed records without Runtime identities, credentials or production profile enablement. v1.25 implements permission Actions (Stage D6b) on both surfaces: list, get and respond, projected from the Session's action.changed records; a WebShell permission Action carries inputRevision, policyRevision, functionCallId, toolName and expiresAt, its option ids allow and deny are stable, a Turn has at most one requested approval at a time, and a response is a durable command operation carrying requestId. v1.26 declares the tenant filter's 403 actor_scope_mismatch on every covered public and WebShell route, including planned routes, and pins the refusal code in contract traffic. v1.27 adds O3 durable tool-result projection, artifact metadata, immutable range/streaming content and WebShell result reads. Artifact capability requires a configured O3 reader and a Workspace-bound Session; execution admission remains independent. v1.28 admits later Turns of a Workspace-bound Session for the SLine truncated
|
||||
},
|
||||
"servers": [
|
||||
{
|
||||
@@ -8436,7 +8436,8 @@
|
||||
"workflow",
|
||||
"background_shell",
|
||||
"monitor",
|
||||
"automation_run"
|
||||
"automation_run",
|
||||
"channel_delivery"
|
||||
],
|
||||
"x-qwen-implementation-status": "partial"
|
||||
},
|
||||
@@ -8465,7 +8466,7 @@
|
||||
"type": "string",
|
||||
"enum": ["cancel", "send_input", "read_output"],
|
||||
"x-qwen-implementation-status": "partial",
|
||||
"description": "Actions the task supports now, the same for every caller. cancel: the cancel route accepts a new command for this task (since 1.40 for child_agent, 1.46 for workflow and 1.54 for background_shell and monitor, tasks of those kinds in pending, running, waiting or degraded, in an active Session); whether a caller may use it is an authorization check (403 on the cancel route). send_input: reserved for a later capability route. read_output: the task events route returns output events for this task to any caller that can read it. read_output does not change during the task's life, and a task without it produces no output events: its output goes only to Artifacts, so the events route never filters out events that exist."
|
||||
"description": "Actions the task supports now, the same for every caller. cancel: the cancel route accepts a new command for this task (since 1.40 for child_agent, 1.46 for workflow and 1.54 for background_shell and monitor, tasks of those kinds in pending, running, waiting or degraded, and since 1.57 for channel_delivery, a delivery still pending because no adapter has claimed it, each in an active Session); whether a caller may use it is an authorization check (403 on the cancel route). send_input: reserved for a later capability route. read_output: the task events route returns output events for this task to any caller that can read it. read_output does not change during the task's life, and a task without it produces no output events: its output goes only to Artifacts, so the events route never filters out events that exist."
|
||||
},
|
||||
"PublicTask": {
|
||||
"type": "object",
|
||||
|
||||
+132
-2
@@ -119,7 +119,8 @@ class HostedChannelFaultGatesIT {
|
||||
@Timeout(600)
|
||||
void fg7dChannelCancellationsSettlePhysicallyOnMySql() throws Exception {
|
||||
runLetter("fg7d", List.of("cancel-planned", "cancel-sending",
|
||||
"cancel-replay", "cancel-unsettled"), null);
|
||||
"cancel-replay", "cancel-unsettled", "task-cancel-planned",
|
||||
"task-cancel-claimed"), null);
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -254,6 +255,7 @@ class HostedChannelFaultGatesIT {
|
||||
"receipt-reply", "route-admit", "delivery-commit",
|
||||
"commit-reply-restart", "cancel-planned",
|
||||
"cancel-sending", "cancel-replay", "cancel-unsettled",
|
||||
"task-cancel-planned", "task-cancel-claimed",
|
||||
"delivery-page-resume", "delivery-state-progress",
|
||||
"provider-ambiguous", "out-of-plan-ordinal",
|
||||
"receipt-never-arrived")) {
|
||||
@@ -507,7 +509,7 @@ class HostedChannelFaultGatesIT {
|
||||
registration.setAsyncSupported(true);
|
||||
registration.setDispatcherTypes(DispatcherType.REQUEST,
|
||||
DispatcherType.ASYNC);
|
||||
registration.addUrlPatterns("/v1/agent-channels/*");
|
||||
registration.addUrlPatterns("/v1/agent-channels/*", "/v1/agents/*");
|
||||
return registration;
|
||||
}
|
||||
|
||||
@@ -585,6 +587,40 @@ class HostedChannelFaultGatesIT {
|
||||
return answer;
|
||||
}
|
||||
|
||||
/** The route Session's one delivery task, read publicly. */
|
||||
private JsonNode deliveryTask(String tenant, String sessionId)
|
||||
throws Exception {
|
||||
HttpResponse<String> response = publicCall(tenant,
|
||||
"/v1/agents/sessions/" + sessionId + "/tasks", null);
|
||||
assertThat(response.statusCode()).as(response.body()).isEqualTo(200);
|
||||
List<JsonNode> tasks = new java.util.ArrayList<>();
|
||||
json.readTree(response.body()).path("data").forEach(task -> {
|
||||
if ("channel_delivery".equals(task.path("kind").asText())) {
|
||||
tasks.add(task);
|
||||
}
|
||||
});
|
||||
assertThat(tasks).as("delivery tasks").hasSize(1);
|
||||
return tasks.getFirst();
|
||||
}
|
||||
|
||||
/** A public GET, or a POST under {@code key} when one is given. */
|
||||
private HttpResponse<String> publicCall(String tenant, String route,
|
||||
String key) throws Exception {
|
||||
HttpRequest.Builder request = HttpRequest.newBuilder(URI.create(
|
||||
"http://127.0.0.1:" + spring.getWebServer().getPort()
|
||||
+ route)).timeout(Duration.ofSeconds(20))
|
||||
.header("X-Qwen-Tenant-Id", tenant)
|
||||
.header("Accept", "application/json");
|
||||
if (key != null) {
|
||||
request.header("Idempotency-Key", key)
|
||||
.POST(HttpRequest.BodyPublishers.noBody());
|
||||
} else {
|
||||
request.GET();
|
||||
}
|
||||
return HttpClient.newHttpClient().send(request.build(),
|
||||
HttpResponse.BodyHandlers.ofString());
|
||||
}
|
||||
|
||||
/** Park one pull mid-send; resume it by creating {@code release}. */
|
||||
private HostedChannelGateSupport.DriverLaunch startParkedPull(
|
||||
Path caseDir, Map<String, Object> base, Path release)
|
||||
@@ -811,6 +847,100 @@ class HostedChannelFaultGatesIT {
|
||||
assertThat(support.ledgerRows(tenant, channelId))
|
||||
.as("nothing ever reaches the ledger").isZero();
|
||||
}
|
||||
// #13746 F3: the public task cancel of a delivery. A planned
|
||||
// one is a pending task that advertises cancel, and the cancel
|
||||
// settles it before any send; once an adapter claims it, the
|
||||
// task runs and refuses a new cancel.
|
||||
case "task-cancel-planned" -> {
|
||||
JsonNode admitted = support.drive(caseDir, "admit", base);
|
||||
String sessionId = support.channelSession(tenant, channelId);
|
||||
String deliveryId = admitted.path("admissionInputIds").get(0)
|
||||
.asText() + ":reply";
|
||||
awaitPlanned(tenant, sessionId, deliveryId);
|
||||
JsonNode task = deliveryTask(tenant, sessionId);
|
||||
assertThat(task.path("state").asText()).isEqualTo("pending");
|
||||
assertThat(task.path("action_capabilities").toString())
|
||||
.isEqualTo("[\"cancel\"]");
|
||||
String route = "/v1/agents/sessions/" + sessionId + "/tasks/"
|
||||
+ task.path("id").asText() + "/cancel";
|
||||
HttpResponse<String> cancel = publicCall(tenant, route,
|
||||
"fg7-task-cancel");
|
||||
assertThat(cancel.statusCode()).as(cancel.body())
|
||||
.isEqualTo(202);
|
||||
String operationId = json.readTree(cancel.body()).path("id")
|
||||
.asText();
|
||||
await("task cancel completed", 90, () -> {
|
||||
try {
|
||||
return "completed".equals(json.readTree(publicCall(
|
||||
tenant, "/v1/agents/sessions/" + sessionId
|
||||
+ "/operations/" + operationId,
|
||||
null).body()).path("status").asText())
|
||||
? true : null;
|
||||
} catch (Exception failure) {
|
||||
throw new IllegalStateException(failure);
|
||||
}
|
||||
});
|
||||
JsonNode replay = json.readTree(publicCall(tenant, route,
|
||||
"fg7-task-cancel").body());
|
||||
assertThat(replay.path("id").asText()).isEqualTo(operationId);
|
||||
assertThat(replay.path("replayed").asBoolean()).isTrue();
|
||||
JsonNode settled = deliveryTask(tenant, sessionId);
|
||||
assertThat(settled.path("state").asText())
|
||||
.isEqualTo("cancelled");
|
||||
assertThat(settled.path("action_capabilities")).isEmpty();
|
||||
assertThat(support.deliveryRecord(tenant, sessionId,
|
||||
deliveryId).path("cancelRequested").asBoolean())
|
||||
.isTrue();
|
||||
JsonNode pull = support.drive(caseDir, "pull", base,
|
||||
Map.of("expectDelivery", false));
|
||||
assertThat(HostedChannelGateSupport.relayCountOk(pull,
|
||||
"claim"))
|
||||
.as("the outbox pull reached the control plane")
|
||||
.isGreaterThanOrEqualTo(1);
|
||||
HostedChannelGateSupport.assertSendsEmpty(pull,
|
||||
"a cancelled delivery task never sends");
|
||||
assertThat(support.ledgerRows(tenant, channelId))
|
||||
.as("nothing ever reaches the ledger").isZero();
|
||||
}
|
||||
case "task-cancel-claimed" -> {
|
||||
JsonNode admitted = support.drive(caseDir, "admit", base);
|
||||
String sessionId = support.channelSession(tenant, channelId);
|
||||
String deliveryId = admitted.path("admissionInputIds").get(0)
|
||||
.asText() + ":reply";
|
||||
awaitPlanned(tenant, sessionId, deliveryId);
|
||||
Path release = caseDir.resolve("smtp-release");
|
||||
HostedChannelGateSupport.DriverLaunch launch = startParkedPull(
|
||||
caseDir, base, release);
|
||||
Process driver = launch.process();
|
||||
try {
|
||||
awaitSending(tenant, sessionId, deliveryId, channelId);
|
||||
JsonNode task = deliveryTask(tenant, sessionId);
|
||||
assertThat(task.path("state").asText())
|
||||
.isEqualTo("running");
|
||||
assertThat(task.path("action_capabilities")).isEmpty();
|
||||
HttpResponse<String> cancel = publicCall(tenant,
|
||||
"/v1/agents/sessions/" + sessionId + "/tasks/"
|
||||
+ task.path("id").asText() + "/cancel",
|
||||
"fg7-task-cancel-claimed");
|
||||
assertThat(cancel.statusCode()).as(cancel.body())
|
||||
.isEqualTo(409);
|
||||
assertThat(json.readTree(cancel.body()).path("error")
|
||||
.path("code").asText())
|
||||
.isEqualTo("task_action_unavailable");
|
||||
Files.writeString(release, "go");
|
||||
assertThat(driver.waitFor(120, TimeUnit.SECONDS))
|
||||
.isTrue();
|
||||
assertThat(driver.exitValue()).isZero();
|
||||
assertThat(support.resultsOf(caseDir, launch.tag())
|
||||
.path("sends")).hasSize(1);
|
||||
assertThat(deliveryTask(tenant, sessionId).path("state")
|
||||
.asText()).isEqualTo("completed");
|
||||
} finally {
|
||||
if (driver.isAlive()) {
|
||||
driver.destroyForcibly();
|
||||
}
|
||||
}
|
||||
}
|
||||
case "cancel-sending" -> {
|
||||
JsonNode admitted = support.drive(caseDir, "admit", base);
|
||||
String sessionId = support.channelSession(tenant, channelId);
|
||||
|
||||
+5
-2
@@ -12,7 +12,6 @@ import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
|
||||
class ManagedChannelRecordContractTest {
|
||||
@@ -25,7 +24,11 @@ class ManagedChannelRecordContractTest {
|
||||
String domain = fixture.get("domain").textValue();
|
||||
String id = fixture.get("id").textValue();
|
||||
var body = ManagedExtensionProjection.RECORD_BODIES.get(domain);
|
||||
assertNull(body.taskKindOf().apply(null));
|
||||
// A delivery projects as a task of its route Session (#13746
|
||||
// F3); a route projects none.
|
||||
assertEquals("channel_delivery".equals(domain)
|
||||
? "channel_delivery" : null,
|
||||
body.taskKindOf().apply(null), id);
|
||||
JsonNode record = merge(fixtures.get("templates").get(domain), fixture.get("patch"));
|
||||
if (fixture.get("valid").booleanValue()) {
|
||||
body.require().accept(record);
|
||||
|
||||
+120
@@ -1027,6 +1027,126 @@ class ManagedExtensionRecordStoreTest {
|
||||
assertThat(records.listRecords(TENANT, sessionId, "channel_delivery").getFirst().path("run").path("delivery").path("state").asText()).isEqualTo("unknown");
|
||||
}
|
||||
|
||||
// #13746 F3: a channel delivery projects as a task of its route
|
||||
// Session, advertising cancel only while no adapter has claimed it.
|
||||
@Test
|
||||
void aChannelDeliveryProjectsAsATaskCancellableOnlyWhilePlanned()
|
||||
throws Exception {
|
||||
JsonNode templates = ManagedChannelRecordContractTest.fixtures()
|
||||
.required("templates");
|
||||
CommitResource policy = channelResource("policy-task", "channel-policy", "{}");
|
||||
CommitResource result = channelResource("reply-task", "managed-tool-result", "reply");
|
||||
CommitResource segment1 = channelResource("part1-task", "channel-delivery-segment", "one");
|
||||
CommitResource segment2 = channelResource("part2-task", "channel-delivery-segment", "two");
|
||||
String sessionId = UUID.randomUUID().toString();
|
||||
jdbc.update("INSERT INTO ma_session (tenant_id,"
|
||||
+ " session_id, agent_id, status, gmt_create, gmt_modified)"
|
||||
+ " VALUES (?, ?, 'qwen-code', 'ACTIVE', 1, 1)", TENANT,
|
||||
sessionId);
|
||||
ExtensionRecordJournal journal = journal(sessionId);
|
||||
commitDomain(journal, "route-task", "channel_route",
|
||||
channelRoute(templates, policy), List.of(policy));
|
||||
ObjectNode delivery = channelDelivery(templates, result, segment1,
|
||||
segment2);
|
||||
commitDomain(journal, "planned-task", "channel_delivery", delivery,
|
||||
List.of(result, segment1, segment2));
|
||||
String taskId = ManagedExtensionProjection.taskId(
|
||||
ManagedExtensionProjection.recordKey(sessionId,
|
||||
"channel_delivery",
|
||||
delivery.path("deliveryId").asText()));
|
||||
var planned = tasks.getPublicTask(TENANT, TENANT, sessionId, taskId);
|
||||
assertThat(planned.kind()).isEqualTo("channel_delivery");
|
||||
assertThat(planned.state()).isEqualTo("pending");
|
||||
assertThat(planned.runtimeState()).isNull();
|
||||
assertThat(planned.actionCapabilities()).containsExactly("cancel");
|
||||
assertThat(records.findTaskTarget(TENANT, sessionId, taskId)
|
||||
.orElseThrow().recordId())
|
||||
.isEqualTo(delivery.path("deliveryId").asText());
|
||||
// A claimed delivery is on its way to the provider: no cancel.
|
||||
ObjectNode sending = delivery.deepCopy();
|
||||
((ObjectNode) sending.path("run")).put("state", "running");
|
||||
((ObjectNode) sending.path("run").path("delivery"))
|
||||
.put("state", "sending");
|
||||
commitDomain(journal, "sending-task", "channel_delivery", sending,
|
||||
List.of());
|
||||
var claimed = tasks.getPublicTask(TENANT, TENANT, sessionId, taskId);
|
||||
assertThat(claimed.state()).isEqualTo("running");
|
||||
assertThat(claimed.actionCapabilities()).isEmpty();
|
||||
// A planned one the cancel settled reads cancelled, with its
|
||||
// request on the record the delivery verdict reads.
|
||||
String otherId = delivery.path("deliveryId").asText() + "-other";
|
||||
ObjectNode other = delivery.deepCopy();
|
||||
other.put("deliveryId", otherId);
|
||||
((ObjectNode) other.path("run")).put("effectId", otherId)
|
||||
.put("deliveryId", otherId);
|
||||
commitDomain(journal, "other-planned", "channel_delivery", other,
|
||||
List.of());
|
||||
ObjectNode cancelled = other.deepCopy();
|
||||
cancelled.put("cancelRequested", true);
|
||||
((ObjectNode) cancelled.path("run")).put("state", "cancelled");
|
||||
((ObjectNode) cancelled.path("run").path("delivery"))
|
||||
.put("state", "cancelled");
|
||||
commitDomain(journal, "other-cancelled", "channel_delivery",
|
||||
cancelled, List.of());
|
||||
String otherTask = ManagedExtensionProjection.taskId(
|
||||
ManagedExtensionProjection.recordKey(sessionId,
|
||||
"channel_delivery", otherId));
|
||||
var settled = tasks.getPublicTask(TENANT, TENANT, sessionId,
|
||||
otherTask);
|
||||
assertThat(settled.state()).isEqualTo("cancelled");
|
||||
assertThat(settled.actionCapabilities()).isEmpty();
|
||||
assertThat(records.findTaskTarget(TENANT, sessionId, otherTask)
|
||||
.orElseThrow().body().path("cancelRequested").asBoolean())
|
||||
.isTrue();
|
||||
}
|
||||
|
||||
// #13746 F3 decision 2: a delivery row written before deliveries
|
||||
// projected tasks keeps no task kind, so a later revision writes no
|
||||
// task state and journals no task event for it.
|
||||
@Test
|
||||
void aDeliveryRowWrittenBeforeTheTaskProjectionStaysANonTask()
|
||||
throws Exception {
|
||||
JsonNode templates = ManagedChannelRecordContractTest.fixtures()
|
||||
.required("templates");
|
||||
CommitResource policy = channelResource("policy-legacy", "channel-policy", "{}");
|
||||
CommitResource result = channelResource("reply-legacy", "managed-tool-result", "reply");
|
||||
CommitResource segment1 = channelResource("part1-legacy", "channel-delivery-segment", "one");
|
||||
CommitResource segment2 = channelResource("part2-legacy", "channel-delivery-segment", "two");
|
||||
String sessionId = UUID.randomUUID().toString();
|
||||
ExtensionRecordJournal journal = journal(sessionId);
|
||||
commitDomain(journal, "route-legacy", "channel_route",
|
||||
channelRoute(templates, policy), List.of(policy));
|
||||
ObjectNode delivery = channelDelivery(templates, result, segment1,
|
||||
segment2);
|
||||
commitDomain(journal, "planned-legacy", "channel_delivery", delivery,
|
||||
List.of(result, segment1, segment2));
|
||||
String recordKey = ManagedExtensionProjection.recordKey(sessionId,
|
||||
"channel_delivery", delivery.path("deliveryId").asText());
|
||||
// The row as a server before the projection wrote it.
|
||||
jdbc.update("UPDATE ma_session_record SET task_kind = NULL,"
|
||||
+ " task_state = NULL WHERE record_key = ?", recordKey);
|
||||
jdbc.update("DELETE FROM ma_public_event WHERE tenant_id = ? AND"
|
||||
+ " session_id = ? AND stream_kind = 'TASK'", TENANT,
|
||||
sessionId);
|
||||
ObjectNode sending = delivery.deepCopy();
|
||||
((ObjectNode) sending.path("run")).put("state", "running");
|
||||
((ObjectNode) sending.path("run").path("delivery"))
|
||||
.put("state", "sending");
|
||||
commitDomain(journal, "sending-legacy", "channel_delivery", sending,
|
||||
List.of());
|
||||
Map<String, Object> row = jdbc.queryForMap("SELECT task_kind,"
|
||||
+ " task_state, delivery_state FROM ma_session_record WHERE"
|
||||
+ " record_key = ?", recordKey);
|
||||
assertThat(row.get("delivery_state")).isEqualTo("sending");
|
||||
assertThat(row.get("task_kind")).isNull();
|
||||
assertThat(row.get("task_state")).isNull();
|
||||
assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_public_event"
|
||||
+ " WHERE tenant_id = ? AND session_id = ? AND stream_kind ="
|
||||
+ " 'TASK'", Integer.class, TENANT, sessionId)).isZero();
|
||||
assertThat(records.findTaskTarget(TENANT, sessionId,
|
||||
ManagedExtensionProjection.taskId(recordKey))).isEmpty();
|
||||
}
|
||||
|
||||
@Test
|
||||
void verifiesTheChannelResourceClosure() throws Exception {
|
||||
JsonNode templates = ManagedChannelRecordContractTest.fixtures()
|
||||
|
||||
+23
@@ -58,6 +58,7 @@ class ManagedTaskCancelOperationTest {
|
||||
private static final String WORKFLOW = "task_" + "e".repeat(64);
|
||||
private static final String SHELL = "task_" + "f".repeat(64);
|
||||
private static final String MONITOR = "task_" + "9".repeat(64);
|
||||
private static final String DELIVERY = "task_" + "8".repeat(64);
|
||||
private final AtomicLong now = new AtomicLong(1_000_000);
|
||||
private ManagedAgentStore store;
|
||||
private JdbcTemplate jdbc;
|
||||
@@ -237,6 +238,28 @@ class ManagedTaskCancelOperationTest {
|
||||
}
|
||||
}
|
||||
|
||||
// #13746 F3: a channel delivery admits a cancel only while pending —
|
||||
// planned, with no adapter claim — and the retained key replays.
|
||||
@Test
|
||||
void aDeliveryTaskAdmitsCancelOnlyWhilePending() {
|
||||
task(DELIVERY, "channel_delivery", "channel_delivery", "pending");
|
||||
OperationAdmission first = begin("delivery-1", "digest-1", DELIVERY);
|
||||
assertThat(first.replayed()).isFalse();
|
||||
assertThat(first.operation().taskId()).isEqualTo(DELIVERY);
|
||||
settle(first.operation(), TaskCancelOutcome.completed());
|
||||
for (String state : new String[] {"running", "waiting", "degraded",
|
||||
"completed", "failed", "cancelled", "recovery_blocked"}) {
|
||||
jdbc.update("UPDATE ma_session_record SET"
|
||||
+ " task_state = ? WHERE record_key = ?", state,
|
||||
"8".repeat(64));
|
||||
refused(() -> begin("delivery-" + state, "digest-" + state,
|
||||
DELIVERY), HttpStatus.CONFLICT,
|
||||
"task_action_unavailable");
|
||||
}
|
||||
assertThat(begin("delivery-1", "digest-1", DELIVERY).replayed())
|
||||
.isTrue();
|
||||
}
|
||||
|
||||
// A bound Session under storage migration admits no new cancel, as
|
||||
// its sibling bound admissions refuse; a retained key still replays.
|
||||
@Test
|
||||
|
||||
+56
@@ -607,6 +607,54 @@ class TaskCancelCoordinatorTest {
|
||||
anyLong());
|
||||
}
|
||||
|
||||
// #13746 F3: a pending delivery takes its cancel through its route
|
||||
// Session's channel operation route; its cancelRequested flag is the
|
||||
// request the outcome is read from.
|
||||
@Test
|
||||
void aPendingDeliveryTakesItsCancelThroughTheChannelRoute() {
|
||||
List<Map<String, Object>> channel = new CopyOnWriteArrayList<>();
|
||||
when(harness.runChannelOperation(eq(TENANT), eq(SESSION), any()))
|
||||
.thenAnswer(args -> {
|
||||
channel.add(Map.copyOf(args.getArgument(2)));
|
||||
target.set(delivery("cancelled", true));
|
||||
return Map.of();
|
||||
});
|
||||
target.set(delivery("pending", false));
|
||||
coordinator.dispatch(TENANT, SESSION, OPERATION);
|
||||
assertThat(sent).isEmpty();
|
||||
assertThat(channel).hasSize(1);
|
||||
assertThat(channel.getFirst()).containsEntry("kind", "cancel_delivery")
|
||||
.containsEntry("deliveryId", "delivery-1")
|
||||
.containsKey("operationId");
|
||||
verify(store).settleTaskCancel(eq(TENANT), eq(SESSION),
|
||||
eq(OPERATION), anyString(), eq(1L),
|
||||
eq(TaskCancelOutcome.completed()), eq(0L));
|
||||
}
|
||||
|
||||
// An adapter that claimed it first keeps sending it: the request is
|
||||
// recorded and the delivery keeps its natural outcome; one that was
|
||||
// delivered without a request fails the cancel definitively.
|
||||
@Test
|
||||
void aDeliveryClaimedFirstKeepsItsOutcome() {
|
||||
when(harness.runChannelOperation(eq(TENANT), eq(SESSION), any()))
|
||||
.thenAnswer(args -> {
|
||||
target.set(delivery("running", true));
|
||||
return Map.of();
|
||||
});
|
||||
target.set(delivery("pending", false));
|
||||
coordinator.dispatch(TENANT, SESSION, OPERATION);
|
||||
verify(store).settleTaskCancel(eq(TENANT), eq(SESSION),
|
||||
eq(OPERATION), anyString(), eq(1L),
|
||||
eq(TaskCancelOutcome.completed()), eq(0L));
|
||||
target.set(delivery("completed", false));
|
||||
coordinator.dispatch(TENANT, SESSION, OPERATION);
|
||||
verify(harness).runChannelOperation(eq(TENANT), eq(SESSION), any());
|
||||
verify(store).settleTaskCancel(eq(TENANT), eq(SESSION),
|
||||
eq(OPERATION), anyString(), eq(1L),
|
||||
eq(TaskCancelOutcome.failed("task_already_settled")),
|
||||
eq(0L));
|
||||
}
|
||||
|
||||
// A Broker process that holds no context for the Runtime Session (a
|
||||
// fresh one before a turn or the close acquires it again) cannot stop
|
||||
// the process on any retry: the cancel parks at once, so it stops
|
||||
@@ -725,6 +773,14 @@ class TaskCancelCoordinatorTest {
|
||||
.put("stopReason", stopReason));
|
||||
}
|
||||
|
||||
private static TaskTarget delivery(String state,
|
||||
boolean cancelRequested) {
|
||||
return new TaskTarget("channel_delivery", "delivery-1",
|
||||
"channel_delivery", state, 3, JSON.createObjectNode()
|
||||
.put("deliveryId", "delivery-1")
|
||||
.put("cancelRequested", cancelRequested));
|
||||
}
|
||||
|
||||
private static TaskTarget workflowRun(String state,
|
||||
boolean stopRequested) {
|
||||
return new TaskTarget("child_run", "run-1", "workflow", state, 3,
|
||||
|
||||
@@ -1132,7 +1132,7 @@ export interface components {
|
||||
workspaceId: string;
|
||||
};
|
||||
/** @enum {string} */
|
||||
TaskKind: "child_agent" | "workflow" | "background_shell" | "monitor" | "automation_run";
|
||||
TaskKind: "child_agent" | "workflow" | "background_shell" | "monitor" | "automation_run" | "channel_delivery";
|
||||
/**
|
||||
* @description Logical run state. completed is terminal and is set when the run's own line ends settled, with or without a physical execution; failed and cancelled are terminal too, and they also cover a run that ended before its execution started (cancelled before dispatch, or failed with no provable start), when no physical execution ever settled. degraded means that the task still runs with reduced guarantees, such as a Monitor that lost its observation source. recovery_blocked means that recovery cannot prove the physical outcome: it is not settled, is never reported as success and is never re-run automatically.
|
||||
* @enum {string}
|
||||
@@ -1144,7 +1144,7 @@ export interface components {
|
||||
*/
|
||||
TaskRuntimeState: "unbound" | "provisioning" | "ready" | "draining" | "lost";
|
||||
/**
|
||||
* @description Actions the task supports now, the same for every caller. cancel: the cancel route accepts a new command for this task (since 1.40 for child_agent, 1.46 for workflow and 1.54 for background_shell and monitor, tasks of those kinds in pending, running, waiting or degraded, in an active Session); whether a caller may use it is an authorization check (403 on the cancel route). send_input: reserved for a later capability route. read_output: the task events route returns output events for this task to any caller that can read it. read_output does not change during the task's life, and a task without it produces no output events: its output goes only to Artifacts, so the events route never filters out events that exist.
|
||||
* @description Actions the task supports now, the same for every caller. cancel: the cancel route accepts a new command for this task (since 1.40 for child_agent, 1.46 for workflow and 1.54 for background_shell and monitor, tasks of those kinds in pending, running, waiting or degraded, and since 1.57 for channel_delivery, a delivery still pending because no adapter has claimed it, each in an active Session); whether a caller may use it is an authorization check (403 on the cancel route). send_input: reserved for a later capability route. read_output: the task events route returns output events for this task to any caller that can read it. read_output does not change during the task's life, and a task without it produces no output events: its output goes only to Artifacts, so the events route never filters out events that exist.
|
||||
* @enum {string}
|
||||
*/
|
||||
TaskActionCapability: "cancel" | "send_input" | "read_output";
|
||||
|
||||
Reference in new issue
Block a user