diff options
Diffstat (limited to 'packages/session-orchestrator/src/orchestrator.ts')
| -rw-r--r-- | packages/session-orchestrator/src/orchestrator.ts | 546 |
1 files changed, 416 insertions, 130 deletions
diff --git a/packages/session-orchestrator/src/orchestrator.ts b/packages/session-orchestrator/src/orchestrator.ts index aaf418a..a2e141a 100644 --- a/packages/session-orchestrator/src/orchestrator.ts +++ b/packages/session-orchestrator/src/orchestrator.ts @@ -131,6 +131,16 @@ export interface StartTurnInput { * system-prompt service when loaded). */ readonly systemPrompt?: string; + /** + * A human-readable title for the conversation tab. When provided, it is + * persisted via `setConversationTitle` AFTER the new-conversation workspace + * setup resolves (so the `meta === null` newness detection still fires and + * `ensureWorkspace` / `setWorkspaceId` / first-turn system-prompt + * construction are NOT skipped) and BEFORE the first message append (so the + * append's auto-title does not overwrite it). Omit to keep the auto-derived + * title. The caller is responsible for trimming/validation. + */ + readonly title?: string; } export type StartTurnResult = @@ -170,6 +180,20 @@ export interface EnqueueResult { readonly queue: readonly QueuedMessage[]; } +/** + * Result of `SessionOrchestrator.cancelQueuedMessage`. `cancelled` is true when + * a message with the given id was found in the conversation's queue and removed + * (it will never run — never delivered as steering, never carried into a new + * turn). `cancelled` is false when the message was not in the queue (already + * drained/delivered, never existed, unknown conversation) OR when the + * message-queue extension isn't loaded (degraded — feature off). `queue` is the + * post-cancel snapshot (empty when no queue extension is loaded). + */ +export interface CancelQueuedMessageResult { + readonly cancelled: boolean; + readonly queue: readonly QueuedMessage[]; +} + export type TurnEventListener = (event: AgentEvent) => void; interface ActiveTurn { @@ -331,6 +355,18 @@ export interface SessionOrchestrator { * returned snapshot is empty (degraded — feature off). */ enqueue(input: EnqueueInput): EnqueueResult; + /** + * Cancel (remove) a SINGLE queued message by id so it never runs. The single + * entry transports call to cancel a queued steering message. Resolves the + * message-queue service lazily (same as `enqueue`); when the extension isn't + * loaded the call degrades to `{ cancelled: false, queue: [] }`. Idempotent — + * cancelling a message that is no longer queued (already drained/delivered) + * returns `{ cancelled: false, ... }` without error. + */ + cancelQueuedMessage(input: { + readonly conversationId: string; + readonly messageId: string; + }): CancelQueuedMessageResult; subscribe(conversationId: string, listener: TurnEventListener): () => void; isActive(conversationId: string): boolean; /** @@ -370,6 +406,8 @@ export interface SessionOrchestrator { systemPrompt?: string; /** Images attached to this turn — see {@link StartTurnInput.images}. */ images?: readonly ImageInput[]; + /** Conversation tab title — see {@link StartTurnInput.title}. */ + title?: string; }): Promise<void>; } @@ -548,6 +586,7 @@ export function createSessionOrchestrator( workspaceId: string, systemPromptOverride: string | undefined, images: readonly ImageInput[] | undefined, + title: string | undefined, ): void { const turnId = generateTurnId(); const promptStartedAt = deps.now?.() ?? Date.now(); @@ -567,14 +606,34 @@ export function createSessionOrchestrator( // The newness flag is also reused to decide whether to construct // (first turn) or get (subsequent turn) the system prompt — see the // providerOpts assembly below. + // + // An explicit `title` (e.g. the CLI `--title` flag) is persisted HERE, + // AFTER the workspace setup resolves — deliberately NOT before the turn. + // Setting it earlier (e.g. in the HTTP route) would pre-create the meta + // row, make `meta !== null`, and fool this newness check into skipping + // `ensureWorkspace` / `setWorkspaceId` / first-turn system-prompt + // construction. By deferring it to here, the title lands after the + // workspace is assigned but BEFORE the first message append (so the + // append's auto-title sees a non-"Untitled" title and preserves it). const workspaceSetupPromise = (async (): Promise<boolean> => { const meta = await deps.conversationStore.getConversationMeta(conversationId); if (meta === null) { await deps.conversationStore.ensureWorkspace(workspaceId); await deps.conversationStore.setWorkspaceId(conversationId, workspaceId); - return true; } - return false; + if (title !== undefined) { + // Best-effort: a title-set failure must NOT break the turn (the + // workspace setup above already succeeded). Log and continue — the + // append's auto-derived title applies instead. + try { + await deps.conversationStore.setConversationTitle(conversationId, title); + } catch (err) { + deps.logger?.child({ conversationId }).warn("orchestrator: title set failure", { + error: err instanceof Error ? err.message : String(err), + }); + } + } + return meta === null; })(); // ALWAYS resolve the effective cwd through getEffectiveCwd, passing the @@ -760,6 +819,11 @@ export function createSessionOrchestrator( conversationId, ...(effectiveCwd !== undefined ? { cwd: effectiveCwd } : {}), ...(effectiveComputerId !== undefined ? { computerId: effectiveComputerId } : {}), + // Thread the turn's abort signal into the filter chain so a filter + // awaiting slow I/O (the MCP tools filter connecting to MCP servers) + // can be interrupted by POST /conversations/:id/stop instead of + // blocking the turn until its own timeout fires. + signal: controller.signal, }); const dispatch = deps.resolveDispatch?.() ?? defaultDispatchPolicy(); const turnLogger = deps.logger?.child({ conversationId, turnId }); @@ -844,17 +908,33 @@ export function createSessionOrchestrator( const drainSteering = queue === undefined ? undefined - : (): readonly ChatMessage[] => { + : async (): Promise<readonly ChatMessage[]> => { const queued = queue.drain(conversationId); if (queued.length === 0) return []; const steerText = queued.map((q) => q.text).join("\n\n"); + const steeringMessage: ChatMessage = { + role: "user", + chunks: [{ type: "text", text: steerText }], + }; + // Persist the injected steering message to the store as part + // of the SAME critical section as the injection, so it is + // never lost. Without this, the message would live only in + // the kernel's in-memory messages array (never persisted), + // so a user could never see it — and in-flight compaction + // (which loads the store) would scrub it. A fire-and-forget + // append would race with the next step's `onStepComplete` + // append and collide on the store's seq counter, so we + // `await` it (the kernel awaits drainSteering). Errors + // propagate (a DB failure ends the turn, matching + // `onStepComplete`'s behavior). + await deps.conversationStore.append(conversationId, [steeringMessage]); emitToHub(conversationId, { type: "steering", conversationId, turnId, text: steerText, }); - return [{ role: "user", chunks: [{ type: "text", text: steerText }] }]; + return [steeringMessage]; }; // Vision handoff: transform the message list for the provider. When the @@ -899,6 +979,81 @@ export function createSessionOrchestrator( ...(effectiveComputerId !== undefined ? { computerId: effectiveComputerId } : {}), ...(deps.now !== undefined ? { now: deps.now } : {}), ...(drainSteering !== undefined ? { drainSteering } : {}), + // In-flight compaction: at every tool-result boundary the kernel + // calls this with the step's usage + the running messages. When the + // context size exceeds the compact-percent threshold (percent of the + // model's context window), the old history is summarized and + // replaced with [summary, ...recent] — mid-turn, without stopping — + // so a long-running turn (e.g. left overnight) does not run out of + // context. This is DISTINCT from the post-seal auto-compact below: + // that one runs AFTER the turn ends (preparing the next turn) and + // refuses while the conversation is active; this one runs DURING the + // turn (saving the running turn) and uses the live step usage (not + // persisted metrics, which are only written at turn end). When the + // threshold is not exceeded, compaction is disabled (percent 0), or + // the model's context window is unknown, it returns void and the + // kernel keeps its history unchanged (a strict no-op). + onStepBoundary: async ({ stepUsage, messages }) => { + const stored = await deps.conversationStore.getCompactPercent(conversationId); + const percent = stored ?? DEFAULT_COMPACT_PERCENT; + if (percent <= 0) return; // auto-compact disabled + // contextSize mirrors the persisted definition: this step's + // inputTokens + outputTokens (the prompt the NEXT step would + // inherit, grown by this step's output). + const contextSize = stepUsage.inputTokens + stepUsage.outputTokens; + if (effectiveModelName === undefined || deps.resolveModelInfo === undefined) return; + const info = await deps.resolveModelInfo(effectiveModelName); + if (info?.contextWindow === undefined) return; + const threshold = Math.floor(info.contextWindow * (percent / 100)); + if (contextSize < threshold) return; // threshold not exceeded + + const keepLastN = DEFAULT_KEEP_LAST_N; + turnLogger?.info("compaction:in-flight", { + conversationId, + turnId, + contextSize, + threshold, + percent, + }); + const outcome = await performCompaction( + { + conversationStore: deps.conversationStore, + resolveProvider: deps.resolveProvider, + ...(deps.resolveModel !== undefined ? { resolveModel: deps.resolveModel } : {}), + ...(deps.resolveSystemPrompt !== undefined + ? { resolveSystemPrompt: deps.resolveSystemPrompt } + : {}), + ...(deps.resolveConcurrencyLimiter !== undefined + ? { resolveConcurrencyLimiter: deps.resolveConcurrencyLimiter } + : {}), + ...(deps.logger !== undefined ? { logger: deps.logger } : {}), + ...(deps.now !== undefined ? { now: deps.now } : {}), + // emit is required by performCompaction; fall back to a no-op + // when the orchestrator was constructed without one (tests). + emit: deps.emit ?? noopEmit, + }, + conversationId, + // Pass the kernel's LIVE messages array (not a store reload): + // it includes mid-turn steering messages (now persisted by + // drainSteering) and is the authoritative prompt state. Using it + // for the split keeps the store write and the kernel's + // replacement aligned (same recent slice) — no DB↔LLM divergence. + { keepLastN, modelName: effectiveModelName, messages }, + ); + if ("error" in outcome) { + turnLogger?.warn("compaction:in-flight:skipped", { + conversationId, + turnId, + error: outcome.error, + }); + return; // too short / empty summary / unknown model → no replacement + } + // Return the compacted history the kernel should adopt. This is + // EXACTLY what performCompaction wrote to the store + // ([summary, ...recent-from-live]), so the kernel's working + // history and the store stay byte-aligned. + return outcome.compactedMessages; + }, }; // Persist the user message at turn start so it has a seq @@ -1021,6 +1176,7 @@ export function createSessionOrchestrator( workspaceId, systemPrompt, images, + title, }) { if (activeTurns.has(conversationId)) { return { started: false, reason: "already-active" }; @@ -1035,6 +1191,7 @@ export function createSessionOrchestrator( workspaceId ?? "default", systemPrompt, images, + title, ); const turn = activeTurns.get(conversationId); const turnId = turn !== undefined ? turn.turnId : ""; @@ -1060,6 +1217,19 @@ export function createSessionOrchestrator( return { startedTurn: false, queue: snapshot }; }, + cancelQueuedMessage({ conversationId, messageId }) { + // When the message-queue extension isn't loaded this degrades: nothing to + // cancel, empty snapshot (feature off). Mirrors `enqueue`'s degraded path. + const queue = deps.resolveQueue?.(); + if (queue === undefined) { + return { cancelled: false, queue: [] }; + } + const beforeLen = queue.getQueue(conversationId).length; + const snapshot = queue.cancel(conversationId, messageId); + const cancelled = snapshot.length < beforeLen; + return { cancelled, queue: snapshot }; + }, + subscribe(conversationId, listener) { let listeners = subscribers.get(conversationId); if (listeners === undefined) { @@ -1140,6 +1310,7 @@ export function createSessionOrchestrator( workspaceId, systemPrompt, images, + title, }) { const turnInput: StartTurnInput = { conversationId, @@ -1151,6 +1322,7 @@ export function createSessionOrchestrator( ...(workspaceId !== undefined ? { workspaceId } : {}), ...(systemPrompt !== undefined ? { systemPrompt } : {}), ...(images !== undefined ? { images } : {}), + ...(title !== undefined ? { title } : {}), }; const result = orchestrator.startTurn(turnInput); if (!result.started) { @@ -1296,6 +1468,16 @@ export function createWarmService( const DEFAULT_KEEP_LAST_N = 10; const DEFAULT_COMPACT_PERCENT = 85; +/** + * No-op emit used as a fallback when the orchestrator is constructed without an + * `emit` (some tests). `performCompaction` requires a non-optional `emit` (it + * emits `conversationCompacted`); the in-flight path degrades to emitting + * nothing rather than skipping compaction entirely. Generic-typed so it + * satisfies `PerformCompactionDeps["emit"]` for any hook payload type. + */ +const noopEmit: <TPayload>(hook: EventHookDescriptor<TPayload>, payload: TPayload) => void = + () => {}; + const COMPACTION_SYSTEM_PROMPT = "You are a conversation summarizer. Summarize the following conversation concord concisely but comprehensively. " + "Focus on key decisions, context, file paths, and any unresolved questions. " + @@ -1317,6 +1499,222 @@ function formatMessagesForSummary(messages: readonly ChatMessage[]): string { .join("\n\n"); } +/** + * Deps for {@link performCompaction} — the subset of `SessionOrchestratorDeps` + * needed to summarize old history, fork it to an archive, and replace it with + * a summary + recent messages. Structural so both the compaction SERVICE + * (`compact`, manual + post-seal auto) and the IN-FLIGHT compaction path (the + * turn loop's `onStepBoundary`) can call the same shared core without duplicating + * the summarization/fork/replace/emit logic. The active-conversation guard and + * the threshold check are the CALLERS' policy (they differ between the two + * paths) and are NOT performed here. + */ +interface PerformCompactionDeps { + readonly conversationStore: ConversationStore; + readonly resolveProvider: () => ProviderContract; + readonly resolveModel?: ( + modelName: string, + ) => { provider: ProviderContract; model: string } | undefined; + readonly resolveSystemPrompt?: () => SystemPromptService | undefined; + readonly resolveConcurrencyLimiter?: () => ConcurrencyLimiter | undefined; + readonly logger?: Logger; + readonly now?: () => number; + readonly emit: <TPayload>(hook: EventHookDescriptor<TPayload>, payload: TPayload) => void; +} + +/** Result of a successful {@link performCompaction}. */ +interface PerformCompactionResult { + readonly summary: string; + readonly newConversationId: string; + readonly messagesSummarized: number; + readonly messagesKept: number; + /** + * The system-role summary message that heads the compacted history + * (`[summaryMessage, ...recentKept]`). Returned so the in-flight caller can + * build the kernel's replacement history with the SAME summary object. + */ + readonly summaryMessage: ChatMessage; + /** + * The full compacted history `[summaryMessage, ...recentKept]` exactly as + * written to the store. The in-flight caller returns this to the kernel so + * the kernel's working history and the store stay byte-aligned (the same + * `recentKept` slice — taken from the caller-supplied live `messages` — is + * used for BOTH the store write and this return value). + */ + readonly compactedMessages: readonly ChatMessage[]; +} + +/** + * The shared compaction core: summarize the oldest `history.length - + * keepLastN` messages via a provider stream, fork the full pre-compaction + * history to an archive (non-destructive), and replace the live history with + * `[summaryMessage, ...recentKept]`. Emits `conversationCompacted`. Returns the + * result (incl. the `summaryMessage` + the `compactedMessages`) or an error. + * + * History source: when `opts.messages` is provided (the in-flight path), it is + * used as the authoritative history — this is the kernel's LIVE messages array, + * which includes mid-turn steering messages (and the vision-transformed + * provider view) that a store reload could miss (the steering persist may not + * have completed, or — before this fix — was never done at all). Using the live + * array keeps the store write and the kernel's replacement aligned (same + * `recentKept` slice), avoiding the DB↔LLM structural divergence where + * independent slices dropped different messages. When `opts.messages` is + * omitted (the post-seal/manual `compact()` path — the turn has ended, so the + * store is stable), the history is loaded from the store. + * + * Performs NO active-conversation guard and NO threshold check — those are the + * callers' policy. No-ops (returns an error) when the conversation is too + * short to compact (≤ keepLastN messages) or the summary is empty. + */ +async function performCompaction( + deps: PerformCompactionDeps, + conversationId: string, + opts: { + readonly keepLastN?: number; + readonly modelName?: string; + /** The kernel's live messages array (in-flight path). Omit to load the store (post-seal/manual). */ + readonly messages?: readonly ChatMessage[]; + }, +): Promise<PerformCompactionResult | { readonly error: string }> { + // Use the caller-supplied live messages (in-flight) or load the store + // (post-seal/manual — the store is stable once the turn has ended). + const history = opts.messages ?? (await deps.conversationStore.load(conversationId)); + const keepLastN = opts?.keepLastN ?? DEFAULT_KEEP_LAST_N; + + if (history.length <= keepLastN) { + return { error: "conversation too short to compact" }; + } + + // Split: old messages to summarize + recent messages to keep. + const toSummarize = history.slice(0, history.length - keepLastN); + const toKeep = history.slice(history.length - keepLastN); + + // Resolve provider + let provider: ProviderContract; + let modelOverride: string | undefined; + if (opts?.modelName !== undefined && deps.resolveModel !== undefined) { + const resolved = deps.resolveModel(opts.modelName); + if (resolved === undefined) return { error: `unknown model: ${opts.modelName}` }; + provider = resolved.provider; + modelOverride = resolved.model; + } else { + provider = deps.resolveProvider(); + } + + // Wrap with concurrency limiting (same as the main turn path). + const compactionLimiter = deps.resolveConcurrencyLimiter?.(); + if (compactionLimiter !== undefined) { + const compactionWorkspaceId = await deps.conversationStore.getWorkspaceId(conversationId); + provider = wrapProviderWithConcurrency( + provider, + compactionLimiter, + conversationId, + compactionWorkspaceId, + deps.now?.() ?? Date.now(), + ); + } + + // Build the summarization request: system prompt + conversation text + instruction + const conversationText = formatMessagesForSummary(toSummarize); + const summaryRequest: ChatMessage = { + role: "user", + chunks: [ + { + type: "text", + text: `Please summarize the following conversation:\n\n${conversationText}`, + }, + ], + }; + + const providerOpts: ProviderStreamOptions = { + maxTokens: 2000, + ...(modelOverride !== undefined ? { model: modelOverride } : {}), + ...(deps.logger !== undefined + ? { logger: deps.logger.child({ conversationId, attrs: { compaction: true } }) } + : {}), + }; + + // Reconstruct the system prompt on compaction (fresh variable + // resolution — files/cwd/time may have changed since construction). + // The construct call also persists the result for future turns. When + // the system-prompt service is unavailable, fall back to the + // compaction-only system prompt (current behavior, no regression). + const systemPromptService = deps.resolveSystemPrompt?.(); + let compactionSystemPrompt: string; + if (systemPromptService !== undefined) { + const cwd = (await deps.conversationStore.getEffectiveCwd(conversationId)) ?? process.cwd(); + const workspaceId = await deps.conversationStore.getWorkspaceId(conversationId); + const computerId = await deps.conversationStore.getEffectiveComputer(conversationId); + const constructed = await systemPromptService.construct(conversationId, cwd, { + ...(opts?.modelName !== undefined ? { model: opts.modelName } : {}), + workspaceId, + ...(computerId !== null ? { computerId } : {}), + }); + compactionSystemPrompt = `${constructed}\n\n${COMPACTION_SYSTEM_PROMPT}`; + } else { + compactionSystemPrompt = COMPACTION_SYSTEM_PROMPT; + } + + // Call the provider and accumulate the summary + let summary = ""; + for await (const event of provider.stream([summaryRequest], [], { + ...providerOpts, + systemPrompt: compactionSystemPrompt, + })) { + if ((event as ProviderEvent).type === "text-delta") { + summary += (event as { delta: string }).delta; + } else if ((event as ProviderEvent).type === "error") { + return { error: (event as { message: string }).message }; + } + } + + if (summary.trim().length === 0) { + return { error: "model produced empty summary" }; + } + + // Non-destructive: fork the full pre-compaction history to a new + // archive conversation. The original conversation keeps its ID + // (so messaging between agents still works) and gets the compacted + // content. The archive inherits the original's compactedFrom, + // creating a chain: A → Y → X → ... + const archiveId = crypto.randomUUID(); + await deps.conversationStore.forkHistory(conversationId, archiveId); + + // Replace history: [system: summary] + the recent kept messages. `toKeep` + // is sliced from the caller-supplied live `messages` (in-flight) — the SAME + // slice returned below as `compactedMessages` — so the store and the kernel's + // working history stay byte-aligned (same messages kept/dropped). + const summaryMessage: ChatMessage = { + role: "system", + chunks: [ + { + type: "text", + text: `The following is a summary of the previous conversation:\n\n${summary}`, + }, + ], + }; + + const compactedMessages: readonly ChatMessage[] = [summaryMessage, ...toKeep]; + await deps.conversationStore.replaceHistory(conversationId, compactedMessages); + await deps.conversationStore.setCompactedFrom(conversationId, archiveId); + + deps.emit(conversationCompacted, { + conversationId, + newConversationId: archiveId, + messagesSummarized: toSummarize.length, + messagesKept: toKeep.length, + }); + + return { + summary, + newConversationId: archiveId, + messagesSummarized: toSummarize.length, + messagesKept: toKeep.length, + summaryMessage, + compactedMessages, + }; +} + export function createCompactionService( deps: SessionOrchestratorDeps & { readonly emit: <TPayload>(hook: EventHookDescriptor<TPayload>, payload: TPayload) => void; @@ -1329,14 +1727,9 @@ export function createCompactionService( return { error: "conversation is generating" }; } - const history = await deps.conversationStore.load(conversationId); - const keepLastN = opts?.keepLastN ?? DEFAULT_KEEP_LAST_N; - - if (history.length <= keepLastN) { - return { error: "conversation too short to compact" }; - } - // Auto mode: check if contextSize exceeds percent of contextWindow. + // The threshold check is the caller's policy (uses persisted turn + // metrics) and is NOT performed by the shared `performCompaction` core. if (opts?.auto === true) { const stored = await deps.conversationStore.getCompactPercent(conversationId); const percent = stored ?? DEFAULT_COMPACT_PERCENT; @@ -1360,129 +1753,22 @@ export function createCompactionService( if (contextSize < threshold) return { error: "threshold not exceeded" }; } - // Split: old messages to summarize + recent messages to keep. - const toSummarize = history.slice(0, history.length - keepLastN); - const toKeep = history.slice(history.length - keepLastN); - - // Resolve provider - let provider: ProviderContract; - let modelOverride: string | undefined; - if (opts?.modelName !== undefined && deps.resolveModel !== undefined) { - const resolved = deps.resolveModel(opts.modelName); - if (resolved === undefined) return { error: `unknown model: ${opts.modelName}` }; - provider = resolved.provider; - modelOverride = resolved.model; - } else { - provider = deps.resolveProvider(); - } - - // Wrap with concurrency limiting (same as the main turn path). - const compactionLimiter = deps.resolveConcurrencyLimiter?.(); - if (compactionLimiter !== undefined) { - const compactionWorkspaceId = await deps.conversationStore.getWorkspaceId(conversationId); - provider = wrapProviderWithConcurrency( - provider, - compactionLimiter, - conversationId, - compactionWorkspaceId, - deps.now?.() ?? Date.now(), - ); - } - - // Build the summarization request: system prompt + conversation text + instruction - const conversationText = formatMessagesForSummary(toSummarize); - const summaryRequest: ChatMessage = { - role: "user", - chunks: [ - { - type: "text", - text: `Please summarize the following conversation:\n\n${conversationText}`, - }, - ], - }; - - const providerOpts: ProviderStreamOptions = { - maxTokens: 2000, - ...(modelOverride !== undefined ? { model: modelOverride } : {}), - ...(deps.logger !== undefined - ? { logger: deps.logger.child({ conversationId, attrs: { compaction: true } }) } - : {}), - }; - - // Reconstruct the system prompt on compaction (fresh variable - // resolution — files/cwd/time may have changed since construction). - // The construct call also persists the result for future turns. When - // the system-prompt service is unavailable, fall back to the - // compaction-only system prompt (current behavior, no regression). - const systemPromptService = deps.resolveSystemPrompt?.(); - let compactionSystemPrompt: string; - if (systemPromptService !== undefined) { - const cwd = (await deps.conversationStore.getEffectiveCwd(conversationId)) ?? process.cwd(); - const workspaceId = await deps.conversationStore.getWorkspaceId(conversationId); - const computerId = await deps.conversationStore.getEffectiveComputer(conversationId); - const constructed = await systemPromptService.construct(conversationId, cwd, { - ...(opts?.modelName !== undefined ? { model: opts.modelName } : {}), - workspaceId, - ...(computerId !== null ? { computerId } : {}), - }); - compactionSystemPrompt = `${constructed}\n\n${COMPACTION_SYSTEM_PROMPT}`; - } else { - compactionSystemPrompt = COMPACTION_SYSTEM_PROMPT; - } - - // Call the provider and accumulate the summary - let summary = ""; - for await (const event of provider.stream([summaryRequest], [], { - ...providerOpts, - systemPrompt: compactionSystemPrompt, - })) { - if ((event as ProviderEvent).type === "text-delta") { - summary += (event as { delta: string }).delta; - } else if ((event as ProviderEvent).type === "error") { - return { error: (event as { message: string }).message }; - } - } - - if (summary.trim().length === 0) { - return { error: "model produced empty summary" }; - } - - // Non-destructive: fork the full pre-compaction history to a new - // archive conversation. The original conversation keeps its ID - // (so messaging between agents still works) and gets the compacted - // content. The archive inherits the original's compactedFrom, - // creating a chain: A → Y → X → ... - const archiveId = crypto.randomUUID(); - await deps.conversationStore.forkHistory(conversationId, archiveId); - - // Replace history: [system: summary] + recent messages - const summaryMessage: ChatMessage = { - role: "system", - chunks: [ - { - type: "text", - text: `The following is a summary of the previous conversation:\n\n${summary}`, - }, - ], - }; - - await deps.conversationStore.replaceHistory(conversationId, [summaryMessage, ...toKeep]); - await deps.conversationStore.setCompactedFrom(conversationId, archiveId); + // Shared summarize + fork + replace + emit core (no active guard, no + // threshold — those are this caller's policy above). The length check + // ("conversation too short to compact") lives inside the core. + const outcome = await performCompaction(deps, conversationId, { + ...(opts?.keepLastN !== undefined ? { keepLastN: opts.keepLastN } : {}), + ...(opts?.modelName !== undefined ? { modelName: opts.modelName } : {}), + }); + if ("error" in outcome) return { error: outcome.error }; + const { summary, newConversationId, messagesSummarized, messagesKept } = outcome; const result: CompactionResult = { summary, - newConversationId: archiveId, - messagesSummarized: toSummarize.length, - messagesKept: toKeep.length, + newConversationId, + messagesSummarized, + messagesKept, }; - - deps.emit(conversationCompacted, { - conversationId, - newConversationId: archiveId, - messagesSummarized: toSummarize.length, - messagesKept: toKeep.length, - }); - return result; }, }; |