From 4f40f62b7b63c13b3a249719394d474ba58d65cb Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Sat, 16 May 2026 00:32:16 +0200 Subject: [PATCH] refactor(agent): harden harness session semantics --- AGENTS.md | 1 + packages/agent/docs/agent-harness.md | 47 +- packages/agent/src/harness/agent-harness.ts | 556 ++++++++++-------- .../compaction/branch-summarization.ts | 167 ++---- .../src/harness/compaction/compaction.ts | 258 ++------ .../agent/src/harness/compaction/utils.ts | 62 +- packages/agent/src/harness/env/nodejs.ts | 154 +++-- .../agent/src/harness/prompt-templates.ts | 64 +- .../agent/src/harness/session/jsonl-repo.ts | 75 ++- .../src/harness/session/jsonl-storage.ts | 155 +++-- .../agent/src/harness/session/memory-repo.ts | 7 +- .../src/harness/session/memory-storage.ts | 51 +- .../agent/src/harness/session/repo-utils.ts | 21 +- packages/agent/src/harness/session/session.ts | 5 +- packages/agent/src/harness/skills.ts | 129 +++- packages/agent/src/harness/types.ts | 107 +++- .../agent/src/harness/utils/shell-output.ts | 16 +- packages/agent/src/index.ts | 3 + .../agent/test/harness/compaction.test.ts | 45 +- .../agent/test/harness/nodejs-env.test.ts | 21 +- packages/agent/test/harness/session.test.ts | 6 +- packages/agent/test/harness/skills.test.ts | 1 + packages/agent/test/harness/storage.test.ts | 34 +- 23 files changed, 1112 insertions(+), 873 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index bf250bc4..e3880791 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -6,6 +6,7 @@ - No emojis in commits, issues, PR comments, or code - No fluff or cheerful filler text - Technical prose only, be kind but direct (e.g., "Thanks @user" not "Thanks so much @user!") +- When the user asks a question, answer it first before making edits or running implementation commands. ## Code Quality diff --git a/packages/agent/docs/agent-harness.md b/packages/agent/docs/agent-harness.md index 3ab1faf0..5a092fa2 100644 --- a/packages/agent/docs/agent-harness.md +++ b/packages/agent/docs/agent-harness.md @@ -21,9 +21,13 @@ A final lifecycle hardening pass should prove these guarantees with a broad list ## Error handling -The target error-handling model is `Result` for fallible harness operations instead of thrown exceptions. Implementations should catch backend/provider/filesystem exceptions at the boundary and normalize them into typed error results; callers should inspect returned results instead of relying on thrown exceptions. +The current split is: -This is currently implemented for `ExecutionEnv`, `NodeExecutionEnv`, shell-output capture, and skill/prompt-template resource loading. Older harness/session/compaction APIs still throw in several paths; finishing that migration is tracked in the cleanup todo. +- low-level capabilities and helpers use `Result` where expected failures are contained and must not throw, such as `ExecutionEnv`, filesystem/shell operations, shell-output capture, resource loading, and compaction helpers +- high-level mutation/orchestration APIs such as `Session` and `AgentHarness` reject/throw instead of returning bare results that can be ignored +- public `AgentHarness` failures are normalized to `AgentHarnessError` where practical; subsystem errors are preserved as `cause` + +Harness events observe committed state. Public mutators validate required input and persistence before committing when practical, then await notifications. If a hook or subscriber fails after commit, the state change is not rolled back and the public method rejects with `AgentHarnessError` code `"hook"`. ## State model @@ -73,6 +77,8 @@ Stream options are shallow-copied when a snapshot is created. `headers` and `met The session contains persisted entries only. Session reads return persisted state and do not include queued writes. +Session storage implementations must persist leaf changes as `leaf` entries. `setLeafId()` is not an in-memory-only cursor update; it appends a durable entry whose `targetId` is the active tree leaf or `null` for root. Reopening storage must reconstruct the current leaf from the latest persisted leaf-affecting entry. + ### Pending session writes Session writes requested while an operation is active are queued as pending session writes. Pending writes are based on session-entry shapes without generated fields (`id`, `parentId`, `timestamp`). @@ -97,7 +103,7 @@ Structural operations require `phase === "idle"` and synchronously set the phase - `compact` - `navigateTree` -Starting another structural operation while the harness is not idle currently throws. This should become a typed result failure as part of the error-handling cleanup. +Starting another structural operation while the harness is not idle rejects with `AgentHarnessError` code `"busy"`. The following operations are allowed during a turn where appropriate: @@ -148,7 +154,7 @@ The low-level loop converts harness `ThinkingLevel` to provider `reasoning` at t No state refresh is needed on `agent_end` except flushing leftover pending session writes and clearing the operation phase. The exact `settled` event timing is still under review. -If the system-prompt callback throws while starting `prompt`, `skill`, or `promptFromTemplate`, the operation currently throws and the harness returns to idle. This should become a typed result failure as part of the error-handling cleanup. If it throws from the save-point snapshot created by `prepareNextTurn`, the low-level agent run records an assistant error message. +If the system-prompt callback throws while starting `prompt`, `skill`, or `promptFromTemplate`, the operation rejects with `AgentHarnessError` and the harness returns to idle. If it throws from the save-point snapshot created by `prepareNextTurn`, the low-level agent run records an assistant error message. ## Hooks and events @@ -331,7 +337,7 @@ Implemented so far: - `setTools(tools, activeToolNames?)` - `setActiveTools(toolNames)` -- invalid active tool names currently throw; convert to result errors +- invalid active tool names reject with `AgentHarnessError` - generic common app tool shape via `AgentHarness` - `QueueMode` exported from core types - `AgentHarnessOptions.steeringMode` / `followUpMode` @@ -363,6 +369,17 @@ Implemented so far: - Removed `liveOperationId`. - Removed `shell()`; use `harness.env`. +Implemented in the hardening pass: + +- Structural compaction/tree operations restore phase with `finally`. +- Public harness failures normalize subsystem causes to `AgentHarnessError`. +- Pending session writes flush one-by-one and are not dropped on failure. +- Queue drains roll back if queue-update notification fails. +- `message_end` persistence happens before subscriber notification. +- `abort()` signals cancellation before notifications and still waits for idle through notification errors. +- Idle model/thinking/tool updates validate and persist before committing in-memory state. +- `setLeafId()` persists durable `leaf` entries so tree navigation survives storage reopen. + Still needed: - Finalize phase/idle semantics. @@ -371,7 +388,6 @@ Still needed: - 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 appends returned messages after the user/next-turn prompt messages. - decide whether replacement, prepend, append, or transform semantics are correct. @@ -379,9 +395,9 @@ Still needed: - Document or change timing for model/thinking/stream-option events that may fire before queued session entries persist while busy. - Audit `abort()` barrier semantics. -### 7. Complete `Result`/non-throwing harness cleanup +### 7. Complete low-level `Result` cleanup -Started. +Current hardening pass complete; future items remain as the API evolves. Implemented so far: @@ -395,17 +411,22 @@ Implemented so far: - Replaced `Buffer` usage in generic truncation utilities with runtime-neutral UTF-8 handling. - Converted compaction summary helpers to typed result returns and added error-path coverage. - Expanded `NodeExecutionEnv` tests for file operations, exec errors, aborts, callbacks, timeouts, and shell-output full-output spill. +- Added `readTextLines()` so JSONL metadata loading reads only the header line instead of whole session files. +- Removed no-op abort handling from Node filesystem methods where cancellation is not meaningful while keeping the `FileSystem` interface unchanged. +- Mapped filesystem errors crossing the session boundary to typed `SessionError`. +- Added typed branch-summary errors and cause-aware public harness error normalization. +- Made resource loaders report structured diagnostics for non-`not_found` filesystem failures. -Still needed: +Ongoing guardrails: -- Remove remaining throws from `src/harness` APIs and helpers, except explicit adapter/test helpers such as `getOrThrow()`. -- Convert session storage/repo/session APIs to typed result returns. -- Convert structural `AgentHarness` operations to typed result returns for busy, missing-resource, auth, compaction, and branch-summary failures. +- Keep low-level capability/helper APIs non-throwing where they return `Result`. +- Keep session storage/repo/session APIs throwing typed `SessionError`. +- Keep structural `AgentHarness` operations rejecting with `AgentHarnessError` for busy, missing-resource, auth, compaction, and branch-summary failures. - Keep Node-specific APIs isolated under `src/harness/env/nodejs.ts` and Node-backed storage/session implementations, or move those implementations behind explicit Node-only entry points. - Audit remaining generic harness utilities for Node globals as new APIs are added. - Audit package exports so browser/generic-JS imports do not pull Node-only modules such as `NodeExecutionEnv`. - Keep expanding `ExecutionEnv` and shell-output contract tests as the API evolves, especially for non-Node implementations. -- Add tests proving harness APIs return `ok: false` instead of throwing for expected failure paths. +- Add tests proving public harness failures reject with `AgentHarnessError` where expected. ### 8. Later coding-agent migration plan diff --git a/packages/agent/src/harness/agent-harness.ts b/packages/agent/src/harness/agent-harness.ts index 5b78316e..6ab5dbef 100644 --- a/packages/agent/src/harness/agent-harness.ts +++ b/packages/agent/src/harness/agent-harness.ts @@ -38,6 +38,7 @@ import type { Session, Skill, } from "./types.js"; +import { AgentHarnessError, BranchSummaryError, CompactionError, SessionError, toError } from "./types.js"; function createUserMessage(text: string, images?: ImageContent[]): UserMessage { const content: Array<{ type: "text"; text: string } | ImageContent> = [{ type: "text", text }]; @@ -131,6 +132,19 @@ const SUBSCRIBER_EVENT_TYPE = "*"; type AgentHarnessHandler = (event: any, signal?: AbortSignal) => Promise | any; +function normalizeHarnessError(error: unknown, fallbackCode: AgentHarnessError["code"]): AgentHarnessError { + if (error instanceof AgentHarnessError) return error; + const cause = toError(error); + if (cause instanceof SessionError) return new AgentHarnessError("session", cause.message, cause); + if (cause instanceof CompactionError) return new AgentHarnessError("compaction", cause.message, cause); + if (cause instanceof BranchSummaryError) return new AgentHarnessError("branch_summary", cause.message, cause); + return new AgentHarnessError(fallbackCode, cause.message, cause); +} + +function normalizeHookError(error: unknown): AgentHarnessError { + return normalizeHarnessError(error, "hook"); +} + interface AgentHarnessTurnState< TSkill extends Skill = Skill, TPromptTemplate extends PromptTemplate = PromptTemplate, @@ -196,13 +210,21 @@ export class AgentHarness< private async emitOwn(event: AgentHarnessOwnEvent, signal?: AbortSignal): Promise { for (const listener of this.getHandlers(SUBSCRIBER_EVENT_TYPE) ?? []) { - await listener(event, signal); + try { + await listener(event, signal); + } catch (error) { + throw normalizeHookError(error); + } } } private async emitAny(event: AgentHarnessEvent, signal?: AbortSignal): Promise { for (const listener of this.getHandlers(SUBSCRIBER_EVENT_TYPE) ?? []) { - await listener(event, signal); + try { + await listener(event, signal); + } catch (error) { + throw normalizeHookError(error); + } } } @@ -213,9 +235,13 @@ export class AgentHarness< if (!handlers || handlers.size === 0) return undefined; let lastResult: AgentHarnessEventResultMap[TType] | undefined; for (const handler of handlers) { - const result = await handler(event); - if (result !== undefined) { - lastResult = result; + try { + const result = await handler(event); + if (result !== undefined) { + lastResult = result; + } + } catch (error) { + throw normalizeHookError(error); } } return lastResult; @@ -230,14 +256,18 @@ export class AgentHarness< 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); + try { + const result = await handler({ + type: "before_provider_request", + model, + sessionId, + streamOptions: cloneStreamOptions(current), + }); + if (result?.streamOptions) { + current = applyStreamOptionsPatch(current, result.streamOptions); + } + } catch (error) { + throw normalizeHookError(error); } } return current; @@ -248,9 +278,13 @@ export class AgentHarness< 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; + try { + const result = await handler({ type: "before_provider_payload", model, payload: current }); + if (result !== undefined) { + current = result.payload; + } + } catch (error) { + throw normalizeHookError(error); } } return current; @@ -356,8 +390,14 @@ export class AgentHarness< private async drainQueuedMessages(queue: AgentMessage[], mode: QueueMode): Promise { const messages = mode === "all" ? queue.splice(0) : queue.splice(0, 1); - if (messages.length > 0) await this.emitQueueUpdate(); - return messages; + if (messages.length === 0) return messages; + try { + await this.emitQueueUpdate(); + return messages; + } catch (error) { + queue.unshift(...messages); + throw normalizeHookError(error); + } } private createLoopConfig( @@ -411,15 +451,14 @@ export class AgentHarness< }; } - private validateToolNames(toolNames: string[]): void { - const missing = toolNames.filter((name) => !this.tools.has(name)); - if (missing.length > 0) throw new Error(`Unknown tool(s): ${missing.join(", ")}`); + private validateToolNames(toolNames: string[], tools: Map = this.tools): void { + const missing = toolNames.filter((name) => !tools.has(name)); + if (missing.length > 0) throw new AgentHarnessError("invalid_argument", `Unknown tool(s): ${missing.join(", ")}`); } private async flushPendingSessionWrites(): Promise { - const writes = this.pendingSessionWrites; - this.pendingSessionWrites = []; - for (const write of writes) { + while (this.pendingSessionWrites.length > 0) { + const write = this.pendingSessionWrites[0]!; if (write.type === "message") { await this.session.appendMessage(write.message); } else if (write.type === "model_change") { @@ -434,28 +473,40 @@ export class AgentHarness< await this.session.appendLabel(write.targetId, write.label); } else if (write.type === "session_info") { await this.session.appendSessionName(write.name ?? ""); + } else if (write.type === "leaf") { + await this.session.getStorage().setLeafId(write.targetId); } + this.pendingSessionWrites.shift(); } } private async handleAgentEvent(event: AgentEvent, signal?: AbortSignal): Promise { - await this.emitAny(event, signal); if (event.type === "message_end") { await this.session.appendMessage(event.message); + await this.emitAny(event, signal); + return; } if (event.type === "turn_end") { + let eventError: unknown; + try { + await this.emitAny(event, signal); + } catch (error) { + eventError = error; + } const hadPendingMutations = this.pendingSessionWrites.length > 0; await this.flushPendingSessionWrites(); - await this.emitOwn({ - type: "save_point", - hadPendingMutations, - }); + if (eventError) throw eventError; + await this.emitOwn({ type: "save_point", hadPendingMutations }); + return; } if (event.type === "agent_end") { await this.flushPendingSessionWrites(); this.phase = "idle"; + await this.emitAny(event, signal); await this.emitOwn({ type: "settled", nextTurnCount: this.nextTurnQueue.length }, signal); + return; } + await this.emitAny(event, signal); } private async emitRunFailure( @@ -480,9 +531,14 @@ export class AgentHarness< let activeTurnState = turnState; let messages: AgentMessage[] = [createUserMessage(text, options?.images)]; if (this.nextTurnQueue.length > 0) { - messages = [...this.nextTurnQueue, messages[0]!]; - this.nextTurnQueue = []; - await this.emitQueueUpdate(); + const queuedMessages = this.nextTurnQueue.splice(0); + try { + await this.emitQueueUpdate(); + } catch (error) { + this.nextTurnQueue.unshift(...queuedMessages); + throw normalizeHookError(error); + } + messages = [...queuedMessages, messages[0]!]; } const beforeResult = await this.emitHook({ type: "before_agent_start", @@ -510,12 +566,20 @@ export class AgentHarness< this.createStreamFn(getTurnState), ); } catch (error) { - return await this.emitRunFailure( - activeTurnState.model, - error, - abortController.signal.aborted, - abortController.signal, - ); + try { + return await this.emitRunFailure( + activeTurnState.model, + error, + abortController.signal.aborted, + abortController.signal, + ); + } catch (failureError) { + const cause = new AggregateError( + [toError(error), toError(failureError)], + "Agent run failed and failure reporting failed", + ); + throw new AgentHarnessError("unknown", cause.message, cause); + } } })(); try { @@ -526,7 +590,7 @@ export class AgentHarness< return message; } } - throw new Error("AgentHarness prompt completed without an assistant message"); + throw new AgentHarnessError("invalid_state", "AgentHarness prompt completed without an assistant message"); } finally { try { await this.flushPendingSessionWrites(); @@ -537,7 +601,7 @@ export class AgentHarness< } async prompt(text: string, options?: { images?: ImageContent[] }): Promise { - if (this.phase !== "idle") throw new Error("AgentHarness is busy"); + if (this.phase !== "idle") throw new AgentHarnessError("busy", "AgentHarness is busy"); this.phase = "turn"; const finishRunPromise = this.startRunPromise(); try { @@ -545,230 +609,229 @@ export class AgentHarness< return await this.executeTurn(turnState, text, options); } catch (error) { this.phase = "idle"; - throw error; + throw normalizeHarnessError(error, "unknown"); } finally { finishRunPromise(); } } async skill(name: string, additionalInstructions?: string): Promise { - if (this.phase !== "idle") throw new Error("AgentHarness is busy"); + if (this.phase !== "idle") throw new AgentHarnessError("busy", "AgentHarness is busy"); this.phase = "turn"; const finishRunPromise = this.startRunPromise(); try { const turnState = await this.createTurnState(); const skill = (turnState.resources.skills ?? []).find((candidate) => candidate.name === name); - if (!skill) throw new Error(`Unknown skill: ${name}`); + if (!skill) throw new AgentHarnessError("invalid_argument", `Unknown skill: ${name}`); return await this.executeTurn(turnState, formatSkillInvocation(skill, additionalInstructions)); } catch (error) { this.phase = "idle"; - throw error; + throw normalizeHarnessError(error, "unknown"); } finally { finishRunPromise(); } } async promptFromTemplate(name: string, args: string[] = []): Promise { - if (this.phase !== "idle") throw new Error("AgentHarness is busy"); + if (this.phase !== "idle") throw new AgentHarnessError("busy", "AgentHarness is busy"); this.phase = "turn"; const finishRunPromise = this.startRunPromise(); try { const turnState = await this.createTurnState(); const template = (turnState.resources.promptTemplates ?? []).find((candidate) => candidate.name === name); - if (!template) throw new Error(`Unknown prompt template: ${name}`); + if (!template) throw new AgentHarnessError("invalid_argument", `Unknown prompt template: ${name}`); return await this.executeTurn(turnState, formatPromptTemplateInvocation(template, args)); } catch (error) { this.phase = "idle"; - throw error; + throw normalizeHarnessError(error, "unknown"); } finally { finishRunPromise(); } } - steer(text: string, options?: { images?: ImageContent[] }): void { - if (this.phase === "idle") throw new Error("Cannot steer while idle"); + async steer(text: string, options?: { images?: ImageContent[] }): Promise { + if (this.phase === "idle") throw new AgentHarnessError("invalid_state", "Cannot steer while idle"); this.steerQueue.push(createUserMessage(text, options?.images)); - void this.emitQueueUpdate(); + await this.emitQueueUpdate(); } - followUp(text: string, options?: { images?: ImageContent[] }): void { - if (this.phase === "idle") throw new Error("Cannot follow up while idle"); + async followUp(text: string, options?: { images?: ImageContent[] }): Promise { + if (this.phase === "idle") throw new AgentHarnessError("invalid_state", "Cannot follow up while idle"); this.followUpQueue.push(createUserMessage(text, options?.images)); - void this.emitQueueUpdate(); + await this.emitQueueUpdate(); } - nextTurn(text: string, options?: { images?: ImageContent[] }): void { + async nextTurn(text: string, options?: { images?: ImageContent[] }): Promise { this.nextTurnQueue.push(createUserMessage(text, options?.images)); - void this.emitQueueUpdate(); + await this.emitQueueUpdate(); } async appendMessage(message: AgentMessage): Promise { - if (this.phase === "idle") { - await this.session.appendMessage(message); - } else { - this.pendingSessionWrites.push({ type: "message", message }); + try { + if (this.phase === "idle") { + await this.session.appendMessage(message); + } else { + this.pendingSessionWrites.push({ type: "message", message }); + } + } catch (error) { + throw normalizeHarnessError(error, "session"); } } async compact( customInstructions?: string, ): Promise<{ summary: string; firstKeptEntryId: string; tokensBefore: number; details?: unknown }> { - if (this.phase !== "idle") throw new Error("compact() requires idle harness"); + if (this.phase !== "idle") throw new AgentHarnessError("busy", "compact() requires idle harness"); this.phase = "compaction"; - const model = this.model; - if (!model) throw new Error("No model set for compaction"); - const auth = await this.getApiKeyAndHeaders?.(model); - if (!auth) throw new Error("No auth available for compaction"); - const branchEntries = await this.session.getBranch(); - const preparation = prepareCompaction(branchEntries, DEFAULT_COMPACTION_SETTINGS); - if (!preparation) throw new Error("Nothing to compact"); - const hookResult = await this.emitHook({ - type: "session_before_compact", - preparation, - branchEntries, - customInstructions, - signal: new AbortController().signal, - }); - if (hookResult?.cancel) { + try { + const model = this.model; + if (!model) throw new AgentHarnessError("invalid_state", "No model set for compaction"); + const auth = await this.getApiKeyAndHeaders?.(model); + if (!auth) throw new AgentHarnessError("auth", "No auth available for compaction"); + const branchEntries = await this.session.getBranch(); + const preparationResult = prepareCompaction(branchEntries, DEFAULT_COMPACTION_SETTINGS); + if (!preparationResult.ok) throw preparationResult.error; + const preparation = preparationResult.value; + if (!preparation) throw new AgentHarnessError("compaction", "Nothing to compact"); + const hookResult = await this.emitHook({ + type: "session_before_compact", + preparation, + branchEntries, + customInstructions, + signal: new AbortController().signal, + }); + if (hookResult?.cancel) throw new AgentHarnessError("compaction", "Compaction cancelled"); + const provided = hookResult?.compaction; + const compactResult = provided + ? { ok: true as const, value: provided } + : await compact( + preparation, + model, + auth.apiKey, + auth.headers, + customInstructions, + undefined, + this.thinkingLevel, + ); + if (!compactResult.ok) throw compactResult.error; + const result = compactResult.value; + const entryId = await this.session.appendCompaction( + result.summary, + result.firstKeptEntryId, + result.tokensBefore, + result.details, + provided !== undefined, + ); + const entry = await this.session.getEntry(entryId); + if (entry?.type === "compaction") { + await this.emitOwn({ type: "session_compact", compactionEntry: entry, fromHook: provided !== undefined }); + } + return result; + } catch (error) { + throw normalizeHarnessError(error, "compaction"); + } finally { this.phase = "idle"; - throw new Error("Compaction cancelled"); } - const provided = hookResult?.compaction; - const compactResult = provided - ? { ok: true as const, value: provided } - : await compact( - preparation, - model, - auth.apiKey, - auth.headers, - customInstructions, - undefined, - this.thinkingLevel, - ); - if (!compactResult.ok) throw compactResult.error; - const result = compactResult.value; - const entryId = await this.session.appendCompaction( - result.summary, - result.firstKeptEntryId, - result.tokensBefore, - result.details, - provided !== undefined, - ); - const entry = await this.session.getEntry(entryId); - if (entry?.type === "compaction") { - await this.emitOwn({ type: "session_compact", compactionEntry: entry, fromHook: provided !== undefined }); - } - this.phase = "idle"; - return result; } async navigateTree( targetId: string, options?: { summarize?: boolean; customInstructions?: string; replaceInstructions?: boolean; label?: string }, ): Promise { - if (this.phase !== "idle") throw new Error("navigateTree() requires idle harness"); + if (this.phase !== "idle") throw new AgentHarnessError("busy", "navigateTree() requires idle harness"); this.phase = "branch_summary"; - const oldLeafId = await this.session.getLeafId(); - if (oldLeafId === targetId) { - this.phase = "idle"; - return { cancelled: false }; - } - const targetEntry = await this.session.getEntry(targetId); - if (!targetEntry) throw new Error(`Entry ${targetId} not found`); - const { entries, commonAncestorId } = await collectEntriesForBranchSummary(this.session, oldLeafId, targetId); - const preparation = { - targetId, - oldLeafId, - commonAncestorId, - entriesToSummarize: entries, - userWantsSummary: options?.summarize ?? false, - customInstructions: options?.customInstructions, - replaceInstructions: options?.replaceInstructions, - label: options?.label, - }; - const signal = new AbortController().signal; - const hookResult = await this.emitHook({ - type: "session_before_tree", - preparation, - signal, - }); - if (hookResult?.cancel) { - this.phase = "idle"; - return { cancelled: true }; - } - let summaryEntry: any | undefined; - let summaryText: string | undefined = hookResult?.summary?.summary; - let summaryDetails: unknown = hookResult?.summary?.details; - if (!summaryText && options?.summarize && entries.length > 0) { - const model = this.model; - if (!model) throw new Error("No model set for branch summary"); - const auth = await this.getApiKeyAndHeaders?.(model); - if (!auth) throw new Error("No auth available for branch summary"); - const branchSummary = await generateBranchSummary(entries, { - model, - apiKey: auth.apiKey, - headers: auth.headers, - signal: new AbortController().signal, - customInstructions: hookResult?.customInstructions ?? options?.customInstructions, - replaceInstructions: hookResult?.replaceInstructions ?? options?.replaceInstructions, - }); - if (branchSummary.aborted) { - this.phase = "idle"; - return { cancelled: true }; - } - if (branchSummary.error) throw new Error(branchSummary.error); - summaryText = branchSummary.summary; - summaryDetails = { - readFiles: branchSummary.readFiles ?? [], - modifiedFiles: branchSummary.modifiedFiles ?? [], + try { + const oldLeafId = await this.session.getLeafId(); + if (oldLeafId === targetId) return { cancelled: false }; + const targetEntry = await this.session.getEntry(targetId); + if (!targetEntry) throw new AgentHarnessError("invalid_argument", `Entry ${targetId} not found`); + const { entries, commonAncestorId } = await collectEntriesForBranchSummary(this.session, oldLeafId, targetId); + const preparation = { + targetId, + oldLeafId, + commonAncestorId, + entriesToSummarize: entries, + userWantsSummary: options?.summarize ?? false, + customInstructions: options?.customInstructions, + replaceInstructions: options?.replaceInstructions, + label: options?.label, }; + const signal = new AbortController().signal; + const hookResult = await this.emitHook({ type: "session_before_tree", preparation, signal }); + if (hookResult?.cancel) return { cancelled: true }; + let summaryEntry: NavigateTreeResult["summaryEntry"]; + let summaryText: string | undefined = hookResult?.summary?.summary; + let summaryDetails: unknown = hookResult?.summary?.details; + if (!summaryText && options?.summarize && entries.length > 0) { + const model = this.model; + if (!model) throw new AgentHarnessError("invalid_state", "No model set for branch summary"); + const auth = await this.getApiKeyAndHeaders?.(model); + if (!auth) throw new AgentHarnessError("auth", "No auth available for branch summary"); + const branchSummary = await generateBranchSummary(entries, { + model, + apiKey: auth.apiKey, + headers: auth.headers, + signal: new AbortController().signal, + customInstructions: hookResult?.customInstructions ?? options?.customInstructions, + replaceInstructions: hookResult?.replaceInstructions ?? options?.replaceInstructions, + }); + if (!branchSummary.ok) { + if (branchSummary.error.code === "aborted") return { cancelled: true }; + throw new AgentHarnessError("branch_summary", branchSummary.error.message, branchSummary.error); + } + summaryText = branchSummary.value.summary; + summaryDetails = { + readFiles: branchSummary.value.readFiles, + modifiedFiles: branchSummary.value.modifiedFiles, + }; + } + let editorText: string | undefined; + let newLeafId: string | null; + if (targetEntry.type === "message" && targetEntry.message.role === "user") { + newLeafId = targetEntry.parentId; + const content = targetEntry.message.content; + editorText = + typeof content === "string" + ? content + : content + .filter((c): c is { readonly type: "text"; readonly text: string } => c.type === "text") + .map((c) => c.text) + .join(""); + } else if (targetEntry.type === "custom_message") { + newLeafId = targetEntry.parentId; + editorText = + typeof targetEntry.content === "string" + ? targetEntry.content + : targetEntry.content + .filter((c): c is { readonly type: "text"; readonly text: string } => c.type === "text") + .map((c) => c.text) + .join(""); + } else { + newLeafId = targetId; + } + const summaryId = await this.session.moveTo( + newLeafId, + summaryText + ? { summary: summaryText, details: summaryDetails, fromHook: hookResult?.summary !== undefined } + : undefined, + ); + if (summaryId) { + const entry = await this.session.getEntry(summaryId); + if (entry?.type === "branch_summary") summaryEntry = entry; + } + await this.emitOwn({ + type: "session_tree", + newLeafId: await this.session.getLeafId(), + oldLeafId, + summaryEntry, + fromHook: hookResult?.summary !== undefined, + }); + return { cancelled: false, editorText, summaryEntry }; + } catch (error) { + throw normalizeHarnessError(error, "branch_summary"); + } finally { + this.phase = "idle"; } - let editorText: string | undefined; - let newLeafId: string | null; - if (targetEntry.type === "message" && targetEntry.message.role === "user") { - newLeafId = targetEntry.parentId; - const content = targetEntry.message.content; - editorText = - typeof content === "string" - ? content - : content - .filter((c): c is { readonly type: "text"; readonly text: string } => c.type === "text") - .map((c) => c.text) - .join(""); - } else if (targetEntry.type === "custom_message") { - newLeafId = targetEntry.parentId; - editorText = - typeof targetEntry.content === "string" - ? targetEntry.content - : targetEntry.content - .filter((c): c is { readonly type: "text"; readonly text: string } => c.type === "text") - .map((c) => c.text) - .join(""); - } else { - newLeafId = targetId; - } - const summaryId = await this.session.moveTo( - newLeafId, - summaryText - ? { - summary: summaryText, - details: summaryDetails, - fromHook: hookResult?.summary !== undefined, - } - : undefined, - ); - if (summaryId) { - summaryEntry = await this.session.getEntry(summaryId); - } - await this.emitOwn({ - type: "session_tree", - newLeafId: await this.session.getLeafId(), - oldLeafId, - summaryEntry, - fromHook: hookResult?.summary !== undefined, - }); - this.phase = "idle"; - return { cancelled: false, editorText, summaryEntry }; } getModel(): Model { @@ -780,37 +843,49 @@ export class AgentHarness< } async setModel(model: Model): Promise { - const previousModel = this.model; - this.model = model; - if (this.phase === "idle") { - await this.session.appendModelChange(model.provider, model.id); - } else { - this.pendingSessionWrites.push({ type: "model_change", provider: model.provider, modelId: model.id }); + try { + const previousModel = this.model; + if (this.phase === "idle") { + await this.session.appendModelChange(model.provider, model.id); + } else { + this.pendingSessionWrites.push({ type: "model_change", provider: model.provider, modelId: model.id }); + } + this.model = model; + await this.emitOwn({ type: "model_select", model, previousModel, source: "set" }); + } catch (error) { + throw normalizeHarnessError(error, "session"); } - await this.emitOwn({ type: "model_select", model, previousModel, source: "set" }); } async setThinkingLevel(level: ThinkingLevel): Promise { - const previousLevel = this.thinkingLevel; - this.thinkingLevel = level; - if (this.phase === "idle") { - await this.session.appendThinkingLevelChange(level); - } else { - this.pendingSessionWrites.push({ type: "thinking_level_change", thinkingLevel: level }); + try { + const previousLevel = this.thinkingLevel; + if (this.phase === "idle") { + await this.session.appendThinkingLevelChange(level); + } else { + this.pendingSessionWrites.push({ type: "thinking_level_change", thinkingLevel: level }); + } + this.thinkingLevel = level; + await this.emitOwn({ type: "thinking_level_select", level, previousLevel }); + } catch (error) { + throw normalizeHarnessError(error, "session"); } - await this.emitOwn({ type: "thinking_level_select", level, previousLevel }); } async setActiveTools(toolNames: string[]): Promise { - this.validateToolNames(toolNames); - this.activeToolNames = [...toolNames]; + try { + this.validateToolNames(toolNames); + this.activeToolNames = [...toolNames]; + } catch (error) { + throw normalizeHarnessError(error, "invalid_argument"); + } } getSteeringMode(): QueueMode { return this.steeringQueueMode; } - setSteeringMode(mode: QueueMode): void { + async setSteeringMode(mode: QueueMode): Promise { this.steeringQueueMode = mode; } @@ -818,7 +893,7 @@ export class AgentHarness< return this.followUpQueueMode; } - setFollowUpMode(mode: QueueMode): void { + async setFollowUpMode(mode: QueueMode): Promise { this.followUpQueueMode = mode; } @@ -842,17 +917,19 @@ export class AgentHarness< return cloneStreamOptions(this.streamOptions); } - setStreamOptions(streamOptions: AgentHarnessStreamOptions): void { + async setStreamOptions(streamOptions: AgentHarnessStreamOptions): Promise { this.streamOptions = cloneStreamOptions(streamOptions); } async setTools(tools: TTool[], activeToolNames?: string[]): Promise { - this.tools = new Map(tools.map((tool) => [tool.name, tool])); - if (activeToolNames) { - this.validateToolNames(activeToolNames); - this.activeToolNames = [...activeToolNames]; - } else { - this.validateToolNames(this.activeToolNames); + try { + const nextTools = new Map(tools.map((tool) => [tool.name, tool])); + const nextActiveToolNames = activeToolNames ? [...activeToolNames] : this.activeToolNames; + this.validateToolNames(nextActiveToolNames, nextTools); + this.tools = nextTools; + this.activeToolNames = [...nextActiveToolNames]; + } catch (error) { + throw normalizeHarnessError(error, "invalid_argument"); } } @@ -861,10 +938,27 @@ export class AgentHarness< const clearedFollowUp = [...this.followUpQueue]; this.steerQueue = []; this.followUpQueue = []; - await this.emitQueueUpdate(); this.runAbortController?.abort(); - await this.waitForIdle(); - await this.emitOwn({ type: "abort", clearedSteer, clearedFollowUp }); + const errors: Error[] = []; + try { + await this.emitQueueUpdate(); + } catch (error) { + errors.push(toError(error)); + } + try { + await this.waitForIdle(); + } catch (error) { + errors.push(toError(error)); + } + try { + await this.emitOwn({ type: "abort", clearedSteer, clearedFollowUp }); + } catch (error) { + errors.push(toError(error)); + } + if (errors.length > 0) { + const cause = errors.length === 1 ? errors[0]! : new AggregateError(errors, "Abort completed with errors"); + throw normalizeHarnessError(cause, "hook"); + } return { clearedSteer, clearedFollowUp }; } diff --git a/packages/agent/src/harness/compaction/branch-summarization.ts b/packages/agent/src/harness/compaction/branch-summarization.ts index 38d3a18e..37af3b15 100644 --- a/packages/agent/src/harness/compaction/branch-summarization.ts +++ b/packages/agent/src/harness/compaction/branch-summarization.ts @@ -1,10 +1,3 @@ -/** - * Branch summarization for tree navigation. - * - * When navigating to a different point in the session tree, this generates - * a summary of the branch being left so context isn't lost. - */ - import type { Model } from "@earendil-works/pi-ai"; import { completeSimple } from "@earendil-works/pi-ai"; import type { AgentMessage } from "../../types.js"; @@ -14,102 +7,75 @@ import { createCompactionSummaryMessage, createCustomMessage, } from "../messages.js"; -import type { Session, SessionTreeEntry } from "../types.js"; -import { estimateTokens } from "./compaction.js"; +import type { BranchSummaryResult, Session, SessionTreeEntry } from "../types.js"; +import { BranchSummaryError, err, ok, type Result, SessionError } from "../types.js"; +import { estimateTokens, SUMMARIZATION_SYSTEM_PROMPT } from "./compaction.js"; import { computeFileLists, createFileOps, extractFileOpsFromMessage, type FileOperations, formatFileOperations, - SUMMARIZATION_SYSTEM_PROMPT, serializeConversation, } from "./utils.js"; -// ============================================================================ -// Types -// ============================================================================ - -export interface BranchSummaryResult { - summary?: string; - readFiles?: string[]; - modifiedFiles?: string[]; - aborted?: boolean; - error?: string; -} - -/** Details stored in BranchSummaryEntry.details for file tracking */ +/** File-operation details stored on generated branch summary entries. */ export interface BranchSummaryDetails { + /** Files read while exploring the summarized branch. */ readFiles: string[]; + /** Files modified while exploring the summarized branch. */ modifiedFiles: string[]; } export type { FileOperations } from "./utils.js"; +/** Prepared branch content for summarization. */ export interface BranchPreparation { - /** Messages extracted for summarization, in chronological order */ + /** Messages selected for the branch summary. */ messages: AgentMessage[]; - /** File operations extracted from tool calls */ + /** File operations extracted from the branch. */ fileOps: FileOperations; - /** Total estimated tokens in messages */ + /** Estimated token count for selected messages. */ totalTokens: number; } +/** Entries selected for branch summarization. */ export interface CollectEntriesResult { - /** Entries to summarize, in chronological order */ + /** Entries to summarize in chronological order. */ entries: SessionTreeEntry[]; - /** Common ancestor between old and new position, if any */ + /** Deepest common ancestor between the previous leaf and target entry. */ commonAncestorId: string | null; } +/** Options for generating a branch summary. */ export interface GenerateBranchSummaryOptions { - /** Model to use for summarization */ + /** Model used for summarization. */ model: Model; - /** API key for the model */ + /** API key forwarded to the provider. */ apiKey: string; - /** Request headers for the model */ + /** Optional request headers forwarded to the provider. */ headers?: Record; - /** Abort signal for cancellation */ + /** Abort signal for the summarization request. */ signal: AbortSignal; - /** Optional custom instructions for summarization */ + /** Optional instructions appended to or replacing the default prompt. */ customInstructions?: string; - /** If true, customInstructions replaces the default prompt instead of being appended */ + /** Replace the default prompt with custom instructions instead of appending them. */ replaceInstructions?: boolean; - /** Tokens reserved for prompt + LLM response (default 16384) */ + /** Tokens reserved for prompt and model output. Defaults to 16384. */ reserveTokens?: number; } -// ============================================================================ -// Entry Collection -// ============================================================================ - -/** - * Collect entries that should be summarized when navigating from one position to another. - * - * Walks from oldLeafId back to the common ancestor with targetId, collecting entries - * along the way. Does NOT stop at compaction boundaries - those are included and their - * summaries become context. - * - * @param session - Session manager (read-only access) - * @param oldLeafId - Current position (where we're navigating from) - * @param targetId - Target position (where we're navigating to) - * @returns Entries to summarize and the common ancestor - */ +/** Collect entries that should be summarized before navigating to a different session tree entry. */ export async function collectEntriesForBranchSummary( session: Session, oldLeafId: string | null, targetId: string, ): Promise { - // If no old position, nothing to summarize if (!oldLeafId) { return { entries: [], commonAncestorId: null }; } - - // Find common ancestor (deepest node that's on both paths) const oldPath = new Set((await session.getBranch(oldLeafId)).map((e) => e.id)); const targetPath = await session.getBranch(targetId); - - // targetPath is root-first, so iterate backwards to find deepest common ancestor let commonAncestorId: string | null = null; for (let i = targetPath.length - 1; i >= 0; i--) { if (oldPath.has(targetPath[i].id)) { @@ -117,36 +83,22 @@ export async function collectEntriesForBranchSummary( break; } } - - // Collect entries from old leaf back to common ancestor const entries: SessionTreeEntry[] = []; let current: string | null = oldLeafId; while (current && current !== commonAncestorId) { const entry = await session.getEntry(current); - if (!entry) break; + if (!entry) throw new SessionError("invalid_session", `Entry ${current} not found`); entries.push(entry as SessionTreeEntry); current = entry.parentId; } - - // Reverse to get chronological order entries.reverse(); return { entries, commonAncestorId }; } - -// ============================================================================ -// Entry to Message Conversion -// ============================================================================ - -/** - * Extract AgentMessage from a session entry. - * Similar to getMessageFromEntry in compaction.ts but also handles compaction entries. - */ function getMessageFromEntry(entry: SessionTreeEntry): AgentMessage | undefined { switch (entry.type) { case "message": - // Skip tool results - context is in assistant's tool call if (entry.message.role === "toolResult") return undefined; return entry.message; @@ -158,38 +110,21 @@ function getMessageFromEntry(entry: SessionTreeEntry): AgentMessage | undefined case "compaction": return createCompactionSummaryMessage(entry.summary, entry.tokensBefore, entry.timestamp); - - // These don't contribute to conversation content case "thinking_level_change": case "model_change": case "custom": case "label": case "session_info": + case "leaf": return undefined; } } -/** - * Prepare entries for summarization with token budget. - * - * Walks entries from NEWEST to OLDEST, adding messages until we hit the token budget. - * This ensures we keep the most recent context when the branch is too long. - * - * Also collects file operations from: - * - Tool calls in assistant messages - * - Existing branch_summary entries' details (for cumulative tracking) - * - * @param entries - Entries in chronological order - * @param tokenBudget - Maximum tokens to include (0 = no limit) - */ +/** Prepare branch entries for summarization within an optional token budget. */ export function prepareBranchEntries(entries: SessionTreeEntry[], tokenBudget: number = 0): BranchPreparation { const messages: AgentMessage[] = []; const fileOps = createFileOps(); let totalTokens = 0; - - // First pass: collect file ops from ALL entries (even if they don't fit in token budget) - // This ensures we capture cumulative file tracking from nested branch summaries - // Only extract from pi-generated summaries (fromHook !== true), not extension-generated ones for (const entry of entries) { if (entry.type === "branch_summary" && !entry.fromHook && entry.details) { const details = entry.details as BranchSummaryDetails; @@ -197,35 +132,26 @@ export function prepareBranchEntries(entries: SessionTreeEntry[], tokenBudget: n for (const f of details.readFiles) fileOps.read.add(f); } if (Array.isArray(details.modifiedFiles)) { - // Modified files go into both edited and written for proper deduplication for (const f of details.modifiedFiles) { fileOps.edited.add(f); } } } } - - // Second pass: walk from newest to oldest, adding messages until token budget for (let i = entries.length - 1; i >= 0; i--) { const entry = entries[i]; const message = getMessageFromEntry(entry); if (!message) continue; - - // Extract file ops from assistant messages (tool calls) extractFileOpsFromMessage(message, fileOps); const tokens = estimateTokens(message); - - // Check budget before adding if (tokenBudget > 0 && totalTokens + tokens > tokenBudget) { - // If this is a summary entry, try to fit it anyway as it's important context if (entry.type === "compaction" || entry.type === "branch_summary") { if (totalTokens < tokenBudget * 0.9) { messages.unshift(message); totalTokens += tokens; } } - // Stop - we've hit the budget break; } @@ -236,10 +162,6 @@ export function prepareBranchEntries(entries: SessionTreeEntry[], tokenBudget: n return { messages, fileOps, totalTokens }; } -// ============================================================================ -// Summary Generation -// ============================================================================ - const BRANCH_SUMMARY_PREAMBLE = `The user explored a different conversation branch before returning here. Summary of that exploration: @@ -274,34 +196,22 @@ Use this EXACT format: Keep each section concise. Preserve exact file paths, function names, and error messages.`; -/** - * Generate a summary of abandoned branch entries. - * - * @param entries - Session entries to summarize (chronological order) - * @param options - Generation options - */ +/** Generate a summary for abandoned branch entries. */ export async function generateBranchSummary( entries: SessionTreeEntry[], options: GenerateBranchSummaryOptions, -): Promise { +): Promise> { const { model, apiKey, headers, signal, customInstructions, replaceInstructions, reserveTokens = 16384 } = options; - - // Token budget = context window minus reserved space for prompt + response const contextWindow = model.contextWindow || 128000; const tokenBudget = contextWindow - reserveTokens; const { messages, fileOps } = prepareBranchEntries(entries, tokenBudget); if (messages.length === 0) { - return { summary: "No content to summarize" }; + return ok({ summary: "No content to summarize", readFiles: [], modifiedFiles: [] }); } - - // Transform to LLM-compatible messages, then serialize to text - // Serialization prevents the model from treating it as a conversation to continue const llmMessages = convertToLlm(messages); const conversationText = serializeConversation(llmMessages); - - // Build prompt let instructions: string; if (replaceInstructions && customInstructions) { instructions = customInstructions; @@ -319,37 +229,34 @@ export async function generateBranchSummary( timestamp: Date.now(), }, ]; - - // Call LLM for summarization const response = await completeSimple( model, { systemPrompt: SUMMARIZATION_SYSTEM_PROMPT, messages: summarizationMessages }, { apiKey, headers, signal, maxTokens: 2048 }, ); - - // Check if aborted or errored if (response.stopReason === "aborted") { - return { aborted: true }; + return err(new BranchSummaryError("aborted", response.errorMessage || "Branch summary aborted")); } if (response.stopReason === "error") { - return { error: response.errorMessage || "Summarization failed" }; + return err( + new BranchSummaryError( + "summarization_failed", + `Branch summary failed: ${response.errorMessage || "Unknown error"}`, + ), + ); } let summary = response.content .filter((c): c is { type: "text"; text: string } => c.type === "text") .map((c) => c.text) .join("\n"); - - // Prepend preamble to provide context about the branch summary summary = BRANCH_SUMMARY_PREAMBLE + summary; - - // Compute file lists and append to summary const { readFiles, modifiedFiles } = computeFileLists(fileOps); summary += formatFileOperations(readFiles, modifiedFiles); - return { + return ok({ summary: summary || "No summary generated", readFiles, modifiedFiles, - }; + }); } diff --git a/packages/agent/src/harness/compaction/compaction.ts b/packages/agent/src/harness/compaction/compaction.ts index 79a76ef1..a04ccd07 100644 --- a/packages/agent/src/harness/compaction/compaction.ts +++ b/packages/agent/src/harness/compaction/compaction.ts @@ -1,10 +1,3 @@ -/** - * Context compaction for long sessions. - * - * Pure functions for compaction logic. The session manager handles I/O, - * and after compaction the session is reloaded. - */ - import type { AssistantMessage, ImageContent, Model, TextContent, Usage } from "@earendil-works/pi-ai"; import { completeSimple } from "@earendil-works/pi-ai"; import type { AgentMessage, ThinkingLevel } from "../../types.js"; @@ -22,35 +15,33 @@ import { extractFileOpsFromMessage, type FileOperations, formatFileOperations, - SUMMARIZATION_SYSTEM_PROMPT, serializeConversation, } from "./utils.js"; -// ============================================================================ -// File Operation Tracking -// ============================================================================ - -/** Details stored in CompactionEntry.details for file tracking */ +/** File-operation details stored on generated compaction entries. */ export interface CompactionDetails { + /** Files read in the compacted history. */ readFiles: string[]; + /** Files modified in the compacted history. */ modifiedFiles: string[]; } +function safeJsonStringify(value: unknown): string { + try { + return JSON.stringify(value) ?? "undefined"; + } catch { + return "[unserializable]"; + } +} -/** - * Extract file operations from messages and previous compaction entries. - */ function extractFileOperations( messages: AgentMessage[], entries: SessionTreeEntry[], prevCompactionIndex: number, ): FileOperations { const fileOps = createFileOps(); - - // Collect from previous compaction's details (if pi-generated) if (prevCompactionIndex >= 0) { const prevCompaction = entries[prevCompactionIndex] as CompactionEntry; if (!prevCompaction.fromHook && prevCompaction.details) { - // fromHook field kept for session file compatibility const details = prevCompaction.details as CompactionDetails; if (Array.isArray(details.readFiles)) { for (const f of details.readFiles) fileOps.read.add(f); @@ -60,23 +51,12 @@ function extractFileOperations( } } } - - // Extract from tool calls in messages for (const msg of messages) { extractFileOpsFromMessage(msg, fileOps); } return fileOps; } - -// ============================================================================ -// Message Extraction -// ============================================================================ - -/** - * Extract AgentMessage from an entry if it produces one. - * Returns undefined for entries that don't contribute to LLM context. - */ function getMessageFromEntry(entry: SessionTreeEntry): AgentMessage | undefined { if (entry.type === "message") { return entry.message as AgentMessage; @@ -106,47 +86,39 @@ function getMessageFromEntryForCompaction(entry: SessionTreeEntry): AgentMessage return getMessageFromEntry(entry); } -/** Result from compact() - SessionManager adds uuid/parentUuid when saving */ +/** Generated compaction data ready to be persisted as a compaction entry. */ export interface CompactionResult { + /** Summary text that replaces compacted history in future context. */ summary: string; + /** Entry id where retained history starts. */ firstKeptEntryId: string; + /** Estimated context tokens before compaction. */ tokensBefore: number; - /** Extension-specific data (e.g., ArtifactIndex, version markers for structured compaction) */ + /** Optional implementation-specific details stored with the compaction entry. */ details?: T; } -// ============================================================================ -// Types -// ============================================================================ - +/** Compaction thresholds and retention settings. */ export interface CompactionSettings { + /** Enable automatic compaction decisions. */ enabled: boolean; + /** Tokens reserved for summary prompt and output. */ reserveTokens: number; + /** Approximate recent-context tokens to keep after compaction. */ keepRecentTokens: number; } +/** Default compaction settings used by the harness. */ export const DEFAULT_COMPACTION_SETTINGS: CompactionSettings = { enabled: true, reserveTokens: 16384, keepRecentTokens: 20000, }; -// ============================================================================ -// Token calculation -// ============================================================================ - -/** - * Calculate total context tokens from usage. - * Uses the native totalTokens field when available, falls back to computing from components. - */ +/** Calculate total context tokens from provider usage. */ export function calculateContextTokens(usage: Usage): number { return usage.totalTokens || usage.input + usage.output + usage.cacheRead + usage.cacheWrite; } - -/** - * Get usage from an assistant message if available. - * Skips aborted and error messages as they don't have valid usage data. - */ function getAssistantUsage(msg: AgentMessage): Usage | undefined { if (msg.role === "assistant" && "usage" in msg) { const assistantMsg = msg as AssistantMessage; @@ -157,9 +129,7 @@ function getAssistantUsage(msg: AgentMessage): Usage | undefined { return undefined; } -/** - * Find the last non-aborted assistant message usage from session entries. - */ +/** Return usage from the last successful assistant message in session entries. */ export function getLastAssistantUsage(entries: SessionTreeEntry[]): Usage | undefined { for (let i = entries.length - 1; i >= 0; i--) { const entry = entries[i]; @@ -171,10 +141,15 @@ export function getLastAssistantUsage(entries: SessionTreeEntry[]): Usage | unde return undefined; } +/** Estimated context-token usage for a message list. */ export interface ContextUsageEstimate { + /** Estimated total context tokens. */ tokens: number; + /** Tokens reported by the most recent assistant usage block. */ usageTokens: number; + /** Estimated tokens after the most recent assistant usage block. */ trailingTokens: number; + /** Index of the message that provided usage, or null when none exists. */ lastUsageIndex: number | null; } @@ -186,10 +161,7 @@ function getLastAssistantUsageInfo(messages: AgentMessage[]): { usage: Usage; in return undefined; } -/** - * Estimate context tokens from messages, using the last assistant usage when available. - * If there are messages after the last usage, estimate their tokens with estimateTokens. - */ +/** Estimate context tokens for messages using provider usage when available. */ export function estimateContextTokens(messages: AgentMessage[]): ContextUsageEstimate { const usageInfo = getLastAssistantUsageInfo(messages); @@ -220,22 +192,13 @@ export function estimateContextTokens(messages: AgentMessage[]): ContextUsageEst }; } -/** - * Check if compaction should trigger based on context usage. - */ +/** Return whether context usage exceeds the configured compaction threshold. */ export function shouldCompact(contextTokens: number, contextWindow: number, settings: CompactionSettings): boolean { if (!settings.enabled) return false; return contextTokens > contextWindow - settings.reserveTokens; } -// ============================================================================ -// Cut point detection -// ============================================================================ - -/** - * Estimate token count for a message using chars/4 heuristic. - * This is conservative (overestimates tokens). - */ +/** Estimate token count for one message using a conservative character heuristic. */ export function estimateTokens(message: AgentMessage): number { let chars = 0; @@ -261,7 +224,7 @@ export function estimateTokens(message: AgentMessage): number { } else if (block.type === "thinking") { chars += block.thinking.length; } else if (block.type === "toolCall") { - chars += block.name.length + JSON.stringify(block.arguments).length; + chars += block.name.length + safeJsonStringify(block.arguments).length; } } return Math.ceil(chars / 4); @@ -276,7 +239,7 @@ export function estimateTokens(message: AgentMessage): number { chars += block.text.length; } if (block.type === "image") { - chars += 4800; // Estimate images as 4000 chars, or 1200 tokens + chars += 4800; } } } @@ -295,14 +258,6 @@ export function estimateTokens(message: AgentMessage): number { return 0; } - -/** - * Find valid cut points: indices of user, assistant, custom, or bashExecution messages. - * Never cut at tool results (they must follow their tool call). - * When we cut at an assistant message with tool calls, its tool results follow it - * and will be kept. - * BashExecutionMessage is treated like a user message (user-initiated context). - */ function findValidCutPoints(entries: SessionTreeEntry[], startIndex: number, endIndex: number): number[] { const cutPoints: number[] = []; for (let i = startIndex; i < endIndex; i++) { @@ -332,10 +287,9 @@ function findValidCutPoints(entries: SessionTreeEntry[], startIndex: number, end case "custom_message": case "label": case "session_info": + case "leaf": break; } - - // branch_summary and custom_message are user-role messages, valid cut points if (entry.type === "branch_summary" || entry.type === "custom_message") { cutPoints.push(i); } @@ -343,15 +297,10 @@ function findValidCutPoints(entries: SessionTreeEntry[], startIndex: number, end return cutPoints; } -/** - * Find the user message (or bashExecution) that starts the turn containing the given entry index. - * Returns -1 if no turn start found before the index. - * BashExecutionMessage is treated like a user message for turn boundaries. - */ +/** Find the user-visible message that starts the turn containing an entry. */ export function findTurnStartIndex(entries: SessionTreeEntry[], entryIndex: number, startIndex: number): number { for (let i = entryIndex; i >= startIndex; i--) { const entry = entries[i]; - // branch_summary and custom_message are user-role messages, can start a turn if (entry.type === "branch_summary" || entry.type === "custom_message") { return i; } @@ -365,31 +314,17 @@ export function findTurnStartIndex(entries: SessionTreeEntry[], entryIndex: numb return -1; } +/** Cut point selected for compaction. */ export interface CutPointResult { - /** Index of first entry to keep */ + /** Index of the first entry retained after compaction. */ firstKeptEntryIndex: number; - /** Index of user message that starts the turn being split, or -1 if not splitting */ + /** Index of the turn-start entry when the cut splits a turn, otherwise -1. */ turnStartIndex: number; - /** Whether this cut splits a turn (cut point is not a user message) */ + /** Whether the selected cut point splits an in-progress turn. */ isSplitTurn: boolean; } -/** - * Find the cut point in session entries that keeps approximately `keepRecentTokens`. - * - * Algorithm: Walk backwards from newest, accumulating estimated message sizes. - * Stop when we've accumulated >= keepRecentTokens. Cut at that point. - * - * Can cut at user OR assistant messages (never tool results). When cutting at an - * assistant message with tool calls, its tool results come after and will be kept. - * - * Returns CutPointResult with: - * - firstKeptEntryIndex: the entry index to start keeping from - * - turnStartIndex: if cutting mid-turn, the user message that started that turn - * - isSplitTurn: whether we're cutting in the middle of a turn - * - * Only considers entries between `startIndex` and `endIndex` (exclusive). - */ +/** Find the compaction cut point that keeps approximately the requested recent-token budget. */ export function findCutPoint( entries: SessionTreeEntry[], startIndex: number, @@ -401,22 +336,15 @@ export function findCutPoint( if (cutPoints.length === 0) { return { firstKeptEntryIndex: startIndex, turnStartIndex: -1, isSplitTurn: false }; } - - // Walk backwards from newest, accumulating estimated message sizes let accumulatedTokens = 0; - let cutIndex = cutPoints[0]; // Default: keep from first message (not header) + let cutIndex = cutPoints[0]; for (let i = endIndex - 1; i >= startIndex; i--) { const entry = entries[i]; if (entry.type !== "message") continue; - - // Estimate this message's size const messageTokens = estimateTokens(entry.message as AgentMessage); accumulatedTokens += messageTokens; - - // Check if we've exceeded the budget if (accumulatedTokens >= keepRecentTokens) { - // Find the closest valid cut point at or after this entry for (let c = 0; c < cutPoints.length; c++) { if (cutPoints[c] >= i) { cutIndex = cutPoints[c]; @@ -426,23 +354,16 @@ export function findCutPoint( break; } } - - // Scan backwards from cutIndex to include any non-message entries (bash, settings, etc.) while (cutIndex > startIndex) { const prevEntry = entries[cutIndex - 1]; - // Stop at session header or compaction boundaries if (prevEntry.type === "compaction") { break; } if (prevEntry.type === "message") { - // Stop if we hit any message break; } - // Include this non-message entry (bash, settings change, etc.) cutIndex--; } - - // Determine if this is a split turn const cutEntry = entries[cutIndex]; const isUserMessage = cutEntry.type === "message" && cutEntry.message.role === "user"; const turnStartIndex = isUserMessage ? -1 : findTurnStartIndex(entries, cutIndex, startIndex); @@ -454,9 +375,9 @@ export function findCutPoint( }; } -// ============================================================================ -// Summarization -// ============================================================================ +export const SUMMARIZATION_SYSTEM_PROMPT = `You are a context summarization assistant. Your task is to read a conversation between a user and an AI coding assistant, then produce a structured summary following the exact format specified. + +Do NOT continue the conversation. Do NOT respond to any questions in the conversation. ONLY output the structured summary.`; const SUMMARIZATION_PROMPT = `The messages above are a conversation to summarize. Create a structured context checkpoint summary that another LLM will use to continue the work. @@ -530,10 +451,7 @@ Use this EXACT format: Keep each section concise. Preserve exact file paths, function names, and error messages.`; -/** - * Generate a summary of the conversation using the LLM. - * If previousSummary is provided, uses the update prompt to merge. - */ +/** Generate or update a conversation summary for compaction. */ export async function generateSummary( currentMessages: AgentMessage[], model: Model, @@ -549,19 +467,12 @@ export async function generateSummary( Math.floor(0.8 * reserveTokens), model.maxTokens > 0 ? model.maxTokens : Number.POSITIVE_INFINITY, ); - - // Use update prompt if we have a previous summary, otherwise initial prompt let basePrompt = previousSummary ? UPDATE_SUMMARIZATION_PROMPT : SUMMARIZATION_PROMPT; if (customInstructions) { basePrompt = `${basePrompt}\n\nAdditional focus: ${customInstructions}`; } - - // Serialize conversation to text so model doesn't try to continue it - // Convert to LLM messages first (handles custom types like bashExecution, custom, etc.) const llmMessages = convertToLlm(currentMessages); const conversationText = serializeConversation(llmMessages); - - // Build the prompt with conversation wrapped in tags let promptText = `\n${conversationText}\n\n\n`; if (previousSummary) { promptText += `\n${previousSummary}\n\n\n`; @@ -581,18 +492,11 @@ export async function generateSummary( ? { maxTokens, signal, apiKey, headers, reasoning: thinkingLevel } : { maxTokens, signal, apiKey, headers }; - const responseResult: Result = await completeSimple( + const response = await completeSimple( model, { systemPrompt: SUMMARIZATION_SYSTEM_PROMPT, messages: summarizationMessages }, completionOptions, - ).then( - (response) => ok(response), - (error: unknown) => - err(new CompactionError("unknown", "Summarization request failed", error)), ); - if (!responseResult.ok) return err(responseResult.error); - - const response = responseResult.value; if (response.stopReason === "aborted") { return err(new CompactionError("aborted", response.errorMessage || "Summarization aborted")); } @@ -613,34 +517,33 @@ export async function generateSummary( return ok(textContent); } -// ============================================================================ -// Compaction Preparation (for extensions) -// ============================================================================ - +/** Prepared inputs for a compaction run. */ export interface CompactionPreparation { - /** UUID of first entry to keep */ + /** Entry id where retained history starts. */ firstKeptEntryId: string; - /** Messages that will be summarized and discarded */ + /** Messages summarized into the history summary. */ messagesToSummarize: AgentMessage[]; - /** Messages that will be turned into turn prefix summary (if splitting) */ + /** Prefix messages summarized separately when compaction splits a turn. */ turnPrefixMessages: AgentMessage[]; - /** Whether this is a split turn (cut point in middle of turn) */ + /** Whether compaction splits a turn. */ isSplitTurn: boolean; + /** Estimated context tokens before compaction. */ tokensBefore: number; - /** Summary from previous compaction, for iterative update */ + /** Previous compaction summary used for iterative updates. */ previousSummary?: string; - /** File operations extracted from messagesToSummarize */ + /** File operations extracted from summarized history. */ fileOps: FileOperations; - /** Compaction settions from settings.jsonl */ + /** Settings used to prepare compaction. */ settings: CompactionSettings; } +/** Prepare session entries for compaction, or return undefined when compaction is not applicable. */ export function prepareCompaction( pathEntries: SessionTreeEntry[], settings: CompactionSettings, -): CompactionPreparation | undefined { - if (pathEntries.length > 0 && pathEntries[pathEntries.length - 1].type === "compaction") { - return undefined; +): Result { + if (pathEntries.length === 0 || pathEntries[pathEntries.length - 1].type === "compaction") { + return ok(undefined); } let prevCompactionIndex = -1; @@ -664,24 +567,18 @@ export function prepareCompaction( const tokensBefore = estimateContextTokens(buildSessionContext(pathEntries).messages).tokens; const cutPoint = findCutPoint(pathEntries, boundaryStart, boundaryEnd, settings.keepRecentTokens); - - // Get UUID of first kept entry const firstKeptEntry = pathEntries[cutPoint.firstKeptEntryIndex]; if (!firstKeptEntry?.id) { - return undefined; // Session needs migration + return err(new CompactionError("invalid_session", "First kept entry has no UUID - session may need migration")); } const firstKeptEntryId = firstKeptEntry.id; const historyEnd = cutPoint.isSplitTurn ? cutPoint.turnStartIndex : cutPoint.firstKeptEntryIndex; - - // Messages to summarize (will be discarded after summary) const messagesToSummarize: AgentMessage[] = []; for (let i = boundaryStart; i < historyEnd; i++) { const msg = getMessageFromEntryForCompaction(pathEntries[i]); if (msg) messagesToSummarize.push(msg); } - - // Messages for turn prefix summary (if splitting a turn) const turnPrefixMessages: AgentMessage[] = []; if (cutPoint.isSplitTurn) { for (let i = cutPoint.turnStartIndex; i < cutPoint.firstKeptEntryIndex; i++) { @@ -689,18 +586,14 @@ export function prepareCompaction( if (msg) turnPrefixMessages.push(msg); } } - - // Extract file operations from messages and previous compaction const fileOps = extractFileOperations(messagesToSummarize, pathEntries, prevCompactionIndex); - - // Also extract file ops from turn prefix if splitting if (cutPoint.isSplitTurn) { for (const msg of turnPrefixMessages) { extractFileOpsFromMessage(msg, fileOps); } } - return { + return ok({ firstKeptEntryId, messagesToSummarize, turnPrefixMessages, @@ -709,13 +602,9 @@ export function prepareCompaction( previousSummary, fileOps, settings, - }; + }); } -// ============================================================================ -// Main compaction function -// ============================================================================ - const TURN_PREFIX_SUMMARIZATION_PROMPT = `This is the PREFIX of a turn that was too large to keep. The SUFFIX (recent work) is retained. Summarize the prefix to provide context for the retained suffix: @@ -731,15 +620,9 @@ Summarize the prefix to provide context for the retained suffix: Be concise. Focus on what's needed to understand the kept suffix.`; -/** - * Generate summaries for compaction using prepared data. - * Returns CompactionResult - SessionManager adds uuid/parentUuid when saving. - * - * @param preparation - Pre-calculated preparation from prepareCompaction() - * @param customInstructions - Optional custom focus for the summary - */ export { serializeConversation } from "./utils.js"; +/** Generate compaction summary data from prepared session history. */ export async function compact( preparation: CompactionPreparation, model: Model, @@ -820,10 +703,6 @@ export async function compact( details: { readFiles, modifiedFiles } as CompactionDetails, }); } - -/** - * Generate a summary for a turn prefix (when splitting a turn). - */ async function generateTurnPrefixSummary( messages: AgentMessage[], model: Model, @@ -836,7 +715,7 @@ async function generateTurnPrefixSummary( const maxTokens = Math.min( Math.floor(0.5 * reserveTokens), model.maxTokens > 0 ? model.maxTokens : Number.POSITIVE_INFINITY, - ); // Smaller budget for turn prefix + ); const llmMessages = convertToLlm(messages); const conversationText = serializeConversation(llmMessages); const promptText = `\n${conversationText}\n\n\n${TURN_PREFIX_SUMMARIZATION_PROMPT}`; @@ -848,22 +727,13 @@ async function generateTurnPrefixSummary( }, ]; - const responseResult: Result = await completeSimple( + const response = await completeSimple( model, { systemPrompt: SUMMARIZATION_SYSTEM_PROMPT, messages: summarizationMessages }, model.reasoning && thinkingLevel && thinkingLevel !== "off" ? { maxTokens, signal, apiKey, headers, reasoning: thinkingLevel } : { maxTokens, signal, apiKey, headers }, - ).then( - (response) => ok(response), - (error: unknown) => - err( - new CompactionError("unknown", "Turn prefix summarization request failed", error), - ), ); - if (!responseResult.ok) return err(responseResult.error); - - const response = responseResult.value; if (response.stopReason === "aborted") { return err(new CompactionError("aborted", response.errorMessage || "Turn prefix summarization aborted")); } diff --git a/packages/agent/src/harness/compaction/utils.ts b/packages/agent/src/harness/compaction/utils.ts index 35cd1938..2b852a46 100644 --- a/packages/agent/src/harness/compaction/utils.ts +++ b/packages/agent/src/harness/compaction/utils.ts @@ -1,20 +1,17 @@ -/** - * Shared utilities for compaction and branch summarization. - */ - import type { Message } from "@earendil-works/pi-ai"; import type { AgentMessage } from "../../types.js"; -// ============================================================================ -// File Operation Tracking -// ============================================================================ - +/** File paths touched by a session branch or compaction range. */ export interface FileOperations { + /** Files read but not necessarily modified. */ read: Set; + /** Files written by full-file write operations. */ written: Set; + /** Files modified by edit operations. */ edited: Set; } +/** Create an empty file-operation accumulator. */ export function createFileOps(): FileOperations { return { read: new Set(), @@ -23,9 +20,7 @@ export function createFileOps(): FileOperations { }; } -/** - * Extract file operations from tool calls in an assistant message. - */ +/** Add file operations from assistant tool calls to an accumulator. */ export function extractFileOpsFromMessage(message: AgentMessage, fileOps: FileOperations): void { if (message.role !== "assistant") return; if (!("content" in message) || !Array.isArray(message.content)) return; @@ -55,10 +50,7 @@ export function extractFileOpsFromMessage(message: AgentMessage, fileOps: FileOp } } -/** - * Compute final file lists from file operations. - * Returns readFiles (files only read, not modified) and modifiedFiles. - */ +/** Compute sorted read-only and modified file lists from accumulated operations. */ export function computeFileLists(fileOps: FileOperations): { readFiles: string[]; modifiedFiles: string[] } { const modified = new Set([...fileOps.edited, ...fileOps.written]); const readOnly = [...fileOps.read].filter((f) => !modified.has(f)).sort(); @@ -66,9 +58,7 @@ export function computeFileLists(fileOps: FileOperations): { readFiles: string[] return { readFiles: readOnly, modifiedFiles }; } -/** - * Format file operations as XML tags for summary. - */ +/** Format file lists as summary metadata tags. */ export function formatFileOperations(readFiles: string[], modifiedFiles: string[]): string { const sections: string[] = []; if (readFiles.length > 0) { @@ -81,31 +71,23 @@ export function formatFileOperations(readFiles: string[], modifiedFiles: string[ return `\n\n${sections.join("\n\n")}`; } -// ============================================================================ -// Message Serialization -// ============================================================================ - -/** Maximum characters for a tool result in serialized summaries. */ const TOOL_RESULT_MAX_CHARS = 2000; -/** - * Truncate text to a maximum character length for summarization. - * Keeps the beginning and appends a truncation marker. - */ +function safeJsonStringify(value: unknown): string { + try { + return JSON.stringify(value) ?? "undefined"; + } catch { + return "[unserializable]"; + } +} + function truncateForSummary(text: string, maxChars: number): string { if (text.length <= maxChars) return text; const truncatedChars = text.length - maxChars; return `${text.slice(0, maxChars)}\n\n[... ${truncatedChars} more characters truncated]`; } -/** - * Serialize LLM messages to text for summarization. - * This prevents the model from treating it as a conversation to continue. - * Call convertToLlm() first to handle custom message types. - * - * Tool results are truncated to keep the summarization request within - * reasonable token budgets. Full content is not needed for summarization. - */ +/** Serialize LLM messages to plain text for summarization prompts. */ export function serializeConversation(messages: Message[]): string { const parts: string[] = []; @@ -132,7 +114,7 @@ export function serializeConversation(messages: Message[]): string { } else if (block.type === "toolCall") { const args = block.arguments as Record; const argsStr = Object.entries(args) - .map(([k, v]) => `${k}=${JSON.stringify(v)}`) + .map(([k, v]) => `${k}=${safeJsonStringify(v)}`) .join(", "); toolCalls.push(`${block.name}(${argsStr})`); } @@ -160,11 +142,3 @@ export function serializeConversation(messages: Message[]): string { return parts.join("\n\n"); } - -// ============================================================================ -// Summarization System Prompt -// ============================================================================ - -export const SUMMARIZATION_SYSTEM_PROMPT = `You are a context summarization assistant. Your task is to read a conversation between a user and an AI coding assistant, then produce a structured summary following the exact format specified. - -Do NOT continue the conversation. Do NOT respond to any questions in the conversation. ONLY output the structured summary.`; diff --git a/packages/agent/src/harness/env/nodejs.ts b/packages/agent/src/harness/env/nodejs.ts index 0b23a2d1..b86f6fe8 100644 --- a/packages/agent/src/harness/env/nodejs.ts +++ b/packages/agent/src/harness/env/nodejs.ts @@ -1,6 +1,6 @@ import { spawn } from "node:child_process"; import { randomUUID } from "node:crypto"; -import { constants } from "node:fs"; +import { constants, createReadStream } from "node:fs"; import { access, appendFile, @@ -15,6 +15,7 @@ import { } from "node:fs/promises"; import { tmpdir } from "node:os"; import { isAbsolute, join, resolve } from "node:path"; +import { createInterface } from "node:readline"; import { type ExecutionEnv, ExecutionError, @@ -24,6 +25,7 @@ import { type FileKind, ok, type Result, + toError, } from "../types.js"; function resolvePath(cwd: string, path: string): string { @@ -62,25 +64,26 @@ function isNodeError(error: unknown): error is NodeJS.ErrnoException { function toFileError(error: unknown, path?: string): FileError { if (error instanceof FileError) return error; + const cause = toError(error); if (isNodeError(error)) { const message = error.message; switch (error.code) { case "ABORT_ERR": - return new FileError("aborted", message, path, error); + return new FileError("aborted", message, path, cause); case "ENOENT": - return new FileError("not_found", message, path, error); + return new FileError("not_found", message, path, cause); case "EACCES": case "EPERM": - return new FileError("permission_denied", message, path, error); + return new FileError("permission_denied", message, path, cause); case "ENOTDIR": - return new FileError("not_directory", message, path, error); + return new FileError("not_directory", message, path, cause); case "EISDIR": - return new FileError("is_directory", message, path, error); + return new FileError("is_directory", message, path, cause); case "EINVAL": - return new FileError("invalid", message, path, error); + return new FileError("invalid", message, path, cause); } } - return new FileError("unknown", error instanceof Error ? error.message : String(error), path, error); + return new FileError("unknown", cause.message, path, cause); } function abortResult(signal: AbortSignal | undefined, path?: string): Result | undefined { @@ -218,18 +221,12 @@ export class NodeExecutionEnv implements ExecutionEnv { this.shellEnv = options.shellEnv; } - async absolutePath(path: string, abortSignal?: AbortSignal): Promise> { - const resolved = resolvePath(this.cwd, path); - const aborted = abortResult(abortSignal, resolved); - if (aborted) return aborted; - return ok(resolved); + async absolutePath(path: string): Promise> { + return ok(resolvePath(this.cwd, path)); } - async joinPath(parts: string[], abortSignal?: AbortSignal): Promise> { - const joined = join(...parts); - const aborted = abortResult(abortSignal, joined); - if (aborted) return aborted; - return ok(joined); + async joinPath(parts: string[]): Promise> { + return ok(join(...parts)); } async exec( @@ -280,9 +277,8 @@ export class NodeExecutionEnv implements ExecutionEnv { stdio: ["ignore", "pipe", "pipe"], }); } catch (error) { - settle( - err(new ExecutionError("spawn_error", error instanceof Error ? error.message : String(error), error)), - ); + const cause = toError(error); + settle(err(new ExecutionError("spawn_error", cause.message, cause))); return; } @@ -311,11 +307,8 @@ export class NodeExecutionEnv implements ExecutionEnv { try { options?.onStdout?.(chunk); } catch (error) { - callbackError = new ExecutionError( - "unknown", - error instanceof Error ? error.message : String(error), - error, - ); + const cause = toError(error); + callbackError = new ExecutionError("callback_error", cause.message, cause); onAbort(); } }); @@ -324,11 +317,8 @@ export class NodeExecutionEnv implements ExecutionEnv { try { options?.onStderr?.(chunk); } catch (error) { - callbackError = new ExecutionError( - "unknown", - error instanceof Error ? error.message : String(error), - error, - ); + const cause = toError(error); + callbackError = new ExecutionError("callback_error", cause.message, cause); onAbort(); } }); @@ -342,14 +332,14 @@ export class NodeExecutionEnv implements ExecutionEnv { settle(err(callbackError)); return; } - if (options?.abortSignal?.aborted) { - settle(err(new ExecutionError("aborted", "aborted"))); - return; - } if (timedOut) { settle(err(new ExecutionError("timeout", `timeout:${options?.timeout}`))); return; } + if (options?.abortSignal?.aborted) { + settle(err(new ExecutionError("aborted", "aborted"))); + return; + } settle(ok({ stdout, stderr, exitCode: code ?? 0 })); }); }); @@ -366,6 +356,37 @@ export class NodeExecutionEnv implements ExecutionEnv { } } + async readTextLines( + path: string, + options?: { maxLines?: number; abortSignal?: AbortSignal }, + ): Promise> { + const resolved = resolvePath(this.cwd, path); + const aborted = abortResult(options?.abortSignal, resolved); + if (aborted) return aborted; + if (options?.maxLines !== undefined && options.maxLines <= 0) return ok([]); + let stream: ReturnType | undefined; + let lineReader: ReturnType | undefined; + try { + stream = createReadStream(resolved, { encoding: "utf8", signal: options?.abortSignal }); + lineReader = createInterface({ input: stream, crlfDelay: Infinity }); + const lines: string[] = []; + for await (const line of lineReader) { + const loopAbort = abortResult(options?.abortSignal, resolved); + if (loopAbort) return loopAbort; + lines.push(line); + if (options?.maxLines !== undefined && lines.length >= options.maxLines) break; + } + const afterReadAbort = abortResult(options?.abortSignal, resolved); + if (afterReadAbort) return afterReadAbort; + return ok(lines); + } catch (error) { + return err(toFileError(error, resolved)); + } finally { + lineReader?.close(); + stream?.destroy(); + } + } + async readBinaryFile(path: string, abortSignal?: AbortSignal): Promise> { const resolved = resolvePath(this.cwd, path); const aborted = abortResult(abortSignal, resolved); @@ -396,31 +417,19 @@ export class NodeExecutionEnv implements ExecutionEnv { } } - async appendFile( - path: string, - content: string | Uint8Array, - abortSignal?: AbortSignal, - ): Promise> { + async appendFile(path: string, content: string | Uint8Array): Promise> { const resolved = resolvePath(this.cwd, path); - const aborted = abortResult(abortSignal, resolved); - if (aborted) return aborted; try { await mkdir(resolve(resolved, ".."), { recursive: true }); - const afterMkdirAbort = abortResult(abortSignal, resolved); - if (afterMkdirAbort) return afterMkdirAbort; await appendFile(resolved, content); - const afterAppendAbort = abortResult(abortSignal, resolved); - if (afterAppendAbort) return afterAppendAbort; return ok(undefined); } catch (error) { return err(toFileError(error, resolved)); } } - async fileInfo(path: string, abortSignal?: AbortSignal): Promise> { + async fileInfo(path: string): Promise> { const resolved = resolvePath(this.cwd, path); - const aborted = abortResult(abortSignal, resolved); - if (aborted) return aborted; try { return fileInfoFromStats(resolved, await lstat(resolved)); } catch (error) { @@ -452,10 +461,8 @@ export class NodeExecutionEnv implements ExecutionEnv { } } - async canonicalPath(path: string, abortSignal?: AbortSignal): Promise> { + async canonicalPath(path: string): Promise> { const resolved = resolvePath(this.cwd, path); - const aborted = abortResult(abortSignal, resolved); - if (aborted) return aborted; try { return ok(await realpath(resolved)); } catch (error) { @@ -463,72 +470,47 @@ export class NodeExecutionEnv implements ExecutionEnv { } } - async exists(path: string, abortSignal?: AbortSignal): Promise> { - const result = await this.fileInfo(path, abortSignal); + async exists(path: string): Promise> { + const result = await this.fileInfo(path); if (result.ok) return ok(true); if (result.error.code === "not_found") return ok(false); return err(result.error); } - async createDir( - path: string, - options?: { recursive?: boolean; abortSignal?: AbortSignal }, - ): Promise> { + async createDir(path: string, options?: { recursive?: boolean }): Promise> { const resolved = resolvePath(this.cwd, path); - const aborted = abortResult(options?.abortSignal, resolved); - if (aborted) return aborted; try { await mkdir(resolved, { recursive: options?.recursive ?? true }); - const afterMkdirAbort = abortResult(options?.abortSignal, resolved); - if (afterMkdirAbort) return afterMkdirAbort; return ok(undefined); } catch (error) { return err(toFileError(error, resolved)); } } - async remove( - path: string, - options?: { recursive?: boolean; force?: boolean; abortSignal?: AbortSignal }, - ): Promise> { + async remove(path: string, options?: { recursive?: boolean; force?: boolean }): Promise> { const resolved = resolvePath(this.cwd, path); - const aborted = abortResult(options?.abortSignal, resolved); - if (aborted) return aborted; try { await rm(resolved, { recursive: options?.recursive ?? false, force: options?.force ?? false }); - const afterRemoveAbort = abortResult(options?.abortSignal, resolved); - if (afterRemoveAbort) return afterRemoveAbort; return ok(undefined); } catch (error) { return err(toFileError(error, resolved)); } } - async createTempDir(prefix: string = "tmp-", abortSignal?: AbortSignal): Promise> { - const aborted = abortResult(abortSignal); - if (aborted) return aborted; + async createTempDir(prefix: string = "tmp-"): Promise> { try { - const path = await mkdtemp(join(tmpdir(), prefix)); - const afterMkdtempAbort = abortResult(abortSignal, path); - if (afterMkdtempAbort) return afterMkdtempAbort; - return ok(path); + return ok(await mkdtemp(join(tmpdir(), prefix))); } catch (error) { return err(toFileError(error)); } } - async createTempFile(options?: { - prefix?: string; - suffix?: string; - abortSignal?: AbortSignal; - }): Promise> { - const dir = await this.createTempDir("tmp-", options?.abortSignal); + async createTempFile(options?: { prefix?: string; suffix?: string }): Promise> { + const dir = await this.createTempDir("tmp-"); if (!dir.ok) return dir; const filePath = join(dir.value, `${options?.prefix ?? ""}${randomUUID()}${options?.suffix ?? ""}`); - const aborted = abortResult(options?.abortSignal, filePath); - if (aborted) return aborted; try { - await writeFile(filePath, "", { signal: options?.abortSignal }); + await writeFile(filePath, ""); return ok(filePath); } catch (error) { return err(toFileError(error, filePath)); diff --git a/packages/agent/src/harness/prompt-templates.ts b/packages/agent/src/harness/prompt-templates.ts index abcaad5a..3e678cfc 100644 --- a/packages/agent/src/harness/prompt-templates.ts +++ b/packages/agent/src/harness/prompt-templates.ts @@ -1,10 +1,14 @@ import { parse } from "yaml"; -import { type ExecutionEnv, type FileInfo, getOrUndefined, type PromptTemplate, type Result } from "./types.js"; +import { type ExecutionEnv, type FileInfo, type PromptTemplate, type Result, toError } from "./types.js"; + +export type PromptTemplateDiagnosticCode = "file_info_failed" | "list_failed" | "read_failed" | "parse_failed"; /** Warning produced while loading prompt templates. */ export interface PromptTemplateDiagnostic { /** Diagnostic severity. Currently only warnings are emitted. */ type: "warning"; + /** Stable diagnostic code. */ + code: PromptTemplateDiagnosticCode; /** Human-readable diagnostic message. */ message: string; /** Path associated with the diagnostic. */ @@ -30,9 +34,20 @@ export async function loadPromptTemplates( const promptTemplates: PromptTemplate[] = []; const diagnostics: PromptTemplateDiagnostic[] = []; for (const path of Array.isArray(paths) ? paths : [paths]) { - const info = getOrUndefined(await env.fileInfo(path)); - if (!info) continue; - const kind = await resolveKind(env, info); + const infoResult = await env.fileInfo(path); + if (!infoResult.ok) { + if (infoResult.error.code !== "not_found") { + diagnostics.push({ + type: "warning", + code: "file_info_failed", + message: infoResult.error.message, + path, + }); + } + continue; + } + const info = infoResult.value; + const kind = await resolveKind(env, info, diagnostics); if (kind === "directory") { const result = await loadTemplatesFromDir(env, info.path); promptTemplates.push(...result.promptTemplates); @@ -87,6 +102,7 @@ async function loadTemplatesFromDir( if (!entriesResult.ok) { diagnostics.push({ type: "warning", + code: "list_failed", message: entriesResult.error.message, path: dir, }); @@ -95,7 +111,7 @@ async function loadTemplatesFromDir( const entries = entriesResult.value; for (const entry of entries.sort((a, b) => a.name.localeCompare(b.name))) { - const kind = await resolveKind(env, entry); + const kind = await resolveKind(env, entry, diagnostics); if (kind !== "file" || !entry.name.endsWith(".md")) continue; const result = await loadTemplateFromFile(env, entry.path); if (result.promptTemplate) promptTemplates.push(result.promptTemplate); @@ -113,6 +129,7 @@ async function loadTemplateFromFile( if (!rawContent.ok) { diagnostics.push({ type: "warning", + code: "read_failed", message: rawContent.error.message, path: filePath, }); @@ -123,6 +140,7 @@ async function loadTemplateFromFile( if (!parsed.ok) { diagnostics.push({ type: "warning", + code: "parse_failed", message: parsed.error.message, path: filePath, }); @@ -146,13 +164,37 @@ async function loadTemplateFromFile( }; } -async function resolveKind(env: ExecutionEnv, info: FileInfo): Promise<"file" | "directory" | undefined> { +async function resolveKind( + env: ExecutionEnv, + info: FileInfo, + diagnostics: PromptTemplateDiagnostic[], +): Promise<"file" | "directory" | undefined> { if (info.kind === "file" || info.kind === "directory") return info.kind; const canonicalPath = await env.canonicalPath(info.path); - if (!canonicalPath.ok) return undefined; - const target = getOrUndefined(await env.fileInfo(canonicalPath.value)); - if (!target) return undefined; - return target.kind === "file" || target.kind === "directory" ? target.kind : undefined; + if (!canonicalPath.ok) { + if (canonicalPath.error.code !== "not_found") { + diagnostics.push({ + type: "warning", + code: "file_info_failed", + message: canonicalPath.error.message, + path: info.path, + }); + } + return undefined; + } + const target = await env.fileInfo(canonicalPath.value); + if (!target.ok) { + if (target.error.code !== "not_found") { + diagnostics.push({ + type: "warning", + code: "file_info_failed", + message: target.error.message, + path: info.path, + }); + } + return undefined; + } + return target.value.kind === "file" || target.value.kind === "directory" ? target.value.kind : undefined; } function parseFrontmatter>( @@ -167,7 +209,7 @@ function parseFrontmatter>( const body = normalized.slice(endIndex + 4).trim(); return { ok: true, value: { frontmatter: (parse(yamlString) ?? {}) as T, body } }; } catch (error) { - return { ok: false, error: error instanceof Error ? error : new Error(String(error)) }; + return { ok: false, error: toError(error) }; } } diff --git a/packages/agent/src/harness/session/jsonl-repo.ts b/packages/agent/src/harness/session/jsonl-repo.ts index e41f0fb3..cd6ac1d8 100644 --- a/packages/agent/src/harness/session/jsonl-repo.ts +++ b/packages/agent/src/harness/session/jsonl-repo.ts @@ -6,9 +6,15 @@ import type { JsonlSessionRepoApi, Session, } from "../types.js"; -import { getOrThrow } from "../types.js"; +import { SessionError, toError } from "../types.js"; import { JsonlSessionStorage, loadJsonlSessionMetadata } from "./jsonl-storage.js"; -import { createSessionId, createTimestamp, getEntriesToFork, toSession } from "./repo-utils.js"; +import { + createSessionId, + createTimestamp, + getEntriesToFork, + getFileSystemResultOrThrow, + toSession, +} from "./repo-utils.js"; type JsonlSessionRepoFileSystem = Pick< FileSystem, @@ -16,6 +22,7 @@ type JsonlSessionRepoFileSystem = Pick< | "absolutePath" | "joinPath" | "readTextFile" + | "readTextLines" | "writeFile" | "appendFile" | "listDir" @@ -40,21 +47,28 @@ export class JsonlSessionRepo implements JsonlSessionRepoApi { private async getSessionsRoot(): Promise { if (!this.sessionsRoot) { - this.sessionsRoot = getOrThrow(await this.fs.absolutePath(this.sessionsRootInput)); + this.sessionsRoot = getFileSystemResultOrThrow( + await this.fs.absolutePath(this.sessionsRootInput), + `Failed to resolve sessions root ${this.sessionsRootInput}`, + ); } return this.sessionsRoot; } private async getSessionDir(cwd: string): Promise { - return getOrThrow(await this.fs.joinPath([await this.getSessionsRoot(), encodeCwd(cwd)])); + return getFileSystemResultOrThrow( + await this.fs.joinPath([await this.getSessionsRoot(), encodeCwd(cwd)]), + `Failed to resolve session directory for ${cwd}`, + ); } private async createSessionFilePath(cwd: string, sessionId: string, timestamp: string): Promise { - return getOrThrow( + return getFileSystemResultOrThrow( await this.fs.joinPath([ await this.getSessionDir(cwd), `${timestamp.replace(/[:.]/g, "-")}_${sessionId}.jsonl`, ]), + `Failed to resolve session file path for ${sessionId}`, ); } @@ -62,7 +76,10 @@ export class JsonlSessionRepo implements JsonlSessionRepoApi { const id = options.id ?? createSessionId(); const createdAt = createTimestamp(); const sessionDir = await this.getSessionDir(options.cwd); - getOrThrow(await this.fs.createDir(sessionDir, { recursive: true })); + getFileSystemResultOrThrow( + await this.fs.createDir(sessionDir, { recursive: true }), + `Failed to create session directory ${sessionDir}`, + ); const filePath = await this.createSessionFilePath(options.cwd, id, createdAt); const storage = await JsonlSessionStorage.create(this.fs, filePath, { cwd: options.cwd, @@ -73,8 +90,10 @@ export class JsonlSessionRepo implements JsonlSessionRepoApi { } async open(metadata: JsonlSessionMetadata): Promise> { - if (!getOrThrow(await this.fs.exists(metadata.path))) { - throw new Error(`Session not found: ${metadata.path}`); + if ( + !getFileSystemResultOrThrow(await this.fs.exists(metadata.path), `Failed to check session ${metadata.path}`) + ) { + throw new SessionError("not_found", `Session not found: ${metadata.path}`); } const storage = await JsonlSessionStorage.open(this.fs, metadata.path); return toSession(storage); @@ -84,15 +103,19 @@ export class JsonlSessionRepo implements JsonlSessionRepoApi { const dirs = options.cwd ? [await this.getSessionDir(options.cwd)] : await this.listSessionDirs(); const sessions: JsonlSessionMetadata[] = []; for (const dir of dirs) { - if (!getOrThrow(await this.fs.exists(dir))) continue; - const files = getOrThrow(await this.fs.listDir(dir)).filter( - (file) => file.kind !== "directory" && file.name.endsWith(".jsonl"), - ); + if (!getFileSystemResultOrThrow(await this.fs.exists(dir), `Failed to check session directory ${dir}`)) { + continue; + } + const files = getFileSystemResultOrThrow( + await this.fs.listDir(dir), + `Failed to list sessions in ${dir}`, + ).filter((file) => file.kind !== "directory" && file.name.endsWith(".jsonl")); for (const file of files) { try { sessions.push(await loadJsonlSessionMetadata(this.fs, file.path)); - } catch { - // Ignore invalid session files when listing a directory. + } catch (error) { + const cause = toError(error); + if (!(cause instanceof SessionError) || cause.code !== "invalid_session") throw cause; } } } @@ -101,7 +124,10 @@ export class JsonlSessionRepo implements JsonlSessionRepoApi { } async delete(metadata: JsonlSessionMetadata): Promise { - getOrThrow(await this.fs.remove(metadata.path, { force: true })); + getFileSystemResultOrThrow( + await this.fs.remove(metadata.path, { force: true }), + `Failed to delete session ${metadata.path}`, + ); } async fork( @@ -113,7 +139,10 @@ export class JsonlSessionRepo implements JsonlSessionRepoApi { const id = options.id ?? createSessionId(); const createdAt = createTimestamp(); const sessionDir = await this.getSessionDir(options.cwd); - getOrThrow(await this.fs.createDir(sessionDir, { recursive: true })); + getFileSystemResultOrThrow( + await this.fs.createDir(sessionDir, { recursive: true }), + `Failed to create session directory ${sessionDir}`, + ); const storage = await JsonlSessionStorage.create( this.fs, await this.createSessionFilePath(options.cwd, id, createdAt), @@ -131,8 +160,18 @@ export class JsonlSessionRepo implements JsonlSessionRepoApi { private async listSessionDirs(): Promise { const sessionsRoot = await this.getSessionsRoot(); - if (!getOrThrow(await this.fs.exists(sessionsRoot))) return []; - const entries = getOrThrow(await this.fs.listDir(sessionsRoot)); + if ( + !getFileSystemResultOrThrow( + await this.fs.exists(sessionsRoot), + `Failed to check sessions root ${sessionsRoot}`, + ) + ) { + return []; + } + const entries = getFileSystemResultOrThrow( + await this.fs.listDir(sessionsRoot), + `Failed to list sessions root ${sessionsRoot}`, + ); return entries.filter((entry) => entry.kind === "directory").map((entry) => entry.path); } } diff --git a/packages/agent/src/harness/session/jsonl-storage.ts b/packages/agent/src/harness/session/jsonl-storage.ts index a6f370f7..eac8334d 100644 --- a/packages/agent/src/harness/session/jsonl-storage.ts +++ b/packages/agent/src/harness/session/jsonl-storage.ts @@ -1,8 +1,9 @@ -import type { FileSystem, JsonlSessionMetadata, SessionStorage, SessionTreeEntry } from "../types.js"; -import { getOrThrow } from "../types.js"; +import type { FileSystem, JsonlSessionMetadata, LeafEntry, SessionStorage, SessionTreeEntry } from "../types.js"; +import { SessionError, toError } from "../types.js"; +import { getFileSystemResultOrThrow } from "./repo-utils.js"; import { uuidv7 } from "./uuid.js"; -type JsonlSessionStorageFileSystem = Pick; +type JsonlSessionStorageFileSystem = Pick; interface SessionHeader { type: "session"; @@ -39,6 +40,76 @@ function generateEntryId(byId: { has(id: string): boolean }): string { return uuidv7(); } +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null; +} + +function invalidSession(filePath: string, message: string, cause?: Error): SessionError { + return new SessionError("invalid_session", `Invalid JSONL session file ${filePath}: ${message}`, cause); +} + +function invalidEntry(filePath: string, lineNumber: number, message: string, cause?: Error): SessionError { + return new SessionError( + "invalid_entry", + `Invalid JSONL session file ${filePath}: line ${lineNumber} ${message}`, + cause, + ); +} + +function parseHeaderLine(line: string, filePath: string): SessionHeader { + let parsed: unknown; + try { + parsed = JSON.parse(line); + } catch (error) { + throw invalidSession(filePath, "first line is not a valid session header", toError(error)); + } + if (!isRecord(parsed)) throw invalidSession(filePath, "first line is not a valid session header"); + if (parsed.type !== "session") throw invalidSession(filePath, "first line is not a valid session header"); + if (parsed.version !== 3) throw invalidSession(filePath, "unsupported session version"); + if (typeof parsed.id !== "string" || !parsed.id) throw invalidSession(filePath, "session header is missing id"); + if (typeof parsed.timestamp !== "string" || !parsed.timestamp) { + throw invalidSession(filePath, "session header is missing timestamp"); + } + if (typeof parsed.cwd !== "string" || !parsed.cwd) throw invalidSession(filePath, "session header is missing cwd"); + if (parsed.parentSession !== undefined && typeof parsed.parentSession !== "string") { + throw invalidSession(filePath, "session header parentSession must be a string"); + } + return { + type: "session", + version: 3, + id: parsed.id, + timestamp: parsed.timestamp, + cwd: parsed.cwd, + parentSession: parsed.parentSession, + }; +} + +function parseEntryLine(line: string, filePath: string, lineNumber: number): SessionTreeEntry { + let parsed: unknown; + try { + parsed = JSON.parse(line); + } catch (error) { + throw invalidEntry(filePath, lineNumber, "is not valid JSON", toError(error)); + } + if (!isRecord(parsed)) throw invalidEntry(filePath, lineNumber, "is not a valid session entry"); + if (typeof parsed.type !== "string") throw invalidEntry(filePath, lineNumber, "is missing entry type"); + if (typeof parsed.id !== "string" || !parsed.id) throw invalidEntry(filePath, lineNumber, "is missing entry id"); + if (parsed.parentId !== null && typeof parsed.parentId !== "string") { + throw invalidEntry(filePath, lineNumber, "has invalid parentId"); + } + if (typeof parsed.timestamp !== "string" || !parsed.timestamp) { + throw invalidEntry(filePath, lineNumber, "is missing timestamp"); + } + if (parsed.type === "leaf" && parsed.targetId !== null && typeof parsed.targetId !== "string") { + throw invalidEntry(filePath, lineNumber, "has invalid targetId"); + } + return parsed as unknown as SessionTreeEntry; +} + +function leafIdAfterEntry(entry: SessionTreeEntry): string | null { + return entry.type === "leaf" ? entry.targetId : entry.id; +} + function headerToSessionMetadata(header: SessionHeader, path: string): JsonlSessionMetadata { return { id: header.id, @@ -53,17 +124,13 @@ export async function loadJsonlSessionMetadata( fs: JsonlSessionStorageFileSystem, filePath: string, ): Promise { - const content = getOrThrow(await fs.readTextFile(filePath)); - for (const line of content.split("\n")) { - if (!line.trim()) break; - try { - const header = JSON.parse(line) as SessionHeader; - return headerToSessionMetadata(header, filePath); - } catch { - throw new Error(`Invalid JSONL session file ${filePath}: first line is not a valid session header`); - } - } - throw new Error(`Invalid JSONL session file ${filePath}: missing session header`); + const lines = getFileSystemResultOrThrow( + await fs.readTextLines(filePath, { maxLines: 1 }), + `Failed to read session header ${filePath}`, + ); + const line = lines[0]; + if (line?.trim()) return headerToSessionMetadata(parseHeaderLine(line, filePath), filePath); + throw invalidSession(filePath, "missing session header"); } async function loadJsonlStorage( @@ -74,29 +141,19 @@ async function loadJsonlStorage( entries: SessionTreeEntry[]; leafId: string | null; }> { - const content = getOrThrow(await fs.readTextFile(filePath)); + const content = getFileSystemResultOrThrow(await fs.readTextFile(filePath), `Failed to read session ${filePath}`); const lines = content.split("\n").filter((line) => line.trim()); if (lines.length === 0) { - throw new Error(`Invalid JSONL session file ${filePath}: missing session header`); - } - - let header: SessionHeader; - try { - header = JSON.parse(lines[0]!) as SessionHeader; - } catch { - throw new Error(`Invalid JSONL session file ${filePath}: first line is not a valid session header`); + throw invalidSession(filePath, "missing session header"); } + const header = parseHeaderLine(lines[0]!, filePath); const entries: SessionTreeEntry[] = []; let leafId: string | null = null; - for (const line of lines.slice(1)) { - try { - const entry = JSON.parse(line) as SessionTreeEntry; - entries.push(entry); - leafId = entry.id; - } catch { - // ignore malformed entry lines - } + for (let i = 1; i < lines.length; i++) { + const entry = parseEntryLine(lines[i]!, filePath, i + 1); + entries.push(entry); + leafId = leafIdAfterEntry(entry); } return { header, entries, leafId }; } @@ -148,7 +205,10 @@ export class JsonlSessionStorage implements SessionStorage cwd: options.cwd, parentSession: options.parentSessionPath, }; - getOrThrow(await fs.writeFile(filePath, `${JSON.stringify(header)}\n`)); + getFileSystemResultOrThrow( + await fs.writeFile(filePath, `${JSON.stringify(header)}\n`), + `Failed to create session ${filePath}`, + ); return new JsonlSessionStorage(fs, filePath, header, [], null); } @@ -157,13 +217,29 @@ export class JsonlSessionStorage implements SessionStorage } async getLeafId(): Promise { + if (this.currentLeafId !== null && !this.byId.has(this.currentLeafId)) { + throw new SessionError("invalid_session", `Entry ${this.currentLeafId} not found`); + } return this.currentLeafId; } async setLeafId(leafId: string | null): Promise { if (leafId !== null && !this.byId.has(leafId)) { - throw new Error(`Entry ${leafId} not found`); + throw new SessionError("not_found", `Entry ${leafId} not found`); } + const entry: LeafEntry = { + type: "leaf", + id: generateEntryId(this.byId), + parentId: this.currentLeafId, + timestamp: new Date().toISOString(), + targetId: leafId, + }; + getFileSystemResultOrThrow( + await this.fs.appendFile(this.filePath, `${JSON.stringify(entry)}\n`), + `Failed to append session leaf ${entry.id}`, + ); + this.entries.push(entry); + this.byId.set(entry.id, entry); this.currentLeafId = leafId; } @@ -172,11 +248,14 @@ export class JsonlSessionStorage implements SessionStorage } async appendEntry(entry: SessionTreeEntry): Promise { - getOrThrow(await this.fs.appendFile(this.filePath, `${JSON.stringify(entry)}\n`)); + getFileSystemResultOrThrow( + await this.fs.appendFile(this.filePath, `${JSON.stringify(entry)}\n`), + `Failed to append session entry ${entry.id}`, + ); this.entries.push(entry); this.byId.set(entry.id, entry); updateLabelCache(this.labelsById, entry); - this.currentLeafId = entry.id; + this.currentLeafId = leafIdAfterEntry(entry); } async getEntry(id: string): Promise { @@ -197,9 +276,13 @@ export class JsonlSessionStorage implements SessionStorage if (leafId === null) return []; const path: SessionTreeEntry[] = []; let current = this.byId.get(leafId); + if (!current) throw new SessionError("not_found", `Entry ${leafId} not found`); while (current) { path.unshift(current); - current = current.parentId ? this.byId.get(current.parentId) : undefined; + if (!current.parentId) break; + const parent = this.byId.get(current.parentId); + if (!parent) throw new SessionError("invalid_session", `Entry ${current.parentId} not found`); + current = parent; } return path; } diff --git a/packages/agent/src/harness/session/memory-repo.ts b/packages/agent/src/harness/session/memory-repo.ts index 85401801..54095514 100644 --- a/packages/agent/src/harness/session/memory-repo.ts +++ b/packages/agent/src/harness/session/memory-repo.ts @@ -1,4 +1,4 @@ -import type { Session, SessionMetadata, SessionRepo } from "../types.js"; +import { type Session, SessionError, type SessionMetadata, type SessionRepo } from "../types.js"; import { InMemorySessionStorage } from "./memory-storage.js"; import { createSessionId, createTimestamp, getEntriesToFork, toSession } from "./repo-utils.js"; @@ -19,7 +19,7 @@ export class InMemorySessionRepo implements SessionRepo> { const session = this.sessions.get(metadata.id); if (!session) { - throw new Error(`Session not found: ${metadata.id}`); + throw new SessionError("not_found", `Session not found: ${metadata.id}`); } return session; } @@ -42,8 +42,7 @@ export class InMemorySessionRepo implements SessionRepo, entry: SessionTreeEntry): void { @@ -27,36 +33,55 @@ function generateEntryId(byId: { has(id: string): boolean }): string { return uuidv7(); } -export class InMemorySessionStorage implements SessionStorage { - private readonly metadata: SessionMetadata; +function leafIdAfterEntry(entry: SessionTreeEntry): string | null { + return entry.type === "leaf" ? entry.targetId : entry.id; +} + +export class InMemorySessionStorage + implements SessionStorage +{ + private readonly metadata: TMetadata; private entries: SessionTreeEntry[]; private byId: Map; private labelsById: Map; private leafId: string | null; - constructor(options?: { entries?: SessionTreeEntry[]; leafId?: string | null; metadata?: SessionMetadata }) { + constructor(options?: { entries?: SessionTreeEntry[]; metadata?: TMetadata }) { this.entries = options?.entries ? [...options.entries] : []; this.byId = new Map(this.entries.map((entry) => [entry.id, entry])); this.labelsById = buildLabelsById(this.entries); - this.leafId = options?.leafId ?? this.entries[this.entries.length - 1]?.id ?? null; + this.leafId = null; + for (const entry of this.entries) this.leafId = leafIdAfterEntry(entry); if (this.leafId !== null && !this.byId.has(this.leafId)) { - throw new Error(`Entry ${this.leafId} not found`); + throw new SessionError("invalid_session", `Entry ${this.leafId} not found`); } - this.metadata = options?.metadata ?? { id: uuidv7(), createdAt: new Date().toISOString() }; + this.metadata = options?.metadata ?? ({ id: uuidv7(), createdAt: new Date().toISOString() } as TMetadata); } - async getMetadata(): Promise { + async getMetadata(): Promise { return this.metadata; } async getLeafId(): Promise { + if (this.leafId !== null && !this.byId.has(this.leafId)) { + throw new SessionError("invalid_session", `Entry ${this.leafId} not found`); + } return this.leafId; } async setLeafId(leafId: string | null): Promise { if (leafId !== null && !this.byId.has(leafId)) { - throw new Error(`Entry ${leafId} not found`); + throw new SessionError("not_found", `Entry ${leafId} not found`); } + const entry: LeafEntry = { + type: "leaf", + id: generateEntryId(this.byId), + parentId: this.leafId, + timestamp: new Date().toISOString(), + targetId: leafId, + }; + this.entries.push(entry); + this.byId.set(entry.id, entry); this.leafId = leafId; } @@ -68,7 +93,7 @@ export class InMemorySessionStorage implements SessionStorage { this.entries.push(entry); this.byId.set(entry.id, entry); updateLabelCache(this.labelsById, entry); - this.leafId = entry.id; + this.leafId = leafIdAfterEntry(entry); } async getEntry(id: string): Promise { @@ -89,9 +114,13 @@ export class InMemorySessionStorage implements SessionStorage { if (leafId === null) return []; const path: SessionTreeEntry[] = []; let current = this.byId.get(leafId); + if (!current) throw new SessionError("not_found", `Entry ${leafId} not found`); while (current) { path.unshift(current); - current = current.parentId ? this.byId.get(current.parentId) : undefined; + if (!current.parentId) break; + const parent = this.byId.get(current.parentId); + if (!parent) throw new SessionError("invalid_session", `Entry ${current.parentId} not found`); + current = parent; } return path; } diff --git a/packages/agent/src/harness/session/repo-utils.ts b/packages/agent/src/harness/session/repo-utils.ts index e842a718..f98b1256 100644 --- a/packages/agent/src/harness/session/repo-utils.ts +++ b/packages/agent/src/harness/session/repo-utils.ts @@ -1,4 +1,11 @@ -import type { SessionMetadata, SessionStorage, SessionTreeEntry } from "../types.js"; +import { + type FileError, + type Result, + SessionError, + type SessionMetadata, + type SessionStorage, + type SessionTreeEntry, +} from "../types.js"; import { Session } from "./session.js"; import { uuidv7 } from "./uuid.js"; @@ -14,6 +21,14 @@ export function toSession(storage: SessionSto return new Session(storage); } +export function getFileSystemResultOrThrow(result: Result, message: string): TValue { + if (!result.ok) { + const code = result.error.code === "not_found" ? "not_found" : "storage"; + throw new SessionError(code, `${message}: ${result.error.message}`, result.error); + } + return result.value; +} + export async function getEntriesToFork( storage: SessionStorage, options: { entryId?: string; position?: "before" | "at" }, @@ -21,14 +36,14 @@ export async function getEntriesToFork( if (!options.entryId) return storage.getEntries(); const target = await storage.getEntry(options.entryId); if (!target) { - throw new Error(`Entry ${options.entryId} not found`); + throw new SessionError("invalid_fork_target", `Entry ${options.entryId} not found`); } let effectiveLeafId: string | null; if ((options.position ?? "before") === "at") { effectiveLeafId = target.id; } else { if (target.type !== "message" || target.message.role !== "user") { - throw new Error(`Entry ${options.entryId} is not a user message`); + throw new SessionError("invalid_fork_target", `Entry ${options.entryId} is not a user message`); } effectiveLeafId = target.parentId; } diff --git a/packages/agent/src/harness/session/session.ts b/packages/agent/src/harness/session/session.ts index f0b42ecf..0b0b3e1e 100644 --- a/packages/agent/src/harness/session/session.ts +++ b/packages/agent/src/harness/session/session.ts @@ -16,6 +16,7 @@ import type { SessionTreeEntry, ThinkingLevelChangeEntry, } from "../types.js"; +import { SessionError } from "../types.js"; export function buildSessionContext(pathEntries: SessionTreeEntry[]): SessionContext { let thinkingLevel = "off"; @@ -206,7 +207,7 @@ export class Session { async appendLabel(targetId: string, label: string | undefined): Promise { if (!(await this.storage.getEntry(targetId))) { - throw new Error(`Entry ${targetId} not found`); + throw new SessionError("not_found", `Entry ${targetId} not found`); } return this.appendTypedEntry({ type: "label", @@ -233,7 +234,7 @@ export class Session { summary?: { summary: string; details?: unknown; fromHook?: boolean }, ): Promise { if (entryId !== null && !(await this.storage.getEntry(entryId))) { - throw new Error(`Entry ${entryId} not found`); + throw new SessionError("not_found", `Entry ${entryId} not found`); } await this.storage.setLeafId(entryId); if (!summary) return undefined; diff --git a/packages/agent/src/harness/skills.ts b/packages/agent/src/harness/skills.ts index d26c8627..7269857e 100644 --- a/packages/agent/src/harness/skills.ts +++ b/packages/agent/src/harness/skills.ts @@ -1,6 +1,6 @@ import ignore from "ignore"; import { parse } from "yaml"; -import { type ExecutionEnv, type FileInfo, getOrUndefined, type Result, type Skill } from "./types.js"; +import { type ExecutionEnv, type FileInfo, type Result, type Skill, toError } from "./types.js"; const MAX_NAME_LENGTH = 64; const MAX_DESCRIPTION_LENGTH = 1024; @@ -8,10 +8,19 @@ const IGNORE_FILE_NAMES = [".gitignore", ".ignore", ".fdignore"]; type IgnoreMatcher = ReturnType; +export type SkillDiagnosticCode = + | "file_info_failed" + | "list_failed" + | "read_failed" + | "parse_failed" + | "invalid_metadata"; + /** Warning produced while loading skills. */ export interface SkillDiagnostic { /** Diagnostic severity. Currently only warnings are emitted. */ type: "warning"; + /** Stable diagnostic code. */ + code: SkillDiagnosticCode; /** Human-readable diagnostic message. */ message: string; /** Path associated with the diagnostic. */ @@ -44,8 +53,20 @@ export async function loadSkills( const skills: Skill[] = []; const diagnostics: SkillDiagnostic[] = []; for (const dir of Array.isArray(dirs) ? dirs : [dirs]) { - const rootInfo = getOrUndefined(await env.fileInfo(dir)); - if (!rootInfo || (await resolveKind(env, rootInfo)) !== "directory") continue; + const rootInfoResult = await env.fileInfo(dir); + if (!rootInfoResult.ok) { + if (rootInfoResult.error.code !== "not_found") { + diagnostics.push({ + type: "warning", + code: "file_info_failed", + message: rootInfoResult.error.message, + path: dir, + }); + } + continue; + } + const rootInfo = rootInfoResult.value; + if ((await resolveKind(env, rootInfo, diagnostics)) !== "directory") continue; const result = await loadSkillsFromDirInternal(env, rootInfo.path, true, ignore(), rootInfo.path); skills.push(...result.skills); diagnostics.push(...result.diagnostics); @@ -89,18 +110,34 @@ async function loadSkillsFromDirInternal( const skills: Skill[] = []; const diagnostics: SkillDiagnostic[] = []; - const dirInfo = getOrUndefined(await env.fileInfo(dir)); - if (!dirInfo || (await resolveKind(env, dirInfo)) !== "directory") return { skills, diagnostics }; + const dirInfoResult = await env.fileInfo(dir); + if (!dirInfoResult.ok) { + if (dirInfoResult.error.code !== "not_found") { + diagnostics.push({ + type: "warning", + code: "file_info_failed", + message: dirInfoResult.error.message, + path: dir, + }); + } + return { skills, diagnostics }; + } + const dirInfo = dirInfoResult.value; + if ((await resolveKind(env, dirInfo, diagnostics)) !== "directory") return { skills, diagnostics }; - await addIgnoreRules(env, ignoreMatcher, dir, rootDir); + await addIgnoreRules(env, ignoreMatcher, dir, rootDir, diagnostics); - const entries = getOrUndefined(await env.listDir(dir)); - if (!entries) return { skills, diagnostics }; + const entriesResult = await env.listDir(dir); + if (!entriesResult.ok) { + diagnostics.push({ type: "warning", code: "list_failed", message: entriesResult.error.message, path: dir }); + return { skills, diagnostics }; + } + const entries = entriesResult.value; for (const entry of entries) { if (entry.name !== "SKILL.md") continue; const fullPath = entry.path; - const kind = await resolveKind(env, entry); + const kind = await resolveKind(env, entry, diagnostics); if (kind !== "file") continue; const relPath = relativeEnvPath(rootDir, fullPath); if (ignoreMatcher.ignores(relPath)) continue; @@ -114,7 +151,7 @@ async function loadSkillsFromDirInternal( for (const entry of entries.sort((a, b) => a.name.localeCompare(b.name))) { if (entry.name.startsWith(".") || entry.name === "node_modules") continue; const fullPath = entry.path; - const kind = await resolveKind(env, entry); + const kind = await resolveKind(env, entry, diagnostics); if (!kind) continue; const relPath = relativeEnvPath(rootDir, fullPath); @@ -137,16 +174,36 @@ async function loadSkillsFromDirInternal( return { skills, diagnostics }; } -async function addIgnoreRules(env: ExecutionEnv, ig: IgnoreMatcher, dir: string, rootDir: string): Promise { +async function addIgnoreRules( + env: ExecutionEnv, + ig: IgnoreMatcher, + dir: string, + rootDir: string, + diagnostics: SkillDiagnostic[], +): Promise { const relativeDir = relativeEnvPath(rootDir, dir); const prefix = relativeDir ? `${relativeDir}/` : ""; for (const filename of IGNORE_FILE_NAMES) { const ignorePath = joinEnvPath(dir, filename); - const info = getOrUndefined(await env.fileInfo(ignorePath)); - if (info?.kind !== "file") continue; + const info = await env.fileInfo(ignorePath); + if (!info.ok) { + if (info.error.code !== "not_found") { + diagnostics.push({ + type: "warning", + code: "file_info_failed", + message: info.error.message, + path: ignorePath, + }); + } + continue; + } + if (info.value.kind !== "file") continue; const content = await env.readTextFile(ignorePath); - if (!content.ok) continue; + if (!content.ok) { + diagnostics.push({ type: "warning", code: "read_failed", message: content.error.message, path: ignorePath }); + continue; + } const patterns = content.value .split(/\r?\n/) .map((line) => prefixIgnorePattern(line, prefix)) @@ -180,13 +237,13 @@ async function loadSkillFromFile( const diagnostics: SkillDiagnostic[] = []; const rawContent = await env.readTextFile(filePath); if (!rawContent.ok) { - diagnostics.push({ type: "warning", message: rawContent.error.message, path: filePath }); + diagnostics.push({ type: "warning", code: "read_failed", message: rawContent.error.message, path: filePath }); return { skill: null, diagnostics }; } const parsed = parseFrontmatter(rawContent.value); if (!parsed.ok) { - diagnostics.push({ type: "warning", message: parsed.error.message, path: filePath }); + diagnostics.push({ type: "warning", code: "parse_failed", message: parsed.error.message, path: filePath }); return { skill: null, diagnostics }; } @@ -196,13 +253,13 @@ async function loadSkillFromFile( const description = typeof frontmatter.description === "string" ? frontmatter.description : undefined; for (const error of validateDescription(description)) { - diagnostics.push({ type: "warning", message: error, path: filePath }); + diagnostics.push({ type: "warning", code: "invalid_metadata", message: error, path: filePath }); } const frontmatterName = typeof frontmatter.name === "string" ? frontmatter.name : undefined; const name = frontmatterName || parentDirName; for (const error of validateName(name, parentDirName)) { - diagnostics.push({ type: "warning", message: error, path: filePath }); + diagnostics.push({ type: "warning", code: "invalid_metadata", message: error, path: filePath }); } if (!description || description.trim() === "") { @@ -255,17 +312,41 @@ function parseFrontmatter>( const body = normalized.slice(endIndex + 4).trim(); return { ok: true, value: { frontmatter: (parse(yamlString) ?? {}) as T, body } }; } catch (error) { - return { ok: false, error: error instanceof Error ? error : new Error(String(error)) }; + return { ok: false, error: toError(error) }; } } -async function resolveKind(env: ExecutionEnv, info: FileInfo): Promise<"file" | "directory" | undefined> { +async function resolveKind( + env: ExecutionEnv, + info: FileInfo, + diagnostics: SkillDiagnostic[], +): Promise<"file" | "directory" | undefined> { if (info.kind === "file" || info.kind === "directory") return info.kind; const canonicalPath = await env.canonicalPath(info.path); - if (!canonicalPath.ok) return undefined; - const target = getOrUndefined(await env.fileInfo(canonicalPath.value)); - if (!target) return undefined; - return target.kind === "file" || target.kind === "directory" ? target.kind : undefined; + if (!canonicalPath.ok) { + if (canonicalPath.error.code !== "not_found") { + diagnostics.push({ + type: "warning", + code: "file_info_failed", + message: canonicalPath.error.message, + path: info.path, + }); + } + return undefined; + } + const target = await env.fileInfo(canonicalPath.value); + if (!target.ok) { + if (target.error.code !== "not_found") { + diagnostics.push({ + type: "warning", + code: "file_info_failed", + message: target.error.message, + path: info.path, + }); + } + return undefined; + } + return target.value.kind === "file" || target.value.kind === "directory" ? target.value.kind : undefined; } function joinEnvPath(base: string, child: string): string { diff --git a/packages/agent/src/harness/types.ts b/packages/agent/src/harness/types.ts index d25f181c..96b60f63 100644 --- a/packages/agent/src/harness/types.ts +++ b/packages/agent/src/harness/types.ts @@ -26,6 +26,17 @@ export function getOrUndefined(result: Result>; /** Read a UTF-8 text file. */ readTextFile(path: string, abortSignal?: AbortSignal): Promise>; + /** Read UTF-8 text lines. Implementations should stop once `maxLines` lines have been read. */ + readTextLines( + path: string, + options?: { maxLines?: number; abortSignal?: AbortSignal }, + ): Promise>; /** Read a binary file. */ readBinaryFile(path: string, abortSignal?: AbortSignal): Promise>; /** Create or overwrite a file, creating parent directories when supported. */ @@ -319,6 +401,11 @@ export interface SessionInfoEntry extends SessionTreeEntryBase { name?: string; } +export interface LeafEntry extends SessionTreeEntryBase { + type: "leaf"; + targetId: string | null; +} + export type SessionTreeEntry = | MessageEntry | ThinkingLevelChangeEntry @@ -328,7 +415,8 @@ export type SessionTreeEntry = | CustomEntry | CustomMessageEntry | LabelEntry - | SessionInfoEntry; + | SessionInfoEntry + | LeafEntry; export interface SessionContext { messages: AgentMessage[]; @@ -350,6 +438,7 @@ export interface JsonlSessionMetadata extends SessionMetadata { export interface SessionStorage { getMetadata(): Promise; getLeafId(): Promise; + /** Persist a leaf entry that records the active session-tree leaf. */ setLeafId(leafId: string | null): Promise; createEntryId(): Promise; appendEntry(entry: SessionTreeEntry): Promise; @@ -688,11 +777,9 @@ export interface GenerateBranchSummaryOptions { } export interface BranchSummaryResult { - summary?: string; - readFiles?: string[]; - modifiedFiles?: string[]; - aborted?: boolean; - error?: string; + summary: string; + readFiles: string[]; + modifiedFiles: string[]; } export interface AgentHarnessOptions< diff --git a/packages/agent/src/harness/utils/shell-output.ts b/packages/agent/src/harness/utils/shell-output.ts index fcb8d538..23f1362c 100644 --- a/packages/agent/src/harness/utils/shell-output.ts +++ b/packages/agent/src/harness/utils/shell-output.ts @@ -1,4 +1,12 @@ -import { type ExecutionEnv, type ExecutionEnvExecOptions, ExecutionError, err, ok, type Result } from "../types.js"; +import { + type ExecutionEnv, + type ExecutionEnvExecOptions, + ExecutionError, + err, + ok, + type Result, + toError, +} from "../types.js"; import { DEFAULT_MAX_BYTES, truncateTail } from "./truncate.js"; export interface ShellCaptureOptions extends Omit { @@ -14,9 +22,9 @@ export interface ShellCaptureResult { } function toExecutionError(error: unknown): ExecutionError { - return error instanceof ExecutionError - ? error - : new ExecutionError("unknown", error instanceof Error ? error.message : String(error), error); + if (error instanceof ExecutionError) return error; + const cause = toError(error); + return new ExecutionError("unknown", cause.message, cause); } export function sanitizeBinaryOutput(str: string): string { diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index 345def15..04275403 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -4,6 +4,9 @@ export * from "./agent.js"; export * from "./agent-loop.js"; export * from "./harness/agent-harness.js"; export { + type BranchPreparation, + type BranchSummaryDetails, + type CollectEntriesResult, collectEntriesForBranchSummary, generateBranchSummary, prepareBranchEntries, diff --git a/packages/agent/test/harness/compaction.test.ts b/packages/agent/test/harness/compaction.test.ts index a9ec08fa..25d740ca 100644 --- a/packages/agent/test/harness/compaction.test.ts +++ b/packages/agent/test/harness/compaction.test.ts @@ -345,7 +345,7 @@ describe("harness compaction", () => { const u3 = createMessageEntry(createUserMessage("user msg 3"), compaction1.id); const a3 = createMessageEntry(createAssistantMessage("assistant msg 3", createMockUsage(8000, 2000)), u3.id); const pathEntries = [u1, a1, u2, a2, compaction1, u3, a3]; - const preparation = prepareCompaction(pathEntries, DEFAULT_COMPACTION_SETTINGS); + const preparation = getOrThrow(prepareCompaction(pathEntries, DEFAULT_COMPACTION_SETTINGS)); expect(preparation).toBeDefined(); expect(preparation?.previousSummary).toBe("First summary"); expect(preparation?.firstKeptEntryId).toBeTruthy(); @@ -365,11 +365,13 @@ describe("harness compaction", () => { }; const u2 = createMessageEntry(createUserMessage("large turn"), compaction1.id); const a2 = createMessageEntry(createAssistantMessage("large assistant message"), u2.id); - const preparation = prepareCompaction([u1, a1, compaction1, u2, a2], { - enabled: true, - reserveTokens: 100, - keepRecentTokens: 1, - }); + const preparation = getOrThrow( + prepareCompaction([u1, a1, compaction1, u2, a2], { + enabled: true, + reserveTokens: 100, + keepRecentTokens: 1, + }), + ); expect(preparation).toMatchObject({ previousSummary: "First summary", isSplitTurn: true }); expect(preparation?.turnPrefixMessages.map((message) => message.role)).toEqual(["user"]); @@ -398,19 +400,21 @@ describe("harness compaction", () => { }; const user = createMessageEntry(createUserMessage("keep"), customMessage.id); const assistant = createMessageEntry(createAssistantMessage("assistant"), user.id); - const preparation = prepareCompaction([branchSummary, customMessage, user, assistant], { - enabled: true, - reserveTokens: 100, - keepRecentTokens: 1, - }); + const preparation = getOrThrow( + prepareCompaction([branchSummary, customMessage, user, assistant], { + enabled: true, + reserveTokens: 100, + keepRecentTokens: 1, + }), + ); expect(preparation?.messagesToSummarize.map((message) => message.role)).toEqual(["branchSummary", "custom"]); }); it("does not prepare compaction when there is nothing valid to compact", () => { const compaction = createCompactionEntry("already compacted", "entry-keep"); - expect(prepareCompaction([compaction], DEFAULT_COMPACTION_SETTINGS)).toBeUndefined(); - expect(prepareCompaction([], DEFAULT_COMPACTION_SETTINGS)).toBeUndefined(); + expect(getOrThrow(prepareCompaction([compaction], DEFAULT_COMPACTION_SETTINGS))).toBeUndefined(); + expect(getOrThrow(prepareCompaction([], DEFAULT_COMPACTION_SETTINGS))).toBeUndefined(); }); it("serializes conversation with truncated tool results", () => { @@ -535,13 +539,6 @@ describe("harness compaction", () => { abortedFaux.setResponses([fauxAssistantMessage("", { stopReason: "aborted", errorMessage: "stopped" })]); const abortedResult = await generateSummary(messages, abortedModel, 2000, "test-key"); expect(abortedResult).toMatchObject({ ok: false, error: { code: "aborted", message: "stopped" } }); - - const missingProviderModel = { ...abortedModel, api: "missing-api" } as Model; - const unknownResult = await generateSummary(messages, missingProviderModel, 2000, "test-key"); - expect(unknownResult).toMatchObject({ - ok: false, - error: { code: "unknown", message: "Summarization request failed" }, - }); }); it("clamps compaction summary maxTokens to the model output cap", async () => { @@ -650,12 +647,6 @@ describe("harness compaction", () => { ok: false, error: { code: "aborted", message: "prefix stopped" }, }); - - const missingProviderModel = { ...abortedModel, api: "missing-api" } as Model; - expect(await compact(preparation, missingProviderModel, "test-key")).toMatchObject({ - ok: false, - error: { code: "unknown", message: "Turn prefix summarization request failed" }, - }); }); it("returns a compaction result with file details", async () => { @@ -667,7 +658,7 @@ describe("harness compaction", () => { const a1 = createMessageEntry(assistantMessage, u1.id); const u2 = createMessageEntry(createUserMessage("continue"), a1.id); const a2 = createMessageEntry(createAssistantMessage("done", createMockUsage(4000, 500)), u2.id); - const preparation = prepareCompaction([u1, a1, u2, a2], DEFAULT_COMPACTION_SETTINGS); + const preparation = getOrThrow(prepareCompaction([u1, a1, u2, a2], DEFAULT_COMPACTION_SETTINGS)); expect(preparation).toBeDefined(); const { faux, model } = createFauxModel(false); faux.setResponses([fauxAssistantMessage("## Goal\nTest summary")]); diff --git a/packages/agent/test/harness/nodejs-env.test.ts b/packages/agent/test/harness/nodejs-env.test.ts index fa3d6aa9..aef098be 100644 --- a/packages/agent/test/harness/nodejs-env.test.ts +++ b/packages/agent/test/harness/nodejs-env.test.ts @@ -27,6 +27,7 @@ describe("NodeExecutionEnv", () => { getOrThrow(await env.writeFile("nested/child/file.txt", "hel")); getOrThrow(await env.appendFile("nested/child/file.txt", "lo")); expect(getOrThrow(await env.readTextFile("nested/child/file.txt"))).toBe("hello"); + expect(getOrThrow(await env.readTextLines("nested/child/file.txt", { maxLines: 1 }))).toEqual(["hello"]); expect(Buffer.from(getOrThrow(await env.readBinaryFile("nested/child/file.txt"))).toString("utf8")).toBe("hello"); const entries = getOrThrow(await env.listDir("nested/child")); @@ -91,6 +92,13 @@ describe("NodeExecutionEnv", () => { ]); }); + it("stops reading text lines at the requested limit", async () => { + const root = createTempDir(); + const env = new NodeExecutionEnv({ cwd: root }); + getOrThrow(await env.writeFile("file.txt", "one\ntwo\nthree")); + expect(getOrThrow(await env.readTextLines("file.txt", { maxLines: 1 }))).toEqual(["one"]); + }); + it("returns FileError for missing paths and keeps exists false for missing paths", async () => { const root = createTempDir(); const env = new NodeExecutionEnv({ cwd: root }); @@ -155,7 +163,7 @@ describe("NodeExecutionEnv", () => { getOrThrow(await env.remove("missing", { force: true })); }); - it("returns aborted results for pre-aborted file operations", async () => { + it("returns aborted results for pre-aborted cancellable file operations", async () => { const root = createTempDir(); const env = new NodeExecutionEnv({ cwd: root }); getOrThrow(await env.writeFile("file.txt", "hello")); @@ -165,17 +173,10 @@ describe("NodeExecutionEnv", () => { const results = await Promise.all([ env.readTextFile("file.txt", signal), + env.readTextLines("file.txt", { abortSignal: signal }), env.readBinaryFile("file.txt", signal), env.writeFile("other.txt", "hello", signal), - env.appendFile("other.txt", "hello", signal), - env.fileInfo("file.txt", signal), env.listDir(".", signal), - env.canonicalPath("file.txt", signal), - env.exists("file.txt", signal), - env.createDir("dir", { abortSignal: signal }), - env.remove("file.txt", { abortSignal: signal }), - env.createTempDir("node-env-test-", signal), - env.createTempFile({ abortSignal: signal }), ]); for (const result of results) { expect(result.ok).toBe(false); @@ -244,7 +245,7 @@ describe("NodeExecutionEnv", () => { }, }); expect(result.ok).toBe(false); - if (!result.ok) expect(result.error).toMatchObject({ code: "unknown", message: "callback failed" }); + if (!result.ok) expect(result.error).toMatchObject({ code: "callback_error", message: "callback failed" }); }); it("returns shell unavailable and spawn errors", async () => { diff --git a/packages/agent/test/harness/session.test.ts b/packages/agent/test/harness/session.test.ts index bfb0a6b3..96510ea4 100644 --- a/packages/agent/test/harness/session.test.ts +++ b/packages/agent/test/harness/session.test.ts @@ -140,10 +140,10 @@ runSessionSuite( const header = JSON.parse(lines[0]!); expect(header.type).toBe("session"); expect(header.version).toBe(3); - for (const line of lines.slice(1)) { - const entry = JSON.parse(line); + const entries = lines.slice(1).map((line) => JSON.parse(line)); + expect(entries.some((entry) => entry.type === "leaf")).toBe(true); + for (const entry of entries) { expect(entry.type).not.toBe("entry"); - expect(entry.type).not.toBe("leaf"); expect(typeof entry.id).toBe("string"); } }, diff --git a/packages/agent/test/harness/skills.test.ts b/packages/agent/test/harness/skills.test.ts index c4d53cbb..41840a3e 100644 --- a/packages/agent/test/harness/skills.test.ts +++ b/packages/agent/test/harness/skills.test.ts @@ -93,6 +93,7 @@ Use this skill. expect(diagnostics).toEqual([ { type: "warning", + code: "invalid_metadata", message: "description is required", path: join(root, "user/broken/SKILL.md"), source: { type: "user" }, diff --git a/packages/agent/test/harness/storage.test.ts b/packages/agent/test/harness/storage.test.ts index 20f7ac8d..ff6a6a95 100644 --- a/packages/agent/test/harness/storage.test.ts +++ b/packages/agent/test/harness/storage.test.ts @@ -4,7 +4,7 @@ import { describe, expect, it } from "vitest"; import { NodeExecutionEnv } from "../../src/harness/env/nodejs.js"; import { JsonlSessionStorage, loadJsonlSessionMetadata } from "../../src/harness/session/jsonl-storage.js"; import { InMemorySessionStorage } from "../../src/harness/session/memory-storage.js"; -import type { MessageEntry, SessionMetadata } from "../../src/harness/types.js"; +import { type MessageEntry, ok, type SessionMetadata } from "../../src/harness/types.js"; import { createAssistantMessage, createTempDir, createUserMessage } from "./session-test-utils.js"; describe("InMemorySessionStorage", () => { @@ -14,7 +14,7 @@ describe("InMemorySessionStorage", () => { expect(await storage.getMetadata()).toEqual(metadata); }); - it("copies initial entries and tracks leaf independently", async () => { + it("copies initial entries and persists leaf changes", async () => { const entry: MessageEntry = { type: "message", id: "entry-1", @@ -29,12 +29,12 @@ describe("InMemorySessionStorage", () => { expect(await storage.getLeafId()).toBe("entry-1"); await storage.setLeafId(null); expect(await storage.getLeafId()).toBeNull(); + expect((await storage.getEntries()).at(-1)).toMatchObject({ type: "leaf", targetId: null }); }); it("rejects invalid leaf ids", async () => { const storage = new InMemorySessionStorage(); await expect(storage.setLeafId("missing")).rejects.toThrow("Entry missing not found"); - expect(() => new InMemorySessionStorage({ leafId: "missing" })).toThrow("Entry missing not found"); }); it("finds entries by type", async () => { @@ -138,7 +138,7 @@ describe("JsonlSessionStorage", () => { await expect(JsonlSessionStorage.open(env, filePath)).rejects.toThrow("first line is not a valid session header"); }); - it("ignores malformed entry lines", async () => { + it("throws for malformed entry lines", async () => { const dir = createTempDir(); const env = new NodeExecutionEnv({ cwd: dir }); const filePath = join(dir, "session.jsonl"); @@ -157,9 +157,7 @@ describe("JsonlSessionStorage", () => { message: createUserMessage("one"), }; writeFileSync(filePath, `${JSON.stringify(header)}\nnot json\n${JSON.stringify(entry)}\n`); - const storage = await JsonlSessionStorage.open(env, filePath); - expect((await storage.getEntries()).map((loadedEntry) => loadedEntry.id)).toEqual(["entry-1"]); - expect(await storage.getLeafId()).toBe("entry-1"); + await expect(JsonlSessionStorage.open(env, filePath)).rejects.toMatchObject({ code: "invalid_entry" }); }); it("creates and reads session metadata from the header", async () => { @@ -211,6 +209,10 @@ describe("JsonlSessionStorage", () => { const loaded = await JsonlSessionStorage.open(env, filePath); expect(await loaded.getLeafId()).toBe("child"); expect((await loaded.getEntries()).map((entry) => entry.id)).toEqual(["root", "child"]); + await loaded.setLeafId("root"); + const reloaded = await JsonlSessionStorage.open(env, filePath); + expect(await reloaded.getLeafId()).toBe("root"); + expect((await reloaded.getEntries()).at(-1)).toMatchObject({ type: "leaf", targetId: "root" }); expect((await loaded.getPathToRoot("child")).map((entry) => entry.id)).toEqual(["root", "child"]); }); @@ -265,9 +267,8 @@ describe("JsonlSessionStorage", () => { expect(await loaded.getLabel("entry-1")).toBeUndefined(); }); - it("reads session metadata from only the first JSONL line", async () => { + it("reads session metadata through the line-reading filesystem operation", async () => { const dir = createTempDir(); - const env = new NodeExecutionEnv({ cwd: dir }); const filePath = join(dir, "session.jsonl"); const header = { type: "session", @@ -276,9 +277,18 @@ describe("JsonlSessionStorage", () => { timestamp: "2026-01-01T00:00:00.000Z", cwd: dir, }; - const malformedSecondLine = "{".repeat(10000); - writeFileSync(filePath, `${JSON.stringify(header)}\n${malformedSecondLine}\n`); - expect(await loadJsonlSessionMetadata(env, filePath)).toEqual({ + const metadata = await loadJsonlSessionMetadata( + { + readTextLines: async () => ok([JSON.stringify(header)]), + readTextFile: async () => { + throw new Error("readTextFile should not be called for metadata"); + }, + writeFile: async () => ok(undefined), + appendFile: async () => ok(undefined), + }, + filePath, + ); + expect(metadata).toEqual({ id: "session-1", createdAt: "2026-01-01T00:00:00.000Z", cwd: dir,