diff --git a/packages/agent/docs/pico/pico-usage-guide.md b/packages/agent/docs/pico/pico-usage-guide.md index db5e420e1..f2f0dc2e8 100644 --- a/packages/agent/docs/pico/pico-usage-guide.md +++ b/packages/agent/docs/pico/pico-usage-guide.md @@ -9,6 +9,7 @@ why things are the way they are. ## Table of Contents - [The Mental Model](#the-mental-model) +- [Calls and Cancellation](#calls-and-cancellation) - [Installation](#installation) - [Quick Start](#quick-start) - [Sessions and Conversations](#sessions-and-conversations) @@ -88,6 +89,35 @@ Nothing runs by itself. Opening a session starts nothing. **Driving** a conversa its tasks execute, and a `prompt` is just accept + drive + read the answer. A UI **watches** a conversation and gets a **view** it can render from directly. +## Calls and cancellation + +Every asynchronous harness, conversation and task-runtime operation takes a required final `Call`. +`Call` is a type alias of Chord `Context`: it carries an abort signal, telemetry parent and, for task +code, a private typed invocation identity. It is not model context. No casts or admission helpers +are needed. Pure accessors, synchronous registrations and methods inside a transaction take no Call. + +```typescript +import type { Call } from '@earendil-works/pi-agent'; +import { BACKGROUND_CONTEXT, withCancel } from '@earendil-works/chord/context'; + +const call: Call = BACKGROUND_CONTEXT; // host call, without cancellation +const { context: waitingCall, cancel } = withCancel(call); +const waiting = conversation.drive(waitingCall); +cancel(); // removes this waiter; does not abort durable work +await waiting; // rejects with cancellation +``` + +Tasks receive `(task, runtime: TaskRuntime, call: Call)` and forward `call`. Tools receive +`(toolCallId, params, out, runtime: ToolRuntime, call: Call)`. Environment/provider/hook operations +interpret the signal; custom handlers must cooperate. Derive a Call for nested telemetry or a tighter +deadline and pass it onward. The driver does not inherit a drive caller's signal into task execution. + +The line reads task identity through a private `createContextKey`; Chord returns the +correct type from `call.value(key)`. Derived calls preserve that exact object. Stale task writes +reject. No invocation identity is serialized over RPC; trusted host bindings supply it locally. +Deliberately using an unrelated host Call or ignoring cancellation is an in-process escape, not +something a facade or type can prevent. + ## Installation ```bash @@ -97,10 +127,13 @@ npm install @earendil-works/pi-agent ## Quick Start ```typescript -import { Harness, JsonlStorage } from '@earendil-works/pi-agent'; +import { Harness, JsonlStorage, type Call } from '@earendil-works/pi-agent'; +import { BACKGROUND_CONTEXT } from '@earendil-works/chord/context'; import { readTool, writeTool, bashTool } from '@earendil-works/pi-agent/tools'; import { builtinModels } from '@earendil-works/pi-ai/providers/all'; +const call: Call = BACKGROUND_CONTEXT; + // One file per session. Reopening the same file resumes it. const storage = await JsonlStorage.open('./session.jsonl'); @@ -109,10 +142,10 @@ const storage = await JsonlStorage.open('./session.jsonl'); const h = await Harness.open(storage, { models: builtinModels(), tools: [readTool, writeTool, bashTool], -}); +}, call); const { generation: generationKind } = h.kinds; // the registered kinds, for config and hooks -const c = await h.root(); +const c = await h.root(call); // Config is declared by the kinds that read it. The generation kind declares model, thinking and // selected tools; `settings` is shorthand for its config. @@ -120,12 +153,12 @@ await c.settings.set({ model: { provider: 'anthropic', modelId: 'claude-opus-5' }, thinking: 'high', selectedTools: ['read', 'write', 'bash'], -}); +}, call); // The system prompt is not stored as config. Each turn the harness asks this hook what the // instructions should be right now, compares with what the transcript says the model was told, // and writes only the difference. See "System Prompt and Tool Loadout". -c.hooks.on(generationKind, 'system_instructions', async ({ config }) => ({ +c.hooks.on(generationKind, 'system_instructions', async ({ config }, call) => ({ sections: { identity: 'You are a careful engineer working in this repository.', cwd: `Working directory: ${process.cwd()}`, @@ -135,7 +168,7 @@ c.hooks.on(generationKind, 'system_instructions', async ({ config }) => ({ // Watch the conversation. `view` is a plain object a UI renders from; attach whenever you like, // the view is complete as of the moment you attach and events follow from there. -const w = await h.watch(c.id, { tail: 100 }); +const w = await h.watch(c.id, { tail: 100 }, call); w.start(event => { if (event.type === 'task_output') { // the generation's preview is the partial assistant message; a tool's is its output so far @@ -146,24 +179,25 @@ w.start(event => { }); // accept the input, drive the conversation until it is idle, return the answer to that input -const answer = await c.prompt('Inspect the parser and list the public API'); +const answer = await c.prompt('Inspect the parser and list the public API', undefined, call); console.log(answer?.model[0]?.content); w.unsubscribe(); -await h.close(); // cancels nothing durable; reopening the file continues exactly here +await h.close(call); // cancels nothing durable; reopening the file continues exactly here ``` Run it, kill it in the middle, run it again: the second run recovers whatever was in flight (publishing a partial answer, rerunning or reporting an interrupted tool) and continues. That is the whole point. -Snippets below assume `h` and `c` set up like this. +Snippets below assume `h`, `c` and `call` set up like this. Optional options before Call are passed +as `undefined` when unused. A task or hook always forwards its supplied Call, not this host root. ## Sessions and Conversations ### Opening -`Harness.open(storage, options)` takes the storage backend and everything that defines behaviour: +`Harness.open(storage, options, call)` takes the storage backend and everything that defines behaviour: | option | what it is | |---|---| @@ -189,8 +223,8 @@ change: ```typescript import { generationKind } from '@earendil-works/pi-agent/kinds'; const h = await Harness.open(storage, { models, tools, replace: { - generation: { ...generationKind, async execute(task, ctx) { await audit(task); return generationKind.execute(task, ctx); } }, -}}); + generation: { ...generationKind, async execute(task, runtime, call) { await audit(task); return generationKind.execute(task, runtime, call); } }, +}}, call); ``` Open checks the recorded kind strings without scanning the transcript and reports live work, but @@ -199,7 +233,7 @@ starts nothing. An unregistered historical entry kind is reported, not rejected: unavailable. A live task kind is still required. ```typescript -const { start, inflight } = await h.inspect(); +const { start, inflight } = await h.inspect(call); // start: tasks that never began or must begin again (a planned tool, a retry, a scheduled job) // inflight: tasks the last process was running when it stopped; recover() will handle them ``` @@ -212,18 +246,31 @@ scope. Each promise resolves when its scope has no live foreground task. Its att work remains served afterwards. A UI usually attaches a watch to what it shows and drives that. ```typescript -const h = await Harness.open(storage, opts); -void h.drive(); // resume everything in the background +const h = await Harness.open(storage, opts, call); +void h.drive(call).catch(error => console.error(error)); // resume; report session faults ``` ### Closing ```typescript -await h.close(); // cancel in-process work, write nothing; everything resumes on the next open -await h.shutdown(); // mark all live work aborted, drain queues, wait for cleanup, then close +await h.close(call); // cancel in-process work, write nothing; everything resumes on the next open +await h.shutdown(call); // mark live tasks, wait for abort cleanup, close; queued input survives ``` -`close` is the normal exit: a crash and a close leave storage in the same state. +`close` is the normal exit: it stops admission on the commit line, then signals and joins owned +invocations outside it, without writing task outcomes. Earlier line operations have already finished; +later mutations reject. Close waits for actual invocation completion, regardless of caller cancellation. +An uncooperative task can delay it indefinitely. + +`shutdown` closes normal admission and atomically marks live tasks only. Queued input and its queued +result records remain stored, including in idle conversations. Fresh abort handlers resolve their +already-running input groups and finish cleanup. Built-in child cleanup marks tasks only, never drains +child queues, including after a crash and reopen. Shutdown waits for both live tasks and running calls +to disappear before closing. Preserved queues create no work by inference on reopen/drive. + +Once admitted, caller cancellation does not abandon shutdown. Repeated lifecycle calls share +completion; explicit close interrupting shutdown makes shutdown reject. Task calls cannot invoke +host lifecycle methods. ## Configuration @@ -250,16 +297,16 @@ a plugin declares its own values the same way (see [Task Kinds](#task-kinds)). ```typescript // all of a kind's values, one batched read -const { model, thinking } = await c.config(generationKind).get(); +const { model, thinking } = await c.config(generationKind).get(call); // some of them, one commit -await c.config(generationKind).set({ thinking: 'low' }); +await c.config(generationKind).set({ thinking: 'low' }, call); // `settings` is config(generationKind), because that is what every UI touches -await c.settings.set({ model: { provider: 'openai', modelId: 'gpt-5.6' } }); +await c.settings.set({ model: { provider: 'openai', modelId: 'gpt-5.6' } }, call); // one value, one point read, by its declared address -const tools = await c.value(generationKind.config.selectedTools).get(); +const tools = await c.value(generationKind.config.selectedTools).get(call); ``` A change is a normal commit: it appears in the watch stream as a `value` event, and the next @@ -303,11 +350,11 @@ in the order given) and the **tool definitions** the model may call. Register it with `subtree: true` so subagents inherit it unless they register their own: ```typescript -c.hooks.on(generationKind, 'system_instructions', async ({ conversationId, config }) => ({ +c.hooks.on(generationKind, 'system_instructions', async ({ conversationId, config }, call) => ({ sections: { identity: IDENTITY, cwd: `Working directory: ${cwd}`, - context_files: await renderContextFiles(cwd), // AGENTS.md and friends + context_files: await renderContextFiles(cwd, call), // AGENTS.md and friends skills: skills.render(), }, tools: h.tools.select(config.selectedTools), // complete definitions, from the catalogue @@ -332,10 +379,10 @@ Every change is either a config write or something the host renders differently. written at that moment; the next generation notices and appends one entry: ```typescript -await c.settings.set({ selectedTools: ['read', 'write', 'bash', 'grep'] }); +await c.settings.set({ selectedTools: ['read', 'write', 'bash', 'grep'] }, call); // next turn → 20 system { toolsAdded: [grep] } -await c.settings.set({ selectedTools: ['read'] }); +await c.settings.set({ selectedTools: ['read'] }, call); // next turn → 30 system { toolsRemoved: [write, bash, grep] } // the user installs a skill; skills.render() changes @@ -357,15 +404,16 @@ A plugin contributes sections from its own state. Handlers for the same point me ```typescript const planMode = conversationValue('plan.mode', { rewind: true }); -h.hooks.on(generationKind, 'system_instructions', async ({ conversationId }) => ({ - sections: { - plan_mode: (await h.conversation(conversationId).value(planMode).get()) ? PLAN_GUIDANCE : undefined, - }, -})); +h.hooks.on(generationKind, 'system_instructions', async ({ conversationId }, call) => { + const conversation = await h.conversation(conversationId, call); + return { sections: { + plan_mode: (await conversation?.value(planMode).get(call)) ? PLAN_GUIDANCE : undefined, + } }; +}); -await c.value(planMode).set(true); +await c.value(planMode).set(true, call); // next turn → 60 system { sections: [plan_mode] } "the plan_mode guidance now applies: ..." -await c.value(planMode).set(false); +await c.value(planMode).set(false, call); // next turn → 70 system { sections: [plan_mode removed] } ``` @@ -398,10 +446,10 @@ const childId = await c.spawn({ [generationKind.config.profile, 'auditor'], ], }, -}); +}, call); -const child = await h.conversation(childId); -child.hooks.on(generationKind, 'system_instructions', async ({ config }) => ({ +const child = await h.conversation(childId, call); +child.hooks.on(generationKind, 'system_instructions', async ({ config }, call) => ({ sections: { identity: AUDITOR_IDENTITY, cwd: `Working directory: ${cwd}` }, tools: h.tools.select(config.selectedTools), })); @@ -420,9 +468,9 @@ Nothing to do in any of them. first turn diffs the two: ```typescript - const b = await c.fork(12); // config as of 12: tools [read, write, bash] - await b.settings.set({ selectedTools: ['read'] }); - await b.prompt('...'); // b's turn: toolsRemoved [write, bash]; the prefix's 11 is still the baseline + const b = await c.fork(12, undefined, call); // config as of 12: tools [read, write, bash] + await b.settings.set({ selectedTools: ['read'] }, call); + await b.prompt('...', undefined, call); // b's turn: toolsRemoved [write, bash]; the prefix's 11 is still the baseline ``` - **A restart** with a changed host (new skill, different cwd) emits "these sections now apply" @@ -436,7 +484,7 @@ Nothing to do in any of them. explicit result for that input. ```typescript -const answer = await c.prompt('Inspect the parser'); +const answer = await c.prompt('Inspect the parser', undefined, call); // AssistantEntry | undefined (the run ended without an answer: aborted, failed) ``` @@ -445,11 +493,11 @@ element, even when idle acceptance places and removes it in the same commit. Gen `post_tools` carry input ids explicitly, so result lookup never scans the transcript: ```typescript -const { inputId } = await c.accept('Inspect the parser', { requestId: 'req-42' }); -const outcome = await c.drive(); // 'idle' | 'closed' -const result = await c.result(inputId); // one sticky-value point read +const { inputId } = await c.accept('Inspect the parser', { requestId: 'req-42' }, call); +const outcome = await c.drive(call); // 'idle' | 'closed' +const result = await c.result(inputId, call); // one sticky-value point read const answer = result?.status === 'done' - ? await h.getEntry(assistantKind, result.resultEntryId) + ? await h.getEntry(assistantKind, result.resultEntryId, call) : undefined; ``` @@ -459,15 +507,15 @@ Context-only writes finish as `placed`. Terminal results never change. A remote caller that lost the accept response performs an explicit request lookup: ```typescript -const existing = await h.acceptance('req-42'); +const existing = await h.acceptance('req-42', call); if (existing) { - const conversation = await h.conversation(existing.conversationId); + const conversation = await h.conversation(existing.conversationId, call); return { inputId: existing.inputId, - result: await conversation?.result(existing.inputId), + result: await conversation?.result(existing.inputId, call), }; } -return c.accept(input, { requestId: 'req-42' }); +return c.accept(input, { requestId: 'req-42' }, call); ``` A concurrent duplicate `accept` rejects as `RequestAlreadyAccepted` and identifies the first @@ -485,13 +533,13 @@ While a turn runs, new input is queued rather than refused. The mode determines input-group behavior: ```typescript -const steer = await c.steer('Focus on the tokenizer first'); // post_tools: joins current group; +const steer = await c.steer('Focus on the tokenizer first', call); // post_tools: joins current group; // final answer: starts next group -const follow = await c.followUp('Then write the tests'); // after final answer; starts next group -const next = await c.nextRun('Remind me to commit'); // waits for the next explicit idle accept +const follow = await c.followUp('Then write the tests', call); // after final answer; starts next group +const next = await c.nextRun('Remind me to commit', call); // waits for the next explicit idle accept const write = await c.write(noteKind, { // next safe boundary; no generation data: { text: 'user stepped away' }, -}); +}, call); ``` When a generation emits calls, its settlement creates every tool plus exactly one `post_tools` @@ -504,22 +552,60 @@ updates are append/remove/clear operations, not whole-array replacements. Any it until it lands: ```typescript -await c.cancelQueued(steer.inputId); // 'cancelled' | 'not_found' +await c.cancelQueued(steer.inputId, call); // 'cancelled' | 'not_found' ``` ### Aborting ```typescript -await c.abort(); // every live foreground task of this conversation, and of conversations they own -await h.abortTask(id); // one background task: a job, a schedule +await c.abort(call); // every live foreground task of this conversation, and of conversations they own +await h.abortTask(id, call); // one background task: a job, a schedule ``` -An abort is a durable mark on the task, then cleanup. The live generation or `post_tools` owns the -active input ids and resolves all of them as cancelled without a normal successor. A generation may -retain its partial for the UI but creates no tool tasks/results and is excluded from later requests; -tool tasks that already exist write their own error results; a subagent tool aborts its child. If -the process dies between the mark and cleanup, the next one finishes it. Queued `steer` and -`followUp` are removed and marked cancelled; `write` and `nextRun` stay queued. +An abort mark is a durable request, not terminal settlement: + +```text +100 generation streaming, execute invocation A running +110 abort=true commits; A can no longer write main state or scratch + line releases; A's signal fires; provider exits; A returns + driver removes A and starts abort invocation B with a fresh Call +120 B commits optional display-only partial, cancelled input results, terminal aborted status + scratch is retired; B returns +``` + +There is no mark/signal branch in normal settlement. If the mark wins, `runtime.commit` rejects +`TaskCancelled` before its closure runs; execute unwinds and the kind's `abort()` writes the durable +cancellation result. If normal settlement wins first, the task is already terminal. Standard effects +observe the signal cooperatively; no effect gate exists, so an operation may start before the signal +is delivered and then cancel. Cancellation does not undo external effects. + +`abortTask` marks exactly one task; if not owned or attached, cleanup awaits a later drive. Conversation +abort marks the current foreground ownership closure. Fresh parent cleanup explicitly cancels its +foreground children and recorded non-detached jobs. Queued `steer`/`followUp` are removed and marked +cancelled; `write`/`nextRun` remain. This is the explicit conversation-abort policy. Shutdown and +built-in task abort cleanup mark tasks only and preserve all queued items and their queued results. + +A generation's abort handler updates its explicit input group in the same commit as settlement. +Here `inputResult(id)` is the sticky value address holding that input's result: + +```typescript +async abort(task, runtime, call) { + await runtime.commit(async tx => { + for (const id of task.state.inputs) { + const address = inputResult(id); + const result = await tx.value(address).get(); + if (result?.status !== 'running') throw new Error(`Invalid active input ${id}`); + tx.value(address).set({ ...result, status: 'cancelled' }); + } + tx.settle(task.id, 'aborted', task.state); + }, call); +} +``` + +The driver never resolves input results itself. Tool and generation kinds own their respective +outcomes. They can recover only committed scratch; missing final usage is unknown, not zero. +Runtime scratch writes reject after cancellation too: await/catch them. Harness sinks handle and +drain their own scratch promises and drop late callbacks, never silently recreate retired scratch. ## Watching @@ -529,10 +615,10 @@ A UI never reads storage or parses commits. It watches a conversation and gets a JSON object the harness keeps current, plus typed events that say what changed. ```typescript -const w = await h.watch(c.id, { tail: 100, values: [myPlugin.config.mode] }); +const w = await h.watch(c.id, { tail: 100, values: [myPlugin.config.mode] }, call); w.view // ConversationView, captured atomically with the subscription w.start(listener); -w.resnapshot(); // fresh capture, same subscription (if the client fell behind) +w.resnapshot(call); // fresh capture, same subscription (if the client fell behind) w.unsubscribe(); ``` @@ -636,7 +722,7 @@ same module. ```typescript // worker (has the harness) // ui process -const w = await h.watch(c.id, { tail: 100 }); on('view', m => { view = m.view; render(view); }); +const w = await h.watch(c.id, { tail: 100 }, call); on('view', m => { view = m.view; render(view); }); send({ type: 'view', view: w.view }); w.start(e => send({ type: 'event', event: e })); on('event', m => { applyEvent(view, m.event); render(view, m.event); }); ``` @@ -647,7 +733,7 @@ What isn't one conversation's: the conversation list, session values, usage tota reports (a hook threw, a task kind misbehaved and was stopped). ```typescript -const sw = await h.watch(); +const sw = await h.watch(call); sw.view.conversations; sw.view.values; sw.view.faulted; sw.start(e => { if (e.type === 'usage') status.setCost(e.totals); @@ -663,10 +749,10 @@ copied and nothing in the source is deleted. It carries the context as it was at rewindable values in force there, and the prompt-as-sent, so its first turn is exact. ```typescript -const alt = await c.fork(answer.id); // the source keeps running, untouched -await alt.prompt('Try a different implementation'); +const alt = await c.fork(answer.id, undefined, call); // the source keeps running, untouched +await alt.prompt('Try a different implementation', undefined, call); -const back = await c.fork(earlier.id, { abort: true }); // "go back": aborts the source's foreground first +const back = await c.fork(earlier.id, { abort: true }, call); // "go back": aborts the source's foreground first ``` Any transcript entry is a valid fork point, including an assistant with unanswered tool calls or one @@ -675,9 +761,9 @@ exchange without inheriting or executing the source tasks. Which conversation a "current" is the UI's business; the harness only has conversations. ```typescript -const all = await h.conversations(); +const all = await h.conversations(undefined, call); const independent = all.items.filter(x => x.owner === undefined); // root and forks: their own drive scopes -const children = await h.conversations({ parent: c.id }); // forks and owned children of c +const children = await h.conversations({ parent: c.id }, call); // forks of c; owned children use ownedFrom ``` ## Compaction and Reset @@ -689,16 +775,16 @@ remain after the prepared boundary. Only a competing head makes the summary stal not. ```typescript -const collapseId = await c.collapse(); -const collapseId = await c.collapse({ instructions: 'keep the API decisions verbatim' }); +const collapseId = await c.collapse(undefined, call); +const collapseId = await c.collapse({ instructions: 'keep the API decisions verbatim' }, call); ``` Threshold and overflow compaction happen inside the generation; nothing to call. Reset starts the context over, with or without a handoff message: ```typescript -await c.reset('Continue from here: we settled on a recursive-descent parser.'); // handoff -await c.reset(); // /clear +await c.reset('Continue from here: we settled on a recursive-descent parser.', call); // handoff +await c.reset(undefined, call); // /clear ``` The transcript keeps everything either way; only the context changes. @@ -721,11 +807,11 @@ subagent({ command: 'stop', id: 88 }) ```typescript // the API const childId = await c.spawn({ prompt: 'Profile the build', context: 'fresh', - values: { inherit: [generationKind.config.model] } }); -const child = await h.conversation(childId); -await child.accept('also check CI'); -await child.drive(); // or let h.drive() / the parent's drive carry it -await child.abort(); + values: { inherit: [generationKind.config.model] } }, call); +const child = await h.conversation(childId, call); +await child.accept('also check CI', undefined, call); +await child.drive(call); // or let h.drive() / the parent's drive carry it +await child.abort(call); ``` `run` keeps the calling tool inflight while it drives the child, so the child is part of the @@ -751,12 +837,12 @@ From the API a job is a task; a schedule is a job with `every`: ```typescript const dev = await c.commit(tx => tx.task(jobKind, { background: true, - state: { cmd: ['npm', 'run', 'dev'], cwd } })); + state: { cmd: 'npm run dev', cwd } }), call); const nightly = await c.commit(tx => tx.task(jobKind, { background: true, - state: { cmd: ['npm', 'test'], cwd, every: 24 * 3600_000, notBefore: tonightAt(2) } })); + state: { cmd: 'npm test', cwd, every: 24 * 3600_000, notBefore: tonightAt(2) } }), call); -await h.abortTask(nightly); // ends the schedule wherever it is +await h.abortTask(nightly, call); // ends the schedule wherever it is ``` A schedule is one task looping `planned → running → planned`; there is no trail of run ids to @@ -777,16 +863,16 @@ const moves = conversationList('game.moves', { rewind: true }); const expanded = conversationValue('ui.expanded', { rewind: false }); const name = sessionValue('pi.session.name'); -await c.value(planMode).set(true); -const on = await c.value(planMode).get(); -const then = await c.value(planMode).get(entryId); // as of an entry: rewindable only +await c.value(planMode).set(true, call); +const on = await c.value(planMode).get(call); +const then = await c.value(planMode).get(entryId, call); // as of an entry: rewindable only -const elementId = await c.list(moves).append({ x: 1, y: 2 }); -const page = await c.list(moves).read({ limit: 50 }); -await c.list(moves).remove(elementId); -await c.list(moves).clear(); +const elementId = await c.list(moves).append({ x: 1, y: 2 }, call); +const page = await c.list(moves).read({ limit: 50 }, call); +await c.list(moves).remove(elementId, call); +await c.list(moves).clear(call); -await h.value(name).set('parser work'); +await h.value(name).set('parser work', call); ``` Rewindable state is what makes plugins survive forks: a fork before the move doesn't see the move, @@ -795,7 +881,9 @@ a fork before plan mode was turned on doesn't have it on. Nothing to register, n ### Atomic Commits Anything that must land together goes in one commit: a closure on the session's write line. Reads -inside it see committed state; ids are final when returned; a throw discards everything. +inside it are asynchronous and see committed state; await them in an async builder. Writes are +synchronous and ids are final when returned; a throw discards everything. Awaiting storage reads does +not release the line. Never await external effects, driver waits or another commit inside a builder. ```typescript const entry = await c.commit(tx => { @@ -805,7 +893,7 @@ const entry = await c.commit(tx => { tx.task(myPlugin.reminderKind, { background: true, // tasks anywhere state: { about: id, at: Date.now() + 3600_000 } }); return id; -}); +}, call); ``` One rule: rewindable conversation value/list writes must precede entries in the same commit. That @@ -836,8 +924,8 @@ export const countLinesTool: Tool<{ i: string; path: string; pattern?: string }, output: { maxBytes: 64_000, maxLines: 200, retain: 'head' }, // the sink enforces this replay: 'safe', // read-only: may be rerun after a crash - async execute(toolCallId, params, signal, out, ctx) { - const lines = getOrThrow(await ctx.env.readTextLines(params.path, {}, { signal })); // ExecutionEnv: FileSystem & Shell + async execute(toolCallId, params, out, runtime, call) { + const lines = getOrThrow(await runtime.env.readTextLines(params.path, {}, call)); // ExecutionEnv: FileSystem & Shell const re = params.pattern ? new RegExp(params.pattern) : undefined; let matching = 0; for (const [i, line] of lines.entries()) { @@ -866,7 +954,8 @@ The sink: | `diag(severity, message, code?)` | commentary about the call, kept out of the data | A tool never touches the transcript, its siblings or the queue. What it may do is create things in -its own conversation through `ctx.conversation.commit(...)`: a job, a child conversation. +its own conversation through `runtime.commit(..., call)`: a job, a child conversation. The runtime +provides conversation lookup, reads and cancellation methods, not a raw host-lifecycle Harness. ### Diagnostics @@ -891,27 +980,46 @@ with the budget from config; if the budget runs out, the call settles with what goes on: ```typescript -async execute(toolCallId, params, signal, out, ctx) { - const job = await ctx.conversation.commit(tx => tx.task(jobKind, { background: true, - state: { cmd: params.cmd, cwd: params.cwd ?? ctx.env.cwd, - origin: { tool: 'bash', task: ctx.taskId, callId: toolCallId } } })); // so UIs render it as bash +async execute(toolCallId, params, out, runtime, call) { + const job = await runtime.commit(tx => { + const id = tx.task(jobKind, { background: true, + state: { cmd: params.cmd, cwd: params.cwd ?? runtime.env.cwd, + origin: { tool: 'bash', task: runtime.taskId, callId: toolCallId } } }); + tx.patch(runtime.taskId, { state: { + jobId: id, cancelJobOnAbort: !params.background, + } }); + return id; + }, call); if (params.background) { out.delegate(job); out.write(`started job ${job}`); return; } - let done: boolean; - try { done = await ctx.waitForTask(job, { budgetMs: ctx.budgetMs, signal }); } - catch (e) { if (isAbortError(e)) await ctx.harness.abortTask(job); throw e; } // user aborted: kill what we started - - const o = await ctx.jobOutput(job); // the job's ToolOutputState, live or settled - out.replace(o.text); out.capture(o.truncation); - if (done) { out.details.exitCode = o.exitCode; if (o.exitCode) out.diag('warn', `exit code ${o.exitCode}`, 'exit'); } - else { out.delegate(job); out.diag('info', `still running as job ${job}; use job wait / status / stop`, 'budget'); - await ctx.conversation.commit(tx => tx.patch(job, { state: { notify: true } })); } + const done = await runtime.waitForTask(job, { budgetMs: runtime.budgetMs }, call); + const output = await runtime.jobOutput(job, call); + out.replace(output.text); out.capture(output.truncation); + if (done) { + out.details.exitCode = output.exitCode; + if (output.exitCode) out.diag('warn', `exit code ${output.exitCode}`, 'exit'); + } else { + out.delegate(job); + out.diag('info', `still running as job ${job}; use job wait / status / stop`, 'budget'); + } } ``` -The job owns its output from the first byte; the tool only ever copies a snapshot. A UI shows the -job's output live either way, rendered with `bash`'s component. +The job owns its output from the first byte; the tool copies a snapshot. Creation stores the cleanup +reference on the tool atomically. A durable abort rejects execute mutations, including catch-handler +writes, so the fresh tool-kind abort handler reads that reference and cancels the non-detached job. +Already-terminal jobs count as successful cleanup; check-and-mark must be atomic or handle that result. + +The tool kind's successful delegation settlement marks the job detached and requests a notice through +a separate sticky notification record keyed by job id. Job completion consumes it atomically; if the +job is already terminal when delegation commits, delegation places the notice itself. Apply this +protocol to exited, killed and lost outcomes, including abort and recovery. The parent never patches +a running job's state. This covers either completion/delegation order. + +A UI renders the job's output with `bash`'s component. Adopting an arbitrary unfinished tool promise +after budget expiry is still an open integration design: it requires an explicit effect/sink ownership +transfer. The initial job-first path above does not race or abandon any invocation. ## Hooks @@ -922,24 +1030,25 @@ like (a human approval is a hook that waits). ```typescript // policy: everywhere -h.hooks.on(toolKind, 'before_tool', async ({ toolName, args, conversationId }) => { - if (toolName === 'write' && await h.conversation(conversationId).value(planMode).get()) +h.hooks.on(toolKind, 'before_tool', async ({ toolName, args, conversationId }, call) => { + const conversation = await h.conversation(conversationId, call); + if (toolName === 'write' && await conversation?.value(planMode).get(call)) return { block: { reason: 'plan mode: no edits' } }; - if (toolName === 'bash' && !(await ui.approve(args))) + if (toolName === 'bash' && !(await ui.approve(args, { signal: call.abortSignal }))) return { block: { reason: 'denied by user' } }; return { args }; // may rewrite arguments }); -h.hooks.on(toolKind, 'after_tool', async ({ toolCallId, output }) => { metrics.record(output.usage); }); +h.hooks.on(toolKind, 'after_tool', async ({ toolCallId, output }, call) => { metrics.record(output.usage); }); -h.hooks.on(generationKind, 'before_request', async ({ request }) => ({ request: withTracing(request) })); +h.hooks.on(generationKind, 'before_request', async ({ request }, call) => ({ request: withTracing(request) })); -h.hooks.on(generationKind, 'on_yield', async ({ answer, conversationId }) => { - if (await goalIncomplete(conversationId)) return { continue: 'The goal is not met yet; continue.' }; +h.hooks.on(generationKind, 'on_yield', async ({ answer, conversationId }, call) => { + if (await goalIncomplete(conversationId, call)) return { continue: 'The goal is not met yet; continue.' }; }); -h.hooks.on(collapseKind, 'before_collapse', async ({ reason, context }) => { - if (reason === 'manual' && context.length < 10) return { decline: true }; +h.hooks.on(collapseKind, 'before_collapse', async ({ reason, entries }, call) => { + if (reason === 'manual' && entries.length < 10) return { decline: true }; }); // instructions: per conversation (see System Prompt and Tool Loadout) @@ -958,8 +1067,10 @@ Points and their fail behaviour: | tool | `after_tool` | nothing | reported | | collapse | `before_collapse` | `{ decline? \| instructions? \| summary? }` | reported, skipped | -A handler may run again after a crash (the task it belongs to is re-executed), so its external side -effects need their own idempotence. +These failure policies exclude cancellation control errors, which propagate to unwind the invocation. +A handler receives the active Call as its final argument and must forward it to waits/effects. The +harness awaits its actual return, not an abandoned raced promise. A handler may run again after a +crash, so its external side effects need their own idempotence. ## Writing Kinds @@ -980,7 +1091,7 @@ export const noteKind = defineEntryKind('myplugin.note'); await c.commit(tx => tx.entry(noteKind, { data: { text: 'Parser plan accepted', pinned: false }, -})); // transcript/UI only; the model sees nothing +}), call); // transcript/UI only; the model sees nothing ``` A plugin that wants both typed data and a model message keeps the append-time conversion in an @@ -1027,8 +1138,8 @@ another head or edit when context should change; no entry-kind callback runs dur Reads are typed by the kind, or untyped and narrowed: ```typescript -const note = await h.getEntry(noteKind, id); // NoteEntry | undefined (also undefined for another kind) -const any = await h.getEntry(id); // Entry | undefined +const note = await h.getEntry(noteKind, id, call); // NoteEntry | undefined (also undefined for another kind) +const any = await h.getEntry(id, call); // Entry | undefined if (noteKind.is(any)) any.data.pinned; ``` @@ -1055,19 +1166,19 @@ export const reminderKind = defineTaskKind({ config: { intervalMs: conversationValue('myplugin.reminder.interval', { rewind: false }) }, hooks: { before_fire: { failClosed: false } }, - async execute(task, ctx) { - if (task.state.at > Date.now()) await ctx.sleep(task.state.at); // throws on abort - await ctx.commit(tx => tx.patch(task.id, { status: 'firing' })); // record intent before any effect - const { skip } = await ctx.hooks(reminderKind).run('before_fire', { about: task.state.about }); - await ctx.commit(tx => { - if (!skip && !tx.getTask(task.id)!.abort) // a marked task creates no successor + async execute(task, runtime, call) { + if (task.state.at > runtime.now()) await runtime.sleep(task.state.at, call); // throws on abort + await runtime.commit(tx => tx.patch(task.id, { status: 'firing' }), call); // record intent before any effect + const { skip } = await runtime.hooks(reminderKind).run('before_fire', { about: task.state.about }, call); + await runtime.commit(tx => { + if (!skip) // marked execution's whole commit rejects tx.entry(noticeKind, { model: [noticeMessage(`Reminder: see entry ${task.state.about}`)] }); tx.settle(task.id, 'done', { ...task.state, fired: !skip }); - }); + }, call); }, - async recover(task, ctx) { return this.execute(task, ctx); }, // safe to redo - async abort(task, ctx) { await ctx.commit(tx => tx.settle(task.id, 'aborted', task.state)); }, + async recover(task, runtime, call) { return this.execute(task, runtime, call); }, // safe to redo + async abort(task, runtime, call) { await runtime.commit(tx => tx.settle(task.id, 'aborted', task.state), call); }, // no preview: nothing to show while sleeping }); @@ -1079,13 +1190,24 @@ The rules an execution follows, and the driver enforces: crash after goes through `recover`. 2. You may block on the world: a provider stream, a process, a child conversation, a sleep. The driver runs executions concurrently; a blocked one holds up nothing. -3. Change your status or settle before returning. Returning unchanged is reported as a precise - contract violation; the driver does not try to diagnose changing but buggy status cycles. -4. Never wait on another task. Depend on it at creation (`after`), or let it be a conversation you - drive. +3. Make a committed status transition or settle before returning, except cancellation/close unwind. + The live task's epoch counts actual transitions, so `planned → running → planned` is valid. + Same-status/state-only patches and abort marks do not count. An abort handler must settle. +4. Prefer `after` for prerequisites; drive children or use the bounded job-wait API. Forward Call + to every wait. Known self/dependency waits reject; never race away an unfinished task/tool/hook. + +The scheduler runs on the commit line, keeps live-task/dependency indexes updated from whole committed +batches, and starts effects outside the line. It scans storage once at open, not after every call. +Only one invocation owns each task, but different tasks run concurrently. Repeated drives share an +attachment and create only temporary waiters. Historical completed children are never traversed. + +An unexpected error escaping a task kind or an unchanged return faults the session: pending drives +reject immediately, admission stops, running invocations are signalled and joined, then storage closes. +Domain tool/provider errors must be settled by their kinds. Reopen starts nothing; inspect and mark +work before driving, or install a replacement kind that can settle a broken persisted task. A kind with a `preview` decides what a UI sees while the task runs: `preview.init(scratch)` builds -it once on attach or reopen, and after that the kind mutates `ctx.preview.state` in place (the +it once on attach or reopen, and after that the kind mutates `runtime.preview.state` in place (the generation applies stream events to a partial message; a tool's preview is its sink). The harness flushes the Chord delta tracker after each scratch commit into `task_output` ops, so a token costs one append op, not a diff of the whole object. @@ -1095,8 +1217,8 @@ one append op, not a diff of the whole object. There is nothing to write. After a crash: ```typescript -const h = await Harness.open(storage, opts); -await h.drive(); +const h = await Harness.open(storage, opts, call); +await h.drive(call); ``` | what was running | what happens | @@ -1121,10 +1243,25 @@ startable. | `JsonlStorage` | everything, from one append-only file plus a scratch file per live task | local sessions; the default | | `SqliteStorage` | only what queries return | long sessions, many sessions in one file, servers | -All three answer the same queries with the same results; the conformance suite replays one -mutation stream into each and compares. JSONL encodes large values as Chord deltas so a plugin's -structured state stays linear in the file; SQLite stores values whole so a point read is never a -replay. +All three answer the same queries with the same results. Entries are immutable and read whole; +indexed queries select the relevant entries before decoding. Values are stored whole on every set; +lists store append/remove/clear operations. There is no automatic value diffing or Chord storage codec. +Chord deltas are only for preview/watch delivery. Streaming scratch appends compact assistant frames +or explicit output operations, not raw provider events with repeated growing partial snapshots. + +JSONL reopens by replaying complete main batches, then scratch for surviving live tasks. Per-file +sequence gaps are expected; the next id follows the maximum complete batch endpoint across both. +Retired scratch is ignored even if unlink failed; its later main settlement already covers its ids. +Only unterminated final lines are discarded and removed before further appends; malformed complete +main/live-scratch batches fail open. + +```text +main ends at 100; live scratch ends at 150 → reopen lastSeq=150, next write=151 +main settles task at 151; scratch unlink fails → ignore retired scratch, lastSeq=151 +``` + +The conformance suite compares one mutation stream across all backends. A repeated whole-value set +can write quadratic bytes when the value grows; incremental lists avoid that without hidden encoding. ```typescript const storage = await SqliteStorage.open('./sessions.db', { session: 'parser-work' }); diff --git a/packages/agent/docs/pico/pico-v3.md b/packages/agent/docs/pico/pico-v3.md index c3d1fd77c..178f97bdc 100644 --- a/packages/agent/docs/pico/pico-v3.md +++ b/packages/agent/docs/pico/pico-v3.md @@ -2,7 +2,9 @@ A design for the agent harness: how a session is stored, how work is scheduled, and how the built-in agent behaviour is composed from that. Code in this document illustrates the model; it is -not a package export. +not a package export. In snippets, `call` is the current `Call` (Chord Context, §6.5), required on +asynchronous public/runtime operations; `runtime` is the task/tool capability object. Transaction +builders do not take another Call. ## 1. Goals @@ -165,8 +167,8 @@ type ToolResultEntry = EntryBase & EntryData & ModelProjection("tool_result"); tx.entry(toolResultKind, conversationId, { data: resultData, model: [message], key: callId }); -const e = await harness.getEntry(summaryKind, id); // SummaryEntry | undefined -const any = await harness.getEntry(id); // Entry | undefined +const e = await harness.getEntry(summaryKind, id, call); // SummaryEntry | undefined +const any = await harness.getEntry(id, call); // Entry | undefined if (summaryKind.is(any)) any.data.summarizedThrough; // narrowed ``` @@ -279,7 +281,7 @@ const entryId = await conv.commit(tx => { const id = tx.entry(userKind, { model: [message] }); // 101 tx.task(generationKind, { state: { input: id } }); // 102 return id; -}); +}, call); // one batch [100–102] persisted and published; entryId === 101 here ``` @@ -300,15 +302,15 @@ history through X: ```ts // tool settlement, one commit -await ctx.commit(tx => { +await runtime.commit(tx => { tx.value(planMode).set(true); // 300 const result = tx.entry(toolResultKind, { data: resultData, model: [message] }); // 301 tx.settle(task, { result }); // 302 -}); +}, call); -await conv.value(model).set("claude-opus-5"); // 303, a later commit +await conv.value(model).set("claude-opus-5", call); // 303, a later commit -const b = await conv.fork(301); // has the result and plan mode; does not have the new model +const b = await conv.fork(301, undefined, call); // has the result and plan mode; does not have the new model ``` The builder throws if a rewindable conversation value/list write follows an entry; the plan fails @@ -435,25 +437,43 @@ Scratch is a task's working state while it runs: streamed frames, a checkpoint, per task, any number of addresses in it, durable across a restart, gone when the task settles. ```ts -const frames = scratchList("frames"); +import { AssistantMessageFrameEncoder, type AssistantMessageFrame } from "@earendil-works/pi-ai"; + +const frames = scratchList("frames"); const checkpoint = scratchValue("checkpoint"); // a new attempt starts clean; then each frame is one small scratch commit -await ctx.scratch(sc => { sc.list(frames).clear(); }); -for await (const frame of stream) { - await ctx.scratch(sc => { sc.list(frames).append(frame); }); +await runtime.scratch(sc => { sc.list(frames).clear(); }, call); +const { context: streamCall, cancel } = withCancel(call); +const stream = runtime.models.stream(model, request, streamCall); +const encoder = new AssistantMessageFrameEncoder(); +try { + for await (const event of stream) { + const frame = encoder.encode(event); + if (frame !== undefined) { + await runtime.scratch(sc => { sc.list(frames).append(frame); }, streamCall); + } + } +} finally { + cancel(); // stop this producer on any early exit, not only a task abort + await stream.result(); // join it; iterator exit alone is not provider completion } // after a restart, recover reads what made it to disk -const partial = await ctx.scratch(sc => sc.list(frames).read()); +const partial = await runtime.scratch(sc => sc.list(frames).read(), call); // settlement: the result becomes an entry; settle retires the scope in the same commit -await ctx.commit(tx => { +await runtime.commit(tx => { const id = tx.entry(assistantKind, { model: [assemble(partial)] }); tx.settle(task, { result: id }); -}); +}, call); ``` +Persist compact frames, not raw provider events with a growing `partial` snapshot on every token. +The encoder stores content deltas and necessary start/end metadata; terminal outcome/usage is recorded +separately. Tool/job scratch similarly appends explicit output operations or bounded checkpoints. +These are application records in ordinary lists, not a Chord storage codec. + Three rules: - A scratch commit writes one task's scratch and nothing else; a main commit never writes scratch. @@ -500,8 +520,9 @@ terminal done; never patched again generation: done, failed, aborted ``` `Tx.task`, `patch` and `settle` derive the role from the registered kind's status map and persist it -on every write. Storage indexes the stored role, so "live tasks in this scope" is one query that -decodes no state and does not need kind code. Callers cannot edit the role independently. +on every write. Storage selects live rows using the role index without inspecting task state or +running kind code. Loading their full records may decode state; terminal history is not decoded. +Callers cannot edit the role independently. ### 5.2 Kinds @@ -514,10 +535,10 @@ interface TaskKind, ctx: TaskContext): Promise; - recover(task: Task, ctx: TaskContext): Promise; - abort(task: Task, ctx: TaskContext): Promise; + }; // afterwards the kind mutates runtime.preview.state in place + execute(task: Task, runtime: TaskRuntime, call: Call): Promise; + recover(task: Task, runtime: TaskRuntime, call: Call): Promise; + abort(task: Task, runtime: TaskRuntime, call: Call): Promise; } ``` @@ -539,31 +560,37 @@ const generationKind = defineTaskKind({ ... }); -await c.config(generationKind).set({ model: "claude-opus-5" }); // or c.settings.set(...) for the generation -const cfg = await ctx.config(generationKind); // { model, thinking, ... } typed, read on the line +await c.config(generationKind).set({ model: "claude-opus-5" }, call); // or c.settings.set(...) for the generation +const cfg = await runtime.config(generationKind, call); // { model, thinking, ... } typed, read on the line await c.spawn({ prompt, values: { inherit: [generationKind.config.model], - set: [[generationKind.config.selectedTools, ["read", "grep"]]] } }); + set: [[generationKind.config.selectedTools, ["read", "grep"]]] } }, call); ``` -`ctx.config(kind)` reads every declared value in one batched read. A UI can list what a +`runtime.config(kind, call)` reads every declared value in one batched read. A UI can list what a conversation is configured with by walking the registered kinds' `config`. Values a plugin defines for itself are declared on the plugin's own kind the same way. -`TaskContext` is what an execution gets from the driver: +`TaskRuntime` is what an execution gets from the driver: ```ts -interface TaskContext { - readonly signal: AbortSignal; // fresh per call; fires on abort mark or close - commit(plan: (tx: ConversationTx) => T): Promise; // bound to the task's conversation (§9.2) - scratch(plan: (sc: ScratchTx) => T): Promise; // bound to the task's scratch (§4.6) - readonly preview: Tracker; // Chord delta tracker; mutate .state, the harness flushes after each scratch commit - sleep(untilMs: number): Promise; // throws AbortError on abort - config(kind: TaskKind): Promise>; // declared values, batched - conversation(id: Id): ConversationHandle; // for driving a child (§8.5) - readonly models: ModelRegistry; // provider streams - readonly tools: ToolRegistry; // tool implementations, by name - hooks(kind: TaskKind): HookRunner; // run this kind's points (§8.7) - readonly env: ExecutionEnv; // FileSystem & Shell: exec, files, temp dirs +interface TaskRuntime { + readonly taskId: Id; + commit(plan: (tx: ConversationTx) => T | Promise, call: Call): Promise; + scratch(plan: (sc: ScratchTx) => T | Promise, call: Call): Promise; + readonly preview: Tracker; // flushed after successful scratch commits + now(): number; // injected clock, not Date.now() in task code + sleep(untilMs: number, call: Call): Promise; + config(kind: TaskKind, call: Call): Promise>; + conversation(id: Id, call: Call): Promise; + abortTask(id: Id, call: Call): Promise; + waitForTask(id: Id, options: { budgetMs?: number } | undefined, call: Call): Promise; + getTask(id: Id, call: Call): Promise; // typed/batched reads also available + getEntry(id: Id, call: Call): Promise; + value(addr) / list(addr) // session state handles; async methods take Call + readonly models: ModelRegistry; // stream/deferred operations take Call + readonly tools: ToolRegistry; + hooks(kind: TaskKind): HookRunner; // run(point, input, call) + readonly env: ExecutionEnv; // existing Context-final methods; no wrappers } ``` @@ -571,12 +598,16 @@ The registry maps kind names to kinds; a replacement must understand the persist state of its live tasks. Open reads the live tasks anyway (`inspect()`), so an unknown live kind rejects before anything runs; terminal tasks of a forgotten kind are only a problem when read typed. -Typed access uses the kind as the witness, on the harness and handles outside a commit and on the -`Tx` inside one: +Typed access uses the kind as witness. Public/runtime reads are asynchronous and require Call; +transaction reads use the transaction view without another Call: ```ts -getTask(kind: TaskKind, id: Id): Task | undefined; -getTasks(kind: TaskKind, ids: readonly Id[]): ReadonlyMap>; +// public/runtime +getTask(kind: TaskKind, id: Id, call: Call): Promise | undefined>; +getTasks(kind: TaskKind, ids: readonly Id[], call: Call): Promise>>; +// inside Tx: same asynchronous reads, but no additional Call +getTask(kind: TaskKind, id: Id): Promise | undefined>; +getTasks(kind: TaskKind, ids: readonly Id[]): Promise>>; ``` The getter returns `Task` without casts, or `undefined` when the task is missing or of another @@ -590,38 +621,40 @@ Four rules, and they are the whole contract: execute from the top; a crash after it goes through recover. 2. You may block on the world: a provider stream, a process, a human inside a hook, a conversation you created. The driver runs executions concurrently; a blocked one holds up nothing. -3. Change your status or settle before returning. An execute that returns with its task unchanged - is a bug; the driver reports it and does not call it again. -4. Never wait on another task. If you need work done first, depend on it at creation (`after`). - If you spawn work you need the result of, it isn't another task: it's a conversation you drive - (§8.5), or a process you wait for. +3. Make a durable status transition or settle before returning, except when cancellation/close + unwinds the invocation. The driver checks a status epoch (§6.1), not final-status equality. +4. Prefer fixed `after` dependencies for prerequisite work. Child conversations use `drive(call)`; + separately running jobs may use the bounded `waitForTask` API (§8.3). Known self/dependency waits + reject (§6.2); never abandon an unfinished effect to release an invocation slot. Timing is the task's business: a retry stores `notBefore` in its state and sleeps in its own -execute (`ctx.sleep`, which throws on abort). Not everything on the signal throws: pi-ai's provider -stream completes with `stopReason: "aborted"`, and the generation settles from that in-band (§8.2). +execute (`runtime.sleep`, which observes Call cancellation). Not everything on the signal throws: +pi-ai may return `stopReason: "aborted"`. The kind handles its domain outcome normally; if a durable +abort mark won, its attempted commit rejects and the driver runs the fresh abort handler (§6.3). A recurring job sets its next `notBefore` and returns in a start status. The scheduler has no timers. A task is one logical operation, not one attempt. Its declared status graph may contain cycles: a generation revisits `streaming` across retries and deferred polls, and a schedule loops from `planned` through `running` back to `planned` under one stable id. Recovery uses only the current -status, state, role and scratch; it does not reconstruct the path taken. Returning with the same -status is still a contract violation caught by the driver, but changing statuses is not a generic -proof of progress and the harness does not try to diagnose a bad cycle. +status, state, role and scratch; it does not reconstruct the path taken. Returning to the starting +status after intermediate committed transitions is valid. Returning without any transition is a +contract violation; transitions are not a generic proof of progress or a diagnosis of bad cycles. ### 5.4 Dependencies mean terminal, not successful `after` lists tasks that must be terminal before this one starts. Terminal, not done: a task whose dependency failed or was aborted still starts and decides for itself what that means. Dependencies are set at creation, reference existing tasks in the same ownership tree, form no cycles, and are -never edited. A task that creates a successor and depends on it in the same commit is fine; a live -task adding a dependency on something it just spawned is not, and that is what rule 4 forbids. +never edited. A task may create a successor whose `after` includes itself in its settlement commit; +it cannot add new dependencies to itself. A dependency outside an attached subtree may require +another explicit drive; attaching one scope never implicitly executes a sibling scope. ### 5.5 post_tools joins an exchange The generation's settlement publishes the exchange atomically: ```ts -await ctx.commit(tx => { +await runtime.commit(tx => { const assistant = tx.entry(assistantKind, { model: [message] }); // calls [A, B] const tools = message.calls.map(call => tx.task(toolKind, { state: { call, assistant } })); @@ -630,7 +663,7 @@ await ctx.commit(tx => { state: { assistant, tools, inputs: task.state.inputs }, }); tx.settle(task, { assistant }); -}); +}, call); ``` Each tool executes and settles itself: hooks, execution, result entry, terminal status. It does not @@ -638,16 +671,16 @@ look at siblings, queues or context. post_tools starts when both are terminal, r typed getter, and decides what happens next: ```ts -async execute(task, ctx) { - await ctx.commit(tx => { - const tools = tx.getTasks(toolKind, task.state.tools); // Map> +async execute(task, runtime, call) { + await runtime.commit(async tx => { + const tools = await tx.getTasks(toolKind, task.state.tools); // Map> const outcomes = [...tools.values()].map(t => t.state.output); // ToolOutputState - if (tx.getTask(task.id)!.abort) { - resolveInputs(tx, task.state.inputs, "cancelled"); - return tx.settle(task, { cancelled: true }); // no successor - } if (outcomes.some(o => o.terminate)) { - resolveInputs(tx, task.state.inputs, "stopped"); + for (const inputId of task.state.inputs) { + const result = await tx.value(inputResult(inputId)).get(); + if (result?.status !== "running") throw new Error(`Invalid active input ${inputId}`); + tx.value(inputResult(inputId)).set({ ...result, status: "stopped" }); + } return tx.settle(task, { stopped: true }); // no successor } const added = outcomes.flatMap(o => o.addedTools ?? []); @@ -657,13 +690,18 @@ async execute(task, ctx) { if (handoff) tx.entry(handoffKind, { data: { text: handoff.handoff }, model: [handoffMessage(handoff.handoff)], head: "self", }); - const inputs = landPostToolsInbox(tx, task.conversationId, task.state.inputs); // writes + steer (§8.1) + const inputs = await landPostToolsInbox(tx, task.conversationId, task.state.inputs); // writes + steer (§8.1) tx.task(generationKind, { state: nextGeneration(task, { selectedTools, inputs }) }); tx.settle(task, {}); - }); + }, call); } ``` +`inputResult(id)` above is the sticky conversation value address for that input (§8.1), not a +driver operation. The abort handler updates the same owned inputs to `cancelled` and settles in one +commit. The runtime rejects a marked execute commit before its closure runs; no branch is needed +in the normal settlement. Inbox-placement helpers above implement exactly the mode table in §8.1. + Sequential tools are the same mechanism with `B after: [A]`. ```text @@ -689,215 +727,276 @@ provenance is a different link (§2.5) and creates no ownership. ## 6. Scheduling and cancellation -### 6.1 The driver +### 6.1 The commit line is the scheduler -The driver keeps one in-memory structure: `owned`, the tasks whose execute, recover or abort this -process is currently running. Everything else it needs is on disk. +Every driver decision runs on the same serialized line as commits. Only task methods, signal +listeners, watcher callbacks and telemetry callbacks run outside it. There is no polling loop, +`Wake`, effect-admission gate, worker pool or second scheduler. -```ts -class Driver { - private owned = new Map }>(); - private waiters: Array<{ scope: Scope; resolve: (o: "idle" | "closed") => void }> = []; - private wake = new Wake(); - private closed = false; +Process state consists of: - constructor(private storage: Storage, private kinds: Map) { void this.loop(); } +| State | Purpose | +|---|---| +| `live: Map` | current non-terminal tasks only | +| `running: Map` | one execute/recover/abort call per task id | +| `byConversation`, `dependents` | live-task membership and reverse live dependency edges | +| ownership links | immutable ancestry needed by live tasks, outstanding calls and attached roots | +| attached roots | stable conversation ids or one session-wide flag | +| waiters | temporary drive/task waiters, indexed by their scope or target | +| phase | open, stopping, closing, faulted, closed | - kick() { this.wake.set(); } // called by the harness after any commit that - // created or patched a task +Open seeds the complete live map before building readiness indexes, but starts no task. After a +successful batch, the line applies every task change to these indexes before scheduling or testing +idle. A settlement removes the task and its dependency edges; creating a successor in the same +batch never exposes an idle gap. `statusEpoch` advances on each actual committed status change, +including intermediate changes in a batch whose final status equals its initial status. State-only +patches, abort marks, same-status patches and failed commits do not advance it. - // attach a scope; resolves when its foreground is idle. The loop keeps serving it afterwards. - drive(scope: Scope): Promise<"idle" | "closed"> { - return new Promise(resolve => { this.waiters.push({ scope, resolve }); this.wake.set(); }); - } - - close() { this.closed = true; for (const o of this.owned.values()) o.controller.abort(); this.wake.set(); } - - private scopes = new Set(); // every scope ever attached, until close - - private async loop() { - while (!this.closed) { - for (const w of this.waiters) this.scopes.add(w.scope); - const convs = new Set(); - for (const scope of this.scopes) for (const c of await scope.conversations()) convs.add(c); - const live = await this.storage.scanTasks({ conversationIds: [...convs], live: true }); - - for (const task of live) { - if (this.owned.has(task.id)) continue; - const kind = this.kinds.get(task.kind)!; - if (task.abort) this.run(task, kind, "abort"); - else if (task.role === "start") { - if (await this.allTerminal(task.after)) this.run(task, kind, "execute"); - } - else if (task.role === "inflight") this.run(task, kind, "recover"); - } - - this.waiters = this.waiters.filter(w => !(w.scope.isIdle(live) && (w.resolve("idle"), true))); - await this.wake.wait(); - } - for (const w of this.waiters) w.resolve("closed"); - } - - private run(task: Task, kind: TaskKind, what: "execute" | "recover" | "abort") { - const controller = new AbortController(); - const ctx = this.contextFor(task, controller.signal); - const done = (async () => { - try { - if (what === "abort") { - const current = this.owned.get(task.id); // already running here? - if (current) { current.controller.abort(); await current.done; } - const fresh = await this.storage.getTask(task.id); - if (fresh && this.roleOf(fresh) !== "terminal") await kind.abort(fresh, ctx); - } else { - await kind[what](task, ctx); - const after = await this.storage.getTask(task.id); - if (after && after.status === task.status && !after.abort) { - this.report(new Error(`${task.kind}.${what} returned without changing status`)); - this.poison(task.id); - } - } - } catch (e) { - if (!isAbortError(e)) { this.report(e); this.poison(task.id); } // report → session watch event (§9.4) - } finally { - this.owned.delete(task.id); - this.wake.set(); - } - })(); - this.owned.set(task.id, { controller, done }); - } - - private allTerminal(ids: readonly number[]) { /* batched getTasks; every role terminal */ } - private roleOf(t: Task) { return t.role; } - private poison(id: Id) { /* remember id; skip it in later passes */ } -} -``` - -`Wake` is a promise that resolves when set and is replaced after each wait; nothing platform-specific: - -```ts -class Wake { - private resolve = () => {}; - private p = new Promise(r => (this.resolve = r)); - set() { this.resolve(); } - wait() { const p = this.p; this.p = new Promise(r => (this.resolve = r)); return p; } -} -``` - -`AbortController` and `AbortSignal` are web standards; the driver runs anywhere JS does. - -A `drive()` call attaches its scope and resolves when the scope's foreground is idle; the loop -doesn't stop there. A spawned child in the scope keeps being served after its parent's `drive()` -has returned, so its chain never stalls, and a fork that nobody attached is never touched. - -Calls are started, not awaited: a task enters `owned` when its call starts and leaves when it -returns, and the loop goes on to the next task. A blocked execute (a stream, a process, a child -conversation, a sleep) holds up nothing. A pass runs after any owned call returns and after any -commit that created or patched a task; between passes the driver waits. It decides nothing else; -whatever should happen next was written by a settlement. - -Two guards: an `AbortError` from a task the driver cancelled is the normal unwind, any other -rejection is reported and the task is poisoned (not re-run in this process); and an execute or -recover that returns with its task's status unchanged is reported and poisoned too, so a buggy -kind can't spin. - -Each call gets a fresh controller. The abort mark on disk is the intent; the controller is how the -intent reaches an execution in this process. The abort path joins the running call before running -the kind's `abort`, so effect and abort never write the same task concurrently, and re-reads the -task first because the effect may have settled in-band (a stream that ended with `aborted`). If -nobody owns a marked task, the mark alone is enough: the next process to open sees it. - -### 6.2 Scope and outcome - -A scope is the set of conversations a `drive()` call attaches to the loop, and the rule for when -that call resolves: - -```ts -interface Scope { - conversations(): Promise; // which conversations' tasks this drive runs - isIdle(live: readonly Task[]): boolean; // when this drive call returns "idle" -} - -// conversation.drive(): the conversation and every conversation owned from it, recursively, -// whether the owning task is still live or not (a spawned child keeps running here) -const conversationScope = (root: Id): Scope => ({ - async conversations() { - const out = [root]; - for (const c of out) - out.push(...(await storage.scanConversations({ ownedFrom: c })).items.map(x => x.id)); - return out; - }, - // idle when the root has no live foreground task. The foreground set (§5.6) reaches into owned - // conversations only through live foreground tasks, so if the root has none, the set is empty; - // a child owned by a settled spawn is driven here but never keeps us here. - isIdle: live => !live.some(t => t.conversationId === root && !t.background), -}); - -// harness.drive(): every conversation, done when no foreground task is live anywhere -const sessionScope: Scope = { - conversations: async () => (await storage.scanConversations({})).items.map(c => c.id), - isIdle: live => !live.some(t => !t.background), -}; -``` - -`ownedFrom: c` is "conversations whose `owner` task belongs to c", one indexed query. - -Forks are not reached through ownership, so nothing drives them by accident; each is its own tree. -Both drive methods start eligible foreground and background tasks in their scope, resolve when that -scope's foreground set is empty, and keep serving its attached background work afterwards. A -session-wide full-quiescence wait, if provided, may remain pending forever while a recurring -schedule is live. Both drive methods return `closed` if the harness closes first. There is no -"suspended": a task waiting on the world is an owned execution, and a host that doesn't want some -work to proceed in this process doesn't drive that scope. - -### 6.3 Cancellation is a mark, then cleanup +Normal execution does not scan storage after every task returns: ```text -abortTask(id) set the mark on one task; reject if terminal -abort(conv) one commit: mark every task in the conversation's foreground set; - cancel queued steer/followUp input; keep write and nextRun input +task commit on line: + persist whole batch + update live/dependency/conversation indexes + reconsider changed tasks, affected dependents and waiters + reserve eligible calls in running + publish complete state; leave line + dispatch calls/signals/listeners outside line + +invocation completion on line: + validate the invocation object matches running[taskId] + inspect that task in live (absence means it already settled) + check outcome and statusEpoch; remove its running slot + reconsider that task and affected waiters + reserve eligible calls; leave line + dispatch outside line ``` -Marks survive restart and are never shrunk as tasks settle. A settlement re-reads its task on the -line and writes its outcome regardless, but creates successors only if the mark is absent; in the -built-ins that check sits in the one helper that decides what happens next, and the harness -enforces it anyway: a commit issued by a marked task that creates a task or conversation fails, -and the kind is reported as buggy. An execution that ignores its signal delays its own cleanup (§6.1 joins it first), and a timeout -must not admit a second writer. `abort` must settle the task and write whatever durable outcome its own -work needs. An aborted tool writes its error result. An aborted generation may retain its partial for -display, but creates no tool tasks/results and its assistant output is excluded from later requests. +Only the current invocation may change its owned task's state/status or settle it. Host cancellation +may set the abort mark, but host or sibling status writes cannot mask a broken invocation's progress +guard. Task code receives a snapshot; later authority is checked against the line's live record and +invocation identity, not that snapshot. No extra storage read is needed for this check. -Either order of a race is correct: if post_tools settles first, the generation it created is marked -by the same abort commit; if the mark lands first, post_tools' abort settles it without a -successor. Background work: `abortTask(job)`, or `abort(childConversation)` for a detached child. +A task can be reserved only when it is served, live and not already in `running`: + +```text +marked → abort, irrespective of after +unmarked, role=start → execute when every dependency is terminal +unmarked, role=inflight → recover +``` + +Existing dependency ids are validated on creation. Once the complete live map is seeded, a valid +dependency absent from it is terminal. Dependency settlement wakes only its live dependents. +New attachments may inspect the current live map once; ordinary completions do not scan all tasks. +An outstanding invocation retains its slot even after it commits terminal status, until its method +actually returns. Different task ids execute concurrently. Retaining the slot is the abort join: +there is no separate invocation that waits for, or overwrites, its current owner. + +A normally returning execute/recover must have settled or advanced its captured `statusEpoch`. +`planned → running → planned` is valid; returning without any committed status transition is not. +Cancellation/close unwind is the exception. An abort handler must settle before returning. The driver +does not diagnose arbitrary changing-but-buggy cycles. + +Unexpected task-contract errors fail-stop the session, rather than poisoning individual tasks and +leaving their dependents waiting forever. Reject all pending drives and shutdown with the fault +immediately; stop admission and claims, signal owned calls outside the line, join them and close. +Do not manufacture terminal outcomes. The report includes task id, kind, method and error. Reopen +starts nothing: the host can inspect and mark a task before driving, or replace a broken kind with +one that understands and settles its stored state. Ordinary provider/tool failures are domain +outcomes handled by their kinds, not driver faults. + +### 6.2 Attachments and waiters + +`conversation.drive(call)` attaches one stable conversation id; `harness.drive(call)` attaches the +session. Each call registers an independent temporary waiter. Repeating a drive never retains another +scope object. Resolving or cancelling a waiter removes it and its signal listener, but keeps the +attachment, so background and recurring work continue. Session attachment subsumes serving filters; +separate conversation waiters still retain their own idle predicate. + +Serving follows ownership, never fork provenance. A conversation attachment serves all its owned +descendants, even when their owner task has settled. Discover membership starting from conversations +with live tasks, walking `conversation.owner → owner task's conversationId` upward. Batch named +ancestor reads through `getConversations` and `getTasks` and cache only the extracted immutable links; +never enumerate historical children or retain terminal owner payloads. +Owner liveness always comes from `live`, not the ancestry cache. Release unused ancestry when no +live task, outstanding invocation or explicit attachment needs it. + +```text +100,000 completed children; 2 live tasks +→ open reads 2 live tasks and only their required ancestry +→ later task completions use current indexes +→ no traversal of the 100,000 historical children +``` + +Idleness is decided on the line against current indexes: + +- Conversation: no direct live foreground task in that conversation. Reaching foreground descendants + requires a live foreground owner in the root, so this is equivalent to an empty foreground set. +- Session: no live foreground task anywhere, including foreground tasks in detached children. + +Both resolve `"idle"`, or `"closed"` when close wins. Session faults reject, not resolve idle. +A full-quiescence wait, if exposed separately, may never complete with a recurring task. + +Drive uses `call.abortSignal` to cancel only the actual registered waiter, not durable work. Register +and check cancellation on the line; a signal listener enqueues removal. Every outcome removes the +waiter and listener exactly once. Do not race an uncancellable drive promise with an outer promise. +Cancelling `prompt` after acceptance also leaves its accepted input durable. + +Task-bearing calls reject known self-waits: a foreground task cannot drive its own foreground scope, +nor wait for a task whose unresolved dependency path leads back to it. The latter also catches a +background task T driving a foreground D with `after:[T]`. Apply the checks before prompt acceptance +and reject new work that would introduce such an indexed wait cycle. A foreground ancestor reached +through a settled/background owner is not automatically the caller's foreground scope. Arbitrary +cycles hidden in plugin promises remain the task author's responsibility. Dependencies elsewhere +in an ownership tree may require another explicit attachment; never silently widen a drive scope. + +### 6.3 Cancellation is a request, then an abort invocation + +```text +abortTask(id, call) mark one live task; reject terminal +conversation.abort(call) + one commit: mark its current foreground ownership closure; + cancel queued steer/followUp in affected conversations; keep write/nextRun +``` + +`abort: true` is a durable request, not terminal status. When that mark commits, the current +execute/recover invocation loses mutation authority immediately. Its later main and scratch writes +reject `TaskCancelled` on the line, before running their builders. This is normal control flow, not +a kind bug. The signal is dispatched after leaving the line; standard effects and waits observe it +cooperatively. No new admission gate is required. An effect may start in the interval before signal +delivery and then be cancelled; cancellation neither rolls it back nor promises exactly-once I/O. + +```text +100 task streaming; execute invocation A owns it +110 abort=true commits; A's mutation authority revoked + line releases; A's signal fires + stream returns/throws; execute finishes local finally cleanup and returns + A's completion enters line; remove A, reserve fresh abort invocation B + line releases; kind.abort(task, runtime, callB) runs +120 partial/error outcome and cancelled input results, if the kind needs them +121 task terminal aborted; scratch retired in the same commit + abort method returns; remove B +``` + +Normal execute/recover settlement needs no mark check and no `tx.origin` branch. If its settlement +wins first, the task is already terminal and a later abort rejects. If the mark wins first, the whole +normal settlement rejects and the driver runs `abort`. A provider's in-band `stopReason: "aborted"` +is not itself a durable mark: without a mark, the kind must commit its failure/retry outcome; with +a mark, its attempted domain commit rejects `TaskCancelled` and cleanup runs in the fresh invocation. +An invocation cancelled only by a task-local deadline may still commit its domain result. + +The driver recognizes its own `TaskCancelled` and the invocation's known cancellation reason, +including supported cancellation errors correlated with the mark/close. An arbitrary exception does +not become harmless merely because a signal fired. Hooks propagate cancellation rather than treating +it as an ordinary fail-open/fail-closed decision. Expected close/stale callback rejections are handled +at their boundary; uncertain persistence still faults the session. + +`abort()` receives a new identity, telemetry span and initially active controller. Repeated marks +never cancel an already-running abort handler. Only close/fault cancels it. Abort ignores `after`, +may perform cleanup effects and write outcomes, but cannot create successor tasks/conversations or +queue future work via accept/nextRun. A target with no owned call and no attached scope remains +marked until driven; marking it does not implicitly attach unrelated work. + +Fresh cleanup can use only durable records. A tool that creates a cancellable background job records +the job id and its cancellation policy on its own task in that same commit. Its abort handler uses +those references, never locals from execute or a post-mark catch that tries another mutation. Child +conversation ownership is already durable. Built-in cleanup marks only the child's live foreground +tasks; it does not call the public conversation abort operation that also cancels queued inputs. +This task-only cleanup policy applies on every abort invocation, including after crash/reopen, so +shutdown cannot accidentally discard preserved queues during recovery. Cleanup's cancellation request treats an already-terminal +target as finished, using an atomic mark-if-live operation internally or handling that specific public +terminal result; a read-then-mark does not close the race. + +Scratch before the mark is authoritative. Harness-owned sinks/progress bridges own and handle their +pending scratch promises, drop late cancelled output and drain callbacks before invocation completion. +Task code must await/catch its own scratch writes; no `void runtime.scratch(...)` without error handling. +Do not turn rejected scratch into a successful no-op or race away an unfinished tool/hook. If code +ignores the signal and never returns or reaches a rejecting harness operation, abort/close can wait +forever. An isolate kill boundary is future work; RPC cancellation alone is not forced termination. ### 6.4 Open, close, shutdown, delete -**Open** opens storage, reports unregistered entry kind strings (§2.1), checks live task kinds -against the registry, exposes `inspect()` (live tasks by role) and queries, and starts nothing. -Unknown entries remain fully usable for context from their stored facets; an unknown live task kind -still rejects. **Close** stops admitting commits, cancels owned executions, waits for them and for -admitted persistence, releases storage; it writes no outcomes, so unfinished work resumes on the -next open. **Shutdown** marks all live work, drains queued input, drives cancellation to settlement, -then closes. **Delete** of a conversation rejects while its ownership subtree has live work, then -rejects new work; entries an independent fork inherits are never erased. +**Open** reports unregistered entry kinds, validates live task kinds, seeds live/ownership indexes, +exposes inspection/queries, and starts nothing. Unknown entries preserve context through stored facets. -### 6.5 Telemetry, storage version, kill boundary +**Close** runs one nonpersistent line operation that enters `closing` and stops admission and claims. +All earlier line operations have already finished. Later storage operations reject; internal invocation +completion messages remain admitted. Outside the line, signal and join every owned call, including +terminal tasks whose methods are still unwinding, then close storage. No new abort marks or outcomes +are written. Never hold the line while awaiting a task that may itself be awaiting a line operation. -Telemetry is one span per task call (execute, recover, abort) from the driver and one per commit -from the line, tagged with task id, kind, conversation and outcome; provider streams and tool -executions nest under the task span. The host supplies the tracer; nothing is recorded in storage. +**Shutdown** runs one line operation: enter `stopping` and commit one batch marking all live tasks. +Queued inputs and their queued result records remain untouched, including in idle conversations. +There is no inbox sweep or special storage query. After this batch, only abort cleanup mutations are +admitted; an abort handler already running remains authorized. Cancelled execute/recover mutations +reject. Abort handlers resolve their already-running input groups, but built-in child cleanup only +marks tasks and never drains queues (§6.3), including after restart. Serve cancellation session-wide; +once both live tasks and running invocation slots are empty, close. A failed abort handler faults +and rejects shutdown rather than leaving an eternal drain. Preserved queues create no work by +inference on reopen/drive; a later acceptance places eligible queued input at the defined boundary. -Session metadata carries a storage version, checked at open; a mismatch rejects with the versions, -and a backend exposes `migrate(from, to)` to be run before open. Durable state is provider -messages, small task states and values, so migrations are rare and mechanical. +Repeated lifecycle calls share completion. Task-token-bearing close/shutdown/session-wide drive reject +rather than waiting on their own invocation. Close joins regardless of its caller's signal; once +shutdown is admitted, caller cancellation cannot abandon marking, joining or closing. Explicit close +may interrupt shutdown; shutdown then rejects and remaining marked tasks recover later. Lifecycle +entry points are exceptions to ordinary admission so repeated close works while already closing. -An execution that ignores its signal delays its own abort indefinitely, because the driver joins it -before running the kind's `abort` (§6.1). First release: everything runs in-process and -cooperatively, as in the lane harness. Later, plugin code moves into an isolate with an API -membrane (mobile-handoff `02-plugins`), which becomes the kill boundary for it: dispose the -isolate, the effect rejects, the abort path settles from the host side. Nothing in this design -changes for that; the types are already JSON across the boundary and the sink and `ExecutionEnv` -are already the objects that would be proxied. +**Delete** rejects while either live tasks or outstanding invocation slots remain in the subtree, +then rejects new work. Entries inherited by independent forks are never erased. + +### 6.5 Call, telemetry and storage version + +`Call` is an alias of Chord `Context`, not a second context implementation. The driver installs a +private typed identity alongside its signal; span/budget derivation preserves it without casts: + +```ts +import type { Context } from "@earendil-works/chord"; +import { createContextKey, withAbortSignal, withContextValue } from "@earendil-works/chord/context"; + +export type Call = Context; + +interface Invocation { + readonly taskId: Id; + readonly method: "execute" | "recover" | "abort"; + readonly controller: AbortController; + readonly initialStatusEpoch: number; +} + +const invocationKey = createContextKey("pico.invocation"); // private + +// parentWithoutCallerCancellation retains the host telemetry parent, not the drive waiter's signal. +const call: Call = withContextValue(invocationKey, invocation, withAbortSignal( + invocation.controller.signal, parentWithoutCallerCancellation, +)); +const identity = call.value(invocationKey); // Invocation | undefined, no assertion +``` + +The line compares object identity with `running.get(identity.taskId)`. `TaskRuntime` is bound to its +expected invocation and rejects absent/foreign/stale identities; public host calls may have no token +while the session is open. Returned ordinary conversation handles also receive Call on every async +operation, so they preserve attribution without a task-specific facade. Capturing a public handle and +deliberately supplying `BACKGROUND_CONTEXT` remains an in-process cooperative escape, not isolation. + +Every async Harness/Conversation/TaskRuntime method takes a required final `call: Call`, including +reads, waits and mutations. Options before it are passed as `undefined` when unused. Pure accessors, +synchronous registration/disposal and operations inside Tx do not require a Call. `TaskRuntime` and +`ToolRuntime` name capabilities; `call` names cancellation/telemetry/identity; model context keeps its +existing meaning. There is no `TaskContext`, `CallContext`, admission helper or signal-injecting facade. + +The driver starts one telemetry span per execute/recover/abort invocation, independent of the drive +caller's cancellation. Commits use the supplied Call's current telemetry parent; nested provider/tool/ +hook spans derive and forward a Call. Existing Context-taking env, provider and hook boundaries +interpret the signal rather than merely carrying it. Node operations without interruptible APIs may +only check before/after; hooks and plugin code must cooperate. Commit admission uses lifecycle, mark, +method and identity, not a blanket signal check: cancellation never abandons admitted persistence. + +RPC transports cancellation and selected telemetry metadata, never the invocation object. A trusted +host-side binding reattaches the local identity for remote task calls. A cancelled RPC waiter does not +prove remote code stopped; no slot is released until completion or a real kill boundary confirms it. + +Telemetry schema names/attributes are a separate implementation package; the host supplies the tracer +and no spans are stored in session data. Storage metadata carries a version; mismatches reject and a +backend exposes migration before open. None of these mechanisms require a renewable session lease. ## 7. Storage @@ -914,17 +1013,19 @@ clear markers, scratch scopes. Context is not stored (§2.2). | Caller | Reads | Never | |---|---|---| -| Driver | live tasks in a scope; their `after` ids in one batch | terminal history | +| Driver open/attach | complete live-task seed and named owner records; then committed-batch index updates | repeated full scans after calls, historical child traversal | +| Shutdown | current live-task index | inbox sweep, historical conversation traversal | | Context | newest head at a target; fork-aware range from its returned boundary | unrelated transcript history | | post_tools | tool tasks by id | sibling scans | | UI | a transcript page before/after an id, with a limit | the whole conversation | -| Validation | named task/conversation records, entry headers | unrelated content | +| Validation | named task/conversation records and full entries by id | unrelated records | | Fork | transcript and rewindable conversation history ≤ the entry id | today's state filtered | | State consumer | latest version by index; a bounded list range | a replay of unrelated records | | Reopen (SQLite) | live tasks and the conversations they need | every historical transition | "Latest" is an indexed descending query with limit 1, never load-and-take-last. Filters apply -before limits. +before limits and decoding. Entries are immutable and read whole; there is no separate header or +projection API. Named owner-task lookups use existing batched `getTasks`, even for terminal owners. ```ts interface Page { readonly items: readonly T[]; readonly next?: Id; readonly readAt: Id } @@ -933,12 +1034,11 @@ interface Cursor { readonly after?: Id; readonly before?: Id; readonly limit: nu interface ConversationQuery extends Cursor { readonly parent?: Id; readonly ownedFrom?: Id } interface EntryQuery extends Cursor { readonly conversationId: Id; readonly kind?: string; readonly key?: string; readonly from?: Id; readonly through?: Id } // inclusive logical range -interface TaskQuery extends Cursor { readonly conversationIds: readonly Id[]; readonly live?: boolean; +interface TaskQuery extends Cursor { readonly conversationIds?: readonly Id[]; readonly live?: boolean; readonly role?: TaskRole; readonly kind?: string; readonly abort?: boolean } interface ListQuery extends Cursor { readonly at?: Id } // at = an entry id (§3.2) interface ValueQuery { readonly scope: Scope; readonly namespace: string; readonly after?: string; readonly limit: number } // keys, ordered -type EntryHeader = Omit; // context fields without arbitrary plugin data interface Version { readonly seq: Id; readonly value: T } interface Element { readonly id: Id; readonly value: T } @@ -950,12 +1050,11 @@ interface Storage { scanConversations(q: ConversationQuery): Promise>; getEntries(ids: readonly Id[]): Promise>; - getEntryHeaders(ids: readonly Id[]): Promise>; - scanEntries(q: EntryQuery): Promise>; // fork-aware - newestHead(conversationId: Id, at: Id): Promise; // target-capped, fork-aware + scanEntries(q: EntryQuery): Promise>; // full entries, fork-aware + newestHead(conversationId: Id, at: Id): Promise; // full entry, target-capped, fork-aware getTasks(ids: readonly Id[]): Promise>; - scanTasks(q: TaskQuery): Promise>; + scanTasks(q: TaskQuery): Promise>; // omitted conversationIds: entire session getValue(addr: Value, at?: Id): Promise | undefined>; // at = an entry id scanValues(q: ValueQuery): Promise }>>; @@ -1033,23 +1132,49 @@ one line per batch: {"type":"task","conversationId":1,"kind":"generation","status":"pending"}]} ``` -Write all bytes, then publish in-memory changes, never before. On open: validate complete batches, -rebuild maps and `lastSeq`, discard only an unterminated final line, fail on a malformed complete one. Durability -against process crash is required; fsync policy is a backend option. +Write all bytes, then publish in-memory changes, never before. Durability against process crash is +required; fsync policy is a backend option. -**Values as deltas.** A JSONL file is one ordered stream replayed from the start, which is what -Chord's delta module (`track` / `flush` / `apply`, with a path-compressing encoder per stream) is -built for. A `value.set` is written as the full value when it is the address's first version or -small, and otherwise as the ops against the previous version; replay applies either. That keeps a -large, frequently updated value (a plugin's structured state) linear in the file instead of -rewriting it on every set. Lists don't need it: elements are immutable and `remove`/`clear` are -already one op. A line encoded as ops is only meaningful after the lines before it, which is fine -for a log and is why this stays out of SQLite. +**Whole values, ordinary list operations.** Every `value.set` stores the complete supplied value, +just like the lane harness. There is no automatic diff, Chord encoding or hidden replay chain for a +value. Lists store append/remove/clear operations. Repeatedly setting a growing whole value can write +quadratic bytes; use compact appended frames/output operations or bounded checkpoints for streaming. +Chord deltas are used only for preview/watch delivery (§9.4), not storage serialization. -**Scratch** goes to `session.scratch/`, one file per scope, the same delta encoding, so a -streamed assistant text or a growing tool output is an append per flush rather than a rewrite. -Retirement in the main file is authoritative; the unlink happens afterwards, and a delayed unlink -must never remove a deliberately reused scope. +**Scratch** goes to `session.scratch/`, one file per task, with the same plain batch format. +Retirement in the main file is authoritative; unlink happens afterwards. Task ids are never reused. +A retry keeps its task and clears its scratch through a persisted list operation, not by deleting +the only record of its latest sequence. + +**Recovery across files.** Main and scratch commits share one session sequence, so gaps inside any +one file are normal: + +```text +main: 100 create generation +scratch: 101–150 frame appends +crash before settlement +reopen: main says generation live; replay its scratch; lastSeq=150; next write=151 + +alternatively: +main: 151 settle generation, retire scratch +crash before unlink +reopen: generation terminal; ignore old scratch; lastSeq=151 +``` + +On open: + +1. Replay complete main batches, reconstructing task liveness and scratch retirement. +2. Replay scratch files only for surviving live tasks. Ignore retired scratch entirely, even if + malformed; it cannot affect state or the sequence high-water mark. +3. Validate increasing, nonoverlapping retained batch ranges and task/scope references. Per-file gaps + are valid; do not require the main log to contain discarded scratch positions. +4. Set `lastSeq` to the maximum complete batch endpoint across main and surviving scratch files, + including scratch clear/remove writes. A retired scratch file's sequences are already below its + later main settlement, so deleting it loses no high-water information. +5. Discard only unterminated final lines in replayed files and physically remove those torn suffixes + before appending again. Any malformed complete main or surviving scratch batch makes open fail. + +No new index, journal or cross-file transaction is required: each commit still writes exactly one file. ### 7.5 SQLite @@ -1063,13 +1188,14 @@ table: entries by (conversation, id), (conversation, kind, key, id) and a partia partial index on non-terminal roles; value_versions by (conversation, namespace, key, seq); conversations by parent and by owner. One commit is one transaction: number the writes from `lastSeq + 1`, insert rows, record the boundary, advance `lastSeq`. No full rewrite of anything per -append. +append; no shutdown-specific inbox index. Values are stored whole, current and historical: a point read must not become a replay chain, and disk is cheaper than that. Scratch is rows in the `scratch` table keyed by (task, address), written in the task's own small transactions, and retired by `DELETE ... WHERE task = ?` inside the settlement transaction: atomic with the result and the terminal status, nothing to unlink. Reopen -reads live tasks and what they reference, not history. +reads live tasks and named records they reference (including terminal owners), not accumulated +history. Once ownership links are extracted, terminal owner payloads need not remain resident. ## 8. Built-in flows @@ -1102,6 +1228,9 @@ type InputResult = readonly placementEntryId?: Id; readonly reason?: string }; ``` +`inputResult(id)` denotes the sticky conversation Value address with namespace `pi.inputResult` +and key `String(id)`, typed as `InputResult`. It is ordinary stored state, not a driver callback. + Acceptance always appends an inbox element and writes `inputResult[inputId]` in the same commit. When the conversation is idle, that commit immediately places and removes the item and creates a generation; append plus remove folds to no inbox watch operation. When it is busy, the item remains @@ -1155,7 +1284,7 @@ execute: if a collapse is needed by threshold: { create collapse C; create g else if sent ≠ desired: system entry with the diff, per section ("the plan_mode guidance now applies: …", "the cwd section changed: …") and toolsAdded / toolsRemoved as full definitions (a changed schema counts); status streaming } - project the context; before_request; re-check status and mark + project model context; before_request with Call; runtime validates writes on the line stream; frames to scratch outcome: calls { assistant; tools; post_tools carrying inputs; settle done } @@ -1166,8 +1295,8 @@ outcome: overflow { settle failed(overflow); create collapse C; C's settlement creates G' } failure { retain partial/error outcome for display if useful; resolve inputs failed; settle failed; no tools/results } - aborted stream ends with stopReason "aborted": { retain partial for display if useful; - resolve inputs cancelled; settle aborted; no tool tasks/results } + aborted provider returns in-band: attempt domain failure/retry commit; + durable mark present → TaskCancelled; execute unwinds; driver calls abort below ``` **Prompt and tools.** Two things the lane harness fuses are kept apart. What the client wants is @@ -1192,13 +1321,16 @@ carrying the attempt count, so nothing live ever waits on the collapse. recover streaming: frames in scratch → retain/publish the partial for display; settle without tools/results no frames → retry within budget, else fail recover deferred: execute again; the handle is in state -abort: only reached if the effect didn't settle in-band (e.g. it was between attempts): - retain partial from scratch if useful; settle aborted without tools/results +abort: fresh invocation after execute/recover returned: + read committed scratch; optionally retain partial for display, excluded from requests; + in one commit mark owned input results cancelled, record known usage, settle aborted; + no tool tasks/results; settlement retires scratch ``` Usage is recorded for failed, deferred and discarded attempts too; a missing report is unknown cost, -not zero. The record is a session list, `pi.usage`, one element per attempt (conversation, task, -model, tokens, cost), appended in the settlement commit together with an update of the session +not zero. A report received only after the cancellation cutoff may not have reached scratch and is +therefore unknown to fresh abort cleanup. The record is a session list, `pi.usage`, one element per +attempt (conversation, task, model, tokens, cost), appended in the settlement commit together with an update of the session value `pi.usage.totals`, so stats are a point read and never a fold. Tools and jobs append the same way through the sink's `usage`. Persistence and invariant failures are not provider errors: they fault the session. @@ -1210,7 +1342,7 @@ tool execute: before_tool → allow(args) | block(reason, terminate?) (a human approval waits inside the hook) block / invalid / unknown tool → own error result, no invocation { status running; effective args; replay policy } - invoke the tool with the signal and the output sink (§9.3); the sink's ops go to scratch + invoke tool(callId, params, out, runtime, call); the sink owns its scratch promises (§9.3) after_tool outside the line { result entry from the folded sink; usage; settle done with result id and ToolOutputState } ``` @@ -1220,7 +1352,9 @@ ordinary tool throw is an error result, not a cancellation. ```text recover: replay only if the stored policy allows; else an interrupted result from the checkpoint -abort: join own execution; drain scratch; own aborted error result; settle +abort: previous invocation already returned; read durable scratch and cleanup references; + mark owned child foreground tasks/non-detached jobs, without touching child queues; + write aborted error result; settle ``` post_tools is the code in §5.5: read the tool tasks, stop on terminate, write the handoff if one @@ -1233,21 +1367,34 @@ budget runs out, the call settles now with what it has, `out.delegate(job)` and work continues as the job: ```ts -const job = await ctx.conversation.commit(tx => tx.task(jobKind, { background: true, - state: { cmd, cwd, origin: { tool: "bash", task: ctx.taskId, callId: toolCallId }, notify: false } })); -const done = await ctx.waitForTask(job, { budgetMs: ctx.budgetMs, signal }); // false when the budget ran out -const o = await ctx.jobOutput(job); // ToolOutputState from the job's preview or terminal state -out.replace(o.text); out.capture(o.truncation); -if (!done) { out.delegate(job); out.diag("info", `still running as job ${job}; use job wait / status / stop`, "budget"); - await ctx.conversation.commit(tx => tx.patch(job, { state: { notify: true } })); } +const job = await runtime.commit(tx => { + const id = tx.task(jobKind, { background: true, + state: { cmd, cwd, origin: { tool: "bash", task: runtime.taskId, callId: toolCallId } } }); + tx.patch(runtime.taskId, { state: { jobId: id, cancelJobOnAbort: true } }); + return id; +}, call); +const done = await runtime.waitForTask(job, { budgetMs: runtime.budgetMs }, call); +const output = await runtime.jobOutput(job, call); +out.replace(output.text); out.capture(output.truncation); +if (!done) { + out.delegate(job); + out.diag("info", `still running as job ${job}; use job wait / status / stop`, "budget"); +} ``` The budget is the generation's config (`budgetMs`, default a few minutes); the tool kind passes it -as `ctx.budgetMs`. `waitForTask` throws on abort like `sleep`; a tool that started a job for a -non-backgrounded call aborts it then (`abortTask`). A tool that awaits arbitrary work instead of -delegating is raced against the budget by the tool kind, which on expiry creates a job that adopts -the promise and retargets the sink into the job's scratch; that path is best-effort across a crash -(the job recovers as `lost`). +as `runtime.budgetMs`. `waitForTask` observes Call cancellation and unregisters its waiter. A marked +tool cannot mutate from its execute catch; its fresh abort handler reads the stored job reference and +cancels non-detached work. Explicit backgrounding records a different cancellation policy. Successful +delegation settlement atomically records that the job is detached and requests its completion notice; +if the job has already finished, that transaction places the notice itself. Use a separate sticky +notification record rather than patching the state of a running job from its former owner. + +**Remaining budget integration question:** adopting arbitrary non-delegating tool work after timeout +requires an explicit transfer of effect and sink ownership before the source invocation releases its +slot. Racing and abandoning the tool promise is not a correct implementation. Preserve this capability +for a separate ownership design; crash recovery remains `lost`. The job-first path above needs no +promise adoption and is the initial implementation path. `new_context` is a tool that calls `out.handoff(message)`; the reset happens here, after the exchange is complete, never inside the tool, because a head landing mid-exchange would cut it in half. Conflicting handoffs reject rather than pick an order. @@ -1296,21 +1443,21 @@ type SubagentCommand = ```text run: { create child conversation (this task owns it); initial values; accept prompt → inputId; first generation } - await child.drive() // blocked on the world; the tool stays inflight - child.result(inputId); { result entry; settle } - recover: the child exists; drive it again - abort: the mark on this tool marks the child's foreground set through ownership; error result + await child.drive(call) // cancels this waiter, not the child + child.result(inputId, call); { result entry; settle } + recover: the child exists; drive it again with the fresh Call + abort: mark child's foreground tasks only, preserve its queues; write own error result and settle spawn: the same creation commit; settle at once with the child's id -send: child.accept(text) → queued if the child is busy +send: child.accept(text, undefined, call) → queued if the child is busy status: the child's tail and live tasks -wait: await child.drive(); child.result(the last sent inputId); settle (recover: drive again) -stop: child.abort() +wait: await child.drive(call); child.result(lastInputId, call); settle (recover: drive again) +stop: child.abort(call) ``` `id` is whatever `spawn` returned, passed back by the model; the tool resolves it with -`harness.conversation(id)`. No map anywhere, nothing to rebuild after a restart. Between `spawn` and -`wait`, whatever drives the parent's tree drives the child. A spawned child is owned by a task that +`runtime.conversation(id, call)`. There is no separate subagent registry to rebuild after a restart. +Between `spawn` and `wait`, whatever drives the parent's tree drives the child. A spawned child is owned by a task that is already terminal, so it is not in the parent's foreground set and aborting the parent leaves it alone; `stop` is the way to cancel it. Inherited context for a child is the parent's context at the last complete exchange before the launching one. @@ -1321,9 +1468,8 @@ A job is a background task that runs a process. Its state: ```ts interface JobState { - cmd: string[]; cwd: string; limits?: ShellOutputLimits; + cmd: string; cwd: string; limits?: ShellOutputLimits; origin?: { tool: string; task: Id; callId: string }; // the call that started it, so UIs render it with that tool's component - notify?: boolean; // the call returned early: append a notice on completion every?: number; notBefore?: number; rerun?: "safe"; // recurrence and recovery policy startedAt?: number; output?: ToolOutputState; exitCode?: number; } @@ -1333,55 +1479,34 @@ Starting one is creating the task; a tool does it in its execute (§8.3), a UI i ```ts const jobId = await conv.commit(tx => - tx.task(jobKind, { background: true, state: { cmd: ["npm", "test"], cwd, every: 6 * 3600_000 } })); + tx.task(jobKind, { background: true, state: { cmd: "npm test", cwd, every: 6 * 3600_000 } }), call); ``` -```ts -const jobKind: TaskKind = { - kind: "job", - initialStatus: "planned", - roles: { planned: "start", running: "inflight", exited: "terminal", killed: "terminal", lost: "terminal" }, - - async execute(task, ctx) { - if (task.state.notBefore && task.state.notBefore > Date.now()) await ctx.sleep(task.state.notBefore); - await ctx.commit(tx => tx.patch(task.id, { status: "running", state: { ...task.state, startedAt: Date.now() } })); - const output = scratchValue("output"); - const result = await ctx.env.exec(task.state.cmd, { // blocked on the world - cwd: task.state.cwd, - capture: { limits: { maxBytes: 64_000, maxLines: 2000, retain: "tail" }, spill: true }, - onUpdate: update => void ctx.scratch(sc => sc.value(output).set(applyShellOutputUpdate(sc.value(output).get(), update))), - }, { signal: ctx.signal }); - await this.finish(task, ctx, result); - }, - - async recover(task, ctx) { // the process died with the old owner; nothing to adopt - if (task.state.rerun === "safe") return this.execute(task, ctx); - await ctx.commit(tx => tx.settle(task.id, "lost", task.state)); - }, - - async abort(task, ctx) { // only reached if exec didn't return on the signal - await ctx.commit(tx => tx.settle(task.id, "killed", task.state)); - }, - - async finish(task, ctx, result) { - const output = await ctx.scratch(sc => sc.value(scratchValue("output")).get()); - await ctx.commit(tx => { - if (ctx.signal.aborted) return tx.settle(task.id, "killed", { ...task.state, output }); - const r = getOrThrow(result); - if (task.state.notify) // the call that started it returned early (§9.3): tell the model - tx.entry(noticeKind, { - model: [noticeMessage(`job ${task.id} (${task.state.origin?.tool}: ${task.state.cmd.join(" ")}) finished, exit ${r.exitCode}`)], - }); - if (task.state.every) // recurring: same task, next time - tx.patch(task.id, { status: "planned", state: { ...task.state, notBefore: Date.now() + task.state.every } }); - else - tx.settle(task.id, "exited", { ...task.state, output, exitCode: r.exitCode }); - }); - }, -}; +```text +roles: planned=start; running=inflight; exited/killed/lost=terminal +execute: sleep until notBefore with Call; { status running; startedAt = runtime.now() } + env.exec(command, options, call), with bounded capture and spill + harness progress bridge owns scratch writes, handles cancellation and drains callbacks + after process and progress bridge return: + { if notification requested: append notice and consume notification request; + if recurring: status planned, notBefore = runtime.now() + every; + otherwise: settle exited with bounded output and exitCode } + a post-mark write rejects TaskCancelled; no in-band killed settlement +recover: safe rerun policy → execute again with the fresh Call + otherwise { consume pending notice request; append lost notice if requested; settle lost } +abort: previous invocation and its process/progress callbacks have returned + read committed output from scratch; + { consume pending notice request; append killed notice if requested; settle killed } ``` -`ctx.env` is the harness's `ExecutionEnv` (`FileSystem & Shell`); `exec` runs the command, captures +The progress bridge is harness-owned sink plumbing (§6.3), not a plugin-created detached promise. +It observes every scratch result, suppresses only expected cancellation/close rejections, reports +persistence faults, and is joined with process execution. Notification requests are current sticky +state keyed by job id; consuming them and publishing the notice is atomic on every terminal path, +including exited, lost and killed. If delegation follows settlement, delegation publishes that same +outcome instead. The two orderings must not change notification behavior. + +`runtime.env` is the harness's `ExecutionEnv` (`FileSystem & Shell`); `exec` runs the command, captures output through `onUpdate` (here into scratch, so a watcher can show it live) and returns when it exits or the signal fires. `running` is inflight, so a restart goes through `recover`; since the shell doesn't hand out an adoptable process, recover reruns if the job declared that safe and @@ -1416,7 +1541,7 @@ interface ToolHooks extends HookPoints { after_tool: { input: { toolCallId: string; output: ToolOutputState }; output: void }; } interface CollapseHooks extends HookPoints { - before_collapse: { input: { reason: "manual" | "threshold" | "overflow"; context: readonly Entry[] }; + before_collapse: { input: { reason: "manual" | "threshold" | "overflow"; entries: readonly Entry[] }; output: { decline?: boolean; instructions?: string; summary?: string } }; } @@ -1424,12 +1549,12 @@ const generationKind: TaskKind = { ..., hooks: const toolKind: TaskKind = { ..., hooks: { before_tool: { failClosed: true }, after_tool: {} } }; // a plugin registers a handler -harness.hooks.on(toolKind, "before_tool", async ({ toolName, args }) => { - if (toolName === "bash" && !(await ui.approve(args))) return { block: { reason: "denied" } }; // may wait for a human +harness.hooks.on(toolKind, "before_tool", async ({ toolName, args }, call) => { + if (toolName === "bash" && !(await ui.approve(args, { signal: call.abortSignal }))) return { block: { reason: "denied" } }; // may wait for a human }); // the kind runs it -const decision = await ctx.hooks(toolKind).run("before_tool", { toolCallId, toolName, args }); +const decision = await runtime.hooks(toolKind).run("before_tool", { toolCallId, toolName, args }, call); ``` A handler can be scoped to a conversation instead of the whole harness; for `system_instructions` @@ -1446,8 +1571,10 @@ when it attaches. Handlers run in registration order, each seeing the merged output so far; the kind decides what to do with the result. A handler runs outside the line and its decision is applied inside a commit -that re-checks the task's status and mark. Handlers may run again after a crash, so their external -side effects need their own idempotence. `failClosed` says what a throw means: for `before_tool` +whose runtime validates current invocation authority. Handlers receive a required final Call, +forward it to waits/effects, and must return before their parent task invocation releases ownership. +Cancellation control errors bypass ordinary hook failure policies. Handlers may run again after a +crash, so their external side effects need their own idempotence. `failClosed` says what a throw means: for `before_tool` a throw blocks the tool; for the others a throw is reported and skipped. `before_tool` may wait as long as it likes (a human approval is a hook that waits); the task is still `planned` meanwhile and a crash simply asks again. A plugin kind declares its own points the same way, and a plugin @@ -1461,48 +1588,51 @@ listeners are not hooks: they observe, never decide, and never await a commit on ```ts interface ConversationHandle { readonly id: Id; - snapshot(): Promise; - accept(input, options?: { requestId?: string }): Promise<{ inputId: Id }>; - prompt(input, options?): Promise; // accept + drive + result(inputId) - result(inputId: Id): Promise; - drive(options?: { signal? }): Promise<"idle" | "closed">; - steer(input): Promise<{ inputId: Id }>; - followUp(input): Promise<{ inputId: Id }>; - nextRun(input): Promise<{ inputId: Id }>; - write(kind: EntryKind, input: EntryInput): Promise<{ inputId: Id }>; - cancelQueued(inputId: Id): Promise<"cancelled" | "not_found">; - abort(): Promise; - collapse(options?: { instructions? }): Promise; // collapse task id - reset(handoff?): Promise; - fork(at: Id | "start", options?: { abort?: boolean; values? }): Promise; // at an entry id - spawn(options): Promise; // child conversation id - value(addr) / list(addr) - config(kind: TaskKind): { get(): Promise>; set(partial: Partial>): Promise }; + snapshot(call: Call): Promise; + accept(input, options: { requestId?: string } | undefined, call: Call): Promise<{ inputId: Id }>; + prompt(input, options: { requestId?: string } | undefined, call: Call): Promise; + result(inputId: Id, call: Call): Promise; + drive(call: Call): Promise<"idle" | "closed">; + steer(input, call: Call): Promise<{ inputId: Id }>; + followUp(input, call: Call): Promise<{ inputId: Id }>; + nextRun(input, call: Call): Promise<{ inputId: Id }>; + write(kind: EntryKind, input: EntryInput, call: Call): Promise<{ inputId: Id }>; + cancelQueued(inputId: Id, call: Call): Promise<"cancelled" | "not_found">; + abort(call: Call): Promise; + collapse(options: { instructions? } | undefined, call: Call): Promise; + reset(handoff: string | undefined, call: Call): Promise; + fork(at: Id | "start", options: { abort?: boolean; values? } | undefined, call: Call): Promise; + spawn(options, call: Call): Promise; + value(addr) / list(addr) // async reads/writes take a final Call + config(kind: TaskKind): { get(call: Call): Promise>; set(partial: Partial>, call: Call): Promise }; readonly hooks: { on(kind, point, handler, o?: { subtree?: boolean }): () => void }; // scoped to this conversation (§8.7) readonly settings: ConfigHandle; // sugar: config(generationKind) - commit(plan: (tx: ConversationTx) => T): Promise; + commit(plan: (tx: ConversationTx) => T | Promise, call: Call): Promise; } interface Harness { - root(): Promise; - conversation(id): Promise; - conversations(q?): Promise>; // independent ones: owner === undefined - acceptance(requestId: string): Promise<{ requestId: string; conversationId: Id; inputId: Id } | undefined>; - inspect(): Promise<{ start: Task[]; inflight: Task[] }>; - drive(options?): Promise<"idle" | "closed">; - getEntry(id); getEntry(kind, id); getEntries(ids); getEntries(kind, ids); entries(conversationId, page); - getTask(id); getTask(kind, id); getTasks(ids); getTasks(kind, ids); - abortTask(id): Promise; - value(addr) / list(addr) (session scope) - commit(plan: (tx: Tx) => T): Promise; - watch(conversationId, { tail: number; values?: Address[]; raw?: boolean }): Promise>; // §9.4 - watch(): Promise>; - deleteConversation(id); shutdown(); close(); + root(call: Call): Promise; + conversation(id: Id, call: Call): Promise; + conversations(query, call: Call): Promise>; + acceptance(requestId: string, call: Call): Promise<{ requestId: string; conversationId: Id; inputId: Id } | undefined>; + inspect(call: Call): Promise<{ start: Task[]; inflight: Task[] }>; + drive(call: Call): Promise<"idle" | "closed">; + getEntry(id, call); getEntry(kind, id, call); getEntries(ids, call); getEntries(kind, ids, call); + entries(conversationId, page, call); + getTask(id, call); getTask(kind, id, call); getTasks(ids, call); getTasks(kind, ids, call); + abortTask(id: Id, call: Call): Promise; + value(addr) / list(addr) // session state handles; async methods take Call + commit(plan: (tx: Tx) => T | Promise, call: Call): Promise; + watch(conversationId: Id, options: { tail: number; values?: Address[]; raw?: boolean }, call: Call): Promise>; + watch(call: Call): Promise>; + deleteConversation(id: Id, call: Call): Promise; + shutdown(call: Call): Promise; + close(call: Call): Promise; } ``` -`drive`'s signal cancels the caller's wait, not the work. `fork({ abort: true })` marks the source's -foreground set in the same commit that creates the fork, for "go back to that point"; which +`Call` is defined in §6.5. `call.abortSignal` cancels the drive caller's wait, not the work. +`fork(at, { abort: true }, call)` marks the source's foreground set in the same commit that creates the fork, for "go back to that point"; which conversation a UI treats as current is the UI's business. `prompt` returns the answer entry only when its explicit input result is `done`; failed, cancelled, stopped or merely placed input returns `undefined`. `result(inputId)` is one sticky-value point read, never a transcript scan. After an @@ -1520,7 +1650,7 @@ const entryId = await harness.commit(tx => { }); // final id, inside the closure tx.task(reminderKind, { background: true, state: { about: id } }); return id; // any value; resolved after commit -}); +}, call); ``` ```ts @@ -1539,8 +1669,8 @@ interface EntryDraft { interface Tx { // reads (committed state) getEntry(id) / getEntry(kind, id) / getEntries(...) / getTask(id) / getTask(kind, id) / getTasks(...) - value(addr: Value): { get(at?): T | undefined; set(v: T): void; delete(): void }; - list(addr: List): { append(v: T): Id; remove(id: Id): void; clear(): void; read(q): Page }; + value(addr: Value): { get(at?): Promise; set(v: T): void; delete(): void }; + list(addr: List): { append(v: T): Id; remove(id: Id): void; clear(): void; read(q): Promise>> }; // writes; rewindable conversation state before entries (§3.2) entry(kind: EntryKind, conversationId: Id, input: EntryInput): Id; task(kind: TaskKind, spec: { state: S; after?: Id[]; background?: true; owns?: Id[] }): Id; @@ -1617,27 +1747,34 @@ class Tx { batch(): CommitBatch { return { kind: "main", writes: this.writes }; } } -// the line: one commit at a time -async commit(plan: (tx: Tx) => T): Promise { - return this.line.run(async () => { +// The line owns admission, persistence, indexes and decisions; callbacks run afterward. +async commit(plan: (tx: Tx) => T | Promise, call: Call): Promise { + const completed = await this.line.run(async () => { + const invocation = call.value(invocationKey); + this.checkAdmission(invocation); // phase, exact running identity, method and mark (§6.3) const tx = new Tx(this.storage, this.kinds); - const result = await plan(tx); // throws → nothing written + const result = await plan(tx); const batch = tx.batch(); - if (batch.writes.length === 0) return result; - this.validate(batch); // ownership, dependencies, exchange rules, marks + this.validate(batch, invocation); // whole-batch authority and structural invariants + if (batch.writes.length === 0) return { result, actions: [] }; await this.storage.commit(batch); - this.publish(batch); // watchers - if (batch.writes.some(w => w.type === "task" || w.type === "patch" || w.type === "settle")) this.driver.kick(); - return result; + const actions = this.driver.applyBatch(batch); // apply ALL writes; then reserve affected work/waiters + this.publishState(batch, actions); // fold views, enqueue delivery; no listener execution + return { result, actions }; }); + this.dispatch(completed.actions); // start effects, signal controllers, deliver callbacks + return completed.result; } ``` `ConversationTx` is the same with `conversationId` bound. `ScratchTx` is the `value`/`list` half -with the batch tagged `scratch`. The closure runs on the line: reads see committed state, writes are buffered and validated -in order, ids are final when returned, and a throw discards everything. The outer promise resolves +with the batch tagged `scratch`. Builders may be async to await storage reads on the line; no other +line operation can interleave. Reads (including Tx/ScratchTx value/list/entry/task reads) return +Promises; writes are synchronous, buffered and validated in order. Never await an external effect, +a driver waiter or another line operation inside the builder. Ids are final when returned, and a +throw discards everything. The outer promise resolves with the closure's value after persistence and publication, which is when ids may escape. -`ConversationHandle.commit` binds the conversation; `TaskContext.commit` is the same for task +`ConversationHandle.commit` binds the conversation; `TaskRuntime.commit` is the same for task code. ### 9.3 Task and tool integration @@ -1647,8 +1784,8 @@ interface Tool { readonly name: string; readonly description: string; readonly parameters: JsonSchema; readonly output?: ShellOutputLimits; // retained window for text output readonly replay?: "safe" | "never"; - execute(toolCallId: string, params: TParams, signal: AbortSignal, - out: ToolOutput, ctx: ToolContext): Promise; + execute(toolCallId: string, params: TParams, + out: ToolOutput, runtime: ToolRuntime, call: Call): Promise; } // the sink; everything a tool produces goes through it (mobile-handoff/01-harness/04-tool-output) @@ -1666,14 +1803,15 @@ interface ToolOutput { diag(severity: "info" | "warn" | "error", message: string, code?: string): void; // commentary, see below } -interface ToolContext { +interface ToolRuntime { readonly taskId: Id; // the tool task; identity for origin/for links readonly budgetMs: number; // blocking budget (§8.3) readonly env: ExecutionEnv; // FileSystem & Shell - readonly conversation: ConversationHandle; // the tool's own conversation (for subagents, §8.5) - readonly harness: Harness; // conversation(id), abortTask, for spawned children - waitForTask(id: Id, o: { budgetMs?: number; signal: AbortSignal }): Promise; // true = terminal - jobOutput(id: Id): Promise; // preview or terminal state + commit(plan: (tx: ConversationTx) => T | Promise, call: Call): Promise; + conversation(id: Id, call: Call): Promise; + abortTask(id: Id, call: Call): Promise; + waitForTask(id: Id, options: { budgetMs?: number } | undefined, call: Call): Promise; + jobOutput(id: Id, call: Call): Promise; } // what the live sink folds to; recorded on the tool task and used to build the result entry @@ -1696,8 +1834,10 @@ type ToolResultData = Omit; rejects, and a tool may throw anything. `terminate` is orthogonal to how execution ended, so a failing tool can still stop the turn (`out.terminate(true); throw error`). Usage and `addTools` go through the sink rather than a return value so that a replayed tool can seed from durable state. -The sink's text ops (`write`, `replace`, and the `slide` the env emits for a moving tail) are what -scratch and the watch stream carry, so a UI shows output live without the tool doing anything. +The sink owns its scratch promises: it handles expected post-mark/close rejections, reports real +persistence errors and drains pending callbacks before the invocation returns. Raw plugin scratch +writes must be awaited/caught. The sink's text ops (`write`, `replace`, and the `slide` the env emits +for a moving tail) are what scratch and the watch stream carry, so a UI shows output live without the tool doing anything. **Diagnostics are a channel, not text.** Commentary about a call (it was truncated, the output spilled to a file, the path was resolved differently, the file changed on disk since it was read, @@ -1731,7 +1871,7 @@ commits: interface WatchHandle { readonly view: View; // captured on the line at watch(); folded before each event is delivered start(listener: (event: Event) => void): void; // delivers everything since the capture, then live - resnapshot(): Promise; // fresh capture, same subscription (when lagging) + resnapshot(call: Call): Promise; // fresh capture, same subscription (when lagging) unsubscribe(): void; } @@ -1763,7 +1903,7 @@ type ConversationEvent = | { type: "context"; ids: readonly Id[] } // a head or edit changed derived context | { type: "fault"; error: unknown } | { type: "closed" }; -const w = await h.watch(c.id, { tail: 100, values: [myPlugin.config.mode] }); +const w = await h.watch(c.id, { tail: 100, values: [myPlugin.config.mode] }, call); render(w.view); w.start(event => render(w.view, event)); // w.view is already folded when the listener runs ``` @@ -1791,7 +1931,8 @@ harness needed a name per case for, and cover plugin kinds without anyone adding Events are delivered per commit, in commit order, all events of one commit before any of the next, so a terminal task and its successor never show as an idle gap. `watch` captures the view and registers the subscription in one serialized step on the line, so nothing falls between them. -Task output travels as delta ops only (§7.4); the folded preview lives in the view, never in the event. `resnapshot` marks a +Task-output watch events travel as Chord delta ops only; durable scratch uses ordinary records (§7.4). +For watch delivery, the folded preview lives in the view, never in the event. `resnapshot` marks a barrier on the delivery tail while the line holds the new capture, so calling it from inside the listener neither deadlocks nor refolds stale state. Buffering is bounded; a lagging client gets a `fault` and calls `resnapshot`. A listener never awaits a commit on the same line. @@ -1806,7 +1947,7 @@ re-diffed for the wire: ```ts // worker // ui process -const w = await h.watch(c.id, { tail: 100 }); on("view", m => { view = m.view; ui.apply(view); }); +const w = await h.watch(c.id, { tail: 100 }, call); on("view", m => { view = m.view; ui.apply(view); }); send({ type: "view", view: w.view }); w.start(e => send({ type: "event", event: e })); on("event", m => { applyEvent(view, m.event); ui.apply(view, m.event); }); ``` @@ -1819,7 +1960,7 @@ clients. That applier is already the mobile port; the reducer adds its ten cases harness package's implementation is the reference, tested by porting `delta.test.ts` first. `task_output.ops` are ops on the task's *preview*, which is a Chord delta tracker the kind mutates -in place (`ctx.preview.state`): the generation applies each stream event to a tracked partial +in place (`runtime.preview.state`): the generation applies each stream event to a tracked partial message, the tool's preview is the sink's tracked state, a job's is the same. The harness flushes the tracker after each scratch commit; a streamed token is one `["a", path, delta]`, computed from dirty paths and a memcmp, never a recomputed object (a reassigned preview would flush as a full @@ -1845,31 +1986,31 @@ type SessionEvent = | { type: "conversation"; conversation: Conversation; change: "created" | "deleted" } | { type: "value"; addr: Address; value: JsonValue | undefined } | { type: "usage"; row: UsageRow; totals: Usage } - | { type: "report"; task?: Id; kind?: string; error: unknown } // hook errors, poisoned tasks + | { type: "report"; task?: Id; kind?: string; method?: string; error: unknown } // hook/task errors | { type: "fault"; error: unknown } | { type: "closed" }; -harness.watch(): Promise>; +harness.watch(call: Call): Promise>; ``` ### 9.5 End to end ```ts -const h = await Harness.open(storage, options); -const c = await h.root(); -const w = await h.watch(c.id, { tail: 100 }); render(w.view); w.start(e => render(w.view, e)); +const h = await Harness.open(storage, options, call); +const c = await h.root(call); +const w = await h.watch(c.id, { tail: 100 }, call); render(w.view); w.start(e => render(w.view, e)); -const answer = await c.prompt("Inspect the parser"); +const answer = await c.prompt("Inspect the parser", undefined, call); -const driving = c.drive(); await c.abort(); await driving; // durable intent, then cleanup +const driving = c.drive(call); await c.abort(call); await driving; // durable intent, then cleanup const child = await c.spawn({ prompt: "Inspect only tests", context: "fresh", - values: { inherit: [model] } }); -await h.drive(); // drives the child too -console.log(await h.conversation(child)); + values: { inherit: [model] } }, call); +await h.drive(call); // drives the child too +console.log(await h.conversation(child, call)); -const alt = await c.fork({ atEntry: answer.id }); -await alt.prompt("Try a different implementation"); // source remains untouched +const alt = await c.fork(answer.id, undefined, call); +await alt.prompt("Try a different implementation", undefined, call); // source remains untouched -w.unsubscribe(); await h.close(); +w.unsubscribe(); await h.close(call); ``` ```ts @@ -1879,7 +2020,7 @@ const h = await Harness.open(storage, { kinds: { entry: pluginEntryKinds, task: pluginTaskKinds }, // added to the built-ins replace: { generation: myGenerationKind }, // swap a built-in by name; same statuses, hook names rootValues, -}); +}, call); h.kinds.generation; h.kinds.tool; h.kinds.postTools; h.kinds.collapse; h.kinds.job; // whatever is registered under the name ``` @@ -1916,7 +2057,20 @@ source cutoffs, sticky state untouched by historical reads. | Torn final JSONL line | only that line discarded | | Malformed complete line | open fails; no silent truncation | | Retirement committed, unlink fails | old scratch invisible | -| Effect returns as abort starts | one owner; join then abort | +| Effect returns as abort starts | one invocation per task; fresh abort only after return | +| Post-mark main/scratch write | reject before builder; sinks handle cancellation and drain callbacks | +| Status cycle returns to initial status | epoch counts intermediate committed changes; no false fault | +| Host patches an owned task | reject lifecycle/state mutation; abort marks remain allowed | +| 100,000 historical children, two live tasks | query only live seed/needed ancestry, no history traversal | +| Repeated drive and cancelled waiter | one attachment; remove waiter/listener exactly once | +| Known dependency self-drive | reject before acceptance or cycle-creating admission | +| Close vs task commit | earlier commit completes; later mutation rejects; completion still admitted | +| Shutdown vs accept / idle nextRun inbox | accept before mark batch or reject; queued items/results preserved | +| Shutdown crash before child cleanup | reopened abort marks tasks only; child queues remain unchanged | +| Main 100, live scratch through 150, crash | reopen recovers lastSeq=150; next write=151 | +| Retired scratch still on disk | ignore it even if malformed; later main settlement covers its sequences | +| Parent cleanup vs already-terminal job | already finished is successful cleanup | +| Broken abort handler | reject drives/shutdown, signal/join and close; replace kind before recovery | | Parallel tools settle in either order | post_tools starts once | | Abort vs post_tools settlement | no unmarked successor | | Child finishes while its owning tool is marked | tool settles aborted, not done | diff --git a/packages/agent/docs/pico/pico-work.md b/packages/agent/docs/pico/pico-work.md index 1e4347537..a8703bbed 100644 --- a/packages/agent/docs/pico/pico-work.md +++ b/packages/agent/docs/pico/pico-work.md @@ -14,8 +14,10 @@ shows the surface each package has to end up with. `Id`, `EntryIdentity`, `EntryBase`, the composable `EntryData` / `ModelProjection` / `ContextHead` / `ContextEdits` facets, `Entry`, `EntryKind`, `EntryInput`, `ContextEdit`, `Task`, `TaskRole`, stored task roles, `Conversation`, `InboxItem`, `InputResult`, `InboxOp`, acceptance -receipts, `Address`/`Value`/`List`/`Scope`, `Write`/`CommitBatch`, `Page`/`Cursor` and the query shapes -(§2, §4.1, §5.1, §7.2–7.3). No code. +receipts, `Call` (an alias of Chord Context), `TaskRuntime`/`ToolRuntime`, private typed invocation +identity, `Address`/`Value`/`List`/`Scope`, `Write`/`CommitBatch`, `Page`/`Cursor` and query shapes. +Every async harness/handle/runtime method takes a required final Call; no duplicate signal options, +task-conversation facades or admission gates. Test: it compiles; fixtures cover entries with no data, no model, every individual facet and the built-in facet combinations. @@ -25,17 +27,20 @@ built-in facet combinations. `Storage` and `MemoryStorage`: `commit` numbering from `lastSeq`, point reads, fork-aware `scanEntries` / `scanTasks` / `scanConversations` with cursors, target-capped `newestHead` over stored numeric boundaries, stored entry data/model/edits, value versions and list elements read at a -position, `ownedFrom`, the set of entry kind strings written. +position, `ownedFrom`, the set of entry kind strings written and session-wide live-task scans. +Use existing batched `getTasks` for named owners; no header/projection APIs or inbox-specific queries. Tests: every row of the §7.2 query table against hand-built batches; ids are `lastSeq + 1 + i`; a -head is found without a kind; entry headers omit arbitrary data but retain model/head/edits; a -live-task scan decodes no terminal rows; `remove` and `clear` hide by position. +head is found without a kind; entry reads/scans/head lookup return full entries after indexed +selection; live-task scans skip terminal rows (named owner reads may fetch them); `remove` and +`clear` hide by position. ## 3. The line and `Tx` -`commit(plan)` on a serialized line: buffered writes, ids final at call time, rewindable +`commit(plan, call)` on a serialized line: buffered writes, ids final at call time, rewindable conversation value/list writes after an entry throw, a throwing plan discards everything, publish -after persist, `kick` when a batch touched a task. Session and sticky conversation state, task +after persist; apply the entire batch to live indexes before scheduling or testing idle. Driver +callbacks, signals and task methods dispatch outside the line. Session and sticky conversation state, task writes and conversation writes may appear anywhere. `task` / `patch` / `settle` materialize the role from the kind's status map. `value` / `list` / `entry` / `task` / `patch` / `settle` build the batch of §7.3. @@ -43,7 +48,11 @@ of §7.3. Tests: concurrent commits serialize; a rejected commit consumes no ids; each builder verb produces the expected write; reads inside a plan see committed state only; session and sticky conversation writes may follow and reference a new entry; rewindable value set/delete and list -append/remove/clear after an entry each reject. +append/remove/clear after an entry each reject; exact invocation-token checks run on the line; +post-mark main/scratch mutation rejects before builder; only an owned task's current invocation may +patch its state/status or settle it (host abort marks remain allowed); caller cancellation never +abandons admitted persistence; Tx/ScratchTx reads are asynchronous and builders may await them +without releasing the line; writes remain synchronous. No external effects or nested line entry in builders. ## 4. Entry kinds and context @@ -71,38 +80,56 @@ results but inherit no tasks; state committed after the entry (a model change) i ## 6. Task kinds and the driver -`TaskKind`, the registry, kind-declared status cycles, `TaskContext` (`commit`, `scratch`, `sleep`, -`signal`, `config`, `hooks`), the driver of §6.1 (`owned`, `Wake`, attached scopes, `drive` as a -waiter, unchanged-status and poison guards, abort join and re-read), scopes of §6.2. Test kinds only: -a counter, a blocker, a cycling schedule, a kind that returns without changing status, a kind that -ignores its signal. There is no parked role or successor chain for retries. +`TaskKind` with `(task, runtime, call)` methods, registry and `TaskRuntime`; one Call-final convention +for operations, existing Context-aware env/provider/hook boundaries. Driver decisions, attachments, +waiters, invocation completion and lifecycle run on the commit line. Only effects/callbacks dispatch +outside it. Seed live tasks once; committed batches update live, reverse-dependency and conversation +indexes. Resolve ownership upward only for live/owned work; evict unused ancestry. Stable attachments +are distinct from temporary waiters. No Wake, polling loop, worker pool, poison graph, gate or parked role. -Tests: persisted roles select execute / recover without a role-map lookup; a blocked execute holds -nothing up; a valid status cycle; the exact unchanged-status guard; `after` gates a start; -conversation and harness drive both start foreground/background work, resolve on foreground idle -and keep serving background work; a full-quiescence wait remains pending on a recurring schedule; -abort marks in both race orders; reopen recovers inflight; a mark written by one process is applied -by the next. +One invocation per task ID, concurrent across IDs; reserve before dispatch and release only after +actual return, even if already terminal. Durable abort revokes execute/recover main and scratch writes, +then signals outside line; fresh abort invocation alone writes cancellation outcomes. Status epochs +count actual committed transitions, including same-start/end cycles. Task-contract faults fail-stop: +reject waiters, stop admission, signal/join and close; preserve durable recovery state. + +Tests: one open scan, no full scan per completion; 100,000 historical children with two live tasks; +whole-batch settle/successor publication; exact epoch rules; owned-task exclusive mutation; one +invocation per ID; blocked calls do not block others; abort bypasses dependencies and never overlaps +execute; repeated mark spares running abort; abort-handler failure; cooperative signal window; +known direct/dependency self-waits, including background caller and new cycle-creating admission; +already-aborted waiter registration and every cancellation/idle race; repeated drive uses one +attachment; foreground idle while attached background work continues; reopen recovers marked work. ## 7. Scratch Scratch batches, one task per batch, `ScratchTx`, retire on settle, sidecar-independent (memory). -Tests: a crash before settle keeps scratch; settle deletes it; a write after settle is rejected; a -new attempt clears its list. +Tests: a crash before settle keeps scratch; settle deletes it; writes after mark/settle/retired token +reject; new attempts clear scratch; persisted assistant frames use the pi-ai compact frame encoder, +not raw cumulative provider events; a rejected stream-frame write cancels and joins the producer +before invocation completion (iterator exit is insufficient); harness progress bridges own every promise, suppress expected +late cancellation only, report persistence errors, and drain before invocation completion. Raw task +scratch calls must be awaited/caught; no successful silent no-op. ## 8. Harness shell and handles `Harness.open` (built-in registries, `kinds` and `replace` options, kinds-set check, `inspect`, `drive`, `close`, `shutdown`), `ConversationHandle` (`commit`, `value` / `list`, `config` / `settings`, `fork` with `abort`, `abort`, `hooks.on` scoped with `subtree`), `acceptance(requestId)`, -`result(inputId)`, `abortTask`, `conversations` with `parent` / independent filtering. No agent -behaviour yet. +`result(inputId, call)`, `abortTask`, `conversations` with `parent` / independent filtering. Call is +required on every async public/runtime operation, including reads and lifecycle; no raw host-lifecycle +Harness exposed to tasks. No agent behaviour yet. Tests: open on empty vs existing storage; an unregistered entry kind is reported without scanning entries and its stored model/head/edits still derive context; an unknown live task kind rejects; replace by name keeps `h.kinds.` consistent; `settings` round-trips; scoped hooks run after -harness-wide ones, innermost last. +harness-wide ones, innermost last; derived Call preserves typed private identity and telemetry; +stale/foreign task token rejects; close stops admission in one nonpersistent line job, joins outside; +shutdown atomically marks live tasks only, then permits abort cleanup; queued items/results remain +unchanged, including idle inbox-only conversations; crash/reopen child cleanup marks tasks only and +preserves queues; repeated close/shutdown share completion; close interrupt rejects shutdown; +lifecycle calls with task identity reject; delete rejects outstanding terminal invocations. ## 9. Generation kind @@ -117,8 +144,11 @@ attempt. Tests: the six system-entry traces from the guide (tool added, removed, host section changed, MCP schema changed, plugin section on and off, `addTools`); fresh baseline after a head; a fork diffs -against its own config; a restart emits only the changed section; in-band `aborted` stop reason; -crash while streaming publishes the partial; retry budget exhausted → failed with no successor. +against its own config; a restart emits only the changed section; in-band provider abort without a durable mark has a +kind-level outcome, but post-mark execute settlement rejects and fresh abort writes optional partial, +cancelled input results and known usage atomically; no mark branch in normal settlement; missing +post-cutoff usage is unknown; crash while streaming publishes the partial; retry budget exhausted +→ failed with no successor. ## 10. Tools, post_tools, exchanges @@ -165,8 +195,9 @@ The `subagent` tool (`run`, `spawn`, `send`, `status`, `wait`, `stop`), ownershi reach through live owners, `run`'s recover driving the child again, `spawn` initialization of config, explicit child input results. -Tests: restart in the middle of `run`; parent abort reaches a `run` child and spares a `spawn` -child; `stop`; nested children; the child's first turn writes its own baseline. +Tests: restart in the middle of `run`; conversation abort reaches a `run` child and spares a `spawn` +child; abortTask marks only owner and fresh cleanup cancels its child; drive waiter observes Call; +cleanup tolerates already-terminal child/job; `stop`; nested children; child's own baseline. ## 14. Jobs and the budget @@ -174,14 +205,20 @@ child; `stop`; nested children; the child's first turn writes its own baseline. delegating first and waiting with the budget, `notify` and the `notice` entry, the `job` tool, schedules. -Tests: budget expiry settles the call with `delegated` and the job continues; the notice appears -on completion; a schedule cycles one stable task id and `abortTask` ends it; it does not block -ordinary harness drive after foreground idle; recover → `lost` or rerun by policy; a user abort -during the wait kills a non-backgrounded job. +Tests: job creation atomically stores job id and cancellation policy on the tool; fresh tool abort +uses those durable references, not execute locals or catch writes; budget expiry settles delegated +and job continues; delegation requests notification in separate sticky state, not a patch to an owned +job; both notification/completion orders publish once for exited, lost and killed; schedule cycles one +stable id; foreground drive +ignores background recurrence; recover → lost or safe rerun; abort kills non-detached job. + +Remaining integration design: arbitrary-promise budget adoption must explicitly transfer effect and +sink ownership before the tool invocation releases. A raced, abandoned promise is not permitted. +Preserve the capability (crash outcome lost), but implement job-first execution until transfer is settled. ## 15. Previews -`ctx.preview` as a Chord tracker per task; the generation applies stream events to a partial +`runtime.preview` as a Chord tracker per task; the generation applies stream events to a partial message, the tool's preview is its sink state, the job's the same; `preview.init` on attach and reopen; flush after each scratch commit. @@ -204,12 +241,17 @@ fold, failed and aborted attempts included. ## 17. JSONL storage Append-only batches, replay on open, torn tail discarded, malformed line fails open, entry -`data`/`model`/`head`/`edits` and the kind-string set from replay, values as Chord deltas (full when -first or small, ops otherwise), scratch sidecars retired after the main-file settle. +`data`/`model`/`head`/`edits` and the kind-string set from replay. Whole value sets and intrinsic list +operations in main and scratch files; no Chord storage codec. Replay main first, then surviving live +scratch; per-file sequence gaps are valid. Recover lastSeq from maximum complete surviving batch +endpoint, including clear/remove. Retired scratch is ignored; its later settlement covers its ids. +Physically remove torn suffixes before appending; malformed complete replayed batches fail. -Tests: conformance against memory (one mutation stream, identical query results); linear file -growth under repeated sets of a large value; accepted payload plus placement has two JSONL copies, -including idle acceptance; crash between the settle write and the sidecar unlink. +Tests: conformance against memory; full replacement values on disk; compact incremental frames/output +operations avoid repeated growing snapshots; main=100/live scratch=150 reopens at 150; settle=151 +with failed unlink ignores retired scratch, even malformed; clear/remove high-water; torn main and +live scratch tails; nonoverlapping ranges; accepted payload plus placement has two JSONL copies, +including idle acceptance. Backend memory and disk growth claims match whole-value behavior. ## 18. SQLite storage @@ -217,15 +259,18 @@ Tables and indexes of §7.5, nullable entry JSON columns for data/model/edits an head boundary, scratch rows deleted in the settle transaction, storage version and `migrate`. Tests: conformance three ways; removed sticky inbox elements may be physically discarded while -input results remain point-readable; a cold reopen decodes only live rows (count them); a version -mismatch rejects. +input results remain point-readable; a cold reopen decodes live tasks and only named required owner +records, including terminal owners; no retained terminal payload cache; a version mismatch rejects. ## 19. Race matrix and telemetry The §10.2 matrix in both orders with faux clocks, fake processes and storage barriers; spans per task call and per commit. -Tests: the matrix; span tree for one turn with a tool and a retry. +Tests: the matrix; span tree for one turn with a tool and retry; nested Call preserves active telemetry +parent and private invocation identity; drive-caller cancellation does not become the task signal; +providers/env interpret signals, hooks propagate cancellation rather than swallowing it; no callback +runs on the line; stale RPC-bound invocation rejected. Metadata transport never carries task authority. ## 20. Clients @@ -254,7 +299,8 @@ Copy (under `packages/agent/src/harness/` unless noted): Depend on, as packages (they are not the old harness): -- `@earendil-works/chord/delta` (15, 16, 17) +- `@earendil-works/chord` Context types and `@earendil-works/chord/context` helpers (1, 6, 8, 19) +- `@earendil-works/chord/delta` (15, 16: preview/watch only; not storage) - `@earendil-works/pi-ai`: `faux` provider for tests, `utils/estimate` for thresholds, `SystemMessage` (4, 9, 12) Read before writing the equivalent, then close the file: