From 7ec1f648e9b21391ab2a886d3ef8325945660d6d Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Fri, 16 Oct 2026 19:30:35 +0000 Subject: [PATCH] feat(durable): accept functions for HarnessOptions.models and settings Like env, models and settings may be a function returning the current value. Generation, compaction, hooks and tools read it at each use (api.models at each access), so a host can replace its model configuration without reopening the Harness. A plain object still works. --- packages/durable/CHANGELOG.md | 1 + packages/durable/docs/spec.md | 16 ++- packages/durable/src/harness/harness.ts | 23 +++- packages/durable/src/harness/scheduler.ts | 10 +- packages/durable/src/harness/tool.ts | 17 ++- packages/durable/src/harness/types.ts | 14 +- packages/durable/src/types.ts | 1 + .../durable/test/harness-host-inputs.test.ts | 124 ++++++++++++++++++ 8 files changed, 185 insertions(+), 21 deletions(-) create mode 100644 packages/durable/test/harness-host-inputs.test.ts diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index d495f0ca4..4e0d75c1d 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -5,6 +5,7 @@ ### Added - `settings.compaction.modelOverrides` sets compaction thresholds per model, keyed by exact `"provider/modelId"`, so one harness can serve small- and large-window models; the threshold check, overflow recovery, and the compaction task use the entry of the model they act for ([#10942](https://github.com/earendil-works/pi/issues/10942)). +- `HarnessOptions.models` and `HarnessOptions.settings` also accept a function returning the current value, like `env`. Generation, compaction, hooks and tools call it at each use (`api.models` reads the current value at each access), so a host can replace its model configuration or settings without reopening the Harness. A plain object still works. ### Changed diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 74fe6193c..e6747aada 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -359,9 +359,11 @@ type AnyTask = { }; type HarnessOptions = { - readonly models: Models; // the pi-ai Models interface + /** The pi-ai Models interface, or a function returning the current one, called at each use. */ + readonly models: Models | (() => Models); readonly registry: RegistryReader; // section 7.1 - readonly settings?: HarnessSettings; + /** Run settings, or a function returning the current ones; read at each use either way. */ + readonly settings?: HarnessSettings | (() => HarnessSettings | undefined); /** Builds a conversation's environment. Never called on the Session line; may be async. */ readonly env?: ( target: { readonly conversationId: ConversationId; readonly cwd?: string; readonly read: DocumentReader }, @@ -819,8 +821,11 @@ selection, request options, retry, compaction thresholds, tool execution, and queue modes. Nothing copies or stores it. Every reader resolves it anew into `Settings`, each absent field over its built-in default and object fields merged key by key, and uses that value for one decision (section 7.1 lists the -readers). A settings object with getters therefore follows the user's -preferences without a Session write. Settings are synchronous because queue +readers). A settings object with getters, or a `settings` function, therefore +follows the user's preferences without a Session write. `models` works the same +way: a function is called at each use (generation and compaction requests, +`runtime.models`, `api.models` of hooks and tools at each access), so a host can +replace its model configuration without reopening the Harness. Settings are synchronous because queue modes are read on the Session line. `stream` is forwarded to generation requests; `stream.maxRetries` are provider retries inside one request, while `retry` governs durable generation attempts (section 8.3). Settings changes make @@ -1752,6 +1757,7 @@ interface TaskRuntime extends DocumentObserver, Docum agent(context: Context): Promise; /** `HarnessOptions.settings`, resolved at each access. */ readonly settings: Settings; + /** `HarnessOptions.models`, resolved at each access. */ readonly models: Models; readonly hooks: HookRunner; /** Calls `HarnessOptions.env` for the task's conversation; rejects with its error. */ @@ -2719,6 +2725,7 @@ 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; @@ -2858,6 +2865,7 @@ interface ToolExecutionApi extends Docum /** The calling conversation's agent, as the tool task's phase resolved it. */ agent(context: Context): Promise; /** `HarnessOptions.models`: the catalog, credentials, and request transforms generation uses. */ + /** The current `HarnessOptions.models`, read at each access. */ readonly models: Models; /** Built by `HarnessOptions.env` for this call. */ readonly env: ExecutionEnv | undefined; diff --git a/packages/durable/src/harness/harness.ts b/packages/durable/src/harness/harness.ts index ff7a3203f..295645299 100644 --- a/packages/durable/src/harness/harness.ts +++ b/packages/durable/src/harness/harness.ts @@ -1,5 +1,6 @@ import type { AttachedReplicatedState, Context, JsonValue } from "@earendil-works/chord"; import { withAbortSignal, withoutAbortSignal } from "@earendil-works/chord/context"; +import type { Models } from "@earendil-works/pi-ai"; import { ResetEntry } from "../entries.ts"; import type { ExecutionEnv } from "../env/index.ts"; import { contained, SessionImpl } from "../session/session.ts"; @@ -43,6 +44,7 @@ import type { ConversationWatch, HarnessInspection, HarnessOptions, + HarnessSettings, Harness as HarnessType, RegistrySnapshot, SettledTask, @@ -165,9 +167,20 @@ class ConversationImpl implements Conversation { } } +/** `HarnessOptions.models` as a function: a plain `Models` is returned at every call. */ +function currentModels(models: HarnessOptions["models"]): () => Models { + return typeof models === "function" ? models : () => models; +} + +/** `HarnessOptions.settings` as a function: plain settings, or none, are returned at every call. */ +function currentSettings(settings: HarnessOptions["settings"]): () => HarnessSettings | undefined { + return typeof settings === "function" ? settings : () => settings; +} + /** Session kernel extended with conversation handles and a registry. */ class HarnessImpl extends SessionImpl implements HarnessType { readonly #options: HarnessOptions; + readonly #hostSettings: () => HarnessSettings | undefined; readonly #report: (error: unknown) => void; readonly #host: ConversationHost; readonly #tasks: TaskScheduler; @@ -177,18 +190,20 @@ class HarnessImpl extends SessionImpl implements constructor(rawStorage: Storage, options: HarnessOptions, context: Context) { const report = safeReport(options.onReport); const now = safeNow(options.now, report); + const hostSettings = currentSettings(options.settings); // Read at each use, as all settings are, but only this field: it is read on every commit. - super(rawStorage, now, () => options.settings?.documentRetentionMs); + super(rawStorage, now, () => hostSettings()?.documentRetentionMs); // Every component reads and writes through the Session's failure guard: `rawStorage` serves only `super()`. const storage = this.storage; this.#options = options; this.#report = report; - const settings = () => resolveSettings(options.settings); + this.#hostSettings = hostSettings; + const settings = () => resolveSettings(hostSettings()); this.#tasks = new TaskScheduler({ session: this, storage, registry: options.registry, - models: options.models, + models: currentModels(options.models), agent: (id, snapshot, callContext) => this.resolveAgent(id, snapshot as RegistrySnapshot, callContext), settings, env: (id, callContext) => this.buildEnv(id, callContext), @@ -223,7 +238,7 @@ class HarnessImpl extends SessionImpl implements ): Promise> { const registry = snapshot ?? this.#options.registry.snapshot(); const state = await this.snapshot(AgentDoc, id, context); - return resolveAgent(state, registry, resolveSettings(this.#options.settings), this.#report); + return resolveAgent(state, registry, resolveSettings(this.#hostSettings()), this.#report); } /** Build a conversation's environment from its current `cwd`; `undefined` without an `env` option. */ diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts index 6f76f2f3b..2fe49e96d 100644 --- a/packages/durable/src/harness/scheduler.ts +++ b/packages/durable/src/harness/scheduler.ts @@ -152,7 +152,8 @@ export type TaskSchedulerOptions = { readonly session: SessionImpl; readonly storage: Storage; readonly registry: RegistryReader; - readonly models: Models; + /** The host's current model access (`HarnessOptions.models`); read at each access. */ + readonly models: () => Models; /** Resolve a conversation's agent against a snapshot; the runtime calls it at most once per phase. */ readonly agent: (conversationId: ConversationId, snapshot: RegistrySnapshot, context: Context) => Promise; /** Resolve the settings; read at each access. */ @@ -200,7 +201,7 @@ export class TaskScheduler { readonly #session: SessionImpl; readonly #storage: Storage; readonly #registry: RegistryReader; - readonly #models: Models; + readonly #models: () => Models; readonly #agent: TaskSchedulerOptions["agent"]; readonly #settings: TaskSchedulerOptions["settings"]; readonly #env: TaskSchedulerOptions["env"]; @@ -1613,11 +1614,14 @@ export class TaskScheduler { }, }; const settings = this.#settings; + const models = this.#models; return { taskId: invocation.taskId as TaskId, conversationId: invocation.conversationId, signal: invocation.controller.signal, - models: this.#models, + get models() { + return models(); + }, agent: (context) => invocation.ended ? Promise.reject(endedError(invocation)) : awaitWithContext(agent(), context), get settings() { diff --git a/packages/durable/src/harness/tool.ts b/packages/durable/src/harness/tool.ts index 50e13203f..56ff63ed0 100644 --- a/packages/durable/src/harness/tool.ts +++ b/packages/durable/src/harness/tool.ts @@ -13,6 +13,7 @@ import { validateToolArguments } from "@earendil-works/pi-ai/utils/validation"; import { Compile } from "typebox/compile"; import { defineDocFamily } from "../documents.ts"; import { AssistantEntry, ToolResultEntry } from "../entries.ts"; +import type { ExecutionEnv } from "../env/index.ts"; import { errorMessage } from "../errors.ts"; import { defineTask } from "../tasks.ts"; import { DEFAULT_MAX_BYTES, DEFAULT_MAX_LINES, utf8ByteLength } from "../truncate.ts"; @@ -295,13 +296,20 @@ async function run( const assertLive = (): void => { if (ended) throw new Error(`Tool call ${call.id} has settled`); }; - const api: Omit = { + /** Built for this call, below, so a rerun after recovery gets the conversation's environment at that time. */ + let env: ExecutionEnv | undefined; + const api: ToolExecutionApi = { taskId: runtime.taskId, conversationId: runtime.conversationId, callId: call.id, registry: runtime.registry, agent: runtime.agent, - models: runtime.models, + get models() { + return runtime.models; + }, + get env() { + return env; + }, output: (chunk, skipped) => { assertLive(); if (reported.output.push(chunk, skipped)) progress.mark(); @@ -399,11 +407,10 @@ async function run( // Execution time of this attempt; a rerun after recovery measures only itself. let durationMs: number | undefined; try { - // Built for this call, so a rerun after recovery gets the conversation's environment at that time. - const env = await runtime.env(context); + env = await runtime.env(context); const startedAt = performance.now(); try { - result = await tool.execute(args, { ...api, env }, context); + result = await tool.execute(args, api, context); } finally { durationMs = Math.round(performance.now() - startedAt); } diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index 7a12804c0..05a91420b 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -211,7 +211,7 @@ export interface ToolExecutionApi extend readonly registry: RegistrySnapshot; /** The calling conversation's agent, as the tool task's phase resolved it. */ agent(context: Context): Promise; - /** `HarnessOptions.models`: the catalog, credentials, and request transforms generation uses. */ + /** The current `HarnessOptions.models`, read at each access: the catalog, credentials, and request transforms. */ readonly models: Models; /** Built by `HarnessOptions.env` for this call; `undefined` without an environment. */ readonly env: ExecutionEnv | undefined; @@ -584,10 +584,14 @@ export type EnvTarget = { }; export type HarnessOptions = { - /** pi-ai model access used by generation. */ - readonly models: Models; + /** + * pi-ai model access used by generation, compaction, hooks and tools: a `Models`, or a function returning the current + * one, called at each use, so a host can replace its model configuration without reopening the Harness. + */ + readonly models: Models | (() => Models); readonly registry: RegistryReader; - readonly settings?: HarnessSettings; + /** Run settings, or a function returning the current ones; read at each use either way. */ + readonly settings?: HarnessSettings | (() => HarnessSettings | undefined); /** Builds a conversation's environment at each use. Never called on the Session line; may be async. */ readonly env?: (target: EnvTarget, context: Context) => ExecutionEnv | undefined | Promise; /** @@ -755,7 +759,7 @@ export interface Harness extends Session { export 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; diff --git a/packages/durable/src/types.ts b/packages/durable/src/types.ts index 804e49ef1..c4d90d77b 100644 --- a/packages/durable/src/types.ts +++ b/packages/durable/src/types.ts @@ -181,6 +181,7 @@ export interface TaskRuntime extends DocumentObserver agent(context: Context): Promise; /** `HarnessOptions.settings`, resolved at each access. */ readonly settings: Settings; + /** `HarnessOptions.models`, resolved at each access. */ readonly models: Models; /** Calls `HarnessOptions.env` for the task's conversation; rejects with its error. */ env(context: Context): Promise; diff --git a/packages/durable/test/harness-host-inputs.test.ts b/packages/durable/test/harness-host-inputs.test.ts new file mode 100644 index 000000000..b1a06fd98 --- /dev/null +++ b/packages/durable/test/harness-host-inputs.test.ts @@ -0,0 +1,124 @@ +import { + createModels, + fauxAssistantMessage, + fauxProvider, + fauxText, + fauxToolCall, + type Models, + type SimpleStreamOptions, + Type, +} from "@earendil-works/pi-ai"; +import { + type Conversation, + createRegistry, + defineExtension, + defineTool, + Harness, + type HarnessOptions, + type HarnessSettings, + MemoryStorage, + type Registry, +} from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import { allEntries, textOf } from "./chat-support.ts"; +import { context } from "./session-support.ts"; + +/** A `Models` whose only provider is a faux one answering every request with `text`. */ +function answering(text: string): Models { + const faux = fauxProvider(); + faux.setResponses(Array.from({ length: 10 }, () => fauxAssistantMessage([fauxText(text)]))); + const models = createModels(); + models.setProvider(faux.provider); + return models; +} + +async function open( + options: Pick, + registry: Registry = createRegistry(), +): Promise<{ harness: Harness; root: Conversation }> { + const harness = await Harness.open(new MemoryStorage(), { ...options, registry }, context); + const root = await harness.root(context, { agent: { model: { provider: "faux", modelId: "faux-1" } } }); + return { harness, root }; +} + +async function turn(root: Conversation, content: string): Promise { + const settled = await (await root.submit({ type: "input", content }, context)).wait(context); + expect(settled.status).toBe("done"); + const entries = await allEntries(root); + return textOf(entries.at(-1)?.model?.[0]); +} + +describe("HarnessOptions.models and settings as functions", () => { + it("generation uses the current return value of a models function at each request", async () => { + let current = answering("first"); + const { harness, root } = await open({ models: () => current }); + expect(await turn(root, "one")).toBe("first"); + current = answering("second"); + expect(await turn(root, "two")).toBe("second"); + await harness.close(context); + }); + + it("still accepts a plain Models object", async () => { + const { harness, root } = await open({ models: answering("plain") }); + expect(await turn(root, "one")).toBe("plain"); + await harness.close(context); + }); + + it("tools read the current models at each access of api.models", async () => { + const first = createModels(); + const second = createModels(); + const faux = fauxProvider(); + faux.setResponses([ + fauxAssistantMessage([fauxToolCall("probe", {}, { id: "call-1" })], { stopReason: "toolUse" }), + fauxAssistantMessage([fauxText("done")]), + ]); + first.setProvider(faux.provider); + second.setProvider(faux.provider); + let current = first; + const seen: Models[] = []; + const registry = createRegistry(); + registry.install( + defineExtension({ + name: "probe", + tools: [ + defineTool({ + name: "probe", + description: "Records api.models before and after the host swaps it", + parameters: Type.Object({}), + execute: async (_args, api) => { + seen.push(api.models); + current = second; + seen.push(api.models); + return { output: [] }; + }, + }), + ], + }), + ); + const { harness, root } = await open({ models: () => current }, registry); + expect(await turn(root, "go")).toBe("done"); + expect(seen).toEqual([first, second]); + expect(seen[0]).toBe(first); + expect(seen[1]).toBe(second); + await harness.close(context); + }); + + it("generation reads the current return value of a settings function", async () => { + const timeouts: (number | undefined)[] = []; + const faux = fauxProvider(); + const reply = (_request: unknown, options?: SimpleStreamOptions) => { + timeouts.push(options?.timeoutMs); + return fauxAssistantMessage([fauxText("ok")]); + }; + faux.setResponses([reply, reply]); + const models = createModels(); + models.setProvider(faux.provider); + let settings: HarnessSettings = { stream: { timeoutMs: 1000 } }; + const { harness, root } = await open({ models, settings: () => settings }); + await turn(root, "one"); + settings = { stream: { timeoutMs: 2000 } }; + await turn(root, "two"); + expect(timeouts).toEqual([1000, 2000]); + await harness.close(context); + }); +});