diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 6511187f0..f76b9543f 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -111,6 +111,7 @@ - Per-conversation turn limits: `AgentState.maxTurns` (`configure({ maxTurns })`, `null` clears; a `RangeError` for anything but an integer 1 or more) bounds the tool rounds one run takes. The round that reaches it runs its tools, then the run ends `unanswered` with the new `RunFailureReason` `turn_limit`, like a failed run, so `settings.afterFailedRun` decides the queued inputs and a continue resumes it on the round's results. The count is `pi.live.run.turns`; every run starts at 0, and an `onYield` continuation starts it over. Task-owned children and forks inherit the limit; `Agent.maxTurns` exposes it ([#11043](https://github.com/earendil-works/pi/issues/11043)) - `whenIdle: "reject"` on input submissions: an idle conversation rejects the input with the new `ConversationIdle` and writes nothing, so it never starts a run. With `whenBusy: "steer"` it steers the current run or is refused, in one admission instead of a host's check and a separate submit that can race the run's end. A known request ID still returns its submission ([#11149](https://github.com/earendil-works/pi/issues/11149)) - `Conversation.submit(tx, draft)` admits a submission inside the caller's commit (`Harness.commit()`, `Conversation.commit()`) with the same rules as `submit(draft, context)`, and returns its ID: a host checks its own documents and `pi.live`, admits, and writes its own state in one commit, which stores all of it or none. `ConversationHandle.submit(tx, draft)` does the same inside a task's or tool's own commit. The Harness supplies the clock, queue modes, the `pi.durable.submit` span (covering the admission only; its `result` is provisional until the commit is stored), and the request for progress. It must come before the commit's first table write, once per commit, and throws `ReadAfterWrite` otherwise on every path ([#11149](https://github.com/earendil-works/pi/issues/11149)) +- `HookApi.entry()` and `entries()`, and `TaskRuntime.entries()`: a hook or task reads the conversation's raw history, which compaction never changes, so a range bounded by entry IDs reads the same before and after a compaction and on a rerun ([#11141](https://github.com/earendil-works/pi/issues/11141)) ### Changed diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 3ef2b232f..e795d5556 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -2144,6 +2144,13 @@ interface TaskRuntime extends DocumentObserver, Docum /** Committed entry visible from the task's conversation. */ entry(id: EntryId, context: Context): Promise; entry(token: Entry, id: EntryId, context: Context): Promise | undefined>; + /** Committed raw history of the task's conversation, fork-aware, as `Conversation.entries()`. */ + entries( + query: Omit, + limit: number, + cursor: Cursor | undefined, + context: Context, + ): Promise>; /** Committed raw active transcript and model context, optionally cut off at `at`. */ context(conversationId: ConversationId, context: Context, options?: { readonly at?: EntryId }): Promise; /** The Harness clock. */ @@ -3279,11 +3286,20 @@ type HookResult = T | undefined | Promise; interface HookApi extends DocumentReader { readonly taskId: TaskId; readonly conversationId: ConversationId; - /** `HarnessOptions.models`. */ /** The current `HarnessOptions.models`, read at each access. */ readonly models: Models; memo(name: string, context: Context): Promise; memo(name: string, candidate: T, context: Context): Promise; + /** As `TaskRuntime.entry()` (section 5.1). */ + entry(id: EntryId, context: Context): Promise; + entry(token: Entry, id: EntryId, context: Context): Promise | undefined>; + /** As `TaskRuntime.entries()` (section 5.1). */ + entries( + query: Omit, + limit: number, + cursor: Cursor | undefined, + context: Context, + ): Promise>; } interface GenerationHooks { @@ -3355,6 +3371,15 @@ after a first decision, does so inside its own `invoke`. Hooks receive the task' runtime as their `HookApi`; hook memos and the task's own memos share one namespace, so hook authors prefix their memo names. +Hooks read committed state as it is when they run, off the line, so a hook a +crash reruns may see more. Entries never change and compaction never removes +them (section 8.7): a range bounded by entry IDs reads the same before and +after a compaction and in a rerun. `HookApi` has no `context()`: the active +transcript changes in the middle of a run, when a compaction summary or a +handoff's reset lands, so a hook asking what a run did reads its history +instead. A range also holds entries of others, such as display-only writes, +compaction summaries, and entries of other tasks; hooks filter by `kind`. + Selection decides which behavior a conversation has; the extension's own document holds that behavior's state: diff --git a/packages/durable/src/harness/harness.ts b/packages/durable/src/harness/harness.ts index 055c05bd7..906fb3539 100644 --- a/packages/durable/src/harness/harness.ts +++ b/packages/durable/src/harness/harness.ts @@ -74,7 +74,7 @@ import type { Wake, } from "./types.ts"; import { addUsageState, UsageDoc, type UsageState } from "./usage.ts"; -import { scanAll } from "./util.ts"; +import { conversationEntries, scanAll } from "./util.ts"; import { type ConversationView, ConversationViews, type ConversationWindow } from "./view.ts"; const SCAN_PAGE_SIZE = 256; @@ -160,13 +160,8 @@ class ConversationImpl implements Conversation { cursor: Cursor | undefined, context: Context, ): Promise> { - const bounded: EntryQuery = { - conversationId: this.id, - ...(query.minEntryId === undefined ? {} : { minEntryId: query.minEntryId }), - ...(query.maxEntryId === undefined ? {} : { maxEntryId: query.maxEntryId }), - ...(query.order === undefined ? {} : { order: query.order }), - }; - return this.#host.harness.readOnLine(() => this.#host.storage.scanEntries(bounded, limit, cursor, context)); + const scan = conversationEntries(this.id, query); + return this.#host.harness.readOnLine(() => this.#host.storage.scanEntries(scan, limit, cursor, context)); } fork(at: EntryId, options: ConversationCreateOptions, context: Context): Promise { diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts index 1fc981be6..de26aceb3 100644 --- a/packages/durable/src/harness/scheduler.ts +++ b/packages/durable/src/harness/scheduler.ts @@ -52,7 +52,7 @@ import type { TaskInspection, Wake, } from "./types.ts"; -import { closedError, optionalExtension, scanAll, Waiters } from "./util.ts"; +import { closedError, conversationEntries, optionalExtension, scanAll, Waiters } from "./util.ts"; type AnyTaskRecord = TaskRecord; /** A record that can still run code: pending, running, waiting, or sleeping. */ @@ -2257,6 +2257,17 @@ export class TaskScheduler { return token === undefined || entry?.kind === token.kind ? entry : undefined; }); }) as ErasedRuntime["entry"], + entries: (query, limit, cursor, context) => + this.#read(invocation, () => + this.#session.readOnLine(() => + this.#storage.scanEntries( + conversationEntries(invocation.conversationId, query), + limit, + cursor, + context, + ), + ), + ), context: (conversationId, context, options) => this.#read(invocation, async () => { const { view, range } = await readContextFrom( diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index 3306ffe09..99a024c6a 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -25,6 +25,7 @@ import type { Cursor, DocumentObserver, DocumentReader, + Entry, EntryDraft, EntryId, EntryQuery, @@ -44,6 +45,7 @@ import type { TaskRecord, TaskState, Tx, + TypedEntry, } from "../types.ts"; import type { TaskGraph, TaskGraphWatch } from "./task-graph.ts"; import type { UsageState } from "./usage.ts"; @@ -933,7 +935,10 @@ export interface Harness extends Session { observeTaskGraph(observer: Observer, context: Context): Promise>; } -/** What a hook may use: committed reads and the asking task's memos, which hooks and the task share. */ +/** + * What a hook may use: committed reads and the asking task's memos, which hooks and the task share. Hooks run off the + * Session line, so reads see what is committed when they run; a hook rerun after a crash may see more. + */ export interface HookApi extends DocumentReader { readonly taskId: TaskId; readonly conversationId: ConversationId; @@ -941,6 +946,17 @@ export interface HookApi extends DocumentReader { readonly models: Models; memo(name: string, context: Context): Promise; memo(name: string, candidate: T, context: Context): Promise; + /** Committed entry visible from the task's conversation, as `TaskRuntime.entry()`. */ + entry(id: EntryId, context: Context): Promise; + /** Undefined when the entry is absent, not visible, or has another kind. */ + entry(token: Entry, id: EntryId, context: Context): Promise | undefined>; + /** Committed raw history of the task's conversation, as `TaskRuntime.entries()`. */ + entries( + query: Omit, + limit: number, + cursor: Cursor | undefined, + context: Context, + ): Promise>; } export type HookResult = T | undefined | Promise; diff --git a/packages/durable/src/harness/util.ts b/packages/durable/src/harness/util.ts index cf2222a5a..f3c8ea374 100644 --- a/packages/durable/src/harness/util.ts +++ b/packages/durable/src/harness/util.ts @@ -1,6 +1,6 @@ import type { Context } from "@earendil-works/chord"; import { SessionFailed } from "../errors.ts"; -import type { Cursor, Page } from "../types.ts"; +import type { ConversationId, Cursor, EntryQuery, Page } from "../types.ts"; /** Pending waits by key. Each settles once: through `resolve`, `rejectAll`, or cancellation of its context. */ export class Waiters { @@ -64,3 +64,16 @@ export function closedError(session: { readonly failure: { readonly error: unkno export function optionalExtension(extension: string | undefined): { readonly extension?: string } { return extension === undefined ? {} : { extension }; } + +/** The scan of conversation `conversationId`'s history that `query` asks for, without fields an untyped caller added. */ +export function conversationEntries( + conversationId: ConversationId, + query: Omit, +): EntryQuery { + return { + conversationId, + ...(query.minEntryId === undefined ? {} : { minEntryId: query.minEntryId }), + ...(query.maxEntryId === undefined ? {} : { maxEntryId: query.maxEntryId }), + ...(query.order === undefined ? {} : { order: query.order }), + }; +} diff --git a/packages/durable/src/types.ts b/packages/durable/src/types.ts index 3ea1f1ca4..4f17a3711 100644 --- a/packages/durable/src/types.ts +++ b/packages/durable/src/types.ts @@ -324,6 +324,17 @@ export interface TaskRuntime extends DocumentObserver entry(id: EntryId, context: Context): Promise; /** Undefined when the entry is absent, not visible, or has another kind. */ entry(token: Entry, id: EntryId, context: Context): Promise | undefined>; + /** + * Committed raw history of the task's conversation, fork-aware, as `Conversation.entries()`: every entry, also those a + * compaction summary or reset left behind. Entries never change, so a range bounded by entry IDs reads the same + * before and after a compaction, and in a rerun after a crash. + */ + entries( + query: Omit, + limit: number, + cursor: Cursor | undefined, + context: Context, + ): Promise>; /** Committed raw active transcript and model context, optionally cut off at the visible entry `at`. */ context(conversationId: ConversationId, context: Context, options?: { readonly at?: EntryId }): Promise; /** The Harness clock. */ diff --git a/packages/durable/test/harness-hook-reads.test.ts b/packages/durable/test/harness-hook-reads.test.ts new file mode 100644 index 000000000..2681d244d --- /dev/null +++ b/packages/durable/test/harness-hook-reads.test.ts @@ -0,0 +1,94 @@ +import { fauxAssistantMessage } from "@earendil-works/pi-ai"; +import { + defineExtension, + defineTask, + type EntryRecord, + GenerationTask, + hook, + MemoryStorage, + UserEntry, +} from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import { allEntries, chatSetup, openChat } from "./chat-support.ts"; +import { context } from "./session-support.ts"; +import { completed } from "./task-support.ts"; + +describe("entry reads for hooks and tasks", () => { + // #11141 + it("reads raw history and single entries from a hook", async () => { + const setup = chatSetup(); + const seen: { kinds: string[]; user?: string }[] = []; + setup.registry.install( + defineExtension({ + name: "reader", + hooks: [ + hook(GenerationTask, { + onYield: async (_answer, api, ctx) => { + const page = await api.entries({ order: "ascending" }, 10, undefined, ctx); + const first = page.items[0]!; + const user = await api.entry(UserEntry, first.id, ctx); + seen.push({ + kinds: page.items.map((entry) => entry.kind), + ...(user === undefined ? {} : { user: user.kind }), + }); + // Another kind: undefined. + expect(await api.entry(GenerationTask as never, first.id, ctx)).toBeUndefined(); + return undefined; + }, + }), + ], + }), + ); + setup.faux.setResponses([fauxAssistantMessage("done")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + // The answer is not committed yet when `onYield` asks. + expect(seen).toEqual([{ kinds: ["pi.user"], user: "pi.user" }]); + await harness.close(context); + }); + + // #11141 + it("pages a task's conversation history by entry ID range", async () => { + const setup = chatSetup(); + const pages: (readonly EntryRecord[])[] = []; + const Reader = defineTask({ + name: "test.reader", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async (_task, runtime, ctx) => { + const all = await runtime.entries({ order: "ascending" }, 1, undefined, ctx); + pages.push(all.items); + const rest = await runtime.entries({ order: "ascending" }, 1, all.next, ctx); + pages.push(rest.items); + const bounded = await runtime.entries( + { minEntryId: rest.items[0]!.id, maxEntryId: rest.items[0]!.id }, + 10, + undefined, + ctx, + ); + pages.push(bounded.items); + await runtime.commit(() => completed(null), ctx); + }, + }, + abort: async () => {}, + }); + setup.registry.install(defineExtension({ name: "reader", tasks: [Reader] })); + setup.faux.setResponses([fauxAssistantMessage("done")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + const id = await root.commit( + (tx) => tx.createTask(Reader, null, { ownership: { kind: "conversation" } }), + context, + ); + await harness.waitForTask(id, context); + const entries = await allEntries(root); + expect(entries.map((entry) => entry.kind)).toEqual(["pi.user", "pi.assistant"]); + expect(pages.map((page) => page.map((entry) => entry.id))).toEqual([ + [entries[0]!.id], + [entries[1]!.id], + [entries[1]!.id], + ]); + await harness.close(context); + }); +});