diff --git a/packages/coding-agent/src/experimental/client-tui-chat.ts b/packages/coding-agent/src/experimental/client-tui-chat.ts index 73200afa5..90abef609 100644 --- a/packages/coding-agent/src/experimental/client-tui-chat.ts +++ b/packages/coding-agent/src/experimental/client-tui-chat.ts @@ -183,6 +183,10 @@ export class ExperimentalChatView { if (message?.role === "user") this.#addText(userText(message.content)); } else if (entry.kind === "pi.reset") { this.#addText(theme.fg("muted", "[new context]")); + } else if (!entry.kind.startsWith("pi.") && message?.role === "user" && entry.model?.length === 1) { + // An input placed as an application entry. + this.transcript.addChild(new Spacer(1)); + this.transcript.addChild(new UserMessageComponent(userText(message.content))); } } diff --git a/packages/coding-agent/src/experimental/durable/runtime.ts b/packages/coding-agent/src/experimental/durable/runtime.ts index fcd6e5239..083f3f115 100644 --- a/packages/coding-agent/src/experimental/durable/runtime.ts +++ b/packages/coding-agent/src/experimental/durable/runtime.ts @@ -104,13 +104,19 @@ async function firstInput(harness: Harness, id: ConversationId): Promise<{ title let cursor: Cursor | undefined; do { const page = await conversation.entries({}, 256, cursor, context); - first = page.items.findLast((entry) => entry.kind === "pi.user") ?? first; + first = page.items.findLast(isInput) ?? first; cursor = page.next; } while (cursor !== undefined); return titleOf(first); } /** The text of a user entry, as a one-line title. */ +/** A placed input: `pi.user`, or an application entry whose model is one user message. */ +function isInput(entry: EntryRecord): boolean { + if (entry.kind === "pi.user") return true; + return !entry.kind.startsWith("pi.") && entry.model?.length === 1 && entry.model[0]!.role === "user"; +} + function titleOf(entry: EntryRecord | undefined): { title?: string } { const message = entry?.model?.[0]; if (message?.role !== "user") return {}; @@ -206,7 +212,7 @@ export async function openDurable(options: OpenDurableOptions = {}): Promise summary.id === id && summary.title === undefined ? { ...summary, ...titleOf(change.value) } : summary, diff --git a/packages/coding-agent/src/experimental/durable/tui.ts b/packages/coding-agent/src/experimental/durable/tui.ts index 397b276cb..6eef2e775 100644 --- a/packages/coding-agent/src/experimental/durable/tui.ts +++ b/packages/coding-agent/src/experimental/durable/tui.ts @@ -462,6 +462,11 @@ class DurableTui { this.#chat.addChild(new Spacer(1)); this.#chat.addChild(summary); } else if (entry.kind === "pi.reset") this.#addText("[new context]"); + else if (!entry.kind.startsWith("pi.") && message?.role === "user" && entry.model?.length === 1) { + // An input placed as an application entry. + this.#chat.addChild(new Spacer(1)); + this.#chat.addChild(new UserMessageComponent(userText(message.content))); + } } #syncStreaming(message: AssistantMessage): void { diff --git a/packages/coding-agent/src/experimental/vacation/runtime.ts b/packages/coding-agent/src/experimental/vacation/runtime.ts index 0c9b8d268..ee7a0c7a6 100644 --- a/packages/coding-agent/src/experimental/vacation/runtime.ts +++ b/packages/coding-agent/src/experimental/vacation/runtime.ts @@ -103,13 +103,19 @@ async function firstInput(harness: Harness, id: ConversationId): Promise<{ title let cursor: Cursor | undefined; do { const page = await conversation.entries({}, 256, cursor, context); - first = page.items.findLast((entry) => entry.kind === "pi.user") ?? first; + first = page.items.findLast(isInput) ?? first; cursor = page.next; } while (cursor !== undefined); return titleOf(first); } /** The text of a user entry, as a one-line title. */ +/** A placed input: `pi.user`, or an application entry whose model is one user message. */ +function isInput(entry: EntryRecord): boolean { + if (entry.kind === "pi.user") return true; + return !entry.kind.startsWith("pi.") && entry.model?.length === 1 && entry.model[0]!.role === "user"; +} + function titleOf(entry: EntryRecord | undefined): { title?: string } { const message = entry?.model?.[0]; if (message?.role !== "user") return {}; @@ -203,7 +209,7 @@ export async function openDurable(options: OpenDurableOptions = {}): Promise summary.id === id && summary.title === undefined ? { ...summary, ...titleOf(change.value) } : summary, diff --git a/packages/coding-agent/src/experimental/vacation/tui.ts b/packages/coding-agent/src/experimental/vacation/tui.ts index 397b276cb..6eef2e775 100644 --- a/packages/coding-agent/src/experimental/vacation/tui.ts +++ b/packages/coding-agent/src/experimental/vacation/tui.ts @@ -462,6 +462,11 @@ class DurableTui { this.#chat.addChild(new Spacer(1)); this.#chat.addChild(summary); } else if (entry.kind === "pi.reset") this.#addText("[new context]"); + else if (!entry.kind.startsWith("pi.") && message?.role === "user" && entry.model?.length === 1) { + // An input placed as an application entry. + this.#chat.addChild(new Spacer(1)); + this.#chat.addChild(new UserMessageComponent(userText(message.content))); + } } #syncStreaming(message: AssistantMessage): void { diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 6cb7b6b8d..f9bb6f61b 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -9,9 +9,11 @@ - `bash` and `powershell` answer a nonzero exit with an error result carrying an `exit_code` diagnostic, instead of throwing (`tool_error`). - Any error a `Storage` method throws, a read included, now fails the Session, which closes itself: every later call and every pending wait rejects with `SessionFailed`, whose `cause` is that error, and nothing is retried. `StorageRejected` is removed; a storage retries its own transient failures and throws the new `StorageRequestError` for an invalid request, which fails only that call. A throw in the scheduler's own commits, or in the commit listeners of the Harness's components, fails the Session too. - `subscribeCommits()` and `subscribeClose()` listeners that throw, or return a rejected promise, are reported through `onReport` instead of failing the operation that ran them. +- A display-only write submission, one of only `kind` and `data` (no `model`, `head`, or `edits`), is appended and settled `done` at once, even while the conversation is busy or has queued items, instead of waiting in `pi.inbox` for the next boundary; only a queued reset still holds it back, so the reset does not hide it. It changes nothing the model sees, so a chat message shows as soon as it is sent, ahead of earlier submissions still queued, and it no longer places items left queued after a failed run ([#10407](https://github.com/earendil-works/pi/issues/10407)). ### Added +- Input submissions take `entry: { kind, data }` (`InputEntry`): the input is placed as an entry of that kind carrying that data, with its content as the user message, instead of `pi.user`, and behaves as any other input; the kind must not start with `pi.`. Queued inputs carry it in their `pi.inbox` item ([#10407](https://github.com/earendil-works/pi/issues/10407)). - `Session.closed` settles once the Session has closed, Storage included: `{ reason: "closed" }` after `close()`, or `{ reason: "failed", error }` after a storage error. `WatchEnd` gains `{ reason: "session_failed", error }`, and the README has an Errors section. - `ToolExecutionApi.executeTool(name, args, context, { key?, progress? })`: a tool calls another tool as a nested call. The nested call is its own `pi.tool` task owned by the caller, runs validation, the `ToolTask` hooks (whose `ToolHookCall` carries `parent`), output limits, and its replay policy, and returns a `NestedToolExecutionResult` (its `taskId`, `structuredOutput`, `details`, `diagnostics`, `usage`, `isError`, `durationMs`, but not the model's `output`) to the caller instead of the transcript. Each nested call is a member of `NestedCallDoc`, a task-scoped document family of the caller keyed by the call's key, holding its `taskId` and, once it ends, its `result` as returned, so it is gone from storage once the caller settles; its task's receipt stays small. Keys, by default the call's position, let a replay-safe caller's rerun reattach to its nested calls; a caller that is not replay-safe has its unfinished nested calls abandoned after a restart. Nested calls show in `pi.live.nestedTools`, with the arguments they run with and a `summary` (error, duration, usage) once done (`progress: false` keeps their running output out), and as tool events with `parentToolCallId` and `parentTaskId`; every tool event also carries its call's `taskId`. A caller aborts the nested calls it left running before it settles. - Structured tool output: a tool declares `structuredOutputSchema` and returns `structuredOutput`, which the Harness validates and gives to programs that call the tool; a tool without a schema gives them its bounded output, one text item as a string, one image as itself, and otherwise the content list, for errors too. A nested call with invalid structured output gets an error result; a model-issued one only loses it, and the host gets a report. `api.retainedOutput()` reads the bounded output the model will see. `bash` and `powershell` return `{ output, truncated, fullOutputPath?, exitCode }`. diff --git a/packages/durable/README.md b/packages/durable/README.md index d7d813990..760dcb853 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -394,11 +394,27 @@ await root.submit({ type: "write", entry: { kind: "app.note", data: "user opened - **Steers** are placed after the current tool round and join the running work. - **Follow-ups** are placed when the run answers, and start the next run. -- **Writes** append an entry without asking the model anything. +- **Writes** append an entry without asking the model anything. A write of only `kind` and `data` changes nothing the model sees, so it is appended at once, even mid-run, and is never queued: a chat message shows in the transcript as soon as it is sent. It can therefore appear before items submitted earlier that are still queued, except behind a queued reset, where it waits so the reset does not hide it. A write with `model`, `head`, or `edits` waits for the next boundary like a steer. - `await submission.abort(context)` withdraws a queued submission. - The [settings](#settings) `steeringMode: "all"` and `followUpMode: "all"` place every queued item at once instead of one per turn. -If a run fails, queued items stay in the inbox until the next submission places them, oldest first. +If a run fails, queued items stay in the inbox until the next input or queued write places them, oldest first. + +An input is placed as a `pi.user` entry. To show it as something else, give it an `entry` with your kind and data; the model still sees `content` as the user message: + +```typescript +await root.submit( + { + type: "input", + content: "ann: can you check the build?", + entry: { kind: "chat.mention", data: { author: "ann", text: "can you check the build?" } }, + }, + context, +); +// placed as { kind: "chat.mention", data: { author: "ann", ... }, model: [{ role: "user", content: "ann: can you check the build?" }] } +``` + +It behaves like any other input: it starts a run, steers, follows up, or rejects, and is answered. While queued, its inbox item carries the entry. The kind must not start with `pi.`. ## Reset and Handoff diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 32e11cfe1..c2a1d3b64 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -2251,7 +2251,8 @@ wait in the built-in inbox document, an ordered list of tagged items: ```ts type InboxItem = - | { readonly id: SubmissionId; readonly mode: "steer" | "followUp"; readonly content: UserInput } + /** `entry` is the input's `InputEntry` (`kind` and optional `data`), present when given. */ + | { readonly id: SubmissionId; readonly mode: "steer" | "followUp"; readonly content: UserInput; readonly entry?: InputEntry } /** `entry` is the write's `EntryDraft`, stored as plain JSON. */ | { readonly id: SubmissionId; readonly mode: "write"; readonly entry: JsonObject }; @@ -2268,8 +2269,24 @@ type InboxState = { items: InboxItem[] }; | view mount | `docs["pi.inbox"]` | | created | with every Harness conversation (section 2.2) | -Items are in ID order. A queued input stores its content; its `pi.user` entry -gets the Harness clock's timestamp at placement. Queued submissions belong to +Items are in ID order. A queued input stores its content and its entry's kind +and data; its entry gets the Harness clock's timestamp at placement. + +An input is placed as one entry whose `model` is the user message of its +`content`: a `pi.user` entry, or, with `entry: { kind, data }`, an entry of that +kind carrying that data. Only `kind` and `data` are taken from `entry`; admission +rejects a kind that is not a non-empty string or starts with `pi.`. Nothing in +the Harness reads input entries by kind; context, compaction, events, and run +accounting go by `model` and the submission records. + +A write of only `kind` and `data`, with no `model`, `head`, or `edits` (present +counts, even if empty), is display-only: it changes no model context, so admission +appends it and settles it `done` in the same commit, busy or idle, ahead of +earlier submissions still queued. Between a tool call and its result it changes +nothing, because results are ordered by message. The one exception is a queued +reset: appended before it, the entry would fall out of the context the reset +starts, so a display-only write queues behind a queued reset and is placed after +it at the same boundary. Queued submissions belong to their conversation, so a fork starts with an empty inbox. Run control lives in the built-in live document `pi.live` (section 8). Its @@ -2288,8 +2305,9 @@ Admission and terminal transitions: | action | submission state | other writes | |---|---|---| | input, idle with empty inbox | `placed`, with user entry | create run and generation | +| display-only write, no queued reset | `done`, with entry | append entry; no run | | write, idle with empty inbox | `done`, with entry | append entry; no run | -| input or write, busy or non-empty inbox | `queued` | append inbox item | +| input or other write, busy or non-empty inbox | `queued` | append inbox item | | boundary places user item | `placed`, with entry | add ID to current or successor run | | boundary places write | `done`, with entry | append entry | | run answers | input `done`, with required answer entry | remove `run`, or hand it to a successor | @@ -2303,8 +2321,9 @@ writes no record and reports `ConversationBusy`. An idle input queues with mode `steer` when `whenBusy` is `steer` and `followUp` otherwise. An idle conversation with a non-empty inbox, for example after a failed run, -queues every new submission behind the waiting items and runs a final boundary -in the same commit. Order is preserved: a follow-up queued before the failure is +queues every new input and every write that is not display-only behind the +waiting items and runs a final boundary in the same commit. A display-only write +is appended without one, so a chat message never restarts failed work. Order is preserved: a follow-up queued before the failure is selected before the new input. A `Submission` waits until `done` or `unanswered`; abort withdraws only a still- @@ -2366,7 +2385,7 @@ Only successful run ends apply the final boundary: an answer, `terminate`, or `handoff`. Failure, a run task's abort handler, fault, and orphan settle the run's inputs `unanswered` and leave the inbox alone (`Conversation.abort()` separately withdraws queued user items); the queued items stay visible in -`pi.inbox` until the next submission's boundary or their withdrawal. +`pi.inbox` until the next input or queued write runs a boundary, or their withdrawal. Selected and stale items are removed positionally while retained item order is preserved. Chord's Astra operation generator must express scattered removals @@ -3404,9 +3423,12 @@ Generation preparation takes these steps against its phase snapshot: append one positional `pi.system` entry when they differ. Preparation does not recheck the transcript before appending: only the Harness -writes to a busy conversation, through submissions, run tasks, boundaries, and a -blocking compaction, which appends only while its generation waits for it -(section 8.7), so the transcript it read is still current (section 12). +writes model context to a busy conversation, through submissions, run tasks, +boundaries, and a blocking compaction, which appends only while its generation +waits for it (section 8.7), so the context it read is still current (section 12). +Display-only writes (section 6) may land meanwhile; they contribute no messages, +and the system entry planning, the compaction threshold, and the request cutoff +depend only on contributions and the newest entry ID. Model and thinking level are request options, not prompt state. @@ -4979,7 +5001,8 @@ These are contracts, not invitations to add defensive machinery: use a write submission. - **Queued items after a failed run:** failure and task abort leave the inbox alone. Queued follow-ups wait in `pi.inbox`, and their `wait()` does not settle, until - the next submission's boundary places them or the host withdraws them. A + the next input or queued write's boundary places them or the host withdraws + them; a display-only write runs no boundary. A compaction summary is such a submission: when a manual or background compaction finishes in that idle conversation, its final boundary places the waiting follow-ups and starts a run. diff --git a/packages/durable/src/harness/generation.ts b/packages/durable/src/harness/generation.ts index 85f5b1d0d..02e6f5dd3 100644 --- a/packages/durable/src/harness/generation.ts +++ b/packages/durable/src/harness/generation.ts @@ -119,8 +119,9 @@ export const GenerationTask = defineTask { const { conversationId } = runtime; diff --git a/packages/durable/src/harness/inbox.ts b/packages/durable/src/harness/inbox.ts index 7ba380291..95f2e8e57 100644 --- a/packages/durable/src/harness/inbox.ts +++ b/packages/durable/src/harness/inbox.ts @@ -1,12 +1,17 @@ -import type { Draft, JsonRepresentation } from "@earendil-works/chord"; +import { copyJson, type Draft, type JsonRepresentation } from "@earendil-works/chord"; import { defineDoc } from "../documents.ts"; import { UserEntry } from "../entries.ts"; -import type { ConversationId, EntryDraft, EntryId, JsonObject, SubmissionId, Tx } from "../types.ts"; -import type { QueueMode, Settings, UserInput } from "./types.ts"; +import type { ConversationId, EntryDraft, EntryId, EntryRecord, JsonObject, SubmissionId, Tx } from "../types.ts"; +import type { InputEntry, QueueMode, Settings, UserInput } from "./types.ts"; -/** A queued submission: user input for a run, or a passive entry write. */ +/** A queued submission: user input for a run, with the entry it is placed as when not `pi.user`, or an entry write. */ export type InboxItem = - | { id: SubmissionId; mode: "steer" | "followUp"; content: JsonRepresentation } + | { + id: SubmissionId; + mode: "steer" | "followUp"; + content: JsonRepresentation; + entry?: JsonRepresentation; + } /** `entry` is an `EntryDraft`, stored as plain JSON. */ | { id: SubmissionId; mode: "write"; entry: JsonObject }; @@ -70,7 +75,7 @@ export async function applyBoundary( ): Promise { const { conversationId, inbox } = boundary; const items = inbox.items; - const reset = items.some((item) => item.mode === "write" && item.entry.head === "self"); + const reset = hasQueuedReset(items); const final = at === "final" || reset; const pick = (mode: "steer" | "followUp", queueMode: QueueMode): number[] => { const indexes = items.flatMap((item, index) => (item.mode === mode ? [index] : [])); @@ -95,8 +100,7 @@ export async function applyBoundary( const placed: SubmissionId[] = []; for (const index of users) { const item = items[index] as Draft>; - const message = { role: "user", content: item.content as UserInput, timestamp: now } as const; - const entry = await tx.appendEntry(UserEntry, conversationId, { model: [message] }); + const entry = await appendInput(tx, conversationId, item.content as UserInput, item.entry as InputEntry, now); tx.placeSubmission(item.id, entry.id); placed.push(item.id); } @@ -105,6 +109,38 @@ export async function applyBoundary( return { users: placed, reset }; } +/** 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: Tx, + conversationId: ConversationId, + content: UserInput, + entry: InputEntry | undefined, + now: number, +): Promise { + const message = { role: "user", content, timestamp: now } as const; + if (entry === undefined) return tx.appendEntry(UserEntry, conversationId, { model: [message] }); + return tx.appendEntry(conversationId, { ...pickInputEntry(entry), 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 }: InputEntry): JsonRepresentation { + const picked = data === undefined ? { kind } : { kind, data }; + return copyJson(picked, { omitUndefinedProperties: true }) as JsonRepresentation; +} + +/** + * Whether a write is display-only: it has no `model`, `head`, or `edits` (present counts, even if empty), so it changes + * nothing the model sees and needs no boundary. + */ +export function isDisplayOnly(entry: EntryDraft): boolean { + return entry.model === undefined && entry.head === undefined && entry.edits === undefined; +} + +/** Whether a reset (`head: "self"` write) is queued; entries appended before it leave the active range when it lands. */ +export function hasQueuedReset(items: readonly InboxItem[]): boolean { + return items.some((item) => item.mode === "write" && item.entry.head === "self"); +} + /** Whether a head write targets an entry before the active range, so placing it would bring back cut history. */ export function isStale(boundary: Boundary, entry: EntryDraft): boolean { return typeof entry.head === "number" && boundary.head !== undefined && entry.head < boundary.head; diff --git a/packages/durable/src/harness/submissions.ts b/packages/durable/src/harness/submissions.ts index 02edc8a33..d84226810 100644 --- a/packages/durable/src/harness/submissions.ts +++ b/packages/durable/src/harness/submissions.ts @@ -1,5 +1,4 @@ import { type Context, copyJson, type JsonRepresentation } from "@earendil-works/chord"; -import { UserEntry } from "../entries.ts"; import { ConversationBusy } from "../errors.ts"; import type { SessionImpl } from "../session/session.ts"; import type { @@ -12,9 +11,20 @@ import type { Tx, } from "../types.ts"; import { startRun } from "./generation.ts"; -import { applyBoundary, InboxDoc, isStale, prepareBoundary, type QueueModes, removeInboxItem } from "./inbox.ts"; +import { + appendInput, + applyBoundary, + hasQueuedReset, + InboxDoc, + isDisplayOnly, + isStale, + pickInputEntry, + prepareBoundary, + type QueueModes, + removeInboxItem, +} from "./inbox.ts"; import { LiveDoc } from "./live.ts"; -import type { SettledSubmissionRecord, Submission, SubmissionDraft, UserInput } from "./types.ts"; +import type { InputEntry, SettledSubmissionRecord, Submission, SubmissionDraft, UserInput } from "./types.ts"; import { closedError, Waiters } from "./util.ts"; type AbortResult = "aborted" | "already_placed" | "settled"; @@ -138,12 +148,25 @@ function isSettled(record: SubmissionRecord): record is SettledSubmissionRecord return record.status === "done" || record.status === "unanswered"; } +/** Reject an input entry an untyped caller got wrong: its kind must be a non-empty string outside the `pi.` prefix. */ +function checkInputEntry(entry: InputEntry): void { + const { kind } = entry; + if (typeof kind !== "string" || kind.length === 0 || kind.startsWith("pi.")) { + throw new TypeError( + `An input entry's kind must be a non-empty string not starting with "pi.", got ${String(kind)}`, + ); + } +} + /** * Admit a submission inside a commit (spec ยง6); `Conversation.submit()` and conversation-owned compactions share it. A - * known request ID returns its existing submission without writing. A busy conversation queues it in `pi.inbox`, or - * rejects `whenBusy: "reject"` input with `ConversationBusy`. An idle conversation with queued items queues it behind - * them and runs a final boundary. Otherwise idle input places a user entry and starts a run, and an idle write appends - * its entry and settles `done`, or `stale` when its head reaches before the active range. + * known request ID returns its existing submission without writing. A display-only write (only `kind` and `data`) is + * appended and settled `done` at once, busy or not, as it changes nothing the model sees; only behind a queued reset it + * queues, so it lands after the reset instead of before it in the context the reset ends. A busy conversation queues anything else in + * `pi.inbox`, or rejects `whenBusy: "reject"` input with `ConversationBusy`. An idle conversation with queued items + * queues it behind them and runs a final boundary. Otherwise idle input places its entry (`pi.user`, or the input's + * own) and starts a run, and an idle write appends its entry and settles `done`, or `stale` when its head reaches + * before the active range. */ export async function admitSubmission( tx: Tx, @@ -152,6 +175,7 @@ export async function admitSubmission( now: number, queueModes: QueueModes, ): Promise { + if (draft.type === "input" && draft.entry !== undefined) checkInputEntry(draft.entry); if (draft.requestId !== undefined) { const existing = await tx.submissionByRequest(conversationId, draft.requestId); if (existing !== undefined) { @@ -161,10 +185,19 @@ export async function admitSubmission( return existing.id; } } + const requestId = draft.requestId === undefined ? {} : { requestId: draft.requestId }; + if ( + draft.type === "write" && + isDisplayOnly(draft.entry) && + !hasQueuedReset((await tx.doc(InboxDoc, conversationId)).items) + ) { + const entry = await tx.appendEntry(conversationId, draft.entry); + const write = { conversationId, ...requestId, type: "write", status: "done", entry: entry.id } as const; + return (await tx.createSubmission(write)).id; + } const live = await tx.doc(LiveDoc, conversationId); const busy = live.run !== undefined; if (busy && draft.type === "input" && draft.whenBusy === "reject") throw new ConversationBusy(conversationId); - const requestId = draft.requestId === undefined ? {} : { requestId: draft.requestId }; // 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); if (boundary === undefined || boundary.inbox.items.length > 0) { @@ -182,7 +215,9 @@ export async function admitSubmission( if (draft.type === "write") items.push({ id, mode: "write", entry: value as JsonObject }); else { const mode = draft.whenBusy === "steer" ? "steer" : "followUp"; - items.push({ id, mode, content: value as JsonRepresentation }); + const content = value as JsonRepresentation; + const entry = draft.entry === undefined ? {} : { entry: pickInputEntry(draft.entry) }; + items.push({ id, mode, content, ...entry }); } if (boundary === undefined) return id; const { users } = await applyBoundary(tx, boundary, "final", now); @@ -198,8 +233,7 @@ export async function admitSubmission( const write = { conversationId, ...requestId, type: "write", status: "done", entry: entry.id } as const; return (await tx.createSubmission(write)).id; } - const message = { role: "user", content: draft.content, timestamp: now } as const; - const entry = await tx.appendEntry(UserEntry, conversationId, { model: [message] }); + const entry = await appendInput(tx, conversationId, draft.content, draft.entry, now); const input = { conversationId, ...requestId, type: "input", status: "placed", entry: entry.id } as const; const { id } = await tx.createSubmission(input); await startRun(tx, conversationId, live, [id]); diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index a81e27e63..60bb8966c 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -51,6 +51,12 @@ export type ModelRef = { export type UserInput = UserMessage["content"]; +/** + * The entry an input is placed as, instead of `pi.user`: an application kind, and data the transcript shows, such as its + * author. Its model message is still the input's user message. + */ +export type InputEntry = { readonly kind: string; readonly data?: JsonValue }; + /** Host submission: user input that may start a run, or a passive entry write. */ export type SubmissionDraft = { readonly requestId?: string; @@ -59,7 +65,8 @@ export type SubmissionDraft = { readonly type: "input"; readonly content: UserInput; readonly whenBusy?: "steer" | "followUp" | "reject"; - readonly entry?: never; + /** Place the input as this entry; default `pi.user` without data. The kind must not start with `pi.`. */ + readonly entry?: InputEntry; } | { readonly type: "write"; @@ -622,8 +629,10 @@ export interface Conversation { configure(change: AgentChange, context: Context): Promise; /** - * Durably admit user input or a passive entry write. A busy conversation, or one with queued items, queues it in - * `pi.inbox`; `whenBusy: "reject"` rejects with `ConversationBusy` instead and writes nothing. + * Durably admit user input or a passive entry write. A busy conversation, or one with queued items, queues input and + * writes that affect model context in `pi.inbox`; `whenBusy: "reject"` rejects input with `ConversationBusy` instead + * and writes nothing. A write of only `kind` and `data` (no `model`, `head`, or `edits`) is appended at once, + * even mid-run, as it changes nothing the model sees. */ submit(submission: SubmissionDraft, context: Context): Promise; /** diff --git a/packages/durable/src/index.ts b/packages/durable/src/index.ts index aab6967a1..8f61a6df4 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -86,6 +86,7 @@ export type { HookRegistration, HookResult, HooksOf, + InputEntry, InputSubmissionDraft, ModelRef, NestedToolExecutionResult, diff --git a/packages/durable/test/harness-compaction.test.ts b/packages/durable/test/harness-compaction.test.ts index 28da7d753..182ac0ad8 100644 --- a/packages/durable/test/harness-compaction.test.ts +++ b/packages/durable/test/harness-compaction.test.ts @@ -261,6 +261,18 @@ describe("range selection", () => { expect(cut).toBe(9); }); + it("never cuts at a display-only entry where the budget is reached (#10407)", () => { + const entries = [ + entry("pi.user", [user(text("u1", 100))]), + entry("pi.assistant", [assistant(text("a1", 100))]), + entry("chat.message", []), + entry("pi.assistant", [assistant(text("a2", 100))]), + ]; + // The budget is reached at a1; the display-only entry after it is skipped for the assistant a2. + expect(selectCut(view(entries), 150)).toBe(1); + expect(selectCut(view(entries), 50)).toBe(3); + }); + it("keeps a huge last tool result together with its assistant", () => { const entries = [ entry("pi.user", [user("u")]), diff --git a/packages/durable/test/harness-inbox.test.ts b/packages/durable/test/harness-inbox.test.ts index 700b7c164..91717e516 100644 --- a/packages/durable/test/harness-inbox.test.ts +++ b/packages/durable/test/harness-inbox.test.ts @@ -77,6 +77,11 @@ function holdTool(setup: ChatSetup, gate: Deferred, result: ToolExecutionResult ); } +/** A write that affects model context, so it queues while busy; a data-only write would be appended at once. */ +function queuedNote(text: string) { + return { kind: "note", model: [{ role: "user", content: text, timestamp: 0 }] } as const; +} + const HOLD = fauxAssistantMessage([fauxToolCall("hold", {}, { id: "c1" })], { stopReason: "toolUse" }); async function status(submission: Submission): Promise { @@ -111,7 +116,7 @@ describe("inbox", () => { const input = await root.submit({ type: "input", content: "a" }, context); await first.reached; const f1 = await root.submit({ type: "input", content: "f1" }, context); - const write = await root.submit({ type: "write", entry: { kind: "note", data: "w" } }, context); + const write = await root.submit({ type: "write", entry: queuedNote("w") }, context); const f2 = await root.submit({ type: "input", content: "f2", whenBusy: "followUp" }, context); for (const submission of [f1, write, f2]) expect((await status(submission)).status).toBe("queued"); expect(await inbox(harness, root)).toEqual([ @@ -126,7 +131,7 @@ describe("inbox", () => { expect(transcript(await allEntries(root))).toEqual([ "pi.user:a", "pi.assistant:first", - "note", + "note:w", "pi.user:f1", "pi.assistant:second", "pi.user:f2", @@ -389,7 +394,7 @@ describe("inbox", () => { const { harness, root } = await openChat(new MemoryStorage(), setup); const input = await root.submit({ type: "input", content: "a" }, context); await first.reached; - const kept = await root.submit({ type: "write", entry: { kind: "note" } }, context); + const kept = await root.submit({ type: "write", entry: queuedNote("kept") }, context); const withdrawn = await root.submit({ type: "input", content: "f" }, context); expect(await withdrawn.abort(context)).toBe("aborted"); expect(await withdrawn.wait(context)).toMatchObject({ status: "unanswered", reason: "aborted" }); @@ -461,17 +466,17 @@ describe("inbox", () => { }); const input = await root.submit({ type: "input", content: "a" }, context); await first.reached; - const w1 = await root.submit({ type: "write", entry: { kind: "note" } }, context); + const w1 = await root.submit({ type: "write", entry: queuedNote("w") }, context); const f1 = await root.submit({ type: "input", content: "f1" }, context); - const w2 = await root.submit({ type: "write", entry: { kind: "note" } }, context); + const w2 = await root.submit({ type: "write", entry: queuedNote("w") }, context); const f2 = await root.submit({ type: "input", content: "f2" }, context); - const w3 = await root.submit({ type: "write", entry: { kind: "note" } }, context); + const w3 = await root.submit({ type: "write", entry: queuedNote("w") }, context); expect(ops).toEqual([ - [["p", ["items"], 0, 0, [{ id: w1.id, mode: "write", entry: { kind: "note" } }]]], + [["p", ["items"], 0, 0, [{ id: w1.id, mode: "write", entry: queuedNote("w") }]]], [["p", ["items"], 1, 0, [{ id: f1.id, mode: "followUp", content: "f1" }]]], - [["p", ["items"], 2, 0, [{ id: w2.id, mode: "write", entry: { kind: "note" } }]]], + [["p", ["items"], 2, 0, [{ id: w2.id, mode: "write", entry: queuedNote("w") }]]], [["p", ["items"], 3, 0, [{ id: f2.id, mode: "followUp", content: "f2" }]]], - [["p", ["items"], 4, 0, [{ id: w3.id, mode: "write", entry: { kind: "note" } }]]], + [["p", ["items"], 4, 0, [{ id: w3.id, mode: "write", entry: queuedNote("w") }]]], ]); first.release(); await input.wait(context); @@ -517,14 +522,14 @@ describe("inbox", () => { const { harness, root } = await openChat(new MemoryStorage(), setup); const input = await root.submit({ type: "input", content: "a" }, context); await first.reached; - const write = await root.submit({ type: "write", entry: { kind: "note" } }, context); + const write = await root.submit({ type: "write", entry: queuedNote("w") }, context); first.release(); await input.wait(context); const entries = await allEntries(root); expect(transcript(entries)).toEqual([ "pi.user:a", "pi.assistant:first", - "note", + "note:w", "pi.user:more", "pi.assistant:second", ]); diff --git a/packages/durable/test/harness-input-entries.test.ts b/packages/durable/test/harness-input-entries.test.ts new file mode 100644 index 000000000..019172d03 --- /dev/null +++ b/packages/durable/test/harness-input-entries.test.ts @@ -0,0 +1,456 @@ +// Inputs placed as application entries, and data-only writes appended at once (#10407). +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + type AssistantMessage, + type FauxResponseStep, + fauxAssistantMessage, + fauxText, + fauxToolCall, + type Context as ModelContext, + Type, +} from "@earendil-works/pi-ai"; +import { + type AgentEvent, + type Conversation, + ConversationBusy, + defineTool, + type EntryRecord, + type Harness, + InboxDoc, + LiveDoc, + MemoryStorage, + type SubmissionDraft, + watchEvents, +} from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; +import { allEntries, type ChatSetup, chatSetup, openChat, textOf, waitFor } from "./chat-support.ts"; +import { addSection, addTool } from "./harness-support.ts"; +import { context } from "./session-support.ts"; +import { aborted, type Deferred, deferred } from "./task-support.ts"; + +function answer(text: string): AssistantMessage { + return fauxAssistantMessage([fauxText(text)]); +} + +/** A faux response held until `release` or cancellation; `reached` resolves when the request is sent. */ +function gated(message: AssistantMessage): { step: FauxResponseStep; reached: Promise; release: () => void } { + const reached = deferred(); + const gate = deferred(); + const step: FauxResponseStep = async (_request, options) => { + reached.resolve(); + await Promise.race([gate.promise, aborted(options!.signal!)]); + return message; + }; + return { step, reached: reached.promise, release: () => gate.resolve() }; +} + +/** Register a `hold` tool whose calls wait for `gate`. */ +function holdTool(setup: ChatSetup, gate: Deferred): void { + addTool( + setup.registry, + defineTool({ + name: "hold", + description: "Waits for the test", + parameters: Type.Object({}), + execute: async () => { + await gate.promise; + return { output: [{ type: "text", text: "held" }] }; + }, + }), + ); +} + +const HOLD = fauxAssistantMessage([fauxToolCall("hold", {}, { id: "c1" })], { stopReason: "toolUse" }); + +/** Role and text of each request message, as the provider receives them. */ +function requestLog(requests: string[][], reply: AssistantMessage): FauxResponseStep { + return (request: ModelContext) => { + requests.push(request.messages.map((message) => `${message.role}:${textOf(message) ?? ""}`)); + return reply; + }; +} + +/** Kind, application data, and model text of each entry, skipping system entries. */ +function transcript(entries: readonly EntryRecord[]): unknown[][] { + return entries + .filter((entry) => entry.kind !== "pi.system") + .map((entry) => { + const message = entry.model?.[0]; + const text = message === undefined || message.role === "toolResult" ? undefined : textOf(message); + const data = entry.data === undefined || entry.kind.startsWith("pi.") ? [] : [entry.data]; + return [entry.kind, ...data, ...(text === undefined ? [] : [text])]; + }); +} + +async function inboxItems(harness: Harness, root: Conversation) { + return (await harness.snapshot(InboxDoc, root.id, context))!.items; +} + +async function toolRunning(harness: Harness, root: Conversation): Promise { + await waitFor(async () => (await harness.snapshot(LiveDoc, root.id, context))?.tools?.[0]?.status === "running"); +} + +const mention = (text: string) => ({ kind: "chat.mention", data: { author: "ann", text } }); + +describe("input entries", () => { + it("places an idle input as its entry: the given kind and data, with the content as its user message", async () => { + const setup = chatSetup(); + setup.now = () => 42; + const requests: string[][] = []; + setup.faux.setResponses([requestLog(requests, answer("hi ann"))]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const submission = await root.submit({ type: "input", content: "ann: hi", entry: mention("hi") }, context); + const settled = await submission.wait(context); + const entries = await allEntries(root); + expect(transcript(entries)).toEqual([ + ["chat.mention", { author: "ann", text: "hi" }, "ann: hi"], + ["pi.assistant", "hi ann"], + ]); + const placed = entries.find((entry) => entry.kind === "chat.mention")!; + expect(placed.model).toEqual([{ role: "user", content: "ann: hi", timestamp: 42 }]); + expect(settled).toMatchObject({ status: "done", entry: placed.id, answer: entries.at(-1)!.id }); + expect(requests).toEqual([["user:ann: hi"]]); + // The context of a fork at the input is the same as for a pi.user entry. + const view = await root.context(context, { at: placed.id }); + expect(view.messages.filter((message) => message.role !== "system")).toEqual(placed.model); + await harness.close(context); + }); + + it("rejects an input entry whose kind is not a non-empty string or starts with pi., writing nothing", async () => { + const setup = chatSetup(); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const before = (await allEntries(root)).length; + for (const kind of [123, "", "pi.user", "pi.compaction"]) { + const draft = { type: "input", content: "x", entry: { kind } } as unknown as SubmissionDraft; + await expect(root.submit(draft, context)).rejects.toThrow("An input entry's kind must be"); + } + expect((await allEntries(root)).length).toBe(before); + expect(await harness.snapshot(LiveDoc, root.id, context)).toEqual({}); + // Writes keep their kinds: reset() writes pi.reset. + await root.reset(undefined, context); + expect((await allEntries(root)).at(-1)?.kind).toBe("pi.reset"); + await harness.close(context); + }); + + it("takes only kind and data from an untyped caller's entry", async () => { + const setup = chatSetup(); + const first = gated(answer("first")); + setup.faux.setResponses([answer("idle"), first.step, answer("queued")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const extra = { + kind: "chat.mention", + model: [{ role: "user", content: "forged", timestamp: 0 }], + head: "self", + edits: [{ target: 1, action: "omit" }], + }; + const draft = (content: string) => ({ type: "input", content, entry: extra }) as unknown as SubmissionDraft; + await (await root.submit(draft("idle"), context)).wait(context); + await root.submit({ type: "input", content: "a" }, context); + await first.reached; + const queued = await root.submit(draft("later"), context); + expect((await inboxItems(harness, root)).at(-1)).toEqual({ + id: queued.id, + mode: "followUp", + content: "later", + entry: { kind: "chat.mention" }, + }); + first.release(); + await queued.wait(context); + const mentions = (await allEntries(root)).filter((entry) => entry.kind === "chat.mention"); + expect(mentions.map((entry) => [entry.head, entry.edits, textOf(entry.model?.[0]), entry.model?.length])).toEqual( + [ + [undefined, undefined, "idle", 1], + [undefined, undefined, "later", 1], + ], + ); + await harness.close(context); + }); + + it("queues busy inputs with their entries in the inbox and places them as those entries at boundaries", async () => { + const setup = chatSetup(); + const gate = deferred(); + holdTool(setup, gate); + setup.faux.setResponses([HOLD, answer("after tools"), answer("follow-up")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const input = await root.submit({ type: "input", content: "a" }, context); + await toolRunning(harness, root); + const steer = await root.submit( + { type: "input", content: "s", entry: mention("s"), whenBusy: "steer", requestId: "r-s" }, + context, + ); + const followUp = await root.submit( + { type: "input", content: "f", entry: { kind: "p2.input" }, whenBusy: "followUp" }, + context, + ); + expect(await inboxItems(harness, root)).toEqual([ + { + id: steer.id, + mode: "steer", + content: "s", + entry: { kind: "chat.mention", data: { author: "ann", text: "s" } }, + }, + { id: followUp.id, mode: "followUp", content: "f", entry: { kind: "p2.input" } }, + ]); + // The request ID deduplicates a queued input with an entry as any other. + const again = await root.submit( + { type: "input", content: "other", entry: mention("other"), whenBusy: "steer", requestId: "r-s" }, + context, + ); + expect(again.id).toBe(steer.id); + await expect( + root.submit({ type: "input", content: "x", entry: mention("x"), whenBusy: "reject" }, context), + ).rejects.toBeInstanceOf(ConversationBusy); + expect(await inboxItems(harness, root)).toHaveLength(2); + + gate.resolve(); + await followUp.wait(context); + const entries = await allEntries(root); + expect(transcript(entries)).toEqual([ + ["pi.user", "a"], + ["pi.assistant"], + ["pi.tool-result"], + ["chat.mention", { author: "ann", text: "s" }, "s"], + ["pi.assistant", "after tools"], + ["p2.input", "f"], + ["pi.assistant", "follow-up"], + ]); + const answers = entries.filter((entry) => entry.kind === "pi.assistant"); + const placed = (kind: string) => entries.find((entry) => entry.kind === kind)!.id; + expect(await input.status(context)).toMatchObject({ status: "done", answer: answers[1]!.id }); + expect(await steer.status(context)).toMatchObject({ + status: "done", + entry: placed("chat.mention"), + answer: answers[1]!.id, + }); + expect(await followUp.status(context)).toMatchObject({ + status: "done", + entry: placed("p2.input"), + answer: answers[2]!.id, + }); + await harness.close(context); + }); + + it("withdraws queued inputs with entries, one by one and by conversation abort", async () => { + const setup = chatSetup(); + const first = gated(answer("first")); + setup.faux.setResponses([first.step]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const input = await root.submit({ type: "input", content: "a" }, context); + await first.reached; + const withdrawn = await root.submit({ type: "input", content: "w", entry: mention("w") }, context); + const aborted = await root.submit({ type: "input", content: "x", entry: mention("x") }, context); + expect(await withdrawn.abort(context)).toBe("aborted"); + expect((await inboxItems(harness, root)).map((item) => item.id)).toEqual([aborted.id]); + await root.abort(context); + for (const submission of [withdrawn, aborted]) { + expect(await submission.wait(context)).toMatchObject({ status: "unanswered", reason: "aborted" }); + } + expect(await input.wait(context)).toMatchObject({ status: "unanswered", reason: "aborted" }); + expect(await inboxItems(harness, root)).toEqual([]); + expect((await allEntries(root)).some((entry) => entry.kind === "chat.mention")).toBe(false); + await harness.close(context); + }); +}); + +describe("data-only writes", () => { + it("appends a data-only write mid-run in time order, leaving the model's messages unchanged", async () => { + const setup = chatSetup(); + const gate = deferred(); + holdTool(setup, gate); + const requests: string[][] = []; + setup.faux.setResponses([requestLog(requests, HOLD), requestLog(requests, answer("done"))]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const events: AgentEvent[] = []; + const stream = await watchEvents(harness, root.id, context); + stream.start(async (batch) => { + events.push(...batch); + }); + const input = await root.submit({ type: "input", content: "a" }, context); + await toolRunning(harness, root); + const chat = { kind: "chat.message", data: { author: "bob", text: "lunch?" } }; + const write = await root.submit({ type: "write", entry: chat, requestId: "bob:1" }, context); + // Settled and appended in its own commit, between the call and its still-running tool's result. + const settled = await write.status(context); + expect(settled.status).toBe("done"); + expect(await inboxItems(harness, root)).toEqual([]); + expect(transcript(await allEntries(root))).toEqual([ + ["pi.user", "a"], + ["pi.assistant"], + ["chat.message", { author: "bob", text: "lunch?" }], + ]); + await waitFor(() => events.some((event) => event.type === "entry_appended")); + const appended = events.find((event) => event.type === "entry_appended"); + expect(appended).toMatchObject({ + entry: { kind: "chat.message", id: settled.status === "done" && settled.entry }, + }); + expect(events.some((event) => event.type === "run_end")).toBe(false); + // A repeated request ID returns the same write without a second entry. + expect((await root.submit({ type: "write", entry: chat, requestId: "bob:1" }, context)).id).toBe(write.id); + + gate.resolve(); + await input.wait(context); + expect(transcript(await allEntries(root))).toEqual([ + ["pi.user", "a"], + ["pi.assistant"], + ["chat.message", { author: "bob", text: "lunch?" }], + ["pi.tool-result"], + ["pi.assistant", "done"], + ]); + expect(requests).toEqual([ + ["system:", "user:a"], + ["system:", "user:a", "assistant:", "toolResult:held"], + ]); + await stream.stop(); + await harness.close(context); + }); + + it("queues a data-only write behind a queued reset, so it lands after the reset and stays visible", async () => { + const setup = chatSetup(); + const first = gated(answer("first")); + setup.faux.setResponses([first.step]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const input = await root.submit({ type: "input", content: "a" }, context); + await first.reached; + await root.reset(undefined, context); + const chat = await root.submit({ type: "write", entry: { kind: "chat.message", data: "hi" } }, context); + expect((await chat.status(context)).status).toBe("queued"); + first.release(); + await input.wait(context); + const settled = await chat.wait(context); + const entries = await allEntries(root); + const reset = entries.find((entry) => entry.kind === "pi.reset")!; + expect(settled.status === "done" && settled.entry).toBeGreaterThan(reset.id); + expect((await root.context(context)).entries.map((entry) => entry.kind)).toEqual(["pi.reset", "chat.message"]); + // Once the reset is placed, display-only writes go in at once again. + const later = await root.submit({ type: "write", entry: { kind: "chat.message", data: "again" } }, context); + expect((await later.status(context)).status).toBe("done"); + await harness.close(context); + }); + + it("leaves the request unchanged by a data-only write that lands while the generation prepares", async () => { + const setup = chatSetup(); + const rendering = deferred(); + const render = deferred(); + let renders = 0; + addSection(setup.registry, "slow", async () => { + if (renders++ === 1) { + rendering.resolve(); + await render.promise; + } + return "SLOW"; + }); + const requests: string[][] = []; + setup.faux.setResponses([requestLog(requests, answer("one")), requestLog(requests, answer("two"))]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "a" }, context)).wait(context); + const second = await root.submit({ type: "input", content: "b" }, context); + await rendering.promise; + await root.submit({ type: "write", entry: { kind: "chat.message", data: "hi" } }, context); + render.resolve(); + await second.wait(context); + expect(requests[1]).toEqual(["system:", "user:a", "assistant:one", "user:b"]); + const kinds = (await allEntries(root)).map((entry) => entry.kind); + // The unchanged prompt adds no second pi.system entry. + expect(kinds).toEqual(["pi.user", "pi.system", "pi.assistant", "pi.user", "chat.message", "pi.assistant"]); + await harness.close(context); + }); + + it("appends a data-only write while idle after a failed run without placing the queued items", async () => { + const setup = chatSetup(); + const failing = gated(fauxAssistantMessage([], { stopReason: "error", errorMessage: "invalid request" })); + setup.faux.setResponses([failing.step, answer("for f")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const input = await root.submit({ type: "input", content: "a" }, context); + await failing.reached; + const f = await root.submit({ type: "input", content: "f", entry: mention("f") }, context); + failing.release(); + expect(await input.wait(context)).toMatchObject({ status: "unanswered", reason: "model_error" }); + await harness.waitForIdle(context); + const note = await root.submit({ type: "write", entry: { kind: "note", data: "n" } }, context); + expect((await note.status(context)).status).toBe("done"); + await harness.waitForIdle(context); + expect((await f.status(context)).status).toBe("queued"); + expect((await inboxItems(harness, root)).map((item) => item.id)).toEqual([f.id]); + expect(await harness.snapshot(LiveDoc, root.id, context)).toEqual({}); + expect(setup.faux.state.callCount).toBe(1); + // The next input places the waiting mention first. + const g = await root.submit({ type: "input", content: "g" }, context); + await g.wait(context); + expect(transcript(await allEntries(root)).filter(([kind]) => kind !== "pi.assistant")).toEqual([ + ["pi.user", "a"], + ["note", "n"], + ["chat.mention", { author: "ann", text: "f" }, "f"], + ["pi.user", "g"], + ]); + await harness.close(context); + }); + + it("keeps a queued input's entry across reopen and places it as that entry", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-input-entries-")); + const path = join(directory, "session.sqlite"); + try { + const setup = chatSetup(); + const first = gated(answer("first")); + setup.faux.setResponses([first.step, answer("first again"), answer("for f")]); + let opened = await openChat(await openNodeSqliteStorage(path), setup); + await opened.root.submit({ type: "input", content: "a" }, context); + await first.reached; + const f = (await opened.root.submit({ type: "input", content: "f", entry: mention("f") }, context)).id; + await opened.harness.close(context); + + opened = await openChat(await openNodeSqliteStorage(path), setup); + expect((await inboxItems(opened.harness, opened.root)).find((item) => item.id === f)).toMatchObject({ + entry: mention("f"), + }); + const settled = await (await opened.harness.submission(f, context))!.wait(context); + const placed = (await allEntries(opened.root)).find((entry) => entry.kind === "chat.mention")!; + expect(settled).toMatchObject({ status: "done", entry: placed.id }); + expect(placed).toMatchObject({ data: mention("f").data, model: [{ role: "user", content: "f" }] }); + await opened.harness.close(context); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it("appends a data-only write at once behind a non-empty inbox, and still queues writes with model or edits", async () => { + const setup = chatSetup(); + const first = gated(answer("first")); + setup.faux.setResponses([first.step, answer("second")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const input = await root.submit({ type: "input", content: "a" }, context); + await first.reached; + const followUp = await root.submit({ type: "input", content: "f" }, context); + const model = await root.submit( + { type: "write", entry: { kind: "note", model: [{ role: "user", content: "m", timestamp: 0 }] } }, + context, + ); + const target = (await allEntries(root)).find((entry) => entry.kind === "pi.user")!; + const edit = await root.submit( + { type: "write", entry: { kind: "redact", edits: [{ target: target.id, action: "omit" }] } }, + context, + ); + const note = await root.submit({ type: "write", entry: { kind: "note", data: "n" } }, context); + expect((await note.status(context)).status).toBe("done"); + expect((await inboxItems(harness, root)).map((item) => [item.id, item.mode])).toEqual([ + [followUp.id, "followUp"], + [model.id, "write"], + [edit.id, "write"], + ]); + first.release(); + await followUp.wait(context); + expect(transcript(await allEntries(root))).toEqual([ + ["pi.user", "a"], + ["note", "n"], + ["pi.assistant", "first"], + ["note", "m"], + ["redact"], + ["pi.user", "f"], + ["pi.assistant", "second"], + ]); + expect(await input.status(context)).toMatchObject({ status: "done" }); + await harness.close(context); + }); +}); diff --git a/packages/durable/test/harness-ownership.test.ts b/packages/durable/test/harness-ownership.test.ts index c502cfa5f..4fd669847 100644 --- a/packages/durable/test/harness-ownership.test.ts +++ b/packages/durable/test/harness-ownership.test.ts @@ -182,6 +182,11 @@ async function status(harness: Harness, id: TaskId) { return (await harness.getTask(id, context))!.state; } +/** A write that affects model context, so it queues while busy; a data-only write would be appended at once. */ +function queuedNote(text: string) { + return { kind: "note", model: [{ role: "user", content: text, timestamp: 0 }] } as const; +} + async function openHarness(storage: Storage = new MemoryStorage()) { const opened = await openTasks(storage, [Hold, Waiter, Spawner]); const root = await opened.harness.root(context); @@ -652,7 +657,7 @@ describe("ownership", () => { const background = await ownedChild(root, "background", { background: true }); await busy(root, foreground.owner); const input = (await root.submit({ type: "input", content: "later" }, context)).id; - const write = (await root.submit({ type: "write", entry: { kind: "note" } }, context)).id; + const write = (await root.submit({ type: "write", entry: queuedNote("w") }, context)).id; await root.abort(context); for (const id of [foreground.owner, foreground.inner]) { expect((await status(harness, id)).status).toBe("terminal"); @@ -693,7 +698,7 @@ describe("ownership", () => { await busy(child, tree.inner); const own = (await root.submit({ type: "input", content: "own" }, context)).id; const below = (await child.submit({ type: "input", content: "below" }, context)).id; - const write = (await child.submit({ type: "write", entry: { kind: "note" } }, context)).id; + const write = (await child.submit({ type: "write", entry: queuedNote("w") }, context)).id; await harness.abortTask(tree.owner, context); await harness.waitForTask(tree.inner, context); const statusOf = async (id: typeof own) => diff --git a/packages/durable/test/types.test.ts b/packages/durable/test/types.test.ts index 8a20bc07f..30ee143f1 100644 --- a/packages/durable/test/types.test.ts +++ b/packages/durable/test/types.test.ts @@ -317,8 +317,8 @@ it("types submissions, task waits, compaction, and tasks erased into extensions" expectTypeOf(typedWaits).toBeFunction(); const compileTimeFailures = (conversation: Conversation) => { - // @ts-expect-error an input submission carries no entry - const inputWithEntry: SubmissionDraft = { type: "input", content: "hi", entry: { kind: "note" } }; + // @ts-expect-error an input's entry has no model: its model message is the input's content + const inputWithModel: SubmissionDraft = { type: "input", content: "hi", entry: { kind: "note", model: [] } }; // @ts-expect-error a write submission carries no content const writeWithContent: SubmissionDraft = { type: "write", entry: { kind: "note" }, content: "hi" }; // @ts-expect-error a write submission never generates, so it has no busy policy @@ -340,7 +340,7 @@ it("types submissions, task waits, compaction, and tasks erased into extensions" phases: { plan: async () => {} }, abort: async () => {}, }); - void [inputWithEntry, writeWithContent, writeWhenBusy, wrongHook, unknownHook, wrongInput, missingPhase]; + void [inputWithModel, writeWithContent, writeWhenBusy, wrongHook, unknownHook, wrongInput, missingPhase]; }; expectTypeOf(compileTimeFailures).toBeFunction(); });