diff --git a/apps/desktop/src/main/__tests__/app-shell-busy-race-settlement.test.ts b/apps/desktop/src/main/__tests__/app-shell-busy-race-settlement.test.ts index b4661d00d6..90b1804730 100644 --- a/apps/desktop/src/main/__tests__/app-shell-busy-race-settlement.test.ts +++ b/apps/desktop/src/main/__tests__/app-shell-busy-race-settlement.test.ts @@ -18,9 +18,9 @@ */ /** - * #1954 busy-race settlement: a sessions:send that raced a root turn another - * client opened can come back `steered` (the send owns no turn) or under a - * Host-chosen turnId. Both results must be interpreted identically by the + * #1954 busy-race settlement: a submitted Message that raced a root turn + * another client opened can come back `steered` (the send owns no turn) or + * under a Host-chosen turnId. Both results must be interpreted identically by the * new-chat and existing-session branches, and a rebind must never overwrite * an authoritative live projection that beat the IPC response. */ diff --git a/apps/desktop/src/main/__tests__/quote-companion-retry.test.ts b/apps/desktop/src/main/__tests__/quote-companion-retry.test.ts index 431d6bf6b2..6a26dfbca6 100644 --- a/apps/desktop/src/main/__tests__/quote-companion-retry.test.ts +++ b/apps/desktop/src/main/__tests__/quote-companion-retry.test.ts @@ -561,6 +561,66 @@ test('replays queued Side Conversation text after Host assigns the ticket to a s assert.equal(probe.getAttribute('data-processing'), 'false'); }); +test('binds an unproven Side Conversation send through the durable transcript', async () => { + let admissionId: string | undefined; + const pendingSend = deferred<{ + ok: false; + reason: 'outcome_unknown'; + messageId: string; + }>(); + // The Host opened a root Turn under its own identity and the answer was lost. + // No `message_admission` event exists for a root Message, so the transcript is + // the only thing that can tie the sent identity back to the Turn. + const { container, emit, send } = await renderOwnershipProbe({ + send: async (_sessionId, command) => { + admissionId = command.turnId; + return pendingSend.promise; + }, + readSettledMessages: async () => ({ + messages: admissionId + ? [ + { + type: 'user' as const, + id: admissionId, + turnId: 'unproven-root', + ts: 1, + text: 'reconcile me', + }, + ] + : [], + settled: true, + }), + }); + + let sendResult!: Promise; + await act(async () => { + sendResult = send('reconcile me'); + await Promise.resolve(); + }); + await act(async () => { + pendingSend.resolve({ + ok: false, + reason: 'outcome_unknown', + messageId: admissionId as string, + }); + assert.equal(await sendResult, true); + await Promise.resolve(); + }); + await act(async () => { + emit(textDeltaEvent('unproven-text', 'unproven-root', 1, 'answer from the lost send')); + await Promise.resolve(); + }); + await act(async () => { + await Promise.resolve(); + }); + + const probe = container.firstElementChild; + assert.ok(probe); + assert.equal(probe.getAttribute('data-live-turn-id'), 'unproven-root'); + assert.equal(probe.getAttribute('data-live-text'), 'answer from the lost send'); + assert.equal(probe.getAttribute('data-processing'), 'false'); +}); + test('clears a stopped Side Conversation admission when its live retraction is lost', async () => { let admissionId: string | undefined; const pendingStop = deferred<{ kind: 'retracted'; messageId: string }>(); 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 887a852605..7a20e9ca77 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 @@ -345,19 +345,16 @@ test('drives the renderer Session execution facade through real UDS framing', as result: { kind: 'managed', access: 'read_only', revision: 2 }, }; }, - 'turn.start': async (input) => { + 'turn.message.submit': async (input) => { assert.equal(input.sessionId, projected.id); + assert.equal(input.messageId, 'turn-1'); + assert.equal(input.placement, 'current_turn'); assert.equal(input.content.text, 'Run through the Host'); return { ok: true, result: { - kind: 'started', - turn: { - sessionId: input.sessionId, - turnId: input.turnId, - runId: 'run-1', - status: 'running', - }, + disposition: 'turn_started', + turnId: 'turn-host-1', skillInvocation: { loaded: [], failed: [], receipts: [] }, }, }; @@ -410,7 +407,7 @@ test('drives the renderer Session execution facade through real UDS framing', as }), { ok: true, - turnId: 'turn-1', + turnId: 'turn-host-1', attachments: [], inlineReferences: [], skillInvocation: { loaded: [], failed: [], receipts: [] }, 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 09ac53260a..78c7c0ca20 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 @@ -359,18 +359,9 @@ test("sends canonical content and uploads owned Attachment bytes through the Hos uploads.push(input); return attachment; }, - startTurn: async (input) => { + submitMessage: async (input) => { starts.push(input); - return { - kind: "started", - turn: { - sessionId: input.sessionId, - turnId: input.turnId, - runId: "run-1", - status: "running", - }, - skillInvocation: { loaded: [], failed: [], receipts: [] }, - }; + return { disposition: "turn_started", turnId: "turn-1" }; }, }); const ipc = ipcHarness(); @@ -406,7 +397,8 @@ test("sends canonical content and uploads owned Attachment bytes through the Hos assert.deepEqual(starts, [ { sessionId: "session-1", - turnId: "turn-1", + messageId: "turn-1", + placement: "current_turn", content: { text: "Read @notes.txt", attachments: [attachment], @@ -472,18 +464,9 @@ test("uploads a selected workspace file as a Host-owned Session Artifact", async uploads.push(input); return attachment; }, - startTurn: async (input) => { + submitMessage: async (input) => { starts.push(input); - return { - kind: "started", - turn: { - sessionId: input.sessionId, - turnId: input.turnId, - runId: "run-1", - status: "running", - }, - skillInvocation: { loaded: [], failed: [], receipts: [] }, - }; + return { disposition: "turn_started", turnId: "turn-1" }; }, }), observer: unusedObserver(), @@ -521,16 +504,11 @@ test("forwards explicit Skill invocation to the Host-owned Turn admission", asyn { client: executionClient({ getSession: async () => session(), - startTurn: async (input) => { + submitMessage: async (input) => { starts.push(input); return { - kind: "started", - turn: { - sessionId: input.sessionId, - turnId: input.turnId, - runId: "run-1", - status: "running", - }, + disposition: "turn_started", + turnId: "turn-skill", skillInvocation: { loaded: [{ id: "review", name: "Review" }], failed: [], @@ -560,7 +538,8 @@ test("forwards explicit Skill invocation to the Host-owned Turn admission", asyn assert.deepEqual(starts, [ { sessionId: "session-1", - turnId: "turn-skill", + messageId: "turn-skill", + placement: "current_turn", content: { text: "", displayText: "/skill:review", inlineReferences: [] }, skillIds: ["review"], }, @@ -585,9 +564,6 @@ test("submits an ordinary composer message once under its stable message identit { client: executionClient({ getSession: async () => session(), - startTurn: async () => { - throw new Error("ordinary composer send must not choose Turn admission"); - }, submitMessage: async (input) => { submits.push(input); return { disposition: "turn_started", turnId: "host-turn" }; @@ -661,9 +637,6 @@ test('submits a slash Skill message and reports the Host Skill outcome', async ( { client: executionClient({ getSession: async () => session(), - startTurn: async () => { - throw new Error('a Skill Message must not route around Host admission'); - }, submitMessage: async (input) => { submits.push(input); return { @@ -711,13 +684,6 @@ test("queues a mid-turn send as steering when the Host reports the session busy" { client: executionClient({ getSession: async () => session(), - startTurn: async () => { - throw new RuntimeHostOperationError( - "turn.start", - "session_busy", - "Session already has an active root Turn", - ); - }, submitMessage: async (input) => { submits.push(input); return { disposition: "steering", queueRevision: 1 }; @@ -762,74 +728,8 @@ test("queues a mid-turn send as steering when the Host reports the session busy" ]); }); -test("retries a dispatched normal send with its original Turn identity", async () => { - const starts: unknown[] = []; - let reconnectQueries = 0; - const ipc = ipcHarness(); - registerExecutionIpc( - { - client: executionClient({ - getSession: async () => { - reconnectQueries += 1; - return sideConversationSession(); - }, - startTurn: async (input) => { - starts.push(input); - if (starts.length === 1) { - throw new RuntimeHostRequestInterruptedError( - "turn.start", - "command", - "dispatched", - "connection_lost", - ); - } - return { - kind: "started", - turn: { - sessionId: input.sessionId, - turnId: input.turnId, - runId: "run-1", - status: "running", - }, - skillInvocation: { loaded: [], failed: [], receipts: [] }, - }; - }, - }), - newId: () => "turn-1", - }, - ipc, - ); - - const result = await ipc.invoke("sessions:send", "session-1", { - type: "send", - turnId: 'message-1', - text: "keep this Turn identity", - }); - - assert.equal(reconnectQueries, 2, 'initial Session lookup plus reconnect probe'); - assert.deepEqual(starts, [ - { - sessionId: "session-1", - turnId: "message-1", - content: { text: "keep this Turn identity", inlineReferences: [] }, - }, - { - sessionId: "session-1", - turnId: "message-1", - content: { text: "keep this Turn identity", inlineReferences: [] }, - }, - ]); - assert.deepEqual(result, { - ok: true, - turnId: "message-1", - attachments: [], - inlineReferences: [], - skillInvocation: { loaded: [], failed: [], receipts: [] }, - }); -}); - -test("does not add admission retry semantics to an ordinary send", async () => { - let starts = 0; +test("resolves a twice-interrupted send as an unknown outcome", async () => { + let submits = 0; let sessionQueries = 0; const ipc = ipcHarness(); registerExecutionIpc( @@ -839,10 +739,10 @@ test("does not add admission retry semantics to an ordinary send", async () => { sessionQueries += 1; return session(); }, - startTurn: async () => { - starts += 1; + submitMessage: async () => { + submits += 1; throw new RuntimeHostRequestInterruptedError( - "turn.start", + "turn.message.submit", "command", "dispatched", "connection_lost", @@ -854,18 +754,25 @@ test("does not add admission retry semantics to an ordinary send", async () => { ipc, ); - await assert.rejects( - ipc.invoke("sessions:send", "session-1", { + // The Message identity is stable, so one retry is safe; a second lost answer + // is still an outcome the renderer must not resolve on its own. + assert.deepEqual( + await ipc.invoke("sessions:send", "session-1", { type: "send", text: "preserve the ordinary send contract", }), - RuntimeHostRequestInterruptedError, + { + ok: false, + reason: "outcome_unknown", + messageId: "turn-1", + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }, ); - assert.equal(starts, 1); - assert.equal(sessionQueries, 1, "only the initial Session lookup runs"); + assert.equal(submits, 2); + assert.equal(sessionQueries, 2, "initial Session lookup plus reconnect probe"); }); -test("retries a dispatched busy fallback with its original message identity", async () => { +test("retries a dispatched send with its original message identity", async () => { const submits: unknown[] = []; let reconnectQueries = 0; const ipc = ipcHarness(); @@ -878,13 +785,6 @@ test("retries a dispatched busy fallback with its original message identity", as ? sideConversationSession(sessionId) : session(); }, - startTurn: async () => { - throw new RuntimeHostOperationError( - "turn.start", - "session_busy", - "Session already has an active root Turn", - ); - }, submitMessage: async (input) => { submits.push(input); if ( @@ -971,7 +871,7 @@ test("retries a dispatched busy fallback with its original message identity", as ); }); -test("starts the turn from the queued message when the busy race resolves idle", async () => { +test("answers a send with the Turn the Host started for it", async () => { const changes: unknown[] = []; const submits: unknown[] = []; const ipc = ipcHarness(); @@ -979,13 +879,6 @@ test("starts the turn from the queued message when the busy race resolves idle", { client: executionClient({ getSession: async () => session(), - startTurn: async () => { - throw new RuntimeHostOperationError( - "turn.start", - "session_busy", - "Session already has an active root Turn", - ); - }, submitMessage: async (input) => { submits.push(input); return { @@ -1030,59 +923,130 @@ test("starts the turn from the queued message when the busy race resolves idle", ]); }); -test("keeps the busy failure for a Skill send instead of degrading it to steering", async () => { +test("propagates a busy explicit Skill send instead of degrading it to steering", async () => { const submits: unknown[] = []; const ipc = ipcHarness(); registerExecutionIpc( { client: executionClient({ getSession: async () => session(), - startTurn: async () => { + submitMessage: async (input) => { + submits.push(input); throw new RuntimeHostOperationError( - "turn.start", + "turn.message.submit", "session_busy", "Session already has an active root Turn", ); }, + }), + newId: () => "id-1", + }, + ipc, + ); + + // Explicit skillIds are exact-Turn intent, so the Host refuses on a busy + // Session. The Desktop no longer carves that case out — it just reports it. + await assert.rejects( + ipc.invoke("sessions:send", "session-1", { + type: "send", + turnId: "turn-1", + text: "", + displayText: "/skill:review", + skillIds: ["review"], + }), + (error: unknown) => + error instanceof RuntimeHostOperationError && error.code === "session_busy", + ); + assert.deepEqual(submits, [ + { + sessionId: "session-1", + messageId: "turn-1", + placement: "current_turn", + content: { + text: "", + displayText: "/skill:review", + inlineReferences: [], + }, + skillIds: ["review"], + }, + ]); +}); + +test("lets the Host queue a textual Skill token as steering", async () => { + const submits: unknown[] = []; + const ipc = ipcHarness(); + registerExecutionIpc( + { + client: executionClient({ + getSession: async () => session(), submitMessage: async (input) => { submits.push(input); return { disposition: "steering", queueRevision: 1 }; }, }), - observer: unusedObserver(), - attachmentApprovals: createAttachmentApprovalRegistry(), - emitSessionsChanged() {}, - stat: async () => ({ size: 0 }), - resizeImage: async (bytes) => bytes, - beforeStop() {}, newId: () => "id-1", }, ipc, ); - // The Desktop composer carries Skills as canonical /skill: tokens in the - // text; explicit skillIds is the protocol-level variant. - await assert.rejects( - ipc.invoke("sessions:send", "session-1", { + // A `/skill:` token in the text is not exact-Turn intent: Host message + // preparation expands it on the queued path too. The Desktop stopped + // sniffing content for it, so this send is reported as the steering the + // Host made of it. + assert.deepEqual( + await ipc.invoke("sessions:send", "session-1", { type: "send", turnId: "turn-1", text: "/skill:review explain the tests", }), - (error: unknown) => - error instanceof RuntimeHostOperationError && error.code === "session_busy", + { + ok: true, + steered: true, + turnId: "turn-1", + attachments: [], + inlineReferences: [], + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }, ); - await assert.rejects( - ipc.invoke("sessions:send", "session-1", { + assert.equal(submits.length, 1); +}); + +test("reports a Host-blocked Skill send as a Skill failure", async () => { + const ipc = ipcHarness(); + registerExecutionIpc( + { + client: executionClient({ + getSession: async () => session(), + submitMessage: async () => ({ + disposition: "blocked", + skillInvocation: { + loaded: [], + failed: [{ request: "missing", reason: "not_found" }], + receipts: [], + }, + }), + }), + newId: () => "id-1", + }, + ipc, + ); + + assert.deepEqual( + await ipc.invoke("sessions:send", "session-1", { type: "send", turnId: "turn-1", - text: "", - displayText: "/skill:review", - skillIds: ["review"], + text: "/skill:missing inspect this", }), - (error: unknown) => - error instanceof RuntimeHostOperationError && error.code === "session_busy", + { + ok: false, + reason: "skill_invocation_failed", + skillInvocation: { + loaded: [], + failed: [{ request: "missing", reason: "not_found" }], + receipts: [], + }, + }, ); - assert.deepEqual(submits, []); }); test("queues explicit Desktop follow-ups", async () => { @@ -1547,7 +1511,6 @@ function executionClient(overrides: Partial): ExecutionClient { updateQueueEntry: unavailable, reorderQueueEntries: unavailable, setSessionReadMarker: unavailable, - startTurn: unavailable, startTurnResume: unavailable, submitMessage: unavailable, updateSessionConfiguration: unavailable, diff --git a/apps/desktop/src/main/__tests__/workhub-controller.test.ts b/apps/desktop/src/main/__tests__/workhub-controller.test.ts index 467fb567c6..6ed05e7962 100644 --- a/apps/desktop/src/main/__tests__/workhub-controller.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-controller.test.ts @@ -1335,6 +1335,52 @@ test('a correction after remount stops a root whose admission is still pending', ]); }); +test('a correction stops an uncertain root under the Turn identity the Host minted', async () => { + const stopped: Array<[string, string]> = []; + const sessions = port([ + session('login', { sessionName: '登录稳定性', updatedAt: 20 }), + session('payment', { sessionName: '支付稳定性', updatedAt: 30 }), + ]); + sessions.reserveTurnId = () => 'reserved-payment'; + // The Host admitted the Message and opened a Turn under its own identity, + // then the answer was lost. Only the transcript can tie the reserved Message + // identity back to that Turn. + sessions.submit = async (target, _text, turnId) => { + if (target.sessionId !== 'payment') return { turnId }; + throw new WorkHubSessionSubmitError('delivery outcome is unknown', 'unknown'); + }; + // The transcript has not caught up at delivery time, so the candidate stays + // uncertain; the correction is the next chance to resolve it. + let transcriptCaughtUp = false; + sessions.reconcileSubmission = async (_target, reservedTurnId) => { + if (reservedTurnId !== 'reserved-payment' || !transcriptCaughtUp) { + transcriptCaughtUp = true; + return { kind: 'unknown' }; + } + return { kind: 'root', turnId: 'turn-payment-host' }; + }; + sessions.stop = async (target, turnId) => { + stopped.push([target.sessionId, turnId]); + }; + const controller = createWorkHubController({ sessions }); + + await assert.rejects(controller.submit({ + requestId: 'request-payment-uncertain', + text: '继续支付稳定性', + explicitTarget: { sessionId: 'payment' }, + })); + + const corrected = await controller.submit({ + requestId: 'request-correct-uncertain-root', + text: '不是这个工作，换成登录稳定性', + }); + + assert.deepEqual(corrected.kind === 'submitted' ? corrected.target : undefined, { + sessionId: 'login', + }); + assert.deepEqual(stopped, [['payment', 'turn-payment-host']]); +}); + test('a correction retries Stop when the same reserved root is admitted before Stop settles', async () => { const stopped: Array<[string, string]> = []; const sessions = port([ diff --git a/apps/desktop/src/main/__tests__/workhub-session-port.test.ts b/apps/desktop/src/main/__tests__/workhub-session-port.test.ts index af3d8d48d7..50e4ec5afd 100644 --- a/apps/desktop/src/main/__tests__/workhub-session-port.test.ts +++ b/apps/desktop/src/main/__tests__/workhub-session-port.test.ts @@ -589,7 +589,7 @@ test('desktop adapter preserves when Session delivery steered an existing root T }); test('desktop adapter distinguishes definite rejection from an unknown delivery outcome', async () => { - let outcome: 'throw' | 'reject' = 'throw'; + let outcome: 'throw' | 'unknown' | 'reject' = 'throw'; const adapter = createDesktopWorkHubSessionPort({ transcripts: unusedTranscripts, sessions: { @@ -598,7 +598,23 @@ test('desktop adapter distinguishes definite rejection from an unknown delivery create: async () => { throw new Error('not used'); }, send: async () => { if (outcome === 'throw') throw new Error('transport disconnected'); - return { ok: false as const, reason: 'archived' as const }; + if (outcome === 'unknown') { + return { + ok: false as const, + reason: 'outcome_unknown' as const, + messageId: 'reserved-turn', + skillInvocation: { loaded: [], failed: [], receipts: [] }, + }; + } + return { + ok: false as const, + reason: 'skill_invocation_failed' as const, + skillInvocation: { + loaded: [], + failed: [{ request: 'missing', reason: 'not_found' as const }], + receipts: [], + }, + }; }, stop: async () => {}, subscribeChanges: () => () => {}, @@ -607,6 +623,13 @@ test('desktop adapter distinguishes definite rejection from an unknown delivery newTurnId: () => 'reserved-turn', }); + await assert.rejects( + adapter.submit({ sessionId: 'payment' }, '继续支付', 'reserved-turn'), + (error) => error instanceof WorkHubSessionSubmitError && error.admission === 'unknown', + ); + // The Host declining to prove the outcome must stay reconcilable; only a + // Host-owned refusal releases the reserved root. + outcome = 'unknown'; await assert.rejects( adapter.submit({ sessionId: 'payment' }, '继续支付', 'reserved-turn'), (error) => error instanceof WorkHubSessionSubmitError && error.admission === 'unknown', 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 c5107d5198..d281173ac0 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 @@ -24,7 +24,6 @@ import { RuntimeHostOperationError, RuntimeHostRequestInterruptedError, } from '@maka/runtime-host/client'; -import { SKILL_INVOCATION_TOKEN_SOURCE } from '@maka/core/skill-invocation-token'; import { isSideConversationSession } from '@maka/core/side-conversation'; import { type SessionChangedEvent, @@ -105,7 +104,6 @@ type RuntimeHostSessionExecutionClient = Pick< | "updateQueueEntry" | "reorderQueueEntries" | "setSessionReadMarker" - | "startTurn" | "startTurnResume" | "submitMessage" | "updateSessionMetadata" @@ -325,9 +323,17 @@ export function registerRuntimeHostSessionExecutionIpc( displayText, workspaceFileReferences: command.workspaceFileReferences, }); - const startInput = { + // Runtime Host is the sole admission authority: one submit answers + // whether the words opened a Turn or joined the running one, and the + // Desktop never routes on content — an explicit Skill or orchestration + // still fails closed on a busy Session, in the Host. The Message identity + // is the Turn id the caller reserved: one submit, one durable Message, + // and a retry the Host recognizes as the same one. + const messageId = turnId; + const submitted = await submitMessageWithReconnect(deps.client, { sessionId, - turnId, + messageId, + placement: "current_turn" as const, content: { text: command.text, ...(command.displayText !== undefined @@ -337,99 +343,49 @@ export function registerRuntimeHostSessionExecutionIpc( ...(command.quotes ? { quotes: command.quotes } : {}), inlineReferences, }, - ...((command.skillIds?.length ?? 0) > 0 - ? { skillIds: command.skillIds } - : {}), + ...((command.skillIds?.length ?? 0) > 0 ? { skillIds: command.skillIds } : {}), ...(command.turnOrchestration ? { turnOrchestration: command.turnOrchestration } : {}), - }; - let startResult; - try { - startResult = sideConversation - ? await retryDispatchedCommand( - () => deps.client.startTurn(startInput), - () => deps.client.getSession(sessionId), - ) - : await deps.client.startTurn(startInput); - } catch (error) { - // The renderer routes text at a session it sees as running to - // `sessions:steer`, but its view can lag the Host: another window, a - // Bot, or a Goal continuation may have opened the root Turn first, and - // that race surfaced here as a session_busy send failure that dropped - // the user's message (#1954). `turn.message.submit` resolves the race - // on the Host: an active session queues the text as steering, an idle - // one starts the Turn. Skill and orchestration sends keep the error — - // their turn semantics cannot be expressed as a queued message — and - // the Desktop composer carries Skills as canonical /skill: tokens in - // the text, not as skillIds. - if ( - !(error instanceof RuntimeHostOperationError) || - error.code !== "session_busy" || - (command.skillIds?.length ?? 0) > 0 || - command.turnOrchestration || - new RegExp(SKILL_INVOCATION_TOKEN_SOURCE).test(command.text) - ) { - throw error; - } - // Preserve the renderer's command identity in the durable message so - // a lost IPC reply can be reconciled as root-vs-steering later. - const messageId = turnId; - const submitInput = { - sessionId, - messageId, - content: startInput.content, - placement: 'current_turn' as const, - }; - const submitted = await submitMessageWithReconnect(deps.client, submitInput); - if (!submitted) { - return { - ok: false as const, - reason: 'outcome_unknown' as const, - messageId, - skillInvocation: EMPTY_SKILL_INVOCATION, - }; - } - if (submitted.disposition === "turn_started") { - deps.emitSessionsChanged("status-change", sessionId, { - turnId: submitted.turnId, - }); - return { - ok: true as const, - turnId: submitted.turnId, - attachments, - inlineReferences, - skillInvocation: EMPTY_SKILL_INVOCATION, - }; - } - // The steering renderer believed this session idle; nudge it to - // refresh so its composer converges on the running turn. - deps.emitSessionsChanged("status-change", sessionId); + }); + if (!submitted) { return { - ok: true as const, - steered: true as const, - turnId, - ...(sideConversation ? { messageId } : {}), - attachments, - inlineReferences, + ok: false as const, + reason: 'outcome_unknown' as const, + messageId, skillInvocation: EMPTY_SKILL_INVOCATION, }; } - if (startResult.kind === "blocked") { + if (submitted.disposition === "blocked") { return { ok: false as const, + reason: "skill_invocation_failed" as const, + skillInvocation: submitted.skillInvocation, + }; + } + if (submitted.disposition === "turn_started") { + deps.emitSessionsChanged("status-change", sessionId, { + turnId: submitted.turnId, + }); + return { + ok: true as const, + turnId: submitted.turnId, attachments, inlineReferences, - skillInvocation: startResult.skillInvocation, + skillInvocation: submitted.skillInvocation ?? EMPTY_SKILL_INVOCATION, }; } - deps.emitSessionsChanged("status-change", sessionId, { turnId }); + // The sending surface believed this Session idle; nudge it to refresh so + // its composer converges on the running Turn. + deps.emitSessionsChanged("status-change", sessionId); return { ok: true as const, + steered: true as const, turnId, + ...(sideConversation ? { messageId } : {}), attachments, inlineReferences, - skillInvocation: startResult.skillInvocation, + skillInvocation: EMPTY_SKILL_INVOCATION, }; }, ); diff --git a/apps/desktop/src/preload/bridge-contract.d.ts b/apps/desktop/src/preload/bridge-contract.d.ts index 94129cd1f7..8d056da319 100644 --- a/apps/desktop/src/preload/bridge-contract.d.ts +++ b/apps/desktop/src/preload/bridge-contract.d.ts @@ -849,11 +849,12 @@ export interface MakaBridge { ): Promise< | { ok: true; - turnId: string; /** - * The send raced a root Turn another client opened first and was - * queued into it as steering instead of starting `turnId`. + * The Turn Runtime Host opened for this Message. Admission mints it, + * so it is not the `turnId` the caller reserved — that identity is + * the Message's, and stays the caller's to reconcile with. */ + turnId: string; steered?: never; messageId?: never; attachments: import('@maka/core/events').AttachmentRef[]; @@ -862,6 +863,10 @@ export interface MakaBridge { } | { ok: true; + /** + * The running Turn this Message was queued into as steering, rather + * than one opened for it. + */ turnId: string; steered: true; /** Host admission identity for the message queued as steering. */ 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 adb5c99a67..68e8ea592a 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 @@ -191,6 +191,7 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan const stopRequestRef = useRef | null>(null); const activeTurnIdRef = useRef(null); const pendingAdmissionRef = useRef(null); + const reconcilingAdmissionRef = useRef(null); const subscriptionReadyRef = useRef>(Promise.resolve()); const submitLockRef = useRef(false); const settlingTurnIdsRef = useRef>(new Set()); @@ -308,6 +309,33 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan [applyOwnedEvent, setPendingAdmission], ); + // A Message whose admission answer was lost is still reconcilable: the Host + // materializes a root Message before its Run starts, so the durable + // transcript names the Turn it opened under the identity the panel sent. + const reconcileUnknownAdmission = useCallback( + async (forkId: string, admission: PendingAdmission): Promise => { + if (reconcilingAdmissionRef.current === admission) return; + reconcilingAdmissionRef.current = admission; + try { + const { messages } = await sideChat.readSettledMessages(forkId); + if (!mountedRef.current || pendingAdmissionRef.current !== admission) return; + const admitted = messages.find( + (message) => message.type === 'user' && message.id === admission.messageId, + ); + if (admitted?.turnId) { + bindAdmittedTurn(forkId, admitted.turnId, { preserveLiveTurn: true }); + } + } catch { + // The next event this fork produces is another chance to reconcile. + } finally { + if (reconcilingAdmissionRef.current === admission) { + reconcilingAdmissionRef.current = null; + } + } + }, + [bindAdmittedTurn, mountedRef, sideChat], + ); + const releaseAdmission = useCallback( (admission: PendingAdmission, message?: string) => { if (pendingAdmissionRef.current !== admission) return; @@ -398,6 +426,9 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan applyOwnedEvent(forkId, event); } else { admission.events.push(event); + // This fork is producing Turn events while the panel still holds an + // unproven Message: the transcript can say whether they are its own. + void reconcileUnknownAdmission(forkId, admission); } return; } @@ -645,6 +676,7 @@ export function useQuoteCompanion(input: UseQuoteCompanionInput): UseQuoteCompan if (outcome?.kind === 'retracted') { return false; } + void reconcileUnknownAdmission(result.forkId, admission); } else if (result.steered) { if (resolveAdmission(result.forkId, admission, result.messageId)?.kind === 'retracted') { return false; diff --git a/apps/desktop/src/renderer/workhub-session-port.ts b/apps/desktop/src/renderer/workhub-session-port.ts index bfa7415323..0fa255bc34 100644 --- a/apps/desktop/src/renderer/workhub-session-port.ts +++ b/apps/desktop/src/renderer/workhub-session-port.ts @@ -211,9 +211,12 @@ export function createDesktopWorkHubSessionPort(deps: { ); } if (!result.ok) { + // `outcome_unknown` is the Host declining to prove what happened, not a + // refusal: the Message may already be running, so it stays reachable + // for reconciliation rather than being released. throw new WorkHubSessionSubmitError( `WorkHub Session send failed: ${result.reason}`, - 'rejected', + result.reason === 'outcome_unknown' ? 'unknown' : 'rejected', ); } return { diff --git a/packages/runtime-host/src/__tests__/execution-host-session-title.test.ts b/packages/runtime-host/src/__tests__/execution-host-session-title.test.ts new file mode 100644 index 0000000000..8ac74faeb0 --- /dev/null +++ b/packages/runtime-host/src/__tests__/execution-host-session-title.test.ts @@ -0,0 +1,102 @@ +/* + * 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 { randomUUID } from 'node:crypto'; +import { test } from 'node:test'; +import { DEFAULT_SESSION_NAME } from '@maka/core/session-name'; +import type { RuntimeHostConnection } from '../client/index.js'; +import { + connectClient, + PROCESS_TIMEOUT_MS, + requireStartedTurn, + waitForTerminalTurn, + withExecutionRoot, +} from './fixtures/execution-host-suite.js'; + +// The Session title effect runs after the Run starts and writes out of band, so +// the name lands after the Turn is already terminal. Without a reachable title +// model the effect falls back to the Message's first line, which is what makes +// this assertion independent of any provider. +async function readSession(client: RuntimeHostConnection, sessionId: string) { + const result = await client.request('session.catalog.query', { kind: 'get', sessionId }); + assert.equal(result.kind, 'session'); + if (result.kind !== 'session') assert.fail('Expected a Session projection'); + const session = result.session; + assert.ok(session && !('reason' in session), 'Expected a wire-representable Session'); + if (!session || 'reason' in session) assert.fail('Expected a wire-representable Session'); + return session; +} + +async function waitForGeneratedName( + client: RuntimeHostConnection, + sessionId: string, +): Promise { + const deadline = Date.now() + PROCESS_TIMEOUT_MS; + let name = DEFAULT_SESSION_NAME; + while (Date.now() < deadline) { + name = (await readSession(client, sessionId)).name; + if (name !== DEFAULT_SESSION_NAME) return name; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + return name; +} + +test('a submitted first Message names its default-named Session', async () => { + await withExecutionRoot(async (fixture) => { + const host = await fixture.startHost(); + const client = await connectClient(fixture.root); + + const submitted = await client.request('turn.message.submit', { + originHostEpoch: host.hostEpoch, + sessionId: fixture.sessionId, + messageId: randomUUID(), + placement: 'current_turn', + content: { text: 'draft the release notes' }, + }); + assert.equal(submitted.disposition, 'turn_started'); + if (submitted.disposition !== 'turn_started') assert.fail('Expected a started Turn'); + await waitForTerminalTurn(client, fixture.sessionId, submitted.turnId); + + assert.equal(await waitForGeneratedName(client, fixture.sessionId), 'draft the release notes'); + // The first user Message is also what freezes the Session's route; no Run + // writes that fact separately. + assert.equal((await readSession(client, fixture.sessionId)).connectionLocked, true); + }); +}); + +test('a started first Turn names its default-named Session', async () => { + await withExecutionRoot(async (fixture) => { + await fixture.startHost(); + const client = await connectClient(fixture.root); + + const turnId = randomUUID(); + requireStartedTurn( + await client.request('turn.start', { + sessionId: fixture.sessionId, + turnId, + content: { text: 'draft the release notes' }, + }), + ); + await waitForTerminalTurn(client, fixture.sessionId, turnId); + + assert.equal(await waitForGeneratedName(client, fixture.sessionId), 'draft the release notes'); + assert.equal((await readSession(client, fixture.sessionId)).connectionLocked, true); + }); +}); diff --git a/packages/runtime-host/src/__tests__/session-effect-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-effect-coordinator.test.ts index 328c6dbb8b..6c4be6855d 100644 --- a/packages/runtime-host/src/__tests__/session-effect-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-effect-coordinator.test.ts @@ -202,21 +202,20 @@ test('Session effect leaves Turn admission free and drain aborts accepted recap ); }); -test('Automatic title generation fences retirement until the effect settles', async () => { +test('Automatic naming fences retirement until the effect settles', async () => { const started = gate(); await withHarness( async ({ coordinator }) => { - const pending = coordinator.generateTitle({ + coordinator.nameSessionFromRootMessage({ sessionId: 'session-1', - header: { id: 'session-1' } as SessionHeader, - sourceText: 'A first user message', + content: { text: 'A first user message' }, }); await started.promise; assert.equal(coordinator.hasLiveSessionState('session-1'), true); assert.equal(coordinator.hasLiveSessionState('session-2'), false); coordinator.beginDrain(); - assert.equal(await pending, undefined); + await coordinator.close(); assert.equal(coordinator.hasLiveSessionState('session-1'), false); }, { @@ -227,7 +226,13 @@ test('Automatic title generation fences retirement until the effect settles', as ); return undefined; }, - generateRecap: async () => assert.fail('title generation must not call recap'), + generateRecap: async () => assert.fail('naming must not call recap'), + }, + { + readSessionHeader: async () => unnamedHeader(), + // Draining retires the effect rather than downgrading it: the fallback + // name answers an unreachable model, not a Host that is shutting down. + nameSessionIfUnnamed: async () => assert.fail('an aborted naming must not write'), }, ); }); @@ -278,6 +283,139 @@ test('Session recap keeps accounting failures non-terminal and unsafe to retry', ); }); +test('Naming writes what the title model generated, not the Message', async () => { + const named = gate(); + const titles: string[] = []; + await withHarness( + async ({ coordinator }) => { + coordinator.nameSessionFromRootMessage({ + sessionId: 'session-1', + content: { text: 'hello world' }, + }); + await named.promise; + assert.deepEqual(titles, ['Model title']); + }, + { + generateTitle: async () => 'Model title', + generateRecap: async () => assert.fail('naming must not call recap'), + }, + { + readSessionHeader: async () => unnamedHeader(), + nameSessionIfUnnamed: async (_sessionId, title) => { + titles.push(title); + return unnamedHeader(); + }, + onSessionNamed: () => named.release(), + }, + ); +}); + +test('Naming falls back to the Message when the title model is unreachable', async () => { + const named = gate(); + const titles: string[] = []; + let notifications = 0; + await withHarness( + async ({ coordinator }) => { + coordinator.nameSessionFromRootMessage({ + sessionId: 'session-1', + content: { text: '\nFallback title\nignored' }, + }); + await named.promise; + assert.deepEqual(titles, ['Fallback title']); + assert.equal(notifications, 1); + }, + { + generateTitle: async () => { + throw new Error('offline'); + }, + generateRecap: async () => assert.fail('naming must not call recap'), + }, + { + readSessionHeader: async () => unnamedHeader(), + nameSessionIfUnnamed: async (_sessionId, title) => { + titles.push(title); + return unnamedHeader(); + }, + onSessionNamed: () => { + notifications += 1; + named.release(); + }, + }, + ); +}); + +test('A named Session is never renamed by the effect', async () => { + let modelCalls = 0; + await withHarness( + async ({ coordinator }) => { + coordinator.nameSessionFromRootMessage({ + sessionId: 'session-1', + content: { text: 'hello' }, + }); + await coordinator.close(); + assert.equal(modelCalls, 0); + }, + { + generateTitle: async () => { + modelCalls += 1; + return 'Generated loses'; + }, + generateRecap: async () => assert.fail('naming must not call recap'), + }, + { + readSessionHeader: async () => ({ + ...unnamedHeader(), + name: 'Manual wins', + titleIsManual: true, + }), + nameSessionIfUnnamed: async () => assert.fail('a named Session must not be renamed'), + onSessionNamed: () => assert.fail('a named Session must not notify'), + }, + ); +}); + +test('A racing manual rename wins over the generated title', async () => { + const attempted = gate(); + let notifications = 0; + await withHarness( + async ({ coordinator }) => { + coordinator.nameSessionFromRootMessage({ + sessionId: 'session-1', + content: { text: 'hello' }, + }); + await attempted.promise; + await coordinator.close(); + assert.equal(notifications, 0); + }, + { + generateTitle: async () => 'Generated loses', + generateRecap: async () => assert.fail('naming must not call recap'), + }, + { + readSessionHeader: async () => unnamedHeader(), + // The store re-checks the name under its own write: the rename landed + // first, so the generated title is refused. + nameSessionIfUnnamed: async () => { + attempted.release(); + return null; + }, + onSessionNamed: () => { + notifications += 1; + }, + }, + ); +}); + +function unnamedHeader(): SessionHeader { + return { + id: 'session-1', + name: 'New Chat', + titleIsManual: false, + isArchived: false, + status: 'active', + } as unknown as SessionHeader; +} + async function withHarness( run: (input: { store: Awaited>; @@ -336,6 +474,8 @@ function createCoordinator( options.readSessionHeader ?? (async () => ({ isArchived: false, status: 'active' }) as unknown as SessionHeader), sessionAdmission: options.admission ?? new SessionAdmissionGate(), + nameSessionIfUnnamed: options.nameSessionIfUnnamed ?? (async () => null), + onSessionNamed: options.onSessionNamed ?? (() => undefined), acquireResidency: () => ({ release: () => undefined }), requestDrain: options.requestDrain ?? @@ -347,6 +487,11 @@ interface CoordinatorOptions { readonly admission?: SessionAdmissionGate; readonly readSessionHeader?: (sessionId: string) => Promise; readonly requestDrain?: () => void; + readonly nameSessionIfUnnamed?: ( + sessionId: string, + title: string, + ) => Promise; + readonly onSessionNamed?: (sessionId: string) => void; } function gate(): { promise: Promise; release(): void } { diff --git a/packages/runtime-host/src/__tests__/session-effect-two-client-uds.test.ts b/packages/runtime-host/src/__tests__/session-effect-two-client-uds.test.ts index 1c2e6c24e3..ad46b006b3 100644 --- a/packages/runtime-host/src/__tests__/session-effect-two-client-uds.test.ts +++ b/packages/runtime-host/src/__tests__/session-effect-two-client-uds.test.ts @@ -77,6 +77,8 @@ test('two Clients share one durable Session recap effect', async () => { readSessionHeader: async () => ({ isArchived: false, status: 'active' }) as unknown as SessionHeader, sessionAdmission: new SessionAdmissionGate(), + nameSessionIfUnnamed: async () => assert.fail('this Host only serves recap effects'), + onSessionNamed: () => assert.fail('this Host only serves recap effects'), acquireResidency: () => context.acquireResidency('session-effect'), requestDrain: context.requestDrain, }); diff --git a/packages/runtime-host/src/__tests__/session-title.test.ts b/packages/runtime-host/src/__tests__/session-title.test.ts new file mode 100644 index 0000000000..158f86e7cd --- /dev/null +++ b/packages/runtime-host/src/__tests__/session-title.test.ts @@ -0,0 +1,69 @@ +/* + * 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, test } from 'node:test'; +import { + cleanGeneratedSessionTitle, + fallbackSessionTitle, + sessionTitleSource, +} from '../server/session-title.js'; + +describe('session title helper', () => { + test('uses display text, strips system reminders, and truncates input on a UTF-8 boundary', () => { + const source = sessionTitleSource({ + text: 'model envelope', + displayText: `secret\n${'🦊'.repeat(3_000)}`, + }); + + assert.equal(source.includes('secret'), false); + assert.equal(new TextEncoder().encode(source).length <= 8 * 1024, true); + assert.equal(source.endsWith('�'), false); + }); + + test('extracts the user message from a raw skill envelope', () => { + assert.equal( + sessionTitleSource({ + text: 'Skills loaded below.\n\nSECRET INSTRUCTIONS\n\n\nAnalyze this code\n', + }), + 'Analyze this code', + ); + }); + + test('builds fallback from the first non-empty line without splitting Unicode code points', () => { + const line = `${'🦊'.repeat(42)}tail`; + assert.equal(fallbackSessionTitle(`\n \n${line}\nignored`), '🦊'.repeat(42)); + assert.equal(fallbackSessionTitle(' \n\t'), undefined); + }); + + test('cleans model reasoning, prefixes, quotes, and extra lines', () => { + assert.equal( + cleanGeneratedSessionTitle( + 'reasoning\nTitle: "Production log analysis"\nextra', + ), + 'Production log analysis', + ); + assert.equal(cleanGeneratedSessionTitle('「生产日志分析」'), '生产日志分析'); + }); + + test('refuses model output that carries no usable name', () => { + assert.equal(cleanGeneratedSessionTitle('x'), undefined); + assert.equal(cleanGeneratedSessionTitle(' \n\t'), undefined); + }); +}); diff --git a/packages/runtime-host/src/server/execution-composition.ts b/packages/runtime-host/src/server/execution-composition.ts index cf46c22b9c..403e7af4c8 100644 --- a/packages/runtime-host/src/server/execution-composition.ts +++ b/packages/runtime-host/src/server/execution-composition.ts @@ -864,6 +864,9 @@ export async function createExecutionRuntimeHostComposition( sessions: stores.sessionStore, readSessionHeader: (sessionId) => stores.sessionStore.readHeaderSnapshot(sessionId), sessionAdmission, + nameSessionIfUnnamed: (sessionId, title) => + stores.sessionStore.setGeneratedTitleIfAbsent(sessionId, title), + onSessionNamed: (sessionId) => continuityCoordinator.enqueueCanonicalRefresh(sessionId), acquireResidency: () => context.acquireResidency('session-effect'), requestDrain: context.requestDrain, }); @@ -905,9 +908,6 @@ export async function createExecutionRuntimeHostComposition( newId: randomUUID, now: Date.now, safeBoundaryResumeEnabled: process.env.MAKA_RUNTIME_SAFE_BOUNDARY_RESUME === '1', - generateSessionTitle: (input) => sessionEffectCoordinator.generateTitle(input), - onSessionTitleChanged: (sessionId) => - continuityCoordinator.enqueueCanonicalRefresh(sessionId), inspectContinuationSafety: createLocalContinuationSafetyInspector({ readSessionCwd: async (sessionId) => (await stores.sessionStore.readHeaderSnapshot(sessionId)).cwd, @@ -1105,6 +1105,7 @@ export async function createExecutionRuntimeHostComposition( ) ).graphId, }, + (input) => sessionEffectCoordinator.nameSessionFromRootMessage(input), ); const coordinator = rootCoordinator; const contextOperations = new HostContextCoordinator({ diff --git a/packages/runtime-host/src/server/execution-model-authority.ts b/packages/runtime-host/src/server/execution-model-authority.ts index 7df90ad64d..1622d4b799 100644 --- a/packages/runtime-host/src/server/execution-model-authority.ts +++ b/packages/runtime-host/src/server/execution-model-authority.ts @@ -39,7 +39,7 @@ import { buildSessionTitlePrompt, cleanGeneratedSessionTitle, SESSION_TITLE_GENERATION_TIMEOUT_MS, -} from '@maka/runtime/session-title'; +} from './session-title.js'; import { createProxiedFetchTransport, type ProxiedFetchProxy, diff --git a/packages/runtime-host/src/server/root-turn-coordinator.ts b/packages/runtime-host/src/server/root-turn-coordinator.ts index 4d4e44b4e1..adbdc14136 100644 --- a/packages/runtime-host/src/server/root-turn-coordinator.ts +++ b/packages/runtime-host/src/server/root-turn-coordinator.ts @@ -331,6 +331,10 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { attachmentValidator?: HostTurnAttachmentValidator, prepareSkillInvocation?: HostSkillInvocationPreparer, private readonly agentGraphEpochs?: HostAgentGraphEpochAuthority, + private readonly nameSessionFromRootMessage?: (input: { + sessionId: string; + content: MessageContent; + }) => void, ) { this.stores = authenticateExecutionStoresWriter(stores, 'interactive'); this.executionProjection = new HostedExecutionProjectionReader(this.stores); @@ -2234,6 +2238,51 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { }; } + /** + * The one root path that carries a user Message. Session naming hangs here + * rather than on the shared run-started hook: a compaction or a continuation + * opens a Run without new words, and neither should name a Session. + */ + private startRootMessageTurn( + input: RootTurnActivationInput, + active: ActiveRootTurn, + content: MessageContent, + messageOrigin: ReturnType, + onRunStarted: () => Promise, + ): AsyncIterable { + return this.manager.sendMessage( + input.sessionId, + { + turnId: input.turnId, + ...content, + ...(active.descriptor.kind === 'regenerate' + ? { + parentTurnId: active.descriptor.sourceTurnId, + regeneratedFromTurnId: active.descriptor.sourceTurnId, + } + : {}), + ...(input.turnOrchestration ? { turnOrchestration: input.turnOrchestration } : {}), + ...(active.descriptor.kind === 'external_message' && + active.descriptor.maxSteps !== undefined + ? { maxSteps: active.descriptor.maxSteps } + : {}), + ...(messageOrigin ? { origin: messageOrigin } : {}), + }, + { + runId: active.runId, + userMessageId: active.userMessageId, + durability: 'required', + onRunStarted: async (startedRunId) => { + if (startedRunId !== active.runId) { + throw new Error('Runtime started a different Run than the admitted identity'); + } + await onRunStarted(); + this.nameSessionFromRootMessage?.({ sessionId: input.sessionId, content }); + }, + }, + ); + } + private async drainTurn( input: RootTurnActivationInput, active: ActiveRootTurn, @@ -2268,37 +2317,12 @@ export class RootTurnCoordinator implements HostedExecutionAuthority { ? this.manager.resumeSafeBoundaryContinuation(active.continuation, { onRunStarted, }) - : this.manager.sendMessage( - input.sessionId, - { - turnId: input.turnId, - ...normalizeMessageContent(requireRootMessageContent(input)), - ...(active.descriptor.kind === 'regenerate' - ? { - parentTurnId: active.descriptor.sourceTurnId, - regeneratedFromTurnId: active.descriptor.sourceTurnId, - } - : {}), - ...(input.turnOrchestration - ? { turnOrchestration: input.turnOrchestration } - : {}), - ...(active.descriptor.kind === 'external_message' && - active.descriptor.maxSteps !== undefined - ? { maxSteps: active.descriptor.maxSteps } - : {}), - ...(messageOrigin ? { origin: messageOrigin } : {}), - }, - { - runId: active.runId, - userMessageId: active.userMessageId, - durability: 'required', - onRunStarted: async (startedRunId) => { - if (startedRunId !== active.runId) { - throw new Error('Runtime started a different Run than the admitted identity'); - } - await onRunStarted(); - }, - }, + : this.startRootMessageTurn( + input, + active, + normalizeMessageContent(requireRootMessageContent(input)), + messageOrigin, + onRunStarted, ); for await (const event of stream) { if (active.execution?.onEvent) { diff --git a/packages/runtime-host/src/server/session-effect-coordinator.ts b/packages/runtime-host/src/server/session-effect-coordinator.ts index 6696d2cfb2..b8fc3e2249 100644 --- a/packages/runtime-host/src/server/session-effect-coordinator.ts +++ b/packages/runtime-host/src/server/session-effect-coordinator.ts @@ -18,8 +18,11 @@ */ import { createHash } from 'node:crypto'; +import type { MessageContent } from '@maka/core/events'; import type { SessionHeader } from '@maka/core/session'; +import { DEFAULT_SESSION_NAME } from '@maka/core/session-name'; import { cleanSessionRecapText } from '@maka/runtime/session-recap'; +import { fallbackSessionTitle, sessionTitleSource } from './session-title.js'; import { type RuntimeReadModelSessionView } from '@maka/runtime/runtime-read-model'; import { authenticateInteractiveArtifactStoreWriter, @@ -52,6 +55,12 @@ export interface HostSessionEffectCoordinatorInput { readonly sessions: SessionPresenceReader; readonly readSessionHeader: (sessionId: string) => Promise; readonly sessionAdmission: SessionAdmissionGate; + /** Names a Session only while it still carries the default name. */ + readonly nameSessionIfUnnamed: ( + sessionId: string, + title: string, + ) => Promise; + readonly onSessionNamed: (sessionId: string) => void; readonly acquireResidency: () => OperationResidency; readonly requestDrain: () => void; } @@ -86,6 +95,11 @@ export class HostSessionEffectCoordinator { readonly #sessions: SessionPresenceReader; readonly #readSessionHeader: (sessionId: string) => Promise; readonly #sessionAdmission: SessionAdmissionGate; + readonly #nameSessionIfUnnamed: ( + sessionId: string, + title: string, + ) => Promise; + readonly #onSessionNamed: (sessionId: string) => void; readonly #acquireResidency: () => OperationResidency; readonly #requestDrain: () => void; readonly #active = new Set>(); @@ -102,27 +116,77 @@ export class HostSessionEffectCoordinator { this.#sessions = input.sessions; this.#readSessionHeader = input.readSessionHeader; this.#sessionAdmission = input.sessionAdmission; + this.#nameSessionIfUnnamed = input.nameSessionIfUnnamed; + this.#onSessionNamed = input.onSessionNamed; this.#acquireResidency = input.acquireResidency; this.#requestDrain = input.requestDrain; } - generateTitle(input: { + /** + * Names a Session from the root Message that opened its Turn. The Turn owns + * nothing here: the name is the only authority for "still unnamed", and an + * unreachable title model falls back to the Message's first line, so a + * Session that carried words is never left at the default name. + */ + nameSessionFromRootMessage(input: { readonly sessionId: string; - readonly header: SessionHeader; - readonly sourceText: string; - }): Promise { - if (!this.#accepting) return Promise.resolve(undefined); + readonly content: MessageContent; + }): void { + if (!this.#accepting) return; + const sourceText = sessionTitleSource(input.content); + if (!sourceText.trim()) return; + // One naming attempt per Session at a time: a queued Message can open its + // Turn while the first title call is still out, and the second call could + // only ever lose the write. + for (const titleSessionId of this.#titleAborts.values()) { + if (titleSessionId === input.sessionId) return; + } const residency = this.#acquireResidency(); const abort = new AbortController(); this.#titleAborts.set(abort, input.sessionId); - return this.#track( - this.#model.generateTitle({ ...input, abortSignal: abort.signal }).finally(() => { + void this.#track( + this.#nameSession(input.sessionId, sourceText, abort.signal).finally(() => { this.#titleAborts.delete(abort); residency.release(); }), ); } + async #nameSession( + sessionId: string, + sourceText: string, + abortSignal: AbortSignal, + ): Promise { + let generated: string | undefined; + try { + const header = await this.#readSessionHeader(sessionId); + if (header.titleIsManual || header.name !== DEFAULT_SESSION_NAME) return; + generated = await this.#model.generateTitle({ + sessionId, + header, + sourceText, + abortSignal, + }); + } catch { + // An unreachable title model is not a Session failure; the fallback name + // below still beats leaving the Session unnamed. + } + // Shutdown aborts the call, and an abort retires the effect rather than + // downgrading it: the next root Message names the Session. + if (abortSignal.aborted) return; + const title = generated ?? fallbackSessionTitle(sourceText); + if (!title) return; + try { + // Naming answers no caller, so nothing here is a Host-level outcome: a + // racing rename wins the conditional write, and a store that cannot + // answer leaves the Session unnamed for the next root Message to retry. + if (!(await this.#nameSessionIfUnnamed(sessionId, title))) return; + this.#onSessionNamed(sessionId); + } catch { + // The Session keeps the name it already had. + } + } + hasLiveSessionState(sessionId: string): boolean { for (const titleSessionId of this.#titleAborts.values()) { if (titleSessionId === sessionId) return true; diff --git a/packages/runtime/src/session-title.ts b/packages/runtime-host/src/server/session-title.ts similarity index 66% rename from packages/runtime/src/session-title.ts rename to packages/runtime-host/src/server/session-title.ts index e505861050..307dcab381 100644 --- a/packages/runtime/src/session-title.ts +++ b/packages/runtime-host/src/server/session-title.ts @@ -17,7 +17,6 @@ * under the License. */ -import { generateText as aiGenerateText, type LanguageModel } from 'ai'; import { normalizeUserSessionName } from '@maka/core/session-name'; const MAX_SOURCE_BYTES = 8 * 1024; @@ -51,46 +50,6 @@ export function fallbackSessionTitle(sourceText: string): string | undefined { return firstLine ? Array.from(firstLine).slice(0, MAX_FALLBACK_CODE_POINTS).join('') : undefined; } -type GenerateText = (options: Record) => Promise<{ - text: string; - finishReason?: string; -}>; - -export async function generateSessionTitle(input: { - model: LanguageModel; - sourceText: string; - providerOptions?: unknown; - generateText?: GenerateText; - timeoutMs?: number; -}): Promise { - if (!input.sourceText.trim()) return undefined; - try { - const generateText: GenerateText = - input.generateText ?? - (async (options) => aiGenerateText(options as Parameters[0])); - const abortSignal = AbortSignal.timeout(input.timeoutMs ?? SESSION_TITLE_GENERATION_TIMEOUT_MS); - let onAbort!: () => void; - const timeout = new Promise((_resolve, reject) => { - onAbort = () => reject(abortSignal.reason); - abortSignal.addEventListener('abort', onAbort, { once: true }); - }); - const result = await Promise.race([ - generateText({ - model: input.model, - prompt: buildSessionTitlePrompt(input.sourceText), - ...(input.providerOptions === undefined ? {} : { providerOptions: input.providerOptions }), - maxOutputTokens: 1024, - abortSignal, - }), - timeout, - ]).finally(() => abortSignal.removeEventListener('abort', onAbort)); - if (result.finishReason === 'length') return undefined; - return cleanGeneratedSessionTitle(result.text); - } catch { - return undefined; - } -} - export function buildSessionTitlePrompt(sourceText: string): string { return `Create a descriptive 5–10 word title for the user message below. Use the user's language; for Chinese and similar languages, use an equivalently brief natural title. Output only the title.\n\n${sourceText}`; } diff --git a/packages/runtime/package.json b/packages/runtime/package.json index 53d05f563c..c4d2a6bad5 100644 --- a/packages/runtime/package.json +++ b/packages/runtime/package.json @@ -87,7 +87,6 @@ "./sandbox-boundary-tool": "./dist/sandbox-boundary-tool.js", "./scheduled-task-tools": "./dist/scheduled-task-tools.js", "./session-recap": "./dist/session-recap.js", - "./session-title": "./dist/session-title.js", "./session-trace-projection": "./dist/session-trace-projection.js", "./shell-detect": "./dist/shell-detect.js", "./shell-run-contract": "./dist/shell-run-contract.js", diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index ea3fd51b75..abd01c1748 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -3816,69 +3816,6 @@ describe('SessionManager child-session runtime primitive', () => { }); }); -describe('SessionManager automatic titles', () => { - test('falls back once on generation failure', async () => { - const store = new MemorySessionStore(); - const backends = new BackendRegistry(); - backends.register('ai-sdk', (ctx) => new TestBackend(ctx)); - const changed = makeGate(); - let calls = 0; - const manager = new SessionManager({ - store, - backends, - newId: nextId(), - now: nextNow(600), - generateSessionTitle: async () => { - calls += 1; - throw new Error('offline'); - }, - onSessionTitleChanged: () => changed.release(), - }); - const session = await manager.createSession(makeInput({ name: 'New Chat' })); - - await drain( - manager.sendMessage(session.id, { turnId: 'first', text: '\nFallback title\nignored' }), - ); - await changed.promise; - await drain(manager.sendMessage(session.id, { turnId: 'second', text: 'second prompt' })); - - expect((await store.readHeader(session.id)).name).toBe('Fallback title'); - expect(calls).toBe(1); - }); - - test('does not overwrite or notify after a racing manual rename', async () => { - const store = new MemorySessionStore(); - const backends = new BackendRegistry(); - backends.register('ai-sdk', (ctx) => new TestBackend(ctx)); - const release = makeGate(); - const attempted = makeGate(); - store.generatedTitleAttempted = attempted; - let notifications = 0; - const manager = new SessionManager({ - store, - backends, - newId: nextId(), - now: nextNow(700), - generateSessionTitle: async () => { - await release.promise; - return 'Generated loses'; - }, - onSessionTitleChanged: () => { - notifications += 1; - }, - }); - const session = await manager.createSession(makeInput({ name: 'New Chat' })); - - await drain(manager.sendMessage(session.id, { turnId: 'turn-race', text: 'hello' })); - await manager.renameSession(session.id, 'Manual wins'); - release.release(); - await attempted.promise; - - expect((await store.readHeader(session.id)).name).toBe('Manual wins'); - expect(notifications).toBe(0); - }); -}); - describe('SessionManager manual compaction and quiescent session changes', () => { test('runs backend history compaction as a runtime turn and persists diagnostics', async () => { const store = new MemorySessionStore(); @@ -16499,7 +16436,6 @@ class MemorySessionStore implements SessionStore { disposeCount = 0; nextReadHeaderGate: { started: Gate; release: Gate } | undefined; nextGraphOperatorProvisionGate: { started: Gate; release: Gate } | undefined; - generatedTitleAttempted: Gate | undefined; async createSubagent( input: CreateSessionInput, @@ -16593,7 +16529,7 @@ class MemorySessionStore implements SessionStore { hasUnread: false, backend: 'ai-sdk', llmConnectionSlug: input.llmConnectionSlug, - connectionLocked: false, + connectionLocked: input.subagentParent !== undefined, model: input.model ?? 'fake-model', ...(input.thinkingLevel !== undefined ? { thinkingLevel: input.thinkingLevel } : {}), permissionMode: input.permissionMode, @@ -16762,13 +16698,6 @@ class MemorySessionStore implements SessionStore { await this.updateHeader(sessionId, { name, titleIsManual: true }); } - async setGeneratedTitleIfAbsent(sessionId: string, title: string): Promise { - const current = await this.readHeader(sessionId); - this.generatedTitleAttempted?.release(); - if (current.titleIsManual || current.name !== 'New Chat') return null; - return this.updateHeader(sessionId, { name: title }); - } - async remove(sessionId: string): Promise { this.headers.delete(sessionId); this.messages.delete(sessionId); diff --git a/packages/runtime/src/__tests__/session-title.test.ts b/packages/runtime/src/__tests__/session-title.test.ts deleted file mode 100644 index 3eaf5b9fc0..0000000000 --- a/packages/runtime/src/__tests__/session-title.test.ts +++ /dev/null @@ -1,143 +0,0 @@ -/* - * 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, test } from 'node:test'; -import { - fallbackSessionTitle, - generateSessionTitle, - sessionTitleSource, -} from '../session-title.js'; - -describe('session title helper', () => { - test('uses display text, strips system reminders, and truncates input on a UTF-8 boundary', () => { - const source = sessionTitleSource({ - text: 'model envelope', - displayText: `secret\n${'🦊'.repeat(3_000)}`, - }); - - assert.equal(source.includes('secret'), false); - assert.equal(new TextEncoder().encode(source).length <= 8 * 1024, true); - assert.equal(source.endsWith('�'), false); - }); - - test('extracts the user message from a raw skill envelope', () => { - assert.equal( - sessionTitleSource({ - text: 'Skills loaded below.\n\nSECRET INSTRUCTIONS\n\n\nAnalyze this code\n', - }), - 'Analyze this code', - ); - }); - - test('builds fallback from the first non-empty line without splitting Unicode code points', () => { - const line = `${'🦊'.repeat(42)}tail`; - assert.equal(fallbackSessionTitle(`\n \n${line}\nignored`), '🦊'.repeat(42)); - assert.equal(fallbackSessionTitle(' \n\t'), undefined); - }); - - test('cleans model reasoning, prefixes, quotes, and extra lines', async () => { - let request: Record | undefined; - const title = await generateSessionTitle({ - model: {} as never, - sourceText: 'Analyze the production logs', - providerOptions: { provider: { required: true } }, - generateText: async (options) => { - request = options; - return { - text: 'reasoning\nTitle: "Production log analysis"\nextra', - finishReason: 'stop', - }; - }, - }); - - assert.equal(title, 'Production log analysis'); - assert.equal(request?.maxOutputTokens, 1024); - assert.deepEqual(request?.providerOptions, { provider: { required: true } }); - assert.equal('tools' in (request ?? {}), false); - }); - - test('returns undefined for empty, truncated, invalid, or failed model output', async () => { - const model = {} as never; - assert.equal( - await generateSessionTitle({ - model, - sourceText: '', - generateText: async () => ({ text: 'unused', finishReason: 'stop' }), - }), - undefined, - ); - assert.equal( - await generateSessionTitle({ - model, - sourceText: 'hello', - generateText: async () => ({ text: 'Title', finishReason: 'length' }), - }), - undefined, - ); - assert.equal( - await generateSessionTitle({ - model, - sourceText: 'hello', - generateText: async () => ({ text: 'x', finishReason: 'stop' }), - }), - undefined, - ); - assert.equal( - await generateSessionTitle({ - model, - sourceText: 'hello', - generateText: async () => { - throw new Error('offline'); - }, - }), - undefined, - ); - }); - - test('aborts title generation when the provider exceeds its deadline', async () => { - let signal: AbortSignal | undefined; - let watchdog: ReturnType | undefined; - try { - const result = await Promise.race([ - generateSessionTitle({ - model: {} as never, - sourceText: 'hello', - timeoutMs: 10, - generateText: (options: Record) => { - signal = options.abortSignal as AbortSignal; - // Never settles: verifies the internal deadline, not provider cooperation. - return new Promise(() => {}); - }, - }), - new Promise((_resolve, reject) => { - watchdog = setTimeout( - () => reject(new Error('title generation did not respect its deadline')), - 250, - ); - }), - ]); - - assert.equal(result, undefined); - assert.equal(signal?.aborted, true); - } finally { - if (watchdog) clearTimeout(watchdog); - } - }); -}); diff --git a/packages/runtime/src/agent-run.ts b/packages/runtime/src/agent-run.ts index ee3cf91e6c..b4b42299c6 100644 --- a/packages/runtime/src/agent-run.ts +++ b/packages/runtime/src/agent-run.ts @@ -226,7 +226,7 @@ export class AgentRun { readonly toolMode: ToolMode; private readonly input: AgentRunInput; - private header: SessionHeader; + private readonly header: SessionHeader; private active: AgentRunActiveSession | undefined; private stopped = false; private abortSource: string | undefined; @@ -686,10 +686,6 @@ export class AgentRun { requireDurableWrite: this.requiresDurablePersistence(), }); - if (!this.header.connectionLocked) { - this.header = await this.input.hooks.updateHeader(this.sessionId, { connectionLocked: true }); - } - this.active = await this.input.hooks.reserveRun(this.sessionId, this.header, this); await this.markRunStarted(this.lastTs); @@ -734,10 +730,6 @@ export class AgentRun { }); } - if (!this.header.connectionLocked) { - this.header = await this.input.hooks.updateHeader(this.sessionId, { connectionLocked: true }); - } - this.active = await this.input.hooks.reserveRun(this.sessionId, this.header, this); await this.markRunStarted(startedAt); @@ -783,10 +775,6 @@ export class AgentRun { }); } - if (!this.header.connectionLocked) { - this.header = await this.input.hooks.updateHeader(this.sessionId, { connectionLocked: true }); - } - this.active = await this.input.hooks.reserveRun(this.sessionId, this.header, this); await this.markRunStarted(startedAt); await this.input.hooks.updateStatus(this.sessionId, 'running', undefined, startedAt); diff --git a/packages/runtime/src/runtime-kernel.ts b/packages/runtime/src/runtime-kernel.ts index 64a94e63f2..ede6ec1f0a 100644 --- a/packages/runtime/src/runtime-kernel.ts +++ b/packages/runtime/src/runtime-kernel.ts @@ -1144,7 +1144,6 @@ export class RuntimeKernel implements RuntimeKernelLike { const childHeader: SessionHeader = { ...parentHeader, permissionMode: definition.permissionMode, - connectionLocked: true, }; const userInput: UserMessageInput = { turnId: input.turnId, @@ -1264,7 +1263,6 @@ export class RuntimeKernel implements RuntimeKernelLike { : { ...parentHeader, permissionMode: definition.permissionMode, - connectionLocked: true, }; const userInput: UserMessageInput = { turnId: continuation.turnId, diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts index a2cb3dec0e..30289e8d3a 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -207,7 +207,6 @@ import { type ResumeContinuationOptions, type TurnStartOptions, } from './runtime-kernel.js'; -import { fallbackSessionTitle, sessionTitleSource } from './session-title.js'; import type { HistoryCompactCleanupRequest } from './history-compact-checkpoint-coordinator.js'; import { fingerprintAgentGraphRunnableIntent } from './stream-graph-admission.js'; import type { AgentGraphRunnableIntent } from './stream-graph-readiness.js'; @@ -687,7 +686,6 @@ export interface SessionStore { ): Promise; setFlagged(sessionId: string, isFlagged: boolean): Promise; rename(sessionId: string, name: string): Promise; - setGeneratedTitleIfAbsent?(sessionId: string, title: string): Promise; remove(sessionId: string): Promise; } @@ -854,12 +852,6 @@ interface SessionManagerBaseDeps { /** Trusted Host-owned graph readers. Hosted graph execution fails closed without them. */ hostedAgentGraphExecution?: RuntimeHostedAgentGraphExecutionCapability; onContinuationLifecycleEvent?: (event: RuntimeContinuationLifecycleEvent) => void | Promise; - generateSessionTitle?: (input: { - sessionId: string; - header: SessionHeader; - sourceText: string; - }) => Promise; - onSessionTitleChanged?: (sessionId: string) => void; } export interface ResolvedChildToolActivation { @@ -2063,40 +2055,7 @@ export class SessionManager { return (await options.admitTurn?.()) ?? 'admitted'; } : options.admitTurn; - const sourceText = sessionTitleSource(input); - const onRunStarted = this.deps.generateSessionTitle - ? async (runId: string, header: SessionHeader) => { - await options.onRunStarted?.(runId, header); - if ( - !header.connectionLocked && - !header.titleIsManual && - header.name === DEFAULT_SESSION_NAME && - sourceText - ) { - void this.generateTitleInBackground(sessionId, header, sourceText); - } - } - : options.onRunStarted; - yield* this.runtimeKernel.startTurn(sessionId, input, { ...options, admitTurn, onRunStarted }); - } - - private async generateTitleInBackground( - sessionId: string, - header: SessionHeader, - sourceText: string, - ): Promise { - let generated: string | undefined; - try { - generated = await this.deps.generateSessionTitle?.({ sessionId, header, sourceText }); - } catch {} - try { - const title = generated ?? fallbackSessionTitle(sourceText); - if (!title) return; - const next = await this.deps.store.setGeneratedTitleIfAbsent?.(sessionId, title); - if (!next) return; - this.runtimeKernel.updateCachedHeader(sessionId, next); - this.deps.onSessionTitleChanged?.(sessionId); - } catch {} + yield* this.runtimeKernel.startTurn(sessionId, input, { ...options, admitTurn }); } async planSafeBoundaryContinuation( diff --git a/packages/storage/src/__tests__/session-store.test.ts b/packages/storage/src/__tests__/session-store.test.ts index cb2be2d93a..0a2cd3b2e9 100644 --- a/packages/storage/src/__tests__/session-store.test.ts +++ b/packages/storage/src/__tests__/session-store.test.ts @@ -25,6 +25,7 @@ import { join } from 'node:path'; import { DatabaseSync } from 'node:sqlite'; import { describe, test } from 'node:test'; import type { CreateSessionInput } from '@maka/core/runtime-inputs'; +import { DEFAULT_SESSION_NAME } from '@maka/core/session-name'; import { WORKHUB_COORDINATION_SESSION_ID, WORKHUB_COORDINATION_SESSION_ROLE, @@ -467,6 +468,123 @@ describe('SQLite SessionStore', () => { } }); + test('a generated title fills an absence and never overwrites a rename', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-session-generated-title-')); + const store = createSessionStore(root); + try { + const unnamed = await store.create(makeInput({ cwd: root, name: DEFAULT_SESSION_NAME })); + assert.equal( + (await store.setGeneratedTitleIfAbsent(unnamed.id, 'draft the release notes'))?.name, + 'draft the release notes', + ); + assert.equal((await store.readHeaderSnapshot(unnamed.id)).name, 'draft the release notes'); + // An already-named Session is never renamed by a later generation. + assert.equal(await store.setGeneratedTitleIfAbsent(unnamed.id, 'a second guess'), null); + + // A rename landing between the check and the write wins: the write is + // conditional on the revision the check read. + const raced = await store.create(makeInput({ cwd: root, name: DEFAULT_SESSION_NAME })); + const readHeaderRecordSnapshot = store.readHeaderRecordSnapshot.bind(store); + let renamed = false; + store.readHeaderRecordSnapshot = async (sessionId: string) => { + const record = await readHeaderRecordSnapshot(sessionId); + if (sessionId === raced.id && !renamed) { + renamed = true; + await store.rename(sessionId, '我自己起的名字'); + } + return record; + }; + assert.equal(await store.setGeneratedTitleIfAbsent(raced.id, 'generated loses'), null); + const header = await readHeaderRecordSnapshot(raced.id); + assert.equal(header.header.name, '我自己起的名字'); + assert.equal(header.header.titleIsManual, true); + + // A revision that moved for any other reason is re-read, not mistaken + // for a rename. + const flagged = await store.create(makeInput({ cwd: root, name: DEFAULT_SESSION_NAME })); + let flaggedOnce = false; + store.readHeaderRecordSnapshot = async (sessionId: string) => { + const record = await readHeaderRecordSnapshot(sessionId); + if (sessionId === flagged.id && !flaggedOnce) { + flaggedOnce = true; + await store.setFlagged(sessionId, true); + } + return record; + }; + assert.equal( + (await store.setGeneratedTitleIfAbsent(flagged.id, 'generated survives'))?.name, + 'generated survives', + ); + + // A Session whose revision moves under every attempt answers null like any + // other lost race, so a caller reading null never has to also expect a throw. + const busy = await store.create(makeInput({ cwd: root, name: DEFAULT_SESSION_NAME })); + let flips = 0; + store.readHeaderRecordSnapshot = async (sessionId: string) => { + const record = await readHeaderRecordSnapshot(sessionId); + if (sessionId === busy.id) { + flips += 1; + await store.setFlagged(sessionId, flips % 2 === 1); + } + return record; + }; + assert.equal(await store.setGeneratedTitleIfAbsent(busy.id, 'never lands'), null); + assert.equal((await readHeaderRecordSnapshot(busy.id)).header.name, DEFAULT_SESSION_NAME); + } finally { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); + + test('a Session freezes its route on the first user message, a subagent at birth', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-session-route-freeze-')); + const store = createSessionStore(root); + try { + const ordinary = await store.create(makeInput({ cwd: root })); + assert.equal(ordinary.connectionLocked, false); + + // A subagent's route is chosen by the spawn that created it and is never + // re-targeted, so it needs no first Message to be frozen. + const child = await store.createSubagent( + makeInput({ + cwd: root, + name: 'Child', + subagentParent: { + kind: 'subagent', + parentSessionId: ordinary.id, + spawnedBy: { + parentRunId: 'parent-run', + parentTurnId: 'parent-turn', + toolCallId: 'tool-call', + }, + lifecycle: 'foreground', + }, + subagentRuntime: { + schemaVersion: 1, + definitionVersion: 1, + agentId: 'local-read', + agentName: 'Local Read', + profile: 'local_read', + systemPrompt: 'Read the assigned workspace task.', + toolNames: ['Read'], + categoryPolicy: { read: 'allow' }, + }, + subagentSpawn: { + schemaVersion: 1, + requestFingerprint: 'a'.repeat(64), + initialTurnId: 'child-turn', + initialRunId: 'child-run', + }, + }), + ); + assert.equal(child.header.connectionLocked, true); + assert.equal((await store.readHeaderSnapshot(child.header.id)).connectionLocked, true); + } finally { + await store.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); + test('appending the first user message locks the session before any read', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-session-lock-heal-')); const store = createSessionStore(root); diff --git a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts index 4411168568..acbfce376a 100644 --- a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts @@ -121,6 +121,60 @@ describe('SqliteSessionMetadataStore', () => { }); } + test('migrates a legacy subagent Session to a frozen model route', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-session-metadata-v32-')); + const path = join(root, 'state.sqlite'); + const child = fullHeader({ + id: 'legacy-child', + parentSessionId: undefined, + branchOfTurnId: undefined, + revisionRootSessionId: undefined, + revisionParentSessionId: undefined, + revisionOfTurnId: undefined, + revisionIndex: undefined, + revisionState: undefined, + connectionLocked: false, + subagentParent: { + kind: 'subagent', + parentSessionId: 'parent-session', + spawnedBy: { + parentRunId: 'parent-run', + parentTurnId: 'parent-turn', + toolCallId: 'tool-call', + }, + lifecycle: 'foreground', + }, + }); + const ordinary = fullHeader({ id: 'legacy-ordinary', connectionLocked: false }); + const setup = createSqliteSessionMetadataStore(path); + try { + await setup.create(child); + await setup.create(ordinary); + } finally { + setup.close(); + } + // A subagent spawned before the route froze at creation, and abandoned + // before its first Message, is the one shape nothing else can lock. + const legacy = new DatabaseSync(path); + try { + legacy.exec(` + UPDATE session_metadata_schema SET version = 32 WHERE scope = 'session_metadata'; + `); + } finally { + legacy.close(); + } + + const migrated = createSqliteSessionMetadataStore(path); + try { + assert.equal(migrated.schemaVersion(), SQLITE_SESSION_METADATA_SCHEMA_VERSION); + assert.equal((await migrated.read('legacy-child')).header.connectionLocked, true); + assert.equal((await migrated.read('legacy-ordinary')).header.connectionLocked, false); + } finally { + migrated.close(); + } + await rm(root, { recursive: true, force: true }); + }); + test('migrates v27 metadata to the current schema without backfilling external origin', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-session-metadata-v27-')); const path = join(root, 'state.sqlite'); diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index 10d045a4d5..e77d5dc093 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -1045,15 +1045,31 @@ class SqliteSessionStore implements SessionAuthorityStore { async setGeneratedTitleIfAbsent(sessionId: string, title: string): Promise { const normalized = normalizeUserSessionName(title); if (!normalized.ok) return null; - const current = await this.readHeaderSnapshot(sessionId); - if ( - current.titleIsManual || - current.name !== DEFAULT_SESSION_NAME || - normalized.value === current.name - ) { - return null; + // A generated title only ever fills an absence. Writing at the revision the + // check read makes a rename that lands between the two a winner rather than + // something this silently overwrites; a revision that moved for any other + // reason is re-read, so losing the race stays the only way to answer null. + for (let attempt = 0; attempt < 3; attempt += 1) { + const record = await this.readHeaderRecordSnapshot(sessionId); + const current = record.header; + if ( + current.titleIsManual || + current.name !== DEFAULT_SESSION_NAME || + normalized.value === current.name + ) { + return null; + } + try { + return ( + await this.updateHeaderVersioned(sessionId, { name: normalized.value }, record.revision) + ).header; + } catch (error) { + if (!(error instanceof SessionMetadataVersionConflictError)) throw error; + } } - return this.updateHeader(sessionId, { name: normalized.value }); + // Losing the race every attempt reads the same as losing it once: the + // Session keeps whichever name the writer that won gave it. + return null; } async remove(sessionId: string): Promise { @@ -1139,7 +1155,10 @@ function buildSessionHeader( hasUnread: false, backend: 'ai-sdk', llmConnectionSlug: input.llmConnectionSlug, - connectionLocked: false, + // A subagent Session's route is chosen by the spawn that created it and is + // never re-targeted, so it is born frozen. Every other Session freezes on + // its first user Message. + connectionLocked: input.subagentParent !== undefined, model: input.model ?? 'default', ...(input.toolProfile !== undefined ? { toolProfile: input.toolProfile } : {}), permissionMode: input.permissionMode, diff --git a/packages/storage/src/sqlite-session-metadata-schema.ts b/packages/storage/src/sqlite-session-metadata-schema.ts index 66b4c04137..9a39e05a65 100644 --- a/packages/storage/src/sqlite-session-metadata-schema.ts +++ b/packages/storage/src/sqlite-session-metadata-schema.ts @@ -19,7 +19,7 @@ import type { DatabaseSync } from 'node:sqlite'; -export const SQLITE_SESSION_METADATA_SCHEMA_VERSION = 32; +export const SQLITE_SESSION_METADATA_SCHEMA_VERSION = 33; export const SQLITE_SESSION_MESSAGE_CHUNK_BYTES = 64 * 1024; export const SQLITE_SESSION_MESSAGE_CHUNK_MARKER = '{"$maka":"session-message-chunks-v1"}'; @@ -1200,6 +1200,22 @@ const MIGRATIONS: ReadonlyMap = new Map([ ALTER TABLE message_admissions ADD COLUMN submitted_intent_json TEXT; `, ], + [ + 33, + ` + UPDATE session_metadata + SET + payload_json = json_set(payload_json, '$.connectionLocked', json('true')), + metadata_version = metadata_version + 1, + committed_at = MAX( + committed_at, + CAST(strftime('%s', 'now') AS INTEGER) * 1000 + ) + WHERE + json_extract(payload_json, '$.connectionLocked') = 0 + AND json_extract(payload_json, '$.subagentParent') IS NOT NULL; + `, + ], ]); if (MIGRATIONS.size !== SQLITE_SESSION_METADATA_SCHEMA_VERSION) {