mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
docs(agent): refine Pico5 plugin state API
This commit is contained in:
Mario Zechner
parent
fde6d778f8
commit
2b04ce27fb
4 changed files with 422 additions and 190 deletions
|
|
@ -30,7 +30,10 @@ terminal and never participate in conversation forks.
|
|||
Examples build on one another. Pico5 names (`defineDoc`, `defineDocFamily`,
|
||||
`Session`, `DocumentSource`, `Id`, `TaskRuntime`, `DocumentObserver`)
|
||||
refer to normative contracts, without a specified import path or runnable Pico5
|
||||
package. These Chord imports exist today:
|
||||
package. In those contracts, `ConversationRecord`, `EntryRecord`, and
|
||||
`TaskRecord` are persisted records, while `Conversation` is the public
|
||||
conversation object and `Entry`/`Task` are typed definitions. These Chord imports
|
||||
exist today:
|
||||
|
||||
```ts
|
||||
import {
|
||||
|
|
@ -39,6 +42,7 @@ import {
|
|||
type ReplicatedState,
|
||||
} from "@earendil-works/chord";
|
||||
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
|
||||
import { applyImmutable } from "@earendil-works/chord/delta";
|
||||
|
||||
// The spec calls this JsonObject; current Chord exports JsonValue, not JsonObject.
|
||||
type JsonObject = { [key: string]: JsonValue };
|
||||
|
|
@ -49,8 +53,8 @@ declare function documentReplicatedState<T extends JsonObject>(
|
|||
): Promise<{ readonly state: ReplicatedState<T | null>; dispose(): void }>;
|
||||
```
|
||||
|
||||
The adapter must register Chord-recognized state, atomically hydrate matching
|
||||
committed value/revision and subscribe, and forward committed ops without another
|
||||
The adapter must register Chord-recognized state, atomically hydrate a matching
|
||||
committed value and adapter-owned delivery sequence, and forward committed ops without another
|
||||
tracker or re-diff. Disposal releases observation, not the document. Even Chord's
|
||||
hydration requests must never flush an uncommitted draft.
|
||||
|
||||
|
|
@ -282,14 +286,17 @@ async function observeJob(
|
|||
): Promise<void> {
|
||||
const target = jobOutputTarget(producerTaskId);
|
||||
const watch = await api.watchDoc(JobOutputDoc, target, context);
|
||||
let value = watch.value;
|
||||
console.log(value === null ? "retired" : value.stdout);
|
||||
try {
|
||||
console.log(watch.revision, watch.value.stdout); // Consume captured initial value.
|
||||
watch.start((value, ops, revision, _context) => {
|
||||
console.log(revision, value === null ? "retired" : value.stdout, ops);
|
||||
}); // Drain buffered changes in order, then continue live delivery.
|
||||
watch.start(async (ops, _context) => {
|
||||
value = applyImmutable(value, ops);
|
||||
await render(value === null ? "retired" : value.stdout);
|
||||
}); // Serialized callbacks never overlap.
|
||||
await finished; // Caller-supplied observation lifetime; outside any commit.
|
||||
} finally {
|
||||
watch.stop(); // Idempotent; discards any remaining buffer.
|
||||
watch.stop(); // Idempotent; prevents another callback from starting.
|
||||
await watch.closed; // Wait for an in-flight callback to settle.
|
||||
}
|
||||
}
|
||||
```
|
||||
|
|
@ -300,16 +307,21 @@ includes that interface. Acquisition validates the target task is live and
|
|||
derives its conversation from the task record.
|
||||
|
||||
```text
|
||||
watchDoc: capture value V at revision 8 + register listener atomically on Session line
|
||||
producer commits revision 9 before start: buffer 9
|
||||
consumer reads watch.value (V), calls start: deliver 9, then subsequent revisions
|
||||
producer becomes terminal: same commit retires output; watch receives null + ["r", null]
|
||||
watchDoc: capture immutable V0 + register for later committed operation batches
|
||||
producer commits D1 and D2 before start: queue D1, D2
|
||||
queue exceeds its limit: replace pending suffix with [["r", V2]]
|
||||
start: caller has initialized from V0; deliver the reset through the async listener
|
||||
producer commits D3 while listener awaits: queue D3; never overlap callbacks
|
||||
producer becomes terminal: queue [["r", null]], deliver it, then close as retired
|
||||
```
|
||||
|
||||
Watches automatically stop when their invocation ends. A Session-acquired watch
|
||||
is caller-owned and stops on Session close. Retirement ends the incarnation's
|
||||
stream; recreation requires a new watch/source and a new numeric
|
||||
document ID. A terminal task cannot create more output. Task-scoped documents
|
||||
is caller-owned and stops on Session close. Previously delivered snapshots never
|
||||
mutate. Slow or unstarted delivery may coalesce an undelivered suffix into a
|
||||
complete reset, so a watch is convergent state observation rather than a
|
||||
transition journal. Retirement ends the incarnation's stream; recreation
|
||||
requires a new watch/source and a new numeric document ID. A terminal task cannot
|
||||
create more output. Task-scoped documents
|
||||
never copy into forks. Preserve required output in result entries or Session- or
|
||||
conversation-scoped documents in the terminal commit before retirement. Use a
|
||||
task-scoped family only when one task needs several independently keyed documents.
|
||||
|
|
@ -337,9 +349,9 @@ dynamic service should then withdraw its instance rather than expose stale data.
|
|||
addStroke -> hold Session mutation line -> await tx.doc -> mutate tracked draft
|
||||
callback succeeds -> tracker flush: incremental ops + candidate value
|
||||
atomic storage commit: all document and record writes; checkpoint predicate selects ops or base
|
||||
storage succeeds -> adopt committed baseline + enqueue source publication, still on line
|
||||
storage succeeds -> materialize immutable published value + enqueue value/ops, still on line
|
||||
release line -> deliver committed source ops -> adapter -> local/remote Chord consumers
|
||||
late subscriber -> atomically capture committed value/revision + subscription
|
||||
late subscriber -> atomically capture committed value + adapter sequence + subscription
|
||||
```
|
||||
|
||||
No visible-undurable path exists. Callback failure restores the unchanged
|
||||
|
|
@ -354,7 +366,10 @@ nothing and poisons the open Session; close and reopen it instead of continuing.
|
|||
Await document access there, not models, processes, network calls, or humans.
|
||||
- Normal `snapshot`, `documentSource`, and `watchDoc` reads are get-or-create.
|
||||
Family `initial` is first-creation input, not an update.
|
||||
- An unstarted watch buffers without a fixed bound. Start or stop promptly.
|
||||
- Initialize from the fixed `watch.value` before `start()`. Slow or unstarted
|
||||
delivery may coalesce an undelivered suffix into a root replacement, omitting
|
||||
intermediate states. Do not use a watch as an audit log.
|
||||
- A listener may call `stop()`, but must not await its own `closed` promise.
|
||||
- Definitions own checkpoints, not storage heuristics. Revise the counting predicates
|
||||
above if mutations change. Keep schema IDs, versions, fork policies, and public
|
||||
paths stable; schema changes require migration, not a source-only rename.
|
||||
|
|
|
|||
|
|
@ -15,9 +15,12 @@ facades, membranes, document routing, view projection, events, or clone chains.
|
|||
|
||||
## 1. Records, cursors, and memory tables
|
||||
|
||||
Implement IDs, sequences, Session metadata with stable root ID, conversations,
|
||||
entries, inputs, live/terminal tasks, document metadata, storage writes, cursors,
|
||||
and detached `MemoryStorage` tables.
|
||||
Implement IDs, sequences, Session metadata with stable root ID,
|
||||
`ConversationRecord`, `EntryRecord`, inputs, live/terminal `TaskRecord` values,
|
||||
document metadata, storage writes, cursors, and detached `MemoryStorage` tables.
|
||||
Reserve `Conversation` for the public conversation object, `Entry` for the typed
|
||||
entry definition, and `Task` for the typed executable definition returned by
|
||||
`defineTask()`.
|
||||
|
||||
Test atomic root-metadata/conversation creation, missing or dangling root
|
||||
metadata rejection, mixed atomic commits, rollback, detached reads/writes,
|
||||
|
|
@ -63,11 +66,14 @@ never reclaimed and default no-fsync behavior matches the specification.
|
|||
|
||||
Keep one tracker per loaded document. Add transaction begin, callback rollback
|
||||
from the unchanged baseline, flush, fatal post-flush failure, committed baseline
|
||||
adoption, and eviction.
|
||||
adoption, and eviction. After storage succeeds, materialize one immutable
|
||||
published value per changed document with `applyImmutable(previous, ops)`,
|
||||
separate from mutable tracker and borrowed storage candidates.
|
||||
|
||||
Do not add membranes, transaction proxy graphs, capability facades, or defensive
|
||||
full-document clones. Test callback failure, no-op transactions, storage failure
|
||||
poisoning, and unload/reload.
|
||||
poisoning, immutable prior published values, replacement-payload ownership, and
|
||||
unload/reload.
|
||||
|
||||
## 7. Document definitions and access
|
||||
|
||||
|
|
@ -114,19 +120,29 @@ unchanged previous immutable snapshots. Internal flushes must remain one commit.
|
|||
Add the thin opaque-source adapter to `ReplicatedState`. It must use committed
|
||||
value/ops directly, with no tracker or re-diff.
|
||||
|
||||
Test contiguous revisions, atomic hydrate/subscribe, attachment races,
|
||||
retirement ending one incarnation, recreation requiring reacquisition, and
|
||||
listener isolation.
|
||||
Test contiguous Chord adapter delivery sequences, atomic hydrate/subscribe,
|
||||
attachment races, retirement ending one incarnation, recreation requiring
|
||||
reacquisition, and listener isolation. Reuse the transaction core's immutable
|
||||
published value; do not materialize another document copy.
|
||||
|
||||
## 12. Document watches
|
||||
|
||||
Implement get-or-create `watchDoc`: capture value/revision and register its
|
||||
listener atomically, buffer without a cap until `start()` or `stop()`, and always
|
||||
deliver value plus ops.
|
||||
Implement get-or-create `watchDoc` as an incarnation-bound `WatchHandle` that
|
||||
atomically captures one fixed immutable value and registers for later committed
|
||||
operation batches. `start()` installs one serialized asynchronous listener.
|
||||
Bound the pending queue by batch count and estimated retained payload; compact an
|
||||
overdue undelivered suffix into one root replacement using the matching latest
|
||||
immutable published value from package 6; never retain mutable tracker or borrowed
|
||||
storage candidates.
|
||||
|
||||
Test ordered draining, live delivery, retirement as absence, idempotent stop,
|
||||
listener errors, and explicit cleanup. Invocation-owned cleanup is integrated in
|
||||
package 15.
|
||||
Test updates between acquisition/return/start; asynchronous consumer
|
||||
initialization; no callback overlap; listener-initiated commits; compaction
|
||||
before start and behind an in-flight callback; one oversized replacement;
|
||||
immutable earlier values; retirement before start and while active; recreation;
|
||||
idempotent stop; second-start rejection; cancellation during acquisition;
|
||||
cancellation/close during a callback; listener-error settlement; `closed`
|
||||
self-join misuse; and invocation-owned
|
||||
cleanup in package 15.
|
||||
|
||||
## 13. Conversations and entries
|
||||
|
||||
|
|
@ -144,7 +160,8 @@ tool-result ordering, and missing post-fork tool results. Replay `content`,
|
|||
ordered named `sections` with `null` removal, then tool removals/additions.
|
||||
|
||||
Test model-less and excluded-stop-reason entries, replacements/omissions,
|
||||
multiple heads, section replacement/removal/re-addition order, rejection of
|
||||
multiple heads, section replacement/removal/re-addition order, order-only
|
||||
configuration changes between separate request preparations, rejection of
|
||||
integer-like section keys, tool addition/removal/replacement order, and raw-view
|
||||
versus model context.
|
||||
|
||||
|
|
@ -198,16 +215,19 @@ specification.
|
|||
Implement `{ conversation, entries, docs }`. Test direct task writes, one
|
||||
publication per Session commit, atomic
|
||||
entry/preview settlement, head changes, contiguous revisions, stable public
|
||||
paths, retry/collapse late-join status, bounded-output truncation metadata, and
|
||||
absence of semantic projection. Specify which diagnostics become entries,
|
||||
paths, immutable acquisition snapshots, asynchronous consumer initialization,
|
||||
serialized updates, reset compaction, retry/collapse late-join status,
|
||||
bounded-output truncation metadata, and absence of semantic projection. Specify which
|
||||
diagnostics become entries,
|
||||
terminal details, or bounded document state.
|
||||
|
||||
## 20. Registries, hooks, and sections
|
||||
|
||||
Implement task/tool/entry/section registries, Session and owned-subtree hooks,
|
||||
positional PR #9548 section/tool updates, complete baselines after a head cut,
|
||||
and preparation revision checks. Do not add a public event stream or
|
||||
plugin-state router.
|
||||
and preparation revision checks. Do not add a Session-kernel semantic event
|
||||
journal or plugin-state router; package 25 adds the thin product notification
|
||||
adapter from specification §9.4.
|
||||
|
||||
Test registration lifetimes, hook replay with memos, exact persisted rendered
|
||||
section strings, minimal section patches and `null` removals, complete baseline
|
||||
|
|
@ -265,22 +285,38 @@ state or entries, and plugin handler recovery.
|
|||
|
||||
Implement the exact Pico3-shaped public surface in specification §2.2:
|
||||
`Harness.open/resume/suspend/close`, lifecycle gates, root/create/lookup
|
||||
conversations, ordered `write` handles, send/input handles, conversation-bound
|
||||
`Conversation` objects, ordered `write` handles, send/input handles,
|
||||
conversation-bound
|
||||
commits, fork/collapse/reset/abort/idle, typed task wait/abort, generic document
|
||||
access, registries, and structural conversation watches. Do not restore Pico3's
|
||||
namespace router, fixed document accessors, semantic view events, or manual Chord
|
||||
view bridge.
|
||||
|
||||
Expose service withdrawal/client detach and product wiring. Implement the v1
|
||||
host-plugin reload path as stop admission, close/join, dispose, rebuild with all
|
||||
new definitions, reopen/migrate, and resume. Test that closing seals commit and
|
||||
Expose service withdrawal/client detach and product wiring. Implement the §9.4
|
||||
agent-mode notification adapter directly from uncoalesced committed publication,
|
||||
without another tracker or persistence authority. Migrate TUI hydration to the
|
||||
structural conversation watch, make print await its own `InputHandle`, and expose
|
||||
JSON/RPC correlated commands plus ordered committed notifications. Test that
|
||||
watch reset compaction cannot erase a subscribed notification lifecycle, late
|
||||
clients use structural hydration rather than event replay, progress notifications
|
||||
reflect durable throttled state rather than every provider frame, and stdout
|
||||
backpressure/disconnect policy stays in the mode adapter.
|
||||
|
||||
Implement the v1 host-plugin reload path as stop admission, close/join, dispose,
|
||||
rebuild with all new definitions, reopen/migrate, and resume. Test that closing seals commit and
|
||||
get-or-create admission, lets storage settlement for already-flushed admitted
|
||||
commits finish despite caller cancellation, writes no abort or terminal outcome,
|
||||
starts no fresh abort invocation, and does not run old and new generations
|
||||
concurrently.
|
||||
commits finish despite caller cancellation, stops watches, joins in-flight watch
|
||||
callbacks and task/tool/hook invocations outside the Session line, writes no abort
|
||||
or terminal outcome, starts no fresh abort invocation, and does not run old and
|
||||
new generations concurrently. Include cancellation during watch acquisition and
|
||||
a non-cooperative watch callback in shutdown/plugin-reload quiescence tests.
|
||||
|
||||
Test stable persisted root identity; atomic conversation/config/section/input
|
||||
creation; fork seed overrides; concrete-entry forks; collapse task-ID return;
|
||||
creation; default `"off"` thinking; every configuration getter/setter; explicit
|
||||
active-tool seed duplicate/unregistered rejection; default active registry
|
||||
snapshot; as-of fork inheritance including unavailable historical names; durable
|
||||
`missing_active_tool` settlement; fork seed overrides; concrete-entry forks;
|
||||
collapse task-ID return;
|
||||
busy reset admission and later placement; mark-only versus signalling abort;
|
||||
conversation abort/join with surviving passive writes and background tasks;
|
||||
quiescence with eligible work; listener initial/future delivery and isolation;
|
||||
|
|
|
|||
|
|
@ -301,7 +301,7 @@ many independently attached presentations</code></pre>
|
|||
<span class="kw">const</span> storage: <span class="type">Storage</span> = <span class="kw">await</span> openAppStorage(context);
|
||||
<span class="kw">const</span> options: <span class="type">HarnessOptions</span> = {
|
||||
models, <span class="comment">// pi-ai Models</span>
|
||||
tools: [read, bash],
|
||||
tools: [read, bash], <span class="comment">// register; new root defaults both active</span>
|
||||
taskKinds: [Reindex],
|
||||
documents: [PlanDoc, JobOutputDoc],
|
||||
sections,
|
||||
|
|
@ -311,7 +311,7 @@ many independently attached presentations</code></pre>
|
|||
<span class="kw">const</span> harness = <span class="kw">await</span> Harness.open(storage, options, context);
|
||||
harness.resume();
|
||||
<span class="kw">const</span> conversation = <span class="kw">await</span> harness.root(context);</code></pre>
|
||||
<p class="subtitle step">Definitions are installed before migration, recovery, and task dispatch.</p>
|
||||
<p class="subtitle step">Registration declares code; only conversation configuration changes an existing active loadout.</p>
|
||||
</div></section>
|
||||
|
||||
<section class="slide" data-slide="21"><div class="frame wide">
|
||||
|
|
@ -385,18 +385,20 @@ console.log(task.state.outcome);</code></pre>
|
|||
<section class="slide" data-slide="25"><div class="frame wide">
|
||||
<h2>Observe the conversation as committed state</h2>
|
||||
<pre class="code-small"><code><span class="kw">const</span> watch = <span class="kw">await</span> conversation.watch(context);
|
||||
render(watch.value);
|
||||
<span class="kw">let</span> view = watch.value;
|
||||
<span class="kw">await</span> render(view);
|
||||
|
||||
watch.start((view, ops, revision, _context) => {
|
||||
render(view);
|
||||
console.log(revision, ops);
|
||||
watch.start(<span class="kw">async</span> (ops, _context) => {
|
||||
view = applyImmutable(view, ops);
|
||||
<span class="kw">await</span> render(view);
|
||||
});
|
||||
|
||||
<span class="comment">// A Session facet may forward the same committed ops.</span>
|
||||
<span class="kw">function</span> disposeUi() { <span class="comment">// on UI teardown</span>
|
||||
<span class="kw">async function</span> disposeUi() { <span class="comment">// on UI teardown</span>
|
||||
watch.stop();
|
||||
<span class="kw">await</span> watch.closed;
|
||||
}</code></pre>
|
||||
<p class="subtitle step">Capture and listener registration are atomic; close stops the watch.</p>
|
||||
<p class="subtitle step">The fixed snapshot plus serialized async updates converge without overlapping callbacks.</p>
|
||||
</div></section>
|
||||
|
||||
<section class="slide" data-slide="26"><div class="frame wide">
|
||||
|
|
@ -642,8 +644,12 @@ commit append assistant entry E
|
|||
<span class="comment">// taskId identifies the supplied live producer task.</span>
|
||||
<span class="kw">const</span> target = { scope: <span class="str">"task"</span>, taskId } <span class="kw">as const</span>;
|
||||
<span class="kw">const</span> watch = <span class="kw">await</span> api.watchDoc(JobOutput, target, context);
|
||||
consume(watch.value);
|
||||
watch.start((value, ops) => consume(value, ops));
|
||||
<span class="kw">let</span> value = watch.value;
|
||||
<span class="kw">await</span> consume(value);
|
||||
watch.start(<span class="kw">async</span> (ops, _context) => {
|
||||
value = applyImmutable(value, ops);
|
||||
<span class="kw">await</span> consume(value);
|
||||
});
|
||||
<span class="comment">// Producer commit: output.stdout = (output.stdout + chunk).slice(-50_000)</span>
|
||||
<span class="comment">// Terminal task commit retires the document.</span></code></pre>
|
||||
<p class="subtitle step">The producer bounds stdout on append; checkpoints only bound replay cost.</p>
|
||||
|
|
@ -669,7 +675,7 @@ crash anywhere → durable marks remain → reopen continues abort</code></pre>
|
|||
<table class="table"><thead><tr><th>Operation</th><th>Durable writes</th><th>Distribution and reopen</th></tr></thead><tbody>
|
||||
<tr class="step"><td>Cancel a Context</td><td>None by itself</td><td>Requests cancellation of one call/wait; remote propagation depends on transport. Committed work is not undone.</td></tr>
|
||||
<tr class="step"><td>Abort task/conversation</td><td>Abort marks, then terminal outcomes</td><td>Traverses ownership; reopen finishes marked aborts.</td></tr>
|
||||
<tr class="step"><td>Close Session</td><td>No abort/terminal writes caused by close</td><td>Admitted commits may settle; then join, stop watches, close storage, and withdraw services. Reopen resumes checkpoints.</td></tr>
|
||||
<tr class="step"><td>Close Session</td><td>No abort/terminal writes caused by close</td><td>Seal admission; signal invocations and stop watches; settle commits and join callbacks; close storage and withdraw services. Reopen resumes checkpoints.</td></tr>
|
||||
</tbody></table>
|
||||
<p class="subtitle step">Disconnecting a client does not silently abort shared work.</p>
|
||||
</div></section>
|
||||
|
|
@ -704,8 +710,8 @@ crash anywhere → durable marks remain → reopen continues abort</code></pre>
|
|||
<section class="slide" data-slide="46"><div class="frame wide">
|
||||
<h2>The conversation view is a mount</h2>
|
||||
<pre class="code-small"><code><span class="kw">type</span> <span class="type">ConversationView</span> = {
|
||||
conversation: <span class="type">Conversation</span>;
|
||||
entries: <span class="kw">readonly</span> <span class="type">Entry</span>[]; <span class="comment">// raw active head → tail</span>
|
||||
conversation: <span class="type">ConversationRecord</span>;
|
||||
entries: <span class="kw">readonly</span> <span class="type">EntryRecord</span>[]; <span class="comment">// raw active head → tail</span>
|
||||
docs: <span class="type">Readonly</span><<span class="type">Record</span><<span class="type">string</span>, <span class="type">JsonObject</span>>>;
|
||||
};
|
||||
|
||||
|
|
@ -715,7 +721,7 @@ crash anywhere → durable marks remain → reopen continues abort</code></pre>
|
|||
</div></section>
|
||||
|
||||
<section class="slide" data-slide="47"><div class="frame wide">
|
||||
<h2>Status is state, not an event</h2>
|
||||
<h2>Status is state, not only an event</h2>
|
||||
<pre><code>live.collapse = { stage: "summarizing", through: 920 }
|
||||
live.collapse = { stage: "retrying", retryAt: 1730000000 }
|
||||
delete live.collapse
|
||||
|
|
@ -723,7 +729,7 @@ delete live.collapse
|
|||
late client during retry:
|
||||
hydrates the current value
|
||||
knows compaction is running and when it retries</code></pre>
|
||||
<p class="subtitle step">A “started” event cannot help a client that attaches later.</p>
|
||||
<p class="subtitle step">Active JSON/RPC clients may receive committed notifications; late clients hydrate state.</p>
|
||||
</div></section>
|
||||
|
||||
<section class="slide" data-slide="48"><div class="frame wide">
|
||||
|
|
@ -750,7 +756,7 @@ late client during retry:
|
|||
<h2>The deliberate sharp edges</h2>
|
||||
<div class="grid2">
|
||||
<div class="panel step"><h3>Ownership</h3><ul class="tight"><li>draft/proxy escape</li><li>inserted and duplicated aliases</li><li>table read after table write</li><li>large values in terminal task records</li></ul></div>
|
||||
<div class="panel step"><h3>Liveness</h3><ul class="tight"><li>external or nested waits on the line</li><li>get-or-create “reads”</li><li>unstarted watch buffering</li><li>checkpoint starvation</li><li>continuing after storage failure</li></ul></div>
|
||||
<div class="panel step"><h3>Liveness</h3><ul class="tight"><li>external or nested waits on the line</li><li>get-or-create “reads”</li><li>watch callback self-join</li><li>checkpoint starvation</li><li>continuing after storage failure</li></ul></div>
|
||||
</div>
|
||||
<p class="subtitle step">Stable definition IDs, mounted paths, and fork settings are persisted protocol.</p>
|
||||
</div></section>
|
||||
|
|
@ -792,7 +798,7 @@ escaped!.strokes.push(otherStroke); <span class="bad">// wrong: transaction ende
|
|||
|
||||
<span class="kw">await</span> session.snapshot(Doc, target, context); <span class="warn">// may create + commit</span>
|
||||
<span class="kw">const</span> watch = <span class="kw">await</span> session.watchDoc(Doc, target, context);
|
||||
<span class="comment">// never start()/stop(): buffered commits are unbounded</span>
|
||||
<span class="comment">// initialize from watch.value; slow pending suffixes may compact to a reset</span>
|
||||
checkpointWhen: () => <span class="kw">false</span> <span class="bad">// replay/storage can grow while live</span>
|
||||
|
||||
known post-flush/checkpoint/storage failure → <span class="bad">close + reopen</span>
|
||||
|
|
|
|||
|
|
@ -5,10 +5,11 @@ Pico5 uses existing package types as follows:
|
|||
|
||||
```ts
|
||||
import type { Context, JsonValue } from "@earendil-works/chord";
|
||||
import type { Op } from "@earendil-works/chord/delta";
|
||||
import { applyImmutable, type Op } from "@earendil-works/chord/delta";
|
||||
import type {
|
||||
Message as ModelMessage,
|
||||
Models,
|
||||
ModelThinkingLevel,
|
||||
TextContent,
|
||||
Tool,
|
||||
ToolReference,
|
||||
|
|
@ -66,7 +67,7 @@ these contracts.
|
|||
type Id = number;
|
||||
type Seq = number;
|
||||
|
||||
type Conversation = {
|
||||
type ConversationRecord = {
|
||||
readonly id: Id;
|
||||
readonly parent?: {
|
||||
readonly conversationId: Id;
|
||||
|
|
@ -107,7 +108,7 @@ type ContextEdit = {
|
|||
readonly messages?: readonly RequestMessage[];
|
||||
};
|
||||
|
||||
type Entry = {
|
||||
type EntryRecord = {
|
||||
readonly id: Id;
|
||||
readonly conversationId: Id;
|
||||
readonly kind: string;
|
||||
|
|
@ -118,7 +119,7 @@ type Entry = {
|
|||
readonly byTaskId?: Id;
|
||||
};
|
||||
|
||||
type EntryDraft = Omit<Entry, "id" | "conversationId" | "byTaskId" | "head"> & {
|
||||
type EntryDraft = Omit<EntryRecord, "id" | "conversationId" | "byTaskId" | "head"> & {
|
||||
readonly head?: Id | "self";
|
||||
};
|
||||
|
||||
|
|
@ -199,9 +200,10 @@ type ConversationSpec = {
|
|||
readonly parent?: { readonly conversationId: Id; readonly at: Id };
|
||||
readonly model?: ModelRef;
|
||||
readonly sections?: readonly SectionSeed[];
|
||||
readonly activeTools?: readonly string[];
|
||||
};
|
||||
|
||||
type AnyTaskKind = {
|
||||
type AnyTask = {
|
||||
readonly definition: {
|
||||
readonly name: string;
|
||||
readonly version: number;
|
||||
|
|
@ -216,7 +218,7 @@ type AnyTaskKind = {
|
|||
type HarnessOptions = {
|
||||
readonly models: Models; // the pi-ai Models interface
|
||||
readonly tools?: readonly ToolRegistration[];
|
||||
readonly taskKinds?: readonly AnyTaskKind[];
|
||||
readonly taskKinds?: readonly AnyTask[];
|
||||
readonly documents?: readonly AnyDocToken[];
|
||||
readonly sections?: readonly SystemSection<JsonValue>[];
|
||||
readonly plugins?: Readonly<Record<string, PluginHandler>>;
|
||||
|
|
@ -226,16 +228,16 @@ type HarnessOptions = {
|
|||
readonly onReport?: (error: unknown) => void;
|
||||
};
|
||||
|
||||
interface EntryKind<E extends Entry = Entry> {
|
||||
interface Entry<E extends EntryRecord = EntryRecord> {
|
||||
readonly kind: string;
|
||||
is(entry: Entry | undefined): entry is E;
|
||||
is(entry: EntryRecord | undefined): entry is E;
|
||||
}
|
||||
|
||||
function defineEntry<E extends Entry>(kind: string): EntryKind<E>;
|
||||
function defineEntry<E extends EntryRecord>(kind: string): Entry<E>;
|
||||
|
||||
type ContextView = {
|
||||
readonly head: Entry | undefined;
|
||||
readonly entries: readonly Entry[];
|
||||
readonly head: EntryRecord | undefined;
|
||||
readonly entries: readonly EntryRecord[];
|
||||
readonly messages: readonly RequestMessage[];
|
||||
};
|
||||
|
||||
|
|
@ -252,30 +254,37 @@ interface InputHandle {
|
|||
): Promise<"aborted" | "already_placed" | "not_found">;
|
||||
}
|
||||
|
||||
type SettledTask<R> = Task<JsonValue, JsonValue, R> & {
|
||||
type SettledTask<R> = TaskRecord<JsonValue, JsonValue, R> & {
|
||||
readonly state: TerminalTask<R>;
|
||||
};
|
||||
|
||||
interface ConversationWatch {
|
||||
readonly value: ConversationView;
|
||||
readonly revision: number;
|
||||
start(listener: (
|
||||
value: ConversationView,
|
||||
ops: readonly Op[],
|
||||
revision: number,
|
||||
context: Context,
|
||||
) => void): void;
|
||||
stop(): void;
|
||||
}
|
||||
type ConversationWatch = WatchHandle<ConversationView>;
|
||||
|
||||
type HooksOf<K> = K extends TaskKind<infer _I, infer _S, infer _R, infer H>
|
||||
type HooksOf<K> = K extends Task<infer _I, infer _S, infer _R, infer H>
|
||||
? H
|
||||
: never;
|
||||
|
||||
interface ConversationHandle {
|
||||
interface Conversation {
|
||||
readonly id: Id;
|
||||
send(input: SendInput, context: Context): Promise<InputHandle>;
|
||||
write(entry: EntryDraft, context: Context): Promise<InputHandle>;
|
||||
|
||||
getModel(context: Context): Promise<ModelRef | undefined>;
|
||||
setModel(model: ModelRef | undefined, context: Context): Promise<void>;
|
||||
getThinkingLevel(context: Context): Promise<ModelThinkingLevel>;
|
||||
setThinkingLevel(level: ModelThinkingLevel, context: Context): Promise<void>;
|
||||
getActiveTools(context: Context): Promise<readonly string[]>;
|
||||
setActiveTools(names: readonly string[], context: Context): Promise<void>;
|
||||
getSection<T extends JsonValue>(
|
||||
section: SystemSection<T>,
|
||||
context: Context,
|
||||
): Promise<T | undefined>;
|
||||
setSection<T extends JsonValue>(
|
||||
section: SystemSection<T>,
|
||||
value: T | undefined,
|
||||
context: Context,
|
||||
): Promise<void>;
|
||||
|
||||
commit<T>(
|
||||
change: (tx: Tx) => T | Promise<T>,
|
||||
context: Context,
|
||||
|
|
@ -285,14 +294,14 @@ interface ConversationHandle {
|
|||
at: Id,
|
||||
spec: Omit<ConversationSpec, "parent">,
|
||||
context: Context,
|
||||
): Promise<ConversationHandle>;
|
||||
): Promise<Conversation>;
|
||||
collapse(instructions: string | undefined, context: Context): Promise<Id>;
|
||||
reset(handoff: string | undefined, context: Context): Promise<void>;
|
||||
abort(context: Context): Promise<void>;
|
||||
waitForIdle(context: Context): Promise<void>;
|
||||
hooks<K extends AnyTaskKind>(
|
||||
hooks<K extends AnyTask>(
|
||||
owner: string,
|
||||
kind: K,
|
||||
task: K,
|
||||
handlers: Partial<HooksOf<K>>,
|
||||
options?: { readonly subtree?: boolean },
|
||||
): () => void;
|
||||
|
|
@ -306,31 +315,31 @@ interface Harness extends Session {
|
|||
hold(): () => void;
|
||||
|
||||
registerDocument(token: AnyDocToken): () => void;
|
||||
registerTaskKind(kind: AnyTaskKind): () => void;
|
||||
registerTaskKind(task: AnyTask): () => void;
|
||||
registerTool(tool: ToolRegistration): () => void;
|
||||
registerSection(section: SystemSection<JsonValue>): () => void;
|
||||
registerEntryKind(kind: EntryKind): () => void;
|
||||
hooks<K extends AnyTaskKind>(
|
||||
registerEntryKind(entry: Entry): () => void;
|
||||
hooks<K extends AnyTask>(
|
||||
owner: string,
|
||||
kind: K,
|
||||
task: K,
|
||||
handlers: Partial<HooksOf<K>>,
|
||||
): () => void;
|
||||
|
||||
root(context: Context): Promise<ConversationHandle>;
|
||||
onConversation(listener: (conversation: ConversationHandle) => void): () => void;
|
||||
conversation(id: Id, context: Context): Promise<ConversationHandle | undefined>;
|
||||
root(context: Context): Promise<Conversation>;
|
||||
onConversation(listener: (conversation: Conversation) => void): () => void;
|
||||
conversation(id: Id, context: Context): Promise<Conversation | undefined>;
|
||||
createConversation(
|
||||
spec: ConversationSpec & { readonly input?: UserInput },
|
||||
context: Context,
|
||||
): Promise<ConversationHandle>;
|
||||
): Promise<Conversation>;
|
||||
entries(
|
||||
query: EntryQuery,
|
||||
cursor: EntryCursor | undefined,
|
||||
limit: number,
|
||||
context: Context,
|
||||
): Promise<Page<Entry, EntryCursor>>;
|
||||
): Promise<Page<EntryRecord, EntryCursor>>;
|
||||
|
||||
getTask(id: Id, context: Context): Promise<Task<JsonValue, JsonValue, JsonValue> | undefined>;
|
||||
getTask(id: Id, context: Context): Promise<TaskRecord<JsonValue, JsonValue, JsonValue> | undefined>;
|
||||
abortInput(
|
||||
id: Id,
|
||||
context: Context,
|
||||
|
|
@ -385,12 +394,49 @@ movement during normal operation, not replacement of executing plugin code;
|
|||
section 7.4 governs code reload.
|
||||
|
||||
Conversation creation atomically commits the conversation, built-in
|
||||
configuration and section seeds, and optional input admission. For a fork,
|
||||
omitted model/section values follow their document definitions' fork policies;
|
||||
provided values override those forked built-in values in the same commit. Other
|
||||
documents follow their own definitions without special handling.
|
||||
configuration and section/tool seeds, and optional input admission. For a fork,
|
||||
omitted model, section, and active-tool values follow their document definition's
|
||||
fork policy; provided values override those forked built-in values in the same
|
||||
commit. Other documents follow their own definitions without special handling.
|
||||
|
||||
A `ConversationHandle.commit()` is a Session commit bound to that conversation.
|
||||
The built-in conversation configuration document contains the selected model,
|
||||
thinking level, section values, and active tool names. Its initial thinking level
|
||||
is `"off"`. It is rewindable with `fork: "asOf"`, so a child starts from the
|
||||
configuration visible at its selected entry unless explicit creation seeds
|
||||
override it. Conversation creation eagerly
|
||||
creates it, so `getModel()`, `getThinkingLevel()`, `getActiveTools()`, and
|
||||
`getSection()` return detached committed snapshots without a get-or-create
|
||||
write. Each setter performs one ordinary Session commit against that document;
|
||||
`setSection(section, undefined)` removes the value. A setter does not start
|
||||
generation or append a system entry. Request preparation later compares the
|
||||
desired committed configuration with transcript history and appends the required
|
||||
positional system baseline or delta.
|
||||
|
||||
Tool declarations and executable functions are process-local Harness registry
|
||||
entries, supplied through `HarnessOptions.tools` or `registerTool()`. The durable
|
||||
configuration document stores only active tool names. A new independent
|
||||
conversation defaults to every tool registered when its creation is admitted;
|
||||
`ConversationSpec.activeTools` overrides that default. A fork with no explicit
|
||||
`activeTools` uses the configuration document's `asOf` value.
|
||||
|
||||
`registerTool()` changes only the runtime registry. It does not activate the tool
|
||||
in existing conversations or write a document. `setActiveTools()` is the only
|
||||
host operation that replaces one existing conversation's durable active set.
|
||||
Explicit `ConversationSpec.activeTools` and `setActiveTools()` values must contain
|
||||
unique names that are registered when their commit is admitted; otherwise the
|
||||
whole operation rejects without a write. Inherited historical names are not
|
||||
revalidated during a fork and may later be unavailable after registry movement.
|
||||
Unregistering a tool does not rewrite any conversation.
|
||||
|
||||
If request preparation finds unavailable active names, it performs no provider
|
||||
request. It atomically terminalizes the generation task as `failed` with
|
||||
`detail: { code: "missing_active_tool", names }`, makes its placed inputs
|
||||
`unanswered` with reason `missing_active_tool`, clears matching turn control, and
|
||||
appends a visible model-less diagnostic entry. It does not silently change the
|
||||
durable loadout. Historical system entries remain replayable because they store
|
||||
the exact declarations actually offered to prior requests.
|
||||
|
||||
A `Conversation.commit()` is a Session commit bound to that conversation.
|
||||
`tx.createTask()` defaults `TaskOptions.conversationId` to the bound conversation.
|
||||
Generic Session-wide document operations remain available directly on `Harness`
|
||||
because `Harness extends Session`.
|
||||
|
|
@ -405,7 +451,7 @@ does not guarantee placement of queued passive writes.
|
|||
`markTask()` only commits `abortRequested`; it neither signals nor joins an
|
||||
active invocation. The scheduler notices the mark on its next drain.
|
||||
`abortTask()` also signals and joins an active run before starting the abort
|
||||
invocation. `ConversationHandle.abort()` withdraws queued non-write inputs, marks
|
||||
invocation. `Conversation.abort()` withdraws queued non-write inputs, marks
|
||||
non-background tasks in the conversation and owned subtree, signals them, and
|
||||
resolves only after that subtree is ordinarily idle. Passive writes and
|
||||
background tasks survive. Conversation idle means no non-background live task
|
||||
|
|
@ -434,14 +480,18 @@ the terminal receipt. Aborting an already terminal task returns `terminal`; an
|
|||
unknown ID rejects. Explicit task abort includes a background task. Cancelling a
|
||||
task or idle wait does not abort work.
|
||||
|
||||
`ConversationHandle.watch()` atomically captures the current structural view and
|
||||
subscribes to later complete Session commits. It has the same stopped-start,
|
||||
unbounded pre-start buffering, listener isolation, and close ownership as
|
||||
`watchDoc()`. It carries no semantic events and owns no second tracker.
|
||||
`Conversation.watch()` atomically captures an immutable structural view and
|
||||
registers for later complete Session commits. Its `WatchHandle` uses the same
|
||||
serialized asynchronous update and reset-compaction contract as `watchDoc()` in
|
||||
section 9.2. It carries no semantic events and owns no second tracker.
|
||||
|
||||
`close()` is equivalent to `suspend()` for v1: it stops reservation, signals and
|
||||
joins invocations, closes watches and storage, and writes no task outcome. Handles
|
||||
belong to that open Harness and must be reacquired after reopen.
|
||||
`close()` is equivalent to `suspend()` for v1. It seals mutation admission and
|
||||
task reservation, signals invocations, and stops watches. Outside the Session
|
||||
line it lets already-admitted storage commits settle, joins task/tool/hook
|
||||
invocations and in-flight watch callbacks, then closes sources and storage. It
|
||||
writes no task outcome. Handles belong to that open Harness and must be
|
||||
reacquired after reopen. A non-cooperative watch callback can delay graceful
|
||||
close just like a non-cooperative task invocation.
|
||||
|
||||
## 3. Documents
|
||||
|
||||
|
|
@ -633,21 +683,21 @@ interface Session extends DocumentObserver {
|
|||
}
|
||||
|
||||
interface Tx {
|
||||
conversation(id: Id): Promise<Conversation | undefined>;
|
||||
entry(id: Id): Promise<Entry | undefined>;
|
||||
conversation(id: Id): Promise<ConversationRecord | undefined>;
|
||||
entry(id: Id): Promise<EntryRecord | undefined>;
|
||||
input(id: Id): Promise<Input | undefined>;
|
||||
task(id: Id): Promise<Task<JsonValue, JsonValue, JsonValue> | undefined>;
|
||||
scanEntries(query: EntryQuery): Promise<readonly Entry[]>;
|
||||
scanTasks(query: TaskQuery): Promise<readonly Task<JsonValue, JsonValue, JsonValue>[]>;
|
||||
task(id: Id): Promise<TaskRecord<JsonValue, JsonValue, JsonValue> | undefined>;
|
||||
scanEntries(query: EntryQuery): Promise<readonly EntryRecord[]>;
|
||||
scanTasks(query: TaskQuery): Promise<readonly TaskRecord<JsonValue, JsonValue, JsonValue>[]>;
|
||||
|
||||
createConversation(value: Omit<Conversation, "id">): Conversation;
|
||||
appendEntry(conversationId: Id, value: EntryDraft): Entry;
|
||||
createConversation(value: Omit<ConversationRecord, "id">): ConversationRecord;
|
||||
appendEntry(conversationId: Id, value: EntryDraft): EntryRecord;
|
||||
createInput(value: Omit<Input, "id">): Input;
|
||||
setInput(value: Input): void;
|
||||
createTask<I, S extends { phase: string }, R, H extends object>(
|
||||
kind: TaskKind<I, S, R, H>, input: I, options?: TaskOptions,
|
||||
task: Task<I, S, R, H>, input: I, options?: TaskOptions,
|
||||
): TaskRef<R>;
|
||||
setTask(value: Task<JsonValue, JsonValue, JsonValue>): void;
|
||||
setTask(value: TaskRecord<JsonValue, JsonValue, JsonValue>): void;
|
||||
|
||||
doc<T extends JsonObject>(token: DocToken<T>, target: DocTarget): Promise<T>;
|
||||
doc<T extends JsonObject, I extends JsonValue>(token: DocFamilyToken<T, I>, target: FamilyTarget<I>): Promise<T>;
|
||||
|
|
@ -722,12 +772,19 @@ callback succeeds
|
|||
Storage.commit evaluates each ordinary mutation's checkpoint predicate exactly once
|
||||
and persists the selected representation while the Session line remains held
|
||||
storage succeeds
|
||||
adopt committed baselines and enqueue publication while holding the line
|
||||
materialize each changed immutable published value with applyImmutable(previous, ops)
|
||||
adopt committed baselines and enqueue value/ops publication while holding the line
|
||||
release the line; invoke listeners later
|
||||
post-flush or storage failure
|
||||
poison Session; publish nothing; close/reopen required
|
||||
```
|
||||
|
||||
The immutable published value is separate from the tracker's mutable working
|
||||
object and is created once per changed document, not once per watch. Initial
|
||||
publication similarly applies the complete base operation to `undefined`.
|
||||
Storage's detached retained copy does not satisfy this Session-side ownership
|
||||
requirement.
|
||||
|
||||
No defensive document copy is required solely for storage failure because the
|
||||
open Session cannot continue after that failure.
|
||||
|
||||
|
|
@ -890,7 +947,7 @@ type TerminalTask<R> = {
|
|||
readonly outcome: TaskOutcome<R>;
|
||||
};
|
||||
|
||||
type Task<I, S, R> = {
|
||||
type TaskRecord<I, S, R> = {
|
||||
readonly id: Id;
|
||||
readonly conversationId: Id;
|
||||
readonly kind: string;
|
||||
|
|
@ -903,7 +960,7 @@ type Task<I, S, R> = {
|
|||
readonly memos?: Readonly<Record<string, JsonValue>>; // live tasks only
|
||||
};
|
||||
|
||||
type RunningTask<I, S, R> = Task<I, S, R> & {
|
||||
type RunningTask<I, S, R> = TaskRecord<I, S, R> & {
|
||||
readonly state: LiveTask<S> & { readonly status: "running" };
|
||||
};
|
||||
|
||||
|
|
@ -940,7 +997,7 @@ type TaskDefinition<I, S extends { phase: string }, R, H extends object> = {
|
|||
readonly phases: {
|
||||
[P in S["phase"]]: PhaseHandler<I, Extract<S, { phase: P }>, S, R, H>;
|
||||
};
|
||||
abort(task: Task<I, S, R>, runtime: TaskRuntime<I, S, R, H>, context: Context): Promise<void>;
|
||||
abort(task: TaskRecord<I, S, R>, runtime: TaskRuntime<I, S, R, H>, context: Context): Promise<void>;
|
||||
migrate?(input: JsonValue, checkpoint: JsonValue, fromVersion: number): {
|
||||
input: I;
|
||||
checkpoint: S;
|
||||
|
|
@ -948,7 +1005,7 @@ type TaskDefinition<I, S extends { phase: string }, R, H extends object> = {
|
|||
readonly hooks?: H;
|
||||
};
|
||||
|
||||
interface TaskKind<I, S extends { phase: string }, R, H extends object> {
|
||||
interface Task<I, S extends { phase: string }, R, H extends object> {
|
||||
readonly definition: TaskDefinition<I, S, R, H>;
|
||||
}
|
||||
|
||||
|
|
@ -962,7 +1019,7 @@ type TaskOptions = {
|
|||
|
||||
function defineTask<I, S extends { phase: string }, R, H extends object = {}>(
|
||||
definition: TaskDefinition<I, S, R, H>,
|
||||
): TaskKind<I, S, R, H>;
|
||||
): Task<I, S, R, H>;
|
||||
```
|
||||
|
||||
The phase map is exhaustive and phase-narrowed. A handler may perform several
|
||||
|
|
@ -989,11 +1046,13 @@ On open, running-task reconciliation changes surviving `running` tasks back to
|
|||
`pending`, preserving their checkpoint and abort mark. Task migration then runs
|
||||
before dispatch. One callback handles every supported older version; newer or
|
||||
unmigratable live tasks become orphaned.
|
||||
`close()` marks the runtime closing, seals admission and reservation, signals and
|
||||
joins invocations, rejects later runtime commits, writes no task outcome, and
|
||||
then closes storage. Closing starts no fresh phase or abort invocation. It does not set
|
||||
abort marks, terminalize tasks, retire task documents, or publish document
|
||||
retirement. Session-owned watches stop. The hosting layer withdraws services and
|
||||
`close()` marks the runtime closing, seals admission and reservation, signals
|
||||
invocations, and stops watches. Outside the Session line it settles admitted
|
||||
commits and joins invocations plus in-flight watch callbacks before closing
|
||||
storage. Later runtime commits reject, and close writes no task outcome. Closing
|
||||
starts no fresh phase or abort invocation. It does not set abort marks,
|
||||
terminalize tasks, retire task documents, or publish document retirement. The
|
||||
hosting layer withdraws services and
|
||||
detaches clients; reconnecting to a reopened Session hydrates the last committed
|
||||
state and resumes recovery from its durable checkpoints.
|
||||
|
||||
|
|
@ -1175,7 +1234,7 @@ type ToolExecutionResult = {
|
|||
readonly control?: ToolControl;
|
||||
};
|
||||
|
||||
interface OwnedConversationHandle {
|
||||
interface OwnedConversation {
|
||||
readonly id: Id;
|
||||
send(input: SendInput, context: Context): Promise<InputHandle>;
|
||||
abort(context: Context): Promise<void>;
|
||||
|
|
@ -1192,17 +1251,17 @@ interface ToolExecutionApi extends DocumentObserver {
|
|||
memo<T extends JsonValue>(name: string, context: Context): Promise<T | undefined>;
|
||||
memo<T extends JsonValue>(name: string, candidate: T, context: Context): Promise<T>;
|
||||
createTask<I, S extends { phase: string }, R, H extends object>(
|
||||
kind: TaskKind<I, S, R, H>,
|
||||
task: Task<I, S, R, H>,
|
||||
input: I,
|
||||
options: Omit<TaskOptions, "conversationId">,
|
||||
context: Context,
|
||||
): Promise<TaskRef<R>>;
|
||||
getTask<R>(ref: TaskRef<R>, context: Context): Promise<Task<JsonValue, JsonValue, R> | undefined>;
|
||||
getTask<R>(ref: TaskRef<R>, context: Context): Promise<TaskRecord<JsonValue, JsonValue, R> | undefined>;
|
||||
waitForTask<R>(ref: TaskRef<R>, context: Context): Promise<SettledTask<R>>;
|
||||
createConversation(
|
||||
spec: Omit<ConversationSpec, "parent"> & { readonly inherit?: boolean },
|
||||
context: Context,
|
||||
): Promise<OwnedConversationHandle>;
|
||||
): Promise<OwnedConversation>;
|
||||
}
|
||||
|
||||
type PluginHandler = (
|
||||
|
|
@ -1308,7 +1367,7 @@ Pico stores prompt and tool changes directly as PR #9548 `SystemMessage` values
|
|||
at their transcript positions:
|
||||
|
||||
```ts
|
||||
type SystemEntry = Entry & {
|
||||
type SystemEntry = EntryRecord & {
|
||||
readonly kind: "pi.system";
|
||||
readonly model: readonly [SystemMessage];
|
||||
};
|
||||
|
|
@ -1342,6 +1401,15 @@ existing section; `null` removes it, and a later re-addition appends it to the
|
|||
ordered section map. Within one message, tool removals happen before additions,
|
||||
so a same-name replacement gets the new declaration and position.
|
||||
|
||||
The built-in configuration document preserves section insertion order.
|
||||
`setSection(section, undefined)` removes the key; setting it later appends it at
|
||||
the end. Preparation compares both values and order. If values can be patched
|
||||
without changing order, it emits the minimal patch. If effective and desired
|
||||
section order differ, one commit appends two `pi.system` entries: the first
|
||||
removes every effective section with `null`, and the second re-adds every desired
|
||||
section in desired order. This makes order-only changes and deletion/re-addition
|
||||
between requests replay exactly; merely restating equal values is insufficient.
|
||||
|
||||
A PR #9548 `SystemMessage` is always a patch, not a reset: it cannot remove
|
||||
previous `content` or restore section order merely by restating current values.
|
||||
Therefore, when a head removes the previous request-visible baseline, the new
|
||||
|
|
@ -1349,10 +1417,12 @@ Therefore, when a head removes the previous request-visible baseline, the new
|
|||
entry still retained after the cut. Its own message is then a complete baseline
|
||||
containing the base `content`, every desired section in order, and every effective
|
||||
tool declaration. Model-context replay sees the new baseline instead of the
|
||||
omitted retained deltas. Otherwise preparation emits only changed section
|
||||
values, `null` removals, and tool additions/removals. Same-name tool replacements
|
||||
remove before adding. Registry or document changes while preparation hooks run
|
||||
cause preparation to retry against a new snapshot.
|
||||
omitted retained deltas. Head rebaselining takes precedence over ordinary
|
||||
order/value patching. Without a head cut, an order mismatch uses the two-entry
|
||||
remove/re-add sequence above; only when order already matches does preparation
|
||||
emit the minimal changed values, `null` removals, and tool additions/removals.
|
||||
Same-name tool replacements remove before adding. Registry or document changes
|
||||
while preparation hooks run cause preparation to retry against a new snapshot.
|
||||
|
||||
Conversation creation may seed section values or explicit removals. Preparation
|
||||
uses a mutable section draft with get/set/delete/wrap; each throwing hook loses
|
||||
|
|
@ -1373,7 +1443,8 @@ In v1, changing host plugin code is a Harness generation boundary:
|
|||
|
||||
1. Stop new admission and task reservation.
|
||||
2. Close the Harness, signalling and joining its active task, tool, and hook
|
||||
invocations without writing abort marks or terminal outcomes.
|
||||
invocations plus in-flight watch callbacks without writing abort marks or
|
||||
terminal outcomes.
|
||||
3. Dispose the old facets and registrations.
|
||||
4. Construct a new Harness over the same storage.
|
||||
5. Register the complete new definition set before open performs task/document
|
||||
|
|
@ -1481,18 +1552,23 @@ interface DocumentSource<T extends JsonObject> {
|
|||
readonly [documentSourceType]: T;
|
||||
}
|
||||
|
||||
interface DocumentWatch<T extends JsonObject> {
|
||||
readonly value: Readonly<T>;
|
||||
readonly revision: number;
|
||||
start(listener: (
|
||||
value: Readonly<T> | null,
|
||||
ops: readonly Op[],
|
||||
revision: number,
|
||||
context: Context,
|
||||
) => void): void;
|
||||
type WatchEnd =
|
||||
| { readonly reason: "stopped" | "cancelled" | "session_closed" | "retired" }
|
||||
| { readonly reason: "listener_error"; readonly error: Error };
|
||||
|
||||
interface WatchHandle<T> {
|
||||
/** Immutable acquisition snapshot. This reference never changes. */
|
||||
readonly value: T;
|
||||
/** Installs the sole serialized asynchronous listener. */
|
||||
start(listener: (ops: readonly Op[], context: Context) => Promise<void>): void;
|
||||
/** Prevents another callback from starting and signals an in-flight callback. */
|
||||
stop(): void;
|
||||
/** Settles after termination and any in-flight callback. */
|
||||
readonly closed: Promise<WatchEnd>;
|
||||
}
|
||||
|
||||
type DocumentWatch<T extends JsonObject> = WatchHandle<Readonly<T> | null>;
|
||||
|
||||
interface DocumentObserver {
|
||||
watchDoc<T extends JsonObject>(token: DocToken<T>, target: DocTarget, context: Context): Promise<DocumentWatch<T>>;
|
||||
watchDoc<T extends JsonObject, I extends JsonValue>(token: DocFamilyToken<T, I>, target: FamilyTarget<I>, context: Context): Promise<DocumentWatch<T>>;
|
||||
|
|
@ -1520,9 +1596,11 @@ and ends that incarnation's stream. The service may then withdraw itself; if it
|
|||
remains exposed, consumers see `null`, never stale state. A later recreation
|
||||
requires acquiring a new source/watch.
|
||||
|
||||
Each document source has a contiguous revision independent of the sparse global
|
||||
Session commit sequence. Hydration atomically captures matching value/revision
|
||||
and subscribes before releasing the Session line.
|
||||
Each Chord adapter assigns its own in-memory contiguous delivery sequence; Pico
|
||||
does not persist or expose that sequence through `WatchHandle`. Hydration
|
||||
atomically captures a matching immutable value and adapter sequence before
|
||||
subsequent source operations are delivered. Reopen creates a new source
|
||||
lifetime and hydration.
|
||||
|
||||
### 9.2 `watchDoc`
|
||||
|
||||
|
|
@ -1530,23 +1608,81 @@ Tasks, hooks, and tools may observe any document for which their code has a toke
|
|||
and target. There is no additional subtree permission system inside trusted
|
||||
Session code.
|
||||
|
||||
`watchDoc()` is available on task, hook, and tool APIs. It atomically registers
|
||||
the internal listener and captures the current value/revision on the Session
|
||||
line. It returns a stopped delivery surface:
|
||||
`watchDoc()` is available on task, hook, and tool APIs. Acquisition performs
|
||||
normal get-or-create and, on the Session line, binds one concrete incarnation,
|
||||
captures its immutable committed value, and registers the handle for every later
|
||||
committed operation batch. `watch.value` is that fixed acquisition snapshot and
|
||||
never changes.
|
||||
|
||||
```ts
|
||||
const watch = await api.watchDoc(JobOutputDoc, target, context);
|
||||
consume(watch.value);
|
||||
watch.start((value, ops) => consume(value, ops));
|
||||
let value = watch.value;
|
||||
try {
|
||||
await initializeConsumer(value, context);
|
||||
watch.start(async (ops, deliveryContext) => {
|
||||
value = applyImmutable(value, ops);
|
||||
await consume(value, deliveryContext);
|
||||
});
|
||||
} catch (error) {
|
||||
watch.stop();
|
||||
await watch.closed;
|
||||
throw error;
|
||||
}
|
||||
```
|
||||
|
||||
Updates between capture and `start()` are buffered in order without a fixed
|
||||
bound. `start()` drains them and continues live delivery. `stop()` discards the
|
||||
buffer and is idempotent. Watches acquired by an invocation stop when it ends;
|
||||
watches acquired from `Session` are caller-owned and stop on `Session.close()`.
|
||||
Listener failures are isolated and cannot affect committed state or other
|
||||
listeners. Retirement delivers `null` with `["r", null]`, the JSON
|
||||
representation of document absence.
|
||||
The caller initializes from `value` before `start()`. `start()` synchronously
|
||||
installs the sole listener, changes the prepared handle to active, and schedules
|
||||
delivery; it never invokes the listener inline. A second `start()`, or `start()`
|
||||
after stop, throws.
|
||||
|
||||
One per-watch delivery line awaits each callback before starting the next, so
|
||||
callbacks never overlap. Commits never wait for callback settlement: while one
|
||||
batch is in flight, later batches append to the pending queue. An update accepted
|
||||
between acquisition and return, or between return and `start()`, is therefore not
|
||||
lost. The listener's promise covers all work the watch serializes; fire-and-forget
|
||||
work started by the listener is outside that guarantee.
|
||||
|
||||
The pending queue is bounded by operation-batch count and estimated retained
|
||||
payload. When either threshold is exceeded, the handle leaves the in-flight batch
|
||||
untouched and replaces the entire pending suffix with one newly allocated root
|
||||
replacement batch:
|
||||
|
||||
```ts
|
||||
[["r", latestImmutablePublishedValue]]
|
||||
```
|
||||
|
||||
Later deltas append behind that reset. A single reset may exceed the payload
|
||||
threshold because compaction cannot bound document size; while it remains
|
||||
pending, further overload replaces it with a newer reset rather than accumulating
|
||||
an unbounded suffix. This watch observes convergent committed state, not every
|
||||
intermediate transition.
|
||||
|
||||
The replacement value is the immutable publication value corresponding exactly
|
||||
to the last batch compacted into it. After storage succeeds, Pico materializes
|
||||
that value once per changed document with
|
||||
`applyImmutable(previousPublishedValue, ops)`; it never retains mutable
|
||||
`Tracker.state`, `Tracker.target`, or a borrowed storage candidate. Watches may
|
||||
share the immutable publication value and operation payloads. A mutable consumer
|
||||
must detach them before using mutable `apply()` or `track()`.
|
||||
|
||||
A watch remains bound to its original incarnation. Retirement replaces the
|
||||
pending suffix with `[["r", null]]`. Before start, `value` remains the original
|
||||
snapshot and the terminal reset becomes the first delivered batch. While active,
|
||||
the reset follows any in-flight callback. Successful terminal delivery closes as
|
||||
`retired`; recreation requires another watch.
|
||||
|
||||
`stop()` is idempotent, unregisters the handle, discards pending batches,
|
||||
prevents another callback from starting, and signals the watch delivery context.
|
||||
An in-flight callback is allowed to settle; `closed` resolves only afterward.
|
||||
The acquisition `Context` governs the watch lifetime. Cancellation during
|
||||
acquisition cleans up any registration before rejecting; an already-admitted
|
||||
get-or-create commit still settles and is not undone. Cancellation immediately
|
||||
after successful acquisition may therefore return an already-stopped handle,
|
||||
whose `start()` throws and whose `closed` reports `cancelled`. Invocation
|
||||
termination and Session close stop owned watches similarly. Listener rejection
|
||||
is reported, discards pending work, and closes only that watch as
|
||||
`listener_error`. The first termination reason wins, and every late rejection is
|
||||
observed.
|
||||
|
||||
### 9.3 Conversation view
|
||||
|
||||
|
|
@ -1554,8 +1690,8 @@ The public view is a fixed structural mount of selected built-in documents:
|
|||
|
||||
```ts
|
||||
type ConversationView = {
|
||||
readonly conversation: Conversation;
|
||||
readonly entries: readonly Entry[];
|
||||
readonly conversation: ConversationRecord;
|
||||
readonly entries: readonly EntryRecord[];
|
||||
readonly docs: Readonly<Record<string, JsonObject>>;
|
||||
};
|
||||
```
|
||||
|
|
@ -1573,15 +1709,48 @@ document op ["s", ["message"], value]
|
|||
|
||||
Entry appends/head changes and every changed mounted document are included in
|
||||
the same publication. The mount owns no tracker and performs no semantic
|
||||
projection. Its revision is contiguous per view.
|
||||
projection. It materializes one immutable published view per batch with
|
||||
`applyImmutable(previousView, mountedOps)` so every conversation watch can share
|
||||
that value for reset compaction. A Chord adapter assigns a contiguous in-memory
|
||||
delivery sequence per view source lifetime.
|
||||
|
||||
`ConversationHandle.watch()` exposes that mount directly. Capture of
|
||||
`ConversationView`, revision, and listener registration is atomic on the Session
|
||||
line. Like `watchDoc()`, it returns stopped, buffers without a fixed bound until
|
||||
`start()` or `stop()`, drains in order, isolates listener failures, and stops on
|
||||
Harness close. Chord Session facets may forward the captured value and committed
|
||||
ops through services, but that product wiring is not part of the Harness facade
|
||||
and must not add another tracker or semantic event envelope.
|
||||
`Conversation.watch()` exposes that mount through the same immutable acquisition
|
||||
snapshot, serialized asynchronous listener, and reset-compacted pending queue as
|
||||
`watchDoc()`. Each operation batch represents one complete Session commit. Chord
|
||||
Session facets may forward the captured value and committed operations through
|
||||
services, but that product wiring is not part of the Harness facade and must not
|
||||
add another tracker or semantic event envelope.
|
||||
|
||||
### 9.4 Agent-mode notifications
|
||||
|
||||
The Session kernel and Chord structural sources do not maintain a semantic event
|
||||
journal. Coding-agent JSON/RPC compatibility uses a thin agent-mode adapter
|
||||
derived from each uncoalesced committed publication before per-watch queue
|
||||
compaction. It owns no tracker or persistence and emits notifications only after
|
||||
the commit that makes them true.
|
||||
|
||||
The adapter protocol covers run start/settlement, committed assistant progress,
|
||||
message entry settlement, tool intent/progress/result, input queue/outcome,
|
||||
retry/deferred/compaction state, configuration changes, and faults. One commit
|
||||
may produce an ordered batch. Progress notifications represent Pico's throttled
|
||||
durable partials, not every raw provider frame. The exact legacy `AgentEvent`
|
||||
wire format is not preserved.
|
||||
|
||||
Notifications have no hydration or replay contract. A consumer requiring a
|
||||
complete lifecycle subscribes before admitting the input; a late or reconnecting
|
||||
consumer hydrates structural state and history instead. Product adapters apply
|
||||
these rules:
|
||||
|
||||
- TUI hydrates and renders `ConversationView`, then applies structural updates;
|
||||
notifications may drive transient animation but are not its authority.
|
||||
- Print awaits its submitted `InputHandle` and prints that input's answer.
|
||||
- JSON/RPC expose correlated commands plus the ordered agent notification
|
||||
protocol, with transport backpressure and disconnect policy owned by that
|
||||
adapter.
|
||||
|
||||
This adapter is allowed even though a public Session-kernel semantic stream is a
|
||||
non-goal. It must not derive notifications from a lossy, reset-compacted watch
|
||||
when complete subscribed lifecycle delivery is promised.
|
||||
|
||||
## 10. Storage contract
|
||||
|
||||
|
|
@ -1640,9 +1809,9 @@ type SessionMetadata = {
|
|||
|
||||
type StorageWrite =
|
||||
| { readonly type: "session.metadata"; readonly value: SessionMetadata }
|
||||
| { readonly type: "conversation"; readonly value: Conversation }
|
||||
| { readonly type: "entry"; readonly value: Entry }
|
||||
| { readonly type: "task"; readonly value: Task<JsonValue, JsonValue, JsonValue> }
|
||||
| { readonly type: "conversation"; readonly value: ConversationRecord }
|
||||
| { readonly type: "entry"; readonly value: EntryRecord }
|
||||
| { readonly type: "task"; readonly value: TaskRecord<JsonValue, JsonValue, JsonValue> }
|
||||
| { readonly type: "input"; readonly value: Input }
|
||||
| {
|
||||
readonly type: "document.create";
|
||||
|
|
@ -1664,15 +1833,15 @@ interface Storage {
|
|||
mintId(): Id;
|
||||
|
||||
metadata(context: Context): Promise<SessionMetadata | undefined>;
|
||||
conversation(id: Id, context: Context): Promise<Conversation | undefined>;
|
||||
scanConversations(cursor: ConversationCursor | undefined, limit: number, context: Context): Promise<Page<Conversation, ConversationCursor>>;
|
||||
conversation(id: Id, context: Context): Promise<ConversationRecord | undefined>;
|
||||
scanConversations(cursor: ConversationCursor | undefined, limit: number, context: Context): Promise<Page<ConversationRecord, ConversationCursor>>;
|
||||
|
||||
entries(ids: readonly Id[], context: Context): Promise<ReadonlyMap<Id, Entry>>;
|
||||
scanEntries(query: EntryQuery, cursor: EntryCursor | undefined, limit: number, context: Context): Promise<Page<Entry, EntryCursor>>;
|
||||
entries(ids: readonly Id[], context: Context): Promise<ReadonlyMap<Id, EntryRecord>>;
|
||||
scanEntries(query: EntryQuery, cursor: EntryCursor | undefined, limit: number, context: Context): Promise<Page<EntryRecord, EntryCursor>>;
|
||||
entryCommit(id: Id, context: Context): Promise<Seq | undefined>;
|
||||
|
||||
task(id: Id, context: Context): Promise<Task<JsonValue, JsonValue, JsonValue> | undefined>;
|
||||
scanTasks(query: TaskQuery, cursor: TaskCursor | undefined, limit: number, context: Context): Promise<Page<Task<JsonValue, JsonValue, JsonValue>, TaskCursor>>;
|
||||
task(id: Id, context: Context): Promise<TaskRecord<JsonValue, JsonValue, JsonValue> | undefined>;
|
||||
scanTasks(query: TaskQuery, cursor: TaskCursor | undefined, limit: number, context: Context): Promise<Page<TaskRecord<JsonValue, JsonValue, JsonValue>, TaskCursor>>;
|
||||
|
||||
input(id: Id, context: Context): Promise<Input | undefined>;
|
||||
inputByRequest(conversationId: Id, requestId: string, context: Context): Promise<Input | undefined>;
|
||||
|
|
@ -1786,9 +1955,15 @@ These are contracts, not invitations to add defensive machinery:
|
|||
- **Schema stability:** definition IDs, mounted document IDs, and visible paths
|
||||
are persisted/public protocol. Value migration cannot rename an ID; an ID
|
||||
change requires explicit copy and retirement.
|
||||
- **Unstarted watches:** `watchDoc()` and `ConversationHandle.watch()` buffer
|
||||
every committed update until `start()` or `stop()`. Delaying both can consume
|
||||
unbounded memory.
|
||||
- **Watch activation:** `watch.value` is the immutable acquisition snapshot.
|
||||
Initialize the consumer from it before `start()`; queued operations are based
|
||||
on exactly that value.
|
||||
- **Coalesced watches:** slow or unstarted watches may replace an undelivered
|
||||
suffix with a complete reset, omitting intermediate committed states. A facet
|
||||
that must audit every transition must persist each fact as an immutable entry
|
||||
or in its own journal and scan that history explicitly.
|
||||
- **Watch self-join:** a listener may call `stop()`, but must not await its own
|
||||
`closed` promise or a Session close that joins that listener.
|
||||
- **Durable progress cadence:** clients see only committed progress. A crash may
|
||||
lose the current uncommitted throttle window.
|
||||
- **Large terminal results:** terminal task records remain queryable. Put large
|
||||
|
|
@ -1808,7 +1983,7 @@ Pico5 initially has no:
|
|||
|
||||
- whole-Session DOM;
|
||||
- visible-undurable publication;
|
||||
- public semantic event stream;
|
||||
- Session-kernel semantic event journal or independently maintained event state;
|
||||
- transaction membrane;
|
||||
- session-scoped rewindable documents;
|
||||
- automatic checkpoint heuristic;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue