docs(agent): replace obsolete Pico prototypes with Pico5 design

This commit is contained in:
Mario Zechner 2026-09-15 11:27:53 +02:00
parent 850f83794d
commit 56cd5989ea
16 changed files with 1593 additions and 4013 deletions

View file

@ -0,0 +1,555 @@
# Pico5 handoff and decision register
> Working coordination document. `pico-v5.md` is the compact design. This file
> records accepted choices, unresolved decisions, examples, and layer sign-offs so
> work survives conversation compaction. No unresolved item may be decided silently
> during implementation.
## Current status
- `packages/agent/src/harness/pico/` and its tests were removed.
- The untracked `pico4` source/tests were removed.
- `pico3` remains as a reference, not an implementation base.
- Storage benchmark artifacts are in `/tmp/pi-pico5-storage-bench/`.
- Astra approved starting the memory + SQLite storage/conformance slice after the
storage decisions below are signed off.
## Accepted direction
1. Entries are immutable and paged; the Session is not one materialized DOM.
2. Authors mutate ordinary JSON documents; Chord produces nested/list/string ops.
3. History policy is per document: `latest` or `rewindable`.
4. Built-ins use three conversation documents:
- `rewindable`: historical configuration;
- `sticky`: present state such as queues and retry policy;
- `live`: high-churn turn, generation, tool, and compaction presentation state.
5. Typed document definitions and dynamic families are required. Family instances
may be task-owned and retire automatically.
6. Task recovery state is a complete replacement, separate from presentation and
application documents.
7. There is no visible-but-undurable publication. Every Chord-visible update follows
a successful durable commit.
8. Interrupted assistant/tool partials are converted into durable aborted/interrupted
entries by their recovering task kind, then removed from `live`.
9. A built-in needs cross-task document observation, so `watchDoc` is initial scope.
10. UI status is state, not an event protocol. There is no public event channel
initially; clients observe document paths. A bounded `live` notice field may be
designed with the built-in view if concrete transient facts require it.
11. Third-party documents initially use their own Chord services. Automatic mounting
into the agent view is deferred.
12. SQLite is the long-running/cloud backend. JSONL remains local and readable.
13. Implementation proceeds layer by layer and stops for user sign-off after each.
## Decisions required before storage implementation
### S1. Document identity and families — accepted
A definition carries type, logical ID, scope, history, initializer, and optional
family ownership rules:
```ts
const live = defineDoc<LiveState>({
id: "pi.live",
scope: "conversation",
history: "latest",
initial: () => ({ tools: [] }),
});
const jobOutput = defineDocFamily<JobState, JobInput>({
id: "pi.job",
scope: "conversation",
history: "latest",
owner: "task",
initial: input => ({ command: input.command, stdout: "" }),
});
```
A persisted instance has a unique, never-reused storage identity plus definition ID,
scope owner, family instance ID, creation sequence, history policy, and retirement
sequence. Reusing one logical family key creates a new incarnation.
### S2. Historical document membership — accepted
A rewindable document retired today still exists in an as-of query from yesterday.
A latest document can be deleted after retirement.
Example:
```text
10 create rewindable review A
20 append entry E
30 retire A
40 fork at E
```
The fork enumerates A at commit 20 and copies its as-of value into a new child-owned
instance. Current enumeration at commit 40 omits A.
### S3. Storage ownership and copying — proposed recommendation
Simplest safe contract:
- Commit inputs are borrowed until `Storage.commit()` settles.
- A backend retaining anything after settlement detaches it first.
- Memory uses a small recursive `copyJson`; JSONL/SQLite detach through serialization.
- Every storage read returns a fresh owned JSON copy. Mutating it cannot corrupt the
backend or another caller. Public types remain readonly, but correctness does not
rely on TypeScript at runtime.
- Large strings are immutable and may share their string value during an in-memory
copy; object/array containers never share mutably.
Example failure this prevents:
```ts
const task = { phase: "running" };
await storage.commit([{ type: "task", task }]);
task.phase = "done"; // must not mutate MemoryStorage behind its back
```
Decision: approve this contract, or replace it with explicit ownership transfer and
state which caller may still use document ops for Chord publication.
### S4. Table reads and writes inside `Tx` — proposed recommendation
Adopt the other design's simple rule:
```text
table reads first -> first table write -> later table reads throw ReadAfterWrite
document reads/mutations always read their transaction draft
```
Example:
```ts
const inputs = await tx.inputs(ids); // read
const head = await tx.newestEntry(c); // read
const entry = tx.appendEntry(c, assistant); // first table write
tx.doc(live, c).message = undefined; // document draft is allowed
// await tx.entry(other) now throws ReadAfterWrite
```
Benefit: no shadow indexes for newly created entries/tasks/inputs. Cost: helpers must
load everything before their first table write. Validate all built-in settlement
traces before accepting.
Decision: adopt this rule, or support read-your-writes for tables and pay for overlays.
### S5. Escaped document draft — open
A callback can retain a document proxy:
```ts
let escaped: LiveState;
await session.commit(tx => { escaped = tx.doc(live, c); });
escaped.message = partial;
```
Options:
- **Detached stale draft:** mutation succeeds locally but has no effect. Simplest,
but silently hides a bug.
- **Revoked draft:** every later read/write throws. Clearest, but requires tracker
proxies to share an active guard or another membrane.
- **Debug assertion:** stale mutation is detected at the next transaction. Cheaper,
but delayed.
Recommendation: first determine whether Chord tracker proxies can cheaply share one
revocation guard. Do not let an escaped proxy mutate committed state or contaminate a
later commit.
### S6. Base policy — partly accepted
Every document may receive bases. Storage gets both incremental ops and a borrowed
candidate value, serializing only one:
```ts
persist(shouldBase(id, ops) ? [["r", value]] : ops);
```
- `latest`: a committed base permits reclaiming older records.
- `rewindable`: bases bound as-of replay; historical segments remain.
- Trigger on both encoded bytes and record count.
Open values:
- byte threshold;
- record-count threshold;
- whether document definitions may request `rebaseWhenInitial`.
`pi.live` benefits from `rebaseWhenInitial`: when a turn ends at `{ tools: [] }`, a
tiny base allows immediate reclamation of streamed output. The definition/document
manager can recognize its initial state; storage still owns physical reclamation.
### S7. Task rows — accepted
Task records are full replacements. SQLite uses one live row and `DELETE` on terminal.
JSONL uses one task sidecar and unlinks it only after terminal publication.
The runtime envelope owns kind/input/dependencies/memos/effect IDs. Kind-defined
recovery state cannot overwrite memos. A terminal transaction transfers any needed
outcome and retires every task-owned document family instance atomically.
This atomic retirement removes the need for an application-level orphan sweep:
- crash before terminal: task and owned docs remain live and recover;
- crash after terminal marker: logical retirement is known, and storage may clean
leftover files during open.
### S8. SQLite representation — accepted direction
Use one SQL transaction per Session commit:
```text
conversations / entries / inputs
document metadata
document base+delta records indexed by document and commit
a current live-task row
```
Do not translate Chord ops into SQLite JSON functions initially. The benchmark found
full JSONB replacement slower than TEXT and generic SQL patching slower with more WAL
writes. Current/as-of reads select a base and indexed tail; they do not scan unrelated
documents.
Still choose schema details, WAL checkpoint cadence, and `synchronous` default during
the SQLite layer.
### S9. JSONL physical layout — current lean: uniform sidecars
Simplest layout from the other design:
```text
main.jsonl tables + one publication marker per commit
doc-<storage-id>.jsonl one file per document instance
task-<storage-id>.jsonl one file per live task
```
Every sidecar record is prepared. Sidecars append first; one main marker is always
last. This deliberately permits records such as:
```json
{"seq":14,"refs":["task-9"]}
{"seq":15,"refs":["task-9"]}
{"seq":16,"refs":["pi.live-1"]}
```
The markers are cosmetic overhead but make every commit use one protocol. They can be
compact; fields such as `writes:[]` and repeated `maxId` are unnecessary if high-water
is represented elsewhere. Without `fsync`, the second append is unlikely to be the
performance bottleneck; the benchmark's no-fsync sidecar backend still handled about
9.5k–30k commits/s depending on workload while opening/closing each sidecar.
Alternative: standalone one-file commits avoid markers but introduce two record modes,
distributed commit discovery, and more recovery cases.
Decision: approve uniform marker publication or select the standalone optimization.
### S10. JSONL process-crash contract — accepted direction
No `fsync` by default. Awaited/full sidecar writes precede the awaited/full main
marker. For an ordinary process crash with the OS/filesystem alive:
- torn final lines are removed;
- sidecar records without a marker are uncommitted and removed;
- a complete marker publishes the transaction.
Power loss, host/kernel failure, or weak filesystem ordering may lose acknowledged
commits. Optional durable mode may fsync sidecars before the marker.
Any uncertain write error poisons the open backend until reopen. Reclamation occurs
only after its base/retirement commits and uses temp + rename; cached descriptors are
then reopened.
### S11. Missing confirmed sidecar — open
Example:
```text
sidecar record 20 was reported written
main marker 20 survives
sidecar record 20 is absent after machine failure
```
This should not arise from the stated ordinary process-crash contract. Options:
- **Fail open:** report corruption and preserve all bytes for diagnosis/repair.
Simplest and never invents state.
- **Roll back globally:** truncate every file to the latest complete global prefix.
Recovers availability but may discard acknowledged later commits and must interact
correctly with reclaimed prefixes and standalone records if those exist.
- **Best-effort skip:** invalid because it exposes a partial atomic commit.
Recommendation: fail open initially. Add a separate repair tool only after physical
fault tests establish a safe prefix algorithm.
### S12. Main JSONL growth — accepted limitation
Document/task sidecars reclaim high-churn data; `main.jsonl` still grows by entries,
lifecycle records, and commit markers. JSONL may scan it on open. It is the local
backend. Years-long products use SQLite. Do not add global JSONL compaction initially.
### S13. Storage query surface — needs exact signatures
The first storage contract must support:
- create/current/retired conversation records;
- paged immutable entries and entry -> commit mapping;
- current task lookup/scan;
- current and as-of document enumeration;
- current and as-of document value;
- family-instance enumeration by definition/scope/owner;
- one atomic mixed commit;
- close/reopen.
As-of results distinguish not-yet-created, retired after target, and unavailable
history. Concrete TypeScript signatures are the first layer's first sign-off.
## Decisions required before document/task implementation
### D1. Three built-in documents — accepted
```ts
rewindable // model/tool/config choices inherited by historical fork
sticky // inbox, queue modes, retry policy
live // current message/tools/generation/collapse and maybe notices
```
`live` is durable because recovery converts its partial output into transcript entries.
Its high churn is independently reclaimable when it returns to initial state.
### D2. Document adapter to Chord — accepted direction
A thin adapter such as `replicatedState.own(documentSource)` exposes committed value
and ops. It owns no second tracker. Each document and mounted view has a contiguous
publication revision separate from sparse global Session commit sequence.
Hydration captures value/revision and subscribes atomically. Listener errors after
persistence are isolated. Current Chord needs a small public source adapter; it does
not need harness knowledge.
### D3. Cross-task `watchDoc` — accepted, exact scope open
Required semantics:
```ts
const watch = await api.watchDoc(jobOutput, job.id, context);
consume(watch.value); // snapshot first
watch.start((ops, value) => consume(value));
```
- capture + subscription are one Session-line operation;
- callbacks run off the line and cannot re-enter a commit synchronously;
- watch is volatile and invocation-owned;
- recovery re-enters the handler, captures a fresh snapshot, and subscribes again;
- retirement appears as document absence;
- `stop()` is idempotent.
Open scope: same conversation only, or any conversation owned by the observing task's
subtree. Initial built-in use must provide the concrete required edge.
### D4. Reopen conversion — accepted
Generation recovery reads `live.message`, appends an assistant entry with
`stopReason: "aborted"`, clears the preview, and retries/fails atomically. Tool
recovery turns persisted output into its interrupted result before clearing the slot.
The entry uses the normal model/render shape while model projection excludes aborted
content.
Conversion runs before retiring any task-owned family document it needs.
### D5. Document versioning/migration — open
Document paths and persisted values are public/durable schema. A plugin upgrade may
change `{ comments: [...] }` to `{ threads: [...] }`.
Options:
- definition has an integer version and migration from complete current value;
- task/document kind owns migration during open;
- reject mismatched versions initially.
Recommendation: persist definition version now even if initial implementation only
rejects mismatches. Retrofitting identity/version after stored products ship is harder.
## Decisions required before view implementation
### V1. Public view shape — open
Fixed top-level fields:
```ts
{ conversation, entries, rewindable, sticky, live }
```
Typed and direct, but every added built-in document changes the protocol.
Document map:
```ts
{
conversation,
entries,
docs: {
"pi.rewindable": RewindableState,
"pi.sticky": StickyState,
"pi.live": LiveState,
},
}
```
Makes document identity/lifetime explicit and can add selected documents later, but
paths and IDs become public protocol. It can remain typed for the three built-ins.
Recommendation: decide before writing any UI adapter. In either shape, mount only the
three built-ins initially and treat every visible path as versioned public API.
### V2. Public events — accepted: none initially
A compaction-start event cannot describe current state to a late joiner. Instead:
```ts
live.collapse = { through, stage: "summarizing" }; // running
live.collapse = undefined; // finished
```
Clients react to state ops by path and always hydrate the current truth. Hooks remain
pre-decision interception. Internal commit subscribers may exist mechanically, but
there is no separate public semantic event protocol initially.
Whether failed attempts need a bounded `live.notices` list is deferred until the UI
is implemented. It remains state with explicit retention, not fire-and-forget events.
### V3. Atomic publication — accepted
The built-in mount consumes the complete Session commit, prefixes every changed
built-in document path, combines entry changes, and publishes one Chord batch. It has
no tracker or semantic projection. Independent third-party Chord services may publish
separately.
### V4. Active transcript bound — open
A Slack conversation cannot hydrate every entry ever written. Options include:
- durable `head` entry: active transcript is the range beginning at its target;
- fixed recent entry/count/byte window;
- product-selected window with a stable cursor.
Example requirement:
```text
10 years / 500,000 entries
late UI attach -> bounded snapshot
older history -> explicit paged query
model context -> independently derived bounded projection
```
Choose before view implementation, not before storage. Entry paging and indexes are
required now regardless.
### V5. Multi-entry fork boundary — open
One commit may append entries E and F and leave one final rewindable state. Fork at E:
- **entry prefix:** child transcript includes E but not F, while document state is the
final state of the containing commit;
- **commit boundary:** selecting E means inherit all entries from its indivisible
commit, including F;
- **restriction:** only designated forkable entries/commits may be targets.
Choose before fork/view API. Storage records entry -> commit either way.
### V6. Sticky fork initialization — current lean: defaults
A historical fork inherits rewindable documents as-of its target and no live tasks.
Copying current sticky/live state can copy queued inputs or running previews from the
future. Simplest initial rule:
```text
rewindable -> historical copy
sticky -> fresh defaults
live -> fresh defaults
```
Explicit sticky-copy options remain deferred.
### V7. Third-party mounting — accepted defer
Third-party documents initially expose application-specific Chord services. Do not add
`inView`, privacy flags, dynamic mount registries, or client-selected arbitrary docs.
A document is not a security boundary; services deliberately choose what crosses a
process boundary.
## Layer plan and sign-off gates
### Layer 0 — cleanup
- Remove `pico` source/tests: done.
- Remove `pico4` source/tests: done.
- Keep `pico3`: done.
- Run repository check after deletions and document additions: pending.
### Layer 1 — logical storage + memory reference
Before code: sign off S3 and S6, retain accepted S1/S2/S7, and approve the exact S13 interfaces.
Acceptance:
- atomic mixed commits;
- document creation, bases, current/as-of membership and values;
- family instances and task ownership;
- full task replacement/retirement;
- entry paging and entry -> commit;
- ownership isolation and failure rollback.
Stop for user review.
### Layer 2 — SQLite
Before code: approve concrete schema and durability defaults.
Acceptance: Layer 1 conformance plus reopen, indexed query plans, latest reclamation,
retired rewindable history, WAL checkpoint behavior, and realistic storage benchmark.
Stop for user review.
### Layer 3 — JSONL
Before code: decide S9 and S11.
Acceptance: semantic conformance plus physical fault injection at every sidecar,
marker, rewrite, rename, retirement, and reopen boundary.
Stop for user review.
### Layer 4 — document manager + Chord source
Before code: decide S4, S5, D3 scope, and D5.
Acceptance: ordinary mutations, no duplicate tracker, failed-commit rollback,
contiguous publication, hydration fencing, listener isolation, eviction, and
`watchDoc` snapshot/update/retirement behavior.
Stop for user review.
### Layer 5 — tasks
Acceptance: phase-narrowed full replacements, effect sandwich, runtime-owned memos,
abortable ownership tree, atomic family retirement/outcome handoff, generation/tool
reopen conversion, and no visible-undurable state.
Stop for user review.
### Layer 6 — built-in documents and conversation view
Before code: decide V1, V4, V5, and V6.
Acceptance: direct built-in writes, fixed structural mount, one publication per commit,
bounded late hydration, crash-visible progress, and no public event dependency.
Stop for user review.
### Layer 7 — remaining harness
Inputs, hooks, tools, system sections, dynamic tools, deferred providers, and product
wiring. Preserve accepted APIs from earlier design work rather than redesigning them
inside the state implementation.

View file

@ -0,0 +1,742 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1, viewport-fit=cover">
<title>Pico5 — durable documents</title>
<style>
:root {
color-scheme: dark;
--bg: #08090d;
--panel: rgba(255,255,255,.045);
--panel-strong: rgba(255,255,255,.075);
--text: rgba(238,241,245,.93);
--muted: rgba(198,207,218,.58);
--faint: rgba(198,207,218,.34);
--line: rgba(255,255,255,.10);
--blue: #9db9ff;
--green: #b7dca2;
--gold: #e4c88f;
--purple: #d2baff;
--red: #efaaa4;
--mono: "SFMono-Regular", ui-monospace, "Cascadia Code", Consolas, monospace;
--pad-x: clamp(24px, 6vw, 112px);
--pad-y: clamp(28px, 5vh, 72px);
}
* { box-sizing: border-box; }
html, body { width: 100%; height: 100%; margin: 0; overflow: hidden; }
body {
background:
radial-gradient(ellipse at 15% 15%, rgba(70,95,160,.12), transparent 42%),
radial-gradient(ellipse at 85% 85%, rgba(105,70,145,.09), transparent 45%),
var(--bg);
color: var(--text);
font-family: var(--mono);
overscroll-behavior: none;
}
.slide {
display: none;
position: relative;
width: 100vw;
height: 100vh;
padding: var(--pad-y) var(--pad-x) clamp(54px, 7vh, 88px);
}
.slide.active { display: flex; flex-direction: column; }
.frame {
flex: 1;
min-height: 0;
width: 100%;
max-width: 1500px;
margin: auto;
display: flex;
flex-direction: column;
justify-content: center;
}
.wide { max-width: 1720px; }
.center { align-items: center; text-align: center; }
h1, h2, h3, p, ul, pre { margin: 0; }
h1 {
max-width: 1400px;
font: 500 clamp(46px, 7vw, 108px)/.98 var(--mono);
letter-spacing: -.045em;
text-wrap: balance;
}
h2 {
font: 500 clamp(30px, 5vw, 72px)/1.04 var(--mono);
letter-spacing: -.035em;
text-wrap: balance;
}
h3 {
margin-bottom: clamp(8px, 1.2vh, 18px);
color: var(--muted);
font: 500 clamp(12px, 1.35vw, 20px)/1.3 var(--mono);
letter-spacing: .09em;
text-transform: uppercase;
}
p, li {
font: 400 clamp(18px, 2.35vw, 38px)/1.34 var(--mono);
letter-spacing: -.015em;
}
ul { padding-left: clamp(22px, 3vw, 44px); margin-top: clamp(12px, 2vh, 28px); }
li { margin: clamp(5px, .8vh, 13px) 0; }
.compact li { font-size: clamp(15px, 1.85vw, 30px); line-height: 1.32; }
.tight li { font-size: clamp(14px, 1.55vw, 25px); line-height: 1.28; }
.subtitle {
margin-top: clamp(14px, 2.4vh, 34px);
color: var(--muted);
font: 400 clamp(15px, 1.8vw, 28px)/1.45 var(--mono);
letter-spacing: .05em;
text-transform: uppercase;
}
.big {
max-width: 1450px;
font: 450 clamp(28px, 4.4vw, 68px)/1.12 var(--mono);
letter-spacing: -.035em;
text-wrap: balance;
}
.small { font-size: clamp(14px, 1.55vw, 24px); color: var(--muted); }
.accent { color: var(--blue); }
.good { color: var(--green); }
.warn { color: var(--gold); }
.bad { color: var(--red); }
.purple { color: var(--purple); }
pre {
margin-top: clamp(12px, 2vh, 28px);
padding: clamp(14px, 2vw, 28px);
overflow: hidden;
border: 1px solid var(--line);
border-radius: 10px;
background: var(--panel);
white-space: pre-wrap;
}
pre code { font: 400 clamp(13px, 1.45vw, 23px)/1.48 var(--mono); }
.code-small code { font-size: clamp(11px, 1.15vw, 18px); line-height: 1.43; }
.code-tight code { font-size: clamp(10px, 1vw, 16px); line-height: 1.38; }
code.inline {
padding: .12em .34em;
border-radius: 4px;
background: var(--panel-strong);
font: .86em var(--mono);
}
.kw { color: var(--blue); }
.str { color: var(--green); }
.num { color: var(--gold); }
.type { color: var(--purple); }
.comment { color: var(--faint); font-style: italic; }
.grid2, .grid3 {
display: grid;
gap: clamp(14px, 2.5vw, 38px);
margin-top: clamp(14px, 2.4vh, 34px);
align-items: stretch;
}
.grid2 { grid-template-columns: repeat(2, minmax(0, 1fr)); }
.grid3 { grid-template-columns: repeat(3, minmax(0, 1fr)); }
.panel {
min-width: 0;
padding: clamp(16px, 2.3vw, 34px);
border: 1px solid var(--line);
border-radius: 12px;
background: var(--panel);
}
.panel p, .panel li { font-size: clamp(14px, 1.7vw, 28px); }
.panel ul { margin-top: 0; padding-left: clamp(18px, 2vw, 30px); }
.flow {
display: flex;
align-items: center;
justify-content: center;
gap: clamp(8px, 1.7vw, 28px);
margin-top: clamp(18px, 4vh, 54px);
}
.node {
flex: 1;
padding: clamp(14px, 2.2vw, 30px);
border: 1px solid var(--line);
border-radius: 10px;
background: var(--panel);
text-align: center;
font-size: clamp(13px, 1.55vw, 25px);
line-height: 1.3;
}
.arrow { color: var(--faint); font-size: clamp(22px, 3vw, 46px); }
.table {
width: 100%;
margin-top: clamp(14px, 2.3vh, 30px);
border-collapse: collapse;
font-size: clamp(13px, 1.45vw, 23px);
}
.table th, .table td {
padding: clamp(8px, 1.2vw, 18px);
border-bottom: 1px solid var(--line);
text-align: left;
vertical-align: top;
}
.table th { color: var(--muted); font-weight: 500; text-transform: uppercase; letter-spacing: .06em; }
.step.hidden { visibility: hidden; }
.footer {
position: absolute;
left: clamp(16px, 2vw, 32px);
right: clamp(16px, 2vw, 32px);
bottom: clamp(12px, 1.6vh, 24px);
display: flex;
justify-content: space-between;
color: var(--faint);
font: 400 clamp(10px, 1vw, 15px)/1 var(--mono);
letter-spacing: .08em;
text-transform: uppercase;
}
.progress {
position: fixed;
z-index: 10;
left: 0;
bottom: 0;
height: 2px;
width: 0;
background: var(--blue);
opacity: .65;
transition: width .15s ease;
}
.hint {
position: fixed;
right: 18px;
top: 14px;
z-index: 10;
color: var(--faint);
font-size: 11px;
letter-spacing: .05em;
}
@media (max-width: 800px) {
.grid2, .grid3 { grid-template-columns: 1fr; }
.flow { flex-wrap: wrap; }
.arrow { transform: rotate(90deg); }
.slide { overflow: auto; }
}
@media print {
html, body { overflow: visible; background: #fff; color: #111; }
.slide, .slide.active { display: flex; break-after: page; }
.progress, .hint { display: none; }
.step.hidden { visibility: visible; }
}
</style>
</head>
<body>
<div class="hint">← → · space · f</div>
<div class="progress" id="progress"></div>
<section class="slide" data-slide="1">
<div class="frame center">
<h1>Pico5</h1>
<p class="big" style="margin-top:28px">Durable documents without a whole-session DOM</p>
<p class="subtitle step">Usage, state → view, and why Lane stops short</p>
</div>
</section>
<section class="slide" data-slide="2">
<div class="frame wide">
<h2>The target is not one coding session</h2>
<div class="grid2">
<div class="panel step"><h3>Products</h3><ul class="compact"><li>coding agent</li><li>personal assistant</li><li>cloud agents in Durable Objects</li><li>Slack bots across years of channels and threads</li></ul></div>
<div class="panel step"><h3>Required invariants</h3><ul class="compact"><li>survive process death mid-effect</li><li>fork historical conversation state</li><li>attach many presentations</li><li>let facets persist application state without inventing storage</li></ul></div>
</div>
</div>
</section>
<section class="slide" data-slide="3">
<div class="frame wide">
<h2>Lane solved a real problem</h2>
<div class="grid3">
<div class="panel step"><h3>Execution</h3><p>Complete operation-state replacements make recovery a switch, not inference.</p></div>
<div class="panel step"><h3>Effects</h3><p>Durable intent → uncertain effect → durable outcome.</p></div>
<div class="panel step"><h3>Transactions</h3><p>Entries, values, and lists commit atomically in memory, JSONL, and SQLite.</p></div>
</div>
<p class="subtitle step">Keep those properties. Replace the application-state substrate.</p>
</div>
</section>
<section class="slide" data-slide="4">
<div class="frame wide">
<h2>Lane exposes values and lists</h2>
<pre><code><span class="kw">const</span> review = value&lt;<span class="type">ReviewState</span>&gt;(<span class="str">"app.review"</span>, id);
<span class="kw">const</span> findings = list&lt;<span class="type">Finding</span>&gt;(<span class="str">"app.findings"</span>, id);
<span class="kw">await</span> session.setValue(review, completeNextReview, context);
<span class="kw">await</span> session.appendList(findings, finding, context);
<span class="kw">await</span> session.deleteList(findings, context);</code></pre>
<ul class="compact steps">
<li class="step">A scalar is replaced whole.</li>
<li class="step">A list appends or is deleted whole.</li>
<li class="step">Nested changes, splices, checkpoints, and reconstruction belong to the caller.</li>
</ul>
</div>
</section>
<section class="slide" data-slide="5">
<div class="frame wide">
<h2>Canvas on Lane: choose your own machinery</h2>
<div class="grid2">
<div class="panel step">
<h3>Store the whole canvas</h3>
<pre class="code-small"><code><span class="kw">const</span> CanvasValue = value&lt;<span class="type">CanvasState</span>&gt;(<span class="str">"app.canvas"</span>);
<span class="kw">const</span> stored = <span class="kw">await</span> session.getValue(CanvasValue, context);
<span class="kw">const</span> next = stored?.value ?? { strokes: [] };
next.strokes.push(stroke);
<span class="kw">await</span> session.setValue(CanvasValue, next, context);</code></pre>
<p class="small">Simple API; complete serialization and replacement for every stroke.</p>
</div>
<div class="panel step">
<h3>Store stroke deltas</h3>
<pre class="code-small"><code><span class="kw">const</span> StrokeLog = list&lt;<span class="type">StrokeOp</span>&gt;(<span class="str">"app.canvas.strokes"</span>);
<span class="kw">await</span> session.appendList(StrokeLog, toStrokeOp(stroke), context);
<span class="comment">// define reducer and operation vocabulary
// define checkpoints
// coordinate append + compact
// reconstruct after restart</span></code></pre>
<p class="small">Efficient writes; every facet authors a private state engine.</p>
</div>
</div>
</div>
</section>
<section class="slide" data-slide="6">
<div class="frame wide">
<h2>Diff review on Lane: durability leaks upward</h2>
<pre class="code-small"><code><span class="kw">const</span> ReviewRecord = (id: <span class="type">string</span>) =&gt; value&lt;<span class="type">Review</span>&gt;(<span class="str">"app.review"</span>, id);
<span class="kw">const</span> activity = env.replicatedState({ review: initialReview });
<span class="kw">return</span> regionFor(reviewId).run(signal, <span class="kw">async</span> () =&gt; {
<span class="kw">const</span> stored = <span class="kw">await</span> session.getValue(ReviewRecord(reviewId), context);
<span class="kw">const</span> next = validateAndReplace(stored!.value, action);
<span class="kw">await</span> session.setValue(ReviewRecord(reviewId), next, context);
activity.state.review = project(next);
activity.publish(context);
});</code></pre>
<ul class="compact steps">
<li class="step">The facet owns locking and read-modify-write correctness.</li>
<li class="step">It reconstructs live state after worker restart.</li>
<li class="step">Persistence and replication are two separately coordinated writes.</li>
</ul>
</div>
</section>
<section class="slide" data-slide="7">
<div class="frame wide">
<h2>Lane forks transcript, not application state</h2>
<pre><code>branch A: e1 ─ e2 ─ e3 ─ e4
└──── fork at e3 → branch B
value("plan-mode") current replacement only
list("canvas-strokes") surviving elements only
question: what was either value at e3? <span class="bad">not representable</span></code></pre>
<p class="subtitle step">The JSONL bytes may contain old writes. The storage contract does not expose them as history.</p>
</div>
</section>
<section class="slide" data-slide="8">
<div class="frame wide">
<h2>The gap is narrow but foundational</h2>
<div class="grid2">
<div class="panel step"><h3>Lane gives facets</h3><ul class="compact"><li>typed scalar replacement</li><li>append/delete list</li><li>atomic storage calls</li><li>separate volatile ReplicatedState</li></ul></div>
<div class="panel step"><h3>Products need</h3><ul class="compact"><li>ordinary nested mutation</li><li>automatic delta + base policy</li><li>historical document reads</li><li>commit and replication as one path</li></ul></div>
</div>
<p class="subtitle step">Lane remains a good execution design. It is not enough as the state substrate for products and facets.</p>
</div>
</section>
<section class="slide" data-slide="9">
<div class="frame center">
<p class="big">A Session is a store of <span class="accent">entries</span>, <span class="good">documents</span>, and <span class="purple">live task bookmarks</span>.</p>
<p class="subtitle step">Not one giant DOM. Not a bag of extension-owned repositories.</p>
</div>
</section>
<section class="slide" data-slide="10">
<div class="frame wide">
<h2>Four durable primitives</h2>
<table class="table">
<thead><tr><th>Primitive</th><th>Use</th><th>Storage</th></tr></thead>
<tbody>
<tr class="step"><td class="accent">Entry</td><td>Transcript and unbounded history</td><td>Immutable, paged</td></tr>
<tr class="step"><td class="good">Document</td><td>Current application and presentation state</td><td>Chord ops + bases</td></tr>
<tr class="step"><td class="purple">Task state</td><td>Private recovery phase/bookmark</td><td>Complete replacement, retireable</td></tr>
<tr class="step"><td class="warn">Memo</td><td>Hook/tool idempotency receipt</td><td>Small, first writer wins</td></tr>
</tbody>
</table>
</div>
</section>
<section class="slide" data-slide="11">
<div class="frame wide">
<h2>Create a durable document</h2>
<pre><code><span class="comment">// Module-level typed declaration token.</span>
<span class="kw">const</span> CanvasDoc = defineDoc&lt;<span class="type">CanvasState</span>&gt;({
id: <span class="str">"@app/canvas"</span>,
scope: <span class="str">"session"</span>,
history: <span class="str">"latest"</span>,
initial: () =&gt; ({ strokes: [] }),
});
<span class="comment">// Tx resolves CanvasDoc to this Session's mutable draft.</span>
<span class="kw">await</span> session.commit(tx =&gt; {
tx.doc(CanvasDoc).strokes.push(stroke);
}, context);</code></pre>
<p class="subtitle step"><span class="code inline">CanvasDoc</span> is the token. <span class="code inline">tx.doc(CanvasDoc)</span> is this transaction's draft.</p>
</div>
</section>
<section class="slide" data-slide="12">
<div class="frame wide">
<h2>Expose the same document through Chord</h2>
<pre class="code-small"><code><span class="kw">interface</span> <span class="type">CanvasService</span> {
<span class="kw">readonly</span> state: ReplicatedState&lt;<span class="type">CanvasState</span>&gt;;
addStroke(stroke: <span class="type">Stroke</span>, context: <span class="type">Context</span>): Promise&lt;<span class="kw">void</span>&gt;;
}
<span class="kw">const</span> Canvas = defineService&lt;<span class="type">CanvasService</span>&gt;(<span class="str">"@app/canvas"</span>);
<span class="comment">// Committed source for the CanvasDoc token declared previously.</span>
<span class="kw">const</span> canvasDocument = session.document(CanvasDoc);
<span class="kw">const</span> canvasState = replicatedState.own(canvasDocument);
env.provide(Canvas, {
state: canvasState,
addStroke: (stroke, context) =&gt; session.commit(tx =&gt; {
tx.doc(CanvasDoc).strokes.push(stroke);
}, context),
});</code></pre>
<ul class="compact steps"><li class="step">No second tracker.</li><li class="step">No facet-authored delta protocol.</li><li class="step">No restart reconstruction code.</li></ul>
</div>
</section>
<section class="slide" data-slide="13">
<div class="frame wide">
<h2>One mutation path</h2>
<div class="flow">
<div class="node step">mutate<br><span class="good">transaction draft</span></div><div class="arrow step">→</div>
<div class="node step">Chord derives<br><span class="accent">ops</span></div><div class="arrow step">→</div>
<div class="node step">storage commits<br><span class="warn">all or nothing</span></div><div class="arrow step">→</div>
<div class="node step">publish committed ops<br><span class="purple">to replicas</span></div>
</div>
<p class="subtitle step">If a client can see it, recovery can reconstruct it.</p>
</div>
</section>
<section class="slide" data-slide="14">
<div class="frame wide">
<h2>No volatile task publication</h2>
<pre><code><span class="kw">for await</span> (<span class="kw">const</span> event <span class="kw">of</span> model.stream(request)) {
applyEvent(partial, event);
<span class="kw">if</span> (flushDue()) {
<span class="kw">await</span> rt.commit(tx =&gt; {
<span class="comment">// LiveDoc is the built-in defineDoc&lt;LiveState&gt; token.</span>
tx.doc(LiveDoc, task.conversationId).message = partial;
}, context);
}
}</code></pre>
<ul class="compact steps"><li class="step">The UI sees only committed throttle points.</li><li class="step">A crash loses at most the uncommitted window.</li><li class="step">There is one tracker baseline: the last durable commit.</li></ul>
</div>
</section>
<section class="slide" data-slide="15">
<div class="frame wide">
<h2>Three built-in conversation documents</h2>
<pre><code><span class="kw">interface</span> <span class="type">ConversationView</span> {
conversation: <span class="type">ConversationInfo</span>;
entries: <span class="type">Entry</span>[]; <span class="comment">// bounded active transcript</span>
rewindable: <span class="type">RewindableState</span>; <span class="comment">// historical fork state</span>
sticky: <span class="type">StickyState</span>; <span class="comment">// inbox and present policy</span>
live: <span class="type">LiveState</span>; <span class="comment">// turn, streaming, tools, collapse</span>
}</code></pre>
<pre class="code-small"><code>rewindable ["s", ["model"], model]
view ["s", ["rewindable", "model"], model]
live ["a", ["message", ...], delta]
view ["a", ["live", "message", ...], delta]</code></pre>
<p class="subtitle step">A fixed structural mount. No config merge, task description pass, or plugin projection.</p>
</div>
</section>
<section class="slide" data-slide="16">
<div class="frame wide">
<h2>Task state stays private and small</h2>
<pre><code><span class="kw">const</span> attempt = task.state.attempt + <span class="num">1</span>;
<span class="comment">// Provisional spelling: transition replaces the complete kind state.</span>
<span class="kw">await</span> rt.transition(
{ phase: <span class="str">"requesting"</span>, attempt, requestId },
tx =&gt; {
tx.doc(LiveDoc, task.conversationId).generation = {
stage: <span class="str">"requesting"</span>, attempt,
};
},
context,
);</code></pre>
<p class="subtitle step">Recovery state and presentation state are updated atomically, but are not the same object.</p>
</div>
</section>
<section class="slide" data-slide="17">
<div class="frame wide">
<h2>Crash-visible tool progress</h2>
<pre><code>TX 40 task state = { phase: "running" }
live.tools[2] = { status: "running", output: "" }
TX 41 live.tools[2].output = "... first 12 KiB ..."
TX 42 live.tools[2].output = "... latest 50 KiB window ..."
<span class="bad">PROCESS DIES</span>
restart task says running
view says exactly how far output reached
recovery writes interrupted result or safely resumes</code></pre>
<p class="subtitle step">The partial is useful UI state even when it never enters model context.</p>
</div>
</section>
<section class="slide" data-slide="18">
<div class="frame wide">
<h2>Rewind belongs to documents</h2>
<pre><code>commit 100 entry E + rewindable plan.enabled = true
commit 101 live streaming progress
commit 102 rewindable plan.enabled = false
fork at E:
rewindable → state at commit 100
sticky → fresh defaults
live → fresh defaults
tasks → not copied
child → new document IDs + initial bases</code></pre>
<ul class="compact steps"><li class="step">Bases bound replay distance.</li><li class="step">Rewindable history remains while ancient forks are supported.</li><li class="step">The parent and its live work remain untouched.</li></ul>
</div>
</section>
<section class="slide" data-slide="19">
<div class="frame wide">
<h2>One atomic Session commit</h2>
<pre class="code-small"><code><span class="comment">// Provisional spelling; all writes below are one transaction.</span>
<span class="kw">await</span> rt.transition({ phase: <span class="str">"tools"</span>, calls }, tx =&gt; {
tx.appendEntry(task.conversationId, finalAssistant);
<span class="kw">const</span> live = tx.doc(LiveDoc, task.conversationId);
<span class="kw">delete</span> live.message;
<span class="kw">delete</span> live.generation;
live.tools = pendingToolSlots(calls);
<span class="kw">for</span> (<span class="kw">const</span> spec <span class="kw">of</span> toolTaskSpecs(calls)) tx.createTask(spec);
}, context);</code></pre>
<p class="subtitle step">Storage publishes all or none. ConversationView emits one combined operation batch.</p>
</div>
</section>
<section class="slide" data-slide="20">
<div class="frame wide">
<h2>Storage chooses delta or base</h2>
<pre><code><span class="kw">interface</span> <span class="type">DocumentChange</span> {
id: <span class="type">DocumentId</span>;
ops: <span class="kw">readonly</span> <span class="type">Op</span>[];
value: <span class="type">JsonValue</span>; <span class="comment">// borrowed; not serialized normally</span>
}
<span class="kw">if</span> (bytesSinceBase + encodedSize(ops) &lt; threshold)
persist(ops);
<span class="kw">else</span>
persist([[<span class="str">"r"</span>, value]]);</code></pre>
<ul class="compact steps"><li class="step">Latest: committed base allows reclaiming the old tail.</li><li class="step">Rewindable: bases accelerate reads; old addressable segments remain.</li><li class="step">Chord clients still receive incremental ops.</li></ul>
</div>
</section>
<section class="slide" data-slide="21">
<div class="frame wide">
<h2>SQLite: the long-running backend</h2>
<div class="grid2">
<div class="panel step"><h3>Documents</h3><ul class="compact"><li>metadata row</li><li>indexed base/delta records</li><li>range query for current or as-of</li><li>transactional latest reclamation</li></ul></div>
<div class="panel step"><h3>Tasks + entries</h3><ul class="compact"><li>one complete row per live task</li><li>DELETE on retirement</li><li>immutable paged entry rows</li><li>all writes in one SQL transaction</li></ul></div>
</div>
<p class="subtitle step">No sidecar publication protocol. No scan of unrelated conversations or documents.</p>
</div>
</section>
<section class="slide" data-slide="22">
<div class="frame wide">
<h2>JSONL: one uniform publication rule</h2>
<pre><code>main.jsonl tables + one publication marker per commit
task-&lt;id&gt;.jsonl retireable task state
doc-&lt;id&gt;.jsonl one file per document instance
commit append every sidecar first
append one compact main marker last
retire task publish terminal marker → unlink task sidecar
rebase latest document publish base marker → atomic sidecar rewrite</code></pre>
<p class="subtitle step">Sidecars are a JSONL backend decision. Harness and facet code never see them.</p>
</div>
</section>
<section class="slide" data-slide="23">
<div class="frame wide">
<h2>The benchmark changed one decision</h2>
<div class="grid2">
<div class="panel step"><h3>Measured</h3><ul class="tight"><li>72 storage runs + 48 SQLite materialization runs validated</li><li>task sidecars: 1.13 MB → 59 KB after retirement</li><li>SQLite task deletion + checkpoint: 4.45 MB → 0.30 MB</li></ul></div>
<div class="panel step"><h3>Rejected optimization</h3><ul class="tight"><li>JSONB replacement 10–18% slower than TEXT</li><li>SQL JSON patching used 2–4× CPU</li><li>large canvas DB+WAL: ~80 MB → ~155 MB</li></ul></div>
</div>
<p class="subtitle step">Persist Chord ops and periodic bases. Do not build a second op interpreter in SQL.</p>
</div>
</section>
<section class="slide" data-slide="24">
<div class="frame wide">
<h2>Years-long sessions stay bounded in memory</h2>
<pre><code>Session
├─ entries paged rows; unbounded on disk
├─ conversation A
│ ├─ active entry window bounded
│ ├─ pi.rewindable loaded while active
│ ├─ pi.sticky loaded while active
│ └─ pi.live loaded only while needed
├─ conversation B evictable while idle
├─ @app/canvas loaded while service is active
└─ task 91 one live row; deleted at terminal</code></pre>
<p class="subtitle step">There is no whole-session materialization, snapshot, or subscriber payload.</p>
</div>
</section>
<section class="slide" data-slide="25">
<div class="frame wide">
<h2>What Pico5 does not add yet</h2>
<div class="grid2">
<div class="panel step"><ul class="compact"><li>generic plugin mounting into ConversationView</li><li>per-subtree history policies</li><li>visible-but-undurable updates</li><li>CRDT or offline writes</li></ul></div>
<div class="panel step"><ul class="compact"><li>SQL JSON patch translation</li><li>one whole-Session DOM</li><li>history pruning semantics</li><li>public semantic event machinery</li><li>pico3 config/plugin projections</li></ul></div>
</div>
<p class="subtitle step">The first release mounts only the three built-in conversation documents.</p>
</div>
</section>
<section class="slide" data-slide="26">
<div class="frame wide">
<h2>Build the owl one piece at a time</h2>
<div class="flow">
<div class="node step"><span class="num">1</span><br>memory + SQLite<br><span class="small">storage semantics</span></div><div class="arrow step">→</div>
<div class="node step"><span class="num">2</span><br>JSONL<br><span class="small">crash matrix</span></div><div class="arrow step">→</div>
<div class="node step"><span class="num">3</span><br>documents + Chord<br><span class="small">narrow proof</span></div><div class="arrow step">→</div>
<div class="node step"><span class="num">4</span><br>tasks<br><span class="small">effect sandwich</span></div><div class="arrow step">→</div>
<div class="node step"><span class="num">5</span><br>view + rest<br><span class="small">policy last</span></div>
</div>
<p class="subtitle step">Do not implement the scheduler while storage semantics are still moving.</p>
</div>
</section>
<section class="slide" data-slide="27">
<div class="frame wide">
<h2>Families: one typed document per task</h2>
<pre class="code-small"><code><span class="kw">const</span> JobOutputDoc = defineDocFamily&lt;<span class="type">JobOutput</span>, <span class="type">JobInput</span>&gt;({
id: <span class="str">"pi.job-output"</span>,
scope: <span class="str">"conversation"</span>,
history: <span class="str">"latest"</span>,
owner: <span class="str">"task"</span>,
initial: input =&gt; ({ command: input.command, stdout: <span class="str">""</span> }),
});
<span class="comment">// task.id selects the family instance owned by this task</span>
<span class="kw">await</span> rt.commit(tx =&gt; {
tx.doc(JobOutputDoc, task.id).stdout += chunk;
}, context);</code></pre>
<p class="subtitle step">Terminal settlement retires the task and every task-owned family instance atomically.</p>
</div>
</section>
<section class="slide" data-slide="28">
<div class="frame wide">
<h2>Watch another task's durable output</h2>
<pre class="code-small"><code><span class="kw">const</span> job = <span class="kw">await</span> api.task(JobKind, input, { background: <span class="kw">true</span> }, context);
<span class="kw">const</span> watch = <span class="kw">await</span> api.watchDoc(JobOutputDoc, job.id, context);
consume(watch.value); <span class="comment">// atomic initial snapshot</span>
watch.start((_ops, value) =&gt; consume(value));
<span class="kw">try</span> {
<span class="kw">return await</span> api.waitForTask(job, context);
} <span class="kw">finally</span> {
watch.stop(); <span class="comment">// idempotent, invocation-owned</span>
}</code></pre>
<ul class="compact steps"><li class="step">Callbacks run off the Session mutation line.</li><li class="step">Recovery re-enters and captures a fresh snapshot; no durable subscription cursor.</li><li class="step">Retirement appears as document absence.</li></ul>
</div>
</section>
<section class="slide" data-slide="29">
<div class="frame center">
<p class="big">Lane made execution durable.</p>
<p class="big step" style="margin-top:30px">Pico5 makes <span class="good">application state</span> durable by default.</p>
<p class="subtitle step">Ordinary objects → one commit → storage + view</p>
</div>
</section>
<script>
(() => {
const slides = Array.from(document.querySelectorAll('.slide'));
const progress = document.getElementById('progress');
let index = 0;
let step = 0;
slides.forEach((slide, i) => {
const footer = document.createElement('div');
footer.className = 'footer';
footer.innerHTML = `<span>pico5 · durable documents</span><span>${i + 1} / ${slides.length}</span>`;
slide.appendChild(footer);
});
const steps = slide => Array.from(slide.querySelectorAll('.step'));
const applySteps = slide => steps(slide).forEach((el, i) => el.classList.toggle('hidden', i >= step));
function show(nextIndex, nextStep = 0) {
index = Math.max(0, Math.min(slides.length - 1, nextIndex));
step = Math.max(0, Math.min(steps(slides[index]).length, nextStep));
slides.forEach((slide, i) => slide.classList.toggle('active', i === index));
applySteps(slides[index]);
progress.style.width = `${((index + 1) / slides.length) * 100}%`;
const suffix = step ? `.${step}` : '';
history.replaceState(null, '', `#${index + 1}${suffix}`);
}
function next() {
if (step < steps(slides[index]).length) { step++; applySteps(slides[index]); show(index, step); }
else if (index < slides.length - 1) show(index + 1);
}
function previous() {
if (step > 0) show(index, step - 1);
else if (index > 0) show(index - 1, steps(slides[index - 1]).length);
}
function fromHash() {
const [rawSlide, rawStep] = location.hash.slice(1).split('.');
const parsedSlide = Number.parseInt(rawSlide, 10);
const parsedStep = Number.parseInt(rawStep, 10);
show(Number.isFinite(parsedSlide) ? parsedSlide - 1 : 0, Number.isFinite(parsedStep) ? parsedStep : 0);
}
document.addEventListener('keydown', event => {
if (['ArrowRight', ' ', 'PageDown', 'j', 'l'].includes(event.key)) { event.preventDefault(); next(); }
if (['ArrowLeft', 'PageUp', 'k', 'h'].includes(event.key)) { event.preventDefault(); previous(); }
if (event.key === 'Home') { event.preventDefault(); show(0); }
if (event.key === 'End') { event.preventDefault(); show(slides.length - 1); }
if (event.key.toLowerCase() === 'f') {
event.preventDefault();
if (document.fullscreenElement) document.exitFullscreen();
else document.documentElement.requestFullscreen().catch(() => {});
}
});
document.addEventListener('click', event => {
if (window.getSelection()?.toString()) return;
if (event.clientX < window.innerWidth / 2) previous(); else next();
});
window.addEventListener('hashchange', fromHash);
fromHash();
})();
</script>
</body>
</html>

View file

@ -0,0 +1,296 @@
# Pico5: durable documents and views
> Design direction, not a complete harness specification. This fixes the state,
> view, and storage foundation. Implement and validate it before tasks or policy.
## 1. Model
Pico has four durable primitives:
| primitive | purpose | storage semantics |
|---|---|---|
| entry | immutable transcript/history | append-only, paged |
| document | current JSON application/presentation state | Chord ops plus bases |
| task state | private recovery bookmark | complete replacement, retireable |
| memo | invocation idempotency receipt | small, first-writer-wins, retireable |
Facet and task authors mutate ordinary tracked JSON objects. They do not define
delta formats, repositories, checkpoints, or restart reconstruction.
Entries hold unbounded history. A document holds current state; it may be large,
but should not double as an unbounded event archive. Sessions and inactive
documents must be loadable and evictable independently.
## 2. Documents
A document has persisted metadata:
```ts
interface DocumentMetadata {
id: DocumentId; // unique incarnation; never reused
key: string;
scope: "session" | { conversationId: Id };
history: "latest" | "rewindable";
}
```
History policy is immutable and applies to the whole document:
- `latest`: only current state is observable; records before a committed base may
be reclaimed.
- `rewindable`: state and document membership are queryable at a historical
Session commit; history remains while forks may address it.
Creation records metadata and a mandatory initial base, including `{}`. Retirement
hides a document from current enumeration. A latest document may then be deleted.
A rewindable document remains historically enumerable, including if retired later.
Recreating a logical key creates a new ID/incarnation.
A Session facet may create its own Session or conversation documents, such as a
canvas or diff review.
## 3. Commits and task state
A Session has one serialized mutation line. One logical commit may atomically:
- create, mutate, or retire multiple documents;
- create, replace, or retire task state and memos;
- append entries.
Document mutation uses transaction-owned drafts. The document manager produces
Chord ops and the candidate current value. Storage commits everything or nothing.
Only after success does the manager adopt drafts and notify views/subscribers.
Failure leaves authoritative values and replicas unchanged.
> If a state change is visible through Chord, it has already committed durably.
There is no volatile `publish()`. Generation and tool progress use throttled durable
commits. A crash may lose the current throttle window, but never state a client saw.
Each task kind defines private JSON recovery state. Phase transitions replace it in
full. The runtime separately owns the task envelope: kind, input, conversation,
ownership/dependencies, memos, and outstanding effect identities. Bulk output lives
in a document, not task state. On reopen, generation and tool kinds convert persisted
partials into aborted/interrupted entries and clear their live presentation state.
The effect sandwich remains:
```text
durable intent -> external effect -> durable outcome
```
Before retirement removes an outcome needed by a waiter, the same commit transfers
it to that waiter, an entry, or another durable receipt. Post-commit publication,
cleanup, or listener failures cannot roll back persistence or suppress later
listeners; they are isolated and reported separately.
## 4. Conversation view
Pico initially owns three documents per conversation:
```text
pi.rewindable history: rewindable; historical configuration
pi.sticky history: latest; queues and long-lived present policy
pi.live history: latest; high-churn turn, generation, tool and collapse state
```
Tasks write presentation state directly. Generation writes `live.message`; tools
write `live.tools`; historically inherited configuration lives in `rewindable`.
Private task state is not projected into view. Returning `live` to its initial state
may trigger a tiny base so streamed history can be reclaimed immediately.
The initial view is a fixed mount:
```ts
interface ConversationView<R, S, L> {
conversation: ConversationInfo;
entries: Entry[]; // bounded active transcript; older entries are paged
rewindable: R;
sticky: S;
live: L;
}
```
The mount consumes a complete Session commit and emits one Chord update. It prefixes
document paths and combines them with entry changes:
```text
["s", ["model"], value] -> ["s", ["rewindable", "model"], value]
["a", ["message"], delta] -> ["a", ["live", "message"], delta]
["r", value] -> ["s", ["live"], value]
```
Thus appending a final entry and clearing its preview is one publication. This is a
structural mount, not semantic projection, and owns no second tracker.
Third-party documents are not mounted initially. Facets expose them through their
own Chord services. Generic plugin mounting is deferred.
Before view implementation, define a durable bounded active-transcript/head rule and
how it changes atomically with appends. A years-long conversation must not hydrate
its entire transcript.
## 5. Chord integration
`DurableDocument` belongs to the harness and need not implement `ReplicatedState`.
A small Chord adapter, provisionally `replicatedState.own(source)`, exposes a durable
document as read-only replicated state without another tracker or re-diff.
The source provides its immutable committed value, a contiguous revision, and
committed ops. Session commit sequence is not a Chord sequence: unrelated commits
create gaps. Each document source and each conversation mount therefore has its own
contiguous publication revision.
Hydration must atomically capture matching value/revision and subscribe without a
gap or duplicate. The durable source's internal `publish()` is a no-op because no
visible mutation is pending. Test reentrant subscriptions and isolate listener
failures after persistence. The ConversationView mount consumes Session commits
directly, not independent document notifications.
A facet can expose a typed durable application document directly. `CanvasDoc` is
the declaration token, `canvasDocument` is the committed source, and `tx.doc(...)`
is a transaction-local draft:
```ts
const CanvasDoc = defineDoc<CanvasState>({
id: "@app/canvas", scope: "session", history: "latest",
initial: () => ({ strokes: [] }),
});
const canvasDocument = session.document(CanvasDoc);
const state = replicatedState.own(canvasDocument);
async function addStroke(stroke: Stroke, context: Context) {
await session.commit(tx => tx.doc(CanvasDoc).strokes.push(stroke), context);
}
```
## 6. Forks
Every Session commit has a global sequence; each entry records its containing commit.
Forking at entry `E`:
1. validates `E` through every ancestor cutoff;
2. enumerates rewindable documents existing at `E`'s commit, including ones retired
later;
3. loads each from the newest base at or before that commit plus later deltas;
4. creates a child-owned document ID and initial base for each;
5. copies no live task state.
A later state-only commit is excluded. Child documents then evolve independently.
Sticky/live documents have no historical lookup. Their simplest initial fork rule is
fresh defaults, never copied queues or running previews.
Before the fork API ships, decide whether selecting one of several entries appended
in the same commit inherits only that entry prefix or treats the commit as an
indivisible transcript boundary. Document state is the final state of the containing
commit either way.
## 7. Storage contract and bases
Storage receives logical metadata and writes, never filenames or table instructions:
```ts
type DocumentWrite =
| { type: "create"; metadata: DocumentMetadata; value: JsonValue }
| { type: "change"; id: DocumentId; ops: readonly Op[]; value: JsonValue }
| { type: "retire"; id: DocumentId };
```
For a change, `value` is a borrowed candidate snapshot. Storage normally persists
`ops`, but may persist a complete base when tail bytes or record count exceed its
policy:
```ts
persist(shouldBase(id, ops) ? [["r", value]] : ops);
```
It serializes only the chosen representation. Chord still publishes the incremental
ops. A committed base permits reclaiming older latest records; rewindable bases bound
replay but do not authorize deleting addressable history.
Storage supports current and as-of document enumeration/value lookup and distinguishes
not-yet-created, live, retired, and unavailable history.
## 8. Backends
### SQLite
One SQL transaction is one Session commit. Store immutable entries with commit
sequence, document metadata, indexed base/delta records by document and commit, and
one complete JSON row per live task. Current/as-of loads select the newest applicable
base and an indexed tail; they never scan unrelated documents. Retiring a task deletes
its row. Retired rewindable document history remains queryable.
Initially store Chord ops/bases directly. Benchmarks found full JSONB replacement
10-18% slower than TEXT and SQL JSON patching 2-4x more CPU with more DB+WAL writes
for a large canvas. SQLite also cannot faithfully map generic middle-array splices or
Chord's UTF-16 truncation. WAL checkpoint, synchronous mode, migrations, backup, and
vacuum are backend concerns.
### JSONL
JSONL prioritizes reclaimable task and latest-document storage. It privately chooses
sidecars; the harness does not know they exist:
```text
main.jsonl entries, lifecycle records, coordinated commits
task-<id>.jsonl live task state
doc-<id>.jsonl independently loadable document records
```
Every sidecar record is prepared. A commit appends all affected sidecars first and
one compact main marker listing their references last. Main-only commits are their
own marker. This uniform protocol intentionally accepts one small main record for an
isolated task/document update instead of introducing standalone and coordinated
record modes.
Recovery applies only records confirmed by a main marker. It removes torn and
unconfirmed tails before reopening for writes. Uncommitted sequences may be reused
only after all records carrying them are removed; committed sequences and document
IDs are never reused. Missing confirmed data is corruption unless a later committed
base or retirement proves it unnecessary.
An uncertain append error poisons the backend until close/reopen recovery. Appends
loop through the complete newline. After a main marker is complete, cleanup failure
cannot roll back the commit. Reclamation runs only after its base/retirement commits,
using serialized temp-file replacement and rename, then invalidates/reopens any
cached descriptor so later appends target the replacement inode.
Without per-commit `fsync`, ordered writes target ordinary process-crash consistency;
power/host/filesystem failure may lose or reorder acknowledged commits. Durable mode
flushes sidecars before the main marker. File descriptors are bounded by per-write
open or an LRU.
## 9. Deferred
Do not initially add per-subtree history, volatile publication, a public semantic
event channel, generic plugin mounting, SQL JSON patch translation, CRDT/offline
writes, whole-Session DOM materialization, arbitrary atomic presentation across
third-party services, sticky copy options, history-policy changes/pruning, or pico3's
config/plugin projections. UI status is durable document state; if transient history
is needed, add a bounded field to `live` with explicit retention.
## 10. Implementation order
1. **Storage semantics: memory + SQLite.** Create two initial-base documents;
atomically change both, replace a task, and append an entry; force latest and
rewindable bases; test current/as-of membership and values, retirement/recreation,
entry-to-commit mapping, reopen, and rollback.
2. **JSONL.** Implement uniform sidecar preparation/main-marker publication,
poisoned writes, committed-only reclamation, and fault injection at every physical
boundary.
3. **Document manager + narrow Chord proof.** Prove draft rollback without
per-success full cloning, one tracker per loaded doc, eviction/forks, contiguous
revisions, hydration fencing, listener isolation, and `watchDoc` snapshot/update/
retirement behavior.
4. **Tasks.** Add runtime envelopes, full-replacement kind state, phases, memos,
effect sandwich, dependency handoff, abort, retirement, durable built-in progress,
and reopen conversion of persisted partials.
5. **Conversation view.** First settle active-transcript and multi-entry fork
boundaries; then mount rewindable, sticky, and live with one publication per
Session commit.
6. **The rest.** Inputs, hooks, tools, system sections, dynamic tools, notices if
required, and product wiring.
Storage is accepted before task implementation. All backends share one semantic
conformance suite; JSONL additionally has a physical crash/failure matrix.

View file

@ -1,273 +0,0 @@
import type { Id, JsonValue } from "./core.ts";
export type Scope =
| { readonly type: "session" }
| { readonly type: "conversation"; readonly conversationId: Id }
| { readonly type: "task"; readonly taskId: Id }
| { readonly type: "shared"; readonly id: Id };
export type PublicScope = Exclude<Scope, { type: "shared" }>;
export type StickyScope = Extract<PublicScope, { type: "session" | "task" }>;
export type ConversationScope = Extract<PublicScope, { type: "conversation" }>;
declare const addressType: unique symbol;
export interface Address<T extends JsonValue = JsonValue> {
readonly scope: Scope;
readonly namespace: string;
readonly key?: string;
readonly kind: "value" | "list";
readonly rewind: boolean;
readonly [addressType]?: T;
}
export interface Value<T extends JsonValue> extends Address<T> {
readonly kind: "value";
readonly default?: T;
}
export interface List<T extends JsonValue> extends Address<T> {
readonly kind: "list";
}
export interface Element<T extends JsonValue> {
readonly id: Id;
readonly value: T;
}
export interface UnboundValue<T extends JsonValue, R extends boolean = boolean> {
readonly namespace: string;
readonly key?: string;
readonly rewind: R;
readonly default?: T;
readonly [addressType]?: T;
bind(conversationId: Id): Value<T>;
}
export interface UnboundList<T extends JsonValue, R extends boolean = boolean> {
readonly namespace: string;
readonly key?: string;
readonly rewind: R;
readonly [addressType]?: T;
bind(conversationId: Id): List<T>;
}
export type ConversationValueDefinition<T extends JsonValue, R extends boolean = boolean> = UnboundValue<T, R>;
export type ConversationListDefinition<T extends JsonValue, R extends boolean = boolean> = UnboundList<T, R>;
export type PayloadOf<D> = D extends Address<infer BoundPayload>
? BoundPayload
: D extends UnboundValue<infer DefinitionPayload, boolean>
? DefinitionPayload
: never;
export type ResolvedValue<D> = D extends { readonly default: infer V } ? V : PayloadOf<D> | undefined;
type DefaultedUnboundValue<T extends JsonValue, R extends boolean> = Omit<UnboundValue<T, R>, "bind" | "default"> & {
readonly default: T;
bind(conversationId: Id): Value<T> & { readonly default: T };
};
type StickyOptions<T extends JsonValue> = { readonly key?: string; readonly default?: T };
type DefaultedStickyOptions<T extends JsonValue> = { readonly key?: string; readonly default: T };
type ConversationOptions<T extends JsonValue> = {
readonly key?: string;
readonly rewind: boolean;
readonly default?: T;
};
type DefaultedConversationOptions<T extends JsonValue> = {
readonly key?: string;
readonly rewind: boolean;
readonly default: T;
};
type InternalOptions<T extends JsonValue> = StickyOptions<T> & { readonly rewind?: boolean };
function assertPublicNamespace(namespace: string): void {
if (namespace.startsWith("pi.")) throw new Error(`Namespace ${namespace} is reserved`);
}
function assertPublicScope(scope: Scope): asserts scope is PublicScope {
if (scope.type === "shared") throw new Error("Shared scope is internal");
}
function value<T extends JsonValue>(scope: Scope, namespace: string, options?: InternalOptions<T>): Value<T> {
return Object.freeze({
...options,
scope,
namespace,
kind: "value" as const,
rewind: options?.rewind ?? false,
});
}
function list<T extends JsonValue>(scope: Scope, namespace: string, options?: InternalOptions<T>): List<T> {
return Object.freeze({
...options,
scope,
namespace,
kind: "list" as const,
rewind: options?.rewind ?? false,
});
}
export function defineValue<T extends JsonValue>(
scope: StickyScope,
namespace: string,
options: DefaultedStickyOptions<T>,
): Value<T> & { readonly default: T };
export function defineValue<T extends JsonValue>(
scope: StickyScope,
namespace: string,
options?: { readonly key?: string },
): Value<T>;
export function defineValue<T extends JsonValue>(
scope: ConversationScope,
namespace: string,
options: DefaultedConversationOptions<T>,
): Value<T> & { readonly default: T };
export function defineValue<T extends JsonValue>(
scope: ConversationScope,
namespace: string,
options: { readonly key?: string; readonly rewind: boolean },
): Value<T>;
export function defineValue<T extends JsonValue>(
scope: Scope,
namespace: string,
options?: InternalOptions<T>,
): Value<T> {
assertPublicScope(scope);
assertPublicNamespace(namespace);
return value(scope, namespace, options);
}
export function defineList<T extends JsonValue>(
scope: StickyScope,
namespace: string,
options?: { readonly key?: string },
): List<T>;
export function defineList<T extends JsonValue>(
scope: ConversationScope,
namespace: string,
options: Omit<ConversationOptions<T>, "default">,
): List<T>;
export function defineList<T extends JsonValue>(
scope: Scope,
namespace: string,
options?: InternalOptions<T>,
): List<T> {
assertPublicScope(scope);
assertPublicNamespace(namespace);
return list(scope, namespace, options);
}
export function conversationValue<T extends JsonValue>(
namespace: string,
options: { readonly key?: string; readonly rewind: true; readonly default: T },
): DefaultedUnboundValue<T, true>;
export function conversationValue<T extends JsonValue>(
namespace: string,
options: { readonly key?: string; readonly rewind: false; readonly default: T },
): DefaultedUnboundValue<T, false>;
export function conversationValue<T extends JsonValue>(
namespace: string,
options: { readonly key?: string; readonly rewind: true },
): UnboundValue<T, true>;
export function conversationValue<T extends JsonValue>(
namespace: string,
options: { readonly key?: string; readonly rewind: false },
): UnboundValue<T, false>;
export function conversationValue<T extends JsonValue>(
namespace: string,
options: ConversationOptions<T>,
): UnboundValue<T> {
assertPublicNamespace(namespace);
return unboundValue(namespace, options);
}
export function conversationList<T extends JsonValue>(
namespace: string,
options: { readonly key?: string; readonly rewind: true },
): UnboundList<T, true>;
export function conversationList<T extends JsonValue>(
namespace: string,
options: { readonly key?: string; readonly rewind: false },
): UnboundList<T, false>;
export function conversationList<T extends JsonValue>(
namespace: string,
options: Omit<ConversationOptions<T>, "default">,
): UnboundList<T> {
assertPublicNamespace(namespace);
return unboundList(namespace, options);
}
export function sessionValue<T extends JsonValue>(
namespace: string,
options: { readonly key?: string; readonly default: T },
): Value<T> & { readonly default: T };
export function sessionValue<T extends JsonValue>(namespace: string, options?: { readonly key?: string }): Value<T>;
export function sessionValue<T extends JsonValue>(namespace: string, options?: StickyOptions<T>): Value<T> {
return defineValue({ type: "session" }, namespace, options);
}
/** Internal constructor for fixed `pi.*` definitions and shared task output state. */
export function defineInternalValue<T extends JsonValue>(
scope: Scope,
namespace: string,
options?: InternalOptions<T>,
): Value<T> {
return value(scope, namespace, options);
}
/** Internal constructor for fixed `pi.*` definitions and shared task output state. */
export function defineInternalList<T extends JsonValue>(
scope: Scope,
namespace: string,
options?: InternalOptions<T>,
): List<T> {
return list(scope, namespace, options);
}
/** Internal constructor for fixed conversation configuration definitions. */
export function internalConversationValue<T extends JsonValue, R extends boolean>(
namespace: string,
options: { readonly key?: string; readonly rewind: R; readonly default: T },
): DefaultedUnboundValue<T, R>;
export function internalConversationValue<T extends JsonValue, R extends boolean>(
namespace: string,
options: { readonly key?: string; readonly rewind: R },
): UnboundValue<T, R>;
export function internalConversationValue<T extends JsonValue, R extends boolean>(
namespace: string,
options: { readonly key?: string; readonly rewind: R; readonly default?: T },
): UnboundValue<T, R> {
return unboundValue(namespace, options);
}
function unboundValue<T extends JsonValue, R extends boolean>(
namespace: string,
options: { readonly key?: string; readonly rewind: R; readonly default?: T },
): UnboundValue<T, R> {
return Object.freeze({
...options,
namespace,
bind: (conversationId: Id) =>
value<T>({ type: "conversation", conversationId }, namespace, {
key: options.key,
rewind: options.rewind,
default: options.default,
}),
});
}
function unboundList<T extends JsonValue, R extends boolean>(
namespace: string,
options: { readonly key?: string; readonly rewind: R },
): UnboundList<T, R> {
return Object.freeze({
...options,
namespace,
bind: (conversationId: Id) =>
list<T>({ type: "conversation", conversationId }, namespace, {
key: options.key,
rewind: options.rewind,
}),
});
}

View file

@ -1,12 +0,0 @@
import type { JsonRepresentation } from "@earendil-works/chord";
export type JsonPrimitive = null | boolean | number | string;
export type JsonValue = JsonPrimitive | JsonObject | readonly JsonValue[];
export type Stored<T> = JsonRepresentation<T>;
export interface JsonObject {
readonly [key: string]: JsonValue;
}
export type Id = number;
export type Seq = number;

View file

@ -1,106 +0,0 @@
import type { AssistantMessage, Message, ToolResultMessage, Usage, UserMessage } from "@earendil-works/pi-ai";
import type { Id, JsonValue, Stored } from "./core.ts";
export type ContextEdit =
| { readonly target: Id; readonly action: "omit" }
| { readonly target: Id; readonly action: "replace"; readonly messages: readonly Stored<Message>[] };
export interface Entry {
readonly id: Id;
readonly conversationId: Id;
readonly kind: string;
readonly byTaskId?: Id;
readonly data?: JsonValue;
readonly model?: readonly Stored<Message>[];
readonly head?: Id;
readonly edits?: readonly ContextEdit[];
}
export type EntryIdentity = Pick<Entry, "id" | "conversationId" | "kind" | "byTaskId">;
export type EntryBase = EntryIdentity;
export interface EntryData<D extends JsonValue = JsonValue> {
readonly data: D;
}
export interface ModelProjection<M extends Message = Message> {
readonly model: readonly Stored<M>[];
}
export interface ContextHead {
readonly head: Id;
}
export interface ContextEdits {
readonly edits: readonly ContextEdit[];
}
export type EntryInput<E extends Entry> = Omit<E, keyof EntryIdentity | "head"> &
(E extends { readonly head: Id } ? { readonly head: Id | "self" } : { readonly head?: never });
export interface EntryKind<E extends Entry = Entry> {
readonly kind: string;
is(entry: Entry | undefined): entry is E;
}
export function defineEntry<E extends Entry>(kind: string): EntryKind<E> {
return Object.freeze({
kind,
is(entry: Entry | undefined): entry is E {
return entry?.kind === kind;
},
});
}
export interface Conversation {
readonly id: Id;
readonly parent?: { readonly conversationId: Id; readonly at: Id };
readonly owner?: Id;
}
export type AssistantEntryData = { readonly attempt: number };
export type UsageEntryData = { readonly attempt: number; readonly usage?: Stored<Usage>; readonly error: string };
export type ToolControl = {
readonly terminate?: true;
readonly handoff?: string;
readonly addTools?: readonly string[];
};
export type ToolDiagnostic = {
readonly severity: "info" | "warn" | "error";
readonly message: string;
readonly code?: string;
};
export type ToolUsage = { readonly [key: string]: number };
export type ToolResultData = {
readonly details?: JsonValue;
readonly usage?: ToolUsage;
readonly diagnostics?: readonly ToolDiagnostic[];
readonly control?: ToolControl;
readonly truncated?: { readonly bytes: number; readonly lines: number };
};
export type SectionRecord =
| { readonly key: string; readonly action: "set"; readonly value: JsonValue; readonly rendered: string }
| { readonly key: string; readonly action: "remove" };
export type SystemEntryData = { readonly baseline?: true; readonly sections: readonly SectionRecord[] };
export type UserEntry = EntryBase & {
readonly model: readonly [Stored<UserMessage>];
readonly data?: { readonly continuation: true; readonly from: Id };
};
export type AssistantEntry = EntryBase & {
readonly model: readonly [Stored<AssistantMessage>];
readonly data?: AssistantEntryData;
};
export type ToolResultEntry = EntryBase & {
readonly model: readonly [Stored<ToolResultMessage>];
readonly data: ToolResultData;
};
export type NoticeEntry = EntryBase & { readonly model: readonly [Stored<UserMessage>]; readonly data?: JsonValue };
export type UsageEntry = EntryBase & { readonly data: UsageEntryData };
export type SummaryEntry = EntryBase & {
readonly model: readonly [Stored<UserMessage>];
readonly data: { readonly through: Id };
readonly head: Id;
};
export type HandoffEntry = EntryBase & { readonly model: readonly [Stored<UserMessage>]; readonly head: Id };
export type ResetEntry = EntryBase & { readonly head: Id };

View file

@ -1,62 +0,0 @@
import type { Context } from "@earendil-works/chord";
import type { Id } from "./core.ts";
declare const hookIn: unique symbol;
declare const hookOut: unique symbol;
export type Fold = "collect" | "first" | "chain";
export interface HookPoint<In, Out, F extends Fold = Fold> {
readonly fold: F;
readonly onThrow: "skip" | "abort";
readonly [hookIn]?: In;
readonly [hookOut]?: Out;
}
export function defineHookPoint<In, Out = void>(options: {
readonly fold: "collect";
readonly onThrow: "skip" | "abort";
}): HookPoint<In, Out, "collect">;
export function defineHookPoint<In, Out = void>(options: {
readonly fold: "first";
readonly onThrow: "skip" | "abort";
}): HookPoint<In, Out, "first">;
export function defineHookPoint<In extends object, Out extends Partial<In>>(options: {
readonly fold: "chain";
readonly onThrow: "skip" | "abort";
}): HookPoint<In, Out, "chain">;
export function defineHookPoint(options: {
readonly fold: Fold;
readonly onThrow: "skip" | "abort";
}): HookPoint<unknown, unknown> {
return Object.freeze(options);
}
export type HookPoints = Record<string, HookPoint<unknown, unknown>>;
export type EmptyHookPoints = Record<never, never>;
export type HookIn<P> = P extends HookPoint<infer In, unknown> ? In : never;
export type HookOut<P> = P extends HookPoint<unknown, infer Out> ? Out : never;
// biome-ignore lint/suspicious/noConfusingVoidType: void allows handlers with no return value.
type HookHandlerResult<P> = HookOut<P> | void;
export type HookHandler<P> = (
input: HookIn<P>,
info: HookInfo,
ctx: Context,
) => HookHandlerResult<P> | Promise<HookHandlerResult<P>>;
export interface HookInfo {
readonly kind: string;
readonly taskId: Id;
readonly conversationId: Id;
}
export interface HookRunner<H extends HookPoints> {
run<P extends H[keyof H]>(point: P, input: HookIn<P>, ctx: Context): Promise<HookResult<P>>;
}
export type HookResult<P> = P extends { fold: "collect" }
? { readonly outputs: readonly HookOut<P>[]; readonly threw?: unknown }
: P extends { fold: "first" }
? { readonly output?: HookOut<P>; readonly threw?: unknown }
: { readonly output: HookIn<P>; readonly threw?: unknown };

View file

@ -1,154 +0,0 @@
export type {
Address,
ConversationListDefinition,
ConversationScope,
ConversationValueDefinition,
Element,
List,
PayloadOf,
PublicScope,
ResolvedValue,
Scope,
StickyScope,
UnboundList,
UnboundValue,
Value,
} from "./addresses.ts";
export {
conversationList,
conversationValue,
defineList,
defineValue,
sessionValue,
} from "./addresses.ts";
export type { Id, JsonObject, JsonPrimitive, JsonValue, Seq, Stored } from "./core.ts";
export type {
AssistantEntry,
AssistantEntryData,
ContextEdit,
ContextEdits,
ContextHead,
Conversation,
Entry,
EntryBase,
EntryData,
EntryIdentity,
EntryInput,
EntryKind,
HandoffEntry,
ModelProjection,
NoticeEntry,
ResetEntry,
SectionRecord,
SummaryEntry,
SystemEntryData,
ToolControl,
ToolDiagnostic,
ToolResultData,
ToolResultEntry,
ToolUsage,
UsageEntry,
UsageEntryData,
UserEntry,
} from "./entries.ts";
export { defineEntry } from "./entries.ts";
export type {
Fold,
HookHandler,
HookIn,
HookInfo,
HookOut,
HookPoint,
HookPoints,
HookResult,
HookRunner,
} from "./hooks.ts";
export { defineHookPoint } from "./hooks.ts";
export { MemoryStorage } from "./memory-storage.ts";
export type {
AbortClosure,
AbortFinalTx,
AbortTaskConversation,
AbortTaskRuntime,
AbortTx,
Acceptance,
BaseTaskTx,
BoundPublicState,
BoundTxState,
ConversationCreateSpec,
ConversationTx,
FinalOutput,
FinalTx,
GenerationOutput,
HostTx,
InputHandle,
InputOutcome,
PublicList,
PublicValue,
QueuedInput,
ReadonlyTaskOutput,
RuntimeOutput,
ScratchReader,
SendInput,
StickyPublicValue,
StoredEntryDraft,
StoredInputResult,
TaskConversation,
TaskOutput,
TaskRuntime,
TerminalClosure,
TerminalInputOutcome,
ToolInput,
ToolOutputState,
TxList,
TxReaders,
TxState,
TxValue,
UserInput,
} from "./runtime.ts";
export { Closed, Faulted, NestedLineOperation, ReadAfterWrite } from "./session.ts";
export type {
ConversationScan,
EntryScan,
NewTask,
Page,
PageQuery,
StateWrite,
Storage,
TaskScan,
Write,
} from "./storage.ts";
export { InvalidHistoryPosition, OutputRetired, ScratchRetired } from "./storage.ts";
export type {
AbortedOf,
AnyDefinedKind,
CheckpointOf,
Completion,
ConfigBundle,
DistributiveOmit,
ExactJsonInput,
FailureOf,
InputOf,
NoCheckpoint,
NoExtra,
OrdinaryKind,
OutputOf,
OutputState,
PayloadsOf,
ResultOf,
RunningTask,
StoredTaskOutputRef,
Task,
TaskBase,
TaskCheckpoint,
TaskKind,
TaskKindDefinition,
TaskKindFactory,
TaskOf,
TaskOutcome,
TaskOutputDefinition,
TaskOutputField,
TaskOutputKind,
TaskOutputRef,
} from "./tasks.ts";
export { defineTask, defineTaskOutput } from "./tasks.ts";

View file

@ -1,578 +0,0 @@
import type { Context } from "@earendil-works/chord";
import type { Address, Element, List, Value } from "./addresses.ts";
import type { Id, JsonValue, Seq } from "./core.ts";
import type { Conversation, Entry } from "./entries.ts";
import {
type ConversationScan,
type EntryScan,
InvalidHistoryPosition,
OutputRetired,
type Page,
type PageQuery,
ScratchRetired,
type Storage,
type StoredConversation,
type TaskScan,
type Write,
} from "./storage.ts";
import type { Task } from "./tasks.ts";
type ValueHistoryWrite =
| { readonly seq: Seq; readonly type: "set"; readonly value: JsonValue }
| { readonly seq: Seq; readonly type: "delete" };
type ListHistoryWrite =
| { readonly seq: Seq; readonly type: "append"; readonly element: Element<JsonValue> }
| { readonly seq: Seq; readonly type: "remove"; readonly elementId: Id }
| { readonly seq: Seq; readonly type: "clear" };
type StoredState =
| { readonly kind: "value"; readonly rewind: false; value?: JsonValue }
| {
readonly kind: "list";
readonly rewind: false;
readonly elements: Map<Id, Element<JsonValue>>;
readonly known: Set<Id>;
}
| { readonly kind: "value"; readonly rewind: true; readonly writes: ValueHistoryWrite[] }
| { readonly kind: "list"; readonly rewind: true; readonly writes: ListHistoryWrite[] };
type RewindListState = Extract<StoredState, { kind: "list"; rewind: true }>;
interface AddressState {
value?: Extract<StoredState, { kind: "value" }>;
list?: Extract<StoredState, { kind: "list" }>;
}
interface NamespaceState {
unkeyed?: AddressState;
readonly keyed: Map<string, AddressState>;
}
type ScopeState = Map<string, NamespaceState>;
interface CappedListState {
readonly state: RewindListState | undefined;
readonly cap: Seq | undefined;
}
interface StoredEntry {
readonly seq: Seq;
readonly entry: Entry;
}
interface ProspectiveWrites {
readonly conversations: Map<Id, StoredConversation>;
readonly entries: Map<Id, StoredEntry>;
readonly listAppends: { readonly seq: Seq; readonly address: List<JsonValue>; readonly elementId: Id }[];
}
function rewindable(
address: Address,
): address is Address & { readonly scope: Extract<Address["scope"], { type: "conversation" }>; readonly rewind: true } {
return address.scope.type === "conversation" && address.rewind;
}
function scopeId(address: Address): Id | undefined {
if (address.scope.type === "conversation") return address.scope.conversationId;
if (address.scope.type === "task") return address.scope.taskId;
if (address.scope.type === "shared") return address.scope.id;
return undefined;
}
function page<T>(
values: Iterable<T>,
query: PageQuery,
id: (value: T) => Id,
matches: (value: T) => boolean,
ascending: boolean,
): Page<T> {
const items: T[] = [];
let more = false;
const ordered = [...values].sort((left, right) => (ascending ? id(left) - id(right) : id(right) - id(left)));
for (const value of ordered) {
const valueId = id(value);
if (
!matches(value) ||
(query.cursor !== undefined && (ascending ? valueId <= query.cursor : valueId >= query.cursor))
) {
continue;
}
if (items.length === query.limit) {
more = true;
break;
}
items.push(value);
}
return { items, ...(more && items.length > 0 ? { next: id(items.at(-1)!) } : {}) };
}
export class MemoryStorage implements Storage {
private nextObjectId: Id = 1;
private committedIdHighWater: Id = 0;
private nextSeq: Seq = 1;
private readonly conversations = new Map<Id, StoredConversation>();
private readonly entries = new Map<Id, StoredEntry>();
private readonly entriesByConversation = new Map<Id, StoredEntry[]>();
private readonly tasks = new Map<Id, Task>();
private readonly outputReferences = new Map<Id, number>();
private readonly sessionState: ScopeState = new Map();
private readonly conversationStickyState = new Map<Id, ScopeState>();
private readonly conversationRewindableState = new Map<Id, ScopeState>();
private readonly taskState = new Map<Id, ScopeState>();
private readonly sharedState = new Map<Id, ScopeState>();
static create(): MemoryStorage {
return new MemoryStorage();
}
nextId(): Id {
return this.nextObjectId++;
}
async commit(writes: readonly Write[], _ctx: Context): Promise<readonly Seq[]> {
this.validateListRemovals(writes);
const seqs = writes.map((_, index) => this.nextSeq + index);
const affectedOutputs = new Set<Id>();
let batchMaxId = this.committedIdHighWater;
for (const write of writes) {
if (write.type === "conversation.create") batchMaxId = Math.max(batchMaxId, write.conversation.id);
else if (write.type === "entry.append") batchMaxId = Math.max(batchMaxId, write.entry.id);
else if (write.type === "task.create") batchMaxId = Math.max(batchMaxId, write.task.id);
else if (write.type === "list.append") batchMaxId = Math.max(batchMaxId, write.element.id);
if (write.type !== "task.create" && write.type !== "task.set") continue;
const currentOutput = this.tasks.get(write.task.id)?.output?.id;
if (currentOutput !== undefined) affectedOutputs.add(currentOutput);
if (write.task.output !== undefined) affectedOutputs.add(write.task.output.id);
}
for (let index = 0; index < writes.length; index++) this.apply(writes[index]!, seqs[index]!);
for (const write of writes) {
if (write.type === "task.set" && write.task.status === "terminal") this.taskState.delete(write.task.id);
}
for (const id of affectedOutputs) {
if (!this.outputReferences.has(id)) this.sharedState.delete(id);
}
this.committedIdHighWater = batchMaxId;
this.nextObjectId = Math.max(this.nextObjectId, batchMaxId + 1);
this.nextSeq += writes.length;
return seqs;
}
async close(_ctx: Context): Promise<void> {
this.nextObjectId = this.committedIdHighWater + 1;
}
async getConversations(ids: readonly Id[], _ctx: Context): Promise<ReadonlyMap<Id, Conversation>> {
const result = new Map<Id, Conversation>();
for (const id of ids) {
const conversation = this.conversations.get(id);
if (conversation !== undefined) result.set(id, this.publicConversation(conversation));
}
return result;
}
async scanConversations(query: ConversationScan, _ctx: Context): Promise<Page<Conversation>> {
return page(
[...this.conversations.values()].map((conversation) => this.publicConversation(conversation)),
query,
(conversation) => conversation.id,
(conversation) =>
(query.parent === undefined || conversation.parent?.conversationId === query.parent) &&
(query.owner === undefined || conversation.owner === query.owner),
true,
);
}
async getEntries(ids: readonly Id[], _ctx: Context): Promise<ReadonlyMap<Id, Entry>> {
const result = new Map<Id, Entry>();
for (const id of ids) {
const stored = this.entries.get(id);
if (stored !== undefined) result.set(id, stored.entry);
}
return result;
}
async scanEntries(query: EntryScan, _ctx: Context): Promise<Page<Entry>> {
const cursor = query.cursor === undefined ? undefined : this.historyPosition(query.conversationId, query.cursor);
const items: Entry[] = [];
let more = false;
for (const stored of this.visibleEntries(query.conversationId, query.through)) {
if (cursor !== undefined && stored.seq >= cursor) continue;
if (query.kind !== undefined && stored.entry.kind !== query.kind) continue;
if (items.length === query.limit) {
more = true;
break;
}
items.push(stored.entry);
}
return { items, ...(more && items.length > 0 ? { next: items.at(-1)!.id } : {}) };
}
async newestHead(conversationId: Id, at: Id, _ctx: Context): Promise<Entry | undefined> {
for (const stored of this.visibleEntries(conversationId, at)) {
if (stored.entry.head !== undefined) return stored.entry;
}
return undefined;
}
async getTasks(ids: readonly Id[], _ctx: Context): Promise<ReadonlyMap<Id, Task>> {
const result = new Map<Id, Task>();
for (const id of ids) {
const task = this.tasks.get(id);
if (task !== undefined) result.set(id, task);
}
return result;
}
async scanTasks(query: TaskScan, _ctx: Context): Promise<Page<Task>> {
return page(
this.tasks.values(),
query,
(task) => task.id,
(task) =>
(query.conversationIds === undefined || query.conversationIds.includes(task.conversationId)) &&
(query.statuses === undefined || query.statuses.includes(task.status)) &&
(query.kind === undefined || task.kind === query.kind) &&
(query.abort === undefined || (task.abort === true) === query.abort) &&
(query.outputId === undefined || task.output?.id === query.outputId),
true,
);
}
async getValue<T extends JsonValue>(address: Value<T>, at: Id | undefined, _ctx: Context): Promise<T | undefined> {
if (at !== undefined) {
if (!rewindable(address)) throw new InvalidHistoryPosition(at);
this.assertScopeReadable(address);
return this.rewindValue(address, this.historyPosition(address.scope.conversationId, at));
}
this.assertScopeReadable(address);
if (rewindable(address)) return this.rewindValue(address, undefined);
const state = this.storedState(address);
return state?.kind === "value" && !state.rewind ? (state.value as T | undefined) : undefined;
}
async readList<T extends JsonValue>(
address: List<T>,
at: Id | undefined,
_ctx: Context,
): Promise<readonly Element<T>[]> {
if (at !== undefined) {
if (!rewindable(address)) throw new InvalidHistoryPosition(at);
this.assertScopeReadable(address);
return this.rewindListElements(address, this.historyPosition(address.scope.conversationId, at));
}
this.assertScopeReadable(address);
if (rewindable(address)) return this.rewindListElements(address, undefined);
const state = this.storedState(address);
if (state?.kind !== "list" || state.rewind) return [];
return [...(state.elements.values() as Iterable<Element<T>>)].sort((left, right) => left.id - right.id);
}
private historyPosition(conversationId: Id, at: Id, prospective?: ProspectiveWrites): Seq {
const stored = prospective?.entries.get(at) ?? this.entries.get(at);
if (stored === undefined) throw new InvalidHistoryPosition(at);
let currentId = conversationId;
let cap: Seq | undefined;
while (true) {
if (stored.entry.conversationId === currentId && (cap === undefined || stored.seq <= cap)) return stored.seq;
const conversation = prospective?.conversations.get(currentId) ?? this.conversations.get(currentId);
if (conversation?.parent === undefined) throw new InvalidHistoryPosition(at);
const parentAtSeq = this.historyPosition(
conversation.parent.conversationId,
conversation.parent.at,
prospective,
);
cap = Math.min(cap ?? parentAtSeq, parentAtSeq);
currentId = conversation.parent.conversationId;
}
}
private rewindValue<T extends JsonValue>(address: Value<T>, cap: Seq | undefined): T | undefined {
let conversationId = address.scope.type === "conversation" ? address.scope.conversationId : undefined;
while (conversationId !== undefined) {
const state = this.conversationStoredState(conversationId, address);
if (state?.kind === "value" && state.rewind) {
for (let index = state.writes.length - 1; index >= 0; index--) {
const write = state.writes[index]!;
if (cap !== undefined && write.seq > cap) continue;
return write.type === "set" ? (write.value as T) : undefined;
}
}
const conversation = this.conversations.get(conversationId);
if (conversation?.parent === undefined) return undefined;
const parentAtSeq = this.historyPosition(conversation.parent.conversationId, conversation.parent.at);
cap = Math.min(cap ?? parentAtSeq, parentAtSeq);
conversationId = conversation.parent.conversationId;
}
return undefined;
}
private rewindListElements<T extends JsonValue>(
address: List<T> & { readonly scope: Extract<Address["scope"], { type: "conversation" }> },
cap: Seq | undefined,
): readonly Element<T>[] {
const ancestry: CappedListState[] = [];
let conversationId = address.scope.conversationId;
while (true) {
const stored = this.conversationStoredState(conversationId, address);
ancestry.push({ state: stored?.kind === "list" && stored.rewind ? stored : undefined, cap });
const conversation = this.conversations.get(conversationId);
if (conversation?.parent === undefined) break;
const parentAtSeq = this.historyPosition(conversation.parent.conversationId, conversation.parent.at);
cap = Math.min(cap ?? parentAtSeq, parentAtSeq);
conversationId = conversation.parent.conversationId;
}
const elements = new Map<Id, Element<T>>();
for (const segment of ancestry.reverse()) {
for (const write of segment.state?.writes ?? []) {
if (segment.cap !== undefined && write.seq > segment.cap) continue;
if (write.type === "append") elements.set(write.element.id, write.element as Element<T>);
else if (write.type === "remove") elements.delete(write.elementId);
else elements.clear();
}
}
return [...elements.values()].sort((left, right) => left.id - right.id);
}
private *visibleEntries(conversationId: Id, through: Id | undefined): Iterable<StoredEntry> {
let conversation = this.conversations.get(conversationId);
let cap = through === undefined ? undefined : this.historyPosition(conversationId, through);
while (conversation !== undefined) {
const entries = this.entriesByConversation.get(conversation.id) ?? [];
for (let index = entries.length - 1; index >= 0; index--) {
const stored = entries[index]!;
if (cap === undefined || stored.seq <= cap) yield stored;
}
if (conversation.parent === undefined) return;
const parentAtSeq = this.historyPosition(conversation.parent.conversationId, conversation.parent.at);
cap = Math.min(cap ?? parentAtSeq, parentAtSeq);
conversation = this.conversations.get(conversation.parent.conversationId);
}
}
private validateListRemovals(writes: readonly Write[]): void {
const prospective: ProspectiveWrites = {
conversations: new Map(),
entries: new Map(),
listAppends: [],
};
for (let index = 0; index < writes.length; index++) {
const write = writes[index]!;
const seq = this.nextSeq + index;
if (write.type === "conversation.create") {
prospective.conversations.set(write.conversation.id, write.conversation);
} else if (write.type === "entry.append") {
prospective.entries.set(write.entry.id, { seq, entry: write.entry });
} else if (write.type === "list.append") {
prospective.listAppends.push({ seq, address: write.address, elementId: write.element.id });
} else if (
write.type === "list.remove" &&
!this.listKnowsElement(write.address, write.elementId, prospective)
) {
throw new Error(`List element ${write.elementId} does not belong to this list lineage`);
}
}
}
private listKnowsElement(address: List<JsonValue>, elementId: Id, prospective?: ProspectiveWrites): boolean {
if (!rewindable(address)) {
const state = this.storedState(address);
return (
(state?.kind === "list" && !state.rewind && state.known.has(elementId)) ||
(prospective?.listAppends.some(
(append) => append.elementId === elementId && this.sameAddress(append.address, address),
) ??
false)
);
}
let conversationId = address.scope.conversationId;
let cap: Seq | undefined;
while (true) {
const state = this.conversationStoredState(conversationId, address);
const appendedProspectively = prospective?.listAppends.some(
(append) =>
append.elementId === elementId &&
append.address.scope.type === "conversation" &&
append.address.scope.conversationId === conversationId &&
append.address.rewind === address.rewind &&
append.address.namespace === address.namespace &&
append.address.key === address.key &&
(cap === undefined || append.seq <= cap),
);
if (
appendedProspectively ||
(state?.kind === "list" &&
state.rewind &&
state.writes.some(
(write) =>
write.type === "append" &&
write.element.id === elementId &&
(cap === undefined || write.seq <= cap),
))
) {
return true;
}
const conversation = prospective?.conversations.get(conversationId) ?? this.conversations.get(conversationId);
if (conversation?.parent === undefined) return false;
const parentAtSeq = this.historyPosition(
conversation.parent.conversationId,
conversation.parent.at,
prospective,
);
cap = Math.min(cap ?? parentAtSeq, parentAtSeq);
conversationId = conversation.parent.conversationId;
}
}
private sameAddress(left: Address, right: Address): boolean {
if (
left.kind !== right.kind ||
left.rewind !== right.rewind ||
left.namespace !== right.namespace ||
left.key !== right.key ||
left.scope.type !== right.scope.type
) {
return false;
}
if (left.scope.type === "session") return true;
if (left.scope.type === "conversation" && right.scope.type === "conversation") {
return left.scope.conversationId === right.scope.conversationId;
}
if (left.scope.type === "task" && right.scope.type === "task") return left.scope.taskId === right.scope.taskId;
return left.scope.type === "shared" && right.scope.type === "shared" && left.scope.id === right.scope.id;
}
private publicConversation(conversation: StoredConversation): Conversation {
return {
id: conversation.id,
...(conversation.parent === undefined ? {} : { parent: conversation.parent }),
...(conversation.owner === undefined ? {} : { owner: conversation.owner }),
};
}
private assertScopeReadable(address: Address): void {
if (address.scope.type === "task") {
const task = this.tasks.get(address.scope.taskId);
if (task === undefined || task.status === "terminal") throw new ScratchRetired(address.scope.taskId);
} else if (address.scope.type === "shared" && !this.outputReferences.has(address.scope.id)) {
throw new OutputRetired(address.scope.id);
}
}
private apply(write: Write, seq: Seq): void {
if (write.type === "conversation.create") {
this.conversations.set(write.conversation.id, write.conversation);
} else if (write.type === "entry.append") {
const entry = write.entry;
const stored = { seq, entry };
this.entries.set(entry.id, stored);
const entries = this.entriesByConversation.get(entry.conversationId);
if (entries === undefined) this.entriesByConversation.set(entry.conversationId, [stored]);
else entries.push(stored);
} else if (write.type === "task.create" || write.type === "task.set") {
const current = this.tasks.get(write.task.id);
if (current?.status !== "terminal" && current?.output !== undefined)
this.removeOutputReference(current.output.id);
this.tasks.set(write.task.id, write.task);
if (write.task.status !== "terminal" && write.task.output !== undefined) {
this.outputReferences.set(write.task.output.id, (this.outputReferences.get(write.task.output.id) ?? 0) + 1);
}
} else if (write.type === "value.set") {
const state = this.valueState(write.address);
if (state.rewind) state.writes.push({ seq, type: "set", value: write.value });
else state.value = write.value;
} else if (write.type === "value.delete") {
const state = this.valueState(write.address);
if (state.rewind) state.writes.push({ seq, type: "delete" });
else state.value = undefined;
} else {
const state = this.listState(write.address);
if (state.rewind) {
if (write.type === "list.append") state.writes.push({ seq, type: "append", element: write.element });
else if (write.type === "list.remove")
state.writes.push({ seq, type: "remove", elementId: write.elementId });
else state.writes.push({ seq, type: "clear" });
} else if (write.type === "list.append") {
state.known.add(write.element.id);
state.elements.set(write.element.id, write.element);
} else if (write.type === "list.remove") state.elements.delete(write.elementId);
else state.elements.clear();
}
}
private removeOutputReference(id: Id): void {
const count = this.outputReferences.get(id)!;
if (count === 1) this.outputReferences.delete(id);
else this.outputReferences.set(id, count - 1);
}
private storedState(address: Address): StoredState | undefined {
return this.stateInScope(this.scopeState(address, false), address);
}
private conversationStoredState(conversationId: Id, address: Address): StoredState | undefined {
const scopes = address.rewind ? this.conversationRewindableState : this.conversationStickyState;
return this.stateInScope(scopes.get(conversationId), address);
}
private stateInScope(scope: ScopeState | undefined, address: Address): StoredState | undefined {
const namespace = scope?.get(address.namespace);
const state = address.key === undefined ? namespace?.unkeyed : namespace?.keyed.get(address.key);
return address.kind === "value" ? state?.value : state?.list;
}
private valueState(address: Value<JsonValue>): Extract<StoredState, { kind: "value" }> {
const initial: StoredState = rewindable(address)
? { kind: "value", rewind: true, writes: [] }
: { kind: "value", rewind: false };
return this.ensureState(address, initial) as Extract<StoredState, { kind: "value" }>;
}
private listState(address: List<JsonValue>): Extract<StoredState, { kind: "list" }> {
const initial: StoredState = rewindable(address)
? { kind: "list", rewind: true, writes: [] }
: { kind: "list", rewind: false, elements: new Map(), known: new Set() };
return this.ensureState(address, initial) as Extract<StoredState, { kind: "list" }>;
}
private ensureState(address: Address, initial: StoredState): StoredState {
const scope = this.scopeState(address, true)!;
let namespace = scope.get(address.namespace);
if (namespace === undefined) {
namespace = { keyed: new Map() };
scope.set(address.namespace, namespace);
}
let state = address.key === undefined ? namespace.unkeyed : namespace.keyed.get(address.key);
if (state === undefined) {
state = {};
if (address.key === undefined) namespace.unkeyed = state;
else namespace.keyed.set(address.key, state);
}
const stored = address.kind === "value" ? state.value : state.list;
if (stored !== undefined) return stored;
if (initial.kind === "value") state.value = initial;
else state.list = initial;
return initial;
}
private scopeState(address: Address, create: boolean): ScopeState | undefined {
if (address.scope.type === "session") return this.sessionState;
const id = scopeId(address)!;
const scopes =
address.scope.type === "conversation"
? address.rewind
? this.conversationRewindableState
: this.conversationStickyState
: address.scope.type === "task"
? this.taskState
: this.sharedState;
let state = scopes.get(id);
if (state === undefined && create) {
state = new Map();
scopes.set(id, state);
}
return state;
}
}

View file

@ -1,343 +0,0 @@
import type { Context } from "@earendil-works/chord";
import type { AssistantMessage, ImageContent, Message, TextContent, ToolCall } from "@earendil-works/pi-ai";
import type { Element, List, PayloadOf, ResolvedValue, UnboundList, UnboundValue, Value } from "./addresses.ts";
import type { Id, JsonValue, Stored } from "./core.ts";
import type { AssistantEntry, ContextEdit, Conversation, Entry, EntryInput, EntryKind, UserEntry } from "./entries.ts";
import type { EmptyHookPoints, HookPoints, HookRunner } from "./hooks.ts";
import type {
AnyDefinedKind,
Completion,
ExactJsonInput,
InputOf,
NoExtra,
OrdinaryKind,
OutputOf,
OutputState,
RunningTask,
Task,
TaskCheckpoint,
TaskOf,
TaskOutputRef,
} from "./tasks.ts";
export interface TxReaders {
getConversation(id: Id): Promise<Conversation | undefined>;
getEntry(id: Id): Promise<Entry | undefined>;
getEntry<E extends Entry>(kind: EntryKind<E>, id: Id): Promise<E | undefined>;
getEntries(ids: readonly Id[]): Promise<ReadonlyMap<Id, Entry>>;
getTask(id: Id): Promise<Task | undefined>;
getTask<K extends AnyDefinedKind>(kind: K, id: Id): Promise<TaskOf<K> | undefined>;
getTasks(ids: readonly Id[]): Promise<ReadonlyMap<Id, Task>>;
}
export interface TxValue<T extends JsonValue, Read = T | undefined> {
get(at?: Id): Promise<Read>;
set(value: T): void;
delete(): void;
}
export interface TxList<T extends JsonValue> {
read(at?: Id): Promise<readonly Element<T>[]>;
append(value: T): Id;
remove(id: Id): void;
clear(): void;
}
export interface TxState {
value<D extends Value<JsonValue>>(address: D): TxValue<PayloadOf<D>, ResolvedValue<D>>;
list<T extends JsonValue>(address: List<T>): TxList<T>;
}
export interface BoundTxState {
value<D extends Value<JsonValue> | UnboundValue<JsonValue>>(address: D): TxValue<PayloadOf<D>, ResolvedValue<D>>;
list<T extends JsonValue>(address: List<T> | UnboundList<T>): TxList<T>;
}
export type Acceptance = {
readonly kind: "send" | "write";
readonly requestId?: string;
readonly conversationId: Id;
readonly inputId: Id;
};
export type TaskSpec<K extends AnyDefinedKind> = {
readonly conversationId?: Id;
readonly input: InputOf<K>;
readonly after?: readonly Id[];
readonly background?: true;
} & ([OutputOf<K>] extends [never] ? { readonly output?: never } : { readonly output?: TaskOutputRef<OutputOf<K>> });
export interface ConversationCreateSpec {
readonly parent?: { readonly conversationId: Id; readonly at: Id | "start" };
}
export type UserInput = string | readonly (TextContent | ImageContent)[];
export type StoredEntryDraft = {
readonly kind: string;
readonly data?: JsonValue;
readonly model?: readonly Stored<Message>[];
readonly head?: Id | "self";
readonly edits?: readonly ContextEdit[];
};
export type QueuedInput =
| { readonly mode: "steer" | "followUp"; readonly input: Stored<UserInput>; readonly requestId?: string }
| { readonly mode: "write"; readonly entry: StoredEntryDraft; readonly requestId?: string };
export type StoredInputResult =
| { readonly status: "queued"; readonly requestId?: string }
| { readonly status: "placed"; readonly requestId?: string; readonly entry: Id }
| { readonly status: "done"; readonly requestId?: string; readonly entry: Id; readonly answer?: Id }
| {
readonly status: "unanswered";
readonly requestId?: string;
readonly entry?: Id;
readonly reason: "terminated" | "aborted" | "failed" | "stale";
readonly detail?: string;
};
export type InputOutcome =
| { readonly status: "queued"; readonly requestId?: string }
| { readonly status: "placed"; readonly requestId?: string; readonly input: UserEntry }
| {
readonly status: "done";
readonly requestId?: string;
readonly input: UserEntry;
readonly answer?: { readonly entry: AssistantEntry; readonly message: AssistantMessage };
}
| {
readonly status: "unanswered";
readonly requestId?: string;
readonly input?: UserEntry;
readonly reason: "terminated" | "aborted" | "failed" | "stale";
readonly detail?: string;
};
export type TerminalInputOutcome = Extract<InputOutcome, { status: "done" | "unanswered" }>;
export interface InternalAcceptOptions {
readonly input: UserInput;
readonly requestId?: string;
readonly whenBusy?: "followUp" | "steer" | "reject";
}
export interface SendInput {
readonly requestId?: string;
readonly content: UserInput;
readonly whenBusy?: "followUp" | "steer" | "reject";
}
export interface InputHandle {
readonly id: Id;
readonly conversationId: Id;
result(ctx: Context): Promise<InputOutcome | undefined>;
wait(ctx: Context): Promise<TerminalInputOutcome>;
abort(ctx: Context): Promise<"aborted" | "already_placed" | "not_found">;
}
interface TaskCreator<Kinds extends AnyDefinedKind, Extra = Record<never, never>> {
task<K extends Kinds, S extends TaskSpec<K> & Extra>(
kind: K,
spec: NoExtra<TaskSpec<K> & Extra, S> & { readonly input: ExactJsonInput<InputOf<K>, S["input"]> },
): Id;
}
export interface HostTx extends TxReaders, TxState, TaskCreator<OrdinaryKind, { readonly conversationId: Id }> {
createConversation(spec: ConversationCreateSpec): Id;
write<E extends Entry>(conversationId: Id, kind: EntryKind<E>, input: EntryInput<E>): Promise<Id>;
}
export interface ConversationTx extends TxReaders, BoundTxState, TaskCreator<OrdinaryKind> {
createConversation(spec: ConversationCreateSpec): Id;
write<E extends Entry>(kind: EntryKind<E>, input: EntryInput<E>): Promise<Id>;
}
interface TaskTxBase<C extends TaskCheckpoint> extends TxReaders, BoundTxState {
checkpoint(value: C): void;
createConversation(spec: ConversationCreateSpec): Id;
}
export interface BaseTaskTx<C extends TaskCheckpoint> extends TaskTxBase<C>, TaskCreator<OrdinaryKind> {
write<E extends Entry>(kind: EntryKind<E>, input: EntryInput<E>): Promise<Id>;
}
export interface InternalAdmissionTx {
accept(conversationId: Id, options: InternalAcceptOptions): Promise<Acceptance>;
queueInput(conversationId: Id, input: QueuedInput): Promise<Acceptance>;
write<E extends Entry>(
conversationId: Id,
kind: EntryKind<E>,
input: EntryInput<E>,
requestId?: string,
): Promise<Acceptance>;
}
export type CoreTaskTx<C extends TaskCheckpoint> = Omit<BaseTaskTx<C>, "task" | "write"> &
TaskCreator<AnyDefinedKind> &
InternalAdmissionTx & {
entry<E extends Entry>(kind: EntryKind<E>, conversationId: Id, input: EntryInput<E>): Id;
};
export interface PublicValue<T extends JsonValue, Read = T | undefined> {
get(ctx: Context): Promise<Read>;
get(at: Id, ctx: Context): Promise<Read>;
set(value: T, ctx: Context): Promise<void>;
delete(ctx: Context): Promise<void>;
}
export interface StickyPublicValue<T extends JsonValue, Read = T | undefined> {
get(ctx: Context): Promise<Read>;
set(value: T, ctx: Context): Promise<void>;
delete(ctx: Context): Promise<void>;
}
export interface PublicList<T extends JsonValue> {
read(ctx: Context): Promise<readonly Element<T>[]>;
read(at: Id, ctx: Context): Promise<readonly Element<T>[]>;
append(value: T, ctx: Context): Promise<Id>;
remove(id: Id, ctx: Context): Promise<void>;
clear(ctx: Context): Promise<void>;
}
export interface BoundPublicState {
value<D extends Value<JsonValue> | UnboundValue<JsonValue>>(address: D): PublicValue<PayloadOf<D>, ResolvedValue<D>>;
list<T extends JsonValue>(address: List<T> | UnboundList<T>): PublicList<T>;
}
export interface TaskOutput<O extends OutputState> {
readonly ref: TaskOutputRef<O>;
read(ctx: Context): Promise<O>;
mutate(mutator: (state: O) => undefined, ctx: Context): Promise<void>;
replace(value: O, ctx: Context): Promise<void>;
}
export interface ReadonlyTaskOutput<O extends OutputState> {
readonly ref: TaskOutputRef<O>;
read(ctx: Context): Promise<O>;
}
export type RuntimeOutput<O extends OutputState> = [O] extends [never]
? Record<never, never>
: { readonly output: TaskOutput<O> };
export type AbortRuntimeOutput<O extends OutputState> = [O] extends [never]
? Record<never, never>
: { readonly output: ReadonlyTaskOutput<O> };
export type FinalOutput<O extends OutputState> = [O] extends [never] ? Record<never, never> : { readonly output: O };
export interface ScratchReader {
value<D extends Value<JsonValue>>(address: D): Pick<TxValue<PayloadOf<D>, ResolvedValue<D>>, "get">;
list<T extends JsonValue>(address: List<T>): Pick<TxList<T>, "read">;
}
export interface TaskConversation extends BoundPublicState {
readonly id: Id;
send(input: SendInput, ctx: Context): Promise<InputHandle>;
}
export interface AbortTaskConversation extends BoundPublicState {
readonly id: Id;
}
interface TaskRuntimeBase<H extends HookPoints> {
readonly taskId: Id;
scratch<T>(build: (tx: TxState) => T | Promise<T>, ctx: Context): Promise<T>;
readonly hooks: HookRunner<H>;
conversation(id: Id, ctx: Context): Promise<TaskConversation | undefined>;
waitForTask(id: Id, ctx: Context): Promise<Task>;
abortTask(id: Id, ctx: Context): Promise<"marked" | "terminal">;
now(): number;
sleep(untilMs: number, ctx: Context): Promise<void>;
}
interface CommitRuntime<I extends JsonValue, C extends TaskCheckpoint, O extends OutputState, Tx> {
commit<T>(build: (tx: Tx, current: RunningTask<I, C, O>) => T | Promise<T>, ctx: Context): Promise<T>;
}
export type TaskRuntime<
I extends JsonValue,
C extends TaskCheckpoint,
O extends OutputState = never,
H extends HookPoints = EmptyHookPoints,
> = TaskRuntimeBase<H> & CommitRuntime<I, C, O, BaseTaskTx<C>> & RuntimeOutput<O>;
export type CoreTaskRuntime<
I extends JsonValue,
C extends TaskCheckpoint,
O extends OutputState = never,
H extends HookPoints = EmptyHookPoints,
> = TaskRuntimeBase<H> & CommitRuntime<I, C, O, CoreTaskTx<C>> & RuntimeOutput<O>;
export type FinalTx<C extends TaskCheckpoint, O extends OutputState> = BaseTaskTx<C> & FinalOutput<O>;
export type CoreFinalTx<C extends TaskCheckpoint, O extends OutputState> = CoreTaskTx<C> & FinalOutput<O>;
export type TerminalClosure<
I extends JsonValue,
C extends TaskCheckpoint,
R extends JsonValue,
F extends JsonValue,
O extends OutputState,
> = (tx: FinalTx<C, O>, current: RunningTask<I, C, O>) => Completion<R, F> | Promise<Completion<R, F>>;
export type CoreTerminalClosure<
I extends JsonValue,
C extends TaskCheckpoint,
R extends JsonValue,
F extends JsonValue,
O extends OutputState,
> = (tx: CoreFinalTx<C, O>, current: RunningTask<I, C, O>) => Completion<R, F> | Promise<Completion<R, F>>;
export interface AbortTx<C extends TaskCheckpoint> extends TxReaders, BoundTxState {
checkpoint(value: C): void;
markTask(id: Id): Promise<"marked" | "terminal">;
}
export interface CoreAbortTx<C extends TaskCheckpoint> extends AbortTx<C> {
entry<E extends Entry>(kind: EntryKind<E>, conversationId: Id, input: EntryInput<E>): Id;
write<E extends Entry>(
conversationId: Id,
kind: EntryKind<E>,
input: EntryInput<E>,
requestId?: string,
): Promise<Acceptance>;
}
export type AbortFinalTx<C extends TaskCheckpoint, O extends OutputState> = AbortTx<C> & FinalOutput<O>;
export type CoreAbortFinalTx<C extends TaskCheckpoint, O extends OutputState> = CoreAbortTx<C> & FinalOutput<O>;
export type AbortClosure<I extends JsonValue, C extends TaskCheckpoint, A extends JsonValue, O extends OutputState> = (
tx: AbortFinalTx<C, O>,
current: RunningTask<I, C, O>,
) => A | Promise<A>;
export type CoreAbortClosure<
I extends JsonValue,
C extends TaskCheckpoint,
A extends JsonValue,
O extends OutputState,
> = (tx: CoreAbortFinalTx<C, O>, current: RunningTask<I, C, O>) => A | Promise<A>;
interface AbortRuntimeBase<H extends HookPoints> {
readonly taskId: Id;
scratch<T>(read: (scratch: ScratchReader) => T | Promise<T>, ctx: Context): Promise<T>;
readonly hooks: HookRunner<H>;
conversation(id: Id, ctx: Context): Promise<AbortTaskConversation | undefined>;
waitForTask(id: Id, ctx: Context): Promise<Task>;
abortTask(id: Id, ctx: Context): Promise<"marked" | "terminal">;
now(): number;
sleep(untilMs: number, ctx: Context): Promise<void>;
}
export type AbortTaskRuntime<
I extends JsonValue,
C extends TaskCheckpoint,
O extends OutputState = never,
H extends HookPoints = EmptyHookPoints,
> = AbortRuntimeBase<H> & CommitRuntime<I, C, O, AbortTx<C>> & AbortRuntimeOutput<O>;
export type CoreAbortTaskRuntime<
I extends JsonValue,
C extends TaskCheckpoint,
O extends OutputState = never,
H extends HookPoints = EmptyHookPoints,
> = AbortRuntimeBase<H> & CommitRuntime<I, C, O, CoreAbortTx<C>> & AbortRuntimeOutput<O>;
export type ToolInput = { readonly assistant: Id; readonly call: Stored<ToolCall> };
export type GenerationOutput = { message?: Stored<AssistantMessage> };
export type ToolOutputState = { progress?: string; log: string[]; details?: JsonValue };

View file

@ -1,440 +0,0 @@
import { AsyncLocalStorage } from "node:async_hooks";
import type { Context } from "@earendil-works/chord";
import { withoutAbortSignal } from "@earendil-works/chord/context";
import type { Element, List, PayloadOf, ResolvedValue, Value } from "./addresses.ts";
import type { Id, JsonValue } from "./core.ts";
import type { Conversation, Entry } from "./entries.ts";
import type { ConversationScan, EntryScan, NewTask, Page, Storage, TaskScan, Write } from "./storage.ts";
import type { Task } from "./tasks.ts";
export class Closed extends Error {
constructor() {
super("Session is closed");
this.name = "Closed";
}
}
export class Faulted extends Error {
constructor() {
super("Session is faulted");
this.name = "Faulted";
}
}
export class NestedLineOperation extends Error {
constructor() {
super("Session line operations cannot be nested");
this.name = "NestedLineOperation";
}
}
export class ReadAfterWrite extends Error {
constructor() {
super("Transaction reads must precede writes");
this.name = "ReadAfterWrite";
}
}
export type TaskCreate = Omit<NewTask, "id" | "output"> & {
readonly output?: { readonly id?: Id; readonly kind: string };
};
interface LineInvocation {
active: boolean;
}
const lineInvocation = new AsyncLocalStorage<LineInvocation>();
const storageOwners = new WeakMap<Storage, object>();
function assertContextActive(ctx: Context): void {
ctx.abortSignal?.throwIfAborted();
}
function resolveStoredValue<D extends Value<JsonValue>>(
address: D,
stored: PayloadOf<D> | undefined,
): ResolvedValue<D> {
const fallback = address.default as PayloadOf<D> | undefined;
return (stored === undefined ? fallback : stored) as ResolvedValue<D>;
}
export interface Transaction {
getConversation(id: Id): Promise<Conversation | undefined>;
getEntry(id: Id): Promise<Entry | undefined>;
getEntries(ids: readonly Id[]): Promise<ReadonlyMap<Id, Entry>>;
getTask(id: Id): Promise<Task | undefined>;
getTasks(ids: readonly Id[]): Promise<ReadonlyMap<Id, Task>>;
createConversation(conversation: Omit<Conversation, "id">): Id;
entry(entry: Omit<Entry, "id">): Id;
task(task: TaskCreate): Id;
setTask(task: Task): void;
value<D extends Value<JsonValue>>(
address: D,
): {
get(at?: Id): Promise<ResolvedValue<D>>;
set(value: PayloadOf<D>): void;
delete(): void;
};
list<T extends JsonValue>(
address: List<T>,
): {
read(at?: Id): Promise<readonly Element<T>[]>;
append(value: T): Id;
remove(id: Id): void;
clear(): void;
};
}
class TransactionState {
private readonly storage: Storage;
private readonly ctx: Context;
private active = true;
private writing = false;
private entryAppended = false;
private pendingReads = 0;
private failure: Error | undefined;
private readonly writes: Write[] = [];
constructor(storage: Storage, ctx: Context) {
this.storage = storage;
this.ctx = ctx;
}
facade(): Transaction {
return Object.freeze({
getConversation: (id: Id) => this.getConversation(id),
getEntry: (id: Id) => this.getEntry(id),
getEntries: (ids: readonly Id[]) => this.getEntries(ids),
getTask: (id: Id) => this.getTask(id),
getTasks: (ids: readonly Id[]) => this.getTasks(ids),
createConversation: (conversation: Omit<Conversation, "id">) => this.createConversation(conversation),
entry: (entry: Omit<Entry, "id">) => this.entry(entry),
task: (task: TaskCreate) => this.task(task),
setTask: (task: Task) => this.setTask(task),
value: <D extends Value<JsonValue>>(address: D) => this.value(address),
list: <T extends JsonValue>(address: List<T>) => this.list(address),
});
}
finish(): readonly Write[] {
this.assertActive();
if (this.pendingReads !== 0) this.poison(new Error("Transaction has pending reads"));
this.active = false;
return this.writes;
}
seal(): void {
this.active = false;
}
async getConversation(id: Id): Promise<Conversation | undefined> {
return (await this.read(() => this.storage.getConversations([id], this.ctx))).get(id);
}
async getEntry(id: Id): Promise<Entry | undefined> {
return (await this.getEntries([id])).get(id);
}
getEntries(ids: readonly Id[]): Promise<ReadonlyMap<Id, Entry>> {
return this.read(() => this.storage.getEntries(ids, this.ctx));
}
async getTask(id: Id): Promise<Task | undefined> {
return (await this.getTasks([id])).get(id);
}
getTasks(ids: readonly Id[]): Promise<ReadonlyMap<Id, Task>> {
return this.read(() => this.storage.getTasks(ids, this.ctx));
}
createConversation(conversation: Omit<Conversation, "id">): Id {
const id = this.allocate();
this.writes.push({ type: "conversation.create", conversation: { ...conversation, id } });
return id;
}
entry(entry: Omit<Entry, "id">): Id {
const id = this.allocate();
this.writes.push({ type: "entry.append", entry: { ...entry, id } });
this.entryAppended = true;
return id;
}
task(task: TaskCreate): Id {
const id = this.allocate();
const { output: requestedOutput, ...base } = task;
const output =
requestedOutput === undefined ? undefined : { id: requestedOutput.id ?? id, kind: requestedOutput.kind };
this.writes.push({
type: "task.create",
task: { ...base, id, ...(output === undefined ? {} : { output }) },
});
return id;
}
setTask(task: Task): void {
this.write({ type: "task.set", task });
}
value<D extends Value<JsonValue>>(
address: D,
): {
get(at?: Id): Promise<ResolvedValue<D>>;
set(value: PayloadOf<D>): void;
delete(): void;
} {
this.assertActive();
return Object.freeze({
get: (at?: Id) =>
this.read(async () => {
const stored = await this.storage.getValue(address as unknown as Value<PayloadOf<D>>, at, this.ctx);
return resolveStoredValue(address, stored);
}),
set: (value: PayloadOf<D>) => {
this.assertStateWriteOrder(address);
this.write({ type: "value.set", address, value });
},
delete: () => {
this.assertStateWriteOrder(address);
this.write({ type: "value.delete", address });
},
});
}
list<T extends JsonValue>(
address: List<T>,
): {
read(at?: Id): Promise<readonly Element<T>[]>;
append(value: T): Id;
remove(id: Id): void;
clear(): void;
} {
this.assertActive();
return Object.freeze({
read: (at?: Id) => this.read(() => this.storage.readList(address, at, this.ctx)),
append: (value: T) => {
this.assertStateWriteOrder(address);
const id = this.allocate();
this.writes.push({ type: "list.append", address, element: { id, value } });
return id;
},
remove: (elementId: Id) => {
this.assertStateWriteOrder(address);
this.write({ type: "list.remove", address, elementId });
},
clear: () => {
this.assertStateWriteOrder(address);
this.write({ type: "list.clear", address });
},
});
}
private assertStateWriteOrder(address: Value<JsonValue> | List<JsonValue>): void {
this.assertActive();
if (address.rewind && this.entryAppended) {
this.poison(new Error("Rewindable state writes must precede entry appends"));
}
}
private allocate(): Id {
this.assertWritable();
return this.storage.nextId();
}
private write(write: Write): void {
this.assertWritable();
this.writes.push(write);
}
private async read<T>(read: () => Promise<T>): Promise<T> {
this.assertReadable();
this.pendingReads++;
try {
return await read();
} finally {
this.pendingReads--;
}
}
private assertReadable(): void {
this.assertActive();
if (this.writing) this.poison(new ReadAfterWrite());
}
private assertWritable(): void {
this.assertActive();
if (this.pendingReads !== 0) this.poison(new Error("Transaction writes must await reads"));
this.writing = true;
}
private assertActive(): void {
if (!this.active) throw new Error("Transaction is closed");
if (this.failure !== undefined) throw this.failure;
}
private poison(error: Error): never {
this.failure ??= error;
throw this.failure;
}
}
export class Session {
private readonly storage: Storage;
private readonly ownership = Object.freeze({});
private tail: Promise<void> = Promise.resolve();
private fault: Faulted | undefined;
private closed = false;
private closeCompletion: Promise<void> | undefined;
private storageCloseCompletion: Promise<void> | undefined;
constructor(storage: Storage) {
if (storageOwners.has(storage)) throw new Error("Storage is already owned by another Session");
storageOwners.set(storage, this.ownership);
this.storage = storage;
}
commit<T>(build: (tx: Transaction) => T | Promise<T>, ctx: Context): Promise<T> {
try {
assertContextActive(ctx);
} catch (error) {
return Promise.reject(error);
}
return this.onLine(async () => {
this.assertUsable();
assertContextActive(ctx);
const tx = new TransactionState(this.storage, ctx);
try {
const result = await build(tx.facade());
const writes = tx.finish();
if (writes.length !== 0) {
try {
await this.storage.commit(writes, withoutAbortSignal(ctx));
} catch (error) {
this.fault = new Faulted();
try {
await this.closeStorage(ctx);
} catch {}
throw error;
}
}
return result;
} finally {
tx.seal();
}
});
}
close(ctx: Context): Promise<void> {
if (this.nested()) return Promise.reject(new NestedLineOperation());
this.closeCompletion ??= this.enqueueLine(async () => {
this.closed = true;
await this.closeStorage(ctx);
});
return this.closeCompletion;
}
getConversations(ids: readonly Id[], ctx: Context): Promise<ReadonlyMap<Id, Conversation>> {
return this.onLine(() => {
this.assertUsable();
return this.storage.getConversations(ids, ctx);
});
}
scanConversations(query: ConversationScan, ctx: Context): Promise<Page<Conversation>> {
return this.onLine(() => {
this.assertUsable();
return this.storage.scanConversations(query, ctx);
});
}
getEntries(ids: readonly Id[], ctx: Context): Promise<ReadonlyMap<Id, Entry>> {
return this.onLine(() => {
this.assertUsable();
return this.storage.getEntries(ids, ctx);
});
}
scanEntries(query: EntryScan, ctx: Context): Promise<Page<Entry>> {
return this.onLine(() => {
this.assertUsable();
return this.storage.scanEntries(query, ctx);
});
}
newestHead(conversationId: Id, at: Id, ctx: Context): Promise<Entry | undefined> {
return this.onLine(() => {
this.assertUsable();
return this.storage.newestHead(conversationId, at, ctx);
});
}
getTasks(ids: readonly Id[], ctx: Context): Promise<ReadonlyMap<Id, Task>> {
return this.onLine(() => {
this.assertUsable();
return this.storage.getTasks(ids, ctx);
});
}
scanTasks(query: TaskScan, ctx: Context): Promise<Page<Task>> {
return this.onLine(() => {
this.assertUsable();
return this.storage.scanTasks(query, ctx);
});
}
getValue<D extends Value<JsonValue>>(address: D, at: Id | undefined, ctx: Context): Promise<ResolvedValue<D>> {
return this.onLine(async () => {
this.assertUsable();
const stored = await this.storage.getValue(address as unknown as Value<PayloadOf<D>>, at, ctx);
return resolveStoredValue(address, stored);
});
}
readList<T extends JsonValue>(address: List<T>, at: Id | undefined, ctx: Context): Promise<readonly Element<T>[]> {
return this.onLine(() => {
this.assertUsable();
return this.storage.readList(address, at, ctx);
});
}
private onLine<T>(operation: () => T | Promise<T>): Promise<T> {
if (this.nested()) return Promise.reject(new NestedLineOperation());
return this.enqueueLine(operation);
}
private enqueueLine<T>(operation: () => T | Promise<T>): Promise<T> {
const previous = this.tail;
let release!: () => void;
this.tail = new Promise((resolve) => {
release = resolve;
});
return (async () => {
await previous;
const invocation: LineInvocation = { active: true };
try {
return await lineInvocation.run(invocation, operation);
} finally {
invocation.active = false;
release();
}
})();
}
private nested(): boolean {
return lineInvocation.getStore()?.active === true;
}
private closeStorage(ctx: Context): Promise<void> {
this.storageCloseCompletion ??= Promise.resolve()
.then(() => this.storage.close(withoutAbortSignal(ctx)))
.then(() => {
if (storageOwners.get(this.storage) === this.ownership) storageOwners.delete(this.storage);
});
return this.storageCloseCompletion;
}
private assertUsable(): void {
if (this.fault !== undefined) throw this.fault;
if (this.closed) throw new Closed();
}
}

View file

@ -1,104 +0,0 @@
import type { Context } from "@earendil-works/chord";
import type { Element, List, Value } from "./addresses.ts";
import type { Id, JsonValue, Seq } from "./core.ts";
import type { Conversation, Entry, SectionRecord } from "./entries.ts";
import type { Task, TaskBase } from "./tasks.ts";
export interface PageQuery {
/** Last item returned by the same scan. */
readonly cursor?: Id;
readonly limit: number;
}
export interface Page<T> {
readonly items: readonly T[];
readonly next?: Id;
}
export class ScratchRetired extends Error {
readonly taskId: Id;
constructor(taskId: Id) {
super(`Scratch for task ${taskId} is retired`);
this.name = "ScratchRetired";
this.taskId = taskId;
}
}
export class OutputRetired extends Error {
readonly outputId: Id;
constructor(outputId: Id) {
super(`Task output ${outputId} is retired`);
this.name = "OutputRetired";
this.outputId = outputId;
}
}
export class InvalidHistoryPosition extends Error {
readonly at: Id;
constructor(at: Id) {
super(`Entry ${at} is not a valid history position for this address`);
this.name = "InvalidHistoryPosition";
this.at = at;
}
}
export type NewTask = Omit<TaskBase, "abort" | "checkpoint" | "owns"> & {
readonly status: "pending";
readonly owns: readonly [];
readonly abort?: never;
readonly checkpoint?: never;
readonly outcome?: never;
};
export type StateWrite =
| { readonly type: "value.set"; readonly address: Value<JsonValue>; readonly value: JsonValue }
| { readonly type: "value.delete"; readonly address: Value<JsonValue> }
| { readonly type: "list.append"; readonly address: List<JsonValue>; readonly element: Element<JsonValue> }
| { readonly type: "list.remove"; readonly address: List<JsonValue>; readonly elementId: Id }
| { readonly type: "list.clear"; readonly address: List<JsonValue> };
export type StoredConversation = Conversation & { readonly sectionSeed?: readonly SectionRecord[] };
export type Write =
| StateWrite
| { readonly type: "conversation.create"; readonly conversation: StoredConversation }
| { readonly type: "entry.append"; readonly entry: Entry }
| { readonly type: "task.create"; readonly task: NewTask }
| { readonly type: "task.set"; readonly task: Task };
export type ConversationScan = PageQuery & {
readonly parent?: Id;
readonly owner?: Id;
};
export type EntryScan = PageQuery & {
readonly conversationId: Id;
readonly kind?: string;
readonly through?: Id;
};
export type TaskScan = PageQuery & {
readonly conversationIds?: readonly Id[];
readonly statuses?: readonly Task["status"][];
readonly kind?: string;
readonly abort?: boolean;
readonly outputId?: Id;
};
export interface Storage {
nextId(): Id;
commit(writes: readonly Write[], ctx: Context): Promise<readonly Seq[]>;
getConversations(ids: readonly Id[], ctx: Context): Promise<ReadonlyMap<Id, Conversation>>;
scanConversations(query: ConversationScan, ctx: Context): Promise<Page<Conversation>>;
getEntries(ids: readonly Id[], ctx: Context): Promise<ReadonlyMap<Id, Entry>>;
scanEntries(query: EntryScan, ctx: Context): Promise<Page<Entry>>;
newestHead(conversationId: Id, at: Id, ctx: Context): Promise<Entry | undefined>;
getTasks(ids: readonly Id[], ctx: Context): Promise<ReadonlyMap<Id, Task>>;
scanTasks(query: TaskScan, ctx: Context): Promise<Page<Task>>;
getValue<T extends JsonValue>(address: Value<T>, at: Id | undefined, ctx: Context): Promise<T | undefined>;
readList<T extends JsonValue>(address: List<T>, at: Id | undefined, ctx: Context): Promise<readonly Element<T>[]>;
close(ctx: Context): Promise<void>;
}

View file

@ -1,312 +0,0 @@
import type { Context } from "@earendil-works/chord";
import type { ConversationValueDefinition } from "./addresses.ts";
import type { Id, JsonObject, JsonValue } from "./core.ts";
import type { EmptyHookPoints, HookPoints } from "./hooks.ts";
import type {
AbortClosure,
AbortTaskRuntime,
CoreAbortClosure,
CoreAbortTaskRuntime,
CoreTaskRuntime,
CoreTerminalClosure,
TaskRuntime,
TerminalClosure,
} from "./runtime.ts";
export interface TaskCheckpoint extends JsonObject {
readonly phase: string;
}
export type NoCheckpoint = never;
export type OutputState = JsonObject | JsonValue[];
export type ConfigBundle = Record<string, ConversationValueDefinition<JsonValue>>;
export type ReservedConfigKeys = "set" | "get";
export type KeysOf<U> = U extends unknown ? keyof U : never;
export type Collides<M, N extends keyof M, Reserved extends PropertyKey> = [
Extract<keyof M[N], KeysOf<M[Exclude<keyof M, N>]> | Reserved>,
] extends [never]
? false
: true;
export type AnyCollision<M, Reserved extends PropertyKey> = true extends {
[N in keyof M]: Collides<M, N, Reserved>;
}[keyof M]
? true
: false;
export type DisjointBundles<M, Reserved extends PropertyKey = never> = AnyCollision<M, Reserved> extends false
? true
: false;
declare const taskOutputType: unique symbol;
const taskKindBrand: unique symbol = Symbol("pico.taskKind");
export interface TaskOutputKind<O extends OutputState> {
readonly kind: string;
readonly [taskOutputType]?: O;
}
export function defineTaskOutput<O extends OutputState>(kind: string): TaskOutputKind<O> {
return Object.freeze({ kind });
}
export interface StoredTaskOutputRef {
readonly id: Id;
readonly kind: string;
}
export interface TaskOutputRef<O extends OutputState> extends StoredTaskOutputRef {
readonly [taskOutputType]?: O;
}
export type TaskOutputField<O extends OutputState> = [O] extends [never]
? Record<never, never>
: { readonly output: TaskOutputRef<O> };
export type TaskOutputDefinition<I extends JsonValue, O extends OutputState> = [O] extends [never]
? { readonly output?: never }
: { readonly output: { readonly kind: TaskOutputKind<O>; initial(input: I): O } };
export type TaskOutcome<
R extends JsonValue = JsonValue,
F extends JsonValue = JsonValue,
A extends JsonValue = JsonValue,
> =
| { readonly status: "completed"; readonly result: R }
| { readonly status: "failed"; readonly failure: F }
| { readonly status: "aborted"; readonly result: A }
| { readonly status: "orphaned" };
export interface TaskBase<I extends JsonValue = JsonValue, C extends TaskCheckpoint = TaskCheckpoint> {
readonly id: Id;
readonly conversationId: Id;
readonly kind: string;
readonly input: I;
readonly checkpoint?: C;
readonly after: readonly Id[];
readonly background?: true;
readonly owns: readonly Id[];
readonly output?: StoredTaskOutputRef;
readonly abort?: true;
}
export type Task<
I extends JsonValue = JsonValue,
C extends TaskCheckpoint = TaskCheckpoint,
R extends JsonValue = JsonValue,
F extends JsonValue = JsonValue,
A extends JsonValue = JsonValue,
> = TaskBase<I, C> &
(
| { readonly status: "pending" | "running"; readonly outcome?: never }
| { readonly status: "terminal"; readonly outcome: TaskOutcome<R, F, A> }
);
export type RunningTask<I extends JsonValue, C extends TaskCheckpoint, O extends OutputState = never> = Omit<
TaskBase<I, C>,
"output"
> &
TaskOutputField<O> & {
readonly status: "running";
readonly outcome?: never;
};
export type Completion<R extends JsonValue, F extends JsonValue> =
| { readonly status: "completed"; readonly result: R }
| { readonly status: "failed"; readonly failure: F };
interface TaskKindMethods<
I extends JsonValue,
C extends TaskCheckpoint,
R extends JsonValue,
F extends JsonValue,
A extends JsonValue,
O extends OutputState,
H extends HookPoints,
> {
readonly kind: string;
readonly hooks?: H;
readonly config?: ConfigBundle;
execute(
task: RunningTask<I, C, O>,
runtime: TaskRuntime<I, C, O, H>,
ctx: Context,
): Promise<TerminalClosure<I, C, R, F, O>>;
recover(
task: RunningTask<I, C, O>,
runtime: TaskRuntime<I, C, O, H>,
ctx: Context,
): Promise<TerminalClosure<I, C, R, F, O>>;
abort(
task: RunningTask<I, C, O>,
runtime: AbortTaskRuntime<I, C, O, H>,
ctx: Context,
): Promise<AbortClosure<I, C, A, O>>;
}
interface CoreTaskKindMethods<
I extends JsonValue,
C extends TaskCheckpoint,
R extends JsonValue,
F extends JsonValue,
A extends JsonValue,
O extends OutputState,
H extends HookPoints,
> {
readonly kind: string;
readonly hooks?: H;
readonly config?: ConfigBundle;
execute(
task: RunningTask<I, C, O>,
runtime: CoreTaskRuntime<I, C, O, H>,
ctx: Context,
): Promise<CoreTerminalClosure<I, C, R, F, O>>;
recover(
task: RunningTask<I, C, O>,
runtime: CoreTaskRuntime<I, C, O, H>,
ctx: Context,
): Promise<CoreTerminalClosure<I, C, R, F, O>>;
abort(
task: RunningTask<I, C, O>,
runtime: CoreAbortTaskRuntime<I, C, O, H>,
ctx: Context,
): Promise<CoreAbortClosure<I, C, A, O>>;
}
export type TaskKind<
I extends JsonValue,
C extends TaskCheckpoint,
R extends JsonValue,
F extends JsonValue,
A extends JsonValue,
O extends OutputState = never,
H extends HookPoints = EmptyHookPoints,
> = TaskKindMethods<I, C, R, F, A, O, H> & TaskOutputDefinition<I, O> & { readonly [taskKindBrand]: "ordinary" };
export type CoreTaskKind<
I extends JsonValue,
C extends TaskCheckpoint,
R extends JsonValue,
F extends JsonValue,
A extends JsonValue,
O extends OutputState = never,
H extends HookPoints = EmptyHookPoints,
> = CoreTaskKindMethods<I, C, R, F, A, O, H> & TaskOutputDefinition<I, O> & { readonly [taskKindBrand]: "core" };
export type NoExtra<Expected, Actual extends Expected> = Actual & Record<Exclude<keyof Actual, keyof Expected>, never>;
export type ExactJsonInput<Expected extends JsonValue, Actual extends Expected> = Expected extends readonly JsonValue[]
? Actual
: Expected extends JsonObject
? NoExtra<Expected, Actual>
: Actual;
export type TaskKindDefinition<
I extends JsonValue,
C extends TaskCheckpoint,
R extends JsonValue,
F extends JsonValue,
A extends JsonValue,
O extends OutputState,
H extends HookPoints,
> = Omit<TaskKind<I, C, R, F, A, O, H>, typeof taskKindBrand>;
export type CoreTaskKindDefinition<
I extends JsonValue,
C extends TaskCheckpoint,
R extends JsonValue,
F extends JsonValue,
A extends JsonValue,
O extends OutputState,
H extends HookPoints,
> = Omit<CoreTaskKind<I, C, R, F, A, O, H>, typeof taskKindBrand>;
export type TaskKindFactory<
I extends JsonValue,
C extends TaskCheckpoint,
R extends JsonValue,
F extends JsonValue,
A extends JsonValue,
O extends OutputState,
> = <H extends HookPoints, D extends TaskKindDefinition<I, C, R, F, A, O, H>>(
definition: NoExtra<TaskKindDefinition<I, C, R, F, A, O, H>, D> & { readonly hooks?: H },
) => D & TaskKind<I, C, R, F, A, O, H>;
export type CoreTaskKindFactory<
I extends JsonValue,
C extends TaskCheckpoint,
R extends JsonValue,
F extends JsonValue,
A extends JsonValue,
O extends OutputState,
> = <H extends HookPoints, D extends CoreTaskKindDefinition<I, C, R, F, A, O, H>>(
definition: NoExtra<CoreTaskKindDefinition<I, C, R, F, A, O, H>, D> & { readonly hooks?: H },
) => D & CoreTaskKind<I, C, R, F, A, O, H>;
export function defineTask<
I extends JsonValue,
C extends TaskCheckpoint,
R extends JsonValue,
F extends JsonValue,
A extends JsonValue,
O extends OutputState = never,
>(): TaskKindFactory<I, C, R, F, A, O> {
const define = <H extends HookPoints, D extends TaskKindDefinition<I, C, R, F, A, O, H>>(
definition: NoExtra<TaskKindDefinition<I, C, R, F, A, O, H>, D> & { readonly hooks?: H },
): D & TaskKind<I, C, R, F, A, O, H> =>
Object.freeze({
...definition,
[taskKindBrand]: "ordinary" as const,
}) as D & TaskKind<I, C, R, F, A, O, H>;
return define as TaskKindFactory<I, C, R, F, A, O>;
}
/** Internal authoring helper for Pico's fixed privileged task kinds. */
export function defineCoreTask<
I extends JsonValue,
C extends TaskCheckpoint,
R extends JsonValue,
F extends JsonValue,
A extends JsonValue,
O extends OutputState = never,
>(): CoreTaskKindFactory<I, C, R, F, A, O> {
const define = <H extends HookPoints, D extends CoreTaskKindDefinition<I, C, R, F, A, O, H>>(
definition: NoExtra<CoreTaskKindDefinition<I, C, R, F, A, O, H>, D> & { readonly hooks?: H },
): D & CoreTaskKind<I, C, R, F, A, O, H> =>
Object.freeze({
...definition,
[taskKindBrand]: "core" as const,
}) as D & CoreTaskKind<I, C, R, F, A, O, H>;
return define as CoreTaskKindFactory<I, C, R, F, A, O>;
}
export type OrdinaryKind = {
readonly kind: string;
readonly [taskKindBrand]: "ordinary";
readonly hooks?: HookPoints;
readonly config?: ConfigBundle;
};
export type CoreKind = {
readonly kind: string;
readonly [taskKindBrand]: "core";
readonly hooks?: HookPoints;
readonly config?: ConfigBundle;
};
export type AnyDefinedKind = OrdinaryKind | CoreKind;
export type PayloadsOf<K> = K extends TaskKind<infer I, infer C, infer R, infer F, infer A, infer O, infer H>
? { input: I; checkpoint: C; result: R; failure: F; aborted: A; output: O; hooks: H }
: K extends CoreTaskKind<infer I, infer C, infer R, infer F, infer A, infer O, infer H>
? { input: I; checkpoint: C; result: R; failure: F; aborted: A; output: O; hooks: H }
: never;
export type InputOf<K> = PayloadsOf<K>["input"];
export type CheckpointOf<K> = PayloadsOf<K>["checkpoint"];
export type ResultOf<K> = PayloadsOf<K>["result"];
export type FailureOf<K> = PayloadsOf<K>["failure"];
export type AbortedOf<K> = PayloadsOf<K>["aborted"];
export type OutputOf<K> = PayloadsOf<K>["output"];
export type DistributiveOmit<T, K extends PropertyKey> = T extends unknown ? Omit<T, K> : never;
export type TaskOf<K> = DistributiveOmit<
Task<InputOf<K>, CheckpointOf<K>, ResultOf<K>, FailureOf<K>, AbortedOf<K>>,
"output"
> &
TaskOutputField<OutputOf<K>>;

File diff suppressed because it is too large Load diff

View file

@ -1,7 +0,0 @@
{
"extends": "../../../../../tsconfig.json",
"compilerOptions": {
"noEmit": true
},
"include": ["../../../src/harness/pico/**/*.ts", "./types.compile.ts"]
}

View file

@ -1,545 +0,0 @@
import type { Context } from "@earendil-works/chord";
import type {
AssistantMessage,
DeferredHandle,
Message,
RetryPolicy,
ToolCall,
ToolResultMessage,
Usage,
UserMessage,
} from "@earendil-works/pi-ai";
import { defineInternalList, internalConversationValue } from "../../../src/harness/pico/addresses.ts";
import type {
Acceptance,
AssistantEntryData,
BaseTaskTx,
BoundTxState,
ConversationTx,
Entry,
EntryBase,
EntryData,
EntryInput,
ExactJsonInput,
GenerationOutput,
HookHandler,
HookPoint,
HookResult,
HostTx,
InputOutcome,
JsonObject,
JsonValue,
List,
OutputState,
PayloadsOf,
PublicScope,
QueuedInput,
ReadonlyTaskOutput,
ResolvedValue,
RunningTask,
ScratchReader,
SectionRecord,
Stored,
StoredEntryDraft,
StoredInputResult,
SummaryEntry,
SystemEntryData,
TaskCheckpoint,
TaskKind,
TaskOf,
TaskOutput,
TaskOutputRef,
TaskRuntime,
TerminalClosure,
ToolControl,
ToolDiagnostic,
ToolInput,
ToolOutputState,
ToolResultData,
TxState,
UsageEntryData,
UserInput,
} from "../../../src/harness/pico/index.ts";
import * as Pico from "../../../src/harness/pico/index.ts";
import {
conversationList,
conversationValue,
defineEntry,
defineHookPoint,
defineList,
defineTask,
defineTaskOutput,
defineValue,
sessionValue,
} from "../../../src/harness/pico/index.ts";
import type { CoreAbortTaskRuntime, CoreTaskRuntime, CoreTaskTx } from "../../../src/harness/pico/runtime.ts";
import type { DisjointBundles } from "../../../src/harness/pico/tasks.ts";
import * as TaskDeclarations from "../../../src/harness/pico/tasks.ts";
import { defineCoreTask } from "../../../src/harness/pico/tasks.ts";
// @ts-expect-error privileged core authoring is not part of the public Pico surface
const publicDefineCoreTask = Pico.defineCoreTask;
// @ts-expect-error the task-kind brand is private even within the tasks module
const publicTaskKindBrand = TaskDeclarations.taskKindBrand;
void publicDefineCoreTask;
void publicTaskKindBrand;
type RemovedPublicExports = [
// @ts-expect-error low-level Session is internal
Pico.Session,
// @ts-expect-error low-level transaction is internal
Pico.Transaction,
// @ts-expect-error low-level task creation shape is internal
Pico.TaskCreate,
// @ts-expect-error stored conversation metadata is internal
Pico.StoredConversation,
// @ts-expect-error core kind construction types are internal
Pico.CoreTaskKind,
// @ts-expect-error erased core kind is internal
Pico.CoreKind,
// @ts-expect-error collision implementation helpers are internal
Pico.DisjointBundles,
];
declare const removedPublicExports: RemovedPublicExports;
void removedPublicExports;
type Equal<Left, Right> = (<Type>() => Type extends Left ? 1 : 2) extends <Type>() => Type extends Right ? 1 : 2
? (<Type>() => Type extends Right ? 1 : 2) extends <Type>() => Type extends Left ? 1 : 2
? true
: false
: false;
type Assert<Condition extends true> = Condition;
type ExtendsJson<T extends JsonValue> = T;
type ExtendsJsonObject<T extends JsonObject> = T;
type StoredAssertions = [
ExtendsJsonObject<Stored<UserMessage>>,
ExtendsJsonObject<Stored<AssistantMessage>>,
ExtendsJsonObject<Stored<ToolResultMessage>>,
ExtendsJsonObject<Stored<ToolCall>>,
ExtendsJsonObject<Stored<Usage>>,
ExtendsJsonObject<Stored<DeferredHandle>>,
ExtendsJsonObject<Stored<RetryPolicy>>,
ExtendsJson<Stored<UserInput>>,
ExtendsJson<ToolInput>,
ExtendsJson<UsageEntryData>,
ExtendsJson<QueuedInput>,
ExtendsJson<StoredEntryDraft>,
ExtendsJson<SystemEntryData>,
ExtendsJson<AssistantEntryData>,
ExtendsJson<ToolResultData>,
ExtendsJson<Acceptance>,
ExtendsJson<SectionRecord>,
ExtendsJson<ToolControl>,
ExtendsJson<ToolDiagnostic>,
Assert<GenerationOutput extends OutputState ? true : false>,
Assert<ToolOutputState extends OutputState ? true : false>,
Assert<Date extends JsonValue ? false : true>,
Assert<{ readonly value: undefined } extends JsonValue ? false : true>,
];
type Input = { readonly prompt: string; readonly nested: { readonly count: number } };
type Checkpoint = { readonly phase: "ready"; readonly cursor: number };
type Result = { readonly answer: string };
type Failure = { readonly code: "failed" };
type Aborted = { readonly cleaned: boolean };
type Progress = { text: string; chunks: string[] };
const pluginEnabled = conversationValue<boolean>("plugin.enabled", { rewind: false, default: false });
const collected = defineHookPoint<{ readonly value: string }, { readonly seen: true }>({
fold: "collect",
onThrow: "skip",
});
const first = defineHookPoint<{ readonly value: string }, { readonly accepted: true }>({
fold: "first",
onThrow: "skip",
});
const beforeRun = defineHookPoint<{ readonly prompt: string }, { readonly prompt?: string }>({
fold: "chain",
onThrow: "skip",
});
const noReturnHandler: HookHandler<typeof beforeRun> = async () => {};
void noReturnHandler;
const ordinaryKind = defineTask<Input, Checkpoint, Result, Failure, Aborted>()({
kind: "plugin.ordinary",
config: { enabled: pluginEnabled },
hooks: { beforeRun },
execute() {
return Promise.resolve(() => ({ status: "completed", result: { answer: "done" } }));
},
recover() {
return Promise.resolve(() => ({ status: "failed", failure: { code: "failed" } }));
},
abort() {
return Promise.resolve(() => ({ cleaned: true }));
},
});
const coreKind = defineCoreTask<Input, Checkpoint, Result, Failure, Aborted>()({
kind: "pi.test_core",
execute() {
return Promise.resolve(() => ({ status: "completed", result: { answer: "done" } }));
},
recover() {
return Promise.resolve(() => ({ status: "failed", failure: { code: "failed" } }));
},
abort() {
return Promise.resolve(() => ({ cleaned: true }));
},
});
const progressOutput = defineTaskOutput<Progress>("plugin.progress");
const outputKind = defineTask<Input, Checkpoint, Result, Failure, Aborted, Progress>()({
kind: "plugin.output",
output: { kind: progressOutput, initial: (input) => ({ text: input.prompt, chunks: [] }) },
execute() {
return Promise.resolve(() => ({ status: "completed", result: { answer: "done" } }));
},
recover() {
return Promise.resolve(() => ({ status: "failed", failure: { code: "failed" } }));
},
abort() {
return Promise.resolve(() => ({ cleaned: true }));
},
});
type HookAssertions = [
Assert<Equal<typeof collected, HookPoint<{ readonly value: string }, { readonly seen: true }, "collect">>>,
Assert<Equal<typeof first, HookPoint<{ readonly value: string }, { readonly accepted: true }, "first">>>,
Assert<Equal<typeof beforeRun, HookPoint<{ readonly prompt: string }, { readonly prompt?: string }, "chain">>>,
Assert<
Equal<
HookResult<typeof collected>,
{ readonly outputs: readonly { readonly seen: true }[]; readonly threw?: unknown }
>
>,
Assert<Equal<HookResult<typeof first>, { readonly output?: { readonly accepted: true }; readonly threw?: unknown }>>,
];
type InferredHooks = PayloadsOf<typeof ordinaryKind>["hooks"];
type KindAssertions = [
Assert<Equal<typeof ordinaryKind.config.enabled, typeof pluginEnabled>>,
Assert<Equal<typeof ordinaryKind.hooks.beforeRun, typeof beforeRun>>,
Assert<Equal<keyof InferredHooks, "beforeRun">>,
Assert<Equal<InferredHooks["beforeRun"], typeof beforeRun>>,
Assert<
Equal<
Parameters<typeof ordinaryKind.execute>[1],
TaskRuntime<Input, Checkpoint, never, typeof ordinaryKind.hooks>
>
>,
Assert<Equal<Parameters<typeof coreKind.execute>[1], CoreTaskRuntime<Input, Checkpoint>>>,
Assert<Equal<Parameters<typeof outputKind.execute>[1]["output"], TaskOutput<Progress>>>,
Assert<Equal<Parameters<typeof outputKind.abort>[1]["output"], ReadonlyTaskOutput<Progress>>>,
Assert<
Equal<
Extract<TaskOf<typeof ordinaryKind>, { status: "terminal" }>["outcome"]["status"],
"completed" | "failed" | "aborted" | "orphaned"
>
>,
];
function verifyNoOutputRuntime(
runtime: TaskRuntime<Input, Checkpoint>,
abortRuntime: Parameters<typeof ordinaryKind.abort>[1],
finalTx: Parameters<TerminalClosure<Input, Checkpoint, Result, Failure, never>>[0],
running: RunningTask<Input, Checkpoint>,
ctx: Context,
): void {
// @ts-expect-error no-output runtimes have no output property
void runtime.output;
// @ts-expect-error no-output abort runtimes have no output property
void abortRuntime.output;
// @ts-expect-error no-output final transactions have no output property
void finalTx.output;
// @ts-expect-error no-output running tasks have no output property
void running.output;
void abortRuntime.sleep(1, ctx);
}
function verifyOutput(output: TaskOutput<Progress>, readonlyOutput: ReadonlyTaskOutput<Progress>, ctx: Context): void {
void output.mutate((state) => {
state.chunks.push("next");
}, ctx);
// @ts-expect-error output mutations must be synchronous
void output.mutate(async (state) => {
state.text = "invalid";
}, ctx);
// @ts-expect-error fresh abort receives read-only output
void readonlyOutput.mutate;
}
type JobOutput = {
stdout: string;
stderr: string;
droppedStdout: number;
droppedStderr: number;
exitCode?: number;
occurrence: number;
};
type JobAssertions = [Assert<JobOutput extends OutputState ? true : false>];
function verifyJobMutation(output: TaskOutput<JobOutput>, snapshot: Readonly<JobOutput>, ctx: Context): void {
void output.mutate((state) => {
state.stdout = snapshot.stdout;
}, ctx);
}
type NoteEntry = EntryBase & EntryData<{ readonly text: string }>;
type PinnedEntry = EntryBase &
EntryData<{ readonly text: string }> & {
readonly model: readonly [Stored<UserMessage>];
};
const noteKind = defineEntry<NoteEntry>("plugin.note");
const pinnedKind = defineEntry<PinnedEntry>("plugin.pinned");
const summaryKind = defineEntry<SummaryEntry>("pi.summary");
function verifyEntryInputs(tx: ConversationTx, text: string): void {
void tx.write(noteKind, { data: { text } });
void tx.write(pinnedKind, {
data: { text },
model: [{ role: "user", content: text, timestamp: 1 }],
});
void tx.write(summaryKind, {
data: { through: 1 },
model: [{ role: "user", content: text, timestamp: 1 }],
head: "self",
});
// @ts-expect-error kind is supplied by the witness
void tx.write(noteKind, { kind: "plugin.note", data: { text } });
// @ts-expect-error id is supplied by Pico
void tx.write(noteKind, { id: 1, data: { text } });
}
type EntryAssertions = [
Assert<NoteEntry extends Entry ? true : false>,
Assert<PinnedEntry extends Entry ? true : false>,
Assert<SummaryEntry extends Entry ? true : false>,
Assert<Equal<EntryInput<SummaryEntry>["head"], number | "self">>,
];
function verifyHostTaskCreation(tx: HostTx): void {
tx.task(ordinaryKind, { conversationId: 1, input: { prompt: "host", nested: { count: 1 } } });
// @ts-expect-error host task creation requires a conversation ID
tx.task(ordinaryKind, { input: { prompt: "host", nested: { count: 1 } } });
}
function verifyTaskCreation(
tx: BaseTaskTx<Checkpoint>,
coreTx: CoreTaskTx<Checkpoint>,
outputRef: TaskOutputRef<Progress>,
): void {
tx.task(ordinaryKind, { input: { prompt: "literal", nested: { count: 1 } } });
const exact = { input: { prompt: "variable", nested: { count: 1 } } };
tx.task(ordinaryKind, exact);
tx.task(ordinaryKind, { ...exact });
// @ts-expect-error ordinary transactions cannot create core tasks
tx.task(coreKind, { input: { prompt: "core", nested: { count: 1 } } });
coreTx.task(coreKind, { input: { prompt: "core", nested: { count: 1 } } });
// @ts-expect-error input literals reject visible top-level extras
tx.task(ordinaryKind, { input: { prompt: "extra", nested: { count: 1 }, extra: 1 } });
const extraInput = { prompt: "extra", nested: { count: 1 }, extra: 1 };
// @ts-expect-error input variables reject visible top-level extras
tx.task(ordinaryKind, { input: extraInput });
const inputBase = { prompt: "extra", nested: { count: 1 } };
// @ts-expect-error input spreads reject visible top-level extras
tx.task(ordinaryKind, { input: { ...inputBase, extra: 1 } });
const extraSpec = { input: { prompt: "extra", nested: { count: 1 } }, extra: 1 };
// @ts-expect-error specs reject visible extras
tx.task(ordinaryKind, extraSpec);
// @ts-expect-error spread specs reject visible extras
tx.task(ordinaryKind, { ...extraSpec });
// Deep exactness is outside the contract.
tx.task(ordinaryKind, { input: { prompt: "nested", nested: { count: 1, extension: true } } });
tx.task(outputKind, { input: { prompt: "output", nested: { count: 1 } }, output: outputRef });
// @ts-expect-error no-output kinds reject output refs
tx.task(ordinaryKind, { input: { prompt: "output", nested: { count: 1 } }, output: outputRef });
}
function verifyAddresses(): void {
const rewindable = conversationValue<boolean>("plugin.plan", { rewind: true });
const defaulted = conversationValue<boolean>("plugin.plan", { key: "defaulted", rewind: true, default: false });
const sticky = conversationValue<string>("plugin.mode", { rewind: false });
const events = conversationList<string>("plugin.events", { rewind: true });
const plainSession = sessionValue<string>("plugin.plain-name");
const defaultedSession = sessionValue<string>("plugin.name", { default: "pico" });
const directPlainSession = defineValue<string>({ type: "session" }, "plugin.direct-session");
const directDefaultedSession = defineValue<string>({ type: "session" }, "plugin.direct-session-default", {
default: "session",
});
const directDefaultedTask = defineValue<string>({ type: "task", taskId: 1 }, "plugin.direct-task-default", {
default: "task",
});
const directPlainConversation = defineValue<boolean>(
{ type: "conversation", conversationId: 1 },
"plugin.direct-conversation",
{ rewind: true },
);
const directDefaultedConversation = defineValue<boolean>(
{ type: "conversation", conversationId: 1 },
"plugin.direct-conversation-default",
{ rewind: false, default: false },
);
const bound = rewindable.bind(1);
const defaultedBound = defaulted.bind(1);
const internalDefaulted = internalConversationValue<string, true>("pi.internal-default", {
rewind: true,
default: "internal",
});
const internalDefaultedBound = internalDefaulted.bind(1);
const boundList: List<string> = events.bind(1);
void sticky;
void bound;
void defaultedBound;
void internalDefaulted;
void internalDefaultedBound;
void boundList;
void plainSession;
void defaultedSession;
void directPlainSession;
void directDefaultedSession;
void directDefaultedTask;
void directPlainConversation;
void directDefaultedConversation;
type AddressAssertions = [
Assert<Equal<(typeof rewindable)["rewind"], true>>,
Assert<Equal<(typeof sticky)["rewind"], false>>,
Assert<Equal<ResolvedValue<typeof plainSession>, string | undefined>>,
Assert<Equal<ResolvedValue<typeof defaultedSession>, string>>,
Assert<Equal<ResolvedValue<typeof directPlainSession>, string | undefined>>,
Assert<Equal<ResolvedValue<typeof directDefaultedSession>, string>>,
Assert<Equal<ResolvedValue<typeof directDefaultedTask>, string>>,
Assert<Equal<ResolvedValue<typeof directPlainConversation>, boolean | undefined>>,
Assert<Equal<ResolvedValue<typeof directDefaultedConversation>, boolean>>,
Assert<Equal<(typeof directDefaultedSession)["default"], string>>,
Assert<Equal<(typeof directDefaultedTask)["default"], string>>,
Assert<Equal<(typeof directDefaultedConversation)["default"], boolean>>,
Assert<Equal<ResolvedValue<typeof rewindable>, boolean | undefined>>,
Assert<Equal<ResolvedValue<typeof defaulted>, boolean>>,
Assert<Equal<ResolvedValue<typeof bound>, boolean | undefined>>,
Assert<Equal<ResolvedValue<typeof defaultedBound>, boolean>>,
Assert<Equal<(typeof defaultedBound)["default"], boolean>>,
Assert<Equal<ResolvedValue<typeof internalDefaulted>, string>>,
Assert<Equal<ResolvedValue<typeof internalDefaultedBound>, string>>,
Assert<Equal<(typeof defaultedSession)["default"], string>>,
Assert<Extract<PublicScope, { type: "shared" }> extends never ? true : false>,
];
const assertions = null as unknown as AddressAssertions;
void assertions;
// @ts-expect-error public bound constructors do not accept shared scope
defineValue<string>({ type: "shared", id: 1 }, "plugin.shared");
// @ts-expect-error public bound constructors do not accept shared scope
defineList<string>({ type: "shared", id: 1 }, "plugin.shared");
// @ts-expect-error conversation addresses require an explicit rewind policy
defineValue<string>({ type: "conversation", conversationId: 1 }, "plugin.value");
// @ts-expect-error session addresses are sticky
defineValue<string>({ type: "session" }, "plugin.value", { rewind: true });
defineInternalList<JsonValue>({ type: "shared", id: 1 }, "pi.output");
}
function verifyResolvedTransactionReads(tx: TxState, boundTx: BoundTxState, scratch: ScratchReader): void {
const plainSession = defineValue<string>({ type: "session" }, "plugin.tx-plain");
const defaultedSession = defineValue<string>({ type: "session" }, "plugin.tx-default", { default: "x" });
const defaultedTask = defineValue<number>({ type: "task", taskId: 1 }, "plugin.scratch-default", {
default: 1,
});
const defaultedConversation = defineValue<boolean>(
{ type: "conversation", conversationId: 1 },
"plugin.tx-conversation-default",
{ rewind: true, default: false },
);
const plainDefinition = conversationValue<number>("plugin.bound-plain", { rewind: true });
const defaultedDefinition = conversationValue<number>("plugin.bound-default", { rewind: false, default: 1 });
const plainRead = tx.value(plainSession).get();
const defaultedRead = tx.value(defaultedSession).get();
const defaultedConversationRead = tx.value(defaultedConversation).get();
const plainDefinitionRead = boundTx.value(plainDefinition).get();
const defaultedDefinitionRead = boundTx.value(defaultedDefinition).get();
const scratchRead = scratch.value(defaultedTask).get();
type ReadAssertions = [
Assert<Equal<Awaited<typeof plainRead>, string | undefined>>,
Assert<Equal<Awaited<typeof defaultedRead>, string>>,
Assert<Equal<Awaited<typeof defaultedConversationRead>, boolean>>,
Assert<Equal<Awaited<typeof plainDefinitionRead>, number | undefined>>,
Assert<Equal<Awaited<typeof defaultedDefinitionRead>, number>>,
Assert<Equal<Awaited<typeof scratchRead>, number>>,
];
const assertions = null as unknown as ReadAssertions;
void assertions;
}
type FakeDuplicateConfigs = { one: { value: 1 }; two: { value: 2 } };
type FakeReservedConfig = { one: { set: 1 }; two: Record<never, never> };
type FakeDuplicateHooks = { one: { before: 1 }; two: { before: 2 } };
type RealConfigs = { one: { model: 1 }; two: { retry: 2 } };
type RealHooks = { one: { before: 1 }; two: { after: 2 } };
type BundleAssertions = [Assert<DisjointBundles<RealConfigs, "set" | "get">>, Assert<DisjointBundles<RealHooks>>];
type BadBundleAssertions = [
// @ts-expect-error duplicate bundle keys are rejected by the assertion
Assert<DisjointBundles<FakeDuplicateConfigs, "set" | "get">>,
// @ts-expect-error reserved config keys are rejected by the assertion
Assert<DisjointBundles<FakeReservedConfig, "set" | "get">>,
// @ts-expect-error duplicate hook keys are rejected by the assertion
Assert<DisjointBundles<FakeDuplicateHooks>>,
];
type ExactInputAssertion = Assert<
Equal<ExactJsonInput<Input, Input & { readonly extension: boolean }>["extension"], never>
>;
type CheckpointAssertion = Assert<
{ readonly phase: "deferred"; readonly handle: Stored<DeferredHandle> } extends TaskCheckpoint ? true : false
>;
type HydrationAssertions = [
Assert<Extract<InputOutcome, { status: "placed" }>["input"] extends Entry ? true : false>,
Assert<
Equal<
Extract<InputOutcome, { status: "done" }>["answer"] extends infer A | undefined
? NonNullable<A> extends { message: infer M }
? M
: never
: never,
AssistantMessage
>
>,
ExtendsJson<StoredInputResult>,
];
type KindShape = TaskKind<Input, Checkpoint, Result, Failure, Aborted>;
type RunningAssertion = Assert<"output" extends keyof RunningTask<Input, Checkpoint> ? false : true>;
declare const storedAssertions: StoredAssertions;
declare const hookAssertions: HookAssertions;
declare const kindAssertions: KindAssertions;
declare const jobAssertions: JobAssertions;
declare const entryAssertions: EntryAssertions;
declare const bundleAssertions: BundleAssertions;
declare const badBundleAssertions: BadBundleAssertions;
declare const exactInputAssertion: ExactInputAssertion;
declare const checkpointAssertion: CheckpointAssertion;
declare const hydrationAssertions: HydrationAssertions;
declare const kindShape: KindShape;
declare const runningAssertion: RunningAssertion;
void storedAssertions;
void hookAssertions;
void kindAssertions;
void jobAssertions;
void entryAssertions;
void bundleAssertions;
void badBundleAssertions;
void exactInputAssertion;
void checkpointAssertion;
void hydrationAssertions;
void kindShape;
void runningAssertion;
void verifyNoOutputRuntime;
void verifyOutput;
void verifyJobMutation;
void verifyEntryInputs;
void verifyHostTaskCreation;
void verifyTaskCreation;
void verifyAddresses;
void verifyResolvedTransactionReads;
void (null as Message | null);
void (null as CoreAbortTaskRuntime<Input, Checkpoint> | null);