mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
feat(durable): inputs placed as application entries, display-only writes appended at once
- Input submissions take entry: { kind, data } (InputEntry): placed as that
kind with that data and the content as user message, instead of pi.user.
Kinds must be non-empty strings outside pi.*. Queued inputs carry it in
their pi.inbox item.
- Writes with only kind and data (no model, head, edits) are appended and
settled done at once, busy or not; only a queued reset holds them back.
- Experimental hosts render such inputs as user messages and title by them.
closes #10407
This commit is contained in:
Mario Zechner
parent
cdaf6c227d
commit
e6b9a6af5d
18 changed files with 683 additions and 57 deletions
|
|
@ -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)));
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<Ope
|
|||
for (const change of publication.changes) {
|
||||
if (change.type === "conversation") {
|
||||
conversations = [...conversations, { id: change.value.id, label: label(change.value.id) }];
|
||||
} else if (change.type === "entry" && change.value.kind === "pi.user") {
|
||||
} else if (change.type === "entry" && isInput(change.value)) {
|
||||
const id = change.value.conversationId;
|
||||
conversations = conversations.map((summary) =>
|
||||
summary.id === id && summary.title === undefined ? { ...summary, ...titleOf(change.value) } : summary,
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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<Ope
|
|||
for (const change of publication.changes) {
|
||||
if (change.type === "conversation") {
|
||||
conversations = [...conversations, { id: change.value.id, label: label(change.value.id) }];
|
||||
} else if (change.type === "entry" && change.value.kind === "pi.user") {
|
||||
} else if (change.type === "entry" && isInput(change.value)) {
|
||||
const id = change.value.conversationId;
|
||||
conversations = conversations.map((summary) =>
|
||||
summary.id === id && summary.title === undefined ? { ...summary, ...titleOf(change.value) } : summary,
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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 }`.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -119,8 +119,9 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
|
|||
phases: {
|
||||
/**
|
||||
* Render the system prompt and tool loadout and append the positional `pi.system` entries they need, then move to
|
||||
* `request`. The agent and settings resolved here are fixed for this request. Only the Harness writes to a busy
|
||||
* conversation, so the transcript read here is still the tail at the commit.
|
||||
* `request`. The agent and settings resolved here are fixed for this request. Only the Harness writes model
|
||||
* context to a busy conversation, so the context read here is still current at the commit; display-only entries
|
||||
* may land meanwhile.
|
||||
*/
|
||||
prepare: async (task, runtime, context) => {
|
||||
const { conversationId } = runtime;
|
||||
|
|
|
|||
|
|
@ -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<UserInput> }
|
||||
| {
|
||||
id: SubmissionId;
|
||||
mode: "steer" | "followUp";
|
||||
content: JsonRepresentation<UserInput>;
|
||||
entry?: JsonRepresentation<InputEntry>;
|
||||
}
|
||||
/** `entry` is an `EntryDraft`, stored as plain JSON. */
|
||||
| { id: SubmissionId; mode: "write"; entry: JsonObject };
|
||||
|
||||
|
|
@ -70,7 +75,7 @@ export async function applyBoundary(
|
|||
): Promise<BoundaryResult> {
|
||||
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<Extract<InboxItem, { mode: "steer" | "followUp" }>>;
|
||||
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<EntryRecord> {
|
||||
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<InputEntry> {
|
||||
const picked = data === undefined ? { kind } : { kind, data };
|
||||
return copyJson(picked, { omitUndefinedProperties: true }) as JsonRepresentation<InputEntry>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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;
|
||||
|
|
|
|||
|
|
@ -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<SubmissionId> {
|
||||
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<UserInput> });
|
||||
const content = value as JsonRepresentation<UserInput>;
|
||||
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]);
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
|
||||
/**
|
||||
* 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<Submission>;
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -86,6 +86,7 @@ export type {
|
|||
HookRegistration,
|
||||
HookResult,
|
||||
HooksOf,
|
||||
InputEntry,
|
||||
InputSubmissionDraft,
|
||||
ModelRef,
|
||||
NestedToolExecutionResult,
|
||||
|
|
|
|||
|
|
@ -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")]),
|
||||
|
|
|
|||
|
|
@ -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<SubmissionRecord> {
|
||||
|
|
@ -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",
|
||||
]);
|
||||
|
|
|
|||
456
packages/durable/test/harness-input-entries.test.ts
Normal file
456
packages/durable/test/harness-input-entries.test.ts
Normal file
|
|
@ -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<void>; 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<void> {
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
|
@ -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) =>
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
});
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue