diff --git a/apps/desktop/src/main/__tests__/session-trace-refresh.test.ts b/apps/desktop/src/main/__tests__/session-trace-refresh.test.ts index ac9364d44e0..19e5fcd1dc4 100644 --- a/apps/desktop/src/main/__tests__/session-trace-refresh.test.ts +++ b/apps/desktop/src/main/__tests__/session-trace-refresh.test.ts @@ -67,7 +67,9 @@ describe('session trace refresh policy', () => { 'token_usage', 'provider_retry', // A mid-turn compaction persists a boundary note the trace must show. + 'context_compaction_started', 'context_compaction_applied', + 'context_compaction_failed', 'complete', 'error', 'abort', diff --git a/apps/desktop/src/renderer/application/contracts/session-inspector/session-trace-refresh.ts b/apps/desktop/src/renderer/application/contracts/session-inspector/session-trace-refresh.ts index e193b7fcb5c..adcf24bc7b2 100644 --- a/apps/desktop/src/renderer/application/contracts/session-inspector/session-trace-refresh.ts +++ b/apps/desktop/src/renderer/application/contracts/session-inspector/session-trace-refresh.ts @@ -37,7 +37,9 @@ const TRACE_RELEVANT_EVENT_TYPES: ReadonlySet = new Set([ 'token_usage', 'provider_retry', // Its whole ledger effect is a persisted boundary note; easy to miss that it counts. + 'context_compaction_started', 'context_compaction_applied', + 'context_compaction_failed', 'error', 'complete', 'abort', diff --git a/packages/cli/src/__tests__/pi-transcript.test.ts b/packages/cli/src/__tests__/pi-transcript.test.ts index 69c9b127129..ff4ae94da4b 100644 --- a/packages/cli/src/__tests__/pi-transcript.test.ts +++ b/packages/cli/src/__tests__/pi-transcript.test.ts @@ -71,6 +71,41 @@ describe('Maka Pi TUI transcript', () => { assert.doesNotMatch(rendered, /context_provider_dropping|truncated or rewritten/); }); + test('shows an interrupted compaction separately from a failed summary', () => { + const state = createMakaPiTranscriptState(); + replaceTranscriptWithStoredMessages(state, [ + { + type: 'system_note', + id: 'interrupted', + turnId: 't1', + ts: 1, + kind: 'context_compaction_failed_open', + data: { failOpenReason: 'interrupted' }, + }, + { + type: 'system_note', + id: 'failed', + turnId: 't1', + ts: 2, + kind: 'context_compaction_failed_open', + data: { failOpenReason: 'write_failed' }, + }, + ]); + + assert.deepEqual( + state.entries + .filter((entry) => entry.kind === 'notice') + .map((entry) => ({ level: entry.level, text: entry.text })), + [ + { level: 'info', text: 'Context compaction was interrupted.' }, + { + level: 'info', + text: 'Context summary failed; the session continued without a new summary.', + }, + ], + ); + }); + test('renders manual compaction from the typed terminal outcome', async () => { for (const [outcome, expected] of [ [{ kind: 'compacted' as const, checkpointId: 'checkpoint-1' }, 'Context compacted.'], diff --git a/packages/cli/src/pi-transcript.ts b/packages/cli/src/pi-transcript.ts index 4b45d0112ec..d9eb7e1c503 100644 --- a/packages/cli/src/pi-transcript.ts +++ b/packages/cli/src/pi-transcript.ts @@ -39,6 +39,10 @@ import { } from '@maka/core/session'; import type { ContextBudgetDiagnostic } from '@maka/core/usage-stats/types'; import { providerRetryDisplaySeconds } from '@maka/core/provider-retry-countdown'; +import { + compactionStateForFailure, + readCompactionFailureReason, +} from '@maka/core/compaction-status'; import type { ThinkingLevel } from '@maka/core/model-thinking'; import { defineUiMessageCatalog, @@ -1361,7 +1365,9 @@ function systemNoteText(message: SystemNoteMessage): string | undefined { case 'context_compacted': return 'Context compacted to keep this task within the model window.'; case 'context_compaction_failed_open': - return 'Context summary failed; the session continued without a new summary.'; + return compactionStateForFailure(readCompactionFailureReason(message.data)) === 'interrupted' + ? 'Context compaction was interrupted.' + : 'Context summary failed; the session continued without a new summary.'; case 'context_overflow_after_compaction': return 'History was compacted and the provider still called this request too large. What remains also carries the system prompt, the tool schemas, the summary and the recent tail; shortening this message is the part you control.'; case 'context_reported_window_exceeded': { diff --git a/packages/core/package.json b/packages/core/package.json index 83b3ba5a835..8b99f90afcf 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -12,6 +12,7 @@ "./durable-tool-result-projection": "./dist/durable-tool-result-projection.js", "./model-projection-transition": "./dist/model-projection-transition.js", "./canonical-runtime-event": "./dist/canonical-runtime-event.js", + "./compaction-status": "./dist/compaction-status.js", "./canonical-json": "./dist/canonical-json.js", "./runtime-boundary": "./dist/runtime-boundary.js", "./runtime-handoff": "./dist/runtime-handoff.js", diff --git a/packages/core/src/backend-types.ts b/packages/core/src/backend-types.ts index 1c757cb530e..6b96e38764f 100644 --- a/packages/core/src/backend-types.ts +++ b/packages/core/src/backend-types.ts @@ -234,7 +234,6 @@ export type BackendSessionEvent = Exclude< | 'message_admission' | 'client_capability_request' | 'client_capability_decision_ack' - | 'context_compaction_started' | 'permission_request' | 'permission_answer_ack' | 'permission_closure_ack' diff --git a/packages/core/src/compaction-status.ts b/packages/core/src/compaction-status.ts new file mode 100644 index 00000000000..fe6c59742ad --- /dev/null +++ b/packages/core/src/compaction-status.ts @@ -0,0 +1,109 @@ +/* + * 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 type { StoredMessage, TurnStatus } from './session.js'; + +export type CompactionTimelineState = 'running' | 'compacted' | 'failed' | 'interrupted'; + +export interface CompactionTimelineMarker { + compactionId: string; + state: CompactionTimelineState; + ts: number; + /** Present only when `state === 'failed'`; cancellation uses `interrupted`. */ + reason?: string; +} + +/** A cancelled attempt is a distinct outcome from a compaction that failed. */ +export function compactionStateForFailure(reason: string | undefined): 'failed' | 'interrupted' { + // `aborted` is how the explicit compaction path names a stopped run; both + // spellings mean the reader cancelled, not that folding failed. + return reason === 'interrupted' || reason === 'aborted' ? 'interrupted' : 'failed'; +} + +/** A failed Turn ends a running attempt as failed; any other terminal Turn cuts it off. */ +export function compactionStateForTurnStatus( + status: Exclude, +): 'failed' | 'interrupted' { + return status === 'failed' ? 'failed' : 'interrupted'; +} + +function recordValue(value: unknown): Record | undefined { + return typeof value === 'object' && value !== null && !Array.isArray(value) + ? (value as Record) + : undefined; +} + +export function readCompactionFailureReason(value: unknown): string | undefined { + const data = recordValue(value); + return typeof data?.failOpenReason === 'string' + ? data.failOpenReason + : typeof data?.reason === 'string' + ? data.reason + : undefined; +} + +/** + * Reads one durable marker of an in-Turn compaction lifecycle. + * + * A timeline row is an attempt, so it requires an explicit `compactionId`. + * Rows written before the lifecycle existed have no attempt identity and stay + * in the transcript as their original notes (`context_compacted` or a visible + * failure note); they never become timeline attempts. Do not fall back to + * `turnId` or turn order: one Turn may contain several compaction attempts. + */ +export function readCompactionTimelineMarker( + message: StoredMessage, +): CompactionTimelineMarker | undefined { + if (message.type !== 'system_note') return undefined; + if ( + message.kind !== 'context_compaction_started' && + message.kind !== 'context_compaction_applied' && + message.kind !== 'context_compaction_failed_open' + ) + return undefined; + const data = recordValue(message.data); + const scope = data?.scope === 'step' ? 'step' : data?.scope === 'turn' ? 'turn' : undefined; + const compactionId = + typeof data?.compactionId === 'string' && data.compactionId.length > 0 + ? data.compactionId + : undefined; + // A marker without an attempt identity cannot be paired with its terminal + // event and must not become a timeline row. + if (compactionId === undefined) return undefined; + const reason = readCompactionFailureReason(message.data); + if (message.kind === 'context_compaction_started') { + if (scope !== 'step') return undefined; + return { compactionId, state: 'running', ts: message.ts }; + } + if (message.kind === 'context_compaction_applied') { + if (scope !== 'step') return undefined; + return { compactionId, state: 'compacted', ts: message.ts }; + } + // Legacy failed_open rows were written at settlement; do not move them into + // the timeline with a misleading position. Only the new step-scoped marker + // carries an attempt identity and an apply-time position. + if (scope !== 'step') return undefined; + const state = compactionStateForFailure(reason); + return { + compactionId, + state, + ts: message.ts, + ...(state === 'failed' && reason !== undefined ? { reason } : {}), + }; +} diff --git a/packages/core/src/events.ts b/packages/core/src/events.ts index eec5e787621..6db85c37724 100644 --- a/packages/core/src/events.ts +++ b/packages/core/src/events.ts @@ -687,7 +687,8 @@ export type SessionEvent = | CompleteEvent | AbortEvent | ContextCompactionAppliedEvent - | ContextCompactionStartedEvent; + | ContextCompactionStartedEvent + | ContextCompactionFailedEvent; export interface TextDeltaEvent extends BaseEvent { type: 'text_delta'; @@ -1437,13 +1438,21 @@ export interface AbortEvent extends BaseEvent { } /** - * A host-owned explicit context-compaction Turn has started. Synthesized by the - * Runtime Host session projector (not the kernel) purely so a client can render - * a "compacting" transcript row while the Turn is in flight; it carries no - * durable state and is excluded from `BackendSessionEvent` like `queue_update`. + * A context-compaction execution started. The Host synthesizes the `turn` + * scope for an explicit compaction Turn; the Runtime emits the `step` scope + * before an automatic in-Turn fold so the client can render the attempt in the + * execution timeline. `compactionId` pairs the start with its terminal event. */ export interface ContextCompactionStartedEvent extends BaseEvent { type: 'context_compaction_started'; + /** + * Required for a `step` attempt. A turn-scoped event (including a legacy / + * manually synthesized one without this field) describes the whole Turn and + * never becomes an attempt timeline row. + */ + compactionId?: string; + /** `turn` = the whole Turn is an explicit compaction; `step` = in-Turn fold. */ + scope?: 'turn' | 'step'; } /** @@ -1453,6 +1462,27 @@ export interface ContextCompactionStartedEvent extends BaseEvent { */ export interface ContextCompactionAppliedEvent extends BaseEvent { type: 'context_compaction_applied'; + /** + * Required for a new `step` attempt. Legacy applied records have no attempt + * identity and no start to pair with; they stay hidden or render as the + * existing settlement note, never as a timeline attempt. Never pair by + * `turnId` or turn order: one Turn may contain several attempts. + */ + compactionId?: string; + /** `step` for an automatic in-Turn fold; `turn` for explicit compaction. */ + scope?: 'turn' | 'step'; +} + +/** + * An automatic in-Turn compaction attempt failed open. The Turn continues with + * the un-compacted projection; this event only settles the timeline row. + */ +export interface ContextCompactionFailedEvent extends BaseEvent { + type: 'context_compaction_failed'; + compactionId: string; + scope: 'step'; + /** `interrupted` means the attempt was cancelled, not that folding failed. */ + reason: string; } // ============================================================================ diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index cec7620a20e..7427ec6e5c7 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -856,10 +856,14 @@ export function userFacingText(message: Pick { id: 'compaction-1', turnId: 'turn-1', ts: 123456789, + compactionId: 'compaction-1', + scope: 'step' as const, }, }; const encoded = encodeProtocolMessage(frame); @@ -651,6 +653,48 @@ describe('Runtime Host bootstrap protocol', () => { } }); + test('roundtrips in-turn compaction lifecycle frames with closed event shapes', () => { + for (const event of [ + { + type: 'context_compaction_started' as const, + id: 'compaction-started-1', + turnId: 'turn-1', + ts: 123456789, + compactionId: 'compact-1', + scope: 'step' as const, + }, + { + type: 'context_compaction_failed' as const, + id: 'compaction-failed-1', + turnId: 'turn-1', + ts: 123456790, + compactionId: 'compact-1', + scope: 'step' as const, + reason: 'write_failed', + }, + ]) { + const frame = { + kind: 'subscription.session_event' as const, + hostEpoch: 'epoch-1', + subscriptionId: 'subscription-1', + sequence: 2, + sessionId: 'session-1', + runId: 'run-1', + event, + }; + const encoded = encodeProtocolMessage(frame); + assert.deepEqual(decodeHostFrame(JSON.parse(encoded.toString('utf8'))), frame); + assert.throws( + () => decodeHostFrame({ ...frame, event: { ...event, scope: 'turn' } }), + isInvalidFrame, + ); + assert.throws( + () => decodeHostFrame({ ...frame, event: { ...event, compactionId: '' } }), + isInvalidFrame, + ); + } + }); + test('preserves opaque nested call and step identities in subscription tool events', () => { const parentId = 'p'.repeat(128); const frame = { diff --git a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts index 91e48fec3da..6fa78ee0bb8 100644 --- a/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts @@ -987,6 +987,8 @@ test('forwards compaction boundaries while the root turn is still running', asyn id: `compaction-${index}`, turnId: 'compaction-turn', ts: 123456789 + index, + compactionId: `compaction-${index}`, + scope: 'step' as const, })); const fixture = await createFailureFixture({ registerBackend: (backends) => diff --git a/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts b/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts index fdc53f2fe8a..416f8268da4 100644 --- a/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts +++ b/packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts @@ -180,6 +180,88 @@ test('forwards the durable steering echo to subscribers as a session event', asy coordinator.close(); }); +test('forwards in-turn compaction lifecycle events with their compaction identity', async () => { + const sink = new RecordingSink(); + const coordinator = new SessionContinuityCoordinator( + HOST_EPOCH, + async () => canonical(), + new SessionAdmissionGate(), + ); + const connection = attachTestConnection(coordinator, 'connection-1', sink); + const opened = await open(coordinator, 'connection-1'); + connection.activate(opened.subscriptionId); + await delayImmediate(); + sink.frames.length = 0; + + const started = { + type: 'context_compaction_started' as const, + id: 'compaction-started-1', + turnId: 'turn-1', + ts: 7, + compactionId: 'compact-1', + scope: 'step' as const, + }; + await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', started); + assert.equal(sink.frames.length, 1); + const startedFrame = sink.frames[0]; + assert.equal(startedFrame?.kind, 'subscription.session_event'); + if (startedFrame?.kind !== 'subscription.session_event') return; + assert.deepEqual(startedFrame.event, started); + + const failed = { + type: 'context_compaction_failed' as const, + id: 'compaction-failed-1', + turnId: 'turn-1', + ts: 8, + compactionId: 'compact-1', + scope: 'step' as const, + reason: 'write_failed', + }; + await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', failed); + assert.equal(sink.frames.length, 2); + const failedFrame = sink.frames[1]; + assert.equal(failedFrame?.kind, 'subscription.session_event'); + if (failedFrame?.kind !== 'subscription.session_event') return; + assert.deepEqual(failedFrame.event, failed); + + connection.abort(opened.subscriptionId); + coordinator.close(); +}); + +test('does not forward a legacy compaction event without an attempt identity', async () => { + const sink = new RecordingSink(); + const coordinator = new SessionContinuityCoordinator( + HOST_EPOCH, + async () => canonical(), + new SessionAdmissionGate(), + ); + const connection = attachTestConnection(coordinator, 'connection-1', sink); + const opened = await open(coordinator, 'connection-1'); + connection.activate(opened.subscriptionId); + await delayImmediate(); + sink.frames.length = 0; + + await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', { + type: 'context_compaction_applied', + id: 'legacy-applied', + turnId: 'turn-1', + ts: 9, + }); + assert.equal(sink.frames.length, 0); + + await coordinator.acceptRuntimeEvent(SESSION_ID, 'run-1', { + type: 'context_compaction_started', + id: 'legacy-started', + turnId: 'turn-1', + ts: 10, + scope: 'step', + }); + assert.equal(sink.frames.length, 0); + + connection.abort(opened.subscriptionId); + coordinator.close(); +}); + test('projects model-only user content out of Guest queue and steering frames', async () => { const grant = { kind: 'session_observation' as const, diff --git a/packages/runtime-host/src/__tests__/session-projector.test.ts b/packages/runtime-host/src/__tests__/session-projector.test.ts index 2a386b86096..815e5c71801 100644 --- a/packages/runtime-host/src/__tests__/session-projector.test.ts +++ b/packages/runtime-host/src/__tests__/session-projector.test.ts @@ -620,6 +620,8 @@ test('projects each live compaction boundary immediately with its original ident id: `compaction-${sequence}`, turnId: 'turn-1', ts: 123456789 + sequence, + compactionId: `compaction-${sequence}`, + scope: 'step' as const, }; const update = projector.accept({ kind: 'subscription.session_event', @@ -634,6 +636,41 @@ test('projects each live compaction boundary immediately with its original ident } }); +test('projects step-scoped compaction lifecycle frames without turning the Turn into a compaction run', () => { + const projector = projectorAt(); + const events = [ + { + type: 'context_compaction_started' as const, + id: 'compaction-started-1', + turnId: 'turn-1', + ts: 123456789, + compactionId: 'compact-1', + scope: 'step' as const, + }, + { + type: 'context_compaction_failed' as const, + id: 'compaction-failed-1', + turnId: 'turn-1', + ts: 123456790, + compactionId: 'compact-1', + scope: 'step' as const, + reason: 'write_failed', + }, + ]; + for (const [index, event] of events.entries()) { + const update = projector.accept({ + kind: 'subscription.session_event', + hostEpoch: 'host-1', + subscriptionId: 'subscription-1', + sequence: index + 1, + sessionId: 'session-1', + runId: 'run-1', + event, + }); + assert.deepEqual(update.events, [event]); + } +}); + test('marks Runtime Host tool results whose durable content is omitted', () => { const projector = projectorAt(); @@ -948,10 +985,16 @@ test('context compaction projects one lifecycle across bootstrap, transition, an }, }); const bootstrapped = projectorAt(compacting); + const seeded = bootstrapped.seedActive(true); assert.deepEqual( - bootstrapped.seedActive(true).map(({ type, turnId }) => [type, turnId]), + seeded.map(({ type, turnId }) => [type, turnId]), [['context_compaction_started', 'turn-compact']], ); + assert.equal(seeded[0]?.type, 'context_compaction_started'); + if (seeded[0]?.type === 'context_compaction_started') { + assert.equal(seeded[0].scope, 'turn'); + assert.equal(seeded[0].compactionId, 'run-compact'); + } for (const previous of [ snapshot(), @@ -995,6 +1038,173 @@ test('context compaction projects one lifecycle across bootstrap, transition, an }); }); +test('seeds only automatic compactions whose durable started marker has no terminal marker', () => { + const transcript: StoredMessage[] = [ + { + type: 'system_note', + id: 'started-complete', + turnId: 'turn-1', + ts: 2, + kind: 'context_compaction_started', + data: { compactionId: 'complete', scope: 'step' }, + }, + { + type: 'system_note', + id: 'applied-complete', + turnId: 'turn-1', + ts: 3, + kind: 'context_compaction_applied', + data: { compactionId: 'complete', scope: 'step' }, + }, + { + type: 'system_note', + id: 'started-failed', + turnId: 'turn-1', + ts: 4, + kind: 'context_compaction_started', + data: { compactionId: 'failed', scope: 'step' }, + }, + { + type: 'system_note', + id: 'failed', + turnId: 'turn-1', + ts: 5, + kind: 'context_compaction_failed_open', + data: { compactionId: 'failed', scope: 'step', failOpenReason: 'write_failed' }, + }, + { + type: 'system_note', + id: 'started-active', + turnId: 'turn-1', + ts: 6, + kind: 'context_compaction_started', + data: { compactionId: 'active', scope: 'step' }, + }, + ]; + const current = snapshot(); + const projector = new RuntimeHostSessionProjector( + current, + createRuntimeHostSessionProjectionSeed(transcript, current), + fixedClock, + ); + + assert.deepEqual( + projector.seedActive(false).filter((event) => event.type === 'context_compaction_started'), + [ + { + type: 'context_compaction_started', + id: 'host-step-compaction-started:active', + turnId: 'turn-1', + ts: 6, + compactionId: 'active', + scope: 'step', + }, + ], + ); +}); + +test('tracks automatic compaction lifecycle events for later reseeding', () => { + const projector = projectorAt(); + const started = projector.accept({ + kind: 'subscription.session_event', + hostEpoch: 'host-1', + subscriptionId: 'subscription-1', + sequence: 1, + sessionId: 'session-1', + runId: 'run-1', + event: { + type: 'context_compaction_started', + id: 'started-1', + turnId: 'turn-1', + ts: 20, + compactionId: 'compaction-1', + scope: 'step', + }, + }); + assert.equal(started.events[0]?.type, 'context_compaction_started'); + assert.deepEqual( + projector.seedActive(false).filter((event) => event.type === 'context_compaction_started'), + [ + { + type: 'context_compaction_started', + id: 'host-step-compaction-started:compaction-1', + turnId: 'turn-1', + ts: 20, + compactionId: 'compaction-1', + scope: 'step', + }, + ], + ); + + projector.accept({ + kind: 'subscription.session_event', + hostEpoch: 'host-1', + subscriptionId: 'subscription-1', + sequence: 2, + sessionId: 'session-1', + runId: 'run-1', + event: { + type: 'context_compaction_applied', + id: 'applied-1', + turnId: 'turn-1', + ts: 30, + compactionId: 'compaction-1', + scope: 'step', + }, + }); + assert.equal( + projector.seedActive(false).some((event) => event.type === 'context_compaction_started'), + false, + ); + + projector.accept({ + kind: 'subscription.session_event', + hostEpoch: 'host-1', + subscriptionId: 'subscription-1', + sequence: 3, + sessionId: 'session-1', + runId: 'run-1', + event: { + type: 'context_compaction_started', + id: 'started-2', + turnId: 'turn-1', + ts: 40, + compactionId: 'compaction-2', + scope: 'step', + }, + }); + assert.equal( + projector + .seedActive(false) + .some( + (event) => + event.type === 'context_compaction_started' && event.compactionId === 'compaction-2', + ), + true, + ); + projector.accept({ + kind: 'subscription.session_event', + hostEpoch: 'host-1', + subscriptionId: 'subscription-1', + sequence: 4, + sessionId: 'session-1', + runId: 'run-1', + event: { + type: 'context_compaction_failed', + id: 'failed-2', + turnId: 'turn-1', + ts: 50, + compactionId: 'compaction-2', + scope: 'step', + reason: 'write_failed', + }, + }); + assert.equal( + projector.seedActive(false).some((event) => event.type === 'context_compaction_started'), + false, + ); +}); + test('seeds an empty queue only for a client that renders the queue', () => { const projector = projectorAt( snapshot({ diff --git a/packages/runtime-host/src/__tests__/session-transcript-pager.test.ts b/packages/runtime-host/src/__tests__/session-transcript-pager.test.ts index 8a5c0516fc7..95695342454 100644 --- a/packages/runtime-host/src/__tests__/session-transcript-pager.test.ts +++ b/packages/runtime-host/src/__tests__/session-transcript-pager.test.ts @@ -23,6 +23,10 @@ import { decodeStoredMessage as decodePersistedStoredMessage, type StoredMessage, } from '@maka/core/session'; +import { + readCompactionTimelineMarker, + readCompactionFailureReason, +} from '@maka/core/compaction-status'; import { markPersisted } from '@maka/core/persisted-value'; import { clientSubscription } from './fixtures/client-session-subscription.js'; import { SESSION_CONTINUITY_SCHEMA_VERSION } from '../protocol/index.js'; @@ -674,3 +678,80 @@ test('shared transcript preserves admitted action identity alongside its physica message, ); }); + +for (const outcome of ['running', 'compacted', 'failed', 'interrupted'] as const) { + test(`shared bootstrap and pages preserve ${outcome} compaction markers`, async () => { + const kind = + outcome === 'running' + ? 'context_compaction_started' + : outcome === 'compacted' + ? 'context_compaction_applied' + : 'context_compaction_failed_open'; + const data = { + scope: 'step', + compactionId: 'attempt-1', + ...(['failed', 'interrupted'].includes(outcome) + ? { failOpenReason: outcome === 'interrupted' ? 'interrupted' : 'timeout' } + : {}), + }; + const message: StoredMessage = { + type: 'system_note', + id: 'marker', + turnId: 'turn-1', + ts: 2, + kind, + data: { ...data, privateDiagnostic: 'must not be shared' }, + }; + const durable: StoredMessage[] = [message]; + const reader = transcriptReader(durable); + const { bootstrap, state } = await createSessionTranscriptBootstrap({ + reader, + sessionId: 'session-1', + subscriptionId: 'shared', + throughSequence: 0, + maxBytes: 4096, + projection: 'shared', + }); + durable.push({ ...message, id: 'later-marker' }); + updateSubscriberTranscriptHighWater(state, 1); + const page = await readSessionTranscriptPage({ + reader, + state, + request: { + subscriptionId: 'shared', + direction: 'newer', + throughSequence: 1, + cursor: null, + anchorSequence: 0, + maxBytes: 4096, + }, + }); + for (const result of [bootstrap.durable, page]) { + const rows = decodeBootstrap(result); + assert.equal(rows.length, 1); + assert.deepEqual(rows[0]?.data, data); + const decoded = decodePersistedStoredMessage(markPersisted(rows[0])); + assert.equal(readCompactionTimelineMarker(decoded)?.state, outcome); + } + }); +} + +test('shared legacy compaction failure preserves reason without inventing an attempt', () => { + for (const reasonField of ['failOpenReason', 'reason']) { + const projected = projectSharedSessionTranscriptMessage( + { + type: 'system_note', + id: 'legacy', + turnId: 'turn-1', + ts: 1, + kind: 'context_compaction_failed_open', + data: { [reasonField]: 'interrupted', privateDiagnostic: 'private' }, + }, + 'session-1', + ); + assert.ok(projected?.type === 'system_note'); + assert.equal(readCompactionFailureReason(projected.data), 'interrupted'); + assert.equal(readCompactionTimelineMarker(projected), undefined); + assert.deepEqual(projected.data, { failOpenReason: 'interrupted' }); + } +}); diff --git a/packages/runtime-host/src/adapter/session-projector.ts b/packages/runtime-host/src/adapter/session-projector.ts index c36d07e32a1..63950f19e42 100644 --- a/packages/runtime-host/src/adapter/session-projector.ts +++ b/packages/runtime-host/src/adapter/session-projector.ts @@ -23,6 +23,7 @@ import { foldAssistantDelta, type SessionEvent, } from '@maka/core/events'; +import { readCompactionTimelineMarker } from '@maka/core/compaction-status'; import type { StoredMessage, TurnRecord } from '@maka/core/session'; import type { InteractionPendingSnapshot, @@ -52,12 +53,18 @@ export interface RuntimeHostSessionProjectionSeed { readonly turnId: string; }[]; readonly activeAssistantMessages: readonly Extract[]; + readonly activeStepCompactions: readonly { + readonly compactionId: string; + readonly turnId: string; + readonly ts: number; + }[]; } export function createRuntimeHostSessionProjectionSeed( transcript: readonly StoredMessage[], snapshot: SessionContinuitySnapshot, ): RuntimeHostSessionProjectionSeed { + const activeStepCompactions = activeStepCompactionsFromTranscript(transcript); return { durableUserMessages: transcript .filter( @@ -71,9 +78,33 @@ export function createRuntimeHostSessionProjectionSeed( (message): message is Extract => message.type === 'assistant' && message.turnId === snapshot.rootTurn?.turnId, ), + activeStepCompactions: + snapshot.rootTurn === null + ? [] + : activeStepCompactions.filter((attempt) => attempt.turnId === snapshot.rootTurn?.turnId), }; } +function activeStepCompactionsFromTranscript( + transcript: readonly StoredMessage[], +): Array<{ compactionId: string; turnId: string; ts: number }> { + const active = new Map(); + for (const message of transcript) { + const marker = readCompactionTimelineMarker(message); + if (!marker || !message.turnId) continue; + if (marker.state === 'running') { + active.set(marker.compactionId, { + compactionId: marker.compactionId, + turnId: message.turnId, + ts: marker.ts, + }); + } else { + active.delete(marker.compactionId); + } + } + return [...active.values()]; +} + export type RuntimeHostTerminalTurn = Extract< TurnSnapshot, { status: 'completed' } | { status: 'failed' } | { status: 'cancelled' } @@ -93,6 +124,10 @@ export class RuntimeHostSessionProjector { readonly #durableTurnByMessage: Map; readonly #accumulators = new Map(); readonly #placedSteeringMessageIds = new Set(); + readonly #activeStepCompactions = new Map< + string, + { compactionId: string; turnId: string; ts: number } + >(); #projectMessageAdmissions: boolean; constructor( @@ -107,6 +142,9 @@ export class RuntimeHostSessionProjector { this.#durableTurnByMessage = new Map( seed.durableUserMessages.map(({ messageId, turnId }) => [messageId, turnId]), ); + for (const attempt of seed.activeStepCompactions) { + this.#activeStepCompactions.set(attempt.compactionId, attempt); + } this.#projectMessageAdmissions = projectMessageAdmissions; const root = snapshot.rootTurn; if (!root) return; @@ -203,6 +241,17 @@ export class RuntimeHostSessionProjector { if (root.rootExecutionKind === 'context_compact') { events.push(contextCompactionStartedEvent(root, this.#now())); } + for (const attempt of this.#activeStepCompactions.values()) { + if (attempt.turnId !== root.turnId) continue; + events.push({ + type: 'context_compaction_started', + id: `host-step-compaction-started:${attempt.compactionId}`, + turnId: attempt.turnId, + ts: attempt.ts, + compactionId: attempt.compactionId, + scope: 'step', + }); + } if (includeAssistantText) { for (const accumulator of this.#accumulators.values()) { if (accumulator.complete) continue; @@ -399,6 +448,22 @@ export class RuntimeHostSessionProjector { } if (frame.kind === 'subscription.session_event') { const event = projectSessionEvent(frame); + if (event.type === 'context_compaction_started' && event.scope === 'step') { + if (event.compactionId !== undefined) { + this.#activeStepCompactions.set(event.compactionId, { + compactionId: event.compactionId, + turnId: event.turnId, + ts: event.ts, + }); + } + } else if ( + event.type === 'context_compaction_applied' || + event.type === 'context_compaction_failed' + ) { + if (event.compactionId !== undefined) { + this.#activeStepCompactions.delete(event.compactionId); + } + } if (event.type === 'steering_message') { if (this.#durableTurnByMessage.has(event.messageId)) return emptyUpdate(events); this.#placedSteeringMessageIds.add(event.messageId); @@ -437,6 +502,7 @@ export class RuntimeHostSessionProjector { ); } if (startedTurn) this.#accumulators.clear(); + if (startedTurn) this.#activeStepCompactions.clear(); // Emit the presentation-only compaction-started event when the root Turn // FIRST becomes a `context_compact` run, not only when the runId changes. // The real lifecycle is `admitted (no rootExecutionKind) → running/ @@ -458,6 +524,7 @@ export class RuntimeHostSessionProjector { ? root : undefined; if (terminalTurn) events.push(...this.#terminalEvents(terminalTurn)); + if (terminalTurn) this.#clearStepCompactionsForTurn(terminalTurn.turnId); return { events, previousSnapshot, @@ -467,6 +534,12 @@ export class RuntimeHostSessionProjector { }; } + #clearStepCompactionsForTurn(turnId: string): void { + for (const [compactionId, attempt] of this.#activeStepCompactions) { + if (attempt.turnId === turnId) this.#activeStepCompactions.delete(compactionId); + } + } + #terminalEvents(root: RuntimeHostTerminalTurn, includeSettled = false): SessionEvent[] { const events: SessionEvent[] = []; for (const accumulator of this.#accumulators.values()) { @@ -564,6 +637,8 @@ function contextCompactionStartedEvent( id: `host-compaction-started:${turn.runId}`, turnId: turn.turnId, ts: now, + compactionId: turn.runId, + scope: 'turn', }; } @@ -637,6 +712,29 @@ function projectSessionEvent( id: event.id, turnId: event.turnId, ts: event.ts, + ...(event.compactionId !== undefined ? { compactionId: event.compactionId } : {}), + ...(event.scope !== undefined ? { scope: event.scope } : {}), + }; + } + if (event.type === 'context_compaction_started') { + return { + type: event.type, + id: event.id, + turnId: event.turnId, + ts: event.ts, + ...(event.compactionId !== undefined ? { compactionId: event.compactionId } : {}), + ...(event.scope !== undefined ? { scope: event.scope } : {}), + }; + } + if (event.type === 'context_compaction_failed') { + return { + type: event.type, + id: event.id, + turnId: event.turnId, + ts: event.ts, + compactionId: event.compactionId, + scope: event.scope, + reason: event.reason, }; } if (event.type === 'steering_message') { diff --git a/packages/runtime-host/src/assistant/README.md b/packages/runtime-host/src/assistant/README.md index 1c1a1d9a4e6..f4a28326ec8 100644 --- a/packages/runtime-host/src/assistant/README.md +++ b/packages/runtime-host/src/assistant/README.md @@ -29,7 +29,7 @@ The feature defaults **off**, including installations with existing assistant da Personal Assistant keeps the existing Session creation rules: use the configured default Project when applicable, then the Host's current Project preference. A local Host with no Project still allocates its existing `~/Maka/tasks/task-*` directory (`~/Maka Dev/tasks/task-*` for development). -Only when an implicit assistant creation on WSL or a remote Host has no resolvable Project, Desktop calls `assistant.allocate-workspace` on that Host. The Host allocates a fresh task directory in its own home and registers its Project identity for ordinary Session creation. It never adopts an arbitrary existing Project or accepts a Client directory. Standard Host development profiles use `~/Maka Dev/tasks/task-*`; other Host profiles use `~/Maka/tasks/task-*`. The connecting Desktop does not choose a remote Host's namespace. Allocation failure is reported without falling back to the Client filesystem. Client and Host must both support compatibility epoch 226. +Only when an implicit assistant creation on WSL or a remote Host has no resolvable Project, Desktop calls `assistant.allocate-workspace` on that Host. The Host allocates a fresh task directory in its own home and registers its Project identity for ordinary Session creation. It never adopts an arbitrary existing Project or accepts a Client directory. Standard Host development profiles use `~/Maka Dev/tasks/task-*`; other Host profiles use `~/Maka/tasks/task-*`. The connecting Desktop does not choose a remote Host's namespace. Allocation failure is reported without falling back to the Client filesystem. Client and Host must both support compatibility epoch 227. Existing assistant bindings retain their conversation, cwd, model and history. Opening them does not allocate another directory or migrate files. This change does not alter ordinary task creation or explicit remote workspace validation. diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index c3caac84e94..9bf21d1b199 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -108,7 +108,10 @@ export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. // Re-check against main before merging; sibling PRs may use the same next epoch. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 229 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 230 as const; +// 230: Session event frames add in-Turn context-compaction started/failed +// events; the applied frame carries `compactionId` and `scope`. Older peers +// reject the new event union or the added fields in their closed decoders. // 229: Session check-in proposes cards for unfinished work across Sessions. // 228: Executor catalog queries support per-agent filtering and process-free inventory. // Older peers reject the executorId and listOnly fields in the strict query decoder. diff --git a/packages/runtime-host/src/protocol/session-continuity.ts b/packages/runtime-host/src/protocol/session-continuity.ts index 48b0c20bbf8..75fd4d5da33 100644 --- a/packages/runtime-host/src/protocol/session-continuity.ts +++ b/packages/runtime-host/src/protocol/session-continuity.ts @@ -238,13 +238,39 @@ export interface SessionContextCompactionAppliedEvent { id: string; turnId: string; ts: number; + compactionId: string; + scope: 'turn' | 'step'; +} + +export interface SessionContextCompactionStartedEvent { + type: 'context_compaction_started'; + id: string; + turnId: string; + ts: number; + compactionId: string; + scope: 'step'; +} + +export interface SessionContextCompactionFailedEvent { + type: 'context_compaction_failed'; + id: string; + turnId: string; + ts: number; + compactionId: string; + scope: 'step'; + reason: string; } export interface SessionEventFrame extends SubscriptionEnvelope { kind: 'subscription.session_event'; sessionId: string; runId: string; - event: SessionToolEvent | SessionSteeringEvent | SessionContextCompactionAppliedEvent; + event: + | SessionToolEvent + | SessionSteeringEvent + | SessionContextCompactionAppliedEvent + | SessionContextCompactionStartedEvent + | SessionContextCompactionFailedEvent; } export interface SessionTranscriptAdvancedFrame extends SubscriptionEnvelope { @@ -816,12 +842,76 @@ function decodeSessionFrameEvent(value: unknown): SessionEventFrame['event'] { 'id', 'turnId', 'ts', + 'compactionId', + 'scope', ]); + if (record.scope !== 'turn' && record.scope !== 'step') { + throw invalidProtocolFrame('Invalid context compaction applied scope'); + } return { type: record.type, id: requireId(record.id, 'Session context compaction applied event id'), turnId: requireEntityId(record.turnId, 'turnId'), ts: requireCount(record.ts, 'Session context compaction applied event timestamp'), + compactionId: requireId( + record.compactionId, + 'Session context compaction applied compactionId', + ), + scope: record.scope, + }; + } + if (record.type === 'context_compaction_started') { + assertExactKeys(record, 'Session context compaction started event', [ + 'type', + 'id', + 'turnId', + 'ts', + 'compactionId', + 'scope', + ]); + if (record.scope !== 'step') { + throw invalidProtocolFrame('Invalid context compaction started scope'); + } + return { + type: record.type, + id: requireId(record.id, 'Session context compaction started event id'), + turnId: requireEntityId(record.turnId, 'turnId'), + ts: requireCount(record.ts, 'Session context compaction started event timestamp'), + compactionId: requireId( + record.compactionId, + 'Session context compaction started compactionId', + ), + scope: 'step', + }; + } + if (record.type === 'context_compaction_failed') { + assertExactKeys(record, 'Session context compaction failed event', [ + 'type', + 'id', + 'turnId', + 'ts', + 'compactionId', + 'scope', + 'reason', + ]); + if (record.scope !== 'step') { + throw invalidProtocolFrame('Invalid context compaction failed scope'); + } + return { + type: record.type, + id: requireId(record.id, 'Session context compaction failed event id'), + turnId: requireEntityId(record.turnId, 'turnId'), + ts: requireCount(record.ts, 'Session context compaction failed event timestamp'), + compactionId: requireId( + record.compactionId, + 'Session context compaction failed compactionId', + ), + scope: 'step', + reason: requireUtf8BoundedString( + record.reason, + 'Session context compaction failure reason', + 256, + ), }; } if (record.type === 'steering_message') return decodeSessionSteeringEvent(record); diff --git a/packages/runtime-host/src/server/root-turn-coordinator.ts b/packages/runtime-host/src/server/root-turn-coordinator.ts index 1f8bd113580..ffe4ac6a1ad 100644 --- a/packages/runtime-host/src/server/root-turn-coordinator.ts +++ b/packages/runtime-host/src/server/root-turn-coordinator.ts @@ -4073,7 +4073,9 @@ function isRuntimeSessionForwardedEvent( event.type === 'tool_result_preview' || event.type === 'tool_result' || event.type === 'steering_message' || + (event.type === 'context_compaction_started' && event.scope !== 'turn') || event.type === 'context_compaction_applied' || + event.type === 'context_compaction_failed' || event.type === 'provider_retry' ); } diff --git a/packages/runtime-host/src/server/session-continuity-coordinator.ts b/packages/runtime-host/src/server/session-continuity-coordinator.ts index 628fc6c09bb..7c110f8a918 100644 --- a/packages/runtime-host/src/server/session-continuity-coordinator.ts +++ b/packages/runtime-host/src/server/session-continuity-coordinator.ts @@ -102,7 +102,9 @@ type RuntimeSessionForwardedEventType = | 'tool_result_preview' | 'tool_result' | 'steering_message' + | 'context_compaction_started' | 'context_compaction_applied' + | 'context_compaction_failed' | 'provider_retry'; export type RuntimeSessionForwardedEvent = Extract< @@ -882,6 +884,12 @@ export class SessionContinuityCoordinator implements SessionContinuityService { this.#deliverInOrder( subscriber, () => { + const projectedEvent = projectSessionEvent( + event, + sessionId, + subscriber.principalKind === 'session_guest', + ); + if (!projectedEvent) return; const frame: SessionEventFrame = { kind: 'subscription.session_event', hostEpoch: this.#hostEpoch, @@ -889,11 +897,7 @@ export class SessionContinuityCoordinator implements SessionContinuityService { sequence: subscriber.nextSequence, sessionId, runId, - event: projectSessionEvent( - event, - sessionId, - subscriber.principalKind === 'session_guest', - ), + event: projectedEvent, }; this.#enqueue(subscriber, frame); }, @@ -1108,6 +1112,12 @@ export class SessionContinuityCoordinator implements SessionContinuityService { if (rootTurn && !isTerminalTurn(rootTurn)) { for (const preview of committed.state.toolResultPreviews.values()) { if (preview.turnId !== rootTurn.turnId) continue; + const projectedEvent = projectSessionEvent( + preview, + sessionId, + subscriber.principalKind === 'session_guest', + ); + if (!projectedEvent) continue; const frame: SessionEventFrame = { kind: 'subscription.session_event', hostEpoch: this.#hostEpoch, @@ -1115,11 +1125,7 @@ export class SessionContinuityCoordinator implements SessionContinuityService { sequence: subscriber.nextSequence, sessionId, runId: rootTurn.runId, - event: projectSessionEvent( - preview, - sessionId, - subscriber.principalKind === 'session_guest', - ), + event: projectedEvent, }; this.#enqueue(subscriber, frame); } @@ -2010,13 +2016,39 @@ function projectSessionEvent( >, sessionId: string, shared = false, -): SessionEventFrame['event'] { +): SessionEventFrame['event'] | undefined { if (event.type === 'context_compaction_applied') { + if (!event.compactionId) return undefined; + return { + type: event.type, + id: event.id, + turnId: event.turnId, + ts: event.ts, + compactionId: event.compactionId, + scope: event.scope ?? 'step', + }; + } + if (event.type === 'context_compaction_started') { + if (event.scope !== 'step' || !event.compactionId) return undefined; + return { + type: event.type, + id: event.id, + turnId: event.turnId, + ts: event.ts, + compactionId: event.compactionId, + scope: 'step', + }; + } + if (event.type === 'context_compaction_failed') { + if (!event.compactionId) return undefined; return { type: event.type, id: event.id, turnId: event.turnId, ts: event.ts, + compactionId: event.compactionId, + scope: 'step', + reason: event.reason, }; } if (event.type === 'steering_message') { diff --git a/packages/runtime-host/src/server/shared-session-transcript.ts b/packages/runtime-host/src/server/shared-session-transcript.ts index ced768c7e65..ba9937c74e8 100644 --- a/packages/runtime-host/src/server/shared-session-transcript.ts +++ b/packages/runtime-host/src/server/shared-session-transcript.ts @@ -18,6 +18,10 @@ */ import type { AttachmentRef, MessageContent } from '@maka/core/events'; +import { + readCompactionFailureReason, + readCompactionTimelineMarker, +} from '@maka/core/compaction-status'; import { projectToolActivityArgs } from '@maka/core/tool-activity-args'; import { isUserVisibleSessionSystemNote, @@ -156,22 +160,59 @@ export function projectSharedSessionTranscriptMessage( ...(message.reasoning === undefined ? {} : { reasoning: message.reasoning }), ...(message.costUsd === undefined ? {} : { costUsd: message.costUsd }), }; - case 'system_note': - return isUserVisibleSessionSystemNote(message.kind) - ? { - type: message.type, - id: message.id, - ...(message.turnId === undefined ? {} : { turnId: message.turnId }), - ts: message.ts, - kind: message.kind, - } - : null; + case 'system_note': { + const isCompactionMarker = readCompactionTimelineMarker(message) !== undefined; + if (!isCompactionMarker && !isUserVisibleSessionSystemNote(message.kind)) return null; + const data = projectSharedSessionSystemNoteData(message); + return { + type: message.type, + id: message.id, + ...(message.turnId === undefined ? {} : { turnId: message.turnId }), + ts: message.ts, + kind: message.kind, + ...(data === undefined ? {} : { data }), + }; + } case 'permission_decision': case 'workhub_coordination': return null; } } +/** Shares only the fields needed to reconstruct a compaction attempt. */ +function projectSharedSessionSystemNoteData( + message: Extract, +): Record | undefined { + if ( + message.kind !== 'context_compaction_started' && + message.kind !== 'context_compaction_applied' && + message.kind !== 'context_compaction_failed_open' + ) { + return undefined; + } + const data = + typeof message.data === 'object' && message.data !== null && !Array.isArray(message.data) + ? (message.data as Record) + : undefined; + const compactionId = + typeof data?.compactionId === 'string' && data.compactionId.length > 0 + ? data.compactionId + : undefined; + const scope = data?.scope === 'step' || data?.scope === 'turn' ? data.scope : undefined; + const failOpenReason = + message.kind === 'context_compaction_failed_open' + ? readCompactionFailureReason(message.data) + : undefined; + if (compactionId === undefined && scope === undefined && failOpenReason === undefined) { + return undefined; + } + return { + ...(scope === undefined ? {} : { scope }), + ...(compactionId === undefined ? {} : { compactionId }), + ...(failOpenReason === undefined ? {} : { failOpenReason }), + }; +} + function isSharedSessionAttachment(attachment: AttachmentRef, sessionId: string): boolean { return attachment.ref.kind === 'session_file' && attachment.ref.sessionId === sessionId; } diff --git a/packages/runtime/src/__tests__/mid-turn-capacity-backend.test.ts b/packages/runtime/src/__tests__/mid-turn-capacity-backend.test.ts index 27e5116c47b..e51d6453fc2 100644 --- a/packages/runtime/src/__tests__/mid-turn-capacity-backend.test.ts +++ b/packages/runtime/src/__tests__/mid-turn-capacity-backend.test.ts @@ -789,6 +789,134 @@ function compactionDecisions( } function defineMidTurnSuite(consumer: ConsumerMode): void { + test('does not label an uncancelled provider AbortError as user interruption', async () => { + const fixture = buildFixture({ + summarize: () => { + throw Object.assign(new Error('provider aborted'), { name: 'AbortError' }); + }, + }); + await runFixtureTurn(fixture, consumer); + assert.equal( + fixture.events.find((event) => event.type === 'context_compaction_failed')?.reason, + 'summarizer_failed', + ); + assert.equal(fixture.summarizerCalls, 1); + }); + + test('records interruption when initial overflow recovery is cancelled', { + timeout: 5_000, + }, async () => { + const started = deferred(); + const fixture = buildFixture({ + firstRequestContextOverflow: true, + summarize: ({ abortSignal }) => + new Promise((_, reject) => { + assert.ok(abortSignal); + abortSignal.addEventListener( + 'abort', + () => reject(Object.assign(new Error('cancelled'), { name: 'AbortError' })), + { once: true }, + ); + started.resolve(); + }), + }); + const turn = runFixtureTurn(fixture, consumer); + await started.promise; + await fixture.backend.stop('user_stop'); + await turn; + assert.equal(fixture.recorded.length, 0); + assert.equal( + fixture.events.find((event) => event.type === 'context_compaction_failed')?.reason, + 'interrupted', + ); + }); + + test('a stop before the summarizer call opens no compaction row', { + timeout: 5_000, + }, async () => { + // The planner loads its transition snapshot after choosing a safe span and + // right before the summarizer. Every request build reads the ledger too, so + // only this read pins the stop between the ledger boundary and the call. + let stopped = false; + const fixture = buildFixture({ + loadTransitions: async () => { + if (!stopped && new Error().stack?.includes('planHistoryCompaction')) { + stopped = true; + void fixture.backend.stop('user_stop'); + } + return EMPTY_PROJECTION_SNAPSHOT; + }, + }); + await runFixtureTurn(fixture, consumer); + assert.equal(stopped, true, 'the stop must land inside the compaction attempt'); + assert.equal(fixture.recorded.length, 0); + assert.equal( + fixture.events.some( + (event) => + event.type === 'context_compaction_started' || event.type === 'context_compaction_failed', + ), + false, + ); + }); + + test('a fail-open before the summarizer call opens no row on any step', async () => { + // No prior turns: each over-window step finds no safe completed span and + // retries on the next step. Those misses must not stack one timeline row + // per step; the send keeps its single settlement note. + const fixture = buildFixture({ contextWindow: 120, withoutPriorTurns: true }); + await runFixtureTurn(fixture, consumer); + + const missedSteps = compactionDecisions(fixture).filter( + (decision) => decision.failOpenReason === 'no_safe_completed_span', + ); + assert.equal(missedSteps.length > 0, true); + // Rows exist only for attempts that reached the summarizer, and each one + // that opened also settled. + const startedIds = fixture.events.flatMap((event) => + event.type === 'context_compaction_started' ? [event.compactionId] : [], + ); + const settledIds = fixture.events.flatMap((event) => + event.type === 'context_compaction_applied' || event.type === 'context_compaction_failed' + ? [event.compactionId] + : [], + ); + assert.equal(startedIds.length <= fixture.summarizerCalls, true); + assert.deepEqual([...settledIds].sort(), [...startedIds].sort()); + assert.equal( + fixture.events.some( + (event) => + event.type === 'context_compaction_failed' && event.reason === 'no_safe_completed_span', + ), + false, + ); + const missNotes = fixture.ledger.filter( + (event) => + event.content?.kind === 'system_note' && + event.content.note === 'context_compaction_failed_open' && + (event.content.data as { failOpenReason?: string } | undefined)?.failOpenReason === + 'no_safe_completed_span', + ); + assert.equal(missNotes.length <= 1, true, 'at most the one settlement note per send'); + }); + + test('an unexpected throw after the summarizer call still settles the open row', async () => { + const fixture = buildFixture({ + loadTransitions: async () => { + if (fixture.summarizerCalls > 0) throw new Error('transition ledger offline'); + return EMPTY_PROJECTION_SNAPSHOT; + }, + }); + await runFixtureTurn(fixture, consumer).catch(() => {}); + + const started = fixture.events.find((event) => event.type === 'context_compaction_started'); + const failed = fixture.events.find((event) => event.type === 'context_compaction_failed'); + assert.ok(started?.type === 'context_compaction_started'); + assert.ok(failed?.type === 'context_compaction_failed'); + assert.equal(failed.compactionId, started.compactionId); + assert.equal(failed.reason, 'unexpected_error'); + assert.equal(fixture.recorded.length, 0); + }); + test('publishes the applied boundary while the post-compaction request is still running', { timeout: 5_000, }, async () => { @@ -824,6 +952,26 @@ function defineMidTurnSuite(consumer: ConsumerMode): void { assert.equal(boundaries[0]?.id, applied?.id); assert.equal(boundaries[0]?.ts, applied?.ts); assert.equal(boundaries[0]?.modelVisibility, 'hidden'); + const started = fixture.events.find((event) => event.type === 'context_compaction_started'); + assert.ok(started?.type === 'context_compaction_started'); + assert.ok(applied?.type === 'context_compaction_applied'); + assert.equal(started.scope, 'step'); + assert.equal(started.compactionId, applied.compactionId); + assert.equal( + fixture.events.indexOf(started) < fixture.events.indexOf(applied), + true, + 'the start row must be published before the applied boundary', + ); + const startedNote = fixture.ledger.find( + (event) => + event.content?.kind === 'system_note' && + event.content.note === 'context_compaction_started', + ); + assert.ok(startedNote && startedNote.content?.kind === 'system_note'); + assert.deepEqual(startedNote.content.data, { + compactionId: started.compactionId, + scope: 'step', + }); } finally { requestHeld.resolve(); await turn; @@ -864,6 +1012,15 @@ function defineMidTurnSuite(consumer: ConsumerMode): void { ); const complete = fixture.events.find((event) => event.type === 'complete'); assert.equal(complete?.type === 'complete' ? complete.stopReason : undefined, 'user_stop'); + const failed = fixture.events.find((event) => event.type === 'context_compaction_failed'); + assert.equal(failed?.reason, 'interrupted'); + const note = fixture.ledger.find( + (event) => + event.content?.kind === 'system_note' && + event.content.note === 'context_compaction_failed_open', + ); + assert.ok(note?.content?.kind === 'system_note'); + assert.equal((note.content.data as { failOpenReason: string }).failOpenReason, 'interrupted'); }); test('compacts after provider usage crosses the declared window, persists first, and continues the same turn', async () => { @@ -1122,6 +1279,25 @@ function defineMidTurnSuite(consumer: ConsumerMode): void { fixture.events.some((event) => event.type === 'context_compaction_applied'), false, ); + const started = fixture.events.find((event) => event.type === 'context_compaction_started'); + const failed = fixture.events.find((event) => event.type === 'context_compaction_failed'); + assert.ok(started?.type === 'context_compaction_started'); + assert.ok(failed?.type === 'context_compaction_failed'); + assert.equal(started.compactionId, failed.compactionId); + assert.equal(failed.reason, 'write_failed'); + assert.equal(fixture.events.indexOf(started) < fixture.events.indexOf(failed), true); + const failedNotes = fixture.ledger.filter( + (event) => + event.content?.kind === 'system_note' && + event.content.note === 'context_compaction_failed_open', + ); + assert.equal(failedNotes.length, 1, 'the terminal event owns one durable failure note'); + assert.ok(failedNotes[0]?.content?.kind === 'system_note'); + assert.deepEqual(failedNotes[0].content.data, { + compactionId: started.compactionId, + scope: 'step', + failOpenReason: 'write_failed', + }); }); test('a malformed summarizer completion fails open end-to-end with its granular reason', async () => { diff --git a/packages/runtime/src/__tests__/overflow-reactive-recovery.test.ts b/packages/runtime/src/__tests__/overflow-reactive-recovery.test.ts index a88b50d414f..4ee3bd633ad 100644 --- a/packages/runtime/src/__tests__/overflow-reactive-recovery.test.ts +++ b/packages/runtime/src/__tests__/overflow-reactive-recovery.test.ts @@ -1246,6 +1246,16 @@ describe('reactive overflow recovery in the streaming backend', () => { assert.equal(fixture.recorded.length, 1); assert.equal(fixture.recorded[0]!.phase, 'mid_turn'); assert.equal(fixture.summarizerCalls(), 1); + const started = fixture.events.find((event) => event.type === 'context_compaction_started'); + const applied = fixture.events.find((event) => event.type === 'context_compaction_applied'); + assert.ok(started?.type === 'context_compaction_started'); + assert.ok(applied?.type === 'context_compaction_applied'); + assert.equal(started.compactionId, applied.compactionId); + assert.equal( + fixture.events.indexOf(started) < fixture.events.indexOf(applied), + true, + 'overflow recovery publishes the start before the applied boundary', + ); // The completed tool step was not re-executed on the retry. assert.deepEqual(fixture.toolExecutions, ['one.md']); // Send-level usage owner (review P1-2): the terminal record carries BOTH diff --git a/packages/runtime/src/__tests__/session-event-runtime-mapper.test.ts b/packages/runtime/src/__tests__/session-event-runtime-mapper.test.ts index 4a07cc6e3a4..d0ed0531e7e 100644 --- a/packages/runtime/src/__tests__/session-event-runtime-mapper.test.ts +++ b/packages/runtime/src/__tests__/session-event-runtime-mapper.test.ts @@ -433,8 +433,36 @@ const PROJECTION_SAMPLES: ProjectionSamples = { text: 'UTC: 2026-10-19T00:00:00Z', }, }, + context_compaction_started: { + subject: { + type: 'context_compaction_started', + id: 'e', + turnId: 'turn-1', + ts: 1, + compactionId: 'compact-1', + scope: 'step', + }, + }, context_compaction_applied: { - subject: { type: 'context_compaction_applied', id: 'e', turnId: 'turn-1', ts: 1 }, + subject: { + type: 'context_compaction_applied', + id: 'e', + turnId: 'turn-1', + ts: 1, + compactionId: 'compact-1', + scope: 'step', + }, + }, + context_compaction_failed: { + subject: { + type: 'context_compaction_failed', + id: 'e', + turnId: 'turn-1', + ts: 1, + compactionId: 'compact-1', + scope: 'step', + reason: 'write_failed', + }, }, text_delta: { subject: { type: 'text_delta', id: 'e', turnId: 'turn-1', ts: 1, messageId: 'm1', text: 'h' }, @@ -720,6 +748,22 @@ describe('SessionEvent projection coverage', () => { ); }); + test('keeps a Host-owned compaction start out of durable Runtime events', () => { + const event: SessionEvent = { + type: 'context_compaction_started', + id: 'host-compaction-started', + turnId: 'turn-1', + ts: 1, + compactionId: 'run-1', + scope: 'turn', + }; + assert.equal(isLiveBackendSessionEvent(event), false); + assert.throws( + () => mapSessionEventToRuntimeEvent(event, ctx), + /turn-scoped context_compaction_started is not a backend event/, + ); + }); + test('keeps Host-owned Client Capability interactions out of backend projection', () => { const request: SessionEvent = { type: 'client_capability_request', diff --git a/packages/runtime/src/ai-sdk-compaction.ts b/packages/runtime/src/ai-sdk-compaction.ts index 8c219c4ed28..5c870d9543c 100644 --- a/packages/runtime/src/ai-sdk-compaction.ts +++ b/packages/runtime/src/ai-sdk-compaction.ts @@ -148,6 +148,107 @@ export interface AutomaticMemoryCompactionDecision { readonly dispatch: boolean; } +/** Presentation lifecycle shared by proactive and overflow compaction. */ +export interface CompactionLifecycle { + /** Allocates the identity shared by the start/terminal presentation rows. */ + newCompactionId: () => string; + /** + * The attempt is about to call the summarizer. Checks that fail before that + * call (no safe span, ledger read, head anchor) open no row: they can repeat + * on every over-window step and stay with the once-per-send settlement note. + */ + onCompactionStarted?: (compactionId: string) => void | Promise; + /** + * Fired after the checkpoint is persisted and the projection takes over, + * before the replacement request. This apply-time boundary is fail-open: + * presentation bookkeeping must never disturb compaction. + */ + onCompactionApplied?: (compactionId: string) => void | Promise; + /** An opened attempt failed open with this diagnostic. */ + onCompactionFailed?: (compactionId: string, reason: string) => void | Promise; +} + +function bindCompactionAttempt(lifecycle: CompactionLifecycle) { + const compactionId = lifecycle.newCompactionId(); + return { + onCompactionStarted: () => lifecycle.onCompactionStarted?.(compactionId), + onCompactionApplied: () => lifecycle.onCompactionApplied?.(compactionId), + onCompactionFailed: (reason: string) => lifecycle.onCompactionFailed?.(compactionId, reason), + }; +} + +/** + * One attempt's visible row: opened at most once, settled at most once, and + * settled only if it was opened. Every callback is fail-open because + * presentation bookkeeping must never disturb compaction. + */ +interface CompactionAttemptPresentation { + open(): Promise; + apply(): Promise; + fail(reason: string): Promise; +} + +function presentCompactionAttempt(callbacks: { + onCompactionStarted?: () => void | Promise; + onCompactionApplied?: () => void | Promise; + onCompactionFailed?: (reason: string) => void | Promise; +}): CompactionAttemptPresentation { + let phase: 'idle' | 'open' | 'settled' = 'idle'; + const open = async (): Promise => { + if (phase !== 'idle') return; + phase = 'open'; + try { + await callbacks.onCompactionStarted?.(); + } catch { + // Losing the start row must not block the summarizer call it precedes. + } + }; + return { + open, + apply: async () => { + // A fold always passes through the summarizer, so the row is normally + // open already; opening here keeps the start/terminal pair complete. + await open(); + if (phase !== 'open') return; + phase = 'settled'; + try { + await callbacks.onCompactionApplied?.(); + } catch { + // The boundary row explains the compaction; losing it must not undo one. + } + }, + fail: async (reason) => { + if (phase !== 'open') return; + phase = 'settled'; + try { + await callbacks.onCompactionFailed?.(reason); + } catch { + // The settlement fallback still records the visible failure note. + } + }, + }; +} + +type ActiveRequestCompactionInput = { + turnId: string; + state: MidTurnCapacityCompactState; + /** The turn this replacement request is built for. */ + origin: ProviderRequestOrigin; + queue: AsyncEventQueue; + minFlushedSteps: number; + activeToolsForStep: readonly string[]; + memoryCompactionDecision?: () => AutomaticMemoryCompactionDecision; + onMemoryCompaction?: (input: AutomaticMemoryCompactionDispatch) => void; + /** See {@link CompactionLifecycle.onCompactionStarted}. */ + onCompactionStarted?: () => void | Promise; + /** See {@link CompactionLifecycle.onCompactionApplied}. */ + onCompactionApplied?: () => void | Promise; + /** See {@link CompactionLifecycle.onCompactionFailed}. */ + onCompactionFailed?: (reason: string) => void | Promise; + phase?: 'pre_turn' | 'mid_turn'; + abortSignal?: AbortSignal; +}; + /** Constructor dependencies for AiSdkCompaction. */ export interface AiSdkCompactionDeps { input: AiSdkCompactionCapabilities; @@ -838,13 +939,7 @@ export class AiSdkCompaction { queue: AsyncEventQueue, providerTools: readonly MakaTool[], onDiagnosticPatch: (patch: Partial) => void, - /** - * Fired once a compaction is durable and applied — checkpoint persisted and the - * projection takes over — before the replacement request goes out. The - * runtime records its apply-time boundary row here; like every system - * note it must never disturb the compaction, so the call is fail-open. - */ - onCompactionApplied: (() => void | Promise) | undefined, + lifecycle: CompactionLifecycle, origin: ProviderRequestOrigin, memoryCompactionDecision?: () => AutomaticMemoryCompactionDecision, onMemoryCompaction?: (input: AutomaticMemoryCompactionDispatch) => void, @@ -918,6 +1013,7 @@ export class AiSdkCompaction { return keepProjection(); } const activeToolsForStep = options.resolveDispatch(options.activeTools).activeTools; + const attempt = bindCompactionAttempt(lifecycle); // Fold a safe completed prefix of the durable turn ledger into a // replacement projection (validate → persist), shared with the reactive // overflow path. This stage maps the outcome to the request-projection contract: @@ -932,7 +1028,7 @@ export class AiSdkCompaction { activeToolsForStep, memoryCompactionDecision, onMemoryCompaction, - onCompactionApplied, + ...attempt, abortSignal, }); if (outcome.decision === 'fail') { @@ -969,28 +1065,36 @@ export class AiSdkCompaction { * covered raw span. It only shapes: the pass/terminate verdict and the * diagnostic emission are the caller's. */ - public async compactActiveRequestHistory(input: { - turnId: string; - state: MidTurnCapacityCompactState; - /** The turn this replacement request is built for. */ - origin: ProviderRequestOrigin; - queue: AsyncEventQueue; - minFlushedSteps: number; - activeToolsForStep: readonly string[]; - memoryCompactionDecision?: () => AutomaticMemoryCompactionDecision; - onMemoryCompaction?: (input: AutomaticMemoryCompactionDispatch) => void; - /** See {@link buildMidTurnCapacityCompactProjection}'s parameter. */ - onCompactionApplied?: () => void | Promise; - phase?: 'pre_turn' | 'mid_turn'; - abortSignal?: AbortSignal; - }): Promise { + public async compactActiveRequestHistory( + input: ActiveRequestCompactionInput, + ): Promise { + const attempt = presentCompactionAttempt(input); + try { + return await this.runActiveRequestCompaction(input, attempt); + } catch (error) { + // An opened row needs a terminal event even when the attempt throws; + // otherwise it spins until the Turn settles and a reattaching client + // replays it as running. + await attempt.fail(input.abortSignal?.aborted ? 'interrupted' : 'unexpected_error'); + throw error; + } + } + + private async runActiveRequestCompaction( + input: ActiveRequestCompactionInput, + attempt: CompactionAttemptPresentation, + ): Promise { const { turnId, state, queue, activeToolsForStep, abortSignal } = input; if (state.summarizerFailure) { - return { - decision: 'fail', - diagnosticReason: state.summarizerFailure, - }; + // The latch means no new provider attempt is made. Keep the first + // attempt's row and diagnostics; do not manufacture another lifecycle for + // every later over-window step. + return { decision: 'fail', diagnosticReason: state.summarizerFailure }; } + const fail = async (diagnosticReason: string): Promise => { + await attempt.fail(diagnosticReason); + return { decision: 'fail', diagnosticReason }; + }; const summarizer = this.input.summarizeHistoryCompact!; const midTurnTracker = this.createProviderRequestTracker({ turnId, @@ -1032,16 +1136,10 @@ export class AiSdkCompaction { // Exits: the boundary, an abort, a detached consumer, or a read failure. for (;;) { if (abortSignal?.aborted) { - return { - decision: 'fail', - diagnosticReason: 'ledger_wait_aborted', - }; + return fail('interrupted'); } if (queue.consumerDetached) { - return { - decision: 'fail', - diagnosticReason: 'ledger_wait_aborted', - }; + return fail('ledger_wait_aborted'); } if (state.flushedSteps >= input.minFlushedSteps && queue.consumedCount >= queue.pushedCount) break; @@ -1051,10 +1149,7 @@ export class AiSdkCompaction { try { turnLedger = await loadTurnRuntimeEvents(turnId); } catch { - return { - decision: 'fail', - diagnosticReason: 'ledger_read_failed', - }; + return fail('ledger_read_failed'); } const currentTurnEvents = turnLedger .filter((event) => event.turnId === turnId) @@ -1063,10 +1158,7 @@ export class AiSdkCompaction { // A fresh anchor belongs to the live ledger; after handoff it belongs to // the authenticated predecessor prefix. Neither path manufactures it. if (!orderedEvents.some((event) => event.id === state.headAnchor.id)) { - return { - decision: 'fail', - diagnosticReason: 'head_anchor_not_durable', - }; + return fail('head_anchor_not_durable'); } const memoryDecision = input.memoryCompactionDecision?.(); const plan = await planHistoryCompaction({ @@ -1098,6 +1190,10 @@ export class AiSdkCompaction { : {}), loadProjectionSnapshot: () => this.loadModelProjectionTransitions(), summarize: async ({ coveredRuntimeEvents, newlyFoldedRuntimeEvents, previousCheckpoint }) => { + // The visible attempt starts with the summarizer call. A stop that + // already landed is settled by the abort check after planning, and + // must not open a row only to close it as interrupted. + if (!abortSignal?.aborted) await attempt.open(); // Same contract as the standalone path: the planner hands the // effective (transition-folded) view to the summarizer and pins its // digest, so a summary can never quote a body a durable transition @@ -1118,6 +1214,11 @@ export class AiSdkCompaction { }, }); + // The planner may translate an aborted summarizer into fail_open. The + // owning signal distinguishes user cancellation from a provider failure, + // and must be checked before latching that failure or persisting a fold. + if (abortSignal?.aborted) return fail('interrupted'); + if (plan.decision === 'fail_open') { const diagnosticReason = plan.diagnosticReason ?? plan.reason; // Latch only a fail-open the summarizer actually produced. The baseline @@ -1135,10 +1236,7 @@ export class AiSdkCompaction { if (plan.reason === 'summarizer_failed') { state.summarizerFailure = diagnosticReason; } - return { - decision: 'fail', - diagnosticReason, - }; + return fail(diagnosticReason); } // Lifecycle order is validate → persist → apply, where validate is @@ -1154,10 +1252,7 @@ export class AiSdkCompaction { hasBlockingReplayDiagnostics(replayPlan) || (replayPlan.hasProviderNativeSemantics && !this.canReplayProviderNative(replayPlan)) ) { - return { - decision: 'fail', - diagnosticReason: 'replacement_unmaterializable', - }; + return fail('replacement_unmaterializable'); } const replacementMessages = await this.materializeRuntimeReplayPlan( replayPlan, @@ -1177,13 +1272,11 @@ export class AiSdkCompaction { // The replacement is valid: durably persist the checkpoint BEFORE // applying the projection — the same order as the pre_turn path. A // persistence failure keeps raw messages and records write_failed. + if (abortSignal?.aborted) return fail('interrupted'); try { await Promise.resolve(recorder(plan.checkpoint, turnId)); } catch { - return { - decision: 'fail', - diagnosticReason: 'write_failed', - }; + return fail('write_failed'); } if (memoryDecision?.dispatch && input.onMemoryCompaction) { try { @@ -1199,11 +1292,7 @@ export class AiSdkCompaction { // The boundary is durable the moment the projection takes over: record it // now, not at turn settlement, so usage readers ordering measurements // against the compaction get the apply time. - try { - await input.onCompactionApplied?.(); - } catch { - // The boundary row explains the compaction; losing it must not undo one. - } + await attempt.apply(); return { decision: 'compacted', checkpoint: plan.checkpoint, @@ -1238,8 +1327,7 @@ export class AiSdkCompaction { origin: ProviderRequestOrigin; memoryCompactionDecision?: () => AutomaticMemoryCompactionDecision; onMemoryCompaction?: (input: AutomaticMemoryCompactionDispatch) => void; - /** See {@link buildMidTurnCapacityCompactProjection}'s parameter. */ - onCompactionApplied?: () => void | Promise; + lifecycle: CompactionLifecycle; abortSignal?: AbortSignal; }): Promise<{ messages: ModelMessage[] } | undefined> { const state = input.midTurnState; @@ -1276,6 +1364,7 @@ export class AiSdkCompaction { } const phase = state.compactionPhase(input.stepNumber); + const attempt = bindCompactionAttempt(input.lifecycle); const outcome = await this.compactActiveRequestHistory({ turnId: input.turnId, phase, @@ -1288,7 +1377,7 @@ export class AiSdkCompaction { activeToolsForStep: input.activeTools, memoryCompactionDecision: input.memoryCompactionDecision, onMemoryCompaction: input.onMemoryCompaction, - onCompactionApplied: input.onCompactionApplied, + ...attempt, abortSignal: input.abortSignal, }); if (outcome.decision !== 'compacted') { diff --git a/packages/runtime/src/ai-sdk-turn.ts b/packages/runtime/src/ai-sdk-turn.ts index 34e1129579f..eb6963f8bd9 100644 --- a/packages/runtime/src/ai-sdk-turn.ts +++ b/packages/runtime/src/ai-sdk-turn.ts @@ -114,6 +114,7 @@ import { finitePositive } from './context-budget-helpers.js'; import type { AutomaticMemoryCompactionDecision, AutomaticMemoryCompactionDispatch, + CompactionLifecycle, ProviderImageBudget, } from './ai-sdk-compaction.js'; import { @@ -1080,7 +1081,22 @@ export class AiSdkTurn { // The apply-time boundary the usage reader orders measurements against: // recorded when each compaction lands, so it exists mid-turn and a stop or // stream error cannot erase it — unlike the settlement-time display note. - const recordCompactionApplied = async (): Promise => { + const recordCompactionStarted = async (compactionId: string): Promise => { + try { + await queue.pushAndWaitUntilConsumed({ + type: 'context_compaction_started', + id: this.deps.newId(), + turnId, + ts: this.deps.now(), + compactionId, + scope: 'step', + }); + } catch { + // The attempt may still run; losing the presentation row must not block + // the compaction or the request it gates. + } + }; + const recordCompactionApplied = async (compactionId: string): Promise => { // The consumer persists the hidden note and forwards the same boundary // to live subscribers. Await its ack before dispatching the replacement // prompt, so even a slow consumer observes the boundary before new usage. @@ -1089,8 +1105,34 @@ export class AiSdkTurn { id: this.deps.newId(), turnId, ts: this.deps.now(), + compactionId, + scope: 'step', }); }; + const recordCompactionFailed = async (compactionId: string, reason: string): Promise => { + try { + await queue.pushAndWaitUntilConsumed({ + type: 'context_compaction_failed', + id: this.deps.newId(), + turnId, + ts: this.deps.now(), + compactionId, + scope: 'step', + reason, + }); + // The event's mapper writes the durable failed_open note at the decision + // point. Keep the settlement fallback from adding the same failure again. + contextCompactionFailedOpenNoteWritten = true; + } catch { + // Leave the flag down so the settlement fallback can still record it. + } + }; + const compactionLifecycle: CompactionLifecycle = { + newCompactionId: () => this.deps.newId(), + onCompactionStarted: recordCompactionStarted, + onCompactionApplied: recordCompactionApplied, + onCompactionFailed: recordCompactionFailed, + }; const trace = new RunTrace({ sessionId: this.deps.backend.sessionId, turnId, @@ -1501,7 +1543,7 @@ export class AiSdkTurn { queue, capacityProviderTools, onMidTurnDiagnosticPatch, - recordCompactionApplied, + compactionLifecycle, this, this.automaticMemoryCompactionSupported() ? () => this.automaticMemoryCompactionDecision() @@ -2078,7 +2120,7 @@ export class AiSdkTurn { activeTools: activeToolsForRequest, queue, onDiagnosticPatch: onMidTurnDiagnosticPatch, - onCompactionApplied: recordCompactionApplied, + lifecycle: compactionLifecycle, origin: this, ...(this.automaticMemoryCompactionSupported() ? { diff --git a/packages/runtime/src/runtime-kernel.ts b/packages/runtime/src/runtime-kernel.ts index 1e3c77756cf..9b35612ed2b 100644 --- a/packages/runtime/src/runtime-kernel.ts +++ b/packages/runtime/src/runtime-kernel.ts @@ -1342,7 +1342,13 @@ export class RuntimeKernel implements RuntimeKernelLike { // a note queued behind one this compaction may already own would be // refused, and the reader would never learn the summary was skipped. if (result.outcome.kind === 'failed') { - await run.recordSystemNote('context_compaction_failed_open').catch(() => {}); + await run + .recordSystemNote('context_compaction_failed_open', { + compactionId: run.runId, + scope: 'turn', + failOpenReason: result.outcome.reason, + }) + .catch(() => {}); notedTerminal = true; } else if (result.outcome.kind === 'compacted') { // Explicit compaction runs on its own turn and never enters the @@ -1352,7 +1358,12 @@ export class RuntimeKernel implements RuntimeKernelLike { // duplicate. // The applied row is the apply-time boundary the usage reader orders // measurements against; the note below stays the display row. - await run.recordSystemNote('context_compaction_applied').catch(() => {}); + await run + .recordSystemNote('context_compaction_applied', { + compactionId: run.runId, + scope: 'turn', + }) + .catch(() => {}); await run.recordSystemNote('context_compacted').catch(() => {}); notedTerminal = true; } @@ -1377,7 +1388,12 @@ export class RuntimeKernel implements RuntimeKernelLike { // because the ledger seals on its terminal fact; the internal compaction // Turn has no user timeline for a failure banner. if (!notedTerminal && !run.isStopped()) { - await run.recordSystemNote('context_compaction_failed_open').catch(() => {}); + await run + .recordSystemNote('context_compaction_failed_open', { + compactionId: run.runId, + scope: 'turn', + }) + .catch(() => {}); } await run.recordFailure(error); throw error; diff --git a/packages/runtime/src/session-event-runtime-mapper.ts b/packages/runtime/src/session-event-runtime-mapper.ts index c4b66f90200..dcfe37a14a4 100644 --- a/packages/runtime/src/session-event-runtime-mapper.ts +++ b/packages/runtime/src/session-event-runtime-mapper.ts @@ -132,12 +132,14 @@ export function mapSessionEventToRuntimeEvent( ctx: RuntimeEventMapContext, memory: SessionEventMapMemory = createSessionEventMapMemory(), ): RuntimeEvent { + if (event.type === 'context_compaction_started' && event.scope !== 'step') { + throw new Error('turn-scoped context_compaction_started is not a backend event'); + } if (isHostProjectionSessionEvent(event)) { // These are Host/kernel projection facts, not backend events. The live // ingress drops them, so reaching this line bypassed that authority boundary. - // `context_compaction_started` is one of these: synthesized by the Runtime - // Host session projector for the renderer's live "compacting" row, never - // produced by a backend or the kernel. + // `context_compaction_started` is deliberately not here: a `step`-scoped + // start is a Runtime fact that must become a durable hidden note. throw new Error(`${event.type} is not a backend event`); } if (isLegacyPermissionSessionEvent(event)) { @@ -148,6 +150,7 @@ export function mapSessionEventToRuntimeEvent( } export function isLiveBackendSessionEvent(event: SessionEvent): event is BackendSessionEvent { + if (event.type === 'context_compaction_started' && event.scope !== 'step') return false; return !isHostProjectionSessionEvent(event) && !isLegacyPermissionSessionEvent(event); } @@ -157,7 +160,6 @@ function isHostProjectionSessionEvent(event: SessionEvent): event is Extract< type: | 'queue_update' | 'message_admission' - | 'context_compaction_started' | 'client_capability_request' | 'client_capability_decision_ack'; } @@ -165,7 +167,6 @@ function isHostProjectionSessionEvent(event: SessionEvent): event is Extract< return ( event.type === 'queue_update' || event.type === 'message_admission' || - event.type === 'context_compaction_started' || event.type === 'client_capability_request' || event.type === 'client_capability_decision_ack' ); @@ -489,13 +490,61 @@ function mapBackendSessionEvent( refs: { toolCallId: event.toolUseId }, }; + case 'context_compaction_started': + return { + ...base, + role: 'system', + author: 'system', + modelVisibility: 'hidden', + content: { + kind: 'system_note', + note: event.type, + ...(event.compactionId !== undefined || event.scope !== undefined + ? { + data: { + ...(event.compactionId !== undefined ? { compactionId: event.compactionId } : {}), + ...(event.scope !== undefined ? { scope: event.scope } : {}), + }, + } + : {}), + }, + }; + case 'context_compaction_applied': return { ...base, role: 'system', author: 'system', modelVisibility: 'hidden', - content: { kind: 'system_note', note: event.type }, + content: { + kind: 'system_note', + note: event.type, + ...(event.compactionId !== undefined || event.scope !== undefined + ? { + data: { + ...(event.compactionId !== undefined ? { compactionId: event.compactionId } : {}), + ...(event.scope !== undefined ? { scope: event.scope } : {}), + }, + } + : {}), + }, + }; + + case 'context_compaction_failed': + return { + ...base, + role: 'system', + author: 'system', + modelVisibility: 'hidden', + content: { + kind: 'system_note', + note: 'context_compaction_failed_open', + data: { + compactionId: event.compactionId, + scope: event.scope, + failOpenReason: event.reason, + }, + }, }; // ── Steering: a user message injected mid-turn at a step boundary ───── diff --git a/packages/ui/src/__tests__/live-turn-buffer.test.ts b/packages/ui/src/__tests__/live-turn-buffer.test.ts index 648c1de1b7a..d87f676c5d3 100644 --- a/packages/ui/src/__tests__/live-turn-buffer.test.ts +++ b/packages/ui/src/__tests__/live-turn-buffer.test.ts @@ -19,9 +19,65 @@ import assert from 'node:assert/strict'; import test from 'node:test'; +import type { StoredMessage } from '@maka/core/session'; import { applyLiveTurnBufferEvent, reconcileLiveTurnBuffer, type LiveTurnBuffer } from '../live-turn-buffer.js'; import { createTranscriptProjection } from '../transcript-projection.js'; +for (const outcome of ['compacted', 'failed', 'interrupted'] as const) { + test(`hands off a durable answer and ${outcome} compaction in one reconciliation`, () => { + let buffer = applyLiveTurnBufferEvent(undefined, { + type: 'context_compaction_started', id: 'start', turnId: 'a', ts: 1, + compactionId: 'compact-a', scope: 'step', + }, 'en'); + const reason = outcome === 'interrupted' ? 'interrupted' : 'write_failed'; + buffer = applyLiveTurnBufferEvent(buffer, outcome === 'compacted' ? { + type: 'context_compaction_applied', id: 'applied', turnId: 'a', ts: 2, + compactionId: 'compact-a', scope: 'step', + } : { + type: 'context_compaction_failed', id: 'failed', turnId: 'a', ts: 2, + compactionId: 'compact-a', scope: 'step', reason, + }, 'en'); + buffer = applyLiveTurnBufferEvent(buffer, { + type: 'text_delta', id: 'delta', turnId: 'a', messageId: 'answer-a', ts: 3, + text: 'Answer', startOffset: 0, + }, 'en'); + buffer = applyLiveTurnBufferEvent(buffer, { + type: 'complete', id: 'complete', turnId: 'a', ts: 4, + stopReason: outcome === 'interrupted' ? 'user_stop' : 'end_turn', + }, 'en'); + assert.ok(buffer); + + const messages: StoredMessage[] = [ + { + type: 'system_note', id: 'compact-terminal', turnId: 'a', ts: 2, + kind: outcome === 'compacted' ? 'context_compaction_applied' : 'context_compaction_failed_open', + data: { compactionId: 'compact-a', scope: 'step', ...(outcome === 'compacted' ? {} : { failOpenReason: reason }) }, + }, + { type: 'assistant', id: 'answer-a', turnId: 'a', ts: 3, text: 'Answer', modelId: 'test' }, + { type: 'turn_state', id: 'terminal-a', turnId: 'a', ts: 4, status: outcome === 'interrupted' ? 'aborted' : 'completed' }, + ]; + + // Answer durability alone must not erase a compaction outcome still missing from history. + const pending = reconcileLiveTurnBuffer(buffer, messages.filter((message) => message.type !== 'system_note')); + assert.equal(pending?.[0]?.steps.length, 0); + assert.equal(pending?.[0]?.compactions?.[0]?.state, outcome); + + assert.equal(reconcileLiveTurnBuffer(buffer, messages), undefined, + 'complete history must finish both handoffs without another refresh or React effect'); + assert.equal(reconcileLiveTurnBuffer(pending!, messages), undefined, + 'a later durable compaction must release an outcome retained during the answer handoff'); + + const withSuccessor = applyLiveTurnBufferEvent(buffer, { + type: 'text_delta', id: 'next-delta', turnId: 'b', messageId: 'answer-b', ts: 5, + text: 'Next answer', startOffset: 0, + }, 'en'); + const successor = withSuccessor?.find((turn) => turn.turnId === 'b'); + const reconciled = reconcileLiveTurnBuffer(withSuccessor!, messages); + assert.equal(reconciled?.length, 1); + assert.equal(reconciled?.[0], successor, 'releasing the predecessor must preserve the live successor'); + }); +} + test('a successor and a delayed predecessor seed retain separate text and settlement', () => { let buffer: LiveTurnBuffer | undefined; const text = (turnId: string, value: string, startOffset: number) => { diff --git a/packages/ui/src/__tests__/live-turn-projection.test.ts b/packages/ui/src/__tests__/live-turn-projection.test.ts index 30dcb5f0ed4..3ea952f27f6 100644 --- a/packages/ui/src/__tests__/live-turn-projection.test.ts +++ b/packages/ui/src/__tests__/live-turn-projection.test.ts @@ -1258,4 +1258,286 @@ describe('context-compaction live row', () => { [getConversationCopy('en').messages.systemNotes.contextCompacted], ); }); + + it('projects a step-scoped attempt into the turn timeline instead of the root row', () => { + let projection = applyLiveTurnEvent(undefined, { + type: 'context_compaction_started', + id: 'compaction-started-1', + turnId: 'turn-1', + ts: 10, + compactionId: 'compaction-1', + scope: 'step', + }); + assert.ok(projection); + assert.equal(projection.rootExecutionKind, undefined); + assert.deepEqual(projection.compactions, [ + { compactionId: 'compaction-1', state: 'running', startedAt: 10 }, + ]); + + projection = applyLiveTurnEvent(projection, { + type: 'context_compaction_applied', + id: 'compaction-applied-1', + turnId: 'turn-1', + ts: 20, + compactionId: 'compaction-1', + scope: 'step', + }); + assert.ok(projection); + assert.deepEqual(projection.compactions, [ + { compactionId: 'compaction-1', state: 'compacted', startedAt: 10 }, + ]); + + const turns = overlayLiveTurn([], projection, 'en'); + assert.equal(turns[0]?.notes.length, 0); + assert.equal(turns[0]?.timeline[0]?.kind, 'compaction'); + if (turns[0]?.timeline[0]?.kind !== 'compaction') return; + assert.equal(turns[0].timeline[0].compactionId, 'compaction-1'); + assert.equal(turns[0].timeline[0].state, 'compacted'); + assert.equal( + turns[0].timeline[0].text, + getConversationCopy('en').messages.systemNotes.contextCompacted, + ); + }); + + it('settles a failed step-scoped attempt and keeps multiple attempts by compactionId', () => { + let projection = applyLiveTurnEvent(undefined, { + type: 'context_compaction_started', + id: 'started-1', + turnId: 'turn-1', + ts: 10, + compactionId: 'compaction-1', + scope: 'step', + }); + projection = applyLiveTurnEvent(projection, { + type: 'context_compaction_started', + id: 'started-2', + turnId: 'turn-1', + ts: 30, + compactionId: 'compaction-2', + scope: 'step', + }); + projection = applyLiveTurnEvent(projection, { + type: 'context_compaction_failed', + id: 'failed-1', + turnId: 'turn-1', + ts: 20, + compactionId: 'compaction-1', + scope: 'step', + reason: 'write_failed', + }); + projection = applyLiveTurnEvent(projection, { + type: 'context_compaction_applied', + id: 'applied-2', + turnId: 'turn-1', + ts: 40, + compactionId: 'compaction-2', + scope: 'step', + }); + assert.deepEqual(projection?.compactions, [ + { compactionId: 'compaction-1', state: 'failed', startedAt: 10, reason: 'write_failed' }, + { compactionId: 'compaction-2', state: 'compacted', startedAt: 30 }, + ]); + + const turns = overlayLiveTurn([], projection, 'en'); + assert.deepEqual( + turns[0]?.timeline.map((item) => item.kind === 'compaction' ? item.state : item.kind), + ['failed', 'compacted'], + ); + }); + + it('records a cancelled attempt as interrupted instead of failed', () => { + let projection = applyLiveTurnEvent(undefined, { + type: 'context_compaction_started', + id: 'started-1', + turnId: 'turn-1', + ts: 10, + compactionId: 'compaction-1', + scope: 'step', + }); + projection = applyLiveTurnEvent(projection, { + type: 'context_compaction_failed', + id: 'failed-1', + turnId: 'turn-1', + ts: 20, + compactionId: 'compaction-1', + scope: 'step', + reason: 'interrupted', + }); + + assert.deepEqual(projection?.compactions, [ + { compactionId: 'compaction-1', state: 'interrupted', startedAt: 10 }, + ]); + }); + + it('retires a compaction-only live projection once the transcript is terminal', () => { + const projection = applyLiveTurnEvent(undefined, { + type: 'context_compaction_started', + id: 'started-1', + turnId: 'turn-1', + ts: 10, + compactionId: 'compaction-1', + scope: 'step', + }); + assert.ok(projection); + const messages: StoredMessage[] = [ + { + type: 'system_note', + id: 'started-note', + turnId: 'turn-1', + ts: 10, + kind: 'context_compaction_started', + data: { compactionId: 'compaction-1', scope: 'step' }, + }, + { type: 'turn_state', id: 'terminal', turnId: 'turn-1', ts: 11, status: 'completed' }, + ]; + assert.equal(reconcileTerminalLiveTurn(projection, messages), undefined); + }); + + it('settles a running live attempt from its durable terminal while the turn continues', () => { + let projection = applyLiveTurnEvent(undefined, { + type: 'context_compaction_started', + id: 'started-1', + turnId: 'turn-1', + ts: 10, + compactionId: 'compaction-1', + scope: 'step', + }); + projection = applyLiveTurnEvent(projection, { + type: 'context_compaction_started', + id: 'started-2', + turnId: 'turn-1', + ts: 30, + compactionId: 'compaction-2', + scope: 'step', + }); + assert.ok(projection); + const messages: StoredMessage[] = [ + { + type: 'user', + id: 'user-1', + turnId: 'turn-1', + ts: 1, + text: 'continue', + }, + { + type: 'system_note', + id: 'started-note-1', + turnId: 'turn-1', + ts: 10, + kind: 'context_compaction_started', + data: { compactionId: 'compaction-1', scope: 'step' }, + }, + { + type: 'system_note', + id: 'applied-note-1', + turnId: 'turn-1', + ts: 20, + kind: 'context_compaction_applied', + data: { compactionId: 'compaction-1', scope: 'step' }, + }, + { + type: 'system_note', + id: 'started-note-2', + turnId: 'turn-1', + ts: 30, + kind: 'context_compaction_started', + data: { compactionId: 'compaction-2', scope: 'step' }, + }, + { + type: 'system_note', + id: 'failed-note-2', + turnId: 'turn-1', + ts: 40, + kind: 'context_compaction_failed_open', + data: { compactionId: 'compaction-2', scope: 'step', failOpenReason: 'write_failed' }, + }, + ]; + + const reconciled = reconcileTerminalLiveTurn(projection, messages); + assert.deepEqual(reconciled?.compactions, [ + { compactionId: 'compaction-1', state: 'compacted', startedAt: 10 }, + { + compactionId: 'compaction-2', + state: 'failed', + startedAt: 30, + reason: 'write_failed', + }, + ]); + const turns = overlayLiveTurn(materializeTurns(messages, 'en'), reconciled, 'en'); + assert.deepEqual( + turns[0]?.timeline + .filter((item) => item.kind === 'compaction') + .map((item) => item.kind === 'compaction' ? item.state : undefined), + ['compacted', 'failed'], + ); + }); +}); + + +describe('compaction-only terminal handoff', () => { + it('does not discard compaction evidence when the last text step hands off', () => { + const projection: LiveTurnProjection = { + turnId: 't', terminal: true, + steps: [{ stepId: 'answer', tools: [], text: { text: 'done', complete: true, truncated: false } }], + compactions: [{ compactionId: 'c', state: 'interrupted', startedAt: 1 }], + }; + const settled = settleLiveTurnStep(projection, 'answer'); + assert.ok(settled); + assert.equal(settled.steps.length, 0); + const terminalOnly: StoredMessage[] = [ + { type: 'turn_state', id: 'end', turnId: 't', ts: 3, status: 'aborted' }, + ]; + assert.ok(reconcileTerminalLiveTurn(settled, terminalOnly)); + }); + + for (const type of ['abort', 'complete'] as const) { + it(`retains interruption through ${type} until history can render it`, () => { + const start = { type: 'context_compaction_started' as const, id: 's', turnId: 't', ts: 1, scope: 'step' as const, compactionId: 'c' }; + const live = applyLiveTurnEvent(undefined, start); + const terminal = applyLiveTurnEvent(live, type === 'complete' + ? { type, id: 'e', turnId: 't', ts: 2, stopReason: 'user_stop' } + : { type, id: 'e', turnId: 't', ts: 2, reason: 'user_stop' }); + assert.ok(terminal); + assert.equal(terminal.compactions?.[0]?.state, 'interrupted'); + assert.equal(terminal.compactions?.[0]?.reason, undefined); + assert.ok(reconcileTerminalLiveTurn(terminal, [])); + const history: StoredMessage[] = [{ type: 'system_note', id: 's', turnId: 't', ts: 1, kind: 'context_compaction_started', data: { scope: 'step', compactionId: 'c' } }]; + assert.ok(reconcileTerminalLiveTurn(terminal, history)); + history.push({ type: 'system_note', id: 'f', turnId: 't', ts: 2, kind: 'context_compaction_failed_open', data: { scope: 'step', compactionId: 'c', failOpenReason: 'interrupted' } }); + assert.equal(reconcileTerminalLiveTurn(terminal, history), undefined); + const row = materializeTurns(history, 'zh-CN')[0]?.timeline[0]; + assert.ok(row?.kind === 'compaction'); + assert.equal(row.text, '上下文压缩已中断。'); + }); + } + + it('settles a running attempt as failed when the turn errors', () => { + const start = { type: 'context_compaction_started' as const, id: 's', turnId: 't', ts: 1, scope: 'step' as const, compactionId: 'c' }; + const live = applyLiveTurnEvent(undefined, start); + const terminal = applyLiveTurnEvent(live, { + type: 'error', + id: 'e', + turnId: 't', + ts: 2, + message: 'provider failed', + recoverable: false, + }); + + assert.deepEqual(terminal?.compactions, [ + { compactionId: 'c', state: 'failed', startedAt: 1 }, + ]); + }); + + for (const outcome of ['compacted', 'write_failed'] as const) { + it(`preserves ${outcome} when the turn later stops`, () => { + const base = { id: 'e', turnId: 't', ts: 1, scope: 'step' as const, compactionId: 'c' }; + const live = applyLiveTurnEvent(undefined, outcome === 'compacted' + ? { ...base, type: 'context_compaction_applied' } + : { ...base, type: 'context_compaction_failed', reason: outcome }); + const terminal = applyLiveTurnEvent(live, { type: 'abort', id: 'a', turnId: 't', ts: 2, reason: 'user_stop' }); + assert.ok(terminal); + assert.equal(terminal.compactions?.[0]?.state, outcome === 'compacted' ? 'compacted' : 'failed'); + assert.equal(terminal.compactions?.[0]?.reason, outcome === 'compacted' ? undefined : outcome); + }); + } }); diff --git a/packages/ui/src/__tests__/materialize.test.ts b/packages/ui/src/__tests__/materialize.test.ts index 33cd49f53c4..b4cdb6b059e 100644 --- a/packages/ui/src/__tests__/materialize.test.ts +++ b/packages/ui/src/__tests__/materialize.test.ts @@ -280,6 +280,295 @@ describe("materializeTurns message metadata", () => { turns.flatMap((turn) => turn.notes.map((note) => note.id)), ["compacted"], ); + assert.equal(turns[0]?.timeline.some((item) => item.kind === "compaction"), false); + }); + + test("projects one in-turn compaction lifecycle into the execution timeline", () => { + const turns = materializeTurns([ + originalUser, + { + type: "system_note", + id: "started", + turnId: "t1", + ts: 2, + kind: "context_compaction_started", + data: { compactionId: "compact-1", scope: "step" }, + }, + { + type: "system_note", + id: "applied", + turnId: "t1", + ts: 3, + kind: "context_compaction_applied", + data: { compactionId: "compact-1", scope: "step" }, + }, + { + type: "system_note", + id: "compacted", + turnId: "t1", + ts: 4, + kind: "context_compacted", + }, + ], "en"); + const turn = turns[0]!; + assert.deepEqual(turn.notes, []); + const compactions = turn.timeline.filter( + (item): item is Extract => + item.kind === "compaction", + ); + assert.equal(compactions.length, 1); + assert.equal(compactions[0]?.compactionId, "compact-1"); + assert.equal(compactions[0]?.state, "compacted"); + assert.equal(compactions[0]?.ts, 2); + assert.equal(compactions[0]?.text, "Earlier context compacted."); + }); + + test("keeps a legacy applied row out of the timeline and leaves its settlement note", () => { + const automatic = materializeTurns([ + originalUser, + { + type: "system_note", + id: "applied", + turnId: "t1", + ts: 3, + kind: "context_compaction_applied", + }, + { + type: "system_note", + id: "compacted", + turnId: "t1", + ts: 4, + kind: "context_compacted", + }, + ], "en")[0]!; + assert.equal(automatic.timeline.some((item) => item.kind === "compaction"), false); + assert.deepEqual(automatic.notes.map((note) => note.id), ["compacted"]); + + const settlementOnly = materializeTurns([ + originalUser, + { + type: "system_note", + id: "compacted-only", + turnId: "t1", + ts: 4, + kind: "context_compacted", + }, + ], "en")[0]!; + assert.equal(settlementOnly.timeline.some((item) => item.kind === "compaction"), false); + assert.deepEqual(settlementOnly.notes.map((note) => note.id), ["compacted-only"]); + }); + + test("requires compactionId before a step row becomes a timeline attempt", () => { + const startedOnly = materializeTurns([ + originalUser, + { + type: "system_note", + id: "started-without-id", + turnId: "t1", + ts: 2, + kind: "context_compaction_started", + data: { scope: "step" }, + }, + { + type: "turn_state", + id: "state", + turnId: "t1", + ts: 3, + status: "completed", + }, + ], "en")[0]!; + assert.equal(startedOnly.timeline.some((item) => item.kind === "compaction"), false); + + const appliedOnly = materializeTurns([ + originalUser, + { + type: "system_note", + id: "applied-without-id", + turnId: "t1", + ts: 2, + kind: "context_compaction_applied", + data: { scope: "step" }, + }, + { + type: "system_note", + id: "compacted", + turnId: "t1", + ts: 3, + kind: "context_compacted", + }, + ], "en")[0]!; + assert.equal(appliedOnly.timeline.some((item) => item.kind === "compaction"), false); + assert.deepEqual(appliedOnly.notes.map((note) => note.id), ["compacted"]); + }); + + test("keeps a failed step-scoped attempt and suppresses its duplicate note", () => { + const turn = materializeTurns([ + originalUser, + { + type: "system_note", + id: "started", + turnId: "t1", + ts: 2, + kind: "context_compaction_started", + data: { compactionId: "compact-1", scope: "step" }, + }, + { + type: "system_note", + id: "failed", + turnId: "t1", + ts: 3, + kind: "context_compaction_failed_open", + data: { compactionId: "compact-1", scope: "step", failOpenReason: "write_failed" }, + }, + ], "en")[0]!; + assert.deepEqual(turn.notes, []); + const item = turn.timeline.find((entry) => entry.kind === "compaction"); + assert.equal(item?.kind, "compaction"); + if (item?.kind !== "compaction") return; + assert.equal(item.state, "failed"); + assert.equal(item.reason, "write_failed"); + }); + + test("reads a cancelled step-scoped attempt as interrupted", () => { + const turn = materializeTurns([ + originalUser, + { + type: "system_note", + id: "started", + turnId: "t1", + ts: 2, + kind: "context_compaction_started", + data: { compactionId: "compact-1", scope: "step" }, + }, + { + type: "system_note", + id: "interrupted", + turnId: "t1", + ts: 3, + kind: "context_compaction_failed_open", + data: { compactionId: "compact-1", scope: "step", failOpenReason: "interrupted" }, + }, + ], "en")[0]!; + assert.deepEqual(turn.notes, []); + const item = turn.timeline.find((entry) => entry.kind === "compaction"); + assert.equal(item?.kind, "compaction"); + if (item?.kind !== "compaction") return; + assert.equal(item.state, "interrupted"); + assert.equal(item.reason, undefined); + assert.equal(item.text, "Context compaction was interrupted."); + }); + + test("keeps legacy compaction rows as notes instead of timeline attempts", () => { + // Old sessions wrote neither an attempt id nor a step scope. Both the + // successful fold and the failure stay transcript notes; neither is an + // attempt that could collide with another fold in the same Turn. + const turn = materializeTurns([ + originalUser, + { + type: "system_note", + id: "applied", + turnId: "t1", + ts: 2, + kind: "context_compaction_applied", + }, + { + type: "system_note", + id: "compacted", + turnId: "t1", + ts: 3, + kind: "context_compacted", + }, + { + type: "system_note", + id: "failed-legacy", + turnId: "t1", + ts: 4, + kind: "context_compaction_failed_open", + data: { failOpenReason: "summarizer_failed" }, + }, + ], "en")[0]!; + const compactions = turn.timeline.filter((item) => item.kind === "compaction"); + assert.deepEqual(compactions, []); + assert.deepEqual( + turn.notes.map((note) => [note.id, note.systemNoteKind]), + [ + ["compacted", "context_compacted"], + ["failed-legacy", "context_compaction_failed_open"], + ], + ); + }); + + test("settles a running attempt as interrupted when the turn is already terminal", () => { + const turn = materializeTurns([ + originalUser, + { + type: "system_note", + id: "started", + turnId: "t1", + ts: 2, + kind: "context_compaction_started", + data: { compactionId: "compact-1", scope: "step" }, + }, + { + type: "turn_state", + id: "state", + turnId: "t1", + ts: 3, + status: "completed", + }, + ], "en")[0]!; + const item = turn.timeline.find((entry) => entry.kind === "compaction"); + assert.equal(item?.kind, "compaction"); + if (item?.kind !== "compaction") return; + assert.equal(item.state, "interrupted"); + assert.equal(item.reason, undefined); + assert.equal(item.text, "Context compaction was interrupted."); + }); + + test("settles a running attempt as failed when the recorded turn failed", () => { + const turn = materializeTurns([ + originalUser, + { + type: "system_note", + id: "started", + turnId: "t1", + ts: 2, + kind: "context_compaction_started", + data: { compactionId: "compact-1", scope: "step" }, + }, + { + type: "turn_state", + id: "state", + turnId: "t1", + ts: 3, + status: "failed", + }, + ], "en")[0]!; + const item = turn.timeline.find((entry) => entry.kind === "compaction"); + assert.equal(item?.kind, "compaction"); + if (item?.kind !== "compaction") return; + assert.equal(item.state, "failed"); + assert.equal(item.text, "Context compaction failed."); + }); + + test("keeps a durable running attempt while the transcript only infers turn completion", () => { + const turn = materializeTurns([ + originalUser, + { + type: "system_note", + id: "started", + turnId: "t1", + ts: 2, + kind: "context_compaction_started", + data: { compactionId: "compact-1", scope: "step" }, + }, + ], "en")[0]!; + const item = turn.timeline.find((entry) => entry.kind === "compaction"); + assert.equal(turn.statusSource, "inferred"); + assert.equal(item?.kind, "compaction"); + if (item?.kind !== "compaction") return; + assert.equal(item.state, "running"); + assert.equal(item.text, "Compacting context…"); }); test("localizes visible system notes", () => { @@ -303,6 +592,22 @@ describe("materializeTurns message metadata", () => { ); }); + test("labels a turn-scoped cancelled compaction as interrupted", () => { + const turn = materializeTurns([ + originalUser, + { + type: "system_note", + id: "cancelled", + turnId: "t1", + ts: 2, + kind: "context_compaction_failed_open", + data: { failOpenReason: "interrupted" }, + }, + ], "en")[0]!; + assert.equal(turn.notes[0]?.compactionState, "interrupted"); + assert.equal(turn.notes[0]?.text, "Context compaction was interrupted."); + }); + test("preserves an explicit empty reference projection as the new-format marker", () => { const messages: StoredMessage[] = [ { diff --git a/packages/ui/src/chat-turn.tsx b/packages/ui/src/chat-turn.tsx index 9fa12f96470..205dc722d63 100644 --- a/packages/ui/src/chat-turn.tsx +++ b/packages/ui/src/chat-turn.tsx @@ -1425,6 +1425,7 @@ const AssistantAnswerBubble = memo(function AssistantAnswerBubble(props: Assista // Semantic keys (no index) so mid-timeline inserts do not remount/collapse disclosures. function timelineEntryKey(item: TurnTimelineItem, index: number): string { if (item.kind === 'tools') return `tools-${item.items[0]?.toolUseId ?? index}`; + if (item.kind === 'compaction') return `compaction-${item.compactionId}`; return `${item.kind}-${item.messageId}`; } @@ -1439,6 +1440,14 @@ const TurnTimelineEntry = memo(function TurnTimelineEntry(props: { initialLiveContent?: ReadonlyMap; }) { const { item } = props; + if (item.kind === 'compaction') { + return ( + + ); + } if (item.kind === 'thinking') { return ( ; + activityObserved?: boolean; +}) { + const copy = getConversationCopy(useUiLocale()).messages; + const running = props.item.state === 'running'; + const observed = props.activityObserved !== false; + const text = running && !observed + ? copy.systemNotes.contextCompactionUnobserved + : props.item.text; + return ( + + + {running && observed && + + ); +} + /** * The turn's execution process (reasoning, intermediate commentary, tool * activity) under one disclosure. Live work stays expanded; a settled turn diff --git a/packages/ui/src/conversation-copy.ts b/packages/ui/src/conversation-copy.ts index 77dafb525bc..e6b64a0b978 100644 --- a/packages/ui/src/conversation-copy.ts +++ b/packages/ui/src/conversation-copy.ts @@ -322,6 +322,7 @@ export interface ConversationCopy { contextCompactionUnobserved: string; contextCompacted: string; contextCompactionFailedOpen: string; + contextCompactionInterrupted: string; contextWindowSuggestion: (tokens: number, declared: number | undefined) => string; contextWindowOverrun: (used: number, declared: number) => string; contextReportedWindowExceeded: (used: number, reported: number) => string; @@ -525,6 +526,7 @@ const conversationCopyZhCn: ConversationCopy = { contextCompactionUnobserved: '上下文压缩状态暂不可用', contextCompacted: '已压缩较早的上下文。', contextCompactionFailedOpen: '上下文压缩失败。', + contextCompactionInterrupted: '上下文压缩已中断。', contextWindowSuggestion: (tokens, declared) => declared === undefined ? `供应商拒绝了这次请求。该模型未声明上下文窗口；上次成功的用量约 ${tokens} tokens，可将窗口设为该值让 Maka 先行压缩。` @@ -649,6 +651,7 @@ const conversationCopyZhTw: ConversationCopy = { contextCompactionUnobserved: '上下文壓縮狀態暫不可用', contextCompacted: '已壓縮較早的上下文。', contextCompactionFailedOpen: '上下文壓縮失敗。', + contextCompactionInterrupted: '上下文壓縮已中斷。', contextWindowSuggestion: (tokens, declared) => declared === undefined ? `供應商拒絕了這次請求。該模型未宣告上下文視窗；上次成功的用量約 ${tokens} tokens，可將視窗設為該值讓 Maka 先行壓縮。` @@ -771,6 +774,7 @@ const conversationCopyEn: ConversationCopy = { contextCompactionUnobserved: 'Context compaction status unavailable', contextCompacted: 'Earlier context compacted.', contextCompactionFailedOpen: 'Context compaction failed.', + contextCompactionInterrupted: 'Context compaction was interrupted.', contextWindowSuggestion: (tokens, declared) => declared === undefined ? `The provider rejected this request. No context window is declared for this model; the last accepted usage was about ${tokens} tokens — set the window to that value so Maka compacts first.` diff --git a/packages/ui/src/live-turn-buffer.ts b/packages/ui/src/live-turn-buffer.ts index 4f571e90eec..af2d31f2233 100644 --- a/packages/ui/src/live-turn-buffer.ts +++ b/packages/ui/src/live-turn-buffer.ts @@ -61,6 +61,8 @@ export function reconcileLiveTurnBuffer(buffer: LiveTurnBuffer, messages: readon } } } - return retained; + // Text handoff may have removed the last step. Recheck durable coverage so + // a covered compaction does not need another refresh to release the Turn. + return retained?.terminal ? reconcileTerminalLiveTurn(retained, messages) : retained; }); } diff --git a/packages/ui/src/live-turn-projection.ts b/packages/ui/src/live-turn-projection.ts index a47a8068104..c6d020a9bce 100644 --- a/packages/ui/src/live-turn-projection.ts +++ b/packages/ui/src/live-turn-projection.ts @@ -31,6 +31,12 @@ import { applyAssistantComplete, applyAssistantDelta } from './assistant-stream. import { projectToolActivityArgs } from '@maka/core/tool-activity-args'; import { toolResultActivityStatus } from '@maka/core/tool-result-status'; import { isInFlightToolStatus } from '@maka/core/tool-result-status'; +import { + compactionStateForFailure, + compactionStateForTurnStatus, + readCompactionTimelineMarker, + type CompactionTimelineState, +} from '@maka/core/compaction-status'; import type { ToolActivityItem } from './materialize.js'; import { applyThinkingComplete, applyThinkingDelta } from './thinking-stream.js'; import type { StreamingDisplayRedactionState } from './streaming-display-redaction.js'; @@ -97,6 +103,14 @@ export interface LiveSteeringProjection { ts: number; } +export interface LiveCompactionProjection { + compactionId: string; + state: CompactionTimelineState; + startedAt: number; + /** Present only when `state === 'failed'`; cancellation uses `interrupted`. */ + reason?: string; +} + export interface LiveTurnProjection { turnId: string; terminal?: true; @@ -123,6 +137,8 @@ export interface LiveTurnProjection { */ unconfirmed?: true; providerRetry?: LiveProviderRetry; + /** In-Turn compaction attempts, keyed by compactionId. */ + compactions?: LiveCompactionProjection[]; steps: LiveTurnStepProjection[]; } @@ -154,6 +170,18 @@ function terminalizeLiveSteps(steps: readonly LiveTurnStepProjection[]): LiveTur })); } +function terminalizeLiveCompactions( + compactions: LiveCompactionProjection[] | undefined, + state: 'failed' | 'interrupted', +): LiveCompactionProjection[] | undefined { + if (!compactions || compactions.length === 0) return compactions; + return compactions.map((compaction) => { + if (compaction.state !== 'running') return compaction; + const { reason: _reason, ...withoutReason } = compaction; + return { ...withoutReason, state }; + }); +} + function terminalThinking(thinking: LiveThinkingProjection): LiveThinkingProjection { const { redactionState: _redactionState, ...safe } = thinking; return { ...safe, complete: true }; @@ -212,6 +240,34 @@ export function applyLiveTurnEvent( return { ...next, startedAt: startedAt ?? event.ts }; } +function updateLiveCompaction( + projection: LiveTurnProjection, + update: LiveCompactionProjection, +): LiveTurnProjection { + const existing = projection.compactions?.find( + (candidate) => candidate.compactionId === update.compactionId, + ); + if (existing) { + if (existing.state !== 'running') return projection; + return { + ...projection, + compactions: projection.compactions!.map((candidate) => + candidate.compactionId === update.compactionId + ? { + ...candidate, + state: update.state, + ...(update.reason !== undefined ? { reason: update.reason } : {}), + } + : candidate, + ), + }; + } + return { + ...projection, + compactions: [...(projection.compactions ?? []), update], + }; +} + function projectLiveTurnEvent( current: LiveTurnProjection | undefined, event: SessionEvent, @@ -250,18 +306,32 @@ function projectLiveTurnEvent( if (event.type === 'error' || event.type === 'abort') { if (!current || current.turnId !== event.turnId) return current; const steps = terminalizeLiveSteps(current.steps); - if (steps.length === 0) return undefined; + if (steps.length === 0 && !current.compactions?.length) return undefined; const { providerRetry: _providerRetry, ...withoutRetry } = confirmed(current); - return { ...withoutRetry, terminal: true, steps }; + const compactions = terminalizeLiveCompactions( + current.compactions, + event.type === 'abort' ? 'interrupted' : 'failed', + ); + return { + ...withoutRetry, + terminal: true, + steps, + ...(compactions !== undefined ? { compactions } : {}), + }; } if (event.type === 'complete') { if (!current || current.turnId !== event.turnId) return current; - if (current.steps.length === 0) return undefined; + if (current.steps.length === 0 && !current.compactions?.length) return undefined; const { providerRetry: _providerRetry, ...withoutRetry } = confirmed(current); + const compactions = terminalizeLiveCompactions( + current.compactions, + event.stopReason === 'error' ? 'failed' : 'interrupted', + ); return { ...withoutRetry, terminal: true, steps: terminalizeLiveSteps(current.steps), + ...(compactions !== undefined ? { compactions } : {}), }; } if (event.type === 'context_compaction_started') { @@ -269,8 +339,41 @@ function projectLiveTurnEvent( current?.turnId === event.turnId ? current : { turnId: event.turnId, steps: [] }; + if (event.scope === 'step') { + if (event.compactionId === undefined) return confirmed(prior); + return updateLiveCompaction(confirmed(prior), { + compactionId: event.compactionId, + state: 'running', + startedAt: event.ts, + }); + } return { ...confirmed(prior), rootExecutionKind: 'context_compact', startedAt: event.ts }; } + if (event.type === 'context_compaction_applied') { + if (event.scope !== 'step' || event.compactionId === undefined) return current; + const prior = + current?.turnId === event.turnId + ? current + : { turnId: event.turnId, steps: [] }; + return updateLiveCompaction(confirmed(prior), { + compactionId: event.compactionId, + state: 'compacted', + startedAt: event.ts, + }); + } + if (event.type === 'context_compaction_failed') { + const prior = + current?.turnId === event.turnId + ? current + : { turnId: event.turnId, steps: [] }; + const state = compactionStateForFailure(event.reason); + return updateLiveCompaction(confirmed(prior), { + compactionId: event.compactionId, + state, + startedAt: event.ts, + ...(state === 'failed' ? { reason: event.reason } : {}), + }); + } if ( event.type !== 'thinking_delta' && event.type !== 'thinking_complete' @@ -550,7 +653,7 @@ export function settleLiveTurnStep( }]; }); if (!found) return current; - if (steps.length === 0 && current.terminal) return undefined; + if (steps.length === 0 && current.terminal && !current.compactions?.length) return undefined; return { ...current, steps }; } @@ -584,6 +687,42 @@ function durableStreamEvidence( return false; } +function reconcileCompactionTerminals( + compactions: LiveCompactionProjection[] | undefined, + turnMessages: readonly StoredMessage[], +): LiveCompactionProjection[] | undefined { + if (!compactions || compactions.length === 0) return compactions; + const terminalById = new Map< + string, + { state: 'compacted' | 'failed' | 'interrupted'; reason?: string } + >(); + for (const message of turnMessages) { + const marker = readCompactionTimelineMarker(message); + if (!marker || marker.state === 'running') continue; + terminalById.set(marker.compactionId, { + state: marker.state, + ...(marker.reason !== undefined ? { reason: marker.reason } : {}), + }); + } + if (terminalById.size === 0) return compactions; + let changed = false; + const next = compactions.map((compaction) => { + const terminal = terminalById.get(compaction.compactionId); + if (!terminal) return compaction; + if (compaction.state === terminal.state && compaction.reason === terminal.reason) { + return compaction; + } + changed = true; + const { reason: _reason, ...withoutReason } = compaction; + return { + ...withoutReason, + state: terminal.state, + ...(terminal.reason !== undefined ? { reason: terminal.reason } : {}), + }; + }); + return changed ? next : compactions; +} + /** * Removes evidence-only steps once the persisted transcript can render the * same durable output, including while a later step is still running. Text @@ -595,16 +734,34 @@ export function reconcileTerminalLiveTurn( messages: readonly StoredMessage[], ): LiveTurnProjection | undefined { const turnMessages = messages.filter((message) => message.turnId === current.turnId); - const transcriptReachedTerminal = turnMessages.some( + const terminalTurn = turnMessages.find( (message) => message.type === 'turn_state' && message.status !== 'running', ); + const transcriptReachedTerminal = terminalTurn !== undefined; + const terminalStatus = terminalTurn?.type === 'turn_state' + && terminalTurn.status !== 'running' + ? terminalTurn.status + : undefined; let projection = current; - if (transcriptReachedTerminal && current.terminal !== true) { + const compactions = reconcileCompactionTerminals(projection.compactions, turnMessages); + if (compactions !== projection.compactions) { + projection = { ...projection, compactions }; + } + if ( + transcriptReachedTerminal + && terminalStatus !== undefined + && current.terminal !== true + ) { const { providerRetry: _providerRetry, ...withoutRetry } = confirmed(current); + const terminalizedCompactions = terminalizeLiveCompactions( + projection.compactions, + compactionStateForTurnStatus(terminalStatus), + ); projection = { ...withoutRetry, terminal: true, steps: terminalizeLiveSteps(current.steps), + ...(terminalizedCompactions !== undefined ? { compactions: terminalizedCompactions } : {}), }; } const userIds = new Set(turnMessages.flatMap((message) => message.type === 'user' ? [message.id] : [])); @@ -627,13 +784,34 @@ export function reconcileTerminalLiveTurn( }); return !toolsCovered; }); + // Terminal live content stays until history can render the same outcome. + // A durable start plus a terminal Turn is settled from the Turn status; a + // start alone must not replace the known terminal state with a spinner. + const durableCompactions = new Map( + turnMessages.flatMap((message) => { + const marker = readCompactionTimelineMarker(message); + return marker ? [[marker.compactionId, marker] as const] : []; + }), + ); + const compactionsCovered = (projection.compactions ?? []).every((compaction) => { + const marker = durableCompactions.get(compaction.compactionId); + if (!marker) return false; + if (marker.state === 'running') { + return transcriptReachedTerminal + && terminalStatus !== undefined + && compaction.state === compactionStateForTurnStatus(terminalStatus); + } + return marker.state === compaction.state && marker.reason === compaction.reason; + }); if ( steps.length === 0 + && compactionsCovered && projection.terminal && ( projection.rootExecutionKind === 'context_compact' || transcriptReachedTerminal || steps.length !== projection.steps.length + || Boolean(projection.compactions?.length) ) ) return undefined; if (steps.length === projection.steps.length) return projection; diff --git a/packages/ui/src/materialize.ts b/packages/ui/src/materialize.ts index d424bb83026..8ef0b5f7cc0 100644 --- a/packages/ui/src/materialize.ts +++ b/packages/ui/src/materialize.ts @@ -39,9 +39,24 @@ import type { } from '@maka/core/events'; import type { ToolActivityStatus } from '@maka/core/tool-result-status'; import type { ShellRunToolResult } from '@maka/core/shell-run-result'; -import type { StoredMessage, TurnRecord, TurnStatus, UserMessage } from '@maka/core/session'; +import { + compactionStateForFailure, + compactionStateForTurnStatus, + readCompactionFailureReason, + readCompactionTimelineMarker, + type CompactionTimelineState, + type CompactionTimelineMarker, +} from '@maka/core/compaction-status'; +import type { + StoredMessage, + SystemNoteKind, + TurnRecord, + TurnStatus, + UserMessage, +} from '@maka/core/session'; import type { UiLocale } from '@maka/core/ui-locale'; import type { + LiveCompactionProjection, LiveSteeringProjection, LiveTurnProjection, } from "./live-turn-projection.js"; @@ -50,7 +65,9 @@ import { getConversationCopy } from "./conversation-copy.js"; export { isCancelledToolResultContent, isInFlightToolStatus, toolResultActivityStatus } from '@maka/core/tool-result-status'; export interface ChatItem { - compactionState?: "running" | "compacted" | "failed"; + compactionState?: CompactionTimelineState; + /** Source kind for runtime-note rows; lets timeline projection suppress its own duplicate. */ + systemNoteKind?: SystemNoteKind; id: string; role: "user" | "assistant" | "system"; text: string; @@ -145,7 +162,11 @@ export interface ToolActivityItem { function systemNoteLabel(kind: string, data: unknown, locale: UiLocale): string { const copy = getConversationCopy(locale).messages.systemNotes; if (kind === "context_compacted") return copy.contextCompacted; - if (kind === "context_compaction_failed_open") return copy.contextCompactionFailedOpen; + if (kind === "context_compaction_failed_open") { + return compactionStateForFailure(readCompactionFailureReason(data)) === "interrupted" + ? copy.contextCompactionInterrupted + : copy.contextCompactionFailedOpen; + } if (kind === "context_overflow_after_compaction") return copy.contextOverflowAfterCompaction; if (kind === "context_reported_window_exceeded") { const exceeded = data as @@ -183,6 +204,17 @@ function systemNoteLabel(kind: string, data: unknown, locale: UiLocale): string return kind; } +function compactionTimelineText( + locale: UiLocale, + state: CompactionTimelineState, +): string { + const notes = getConversationCopy(locale).messages.systemNotes; + if (state === "running") return notes.contextCompacting; + if (state === "compacted") return notes.contextCompacted; + if (state === "interrupted") return notes.contextCompactionInterrupted; + return notes.contextCompactionFailedOpen; +} + export function materializeTools( messages: readonly StoredMessage[], ): ToolActivityItem[] { @@ -350,7 +382,16 @@ export type TurnTimelineItem = complete?: boolean; truncated?: boolean; } - | { kind: "tools"; items: ToolActivityItem[] }; + | { kind: "tools"; items: ToolActivityItem[] } + | { + kind: "compaction"; + compactionId: string; + state: CompactionTimelineState; + text: string; + ts: number; + /** Present only when `state === 'failed'`; cancellation uses `interrupted`. */ + reason?: string; + }; /** * A single conversational turn — typically one user message, the assistant's @@ -458,14 +499,22 @@ export function overlayLiveTurn( } satisfies TurnViewModel, ]; } - if (targetIndex >= 0 && liveTurn.steps.length === 0) { + if ( + targetIndex >= 0 && + liveTurn.steps.length === 0 && + (liveTurn.compactions?.length ?? 0) === 0 + ) { return turns; } // A send arm is only a presentation claim that the next message may still // arrive. It is not a Turn record and must not manufacture one while the // canonical transcript is catching up. A real live step (or steering // message) is sufficient evidence to project a missing external Turn. - if (targetIndex < 0 && liveTurn.steps.length === 0) { + if ( + targetIndex < 0 && + liveTurn.steps.length === 0 && + (liveTurn.compactions?.length ?? 0) === 0 + ) { return turns; } const current = @@ -505,6 +554,9 @@ export function overlayLiveTurn( ); } } + for (const compaction of liveTurn.compactions ?? []) { + liveContentKeys.add(`compaction\0${compaction.compactionId}`); + } const liveTimeline: TurnTimelineItem[] = []; const liveItemTs: number[] = []; const pushLive = (item: TurnTimelineItem, ts: number | undefined): void => { @@ -518,7 +570,30 @@ export function overlayLiveTurn( messageId: message.id, }, message.ts); }; + const pendingCompactions = [...(liveTurn.compactions ?? [])].sort( + (left, right) => left.startedAt - right.startedAt, + ); + let nextCompaction = 0; + const pushCompactionsThrough = (ts: number | undefined): void => { + while (nextCompaction < pendingCompactions.length) { + const compaction = pendingCompactions[nextCompaction]!; + if (ts !== undefined && compaction.startedAt > ts) return; + pushLive( + { + kind: "compaction", + compactionId: compaction.compactionId, + state: compaction.state, + text: compactionTimelineText(locale, compaction.state), + ts: compaction.startedAt, + ...(compaction.reason !== undefined ? { reason: compaction.reason } : {}), + }, + compaction.startedAt, + ); + nextCompaction += 1; + } + }; for (const step of liveTurn.steps) { + pushCompactionsThrough(step.startedAt); if (step.steering !== undefined) appendLiveSteering(step.steering); const contentOrder = step.contentOrder ?? [ ...(step.thinking ? ["thinking" as const] : []), @@ -554,6 +629,7 @@ export function overlayLiveTurn( } } } + pushCompactionsThrough(undefined); // Shared entries are handoff points: replace them in place while preserving // live production order. Appending all live content after settled rows moved // an earlier answer (and its steering anchor) behind later persisted steps. @@ -786,7 +862,12 @@ export function materializeTurns( id: message.id, role: "system", text: systemNoteLabel(message.kind, message.data, locale), - compactionState: message.kind === "context_compacted" ? "compacted" : message.kind === "context_compaction_failed_open" ? "failed" : undefined, + compactionState: message.kind === "context_compacted" + ? "compacted" + : message.kind === "context_compaction_failed_open" + ? compactionStateForFailure(readCompactionFailureReason(message.data)) + : undefined, + systemNoteKind: message.kind, ts: message.ts, }); } else if (message.type === "token_usage") { @@ -837,8 +918,47 @@ export function materializeTurns( turn.timeline = buildTurnTimeline( turnMessages, toolItemsByTurnId.get(turnId) ?? new Map(), + locale, ); + if (turn.statusSource === "recorded" && turn.status !== "running") { + const terminalState = compactionStateForTurnStatus(turn.status); + turn.timeline = turn.timeline.map((item) => { + if (item.kind !== "compaction" || item.state !== "running") return item; + const { reason: _reason, ...withoutReason } = item; + return { + ...withoutReason, + state: terminalState, + text: compactionTimelineText(locale, terminalState), + }; + }); + } turn.tools = timelineTools(turn.timeline); + if (turn.timeline.some((item) => item.kind === "compaction")) { + // The execution item is the single display row for an in-Turn fold. Keep + // the durable notes available to other clients, but do not render the + // same fold again above the timeline. Match by attempt: a settlement-only + // failure has no step-scoped marker of its own, so a successful fold in + // the same Turn must not take that independent failure note down with it. + const timelineNoteIds = new Set(); + for (const message of turnMessages) { + if ( + message.type === "system_note" && + readCompactionTimelineMarker(message) + ) { + timelineNoteIds.add(message.id); + } + } + const hasCompactedTimelineItem = turn.timeline.some( + (item) => item.kind === "compaction" && item.state === "compacted", + ); + turn.notes = turn.notes.filter((note) => { + if (note.systemNoteKind === "context_compacted") return !hasCompactedTimelineItem; + if (note.systemNoteKind === "context_compaction_failed_open") { + return !timelineNoteIds.has(note.id); + } + return true; + }); + } } return order.map((turnId) => byId.get(turnId)!); @@ -1022,13 +1142,44 @@ export function projectTurnTools( function buildTurnTimeline( turnMessages: readonly StoredMessage[], toolItemByUseId: ReadonlyMap, + locale: UiLocale, ): TurnTimelineItem[] { const raw: TurnTimelineItem[] = []; let pending: ToolActivityItem[] = []; let sawRootUser = false; + const compactionIndexById = new Map(); const flushTools = (items: ToolActivityItem[]): void => { if (items.length > 0) raw.push({ kind: "tools", items }); }; + const upsertCompaction = (marker: CompactionTimelineMarker): void => { + const index = compactionIndexById.get(marker.compactionId); + const existing = index === undefined ? undefined : raw[index]; + if (index !== undefined && existing?.kind === "compaction") { + // A terminal state never downgrades back to running on a replayed seed. + if (existing.state !== "running") return; + raw[index] = { + ...existing, + state: marker.state, + text: compactionTimelineText(locale, marker.state), + ...(marker.reason !== undefined ? { reason: marker.reason } : {}), + }; + return; + } + // Compaction lands at a request boundary: everything buffered before it is + // earlier tool work and must keep its place ahead of the compaction row. + flushTools(pending); + pending = []; + const item: Extract = { + kind: "compaction", + compactionId: marker.compactionId, + state: marker.state, + text: compactionTimelineText(locale, marker.state), + ts: marker.ts, + ...(marker.reason !== undefined ? { reason: marker.reason } : {}), + }; + compactionIndexById.set(marker.compactionId, raw.length); + raw.push(item); + }; for (const message of turnMessages) { if (message.type === "user") { if (message.steeringEventId === undefined && !sawRootUser) { @@ -1115,6 +1266,9 @@ function buildTurnTimeline( } flushTools(matched); } + } else if (message.type === "system_note") { + const marker = readCompactionTimelineMarker(message); + if (marker) upsertCompaction(marker); } } flushTools(pending); @@ -1189,7 +1343,9 @@ function projectImportedCodexQuotes(content: MessageContent): Pick { if (buffer.length === 0) return; // Imported transcripts can record reasoning after the visible reply. - // Ignore that trailing reasoning when locating the answer, but stop at - // tool activity: text before tools is still process commentary. - const replyIndex = buffer.findLastIndex((item) => item.kind !== 'thinking'); + // A compaction row is execution evidence too: it may settle after the + // answer, so it must not become the answer candidate. Stop at tool + // activity: text before tools is still process commentary. + const replyIndex = buffer.findLastIndex( + (item) => item.kind !== 'thinking' && item.kind !== 'compaction', + ); const candidate = buffer[replyIndex]; const answer = candidate?.kind === 'text' ? candidate : undefined; if (answer) buffer.splice(replyIndex, 1);