From 68ccef17608c5964bc91d1c567b9d1fe5b0a590a Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Tue, 6 Oct 2026 16:58:39 +0200 Subject: [PATCH] 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 | 4 + packages/durable/src/harness/context.ts | 89 +++++++++++---- packages/durable/src/harness/scheduler.ts | 24 ++++- packages/durable/test/harness-context.test.ts | 102 +++++++++++++++++- 4 files changed, 194 insertions(+), 25 deletions(-) diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 0fcbae79b..d0b27a67c 100644 --- a/packages/durable/CHANGELOG.md +++ b/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 diff --git a/packages/durable/src/harness/context.ts b/packages/durable/src/harness/context.ts index 8dd584f05..96827b13d 100644 --- a/packages/durable/src/harness/context.ts +++ b/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(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 { - 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(); // 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 { +): Promise { 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 { + 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, context: Context, ): Promise { - 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)]; } diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts index 7882a711b..2c9e9fd0f 100644 --- a/packages/durable/src/harness/scheduler.ts +++ b/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>; + /** + * 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; 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 { 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); diff --git a/packages/durable/test/harness-context.test.ts b/packages/durable/test/harness-context.test.ts index f01b87326..29ff26761 100644 --- a/packages/durable/test/harness-context.test.ts +++ b/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(memory, { + get(target, key) { + const value: unknown = Reflect.get(target, key, target); + if (key === "scanEntries") { + return async (...args: Parameters) => { + 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 => + 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, { 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); + }); });