fix(durable): reuse the scanned context range within a task invocation

runtime.context() keeps the range its last read scanned for the task
invocation. A later read with the same conversation and head marker scans
only entries committed after that range's tail and re-derives the view in
memory. A generation's request no longer rereads the whole transcript after
its prepare. The range is dropped when the invocation ends or sleeps.

packages/durable/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Changed
+
+- Within one task invocation, `runtime.context()` reuses the range its previous read scanned and reads only entries committed since then, unless the head marker changed. A generation's request no longer rereads the whole transcript after its prepare, which roughly halves context reads per turn. The range is dropped when the invocation ends or sleeps.
+
 ## [1.0.4] - 2026-10-05
 
 ### Breaking Changes

packages/durable/src/harness/context.ts

@@ -1,7 +1,8 @@
 import type { Context } from "@earendil-works/chord";
 import type { AssistantMessage, Message, ToolCall, ToolResultMessage } from "@earendil-works/pi-ai";
 import type { SessionImpl } from "../session/session.ts";
-import type { ContextEdit, ConversationId, Cursor, EntryId, EntryRecord, Storage } from "../types.ts";
+import { idFromNumber } from "../ids.ts";
+import type { ContextEdit, ConversationId, Cursor, EntryId, EntryRecord, EntryQuery, Storage } from "../types.ts";
 import type { ContextView } from "./types.ts";
 
 const SCAN_PAGE_SIZE = 256;
@@ -38,6 +39,47 @@ export async function captureContextBounds(
 	return { head: await storage.findLatestHeadMarker(conversationId, tail, context), tail };
 }
 
+/**
+ * Visible entries one context read scanned: from the head marker's head, or transcript start, through `bounds.tail`,
+ * oldest first. Entries at or below the tail never change, so a later read with the same head marker extends it.
+ */
+export type ContextRange = {
+	readonly conversationId: ConversationId;
+	readonly bounds: ContextBounds;
+	readonly entries: readonly EntryRecord[];
+};
+
+/**
+ * `readContext()` that reuses `previous`: with the same conversation and head marker, only entries after its tail are
+ * scanned. Returns the view and the range to pass to the next read.
+ */
+export async function readContextFrom(
+	session: SessionImpl,
+	storage: Storage,
+	conversationId: ConversationId,
+	context: Context,
+	at: EntryId | undefined,
+	previous: ContextRange | undefined,
+): Promise<{ readonly view: ContextView; readonly range: ContextRange | undefined }> {
+	const bounds = await session.readOnLine(() => captureContextBounds(storage, conversationId, context, at));
+	if (bounds === undefined) return { view: emptyView(), range: undefined };
+	let entries: readonly EntryRecord[];
+	if (
+		previous === undefined ||
+		previous.conversationId !== conversationId ||
+		previous.bounds.head?.id !== bounds.head?.id
+	) {
+		entries = await scanRange(storage, conversationId, rangeQuery(bounds), context);
+	} else if (bounds.tail <= previous.bounds.tail) {
+		entries = previous.entries.filter((entry) => entry.id <= bounds.tail);
+	} else {
+		const minEntryId = idFromNumber<EntryId>(previous.bounds.tail + 1);
+		const added = await scanRange(storage, conversationId, { minEntryId, maxEntryId: bounds.tail }, context);
+		entries = [...previous.entries, ...added];
+	}
+	return { view: deriveView(bounds.head, entries), range: { conversationId, bounds, entries } };
+}
+
 /** Committed context of one conversation: bounds captured on the Session line, entries derived off it. */
 export async function readContext(
 	session: SessionImpl,
@@ -62,9 +104,16 @@ export async function deriveContext(
 	bounds: ContextBounds | undefined,
 	context: Context,
 ): Promise<ContextView> {
-	if (bounds === undefined) return { head: undefined, entries: [], contributions: [], messages: [] };
-	const head = bounds.head;
-	const range = await scanRange(storage, conversationId, bounds, context);
+	if (bounds === undefined) return emptyView();
+	return deriveView(bounds.head, await scanRange(storage, conversationId, rangeQuery(bounds), context));
+}
+
+function emptyView(): ContextView {
+	return { head: undefined, entries: [], contributions: [], messages: [] };
+}
+
+/** Derive the context view from the scanned range of `head`. */
+function deriveView(head: ContextBounds["head"], range: readonly EntryRecord[]): ContextView {
 	const edits = new Map<EntryId, ContextEdit>();
 	// Edits of every entry in the range count, including older head markers that `selectActive()` drops.
 	for (const entry of range) for (const edit of entry.edits ?? []) edits.set(edit.target, edit);
@@ -87,38 +136,36 @@ export async function activeEntries(
 	conversationId: ConversationId,
 	bounds: ContextBounds | undefined,
 	context: Context,
-): Promise<EntryRecord[]> {
+): Promise<readonly EntryRecord[]> {
 	if (bounds === undefined) return [];
-	return selectActive(bounds.head, await scanRange(storage, conversationId, bounds, context));
+	return selectActive(bounds.head, await scanRange(storage, conversationId, rangeQuery(bounds), context));
 }
 
-/** Visible entries from the head marker's head, or transcript start, through the tail, oldest first. */
+/** The visible range of `bounds`: from the head marker's head, or transcript start, through the tail. */
+function rangeQuery(bounds: ContextBounds): Omit<EntryQuery, "conversationId"> {
+	const head = bounds.head;
+	return head === undefined ? { maxEntryId: bounds.tail } : { minEntryId: head.head, maxEntryId: bounds.tail };
+}
+
+/** Visible entries within `range`, oldest first. */
 async function scanRange(
 	storage: Storage,
 	conversationId: ConversationId,
-	bounds: ContextBounds,
+	range: Omit<EntryQuery, "conversationId">,
 	context: Context,
 ): Promise<EntryRecord[]> {
-	const head = bounds.head;
-	const range: EntryRecord[] = [];
+	const entries: EntryRecord[] = [];
 	let cursor: Cursor | undefined;
 	do {
-		const page = await storage.scanEntries(
-			head === undefined
-				? { conversationId, maxEntryId: bounds.tail }
-				: { conversationId, minEntryId: head.head, maxEntryId: bounds.tail },
-			SCAN_PAGE_SIZE,
-			cursor,
-			context,
-		);
-		range.push(...page.items);
+		const page = await storage.scanEntries({ conversationId, ...range }, SCAN_PAGE_SIZE, cursor, context);
+		entries.push(...page.items);
 		cursor = page.next;
 	} while (cursor !== undefined);
-	return range.reverse();
+	return entries.reverse();
 }
 
 /** The head marker followed by the range's non-head entries, or the whole range without a marker. */
-function selectActive(head: ContextBounds["head"], range: EntryRecord[]): EntryRecord[] {
+function selectActive(head: ContextBounds["head"], range: readonly EntryRecord[]): readonly EntryRecord[] {
 	return head === undefined ? range : [head, ...range.filter((entry) => entry.head === undefined)];
 }
 

packages/durable/src/harness/scheduler.ts

@@ -23,7 +23,7 @@ import type {
 	TaskState,
 } from "../types.ts";
 import { agentHooks } from "./agent.ts";
-import { readContext } from "./context.ts";
+import { type ContextRange, readContextFrom } from "./context.ts";
 import type {
 	Agent,
 	AnyTask,
@@ -75,6 +75,11 @@ type Invocation = {
 	readonly context: Context;
 	/** Watches acquired through the runtime; stopped at invocation end. */
 	readonly watches: Set<DocumentWatch<JsonObject>>;
+	/**
+	 * Range of the invocation's last context read, so the next read, such as a generation's request after its prepare,
+	 * scans only newer entries. Derived and never persisted; dropped at invocation end and before sleeping.
+	 */
+	contextRange: ContextRange | undefined;
 	ended: boolean;
 	readonly done: Promise<void>;
 	readonly finish: () => void;
@@ -825,6 +830,7 @@ export class TaskScheduler {
 			controller,
 			context: withAbortSignal(controller.signal, this.#context),
 			watches: new Set(),
+			contextRange: undefined,
 			ended: false,
 			done,
 			finish: () => finish(),
@@ -1028,6 +1034,7 @@ export class TaskScheduler {
 	#end(invocation: Invocation): void {
 		if (invocation.ended) return;
 		invocation.ended = true;
+		invocation.contextRange = undefined;
 		if (this.#invocations.get(invocation.taskId) === invocation) this.#invocations.delete(invocation.taskId);
 		for (const watch of invocation.watches) void watch.stop();
 		// Pending waits bound to the invocation, such as a tool's waitForTask(), reject with it.
@@ -1168,7 +1175,18 @@ export class TaskScheduler {
 				});
 			}) as ErasedRuntime["entry"],
 			context: (conversationId, context, at) =>
-				this.#read(invocation, () => readContext(this.#session, this.#storage, conversationId, context, at)),
+				this.#read(invocation, async () => {
+					const { view, range } = await readContextFrom(
+						this.#session,
+						this.#storage,
+						conversationId,
+						context,
+						at,
+						invocation.contextRange,
+					);
+					if (!invocation.ended) invocation.contextRange = range;
+					return view;
+				}),
 			now: () => {
 				if (invocation.ended) throw endedError(invocation);
 				return this.#now();
@@ -1214,6 +1232,8 @@ export class TaskScheduler {
 	/** Wait until the Harness clock reaches `until`, rechecking it after every timer. */
 	async #sleep(invocation: Invocation, until: number, context: Context): Promise<void> {
 		if (invocation.ended) throw endedError(invocation);
+		// A sleeping task holds no context range; a retry or poll can wait for a long time.
+		invocation.contextRange = undefined;
 		const signals = [invocation.controller.signal];
 		if (context.abortSignal !== undefined) signals.push(context.abortSignal);
 		const signal = AbortSignal.any(signals);

packages/durable/test/harness-context.test.ts

@@ -1,7 +1,16 @@
 import type { Message } from "@earendil-works/pi-ai";
-import { type EntryDraft, type EntryId, type EntryRecord, MemoryStorage } from "@earendil-works/pi-durable";
+import {
+	type ContextView,
+	createRegistry,
+	defineTask,
+	type EntryDraft,
+	type EntryId,
+	type EntryRecord,
+	MemoryStorage,
+	type Storage,
+} from "@earendil-works/pi-durable";
 import { describe, expect, it } from "vitest";
-import { assistant, describeMessage, openHarness, system, toolResult, user } from "./harness-support.ts";
+import { addTask, assistant, describeMessage, openHarness, system, toolResult, user } from "./harness-support.ts";
 import { context } from "./session-support.ts";
 
 async function setup() {
@@ -132,4 +141,93 @@ describe("conversation context", () => {
 		expect(parentView.messages.map(describeMessage)).toEqual([]);
 		expect(parentView.entries.map((entry) => entry.kind)).toEqual(["reset", "message"]);
 	});
+
+	it("extends a task invocation's context read with only newer entries", async () => {
+		let scanned = 0;
+		const memory = new MemoryStorage();
+		const storage = new Proxy<Storage>(memory, {
+			get(target, key) {
+				const value: unknown = Reflect.get(target, key, target);
+				if (key === "scanEntries") {
+					return async (...args: Parameters<Storage["scanEntries"]>) => {
+						const page = await target.scanEntries(...args);
+						scanned += page.items.length;
+						return page;
+					};
+				}
+				return typeof value === "function" ? value.bind(target) : value;
+			},
+		});
+		const registry = createRegistry();
+		const { harness } = await openHarness(storage, [], { registry });
+		const root = await harness.root(context);
+		const append = (draft: EntryDraft): Promise<EntryRecord> =>
+			root.commit((tx) => tx.appendEntry(root.id, draft), context);
+		const first = await append({ kind: "message", model: [user("first")] });
+		for (let index = 0; index < 20; index++) await append({ kind: "message", model: [assistant(`old ${index}`)] });
+		const Reads = defineTask<Record<string, never>, { phase: "run" }, null>({
+			name: "test.context-reads",
+			version: 1,
+			initial: () => ({ phase: "run" }),
+			phases: {
+				run: async (_task, runtime, taskContext) => {
+					const write = (draft: EntryDraft) =>
+						runtime.commit(async (tx) => {
+							await tx.appendEntry(root.id, draft);
+							return undefined;
+						}, taskContext);
+					const read = async (at?: EntryId): Promise<{ view: ContextView; rows: number }> => {
+						const before = scanned;
+						const view = await runtime.context(root.id, taskContext, at);
+						return { view, rows: scanned - before };
+					};
+					const initial = await read();
+					expect(initial.view).toEqual(await root.context(taskContext));
+
+					// Three new entries: the bounds probe reads one row, the extension the three new ones.
+					await write({ kind: "message", model: [user("new")] });
+					await write({ kind: "note", data: { text: "display only" } });
+					await write({ kind: "message", model: [assistant("answer")] });
+					const extended = await read();
+					expect(extended.rows).toBe(4);
+					expect(extended.view).toEqual(await root.context(taskContext));
+
+					// A newer edit of an older entry applies to the extended range.
+					await write({ kind: "edit", edits: [{ target: first.id, action: "omit" }] });
+					const edited = await read();
+					expect(edited.rows).toBe(2);
+					expect(edited.view).toEqual(await root.context(taskContext));
+					expect(edited.view.messages.map(describeMessage)).not.toContain("user:first");
+
+					// An earlier cutoff reuses the range.
+					const cutoff = await read(extended.view.entries.at(-1)!.id);
+					expect(cutoff.rows).toBe(0);
+					expect(cutoff.view).toEqual(extended.view);
+
+					// A new head marker changes the range: read it whole.
+					await write({ kind: "reset", head: "self", model: [user("fresh")] });
+					const reset = await read();
+					expect(reset.view).toEqual(await root.context(taskContext));
+					expect(reset.view.messages.map(describeMessage)).toEqual(["user:fresh"]);
+
+					await runtime.commit(
+						() => ({ status: "terminal", outcome: { status: "completed", result: null } }),
+						taskContext,
+					);
+				},
+			},
+			abort: async (_task, runtime, taskContext) => {
+				await runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), taskContext);
+			},
+		});
+		addTask(registry, Reads);
+		const id = await harness.commit(
+			(tx) => tx.createTask(Reads, {}, { ownership: { kind: "conversation" }, conversationId: root.id }),
+			context,
+		);
+		harness.resume();
+		const settled = await harness.waitForTask(id, context);
+		expect(settled.state.outcome).toEqual({ status: "completed", result: null });
+		await harness.close(context);
+	});
 });