feat(durable): take the TaskRuntime.context() cutoff as an options object

Matches Conversation.context(context, { at }). refs #10512

packages/durable/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Breaking Changes
+
+- `TaskRuntime.context()` takes the cutoff as an options object, matching `Conversation.context()`: `runtime.context(conversationId, context, { at })` replaces `runtime.context(conversationId, context, at)`.
+
 ### Added
 
 - `ToolExecutionApi.models` and `HookApi.models`: the Harness's `models`, so tools and hooks can resolve models and make requests with the same catalog, credentials, and request transforms as generation ([#10395](https://github.com/earendil-works/pi/issues/10395)).

packages/durable/docs/spec.md

@@ -1698,7 +1698,7 @@ interface TaskRuntime<I, S, R, H extends object> extends DocumentObserver, Docum
   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 active transcript and model context, optionally cut off at `at`. */
-  context(conversationId: ConversationId, context: Context, at?: EntryId): Promise<ContextView>;
+  context(conversationId: ConversationId, context: Context, options?: { readonly at?: EntryId }): Promise<ContextView>;
   /** The Harness clock. */
   now(): number;
   /** Forward a non-fatal failure to `HarnessOptions.onReport`. */

packages/durable/src/harness/compaction.ts

@@ -151,7 +151,7 @@ export const CompactionTask = defineTask<CompactionInput, CompactionCheckpoint,
 			const model = runtime.models.getModel(ref.provider, ref.modelId);
 			if (model === undefined) return failNoModel(runtime, ref, context);
 			// The context at `tail` is immutable, so this is the range `select` chose.
-			const view = await runtime.context(runtime.conversationId, context, tail);
+			const view = await runtime.context(runtime.conversationId, context, { at: tail });
 			const cut = view.entries.findIndex((entry) => entry.id === firstKept);
 			const now = runtime.now();
 			const messages: Message[] = [

packages/durable/src/harness/generation.ts

@@ -192,7 +192,7 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
 			}, context);
 			const model = runtime.models.getModel(ref.provider, ref.modelId);
 			if (model === undefined) return failNoModel(runtime, ref, context);
-			const view = await runtime.context(conversationId, context, cutoff);
+			const view = await runtime.context(conversationId, context, { at: cutoff });
 			let messages = view.messages;
 			await runtime.hooks.each("beforeRequest", async (hook) => {
 				const replaced = await hook({ messages }, runtime, context);
@@ -460,7 +460,7 @@ async function classify(
 	const overflow = message.stopReason === "error" && isContextOverflow(message);
 	if (overflow && compacted === undefined && settings.compaction.enabled) {
 		const policy = settings.compaction;
-		const view = await runtime.context(conversationId, context, cutoff);
+		const view = await runtime.context(conversationId, context, { at: cutoff });
 		if (selectCut(view, policy.keepRecentTokens) !== undefined) {
 			const text = message.errorMessage ?? "Context overflow";
 			await runtime.commit(async (tx): Promise<Next> => {
@@ -550,7 +550,8 @@ async function startToolRound(
 	context: Context,
 ): Promise<void> {
 	const conversationId = runtime.conversationId;
-	const messages = request.messages ?? (await runtime.context(conversationId, context, request.cutoff)).messages;
+	const messages =
+		request.messages ?? (await runtime.context(conversationId, context, { at: request.cutoff })).messages;
 	const offered = new Set(getCurrentTools(messages).map((tool) => tool.name));
 	// Read as the round starts; a tool is resolved as its tool task resolves it.
 	const tools = (await runtime.agent(context)).tools;

packages/durable/src/harness/scheduler.ts

@@ -1174,14 +1174,14 @@ export class TaskScheduler {
 					return token === undefined || entry?.kind === token.kind ? entry : undefined;
 				});
 			}) as ErasedRuntime["entry"],
-			context: (conversationId, context, at) =>
+			context: (conversationId, context, options) =>
 				this.#read(invocation, async () => {
 					const { view, range } = await readContextFrom(
 						this.#session,
 						this.#storage,
 						conversationId,
 						context,
-						at,
+						options?.at,
 						invocation.contextRange,
 					);
 					if (!invocation.ended) invocation.contextRange = range;

packages/durable/src/types.ts

@@ -220,7 +220,7 @@ export interface TaskRuntime<I, S, R, H extends object> extends DocumentObserver
 	/** 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 active transcript and model context, optionally cut off at the visible entry `at`. */
-	context(conversationId: ConversationId, context: Context, at?: EntryId): Promise<ContextView>;
+	context(conversationId: ConversationId, context: Context, options?: { readonly at?: EntryId }): Promise<ContextView>;
 	/** The Harness clock. */
 	now(): number;
 	/** Forward a non-fatal failure to `HarnessOptions.onReport`. */

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

@@ -210,7 +210,7 @@ describe("conversation context", () => {
 						}, taskContext);
 					const read = async (at?: EntryId): Promise<{ view: ContextView; rows: number }> => {
 						const before = scanned;
-						const view = await runtime.context(root.id, taskContext, at);
+						const view = await runtime.context(root.id, taskContext, { at });
 						return { view, rows: scanned - before };
 					};
 					const initial = await read();

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

@@ -544,8 +544,8 @@ describe("task runtime", () => {
 			);
 			seen.push((await runtime.snapshot(Notes, runtime.conversationId, ctx))?.text);
 			seen.push((await runtime.snapshotAsOf(Notes, runtime.conversationId, first!, ctx))?.text);
-			seen.push((await runtime.context(runtime.conversationId, ctx, first)).entries.length);
-			seen.push((await runtime.context(runtime.conversationId, ctx, second)).messages.length);
+			seen.push((await runtime.context(runtime.conversationId, ctx, { at: first })).entries.length);
+			seen.push((await runtime.context(runtime.conversationId, ctx, { at: second })).messages.length);
 			seen.push(runtime.now());
 			runtime.report(new Error("reported"));
 			await runtime.commit(() => completed(null), ctx);