diff --git a/packages/agent/docs/pico/pico-usage-guide.md b/packages/agent/docs/pico/pico-usage-guide.md index f2f0dc2e8..d7c2d2c78 100644 --- a/packages/agent/docs/pico/pico-usage-guide.md +++ b/packages/agent/docs/pico/pico-usage-guide.md @@ -233,9 +233,11 @@ 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(call); +const { start, inflight, orphaned, parked } = 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 +// orphaned: foreground tasks whose kind is missing (a plugin was uninstalled); settled at open +// parked: background tasks whose kind is missing; they resume when it is registered again ``` ### What Happens on Reopen @@ -837,10 +839,10 @@ 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 } }), call); + state: { status: 'planned', 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) } }), call); + state: { status: 'planned', cmd: 'npm test', cwd, every: 24 * 3600_000, notBefore: tonightAt(2) } }), call); await h.abortTask(nightly, call); // ends the schedule wherever it is ``` @@ -891,7 +893,7 @@ const entry = await c.commit(tx => { const id = tx.entry(myPlugin.noteKind, { data: { text: 'plan accepted' } }); tx.value(expanded).set(true); // sticky state may follow tx.task(myPlugin.reminderKind, { background: true, // tasks anywhere - state: { about: id, at: Date.now() + 3600_000 } }); + state: { status: 'scheduled', about: id, at: Date.now() + 3600_000 } }); return id; }, call); ``` @@ -983,11 +985,9 @@ goes on: 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, + state: { status: 'planned', 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, - } }); + tx.patch(task, { jobId: id, cancelJobOnAbort: !params.background }); // partial: still 'running' return id; }, call); @@ -1157,7 +1157,11 @@ only the current status, state, role and scratch. Here is a reminder that fires once: ```typescript -interface ReminderState { about: Id; at: number; fired?: boolean } +type ReminderStates = + | { status: 'scheduled'; about: Id; at: number } + | { status: 'firing'; about: Id; at: number } + | { status: 'done'; about: Id; at: number; fired: boolean } + | { status: 'aborted'; about: Id; at: number }; export const reminderKind = defineTaskKind({ kind: 'myplugin.reminder', @@ -1167,23 +1171,33 @@ export const reminderKind = defineTaskKind({ hooks: { before_fire: { failClosed: false } }, 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 + if (task.state.at > runtime.now()) await runtime.sleep(task.state.at, call); // throws on cancellation + await runtime.commit(tx => tx.patch(task, { status: 'firing', ...common(task) }), call); // 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 }); + if (!skip) tx.entry(noticeKind, { model: [noticeMessage(`Reminder: see entry ${task.state.about}`)] }); + tx.settle(task, { status: 'done', ...common(task), fired: !skip }); // a marked task's commit rejects }, call); }, - 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); }, + 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, { status: 'aborted', ...common(task) }), call); + }, // no preview: nothing to show while sleeping }); + +const common = (t: Task) => ({ about: t.state.about, at: t.state.at }); ``` +State is a tagged union over status: one variant per status carrying exactly the fields that exist +in it, so a reader narrows instead of checking optionals, and `patch` within a variant takes a +partial (`tx.patch(task, { at: later })`) while a transition takes the whole variant. The harness +adds an `orphaned` terminal variant with the fields common to all of yours, for the case where this +plugin is not installed when the session is opened; you never write it, and a task that depends on +yours handles it in the same `switch` that handles your other terminal statuses. + The rules an execution follows, and the driver enforces: 1. Commit an inflight status before any external effect. A crash before that re-runs `execute`; a diff --git a/packages/agent/docs/pico/pico-v3.md b/packages/agent/docs/pico/pico-v3.md index 178f97bdc..ce87650d6 100644 --- a/packages/agent/docs/pico/pico-v3.md +++ b/packages/agent/docs/pico/pico-v3.md @@ -279,7 +279,7 @@ One session-wide sequence orders every mutation. A creation's sequence number is const entryId = await conv.commit(tx => { tx.value(planMode).set(true); // 100 const id = tx.entry(userKind, { model: [message] }); // 101 - tx.task(generationKind, { state: { input: id } }); // 102 + tx.task(generationKind, { state: { status: "pending", input: id } }); // 102 return id; }, call); // one batch [100–102] persisted and published; entryId === 101 here @@ -305,7 +305,7 @@ history through X: 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 + tx.settle(task, { status: "done", call: task.state.call, assistant: task.state.assistant, output, result }); // 302 }, call); await conv.value(model).set("claude-opus-5", call); // 303, a later commit @@ -465,7 +465,7 @@ 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 runtime.commit(tx => { const id = tx.entry(assistantKind, { model: [assemble(partial)] }); - tx.settle(task, { result: id }); + tx.settle(task, { status: "done", call: task.state.call, assistant: task.state.assistant, output, result: id }); }, call); ``` @@ -497,13 +497,15 @@ supplies the inputs, progress and outcome that must survive a process. ```ts type TaskRole = "start" | "inflight" | "terminal"; -interface Task { +interface TaskStateBase { readonly status: string } + +interface Task { readonly id: Id; readonly conversationId: Id; readonly kind: string; - readonly status: string; // kind-specific + readonly status: string; // = state.status, stored and indexed so storage never decodes state readonly role: TaskRole; // materialized from the kind's roles map on every write - readonly state: State; // JSON + readonly state: State; // a tagged union; the status is its discriminant readonly after: readonly Id[]; // dependencies, fixed at creation readonly background?: true; // fixed at creation; absent = foreground (§5.6) readonly owns?: readonly Id[]; // conversations this task created (§5.7) @@ -511,6 +513,23 @@ interface Task { } ``` +**State is a tagged union over status.** A kind declares one variant per status, carrying exactly +the fields that exist in it, so nothing is an optional that a reader has to guess about and a +dependent narrows instead of probing: + +```ts +type ToolStates = + | { status: "planned"; call: ToolCall; assistant: Id } + | { status: "running"; call: ToolCall; assistant: Id; args: JsonObject; replay: "safe" | "never"; + jobId?: Id; cancelJobOnAbort?: boolean } + | { status: "done"; call: ToolCall; assistant: Id; output: ToolOutputState; result: Id } + | { status: "aborted"; call: ToolCall; assistant: Id; output: ToolOutputState; result: Id }; +``` + +`Task.status` duplicates `state.status` because storage indexes it and the driver compares it +without decoding state; `Tx` writes both from the one variant it was given, so they cannot diverge. +Fields common to every variant must have the same type in each. + A kind maps its statuses to three roles: ```text @@ -524,21 +543,40 @@ on every write. Storage selects live rows using the role index without inspectin running kind code. Loading their full records may decode state; terminal history is not decoded. Callers cannot edit the role independently. +**`orphaned` is derived, not declared.** A kind never writes it and never lists it. The harness adds +one terminal variant to every state union, carrying the fields common to all of the kind's own +variants: + +```ts +type Common = UnionToIntersection; // fields present in every variant +type Orphaned = Omit, "status"> & { status: "orphaned" }; +type ToolState = ToolStates | Orphaned; // { status: "orphaned"; call; assistant } +``` + +It is written in one situation only: at open, for a live foreground task whose kind is not +registered in this process (§6.4). The stored record keeps whatever fields it had; the type +promises only the common ones, which is all a dependent may rely on, because the kind never ran its +cleanup. Background tasks of unregistered kinds are parked instead, and resume through `recover` +when their kind returns. This is the only case in which the harness settles a task it does not own. + ### 5.2 Kinds ```ts -interface TaskKind { +interface TaskKind { readonly kind: string; - readonly initialStatus: string; - readonly roles: Readonly>; + readonly initialStatus: States["status"]; + readonly roles: Readonly>; // `orphaned` is added as terminal (§5.1) readonly hooks?: HookSpecs; // the points this kind runs (§8.7) readonly config?: Config; // the values this kind reads (below) preview?: { // what UIs see while the task runs (§9.4) init(scratch: ScratchReader): Preview; // built once, on attach or reopen, from durable scratch }; // 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; + execute(task: Task, runtime: TaskRuntime, call: Call): Promise; + recover(task: Task, runtime: TaskRuntime, call: Call): Promise; + abort(task: Task, runtime: TaskRuntime, call: Call): Promise; +} + +type TaskState = K extends TaskKind ? S | Orphaned : never; // what readers see } ``` @@ -657,31 +695,39 @@ The generation's settlement publishes the exchange atomically: 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 } })); + tx.task(toolKind, { state: { status: "planned", call, assistant } })); tx.task(postToolsKind, { after: tools, - state: { assistant, tools, inputs: task.state.inputs }, + state: { status: "waiting", assistant, tools, inputs: task.state.inputs }, }); - tx.settle(task, { assistant }); + tx.settle(task, { status: "done", inputs: task.state.inputs, assistant }); }, call); ``` -Each tool executes and settles itself: hooks, execution, result entry, terminal status. It does not +Each tool executes and settles itself: hooks, execution, result entry, terminal variant. It does not look at siblings, queues or context. post_tools starts when both are terminal, reads them with the -typed getter, and decides what happens next: +typed getter, narrows on their status, and decides what happens next: ```ts 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 + const outcomes: ToolOutputState[] = []; + for (const t of tools.values()) { + switch (t.state.status) { + case "done": case "aborted": outcomes.push(t.state.output); break; // both wrote their result entry + case "orphaned": // kind unregistered at open (§5.1) + tx.entry(toolResultKind, unavailableResult(t.state.call)); break; + default: throw new Error(`tool ${t.id} still live in post_tools`); + } + } if (outcomes.some(o => o.terminate)) { 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 + return tx.settle(task, { status: "stopped", assistant: task.state.assistant, tools: task.state.tools }); // no successor } const added = outcomes.flatMap(o => o.addedTools ?? []); const selectedTools = added.length ? [...current, ...added] : current; @@ -691,8 +737,8 @@ async execute(task, runtime, call) { data: { text: handoff.handoff }, model: [handoffMessage(handoff.handoff)], head: "self", }); 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, {}); + tx.task(generationKind, { state: { status: "pending", ...nextGeneration(task, { selectedTools, inputs }) } }); + tx.settle(task, { status: "done", assistant: task.state.assistant, tools: task.state.tools, inputs }); }, call); } ``` @@ -916,8 +962,11 @@ forever. An isolate kill boundary is future work; RPC cancellation alone is not ### 6.4 Open, close, shutdown, delete -**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. +**Open** reports unregistered entry kinds, seeds live/ownership indexes, exposes inspection/queries, +and starts nothing. A live task whose kind is not registered cannot run: a background one is parked +(not started, not recovered, not counted for idleness) and resumes through `recover` when its kind +returns; a foreground one would keep its conversation busy forever, so open settles it in the +derived `orphaned` variant (§5.1) and reports it. `inspect` returns those separately. Unknown entries preserve context through stored facets. **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 @@ -1369,8 +1418,8 @@ work continues as the job: ```ts 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 } }); + state: { status: "planned", cmd, cwd, origin: { tool: "bash", task: runtime.taskId, callId: toolCallId } } }); + tx.patch(task, { jobId: id, cancelJobOnAbort: true }); // partial: still "running" (§9.2) return id; }, call); const done = await runtime.waitForTask(job, { budgetMs: runtime.budgetMs }, call); @@ -1479,10 +1528,12 @@ 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 } }), call); + tx.task(jobKind, { background: true, state: { status: "planned", cmd: "npm test", cwd, every: 6 * 3600_000 } }), call); ``` ```text +states: planned { cmd, cwd, every?, notBefore? } | running { …, startedAt } | + exited { …, output, exitCode } | killed { …, output } | lost { … } 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 @@ -1615,7 +1666,7 @@ interface Harness { 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[] }>; + inspect(call: Call): Promise<{ start: Task[]; inflight: Task[]; orphaned: Task[]; parked: 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); @@ -1648,7 +1699,7 @@ const entryId = await harness.commit(tx => { data: { text: "plan accepted" }, model: [{ role: "user", content: "plan accepted", timestamp: 0 }], }); // final id, inside the closure - tx.task(reminderKind, { background: true, state: { about: id } }); + tx.task(reminderKind, { background: true, state: { status: "scheduled", about: id } }); return id; // any value; resolved after commit }, call); ``` @@ -1674,8 +1725,20 @@ interface Tx { // 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; - patch(task: Task | Id, changes: { status?: string; state?: Partial }): void; - settle(task: Task | Id, outcome: Partial & { status?: string }): void; // terminal; retires scratch + // stay in the current variant: a partial of that variant, no status + patch( + task: Task & { state: { status: K } }, + changes: Partial, "status">>): void; + // move to another variant, or patch by id: the whole variant + patch(task: Task | Id, state: NonTerminal): void; + settle(task: Task | Id, state: Terminal): void; // terminal; retires scratch + // NonTerminal / Terminal are S filtered by the kind's roles map. The partial overload needs a task + // whose variant the compiler can see (the normal case inside a kind's methods); a bare id or an unnarrowed + // union falls back to the whole-variant form, so no field of another variant is smuggled in and a partial + // cannot change status: + // tx.patch(task, { jobId }); // still "running" + // tx.patch(task, { status: "running", call, assistant, args, replay }); // planned → running + // tx.settle(task, { status: "done", call, assistant, output, result }); // terminal createConversation(spec): Id; deleteConversation(id: Id): void; } ```