diff --git a/packages/agent/docs/agent-harness.md b/packages/agent/docs/agent-harness.md index c88cf7aa..b3a57377 100644 --- a/packages/agent/docs/agent-harness.md +++ b/packages/agent/docs/agent-harness.md @@ -31,6 +31,7 @@ Harness config is the latest runtime configuration set by the application or ext - tools - active tool names - resources +- stream options - system prompt or system prompt provider Getters return harness config. They do not return the snapshot used by an in-flight provider request. @@ -52,11 +53,15 @@ A turn snapshot is the concrete state used for one LLM turn. It is created by `c - thinking level - all tools - active tools +- stream options +- derived session id Static option values are used directly. System-prompt provider callbacks are invoked once per `createTurnState()` call. All logic for that turn uses the same snapshot. Resource arrays are shallow-copied when a snapshot is created. Individual skill and prompt-template objects are not deep-copied. +Stream options are shallow-copied when a snapshot is created. `headers` and `metadata` maps are shallow-copied; their values are not deep-copied. Credentials from `getApiKeyAndHeaders()` are resolved per provider request so expiring tokens can refresh, but the configured stream options and derived session id come from the current turn snapshot. + ### Session The session contains persisted entries only. Session reads return persisted state and do not include queued writes. @@ -125,9 +130,9 @@ At a save point the harness: 1. flushes pending session writes after the agent-emitted messages for that turn 2. creates a fresh turn snapshot if the low-level loop may continue -3. applies the fresh context/model/thinking-level state before the next provider request +3. applies the fresh context/model/thinking-level/stream-options/session-id state before the next provider request -This lets model, thinking level, tool, resource, and system prompt changes made during a turn affect the next turn in the same run, while never mutating an in-flight provider request. The loop callbacks are not recreated at save points. +This lets model, thinking level, tool, resource, stream option, and system prompt changes made during a turn affect the next turn in the same run, while never mutating an in-flight provider request. The loop callbacks are not recreated at save points. The low-level loop converts harness `ThinkingLevel` to provider `reasoning` at the provider boundary: @@ -188,12 +193,154 @@ Branch summary generation is part of the tree navigation operation. Auto-compaction and retry decision points are not implemented in `AgentHarness` yet. -## Final lifecycle hardening todo +## Implementation todo -Before treating `AgentHarness` as migration-ready, add a broad test suite that exercises listeners and hooks closing over the harness and calling public APIs during every relevant event: +This list tracks the remaining work before treating `AgentHarness` as migration-ready. + +### 1. Finish curated provider/stream configuration + +Implemented so far: + +- `AgentHarnessOptions.streamOptions` provides curated request configuration. +- `getStreamOptions()` returns a shallow copy of current harness config. +- `setStreamOptions()` replaces current harness config. +- Stream options are snapshotted in `createTurnState()` and applied with `applyTurnState()`. +- `headers` and `metadata` maps are shallow-copied when stream options are copied. +- `sessionId` is derived from `session.getMetadata().id` in the turn snapshot. +- The harness installs its own internal stream wrapper and calls `streamSimple()`. +- The wrapper ignores raw incoming provider options except lifecycle-owned fields that must come from the low-level loop: `signal` and `reasoning`. +- Credentials and auth headers from `getApiKeyAndHeaders()` are resolved per provider request. + +Implemented provider hook behavior: + +- `before_provider_request` runs before `streamSimple()` and can patch curated stream options for the current request only. +- `before_provider_payload` maps to the underlying `pi-ai` `onPayload` and can inspect/replace provider-specific payloads. +- `after_provider_response` maps to the underlying `pi-ai` `onResponse` and observes response status/headers before body consumption. +- `AgentHarnessStreamOptionsPatch` has explicit deletion semantics: + - top-level fields present with `undefined` clear that option. + - `headers` and `metadata` patches may set individual keys to `undefined` to delete them. + - `headers: undefined` or `metadata: undefined`, when explicitly present, clears the whole map. +- Current-request stream option merge order is: + 1. snapshotted `streamOptions` + 2. auth headers from `getApiKeyAndHeaders()` + 3. `before_provider_request` patches, in hook registration order +- `before_provider_request` does not patch `reasoning`; add that only if a concrete use case appears. + +Still needed: + +- Add tests proving stream options are snapshotted per turn and changes while busy affect only future provider requests/save-point snapshots. +- Add tests proving provider hook patch/deletion/chaining semantics. + +### 2. Design per-`AgentHarness` model registry + +Not started. + +Still needed: + +- Decide how applications supply the model registry. +- Decide whether the harness stores concrete `Model` objects, model references, or both. +- Validate model selection against the registry. +- Define model change semantics during active turns and save points. +- Preserve current `setModel()` behavior until the registry model is designed. + +### 3. Design generic hook/event extension mechanism + +Current cleanup already done: + +- Removed `AgentHarnessContext`. +- Hooks receive only event payloads. +- `emitHook(event)` derives the hook type from `event.type`. + +Still needed: + +- Define extension context shape. +- Likely expose a harness facade plus a session facade rather than raw internals. +- Decide which public harness APIs are allowed from each hook/event. +- Decide whether hooks can mutate turn snapshots directly or only through explicit hook results/public APIs. +- Clarify event payload semantics versus harness getter semantics. +- Revisit `AgentHarnessOwnEvent` versus `AgentHarnessEvent`. +- Define hook result chaining where it has clean transform semantics: + - `before_provider_request`: each hook receives the stream options produced by previous hooks. + - `before_provider_payload`: each hook receives the payload produced by previous hooks. + - possibly `context`: each hook receives the messages produced by previous hooks. + - possibly `tool_result`: each hook receives the result fields produced by previous hooks. +- Do not chain hooks where semantics are policy-based or ambiguous until explicitly designed, such as `tool_call`, `session_before_compact`, `session_before_tree`, and `before_agent_start`. + +### 4. Add explicit tool registry read/update semantics + +Implemented so far: + +- `setTools(tools, activeToolNames?)` +- `setActiveTools(toolNames)` +- invalid active tool names throw +- generic common app tool shape via `AgentHarness` +- `QueueMode` exported from `Agent` +- `AgentHarnessOptions.steeringMode` / `followUpMode` +- live `steeringMode` / `followUpMode` getters/setters +- queue modes are immediate/live, matching coding-agent behavior + +Still needed: + +- Add `getTools()` semantics. +- Add `getActiveTools()` semantics. +- Decide and implement tool update observability events. +- Include active-tool-only updates in the uniform runtime config observability plan. + +### 5. Full `AgentHarness` lifecycle/state pass + +Implemented so far: + +- Removed constructor `void syncFromTree()`. +- Removed `syncFromTree()`. +- Added `createTurnState()`, `applyTurnState()`, and `executeTurn()`. +- Low-level `AgentLoopConfig.prepareNextTurn` save-point update exists. +- `prepareNextTurn` updates low-level context/model/thinking-level and harness-applied stream/session snapshot state. +- The loop converts `ThinkingLevel` to provider `reasoning` internally. +- `phase` replaces boolean idle. +- Pending session writes are based on session-entry shapes without generated fields. +- Pending session writes flush at save points, settlement, and failure cleanup. +- `steer`, `followUp`, and `nextTurn` accept text plus optional images and create `UserMessage` internally. +- `nextTurn` ordering is fixed: queued messages before the new user message. +- Removed `liveOperationId`. +- Removed `shell()`; use `harness.env`. + +Still needed: + +- Finalize phase/idle semantics. +- Audit whether `settled` can fire too early. +- Make session writes inside `settled` callbacks deterministic. +- Audit follow-up behavior around `agent_end`. +- Implement auto-compaction decision point. +- Implement retry handling. +- Ensure structural operations use consistent `try/finally` phase cleanup. +- Verify `before_agent_start` hook semantics against coding-agent: + - current behavior prepends returned messages. + - decide whether replacement, prepend, append, or transform semantics are correct. +- Decide if `before_agent_start` needs more turn info such as tools/tool snippets. +- Document or change timing for model/thinking/stream-option events that may fire before queued session entries persist while busy. +- Audit `abort()` barrier semantics. + +### 6. Later coding-agent migration plan + +Not started. + +Still needed: + +- Map coding-agent resources to sourced loaders. +- Keep app-level resource dedupe/provenance outside the harness. +- Adapt extension loader to the future hook/session facade. +- Preserve UI/session behavior outside core. +- Move coding-agent stream/auth/retry/header behavior onto the harness stream configuration and provider hooks. + +### 7. Final lifecycle hardening suite + +Before treating `AgentHarness` as migration-ready, add a broad test suite that exercises listeners and hooks closing over the harness and calling public APIs during every relevant event. + +Needs broad tests for: - runtime config setters from low-level lifecycle events and harness events -- resource/tool/model/thinking updates during active turns and save points +- uniform runtime config observability events for model, thinking, resources, tools, active tools, and stream options +- resource/tool/model/thinking/stream-option updates during active turns and save points - session writes from listeners and hooks, including writes from `settled` - queue operations from turn events, tool events, and provider hooks - rejected structural operations while busy diff --git a/packages/agent/src/harness/agent-harness.ts b/packages/agent/src/harness/agent-harness.ts index 33b96b37..879a3599 100644 --- a/packages/agent/src/harness/agent-harness.ts +++ b/packages/agent/src/harness/agent-harness.ts @@ -1,4 +1,10 @@ -import type { AssistantMessage, ImageContent, Model, UserMessage } from "@earendil-works/pi-ai"; +import { + type AssistantMessage, + type ImageContent, + type Model, + streamSimple, + type UserMessage, +} from "@earendil-works/pi-ai"; import { Agent, type QueueMode } from "../agent.js"; import type { AgentEvent, AgentMessage, AgentTool, ThinkingLevel } from "../types.js"; import { collectEntriesForBranchSummary, generateBranchSummary } from "./compaction/branch-summarization.js"; @@ -13,7 +19,8 @@ import type { AgentHarnessOwnEvent, AgentHarnessPhase, AgentHarnessResources, - AgentHarnessTurnState, + AgentHarnessStreamOptions, + AgentHarnessStreamOptionsPatch, ExecutionEnv, NavigateTreeResult, PendingSessionWrite, @@ -28,6 +35,87 @@ function createUserMessage(text: string, images?: ImageContent[]): UserMessage { return { role: "user", content, timestamp: Date.now() }; } +function cloneStreamOptions(streamOptions?: AgentHarnessStreamOptions): AgentHarnessStreamOptions { + return { + ...streamOptions, + headers: streamOptions?.headers ? { ...streamOptions.headers } : undefined, + metadata: streamOptions?.metadata ? { ...streamOptions.metadata } : undefined, + }; +} + +function mergeHeaders(...headers: Array | undefined>): Record | undefined { + const merged: Record = {}; + let hasHeaders = false; + for (const entry of headers) { + if (!entry) continue; + Object.assign(merged, entry); + hasHeaders = true; + } + return hasHeaders ? merged : undefined; +} + +function hasOwn(object: object, key: PropertyKey): boolean { + return Object.hasOwn(object, key); +} + +function applyStreamOptionsPatch( + base: AgentHarnessStreamOptions, + patch?: AgentHarnessStreamOptionsPatch, +): AgentHarnessStreamOptions { + const result = cloneStreamOptions(base); + if (!patch) return result; + + if (hasOwn(patch, "transport")) result.transport = patch.transport; + if (hasOwn(patch, "timeoutMs")) result.timeoutMs = patch.timeoutMs; + if (hasOwn(patch, "maxRetries")) result.maxRetries = patch.maxRetries; + if (hasOwn(patch, "maxRetryDelayMs")) result.maxRetryDelayMs = patch.maxRetryDelayMs; + if (hasOwn(patch, "cacheRetention")) result.cacheRetention = patch.cacheRetention; + + if (hasOwn(patch, "headers")) { + if (patch.headers === undefined) { + result.headers = undefined; + } else { + const headers = { ...(result.headers ?? {}) }; + for (const [key, value] of Object.entries(patch.headers)) { + if (value === undefined) delete headers[key]; + else headers[key] = value; + } + result.headers = Object.keys(headers).length > 0 ? headers : undefined; + } + } + + if (hasOwn(patch, "metadata")) { + if (patch.metadata === undefined) { + result.metadata = undefined; + } else { + const metadata = { ...(result.metadata ?? {}) }; + for (const [key, value] of Object.entries(patch.metadata)) { + if (value === undefined) delete metadata[key]; + else metadata[key] = value; + } + result.metadata = Object.keys(metadata).length > 0 ? metadata : undefined; + } + } + + return result; +} + +interface AgentHarnessTurnState< + TSkill extends Skill = Skill, + TPromptTemplate extends PromptTemplate = PromptTemplate, + TTool extends AgentTool = AgentTool, +> { + messages: AgentMessage[]; + resources: AgentHarnessResources; + streamOptions: AgentHarnessStreamOptions; + sessionId: string; + systemPrompt: string; + model: Model; + thinkingLevel: ThinkingLevel; + tools: TTool[]; + activeTools: TTool[]; +} + export class AgentHarness< TSkill extends Skill = Skill, TPromptTemplate extends PromptTemplate = PromptTemplate, @@ -45,6 +133,9 @@ export class AgentHarness< private followUpQueue: UserMessage[] = []; private pendingSessionWrites: PendingSessionWrite[] = []; private resources: AgentHarnessResources; + private streamOptions: AgentHarnessStreamOptions; + private appliedStreamOptions: AgentHarnessStreamOptions = {}; + private appliedSessionId?: string; private systemPrompt: AgentHarnessOptions["systemPrompt"]; private getApiKeyAndHeaders?: AgentHarnessOptions["getApiKeyAndHeaders"]; private tools = new Map(); @@ -60,12 +151,46 @@ export class AgentHarness< thinkingLevel: options.thinkingLevel, tools: options.tools ?? [], }, + streamFn: async (model, context, streamOptions) => { + const auth = await this.getApiKeyAndHeaders?.(model); + const snapshotOptions: AgentHarnessStreamOptions = { + ...this.appliedStreamOptions, + headers: mergeHeaders(this.appliedStreamOptions.headers, auth?.headers), + }; + const requestOptions = await this.emitBeforeProviderRequest( + model, + this.appliedSessionId ?? "", + snapshotOptions, + ); + return streamSimple(model, context, { + cacheRetention: requestOptions.cacheRetention, + headers: requestOptions.headers, + maxRetries: requestOptions.maxRetries, + maxRetryDelayMs: requestOptions.maxRetryDelayMs, + metadata: requestOptions.metadata, + onPayload: async (payload) => await this.emitBeforeProviderPayload(model, payload), + onResponse: async (response) => { + const headers = { ...(response.headers as Record) }; + await this.emitOwn( + { type: "after_provider_response", status: response.status, headers }, + this.agent.signal, + ); + }, + reasoning: streamOptions?.reasoning, + signal: streamOptions?.signal, + sessionId: this.appliedSessionId, + timeoutMs: requestOptions.timeoutMs, + transport: requestOptions.transport, + apiKey: auth?.apiKey, + }); + }, steeringMode: options.steeringMode, followUpMode: options.followUpMode, }); this.env = options.env; this.session = options.session; this.resources = options.resources ?? {}; + this.streamOptions = cloneStreamOptions(options.streamOptions); this.systemPrompt = options.systemPrompt; this.getApiKeyAndHeaders = options.getApiKeyAndHeaders; for (const tool of options.tools ?? []) { @@ -76,11 +201,6 @@ export class AgentHarness< this.activeToolNames = options.activeToolNames ?? (options.tools ?? []).map((tool) => tool.name); this.agent.state.model = this.model; this.agent.state.thinkingLevel = this.thinkingLevel; - this.agent.getApiKey = async (provider) => { - const model = this.model; - if (!this.getApiKeyAndHeaders || model.provider !== provider) return undefined; - return (await this.getApiKeyAndHeaders(model))?.apiKey; - }; this.agent.transformContext = async (messages) => { const result = await this.emitHook({ type: "context", messages: [...messages] }); return result?.messages ?? messages; @@ -108,14 +228,6 @@ export class AgentHarness< ? { content: patch.content, details: patch.details, isError: patch.isError, terminate: patch.terminate } : undefined; }; - this.agent.onPayload = async (payload) => { - const result = await this.emitHook({ type: "before_provider_request", payload }); - return result?.payload ?? payload; - }; - this.agent.onResponse = async (response) => { - const headers = { ...(response.headers as Record) }; - await this.emitOwn({ type: "after_provider_response", status: response.status, headers }, this.agent.signal); - }; this.agent.prepareNextTurn = async () => { await this.flushPendingSessionWrites(); const turnState = await this.createTurnState(); @@ -162,6 +274,41 @@ export class AgentHarness< return lastResult; } + private async emitBeforeProviderRequest( + model: Model, + sessionId: string, + streamOptions: AgentHarnessStreamOptions, + ): Promise { + const handlers = this.hooks.get("before_provider_request"); + let current = cloneStreamOptions(streamOptions); + if (!handlers || handlers.size === 0) return current; + for (const handler of handlers) { + const result = await handler({ + type: "before_provider_request", + model, + sessionId, + streamOptions: cloneStreamOptions(current), + }); + if (result?.streamOptions) { + current = applyStreamOptionsPatch(current, result.streamOptions); + } + } + return current; + } + + private async emitBeforeProviderPayload(model: Model, payload: unknown): Promise { + const handlers = this.hooks.get("before_provider_payload"); + let current = payload; + if (!handlers || handlers.size === 0) return current; + for (const handler of handlers) { + const result = await handler({ type: "before_provider_payload", model, payload: current }); + if (result !== undefined) { + current = result.payload; + } + } + return current; + } + private async emitQueueUpdate(): Promise { await this.emitOwn({ type: "queue_update", @@ -174,6 +321,7 @@ export class AgentHarness< private async createTurnState(): Promise> { const context = await this.session.buildContext(); const resources = this.getResources(); + const sessionMetadata = await this.session.getMetadata(); const tools = [...this.tools.values()]; const activeTools = this.activeToolNames .map((name) => this.tools.get(name)) @@ -194,6 +342,8 @@ export class AgentHarness< return { messages: context.messages, resources, + streamOptions: cloneStreamOptions(this.streamOptions), + sessionId: sessionMetadata.id, systemPrompt, model: this.model, thinkingLevel: this.thinkingLevel, @@ -204,6 +354,8 @@ export class AgentHarness< private applyTurnState(turnState: AgentHarnessTurnState): void { this.agent.state.messages = turnState.messages; + this.appliedStreamOptions = cloneStreamOptions(turnState.streamOptions); + this.appliedSessionId = turnState.sessionId; this.agent.state.systemPrompt = turnState.systemPrompt; this.agent.state.model = turnState.model; this.agent.state.thinkingLevel = turnState.thinkingLevel; @@ -602,6 +754,14 @@ export class AgentHarness< await this.emitOwn({ type: "resources_update", resources: this.getResources(), previousResources }); } + getStreamOptions(): AgentHarnessStreamOptions { + return cloneStreamOptions(this.streamOptions); + } + + setStreamOptions(streamOptions: AgentHarnessStreamOptions): void { + this.streamOptions = cloneStreamOptions(streamOptions); + } + async setTools(tools: TTool[], activeToolNames?: string[]): Promise { this.tools = new Map(tools.map((tool) => [tool.name, tool])); if (activeToolNames) { diff --git a/packages/agent/src/harness/types.ts b/packages/agent/src/harness/types.ts index 24b5feeb..f50337f8 100644 --- a/packages/agent/src/harness/types.ts +++ b/packages/agent/src/harness/types.ts @@ -1,4 +1,4 @@ -import type { ImageContent, Model, TextContent } from "@earendil-works/pi-ai"; +import type { ImageContent, Model, SimpleStreamOptions, TextContent, Transport } from "@earendil-works/pi-ai"; import type { QueueMode } from "../agent.js"; import type { AgentEvent, AgentMessage, AgentTool, ThinkingLevel } from "../index.js"; import type { Session } from "./session/session.js"; @@ -43,6 +43,33 @@ export interface AgentHarnessResources< skills?: TSkill[]; } +/** Curated provider request options owned by the harness and snapshotted per turn. */ +export interface AgentHarnessStreamOptions { + /** Preferred transport forwarded to the stream function. */ + transport?: Transport; + /** Provider request timeout in milliseconds. */ + timeoutMs?: number; + /** Maximum provider retry attempts. */ + maxRetries?: number; + /** Optional cap for provider-requested retry delays. */ + maxRetryDelayMs?: number; + /** Additional request headers merged with auth and lifecycle headers. */ + headers?: Record; + /** Provider metadata forwarded with requests. */ + metadata?: SimpleStreamOptions["metadata"]; + /** Provider cache retention hint. */ + cacheRetention?: SimpleStreamOptions["cacheRetention"]; +} + +/** Per-request stream option patch returned by provider hooks. */ +export interface AgentHarnessStreamOptionsPatch + extends Omit, "headers" | "metadata"> { + /** Header patch. `undefined` values delete keys; explicit `headers: undefined` clears all headers. */ + headers?: Record; + /** Metadata patch. `undefined` values delete keys; explicit `metadata: undefined` clears all metadata. */ + metadata?: Record; +} + /** Kind of filesystem object as addressed by an {@link ExecutionEnv}. Symlinks are not followed automatically. */ export type FileKind = "file" | "directory" | "symlink"; @@ -298,20 +325,6 @@ export type PendingSessionWrite = SessionTreeEntry extends infer TEntry : never : never; -export interface AgentHarnessTurnState< - TSkill extends Skill = Skill, - TPromptTemplate extends PromptTemplate = PromptTemplate, - TTool extends AgentTool = AgentTool, -> { - messages: AgentMessage[]; - resources: AgentHarnessResources; - systemPrompt: string; - model: Model; - thinkingLevel: ThinkingLevel; - tools: TTool[]; - activeTools: TTool[]; -} - export interface QueueUpdateEvent { type: "queue_update"; steer: AgentMessage[]; @@ -353,6 +366,14 @@ export interface ContextEvent { export interface BeforeProviderRequestEvent { type: "before_provider_request"; + model: Model; + sessionId: string; + streamOptions: AgentHarnessStreamOptions; +} + +export interface BeforeProviderPayloadEvent { + type: "before_provider_payload"; + model: Model; payload: unknown; } @@ -440,6 +461,7 @@ export type AgentHarnessOwnEvent< | BeforeAgentStartEvent | ContextEvent | BeforeProviderRequestEvent + | BeforeProviderPayloadEvent | AfterProviderResponseEvent | ToolCallEvent | ToolResultEvent @@ -465,6 +487,10 @@ export interface ContextResult { } export interface BeforeProviderRequestResult { + streamOptions?: AgentHarnessStreamOptionsPatch; +} + +export interface BeforeProviderPayloadResult { payload: unknown; } @@ -497,6 +523,7 @@ export type AgentHarnessEventResultMap = { before_agent_start: BeforeAgentStartResult | undefined; context: ContextResult | undefined; before_provider_request: BeforeProviderRequestResult | undefined; + before_provider_payload: BeforeProviderPayloadResult | undefined; after_provider_response: undefined; tool_call: ToolCallResult | undefined; tool_result: ToolResultPatch | undefined; @@ -613,6 +640,8 @@ export interface AgentHarnessOptions< getApiKeyAndHeaders?: ( model: Model, ) => Promise<{ apiKey: string; headers?: Record } | undefined>; + /** Curated stream/provider request options. Snapshotted at turn start. */ + streamOptions?: AgentHarnessStreamOptions; model: Model; thinkingLevel?: ThinkingLevel; activeToolNames?: string[];