feat(durable): entry reads for hooks and tasks

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
in a rerun.

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
 

packages/durable/docs/spec.md

@@ -2144,6 +2144,13 @@ interface TaskRuntime<I, S, R, H extends object> extends DocumentObserver, Docum
   /** Committed entry visible from the task's conversation. */
   entry(id: EntryId, context: Context): Promise<EntryRecord | undefined>;
   entry<D extends JsonValue>(token: Entry<D>, id: EntryId, context: Context): Promise<TypedEntry<D> | undefined>;
+  /** Committed raw history of the task's conversation, fork-aware, as `Conversation.entries()`. */
+  entries(
+    query: Omit<EntryQuery, "conversationId">,
+    limit: number,
+    cursor: Cursor | undefined,
+    context: Context,
+  ): Promise<Page<EntryRecord, Cursor>>;
   /** Committed raw active transcript and model context, optionally cut off at `at`. */
   context(conversationId: ConversationId, context: Context, options?: { readonly at?: EntryId }): Promise<ContextView>;
   /** The Harness clock. */
@@ -3279,11 +3286,20 @@ type HookResult<T> = T | undefined | Promise<T | undefined>;
 interface HookApi extends DocumentReader {
   readonly taskId: TaskId;
   readonly conversationId: ConversationId;
-  /** `HarnessOptions.models`. */
   /** The current `HarnessOptions.models`, read at each access. */
   readonly models: Models;
   memo<T extends JsonValue>(name: string, context: Context): Promise<T | undefined>;
   memo<T extends JsonValue>(name: string, candidate: T, context: Context): Promise<T>;
+  /** As `TaskRuntime.entry()` (section 5.1). */
+  entry(id: EntryId, context: Context): Promise<EntryRecord | undefined>;
+  entry<D extends JsonValue>(token: Entry<D>, id: EntryId, context: Context): Promise<TypedEntry<D> | undefined>;
+  /** As `TaskRuntime.entries()` (section 5.1). */
+  entries(
+    query: Omit<EntryQuery, "conversationId">,
+    limit: number,
+    cursor: Cursor | undefined,
+    context: Context,
+  ): Promise<Page<EntryRecord, Cursor>>;
 }
 
 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:
 

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<Tool extends ToolRegistration> implements Conversation {
 		cursor: Cursor | undefined,
 		context: Context,
 	): Promise<Page<EntryRecord, Cursor>> {
-		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<Conversation> {

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<JsonValue, JsonValue, JsonValue>;
 /** 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(

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<TaskGraph>, context: Context): Promise<Observation<TaskGraph>>;
 }
 
-/** 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<T extends JsonValue>(name: string, context: Context): Promise<T | undefined>;
 	memo<T extends JsonValue>(name: string, candidate: T, context: Context): Promise<T>;
+	/** Committed entry visible from the task's conversation, as `TaskRuntime.entry()`. */
+	entry(id: EntryId, context: Context): Promise<EntryRecord | undefined>;
+	/** Undefined when the entry is absent, not visible, or has another kind. */
+	entry<D extends JsonValue>(token: Entry<D>, id: EntryId, context: Context): Promise<TypedEntry<D> | undefined>;
+	/** Committed raw history of the task's conversation, as `TaskRuntime.entries()`. */
+	entries(
+		query: Omit<EntryQuery, "conversationId">,
+		limit: number,
+		cursor: Cursor | undefined,
+		context: Context,
+	): Promise<Page<EntryRecord, Cursor>>;
 }
 
 export type HookResult<T> = T | undefined | Promise<T | undefined>;

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<K, T> {
@@ -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, "conversationId">,
+): EntryQuery {
+	return {
+		conversationId,
+		...(query.minEntryId === undefined ? {} : { minEntryId: query.minEntryId }),
+		...(query.maxEntryId === undefined ? {} : { maxEntryId: query.maxEntryId }),
+		...(query.order === undefined ? {} : { order: query.order }),
+	};
+}

packages/durable/src/types.ts

@@ -324,6 +324,17 @@ export interface TaskRuntime<I, S, R, H extends object> extends DocumentObserver
 	entry(id: EntryId, context: Context): Promise<EntryRecord | undefined>;
 	/** Undefined when the entry is absent, not visible, or has another kind. */
 	entry<D extends JsonValue>(token: Entry<D>, id: EntryId, context: Context): Promise<TypedEntry<D> | 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<EntryQuery, "conversationId">,
+		limit: number,
+		cursor: Cursor | undefined,
+		context: Context,
+	): Promise<Page<EntryRecord, Cursor>>;
 	/** 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<ContextView>;
 	/** The Harness clock. */

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<null, { phase: "run" }, null>({
+			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);
+	});
+});