diff --git a/apps/desktop/src/main/__tests__/session-sticky-model-contract.test.ts b/apps/desktop/src/main/__tests__/session-sticky-model-contract.test.ts index 6210e60acb..e2558405ac 100644 --- a/apps/desktop/src/main/__tests__/session-sticky-model-contract.test.ts +++ b/apps/desktop/src/main/__tests__/session-sticky-model-contract.test.ts @@ -56,10 +56,14 @@ describe('PR-SESSION-STICKY-MODEL-0 contract', () => { it('preserves sticky model through branch sessions and session summaries', async () => { const runtime = await readFile(resolve(REPO_ROOT, 'packages/runtime/src/session-manager.ts'), 'utf8'); + // #1084 moved branch-session creation into session-branch.ts; the sticky + // model is captured there, behind the branchFromTurn entry point. + const runtimeBranch = await readFile(resolve(REPO_ROOT, 'packages/runtime/src/session-branch.ts'), 'utf8'); const storage = await readFile(resolve(REPO_ROOT, 'packages/storage/src/session-store.ts'), 'utf8'); const core = await readFile(resolve(REPO_ROOT, 'packages/core/src/session.ts'), 'utf8'); - assert.match(runtime, /branchFromTurn[\s\S]*model: header\.model/); + assert.match(runtime, /branchFromTurn[\s\S]*createBranchSession/); + assert.match(runtimeBranch, /createBranchSession[\s\S]*model: header\.model/); assert.match(runtime, /model: h\.model/); assert.match(storage, /model: header\.model/); assert.match(core, /Sticky session default model id, captured when the session is created/); diff --git a/packages/runtime/src/agent-run-recovery.ts b/packages/runtime/src/agent-run-recovery.ts index bb21a27e47..7d1ea70f1d 100644 --- a/packages/runtime/src/agent-run-recovery.ts +++ b/packages/runtime/src/agent-run-recovery.ts @@ -134,7 +134,7 @@ function diagnostic( }; } -function headerLineage( +export function headerLineage( header: AgentRunHeader, ): Partial< Pick< diff --git a/packages/runtime/src/session-branch.ts b/packages/runtime/src/session-branch.ts new file mode 100644 index 0000000000..30859dbf22 --- /dev/null +++ b/packages/runtime/src/session-branch.ts @@ -0,0 +1,220 @@ +/** + * session-branch — branch-session creation and runtime-ledger cloning. + * + * Owns the mechanics of forking a new session from a point in an existing + * session's history: copying conversation messages up to (or before) a turn + * boundary, cloning the corresponding slice of the runtime ledger (AgentRun + * headers + RuntimeEvents) onto the new session, and creating the child + * session header itself. `SessionManager.branchFromTurn` / `branchBeforeTurn` + * call into this module and then translate the result for their own public + * return shape. + */ + +import type { AgentRunHeader, AgentRunStore, RuntimeEvent, RuntimeEventStore } from '@maka/core'; +import type { BranchFromTurnInput, CreateSessionInput } from '@maka/core/runtime-inputs'; +import type { SessionHeader, StoredMessage } from '@maka/core/session'; +import { + classifyTerminalRuntimeLedger, + commitTerminalRunWithRuntimeFact, +} from './terminal-run-commit.js'; +import type { RuntimeReadModelSessionView } from './runtime-read-model.js'; + +export interface SessionBranchDeps { + store: { + readHeader(sessionId: string): Promise; + create(input: CreateSessionInput): Promise; + appendMessage(sessionId: string, m: StoredMessage): Promise; + appendMessages(sessionId: string, ms: StoredMessage[]): Promise; + }; + runStore?: AgentRunStore; + runtimeEventStore?: RuntimeEventStore; + newId: () => string; + now: () => number; +} + +export async function createBranchSession( + deps: SessionBranchDeps, + sessionId: string, + sourceView: RuntimeReadModelSessionView, + copied: StoredMessage[], + input: BranchFromTurnInput, +): Promise { + const header = await deps.store.readHeader(sessionId); + const next = await deps.store.create({ + cwd: header.cwd, + backend: header.backend, + llmConnectionSlug: header.llmConnectionSlug, + model: header.model, + thinkingLevel: header.thinkingLevel, + permissionMode: header.permissionMode, + name: input.name ?? `${header.name} · 分支`, + labels: header.labels, + parentSessionId: sessionId, + branchOfTurnId: input.sourceTurnId, + status: 'active', + }); + await cloneBranchRuntimeLedger(deps, next.id, sourceView, copied); + if (copied.length > 0) await deps.store.appendMessages(next.id, copied); + await deps.store.appendMessage(next.id, { + type: 'system_note', + id: deps.newId(), + ts: deps.now(), + kind: 'session_start', + data: { parentSessionId: sessionId, branchOfTurnId: input.sourceTurnId }, + }); + return deps.store.readHeader(next.id); +} + +async function cloneBranchRuntimeLedger( + deps: SessionBranchDeps, + childSessionId: string, + sourceView: RuntimeReadModelSessionView, + copiedMessages: readonly StoredMessage[], +): Promise { + if (!deps.runStore || !deps.runtimeEventStore) return; + const copiedTurnIds = new Set(); + for (const message of copiedMessages) { + if ('turnId' in message && typeof message.turnId === 'string') + copiedTurnIds.add(message.turnId); + } + if (copiedTurnIds.size === 0) return; + + for (const sourceRun of sourceView.runs) { + if (!copiedTurnIds.has(sourceRun.turnId)) continue; + const sourceEvents = sourceView.events.filter( + (event) => event.runId === sourceRun.runId && copiedTurnIds.has(event.turnId), + ); + if (sourceEvents.length === 0) continue; + + const runId = deps.newId(); + const invocationId = deps.newId(); + const clonedRun = cloneRunHeaderForBranchCreate(sourceRun, childSessionId, runId, invocationId); + await deps.runStore.createRun(clonedRun); + + const sourceTerminalLedger = classifyTerminalRuntimeLedger(sourceRun, sourceEvents); + const clonedEventBySourceId = new Map(); + for (const event of sourceEvents) { + const clonedEvent = cloneRuntimeEventForBranch(event, { + sessionId: childSessionId, + runId, + eventId: deps.newId(), + invocationId, + }); + await deps.runtimeEventStore.appendRuntimeEvent(childSessionId, runId, clonedEvent); + clonedEventBySourceId.set(event.id, clonedEvent); + } + + if (sourceTerminalLedger.kind === 'fact' && isTerminalRunStatus(sourceRun.status)) { + const terminalEvent = clonedEventBySourceId.get(sourceTerminalLedger.fact.terminalEvent.id); + if (!terminalEvent) continue; + await commitTerminalRunWithRuntimeFact({ + runStore: deps.runStore, + runtimeEventStore: deps.runtimeEventStore, + newId: deps.newId, + sessionId: childSessionId, + runId, + turnId: sourceRun.turnId, + status: sourceTerminalLedger.fact.runStatus, + ts: terminalEvent.ts, + terminalEvent, + ...(sourceTerminalLedger.fact.failureClass + ? { failureClass: sourceTerminalLedger.fact.failureClass } + : {}), + ...(sourceRun.failureMessage ? { failureMessage: sourceRun.failureMessage } : {}), + ...(sourceTerminalLedger.fact.abortSource + ? { abortSource: sourceTerminalLedger.fact.abortSource } + : {}), + runEventData: { + recovered: true, + recoveryReason: 'branch_runtime_ledger_clone', + sourceSessionId: sourceRun.sessionId, + sourceRunId: sourceRun.runId, + }, + }); + } + } +} + +function cloneRuntimeEventForBranch( + event: RuntimeEvent, + ids: { sessionId: string; runId: string; eventId: string; invocationId: string }, +): RuntimeEvent { + return { + ...event, + id: ids.eventId, + invocationId: ids.invocationId, + sessionId: ids.sessionId, + runId: ids.runId, + }; +} + +function cloneRunHeaderForBranchCreate( + sourceRun: AgentRunHeader, + childSessionId: string, + runId: string, + invocationId: string, +): AgentRunHeader { + const cloned = { ...sourceRun, invocationId, sessionId: childSessionId, runId }; + if (isTerminalRunStatus(sourceRun.status)) { + cloned.status = 'running'; + delete cloned.completedAt; + delete cloned.failureClass; + delete cloned.failureMessage; + delete cloned.abortSource; + } + return cloned; +} + +export function copyMessagesThroughTurnBoundary( + messages: readonly StoredMessage[], + turnId: string, +): StoredMessage[] { + let lastIndex = -1; + for (let index = 0; index < messages.length; index += 1) { + const message = messages[index]!; + if ((message as { turnId?: string }).turnId === turnId) { + lastIndex = index; + } + } + if (lastIndex < 0) return []; + // Branch v1 copies conversation context only. Turn metadata is intentionally + // not copied into the child session; lineage lives on the child session + // header (`parentSessionId` + `branchOfTurnId`) and future turns. + return messages.slice(0, lastIndex + 1).filter((message) => message.type !== 'turn_state'); +} + +// Exclusive dual of copyMessagesThroughTurnBoundary: every message belonging to +// a turn strictly before the chosen one, dropping it and every later turn. +// Returns null when the turn is absent (so the caller can reject an unknown +// turn), and an empty array when the turn is the first one (a valid branch into +// empty context). Membership, not array position, decides what to keep: the read +// model does not guarantee a turn's messages are contiguous or that a user +// prompt precedes its turn_state in array order, so a positional slice could +// drop an earlier turn's prompt. turn_state is dropped for the same reason as in +// the inclusive copy — lineage lives on the child header, not copied metadata. +export function copyMessagesBeforeTurn( + messages: readonly StoredMessage[], + turnId: string, +): StoredMessage[] | null { + const turnOrder: string[] = []; + const seen = new Set(); + for (const message of messages) { + const messageTurnId = (message as { turnId?: string }).turnId; + if (messageTurnId && !seen.has(messageTurnId)) { + seen.add(messageTurnId); + turnOrder.push(messageTurnId); + } + } + const cut = turnOrder.indexOf(turnId); + if (cut < 0) return null; + const keep = new Set(turnOrder.slice(0, cut)); + return messages.filter((message) => { + if (message.type === 'turn_state') return false; + const messageTurnId = (message as { turnId?: string }).turnId; + return messageTurnId !== undefined && keep.has(messageTurnId); + }); +} + +function isTerminalRunStatus(status: AgentRunHeader['status']): boolean { + return status === 'completed' || status === 'failed' || status === 'cancelled'; +} diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts index 99ef343881..b2a89a9191 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -76,6 +76,11 @@ import { effectiveRunHeaderFromTerminalFact, terminalRunStatusFromRuntimeEvent, } from './terminal-run-commit.js'; +import { + copyMessagesBeforeTurn, + copyMessagesThroughTurnBoundary, + createBranchSession, +} from './session-branch.js'; import type { AgentBackend, BackendStopMode } from '@maka/core/backend-types'; import type { AgentTeamExecutionContext, MakaTool } from './tool-runtime.js'; @@ -85,7 +90,11 @@ import type { ActiveFullCompactBlock } from './active-full-compact.js'; import type { SemanticCompactBlock } from './semantic-compact.js'; import type { HistoryCompactCheckpoint } from './history-compact-checkpoint.js'; import type { AgentRunLineage, RuntimeContinuationFailpoint } from './agent-run.js'; -import { classifyAgentRunRecovery, type AgentRunRecoveryDecision } from './agent-run-recovery.js'; +import { + classifyAgentRunRecovery, + headerLineage, + type AgentRunRecoveryDecision, +} from './agent-run-recovery.js'; import type { InvocationResult, InvocationSource } from './invocation-context.js'; import { RuntimeKernel, type RuntimeKernelLike, type TurnStartOptions } from './runtime-kernel.js'; import { fallbackSessionTitle, sessionTitleSource } from './session-title.js'; @@ -1157,7 +1166,9 @@ export class SessionManager { const copied = copyMessagesThroughTurnBoundary(sourceView.messages, input.sourceTurnId); if (copied.length === 0) throw new Error(`Cannot branch from unknown turn ${input.sourceTurnId}`); - return this.createBranchSession(sessionId, sourceView, copied, input); + return headerToSummary( + await createBranchSession(this.deps, sessionId, sourceView, copied, input), + ); } async branchBeforeTurn(sessionId: string, input: BranchFromTurnInput): Promise { @@ -1167,39 +1178,9 @@ export class SessionManager { // here (the turn is the first one) — it branches to a fresh, empty context. const copied = copyMessagesBeforeTurn(sourceView.messages, input.sourceTurnId); if (copied === null) throw new Error(`Cannot branch before unknown turn ${input.sourceTurnId}`); - return this.createBranchSession(sessionId, sourceView, copied, input); - } - - private async createBranchSession( - sessionId: string, - sourceView: RuntimeReadModelSessionView, - copied: StoredMessage[], - input: BranchFromTurnInput, - ): Promise { - const header = await this.deps.store.readHeader(sessionId); - const next = await this.deps.store.create({ - cwd: header.cwd, - backend: header.backend, - llmConnectionSlug: header.llmConnectionSlug, - model: header.model, - thinkingLevel: header.thinkingLevel, - permissionMode: header.permissionMode, - name: input.name ?? `${header.name} · 分支`, - labels: header.labels, - parentSessionId: sessionId, - branchOfTurnId: input.sourceTurnId, - status: 'active', - }); - await this.cloneBranchRuntimeLedger(next.id, sourceView, copied); - if (copied.length > 0) await this.deps.store.appendMessages(next.id, copied); - await this.deps.store.appendMessage(next.id, { - type: 'system_note', - id: this.deps.newId(), - ts: this.deps.now(), - kind: 'session_start', - data: { parentSessionId: sessionId, branchOfTurnId: input.sourceTurnId }, - }); - return headerToSummary(await this.deps.store.readHeader(next.id)); + return headerToSummary( + await createBranchSession(this.deps, sessionId, sourceView, copied, input), + ); } async respondToPermission(sessionId: string, response: PermissionResponse): Promise { @@ -1385,80 +1366,6 @@ export class SessionManager { ); } - private async cloneBranchRuntimeLedger( - childSessionId: string, - sourceView: RuntimeReadModelSessionView, - copiedMessages: readonly StoredMessage[], - ): Promise { - if (!this.deps.runStore || !this.deps.runtimeEventStore) return; - const copiedTurnIds = new Set(); - for (const message of copiedMessages) { - if ('turnId' in message && typeof message.turnId === 'string') - copiedTurnIds.add(message.turnId); - } - if (copiedTurnIds.size === 0) return; - - for (const sourceRun of sourceView.runs) { - if (!copiedTurnIds.has(sourceRun.turnId)) continue; - const sourceEvents = sourceView.events.filter( - (event) => event.runId === sourceRun.runId && copiedTurnIds.has(event.turnId), - ); - if (sourceEvents.length === 0) continue; - - const runId = this.deps.newId(); - const invocationId = this.deps.newId(); - const clonedRun = cloneRunHeaderForBranchCreate( - sourceRun, - childSessionId, - runId, - invocationId, - ); - await this.deps.runStore.createRun(clonedRun); - - const sourceTerminalLedger = classifyTerminalRuntimeLedger(sourceRun, sourceEvents); - const clonedEventBySourceId = new Map(); - for (const event of sourceEvents) { - const clonedEvent = cloneRuntimeEventForBranch(event, { - sessionId: childSessionId, - runId, - eventId: this.deps.newId(), - invocationId, - }); - await this.deps.runtimeEventStore.appendRuntimeEvent(childSessionId, runId, clonedEvent); - clonedEventBySourceId.set(event.id, clonedEvent); - } - - if (sourceTerminalLedger.kind === 'fact' && isTerminalRunStatus(sourceRun.status)) { - const terminalEvent = clonedEventBySourceId.get(sourceTerminalLedger.fact.terminalEvent.id); - if (!terminalEvent) continue; - await commitTerminalRunWithRuntimeFact({ - runStore: this.deps.runStore, - runtimeEventStore: this.deps.runtimeEventStore, - newId: this.deps.newId, - sessionId: childSessionId, - runId, - turnId: sourceRun.turnId, - status: sourceTerminalLedger.fact.runStatus, - ts: terminalEvent.ts, - terminalEvent, - ...(sourceTerminalLedger.fact.failureClass - ? { failureClass: sourceTerminalLedger.fact.failureClass } - : {}), - ...(sourceRun.failureMessage ? { failureMessage: sourceRun.failureMessage } : {}), - ...(sourceTerminalLedger.fact.abortSource - ? { abortSource: sourceTerminalLedger.fact.abortSource } - : {}), - runEventData: { - recovered: true, - recoveryReason: 'branch_runtime_ledger_clone', - sourceSessionId: sourceRun.sessionId, - sourceRunId: sourceRun.runId, - }, - }); - } - } - } - private async recoverAgentRunsFromLedger( sessionId: string, policy: RecoveryPolicy = { kind: 'best_effort' }, @@ -1870,86 +1777,6 @@ function turnStateLineage( }; } -function cloneRuntimeEventForBranch( - event: RuntimeEvent, - ids: { sessionId: string; runId: string; eventId: string; invocationId: string }, -): RuntimeEvent { - return { - ...event, - id: ids.eventId, - invocationId: ids.invocationId, - sessionId: ids.sessionId, - runId: ids.runId, - }; -} - -function cloneRunHeaderForBranchCreate( - sourceRun: AgentRunHeader, - childSessionId: string, - runId: string, - invocationId: string, -): AgentRunHeader { - const cloned = { ...sourceRun, invocationId, sessionId: childSessionId, runId }; - if (isTerminalRunStatus(sourceRun.status)) { - cloned.status = 'running'; - delete cloned.completedAt; - delete cloned.failureClass; - delete cloned.failureMessage; - delete cloned.abortSource; - } - return cloned; -} - -function copyMessagesThroughTurnBoundary( - messages: readonly StoredMessage[], - turnId: string, -): StoredMessage[] { - let lastIndex = -1; - for (let index = 0; index < messages.length; index += 1) { - const message = messages[index]!; - if ((message as { turnId?: string }).turnId === turnId) { - lastIndex = index; - } - } - if (lastIndex < 0) return []; - // Branch v1 copies conversation context only. Turn metadata is intentionally - // not copied into the child session; lineage lives on the child session - // header (`parentSessionId` + `branchOfTurnId`) and future turns. - return messages.slice(0, lastIndex + 1).filter((message) => message.type !== 'turn_state'); -} - -// Exclusive dual of copyMessagesThroughTurnBoundary: every message belonging to -// a turn strictly before the chosen one, dropping it and every later turn. -// Returns null when the turn is absent (so the caller can reject an unknown -// turn), and an empty array when the turn is the first one (a valid branch into -// empty context). Membership, not array position, decides what to keep: the read -// model does not guarantee a turn's messages are contiguous or that a user -// prompt precedes its turn_state in array order, so a positional slice could -// drop an earlier turn's prompt. turn_state is dropped for the same reason as in -// the inclusive copy — lineage lives on the child header, not copied metadata. -function copyMessagesBeforeTurn( - messages: readonly StoredMessage[], - turnId: string, -): StoredMessage[] | null { - const turnOrder: string[] = []; - const seen = new Set(); - for (const message of messages) { - const messageTurnId = (message as { turnId?: string }).turnId; - if (messageTurnId && !seen.has(messageTurnId)) { - seen.add(messageTurnId); - turnOrder.push(messageTurnId); - } - } - const cut = turnOrder.indexOf(turnId); - if (cut < 0) return null; - const keep = new Set(turnOrder.slice(0, cut)); - return messages.filter((message) => { - if (message.type === 'turn_state') return false; - const messageTurnId = (message as { turnId?: string }).turnId; - return messageTurnId !== undefined && keep.has(messageTurnId); - }); -} - function isTerminalRunStatus(status: AgentRunHeader['status']): boolean { return status === 'completed' || status === 'failed' || status === 'cancelled'; } @@ -2000,19 +1827,6 @@ function runtimeTerminalFactToRecoveryDecision( }; } -function headerLineage(header: AgentRunHeader): AgentRunRecoveryDecision['lineage'] { - return { - ...(header.parentRunId ? { parentRunId: header.parentRunId } : {}), - ...(header.parentTurnId ? { parentTurnId: header.parentTurnId } : {}), - ...(header.retriedFromTurnId ? { retriedFromTurnId: header.retriedFromTurnId } : {}), - ...(header.regeneratedFromTurnId - ? { regeneratedFromTurnId: header.regeneratedFromTurnId } - : {}), - ...(header.branchOfTurnId ? { branchOfTurnId: header.branchOfTurnId } : {}), - ...(header.parentSessionId ? { parentSessionId: header.parentSessionId } : {}), - }; -} - function normalizeAgentOutputMaxEvents(value: number | undefined): number { if (typeof value !== 'number' || !Number.isFinite(value)) return 20; return Math.min(100, Math.max(1, Math.floor(value)));