diff --git a/packages/agent/src/agent-loop.ts b/packages/agent/src/agent-loop.ts index c7f02810d..934dae05c 100644 --- a/packages/agent/src/agent-loop.ts +++ b/packages/agent/src/agent-loop.ts @@ -107,17 +107,24 @@ export async function runAgentLoop( signal: AbortSignal | undefined, streamFn: StreamFn, ): Promise { - const newMessages: AgentMessage[] = [...prompts]; + const newMessages: AgentMessage[] = []; const currentContext: AgentContext = { ...context, - messages: [...context.messages, ...prompts], + messages: [...context.messages], }; await emit({ type: "agent_start" }); await emit({ type: "turn_start" }); for (const prompt of prompts) { + if (config.shouldDeliverMessage?.(prompt) === false) continue; await emit({ type: "message_start", message: prompt }); await emit({ type: "message_end", message: prompt }); + currentContext.messages.push(prompt); + newMessages.push(prompt); + } + if (newMessages.length === 0) { + await emit({ type: "agent_end", messages: newMessages }); + return newMessages; } await runLoop(currentContext, newMessages, config, signal, emit, streamFn ?? getDefaultStreamFn()); @@ -188,6 +195,7 @@ async function runLoop( let config = initialConfig; let firstTurn = true; let firstProviderRequest = true; + let continuingFromTerminatingQueue = false; // Check for steering messages at start (user may have typed while waiting) let pendingMessages: AgentMessage[] = (await config.getSteeringMessages?.()) || []; let drainedTerminatingQueue: "steering" | "followUp" | undefined; @@ -204,7 +212,8 @@ async function runLoop( // Outer loop: continues when queued follow-up messages arrive after agent would stop while (true) { - let hasMoreToolCalls = true; + let hasMoreToolCalls = !continuingFromTerminatingQueue; + continuingFromTerminatingQueue = false; // Inner loop: process tool calls and steering messages while (hasMoreToolCalls || pendingMessages.length > 0) { @@ -224,13 +233,18 @@ async function runLoop( // Process pending messages (inject before next assistant response) if (pendingMessages.length > 0) { + let deliveredPendingMessage = false; for (const message of pendingMessages) { + if (config.shouldDeliverMessage?.(message) === false) continue; + deliveredPendingMessage = true; await emit({ type: "message_start", message }); await emit({ type: "message_end", message }); currentContext.messages.push(message); newMessages.push(message); } pendingMessages = []; + if (!deliveredPendingMessage && !hasMoreToolCalls) break; + if (deliveredPendingMessage) hasMoreToolCalls = true; } // Stream assistant response. Continuation-scoped overrides apply to one @@ -381,6 +395,7 @@ async function runLoop( if (followUpMessages.length > 0) { // Set as pending so inner loop processes them pendingMessages = followUpMessages; + continuingFromTerminatingQueue = true; continue; } diff --git a/packages/agent/src/agent.ts b/packages/agent/src/agent.ts index 0511c8d3d..f65f91b1d 100644 --- a/packages/agent/src/agent.ts +++ b/packages/agent/src/agent.ts @@ -173,6 +173,13 @@ class PendingMessageQueue { this.messages = [...messages, ...this.messages]; } + remove(message: AgentMessage): boolean { + const index = this.messages.indexOf(message); + if (index === -1) return false; + this.messages.splice(index, 1); + return true; + } + clear(): void { this.messages = []; this.clearGeneration++; @@ -219,6 +226,7 @@ export class Agent { context: PrepareNextTurnContext, signal?: AbortSignal, ) => Promise | AgentLoopTurnUpdate | undefined; + private messageFilter?: (message: AgentMessage) => boolean; private activeRun?: ActiveRun; /** Session identifier forwarded to providers for cache-aware backends. */ public sessionId?: string; @@ -333,6 +341,16 @@ export class Agent { this.clearFollowUpQueue(); } + /** Remove one exact queued message without disturbing identical siblings. */ + removeQueuedMessage(message: AgentMessage): boolean { + return this.steeringQueue.remove(message) || this.followUpQueue.remove(message); + } + + /** Install a last-moment admission check for prompt and drained queue messages. */ + setMessageFilter(filter: ((message: AgentMessage) => boolean) | undefined): void { + this.messageFilter = filter; + } + /** Returns true when either queue still contains pending messages. */ hasQueuedMessages(): boolean { return this.steeringQueue.hasItems() || this.followUpQueue.hasItems(); @@ -546,6 +564,7 @@ export class Agent { let followUpQueueGeneration = this.followUpQueue.getClearGeneration(); return { model: this._state.model, + shouldDeliverMessage: this.messageFilter, reasoning: this._state.thinkingLevel === "off" ? undefined : this._state.thinkingLevel, sessionId: this.sessionId, onPayload: this.onPayload, diff --git a/packages/agent/src/changes.md b/packages/agent/src/changes.md index d267f3d77..2819bab90 100644 --- a/packages/agent/src/changes.md +++ b/packages/agent/src/changes.md @@ -1,5 +1,12 @@ # Changes +## 2026-07-31 - Exact queued-message cancellation + +### What changed and why + +- `Agent.removeQueuedMessage()` removes one exact steering or follow-up message object without clearing identical siblings. +- Coding-agent extension delivery receipts use this identity-safe primitive to revoke superseded hidden work while preserving unrelated queued user and extension messages. + ## 2026-07-30 - Bound empty Kimi assistant responses ### What changed and why diff --git a/packages/agent/src/types.ts b/packages/agent/src/types.ts index b1d88ca0f..78fd80d13 100644 --- a/packages/agent/src/types.ts +++ b/packages/agent/src/types.ts @@ -145,6 +145,8 @@ export interface PrepareNextTurnContext extends ShouldStopAfterTurnContext {} export interface AgentLoopConfig extends SimpleStreamOptions { model: Model; + /** Last-moment admission check before a queued message enters loop-owned context. */ + shouldDeliverMessage?: (message: AgentMessage) => boolean; /** * Maximum time in milliseconds to wait for the FIRST provider stream event. diff --git a/packages/coding-agent/docs/extensions.md b/packages/coding-agent/docs/extensions.md index 27834b83d..026df22c5 100644 --- a/packages/coding-agent/docs/extensions.md +++ b/packages/coding-agent/docs/extensions.md @@ -1488,7 +1488,7 @@ extension-declared MCP servers. Inject a custom message into the session. Custom messages participate in LLM context. For durable TUI-only content that should not be sent to the LLM, use [`pi.appendEntry()`](#piappendentrycustomtype-data) with [`pi.registerEntryRenderer()`](#piregisterentryrenderercustomtype-renderer). ```typescript -pi.sendMessage({ +const delivery = pi.sendMessage({ customType: "my-extension", content: "Message text", display: true, @@ -1497,6 +1497,13 @@ pi.sendMessage({ triggerTurn: true, deliverAs: "steer", }); + +delivery.onStarted(() => { + // This exact custom message began a model turn. +}); + +// Removes only this pending delivery; returns false after it started or was already cancelled. +delivery.cancel(); ``` **Options:** @@ -1506,6 +1513,10 @@ pi.sendMessage({ - `"nextTurn"` - Queued for next user prompt. Does not interrupt or trigger anything. - `triggerTurn: true` - If agent is idle, trigger an LLM response immediately. Only applies to `"steer"` and `"followUp"` modes (ignored for `"nextTurn"`). +The returned `MessageDelivery` has an opaque `id`, `cancel()`, `onStarted()`, and `onCancelled()`. Cancellation is +identity-based, so equal-content sibling messages remain queued. Clearing the session queue or disposing the session +also cancels pending receipts. + ### pi.sendUserMessage(content, options?) Send a user message to the agent. Unlike `sendMessage()` which sends custom messages, this sends an actual user message that appears as if typed by the user. Always triggers a turn. diff --git a/packages/coding-agent/docs/sdk.md b/packages/coding-agent/docs/sdk.md index 1dc32a14f..8a81224a4 100644 --- a/packages/coding-agent/docs/sdk.md +++ b/packages/coding-agent/docs/sdk.md @@ -73,8 +73,8 @@ interface AgentSession { prompt(text: string, options?: PromptOptions): Promise; // Queue messages during streaming - steer(text: string): Promise; - followUp(text: string): Promise; + steer(text: string, images?: ImageContent[], options?: QueuedInputOptions): Promise; + followUp(text: string, images?: ImageContent[], options?: QueuedInputOptions): Promise; // Subscribe to events (returns unsubscribe function) subscribe(listener: (event: AgentSessionEvent) => void): () => void; @@ -232,6 +232,8 @@ await session.followUp("After you're done, also do this"); ``` Both `steer()` and `followUp()` expand file-based prompt templates but error on extension commands (extension commands cannot be queued). +They also emit the same correlated extension input/disposition lifecycle as queued `prompt()` calls. SDK calls default to +`source: "interactive"`; transport adapters pass `source: "rpc"` when the input came from RPC. ### Agent and AgentState diff --git a/packages/coding-agent/src/changes.md b/packages/coding-agent/src/changes.md index 8397ffa5a..942e46169 100644 --- a/packages/coding-agent/src/changes.md +++ b/packages/coding-agent/src/changes.md @@ -1,3 +1,14 @@ +## Automatic compaction timeout recovery preserves active work (2026-08-01) + +- First-pass automatic compaction timeouts no longer fall back to a context-free marker: the deterministic checkpoint includes bounded recent user intent and the latest active todo state. +- Compaction todo recovery now snapshots only the latest state and correctly restores it when all surviving todo records predate the newest compaction boundary. +- Regression coverage drives both the required automatic fallback handler and the post-compaction todo bridge. + +## Direct input admission and cancellable extension deliveries (2026-07-31) + +- Public `AgentSession.steer()` and `followUp()` now run the same correlated extension input/disposition admission as queued `prompt()` calls. Classic RPC and app-server steering preserve `source: "rpc"`, while compaction queue transfer reuses the same path without duplicate events. +- `pi.sendMessage()` returns an identity-safe delivery receipt with cancellation and started/cancelled subscriptions. Queue clear, disposal, and dispatch failure revoke pending receipts without removing identical sibling messages. +- Direct reload retires the old extension runner only after removed-extension notifications finish, so captured pre-reload APIs are usable during teardown and stale afterward. ## Backfill: injected app-server turns (2026-08-01) ### What changed diff --git a/packages/coding-agent/src/config.ts b/packages/coding-agent/src/config.ts index 77887ff5e..9e45e3163 100644 --- a/packages/coding-agent/src/config.ts +++ b/packages/coding-agent/src/config.ts @@ -304,11 +304,13 @@ function getEntrypointPackageDir(): string | undefined { return undefined; } -function isSelfUpdatePathWritable(): boolean { - const packageDir = getPackageDir(); +export function isSelfUpdatePathWritable( + packageDir = getPackageDir(), + access: (path: string, mode: number) => void = accessSync, +): boolean { try { - accessSync(packageDir, constants.W_OK); - accessSync(dirname(packageDir), constants.W_OK); + access(packageDir, constants.W_OK); + access(dirname(packageDir), constants.W_OK); return true; } catch { return false; diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 4049f26b7..a08d7220d 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -90,6 +90,7 @@ import { type ExtensionToolHookLifecycleEvent, type ExtensionUIContext, type InputSource, + type MessageDelivery, type MessageEndEvent, type MessageStartEvent, type MessageUpdateEvent, @@ -448,6 +449,13 @@ export type QueuedInput = { readonly enqueueOrder: number; }; +export type QueuedInputOptions = { + /** Source of input for extension input event handlers. Defaults to "interactive". */ + readonly source?: InputSource; + /** Recovery-only global order retained across compaction queue transfers. */ + readonly enqueueOrder?: number; +}; + export type ClearedQueue = { steering: string[]; followUp: string[]; @@ -477,6 +485,15 @@ export interface PromptOptions { sessionTitlePrompt?: string | false; } +type MessageDeliveryRecord = { + readonly id: string; + readonly message: CustomMessage; + readonly startedListeners: Set<() => void>; + readonly cancelledListeners: Set<() => void>; + state: "pending" | "started" | "cancelled"; + readonly handle: MessageDelivery; +}; + /** Result from cycleModel() */ export interface ModelCycleResult { model: Model; @@ -580,6 +597,12 @@ export class AgentSession { private _sessionLogger: SessionLogger; /** Messages queued to be included with the next user prompt as context ("asides"). */ private _pendingNextTurnMessages: CustomMessage[] = []; + private _messageDeliveries = new Map(); + private _cancelledMessageDeliveries = new WeakSet(); + private _claimedMessageDeliveries = new Map< + CustomMessage, + { promptMessages: AgentMessage[]; sourceMessages: CustomMessage[] } + >(); // Queues held while the first post-compaction response is classified. Agent // core otherwise drains steering immediately before AgentSession can consume // the stale-usage exemption and schedule the continuation itself. @@ -673,6 +696,12 @@ export class AgentSession { constructor(config: AgentSessionConfig) { this.agent = config.agent; + this.agent.setMessageFilter((message) => { + if (message.role !== "custom") return true; + if (this._cancelledMessageDeliveries.has(message)) return false; + this._startMessageDelivery(message); + return true; + }); this.sessionManager = config.sessionManager; this.settingsManager = config.settingsManager; this.agent.abortServerSideFallback = this.settingsManager.getAbortServerSideFallback(); @@ -1193,6 +1222,14 @@ export class AgentSession { /** Internal handler for agent events - shared by subscribe and reconnect */ private _handleAgentEvent = (event: AgentEvent, signal: AbortSignal): void => { + // A queued custom message is no longer revocable once Agent core begins its + // exact message lifecycle. Mark it before deferring event processing so a + // synchronous core subscriber cannot claim cancellation after the message + // has been moved into the local provider batch. + if (event.type === "message_start" && event.message.role === "custom") { + this._startMessageDelivery(event.message); + } + // Agent core drains native steer/follow-up queues immediately after its // final agent_end. This subscriber intentionally processes its own event // queue asynchronously, so a later recovery rejection cannot abort that @@ -1874,6 +1911,7 @@ export class AgentSession { this.abortSessionTitleGeneration(); this.abortBash(); this.agent.abort(); + this._cancelAllMessageDeliveries(); } catch { // Dispose must succeed even if an abort hook throws. } @@ -2542,6 +2580,10 @@ export class AgentSession { this._pendingNextTurnMessages = []; for (const msg of consumedNextTurnMessages) { messages.push(msg); + this._claimedMessageDeliveries.set(msg, { + promptMessages: messages, + sourceMessages: consumedNextTurnMessages, + }); } // Emit before_agent_start extension event @@ -2584,6 +2626,7 @@ export class AgentSession { } catch (error) { await emitInputDisposition("rejected"); if (consumedNextTurnMessages && consumedNextTurnMessages.length > 0) { + for (const message of consumedNextTurnMessages) this._claimedMessageDeliveries.delete(message); this._pendingNextTurnMessages = [...consumedNextTurnMessages, ...this._pendingNextTurnMessages]; } preflightResult?.(false); @@ -2722,17 +2765,8 @@ export class AgentSession { * @param images Optional image attachments to include with the message * @throws Error if text is an extension command */ - async steer(text: string, images?: ImageContent[], recovery?: { enqueueOrder?: number }): Promise { - // Check for extension commands (cannot be queued) - if (text.startsWith("/")) { - this._throwIfExtensionCommand(text); - } - - // Expand skill commands and prompt templates - let expandedText = this._expandSkillCommand(text); - expandedText = expandPromptTemplate(expandedText, [...this.promptTemplates]); - - await this._queueSteer(expandedText, images, recovery?.enqueueOrder); + async steer(text: string, images?: ImageContent[], options?: QueuedInputOptions): Promise { + await this._admitQueuedInput("steer", text, images, options); } /** @@ -2742,17 +2776,61 @@ export class AgentSession { * @param images Optional image attachments to include with the message * @throws Error if text is an extension command */ - async followUp(text: string, images?: ImageContent[], recovery?: { enqueueOrder?: number }): Promise { + async followUp(text: string, images?: ImageContent[], options?: QueuedInputOptions): Promise { + await this._admitQueuedInput("followUp", text, images, options); + } + + private async _admitQueuedInput( + mode: "steer" | "followUp", + text: string, + images: ImageContent[] | undefined, + options: QueuedInputOptions | undefined, + ): Promise { // Check for extension commands (cannot be queued) if (text.startsWith("/")) { this._throwIfExtensionCommand(text); } - // Expand skill commands and prompt templates - let expandedText = this._expandSkillCommand(text); - expandedText = expandPromptTemplate(expandedText, [...this.promptTemplates]); + let inputId: string | undefined; + const emitDisposition = async (disposition: "handled" | "queued" | "rejected"): Promise => { + if (inputId === undefined) return; + await this._extensionRunner.emit({ type: "input_disposition", inputId, disposition }); + }; + + try { + let currentText = text; + let currentImages = images; + if (this._extensionRunner.hasHandlers("input")) { + inputId = `${this.sessionManager.getSessionId()}:${++this._nextInputId}`; + const inputResult = await this._extensionRunner.emitInput( + currentText, + currentImages, + options?.source ?? "interactive", + mode, + inputId, + ); + if (inputResult.action === "handled") { + await emitDisposition("handled"); + return; + } + if (inputResult.action === "transform") { + currentText = inputResult.text; + currentImages = inputResult.images ?? currentImages; + } + } - await this._queueFollowUp(expandedText, images, recovery?.enqueueOrder); + let expandedText = this._expandSkillCommand(currentText); + expandedText = expandPromptTemplate(expandedText, [...this.promptTemplates]); + if (mode === "followUp") { + await this._queueFollowUp(expandedText, currentImages, options?.enqueueOrder); + } else { + await this._queueSteer(expandedText, currentImages, options?.enqueueOrder); + } + await emitDisposition("queued"); + } catch (error) { + await emitDisposition("rejected"); + throw error; + } } private _startSessionTitleGeneration(firstPrompt: string): void { @@ -2912,7 +2990,14 @@ export class AgentSession { message: Pick, "customType" | "content" | "display" | "details">, options?: { triggerTurn?: boolean; deliverAs?: "steer" | "followUp" | "nextTurn" }, ): Promise { - const appMessage = { + const appMessage = this._createCustomMessage(message); + await this._deliverCustomMessage(appMessage, options); + } + + private _createCustomMessage( + message: Pick, "customType" | "content" | "display" | "details">, + ): CustomMessage { + return { role: "custom" as const, customType: message.customType, // Untyped extensions can pass null/missing content; normalize at ingestion. @@ -2921,6 +3006,27 @@ export class AgentSession { details: message.details, timestamp: Date.now(), } satisfies CustomMessage; + } + + private _sendExtensionMessage( + message: Pick, "customType" | "content" | "display" | "details">, + options: { triggerTurn?: boolean; deliverAs?: "steer" | "followUp" | "nextTurn" } | undefined, + onError: (error: unknown) => void, + ): MessageDelivery { + const appMessage = this._createCustomMessage(message); + const delivery = this._createMessageDelivery(appMessage); + queueMicrotask(() => { + void this._deliverCustomMessage(appMessage, options, delivery).catch(onError); + }); + return delivery.handle; + } + + private async _deliverCustomMessage( + appMessage: CustomMessage, + options?: { triggerTurn?: boolean; deliverAs?: "steer" | "followUp" | "nextTurn" }, + delivery?: MessageDeliveryRecord, + ): Promise { + if (this._isMessageDeliveryCancelled(delivery)) return; const waitForExistingSessionWork = options?.triggerTurn === true && options.deliverAs !== "nextTurn" && @@ -2933,6 +3039,7 @@ export class AgentSession { try { if (waitForExistingSessionWork) { await this._waitForSettledSessionWork(); + if (this._isMessageDeliveryCancelled(delivery)) return; finishSessionWork = this._sessionWorkBarrier.begin(); } @@ -2958,25 +3065,128 @@ export class AgentSession { } } else if (options?.triggerTurn) { await this._enforceCompactionBeforeProvider(this._findLastAssistantMessage(), false, "pre_prompt"); + if (this._isMessageDeliveryCancelled(delivery)) return; await this._enforceFinalProviderAdmission([appMessage]); + if (this._isMessageDeliveryCancelled(delivery)) return; + if (delivery !== undefined) this._startMessageDelivery(appMessage); await this._promptAgent(appMessage); } else { + if (delivery !== undefined) this._startMessageDelivery(appMessage); this.agent.state.messages.push(appMessage); this.sessionManager.appendCustomMessageEntry( - message.customType, - message.content, - message.display, - message.details, + appMessage.customType, + appMessage.content, + appMessage.display, + appMessage.details, ); this._incrementMessageRevision(); this._emit({ type: "message_start", message: appMessage }); this._emit({ type: "message_end", message: appMessage }); } + } catch (error) { + if (delivery !== undefined) this._cancelMessageDelivery(delivery); + throw error; } finally { finishSessionWork?.(); } } + private _isMessageDeliveryCancelled(delivery: MessageDeliveryRecord | undefined): boolean { + return delivery?.state === "cancelled"; + } + + private _createMessageDelivery(message: CustomMessage): MessageDeliveryRecord { + const id = randomUUID(); + let record: MessageDeliveryRecord; + const handle: MessageDelivery = { + id, + cancel: () => this._cancelMessageDelivery(record), + onStarted: (listener) => this._subscribeMessageDelivery(record, "started", listener), + onCancelled: (listener) => this._subscribeMessageDelivery(record, "cancelled", listener), + }; + record = { + id, + message, + startedListeners: new Set<() => void>(), + cancelledListeners: new Set<() => void>(), + state: "pending", + handle, + }; + this._messageDeliveries.set(message, record); + return record; + } + + private _subscribeMessageDelivery( + record: MessageDeliveryRecord, + state: "started" | "cancelled", + listener: () => void, + ): () => void { + if (record.state === state) { + listener(); + return () => {}; + } + if (record.state !== "pending") return () => {}; + const listeners = state === "started" ? record.startedListeners : record.cancelledListeners; + listeners.add(listener); + return () => listeners.delete(listener); + } + + private _startMessageDelivery(message: CustomMessage): void { + this._cancelledMessageDeliveries.delete(message); + this._claimedMessageDeliveries.delete(message); + const record = this._messageDeliveries.get(message); + if (record === undefined || record.state !== "pending") return; + record.state = "started"; + this._messageDeliveries.delete(message); + this._notifyMessageDeliveryListeners(record, record.startedListeners, "started"); + } + + private _cancelMessageDelivery(record: MessageDeliveryRecord): boolean { + if (record.state !== "pending") return false; + const claimed = this._claimedMessageDeliveries.get(record.message); + if (claimed !== undefined) { + const promptIndex = claimed.promptMessages.indexOf(record.message); + if (promptIndex !== -1) claimed.promptMessages.splice(promptIndex, 1); + const sourceIndex = claimed.sourceMessages.indexOf(record.message); + if (sourceIndex !== -1) claimed.sourceMessages.splice(sourceIndex, 1); + this._claimedMessageDeliveries.delete(record.message); + } + record.state = "cancelled"; + this._cancelledMessageDeliveries.add(record.message); + this._messageDeliveries.delete(record.message); + this.agent.removeQueuedMessage(record.message); + const nextTurnIndex = this._pendingNextTurnMessages.indexOf(record.message); + if (nextTurnIndex !== -1) this._pendingNextTurnMessages.splice(nextTurnIndex, 1); + this._notifyMessageDeliveryListeners(record, record.cancelledListeners, "cancelled"); + return true; + } + + private _cancelAllMessageDeliveries(): void { + for (const record of [...this._messageDeliveries.values()]) { + this._cancelMessageDelivery(record); + } + } + + private _notifyMessageDeliveryListeners( + record: MessageDeliveryRecord, + listeners: Set<() => void>, + state: "started" | "cancelled", + ): void { + for (const listener of [...listeners]) { + try { + listener(); + } catch (error) { + this._sessionLogger.warn("extension_message_delivery_listener_failed", { + deliveryId: record.id, + state, + error: error instanceof Error ? error.message : String(error), + }); + } + } + record.startedListeners.clear(); + record.cancelledListeners.clear(); + } + /** * Send a user message to the agent. Always triggers a turn. * When the agent is streaming, use deliverAs to specify how to queue the message. @@ -3097,6 +3307,7 @@ export class AgentSession { this._queuedInputOrder = []; this._postCompactionDeferredSteeringMessages = []; this._postCompactionDeferredFollowUpMessages = []; + this._cancelAllMessageDeliveries(); this.agent.clearAllQueues(); this._emitQueueUpdate(); const cleared = { steering, followUp } as ClearedQueue; @@ -4890,7 +5101,7 @@ export class AgentSession { runner.bindCore( { sendMessage: (message, options) => { - this.sendCustomMessage(message, options).catch((err) => { + return this._sendExtensionMessage(message, options, (err) => { runner.emitError({ extensionPath: RUNTIME_EXTENSION_PATH, event: "send_message", @@ -5303,19 +5514,24 @@ export class AgentSession { includeAllExtensionTools: true, }); } finally { - // An extension removed by this reload must be told even if the rebuild throws - // (e.g. _refreshToolRegistry rejecting an extension's tool metadata): the new - // runner is already installed without it, so nothing else would dispose it. - const newExtensionResolvedPaths = new Set( - this._extensionRunner.getExtensionIdentities().map((extension) => extension.resolvedPath), - ); - const removed = oldExtensionIdentities.filter( - (extension) => !newExtensionResolvedPaths.has(extension.resolvedPath), - ); - if (removed.length > 0) { - await oldExtensionRunner.emit({ type: "session_extensions_removed", reason: "reload", removed }); + const replacementInstalled = this._extensionRunner !== oldExtensionRunner; + try { + // An extension removed by this reload must be told even if the rebuild throws + // (e.g. _refreshToolRegistry rejecting an extension's tool metadata): the new + // runner is already installed without it, so nothing else would dispose it. + const newExtensionResolvedPaths = new Set( + this._extensionRunner.getExtensionIdentities().map((extension) => extension.resolvedPath), + ); + const removed = oldExtensionIdentities.filter( + (extension) => !newExtensionResolvedPaths.has(extension.resolvedPath), + ); + if (removed.length > 0) { + await oldExtensionRunner.emit({ type: "session_extensions_removed", reason: "reload", removed }); + } + } finally { + if (replacementInstalled) oldExtensionRunner.invalidate(); + time("runtime", "reload"); } - time("runtime", "reload"); } const hasBindings = diff --git a/packages/coding-agent/src/core/compaction/compaction.ts b/packages/coding-agent/src/core/compaction/compaction.ts index 94c176213..8e7297b41 100644 --- a/packages/coding-agent/src/core/compaction/compaction.ts +++ b/packages/coding-agent/src/core/compaction/compaction.ts @@ -657,13 +657,15 @@ export async function completeSummarization( const responseStream = Promise.resolve( streamFn ? streamFn(model, context, requestOptions) : streamSimple(model, context, requestOptions), ); - await consumeStreamWithIdleTimeout(responseStream, { + const result = await consumeStreamWithIdleTimeout(responseStream, { idleTimeoutMs: DEFAULT_SUMMARIZATION_IDLE_TIMEOUT_MS, maxDurationMs: DEFAULT_SUMMARIZATION_MAX_DURATION_MS, abort: () => requestController.abort(), signal: callerSignal, + getResult: (stream) => stream.result(), }); - return await (await responseStream).result(); + if (result === undefined) throw callerSignal?.reason ?? new Error("Summarization cancelled"); + return result; } finally { if (callerSignal) callerSignal.removeEventListener("abort", onCallerAbort); } diff --git a/packages/coding-agent/src/core/compaction/stream-watchdog.ts b/packages/coding-agent/src/core/compaction/stream-watchdog.ts index 99cd78e7f..87be577f4 100644 --- a/packages/coding-agent/src/core/compaction/stream-watchdog.ts +++ b/packages/coding-agent/src/core/compaction/stream-watchdog.ts @@ -45,14 +45,18 @@ export const DEFAULT_SUMMARIZATION_IDLE_TIMEOUT_MS = 300_000; */ export const DEFAULT_SUMMARIZATION_MAX_DURATION_MS = 120_000; -export interface ConsumeStreamWithIdleTimeoutOptions { +type StreamEvent> = Stream extends AsyncIterable ? Event : never; + +export interface ConsumeStreamWithIdleTimeoutOptions, Result = unknown> { /** Silence budget per read; the timer resets on every event. */ readonly idleTimeoutMs: number; /** Total wall-clock budget for the whole stream; omit to leave it unbounded. */ readonly maxDurationMs?: number; /** Tear down the underlying request (abort the request-local controller). */ readonly abort: () => void; - readonly onEvent?: (event: T) => void; + readonly onEvent?: (event: StreamEvent) => void; + /** Resolve the stream's final value under the same absolute duration budget. */ + readonly getResult?: (stream: Stream) => PromiseLike; /** Caller cancellation; an abort here ends the wait without an idle error. */ readonly signal?: AbortSignal; } @@ -66,12 +70,21 @@ const CALLER_ABORTED = "caller-aborted" as const; * event arrives within `idleTimeoutMs`. Caller aborts propagate as the * stream's own abort outcome, never masked as an idle timeout. */ -export async function consumeStreamWithIdleTimeout( - stream: AsyncIterable | PromiseLike>, - options: ConsumeStreamWithIdleTimeoutOptions, -): Promise { - const { idleTimeoutMs, maxDurationMs, abort, onEvent, signal } = options; - let iterator: AsyncIterator | undefined; +export function consumeStreamWithIdleTimeout, Result>( + stream: Stream | PromiseLike, + options: ConsumeStreamWithIdleTimeoutOptions & { + readonly getResult: (stream: Stream) => PromiseLike; + }, +): Promise; +export function consumeStreamWithIdleTimeout>( + stream: Stream | PromiseLike, + options: ConsumeStreamWithIdleTimeoutOptions, +): Promise; +export async function consumeStreamWithIdleTimeout, Result>( + stream: Stream | PromiseLike, + options: ConsumeStreamWithIdleTimeoutOptions, +): Promise { + const { idleTimeoutMs, maxDurationMs, abort, onEvent, signal, getResult } = options; let removeAbortListener: (() => void) | undefined; let callerAbortPromise: Promise | undefined; if (signal?.aborted) { @@ -97,11 +110,11 @@ export async function consumeStreamWithIdleTimeout( callerAbortPromise = promise; } try { - let resolvedStream: AsyncIterable; + let resolvedStream: Stream; if (Symbol.asyncIterator in stream) { resolvedStream = stream; } else { - const streamContenders: Array | typeof BUDGET_TRIP | typeof CALLER_ABORTED>> = [ + const streamContenders: Array> = [ Promise.resolve(stream), ]; if (callerAbortPromise) streamContenders.push(callerAbortPromise); @@ -114,18 +127,22 @@ export async function consumeStreamWithIdleTimeout( if (resolution === CALLER_ABORTED) return; resolvedStream = resolution; } - iterator = resolvedStream[Symbol.asyncIterator](); + const iterator = resolvedStream[Symbol.asyncIterator]() as AsyncIterator>; while (true) { const { promise: idlePromise, resolve: resolveIdle } = Promise.withResolvers(); const timer = setTimeout(() => resolveIdle(IDLE_TRIP), idleTimeoutMs); timer.unref?.(); const contenders: Array< - Promise | typeof IDLE_TRIP | typeof BUDGET_TRIP | typeof CALLER_ABORTED> + Promise> | typeof IDLE_TRIP | typeof BUDGET_TRIP | typeof CALLER_ABORTED> > = [iterator.next(), idlePromise]; if (callerAbortPromise) contenders.push(callerAbortPromise); if (budgetPromise) contenders.push(budgetPromise); - let result: IteratorResult | typeof IDLE_TRIP | typeof BUDGET_TRIP | typeof CALLER_ABORTED; + let result: + | IteratorResult> + | typeof IDLE_TRIP + | typeof BUDGET_TRIP + | typeof CALLER_ABORTED; try { result = await Promise.race(contenders); } finally { @@ -133,19 +150,41 @@ export async function consumeStreamWithIdleTimeout( } if (result === IDLE_TRIP) { abort(); - void iterator?.return?.(); + void iterator.return?.(); throw new StreamIdleTimeoutError(idleTimeoutMs); } if (result === BUDGET_TRIP) { abort(); - void iterator?.return?.(); + void iterator.return?.(); throw new StreamDurationBudgetError(budgetMs); } if (result === CALLER_ABORTED) { - void iterator?.return?.(); - return; + void iterator.return?.(); + if (!getResult) return undefined; + const abortedResultContenders: Array> = [ + Promise.resolve(getResult(resolvedStream)), + ]; + if (budgetPromise) abortedResultContenders.push(budgetPromise); + const abortedResult = await Promise.race(abortedResultContenders); + if (abortedResult === BUDGET_TRIP) { + abort(); + throw new StreamDurationBudgetError(budgetMs); + } + return abortedResult; + } + if (result.done) { + if (!getResult) return undefined; + const finalResultContenders: Array> = [ + Promise.resolve(getResult(resolvedStream)), + ]; + if (budgetPromise) finalResultContenders.push(budgetPromise); + const finalResult = await Promise.race(finalResultContenders); + if (finalResult === BUDGET_TRIP) { + abort(); + throw new StreamDurationBudgetError(budgetMs); + } + return finalResult; } - if (result.done) return; onEvent?.(result.value); } } finally { diff --git a/packages/coding-agent/src/core/export-html/index.ts b/packages/coding-agent/src/core/export-html/index.ts index a1060bb7a..3c9d12b8d 100644 --- a/packages/coding-agent/src/core/export-html/index.ts +++ b/packages/coding-agent/src/core/export-html/index.ts @@ -137,6 +137,18 @@ interface SessionData { renderedTools?: Record; } +function sanitizeEntriesForHtmlExport(entries: SessionEntry[]): SessionEntry[] { + return entries.map((entry) => { + if (entry.type === "custom") { + const { data: _data, ...sanitizedEntry } = entry; + return sanitizedEntry; + } + if (entry.type !== "custom_message" || entry.display) return entry; + const { content: _content, details: _details, ...sanitizedEntry } = entry; + return { ...sanitizedEntry, content: "" }; + }); +} + /** * Core HTML generation logic shared by both export functions. */ @@ -157,7 +169,9 @@ function generateHtml(sessionData: SessionData, themeName?: string): string { const infoBg = themeExport.infoBg ?? derivedExportColors.infoBg; // Base64 encode session data to avoid escaping issues - const sessionDataBase64 = Buffer.from(JSON.stringify(sessionData)).toString("base64"); + const sessionDataBase64 = Buffer.from( + JSON.stringify({ ...sessionData, entries: sanitizeEntriesForHtmlExport(sessionData.entries) }), + ).toString("base64"); // Build the CSS with theme variables injected const css = templateCss diff --git a/packages/coding-agent/src/core/extensions/builtin/compaction/changes.md b/packages/coding-agent/src/core/extensions/builtin/compaction/changes.md index 28807b5c9..82d288c48 100644 --- a/packages/coding-agent/src/core/extensions/builtin/compaction/changes.md +++ b/packages/coding-agent/src/core/extensions/builtin/compaction/changes.md @@ -1,5 +1,12 @@ # Builtin compaction extension changes +## Automatic timeout recovery retains current task state (2026-08-01) + +- Required automatic compaction now carries bounded recent user intent plus the latest todo state into its deterministic checkpoint when provider summarization exceeds the wall-clock budget or terminates with a typed transient truncation. +- User intent recovery skips Senpi control envelopes and embeds the recovered intent only in the bounded recovery summary; durable checkpoint details omit task text. +- Todo snapshots persist only the latest state instead of the full historical sequence, and post-compaction restore ignores todo records that exist only before the newest compaction boundary. +- Compaction details remain canonical and bounded: todo/checkpoint objects stay out of durable detail metadata while the user-facing recovery summary contains only the formatted current work items. +- Coverage: `test/compaction/required-compaction-deterministic-fallback.test.ts` and `test/compaction/todo-preservation.test.ts`. ## Blocking compaction route guards (2026-08-01) ### What changed diff --git a/packages/coding-agent/src/core/extensions/builtin/compaction/deterministic-fallback.ts b/packages/coding-agent/src/core/extensions/builtin/compaction/deterministic-fallback.ts index f3b3f0304..bf3aacaf1 100644 --- a/packages/coding-agent/src/core/extensions/builtin/compaction/deterministic-fallback.ts +++ b/packages/coding-agent/src/core/extensions/builtin/compaction/deterministic-fallback.ts @@ -2,7 +2,7 @@ import { type CompactionPreparation, type CompactionResult, estimateContextToken import { StreamDurationBudgetError, StreamIdleTimeoutError } from "../../../compaction/stream-watchdog.ts"; import { buildSessionContext, type CompactionEntry, type SessionEntry } from "../../../session-manager.ts"; import { SummaryRequestError } from "./speculative.ts"; -import { capUtf8Bytes } from "./task-intent.ts"; +import { capUtf8Bytes, sanitizeTaskIntent } from "./task-intent.ts"; export type RequiredCompactionFallbackFailure = "summarization-timeout" | "upstream-stream-truncated"; @@ -19,6 +19,95 @@ interface DeterministicFallbackDetails { retainedSuffix?: "prepared"; } +const TODO_RECOVERY_BYTE_CAP = 8_192; +const USER_INTENT_BYTE_CAP = 4_096; +const USER_MESSAGE_BYTE_CAP = 1_024; +const MAX_RECENT_USER_MESSAGES = 8; + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function latestTodoItems(snapshot: unknown): unknown[] { + if (!isRecord(snapshot) || !Array.isArray(snapshot.todos)) return []; + const todos = snapshot.todos; + for (let index = todos.length - 1; index >= 0; index -= 1) { + const candidate = todos[index]; + if (!isRecord(candidate) || candidate.type !== "custom" || !isRecord(candidate.data)) continue; + if (Array.isArray(candidate.data.phases)) return candidate.data.phases; + if (Array.isArray(candidate.data.todos)) return candidate.data.todos; + } + return todos; +} + +function formatTodoItem(value: unknown): string | undefined { + if (!isRecord(value)) return undefined; + const content = + typeof value.content === "string" + ? value.content.trim() + : typeof value.text === "string" + ? value.text.trim() + : ""; + if (!content) return undefined; + const status = typeof value.status === "string" && value.status.trim() ? value.status.trim() : "pending"; + return `[${status}] ${content}`; +} + +function formatTodoRecovery(snapshot: unknown): string | undefined { + const lines: string[] = []; + for (const item of latestTodoItems(snapshot)) { + if (isRecord(item) && typeof item.name === "string" && Array.isArray(item.tasks)) { + const tasks = item.tasks.flatMap((task) => { + const formatted = formatTodoItem(task); + return formatted ? [`- ${formatted}`] : []; + }); + if (tasks.length > 0) lines.push(`${item.name.trim() || "Tasks"}:`, ...tasks); + continue; + } + const formatted = formatTodoItem(item); + if (formatted) lines.push(`- ${formatted}`); + } + if (lines.length === 0) return undefined; + return capUtf8Bytes(lines.join("\n"), TODO_RECOVERY_BYTE_CAP); +} + +function appendBoundedSection(base: string, heading: string, value: string | undefined, maxBytes: number): string { + if (!value) return base; + const prefix = `\n\n${heading}:\n`; + const availableBytes = Math.max(0, maxBytes - Buffer.byteLength(`${base}${prefix}`)); + if (availableBytes === 0) return base; + const bounded = capUtf8Bytes(value, availableBytes); + return bounded ? `${base}${prefix}${bounded}` : base; +} + +function readUserText(entry: SessionEntry): string | undefined { + if (entry.type !== "message" || entry.message.role !== "user") return undefined; + const content = entry.message.content; + const text = + typeof content === "string" + ? content + : content + .flatMap((part) => (part.type === "text" && typeof part.text === "string" ? [part.text] : [])) + .join("\n"); + const trimmed = text.trim(); + if (!trimmed || trimmed.startsWith("") || trimmed.startsWith("")) { + return undefined; + } + return capUtf8Bytes(sanitizeTaskIntent(trimmed), USER_MESSAGE_BYTE_CAP); +} + +function resolveDroppedUserIntent(branchEntries: SessionEntry[], firstKeptEntryId: string): string | undefined { + const firstKeptIndex = branchEntries.findIndex((entry) => entry.id === firstKeptEntryId); + if (firstKeptIndex <= 0) return undefined; + const recent: string[] = []; + for (let index = firstKeptIndex - 1; index >= 0 && recent.length < MAX_RECENT_USER_MESSAGES; index -= 1) { + const text = readUserText(branchEntries[index]!); + if (text) recent.push(text); + } + if (recent.length === 0) return undefined; + return capUtf8Bytes(recent.reverse().join("\n\n"), USER_INTENT_BYTE_CAP); +} + export function classifyRequiredCompactionFallbackFailure( error: unknown, ): RequiredCompactionFallbackFailure | undefined { @@ -47,9 +136,18 @@ export function createRequiredCompactionFallback( "Generated summarization did not complete, so older context was reduced without another provider request.", "Continue from the retained messages after this checkpoint. Treat omitted transcript details as unknown.", ].join("\n"); - const taskIntent = metadata.taskIntent?.trim(); - const fixedText = taskIntent ? `${marker}\n\nTask intent:\n${taskIntent}` : marker; + const explicitTaskIntent = metadata.taskIntent?.trim(); + const taskIntent = explicitTaskIntent + ? capUtf8Bytes(sanitizeTaskIntent(explicitTaskIntent), USER_INTENT_BYTE_CAP) + : resolveDroppedUserIntent(branchEntries, preparation.firstKeptEntryId); const maxSummaryBytes = Math.max(1_024, Math.floor(contextWindow * 0.4)); + let fixedText = appendBoundedSection(marker, "Task intent", taskIntent, maxSummaryBytes); + fixedText = appendBoundedSection( + fixedText, + "Current todo state", + formatTodoRecovery(metadata.todoSnapshot), + maxSummaryBytes, + ); const previousSummary = preparation.previousSummary?.trim(); let summary = fixedText; if (previousSummary) { @@ -66,7 +164,6 @@ export function createRequiredCompactionFallback( schema: "senpi.compaction.deterministic-fallback.v1", origin: "required-compaction-recovery", failureKind, - ...(taskIntent ? { taskIntent } : {}), }; const result: CompactionResult = { summary, diff --git a/packages/coding-agent/src/core/extensions/builtin/compaction/index.ts b/packages/coding-agent/src/core/extensions/builtin/compaction/index.ts index f79dd2f40..c86a15925 100644 --- a/packages/coding-agent/src/core/extensions/builtin/compaction/index.ts +++ b/packages/coding-agent/src/core/extensions/builtin/compaction/index.ts @@ -10,6 +10,8 @@ import type { SessionBeforeCompactEvent, SessionCompactEvent, } from "../../types.ts"; +import { readGoal } from "../goal/store.ts"; +import { goalStoreRef } from "../goal/store-ref.ts"; import * as checkpointState from "./checkpoint-state.ts"; import * as breaker from "./circuit-breaker.ts"; import { @@ -55,6 +57,7 @@ import { runExtensionCompaction, type SpeculativeCompactionResult, type SpeculativeCompactionSnapshot, + type SummarizationWatchdogBudget, SummaryGenerationError, } from "./speculative.ts"; import { type CompactionExtensionState, createInitialState, resetTurnCounter } from "./state.ts"; @@ -64,6 +67,9 @@ import { isTransientSummarizationFailure } from "./transient-failure.ts"; import { isIneffectiveCompaction } from "./yield.ts"; const DEFAULT_CONTEXT_WINDOW = 200_000; +export interface CompactionExtensionDependencies extends OpenAiRemoteCompactionDependencies { + summarizationWatchdog?: SummarizationWatchdogBudget; +} const EMERGENCY_COMPACTION_INSTRUCTIONS = "EMERGENCY: hard context limit reached. Produce an aggressive recovery summary that preserves current goal, constraints, files touched, tool outcomes, and exact next steps. Prefer concise factual state over transcript detail."; const PROACTIVE_COMPACTION_INSTRUCTIONS = "Proactively compact before the next agent turn."; @@ -74,6 +80,7 @@ const MAX_OUTPUT_RESERVE_RATIO = 0.5; interface PendingCompactionMetadata { checkpoint: checkpointState.AgentCheckpoint; todoSnapshot: todoBridge.TodoSnapshotPayload; + taskIntent?: string; } function approxTokens(text: string): number { @@ -167,7 +174,7 @@ function createBlockingRemoteCompactionEvent( export default function compactionExtension( pi: ExtensionAPI, - remoteCompactionDependencies: OpenAiRemoteCompactionDependencies = {}, + remoteCompactionDependencies: CompactionExtensionDependencies = {}, ): void { let state: CompactionExtensionState = createInitialState(); const emergencyPruneLatch = createEmergencyPruneLatch(); @@ -189,7 +196,8 @@ export default function compactionExtension( interface CompactionContext extends ExtensionContext { agentDir?: string; } - const getLogger = (ctx: CompactionContext): CompactionLogger => (logger ??= createCompactionLogger(ctx.agentDir)); + const getLogger = (ctx: CompactionContext): CompactionLogger => + (logger ??= createCompactionLogger(ctx.getLoadedHookSources?.().agentDir ?? ctx.agentDir)); function getSummarizationTools(): Tool[] { if (typeof pi.getAllTools !== "function" || typeof pi.getActiveTools !== "function") return []; @@ -224,7 +232,13 @@ export default function compactionExtension( if (!snapshot) return; getLogger(ctx).debug("speculative_started", { generation, origin: "speculative" }); const controller = new AbortController(); - const settled = runExtensionCompaction(ctx, snapshot, controller.signal).then( + const settled = runExtensionCompaction( + ctx, + snapshot, + controller.signal, + undefined, + remoteCompactionDependencies.summarizationWatchdog, + ).then( (result) => ({ result, error: undefined }), (error: unknown) => ({ result: undefined, error: error instanceof Error ? error : new Error(String(error)) }), ); @@ -233,10 +247,14 @@ export default function compactionExtension( speculativeJob = { generation, snapshot, controller, promise, failure }; } - function capturePendingMetadata(requestId: string, ctx: ExtensionContext): void { + async function capturePendingMetadata(requestId: string, ctx: ExtensionContext): Promise { + const ref = goalStoreRef(ctx.sessionManager, ctx.cwd); + const goal = await readGoal(ref); + const taskIntent = goal === null || goal.status === "complete" ? undefined : goal.objective; pendingMetadata.set(requestId, { checkpoint: checkpointState.captureAgentCheckpoint(pi, ctx), todoSnapshot: todoBridge.createTodoSnapshot(ctx), + ...(taskIntent === undefined ? {} : { taskIntent }), }); while (pendingMetadata.size > MAX_PENDING_METADATA) { const oldestRequestId = pendingMetadata.keys().next().value; @@ -372,12 +390,17 @@ export default function compactionExtension( } let compaction: CompactionResult | undefined; try { - compaction = await runExtensionCompaction(ctx, snapshot, feedbackSignal, (delta) => - ctx.updateCompaction?.({ - reason: "extension", - signal: feedbackSignal, - delta, - }), + compaction = await runExtensionCompaction( + ctx, + snapshot, + feedbackSignal, + (delta) => + ctx.updateCompaction?.({ + reason: "extension", + signal: feedbackSignal, + delta, + }), + remoteCompactionDependencies.summarizationWatchdog, ); } catch (error) { if (!(error instanceof SummaryGenerationError)) throw error; @@ -430,7 +453,7 @@ export default function compactionExtension( }; } - capturePendingMetadata(event.requestId, ctx); + await capturePendingMetadata(event.requestId, ctx); const model = ctx.model; if (!model) return undefined; @@ -460,18 +483,27 @@ export default function compactionExtension( }; let compaction: CompactionResult | undefined; try { - compaction = await runExtensionCompaction(ctx, snapshot, event.signal, (delta) => - ctx.updateCompaction?.({ reason: event.reason, signal: event.signal, delta }), + compaction = await runExtensionCompaction( + ctx, + snapshot, + event.signal, + (delta) => ctx.updateCompaction?.({ reason: event.reason, signal: event.signal, delta }), + remoteCompactionDependencies.summarizationWatchdog, ); } catch (error) { const message = error instanceof Error ? error.message : String(error); const failureKind = classifyRequiredCompactionFallbackFailure(error); if (isRequiredCompactionFallbackReason(event.reason) && failureKind !== undefined && !event.signal.aborted) { + const recoveryMetadata = pendingMetadata.get(event.requestId); const fallback = createRequiredCompactionFallback( snapshot.preparation, snapshot.contextWindow, failureKind, - { taskIntent: resolveInheritedTaskIntent(event.branchEntries) }, + { + taskIntent: recoveryMetadata?.taskIntent ?? resolveInheritedTaskIntent(event.branchEntries), + todoSnapshot: recoveryMetadata?.todoSnapshot, + checkpoint: recoveryMetadata?.checkpoint, + }, event.branchEntries, ); if (fallback) return { compaction: fallback }; @@ -539,8 +571,36 @@ export default function compactionExtension( state = cap.incrementAccepted(state); state = breaker.recordSuccess(state); const details = compactEvent.compactionEntry.details as - | { structuralYield?: { savedTokens: number; savingsRatio: number } } + | { + schema?: string; + origin?: string; + failureKind?: string; + taskIntent?: string; + structuralYield?: { savedTokens: number; savingsRatio: number }; + } | undefined; + if ( + details?.schema === "senpi.compaction.deterministic-fallback.v1" && + details.origin === "required-compaction-recovery" + ) { + ctx.ui.notify( + "Automatic compaction used a local recovery checkpoint because summarization did not finish; older context may be incomplete.", + "warning", + ); + getLogger(ctx).info("deterministic_fallback_applied", { + origin: details.origin, + route: "session_before_compact", + reason: compactEvent.reason, + requestId: compactEvent.requestId, + failureKind: details.failureKind, + contextWindow: + ctx.getContextUsage()?.contextWindow ?? ctx.model?.contextWindow ?? DEFAULT_CONTEXT_WINDOW, + tokensBefore: compactEvent.compactionEntry.tokensBefore, + retainedEntryCount: keptEntries.length, + summaryBytes: Buffer.byteLength(compactEvent.compactionEntry.summary), + hasTaskIntent: details.taskIntent !== undefined, + }); + } const sy = details?.structuralYield; if (sy && typeof sy.savedTokens === "number" && typeof sy.savingsRatio === "number") { state = { diff --git a/packages/coding-agent/src/core/extensions/builtin/compaction/log.ts b/packages/coding-agent/src/core/extensions/builtin/compaction/log.ts index f8fabe884..1ccaf1b64 100644 --- a/packages/coding-agent/src/core/extensions/builtin/compaction/log.ts +++ b/packages/coding-agent/src/core/extensions/builtin/compaction/log.ts @@ -18,6 +18,11 @@ const ALLOWED_KEYS = new Set([ "remainingSec", "count", "durationMs", + "failureKind", + "retainedEntryCount", + "todoItemCount", + "summaryBytes", + "hasTaskIntent", ]); const DEBUG_PREFIX = "[senpi-compaction]"; const EVENTS = new Set([ @@ -35,6 +40,7 @@ const EVENTS = new Set([ "emergency_prune", "ineffective_counted", "summary_failed", + "deterministic_fallback_applied", ]); export type CompactionLoggerEvent = @@ -52,7 +58,8 @@ export type CompactionLoggerEvent = | "emergency_prune" | "ineffective_counted" | "idle_trigger" - | "summary_failed"; + | "summary_failed" + | "deterministic_fallback_applied"; export interface CompactionLoggerData { origin?: string; @@ -70,6 +77,11 @@ export interface CompactionLoggerData { remainingSec?: number; count?: number; durationMs?: number; + failureKind?: string; + retainedEntryCount?: number; + todoItemCount?: number; + summaryBytes?: number; + hasTaskIntent?: boolean; } export interface CompactionLogger { diff --git a/packages/coding-agent/src/core/extensions/builtin/compaction/speculative.ts b/packages/coding-agent/src/core/extensions/builtin/compaction/speculative.ts index 3e7220b82..6e9f6b1b3 100644 --- a/packages/coding-agent/src/core/extensions/builtin/compaction/speculative.ts +++ b/packages/coding-agent/src/core/extensions/builtin/compaction/speculative.ts @@ -68,6 +68,11 @@ export interface SpeculativeCompactionContext { ): Promise; } +export interface SummarizationWatchdogBudget { + readonly idleTimeoutMs: number; + readonly maxDurationMs: number; +} + export interface SpeculativeCompactionSnapshot { generation: number; expectedRevision: number; @@ -219,6 +224,7 @@ async function generateSummaryMessage(options: { headers?: Record; extraBody?: Record; }; + watchdogBudget?: SummarizationWatchdogBudget; }): Promise { // Send the conversation as native LLM messages with the summarization // instruction as a trailing user message, mirroring normal agent traffic. @@ -264,18 +270,18 @@ async function generateSummaryMessage(options: { signal: requestController.signal, ...summarizationReasoningOptions(options.snapshot.model), }); - await consumeStreamWithIdleTimeout(responseStream, { - idleTimeoutMs: DEFAULT_SUMMARIZATION_IDLE_TIMEOUT_MS, - maxDurationMs: DEFAULT_SUMMARIZATION_MAX_DURATION_MS, + return await consumeStreamWithIdleTimeout(responseStream, { + idleTimeoutMs: options.watchdogBudget?.idleTimeoutMs ?? DEFAULT_SUMMARIZATION_IDLE_TIMEOUT_MS, + maxDurationMs: options.watchdogBudget?.maxDurationMs ?? DEFAULT_SUMMARIZATION_MAX_DURATION_MS, abort: () => requestController.abort(), signal: options.signal, + getResult: (stream) => stream.result(), onEvent: (event) => { if (event.type === "text_delta" && event.delta) { options.onProgress?.(event.delta); } }, }); - return await responseStream.result(); } finally { if (options.signal) options.signal.removeEventListener("abort", onCallerAbort); } @@ -505,6 +511,7 @@ export async function runExtensionCompaction( snapshot: SpeculativeCompactionSnapshot, signal?: AbortSignal, onProgress?: CompactionProgressCallback, + watchdogBudget?: SummarizationWatchdogBudget, ): Promise { if (signal?.aborted) return undefined; const auth = await context.modelRegistry?.getApiKeyAndHeaders(snapshot.model); @@ -544,6 +551,7 @@ export async function runExtensionCompaction( headers: auth.headers, extraBody: auth.extraBody, }, + watchdogBudget, }); if (!response) return undefined; diff --git a/packages/coding-agent/src/core/extensions/builtin/compaction/todo-bridge.ts b/packages/coding-agent/src/core/extensions/builtin/compaction/todo-bridge.ts index f6fa35306..c9ecc0921 100644 --- a/packages/coding-agent/src/core/extensions/builtin/compaction/todo-bridge.ts +++ b/packages/coding-agent/src/core/extensions/builtin/compaction/todo-bridge.ts @@ -81,8 +81,27 @@ function readTodosFromEntry(entry: CustomEntry): TodoEntry[] { return []; } +function readTodoSnapshotItems(entry: SessionEntry | undefined): TodoSnapshotItems { + if (entry?.type !== "custom" || !isRecord(entry.data)) return []; + if (Array.isArray(entry.data.phases)) return entry.data.phases.filter(isTodoPhase); + if (Array.isArray(entry.data.todos)) return entry.data.todos.filter(isTodoEntry); + return []; +} + +function hasTodoStateAfterLatestCompaction(ctx: ExtensionContext): boolean { + const entries = ctx.sessionManager.getBranch(); + let searchFrom = 0; + for (let index = entries.length - 1; index >= 0; index -= 1) { + if (entries[index]?.type === "compaction") { + searchFrom = index + 1; + break; + } + } + return entries.slice(searchFrom).some(isCustomTodoEntry); +} + function findLatestTodoSnapshot(ctx: ExtensionContext): TodoSnapshotPayload | null { - const entries = ctx.sessionManager.getEntries(); + const entries = ctx.sessionManager.getBranch(); for (let index = entries.length - 1; index >= 0; index -= 1) { const entry = entries[index]; if (entry.type !== "custom" || entry.customType !== TODO_SNAPSHOT_CUSTOM_TYPE) continue; @@ -107,13 +126,14 @@ export function findTodoEntries( .flatMap(readTodosFromEntry); } - return ctxOrEntries.sessionManager.getEntries().filter(isCustomTodoEntry); + return ctxOrEntries.sessionManager.getBranch().filter(isCustomTodoEntry); } export function createTodoSnapshot(ctx: ExtensionContext): TodoSnapshotPayload { + const latestEntry = findTodoEntries(ctx).at(-1); return { schema: TODO_SNAPSHOT_SCHEMA, - todos: findTodoEntries(ctx), + todos: readTodoSnapshotItems(latestEntry), capturedAt: Date.now(), }; } @@ -162,7 +182,7 @@ export function restoreTodosIfMissing( const pi = piOrSnapshot as SendMessageTarget; const ctx = ctxOrCurrentTodos as ExtensionContext; - if (findTodoEntries(ctx).length > 0) return; + if (hasTodoStateAfterLatestCompaction(ctx)) return; const snapshot = findLatestTodoSnapshot(ctx); if (!snapshot || snapshot.todos.length === 0) return; diff --git a/packages/coding-agent/src/core/extensions/builtin/config-reload/log.ts b/packages/coding-agent/src/core/extensions/builtin/config-reload/log.ts index f372fcdc6..ace461d23 100644 --- a/packages/coding-agent/src/core/extensions/builtin/config-reload/log.ts +++ b/packages/coding-agent/src/core/extensions/builtin/config-reload/log.ts @@ -68,6 +68,7 @@ export interface ConfigReloadLogger { export interface ConfigReloadLoggerOptions { maxBytes?: number; + writeLine?: (filePath: string, line: string, maxBytes: number) => void; } export function createConfigReloadLogger( @@ -76,6 +77,7 @@ export function createConfigReloadLogger( ): ConfigReloadLogger { const filePath = join(agentDir, "logs", "config-reload.log"); const maxBytes = validMaxBytes(options.maxBytes); + const write = options.writeLine ?? writeLine; let disabled = false; function log( @@ -85,7 +87,7 @@ export function createConfigReloadLogger( ): ConfigReloadLogStatus { if (disabled) return { written: false, disabled: true }; try { - writeLine(filePath, JSON.stringify(formatEntry(level, event, details)), maxBytes); + write(filePath, JSON.stringify(formatEntry(level, event, details)), maxBytes); return { written: true, disabled: false }; } catch { disabled = true; diff --git a/packages/coding-agent/src/core/extensions/builtin/goal/changes.md b/packages/coding-agent/src/core/extensions/builtin/goal/changes.md index 048cb8b14..ae4575f2b 100644 --- a/packages/coding-agent/src/core/extensions/builtin/goal/changes.md +++ b/packages/coding-agent/src/core/extensions/builtin/goal/changes.md @@ -1,5 +1,18 @@ # goal Extension Changes +## Serialized continuation admission and full-objective recovery (2026-07-31) + +### What changed + +- Every Goal store mutation is serialized by canonical goal-file path. Continuation delivery and guard denial use Goal ID, status, count, and signature expectations, preventing stale snapshots and concurrent writers from mutating or overwriting a replacement Goal. +- Hidden continuations retain their exact `pi.sendMessage()` delivery receipt. Accepted newer direct input and queue clear cancel only that pending continuation; unrelated agent starts no longer consume a boolean latch. +- Direct `steer`/`followUp`, classic RPC, app-server steering, and compaction-transferred queues now reach the same accepted-input Goal recovery lifecycle. +- Long objectives remain truncated in persisted/UI Goal state, while validated matching sidecars are deterministically restored for startup and monitor continuation prompts after compaction. + +### Verification + +- Regression coverage pins direct RPC-style queue recovery, concurrent mutation preservation and CAS, exact delivery cancellation, reload invalidation, queue clear, and full-objective startup/monitor reinjection. + ## Legacy `pi-goal` state is imported once at session start (2026-07-31) ### What changed diff --git a/packages/coding-agent/src/core/extensions/builtin/goal/direct-input-lifecycle.ts b/packages/coding-agent/src/core/extensions/builtin/goal/direct-input-lifecycle.ts index 22e123ddc..d954fa68e 100644 --- a/packages/coding-agent/src/core/extensions/builtin/goal/direct-input-lifecycle.ts +++ b/packages/coding-agent/src/core/extensions/builtin/goal/direct-input-lifecycle.ts @@ -11,6 +11,7 @@ type DirectInputCandidate = { type DirectInputLifecycleDependencies = { readonly monitor: MonitorAwareGoalContinuation; readonly goalStoreRef: (ctx: ExtensionContext) => GoalStoreRef; + readonly cancelPendingContinuation: () => void; readonly beginAgentGoalAccounting: (goal: Goal) => void; readonly refreshGoalUi: (ctx: ExtensionContext, goal: Goal) => void; }; @@ -43,22 +44,27 @@ export class GoalDirectInputLifecycle { this.#candidates.delete(event.inputId); const accepted = event.disposition === "started" || event.disposition === "queued"; this.#dependencies.monitor.resolveDirectInput(event.inputId, accepted); - if (!accepted || candidate.goalId === null) return; + if (!accepted) return; + this.#dependencies.cancelPendingContinuation(); + if (candidate.goalId === null) return; const ref = this.#dependencies.goalStoreRef(ctx); const currentGoal = await readGoal(ref); if (currentGoal?.id !== candidate.goalId) return; if (currentGoal.status === "blocked" && isMechanicalContinuationBlock(currentGoal.blockedReason)) { - await resetContinuationStreak(ref); - const reactivated = await updateGoal(ref, { status: "active" }, "user"); + const reactivated = await updateGoal(ref, { status: "active" }, "user", { + id: currentGoal.id, + status: "blocked", + }); + if (reactivated === null) return; this.#dependencies.beginAgentGoalAccounting(reactivated); this.#dependencies.refreshGoalUi(ctx, reactivated); return; } if (currentGoal.status !== "active") return; - const reset = await resetContinuationStreak(ref); + const reset = await resetContinuationStreak(ref, { id: currentGoal.id, status: "active" }); if (reset !== null) this.#dependencies.refreshGoalUi(ctx, reset); } } diff --git a/packages/coding-agent/src/core/extensions/builtin/goal/index.ts b/packages/coding-agent/src/core/extensions/builtin/goal/index.ts index 9d1a82291..5f461a75a 100644 --- a/packages/coding-agent/src/core/extensions/builtin/goal/index.ts +++ b/packages/coding-agent/src/core/extensions/builtin/goal/index.ts @@ -1,6 +1,6 @@ import { GOAL_CONTINUATION_MESSAGE_TYPE } from "../../../messages.ts"; import type { SessionEntry } from "../../../session-manager.ts"; -import type { AgentEndEvent, ExtensionAPI, ExtensionContext } from "../../types.ts"; +import type { AgentEndEvent, ExtensionAPI, ExtensionContext, MessageDelivery } from "../../types.ts"; import { GOAL_CACHE_WARMUP_ENTRY_TYPE } from "./cache-warm.ts"; import { renderGoalCacheWarmupEntry } from "./cache-warm-renderer.ts"; import { registerGoalCommand } from "./command-registration.ts"; @@ -34,18 +34,31 @@ export default function goalExtension(pi: ExtensionAPI): void { let agentGoalAccounting: AgentGoalAccounting | null = null; let blockedThisTurnGoalId: string | null = null; let completedThisTurnGoalId: string | null = null; - let continuationPending = false; + let pendingContinuation: MessageDelivery | undefined; const turnUsage = new TurnUsageTracker(); - const monitorContinuation = new MonitorAwareGoalContinuation( + let monitorContinuation: MonitorAwareGoalContinuation; + const markContinuationPending = (delivery: MessageDelivery): void => { + pendingContinuation = delivery; + delivery.onStarted(() => { + if (pendingContinuation !== delivery) return; + pendingContinuation = undefined; + monitorContinuation.noteContinuationStarted(); + }); + delivery.onCancelled(() => { + if (pendingContinuation === delivery) pendingContinuation = undefined; + }); + }; + monitorContinuation = new MonitorAwareGoalContinuation( pi, - () => continuationPending, - () => { - continuationPending = true; - }, + () => pendingContinuation !== undefined, + markContinuationPending, ); const directInputLifecycle = new GoalDirectInputLifecycle({ monitor: monitorContinuation, goalStoreRef, + cancelPendingContinuation: () => { + pendingContinuation?.cancel(); + }, beginAgentGoalAccounting, refreshGoalUi: refreshGoalUiBestEffort, }); @@ -143,9 +156,6 @@ export default function goalExtension(pi: ExtensionAPI): void { }); pi.on("agent_start", async (_event, ctx) => { - const continuationStarted = continuationPending; - continuationPending = false; - if (continuationStarted) monitorContinuation.noteContinuationStarted(); agentTurnInProgress = true; blockedThisTurnGoalId = null; completedThisTurnGoalId = null; @@ -226,6 +236,8 @@ export default function goalExtension(pi: ExtensionAPI): void { } clearAgentGoalAccounting(); goalTicker.stop(); + pendingContinuation?.cancel(); + pendingContinuation = undefined; monitorContinuation.dispose(); }); @@ -259,10 +271,8 @@ export default function goalExtension(pi: ExtensionAPI): void { goal: Goal, ): Promise { const continuedGoal = await queueGoalContinuation(extensionApi, ctx, goal, { - continuationPending, - markContinuationPending: () => { - continuationPending = true; - }, + continuationPending: pendingContinuation !== undefined, + markContinuationPending, }); if (continuedGoal.status === goal.status) return; if (continuedGoal.status === "active") beginAgentGoalAccounting(continuedGoal); diff --git a/packages/coding-agent/src/core/extensions/builtin/goal/lifecycle-helpers.ts b/packages/coding-agent/src/core/extensions/builtin/goal/lifecycle-helpers.ts index 5751025f9..8db4daacb 100644 --- a/packages/coding-agent/src/core/extensions/builtin/goal/lifecycle-helpers.ts +++ b/packages/coding-agent/src/core/extensions/builtin/goal/lifecycle-helpers.ts @@ -1,7 +1,7 @@ import type { AgentMessage } from "@earendil-works/pi-agent-core"; import { GOAL_CONTINUATION_MESSAGE_TYPE } from "../../../messages.ts"; import type { SessionEntry } from "../../../session-manager.ts"; -import type { ExtensionAPI, ExtensionContext } from "../../types.ts"; +import type { ExtensionAPI, ExtensionContext, MessageDelivery } from "../../types.ts"; import { getLatestPhasesFromBranchEntries } from "../todotools/state.ts"; import { buildGoalContinuationSignature, @@ -17,22 +17,98 @@ import { REPETITION_BLOCKED_REASON, } from "./continuation-recovery.ts"; import { buildContinuationPrompt } from "./prompt.ts"; -import { recordContinuationDelivered, updateGoal } from "./store.ts"; +import { + readObjectiveForPrompt, + recordContinuationDelivered, + rollbackContinuationDelivered, + updateGoal, +} from "./store.ts"; import { goalStoreRef } from "./store-ref.ts"; import { openTodoTaskContents } from "./todo-gate.ts"; import type { Goal } from "./types.ts"; type ContinuingGoalContinuationVerdict = Extract; +type ReservedMessageDelivery = MessageDelivery & { + readonly cancelled: boolean; + bind(delivery: MessageDelivery): boolean; +}; + +export type GoalContinuationDeliveryOutcome = { + readonly goal: Goal; + readonly admitted: boolean; + readonly delivery?: MessageDelivery; +}; + +type DeliveryState = "pending" | "started" | "cancelled"; + +let nextReservationId = 0; + +function reserveMessageDelivery(): ReservedMessageDelivery { + let actual: MessageDelivery | undefined; + let state: DeliveryState = "pending"; + const startedListeners = new Set<() => void>(); + const cancelledListeners = new Set<() => void>(); + const isCancelled = (): boolean => state === "cancelled"; + const settle = (nextState: Exclude): void => { + if (state !== "pending") return; + state = nextState; + const listeners = nextState === "started" ? startedListeners : cancelledListeners; + for (const listener of listeners) listener(); + startedListeners.clear(); + cancelledListeners.clear(); + }; + return { + id: `goal-continuation-reservation-${++nextReservationId}`, + get cancelled() { + return isCancelled(); + }, + cancel() { + if (state !== "pending") return false; + if (actual !== undefined && !actual.cancel()) return false; + settle("cancelled"); + return true; + }, + onStarted(listener) { + if (state === "started") { + listener(); + return () => {}; + } + if (state !== "pending") return () => {}; + startedListeners.add(listener); + return () => startedListeners.delete(listener); + }, + onCancelled(listener) { + if (state === "cancelled") { + listener(); + return () => {}; + } + if (state !== "pending") return () => {}; + cancelledListeners.add(listener); + return () => cancelledListeners.delete(listener); + }, + bind(delivery) { + if (state !== "pending") { + delivery.cancel(); + return false; + } + actual = delivery; + delivery.onStarted(() => settle("started")); + delivery.onCancelled(() => settle("cancelled")); + return !isCancelled(); + }, + }; +} + type GoalContinuationDeliveryOptions = { readonly input: Omit; - readonly content: (verdict: ContinuingGoalContinuationVerdict) => string; - readonly markContinuationPending: () => void; + readonly content: (verdict: ContinuingGoalContinuationVerdict) => string | Promise; + readonly markContinuationPending: (delivery: MessageDelivery) => void; }; export type SessionStartContinuationOptions = { readonly continuationPending: boolean; - readonly markContinuationPending: () => void; + readonly markContinuationPending: (delivery: MessageDelivery) => void; }; export function isResumeOfPausedGoal( @@ -55,21 +131,59 @@ export async function admitAndQueueGoalContinuation( ctx: ExtensionContext, goal: Goal, options: GoalContinuationDeliveryOptions, -): Promise { +): Promise { const verdict = evaluateGoalContinuation({ goal, ...options.input }); - if (verdict.kind === "deny") return handleDeniedContinuation(pi, ctx, goal, options.input, verdict.reason); + if (verdict.kind === "deny") { + return { goal: await handleDeniedContinuation(pi, ctx, goal, options.input, verdict.reason), admitted: false }; + } if (options.input.currentSignature === undefined) { throw new Error("Cannot queue a goal continuation without a progress signature"); } - options.markContinuationPending(); - const recordedGoal = await recordContinuationDelivered( - goalStoreRef(ctx.sessionManager, ctx.cwd), - options.input.currentSignature, - ); - if (recordedGoal === null) throw new Error("Cannot persist goal continuation delivery without an active goal"); - queueHiddenGoalPrompt(pi, options.content(verdict)); - return recordedGoal; + const reservation = reserveMessageDelivery(); + options.markContinuationPending(reservation); + try { + const content = await options.content(verdict); + if (reservation.cancelled) return { goal, admitted: false }; + const recordedGoal = await recordContinuationDelivered( + goalStoreRef(ctx.sessionManager, ctx.cwd), + options.input.currentSignature, + goalContinuationExpectation(goal), + ); + if (recordedGoal === null) { + reservation.cancel(); + return { goal, admitted: false }; + } + reservation.onCancelled(() => { + void rollbackContinuationDelivered( + goalStoreRef(ctx.sessionManager, ctx.cwd), + goal, + options.input.currentSignature ?? "", + ).then( + (rolledBack) => { + if (rolledBack !== null) { + pi.events?.emit("goal_continuation_delivery_rolled_back", { goalId: goal.id }); + } + }, + (error) => { + pi.events?.emit("goal_continuation_delivery_rollback_failed", { + goalId: goal.id, + error: error instanceof Error ? error.message : String(error), + }); + }, + ); + }); + if (reservation.cancelled) return { goal: recordedGoal, admitted: false }; + const admitted = reservation.bind(queueHiddenGoalPrompt(pi, content)); + return { + goal: recordedGoal, + admitted, + ...(admitted ? { delivery: reservation } : {}), + }; + } catch (error) { + reservation.cancel(); + throw error; + } } /** Routes startup and resume continuations through the same verdict and delivery accounting as agent-end paths. */ @@ -79,12 +193,14 @@ export async function queueGoalContinuation( goal: Goal, options: SessionStartContinuationOptions, ): Promise { + const ref = goalStoreRef(ctx.sessionManager, ctx.cwd); + const objective = await readObjectiveForPrompt(ref, goal); const signature = buildCurrentGoalContinuationSignature( ctx, goal, lastAssistantTextFromEntries(ctx.sessionManager.getBranch()), ); - return admitAndQueueGoalContinuation(pi, ctx, goal, { + const outcome = await admitAndQueueGoalContinuation(pi, ctx, goal, { input: { isIdle: ctx.isIdle(), hasPendingMessages: ctx.hasPendingMessages(), @@ -98,9 +214,10 @@ export async function queueGoalContinuation( toollessContinuationStreak: 0, continuationPending: options.continuationPending, }, - content: () => buildContinuationPrompt(goal), + content: () => buildContinuationPrompt(goal, objective), markContinuationPending: options.markContinuationPending, }); + return outcome.goal; } export function buildCurrentGoalContinuationSignature( @@ -151,7 +268,9 @@ async function handleDeniedContinuation( goalStoreRef(ctx.sessionManager, ctx.cwd), { status: "blocked", reason: blockedReason }, "model", + goalContinuationExpectation(goal), ); + if (blocked === null) return goal; if (ctx.hasUI) ctx.ui.notify(continuationCapRecoveryHint(blockedReason), "warning"); pi.events?.emit("goal_continuation_guard_tripped", { goalId: goal.id, @@ -178,8 +297,19 @@ function blockedReasonForContinuationGuard( } } -export function queueHiddenGoalPrompt(pi: ExtensionAPI, content: string): void { - pi.sendMessage( +function goalContinuationExpectation(goal: Goal) { + return { + id: goal.id, + status: goal.status, + continuation: { + consecutiveContinuations: goal.consecutiveContinuations ?? 0, + lastContinuationSignature: goal.lastContinuationSignature, + }, + } as const; +} + +export function queueHiddenGoalPrompt(pi: ExtensionAPI, content: string): MessageDelivery { + return pi.sendMessage( { customType: GOAL_CONTINUATION_MESSAGE_TYPE, content, display: false }, { triggerTurn: true, deliverAs: "followUp" }, ); diff --git a/packages/coding-agent/src/core/extensions/builtin/goal/monitor-continuation.ts b/packages/coding-agent/src/core/extensions/builtin/goal/monitor-continuation.ts index 187ea6c20..5b58586a8 100644 --- a/packages/coding-agent/src/core/extensions/builtin/goal/monitor-continuation.ts +++ b/packages/coding-agent/src/core/extensions/builtin/goal/monitor-continuation.ts @@ -1,5 +1,5 @@ import type { AgentMessage } from "@earendil-works/pi-agent-core"; -import type { ExtensionAPI, ExtensionContext } from "../../types.ts"; +import type { ExtensionAPI, ExtensionContext, MessageDelivery } from "../../types.ts"; import { isTerminalMonitorStateEvent, TERMINAL_MONITOR_STATE_EVENT } from "../monitor-state-event.ts"; import { buildCacheWarmResumedNotice, @@ -23,10 +23,11 @@ import { import { admitAndQueueGoalContinuation, buildCurrentGoalContinuationSignature, + type GoalContinuationDeliveryOutcome, lastAssistantText, } from "./lifecycle-helpers.ts"; import { buildContinuationPrompt, buildGoalStallNotice, buildTruncationRecoveryPrompt } from "./prompt.ts"; -import { resetContinuationStreak } from "./store.ts"; +import { readGoal, readObjectiveForPrompt, resetContinuationStreak } from "./store.ts"; import { goalStoreRef } from "./store-ref.ts"; import { collectAssistantUsage } from "./turn-usage.ts"; import type { Goal, TokenUsageSnapshot } from "./types.ts"; @@ -46,15 +47,10 @@ interface AgentEndOptions { type ContinuingGoalContinuationVerdict = Extract; -type GoalContinuationAdmission = { - readonly goal: Goal; - readonly admitted: boolean; -}; - export class MonitorAwareGoalContinuation { readonly #pi: ExtensionAPI; readonly #isContinuationPending: () => boolean; - readonly #markContinuationPending: () => void; + readonly #markContinuationPending: (delivery: MessageDelivery) => void; #activeMonitorCount = 0; #ctx: ExtensionContext | undefined; #goal: Goal | null = null; @@ -77,7 +73,7 @@ export class MonitorAwareGoalContinuation { constructor( pi: ExtensionAPI, isContinuationPending: () => boolean = () => false, - markContinuationPending: () => void = () => {}, + markContinuationPending: (delivery: MessageDelivery) => void = () => {}, ) { this.#pi = pi; this.#isContinuationPending = isContinuationPending; @@ -120,11 +116,25 @@ export class MonitorAwareGoalContinuation { } this.#recordToollessContinuationTurn(options.goal, turnUsedTools); const immediateInput = this.#buildVerdictInput(options.ctx, options.goal, "immediate", options.messages); - const goal = - !this.#endedTurnWasUserInitiated && (turnUsedTools || hasGoalContinuationProgress(immediateInput)) - ? ((await resetContinuationStreak(goalStoreRef(options.ctx.sessionManager, options.ctx.cwd))) ?? - options.goal) - : options.goal; + let goal = options.goal; + if (!this.#endedTurnWasUserInitiated && (turnUsedTools || hasGoalContinuationProgress(immediateInput))) { + const ref = goalStoreRef(options.ctx.sessionManager, options.ctx.cwd); + const resetGoal = await resetContinuationStreak(ref, { + id: options.goal.id, + status: options.goal.status, + continuation: { + consecutiveContinuations: options.goal.consecutiveContinuations ?? 0, + lastContinuationSignature: options.goal.lastContinuationSignature, + }, + }); + if (resetGoal === null) { + const currentGoal = await readGoal(ref); + this.#goal = currentGoal; + this.#cancelTimer(); + return currentGoal; + } + goal = resetGoal; + } this.#goal = goal; if (this.#endedTurnWasUserInitiated) { this.#endedTurnWasUserInitiated = false; @@ -263,25 +273,27 @@ export class MonitorAwareGoalContinuation { if (ctx === undefined || goal?.status !== "active" || !ctx.isIdle() || ctx.hasPendingMessages()) return; if (this.#activeMonitorCount === 0) return; const admission = await this.#admitAndQueue(ctx, goal, "monitorDelayed", this.#lastAgentEndMessages); - if (!admission.admitted) return; - this.#pi.events?.emit(GOAL_CONTINUATION_RESUMED_EVENT, { - goalId: goal.id, - delayMs, - waitedMs, - activeMonitorCount: this.#activeMonitorCount, - cache, - }); - this.#appendWarmupEntry({ - phase: "resumed", - goalId: goal.id, - delayMs, - waitedMs, - activeMonitorCount: this.#activeMonitorCount, - ...(cache !== undefined ? { cache } : {}), + if (!admission.admitted || admission.delivery === undefined) return; + admission.delivery.onStarted(() => { + this.#pi.events?.emit(GOAL_CONTINUATION_RESUMED_EVENT, { + goalId: goal.id, + delayMs, + waitedMs, + activeMonitorCount: this.#activeMonitorCount, + cache, + }); + this.#appendWarmupEntry({ + phase: "resumed", + goalId: goal.id, + delayMs, + waitedMs, + activeMonitorCount: this.#activeMonitorCount, + ...(cache !== undefined ? { cache } : {}), + }); + if (ctx.hasUI) { + ctx.ui.notify(buildCacheWarmResumedNotice(waitedMs, this.#activeMonitorCount, cache), "info"); + } }); - if (ctx.hasUI) { - ctx.ui.notify(buildCacheWarmResumedNotice(waitedMs, this.#activeMonitorCount, cache), "info"); - } } async #admitAndQueue( @@ -289,23 +301,30 @@ export class MonitorAwareGoalContinuation { goal: Goal, path: GoalContinuationPath, messages: readonly AgentMessage[], - ): Promise { + ): Promise { const input = this.#buildVerdictInput(ctx, goal, path, messages); const verdict = evaluateGoalContinuation({ goal, ...input }); - const admittedGoal = await admitAndQueueGoalContinuation(this.#pi, ctx, goal, { + const outcome = await admitAndQueueGoalContinuation(this.#pi, ctx, goal, { input, content: (continuationVerdict) => this.#buildContinuationContent(ctx, goal, continuationVerdict), markContinuationPending: this.#markContinuationPending, }); - if (verdict.kind === "continue" && input.lastStopReason === "length") { - this.#consecutiveLengthRecoveries.set(goal.id, input.consecutiveLengthRecoveries + 1); + if ( + outcome.admitted && + outcome.delivery !== undefined && + verdict.kind === "continue" && + input.lastStopReason === "length" + ) { + outcome.delivery.onStarted(() => { + this.#consecutiveLengthRecoveries.set(goal.id, input.consecutiveLengthRecoveries + 1); + }); } - this.#goal = admittedGoal; - if (admittedGoal.status !== "active") { + this.#goal = outcome.goal; + if (outcome.goal.status !== "active") { this.#cancelTimer(); this.#resetToollessContinuationStreak(); } - return { goal: admittedGoal, admitted: verdict.kind === "continue" }; + return outcome; } #appendWarmupEntry(data: GoalCacheWarmupEntryData): void { @@ -334,8 +353,19 @@ export class MonitorAwareGoalContinuation { }; } - #buildContinuationContent(ctx: ExtensionContext, goal: Goal, verdict: ContinuingGoalContinuationVerdict): string { - let content = verdict.prompt === "minimal" ? buildTruncationRecoveryPrompt() : buildContinuationPrompt(goal); + async #buildContinuationContent( + ctx: ExtensionContext, + goal: Goal, + verdict: ContinuingGoalContinuationVerdict, + ): Promise { + const contentObjective = + verdict.prompt === "minimal" + ? undefined + : await readObjectiveForPrompt(goalStoreRef(ctx.sessionManager, ctx.cwd), goal); + let content = + verdict.prompt === "minimal" + ? buildTruncationRecoveryPrompt() + : buildContinuationPrompt(goal, contentObjective); if (!verdict.stallNotice) return content; const monitorsActive = this.#activeMonitorCount > 0; diff --git a/packages/coding-agent/src/core/extensions/builtin/goal/persistence.ts b/packages/coding-agent/src/core/extensions/builtin/goal/persistence.ts index f4df07017..e8c376d0e 100644 --- a/packages/coding-agent/src/core/extensions/builtin/goal/persistence.ts +++ b/packages/coding-agent/src/core/extensions/builtin/goal/persistence.ts @@ -3,12 +3,15 @@ import type { Dirent } from "node:fs"; import { mkdir, readdir, readFile, rename, rm, writeFile } from "node:fs/promises"; import { homedir } from "node:os"; import { dirname, join, sep } from "node:path"; +import { withFileMutationQueue } from "../../../tools/file-mutation-queue.ts"; import { InvalidGoalStoreError, UnsupportedGoalStoreVersionError } from "./errors.ts"; import type { Goal, GoalFile, GoalStatus, GoalStoreRef } from "./types.ts"; import { isRecord } from "./types.ts"; import { isGoalStatus, isNonNegativeSafeInteger } from "./validation.ts"; const STORE_VERSION = 1; +const PRIVATE_FILE_MODE = 0o600; +const goalWriteListeners = new Map void>>(); const CURRENT_STORE_DIRECTORY = "goal"; const LEGACY_STORE_DIRECTORY = "pi-goal"; const LEGACY_BUDGET_LIMITED_STATUSES = ["budgetLimited", "budget_limited"]; @@ -21,6 +24,10 @@ export function goalFilePath(ref: GoalStoreRef): string { return join(ref.baseDir, `${encodedThreadId(ref)}.json`); } +export function withGoalStoreMutation(ref: GoalStoreRef, fn: () => Promise): Promise { + return withFileMutationQueue(goalFilePath(ref), fn); +} + export async function readGoalFile(ref: GoalStoreRef): Promise { try { return parseGoalFile(await readFile(goalFilePath(ref), "utf8")).goal; @@ -46,6 +53,10 @@ export async function readGoalFile(ref: GoalStoreRef): Promise { * Returns the imported goal, or null when nothing was migrated. */ export async function migrateLegacyGoalFile(ref: GoalStoreRef): Promise { + return withGoalStoreMutation(ref, async () => migrateLegacyGoalFileUnlocked(ref)); +} + +async function migrateLegacyGoalFileUnlocked(ref: GoalStoreRef): Promise { try { await readFile(goalFilePath(ref), "utf8"); return null; @@ -172,14 +183,26 @@ function pathSegments(path: string): string[] { export async function writeGoalFile(ref: GoalStoreRef, goal: Goal | null): Promise { const filePath = goalFilePath(ref); - await mkdir(dirname(filePath), { recursive: true }); - await writeGoalFileAtomic(filePath, `${JSON.stringify({ version: STORE_VERSION, goal }, null, 2)}\n`); + await writePrivateFileAtomic(filePath, `${JSON.stringify({ version: STORE_VERSION, goal }, null, 2)}\n`); + for (const listener of goalWriteListeners.get(filePath) ?? []) listener(); } -async function writeGoalFileAtomic(filePath: string, contents: string): Promise { +export function subscribeGoalFileWrites(ref: GoalStoreRef, listener: () => void): () => void { + const filePath = goalFilePath(ref); + const listeners = goalWriteListeners.get(filePath) ?? new Set<() => void>(); + listeners.add(listener); + goalWriteListeners.set(filePath, listeners); + return () => { + listeners.delete(listener); + if (listeners.size === 0) goalWriteListeners.delete(filePath); + }; +} + +export async function writePrivateFileAtomic(filePath: string, contents: string): Promise { + await mkdir(dirname(filePath), { recursive: true }); const tempPath = join(dirname(filePath), `.goal-${randomUUID()}.tmp`); try { - await writeFile(tempPath, contents, { encoding: "utf8", mode: 0o600 }); + await writeFile(tempPath, contents, { encoding: "utf8", mode: PRIVATE_FILE_MODE }); await rename(tempPath, filePath); } catch (error) { try { @@ -199,7 +222,8 @@ async function publishMigratedGoalFile(ref: GoalStoreRef, goal: Goal): Promise", - escapeXmlText(goal.objective), + escapeXmlText(objective), "", "", "Usage so far:", diff --git a/packages/coding-agent/src/core/extensions/builtin/goal/store.ts b/packages/coding-agent/src/core/extensions/builtin/goal/store.ts index f3b0a2c9f..4ff7c7354 100644 --- a/packages/coding-agent/src/core/extensions/builtin/goal/store.ts +++ b/packages/coding-agent/src/core/extensions/builtin/goal/store.ts @@ -1,18 +1,25 @@ import { randomUUID } from "node:crypto"; -import { appendFile, mkdir, writeFile } from "node:fs/promises"; +import { appendFile, chmod, mkdir, readFile, rm } from "node:fs/promises"; import { dirname, join } from "node:path"; import { GoalAlreadyExistsError, GoalNotFoundError } from "./errors.ts"; -import { encodedThreadId, readGoalFile, writeGoalFile } from "./persistence.ts"; +import { + encodedThreadId, + readGoalFile, + withGoalStoreMutation, + writeGoalFile, + writePrivateFileAtomic, +} from "./persistence.ts"; import { transitionGoalStatus } from "./transitions.ts"; import type { Goal, GoalAccountingMode, + GoalExpectation, GoalStoreRef, GoalUpdate, GoalUpdateSource, TokenUsageSnapshot, } from "./types.ts"; -import { resolveTokenBudget, validateObjective, validateTokenBudget } from "./validation.ts"; +import { resolveTokenBudget, truncationMarker, validateObjective, validateTokenBudget } from "./validation.ts"; export { goalFilePath } from "./persistence.ts"; @@ -29,46 +36,91 @@ export function objectiveFullTextFilePath(ref: GoalStoreRef): string { } export async function readGoal(ref: GoalStoreRef): Promise { - return readGoalFile(ref); + return withHardenedGoalStoreMutation(ref, async () => readGoalFile(ref)); +} + +export async function readObjectiveForPrompt(ref: GoalStoreRef, goal: Pick): Promise { + return withHardenedGoalStoreMutation(ref, async () => { + const current = await readGoalFile(ref); + if (current?.id !== goal.id || current.objective !== goal.objective) return goal.objective; + await hardenGoalAuxiliaryFilePermissions(ref); + + const fullTextFileName = objectiveFullTextFileName(ref); + if (!goal.objective.endsWith(truncationMarker(fullTextFileName))) return goal.objective; + + let fullObjective: string; + try { + fullObjective = (await readFile(objectiveFullTextFilePath(ref), "utf8")).trim(); + } catch (error) { + if (isMissingFileError(error) || isFileNameTooLongError(error)) return goal.objective; + throw error; + } + + try { + const validated = validateObjective(fullObjective, fullTextFileName); + return validated.truncated && validated.objective === goal.objective ? fullObjective : goal.objective; + } catch { + return goal.objective; + } + }); } export async function writeGoal(ref: GoalStoreRef, goal: Goal | null): Promise { - await writeGoalFile(ref, goal); + await withHardenedGoalStoreMutation(ref, async () => writeGoalFile(ref, goal)); } export async function createGoal(ref: GoalStoreRef, objective: string, tokenBudget?: number): Promise { - const validatedObjective = validateObjective(objective, objectiveFullTextFileName(ref)); - const current = await readGoal(ref); - if (current !== null && current.status !== "complete") { - throw new GoalAlreadyExistsError("cannot create a new goal because this thread already has a goal"); - } - if (validatedObjective.truncated) await writeFullObjectiveText(ref, objective); - if (current?.status === "complete") await archiveGoal(ref, current); - const now = nowSeconds(); - const goal: Goal = { - id: randomUUID(), - threadId: ref.threadId, - objective: validatedObjective.objective, - status: "active", - tokensUsed: 0, - timeUsedSeconds: 0, - consecutiveContinuations: 0, - createdAt: now, - updatedAt: now, - lastStartedAt: now, - ...(tokenBudget === undefined ? {} : { tokenBudget: validateTokenBudget(tokenBudget) }), - }; - await writeGoal(ref, goal); - return goal; + return withHardenedGoalStoreMutation(ref, async () => { + const validatedObjective = validateObjective(objective, objectiveFullTextFileName(ref)); + const current = await readGoalFile(ref); + if (current !== null && current.status !== "complete") { + throw new GoalAlreadyExistsError("cannot create a new goal because this thread already has a goal"); + } + if (current?.status === "complete") await archiveGoalUnlocked(ref, current); + const now = nowSeconds(); + const goal: Goal = { + id: randomUUID(), + threadId: ref.threadId, + objective: validatedObjective.objective, + status: "active", + tokensUsed: 0, + timeUsedSeconds: 0, + consecutiveContinuations: 0, + createdAt: now, + updatedAt: now, + lastStartedAt: now, + ...(tokenBudget === undefined ? {} : { tokenBudget: validateTokenBudget(tokenBudget) }), + }; + await writeGoalAndSidecar(ref, current, goal, validatedObjective.truncated ? objective : null); + return goal; + }); } +export function updateGoal(ref: GoalStoreRef, update: GoalUpdate, source?: GoalUpdateSource): Promise; +export function updateGoal( + ref: GoalStoreRef, + update: GoalUpdate, + source: GoalUpdateSource, + expected: GoalExpectation, +): Promise; export async function updateGoal( ref: GoalStoreRef, update: GoalUpdate, source: GoalUpdateSource = "model", -): Promise { - const current = await readGoal(ref); + expected?: GoalExpectation, +): Promise { + return withHardenedGoalStoreMutation(ref, async () => updateGoalUnlocked(ref, update, source, expected)); +} + +async function updateGoalUnlocked( + ref: GoalStoreRef, + update: GoalUpdate, + source: GoalUpdateSource, + expected: GoalExpectation | undefined, +): Promise { + const current = await readGoalFile(ref); if (!current) throw new GoalNotFoundError("cannot update goal: no goal exists"); + if (expected !== undefined && !matchesGoalExpectation(current, expected)) return null; const validatedObjective = update.objective === undefined ? undefined : validateObjective(update.objective, objectiveFullTextFileName(ref)); @@ -76,7 +128,13 @@ export async function updateGoal( const tokenBudget = resolveTokenBudget(current.tokenBudget, update.tokenBudget); const now = nextUpdatedAt(current.updatedAt); const hasObjectiveUpdate = update.objective !== undefined; - const replacesGoal = hasObjectiveUpdate && (objective !== current.objective || current.status === "complete"); + const replacesSameTruncatedObjective = + validatedObjective?.truncated === true && + objective === current.objective && + (await readFullObjectiveText(ref))?.trim() !== update.objective?.trim(); + const replacesGoal = + hasObjectiveUpdate && + (objective !== current.objective || current.status === "complete" || replacesSameTruncatedObjective); const requestedStatus = update.status ?? (hasObjectiveUpdate ? "active" : undefined); if (replacesGoal) { @@ -96,8 +154,7 @@ export async function updateGoal( }; if (status === "active") next.lastStartedAt = now; if (status === "complete") next.completedAt = now; - if (validatedObjective?.truncated) await writeFullObjectiveText(ref, update.objective ?? ""); - await writeGoal(ref, next); + await writeGoalAndSidecar(ref, current, next, validatedObjective?.truncated ? (update.objective ?? "") : null); return next; } @@ -114,27 +171,127 @@ export async function updateGoal( } if (tokenBudget === undefined) delete next.tokenBudget; else next.tokenBudget = tokenBudget; - if (validatedObjective?.truncated) await writeFullObjectiveText(ref, update.objective ?? ""); - await writeGoal(ref, next); + await writeGoalAndSidecar( + ref, + current, + next, + validatedObjective === undefined ? undefined : validatedObjective.truncated ? (update.objective ?? "") : null, + ); return next; } export async function archiveGoal(ref: GoalStoreRef, goal: Goal): Promise { + await withHardenedGoalStoreMutation(ref, async () => archiveGoalUnlocked(ref, goal)); +} + +async function archiveGoalUnlocked(ref: GoalStoreRef, goal: Goal): Promise { const filePath = goalHistoryFilePath(ref); await mkdir(dirname(filePath), { recursive: true }); - await appendFile(filePath, `${JSON.stringify(goal)}\n`, "utf8"); + try { + await chmod(filePath, 0o600); + } catch (error) { + if (!isMissingFileError(error)) throw error; + } + await appendFile(filePath, `${JSON.stringify(goal)}\n`, { encoding: "utf8", mode: 0o600 }); + await chmod(filePath, 0o600); +} + +async function writeGoalAndSidecar( + ref: GoalStoreRef, + previousGoal: Goal | null, + goal: Goal | null, + sidecar: string | null | undefined, +): Promise { + await hardenGoalAuxiliaryFilePermissions(ref); + if (sidecar === undefined) { + await writeGoalFile(ref, goal); + return; + } + if (sidecar === null) { + await writeGoalFile(ref, goal); + try { + await removeFullObjectiveText(ref); + } catch (error) { + await restoreGoalFile(ref, previousGoal, error); + } + return; + } + + const previousSidecar = await readFullObjectiveText(ref); + await writeFullObjectiveText(ref, sidecar); + try { + await writeGoalFile(ref, goal); + } catch (error) { + try { + if (previousSidecar === undefined) await removeFullObjectiveText(ref); + else await writeFullObjectiveText(ref, previousSidecar); + } catch (cleanupError) { + throw new AggregateError( + [error, cleanupError], + "goal store write failed and its full-objective sidecar could not be restored", + ); + } + throw error; + } +} + +async function hardenGoalAuxiliaryFilePermissions(ref: GoalStoreRef): Promise { + for (const filePath of [objectiveFullTextFilePath(ref), goalHistoryFilePath(ref)]) { + try { + await chmod(filePath, 0o600); + } catch (error) { + if (!isMissingFileError(error) && !isFileNameTooLongError(error)) throw error; + } + } +} + +async function withHardenedGoalStoreMutation(ref: GoalStoreRef, operation: () => Promise): Promise { + return withGoalStoreMutation(ref, async () => { + await hardenGoalAuxiliaryFilePermissions(ref); + return operation(); + }); +} + +async function restoreGoalFile(ref: GoalStoreRef, previousGoal: Goal | null, error: unknown): Promise { + try { + await writeGoalFile(ref, previousGoal); + } catch (cleanupError) { + throw new AggregateError( + [error, cleanupError], + "full-objective sidecar cleanup failed and the goal store could not be restored", + ); + } + throw error; +} + +async function readFullObjectiveText(ref: GoalStoreRef): Promise { + try { + return await readFile(objectiveFullTextFilePath(ref), "utf8"); + } catch (error) { + if (isMissingFileError(error) || isFileNameTooLongError(error)) return undefined; + throw error; + } } async function writeFullObjectiveText(ref: GoalStoreRef, objective: string): Promise { - const filePath = objectiveFullTextFilePath(ref); - await mkdir(dirname(filePath), { recursive: true }); - await writeFile(filePath, objective, "utf8"); + await writePrivateFileAtomic(objectiveFullTextFilePath(ref), objective); +} + +async function removeFullObjectiveText(ref: GoalStoreRef): Promise { + try { + await rm(objectiveFullTextFilePath(ref), { force: true }); + } catch (error) { + if (isFileNameTooLongError(error)) return; + throw error; + } } export async function clearGoal(ref: GoalStoreRef): Promise { - const hadGoal = (await readGoal(ref)) !== null; - await writeGoal(ref, null); - return hadGoal; + return withHardenedGoalStoreMutation(ref, async () => { + const current = await readGoalFile(ref); + await writeGoalAndSidecar(ref, current, null, null); + return current !== null; + }); } export async function accountGoalUsage( @@ -144,39 +301,83 @@ export async function accountGoalUsage( mode: GoalAccountingMode = "active", expectedGoalId?: string, ): Promise { - const goal = await readGoal(ref); - if (!goal || (expectedGoalId !== undefined && goal.id !== expectedGoalId) || !canAccountGoalUsage(goal, mode)) { - return goal; - } - const next: Goal = { - ...goal, - tokensUsed: goal.tokensUsed + Math.max(0, usage.input) + Math.max(0, usage.output), - timeUsedSeconds: goal.timeUsedSeconds + Math.max(0, Math.trunc(elapsedSeconds)), - updatedAt: nextUpdatedAt(goal.updatedAt), - }; - await writeGoal(ref, next); - return next; + return withHardenedGoalStoreMutation(ref, async () => { + const goal = await readGoalFile(ref); + if (!goal || (expectedGoalId !== undefined && goal.id !== expectedGoalId) || !canAccountGoalUsage(goal, mode)) { + return goal; + } + const next: Goal = { + ...goal, + tokensUsed: goal.tokensUsed + Math.max(0, usage.input) + Math.max(0, usage.output), + timeUsedSeconds: goal.timeUsedSeconds + Math.max(0, Math.trunc(elapsedSeconds)), + updatedAt: nextUpdatedAt(goal.updatedAt), + }; + await writeGoalFile(ref, next); + return next; + }); } -export async function recordContinuationDelivered(ref: GoalStoreRef, signature: string): Promise { - const goal = await readGoal(ref); - if (!goal) return goal; - const next: Goal = { - ...goal, - consecutiveContinuations: (goal.consecutiveContinuations ?? 0) + 1, - lastContinuationSignature: signature, - }; - await writeGoal(ref, next); - return next; +export async function recordContinuationDelivered( + ref: GoalStoreRef, + signature: string, + expected?: GoalExpectation, +): Promise { + return withHardenedGoalStoreMutation(ref, async () => { + const goal = await readGoalFile(ref); + if (!goal || (expected !== undefined && !matchesGoalExpectation(goal, expected))) return null; + const next: Goal = { + ...goal, + consecutiveContinuations: (goal.consecutiveContinuations ?? 0) + 1, + lastContinuationSignature: signature, + }; + await writeGoalFile(ref, next); + return next; + }); } -export async function resetContinuationStreak(ref: GoalStoreRef): Promise { - const goal = await readGoal(ref); - if (!goal) return goal; - const next: Goal = { ...goal, consecutiveContinuations: 0 }; - delete next.lastContinuationSignature; - await writeGoal(ref, next); - return next; +export async function rollbackContinuationDelivered( + ref: GoalStoreRef, + previous: Pick, + signature: string, +): Promise { + return withHardenedGoalStoreMutation(ref, async () => { + const goal = await readGoalFile(ref); + const previousCount = previous.consecutiveContinuations ?? 0; + if ( + !goal || + goal.id !== previous.id || + goal.status !== previous.status || + goal.consecutiveContinuations !== previousCount + 1 || + goal.lastContinuationSignature !== signature + ) { + return null; + } + const next: Goal = { ...goal, consecutiveContinuations: previousCount }; + if (previous.lastContinuationSignature === undefined) delete next.lastContinuationSignature; + else next.lastContinuationSignature = previous.lastContinuationSignature; + await writeGoalFile(ref, next); + return next; + }); +} + +export async function resetContinuationStreak(ref: GoalStoreRef, expected?: GoalExpectation): Promise { + return withHardenedGoalStoreMutation(ref, async () => { + const goal = await readGoalFile(ref); + if (!goal || (expected !== undefined && !matchesGoalExpectation(goal, expected))) return null; + const next: Goal = { ...goal, consecutiveContinuations: 0 }; + delete next.lastContinuationSignature; + await writeGoalFile(ref, next); + return next; + }); +} + +function matchesGoalExpectation(goal: Goal, expected: GoalExpectation): boolean { + if (goal.id !== expected.id || goal.status !== expected.status) return false; + if (expected.continuation === undefined) return true; + return ( + (goal.consecutiveContinuations ?? 0) === expected.continuation.consecutiveContinuations && + goal.lastContinuationSignature === expected.continuation.lastContinuationSignature + ); } function canAccountGoalUsage(goal: Goal, mode: GoalAccountingMode): boolean { @@ -192,3 +393,11 @@ function nextUpdatedAt(previousUpdatedAt: number): number { function nowSeconds(): number { return Math.trunc(Date.now() / 1000); } + +function isMissingFileError(error: unknown): boolean { + return typeof error === "object" && error !== null && "code" in error && error.code === "ENOENT"; +} + +function isFileNameTooLongError(error: unknown): boolean { + return typeof error === "object" && error !== null && "code" in error && error.code === "ENAMETOOLONG"; +} diff --git a/packages/coding-agent/src/core/extensions/builtin/goal/types.ts b/packages/coding-agent/src/core/extensions/builtin/goal/types.ts index 86ac01a13..82e5e9bd9 100644 --- a/packages/coding-agent/src/core/extensions/builtin/goal/types.ts +++ b/packages/coding-agent/src/core/extensions/builtin/goal/types.ts @@ -12,6 +12,15 @@ export type GoalStoreRef = { export type GoalAccountingMode = "active" | "activeOrBlocked" | "activeOrComplete"; export type GoalUpdateSource = "model" | "user"; +export type GoalExpectation = Readonly<{ + id: string; + status: GoalStatus; + continuation?: Readonly<{ + consecutiveContinuations: number; + lastContinuationSignature: string | undefined; + }>; +}>; + export type Goal = { id: string; threadId: string; diff --git a/packages/coding-agent/src/core/extensions/index.ts b/packages/coding-agent/src/core/extensions/index.ts index 1822396f2..4052c07df 100644 --- a/packages/coding-agent/src/core/extensions/index.ts +++ b/packages/coding-agent/src/core/extensions/index.ts @@ -108,6 +108,7 @@ export type { LsToolResultEvent, MarkdownTransformContext, MarkdownTransformer, + MessageDelivery, // Events - Message MessageEndEvent, MessageRenderer, diff --git a/packages/coding-agent/src/core/extensions/loader.ts b/packages/coding-agent/src/core/extensions/loader.ts index ae1d2376a..bd7eb1831 100644 --- a/packages/coding-agent/src/core/extensions/loader.ts +++ b/packages/coding-agent/src/core/extensions/loader.ts @@ -448,9 +448,9 @@ function createExtensionAPI( }, // Action methods - delegate to shared runtime - sendMessage(message, options): void { + sendMessage(message, options) { runtime.assertActive(); - runtime.sendMessage(message, options); + return runtime.sendMessage(message, options); }, sendUserMessage(content, options): void { diff --git a/packages/coding-agent/src/core/extensions/types.ts b/packages/coding-agent/src/core/extensions/types.ts index 4e0800d56..6ad2288f8 100644 --- a/packages/coding-agent/src/core/extensions/types.ts +++ b/packages/coding-agent/src/core/extensions/types.ts @@ -1593,7 +1593,7 @@ export interface ExtensionAPI { sendMessage( message: Pick, "customType" | "content" | "display" | "details">, options?: { triggerTurn?: boolean; deliverAs?: "steer" | "followUp" | "nextTurn" }, - ): void; + ): MessageDelivery; /** * Send a user message to the agent. Always triggers a turn. @@ -1874,10 +1874,17 @@ export interface ExtensionShortcut { type HandlerFn = (...args: unknown[]) => Promise; +export interface MessageDelivery { + readonly id: string; + cancel(): boolean; + onStarted(listener: () => void): () => void; + onCancelled(listener: () => void): () => void; +} + export type SendMessageHandler = ( message: Pick, "customType" | "content" | "display" | "details">, options?: { triggerTurn?: boolean; deliverAs?: "steer" | "followUp" | "nextTurn" }, -) => void; +) => MessageDelivery; export type SendUserMessageHandler = ( content: string | (TextContent | ImageContent)[], diff --git a/packages/coding-agent/src/core/tools/edit-diff.ts b/packages/coding-agent/src/core/tools/edit-diff.ts index ed7173321..ae1bf81bd 100644 --- a/packages/coding-agent/src/core/tools/edit-diff.ts +++ b/packages/coding-agent/src/core/tools/edit-diff.ts @@ -509,6 +509,16 @@ export interface EditDiffError { error: string; } +export interface EditDiffOperations { + access: (path: string, mode: number) => Promise; + readFile: (path: string, encoding: BufferEncoding) => Promise; +} + +const defaultEditDiffOperations: EditDiffOperations = { + access, + readFile, +}; + /** * Compute the diff for one or more edit operations without applying them. * Used for preview rendering in the TUI before the tool executes. @@ -517,20 +527,21 @@ export async function computeEditsDiff( path: string, edits: Edit[], cwd: string, + operations: EditDiffOperations = defaultEditDiffOperations, ): Promise { const absolutePath = resolveToCwd(path, cwd); try { // Check if file exists and is readable try { - await access(absolutePath, constants.R_OK); + await operations.access(absolutePath, constants.R_OK); } catch (error: unknown) { const errorMessage = error instanceof Error && "code" in error ? `Error code: ${error.code}` : String(error); return { error: `Could not edit file: ${path}. ${errorMessage}.` }; } // Read the file - const rawContent = await readFile(absolutePath, "utf-8"); + const rawContent = await operations.readFile(absolutePath, "utf-8"); // Strip BOM before matching (LLM won't include invisible BOM in oldText) const { text: content } = stripBom(rawContent); diff --git a/packages/coding-agent/src/modes/app-server/threads/turn-runtime.ts b/packages/coding-agent/src/modes/app-server/threads/turn-runtime.ts index a1f04da67..43d652ffa 100644 --- a/packages/coding-agent/src/modes/app-server/threads/turn-runtime.ts +++ b/packages/coding-agent/src/modes/app-server/threads/turn-runtime.ts @@ -24,7 +24,7 @@ export interface TurnEngineSession { text: string, options?: { readonly source?: "rpc"; readonly preflightResult?: (success: boolean) => void }, ): Promise; - steer(text: string): Promise; + steer(text: string, options?: { readonly source?: "rpc" }): Promise; abort(): Promise; subscribe(listener: (event: TurnEngineSessionEvent) => void): () => void; } diff --git a/packages/coding-agent/src/modes/app-server/threads/turns.ts b/packages/coding-agent/src/modes/app-server/threads/turns.ts index 82b644e91..5a2b7228a 100644 --- a/packages/coding-agent/src/modes/app-server/threads/turns.ts +++ b/packages/coding-agent/src/modes/app-server/threads/turns.ts @@ -192,7 +192,7 @@ class TurnEngine { ); } const parsedInput = parseInput(params.input); - await entry.session.steer(parsedInput.text); + await entry.session.steer(parsedInput.text, { source: "rpc" }); this.emitUserMessage( params.threadId, activeTurn.turnId, diff --git a/packages/coding-agent/src/modes/app-server/turn-adapter.ts b/packages/coding-agent/src/modes/app-server/turn-adapter.ts index c56614104..bcf7a9081 100644 --- a/packages/coding-agent/src/modes/app-server/turn-adapter.ts +++ b/packages/coding-agent/src/modes/app-server/turn-adapter.ts @@ -218,8 +218,8 @@ class ModeTurnSession implements TurnEngineSession { return this.session.prompt(text, options); } - steer(text: string): Promise { - return this.session.steer(text); + steer(text: string, options?: { readonly source?: "rpc" }): Promise { + return this.session.steer(text, undefined, options); } abort(): Promise { diff --git a/packages/coding-agent/src/modes/rpc/connection-handler.ts b/packages/coding-agent/src/modes/rpc/connection-handler.ts index e865aa443..81fba1dc4 100644 --- a/packages/coding-agent/src/modes/rpc/connection-handler.ts +++ b/packages/coding-agent/src/modes/rpc/connection-handler.ts @@ -551,12 +551,12 @@ export function createRpcConnectionHandler( } case "steer": { - await session.steer(command.message, command.images); + await session.steer(command.message, command.images, { source: "rpc" }); return success(id, "steer"); } case "follow_up": { - await session.followUp(command.message, command.images); + await session.followUp(command.message, command.images, { source: "rpc" }); return success(id, "follow_up"); } diff --git a/packages/coding-agent/test/compaction/before-compact-error-surfacing.test.ts b/packages/coding-agent/test/compaction/before-compact-error-surfacing.test.ts index 42fcec231..e85ebff31 100644 --- a/packages/coding-agent/test/compaction/before-compact-error-surfacing.test.ts +++ b/packages/coding-agent/test/compaction/before-compact-error-surfacing.test.ts @@ -92,6 +92,9 @@ function createHarness(options?: { withAuth?: boolean }): Harness { sessionManager: { getEntries: () => [], getBranch: () => [], + getSessionFile: () => `/tmp/senpi-before-compact-${process.pid}/session.jsonl`, + getSessionDir: () => `/tmp/senpi-before-compact-${process.pid}`, + getSessionId: () => `before-compact-${process.pid}`, } as unknown as ExtensionContext["sessionManager"], modelRegistry, model, diff --git a/packages/coding-agent/test/compaction/compaction-log.test.ts b/packages/coding-agent/test/compaction/compaction-log.test.ts index 06ec8224e..039060496 100644 --- a/packages/coding-agent/test/compaction/compaction-log.test.ts +++ b/packages/coding-agent/test/compaction/compaction-log.test.ts @@ -89,4 +89,32 @@ describe("compaction logger", () => { expect(entry).not.toHaveProperty("message"); expect(entry).not.toHaveProperty("summary"); }); + + it("Given deterministic fallback diagnostics When logging Then content is never persisted", () => { + const sink: string[] = []; + const logger = createCompactionLogger("/tmp/senpi-compaction-log-fallback", { sink: (line) => sink.push(line) }); + + logger.info("deterministic_fallback_applied", { + origin: "required-compaction-recovery", + failureKind: "summarization-timeout", + retainedEntryCount: 8, + todoItemCount: 2, + summaryBytes: 512, + hasTaskIntent: true, + summary: "private recovery content", + taskIntent: "private user task", + } as unknown as CompactionLoggerData); + + const entry = JSON.parse(sink[0] as string) as Record; + expect(entry).toMatchObject({ + event: "deterministic_fallback_applied", + failureKind: "summarization-timeout", + retainedEntryCount: 8, + todoItemCount: 2, + summaryBytes: 512, + hasTaskIntent: true, + }); + expect(entry).not.toHaveProperty("summary"); + expect(entry).not.toHaveProperty("taskIntent"); + }); }); diff --git a/packages/coding-agent/test/compaction/metadata-side-effects.test.ts b/packages/coding-agent/test/compaction/metadata-side-effects.test.ts index d0a11d59f..8dd5111a1 100644 --- a/packages/coding-agent/test/compaction/metadata-side-effects.test.ts +++ b/packages/coding-agent/test/compaction/metadata-side-effects.test.ts @@ -61,6 +61,9 @@ function createExtensionContext(entries: SessionEntry[]): ExtensionContext { const sessionManager = { getEntries: () => entries, getBranch: () => entries, + getSessionFile: () => `/tmp/senpi-metadata-side-effects-${process.pid}/session.jsonl`, + getSessionDir: () => `/tmp/senpi-metadata-side-effects-${process.pid}`, + getSessionId: () => `metadata-side-effects-${process.pid}`, } as ExtensionContext["sessionManager"]; return { @@ -132,6 +135,38 @@ function createCompactionEntry(id: string, firstKeptEntryId: string): Compaction } describe("compaction metadata side effects", () => { + it("warns once when an accepted automatic compaction used deterministic recovery", async () => { + const harness = createCompactionExtensionHarness(); + const ctx = createExtensionContext([]); + const compactionEntry = { + ...createCompactionEntry("fallback", "kept"), + details: { + schema: "senpi.compaction.deterministic-fallback.v1", + origin: "required-compaction-recovery", + failureKind: "summarization-timeout", + }, + }; + + await harness.sessionCompact( + { + type: "session_compact", + reason: "overflow", + requestId: "fallback-request", + accepted: true, + compactionEntry, + fromExtension: true, + willRetry: true, + }, + ctx, + ); + + expect(ctx.ui.notify).toHaveBeenCalledOnce(); + expect(ctx.ui.notify).toHaveBeenCalledWith( + "Automatic compaction used a local recovery checkpoint because summarization did not finish; older context may be incomplete.", + "warning", + ); + }); + describe("Given a compaction request that has not been accepted yet", () => { describe("When session_before_compact runs", () => { it("Then checkpoint and todo metadata are not persisted until session_compact succeeds", async () => { diff --git a/packages/coding-agent/test/compaction/required-compaction-deterministic-fallback.test.ts b/packages/coding-agent/test/compaction/required-compaction-deterministic-fallback.test.ts index 17bae7938..d35e8d75c 100644 --- a/packages/coding-agent/test/compaction/required-compaction-deterministic-fallback.test.ts +++ b/packages/coding-agent/test/compaction/required-compaction-deterministic-fallback.test.ts @@ -149,6 +149,124 @@ describe("required compaction deterministic fallback", () => { ).toBe("upstream-stream-truncated"); }); + it("preserves current todo state when the first required automatic summary times out", () => { + const harness = createBlockingContext({ usageTokens: 9_900 }); + const branchEntries = harness.ctx.sessionManager.getBranch(); + const preparation = prepareCompaction(branchEntries, harness.ctx.getCompactionSettings(), true); + expect(preparation).toBeDefined(); + + const result = createRequiredCompactionFallback( + { + ...preparation!, + firstKeptEntryId: branchEntries.at(-1)?.id ?? "", + }, + 100_000, + "summarization-timeout", + { + todoSnapshot: { + schema: "senpi.compaction.todo-snapshot.v1", + todos: [ + { + name: "Repair", + tasks: [ + { content: "Preserve automatic compaction task state", status: "in_progress" }, + { content: "Run automatic compaction regression", status: "pending" }, + ], + }, + ], + capturedAt: 0, + }, + }, + branchEntries, + ); + + expect(result).toBeDefined(); + expect(result!.summary).toContain("Current todo state:"); + expect(result!.summary).toContain("[in_progress] Preserve automatic compaction task state"); + expect(result!.summary).toContain("[pending] Run automatic compaction regression"); + expect(result!.details).not.toHaveProperty("todoSnapshot"); + }); + + it("passes the latest todo snapshot through the required automatic fallback handler", async () => { + const handlers = createCompactionHandlers(); + const harness = createBlockingContext({ usageTokens: 99_000, contextWindow: 100_000 }); + harness.sessionManager.appendCustomEntry("senpi.todo-state", { + schema: "v2", + phases: [{ name: "Old", tasks: [{ content: "Do not restore obsolete work", status: "in_progress" }] }], + }); + harness.sessionManager.appendCustomEntry("senpi.todo-state", { + schema: "v2", + phases: [ + { + name: "Current", + tasks: [{ content: "Restore the current automatic compaction task", status: "in_progress" }], + }, + ], + }); + harness.registration.setResponses([ + fauxAssistantMessage("", { + stopReason: "error", + errorMessage: "upstream_stream_truncated: Responses stream ended before a terminal event", + }), + ]); + const branchEntries = harness.ctx.sessionManager.getBranch(); + const preparation = prepareCompaction(branchEntries, harness.ctx.getCompactionSettings(), true); + expect(preparation).toBeDefined(); + + const result = await handlers.sessionBeforeCompact( + { + type: "session_before_compact", + reason: "threshold", + willRetry: false, + requestId: "automatic-fallback-with-current-todos", + preparation: preparation!, + branchEntries, + signal: new AbortController().signal, + }, + harness.ctx, + ); + + expect(result).toHaveProperty("compaction"); + expect(result?.compaction?.summary).toContain("[in_progress] Restore the current automatic compaction task"); + expect(result?.compaction?.summary).not.toContain("Do not restore obsolete work"); + }); + + it("preserves recent dropped user intent when automatic fallback has no todo state", () => { + const harness = createBlockingContext({ usageTokens: 9_900 }); + const branchEntries = harness.ctx.sessionManager.getBranch(); + const preparation = prepareCompaction(branchEntries, harness.ctx.getCompactionSettings(), true); + expect(preparation).toBeDefined(); + branchEntries.splice(-1, 0, { + type: "message", + id: "control-envelope", + parentId: branchEntries.at(-2)?.id ?? null, + timestamp: new Date(4).toISOString(), + message: { + role: "user", + content: [ + { type: "text", text: "Do not preserve this control envelope" }, + ], + timestamp: 4, + }, + }); + + const result = createRequiredCompactionFallback( + { + ...preparation!, + firstKeptEntryId: branchEntries.at(-1)?.id ?? "", + }, + 100_000, + "summarization-timeout", + {}, + branchEntries, + ); + + expect(result).toBeDefined(); + expect(result!.summary).toContain("Task intent:"); + expect(result!.summary).toContain("Summarize old context"); + expect(result!.summary).not.toContain("Do not preserve this control envelope"); + }); + it("requires a real retained suffix, keeps only canonical detail metadata, and uses UTF-8-safe bounds", () => { const harness = createBlockingContext({ usageTokens: 9_900 }); const branchEntries = harness.ctx.sessionManager.getBranch(); @@ -173,7 +291,7 @@ describe("required compaction deterministic fallback", () => { 100_000, "summarization-timeout", { - taskIntent: "Finish the current repair", + taskIntent: `Finish the current repair\n${"😀".repeat(3_000)}PRIVATE_TAIL`, todoSnapshot: { items: ["verify recovery"] }, checkpoint: { files: ["agent-session.ts"] }, }, @@ -182,10 +300,11 @@ describe("required compaction deterministic fallback", () => { expect(result).toBeDefined(); expect(result!.summary).not.toContain("�"); + expect(result!.summary).not.toContain("PRIVATE_TAIL"); expect(result!.details).toMatchObject({ - taskIntent: "Finish the current repair", retainedSuffix: "prepared", }); + expect(result!.details).not.toHaveProperty("taskIntent"); expect(result!.details).not.toHaveProperty("todoSnapshot"); expect(result!.details).not.toHaveProperty("checkpoint"); harness.sessionManager.appendCompaction( diff --git a/packages/coding-agent/test/compaction/speculative-compaction.test.ts b/packages/coding-agent/test/compaction/speculative-compaction.test.ts index e7bcc8249..219983efa 100644 --- a/packages/coding-agent/test/compaction/speculative-compaction.test.ts +++ b/packages/coding-agent/test/compaction/speculative-compaction.test.ts @@ -13,9 +13,10 @@ import { type StreamOptions, unregisterApiProviders, } from "@earendil-works/pi-ai/compat"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { AuthStorage } from "../../src/core/auth-storage.ts"; import { DEFAULT_COMPACTION_SETTINGS } from "../../src/core/compaction/index.ts"; +import { StreamDurationBudgetError } from "../../src/core/compaction/stream-watchdog.ts"; import compactionExtension from "../../src/core/extensions/builtin/compaction/index.ts"; import { shouldStartSpeculativeCompaction } from "../../src/core/extensions/builtin/compaction/policy.ts"; import { @@ -38,6 +39,7 @@ type TestSpeculativeCompactionContext = SpeculativeCompactionContext & { }; afterEach(() => { + vi.useRealTimers(); for (const registration of registrations.splice(0)) { registration.unregister(); } @@ -113,6 +115,22 @@ function createContext(options?: { } describe("speculative compaction", () => { + it("uses an injected watchdog budget without waiting in real time", async () => { + vi.useFakeTimers(); + const context = createContext({ shrink: true }); + const snapshot = createSpeculativeCompactionSnapshot(context, { generation: 1 }); + expect(snapshot).toBeDefined(); + context.registration.setResponses([() => new Promise(() => {})]); + + const outcome = runExtensionCompaction(context, snapshot!, undefined, undefined, { + idleTimeoutMs: 1_000, + maxDurationMs: 5, + }).catch((error: unknown) => error); + await vi.advanceTimersByTimeAsync(6); + + await expect(outcome).resolves.toBeInstanceOf(StreamDurationBudgetError); + }); + it("starts at the 37.5 percent default trigger for a 32k context window", () => { // Given const contextWindow = 32_000; diff --git a/packages/coding-agent/test/compaction/summarization-stream-watchdog.test.ts b/packages/coding-agent/test/compaction/summarization-stream-watchdog.test.ts index dcefdca73..9d1c5e54f 100644 --- a/packages/coding-agent/test/compaction/summarization-stream-watchdog.test.ts +++ b/packages/coding-agent/test/compaction/summarization-stream-watchdog.test.ts @@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { consumeStreamWithIdleTimeout, DEFAULT_SUMMARIZATION_IDLE_TIMEOUT_MS, + StreamDurationBudgetError, StreamIdleTimeoutError, } from "../../src/core/compaction/stream-watchdog.ts"; @@ -87,6 +88,49 @@ describe("consumeStreamWithIdleTimeout", () => { expect(seen).toEqual(["a", "b", "c"]); }); + it("keeps the duration budget active while the final result settles", async () => { + let aborted = false; + const stream = { + async *[Symbol.asyncIterator]() {}, + result: () => new Promise(() => {}), + }; + const outcome = consumeStreamWithIdleTimeout(stream, { + idleTimeoutMs: 1_000, + maxDurationMs: 50, + abort: () => { + aborted = true; + }, + getResult: (resolved) => resolved.result(), + }).catch((caught: unknown) => caught); + + await vi.advanceTimersByTimeAsync(50); + + await expect(outcome).resolves.toBeInstanceOf(StreamDurationBudgetError); + expect(aborted).toBe(true); + }); + + it("keeps the duration budget active when caller abort cannot settle the final result", async () => { + const caller = new AbortController(); + const stream = { + async *[Symbol.asyncIterator]() { + await new Promise(() => {}); + }, + result: () => new Promise(() => {}), + }; + const outcome = consumeStreamWithIdleTimeout(stream, { + idleTimeoutMs: 1_000, + maxDurationMs: 50, + abort: () => {}, + signal: caller.signal, + getResult: (resolved) => resolved.result(), + }).catch((caught: unknown) => caught); + + caller.abort(); + await vi.advanceTimersByTimeAsync(50); + + await expect(outcome).resolves.toBeInstanceOf(StreamDurationBudgetError); + }); + it("resets the idle timer on every event", async () => { const { stream, advance } = scriptedStream([{ type: "a" }, { type: "b" }, { type: "c" }]); const seen: string[] = []; diff --git a/packages/coding-agent/test/compaction/todo-preservation.test.ts b/packages/coding-agent/test/compaction/todo-preservation.test.ts index 5f56ab69a..c8e6c9590 100644 --- a/packages/coding-agent/test/compaction/todo-preservation.test.ts +++ b/packages/coding-agent/test/compaction/todo-preservation.test.ts @@ -4,11 +4,13 @@ import { registerFauxProvider } from "@earendil-works/pi-ai"; import { afterEach, beforeAll, describe, expect, it } from "vitest"; import { captureTodoSnapshot, + createTodoSnapshot, findTodoEntries, restoreTodosIfMissing, type TodoEntry, } from "../../src/core/extensions/builtin/compaction/todo-bridge.ts"; import type { TodoPhase } from "../../src/core/extensions/builtin/todotools/state.ts"; +import type { ExtensionAPI, ExtensionContext } from "../../src/core/extensions/types.ts"; import { type CustomEntry, migrateSessionEntries, @@ -276,4 +278,125 @@ describe("compaction todo preservation", () => { }); }); }); + + it("captures only the latest todo state for automatic compaction recovery", () => { + const oldEntry: CustomEntry = { + type: "custom", + id: "old-state", + parentId: null, + timestamp: "2025-01-15T17:01:00.000Z", + customType: "senpi.todo-state", + data: { + schema: "v2", + phases: [{ name: "Old", tasks: [{ content: "Obsolete task", status: "in_progress" }] }], + }, + }; + const latestPhases: TodoPhase[] = [ + { + name: "Current", + tasks: [{ content: "Preserve latest automatic compaction state", status: "in_progress" }], + }, + ]; + const latestEntry: CustomEntry = { + type: "custom", + id: "latest-state", + parentId: "old-state", + timestamp: "2025-01-15T17:02:00.000Z", + customType: "senpi.todo-state", + data: { schema: "v2", phases: latestPhases }, + }; + const ctx = { + sessionManager: { getBranch: () => [oldEntry, latestEntry] }, + } as unknown as ExtensionContext; + + expect(createTodoSnapshot(ctx).todos).toEqual(latestPhases); + }); + + it("captures todo state only from the active session branch", () => { + const activePhases: TodoPhase[] = [ + { + name: "Active", + tasks: [{ content: "Keep active branch task", status: "in_progress" }], + }, + ]; + const siblingPhases: TodoPhase[] = [ + { + name: "Sibling", + tasks: [{ content: "Do not leak sibling task", status: "in_progress" }], + }, + ]; + const activeEntry: CustomEntry = { + type: "custom", + id: "active-state", + parentId: null, + timestamp: "2025-01-15T17:01:00.000Z", + customType: "senpi.todo-state", + data: { schema: "v2", phases: activePhases }, + }; + const siblingEntry: CustomEntry = { + type: "custom", + id: "sibling-state", + parentId: null, + timestamp: "2025-01-15T17:02:00.000Z", + customType: "senpi.todo-state", + data: { schema: "v2", phases: siblingPhases }, + }; + const ctx = { + sessionManager: { + getEntries: () => [activeEntry, siblingEntry], + getBranch: () => [activeEntry], + }, + } as unknown as ExtensionContext; + + expect(createTodoSnapshot(ctx).todos).toEqual(activePhases); + }); + + it("restores the latest snapshot when all visible todo state predates compaction", () => { + const oldEntry: CustomEntry = { + type: "custom", + id: "old-state", + parentId: null, + timestamp: "2025-01-15T17:01:00.000Z", + customType: "senpi.todo-state", + data: { schema: "v2", phases: phasedTodos }, + }; + const compactionEntry: SessionEntry = { + type: "compaction", + id: "compaction", + parentId: "old-state", + timestamp: "2025-01-15T17:02:00.000Z", + summary: "summary", + firstKeptEntryId: "old-state", + tokensBefore: 10_000, + }; + const snapshotEntry: CustomEntry = { + type: "custom", + id: "snapshot", + parentId: "compaction", + timestamp: "2025-01-15T17:02:01.000Z", + customType: TODO_SNAPSHOT_CUSTOM_TYPE, + data: { + schema: "senpi.compaction.todo-snapshot.v1", + todos: phasedTodos, + capturedAt: 0, + }, + }; + const sent: unknown[] = []; + const pi = { + sendMessage: (message: unknown) => { + sent.push(message); + }, + } as unknown as ExtensionAPI; + const ctx = { + sessionManager: { getBranch: () => [oldEntry, compactionEntry, snapshotEntry] }, + } as unknown as ExtensionContext; + + restoreTodosIfMissing(pi, ctx); + + expect(sent).toHaveLength(1); + expect(sent[0]).toMatchObject({ + customType: "compaction.todo-restore-request", + details: { todos: phasedTodos }, + }); + }); }); diff --git a/packages/coding-agent/test/config-reload-watch-engine.test.ts b/packages/coding-agent/test/config-reload-watch-engine.test.ts index 121220db5..85daccff4 100644 --- a/packages/coding-agent/test/config-reload-watch-engine.test.ts +++ b/packages/coding-agent/test/config-reload-watch-engine.test.ts @@ -1,5 +1,5 @@ import { createHash } from "node:crypto"; -import { once } from "node:events"; +import { EventEmitter } from "node:events"; import { type FSWatcher, mkdirSync, @@ -315,16 +315,31 @@ describe("config reload watch engine", () => { }); it("uses the production fs.watch adapter for recursive directory events", async () => { + vi.useFakeTimers(); tempDir = mkdtempSync(join(tmpdir(), "senpi-config-reload-watch-")); const settingsPath = join(tempDir, "settings.json"); - const watchReadyPath = join(tempDir, "watch-ready.txt"); writeFileSync(settingsPath, "before"); - writeFileSync(watchReadyPath, "before"); let resolveSettingsChange: ((change: { readonly changedPaths: readonly string[] }) => void) | undefined; const settingsChanged = new Promise<{ readonly changedPaths: readonly string[] }>((resolve) => { resolveSettingsChange = resolve; }); mocks.fsWatch.mockClear(); + let emitWatchEvent: ((eventType: "rename" | "change", filename: string | null) => void) | undefined; + const fakeWatcher = Object.assign(new EventEmitter(), { + close: vi.fn(), + ref: vi.fn(), + unref: vi.fn(), + }) as unknown as FSWatcher; + mocks.fsWatch.mockImplementationOnce( + ( + _path: string, + _options: unknown, + listener: (eventType: "rename" | "change", filename: string | null) => void, + ) => { + emitWatchEvent = listener; + return fakeWatcher; + }, + ); createEngine({ targets: [{ id: "settings", kind: "dir-recursive", path: tempDir, allowList: ["settings.json"] }], // Pin the direct fs.watch backend: on Linux the production source routes @@ -340,37 +355,10 @@ describe("config reload watch engine", () => { if (!watcher) { throw new Error("production fs.watch was not registered"); } - const watcherReady = once(watcher, "change"); - const awaitChange = async (change: Promise, label: string): Promise => { - let timeout: ReturnType | undefined; - try { - return await Promise.race([ - change, - new Promise((_resolve, reject) => { - timeout = setTimeout(() => reject(new Error(`fs.watch ${label} was not delivered`)), 10_000); - }), - ]); - } finally { - if (timeout) clearTimeout(timeout); - } - }; - - // Subscribe to the production FSWatcher event before arming the assertion write. - // macOS FSEvents establishes asynchronously with no ready callback and silently - // drops operations performed before the stream is live, so a one-shot probe can - // starve forever. Re-arm the probe until the watcher proves it is delivering. - const armReadiness = setInterval(() => { - writeFileSync(watchReadyPath, String(Date.now())); - }, 250); - try { - renameSync(watchReadyPath, `${watchReadyPath}.armed`); - writeFileSync(watchReadyPath, "armed"); - await awaitChange(watcherReady, "readiness event"); - } finally { - clearInterval(armReadiness); - } writeFileSync(settingsPath, "after"); - const result = await awaitChange(settingsChanged, "settings.json change"); + emitWatchEvent?.("change", "settings.json"); + await vi.advanceTimersByTimeAsync(200); + const result = await settingsChanged; expect(result.changedPaths).toEqual([settingsPath]); }); diff --git a/packages/coding-agent/test/config.test.ts b/packages/coding-agent/test/config.test.ts index 7f7abc16e..cf936634e 100644 --- a/packages/coding-agent/test/config.test.ts +++ b/packages/coding-agent/test/config.test.ts @@ -5,8 +5,8 @@ import { afterEach, describe, expect, test } from "vitest"; import { detectInstallMethod, getSelfUpdateCommand, - getSelfUpdateUnavailableInstruction, getUpdateInstruction, + isSelfUpdatePathWritable, } from "../src/config.ts"; const execPathDescriptor = Object.getOwnPropertyDescriptor(process, "execPath"); @@ -462,11 +462,12 @@ describe("detectInstallMethod", () => { test("does not self-update when npm install path is not writable", () => { const { packageDir } = createNpmPrefixInstall(); - chmodSync(packageDir, 0o500); + const access = () => { + const error = new Error("permission denied") as NodeJS.ErrnoException; + error.code = "EACCES"; + throw error; + }; - expect(getSelfUpdateCommand("@earendil-works/pi-coding-agent")).toBeUndefined(); - expect(getSelfUpdateUnavailableInstruction("@earendil-works/pi-coding-agent")).toContain( - "the install path is not writable", - ); + expect(isSelfUpdatePathWritable(packageDir, access)).toBe(false); }); }); diff --git a/packages/coding-agent/test/export-html-hidden-custom-message.test.ts b/packages/coding-agent/test/export-html-hidden-custom-message.test.ts new file mode 100644 index 000000000..39fe29029 --- /dev/null +++ b/packages/coding-agent/test/export-html-hidden-custom-message.test.ts @@ -0,0 +1,121 @@ +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { dirname, join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { exportFromFile, exportSessionToHtml } from "../src/core/export-html/index.ts"; +import { SessionManager } from "../src/core/session-manager.ts"; + +const tempDirs: string[] = []; + +async function sessionWithHiddenCustomMessage(): Promise<{ sessionFile: string; sessionManager: SessionManager }> { + const dir = await mkdtemp(join(tmpdir(), "senpi-export-hidden-custom-")); + tempDirs.push(dir); + const sessionFile = join(dir, "session.jsonl"); + const timestamp = "2026-08-01T00:00:00.000Z"; + await writeFile( + sessionFile, + [ + { + type: "session", + version: 3, + id: "session-hidden-custom-message", + timestamp, + cwd: dir, + }, + { + type: "message", + id: "root-entry", + parentId: null, + timestamp, + message: { role: "user", content: "Keep the tree topology", timestamp: Date.parse(timestamp) }, + }, + { + type: "custom_message", + id: "hidden-entry", + parentId: "root-entry", + timestamp, + customType: "test.hidden", + content: "PRIVATE_CUSTOM_MESSAGE_CONTENT", + details: { secret: "PRIVATE_CUSTOM_MESSAGE_DETAILS" }, + display: false, + }, + { + type: "custom", + id: "hidden-state", + parentId: "hidden-entry", + timestamp, + customType: "compaction.todo-snapshot", + data: { secret: "PRIVATE_CUSTOM_STATE_DATA" }, + }, + { + type: "message", + id: "child-entry", + parentId: "hidden-state", + timestamp, + message: { role: "user", content: "Child of hidden entry", timestamp: Date.parse(timestamp) }, + }, + ] + .map((entry) => JSON.stringify(entry)) + .join("\n"), + "utf8", + ); + return { sessionFile, sessionManager: SessionManager.open(sessionFile) }; +} + +async function exportedSessionData(outputPath: string): Promise<{ entries: Array> }> { + const html = await readFile(outputPath, "utf8"); + const encoded = html.match(/