diff --git a/docs/side-conversation.md b/docs/side-conversation.md index 77e3f46c2c..a9cccd3821 100644 --- a/docs/side-conversation.md +++ b/docs/side-conversation.md @@ -44,8 +44,10 @@ The generic side-conversation entry extends that foundation: with an empty context (inheriting the source's model, cwd, and permission but no transcript) when the source has not completed a turn yet, so opening the panel never depends on the main Session's turn state; -- 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 @@ -125,7 +127,7 @@ turn yet. Opening the panel never depends on the parent's turn state, and no mid-flight turn is ever copied. 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; @@ -224,7 +226,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..3bdab3d088 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,84 @@ 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( + 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/__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())); + }); +} 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 fe67a5fffb..27fe72f20e 100644 --- a/packages/runtime/src/ai-sdk-backend.ts +++ b/packages/runtime/src/ai-sdk-backend.ts @@ -70,6 +70,10 @@ import type { BackendKind, SessionHeader, } from '@maka/core/session'; +import { + applySideConversationUserMessageBoundary, + resolveSideConversationPromptCacheSessionId, +} from '@maka/core/side-conversation'; import type { AgentBackend, BackendCompactHistoryInput, @@ -1093,6 +1097,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, @@ -1835,7 +1844,7 @@ export class AiSdkBackend implements AgentBackend { input.quotes, input.headAnchorRuntimeEvent?.id, ); - const messages = + const messages = applySideConversationUserMessageBoundary( currentUserContent === undefined ? [...priorReplay.messages] : [ @@ -1844,7 +1853,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) { @@ -1910,9 +1924,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;