surma/sausage
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Carry durable input hooks and add a headless harness channel

Browse Source
This commit is contained in:
surma
2026-11-07 01:03:17 +00:00
parent aaf6d9b54f
commit c8525c3acb
16 changed files with 21810 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
host/js/README.md
+31 -3
View File
@@ -1,7 +1,7 @@
# Host JavaScript
The console uses pi-durable in V8. Rust owns the SQLite connection.
The console runs V8 on a dedicated current-thread Tokio runtime.
The headless harness uses pi-durable in V8. Rust owns the SQLite connection.
The harness runs V8 on a dedicated current-thread Tokio runtime.
This satisfies Deno's runtime requirement without a change to the CLI runtime.
The database lives at `<dir>/host/harness.sqlite`, outside the world.
WAL preserves atomic transactions after SIGKILL. `synchronous=FULL` flushes each commit before SQLite reports success.
@@ -28,7 +28,13 @@ npm run --prefix host/js bundle
CARGO_TARGET_DIR=target/harness nix develop -c cargo test -p sausage --test harness
```
The lock file pins npm dependencies. The checked-in bundles let Cargo build without npm or network access for JavaScript.
The lock file pins npm dependencies. The bundle command requires Nushell, GNU patch, and npm.
It applies `patches/pi-durable-1.5.0.patch` before esbuild runs.
This is Ultimate's patch at `7ec0b4f`, without the Node execution environment hunks (D65).
The patch adds transactional `beforeInput`, input markers, scheduler hold/release, and the failed-run boundary fix.
The step verifies an existing application before another bundle build. A mismatched or partially applied patch fails the command.
`npm ci` restores the unpatched dependency. The next bundle command applies the patch again.
The checked-in bundles let Cargo build without npm or network access for JavaScript.
Cargo still needs its normal Rust dependencies and the V8 archive.
For Nix, use a separate npm derivation with `importNpmLock` and the checked-in lock file.
@@ -52,3 +58,25 @@ A decided task admits one plain-language input and completes in one SQLite commi
The input starts a new turn when the conversation is idle. Neither the input nor the task contains a secret.
The task and its input survive SIGKILL. The input uses a stable request id to prevent duplicate admission.
Hosts with a shared credentials handle can use `run_module_with_credentials`.
## Channel input and shutdown
`run_module_with_input` accepts a receiver from `harness::input::channel()` and runs without console input or output.
Its caller supplies the world, process interface, shared credentials handle, and bundled module.
`Input::Submit` contains a conversation id, JSON content, and a request id.
The harness admits that input into the specified conversation. The request id preserves submission deduplication.
Headless admission does not wait for a model answer.
The harness groups queued follow-ups into one new run.
`Input::Stop` waits for admitted foreground work, then closes the harness cleanly.
The JavaScript `serve` function implements this channel protocol.
The main module exports `storage`, including its `scanSubmissions` method.
`run_console` connects stdin and stop signals to the same channel.
The console submits to the durable root conversation (#1).
EOF closes only the stdin client. `sausage run` stays active until SIGINT or SIGTERM.
The signal client sends `Input::Stop`. The harness closes its watches, credential coordinator, and SQLite connection.
The host then stops its process services and HTTP listener.
Console output still streams assistant messages, without replay of the historical transcript.
`harness_foundation.rs` runs the three Ultimate input tests and the scheduler phase-boundary test through bundled V8 modules.
It also tests the channel with closed stdin, CLI EOF and SIGTERM, submission scans, and the patched bundle hooks.
host/js/apply-patch.nu
+15
View File
@@ -0,0 +1,15 @@
# Apply the versioned dist patch, or verify that all its hunks already exist.
let args = [-p1 -d node_modules/@earendil-works/pi-durable -i ../../../patches/pi-durable-1.5.0.patch --batch --fuzz=0]
let forward = (^patch ...$args --forward --dry-run | complete)
if $forward.exit_code == 0 {
^patch ...$args --forward
if $env.LAST_EXIT_CODE != 0 { error make {msg: "pi-durable patch failed"} }
} else {
let reverse = (^patch ...$args --force --reverse --dry-run | complete)
if $reverse.exit_code != 0 {
print $forward.stdout
print --stderr $forward.stderr
error make {msg: "pi-durable 1.5.0 patch does not match the installed package"}
}
print "pi-durable patch already applied"
}
host/js/bundle.js
+162 -94
View File
@@ -30494,8 +30494,8 @@ var CachedStorage = class {
#submissions = new Lru(256);
/** Per record with a read in flight: the reads, and how many published commits wrote it since the first began. */
#reads = /* @__PURE__ */ new Map();
constructor(storage) {
this.#storage = storage;
constructor(storage2) {
this.#storage = storage2;
}
/** The backend's own scope for a publication (`Storage.publishing`), when it has one. */
get publishing() {
@@ -30686,10 +30686,10 @@ var CachedStorage = class {
return this.#reads.size;
}
};
function scopeOf(storage) {
if (storage.publishing === void 0)
function scopeOf(storage2) {
if (storage2.publishing === void 0)
return void 0;
return (publish2, context) => storage.publishing(publish2, context);
return (publish2, context) => storage2.publishing(publish2, context);
}
var Lru = class {
#map = /* @__PURE__ */ new Map();
@@ -30877,8 +30877,8 @@ function tableKeys(writes, keys) {
var GatedStorage = class {
#storage;
#touch;
constructor(storage, touch) {
this.#storage = storage;
constructor(storage2, touch) {
this.#storage = storage2;
this.#touch = touch;
}
commit(batches, receive, context) {
@@ -30954,24 +30954,24 @@ var ROOT_CONVERSATION_ID = 1;
// node_modules/@earendil-works/pi-durable/dist/session/forks.js
var SCAN_PAGE_SIZE = 256;
async function prepareForkDocumentCopies(storage, parentConversationId, at, childConversationId, context) {
const entry2 = await storage.entry(parentConversationId, at, context);
async function prepareForkDocumentCopies(storage2, parentConversationId, at, childConversationId, context) {
const entry2 = await storage2.entry(parentConversationId, at, context);
if (entry2 === void 0)
throw new Error(`Entry ${at} is not visible from conversation ${parentConversationId}`);
const copies = [];
const copiedAddresses = /* @__PURE__ */ new Set();
await collectCopies(storage, { kind: "conversation", conversationId: entry2.entry.conversationId }, entry2.commitSeq, "asOf", childConversationId, copies, copiedAddresses, context);
await collectCopies(storage, { kind: "conversation", conversationId: parentConversationId }, "current", "current", childConversationId, copies, copiedAddresses, context);
await collectCopies(storage2, { kind: "conversation", conversationId: entry2.entry.conversationId }, entry2.commitSeq, "asOf", childConversationId, copies, copiedAddresses, context);
await collectCopies(storage2, { kind: "conversation", conversationId: parentConversationId }, "current", "current", childConversationId, copies, copiedAddresses, context);
return copies;
}
async function collectCopies(storage, scope, at, policy, childConversationId, copies, copiedAddresses, context) {
async function collectCopies(storage2, scope, at, policy, childConversationId, copies, copiedAddresses, context) {
let cursor;
do {
const page2 = await storage.scanDocuments({ scope, at }, SCAN_PAGE_SIZE, cursor, context);
const page2 = await storage2.scanDocuments({ scope, at }, SCAN_PAGE_SIZE, cursor, context);
for (const source of page2.items) {
if (source.scope.kind !== "conversation" || source.fork !== policy)
continue;
const id = await storage.mintId();
const id = await storage2.mintId();
const identity = {
id,
kind: source.kind,
@@ -31140,6 +31140,9 @@ var Transaction = class {
this.#assertOpen();
return record;
}
beforeInput(conversationId, input) {
return this.#read("beforeInput", () => this.#host.beforeInput(this, conversationId, input, this.#context));
}
appendEntry(first, second, third) {
if (typeof first === "number")
return this.#appendEntry(first, second);
@@ -31612,7 +31615,7 @@ var Transaction = class {
return { observed: publications, stored: stored ?? publications };
}
async #assemble() {
const storage = this.#host.storage;
const storage2 = this.#host.storage;
const plans = this.#plans;
for (const document of this.#documents) {
const plan = planDocument(document);
@@ -31667,7 +31670,7 @@ var Transaction = class {
}
let cursor;
do {
const page2 = await storage.scanDocuments({ scope: { kind: "task", taskId }, at: "current" }, INTERNAL_SCAN_PAGE_SIZE, cursor, this.#context);
const page2 = await storage2.scanDocuments({ scope: { kind: "task", taskId }, at: "current" }, INTERNAL_SCAN_PAGE_SIZE, cursor, this.#context);
for (const record of page2.items)
retire(record);
cursor = page2.next;
@@ -31691,7 +31694,7 @@ var Transaction = class {
plan.conversationId = task.publicationConversationId;
}
for (const { id, change } of this.#submissionChanges) {
const current = this.#submissions.get(id) ?? await storage.submission(id, this.#context);
const current = this.#submissions.get(id) ?? await storage2.submission(id, this.#context);
if (current === void 0)
throw new Error(`Submission ${id} does not exist`);
const next = applySubmissionChange(current, change);
@@ -32262,12 +32265,12 @@ var SessionImpl = class {
get failed() {
return this.#failed.promise;
}
constructor(storage, now = Date.now, documentRetentionMs = () => void 0, setTimer, readOnly = false, persistWithinMs = () => void 0) {
constructor(storage2, now = Date.now, documentRetentionMs = () => void 0, setTimer, readOnly = false, persistWithinMs = () => void 0) {
this.#failed.promise.catch(() => {
});
this.#readOnly = readOnly;
this.#cached = readOnly ? void 0 : new CachedStorage(storage);
this.#storage = new GuardedStorage(this.#cached ?? storage, this);
this.#cached = readOnly ? void 0 : new CachedStorage(storage2);
this.#storage = new GuardedStorage(this.#cached ?? storage2, this);
this.#gated = new GatedStorage(this.#storage, (key) => this.touch(key));
this.#writer = new Writer({
commit: (batches, receive, context) => this.#storage.commit(batches, receive, context),
@@ -32322,6 +32325,7 @@ var SessionImpl = class {
if (this.#documents.get(id)?.record.id === recordId)
this.#dropDocument(id);
},
beforeInput: (tx, id, input, context) => this.beforeInput(tx, id, input, context),
conversationCreated: (tx, record) => this.conversationCreated(tx, record),
taskDocuments: (taskId) => this.#taskDocuments.get(taskId)?.values(),
latestEntryId: (conversationId, context) => this.latestEntryId(conversationId, context),
@@ -32646,6 +32650,9 @@ var SessionImpl = class {
*/
report(_error, _info = LISTENER, _context) {
}
async beforeInput(_tx, _id, _input, _context) {
return [];
}
/**
* Runs inside every transaction that creates or forks a conversation, after the conversation record is staged. A
* plain Session stages nothing; a Harness stages its built-in documents.
@@ -33561,8 +33568,8 @@ var GuardedStorage = class {
/** Backend calls underway, also those off the Session line, which close waits for before it closes the backend. */
#underway = /* @__PURE__ */ new Set();
#closing = false;
constructor(storage, latch) {
this.#storage = storage;
constructor(storage2, latch) {
this.#storage = storage2;
this.#latch = latch;
}
/** The backend's own scope for a publication (`Storage.publishing`), when it has one; not a call to guard. */
@@ -34072,21 +34079,21 @@ function defineTask(definition) {
var SCAN_PAGE_SIZE2 = 256;
var EXCLUDED_STOP_REASONS = /* @__PURE__ */ new Set(["aborted", "error", "deferred"]);
var MISSING_RESULT_TEXT = "Tool result unavailable: history ends before this call completed.";
async function captureContextBounds(session, storage, conversationId, context, at, visible = false) {
async function captureContextBounds(session, storage2, conversationId, context, at, visible = false) {
let tail;
if (at === void 0) {
tail = await session.latestEntryId(conversationId, context);
if (tail === void 0)
return void 0;
} else {
if (!visible && await storage.entry(conversationId, at, context) === void 0) {
if (!visible && await storage2.entry(conversationId, at, context) === void 0) {
throw new Error(`Entry ${at} is not visible from conversation ${conversationId}`);
}
tail = at;
}
return { head: await session.latestHeadMarker(conversationId, tail, context), tail };
}
async function readContextFrom(session, storage, conversationId, context, at, previous) {
async function readContextFrom(session, storage2, conversationId, context, at, previous) {
const visible = at !== void 0 && previous !== void 0 && rangeEntry(previous, at) !== void 0;
const bounds = await session.readOnLine((line) => captureContextBounds(session, line, conversationId, context, at, visible));
if (bounds === void 0)
@@ -34095,7 +34102,7 @@ async function readContextFrom(session, storage, conversationId, context, at, pr
const current = at === void 0;
let range;
if (previous === void 0 || previous.bounds.head?.id !== bounds.head?.id) {
const entries2 = freezeJson(await scanRange(storage, conversationId, rangeQuery(bounds), context));
const entries2 = freezeJson(await scanRange(storage2, conversationId, rangeQuery(bounds), context));
range = deriveRange(bounds, entries2, current);
} else if (bounds.tail === previous.bounds.tail) {
range = previous;
@@ -34103,7 +34110,7 @@ async function readContextFrom(session, storage, conversationId, context, at, pr
range = deriveRange(bounds, previous.entries.filter((entry2) => entry2.id <= bounds.tail), false);
} else {
const minEntryId = idFromNumber(previous.bounds.tail + 1);
const added = await scanRange(storage, conversationId, { minEntryId, maxEntryId: bounds.tail }, context);
const added = await scanRange(storage2, conversationId, { minEntryId, maxEntryId: bounds.tail }, context);
range = extendRange(previous, bounds, freezeJson(added), current);
}
if (range.current && session.knownLatestEntryId(conversationId) !== range.bounds.tail) {
@@ -34120,14 +34127,14 @@ async function readContextFrom(session, storage, conversationId, context, at, pr
range
};
}
async function readContext(session, storage, conversationId, context, at) {
async function readContext(session, storage2, conversationId, context, at) {
const bounds = await session.readOnLine((line) => captureContextBounds(session, line, conversationId, context, at));
return deriveContext(storage, conversationId, bounds, context);
return deriveContext(storage2, conversationId, bounds, context);
}
async function deriveContext(storage, conversationId, bounds, context) {
async function deriveContext(storage2, conversationId, bounds, context) {
if (bounds === void 0)
return emptyView();
return deriveRange(bounds, await scanRange(storage, conversationId, rangeQuery(bounds), context), false).view;
return deriveRange(bounds, await scanRange(storage2, conversationId, rangeQuery(bounds), context), false).view;
}
function emptyView() {
return { head: void 0, entries: [], contributions: [], messages: [] };
@@ -34245,11 +34252,11 @@ function rangeQuery(bounds) {
const head = bounds.head;
return head === void 0 ? { maxEntryId: bounds.tail } : { minEntryId: head.head, maxEntryId: bounds.tail };
}
async function scanRange(storage, conversationId, range, context) {
async function scanRange(storage2, conversationId, range, context) {
const entries2 = [];
let cursor;
do {
const page2 = await storage.scanEntries({ conversationId, ...range }, SCAN_PAGE_SIZE2, cursor, context);
const page2 = await storage2.scanEntries({ conversationId, ...range }, SCAN_PAGE_SIZE2, cursor, context);
entries2.push(...page2.items);
cursor = page2.next;
} while (cursor !== void 0);
@@ -34403,10 +34410,14 @@ var InboxDoc = defineDoc({
initial: () => ({ items: [] }),
checkpointWhen: (value2) => value2.items.length === 0
});
async function prepareBoundary(tx, conversationId, modes) {
async function prepareBoundary(tx, conversationId, modes, at, now, incoming) {
const head = (await tx.latestHeadMarker(conversationId))?.head;
const inbox = await tx.doc(InboxDoc, conversationId);
const first = inbox.items.find((item) => item.mode === "steer" || item.mode === "followUp");
const input = first === void 0 ? incoming : inputDraft(first.content, first.entry, now);
const entries2 = input !== void 0 && (at === "final" || hasQueuedReset(inbox.items)) ? await tx.beforeInput(conversationId, input) : [];
return {
entries: entries2,
conversationId,
inbox,
steeringMode: modes.steeringMode,
@@ -34442,6 +34453,9 @@ async function applyBoundary(tx, boundary, at, now) {
}
const placed = [];
for (const index of users) {
if (placed.length === 0)
for (const draft of boundary.entries)
await tx.appendEntry(conversationId, draft);
const item = items[index];
const entry2 = await appendInput(tx, conversationId, item.content, item.entry, now);
tx.placeSubmission(item.id, entry2.id);
@@ -34454,10 +34468,11 @@ async function applyBoundary(tx, boundary, at, now) {
return { users: placed, first, reset };
}
function appendInput(tx, conversationId, content, entry2, now) {
return tx.appendEntry(conversationId, inputDraft(content, entry2, now));
}
function inputDraft(content, entry2, now) {
const message = { role: "user", content, timestamp: now };
if (entry2 === void 0)
return tx.appendEntry(UserEntry, conversationId, { model: [message] });
return tx.appendEntry(conversationId, { ...pickInputEntry(entry2), model: [message] });
return { ...entry2 === void 0 ? { kind: UserEntry.kind } : pickInputEntry(entry2), input: true, model: [message] };
}
function pickInputEntry({ kind, data }) {
const picked = data === void 0 ? { kind } : { kind, data };
@@ -35177,6 +35192,7 @@ var TaskScheduler = class {
#unsubscribeRegistry = () => {
};
#enabled = false;
#held = false;
#readOnly;
#closing = false;
#dirty = false;
@@ -35205,6 +35221,14 @@ var TaskScheduler = class {
this.#context = options.context;
this.#readOnly = options.readOnly === true;
}
/** Start no run invocation until `release()`; running invocations and abort handlers go on. Overrides `resume()`. */
hold() {
this.#held = true;
}
release() {
this.#held = false;
this.#kick();
}
/**
* Load live tasks and change surviving `running` tasks back to `pending`, counting the restart. Dispatches nothing;
* a task cut off `settings.restartLimit` times ends when it is next reserved. Read-only, it only loads them: a
@@ -35394,13 +35418,13 @@ var TaskScheduler = class {
return marked.result;
}
async waitForTask(id, context) {
const found = await this.#session.readOnLine(async (storage) => {
const found = await this.#session.readOnLine(async (storage2) => {
if (this.#closing)
throw closedError(this.#session);
if (await this.#region(() => this.#liveRecord(id) !== void 0)) {
return { promise: this.#taskWaiters.add(id, context) };
}
const record = await storage.task(id, context);
const record = await storage2.task(id, context);
if (record === void 0)
throw new Error(`Task ${id} does not exist`);
return { promise: Promise.resolve(record) };
@@ -36659,6 +36683,8 @@ var TaskScheduler = class {
for (const record of plan.ready) {
const id = record.id;
const mode = record.abortRequested ? "abort" : "run";
if (mode === "run" && this.#held)
continue;
const cutOff = record.restarts !== void 0 && record.restarts >= this.#restartLimit();
if (cutOff && (mode === "abort" || !this.#limitsRestarts(record.kind))) {
const message = mode === "abort" ? `Abort handler of task ${id} (${record.kind}) was cut off ${record.restarts} times` : `Task ${id} (${record.kind}) was cut off ${record.restarts} times without progress`;
@@ -36844,7 +36870,7 @@ var TaskScheduler = class {
const tasks = [];
for (const record of this.#live.values())
tasks.push({ record, state: this.#inspectTask(record, snapshot) });
const scheduling = this.#closing ? "closing" : this.#enabled ? "running" : "paused";
const scheduling = this.#closing ? "closing" : this.#enabled && !this.#held ? "running" : "paused";
return { scheduling, tasks };
}
#inspectTask(record, snapshot) {
@@ -36999,6 +37025,10 @@ var TaskScheduler = class {
* invocation continues.
*/
#progress(tx, current, state2, extension, context) {
if (this.#held) {
tx.setTask(withState(current, { status: "pending", checkpoint: current.state.checkpoint }));
return false;
}
state2.snapshot = this.#registry.snapshot();
const next = state2.snapshot.task(current.kind);
if (next === state2.task)
@@ -37366,7 +37396,7 @@ var TaskScheduler = class {
watchDoc: ((...args) => this.#watchDoc(invocation, args)),
snapshot: ((...args) => this.#read(invocation, () => sessionMethod(this.#session, "snapshot")(...args))),
snapshotAsOf: ((...args) => this.#read(invocation, () => sessionMethod(this.#session, "snapshotAsOf")(...args))),
getTask: ((id, context) => this.#read(invocation, () => this.#session.readOnLine((storage) => storage.task(id, context)))),
getTask: ((id, context) => this.#read(invocation, () => this.#session.readOnLine((storage2) => storage2.task(id, context)))),
waitForTask: ((id, context) => this.#read(invocation, () => this.waitForTask(id, withAbortSignal(invocation.controller.signal, context)))),
ownedTasks: (context) => this.#read(invocation, () => this.#session.readOnLine(async () => {
if (invocation.ended)
@@ -37386,7 +37416,7 @@ var TaskScheduler = class {
})),
abortOwned: (id, context) => this.#read(invocation, async () => {
const bound = withAbortSignal(invocation.controller.signal, context);
const record = await this.#session.readOnLine((storage) => storage.task(id, bound));
const record = await this.#session.readOnLine((storage2) => storage2.task(id, bound));
if (record?.owner !== invocation.taskId) {
throw new Error(`Task ${id} is not owned by task ${invocation.taskId}`);
}
@@ -37395,10 +37425,10 @@ var TaskScheduler = class {
await this.abort(id, bound, true);
await this.waitForTask(id, bound);
}),
outcomes: ((ids, context) => this.#read(invocation, () => this.#session.readOnLine(async (storage) => {
outcomes: ((ids, context) => this.#read(invocation, () => this.#session.readOnLine(async (storage2) => {
const outcomes = [];
for (const id of ids) {
const state2 = (await storage.task(id, context))?.state;
const state2 = (await storage2.task(id, context))?.state;
if (state2?.status !== "terminal")
throw new Error(`Task ${id} is not terminal`);
outcomes.push(state2.outcome);
@@ -37416,11 +37446,11 @@ var TaskScheduler = class {
const [token, id, context] = args.length === 2 ? [void 0, args[0], args[1]] : [args[0], args[1], args[2]];
return this.#read(invocation, async () => {
const kept = this.#contexts.get(invocation.conversationId);
const entry2 = (kept === void 0 ? void 0 : rangeEntry(kept.range, id)) ?? (await this.#session.readOnLine((storage) => storage.entry(invocation.conversationId, id, context)))?.entry;
const entry2 = (kept === void 0 ? void 0 : rangeEntry(kept.range, id)) ?? (await this.#session.readOnLine((storage2) => storage2.entry(invocation.conversationId, id, context)))?.entry;
return token === void 0 || entry2?.kind === token.kind ? entry2 : void 0;
});
}),
entries: (query, limit2, cursor, context) => this.#read(invocation, () => this.#session.readOnLine((storage) => storage.scanEntries(conversationEntries(invocation.conversationId, query), limit2, cursor, context))),
entries: (query, limit2, cursor, context) => this.#read(invocation, () => this.#session.readOnLine((storage2) => storage2.scanEntries(conversationEntries(invocation.conversationId, query), limit2, cursor, context))),
context: (conversationId, context, options) => this.#read(invocation, async () => {
const read2 = await readContextFrom(this.#session, this.#storage, conversationId, context, options?.at, this.#contexts.get(conversationId)?.range);
const { view: view2 } = read2;
@@ -39732,7 +39762,7 @@ async function prepareFailure(tx, runtime) {
const policy = settings.afterFailedRun;
if (policy !== "continue")
return { policy };
return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings) };
return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings, "final", runtime.now()) };
}
async function endFailedRun(tx, runtime, live, failure, end) {
const conversationId = runtime.conversationId;
@@ -39900,7 +39930,7 @@ async function answer(runtime, message, context) {
const afterRun = await asksAfterRun(runtime, context);
const conversationId = runtime.conversationId;
await runtime.commit(async (tx) => {
const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
const live = await tx.doc(LiveDoc, conversationId);
const entry2 = await appendAssistant(tx, conversationId, message);
const result = { status: "terminal", outcome: { status: "completed", result: { entryId: entry2.id } } };
@@ -39973,7 +40003,7 @@ function prepareRunEndTasks(runtime, requests, prepared) {
async function endAnsweredRun(runtime, answer2, context) {
const conversationId = runtime.conversationId;
await runtime.commit(async (tx) => {
const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
const live = await tx.doc(LiveDoc, conversationId);
const { users, first } = await applyBoundary(tx, boundary, "final", runtime.now());
endRun(tx, live, runtime.taskId, { status: "done", answer: answer2 });
@@ -40046,7 +40076,7 @@ async function finishToolRound(runtime, assistant, tools, context) {
const maxTurns = (await runtime.agent(context)).maxTurns;
const afterRun = terminate && await asksAfterRun(runtime, context);
await runtime.commit(async (tx) => {
const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
const boundary = await prepareBoundary(tx, conversationId, runtime.settings, terminate || handoff !== void 0 ? "final" : "postTools", runtime.now());
if (added.length > 0)
await addTools(tx, conversationId, added);
const live = await tx.doc(LiveDoc, conversationId);
@@ -40354,19 +40384,19 @@ var Submissions = class {
}
/** Handle for an existing submission, or `undefined`. */
async get(id, context) {
const record = await this.#session.readOnLine((storage) => storage.submission(id, context));
const record = await this.#session.readOnLine((storage2) => storage2.submission(id, context));
return record === void 0 ? void 0 : new SubmissionHandle(record.id, this);
}
async status(id, context) {
const record = await this.#session.readOnLine((storage) => storage.submission(id, context));
const record = await this.#session.readOnLine((storage2) => storage2.submission(id, context));
if (record === void 0)
throw new Error(`Submission ${id} does not exist`);
return record;
}
async wait(id, context) {
this.#resume();
const found = await this.#session.readOnLine(async (storage) => {
const record = await storage.submission(id, context);
const found = await this.#session.readOnLine(async (storage2) => {
const record = await storage2.submission(id, context);
if (record === void 0)
throw new Error(`Submission ${id} does not exist`);
if (isSettled(record))
@@ -40463,13 +40493,13 @@ async function admitSubmission(tx, conversationId, draft, now, queueModes, span)
if (draft.type === "continue") {
if (busy)
throw new ConversationBusy(conversationId);
const boundary2 = await prepareBoundary(tx, conversationId, queueModes);
const boundary2 = await prepareBoundary(tx, conversationId, queueModes, "final", now);
const { id: id2 } = await tx.createSubmission({ ...fields, type: "continue", status: "placed" });
const { users, first } = await applyBoundary(tx, boundary2, "final", now);
await startRun(tx, conversationId, live, { inputs: [...users, id2], first: live.runFailure?.first ?? first });
return { id: id2, status: "placed" };
}
const boundary = busy ? void 0 : await prepareBoundary(tx, conversationId, queueModes);
const boundary = busy ? void 0 : await prepareBoundary(tx, conversationId, queueModes, "final", now, draft.type === "input" ? inputDraft(draft.content, draft.entry, now) : void 0);
if (boundary === void 0 || boundary.inbox.items.length > 0) {
const { id: id2 } = await tx.createSubmission({
...fields,
@@ -40507,6 +40537,8 @@ async function admitSubmission(tx, conversationId, draft, now, queueModes, span)
const write = { ...fields, type: "write", status: "done", entry: entry3.id };
return { id: (await tx.createSubmission(write)).id, status: "done" };
}
for (const before of boundary.entries)
await tx.appendEntry(conversationId, before);
const entry2 = await appendInput(tx, conversationId, draft.content, draft.entry, now);
const input = { ...fields, type: "input", status: "placed", entry: entry2.id };
const { id } = await tx.createSubmission(input);
@@ -41037,9 +41069,9 @@ var TaskGraphView = class {
}
/** Register an observer created from the current revision, atomically on the Session line. */
#attach(create, context) {
return this.#session.readOnLine(async (storage) => {
return this.#session.readOnLine(async (storage2) => {
const mount = this.#mount ?? {
...await this.#build(storage, context),
...await this.#build(storage2, context),
observers: /* @__PURE__ */ new Set()
};
const detach = () => {
@@ -41076,16 +41108,16 @@ var TaskGraphView = class {
});
}
/** Read the graph on the line, through the line's `storage`. */
async #build(storage, context) {
async #build(storage2, context) {
const tasks = {};
const live = /* @__PURE__ */ new Map();
const records = [];
for (const status of LIVE_STATUSES2) {
records.push(...await scanAll((cursor) => storage.scanTasks({ status }, SCAN_PAGE_SIZE4, cursor, context)));
records.push(...await scanAll((cursor) => storage2.scanTasks({ status }, SCAN_PAGE_SIZE4, cursor, context)));
}
records.sort((a, b) => a.id - b.id);
for (const record of records) {
const owned = await scanAll((cursor) => storage.scanConversations({ ownerTaskId: record.id }, SCAN_PAGE_SIZE4, cursor, context));
const owned = await scanAll((cursor) => storage2.scanConversations({ ownerTaskId: record.id }, SCAN_PAGE_SIZE4, cursor, context));
const conversations = owned.map((conversation) => conversation.id).sort((a, b) => a - b);
const subject = blockedSubject(record);
tasks[String(record.id)] = nodeOf2(record, conversations, this.#blocking.blocked(subject));
@@ -41262,9 +41294,9 @@ var ConversationViews = class {
*/
async attach(id, window2, create, context) {
const bounds = resolveWindow(window2);
return this.#session.readOnLine(async (storage) => {
const mount = this.#mounts.get(id) ?? await this.#build(storage, id, context);
const { entries: entries2, before } = await loadWindow(storage, id, bounds, context);
return this.#session.readOnLine(async (storage2) => {
const mount = this.#mounts.get(id) ?? await this.#build(storage2, id, context);
const { entries: entries2, before } = await loadWindow(storage2, id, bounds, context);
const value2 = view(mount.conversation, entries2, before, mount.docs);
let windowed;
const detach = () => {
@@ -41284,8 +41316,8 @@ var ConversationViews = class {
});
}
/** Hydrate a mount on the line, through the line's `storage`. */
async #build(storage, id, context) {
const conversation = await storage.conversation(id, context);
async #build(storage2, id, context) {
const conversation = await storage2.conversation(id, context);
if (conversation === void 0)
throw new Error(`Conversation ${id} does not exist`);
const docs = {};
@@ -41303,25 +41335,25 @@ var ConversationViews = class {
function view(conversation, entries2, before, docs) {
return before === void 0 ? { conversation, entries: entries2, docs } : { conversation, entries: entries2, before, docs };
}
async function loadWindow(storage, conversationId, bounds, context) {
async function loadWindow(storage2, conversationId, bounds, context) {
const { from, min, max } = bounds;
if (from !== void 0) {
const since = await scanUpTo(storage, { conversationId, minEntryId: from, order: "ascending" }, max + 1, context);
const since = await scanUpTo(storage2, { conversationId, minEntryId: from, order: "ascending" }, max + 1, context);
if (since.length <= max) {
const query = { conversationId, maxEntryId: idFromNumber(from - 1) };
const previous = (await storage.scanEntries(query, 1, void 0, context)).items[0];
const previous = (await storage2.scanEntries(query, 1, void 0, context)).items[0];
return { entries: since, before: previous?.id };
}
}
const newest = await scanUpTo(storage, { conversationId }, min + 1, context);
const newest = await scanUpTo(storage2, { conversationId }, min + 1, context);
const before = newest.length > min ? newest.pop().id : void 0;
return { entries: newest.reverse(), before };
}
async function scanUpTo(storage, query, limit2, context) {
async function scanUpTo(storage2, query, limit2, context) {
const entries2 = [];
let cursor;
do {
const page2 = await storage.scanEntries(query, Math.min(SCAN_PAGE_SIZE5, limit2 - entries2.length), cursor, context);
const page2 = await storage2.scanEntries(query, Math.min(SCAN_PAGE_SIZE5, limit2 - entries2.length), cursor, context);
entries2.push(...page2.items);
cursor = page2.next;
} while (cursor !== void 0 && entries2.length < limit2);
@@ -41441,7 +41473,7 @@ var ConversationImpl = class {
}
entries(query, limit2, cursor, context) {
const scan = conversationEntries(this.id, query);
return this.#host.harness.readOnLine((storage) => storage.scanEntries(scan, limit2, cursor, context));
return this.#host.harness.readOnLine((storage2) => storage2.scanEntries(scan, limit2, cursor, context));
}
fork(at, options, context) {
return this.#host.create({ kind: "fork", parentId: this.id, at, ownership: options.ownership }, options, context);
@@ -41482,7 +41514,7 @@ var HarnessImpl = class extends SessionImpl {
const now = safeNow(options.now, report2);
const hostSettings = currentSettings(options.settings);
super(rawStorage, now, () => hostSettings()?.documentRetentionMs, options.setTimer, options.readOnly === true, () => hostSettings()?.progress?.persistWithinMs);
const storage = this.storage;
const storage2 = this.storage;
this.#options = options;
this.#report = report2;
this.#hostSettings = hostSettings;
@@ -41490,7 +41522,7 @@ var HarnessImpl = class extends SessionImpl {
this.#tasks = new TaskScheduler({
telemetry: options.telemetry,
session: this,
storage,
storage: storage2,
registry: options.registry,
models: currentModels(options.models, options.identity),
agent: (id, snapshot, callContext) => this.resolveAgent(id, snapshot, callContext),
@@ -41518,7 +41550,7 @@ var HarnessImpl = class extends SessionImpl {
});
this.#host = {
harness: this,
storage,
storage: storage2,
tasks: this.#tasks,
submissions: this.#submissions,
views: new ConversationViews(this),
@@ -41526,6 +41558,15 @@ var HarnessImpl = class extends SessionImpl {
create: (target, createOptions, context2) => this.#create(target, createOptions, context2)
};
}
async beforeInput(tx, conversationId, input, context) {
const agent = resolveAgent(await tx.doc(AgentDoc, conversationId), this.#options.registry.snapshot(), resolveSettings(this.#hostSettings()), (error, extension) => this.#report(error, { kind: "extension_error", conversationId, extension }, context));
const entries2 = [];
for (const { handlers } of agentHooks(agent, "pi.generation")) {
const result = await handlers.beforeInput?.(input, { conversationId, tx }, context);
entries2.push(...result?.entries ?? []);
}
return entries2;
}
/** Resolve a conversation's committed `pi.agent` against `snapshot`, or the current one, and the current settings. */
async resolveAgent(id, snapshot, context) {
const registry2 = snapshot ?? this.#options.registry.snapshot();
@@ -41548,12 +41589,18 @@ var HarnessImpl = class extends SessionImpl {
this.assertUsable();
this.#tasks.resume();
}
hold() {
this.#tasks.hold();
}
release() {
this.#tasks.release();
}
wake() {
this.assertUsable();
return this.#tasks.wake();
}
getTask(id, context) {
return this.readOnLine((storage) => storage.task(id, context));
return this.readOnLine((storage2) => storage2.task(id, context));
}
/** Internal, for tests: sizes of the scheduler's ownership indexes. */
get [SCHEDULER_INDEX_SIZES]() {
@@ -41564,9 +41611,9 @@ var HarnessImpl = class extends SessionImpl {
return this.#tasks.idleState;
}
inspect(context) {
return this.readOnLine(async (storage) => {
return this.readOnLine(async (storage2) => {
const { scheduling, tasks } = await this.#tasks.inspect(this.#options.registry.snapshot());
const scan = (status) => scanAll((cursor) => storage.scanSubmissions({ status }, SCAN_PAGE_SIZE6, cursor, context));
const scan = (status) => scanAll((cursor) => storage2.scanSubmissions({ status }, SCAN_PAGE_SIZE6, cursor, context));
const submissions = [...await scan("queued"), ...await scan("placed")].sort((a, b) => a.id - b.id);
return { scheduling, tasks, submissions };
});
@@ -41590,7 +41637,7 @@ var HarnessImpl = class extends SessionImpl {
}
/** Sum every conversation's committed `pi.usage`. Each document is read at its own point; totals only grow. */
async usage(context) {
const conversations = await this.readOnLine((storage) => scanAll((cursor) => storage.scanConversations({}, SCAN_PAGE_SIZE6, cursor, context)));
const conversations = await this.readOnLine((storage2) => scanAll((cursor) => storage2.scanConversations({}, SCAN_PAGE_SIZE6, cursor, context)));
const total = UsageDoc.definition.initial();
for (const { id } of conversations) {
const state2 = await this.snapshot(UsageDoc, id, context);
@@ -41615,7 +41662,7 @@ var HarnessImpl = class extends SessionImpl {
}
async conversation(id, context) {
this.assertUsable();
const record = await this.readOnLine((storage) => storage.conversation(id, context));
const record = await this.readOnLine((storage2) => storage2.conversation(id, context));
return record === void 0 ? void 0 : new ConversationImpl(record.id, this.#host);
}
createConversation(options, context) {
@@ -41746,7 +41793,7 @@ function boundConversation(id, binding, submissions, tasks) {
}
var Harness = {
/** Open a Harness over storage. The registry may keep changing while the Harness runs. */
async open(storage, options, context) {
async open(storage2, options, context) {
context.abortSignal?.throwIfAborted();
const snapshot = options.registry.snapshot();
const missing = BUILTIN_TASKS.filter((task) => snapshot.task(task.definition.name) === void 0);
@@ -41754,7 +41801,7 @@ var Harness = {
const names2 = missing.map((task) => task.definition.name).join(", ");
throw new Error(`Registry lacks built-in tasks ${names2}; create it with createRegistry()`);
}
const harness2 = new HarnessImpl(storage, options, context);
const harness2 = new HarnessImpl(storage2, options, context);
try {
await harness2.openTasks(context);
} catch (error) {
@@ -44482,6 +44529,29 @@ async function startRequests(harness2, context) {
};
}
// input.ts
async function serve(harness2, context, afterSubmit) {
while (true) {
const input = await Deno.core.ops.op_input();
if (input === null || input.type === "stop") {
await harness2.waitForIdle(context);
return;
}
const conversation = await harness2.conversation(input.conversation, context);
if (!conversation) throw new Error(`Unknown conversation: ${input.conversation}`);
const submission2 = await conversation.submit({
type: "input",
content: input.content,
requestId: input.requestId
}, context);
if (afterSubmit) {
const result = await submission2.wait(context);
await afterSubmit();
if (result.status !== "done") throw new Error(`Run ended: ${JSON.stringify(result)}`);
}
}
}
// main.ts
var proxy = Deno.core.ops.op_model_proxy();
if (!proxy?.base_url || !proxy?.api_key) throw new Error("The llm-proxy credential is required");
@@ -44516,9 +44586,11 @@ models.setProvider({
var registry = createRegistry();
registry.install(CodingTools);
registry.install(CredentialRequests);
var harness = await Harness.open(await openStorage(globalThis.storagePath), {
var storage = await openStorage(globalThis.storagePath);
var harness = await Harness.open(storage, {
models,
registry,
settings: { followUpMode: "all" },
env: (target) => environment(target.cwd ?? "/", target.conversationId)
}, BACKGROUND_CONTEXT);
var root = await harness.root(BACKGROUND_CONTEXT, { agent: { model: { provider: "openai", modelId: model2.id }, thinkingLevel: "low", cwd: "/", extensions: [CodingTools, CredentialRequests] } });
@@ -44543,25 +44615,21 @@ var render = (view2) => {
for (const entry2 of watch.value.entries) {
if (entry2.kind === "pi.assistant") emitted.set(entry2.id, text(entry2.model?.[0]));
}
watch.start((event) => {
if (globalThis.consoleEnabled) watch.start((event) => {
if (event.kind === "value") render(event.value);
});
var stopRequests = await startRequests(harness, BACKGROUND_CONTEXT);
harness.resume();
await harness.waitForIdle(BACKGROUND_CONTEXT);
while (true) {
const line = await Deno.core.ops.op_input();
if (line === null) break;
if (!line.trim()) continue;
const submission2 = await root.submit({ type: "input", content: line, requestId: crypto.randomUUID() }, BACKGROUND_CONTEXT);
const result = await submission2.wait(BACKGROUND_CONTEXT);
await serve(harness, BACKGROUND_CONTEXT, globalThis.consoleEnabled ? async () => {
await harness.waitForIdle(BACKGROUND_CONTEXT);
const final = await root.watch(BACKGROUND_CONTEXT);
render(final.value);
await final.stop();
Deno.core.ops.op_output("\n");
if (result.status !== "done") throw new Error(`Run ended: ${JSON.stringify(result)}`);
}
} : void 0);
await stopRequests();
await watch.stop();
await harness.close(BACKGROUND_CONTEXT);
export {
storage
};
host/js/durable-test-bundle.js
+58 -11
View File
@@ -8883,6 +8883,9 @@ var Transaction = class {
this.#assertOpen();
return record;
}
beforeInput(conversationId, input) {
return this.#read("beforeInput", () => this.#host.beforeInput(this, conversationId, input, this.#context));
}
appendEntry(first, second, third) {
if (typeof first === "number")
return this.#appendEntry(first, second);
@@ -10065,6 +10068,7 @@ var SessionImpl = class {
if (this.#documents.get(id)?.record.id === recordId)
this.#dropDocument(id);
},
beforeInput: (tx, id, input, context) => this.beforeInput(tx, id, input, context),
conversationCreated: (tx, record) => this.conversationCreated(tx, record),
taskDocuments: (taskId) => this.#taskDocuments.get(taskId)?.values(),
latestEntryId: (conversationId, context) => this.latestEntryId(conversationId, context),
@@ -10389,6 +10393,9 @@ var SessionImpl = class {
*/
report(_error, _info = LISTENER, _context) {
}
async beforeInput(_tx, _id, _input, _context) {
return [];
}
/**
* Runs inside every transaction that creates or forks a conversation, after the conversation record is staged. A
* plain Session stages nothing; a Harness stages its built-in documents.
@@ -12193,10 +12200,14 @@ var InboxDoc = defineDoc({
initial: () => ({ items: [] }),
checkpointWhen: (value2) => value2.items.length === 0
});
async function prepareBoundary(tx, conversationId, modes) {
async function prepareBoundary(tx, conversationId, modes, at, now, incoming) {
const head = (await tx.latestHeadMarker(conversationId))?.head;
const inbox = await tx.doc(InboxDoc, conversationId);
const first = inbox.items.find((item) => item.mode === "steer" || item.mode === "followUp");
const input = first === void 0 ? incoming : inputDraft(first.content, first.entry, now);
const entries2 = input !== void 0 && (at === "final" || hasQueuedReset(inbox.items)) ? await tx.beforeInput(conversationId, input) : [];
return {
entries: entries2,
conversationId,
inbox,
steeringMode: modes.steeringMode,
@@ -12232,6 +12243,9 @@ async function applyBoundary(tx, boundary, at, now) {
}
const placed = [];
for (const index of users) {
if (placed.length === 0)
for (const draft of boundary.entries)
await tx.appendEntry(conversationId, draft);
const item = items[index];
const entry2 = await appendInput(tx, conversationId, item.content, item.entry, now);
tx.placeSubmission(item.id, entry2.id);
@@ -12244,10 +12258,11 @@ async function applyBoundary(tx, boundary, at, now) {
return { users: placed, first, reset };
}
function appendInput(tx, conversationId, content, entry2, now) {
return tx.appendEntry(conversationId, inputDraft(content, entry2, now));
}
function inputDraft(content, entry2, now) {
const message = { role: "user", content, timestamp: now };
if (entry2 === void 0)
return tx.appendEntry(UserEntry, conversationId, { model: [message] });
return tx.appendEntry(conversationId, { ...pickInputEntry(entry2), model: [message] });
return { ...entry2 === void 0 ? { kind: UserEntry.kind } : pickInputEntry(entry2), input: true, model: [message] };
}
function pickInputEntry({ kind, data }) {
const picked = data === void 0 ? { kind } : { kind, data };
@@ -12959,6 +12974,7 @@ var TaskScheduler = class {
#unsubscribeRegistry = () => {
};
#enabled = false;
#held = false;
#readOnly;
#closing = false;
#dirty = false;
@@ -12987,6 +13003,14 @@ var TaskScheduler = class {
this.#context = options.context;
this.#readOnly = options.readOnly === true;
}
/** Start no run invocation until `release()`; running invocations and abort handlers go on. Overrides `resume()`. */
hold() {
this.#held = true;
}
release() {
this.#held = false;
this.#kick();
}
/**
* Load live tasks and change surviving `running` tasks back to `pending`, counting the restart. Dispatches nothing;
* a task cut off `settings.restartLimit` times ends when it is next reserved. Read-only, it only loads them: a
@@ -14441,6 +14465,8 @@ var TaskScheduler = class {
for (const record of plan.ready) {
const id = record.id;
const mode = record.abortRequested ? "abort" : "run";
if (mode === "run" && this.#held)
continue;
const cutOff = record.restarts !== void 0 && record.restarts >= this.#restartLimit();
if (cutOff && (mode === "abort" || !this.#limitsRestarts(record.kind))) {
const message = mode === "abort" ? `Abort handler of task ${id} (${record.kind}) was cut off ${record.restarts} times` : `Task ${id} (${record.kind}) was cut off ${record.restarts} times without progress`;
@@ -14626,7 +14652,7 @@ var TaskScheduler = class {
const tasks = [];
for (const record of this.#live.values())
tasks.push({ record, state: this.#inspectTask(record, snapshot) });
const scheduling = this.#closing ? "closing" : this.#enabled ? "running" : "paused";
const scheduling = this.#closing ? "closing" : this.#enabled && !this.#held ? "running" : "paused";
return { scheduling, tasks };
}
#inspectTask(record, snapshot) {
@@ -14781,6 +14807,10 @@ var TaskScheduler = class {
* invocation continues.
*/
#progress(tx, current, state, extension, context) {
if (this.#held) {
tx.setTask(withState(current, { status: "pending", checkpoint: current.state.checkpoint }));
return false;
}
state.snapshot = this.#registry.snapshot();
const next = state.snapshot.task(current.kind);
if (next === state.task)
@@ -17429,7 +17459,7 @@ async function prepareFailure(tx, runtime) {
const policy = settings.afterFailedRun;
if (policy !== "continue")
return { policy };
return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings) };
return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings, "final", runtime.now()) };
}
async function endFailedRun(tx, runtime, live, failure, end) {
const conversationId = runtime.conversationId;
@@ -17597,7 +17627,7 @@ async function answer(runtime, message, context) {
const afterRun = await asksAfterRun(runtime, context);
const conversationId = runtime.conversationId;
await runtime.commit(async (tx) => {
const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
const live = await tx.doc(LiveDoc, conversationId);
const entry2 = await appendAssistant(tx, conversationId, message);
const result = { status: "terminal", outcome: { status: "completed", result: { entryId: entry2.id } } };
@@ -17670,7 +17700,7 @@ function prepareRunEndTasks(runtime, requests, prepared) {
async function endAnsweredRun(runtime, answer2, context) {
const conversationId = runtime.conversationId;
await runtime.commit(async (tx) => {
const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
const live = await tx.doc(LiveDoc, conversationId);
const { users, first } = await applyBoundary(tx, boundary, "final", runtime.now());
endRun(tx, live, runtime.taskId, { status: "done", answer: answer2 });
@@ -17743,7 +17773,7 @@ async function finishToolRound(runtime, assistant, tools, context) {
const maxTurns = (await runtime.agent(context)).maxTurns;
const afterRun = terminate && await asksAfterRun(runtime, context);
await runtime.commit(async (tx) => {
const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
const boundary = await prepareBoundary(tx, conversationId, runtime.settings, terminate || handoff !== void 0 ? "final" : "postTools", runtime.now());
if (added.length > 0)
await addTools(tx, conversationId, added);
const live = await tx.doc(LiveDoc, conversationId);
@@ -18159,13 +18189,13 @@ async function admitSubmission(tx, conversationId, draft, now, queueModes, span)
if (draft.type === "continue") {
if (busy)
throw new ConversationBusy(conversationId);
const boundary2 = await prepareBoundary(tx, conversationId, queueModes);
const boundary2 = await prepareBoundary(tx, conversationId, queueModes, "final", now);
const { id: id2 } = await tx.createSubmission({ ...fields, type: "continue", status: "placed" });
const { users, first } = await applyBoundary(tx, boundary2, "final", now);
await startRun(tx, conversationId, live, { inputs: [...users, id2], first: live.runFailure?.first ?? first });
return { id: id2, status: "placed" };
}
const boundary = busy ? void 0 : await prepareBoundary(tx, conversationId, queueModes);
const boundary = busy ? void 0 : await prepareBoundary(tx, conversationId, queueModes, "final", now, draft.type === "input" ? inputDraft(draft.content, draft.entry, now) : void 0);
if (boundary === void 0 || boundary.inbox.items.length > 0) {
const { id: id2 } = await tx.createSubmission({
...fields,
@@ -18203,6 +18233,8 @@ async function admitSubmission(tx, conversationId, draft, now, queueModes, span)
const write = { ...fields, type: "write", status: "done", entry: entry3.id };
return { id: (await tx.createSubmission(write)).id, status: "done" };
}
for (const before of boundary.entries)
await tx.appendEntry(conversationId, before);
const entry2 = await appendInput(tx, conversationId, draft.content, draft.entry, now);
const input = { ...fields, type: "input", status: "placed", entry: entry2.id };
const { id } = await tx.createSubmission(input);
@@ -19215,6 +19247,15 @@ var HarnessImpl = class extends SessionImpl {
create: (target, createOptions, context2) => this.#create(target, createOptions, context2)
};
}
async beforeInput(tx, conversationId, input, context) {
const agent = resolveAgent(await tx.doc(AgentDoc, conversationId), this.#options.registry.snapshot(), resolveSettings(this.#hostSettings()), (error, extension) => this.#report(error, { kind: "extension_error", conversationId, extension }, context));
const entries2 = [];
for (const { handlers } of agentHooks(agent, "pi.generation")) {
const result = await handlers.beforeInput?.(input, { conversationId, tx }, context);
entries2.push(...result?.entries ?? []);
}
return entries2;
}
/** Resolve a conversation's committed `pi.agent` against `snapshot`, or the current one, and the current settings. */
async resolveAgent(id, snapshot, context) {
const registry2 = snapshot ?? this.#options.registry.snapshot();
@@ -19237,6 +19278,12 @@ var HarnessImpl = class extends SessionImpl {
this.assertUsable();
this.#tasks.resume();
}
hold() {
this.#tasks.hold();
}
release() {
this.#tasks.release();
}
wake() {
this.assertUsable();
return this.#tasks.wake();
host/js/foundation-test-bundle.js
+20652
View File
File diff suppressed because it is too large Load Diff
host/js/foundation-test.ts
+127
View File
@@ -0,0 +1,127 @@
// Ultimate's input-placement cases, scheduler hold, and the headless channel contract.
import { BACKGROUND_CONTEXT as context } from '@earendil-works/chord/context';
import { createModels } from '@earendil-works/pi-ai/models';
import { fauxProvider, fauxAssistantMessage, fauxToolCall } from '@earendil-works/pi-ai/providers/faux';
import { Harness, createRegistry, defineDoc, defineExtension, defineTask, GenerationTask, hook, InboxDoc } from '@earendil-works/pi-durable';
import { openStorage } from './sqlite';
import { serve } from './input';
const assert = (value: any, message: string) => { if (!value) throw new Error(message); };
const equal = (a: any, b: any) => assert(JSON.stringify(a) === JSON.stringify(b), `Expected ${JSON.stringify(b)}, got ${JSON.stringify(a)}`);
const until = async (condition: () => any) => {
const deadline = Date.now() + 10000;
while (!await condition()) {
if (Date.now() > deadline) throw new Error('Condition deadline');
await new Promise(resolve => setTimeout(resolve, 0));
}
};
const Placed = defineDoc<{ count: number }>({ kind: 'input-test.placed', version: 1,
checkpointWhen: () => false, scope: 'conversation', history: 'latest', fork: 'initial', initial: () => ({ count: 0 }) });
const gate = Promise.withResolvers<void>();
let held = false, reject = false, calls = 0, second = false;
const Notes = defineExtension({ name: 'input-test', hooks: [hook(GenerationTask, {
beforeInput: async (_input, { tx, conversationId }) => {
calls++;
const state = await tx.doc(Placed, conversationId);
state.count++;
if (reject) throw new Error('reject placement');
return { entries: [{ kind: 'input-test.note', data: { count: state.count },
model: [{ role: 'user', content: `before ${state.count}`, timestamp: 1 }] }] };
},
})] });
const Park = defineTask<null, { phase: 'first' | 'second' }, null>({ name: 'hold-test', version: 1,
initial: () => ({ phase: 'first' }), phases: {
first: async (_task, runtime, ctx) => {
held = true;
await gate.promise;
await runtime.commit(() => ({ status: 'pending', checkpoint: { phase: 'second' } }), ctx);
},
second: async (_task, runtime, ctx) => {
second = true;
await runtime.commit(() => ({ status: 'terminal', outcome: { status: 'completed', result: null } }), ctx);
},
}, abort: (_task, runtime, ctx) => runtime.commit(() => ({ status: 'terminal', outcome: { status: 'aborted' } }), ctx),
});
const models = createModels();
const faux = fauxProvider({ tokensPerSecond: 100000 });
models.setProvider(faux.provider);
const registry = createRegistry();
registry.install(Notes);
registry.install(defineExtension({ name: 'hold-test', tasks: [Park] }));
const storage = await openStorage(globalThis.storagePath);
const harness = await Harness.open(storage, { models, registry, settings: { followUpMode: 'all' } }, context);
const scenario = globalThis.foundationCase;
const root = await harness.root(context, { agent: { model: { provider: 'faux', modelId: 'faux-1' },
extensions: scenario === 'placement' ? [Notes] : [] } });
try {
if (scenario === 'placement' || scenario === 'followups') {
faux.setResponses([async () => {
held = true;
await gate.promise;
return fauxAssistantMessage([fauxToolCall('unknown', {})], { stopReason: 'toolUse' });
}, fauxAssistantMessage('first answer'), fauxAssistantMessage('follow-ups answer'),
fauxAssistantMessage('extra'), fauxAssistantMessage('extra')]);
const first = await root.submit({ type: 'input', content: 'first', requestId: 'one' }, context);
await until(() => held);
const queued = [];
const followups = scenario === 'placement' ? ['next'] : ['one', 'two', 'three'];
for (const content of followups) queued.push(await root.submit({ type: 'input', content, whenBusy: 'followUp' }, context));
const steer = await root.submit({ type: 'input', content: 'steer', whenBusy: 'steer' }, context);
if (scenario === 'placement') equal(calls, 1);
gate.resolve();
await Promise.all([first, steer, ...queued].map(submission => submission.wait(context)));
const view = await root.context(context);
if (scenario === 'placement') {
equal(view.messages.filter(m => m.role === 'user').map(m => m.content), ['before 1', 'first', 'steer', 'before 2', 'next']);
const notes = view.entries.filter(e => e.kind === 'input-test.note');
equal(notes.length, 2);
assert(notes.every(e => e.edits === undefined), 'Hook notes must be placed, not edits');
equal(calls, 2);
assert(view.entries.filter(e => e.kind === 'pi.user').every(e => e.input === true), 'Inputs need their marker');
await root.submit({ type: 'input', content: 'first', requestId: 'one' }, context);
equal(calls, 2);
reject = true;
let rejected = false;
try { await root.submit({ type: 'input', content: 'rejected' }, context); }
catch (error) { rejected = String(error).includes('reject placement'); }
assert(rejected, 'Placement must reject');
equal((await harness.snapshot(Placed, root.id, context))?.count, 2);
equal((await root.context(context)).entries.filter(e => e.kind === 'input-test.note').length, 2);
} else {
equal(view.messages.flatMap(m => m.role === 'user' ? [m.content] : m.role === 'assistant'
? m.content.flatMap(p => p.type === 'text' ? [`> ${p.text}`] : []) : []),
['first', 'steer', '> first answer', 'one', 'two', 'three', '> follow-ups answer']);
equal(await harness.snapshot(InboxDoc, root.id, context), { items: [] });
}
} else if (scenario === 'thinking') {
faux.setResponses([fauxAssistantMessage('unset'), fauxAssistantMessage('high')]);
await (await root.submit({ type: 'input', content: 'one' }, context)).wait(context);
await root.configure({ thinkingLevel: 'high' }, context);
await (await root.submit({ type: 'input', content: 'two' }, context)).wait(context);
equal((await root.context(context)).messages.flatMap(m => m.role === 'assistant' ? [m.thinkingLevel] : []), ['off', 'high']);
} else if (scenario === 'hold') {
const task = await root.commit(tx => tx.createTask(Park, null, { ownership: { kind: 'conversation' } }), context);
harness.resume();
await until(() => held);
harness.hold();
gate.resolve();
await until(async () => (await harness.getTask(task, context))?.state.status === 'pending');
equal((await harness.getTask(task, context))?.state.checkpoint.phase, 'second');
harness.resume();
equal((await harness.inspect(context)).scheduling, 'paused');
assert(!second, 'Held scheduler started another phase');
harness.release();
await until(() => second);
await harness.waitForIdle(context);
equal((await harness.getTask(task, context))?.state.status, 'terminal');
} else if (scenario === 'headless') {
faux.setResponses([fauxAssistantMessage('channel answer')]);
await serve(harness, context);
const submissions = await storage.scanSubmissions({ conversationId: root.id }, 100, undefined, context);
equal(submissions.items.length, 1);
equal(submissions.items[0].requestId, 'channel-request');
assert(JSON.stringify(await root.context(context)).includes('channel answer'), 'Channel answer missing');
} else throw new Error('Unknown foundation case');
} finally {
gate.resolve();
await harness.close(context);
}
host/js/input.ts
+20
View File
@@ -0,0 +1,20 @@
// Channel submissions target any conversation. EOF and Stop close the harness cleanly.
export async function serve(harness: any, context: any, afterSubmit?: () => Promise<void>) {
while (true) {
const input = await Deno.core.ops.op_input();
if (input === null || input.type === 'stop') {
await harness.waitForIdle(context);
return;
}
const conversation = await harness.conversation(input.conversation, context);
if (!conversation) throw new Error(`Unknown conversation: ${input.conversation}`);
const submission = await conversation.submit({
type: 'input', content: input.content, requestId: input.requestId,
}, context);
if (afterSubmit) {
const result = await submission.wait(context);
await afterSubmit();
if (result.status !== 'done') throw new Error(`Run ended: ${JSON.stringify(result)}`);
}
}
}
host/js/main.ts
+8 -12
View File
@@ -7,6 +7,7 @@ import { CodingTools } from '@earendil-works/pi-durable/tools';
import { openStorage } from './sqlite';
import { environment } from './environment';
import { CredentialRequests, startRequests } from './credentials';
import { serve } from './input';
const proxy = Deno.core.ops.op_model_proxy();
if (!proxy?.base_url || !proxy?.api_key) throw new Error('The llm-proxy credential is required');
const models = createModels();
@@ -27,8 +28,10 @@ models.setProvider({ ...provider, baseUrl: model.baseUrl, getModels: () => [mode
const registry = createRegistry();
registry.install(CodingTools);
registry.install(CredentialRequests);
const harness = await Harness.open(await openStorage(globalThis.storagePath), {
models, registry, env: (target: any) => environment(target.cwd ?? '/', target.conversationId),
export const storage = await openStorage(globalThis.storagePath);
const harness = await Harness.open(storage, {
models, registry, settings: { followUpMode: 'all' },
env: (target: any) => environment(target.cwd ?? '/', target.conversationId),
}, context);
const root = await harness.root(context, { agent: { model: { provider: 'openai', modelId: model.id }, thinkingLevel: 'low', cwd: '/', extensions: [CodingTools, CredentialRequests] } });
const watch = await root.watch(context);
@@ -53,23 +56,16 @@ const render = (view: any) => {
for (const entry of watch.value.entries) {
if (entry.kind === 'pi.assistant') emitted.set(entry.id, text(entry.model?.[0]));
}
watch.start((event: any) => { if (event.kind === 'value') render(event.value); });
if (globalThis.consoleEnabled) watch.start((event: any) => { if (event.kind === 'value') render(event.value); });
const stopRequests = await startRequests(harness, context);
harness.resume();
await harness.waitForIdle(context);
while (true) {
const line = await Deno.core.ops.op_input();
if (line === null) break;
if (!line.trim()) continue;
const submission = await root.submit({ type: 'input', content: line, requestId: crypto.randomUUID() }, context);
const result = await submission.wait(context);
await serve(harness, context, globalThis.consoleEnabled ? async () => {
await harness.waitForIdle(context);
const final = await root.watch(context);
render(final.value);
await final.stop();
Deno.core.ops.op_output('\n');
if (result.status !== 'done') throw new Error(`Run ended: ${JSON.stringify(result)}`);
}
} : undefined);
await stopRequests();
await watch.stop();
await harness.close(context);
host/js/package.json
+1 -1
View File
@@ -1,7 +1,7 @@
{
"private": true,
"type": "module",
"scripts": { "bundle": "esbuild main.ts --bundle --format=esm --platform=browser --outfile=bundle.js && esbuild test.ts --bundle --format=esm --platform=browser --outfile=test-bundle.js && esbuild durable-test.ts --bundle --format=esm --platform=browser --outfile=durable-test-bundle.js" },
"scripts": { "bundle": "nu apply-patch.nu && esbuild main.ts --bundle --format=esm --platform=browser --outfile=bundle.js && esbuild test.ts --bundle --format=esm --platform=browser --outfile=test-bundle.js && esbuild durable-test.ts --bundle --format=esm --platform=browser --outfile=durable-test-bundle.js && esbuild foundation-test.ts --bundle --format=esm --platform=browser --outfile=foundation-test-bundle.js" },
"dependencies": {
"@earendil-works/pi-durable": "1.5.0",
"@earendil-works/pi-ai": "1.5.0",
host/js/patches/pi-durable-1.5.0.patch
+387
View File
@@ -0,0 +1,387 @@
diff --git a/dist/harness/generation.js b/dist/harness/generation.js
index ac42dfe741b20726ab8eb3d8139b1296d664210d..1e7f1ac26d17b83cb15bf00408fa70b0aee4e747 100644
--- a/dist/harness/generation.js
+++ b/dist/harness/generation.js
@@ -458,7 +458,7 @@ async function prepareFailure(tx, runtime) {
const policy = settings.afterFailedRun;
if (policy !== "continue")
return { policy };
- return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings) };
+ return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings, "final", runtime.now()) };
}
/**
* End a failed run, settling its inputs with `settlement`, and do with the inputs queued behind it what
@@ -668,7 +668,7 @@ async function answer(runtime, message, context) {
const conversationId = runtime.conversationId;
await runtime.commit(async (tx) => {
// Queue modes are read on the Session line, when the boundary is decided.
- const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
+ const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
const live = await tx.doc(LiveDoc, conversationId);
const entry = await appendAssistant(tx, conversationId, message);
const result = { status: "terminal", outcome: { status: "completed", result: { entryId: entry.id } } };
@@ -762,7 +762,7 @@ function prepareRunEndTasks(runtime, requests, prepared) {
async function endAnsweredRun(runtime, answer, context) {
const conversationId = runtime.conversationId;
await runtime.commit(async (tx) => {
- const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
+ const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
const live = await tx.doc(LiveDoc, conversationId);
const { users, first } = await applyBoundary(tx, boundary, "final", runtime.now());
endRun(tx, live, runtime.taskId, { status: "done", answer });
@@ -868,7 +868,7 @@ async function finishToolRound(runtime, assistant, tools, context) {
const maxTurns = (await runtime.agent(context)).maxTurns;
const afterRun = terminate && (await asksAfterRun(runtime, context));
await runtime.commit(async (tx) => {
- const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
+ const boundary = await prepareBoundary(tx, conversationId, runtime.settings, terminate || handoff !== undefined ? "final" : "postTools", runtime.now());
if (added.length > 0)
await addTools(tx, conversationId, added);
const live = await tx.doc(LiveDoc, conversationId);
diff --git a/dist/harness/harness.js b/dist/harness/harness.js
index e286f00f872ebec79367a594938040c6a37a879f..44979b05c2bcf30580841797175c099c8c621462 100644
--- a/dist/harness/harness.js
+++ b/dist/harness/harness.js
@@ -5,7 +5,7 @@ import { ResetEntry } from "../entries.js";
import { contained, SessionImpl } from "../session/session.js";
import { hostTimers } from "../timers.js";
import { ROOT_CONVERSATION_ID } from "../types.js";
-import { AgentDoc, configure, createAgent, resolveAgent, resolveSettings } from "./agent.js";
+import { AgentDoc, agentHooks, configure, createAgent, resolveAgent, resolveSettings } from "./agent.js";
import { createCompaction } from "./compaction.js";
import { readContext } from "./context.js";
import { InboxDoc, withdrawQueuedInputs } from "./inbox.js";
@@ -157,6 +157,15 @@ class HarnessImpl extends SessionImpl {
create: (target, createOptions, context) => this.#create(target, createOptions, context),
};
}
+ async beforeInput(tx, conversationId, input, context) {
+ const agent = resolveAgent(await tx.doc(AgentDoc, conversationId), this.#options.registry.snapshot(), resolveSettings(this.#hostSettings()), (error, extension) => this.#report(error, { kind: "extension_error", conversationId, extension }, context));
+ const entries = [];
+ for (const { handlers } of agentHooks(agent, "pi.generation")) {
+ const result = await handlers.beforeInput?.(input, { conversationId, tx }, context);
+ entries.push(...(result?.entries ?? []));
+ }
+ return entries;
+ }
/** Resolve a conversation's committed `pi.agent` against `snapshot`, or the current one, and the current settings. */
async resolveAgent(id, snapshot, context) {
const registry = snapshot ?? this.#options.registry.snapshot();
@@ -179,6 +188,12 @@ class HarnessImpl extends SessionImpl {
this.assertUsable();
this.#tasks.resume();
}
+ hold() {
+ this.#tasks.hold();
+ }
+ release() {
+ this.#tasks.release();
+ }
wake() {
this.assertUsable();
return this.#tasks.wake();
diff --git a/dist/harness/inbox.d.ts b/dist/harness/inbox.d.ts
index 1197339369188857a1ad62b528cea65e3833f27e..a61d9712ce55e80441847442772da6f3e354fb2b 100644
--- a/dist/harness/inbox.d.ts
+++ b/dist/harness/inbox.d.ts
@@ -23,6 +23,7 @@ export declare const InboxDoc: import("../types.ts").ConversationDocToken<InboxS
export type QueueModes = Pick<Settings, "steeringMode" | "followUpMode">;
/** What a boundary reads before the commit's first table write, and the newest head it has seen so far. */
export type Boundary = {
+ readonly entries: readonly EntryDraft[];
readonly conversationId: ConversationId;
readonly inbox: Draft<InboxState>;
readonly steeringMode: QueueMode;
@@ -43,7 +44,7 @@ export type BoundaryResult = {
* Read what a boundary needs. Table reads must precede the commit's first table write, so callers prepare the
* boundary at the start of their commit.
*/
-export declare function prepareBoundary(tx: Tx, conversationId: ConversationId, modes: QueueModes): Promise<Boundary>;
+export declare function prepareBoundary(tx: Tx, conversationId: ConversationId, modes: QueueModes, at: "postTools" | "final", now: number, incoming?: EntryDraft): Promise<Boundary>;
/**
* Place the queued items a boundary selects (spec §6): every write, the first or all steers, and at `final` the first
* or all follow-ups. A selected reset turns a `postTools` boundary into `final`. Writes are placed first and user
@@ -53,6 +54,7 @@ export declare function prepareBoundary(tx: Tx, conversationId: ConversationId,
*/
export declare function applyBoundary(tx: Tx, boundary: Boundary, at: "postTools" | "final", now: number): Promise<BoundaryResult>;
/** Append the entry an input is placed as: `entry`'s kind and data, or `pi.user`, with the user message of `content`. */
+export declare function inputDraft(content: UserInput, entry: InputEntry | undefined, now: number): EntryDraft;
export declare function appendInput(tx: Tx, conversationId: ConversationId, content: UserInput, entry: InputEntry | undefined, now: number): Promise<EntryRecord>;
/** The kind and data of an input's entry as strict JSON, without other fields an untyped caller might pass. */
export declare function pickInputEntry({ kind, data }: InputEntry): JsonRepresentation<InputEntry>;
diff --git a/dist/harness/inbox.js b/dist/harness/inbox.js
index 5ff8c1703808c5ffd703164e6c339150e1f1e131..ec4cf6d5adc599e5fe428514a7cd7e0dac1a7c9d 100644
--- a/dist/harness/inbox.js
+++ b/dist/harness/inbox.js
@@ -14,10 +14,15 @@ export const InboxDoc = defineDoc({
* Read what a boundary needs. Table reads must precede the commit's first table write, so callers prepare the
* boundary at the start of their commit.
*/
-export async function prepareBoundary(tx, conversationId, modes) {
+export async function prepareBoundary(tx, conversationId, modes, at, now, incoming) {
const head = (await tx.latestHeadMarker(conversationId))?.head;
const inbox = await tx.doc(InboxDoc, conversationId);
+ const first = inbox.items.find(item => item.mode === "steer" || item.mode === "followUp");
+ const input = first === undefined ? incoming : inputDraft(first.content, first.entry, now);
+ const entries = input !== undefined && (at === "final" || hasQueuedReset(inbox.items))
+ ? await tx.beforeInput(conversationId, input) : [];
return {
+ entries,
conversationId,
inbox,
steeringMode: modes.steeringMode,
@@ -61,6 +66,9 @@ export async function applyBoundary(tx, boundary, at, now) {
}
const placed = [];
for (const index of users) {
+ if (placed.length === 0)
+ for (const draft of boundary.entries)
+ await tx.appendEntry(conversationId, draft);
const item = items[index];
const entry = await appendInput(tx, conversationId, item.content, item.entry, now);
tx.placeSubmission(item.id, entry.id);
@@ -74,10 +82,11 @@ export async function applyBoundary(tx, boundary, at, now) {
}
/** Append the entry an input is placed as: `entry`'s kind and data, or `pi.user`, with the user message of `content`. */
export function appendInput(tx, conversationId, content, entry, now) {
+ return tx.appendEntry(conversationId, inputDraft(content, entry, now));
+}
+export function inputDraft(content, entry, now) {
const message = { role: "user", content, timestamp: now };
- if (entry === undefined)
- return tx.appendEntry(UserEntry, conversationId, { model: [message] });
- return tx.appendEntry(conversationId, { ...pickInputEntry(entry), model: [message] });
+ return { ...(entry === undefined ? { kind: UserEntry.kind } : pickInputEntry(entry)), input: true, model: [message] };
}
/** The kind and data of an input's entry as strict JSON, without other fields an untyped caller might pass. */
export function pickInputEntry({ kind, data }) {
diff --git a/dist/harness/scheduler.d.ts b/dist/harness/scheduler.d.ts
index f6189edb71ac0557060b623625fc4be4bec5e904..269a3c4ab1676da22db5a723f30d2974f84b6bea 100644
--- a/dist/harness/scheduler.d.ts
+++ b/dist/harness/scheduler.d.ts
@@ -107,6 +107,8 @@ export declare class TaskScheduler {
* Idempotent; the kick does nothing once closing.
*/
resume(): void;
+ hold(): void;
+ release(): void;
/**
* Why a live task cannot run here, if it cannot: its definition is missing, too old, or failed to migrate. A running
* task, an abort-marked one, and one holding its outcome are never blocked.
diff --git a/dist/harness/scheduler.js b/dist/harness/scheduler.js
index 365ea3ab42dd4406ab11a03d6702b238dfd0776b..3e8170fdb790e492e95a390fa6ea4693eb3d2f97 100644
--- a/dist/harness/scheduler.js
+++ b/dist/harness/scheduler.js
@@ -179,6 +179,7 @@ export class TaskScheduler {
#cascadePending = false;
#unsubscribeRegistry = () => { };
#enabled = false;
+ #held = false;
#readOnly;
#closing = false;
#dirty = false;
@@ -207,6 +208,14 @@ export class TaskScheduler {
this.#context = options.context;
this.#readOnly = options.readOnly === true;
}
+ /** Start no run invocation until `release()`; running invocations and abort handlers go on. Overrides `resume()`. */
+ hold() {
+ this.#held = true;
+ }
+ release() {
+ this.#held = false;
+ this.#kick();
+ }
/**
* Load live tasks and change surviving `running` tasks back to `pending`, counting the restart. Dispatches nothing;
* a task cut off `settings.restartLimit` times ends when it is next reserved. Read-only, it only loads them: a
@@ -1771,6 +1780,8 @@ export class TaskScheduler {
for (const record of plan.ready) {
const id = record.id;
const mode = record.abortRequested ? "abort" : "run";
+ if (mode === "run" && this.#held)
+ continue;
// Work cut off `restartLimit` times in a row would run forever: an abort handler, or a phase of a kind that
// does not end itself at the limit, ends the task `faulted` instead.
// The settings are read only for a task that was cut off at all.
@@ -1973,7 +1984,7 @@ export class TaskScheduler {
const tasks = [];
for (const record of this.#live.values())
tasks.push({ record, state: this.#inspectTask(record, snapshot) });
- const scheduling = this.#closing ? "closing" : this.#enabled ? "running" : "paused";
+ const scheduling = this.#closing ? "closing" : this.#enabled && !this.#held ? "running" : "paused";
return { scheduling, tasks };
}
#inspectTask(record, snapshot) {
@@ -2151,6 +2162,11 @@ export class TaskScheduler {
* invocation continues.
*/
#progress(tx, current, state, extension, context) {
+ // Held: park at this phase boundary, as a handover does; reservation starts it again after `release()`.
+ if (this.#held) {
+ tx.setTask(withState(current, { status: "pending", checkpoint: current.state.checkpoint }));
+ return false;
+ }
state.snapshot = this.#registry.snapshot();
const next = state.snapshot.task(current.kind);
if (next === state.task)
diff --git a/dist/harness/submissions.js b/dist/harness/submissions.js
index 6e882a8c252dda942b8b45e5b1837cf2fee28341..89a85e8a52bf67044ff15c3c856a650dea4c7c27 100644
--- a/dist/harness/submissions.js
+++ b/dist/harness/submissions.js
@@ -3,7 +3,7 @@ import { withTelemetry } from "@earendil-works/chord/context";
import { createTypedSpanStarter, NOOP_TELEMETRY_CONTEXT, } from "@earendil-works/pi-telemetry";
import { ConversationBusy, ConversationIdle, ReadAfterWrite } from "../errors.js";
import { startRun } from "./generation.js";
-import { appendInput, applyBoundary, hasQueuedReset, InboxDoc, isDisplayOnly, isStale, pickInputEntry, prepareBoundary, removeInboxItem, } from "./inbox.js";
+import { appendInput, inputDraft, applyBoundary, hasQueuedReset, InboxDoc, isDisplayOnly, isStale, pickInputEntry, prepareBoundary, removeInboxItem, } from "./inbox.js";
import { LiveDoc } from "./live.js";
import { DURABLE_TELEMETRY_SCHEMA } from "./telemetry.js";
import { closedError, Waiters } from "./util.js";
@@ -205,7 +205,7 @@ export async function admitSubmission(tx, conversationId, draft, now, queueModes
// Never queued: whether a run can continue depends on how the current one ends. Rejected before any write.
if (busy)
throw new ConversationBusy(conversationId);
- const boundary = await prepareBoundary(tx, conversationId, queueModes);
+ const boundary = await prepareBoundary(tx, conversationId, queueModes, "final", now);
const { id } = await tx.createSubmission({ ...fields, type: "continue", status: "placed" });
// Items an earlier failed run left queued (`afterFailedRun: "keep"`) are placed first and join the run. Its history
// starts with the failed run's, so `afterRun` hooks see what that run did.
@@ -214,7 +214,7 @@ export async function admitSubmission(tx, conversationId, draft, now, queueModes
return { id, status: "placed" };
}
// A boundary reads the table, so it is prepared before the first table write; a busy one needs none.
- const boundary = busy ? undefined : await prepareBoundary(tx, conversationId, queueModes);
+ const boundary = busy ? undefined : await prepareBoundary(tx, conversationId, queueModes, "final", now, draft.type === "input" ? inputDraft(draft.content, draft.entry, now) : undefined);
if (boundary === undefined || boundary.inbox.items.length > 0) {
const { id } = await tx.createSubmission({
...fields,
@@ -253,6 +253,8 @@ export async function admitSubmission(tx, conversationId, draft, now, queueModes
const write = { ...fields, type: "write", status: "done", entry: entry.id };
return { id: (await tx.createSubmission(write)).id, status: "done" };
}
+ for (const before of boundary.entries)
+ await tx.appendEntry(conversationId, before);
const entry = await appendInput(tx, conversationId, draft.content, draft.entry, now);
const input = { ...fields, type: "input", status: "placed", entry: entry.id };
const { id } = await tx.createSubmission(input);
diff --git a/dist/harness/types.d.ts b/dist/harness/types.d.ts
index eb64e474162b158a43563e6c68d8ffea8c18b74a..cce695c6d87c79dadea54560bafcca62cf073ca1 100644
--- a/dist/harness/types.d.ts
+++ b/dist/harness/types.d.ts
@@ -816,6 +816,9 @@ export interface Harness extends Session {
* never do.
*/
resume(): void;
+ /** Start no task run until `release()`, whatever `resume()` says; running invocations and abort handlers go on. */
+ hold(): void;
+ release(): void;
/**
* When this Session next needs a running process, from the scheduler's memory: `active` while some task can run now
* or is running, `at` for the earliest sleep. Answers also while scheduling is paused. A host that wakes the
@@ -888,6 +891,8 @@ export interface HookApi extends DocumentReader {
export type HookResult<T> = T | undefined | Promise<T | undefined>;
/** Hooks of the built-in generation task. */
export interface GenerationHooks {
+ /** Before a new run’s first input is placed. Use only tx for storage access; effects commit with the input. */
+ beforeInput(input: EntryDraft, api: { readonly conversationId: ConversationId; readonly tx: Tx }, context: Context): HookResult<{ readonly entries?: readonly EntryDraft[] }>;
/** Before every request attempt, including recovery; the result is used for that request only. */
beforeRequest(request: {
readonly messages: readonly Message[];
diff --git a/dist/session/session.d.ts b/dist/session/session.d.ts
index 8201d7cfefb710516a3d4d89ba208adc2cbebd19..37c9f362401b5bc81f2f24233aa3dd7f57e29204 100644
--- a/dist/session/session.d.ts
+++ b/dist/session/session.d.ts
@@ -1,5 +1,5 @@
import { type Context, type JsonValue, type WatchEnd } from "@earendil-works/chord";
-import type { CommitChange, CommitObservation, CommitObserver, CommitOptions, ConversationDocFamilyToken, ConversationDocToken, ConversationId, ConversationRecord, DocumentCommitChange, DocumentRecord, DocumentWatch, EntryId, JsonObject, Observation, Observer, ReportInfo, RewindableConversationDocFamilyToken, RewindableConversationDocToken, Seq, Session, SessionDocFamilyToken, SessionDocToken, SessionEnd, SetTimer, Storage, StorageWrite, TaskDocFamilyToken, TaskDocToken, TaskId, Tx } from "../types.ts";
+import type { CommitChange, CommitObservation, CommitObserver, CommitOptions, ConversationDocFamilyToken, ConversationDocToken, EntryDraft, ConversationId, ConversationRecord, DocumentCommitChange, DocumentRecord, DocumentWatch, EntryId, JsonObject, Observation, Observer, ReportInfo, RewindableConversationDocFamilyToken, RewindableConversationDocToken, Seq, Session, SessionDocFamilyToken, SessionDocToken, SessionEnd, SetTimer, Storage, StorageWrite, TaskDocFamilyToken, TaskDocToken, TaskId, Tx } from "../types.ts";
import { type HeadMarker } from "./conversation-tails.ts";
import { Transaction, type TransactionScope } from "./transaction.ts";
/**
@@ -171,6 +171,7 @@ export declare class SessionImpl implements Session {
* listener. A plain Session drops them.
*/
report(_error: unknown, _info?: ReportInfo, _context?: Context): void;
+ protected beforeInput(tx: Transaction, id: ConversationId, input: EntryDraft, context: Context): Promise<readonly EntryDraft[]>;
/**
* Runs inside every transaction that creates or forks a conversation, after the conversation record is staged. A
* plain Session stages nothing; a Harness stages its built-in documents.
diff --git a/dist/session/session.js b/dist/session/session.js
index e9ae9ebfc2c45765ccda141d7e60cdc9a15a0894..e1e5c69deed9f0d09cf67c26cdb903fe4593f9c6 100644
--- a/dist/session/session.js
+++ b/dist/session/session.js
@@ -224,6 +224,7 @@ export class SessionImpl {
if (this.#documents.get(id)?.record.id === recordId)
this.#dropDocument(id);
},
+ beforeInput: (tx, id, input, context) => this.beforeInput(tx, id, input, context),
conversationCreated: (tx, record) => this.conversationCreated(tx, record),
taskDocuments: (taskId) => this.#taskDocuments.get(taskId)?.values(),
latestEntryId: (conversationId, context) => this.latestEntryId(conversationId, context),
@@ -570,6 +571,9 @@ export class SessionImpl {
* listener. A plain Session drops them.
*/
report(_error, _info = LISTENER, _context) { }
+ async beforeInput(_tx, _id, _input, _context) {
+ return [];
+ }
/**
* Runs inside every transaction that creates or forks a conversation, after the conversation record is staged. A
* plain Session stages nothing; a Harness stages its built-in documents.
diff --git a/dist/session/transaction.d.ts b/dist/session/transaction.d.ts
index 5508fa2d021d2153f4f104b4ef69b99ef98a6a74..1c94a4a6e29b8371d49f7d6cb4d2e4e7cd6aec50 100644
--- a/dist/session/transaction.d.ts
+++ b/dist/session/transaction.d.ts
@@ -48,6 +48,7 @@ export interface TransactionHost {
install(document: LoadedDocument): void;
/** Remove a retired incarnation if it is still the cached occupant of its address. */
evict(addressId: string, recordId: DocumentId): void;
+ beforeInput(tx: Transaction, conversationId: ConversationId, input: EntryDraft, context: Context): Promise<readonly EntryDraft[]>;
/** Stage writes that belong to every newly created or forked conversation, in its creating transaction. */
conversationCreated(tx: Transaction, record: ConversationRecord): Promise<void>;
/** The newest entry ID visible from a conversation, known from the Session's commits. */
@@ -112,6 +113,7 @@ export declare class Transaction implements Tx {
forkConversation(parentConversationId: ConversationId, at: EntryId, options: {
readonly ownership: ConversationOwnership;
}): Promise<ConversationRecord>;
+ beforeInput(conversationId: ConversationId, input: EntryDraft): Promise<readonly EntryDraft[]>;
appendEntry(conversationId: ConversationId, value: EntryDraft): Promise<EntryRecord>;
appendEntry<D extends JsonValue>(token: Entry<D>, conversationId: ConversationId, value: TypedEntryDraft<NoInfer<D>>): Promise<TypedEntry<D>>;
createTask<I, S extends {
diff --git a/dist/session/transaction.js b/dist/session/transaction.js
index 7fbe75b11bb2a9ab28f4f23800143231172d71b7..a516853e2f9c4a4437dbc2bb8f1c8d2117386e9a 100644
--- a/dist/session/transaction.js
+++ b/dist/session/transaction.js
@@ -167,6 +167,9 @@ export class Transaction {
this.#assertOpen();
return record;
}
+ beforeInput(conversationId, input) {
+ return this.#read("beforeInput", () => this.#host.beforeInput(this, conversationId, input, this.#context));
+ }
appendEntry(first, second, third) {
if (typeof first === "number")
return this.#appendEntry(first, second);
diff --git a/dist/types.d.ts b/dist/types.d.ts
index b43143c5e7998b36034ffe9577f74f3d88ab932b..a9ca10ae41c48ad0ebb813e62caa3699dec5c51e 100644
--- a/dist/types.d.ts
+++ b/dist/types.d.ts
@@ -409,6 +409,8 @@ export type ContextEdit = {
});
/** Immutable transcript event with separate model-facing and application-facing payloads. */
export type EntryRecord = {
+ /** Placed by the harness from an input submission, not a write. */
+ readonly input?: true;
readonly id: EntryId;
readonly conversationId: ConversationId;
/** Application-defined entry discriminator. */
@@ -891,6 +893,8 @@ export type CommitPublication = {
* Session may freeze them (its caches and context ranges keep them), so mutating one can throw.
*/
export interface Tx {
+ /** Prepare entries before a new run’s first input; called before table writes, in the placement transaction. */
+ beforeInput(conversationId: ConversationId, input: EntryDraft): Promise<readonly EntryDraft[]>;
conversation(id: ConversationId): Promise<ConversationRecord | undefined>;
entry(id: EntryId): Promise<EntryRecord | undefined>;
/** Undefined when the entry is absent or has another kind. */
host/js/test-bundle.js
+58 -11
View File
@@ -8940,6 +8940,9 @@ var Transaction = class {
this.#assertOpen();
return record;
}
beforeInput(conversationId, input) {
return this.#read("beforeInput", () => this.#host.beforeInput(this, conversationId, input, this.#context));
}
appendEntry(first, second, third) {
if (typeof first === "number")
return this.#appendEntry(first, second);
@@ -10122,6 +10125,7 @@ var SessionImpl = class {
if (this.#documents.get(id)?.record.id === recordId)
this.#dropDocument(id);
},
beforeInput: (tx, id, input, context) => this.beforeInput(tx, id, input, context),
conversationCreated: (tx, record) => this.conversationCreated(tx, record),
taskDocuments: (taskId) => this.#taskDocuments.get(taskId)?.values(),
latestEntryId: (conversationId, context) => this.latestEntryId(conversationId, context),
@@ -10446,6 +10450,9 @@ var SessionImpl = class {
*/
report(_error, _info = LISTENER, _context) {
}
async beforeInput(_tx, _id, _input, _context) {
return [];
}
/**
* Runs inside every transaction that creates or forks a conversation, after the conversation record is staged. A
* plain Session stages nothing; a Harness stages its built-in documents.
@@ -12250,10 +12257,14 @@ var InboxDoc = defineDoc({
initial: () => ({ items: [] }),
checkpointWhen: (value2) => value2.items.length === 0
});
async function prepareBoundary(tx, conversationId, modes) {
async function prepareBoundary(tx, conversationId, modes, at, now, incoming) {
const head = (await tx.latestHeadMarker(conversationId))?.head;
const inbox = await tx.doc(InboxDoc, conversationId);
const first = inbox.items.find((item) => item.mode === "steer" || item.mode === "followUp");
const input = first === void 0 ? incoming : inputDraft(first.content, first.entry, now);
const entries2 = input !== void 0 && (at === "final" || hasQueuedReset(inbox.items)) ? await tx.beforeInput(conversationId, input) : [];
return {
entries: entries2,
conversationId,
inbox,
steeringMode: modes.steeringMode,
@@ -12289,6 +12300,9 @@ async function applyBoundary(tx, boundary, at, now) {
}
const placed = [];
for (const index of users) {
if (placed.length === 0)
for (const draft of boundary.entries)
await tx.appendEntry(conversationId, draft);
const item = items[index];
const entry2 = await appendInput(tx, conversationId, item.content, item.entry, now);
tx.placeSubmission(item.id, entry2.id);
@@ -12301,10 +12315,11 @@ async function applyBoundary(tx, boundary, at, now) {
return { users: placed, first, reset };
}
function appendInput(tx, conversationId, content, entry2, now) {
return tx.appendEntry(conversationId, inputDraft(content, entry2, now));
}
function inputDraft(content, entry2, now) {
const message = { role: "user", content, timestamp: now };
if (entry2 === void 0)
return tx.appendEntry(UserEntry, conversationId, { model: [message] });
return tx.appendEntry(conversationId, { ...pickInputEntry(entry2), model: [message] });
return { ...entry2 === void 0 ? { kind: UserEntry.kind } : pickInputEntry(entry2), input: true, model: [message] };
}
function pickInputEntry({ kind, data }) {
const picked = data === void 0 ? { kind } : { kind, data };
@@ -13016,6 +13031,7 @@ var TaskScheduler = class {
#unsubscribeRegistry = () => {
};
#enabled = false;
#held = false;
#readOnly;
#closing = false;
#dirty = false;
@@ -13044,6 +13060,14 @@ var TaskScheduler = class {
this.#context = options.context;
this.#readOnly = options.readOnly === true;
}
/** Start no run invocation until `release()`; running invocations and abort handlers go on. Overrides `resume()`. */
hold() {
this.#held = true;
}
release() {
this.#held = false;
this.#kick();
}
/**
* Load live tasks and change surviving `running` tasks back to `pending`, counting the restart. Dispatches nothing;
* a task cut off `settings.restartLimit` times ends when it is next reserved. Read-only, it only loads them: a
@@ -14498,6 +14522,8 @@ var TaskScheduler = class {
for (const record of plan.ready) {
const id = record.id;
const mode = record.abortRequested ? "abort" : "run";
if (mode === "run" && this.#held)
continue;
const cutOff = record.restarts !== void 0 && record.restarts >= this.#restartLimit();
if (cutOff && (mode === "abort" || !this.#limitsRestarts(record.kind))) {
const message = mode === "abort" ? `Abort handler of task ${id} (${record.kind}) was cut off ${record.restarts} times` : `Task ${id} (${record.kind}) was cut off ${record.restarts} times without progress`;
@@ -14683,7 +14709,7 @@ var TaskScheduler = class {
const tasks = [];
for (const record of this.#live.values())
tasks.push({ record, state: this.#inspectTask(record, snapshot) });
const scheduling = this.#closing ? "closing" : this.#enabled ? "running" : "paused";
const scheduling = this.#closing ? "closing" : this.#enabled && !this.#held ? "running" : "paused";
return { scheduling, tasks };
}
#inspectTask(record, snapshot) {
@@ -14838,6 +14864,10 @@ var TaskScheduler = class {
* invocation continues.
*/
#progress(tx, current, state, extension, context) {
if (this.#held) {
tx.setTask(withState(current, { status: "pending", checkpoint: current.state.checkpoint }));
return false;
}
state.snapshot = this.#registry.snapshot();
const next = state.snapshot.task(current.kind);
if (next === state.task)
@@ -17571,7 +17601,7 @@ async function prepareFailure(tx, runtime) {
const policy = settings.afterFailedRun;
if (policy !== "continue")
return { policy };
return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings) };
return { policy, boundary: await prepareBoundary(tx, runtime.conversationId, settings, "final", runtime.now()) };
}
async function endFailedRun(tx, runtime, live, failure, end) {
const conversationId = runtime.conversationId;
@@ -17739,7 +17769,7 @@ async function answer(runtime, message, context) {
const afterRun = await asksAfterRun(runtime, context);
const conversationId = runtime.conversationId;
await runtime.commit(async (tx) => {
const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
const live = await tx.doc(LiveDoc, conversationId);
const entry2 = await appendAssistant(tx, conversationId, message);
const result2 = { status: "terminal", outcome: { status: "completed", result: { entryId: entry2.id } } };
@@ -17812,7 +17842,7 @@ function prepareRunEndTasks(runtime, requests, prepared) {
async function endAnsweredRun(runtime, answer2, context) {
const conversationId = runtime.conversationId;
await runtime.commit(async (tx) => {
const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
const boundary = await prepareBoundary(tx, conversationId, runtime.settings, "final", runtime.now());
const live = await tx.doc(LiveDoc, conversationId);
const { users, first } = await applyBoundary(tx, boundary, "final", runtime.now());
endRun(tx, live, runtime.taskId, { status: "done", answer: answer2 });
@@ -17885,7 +17915,7 @@ async function finishToolRound(runtime, assistant, tools, context) {
const maxTurns = (await runtime.agent(context)).maxTurns;
const afterRun = terminate && await asksAfterRun(runtime, context);
await runtime.commit(async (tx) => {
const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
const boundary = await prepareBoundary(tx, conversationId, runtime.settings, terminate || handoff !== void 0 ? "final" : "postTools", runtime.now());
if (added.length > 0)
await addTools(tx, conversationId, added);
const live = await tx.doc(LiveDoc, conversationId);
@@ -18301,13 +18331,13 @@ async function admitSubmission(tx, conversationId, draft, now, queueModes, span)
if (draft.type === "continue") {
if (busy)
throw new ConversationBusy(conversationId);
const boundary2 = await prepareBoundary(tx, conversationId, queueModes);
const boundary2 = await prepareBoundary(tx, conversationId, queueModes, "final", now);
const { id: id2 } = await tx.createSubmission({ ...fields, type: "continue", status: "placed" });
const { users, first } = await applyBoundary(tx, boundary2, "final", now);
await startRun(tx, conversationId, live, { inputs: [...users, id2], first: live.runFailure?.first ?? first });
return { id: id2, status: "placed" };
}
const boundary = busy ? void 0 : await prepareBoundary(tx, conversationId, queueModes);
const boundary = busy ? void 0 : await prepareBoundary(tx, conversationId, queueModes, "final", now, draft.type === "input" ? inputDraft(draft.content, draft.entry, now) : void 0);
if (boundary === void 0 || boundary.inbox.items.length > 0) {
const { id: id2 } = await tx.createSubmission({
...fields,
@@ -18345,6 +18375,8 @@ async function admitSubmission(tx, conversationId, draft, now, queueModes, span)
const write = { ...fields, type: "write", status: "done", entry: entry3.id };
return { id: (await tx.createSubmission(write)).id, status: "done" };
}
for (const before of boundary.entries)
await tx.appendEntry(conversationId, before);
const entry2 = await appendInput(tx, conversationId, draft.content, draft.entry, now);
const input = { ...fields, type: "input", status: "placed", entry: entry2.id };
const { id } = await tx.createSubmission(input);
@@ -19360,6 +19392,15 @@ var HarnessImpl = class extends SessionImpl {
create: (target, createOptions, context2) => this.#create(target, createOptions, context2)
};
}
async beforeInput(tx, conversationId, input, context) {
const agent = resolveAgent(await tx.doc(AgentDoc, conversationId), this.#options.registry.snapshot(), resolveSettings(this.#hostSettings()), (error, extension) => this.#report(error, { kind: "extension_error", conversationId, extension }, context));
const entries2 = [];
for (const { handlers } of agentHooks(agent, "pi.generation")) {
const result2 = await handlers.beforeInput?.(input, { conversationId, tx }, context);
entries2.push(...result2?.entries ?? []);
}
return entries2;
}
/** Resolve a conversation's committed `pi.agent` against `snapshot`, or the current one, and the current settings. */
async resolveAgent(id, snapshot, context) {
const registry2 = snapshot ?? this.#options.registry.snapshot();
@@ -19382,6 +19423,12 @@ var HarnessImpl = class extends SessionImpl {
this.assertUsable();
this.#tasks.resume();
}
hold() {
this.#tasks.hold();
}
release() {
this.#tasks.release();
}
wake() {
this.assertUsable();
return this.#tasks.wake();
host/src/api/server.rs
+1 -6
View File
@@ -124,12 +124,7 @@ pub async fn run_with_userland(
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?
.block_on(crate::harness::run_module_with_credentials(
world,
Arc::new(runtime),
cred,
include_str!("../../js/bundle.js"),
))
.block_on(crate::harness::run_console(world, Arc::new(runtime), cred))
})
.await
.map_err(anyhow::Error::from)
host/src/harness/environment.rs
-13
View File
@@ -129,19 +129,6 @@ pub async fn op_exec_next(
Ok(value)
}
#[op2]
#[string]
pub async fn op_input() -> Result<Option<String>, JsErrorBox> {
use tokio::io::AsyncBufReadExt;
let mut reader = tokio::io::BufReader::with_capacity(1, tokio::io::stdin());
let mut line = String::new();
let n = reader
.read_line(&mut line)
.await
.map_err(|e| JsErrorBox::generic(e.to_string()))?;
Ok((n != 0).then_some(line))
}
#[op2(fast)]
pub fn op_output(#[string] text: &str) {
print!("{text}");
host/src/harness/input.rs
+74
View File
@@ -0,0 +1,74 @@
//! Channel input for a headless harness. The console and stop signals are clients.
use deno_core::{OpState, op2};
use deno_error::JsErrorBox;
use serde::Serialize;
use serde_json::Value;
use std::{cell::RefCell, rc::Rc};
use tokio::sync::mpsc;
#[derive(Serialize)]
#[serde(tag = "type", rename_all = "camelCase")]
pub enum Input {
Submit {
conversation: u64,
content: Value,
#[serde(rename = "requestId")]
request_id: String,
},
Stop,
}
pub type Sender = mpsc::UnboundedSender<Input>;
pub type Receiver = mpsc::UnboundedReceiver<Input>;
pub fn channel() -> (Sender, Receiver) {
mpsc::unbounded_channel()
}
pub(super) struct Inbox(pub tokio::sync::Mutex<Receiver>);
#[op2]
#[serde]
pub async fn op_input(state: Rc<RefCell<OpState>>) -> Result<serde_json::Value, JsErrorBox> {
let inbox = state.borrow().borrow::<Rc<Inbox>>().clone();
serde_json::to_value(inbox.0.lock().await.recv().await)
.map_err(|error| JsErrorBox::generic(error.to_string()))
}
pub(super) fn console(sender: Sender) {
std::thread::spawn(move || {
use std::io::BufRead;
for line in std::io::stdin().lock().lines() {
let Ok(line) = line else { break };
if line.trim().is_empty() {
continue;
}
if sender
.send(Input::Submit {
conversation: 1,
content: Value::String(line),
request_id: uuid::Uuid::new_v4().to_string(),
})
.is_err()
{
break;
}
}
// EOF closes only this client. The signal client keeps the harness alive.
});
}
pub(super) fn stop_signal(
sender: Sender,
) -> std::io::Result<impl std::future::Future<Output = ()> + Send> {
// Register before V8 starts, so a stop cannot arrive before the listener.
let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
let mut interrupt = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())?;
Ok(async move {
tokio::select! {
_ = interrupt.recv() => {}
_ = terminate.recv() => {}
}
let _ = sender.send(Input::Stop);
})
}
host/src/harness/mod.rs
+44 -2
View File
@@ -1,6 +1,7 @@
//! Durable V8 harness, host-side SQLite storage, and world tool operations.
mod credentials;
mod environment;
pub mod input;
use crate::{exec::Exec, world::World};
use deno_core::{JsRuntime, OpState, RuntimeOptions, op2};
use deno_error::JsErrorBox;
@@ -85,7 +86,7 @@ deno_core::extension!(
environment::op_file,
environment::op_exec,
environment::op_exec_next,
environment::op_input,
input::op_input,
environment::op_output
]
);
@@ -97,7 +98,10 @@ pub async fn run_with_exec(world: World, exec: Arc<dyn Exec>) -> anyhow::Result<
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?
.block_on(run_module(world, exec, include_str!("../../js/bundle.js")))
.block_on(async {
let cred = Arc::new(crate::cred::store::Cred::open(&world.host_dir)?);
run_console(world, exec, cred).await
})
})
.await?
.map_err(|error: anyhow::Error| {
@@ -126,6 +130,19 @@ pub async fn run_module_with_credentials(
exec: Arc<dyn Exec>,
cred: Arc<crate::cred::store::Cred>,
source: &str,
) -> anyhow::Result<()> {
let (_, receiver) = input::channel();
run_module_with_input(world, exec, cred, receiver, source).await
}
/// Run without the console. Submissions and stop requests arrive through the channel.
/// The caller must use a current-thread Tokio runtime.
pub async fn run_module_with_input(
world: World,
exec: Arc<dyn Exec>,
cred: Arc<crate::cred::store::Cred>,
receiver: input::Receiver,
source: &str,
) -> anyhow::Result<()> {
let proxy = credentials::proxy(&cred)?;
let mut runtime = JsRuntime::new(RuntimeOptions {
@@ -183,6 +200,12 @@ pub async fn run_module_with_credentials(
.borrow_mut()
.put(credentials::Notifications::new(&cred));
runtime.op_state().borrow_mut().put(cred);
runtime
.op_state()
.borrow_mut()
.put(std::rc::Rc::new(input::Inbox(tokio::sync::Mutex::new(
receiver,
))));
runtime
.op_state()
.borrow_mut()
@@ -224,6 +247,25 @@ pub async fn run_module_with_credentials(
})
}
/// Run the headless harness with console input and signal clients.
/// The caller must use a current-thread Tokio runtime.
pub async fn run_console(
world: World,
exec: Arc<dyn Exec>,
cred: Arc<crate::cred::store::Cred>,
) -> anyhow::Result<()> {
let (sender, receiver) = input::channel();
input::console(sender.clone());
let signals = tokio::spawn(input::stop_signal(sender)?);
let source = concat!(
"globalThis.consoleEnabled = true;\n",
include_str!("../../js/bundle.js")
);
let result = run_module_with_input(world, exec, cred, receiver, source).await;
signals.abort();
result
}
/// Open a world and resume its durable root conversation.
pub async fn run(path: &Path) -> anyhow::Result<()> {
run_with_userland(path, None).await
host/tests/harness_foundation.rs
+172
View File
@@ -0,0 +1,172 @@
//! Patched pi-durable input placement, hold boundaries, and headless channel lifecycle.
use sausage::{
cred::Cred,
harness::{
input::{self, Input},
run_module, run_module_with_input,
},
world::World,
};
use std::{
process::{Command, Stdio},
sync::Arc,
time::{Duration, Instant},
};
const BUNDLE: &str = include_str!("../js/foundation-test-bundle.js");
#[test]
fn bundle_has_patched_hooks() {
for bundle in [BUNDLE, include_str!("../js/bundle.js")] {
assert!(bundle.contains("handlers.beforeInput"));
assert!(bundle.contains("input: true"));
assert!(bundle.contains("#held"));
}
assert!(!include_str!("../js/patches/pi-durable-1.5.0.patch").contains("dist/env/node"));
}
async fn probe(case: &str) -> anyhow::Result<()> {
let dir = tempfile::tempdir()?;
let world = World::open(dir.path())?;
let exec = sausage::process::exec(World::open(dir.path())?)?;
run_module(
world,
exec,
&format!("globalThis.foundationCase = {case:?};\n{BUNDLE}"),
)
.await
}
#[tokio::test(flavor = "current_thread")]
async fn input_placement() -> anyhow::Result<()> {
probe("placement").await
}
#[tokio::test(flavor = "current_thread")]
async fn followups_start_one_run() -> anyhow::Result<()> {
probe("followups").await
}
#[tokio::test(flavor = "current_thread")]
async fn response_thinking_level() -> anyhow::Result<()> {
probe("thinking").await
}
#[tokio::test(flavor = "current_thread")]
async fn hold_parks_at_phase_boundary() -> anyhow::Result<()> {
probe("hold").await
}
#[tokio::test(flavor = "current_thread")]
async fn headless_child() -> anyhow::Result<()> {
if std::env::var_os("HEADLESS_CHILD").is_none() {
return Ok(());
}
let dir = tempfile::tempdir()?;
let world = World::open(dir.path())?;
let cred = Arc::new(Cred::open(&world.host_dir)?);
let exec = sausage::process::exec(World::open(dir.path())?)?;
let (sender, receiver) = input::channel();
sender.send(Input::Submit {
conversation: 1,
content: serde_json::json!("channel input"),
request_id: "channel-request".into(),
})?;
sender.send(Input::Stop)?;
tokio::time::timeout(
Duration::from_secs(30),
run_module_with_input(
world,
exec,
cred,
receiver,
&format!("globalThis.foundationCase = 'headless';\n{BUNDLE}"),
),
)
.await??;
Ok(())
}
#[test]
fn channel_with_closed_stdin() -> anyhow::Result<()> {
let mut child = Command::new(std::env::current_exe()?)
.args(["--exact", "headless_child", "--nocapture"])
.env("HEADLESS_CHILD", "1")
.stdin(Stdio::null())
.spawn()?;
let deadline = Instant::now() + Duration::from_secs(40);
loop {
if let Some(status) = child.try_wait()? {
assert!(status.success());
break;
}
if Instant::now() > deadline {
child.kill()?;
child.wait()?;
anyhow::bail!("Headless channel did not stop");
}
std::thread::yield_now();
}
Ok(())
}
#[test]
fn cli_stays_up_after_stdin_eof() -> anyhow::Result<()> {
let dir = tempfile::tempdir()?;
let mut child = Command::new(env!("CARGO_BIN_EXE_sausage"))
.arg("run")
.arg(dir.path())
.args(["--api-listen", "127.0.0.1:0"])
.env("PI_PROXY_BASE_URL", "https://proxy.invalid")
.env("PI_PROXY_API_KEY", "faux-token")
.env_remove("SAUSAGE_USERLAND")
.stdin(Stdio::null())
.spawn()?;
let deadline = Instant::now() + Duration::from_secs(30);
loop {
if let Some(status) = child.try_wait()? {
anyhow::bail!("CLI exited at EOF: {status}");
}
if let Ok(db) = rusqlite::Connection::open(dir.path().join("host/harness.sqlite"))
&& db
.query_row("SELECT count(*) FROM conversations", [], |r| {
r.get::<_, i64>(0)
})
.unwrap_or(0)
> 0
{
break;
}
if Instant::now() > deadline {
child.kill()?;
child.wait()?;
anyhow::bail!("CLI did not initialize");
}
std::thread::yield_now();
}
// EOF shutdown previously took one credential tick. Check beyond that tick.
let alive_until = Instant::now() + Duration::from_secs(2);
while Instant::now() < alive_until {
if let Some(status) = child.try_wait()? {
anyhow::bail!("CLI exited after EOF: {status}");
}
std::thread::yield_now();
}
assert!(
Command::new("kill")
.args(["-TERM", &child.id().to_string()])
.status()?
.success()
);
let deadline = Instant::now() + Duration::from_secs(10);
loop {
if let Some(status) = child.try_wait()? {
assert!(status.success());
break;
}
if Instant::now() > deadline {
child.kill()?;
child.wait()?;
anyhow::bail!("CLI did not stop cleanly");
}
std::thread::yield_now();
}
Ok(())
}