From 5935375ff12b6b43ec44c31c6f362b365da80202 Mon Sep 17 00:00:00 2001 From: Changsu Seong <110822847+scs0209@users.noreply.github.com> Date: Wed, 2 Sep 2026 15:44:54 +0900 Subject: [PATCH 1/2] fix: preserve prompt-cache prefix for side conversation forks Move the side-conversation boundary from the system prompt into the first fork-owned user turn and route OpenAI promptCacheKey through the parent session id so fork requests share the parent prefix. Fixes #4540 Generated-by: Cursor --- docs/side-conversation.md | 10 ++- .../src/__tests__/side-conversation.test.ts | 88 +++++++++++++++++++ packages/core/src/side-conversation.ts | 87 +++++++++++++++++- .../src/server/interactive-run-composer.ts | 9 -- packages/runtime/src/ai-sdk-backend.ts | 30 +++++-- packages/runtime/src/model-adapter.ts | 4 +- 6 files changed, 208 insertions(+), 20 deletions(-) create mode 100644 packages/core/src/__tests__/side-conversation.test.ts diff --git a/docs/side-conversation.md b/docs/side-conversation.md index e3393a2fbb..03bb2340a0 100644 --- a/docs/side-conversation.md +++ b/docs/side-conversation.md @@ -41,8 +41,10 @@ The generic side-conversation entry extends that foundation: - selecting transcript text still appends quote context to the same panel; - the fork is created eagerly when the panel opens, with a short preparation state before the Composer receives focus; -- the child receives the `mode:side_conversation` label, which adds a system - boundary declaring inherited parent history reference-only; +- the child receives the `mode:side_conversation` label; the boundary declaring + inherited parent history reference-only is prepended to the first fork-owned + user turn instead of the system prompt so prompt-cache prefixes stay aligned + with the parent session; - the main Session and its active turn continue independently; - only instructions submitted in the side chat are active; explicit side-chat actions may use the inherited permission profile, and the permission can be @@ -118,7 +120,7 @@ moved until the eager fork is ready. It then becomes the ordinary closeable Side Chat tab without remounting the panel draft. The fork is marked both ephemeral and side-conversation, excluded from recent -conversation surfaces, and receives a developer boundary that: +conversation surfaces, and prepends a user-turn boundary that: - treats inherited history and tools as reference-only; - activates only instructions submitted after the side-chat boundary; @@ -217,7 +219,7 @@ authority. | --- | --- | --- | | Entry | `/side`, keyboard shortcut, Desktop actions | `/side`, titlebar, command palette, keyboard shortcut, and selected-text actions | | Initial transcript | parent history hidden | parent history hidden; only side turns render | -| Parent history | reference-only developer instruction plus hidden boundary | reference-only system prompt from the side label | +| Parent history | reference-only developer instruction plus hidden boundary | reference-only user-turn boundary from the side label | | Tool policy | read-mostly guidance; explicit side requests may mutate under the active permission profile | inherited permission profile; only explicit side-chat requests are active | | Lifetime | temporary, with Desktop confirmation and some retained-tab behavior | temporary; close deletes the fork with durable cleanup recovery | | Conversation list | suppressed | suppressed | diff --git a/packages/core/src/__tests__/side-conversation.test.ts b/packages/core/src/__tests__/side-conversation.test.ts new file mode 100644 index 0000000000..7569894de7 --- /dev/null +++ b/packages/core/src/__tests__/side-conversation.test.ts @@ -0,0 +1,88 @@ +/* + * 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 { describe, it } from 'node:test'; +import { + applySideConversationUserMessageBoundary, + buildSideConversationUserMessageBoundary, + resolveSideConversationPromptCacheSessionId, + SIDE_CONVERSATION_SESSION_LABEL, + userContentIncludesSideConversationBoundary, +} from '../side-conversation.js'; + +describe('side-conversation prompt cache helpers', () => { + it('prepends the boundary to the first fork-owned user message', () => { + const messages = applySideConversationUserMessageBoundary( + [ + { role: 'assistant', content: 'parent reply' }, + { role: 'user', content: 'side question' }, + ], + { + inheritedPrefixLength: 1, + labels: [SIDE_CONVERSATION_SESSION_LABEL], + }, + ); + + assert.equal(messages[0]?.role, 'assistant'); + assert.equal(messages[1]?.role, 'user'); + assert.match(String(messages[1]?.content), /Side conversation boundary:/); + assert.match(String(messages[1]?.content), /side question/); + }); + + it('is idempotent when the boundary is already present', () => { + const boundary = buildSideConversationUserMessageBoundary(); + const messages = applySideConversationUserMessageBoundary( + [{ role: 'user', content: `${boundary}\n\nalready there` }], + { + inheritedPrefixLength: 0, + labels: [SIDE_CONVERSATION_SESSION_LABEL], + }, + ); + + assert.equal(messages[0]?.content, `${boundary}\n\nalready there`); + }); + + it('routes OpenAI prompt cache keys through the parent session id', () => { + assert.equal( + resolveSideConversationPromptCacheSessionId({ + sessionId: 'fork-session', + parentSessionId: 'parent-session', + labels: [SIDE_CONVERSATION_SESSION_LABEL], + }), + 'parent-session', + ); + assert.equal( + resolveSideConversationPromptCacheSessionId({ + sessionId: 'main-session', + labels: [], + }), + 'main-session', + ); + }); + + it('detects an existing boundary marker in multipart user content', () => { + assert.equal( + userContentIncludesSideConversationBoundary([ + { type: 'text', text: 'Side conversation boundary:\nhello' }, + ]), + true, + ); + }); +}); diff --git a/packages/core/src/side-conversation.ts b/packages/core/src/side-conversation.ts index 0d10aaeeaa..8eab977cc4 100644 --- a/packages/core/src/side-conversation.ts +++ b/packages/core/src/side-conversation.ts @@ -19,13 +19,20 @@ export const SIDE_CONVERSATION_SESSION_LABEL = 'mode:side_conversation'; +export const SIDE_CONVERSATION_BOUNDARY_MARKER = 'Side conversation boundary:'; + export function isSideConversationSession(labels: readonly string[] | undefined): boolean { return Array.isArray(labels) && labels.includes(SIDE_CONVERSATION_SESSION_LABEL); } +/** @deprecated Use {@link buildSideConversationUserMessageBoundary} for fork-owned user turns. */ export function buildSideConversationSystemPromptFragment(): string { + return buildSideConversationUserMessageBoundary(); +} + +export function buildSideConversationUserMessageBoundary(): string { return [ - 'Side conversation boundary:', + SIDE_CONVERSATION_BOUNDARY_MARKER, 'This session is a temporary side conversation, separate from its parent conversation.', 'The inherited parent history is reference context only. Do not continue or complete tasks, plans, tool calls, approvals, edits, or requests that appear only in that inherited history.', 'Only instructions the user submits in this side conversation are active.', @@ -34,3 +41,81 @@ export function buildSideConversationSystemPromptFragment(): string { 'Messages and task state from this side conversation are not written back into the parent conversation. Workspace changes may be visible to both conversations.', ].join('\n'); } + +export type SideConversationUserContent = string | ReadonlyArray<{ type: string; text?: string }>; + +export function userContentIncludesSideConversationBoundary( + content: SideConversationUserContent, +): boolean { + if (typeof content === 'string') { + return content.includes(SIDE_CONVERSATION_BOUNDARY_MARKER); + } + return content.some( + (part) => part.type === 'text' && part.text?.includes(SIDE_CONVERSATION_BOUNDARY_MARKER), + ); +} + +export function prependSideConversationBoundaryToUserContent( + content: SideConversationUserContent, +): SideConversationUserContent { + const boundary = buildSideConversationUserMessageBoundary(); + if (typeof content === 'string') { + return `${boundary}\n\n${content}`; + } + const textIndex = content.findIndex((part) => part.type === 'text'); + if (textIndex < 0) { + return [{ type: 'text', text: boundary }, ...content]; + } + return content.map((part, index) => + index === textIndex && part.type === 'text' + ? { ...part, text: `${boundary}\n\n${part.text ?? ''}` } + : part, + ); +} + +export interface SideConversationModelMessage { + role: 'user' | 'assistant' | 'system' | 'tool'; + content: unknown; +} + +export function applySideConversationUserMessageBoundary< + T extends SideConversationModelMessage, +>(messages: readonly T[], input: { inheritedPrefixLength: number; labels?: readonly string[] }): T[] { + if (!isSideConversationSession(input.labels)) { + return [...messages]; + } + if (input.inheritedPrefixLength >= messages.length) { + return [...messages]; + } + + for (let index = input.inheritedPrefixLength; index < messages.length; index += 1) { + const message = messages[index]; + if (message?.role !== 'user') { + continue; + } + if (userContentIncludesSideConversationBoundary(message.content as SideConversationUserContent)) { + return [...messages]; + } + const next = [...messages]; + next[index] = { + ...message, + content: prependSideConversationBoundaryToUserContent( + message.content as SideConversationUserContent, + ), + }; + return next; + } + + return [...messages]; +} + +export function resolveSideConversationPromptCacheSessionId(input: { + sessionId: string; + labels?: readonly string[]; + parentSessionId?: string; +}): string { + if (isSideConversationSession(input.labels) && input.parentSessionId) { + return input.parentSessionId; + } + return input.sessionId; +} diff --git a/packages/runtime-host/src/server/interactive-run-composer.ts b/packages/runtime-host/src/server/interactive-run-composer.ts index dc35d96779..b9016cefa4 100644 --- a/packages/runtime-host/src/server/interactive-run-composer.ts +++ b/packages/runtime-host/src/server/interactive-run-composer.ts @@ -17,10 +17,6 @@ * under the License. */ -import { - buildSideConversationSystemPromptFragment, - isSideConversationSession, -} from '@maka/core/side-conversation'; import { type RunCompositionSourceRevision } from '@maka/core/run-composition'; import { buildDeepResearchSystemPromptFragment, @@ -95,7 +91,6 @@ export interface InteractiveRunComposerInput { readonly memory: HostMemoryCoordinator; readonly sessionTodo: SessionTodoToolStore; readonly childInstruction?: string; - readonly sideConversation?: boolean; readonly boundTools?: readonly MakaTool[]; readonly toolProfile?: SessionToolProfile; readonly skillBudget?: SkillCatalogBudgetOptions; @@ -229,7 +224,6 @@ export function createInteractiveRunComposer(input: InteractiveRunComposerInput) ? renderPlanModePrompt({ fullAccess: input.plan.permissionMode === 'bypass' }) : undefined, input.deepResearch ? buildDeepResearchSystemPromptFragment() : undefined, - input.sideConversation ? buildSideConversationSystemPromptFragment() : undefined, ]); return Object.freeze({ text, @@ -384,9 +378,6 @@ export function createInteractiveRunComposerFactory( memory: input.memory, sessionTodo: input.sessionTodo, ...(backendContext.systemPrompt ? { childInstruction: backendContext.systemPrompt } : {}), - ...(isSideConversationSession(backendContext.header.labels) - ? { sideConversation: true } - : {}), ...(boundTools ? { boundTools } : {}), ...(!boundTools && backendContext.header.toolProfile ? { toolProfile: backendContext.header.toolProfile } diff --git a/packages/runtime/src/ai-sdk-backend.ts b/packages/runtime/src/ai-sdk-backend.ts index 69f83d569f..ef600a8ea2 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -71,6 +71,10 @@ import type { BackendKind, SessionHeader, } from '@maka/core/session'; +import { + applySideConversationUserMessageBoundary, + resolveSideConversationPromptCacheSessionId, +} from '@maka/core/side-conversation'; import type { AgentBackend, BackendCompactHistoryInput, @@ -1098,6 +1102,11 @@ export class AiSdkBackend implements AgentBackend { buildProviderOptions(input.connection, input.modelId, input.header.thinkingLevel); this.modelAdapter = new ModelAdapter({ sessionId: input.sessionId, + promptCacheSessionId: resolveSideConversationPromptCacheSessionId({ + sessionId: input.sessionId, + labels: input.header.labels, + parentSessionId: input.header.parentSessionId, + }), connection: input.connection, apiKey: input.apiKey, modelId: input.modelId, @@ -1853,7 +1862,7 @@ export class AiSdkBackend implements AgentBackend { input.quotes, input.headAnchorRuntimeEvent?.id, ); - const messages = + const messages = applySideConversationUserMessageBoundary( currentUserContent === undefined ? [...priorReplay.messages] : [ @@ -1862,7 +1871,12 @@ export class AiSdkBackend implements AgentBackend { role: 'user' as const, content: currentUserContent, } as ModelMessage, - ]; + ], + { + inheritedPrefixLength: priorReplay.messages.length, + labels: this.input.header.labels, + }, + ); const loadDurableTurnEvents = async (): Promise => { const loadTurnRuntimeEvents = this.input.loadTurnRuntimeEvents; if (!loadTurnRuntimeEvents) { @@ -1928,9 +1942,15 @@ export class AiSdkBackend implements AgentBackend { scope.runId, ), ); - return projectionCheckpoint - ? currentTurnMessages - : [...priorReplay.messages, ...currentTurnMessages]; + return applySideConversationUserMessageBoundary( + projectionCheckpoint + ? currentTurnMessages + : [...priorReplay.messages, ...currentTurnMessages], + { + inheritedPrefixLength: priorReplay.messages.length, + labels: this.input.header.labels, + }, + ); }; // Tool Availability describes the provider-visible (active) subset. A // group loaded this turn expands that subset on later requests, so the diff --git a/packages/runtime/src/model-adapter.ts b/packages/runtime/src/model-adapter.ts index b5444bb529..3ef6e4d369 100644 --- a/packages/runtime/src/model-adapter.ts +++ b/packages/runtime/src/model-adapter.ts @@ -108,6 +108,8 @@ export interface RepairableAiSdkToolCall { export interface ModelAdapterInput { sessionId?: string; + /** OpenAI prompt cache routing key; defaults to sessionId when omitted. */ + promptCacheSessionId?: string; connection: RuntimeExecutionConnection; apiKey: string; modelId: string; @@ -295,7 +297,7 @@ export class ModelAdapter { const providerOptions = usesNativeOpenAiResponses(this.input.connection, this.runtime) ? mergeOpenAiResponsesProviderOptions( this.input.providerOptions, - this.input.sessionId ?? this.input.connection.slug, + this.input.promptCacheSessionId ?? this.input.sessionId ?? this.input.connection.slug, continuation.previousResponseId, ) : this.input.providerOptions; From 4f71fa45307ce9ceb427c84f8a9cb80150786fb3 Mon Sep 17 00:00:00 2001 From: Changsu Seong <110822847+scs0209@users.noreply.github.com> Date: Wed, 2 Sep 2026 15:58:53 +0900 Subject: [PATCH 2/2] fix: preserve prompt-cache prefix for side conversation forks Move the side-conversation boundary from the system prompt into the first fork-owned user turn, route OpenAI promptCacheKey through the parent session id, and add a runtime-host regression test that compares parent and fork provider request bodies. Fixes #4540 Generated-by: Cursor --- packages/core/src/side-conversation.ts | 11 +- .../side-conversation-prompt-cache.test.ts | 475 ++++++++++++++++++ 2 files changed, 482 insertions(+), 4 deletions(-) create mode 100644 packages/runtime-host/src/__tests__/side-conversation-prompt-cache.test.ts diff --git a/packages/core/src/side-conversation.ts b/packages/core/src/side-conversation.ts index 8eab977cc4..3bdab3d088 100644 --- a/packages/core/src/side-conversation.ts +++ b/packages/core/src/side-conversation.ts @@ -78,9 +78,10 @@ export interface SideConversationModelMessage { content: unknown; } -export function applySideConversationUserMessageBoundary< - T extends SideConversationModelMessage, ->(messages: readonly T[], input: { inheritedPrefixLength: number; labels?: readonly string[] }): T[] { +export function applySideConversationUserMessageBoundary( + messages: readonly T[], + input: { inheritedPrefixLength: number; labels?: readonly string[] }, +): T[] { if (!isSideConversationSession(input.labels)) { return [...messages]; } @@ -93,7 +94,9 @@ export function applySideConversationUserMessageBoundary< if (message?.role !== 'user') { continue; } - if (userContentIncludesSideConversationBoundary(message.content as SideConversationUserContent)) { + if ( + userContentIncludesSideConversationBoundary(message.content as SideConversationUserContent) + ) { return [...messages]; } const next = [...messages]; diff --git a/packages/runtime-host/src/__tests__/side-conversation-prompt-cache.test.ts b/packages/runtime-host/src/__tests__/side-conversation-prompt-cache.test.ts new file mode 100644 index 0000000000..dcba5a1142 --- /dev/null +++ b/packages/runtime-host/src/__tests__/side-conversation-prompt-cache.test.ts @@ -0,0 +1,475 @@ +/* + * 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 { createServer, type IncomingMessage, type Server, type ServerResponse } from 'node:http'; +import { mkdir, mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { test } from 'node:test'; +import { + SIDE_CONVERSATION_BOUNDARY_MARKER, + SIDE_CONVERSATION_SESSION_LABEL, +} from '@maka/core/side-conversation'; +import { openInteractiveExecutionStoresForWrite } from '@maka/storage/execution-stores'; +import type { InteractiveExecutionStoresWriter } from '@maka/storage/execution-stores'; +import { openInteractiveRuntimePolicyStoresForWrite } from '@maka/storage/runtime-policy-stores'; +import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '@maka/storage/root-authority'; +import type { TurnSnapshot, SessionCatalogProjection } from '../protocol/index.js'; +import { createExecutionRuntimeHostComposition } from '../server/execution-composition.js'; +import type { ConnectionContext } from '../server/operation-dispatcher.js'; +import type { RuntimePolicyStoresWriter } from '@maka/storage/runtime-policy-stores'; + +const MODEL_ID = 'side-conversation-cache-model'; +const API_KEY = 'side-conversation-cache-key'; +const RESPONSE_TEXT = 'Side conversation cache regression response.'; + +interface ProviderRequest { + readonly url: string; + readonly body: Record; +} + +test('side conversation fork preserves the parent provider prefix before the fork-owned turn', async () => { + const base = await mkdtemp(join(tmpdir(), 'maka-side-conversation-prompt-cache-')); + const root = join(base, 'interactive'); + const project = join(base, 'project'); + const provider = await startProvider(); + const capability = await resolveStorageRoot({ path: root, kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + if (!owner) return; + const context: ConnectionContext = { + hostEpoch: 'side-conversation-prompt-cache-epoch', + connectionId: 'side-conversation-prompt-cache-client', + principal: 'local_os_user', + acquireResidency: () => ({ release() {} }), + }; + let composition: Awaited> | undefined; + try { + await mkdir(project); + const policy = await openInteractiveRuntimePolicyStoresForWrite(owner.lease); + const created = await policy.connectionCatalog.create({ + expectedCatalogRevision: 0, + connection: { + slug: 'side-conversation-cache-provider', + name: 'Side conversation cache provider', + providerType: 'moonshot', + baseUrl: provider.baseUrl, + enabled: true, + enabledModelIds: [MODEL_ID], + }, + }); + assert.equal(created.kind, 'committed'); + if (created.kind !== 'committed') return; + const connection = created.snapshot.connections[0]; + assert.ok(connection); + if (!connection) return; + assert.equal( + ( + await policy.credentialVault.set({ + locator: { + scope: 'connection', + connectionId: connection.connectionId, + kind: 'api_key', + }, + expected: null, + secret: API_KEY, + }) + ).kind, + 'committed', + ); + await publishConnectionModel(policy, connection.connectionId, MODEL_ID); + + const execution = await openInteractiveExecutionStoresForWrite(owner.lease); + const parentSession = await execution.sessionStore.create({ + cwd: project, + llmConnectionId: connection.connectionId, + llmConnectionSlug: 'side-conversation-cache-provider', + model: MODEL_ID, + permissionMode: 'bypass', + }); + + composition = await createExecutionRuntimeHostComposition({ + owner, + hostEpoch: context.hostEpoch, + acquireResidency: context.acquireResidency, + retainUntilProcessExit: () => undefined, + requestDrain: () => undefined, + }); + await composition.recover(); + + const parentTurnId = 'parent-turn-1'; + const parentTerminal = await waitForTerminal( + composition, + parentSession.id, + parentTurnId, + await startTurn( + composition, + parentSession.id, + parentTurnId, + 'Summarize the workspace in one sentence.', + context, + ), + context, + ); + assert.equal(parentTerminal.status, 'completed'); + + const parentStreamRequests = streamProviderRequests(provider.requests); + + const forkSessionId = 'side-conversation-fork'; + const branch = await branchSideConversation({ + composition, + context, + execution, + sourceSessionId: parentSession.id, + targetSessionId: forkSessionId, + sourceTurnId: parentTurnId, + }); + assert.ok(branch.labels.includes(SIDE_CONVERSATION_SESSION_LABEL)); + + const parentStreamCountBeforeFork = parentStreamRequests.length; + const forkTurnId = 'fork-turn-1'; + const forkTerminal = await waitForTerminal( + composition, + forkSessionId, + forkTurnId, + await startTurn( + composition, + forkSessionId, + forkTurnId, + 'What did I just ask in this side chat?', + context, + ), + context, + ); + assert.equal(forkTerminal.status, 'completed'); + + const streamRequestsAfterFork = streamProviderRequests(provider.requests); + assert.ok(streamRequestsAfterFork.length > parentStreamRequests.length); + const forkFirstRequest = firstMainTurnStreamRequest( + streamRequestsAfterFork, + parentStreamRequests.length, + ); + const parentLastRequest = streamRequestsAfterFork[parentStreamRequests.length - 1]; + assert.ok(parentLastRequest, 'Parent Session did not emit a streaming provider request'); + assert.equal(isAuxiliaryProviderRequest(parentLastRequest.body), false); + + const parentTools = parentLastRequest.body.tools; + const forkTools = forkFirstRequest.body.tools; + assert.deepEqual(forkTools, parentTools); + + const parentSystem = systemPromptText(parentLastRequest.body); + const forkSystem = systemPromptText(forkFirstRequest.body); + assert.equal(forkSystem, parentSystem); + assert.doesNotMatch(forkSystem, new RegExp(SIDE_CONVERSATION_BOUNDARY_MARKER)); + + const parentMessages = providerMessages(parentLastRequest.body); + const forkMessages = providerMessages(forkFirstRequest.body); + assert.ok(forkMessages.length > parentMessages.length); + assert.deepEqual(forkMessages.slice(0, parentMessages.length), parentMessages); + + const extensionMessages = forkMessages.slice(parentMessages.length); + const boundaryUserMessages = extensionMessages.filter( + (message) => + message.role === 'user' && messageText(message).includes(SIDE_CONVERSATION_BOUNDARY_MARKER), + ); + assert.equal(boundaryUserMessages.length, 1); + const divergentMessage = extensionMessages.at(-1); + assert.equal(divergentMessage?.role, 'user'); + const divergentContent = messageText(divergentMessage); + assert.match(divergentContent, new RegExp(SIDE_CONVERSATION_BOUNDARY_MARKER)); + assert.match(divergentContent, /What did I just ask in this side chat\?/); + } finally { + await composition?.close(); + await owner?.close(); + await provider.close(); + await rm(base, { recursive: true, force: true }); + } +}); + +async function branchSideConversation(input: { + composition: Awaited>; + context: ConnectionContext; + execution: InteractiveExecutionStoresWriter; + sourceSessionId: string; + targetSessionId: string; + sourceTurnId: string; +}): Promise { + for (let attempt = 0; attempt < 100; attempt += 1) { + const record = await input.execution.sessionStore.readCatalogRecord(input.sourceSessionId); + const branch = await input.composition.handlers['session.branch.create']( + { + sourceSessionId: input.sourceSessionId, + targetSessionId: input.targetSessionId, + sourceTurnId: input.sourceTurnId, + expectedSourceRevision: record.revision, + intent: 'side_conversation', + }, + input.context, + ); + if (!branch.ok) { + assert.fail(`Side conversation branch failed: ${JSON.stringify(branch)}`); + } + if (branch.result.kind === 'committed') { + if ('kind' in branch.result.session) { + assert.fail('Fork Session catalog projection was unsupported'); + } + return branch.result.session; + } + if (branch.result.kind === 'source_revision_conflict') { + await new Promise((resolve) => setTimeout(resolve, 10)); + continue; + } + assert.fail(`Side conversation branch failed: ${JSON.stringify(branch.result)}`); + } + assert.fail('Side conversation branch never observed a stable source revision'); +} + +async function startTurn( + composition: Awaited>, + sessionId: string, + turnId: string, + text: string, + context: ConnectionContext, +): Promise { + for (let attempt = 0; attempt < 200; attempt += 1) { + const started = await composition.handlers['turn.start']( + { sessionId, turnId, content: { text } }, + context, + ); + if (started.ok) { + if (started.result.kind === 'started') return started.result.turn; + throw new Error(`Side conversation turn start was blocked: ${JSON.stringify(started)}`); + } + if (started.error.code !== 'session_busy') { + throw new Error(`Side conversation turn start failed: ${JSON.stringify(started.error)}`); + } + await new Promise((resolve) => setTimeout(resolve, 10)); + } + throw new Error('Side conversation Session did not become idle'); +} + +async function waitForTerminal( + composition: Awaited>, + sessionId: string, + turnId: string, + initial: TurnSnapshot, + context: ConnectionContext, +): Promise { + let snapshot = initial; + for (let attempt = 0; attempt < 200; attempt += 1) { + if (isTerminal(snapshot)) return snapshot; + await new Promise((resolve) => setTimeout(resolve, 10)); + const queried = await composition.handlers['turn.query']({ sessionId, turnId }, context); + assert.equal(queried.ok, true); + snapshot = queried.result; + } + throw new Error('Side conversation turn did not become terminal'); +} + +function isTerminal(snapshot: TurnSnapshot): boolean { + return ( + snapshot.status === 'completed' || + snapshot.status === 'failed' || + snapshot.status === 'cancelled' + ); +} + +async function publishConnectionModel( + policy: RuntimePolicyStoresWriter, + connectionId: string, + modelId: string, +): Promise { + const prepared = await policy.operations.beginModelFetch(connectionId); + assert.equal(prepared.kind, 'ready'); + if (prepared.kind !== 'ready') throw new Error('Model discovery was not ready'); + const committed = await policy.operations.completeModelFetch(prepared.ticket, { + models: [ + { + id: modelId, + capabilities: { chat: true, functionCalling: true }, + contextWindow: 32_768, + maxOutputTokens: 256, + }, + ], + source: 'fetched', + fetchedAt: Date.now(), + }); + assert.equal(committed.kind, 'committed'); +} + +function streamProviderRequests(requests: readonly ProviderRequest[]): ProviderRequest[] { + return requests.filter((request) => request.body.stream === true); +} + +function isAuxiliaryProviderRequest(body: Record): boolean { + const serialized = JSON.stringify(body); + return ( + /Perform the first stage of long-term-memory extraction/.test(serialized) || + /context summarization assistant/.test(serialized) + ); +} + +function firstMainTurnStreamRequest( + requests: readonly ProviderRequest[], + start: number, +): ProviderRequest { + for (let index = start; index < requests.length; index += 1) { + const request = requests[index]; + if (!request || isAuxiliaryProviderRequest(request.body)) { + continue; + } + return request; + } + assert.fail('Fork Session did not emit a main streaming provider request'); +} + +function providerMessages(body: Record): Array> { + const messages = body.messages; + assert.ok(Array.isArray(messages), JSON.stringify(body)); + return messages.filter( + (message): message is Record => + Boolean(message) && typeof message === 'object' && !Array.isArray(message), + ); +} + +function systemPromptText(body: Record): string { + return providerMessages(body) + .filter((message) => message.role === 'system') + .map((message) => messageText(message)) + .join('\n\n'); +} + +function messageText(message: Record | undefined): string { + if (!message) return ''; + const content = message.content; + if (typeof content === 'string') return content; + if (!Array.isArray(content)) return JSON.stringify(content ?? ''); + return content + .flatMap((part) => { + if (!part || typeof part !== 'object') return []; + const text = (part as { text?: unknown }).text; + return typeof text === 'string' ? [text] : []; + }) + .join('\n'); +} + +async function startProvider(): Promise<{ + readonly baseUrl: string; + readonly requests: ProviderRequest[]; + close(): Promise; +}> { + const requests: ProviderRequest[] = []; + const server = createServer((request, response) => { + void handleProviderRequest(request, response, requests).catch((error) => { + response.destroy(error as Error); + }); + }); + await listen(server); + const address = server.address(); + assert.ok(address && typeof address === 'object'); + return { + baseUrl: `http://127.0.0.1:${address.port}/v1`, + requests, + close: () => closeServer(server), + }; +} + +async function handleProviderRequest( + request: IncomingMessage, + response: ServerResponse, + requests: ProviderRequest[], +): Promise { + assert.equal(request.method, 'POST'); + const body = JSON.parse(await readBody(request)) as Record; + requests.push({ url: request.url ?? '', body }); + if (body.stream !== true) { + response.writeHead(200, { 'content-type': 'application/json' }); + response.end( + JSON.stringify({ + id: 'chatcmpl-side-conversation-summary', + object: 'chat.completion', + created: 1, + model: MODEL_ID, + choices: [ + { + index: 0, + message: { role: 'assistant', content: RESPONSE_TEXT }, + finish_reason: 'stop', + }, + ], + usage: { prompt_tokens: 7, completion_tokens: 3, total_tokens: 10 }, + }), + ); + return; + } + respondProviderText(response, RESPONSE_TEXT); +} + +function respondProviderText(response: ServerResponse, text: string): void { + response.writeHead(200, { 'content-type': 'text/event-stream' }); + response.write( + `data: ${JSON.stringify({ + id: 'chatcmpl-side-conversation', + object: 'chat.completion.chunk', + created: 1, + model: MODEL_ID, + choices: [{ index: 0, delta: { role: 'assistant', content: text }, finish_reason: null }], + })}\n\n`, + ); + response.write( + `data: ${JSON.stringify({ + id: 'chatcmpl-side-conversation', + object: 'chat.completion.chunk', + created: 1, + model: MODEL_ID, + choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], + usage: { prompt_tokens: 11, completion_tokens: 5, total_tokens: 16 }, + })}\n\n`, + ); + response.end('data: [DONE]\n\n'); +} + +function readBody(request: IncomingMessage): Promise { + return new Promise((resolve, reject) => { + let body = ''; + request.setEncoding('utf8'); + request.on('data', (chunk) => { + body += chunk; + }); + request.on('end', () => resolve(body)); + request.on('error', reject); + }); +} + +function listen(server: Server): Promise { + return new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(0, '127.0.0.1', () => { + server.off('error', reject); + resolve(); + }); + }); +} + +function closeServer(server: Server): Promise { + return new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); +}