diff --git a/apps/desktop/src/main/__tests__/quote-companion-cleanup.test.ts b/apps/desktop/src/main/__tests__/quote-companion-cleanup.test.ts index 90a268e9f6..0c9b9cac64 100644 --- a/apps/desktop/src/main/__tests__/quote-companion-cleanup.test.ts +++ b/apps/desktop/src/main/__tests__/quote-companion-cleanup.test.ts @@ -38,6 +38,44 @@ afterEach(async () => { }); describe('quote companion cleanup authority', () => { + it('forgets a known rejected creation without trying to resume or remove it', async () => { + const workspaceRoot = await createWorkspace(); + let resumes = 0; + let removals = 0; + const authority = createSessionCopyCleanupAuthority({ + workspaceRoot, + resumeSessionCopy: async () => { + resumes += 1; + }, + removeSession: async () => { + removals += 1; + }, + }); + const creation = { + sessionId: 'fork-rejected', + kind: 'branch' as const, + sourceSessionId: 'source-session', + sourceTurnId: 'source-turn', + intent: 'side_conversation' as const, + ownerId: 'web-contents:1', + }; + + await assert.rejects( + authority.ownCreation(creation, async () => { + throw new Error('session busy'); + }), + /session busy/, + ); + assert.deepEqual(await readPendingIds(workspaceRoot), ['fork-rejected']); + + await authority.rejectCreation('fork-rejected'); + + assert.deepEqual(await readPendingIds(workspaceRoot), []); + assert.equal(resumes, 0); + assert.equal(removals, 0); + assert.equal(await authority.ownCreation(creation, async () => 'retried'), 'retried'); + }); + it('releases a rejected creation lease so the same identity can retry', async () => { const workspaceRoot = await createWorkspace(); let removalFails = true; @@ -124,6 +162,7 @@ describe('quote companion cleanup authority', () => { kind: 'branch', sourceSessionId: 'source-session', sourceTurnId: 'source-turn', + intent: 'side_conversation', ownerId: 'web-contents:2', }, async () => { @@ -138,7 +177,7 @@ describe('quote companion cleanup authority', () => { workspaceRoot, processId: 'process-after-crash', resumeSessionCopy: async (creation) => { - events.push(`resume:${creation.sessionId}:${creation.sourceTurnId}`); + events.push(`resume:${creation.sessionId}:${creation.sourceTurnId}:${creation.intent}`); }, removeSession: async (sessionId) => { events.push(`remove:${sessionId}`); @@ -150,7 +189,7 @@ describe('quote companion cleanup authority', () => { failed: [], }); assert.deepEqual(events, [ - 'resume:fork-unknown-create:source-turn', + 'resume:fork-unknown-create:source-turn:side_conversation', 'remove:fork-unknown-create', ]); assert.deepEqual(await readPendingIds(workspaceRoot), []); diff --git a/apps/desktop/src/main/__tests__/quote-companion-disposal.test.ts b/apps/desktop/src/main/__tests__/quote-companion-disposal.test.ts index 6bebbd8a21..8553509e2d 100644 --- a/apps/desktop/src/main/__tests__/quote-companion-disposal.test.ts +++ b/apps/desktop/src/main/__tests__/quote-companion-disposal.test.ts @@ -23,6 +23,7 @@ import type { SessionSummary, TurnRecord } from '@maka/core/session'; import { abandonPendingCompanionCopy, createFakeWorkbarServices, + ensureCompanionFork, performCompanionTurn, type PerformCompanionTurnDeps, type WorkbarServices, @@ -92,6 +93,26 @@ afterEach(async () => { }); describe('quote companion disposal fencing', () => { + it('preserves a retryable busy reason from Side Conversation creation', async () => { + const defaults = createFakeWorkbarServices(); + const sideChat = { + ...defaults.sideChat, + listTurns: async () => [settledTurn('source-turn')], + branchFromTurn: async () => ({ ok: false as const, reason: 'session_busy' as const }), + }; + + assert.deepEqual( + await ensureCompanionFork({ + api: sideChat, + sourceSession, + panelId, + name: 'Side chat', + isDisposed: () => false, + }), + { status: 'error', code: 'fork_source_busy' }, + ); + }); + it('does not start a send when the panel was disposed after fork setup', async () => { let sends = 0; let armed = 0; @@ -152,7 +173,7 @@ describe('quote companion disposal fencing', () => { const sideChat = { ...defaults.sideChat, listTurns: async () => [settledTurn('source-turn')], - branchFromTurn: () => pendingFork.promise, + branchFromTurn: async () => ({ ok: true as const, session: await pendingFork.promise }), cleanupSessionCopy: async (sessionId: string) => { cleaned.push(sessionId); }, diff --git a/apps/desktop/src/main/__tests__/quote-companion-retry.test.ts b/apps/desktop/src/main/__tests__/quote-companion-retry.test.ts new file mode 100644 index 0000000000..4070741470 --- /dev/null +++ b/apps/desktop/src/main/__tests__/quote-companion-retry.test.ts @@ -0,0 +1,261 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { afterEach, test } from 'node:test'; +import { parseHTML } from 'linkedom'; +import { act, createElement } from 'react'; +import { createRoot, type Root } from 'react-dom/client'; +import type { SessionChangedEvent, SessionSummary, TurnRecord } from '@maka/core/session'; +import { + createFakeWorkbarServices, + useQuoteCompanion, + WorkbarServicesProvider, + type WorkbarServices, +} from '../../renderer/features/workbar/testing.js'; + +const originalGlobals = { + document: globalThis.document, + window: globalThis.window, + HTMLElement: globalThis.HTMLElement, + HTMLIFrameElement: globalThis.HTMLIFrameElement, + Event: globalThis.Event, + Node: globalThis.Node, + IS_REACT_ACT_ENVIRONMENT: (globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }) + .IS_REACT_ACT_ENVIRONMENT, +}; + +let mountedRoot: Root | undefined; +const SOURCE_SESSION = session('source-session'); + +afterEach(async () => { + if (mountedRoot) { + await act(async () => { + mountedRoot?.unmount(); + await Promise.resolve(); + }); + } + mountedRoot = undefined; + Object.assign(globalThis, originalGlobals); +}); + +test('retries a busy Side Conversation at the newest settled boundary and clears its banner', async () => { + const parsed = parseHTML('
'); + const { document, window } = parsed; + Object.assign(globalThis, { + document, + window, + HTMLElement: window.HTMLElement, + HTMLIFrameElement: window.HTMLIFrameElement ?? class HTMLIFrameElement {}, + Event: window.Event, + Node: window.Node, + IS_REACT_ACT_ENVIRONMENT: true, + }); + + let listCount = 0; + let sessionChange: ((event: SessionChangedEvent) => void) | undefined; + let releaseRetry: (() => void) | undefined; + const branchInputs: Array<{ sourceTurnId: string; copyId: string }> = []; + const defaults = createFakeWorkbarServices(); + const services: WorkbarServices = { + ...defaults, + sideChat: { + ...defaults.sideChat, + listTurns: async () => { + listCount += 1; + return listCount === 1 + ? [settledTurn('turn-before-busy')] + : [settledTurn('turn-before-busy'), settledTurn('turn-after-busy')]; + }, + branchFromTurn: async (_sessionId, input) => { + branchInputs.push({ sourceTurnId: input.sourceTurnId, copyId: input.copyId }); + if (branchInputs.length === 1) { + return { ok: false as const, reason: 'session_busy' as const }; + } + await new Promise((resolve) => { + releaseRetry = resolve; + }); + return { ok: true as const, session: session('side-conversation') }; + }, + subscribeSessionChanges: (handler) => { + sessionChange = handler; + return () => { + if (sessionChange === handler) sessionChange = undefined; + }; + }, + }, + }; + const container = document.querySelector('#root'); + assert.ok(container); + const root = createRoot(container); + mountedRoot = root; + + await act(async () => { + root.render( + createElement(WorkbarServicesProvider, { + services, + children: createElement(QuoteCompanionProbe), + }), + ); + await Promise.resolve(); + }); + await waitUntil(() => branchInputs.length === 1 && sessionChange !== undefined); + assert.match(container.textContent, /main conversation or a linked task is still running/i); + const probe = container.firstElementChild; + assert.ok(probe); + + await act(async () => { + sessionChange?.({ + reason: 'turn-status-change', + sessionId: 'source-session', + turnId: 'turn-after-busy', + ts: Date.now(), + }); + await Promise.resolve(); + }); + await waitUntil(() => branchInputs.length === 2 && releaseRetry !== undefined); + assert.equal(probe.getAttribute('data-preparing'), 'false'); + assert.match(container.textContent, /main conversation or a linked task is still running/i); + + await act(async () => { + releaseRetry?.(); + await Promise.resolve(); + }); + await waitUntil( + () => probe.getAttribute('data-companion-id') === 'side-conversation', + () => + `branch inputs: ${JSON.stringify(branchInputs)}; companion: ${probe.getAttribute('data-companion-id')}; error: ${probe.getAttribute('data-error')}`, + ); + + assert.deepEqual( + branchInputs.map(({ sourceTurnId }) => sourceTurnId), + ['turn-before-busy', 'turn-after-busy'], + ); + assert.notEqual(branchInputs[0]?.copyId, branchInputs[1]?.copyId); + assert.equal(probe.getAttribute('data-error'), ''); +}); + +test('does not restart foreground setup when the source Session object refreshes', async () => { + const parsed = parseHTML('
'); + const { document, window } = parsed; + Object.assign(globalThis, { + document, + window, + HTMLElement: window.HTMLElement, + HTMLIFrameElement: window.HTMLIFrameElement ?? class HTMLIFrameElement {}, + Event: window.Event, + Node: window.Node, + IS_REACT_ACT_ENVIRONMENT: true, + }); + + let branchCount = 0; + const defaults = createFakeWorkbarServices(); + const services: WorkbarServices = { + ...defaults, + sideChat: { + ...defaults.sideChat, + listTurns: async () => [settledTurn('settled-turn')], + branchFromTurn: async () => { + branchCount += 1; + if (branchCount === 1) { + return { ok: false as const, reason: 'session_busy' as const }; + } + return await new Promise(() => undefined); + }, + }, + }; + const container = document.querySelector('#root'); + assert.ok(container); + const root = createRoot(container); + mountedRoot = root; + + const render = (sourceSession: SessionSummary) => + root.render( + createElement(WorkbarServicesProvider, { + services, + children: createElement(QuoteCompanionProbe, { sourceSession }), + }), + ); + + await act(async () => { + render(session('source-session')); + await Promise.resolve(); + }); + const probe = container.firstElementChild; + assert.ok(probe); + await waitUntil( + () => branchCount === 1 && probe.getAttribute('data-preparing') === 'false', + ); + + await act(async () => { + render(session('source-session')); + await Promise.resolve(); + }); + + assert.equal(branchCount, 1); + assert.equal(probe.getAttribute('data-preparing'), 'false'); +}); + +function QuoteCompanionProbe(props: { sourceSession?: SessionSummary }) { + const companion = useQuoteCompanion({ + panelId: 'retry-panel', + pendingQuotes: [], + sourceSession: props.sourceSession ?? SOURCE_SESSION, + locale: 'en', + onQuotesConsumed: () => undefined, + }); + return createElement('div', { + 'data-error': companion.error ?? '', + 'data-companion-id': companion.companionSession?.id ?? '', + 'data-preparing': String(companion.preparing), + }, companion.error); +} + +function session(id: string): SessionSummary { + return { + id, + name: id, + isFlagged: false, + isArchived: false, + labels: [], + hasUnread: false, + status: 'active', + backend: 'ai-sdk', + llmConnectionSlug: 'test', + connectionLocked: false, + model: 'test-model', + permissionMode: 'ask', + }; +} + +function settledTurn(turnId: string): TurnRecord { + return { turnId, status: 'completed', partialOutputRetained: false }; +} + +async function waitUntil(predicate: () => boolean, diagnostics?: () => string): Promise { + for (let attempt = 0; attempt < 50; attempt += 1) { + if (predicate()) return; + await act(async () => { + await new Promise((resolve) => setImmediate(resolve)); + }); + } + assert.fail( + `Timed out waiting for the Side Conversation state${diagnostics ? ` (${diagnostics()})` : ''}`, + ); +} diff --git a/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts b/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts index 941b9d72e4..887a852605 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-client-uds.test.ts @@ -257,6 +257,7 @@ test('drives the renderer Session catalog facade through real UDS framing', asyn completeComputerUseTurn() {}, createSessionCopyCleanup: () => ({ ownCreation: (_creation, operation) => operation(), + rejectCreation: async () => undefined, cleanup: async () => undefined, schedule: async () => undefined, abandonOwner: async () => undefined, @@ -571,6 +572,7 @@ function ipcHarness() { function unusedSessionCopyCleanup() { return { ownCreation: async (_creation: unknown, operation: () => Promise) => operation(), + async rejectCreation() {}, async cleanup() {}, async schedule() {}, async abandonOwner() {}, diff --git a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts index a576c9c170..ea2d4b24b7 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts @@ -524,6 +524,7 @@ test('does not release or report a Revision the Host retained during cleanup', a removeSessionCopy = removeSession; return { ownCreation: (_creation, operation) => operation(), + rejectCreation: async () => undefined, cleanup: async () => undefined, schedule: async () => undefined, abandonOwner: async () => undefined, @@ -864,6 +865,7 @@ function deps( completeComputerUseTurn() {}, createSessionCopyCleanup: () => ({ ownCreation: (_creation, operation) => operation(), + rejectCreation: async () => undefined, cleanup: async () => undefined, schedule: async () => undefined, abandonOwner: async () => undefined, diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-catalog-running-turns.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-catalog-running-turns.test.ts index 7d2fca2de3..cba9d1a395 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-catalog-running-turns.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-catalog-running-turns.test.ts @@ -36,6 +36,7 @@ test('projects observed running Turn identities into renderer Session lists', as emitSessionsChanged() {}, releaseSessionResources() {}, sessionCopyCleanup: { + async rejectCreation() {}, recover: async () => ({ failed: [] }), } as never, }, @@ -76,6 +77,7 @@ test('merges catalog and observed running Turn identities in stable order', asyn emitSessionsChanged() {}, releaseSessionResources() {}, sessionCopyCleanup: { + async rejectCreation() {}, recover: async () => ({ failed: [] }), } as never, }, diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts index b03be46a23..a59a82cf65 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts @@ -201,13 +201,15 @@ test("retries committed Branch and Revision copies with the renderer-owned ident assert.equal(committed.size, 3); }); -test("marks Runtime Host Branch copies as side conversations", async () => { +test("sends Side Conversation intent and metadata atomically to Runtime Host", async () => { + const copyInputs: unknown[] = []; const metadataUpdates: unknown[] = []; const abandonedOwners: string[] = []; const backgroundErrors: unknown[] = []; const ipc = ipcHarness(); const sessionCopyCleanup = { ownCreation: (_creation: unknown, operation: () => Promise) => operation(), + async rejectCreation() {}, async cleanup() {}, async schedule() {}, async abandonOwner(ownerId: string) { @@ -221,17 +223,20 @@ test("marks Runtime Host Branch copies as side conversations", async () => { registerExecutionIpc( { client: executionClient({ - copySession: async (_kind, input) => ({ - ...session(), - id: input.targetSessionId, - labels: ["source-label"], - }), + copySession: async (_kind, input) => { + copyInputs.push(input); + return { + ...session(), + id: input.targetSessionId, + labels: ["source-label", SIDE_CONVERSATION_SESSION_LABEL], + }; + }, updateSessionMetadata: async (sessionId, patch) => { metadataUpdates.push({ sessionId, patch }); return { ...session(), id: sessionId, - labels: patch.labels ?? [], + labels: patch.labels ?? ['source-label', SIDE_CONVERSATION_SESSION_LABEL], }; }, }), @@ -247,23 +252,26 @@ test("marks Runtime Host Branch copies as side conversations", async () => { ipc, ); - const branch = (await ipc.invoke("sessions:branchFromTurn", "source-session", { + const branchResult = (await ipc.invoke("sessions:branchFromTurn", "source-session", { sourceTurnId: "source-turn", copyId: "side-copy", name: "Side chat", sideConversation: true, - })) as { labels: string[] }; + })) as { ok: true; session: { labels: string[] } }; - assert.deepEqual(metadataUpdates, [ + assert.deepEqual(copyInputs, [ { - sessionId: "side-copy", - patch: { - name: "Side chat", - labels: ["source-label", SIDE_CONVERSATION_SESSION_LABEL], - }, + sourceSessionId: 'source-session', + targetSessionId: 'side-copy', + sourceTurnId: 'source-turn', + intent: 'side_conversation', }, ]); - assert.deepEqual(branch.labels, [ + assert.deepEqual(metadataUpdates, [ + { sessionId: 'side-copy', patch: { name: 'Side chat' } }, + ]); + assert.equal(branchResult.ok, true); + assert.deepEqual(branchResult.session.labels, [ "source-label", SIDE_CONVERSATION_SESSION_LABEL, ]); @@ -276,6 +284,50 @@ test("marks Runtime Host Branch copies as side conversations", async () => { ]); }); +test('returns structured Side Conversation setup failures across IPC', async () => { + for (const reason of ['session_busy', 'operation_unavailable'] as const) { + const ipc = ipcHarness(); + const rejectedCreations: string[] = []; + registerExecutionIpc( + { + client: executionClient({ + copySession: async () => { + throw new RuntimeHostOperationError( + 'session.branch.create', + reason, + 'Side Conversation setup failed', + ); + }, + }), + observer: unusedObserver(), + attachmentApprovals: createAttachmentApprovalRegistry(), + emitSessionsChanged() {}, + stat: async () => ({ size: 0 }), + resizeImage: async (bytes) => bytes, + beforeStop() {}, + newId: () => 'id-1', + sessionCopyCleanup: { + ...unusedSessionCopyCleanup(), + async rejectCreation(sessionId) { + rejectedCreations.push(sessionId); + }, + }, + }, + ipc, + ); + + assert.deepEqual( + await ipc.invoke('sessions:branchFromTurn', 'source-session', { + sourceTurnId: 'source-turn', + copyId: `side-copy-${reason}`, + sideConversation: true, + }), + { ok: false, reason }, + ); + assert.deepEqual(rejectedCreations, [`side-copy-${reason}`]); + } +}); + test("sends canonical content and uploads owned Attachment bytes through the Host", async () => { const starts: unknown[] = []; const uploads: unknown[] = []; @@ -1089,6 +1141,7 @@ function registerExecutionIpc( function unusedSessionCopyCleanup(): RuntimeHostSessionExecutionIpcDeps['sessionCopyCleanup'] { return { ownCreation: async (_creation, operation) => operation(), + async rejectCreation() {}, async cleanup() {}, async schedule() {}, async abandonOwner() {}, diff --git a/apps/desktop/src/main/quote-companion-cleanup.ts b/apps/desktop/src/main/quote-companion-cleanup.ts index 09d4576255..a5e2ed8952 100644 --- a/apps/desktop/src/main/quote-companion-cleanup.ts +++ b/apps/desktop/src/main/quote-companion-cleanup.ts @@ -25,6 +25,7 @@ export interface SessionCopyCreationLease { kind: 'branch' | 'revision'; sourceSessionId: string; sourceTurnId: string; + intent?: 'side_conversation'; ownerId: string; } @@ -61,6 +62,7 @@ export interface SessionCopyCleanupRecovery { export interface SessionCopyCleanupAuthority { ownCreation(creation: SessionCopyCreationLease, operation: () => Promise): Promise; + rejectCreation(sessionId: string): Promise; cleanup(sessionId: string): Promise; schedule(sessionId: string): Promise; abandonOwner(ownerId: string): Promise; @@ -125,6 +127,17 @@ class SessionCopyCleanupAuthorityImpl implements SessionCopyCleanupAuthority { return task; } + async rejectCreation(sessionId: string): Promise { + const normalized = normalizeSessionId(sessionId); + await this.creations.get(normalized)?.operation.catch(() => undefined); + const record = await this.store.read(normalized); + if (!record) return; + if (record.phase !== 'creating') { + throw new Error(`Session copy ${normalized} is no longer awaiting creation`); + } + await this.store.forget(normalized); + } + async cleanup(sessionId: string): Promise { const normalized = normalizeSessionId(sessionId); const active = this.cleanups.get(normalized); @@ -258,6 +271,7 @@ class SqliteSessionCopyCleanupStore implements SessionCopyCleanupStore { kind: creation.kind, sourceSessionId: creation.sourceSessionId, sourceTurnId: creation.sourceTurnId, + ...(creation.intent ? { intent: creation.intent } : {}), }, }; }); @@ -396,6 +410,7 @@ function normalizeCreationLease(creation: SessionCopyCreationLease): SessionCopy kind: creation.kind, sourceSessionId: normalizeSessionId(creation.sourceSessionId), sourceTurnId: normalizeSessionId(creation.sourceTurnId), + ...(creation.intent === 'side_conversation' ? { intent: creation.intent } : {}), ownerId: normalizeOwnerId(creation.ownerId), }; } @@ -409,6 +424,7 @@ function sameCreation( left.kind === right.kind && left.sourceSessionId === right.sourceSessionId && left.sourceTurnId === right.sourceTurnId && + left.intent === right.intent && left.ownerId === right.ownerId ); } @@ -420,7 +436,8 @@ function samePersistedCreation( return ( left.kind === right.kind && left.sourceSessionId === right.sourceSessionId && - left.sourceTurnId === right.sourceTurnId + left.sourceTurnId === right.sourceTurnId && + left.intent === right.intent ); } diff --git a/apps/desktop/src/main/runtime-host-desktop-candidate.ts b/apps/desktop/src/main/runtime-host-desktop-candidate.ts index 37396022b1..071e86bd8a 100644 --- a/apps/desktop/src/main/runtime-host-desktop-candidate.ts +++ b/apps/desktop/src/main/runtime-host-desktop-candidate.ts @@ -130,6 +130,7 @@ export interface DesktopRuntimeHostCandidateDeps { kind: 'branch' | 'revision'; sourceSessionId: string; sourceTurnId: string; + intent?: 'side_conversation'; }) => Promise; }) => SessionCopyCleanupAuthority; readonly registerClientIpc?: ( @@ -638,11 +639,12 @@ export async function createDesktopRuntimeHostCandidate( emitSessionsChanged("deleted", sessionId); return disposition; }, - resumeSessionCopy: async ({ sessionId, kind, sourceSessionId, sourceTurnId }) => { + resumeSessionCopy: async ({ sessionId, kind, sourceSessionId, sourceTurnId, intent }) => { await client.copySession(kind, { sourceSessionId, targetSessionId: sessionId, sourceTurnId, + ...(intent ? { intent } : {}), }); }, }); diff --git a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts index a6653fedb0..3be6608f6a 100644 --- a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts @@ -26,7 +26,6 @@ import { type SessionChangedEvent, type SessionChangedReason, } from '@maka/core/session'; -import { SIDE_CONVERSATION_SESSION_LABEL } from '@maka/core/side-conversation'; import { type ActiveInteractionRequestEvent, type AttachmentRef } from '@maka/core/events'; import { type PermissionMode } from '@maka/core/permission'; import { type SandboxBoundaryResponse } from '@maka/core/sandbox-boundary'; @@ -60,6 +59,10 @@ import type { DesktopTranscriptRangeRequest } from '../preload/transcript-contra import { toDesktopHostSessionSummary } from "./runtime-host-session-catalog-ipc-main.js"; import { mergeWorkspaceFileInlineReferences } from "./session-workspace-inline-references.js"; +type SideConversationBranchResult = + | { readonly ok: true; readonly session: ReturnType } + | { readonly ok: false; readonly reason: 'session_busy' | 'operation_unavailable' }; + type RuntimeHostSessionExecutionClient = Pick< DesktopRuntimeHostClient, | "answerInteraction" @@ -624,36 +627,47 @@ export function registerRuntimeHostSessionExecutionIpc( sourceSessionId: sessionId, targetSessionId: normalized.copyId, sourceTurnId: normalized.sourceTurnId, + ...(normalized.sideConversation ? { intent: 'side_conversation' as const } : {}), }); - let branch = normalized.sideConversation - ? await deps.sessionCopyCleanup.ownCreation( - { - sessionId: normalized.copyId, - kind: 'branch', - sourceSessionId: sessionId, - sourceTurnId: normalized.sourceTurnId, - ownerId: bindCopyOwner(event), - }, - createBranch, - ) - : await createBranch(); - if (normalized.name || normalized.sideConversation) { + let branch; + try { + branch = normalized.sideConversation + ? await deps.sessionCopyCleanup.ownCreation( + { + sessionId: normalized.copyId, + kind: 'branch', + sourceSessionId: sessionId, + sourceTurnId: normalized.sourceTurnId, + intent: 'side_conversation', + ownerId: bindCopyOwner(event), + }, + createBranch, + ) + : await createBranch(); + } catch (error) { + if ( + normalized.sideConversation && + error instanceof RuntimeHostOperationError && + (error.code === 'session_busy' || error.code === 'operation_unavailable') + ) { + await deps.sessionCopyCleanup.rejectCreation(normalized.copyId); + return { + ok: false, + reason: error.code, + } satisfies SideConversationBranchResult; + } + throw error; + } + if (normalized.name) { branch = await deps.client.updateSessionMetadata(branch.id, { - ...(normalized.name ? { name: normalized.name } : {}), - ...(normalized.sideConversation - ? { - labels: [ - ...new Set([ - ...branch.labels, - SIDE_CONVERSATION_SESSION_LABEL, - ]), - ], - } - : {}), + name: normalized.name, }); } deps.emitSessionsChanged("created", branch.id); - return toDesktopHostSessionSummary(branch); + const summary = toDesktopHostSessionSummary(branch); + return normalized.sideConversation + ? ({ ok: true, session: summary } satisfies SideConversationBranchResult) + : summary; }, ); ipcMain.handle( diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts index f05bdac90d..01fc780311 100644 --- a/apps/desktop/src/preload/bridge-contract.d.ts +++ b/apps/desktop/src/preload/bridge-contract.d.ts @@ -200,6 +200,10 @@ export type DesktopBranchFromTurnInput = BranchFromTurnInput & { copyId: string; }; +export type DesktopSideConversationBranchResult = + | { ok: true; session: DesktopSessionSummary } + | { ok: false; reason: 'session_busy' | 'operation_unavailable' }; + export type DesktopReviseBeforeTurnInput = ReviseBeforeTurnInput & { /** Stable target identity for retrying one Desktop copy action. */ copyId: string; @@ -772,7 +776,14 @@ export interface MakaBridge { | { disposition: 'park'; rejectionReasons: string[]; diagnostics: unknown[] } >; regenerateTurn(sessionId: string, input: RegenerateTurnInput): Promise; - branchFromTurn(sessionId: string, input: DesktopBranchFromTurnInput): Promise; + branchFromTurn( + sessionId: string, + input: DesktopBranchFromTurnInput & { sideConversation: true }, + ): Promise; + branchFromTurn( + sessionId: string, + input: DesktopBranchFromTurnInput & { sideConversation?: false }, + ): Promise; reviseBeforeTurn(sessionId: string, input: DesktopReviseBeforeTurnInput): Promise; respondToSandboxBoundary(sessionId: string, response: SandboxBoundaryResponse): Promise; respondToUserQuestion(sessionId: string, response: UserQuestionResponse): Promise; diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts index 67b2c15bb7..ac25fc6ce0 100644 --- a/apps/desktop/src/preload/preload.ts +++ b/apps/desktop/src/preload/preload.ts @@ -30,6 +30,7 @@ import type { PermissionOverlayStartResult, RendererIngestInput, DesktopBranchFromTurnInput, + DesktopSideConversationBranchResult, DesktopReviseBeforeTurnInput, AppUpdateInstallRequest, AppUpdateInstallResult, @@ -608,6 +609,34 @@ async function invokeSessionSummary( return projectSessionSummary(session.scope, summary); } +async function invokeBranchFromTurn( + sessionId: string, + input: DesktopBranchFromTurnInput & { sideConversation: true }, +): Promise; +async function invokeBranchFromTurn( + sessionId: string, + input: DesktopBranchFromTurnInput & { sideConversation?: false }, +): Promise; +async function invokeBranchFromTurn( + sessionId: string, + input: DesktopBranchFromTurnInput, +): Promise { + const ref = await runtimeHostSessionRef(sessionId); + const result = await ipcRenderer.invoke( + 'sessions:branchFromTurn', + ref.scope, + ref.sessionId, + input, + ) as SessionSummary | { ok: true; session: SessionSummary } | { ok: false; reason: string }; + if (input.sideConversation) { + if (!('ok' in result) || result.ok === false) { + return result as DesktopSideConversationBranchResult; + } + return { ok: true, session: projectSessionSummary(ref.scope, result.session) }; + } + return projectSessionSummary(ref.scope, result as SessionSummary); +} + async function invokeSessionInput( channel: string, input: I, @@ -1619,13 +1648,7 @@ const makaBridge = { regenerateTurn(sessionId: string, input: RegenerateTurnInput): Promise { return invokeSessionRuntimeHost('sessions:regenerateTurn', sessionId, input); }, - async branchFromTurn(sessionId: string, input: DesktopBranchFromTurnInput): Promise { - const ref = await runtimeHostSessionRef(sessionId); - const summary = await ipcRenderer.invoke( - 'sessions:branchFromTurn', ref.scope, ref.sessionId, input, - ) as SessionSummary; - return projectSessionSummary(ref.scope, summary); - }, + branchFromTurn: invokeBranchFromTurn, async reviseBeforeTurn(sessionId: string, input: DesktopReviseBeforeTurnInput): Promise { const ref = await runtimeHostSessionRef(sessionId); const summary = await ipcRenderer.invoke( diff --git a/apps/desktop/src/renderer/features/workbar/ports.ts b/apps/desktop/src/renderer/features/workbar/ports.ts index ca959adc65..fb14676742 100644 --- a/apps/desktop/src/renderer/features/workbar/ports.ts +++ b/apps/desktop/src/renderer/features/workbar/ports.ts @@ -36,6 +36,7 @@ import type { PermissionMode } from '@maka/core/permission'; import type { RegenerateTurnInput } from '@maka/core/runtime-inputs'; import type { SandboxBoundaryResponse } from '@maka/core/sandbox-boundary'; import type { + SessionChangedEvent, SessionSummary, StoredMessage, TurnRecord, @@ -212,9 +213,12 @@ export interface SideChatSessionPort { sourceTurnId: string; name?: string; copyId: string; - sideConversation?: boolean; + sideConversation: true; }, - ): Promise; + ): Promise< + | { ok: true; session: SessionSummary } + | { ok: false; reason: 'session_busy' | 'operation_unavailable' } + >; cleanupSessionCopy(sessionId: string): Promise; abandonSessionCopy(sourceSessionId: string, copyId: string): Promise; send( @@ -246,6 +250,7 @@ export interface SideChatSessionPort { sessionId: string, handler: (event: SessionEvent) => void, ): WorkbarUnsubscribe; + subscribeSessionChanges(handler: (event: SessionChangedEvent) => void): WorkbarUnsubscribe; } export interface WorkbarServices { diff --git a/apps/desktop/src/renderer/features/workbar/testing.ts b/apps/desktop/src/renderer/features/workbar/testing.ts index 576cd77fde..f2e3e940b5 100644 --- a/apps/desktop/src/renderer/features/workbar/testing.ts +++ b/apps/desktop/src/renderer/features/workbar/testing.ts @@ -37,6 +37,7 @@ export * from './tools/inspector/session-inspector-overview-model.js'; export * from './tools/side-chat/quote-companion-panel-state.js'; export * from './tools/side-chat/quote-companion-core.js'; export * from './tools/side-chat/quote-companion-visibility.js'; +export { useQuoteCompanion } from './tools/side-chat/use-quote-companion.js'; export * from './tools/terminal/session-terminal-hydration.js'; export * from './tools/terminal/session-terminal-query.js'; export * from './tools/terminal/session-terminal-frame.js'; @@ -136,6 +137,7 @@ export function createFakeWorkbarServices( respondToSandboxBoundary: async () => undefined, respondToUserQuestion: async () => undefined, subscribeEvents: noopSubscription, + subscribeSessionChanges: noopSubscription, }, ...overrides, }; diff --git a/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-core.ts b/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-core.ts index ec5897b7cc..dceb846301 100644 --- a/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-core.ts +++ b/apps/desktop/src/renderer/features/workbar/tools/side-chat/quote-companion-core.ts @@ -50,6 +50,8 @@ import { sessionEventErrorMessage } from '../../../../model-connection-errors.js * `{ok:false}`. */ export type CompanionErrorCode = | 'fork_setup_failed' + | 'fork_source_busy' + | 'fork_unsupported' | 'send_failed' | 'send_rejected'; @@ -247,12 +249,20 @@ export async function ensureCompanionFork( ) { return { status: 'error', code: 'fork_setup_failed' }; } - created = await api.branchFromTurn(sourceSession.id, { + const result = await api.branchFromTurn(sourceSession.id, { sourceTurnId: copyAttempt.sourceTurnId, name, copyId: copyAttempt.copyId, sideConversation: true, }); + if (!result.ok) { + copyAttempt.complete(); + return { + status: 'error', + code: result.reason === 'session_busy' ? 'fork_source_busy' : 'fork_unsupported', + }; + } + created = result.session; } catch { return { status: 'error', code: 'fork_setup_failed' }; } diff --git a/apps/desktop/src/renderer/features/workbar/tools/side-chat/use-quote-companion.ts b/apps/desktop/src/renderer/features/workbar/tools/side-chat/use-quote-companion.ts index 8ba10b32bb..7c4bcb5be8 100644 --- a/apps/desktop/src/renderer/features/workbar/tools/side-chat/use-quote-companion.ts +++ b/apps/desktop/src/renderer/features/workbar/tools/side-chat/use-quote-companion.ts @@ -51,15 +51,15 @@ import { performCompanionTurn, type CompanionErrorCode, type EnsureCompanionForkResult, -} from './quote-companion-core'; +} from './quote-companion-core.js'; import { mergeSettledMessages } from '../../../../settled-message-merge.js'; import { getDesktopConversationCopy } from '../../../../locales/conversation-copy.js'; import { snapshotCompanionQuotes, type CompanionQuoteSnapshot, type StagedCompanionQuote, -} from './quote-companion-panel-state'; -import type { CompanionForkVisibilityEvent } from './quote-companion-visibility'; +} from './quote-companion-panel-state.js'; +import type { CompanionForkVisibilityEvent } from './quote-companion-visibility.js'; export interface UseQuoteCompanionInput { /** Stable owner for the currently mounted panel generation. */ @@ -151,8 +151,11 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan // A created fork is hidden immediately, but is not considered usable until // onForkCommitted promotes it. const pendingForkIdRef = useRef(null); + const sourceSessionRef = useRef(sourceSession); + sourceSessionRef.current = sourceSession; + const sourceSessionId = sourceSession?.id; const sourceSessionIdRef = useRef(sourceSession?.id); - sourceSessionIdRef.current = sourceSession?.id; + sourceSessionIdRef.current = sourceSessionId; const forkSetupPromiseRef = useRef | null>(null); const stopRequestedRef = useRef(false); const activeTurnIdRef = useRef(null); @@ -178,6 +181,7 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan ); const [hasContent, setHasContent] = useState(false); const [error, setError] = useState(null); + const [forkRetryPending, setForkRetryPending] = useState(false); // Bumped whenever the own-turn set changes so the render picks up the new // filter result (the set lives in a ref to stay stable for the event handler). const [, setOwnTurnTick] = useState(0); @@ -264,18 +268,23 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan ); const ensureFork = useCallback( - (name: string): Promise => { + ( + name: string, + options: { readonly showPreparing?: boolean } = {}, + ): Promise => { const existing = companionRef.current; if (existing) return Promise.resolve({ status: 'ready', session: existing }); if (forkSetupPromiseRef.current) return forkSetupPromiseRef.current; - if (!sourceSession) { + const currentSourceSession = sourceSessionRef.current; + if (!currentSourceSession) { return Promise.resolve({ status: 'error', code: 'fork_setup_failed' }); } - setPreparing(true); + const showPreparing = options.showPreparing ?? true; + if (showPreparing) setPreparing(true); const promise = ensureCompanionFork({ api: sideChat, - sourceSession, + sourceSession: currentSourceSession, panelId, name, isDisposed: () => !mountedRef.current, @@ -298,25 +307,67 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan }) .then((result) => { if (result.status === 'ready' && mountedRef.current) { + setForkRetryPending(false); + setError(null); commitFork(result.session); } else if (result.status === 'error' && mountedRef.current) { - setError(copyRef.current.errors.forkSetupFailed); + setForkRetryPending(result.code === 'fork_source_busy'); + const errors = copyRef.current.errors; + setError( + result.code === 'fork_source_busy' + ? errors.forkSourceBusy + : result.code === 'fork_unsupported' + ? errors.forkUnsupported + : errors.forkSetupFailed, + ); } return result; }) .finally(() => { forkSetupPromiseRef.current = null; - if (mountedRef.current) setPreparing(false); + if (showPreparing && mountedRef.current) setPreparing(false); }); forkSetupPromiseRef.current = promise; return promise; }, - [commitFork, mountedRef, panelId, sideChat, sourceSession], + [commitFork, mountedRef, panelId, sideChat], ); useEffect(() => { - if (sourceSession) void ensureFork(copyRef.current.defaultName); - }, [ensureFork, sourceSession]); + if (sourceSessionId) void ensureFork(copyRef.current.defaultName); + }, [ensureFork, sourceSessionId]); + + useEffect(() => { + if (!sourceSessionId || !forkRetryPending) return; + let retrying = false; + const retry = () => { + if (retrying || !mountedRef.current || companionRef.current) return; + retrying = true; + const currentSetup = forkSetupPromiseRef.current; + void (async () => { + if (currentSetup) await currentSetup; + if (!mountedRef.current || companionRef.current) return; + await ensureFork(copyRef.current.defaultName, { showPreparing: false }); + })().finally(() => { + retrying = false; + }); + }; + const unsubscribe = sideChat.subscribeSessionChanges((event) => { + if ( + event.sessionId === sourceSessionId && + (event.reason === 'turn-status-change' || + event.reason === 'status-change' || + event.reason === 'message-appended') + ) { + retry(); + } + }); + const retryTimer = globalThis.setInterval(retry, 2_000); + return () => { + globalThis.clearInterval(retryTimer); + unsubscribe(); + }; + }, [ensureFork, forkRetryPending, mountedRef, sideChat, sourceSessionId]); // The fork is ephemeral (用完即弃): when the panel is dismissed — 退出, // switching source session — unsubscribe and remove the fork so it never @@ -427,6 +478,8 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan const errors = copyRef.current.errors; const byCode: Record = { fork_setup_failed: errors.forkSetupFailed, + fork_source_busy: errors.forkSourceBusy, + fork_unsupported: errors.forkUnsupported, send_failed: errors.sendFailed, send_rejected: errors.sendRejected, }; diff --git a/apps/desktop/src/renderer/locales/conversation-copy.ts b/apps/desktop/src/renderer/locales/conversation-copy.ts index 9fc6c9f9a8..f4ad6a5e3e 100644 --- a/apps/desktop/src/renderer/locales/conversation-copy.ts +++ b/apps/desktop/src/renderer/locales/conversation-copy.ts @@ -280,6 +280,10 @@ export interface DesktopConversationCopy { errors: { /** Reading the source boundary or creating the companion fork failed. */ forkSetupFailed: string; + /** The source or one of its linked child runs is still active. */ + forkSourceBusy: string; + /** The retained source context cannot be represented safely. */ + forkUnsupported: string; /** `sessions.send` was rejected without throwing (e.g. an unresolved skill). */ sendRejected: string; /** `sessions.send` threw / the turn could not be started. */ @@ -558,6 +562,8 @@ const COPY = { }, errors: { forkSetupFailed: '无法创建侧边对话，请稍后重试。', + forkSourceBusy: '主对话或子任务仍在运行，请等待完成后重试。', + forkUnsupported: '当前对话上下文暂不支持创建侧边对话。', sendRejected: '追问未能开始，请稍后重试。', sendFailed: '追问失败，请稍后重试。', settlementFailed: '运行已结束，但消息加载失败。请重试或重新打开侧边对话。', @@ -761,6 +767,9 @@ const COPY = { }, errors: { forkSetupFailed: 'Could not open the side chat. Please try again.', + forkSourceBusy: + 'The main conversation or a linked task is still running. Try again when it finishes.', + forkUnsupported: 'This conversation context cannot be opened as a side chat yet.', sendRejected: 'The companion could not start. Please try again.', sendFailed: 'The companion request failed. Please try again.', settlementFailed: 'The run ended, but its messages could not be loaded. Retry or reopen the side chat.', diff --git a/apps/desktop/src/renderer/platform/desktop/create-workbar-services.ts b/apps/desktop/src/renderer/platform/desktop/create-workbar-services.ts index 3f22d6e2c9..86c9d0ef00 100644 --- a/apps/desktop/src/renderer/platform/desktop/create-workbar-services.ts +++ b/apps/desktop/src/renderer/platform/desktop/create-workbar-services.ts @@ -132,6 +132,7 @@ export function createDesktopWorkbarServices( bridge.sessions.respondToUserQuestion(sessionId, response), subscribeEvents: (sessionId, handler) => bridge.sessions.subscribeEvents(sessionId, handler), + subscribeSessionChanges: (handler) => bridge.sessions.subscribeChanges(handler), }, }; } diff --git a/apps/desktop/stories/session-workbar.stories.tsx b/apps/desktop/stories/session-workbar.stories.tsx index f2f6d14fd6..600fd6ebfa 100644 --- a/apps/desktop/stories/session-workbar.stories.tsx +++ b/apps/desktop/stories/session-workbar.stories.tsx @@ -706,7 +706,7 @@ function bridge(options: { }, ], readSettledMessages: async () => ({ messages: [], settled: true }), - branchFromTurn: async () => SIDE_CHAT_SESSION, + branchFromTurn: async () => ({ ok: true, session: SIDE_CHAT_SESSION }), cleanupSessionCopy: async () => undefined, abandonSessionCopy: async () => undefined, send: async () => ({ ok: true }), @@ -720,6 +720,7 @@ function bridge(options: { respondToSandboxBoundary: async () => undefined, respondToUserQuestion: async () => undefined, subscribeEvents: unsubscribe, + subscribeSessionChanges: unsubscribe, }, }); return (Story) => ( diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 48b545754f..b1dd1e694c 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -158,6 +158,7 @@ export interface SessionConversationCopy { sourceTurnId: string; requestFingerprint: `sha256:${string}`; state: 'preparing' | 'committed'; + intent?: 'side_conversation'; } export type SubagentSessionRuntimeSummary = Omit< @@ -452,7 +453,7 @@ const SUBAGENT_SESSION_SPAWN_IDENTITY_SHAPE = defineObjectShape()( ['kind', 'sourceSessionId', 'sourceTurnId', 'requestFingerprint', 'state'], - [], + ['intent'], ); const SESSION_LINEAGE_ID_MAX_CHARS = 512; const SESSION_LINEAGE_CONTROL_CHARACTERS = /[\u0000-\u001f\u007f]/; @@ -551,6 +552,8 @@ export function isSessionConversationCopy(value: unknown): value is SessionConve isRecord(value) && hasExactShape(value, SESSION_CONVERSATION_COPY_SHAPE) && (value.kind === 'branch' || value.kind === 'revision') && + (value.intent === undefined || + (value.kind === 'branch' && value.intent === 'side_conversation')) && isSessionLineageId(value.sourceSessionId) && isSessionLineageId(value.sourceTurnId) && typeof value.requestFingerprint === 'string' && diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 6726905f71..c4f2fad83b 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -182,6 +182,13 @@ describe('Runtime Host bootstrap protocol', () => { assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 46); }); + test('publishes a new compatibility epoch for Side Conversation copy intent', () => { + // Epoch 47 belongs to project registration preferences on current main. + // Side Conversation adds another closed branch-copy input and therefore + // needs its own later handshake boundary. + assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 47); + }); + test('adds credential rotation without changing existing credential inputs', () => { const issueInput = { principalKind: 'remote_owner', diff --git a/packages/runtime-host/src/__tests__/session-revision-graph-references.test.ts b/packages/runtime-host/src/__tests__/session-revision-graph-references.test.ts index fa4aab83e2..1c61f354ba 100644 --- a/packages/runtime-host/src/__tests__/session-revision-graph-references.test.ts +++ b/packages/runtime-host/src/__tests__/session-revision-graph-references.test.ts @@ -72,6 +72,95 @@ test('Agent Graph revision references preserve only exact terminal provenance', assert.equal(archived.ok, true); }); +test('Side Conversation references accept terminal linked children as snapshots', async () => { + const accepted = await prepare({ kind: 'side_conversation' }); + assert.equal(accepted.ok, true); + if (!accepted.ok) assert.fail('Expected accepted Side Conversation references'); + assert.deepEqual([...accepted.references.keys()], [CHILD_SESSION_ID]); +}); + +test('Side Conversation references accept terminal non-Graph child Sessions as snapshots', async () => { + const accepted = await prepare({ + kind: 'side_conversation', + messages: [linkedSubagentResult('completed')], + sessionHeaders: [sessionHeader(ROOT_SESSION_ID), childHeader({ graph: false })], + }); + assert.equal(accepted.ok, true); + if (!accepted.ok) assert.fail('Expected accepted linked-child snapshot'); + assert.deepEqual([...accepted.references.keys()], [CHILD_SESSION_ID]); +}); + +test('Side Conversation references wait for live Graph and child state', async () => { + for (const input of [ + { graphState: 'live' as const }, + { childActive: true }, + { messages: [linkedSubagentResult('running')] }, + ]) { + const outcome = await prepare({ kind: 'side_conversation', ...input }); + assert.equal(outcome.ok, false); + if (!outcome.ok) assert.equal(outcome.code, 'session_busy'); + } +}); + +test('Side Conversation validates the retained Graph instead of a newer live Graph', async () => { + const sideConversation = await prepare({ + kind: 'side_conversation', + sessionGraphState: 'live', + graphState: 'terminal', + }); + assert.equal(sideConversation.ok, true); + + const revision = await prepare({ + sessionGraphState: 'live', + graphState: 'terminal', + }); + assert.deepEqual(revision, { + ok: false, + code: 'session_busy', + message: 'A retained Agent Graph is not terminal', + }); +}); + +test('Side Conversation rejects a retained child without a terminal result snapshot', async () => { + const outcome = await prepare({ + kind: 'side_conversation', + messages: [], + sessionHeaders: [sessionHeader(ROOT_SESSION_ID), childHeader({ graph: false })], + }); + assert.deepEqual(outcome, { + ok: false, + code: 'operation_unavailable', + message: 'Side Conversation requires a terminal result for every retained linked child', + }); +}); + +test('Side Conversation waits for a live retained child before its result is committed', async () => { + const outcome = await prepare({ + kind: 'side_conversation', + messages: [], + childActive: true, + sessionHeaders: [sessionHeader(ROOT_SESSION_ID), childHeader({ graph: false })], + }); + assert.deepEqual(outcome, { + ok: false, + code: 'session_busy', + message: 'A retained linked child is still active', + }); +}); + +test('Side Conversation waits for a live retained Graph before its result is committed', async () => { + const outcome = await prepare({ + kind: 'side_conversation', + messages: [], + graphState: 'live', + }); + assert.deepEqual(outcome, { + ok: false, + code: 'session_busy', + message: 'A retained Agent Graph is not terminal', + }); +}); + test('Agent Graph revision references reject incomplete or mismatched provenance', async () => { const cases: ReadonlyArray<{ name: string; @@ -228,11 +317,12 @@ test('Agent Graph revision admission includes only retained direct and reference }); interface PrepareOverrides { - readonly kind?: 'branch' | 'revision'; + readonly kind?: 'branch' | 'revision' | 'side_conversation'; readonly messages?: readonly StoredMessage[]; readonly archivedResults?: readonly string[]; readonly sessionHeaders?: readonly SessionHeader[]; readonly runs?: readonly AgentRunHeader[]; + readonly sessionGraphState?: 'absent' | 'live' | 'terminal'; readonly graphState?: 'absent' | 'live' | 'terminal'; readonly artifactTurnId?: string; readonly artifactStatus?: 'live' | 'deleted'; @@ -279,7 +369,8 @@ async function prepare(overrides: PrepareOverrides = {}) { }), }, graph: { - readSessionState: async () => overrides.graphState ?? 'terminal', + readSessionState: async () => + overrides.sessionGraphState ?? overrides.graphState ?? 'terminal', readGraphState: async (_rootSessionId, graphId) => { if (graphId !== agentGraphIdForRootSession(ROOT_SESSION_ID)) { throw new Error('Graph is not bound to this root Session'); diff --git a/packages/runtime-host/src/__tests__/session-revision-protocol.test.ts b/packages/runtime-host/src/__tests__/session-revision-protocol.test.ts index da25a96a24..46c50e0439 100644 --- a/packages/runtime-host/src/__tests__/session-revision-protocol.test.ts +++ b/packages/runtime-host/src/__tests__/session-revision-protocol.test.ts @@ -28,6 +28,63 @@ import { } from '../protocol/index.js'; describe('Session revision protocol', () => { + test('accepts only the Side Conversation branch intent', () => { + assert.deepEqual( + decodeClientFrame({ + requestId: 'request-side-conversation', + operation: 'session.branch.create', + input: { + sourceSessionId: 'source-session', + targetSessionId: 'target-session', + sourceTurnId: 'turn-1', + expectedSourceRevision: 1, + intent: 'side_conversation', + }, + }), + { + requestId: 'request-side-conversation', + operation: 'session.branch.create', + input: { + sourceSessionId: 'source-session', + targetSessionId: 'target-session', + sourceTurnId: 'turn-1', + expectedSourceRevision: 1, + intent: 'side_conversation', + }, + }, + ); + assert.throws( + () => + decodeClientFrame({ + requestId: 'request-invalid-purpose', + operation: 'session.branch.create', + input: { + sourceSessionId: 'source-session', + targetSessionId: 'target-session', + sourceTurnId: 'turn-1', + expectedSourceRevision: 1, + intent: 'ordinary', + }, + }), + isInvalidFrame, + ); + assert.throws( + () => + decodeClientFrame({ + requestId: 'request-revision-purpose', + operation: 'session.revision.create', + input: { + sourceSessionId: 'source-session', + targetSessionId: 'target-session', + sourceTurnId: 'turn-1', + expectedSourceRevision: 1, + intent: 'side_conversation', + }, + }), + isInvalidFrame, + ); + }); + test('rejects aliasing, unknown fields, and mismatched response identities', () => { assert.throws( () => diff --git a/packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts index 20a7ecafa1..5ccf51c26a 100644 --- a/packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts +++ b/packages/runtime-host/src/__tests__/session-revision-two-client-uds.test.ts @@ -63,6 +63,10 @@ const ADMITTED_REVISION_TARGET_ID = 'admitted-revision-target'; const LINEAGE_REVISION_TARGET_ID = 'lineage-revision-target'; const LINEAGE_BRANCH_TARGET_ID = 'lineage-branch-target'; const GRAPH_REVISION_TARGET_ID = 'graph-revision-target'; +const GRAPH_SIDE_CONVERSATION_TARGET_ID = 'graph-side-conversation-target'; +const GRAPH_SIDE_CONVERSATION_REMOVAL_TARGET_ID = 'graph-side-conversation-removal-target'; +const ARCHIVED_SIDE_CONVERSATION_TARGET_ID = 'archived-side-conversation-target'; +const ACTIVE_SOURCE_SIDE_CONVERSATION_TARGET_ID = 'active-source-side-conversation-target'; test('two Clients share exact retryable Session branch and revision authority', { skip: process.platform === 'win32' ? 'Windows SQLite shutdown lifecycle' : false, @@ -116,6 +120,9 @@ test('two Clients share exact retryable Session branch and revision authority', LINEAGE_REVISION_TARGET_ID, LINEAGE_BRANCH_TARGET_ID, GRAPH_REVISION_TARGET_ID, + GRAPH_SIDE_CONVERSATION_TARGET_ID, + ARCHIVED_SIDE_CONVERSATION_TARGET_ID, + ACTIVE_SOURCE_SIDE_CONVERSATION_TARGET_ID, graphChildSessionId, ); } finally { @@ -178,6 +185,50 @@ async function verifyConcurrentRevisionAuthority( }), { kind: 'session', session: null }, ); + const sideConversation = await desktop.request('session.branch.create', { + sourceSessionId: linkedChildSourceSessionId, + targetSessionId: GRAPH_SIDE_CONVERSATION_TARGET_ID, + sourceTurnId: 'linked-turn', + expectedSourceRevision: linkedChildSource.revision, + intent: 'side_conversation', + }); + assert.equal(sideConversation.kind, 'committed'); + if (sideConversation.kind !== 'committed') { + assert.fail('Side Conversation must commit'); + } + const sideConversationSession = requireSessionProjection(sideConversation.session); + assert.ok(sideConversationSession.labels.includes('mode:side_conversation')); + await assert.rejects( + desktop.request('session.branch.create', { + sourceSessionId: linkedChildSourceSessionId, + targetSessionId: GRAPH_SIDE_CONVERSATION_TARGET_ID, + sourceTurnId: 'linked-turn', + expectedSourceRevision: linkedChildSource.revision, + }), + operationError('operation_conflict'), + ); + const removableSideConversation = await desktop.request('session.branch.create', { + sourceSessionId: linkedChildSourceSessionId, + targetSessionId: GRAPH_SIDE_CONVERSATION_REMOVAL_TARGET_ID, + sourceTurnId: 'linked-turn', + expectedSourceRevision: linkedChildSource.revision, + intent: 'side_conversation', + }); + assert.equal(removableSideConversation.kind, 'committed'); + if (removableSideConversation.kind !== 'committed') { + assert.fail('Removable Side Conversation must commit'); + } + const removableSideConversationSession = requireSessionProjection( + removableSideConversation.session, + ); + assert.deepEqual( + await desktop.request('session.remove', { + sessionId: GRAPH_SIDE_CONVERSATION_REMOVAL_TARGET_ID, + expectedRevision: removableSideConversationSession.revision, + }), + { kind: 'removed', sessionId: GRAPH_SIDE_CONVERSATION_REMOVAL_TARGET_ID }, + ); + assert.equal((await querySession(tui, graphChildSessionId)).id, graphChildSessionId); const graphRevision = await desktop.request('session.revision.create', { sourceSessionId: linkedChildSourceSessionId, targetSessionId: GRAPH_REVISION_TARGET_ID, @@ -240,6 +291,14 @@ async function verifyConcurrentRevisionAuthority( }), operationError('operation_unavailable'), ); + const archivedSideConversation = await desktop.request('session.branch.create', { + sourceSessionId: archivedOwnedSourceSessionId, + targetSessionId: ARCHIVED_SIDE_CONVERSATION_TARGET_ID, + sourceTurnId: 'archived-owned-turn', + expectedSourceRevision: archivedOwnedSource.revision, + intent: 'side_conversation', + }); + assert.equal(archivedSideConversation.kind, 'committed'); for (const sessionId of ['metadata-linked-copy-target', 'archived-owned-copy-target']) { assert.deepEqual( await tui.request('session.catalog.query', { @@ -409,6 +468,66 @@ async function verifyConcurrentRevisionAuthority( throw cleanupError; } if (assertionError !== undefined) throw assertionError; + + const activeSourceTurn = requireStartedTurn( + await desktop.startTurn({ + sessionId: sourceSessionId, + turnId: 'active-source-turn', + content: { text: FAKE_ASK_USER_QUESTION_PROMPT }, + }), + ); + assertionError = undefined; + try { + const activeSource = await querySession(desktop, sourceSessionId); + const historicalCopyInput = { + sourceSessionId, + sourceTurnId: 'turn-2', + expectedSourceRevision: activeSource.revision, + }; + await assert.rejects( + tui.request('session.branch.create', { + ...historicalCopyInput, + targetSessionId: 'active-source-ordinary-branch-target', + }), + operationError('session_busy'), + ); + const sideConversation = await tui.request('session.branch.create', { + ...historicalCopyInput, + targetSessionId: ACTIVE_SOURCE_SIDE_CONVERSATION_TARGET_ID, + intent: 'side_conversation', + }); + assert.equal(sideConversation.kind, 'committed'); + if (sideConversation.kind !== 'committed') { + assert.fail('Side Conversation must fork a settled Turn while the source keeps running'); + } + assert.ok( + requireSessionProjection(sideConversation.session).labels.includes( + 'mode:side_conversation', + ), + ); + } catch (error) { + assertionError = error; + } + try { + const stopped = await desktop.stopTurn( + { + sessionId: sourceSessionId, + turnId: 'active-source-turn', + runId: activeSourceTurn.runId, + }, + PROCESS_TIMEOUT_MS, + ); + assert.equal(stopped.status, 'cancelled'); + } catch (cleanupError) { + if (assertionError !== undefined) { + throw new AggregateError( + [assertionError, cleanupError], + 'active-source Side Conversation check failed and parked-turn cleanup failed', + ); + } + throw cleanupError; + } + if (assertionError !== undefined) throw assertionError; } finally { await Promise.allSettled([desktop.close(), tui.close()]); } @@ -1357,6 +1476,9 @@ async function verifyDurableBranch( lineageRevisionTargetId: string, lineageBranchTargetId: string, graphRevisionTargetId: string, + graphSideConversationTargetId: string, + archivedSideConversationTargetId: string, + activeSourceSideConversationTargetId: string, graphChildSessionId: string, ): Promise { const owner = await tryAcquireInteractiveRootOwner(capability); @@ -1408,6 +1530,112 @@ async function verifyDurableBranch( (await execution.sessionStore.readHeaderSnapshot(lineageBranchTargetId)).parentSessionId, lineageRevisionTargetId, ); + const sideConversationHeader = await execution.sessionStore.readHeaderSnapshot( + graphSideConversationTargetId, + ); + assert.equal(sideConversationHeader.conversationCopy?.intent, 'side_conversation'); + assert.ok(sideConversationHeader.labels.includes('mode:side_conversation')); + const sideConversationMessages = await execution.sessionStore.readMessagesSnapshot( + graphSideConversationTargetId, + ); + const sideConversationResult = sideConversationMessages.find( + (message) => message.type === 'tool_result' && message.content.kind === 'agent_swarm', + ); + assert.ok(sideConversationResult?.type === 'tool_result'); + if ( + sideConversationResult?.type !== 'tool_result' || + sideConversationResult.content.kind !== 'agent_swarm' + ) { + assert.fail('Side Conversation must retain the Agent Graph summary'); + } + assert.equal(sideConversationResult.content.items[0]?.summary, 'done'); + assert.equal(sideConversationResult.content.items[0]?.childSessionId, undefined); + assert.equal(sideConversationResult.content.items[0]?.runId, undefined); + const sideConversationArtifactId = sideConversationResult.content.items[0]?.artifactIds[0]; + assert.ok(sideConversationArtifactId); + const activeSourceSideConversationMessages = await execution.sessionStore.readMessagesSnapshot( + activeSourceSideConversationTargetId, + ); + assert.ok(activeSourceSideConversationMessages.some((message) => message.turnId === 'turn-2')); + assert.ok( + activeSourceSideConversationMessages.every( + (message) => message.turnId !== 'active-source-turn', + ), + ); + const sideConversationRuns = await execution.agentRunStore.listSessionRuns( + graphSideConversationTargetId, + ); + const sideConversationRun = sideConversationRuns.find((run) => run.turnId === 'linked-turn'); + assert.ok(sideConversationRun); + const sideConversationRuntimeResult = ( + await execution.runtimeEventStore.readRuntimeEvents( + graphSideConversationTargetId, + sideConversationRun.runId, + ) + ).find((event) => event.content?.kind === 'function_response')?.content; + assert.ok(sideConversationRuntimeResult?.kind === 'function_response'); + if (sideConversationRuntimeResult?.kind !== 'function_response') { + assert.fail('Side Conversation must retain its RuntimeEvent result snapshot'); + } + const runtimeSideConversationResult = decodeCanonicalToolResultContent( + sideConversationRuntimeResult.result, + ); + assert.equal(runtimeSideConversationResult.kind, 'agent_swarm'); + if (runtimeSideConversationResult.kind !== 'agent_swarm') { + assert.fail('Copied RuntimeEvent result must remain an Agent Graph result'); + } + assert.equal(runtimeSideConversationResult.items[0]?.childSessionId, undefined); + assert.equal(runtimeSideConversationResult.items[0]?.runId, undefined); + assert.deepEqual(runtimeSideConversationResult.items[0]?.artifactIds, [ + sideConversationArtifactId, + ]); + assert.deepEqual( + await artifacts.readTextInSession(graphSideConversationTargetId, sideConversationArtifactId), + { + ok: true, + text: 'graph child result', + }, + ); + const archivedSideConversationRuns = await execution.agentRunStore.listSessionRuns( + archivedSideConversationTargetId, + ); + const archivedSideConversationChildRun = archivedSideConversationRuns.find( + (run) => run.turnId === 'archived-owned-child-turn', + ); + assert.ok(archivedSideConversationChildRun); + const archivedSideConversationResult = ( + await execution.runtimeEventStore.readRuntimeEvents( + archivedSideConversationTargetId, + archivedSideConversationChildRun.runId, + ) + ).find((event) => event.content?.kind === 'function_response')?.content; + assert.ok(archivedSideConversationResult?.kind === 'function_response'); + if (archivedSideConversationResult?.kind !== 'function_response') { + assert.fail('Side Conversation must retain its archived tool result placeholder'); + } + const archivedSideConversationContent = decodeCanonicalToolResultContent( + archivedSideConversationResult.result, + ); + assert.equal(archivedSideConversationContent.kind, 'subagent'); + if (archivedSideConversationContent.kind !== 'subagent') { + assert.fail('Copied archived child result must be restored as a static snapshot'); + } + assert.notEqual(archivedSideConversationContent.runId, 'archived-owned-child-run'); + assert.deepEqual(archivedSideConversationContent, { + kind: 'subagent', + agentName: 'Worker', + turnId: 'archived-owned-child-turn', + runId: archivedSideConversationChildRun.runId, + status: 'completed', + permissionMode: 'ask', + summary: 'done', + artifactIds: [], + }); + const archivedSideConversationArtifacts = await artifacts.listPage( + archivedSideConversationTargetId, + { offset: 0, limit: 10 }, + ); + assert.equal(archivedSideConversationArtifacts.total, 0); const graphRevisionMessages = await execution.sessionStore.readMessagesSnapshot(graphRevisionTargetId); const graphResult = graphRevisionMessages.find( diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 2d3569b957..19d5dd1f5b 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -91,7 +91,9 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 47 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 48 as const; +// 48: Session branch creation accepts an explicit Side Conversation intent. +// Older peers reject the strict input shape or cannot apply its snapshot semantics. // 47: Project registration can carry an explicit location preference. Epoch-46 // hosts reject that optional field on the closed registration input. // 46: Queued message content can be edited in place (queue.entry.update). diff --git a/packages/runtime-host/src/protocol/session-revision.ts b/packages/runtime-host/src/protocol/session-revision.ts index a61d995be9..c48f1fd533 100644 --- a/packages/runtime-host/src/protocol/session-revision.ts +++ b/packages/runtime-host/src/protocol/session-revision.ts @@ -17,7 +17,13 @@ * under the License. */ -import { requireCount, requireEntityId, requireExactRecord, requireRecord } from './codec.js'; +import { + requireCount, + requireEntityId, + requireExactRecord, + requireRecord, + requireShapedRecord, +} from './codec.js'; import { invalidProtocolFrame } from './errors.js'; import { defineOperation } from './operation-spec.js'; import { decodeSessionCatalogItem, type SessionCatalogItem } from './session-catalog.js'; @@ -40,6 +46,7 @@ export interface SessionConversationCopyInput { readonly targetSessionId: string; readonly sourceTurnId: string; readonly expectedSourceRevision: number; + readonly intent?: 'side_conversation'; } export type SessionConversationCopyResult = @@ -82,7 +89,7 @@ export const SESSION_REVISION_OPERATION_SPECS = { mode: 'command', availability: 'ready', errors: SESSION_COPY_ERRORS, - decodeInput: decodeSessionConversationCopyInput, + decodeInput: decodeSessionRevisionCopyInput, decodeOutput: decodeSessionConversationCopyResult, assertOutputForInput: assertConversationCopyOutput, }), @@ -104,6 +111,14 @@ export const SESSION_REVISION_OPERATION_SPECS = { }), } as const; +function decodeSessionRevisionCopyInput(value: unknown): SessionConversationCopyInput { + const input = decodeSessionConversationCopyInput(value); + if (input.intent !== undefined) { + throw invalidProtocolFrame('Session revision copy does not support an intent'); + } + return input; +} + function decodeSessionRevisionAbandonInput(value: unknown): SessionRevisionAbandonInput { const input = requireExactRecord(value, 'Session revision abandon input', ['targetSessionId']); return { targetSessionId: requireEntityId(input.targetSessionId, 'targetSessionId') }; @@ -124,17 +139,20 @@ function decodeSessionRevisionAbandonResult(value: unknown): SessionRevisionAban } export function decodeSessionConversationCopyInput(value: unknown): SessionConversationCopyInput { - const input = requireExactRecord(value, 'Session conversation-copy input', [ - 'sourceSessionId', - 'targetSessionId', - 'sourceTurnId', - 'expectedSourceRevision', - ]); + const input = requireShapedRecord( + value, + 'Session conversation-copy input', + ['sourceSessionId', 'targetSessionId', 'sourceTurnId', 'expectedSourceRevision'], + ['intent'], + ); const sourceSessionId = requireEntityId(input.sourceSessionId, 'sourceSessionId'); const targetSessionId = requireEntityId(input.targetSessionId, 'targetSessionId'); if (sourceSessionId === targetSessionId) { throw invalidProtocolFrame('Session conversation copy requires distinct Sessions'); } + if (input.intent !== undefined && input.intent !== 'side_conversation') { + throw invalidProtocolFrame('Invalid Session conversation-copy intent'); + } return { sourceSessionId, targetSessionId, @@ -143,6 +161,7 @@ export function decodeSessionConversationCopyInput(value: unknown): SessionConve input.expectedSourceRevision, 'expected source Session revision', ), + ...(input.intent === 'side_conversation' ? { intent: input.intent } : {}), }; } diff --git a/packages/runtime-host/src/server/session-revision-coordinator.ts b/packages/runtime-host/src/server/session-revision-coordinator.ts index be72f6fc29..4f3fc24538 100644 --- a/packages/runtime-host/src/server/session-revision-coordinator.ts +++ b/packages/runtime-host/src/server/session-revision-coordinator.ts @@ -19,6 +19,7 @@ import { createHash, randomUUID } from 'node:crypto'; import { isDeepResearchSession } from '@maka/core/explore-agent'; +import { SIDE_CONVERSATION_SESSION_LABEL } from '@maka/core/side-conversation'; import type { RuntimeEvent } from '@maka/core/runtime-event'; import type { CreateSessionInput } from '@maka/core/runtime-inputs'; import { @@ -28,6 +29,7 @@ import { type StoredMessage, } from '@maka/core/session'; import { + archivedToolResultContainsLinkedChildReferences, archivedToolResultContainsConversationOwnedReferences, cloneConversationRuntimeLedger, collectConversationCopyLinkedChildReferences, @@ -71,6 +73,7 @@ import { import { purgeSessionSidecars } from './session-sidecar-purge.js'; type ConversationCopyKind = 'branch' | 'revision'; +type ConversationCopySemanticKind = ConversationCopyKind | 'side_conversation'; type ConversationCopyOperationKey = Exclude< SessionRevisionOperationKey, 'session.revision.abandon' @@ -172,9 +175,10 @@ export class HostSessionRevisionCoordinator { kind: ConversationCopyKind, input: SessionConversationCopyInput, ): Promise { - const requestFingerprint = conversationCopyFingerprint(kind, input); + const semanticKind = conversationCopySemanticKind(kind, input); + const requestFingerprint = conversationCopyFingerprint(semanticKind, input); const retry = await this.options.admission.run(input.targetSessionId, async () => - this.#resolveExistingTarget(kind, input, requestFingerprint, true), + this.#resolveExistingTarget(semanticKind, input, requestFingerprint, true), ); if (retry) return retry; @@ -200,7 +204,7 @@ export class HostSessionRevisionCoordinator { // lanes and keep the complete lease through validation and publication. for (let pass = 0; pass < CONVERSATION_COPY_ADMISSION_PASSES; pass += 1) { const result = await this.options.admission.runMany([...admittedSessionIds], (lease) => - this.#copyAdmitted(kind, input, requestFingerprint, lease, admittedSessionIds), + this.#copyAdmitted(semanticKind, input, requestFingerprint, lease, admittedSessionIds), ); if ('ok' in result) return result; for (const sessionId of result.sessionIds) admittedSessionIds.add(sessionId); @@ -245,7 +249,7 @@ export class HostSessionRevisionCoordinator { } async #copyAdmitted( - kind: ConversationCopyKind, + kind: ConversationCopySemanticKind, input: SessionConversationCopyInput, requestFingerprint: `sha256:${string}`, lease: SessionAdmissionLease, @@ -293,7 +297,7 @@ export class HostSessionRevisionCoordinator { 'Deep Research Sessions cannot be copied without an exact research ledger boundary', ); } - if (this.options.isSessionActive(input.sourceSessionId)) { + if (kind !== 'side_conversation' && this.options.isSessionActive(input.sourceSessionId)) { return copyFailure('session_busy', 'Source Session has an active Turn'); } @@ -306,7 +310,7 @@ export class HostSessionRevisionCoordinator { const slice = createConversationCopySlice( source.messages, input.sourceTurnId, - kind === 'branch' ? 'through' : 'before', + kind === 'revision' ? 'before' : 'through', ); if (!slice) { return copyFailure('invalid_request', 'Source turn does not exist'); @@ -388,6 +392,7 @@ export class HostSessionRevisionCoordinator { return copyFailure(linkedReferences.code, linkedReferences.message); } if ( + kind !== 'side_conversation' && archivePreflight.serializedResults.some((serializedResult) => archivedToolResultContainsConversationOwnedReferences( serializedResult, @@ -447,10 +452,34 @@ export class HostSessionRevisionCoordinator { } try { + const archivedSnapshotResults = new Map( + archivePreflight.results + .filter( + ({ serializedResult }) => + archivedToolResultContainsLinkedChildReferences(serializedResult) || + archivedToolResultContainsConversationOwnedReferences( + serializedResult, + input.sourceSessionId, + linkedReferences.references, + ), + ) + .map(({ descriptor, serializedResult }) => [descriptor.artifactId, serializedResult]), + ); const artifactCopy = await this.#artifacts.copyConversationArtifacts({ sourceSessionId: input.sourceSessionId, targetSessionId: input.targetSessionId, turnIds: copyTurnIds, + ...(kind === 'side_conversation' && archivedSnapshotResults.size > 0 + ? { excludeArtifactIds: [...archivedSnapshotResults.keys()] } + : {}), + ...(kind === 'side_conversation' && linkedReferences.references.size > 0 + ? { + linkedArtifacts: [...linkedReferences.references].map(([sessionId, references]) => ({ + sessionId, + artifactIds: [...references.artifactIds], + })), + } + : {}), }); const references = { mode: 'exact' as const, @@ -459,12 +488,17 @@ export class HostSessionRevisionCoordinator { artifactIds: artifactCopy.artifactIds, relativePaths: artifactCopy.relativePaths, linkedChildren: - linkedReferences.references.size > 0 + kind === 'side_conversation' ? { - mode: 'preserve_validated' as const, - references: linkedReferences.references, + mode: 'snapshot' as const, + archivedResults: archivedSnapshotResults, } - : { mode: 'reject' as const }, + : linkedReferences.references.size > 0 + ? { + mode: 'preserve_validated' as const, + references: linkedReferences.references, + } + : { mode: 'reject' as const }, }; const runtimeCopy = await cloneConversationRuntimeLedger({ plan, @@ -481,6 +515,7 @@ export class HostSessionRevisionCoordinator { turnIds: copyTurnIds, ...(slice.beforeTs === undefined ? {} : { beforeTs: slice.beforeTs }), runIdMap: runtimeCopy.runIdMap, + ...(kind === 'side_conversation' ? { linkedChildren: 'snapshot' as const } : {}), }); if (copiedMessages.length > 0) { await this.#stores.sessionStore.appendMessages(input.targetSessionId, [...copiedMessages]); @@ -523,7 +558,14 @@ export class HostSessionRevisionCoordinator { copiedMessages: readonly StoredMessage[], copyTurnIds: readonly string[], ): Promise< - | { readonly ok: true; readonly serializedResults: readonly string[] } + | { + readonly ok: true; + readonly results: readonly { + readonly descriptor: ArchivedToolResultCopyDescriptor; + readonly serializedResult: string; + }[]; + readonly serializedResults: readonly string[]; + } | { readonly ok: false; readonly outcome: ConversationCopyOutcome } > { const archives = collectArchivedToolResultPlaceholders( @@ -537,7 +579,10 @@ export class HostSessionRevisionCoordinator { outcome: copyFailure('persistence_failed', 'Archived tool result metadata is invalid'), }; } - const serializedResults: string[] = []; + const results: Array<{ + descriptor: ArchivedToolResultCopyDescriptor; + serializedResult: string; + }> = []; for (const archive of archives) { const read = await this.#artifacts .readTextInSession(sourceSessionId, archive.artifactId, { @@ -557,13 +602,17 @@ export class HostSessionRevisionCoordinator { ), }; } - serializedResults.push(read.text); + results.push({ descriptor: archive, serializedResult: read.text }); } - return { ok: true, serializedResults }; + return { + ok: true, + results, + serializedResults: results.map(({ serializedResult }) => serializedResult), + }; } async #createInput( - kind: ConversationCopyKind, + kind: ConversationCopySemanticKind, input: SessionConversationCopyInput, requestFingerprint: `sha256:${string}`, source: SessionHeader, @@ -578,17 +627,21 @@ export class HostSessionRevisionCoordinator { collaborationMode: source.collaborationMode ?? 'agent', orchestrationMode: source.orchestrationMode ?? 'default', name: source.name, - labels: [...source.labels], + labels: + kind === 'side_conversation' + ? [...new Set([...source.labels, SIDE_CONVERSATION_SESSION_LABEL])] + : [...source.labels], conversationCopy: { - kind, + kind: persistedConversationCopyKind(kind), sourceSessionId: input.sourceSessionId, sourceTurnId: input.sourceTurnId, requestFingerprint, state: 'preparing', + ...(kind === 'side_conversation' ? { intent: kind } : {}), }, status: 'active', }; - if (kind === 'branch') { + if (kind !== 'revision') { return { ...common, parentSessionId: input.sourceSessionId, @@ -617,7 +670,7 @@ export class HostSessionRevisionCoordinator { } async #resolveExistingTarget( - kind: ConversationCopyKind, + kind: ConversationCopySemanticKind, input: SessionConversationCopyInput, requestFingerprint: `sha256:${string}`, discardPreparing: boolean, @@ -640,7 +693,8 @@ export class HostSessionRevisionCoordinator { } const copy = probe.record.header.conversationCopy; if ( - copy?.kind !== kind || + copy?.kind !== persistedConversationCopyKind(kind) || + copy.intent !== (kind === 'side_conversation' ? kind : undefined) || copy.sourceSessionId !== input.sourceSessionId || copy.sourceTurnId !== input.sourceTurnId || copy.requestFingerprint !== requestFingerprint @@ -679,7 +733,7 @@ export class HostSessionRevisionCoordinator { } async #rollbackIncompleteCopy( - kind: ConversationCopyKind, + kind: ConversationCopySemanticKind, input: SessionConversationCopyInput, requestFingerprint: `sha256:${string}`, message: string, @@ -703,7 +757,7 @@ export class HostSessionRevisionCoordinator { } async #unknownAfterCommitAttempt( - kind: ConversationCopyKind, + kind: ConversationCopySemanticKind, input: SessionConversationCopyInput, requestFingerprint: `sha256:${string}`, message: string, @@ -771,27 +825,24 @@ function isConversationRuntimeFactRewriteUnsupported(error: unknown): boolean { } function conversationCopyFingerprint( - kind: ConversationCopyKind, + kind: ConversationCopySemanticKind, input: SessionConversationCopyInput, ): `sha256:${string}` { // The optimistic source revision guards only the initial create. Once this // target exists, its stable identity must resolve the committed outcome even // if a reconnecting Client observes a newer source revision. - return `sha256:${createHash('sha256') - .update( - JSON.stringify([ - 'session.conversation-copy.v1', - kind, - input.sourceSessionId, - input.targetSessionId, - input.sourceTurnId, - ]), - ) - .digest('hex')}`; + const identity = [ + kind === 'side_conversation' ? 'session.conversation-copy.v2' : 'session.conversation-copy.v1', + kind, + input.sourceSessionId, + input.targetSessionId, + input.sourceTurnId, + ]; + return `sha256:${createHash('sha256').update(JSON.stringify(identity)).digest('hex')}`; } function conversationCopyStartNote( - kind: ConversationCopyKind, + kind: ConversationCopySemanticKind, input: SessionConversationCopyInput, createInput: ConversationCopyCreateInput, ): StoredMessage { @@ -801,7 +852,7 @@ function conversationCopyStartNote( ts: Date.now(), kind: 'session_start', data: - kind === 'branch' + kind !== 'revision' ? { parentSessionId: input.sourceSessionId, branchOfTurnId: input.sourceTurnId, @@ -816,6 +867,17 @@ function conversationCopyStartNote( }; } +function conversationCopySemanticKind( + kind: ConversationCopyKind, + input: SessionConversationCopyInput, +): ConversationCopySemanticKind { + return kind === 'branch' && input.intent === 'side_conversation' ? input.intent : kind; +} + +function persistedConversationCopyKind(kind: ConversationCopySemanticKind): ConversationCopyKind { + return kind === 'revision' ? 'revision' : 'branch'; +} + function isRevisionStartData(value: unknown): boolean { return ( !!value && diff --git a/packages/runtime-host/src/server/session-revision-graph-references.ts b/packages/runtime-host/src/server/session-revision-graph-references.ts index 6aba82b59c..9cd3070324 100644 --- a/packages/runtime-host/src/server/session-revision-graph-references.ts +++ b/packages/runtime-host/src/server/session-revision-graph-references.ts @@ -26,7 +26,7 @@ import { } from '@maka/runtime/conversation-copy'; import type { InteractiveArtifactStoreWriter } from '@maka/storage/artifact-stores'; -type ConversationCopyKind = 'branch' | 'revision'; +type ConversationCopyKind = 'branch' | 'revision' | 'side_conversation'; export type AgentGraphRevisionReferencePreparation = | { @@ -98,49 +98,81 @@ export async function prepareAgentGraphRevisionReferences( return { ok: true, references: new Map() }; } const requestedChildIds = new Set(requests.map((request) => request.childSessionId)); + const unrepresentedChildren = directChildren.filter((child) => !requestedChildIds.has(child.id)); if ( - directChildren.some((child) => !child.subagentParent?.graph || !requestedChildIds.has(child.id)) + input.kind === 'side_conversation' && + unrepresentedChildren.some((child) => dependencies.isSessionActive(child.id)) ) { - return failure( - 'operation_unavailable', - 'Session revision requires a terminal result for every retained Agent Graph child', - ); + return failure('session_busy', 'A retained linked child is still active'); } - const headersById = new Map(input.sessionHeaders.map((header) => [header.id, header])); - try { - if ((await dependencies.graph.readSessionState(input.sourceSessionId)) === 'live') { - return failure('session_busy', 'A retained Agent Graph is not terminal'); - } - } catch { - return failure('operation_unavailable', 'Retained Agent Graph state is unavailable'); - } const referencedGraphs = new Map>(); - for (const request of requests) { - const parent = headersById.get(request.childSessionId)?.subagentParent; - if (!parent?.graph) continue; + const retainGraph = (header: SessionHeader | undefined) => { + const parent = header?.subagentParent; + if (!parent?.graph) return; const graphIds = referencedGraphs.get(parent.parentSessionId) ?? new Set(); graphIds.add(parent.graph.graphId); referencedGraphs.set(parent.parentSessionId, graphIds); + }; + for (const request of requests) retainGraph(headersById.get(request.childSessionId)); + if (input.kind === 'side_conversation') { + for (const child of directChildren) retainGraph(child); } - for (const [rootSessionId, graphIds] of referencedGraphs) { - for (const graphId of graphIds) { - let state: 'absent' | 'live' | 'terminal'; - try { - state = await dependencies.graph.readGraphState(rootSessionId, graphId); - } catch { - return failure('operation_unavailable', 'Retained Agent Graph state is unavailable'); - } - if (state === 'live') { + const retainedSessionGraphFailure = async () => { + try { + if ((await dependencies.graph.readSessionState(input.sourceSessionId)) === 'live') { return failure('session_busy', 'A retained Agent Graph is not terminal'); } - if (state === 'absent') { - return failure( - 'operation_unavailable', - 'Retained Agent Graph control state is unavailable', - ); + } catch { + return failure('operation_unavailable', 'Retained Agent Graph state is unavailable'); + } + return undefined; + }; + const retainedExactGraphFailure = async () => { + for (const [rootSessionId, graphIds] of referencedGraphs) { + for (const graphId of graphIds) { + let state: 'absent' | 'live' | 'terminal'; + try { + state = await dependencies.graph.readGraphState(rootSessionId, graphId); + } catch { + return failure('operation_unavailable', 'Retained Agent Graph state is unavailable'); + } + if (state === 'live') { + return failure('session_busy', 'A retained Agent Graph is not terminal'); + } + if (state === 'absent') { + return failure( + 'operation_unavailable', + 'Retained Agent Graph control state is unavailable', + ); + } } } + return undefined; + }; + if (input.kind === 'side_conversation') { + const graphFailure = await retainedExactGraphFailure(); + if (graphFailure) return graphFailure; + } + if ( + directChildren.some( + (child) => + (input.kind === 'revision' && !child.subagentParent?.graph) || + !requestedChildIds.has(child.id), + ) + ) { + return failure( + 'operation_unavailable', + input.kind === 'side_conversation' + ? 'Side Conversation requires a terminal result for every retained linked child' + : 'Session revision requires a terminal result for every retained Agent Graph child', + ); + } + if (input.kind === 'revision') { + const graphFailure = await retainedSessionGraphFailure(); + if (graphFailure) return graphFailure; + const exactGraphFailure = await retainedExactGraphFailure(); + if (exactGraphFailure) return exactGraphFailure; } const references = new Map(); @@ -151,9 +183,11 @@ export async function prepareAgentGraphRevisionReferences( const parent = child?.subagentParent; if ( !child || - !parent?.graph || + !parent || !familySessionIds.has(parent.parentSessionId) || - !referencedGraphs.get(parent.parentSessionId)?.has(parent.graph.graphId) || + (input.kind === 'revision' && !parent.graph) || + (parent.graph !== undefined && + !referencedGraphs.get(parent.parentSessionId)?.has(parent.graph.graphId)) || !retainedTurnIds.has(parent.spawnedBy.parentTurnId) ) { return failure( diff --git a/packages/runtime/src/__tests__/conversation-copy.test.ts b/packages/runtime/src/__tests__/conversation-copy.test.ts index 8b9b4116ed..0a40519375 100644 --- a/packages/runtime/src/__tests__/conversation-copy.test.ts +++ b/packages/runtime/src/__tests__/conversation-copy.test.ts @@ -35,6 +35,7 @@ import { createWorkspaceRuntimeStore, } from '@maka/storage'; import { + archivedToolResultContainsLinkedChildReferences, archivedToolResultContainsConversationOwnedReferences, cloneConversationRuntimeLedger, collectConversationCopyLinkedChildReferences, @@ -231,6 +232,327 @@ test('conversation copy discovers linked children in persisted retired tool resu ); }); +test('Side Conversation preflight identifies linked-child archive bodies', () => { + assert.equal( + archivedToolResultContainsLinkedChildReferences( + JSON.stringify({ + kind: 'subagent', + childSessionId: 'child-session', + agentName: 'Researcher', + turnId: 'child-turn', + runId: 'child-run', + status: 'completed', + permissionMode: 'ask', + summary: 'done', + artifactIds: ['child-artifact'], + }), + ), + true, + ); + assert.equal( + archivedToolResultContainsLinkedChildReferences( + JSON.stringify({ kind: 'text', text: 'safe result' }), + ), + false, + ); +}); + +test('Side Conversation snapshots remove linked child ownership identifiers', () => { + const message: Extract = { + type: 'tool_result', + id: 'linked-result', + turnId: 'turn-1', + ts: 1, + toolUseId: 'linked-call', + isError: false, + content: { + kind: 'agent_swarm', + status: 'completed', + items: [ + { + itemId: 'item-1', + index: 0, + profile: 'default', + started: true, + childSessionId: 'child-session', + turnId: 'child-turn', + runId: 'child-run', + resumedFromRunId: 'child-parent-run', + status: 'completed', + summary: 'The delegated review found one issue.', + artifactIds: ['child-artifact'], + }, + ], + startedAt: 1, + completedAt: 2, + durationMs: 1, + }, + }; + const rewritten = rewriteConversationCopyMessage(message, { + mode: 'exact', + sourceSessionId: 'session-source', + targetSessionId: 'session-target', + artifactIds: new Map([['child-artifact', 'child-artifact-snapshot']]), + relativePaths: new Map(), + linkedChildren: { + mode: 'snapshot', + archivedResults: new Map(), + }, + runIds: new Map(), + runtimeEventIds: new Map(), + providerTraceIds: new Map(), + }); + + assert.equal(rewritten.type, 'tool_result'); + if (rewritten.type !== 'tool_result' || rewritten.content.kind !== 'agent_swarm') { + assert.fail('Expected the Agent Graph result snapshot'); + } + assert.deepEqual(rewritten.content.items, [ + { + itemId: 'item-1', + index: 0, + profile: 'default', + started: true, + turnId: 'child-turn', + status: 'completed', + summary: 'The delegated review found one issue.', + artifactIds: ['child-artifact-snapshot'], + }, + ]); +}); + +test('Side Conversation snapshots rewrite source-owned Agent Swarm identities', () => { + const message: Extract = { + type: 'tool_result', + id: 'source-owned-result', + turnId: 'turn-1', + ts: 1, + toolUseId: 'source-owned-call', + isError: false, + content: { + kind: 'agent_swarm', + status: 'completed', + items: [ + { + itemId: 'item-1', + index: 0, + profile: 'default', + started: true, + runId: 'run-source', + resumedFromRunId: 'run-parent-source', + status: 'completed', + summary: 'The source-owned run completed.', + artifactIds: ['artifact-source'], + }, + ], + startedAt: 1, + completedAt: 2, + durationMs: 1, + }, + }; + const rewritten = rewriteConversationCopyMessage(message, { + mode: 'exact', + sourceSessionId: 'session-source', + targetSessionId: 'session-target', + artifactIds: new Map([['artifact-source', 'artifact-target']]), + relativePaths: new Map(), + linkedChildren: { + mode: 'snapshot', + archivedResults: new Map(), + }, + runIds: new Map([ + ['run-source', 'run-target'], + ['run-parent-source', 'run-parent-target'], + ]), + runtimeEventIds: new Map(), + providerTraceIds: new Map(), + }); + + assert.equal(rewritten.type, 'tool_result'); + if (rewritten.type !== 'tool_result' || rewritten.content.kind !== 'agent_swarm') { + assert.fail('Expected the source-owned Agent Swarm result'); + } + assert.deepEqual(rewritten.content.items, [ + { + itemId: 'item-1', + index: 0, + profile: 'default', + started: true, + runId: 'run-target', + resumedFromRunId: 'run-parent-target', + status: 'completed', + summary: 'The source-owned run completed.', + artifactIds: ['artifact-target'], + }, + ]); +}); + +test('Side Conversation snapshots rewrite source-owned subagent identities', () => { + const message: Extract = { + type: 'tool_result', + id: 'source-owned-result', + turnId: 'turn-1', + ts: 1, + toolUseId: 'source-owned-call', + isError: false, + content: { + kind: 'subagent', + agentName: 'Researcher', + turnId: 'turn-1', + runId: 'run-source', + status: 'completed', + permissionMode: 'ask', + summary: 'The source-owned run completed.', + artifactIds: ['artifact-source'], + }, + }; + const rewritten = rewriteConversationCopyMessage(message, { + mode: 'exact', + sourceSessionId: 'session-source', + targetSessionId: 'session-target', + artifactIds: new Map([['artifact-source', 'artifact-target']]), + relativePaths: new Map(), + linkedChildren: { + mode: 'snapshot', + archivedResults: new Map(), + }, + runIds: new Map([['run-source', 'run-target']]), + runtimeEventIds: new Map(), + providerTraceIds: new Map(), + }); + + assert.equal(rewritten.type, 'tool_result'); + if (rewritten.type !== 'tool_result' || rewritten.content.kind !== 'subagent') { + assert.fail('Expected the source-owned subagent result'); + } + assert.equal(rewritten.content.runId, 'run-target'); + assert.deepEqual(rewritten.content.artifactIds, ['artifact-target']); +}); + +test('Side Conversation snapshots preserve ordinary archived tool results', () => { + const message: Extract = { + type: 'tool_result', + id: 'archived-result', + turnId: 'turn-1', + ts: 1, + toolUseId: 'tool-1', + isError: false, + content: { + kind: 'json', + value: { + kind: 'maka.archived_tool_result', + rewriteVersion: 1, + artifactId: 'artifact-source', + runtimeEventId: 'event-source', + toolCallId: 'tool-1', + toolName: 'search', + bodySha256: 'a'.repeat(64), + originalEstimatedTokens: 42, + originalBytes: 128, + reason: 'stale_tool_result_pruned_before_compact', + }, + }, + }; + const rewritten = rewriteConversationCopyMessage(message, { + mode: 'exact', + sourceSessionId: 'session-source', + targetSessionId: 'session-target', + artifactIds: new Map([['artifact-source', 'artifact-target']]), + relativePaths: new Map(), + linkedChildren: { + mode: 'snapshot', + archivedResults: new Map(), + }, + runIds: new Map(), + runtimeEventIds: new Map([['event-source', 'event-target']]), + providerTraceIds: new Map(), + }); + + assert.equal(rewritten.type, 'tool_result'); + if (rewritten.type !== 'tool_result' || rewritten.content.kind !== 'json') { + assert.fail('Expected an archived JSON tool result'); + } + assert.deepEqual(rewritten.content.value, { + kind: 'maka.archived_tool_result', + rewriteVersion: 1, + artifactId: 'artifact-target', + runtimeEventId: 'event-target', + toolCallId: 'tool-1', + toolName: 'search', + bodySha256: 'a'.repeat(64), + originalEstimatedTokens: 42, + originalBytes: 128, + reason: 'stale_tool_result_pruned_before_compact', + }); +}); + +test('Side Conversation snapshots retire archived linked-child results', () => { + const message: Extract = { + type: 'tool_result', + id: 'archived-result', + turnId: 'turn-1', + ts: 1, + toolUseId: 'tool-1', + isError: false, + content: { + kind: 'json', + value: { + kind: 'maka.archived_tool_result', + rewriteVersion: 1, + artifactId: 'artifact-source', + runtimeEventId: 'event-source', + toolCallId: 'tool-1', + toolName: 'subagent', + bodySha256: 'b'.repeat(64), + originalEstimatedTokens: 42, + originalBytes: 128, + reason: 'stale_tool_result_pruned_before_compact', + }, + }, + }; + const rewritten = rewriteConversationCopyMessage(message, { + mode: 'exact', + sourceSessionId: 'session-source', + targetSessionId: 'session-target', + artifactIds: new Map([['child-artifact', 'child-artifact-snapshot']]), + relativePaths: new Map(), + linkedChildren: { + mode: 'snapshot', + archivedResults: new Map([ + [ + 'artifact-source', + JSON.stringify({ + kind: 'subagent', + childSessionId: 'child-session', + agentName: 'Researcher', + turnId: 'child-turn', + runId: 'child-run', + status: 'completed', + permissionMode: 'ask', + summary: 'The archived review found one issue.', + artifactIds: ['child-artifact'], + }), + ], + ]), + }, + runIds: new Map(), + runtimeEventIds: new Map([['event-source', 'event-target']]), + providerTraceIds: new Map(), + }); + + assert.equal(rewritten.type, 'tool_result'); + if (rewritten.type !== 'tool_result') assert.fail('Expected a tool result'); + assert.deepEqual(rewritten.content, { + kind: 'subagent', + agentName: 'Researcher', + turnId: 'child-turn', + status: 'completed', + permissionMode: 'ask', + summary: 'The archived review found one issue.', + artifactIds: ['child-artifact-snapshot'], + }); +}); + test('conversation copy slices exact turns on inclusive and exclusive boundaries', () => { const messages = [ { type: 'user', id: 'user-1', turnId: 'turn-1', ts: 1, text: 'first' }, diff --git a/packages/runtime/src/conversation-copy.ts b/packages/runtime/src/conversation-copy.ts index bff9cc5309..d013f0b7d7 100644 --- a/packages/runtime/src/conversation-copy.ts +++ b/packages/runtime/src/conversation-copy.ts @@ -87,6 +87,10 @@ export type ConversationCopyArtifactReferenceMap = readonly relativePaths: ReadonlyMap; readonly linkedChildren: | { readonly mode: 'reject' } + | { + readonly mode: 'snapshot'; + readonly archivedResults: ReadonlyMap; + } | { readonly mode: 'preserve_validated'; readonly references: ReadonlyMap; @@ -532,6 +536,20 @@ export function archivedToolResultContainsConversationOwnedReferences( return false; } +export function archivedToolResultContainsLinkedChildReferences(serializedResult: string): boolean { + const value = deserializeToolResultArchive(serializedResult); + if (isArchivedToolResultPlaceholder(value)) return false; + try { + return ( + conversationCopyLinkedChildReferences( + decodePersistedToolResultContent(markPersisted(value)), + ).length > 0 + ); + } catch { + return false; + } +} + export function conversationCopyLinkedChildReferences( content: ToolResultContent, ): readonly ConversationCopyLinkedChildReference[] { @@ -1010,9 +1028,11 @@ function rewriteRuntimeEventReferences( } : {}), ...(event.refs.artifactId - ? { - artifactId: rewriteOwnedArtifactId(event.refs.artifactId, references), - } + ? archivedSnapshotResult(event.refs.artifactId, references) !== undefined + ? {} + : { + artifactId: rewriteOwnedArtifactId(event.refs.artifactId, references), + } : {}), ...(event.refs.sourceInvocationId ? { @@ -1111,6 +1131,8 @@ function rewriteToolResultContent( return { ...content, ref: rewriteStorageRef(content.ref, references) }; } if (content.kind === 'archived_tool_result') { + const snapshot = rewriteArchivedSnapshot(content, references); + if (snapshot) return snapshot; return { ...content, runtimeEventId: rewriteOwnedId( @@ -1124,12 +1146,21 @@ function rewriteToolResultContent( }; } if (content.kind === 'json' && isArchivedToolResultPlaceholder(content.value)) { + const snapshot = rewriteArchivedSnapshot(content.value, references); + if (snapshot) return snapshot; return { ...content, value: rewriteArchivedToolResult(content.value, references), }; } if (content.kind === 'subagent') { + if (linkedChildrenAreSnapshots(references) && content.childSessionId) { + const { childSessionId: _childSessionId, runId: _runId, ...snapshot } = content; + return { + ...snapshot, + artifactIds: rewriteSnapshotArtifactIds(content.artifactIds, references), + }; + } return { ...content, ...(content.runId @@ -1153,6 +1184,18 @@ function rewriteToolResultContent( return { ...content, items: content.items.map((item) => { + if (linkedChildrenAreSnapshots(references) && item.childSessionId) { + const { + childSessionId: _childSessionId, + runId: _runId, + resumedFromRunId: _resumedFromRunId, + ...snapshot + } = item; + return { + ...snapshot, + artifactIds: rewriteSnapshotArtifactIds(item.artifactIds, references), + }; + } return { ...item, ...(item.runId @@ -1183,6 +1226,8 @@ function rewriteRuntimeToolResult( references: ConversationCopyMessageReferenceMap, ): unknown { if (isArchivedToolResultPlaceholder(value)) { + const snapshot = rewriteArchivedSnapshot(value, references); + if (snapshot) return snapshot; return rewriteArchivedToolResult(value, references); } let content: ToolResultContent; @@ -1206,6 +1251,7 @@ function validatedExternalChildReferences( references: ConversationCopyMessageReferenceMap, ): ConversationCopyExternalChildReferences | undefined { if (references.mode === 'preserve_external') return undefined; + if (references.linkedChildren.mode === 'snapshot') return undefined; if (references.linkedChildren.mode === 'reject') { throw new Error(`Conversation copy cannot retain linked child Session ${childSessionId}`); } @@ -1216,6 +1262,81 @@ function validatedExternalChildReferences( return external; } +function linkedChildrenAreSnapshots(references: ConversationCopyMessageReferenceMap): boolean { + return references.mode === 'exact' && references.linkedChildren.mode === 'snapshot'; +} + +function rewriteSnapshotArtifactIds( + artifactIds: readonly string[], + references: ConversationCopyMessageReferenceMap, +): readonly string[] { + if (references.mode !== 'exact' || references.linkedChildren.mode !== 'snapshot') { + return artifactIds; + } + return artifactIds.map((artifactId) => + requiredMappedId(references.artifactIds, artifactId, 'linked Artifact'), + ); +} + +function rewriteArchivedSnapshot( + value: + | ArchivedToolResultPlaceholder + | Extract, + references: ConversationCopyMessageReferenceMap, +): ToolResultContent | undefined { + const serializedResult = archivedSnapshotResult(value.artifactId, references); + if (serializedResult === undefined) return undefined; + const archived = deserializeToolResultArchive(serializedResult); + if (isArchivedToolResultPlaceholder(archived)) { + return unavailableArchivedToolResult(value, references); + } + try { + const decoded = decodePersistedToolResultContent(markPersisted(archived)); + return decoded.kind === 'archived_tool_result' + ? unavailableArchivedToolResult(value, references) + : rewriteToolResultContent(decoded, references); + } catch { + return unavailableArchivedToolResult(value, references); + } +} + +function archivedSnapshotResult( + artifactId: string | undefined, + references: ConversationCopyMessageReferenceMap, +): string | undefined { + if ( + artifactId === undefined || + references.mode !== 'exact' || + references.linkedChildren.mode !== 'snapshot' + ) { + return undefined; + } + return references.linkedChildren.archivedResults.get(artifactId); +} + +function unavailableArchivedToolResult( + value: + | ArchivedToolResultPlaceholder + | Extract, + references: ConversationCopyMessageReferenceMap, +): Extract { + return { + kind: 'archived_tool_result', + status: 'missing', + runtimeEventId: rewriteOwnedId( + value.runtimeEventId, + references.runtimeEventIds, + 'RuntimeEvent', + ), + toolCallId: value.toolCallId, + toolName: value.toolName, + originalEstimatedTokens: value.originalEstimatedTokens, + originalBytes: value.originalBytes, + rewriteVersion: value.rewriteVersion, + reason: value.reason, + }; +} + function rewriteLinkedRunId( sourceId: string, childSessionId: string | undefined, diff --git a/packages/storage/src/__tests__/artifact-store.test.ts b/packages/storage/src/__tests__/artifact-store.test.ts index db16676269..245a57980b 100644 --- a/packages/storage/src/__tests__/artifact-store.test.ts +++ b/packages/storage/src/__tests__/artifact-store.test.ts @@ -322,6 +322,63 @@ describe('SQLite Artifact store', () => { }); }); + test('excludes selected Artifacts from a conversation snapshot', async () => { + await withWorkspace(async (root) => { + const authority = createArtifactStoreWriteAuthority(root); + await authority.recover(); + const { store } = authority; + await store.create({ + ...artifactInput('retained-artifact', 'retained', 10), + turnId: 'turn-retained', + }); + await store.create({ + ...artifactInput('excluded-archive', 'archived child result', 11), + turnId: 'turn-retained', + source: 'tool_result_archive', + }); + + const copied = await store.copyConversationArtifacts({ + sourceSessionId: 'session-1', + targetSessionId: 'session-copy', + turnIds: ['turn-retained'], + excludeArtifactIds: ['excluded-archive'], + }); + + assert.equal(copied.artifactIds.has('excluded-archive'), false); + assert.deepEqual( + (await store.list('session-copy')).map((record) => record.name), + ['retained-artifact.txt'], + ); + assert.equal((await store.get('excluded-archive'))?.sessionId, 'session-1'); + }); + }); + + test('copies explicit linked child Artifacts into a conversation snapshot', async () => { + await withWorkspace(async (root) => { + const authority = createArtifactStoreWriteAuthority(root); + await authority.recover(); + const { store } = authority; + await store.create({ + ...artifactInput('child-artifact', 'child result', 10), + sessionId: 'child-session', + turnId: 'child-turn', + }); + + const copied = await store.copyConversationArtifacts({ + sourceSessionId: 'session-1', + targetSessionId: 'session-copy', + turnIds: ['turn-retained'], + linkedArtifacts: [{ sessionId: 'child-session', artifactIds: ['child-artifact'] }], + }); + + const copiedId = copied.artifactIds.get('child-artifact'); + assert.ok(copiedId); + assert.deepEqual(await store.readText(copiedId), { ok: true, text: 'child result' }); + assert.equal((await store.get(copiedId))?.sessionId, 'session-copy'); + assert.equal((await store.get('child-artifact'))?.sessionId, 'child-session'); + }); + }); + test('user delete evaluates current-generation policy before tombstone state', async () => { await withWorkspace(async (root) => { const authority = createArtifactStoreWriteAuthority(root); diff --git a/packages/storage/src/__tests__/task-ledger-authority.test.ts b/packages/storage/src/__tests__/task-ledger-authority.test.ts index d4c9d5b7c6..d0ca59dc87 100644 --- a/packages/storage/src/__tests__/task-ledger-authority.test.ts +++ b/packages/storage/src/__tests__/task-ledger-authority.test.ts @@ -112,6 +112,98 @@ describe('interactive task ledger authority', () => { ); }); }); + + test('copies child-owned tasks into an ownership-free Side Conversation snapshot', async () => { + await withInteractiveOwner(async ({ writer }) => { + const sourceSessionId = 'source-session'; + const targetSessionId = 'side-conversation'; + const created = await writer.create(sourceSessionId, [{ subject: 'Delegated review' }], { + turnId: 'root-turn', + source: 'tool', + actor: 'main_agent', + }); + const taskId = created.created[0]!.id; + await writer.claim( + sourceSessionId, + taskId, + { + actor: 'child_agent', + sessionId: 'child-session', + agentId: 'reviewer', + runId: 'child-run', + turnId: 'child-turn', + }, + { + turnId: 'root-turn', + runId: 'child-run', + source: 'tool', + actor: 'child_agent', + }, + ); + const legacy = await writer.create(sourceSessionId, [{ subject: 'Legacy child note' }], { + turnId: 'root-turn', + runId: 'legacy-child-run', + source: 'tool', + actor: 'child_agent', + }); + + await writer.copyConversationTaskLedger({ + sourceSessionId, + targetSessionId, + turnIds: ['root-turn'], + runIdMap: [], + linkedChildren: 'snapshot', + }); + + const copied = await writer.list(targetSessionId); + assert.deepEqual( + copied.map((task) => ({ id: task.id, subject: task.subject, status: task.status })), + [ + { id: taskId, subject: 'Delegated review', status: 'in_progress' }, + { id: legacy.created[0]!.id, subject: 'Legacy child note', status: 'pending' }, + ], + ); + assert.ok(copied.every((task) => task.owner === undefined)); + assert.equal((await writer.get(sourceSessionId, taskId))?.owner?.sessionId, 'child-session'); + assert.equal((await writer.get(sourceSessionId, taskId))?.owner?.runId, 'child-run'); + }); + }); + + test('preserves main-agent ownership on child-authored snapshot events', async () => { + await withInteractiveOwner(async ({ writer }) => { + const sourceSessionId = 'source-main-owned'; + const targetSessionId = 'side-main-owned'; + const created = await writer.create(sourceSessionId, [{ subject: 'Parent-owned work' }], { + turnId: 'root-turn', + runId: 'root-run', + source: 'tool', + actor: 'main_agent', + }); + await writer.update( + sourceSessionId, + created.created[0]!.id, + { subject: 'Child reported progress' }, + { + turnId: 'root-turn', + runId: 'child-run', + source: 'tool', + actor: 'child_agent', + }, + ); + + await writer.copyConversationTaskLedger({ + sourceSessionId, + targetSessionId, + turnIds: ['root-turn'], + runIdMap: [{ sourceRunId: 'root-run', targetRunId: 'copied-root-run' }], + linkedChildren: 'snapshot', + }); + + const [copied] = await writer.list(targetSessionId); + assert.equal(copied?.owner?.actor, 'main_agent'); + assert.equal(copied?.owner?.runId, 'copied-root-run'); + }); + }); }); async function withInteractiveOwner( diff --git a/packages/storage/src/artifact-store.ts b/packages/storage/src/artifact-store.ts index bcc802d443..dc43a65241 100644 --- a/packages/storage/src/artifact-store.ts +++ b/packages/storage/src/artifact-store.ts @@ -158,6 +158,11 @@ export interface ConversationArtifactCopyInput { readonly sourceSessionId: string; readonly targetSessionId: string; readonly turnIds: readonly string[]; + readonly excludeArtifactIds?: readonly string[]; + readonly linkedArtifacts?: readonly { + readonly sessionId: string; + readonly artifactIds: readonly string[]; + }[]; } export interface ConversationArtifactCopyResult { @@ -383,21 +388,52 @@ class SqliteArtifactStore implements ArtifactAuthorityStore { throw new Error('Artifact conversation copy requires distinct Sessions'); } const turnIds = new Set(input.turnIds); + const excludedArtifactIds = new Set(input.excludeArtifactIds ?? []); for (const turnId of turnIds) assertArtifactTurnKey(turnId); + const linkedArtifacts = input.linkedArtifacts ?? []; + const requestedLinkedArtifactIds = new Map>(); + for (const linked of linkedArtifacts) { + assertCanonicalArtifactEntityId(linked.sessionId, 'sessionId'); + if (linked.sessionId === input.targetSessionId) { + throw new Error('Linked Artifact copy requires a distinct source Session'); + } + const artifactIds = requestedLinkedArtifactIds.get(linked.sessionId) ?? new Set(); + for (const artifactId of linked.artifactIds) { + assertCanonicalArtifactEntityId(artifactId, 'id'); + artifactIds.add(artifactId); + } + requestedLinkedArtifactIds.set(linked.sessionId, artifactIds); + } const records = await this.enqueue(async () => { await this.load(); - return this.records + const selected = this.records .filter( - (record) => record.sessionId === input.sourceSessionId && turnIds.has(record.turnId), + (record) => + record.sessionId === input.sourceSessionId && + turnIds.has(record.turnId) && + !excludedArtifactIds.has(record.id), ) .map((record) => ({ ...record })); + for (const [sessionId, artifactIds] of requestedLinkedArtifactIds) { + for (const artifactId of artifactIds) { + const record = this.records.find( + (candidate) => + candidate.sessionId === sessionId && + candidate.id === artifactId && + candidate.status !== 'deleted', + ); + if (!record) throw new Error(`Linked Artifact ${artifactId} could not be copied`); + selected.push({ ...record }); + } + } + return selected; }); const artifactIds = new Map(); const relativePaths = new Map(); for (const record of records) { const targetId = conversationCopyArtifactId( - input.sourceSessionId, + record.sessionId, input.targetSessionId, record.id, ); diff --git a/packages/storage/src/artifact-stores.ts b/packages/storage/src/artifact-stores.ts index 4cdd20ae1d..feaa5a94e2 100644 --- a/packages/storage/src/artifact-stores.ts +++ b/packages/storage/src/artifact-stores.ts @@ -152,6 +152,21 @@ function createWriterFacade( const acceptedInput: ConversationArtifactCopyInput = Object.freeze({ ...input, turnIds: Object.freeze([...input.turnIds]), + ...(input.excludeArtifactIds + ? { excludeArtifactIds: Object.freeze([...input.excludeArtifactIds]) } + : {}), + ...(input.linkedArtifacts + ? { + linkedArtifacts: Object.freeze( + input.linkedArtifacts.map((linked) => + Object.freeze({ + sessionId: linked.sessionId, + artifactIds: Object.freeze([...linked.artifactIds]), + }), + ), + ), + } + : {}), }); return run(() => store.copyConversationArtifacts(acceptedInput)); }, diff --git a/packages/storage/src/session-conversation-copy.ts b/packages/storage/src/session-conversation-copy.ts index db33111d7d..eff3083ae3 100644 --- a/packages/storage/src/session-conversation-copy.ts +++ b/packages/storage/src/session-conversation-copy.ts @@ -41,6 +41,7 @@ export function isValidConversationCopyTransition( previous.sourceSessionId === next.sourceSessionId && previous.sourceTurnId === next.sourceTurnId && previous.requestFingerprint === next.requestFingerprint && + previous.intent === next.intent && (previous.state !== 'committed' || next.state === 'committed') && (previous.state !== 'preparing' || next.state === 'preparing' || next.state === 'committed') ); diff --git a/packages/storage/src/task-ledger-authority.ts b/packages/storage/src/task-ledger-authority.ts index 9d2f2d8e47..b63328b706 100644 --- a/packages/storage/src/task-ledger-authority.ts +++ b/packages/storage/src/task-ledger-authority.ts @@ -145,6 +145,7 @@ function createInteractiveWriterFacade( ...input, turnIds: Object.freeze([...input.turnIds]), runIdMap: Object.freeze(input.runIdMap.map((entry) => Object.freeze({ ...entry }))), + ...(input.linkedChildren ? { linkedChildren: input.linkedChildren } : {}), }); return run(() => store.copyConversationTaskLedger(acceptedInput)); }, diff --git a/packages/storage/src/task-ledger-store.ts b/packages/storage/src/task-ledger-store.ts index 78d40d508b..ceb62d1df6 100644 --- a/packages/storage/src/task-ledger-store.ts +++ b/packages/storage/src/task-ledger-store.ts @@ -65,6 +65,7 @@ export interface ConversationTaskLedgerCopyInput { readonly sourceRunId: string; readonly targetRunId: string; }[]; + readonly linkedChildren?: 'preserve' | 'snapshot'; } export interface TaskLedgerAuthorityStore extends TaskLedgerStore { @@ -174,7 +175,13 @@ class SqliteTaskLedgerStoreImpl implements SqliteTaskLedgerStore { throw new Error('Task Ledger events cross the conversation-copy boundary'); } selected.push( - rewriteConversationTaskEvent(event, input.sourceSessionId, input.targetSessionId, runIds), + rewriteConversationTaskEvent( + event, + input.sourceSessionId, + input.targetSessionId, + runIds, + input.linkedChildren ?? 'preserve', + ), ); } if (selected.length === 0) return; @@ -787,29 +794,37 @@ function rewriteConversationTaskEvent( sourceSessionId: string, targetSessionId: string, runIds: ReadonlyMap, + linkedChildren: 'preserve' | 'snapshot', ): TaskLedgerEvent { - const owner = event.task.owner; + const { owner, ...task } = event.task; + const snapshotChildOwner = linkedChildren === 'snapshot' && owner?.actor === 'child_agent'; + const snapshotChildRun = linkedChildren === 'snapshot' && event.actor === 'child_agent'; const rewrittenOwner = owner === undefined ? undefined - : { - ...owner, - ...(owner.sessionId === sourceSessionId ? { sessionId: targetSessionId } : {}), - ...(owner.runId - ? { - runId: requiredConversationCopyRunId(runIds, owner.runId), - } - : {}), - }; + : snapshotChildOwner + ? undefined + : { + ...owner, + ...(owner.sessionId === sourceSessionId ? { sessionId: targetSessionId } : {}), + ...(owner.runId + ? { + runId: requiredConversationCopyRunId(runIds, owner.runId), + } + : {}), + }; const refs = event.refs === undefined ? undefined - : { - ...event.refs, - ...(event.refs.runId - ? { runId: requiredConversationCopyRunId(runIds, event.refs.runId) } - : {}), - }; + : (() => { + const { runId: sourceRunId, ...preserved } = event.refs; + return { + ...preserved, + ...(sourceRunId && !snapshotChildRun + ? { runId: requiredConversationCopyRunId(runIds, sourceRunId) } + : {}), + }; + })(); return { ...event, eventId: `task-copy-${createHash('sha256') @@ -817,7 +832,7 @@ function rewriteConversationTaskEvent( .digest('hex')}`, sessionId: targetSessionId, task: { - ...event.task, + ...task, ...(rewrittenOwner ? { owner: rewrittenOwner } : {}), }, ...(refs ? { refs } : {}),