feat(tasks): one durable state-machine engine, two APIs (agents/state-machine + agents/tasks)

.changeset/chat-hosts-agent-streams.md

@@ -0,0 +1,6 @@
+---
+"@cloudflare/ai-chat": patch
+"@cloudflare/think": patch
+---
+
+`AIChatAgent` and `Think` no longer construct their own `Streams` capability: `Agent` installs one as `this.streams`, and `ResumableStream` raises the per-chunk ceiling chat needs for its own writes through that shared instance. `this.streams` is still there, now inherited from `Agent`; `createChatStreams()` is deprecated and only for a plain Lifecycle host that installs its own.

.changeset/lifecycle-jobs-push-sync.md

@@ -0,0 +1,7 @@
+---
+"agents": patch
+---
+
+`LifecycleJobs` gains `pushSync()` and `cancelSync()`: the same queue writes as `push()`/`cancel()` without the physical alarm re-arm, so a caller can write its job row inside its own `storage.transactionSync()` — where the async re-arm cannot be awaited — and have the job commit or roll back with the caller's own rows. The caller re-arms with `jobs.rearm()` once its transaction commits; a sync mutation made during startup needs no explicit call, because the re-arm is already deferred and coalesced to the end of startup. Nothing repairs a re-arm lost elsewhere, so an owner that pushes this way outside startup needs a startup reconcile of its own.
+
+The job queue also repairs its cached "table exists" flag when a rolled-back caller transaction was the one that lazily created the table, so a read or write made inside such a transaction no longer breaks the queue for the rest of the isolate.

.changeset/streams-writer-on-commit.md

@@ -0,0 +1,7 @@
+---
+"agents": patch
+---
+
+`StreamWriter.onCommit(fn)` registers synchronous callbacks to run inside the stream's settle transaction, for code that holds the writer but does not own the `close()` call. Callbacks run in registration order, before the `commit` passed to `close()`/`error()`, under the same contract — no awaiting, and a throw rolls the settle back and leaves the stream live — and only for the call that ends the stream. The set is fixed when `close()`/`error()` is called, `onCommit` returns an unregister function, and registrations belong to the writer object rather than the stream: reopening a live stream returns a fresh writer with none, and a writer stops accepting registrations once its own call has settled the stream.
+
+A writer with callbacks registered settles through the transaction path rather than the cheaper non-transactional one; writers that never call `onCommit` keep their current cost.

.changeset/tasks-state-machine.md

@@ -0,0 +1,36 @@
+---
+"agents": minor
+---
+
+`agents/tasks` becomes one durable state-machine engine with two APIs, and the engine ships as its own entry point, `agents/state-machine`.
+
+`agents/state-machine` exports `StateMachine`, the Lifecycle capability that owns the run tables, the wake stream and routed dispatch, registering machine definitions only, under a `StateMachine`-prefixed vocabulary (`StateMachineDefinition`, `StateMachineContext`, `StateMachineRunHandle`, `StateMachineOptions`, `StateMachine*Error`, plus `defineAsk`). `Tasks` now extends `StateMachine` — same tables, same handles, one capability instance — adding the durable-function form, compiled onto the engine as it is registered, and keeping every `Task*` name as an alias of its `StateMachine*` type (each `Task*Error` is the same class). `Agent` still installs only `this.tasks`; a machine registers through `taskDefinitions` or `this.tasks.register()`. Storage tables and event types keep the `task` namespace. One observable change: the engine errors' runtime `name` (as persisted in a failed run's `error.name` and seen in `onError`) now reads `StateMachine…` rather than `Task…`; `instanceof` under either alias is unchanged.
+
+The Workflows-shaped step API is unchanged — `(input, step) => result`, replay from the first line, `step.do`/`sleep`/`sleepUntil`/`status`/`idempotencyKey`/`signal`/`attempt`/`interrupted`, `NonRetryableError`, run `deadline` and `interruptions`. A step definition is compiled onto the engine as a single-phase machine whose one handler replays the function, on the same journal, with byte-identical step idempotency keys: there is no second journal, no second claim path, and no second abort protocol.
+
+Alongside it, a definition may now be a durable state machine: `{ initial, phases, onCancel?, migrate? }`. `initial` is the first checkpoint — a value, or a function of the run's seed. `phases` holds one handler per phase, `(state, ctx) => next`, keyed by the state union's `phase` discriminant: it receives the durable checkpoint, does work, and returns the next one. Returning is the commit, one generation-fenced row write. Annotate the definition `satisfies TaskMachine<State, Mailbox, Result, Seed>` and each handler's state narrows to its own phase; no helper function, no registration step. `ctx` is the per-handler runtime, and it EXTENDS `step` — the same object, widened with a durable mailbox, correlated durable asks (`defineAsk`), `memo`, owned children, engine-owned streams, and progress credit.
+
+Also: `run(name, input, { start, background, turnTimeout })` chooses who drives a run's first attempt — `warm` (the default), `queued`, or `attached`, which drives it in the caller's invocation and resolves at its next durable boundary — and `register(name, definition)` returns a `TaskInternalHandle` whose `run(input, { start })` is the one way to start a reserved `__cf`-prefixed definition. The `__DO_NOT_USE_WILL_BREAK__runAttached` and `__DO_NOT_USE_WILL_BREAK__enqueue` apertures are removed; every in-repo host and harness now starts its reserved definitions through the handle `register()` returns. `AIChatAgent` and `Think` gain a protected `_reservedTask(name)` accessor for those handles, and their `_enqueueChatRecovery` is now protected, so a subclass can queue a recovery run without the apertures. `tasks.at(name, runId)` returns a per-run typed handle beside the `handle(name)` lens, which is now typed by its definition, so the machine verbs are `never` — uncallable — on a function definition. `cancel(runId, reason?, { wait? })` can await terminality rather than acceptance.
+
+Schema version 3 adds the checkpoint, turn, progress, transition, abort-mark, parent, background and stream-epoch columns to `cf_agents_task_runs` by plain `ALTER TABLE ADD COLUMN`, adds mailbox, ask and routed-owner tables, and rebuilds `cf_agents_task_steps` as `cf_agents_task_journal` keyed by `(run_id, turn, name)`. Host SQL that reads the old table name must follow the rename. The rebuild runs in batches under a durable cursor at an intermediate version, each batch in its own transaction, and no definition dispatches until it completes; an object with nothing to rebuild skips it and stamps version 3 directly. A requested cancellation becomes the run's abort mark. In-flight runs migrate at turn 0 and resume from their journals; nothing is reset. Objects holding a large retained journal should call `tasks.delete({ settledBefore })` before upgrading, since the rebuild copies every retained row once.
+
+**Two type changes:** `TaskWaitReason` widens to `sleep | retry | interrupted | mailbox | event | ask | child | paused`, and `TaskRunSnapshot`'s `waiting` arm makes `wakeAt` OPTIONAL, because a run parked on a mailbox, an ask or a child has no wake time and reporting `updated_at` as a future wake was wrong. The `running` and `waiting` arms gain `abortRequested`/`abortReason` for the window between a `cancel()` taking the mark and the run settling. `TaskRunState` keeps its six values; `faulted` and `orphaned` ride a new `outcome` field beside `failed`.
+
+New exports on `agents/tasks`: the machine types (`TaskMachine`, `TaskContext`, `TaskDefinition`, `TaskDefinitions`, `TaskPhased`, `TaskTerminal`, `TaskTimedOut`), the extractors (`TaskInput`, `TaskOutput`, `TaskState`, `TaskMailbox`), the handles (`TaskHandle`, `TaskRunHandle`, `TaskInternalHandle`, with `Task` kept as the two-parameter alias), the mailbox, ask, stream, spawn and change types, `defineAsk` with `AskKind`/`Pending`, `AssertJson`, `MAX_CHECKPOINT_BYTES`, and the new errors `TaskNoProgressError`, `TaskTransitionBudgetError`, `TaskTurnDeadlineExceededError`, `TaskCancelCannotParkError`, `TaskConcurrentParkError`, `TaskEventTimeoutError`, `TaskMailboxFullError`, `TaskCheckpointTooLargeError` (a `TaskSerializationError`) and `TaskOrphanedDefinitionError`.
+
+Children and streams: `ctx.spawn(definition, input, { background?, notify?, ...run options })` accepts a child run whose settlement lands in the parent's mailbox as a `kind: "child"` note, `ctx.join(children, { within })` is sugar over receiving those notes, cancelling a parent cascades to its in-tree children (background children are detached), and `terminate` takes the children with it. `ctx.stream(name?)` opens an engine-owned stream — `${runId}:${name}#${epoch}`, tagged `${runId}:${name}` — that settles in the same transaction as the next checkpoint, is sealed and rotated when a lost attempt is reclaimed, and credits its appends as progress; pass the `Streams` capability as `streams` in the Tasks options. `view()` reports the run's streams. `Agent` now installs the `Streams` capability as `this.streams` and hands it to `this.tasks`, so a machine on an Agent can open engine-owned streams without further wiring. `streams.transaction(closure)` settles several streams as one unit, holding their events until it commits; the engine settles a run's streams together with the run's own write through it.
+
+Cross-facet children: `ctx.spawn(definition, input, { owner })` accepts the child on one of the Agent's sub-agents, addressed by the new `Agent.subAgentRouteAddress(cls, name)`. The child runs and journals there; its settlement note returns to the parent through the root, the cancel cascade and `terminate` reach it, and `view().children` lists it with its `ownerKey`. Every run a sub-agent accepts is now indexed on the root (`cf_agents_task_routes` gains `definition`, `background` and `settled_at`), and a verb that misses locally — `get`, `view`, `send`, `sendEvent`, `withdraw`, `answer`, `withdrawAsk`, `cancel`, `terminate`, `pause`, `resume` — is forwarded to the owner, so an approval answered on the root reaches a facet-hosted run parked with no wake. `watch()` and `list()` stay local.
+
+Per-effect compensation: `step.do(name, { compensate: (result) => … }, cb)` declares how to undo a completed step. When a run without `onCancel` is aborted — cancel, deadline, watchdog, cascade or memory limit — the inline default replays the current turn's journal in a fresh, fenced invocation (no step executes again), collects the compensations of completed steps, runs them newest first, each bounded by its step's timeout, records each success on the journal row (`compensated_at`) so an interrupted pass never repeats one, then settles by the mark. A failing compensation is recorded (`task:step:compensation-failed`) and the rest still run; `watch()` reports a `compensating` change. `terminate()` runs none. A live transition is no longer joined before the inline default: the pass's fresh generation fences it.
+
+The repository's `check` gate now runs `check:machines`, a rule that requires every object-literal machine definition to carry `satisfies TaskMachine<…>` (or `StateMachineDefinition`), since only the annotation narrows each handler's state to its phase.
+
+Review round two: a machine run's transition watchdog defaults to the step timeout (off for a durable function) and is enforced on routed runs too; a terminal the fence refuses consumes no mailbox items; a duplicate `requestId` is refused before `latest` can clear the original; ask ids and default child run ids derive from the turn so a replayed transition finds what it already raised or spawned; a park re-checks for a send or answer that raced it; `reopen()` accepts any failed or cancelled run whose row is still present.
+
+Two engine knobs for one-machine-per-address hosts:
+
+- A definition may declare `transitionBudget`, overriding the capability's for its own runs, so a machine that legitimately transitions many times between parks is not faulted by progress Rule B.
+- `run()` at an existing `runId` or `idempotencyKey` joins across versions of one base — the conflict check compares base names — and the receipt's `definition` is the name the run is stored under. A handle scopes by base for the same reason, so the handle that joined the run can still read it. A different base still throws.
+
+Breaking, experimental surface.

design/AGENTS.md

@@ -110,6 +110,8 @@ Keep it concise. A few paragraphs is fine. These are records, not essays.
 | `rfc-streams.md`                         | RFC        | Streams — durable incremental output as a Lifecycle capability: chunk log, cursor, replay-then-tail reads, composed with Tasks via checkpointed cursors (proposed)              |
 | `rfc-pi-harness-example.md`              | RFC        | Pi `AgentHarness` as an example-local Lifecycle capability over Tasks, Streams, and WebSockets; what must land before it becomes a package export (proposed)                    |
 | `rfc-self-modifying-harness.md`          | RFC        | Editable harness source in Workspace, activation through Worker Bundler, and one fresh Dynamic Worker per turn (proposed)                                                       |
+| `rfc-tasks-state-machine.md`             | RFC        | Tasks as one durable state-machine engine with two definition forms — the step API compiled onto it — plus a normative specification appendix (proposed)                        |
+| `tasks-cleanup-plan.md`                  | design doc | Forward plan, not current state: surface-by-surface verdicts, consumer migrations and a sequenced PR order for the Tasks engine cutover; delete once it lands                   |
 
 ## Relationship to `/docs`
 

design/README.md

@@ -22,3 +22,5 @@ The goal is to give contributors (and future-us) a quick way to understand _why_
 | [rfc-harness-model-provider-boundary.md](./rfc-harness-model-provider-boundary.md) | RFC: LanguageModelV4 for Codex and pi-ai providers for Pi in v1               |
 | [test-coverage-matrix.md](./test-coverage-matrix.md)                               | Feature × test-layer coverage rollup, CI mapping, skipped-test debt, hygiene  |
 | [rfc-pi-harness-example.md](./rfc-pi-harness-example.md)                           | RFC: pi `AgentHarness` as an example-local Lifecycle capability               |
+| [rfc-tasks-state-machine.md](./rfc-tasks-state-machine.md)                         | RFC: Tasks — one durable state-machine engine, two APIs (proposed)            |
+| [tasks-cleanup-plan.md](./tasks-cleanup-plan.md)                                   | Tasks cleanup — surface verdicts, consumer migrations, sequenced PR plan      |

design/alarm-coordination.md

@@ -25,10 +25,14 @@ capabilities push jobs instead of contributing wake times.
 - **Deadman pre-alarm** — armed before the event loop drives any due job, so
   an isolate death mid-drive still wakes the object to resume its queue.
 - **Tasks** — the capability for durable replayable execution. Every
-  non-terminal run's authoritative `next_at` deadline is mirrored as one queue
-  job per run. Task wakes use one job-dispatch attempt because ReplayStep owns
-  their durable retry budget; a propagated platform failure must reach a fresh
-  alarm invocation instead of entering JobDriver's generic retry loop.
+  non-terminal run's authoritative wake — its park or claim deadline, its run
+  deadline, and while a transition is live its `turn_deadline_at` watchdog —
+  is mirrored as one queue job per run. A run parked on its mailbox, an ask
+  or a child with no `within` has no wake at all and holds no job: a send or
+  an answer schedules one directly. Task wakes use one job-dispatch attempt
+  because ReplayStep owns their durable retry budget; a propagated platform
+  failure must reach a fresh alarm invocation instead of entering JobDriver's
+  generic retry loop.
 
 ## How the alarm is derived
 
@@ -80,9 +84,8 @@ Scheduler rows. Root chat recovery uses the reserved
 temporarily keeps its root-owned Scheduler rows; its callback executes on the
 facet's own Lifecycle, outside any alarm, so `trackAlarmWork` declines there
 and the facet's post-handoff turn is bounded by the incident's own memory-reset
-budget rather than the breaker until Tasks can mirror child wakes to the alarm
-owner. A sealing strike still reaches the facet through the routed
-compatibility bridge.
+budget rather than the breaker. A sealing strike still reaches the facet
+through the routed compatibility bridge.
 
 ## Agent integration
 

design/rfc-fibers.md

@@ -2,7 +2,11 @@
 
 Status: shipped as `agents/tasks` (originally proposed under the working
 name "Fibers"; this document records the shipped design and how it evolved
-during implementation).
+during implementation). Its execution half — the replay engine, the step
+journal, the claim and wake rules — is superseded by
+[`rfc-tasks-state-machine.md`](./rfc-tasks-state-machine.md), which
+rebuilds it as one state-machine engine with the step API compiled on top;
+the "deferred: facets" line below was already stale by then.
 
 ## The problem