fix(transcript): recover from oversized invocations

apps/desktop/src/main/__tests__/runtime-host-desktop-candidate.test.ts

@@ -1034,6 +1034,68 @@ test('keeps a restored observation retryable until replacement seeding succeeds'
   assert.equal(pendingAt >= 0 && readyAt > pendingAt, true);
   await replacementCandidate.close();
 }); // A failed replacement leaves the registry available to the next candidate.
+test('keeps replacement Host ready when one transcript observation cannot seed', async () => {
+  const observations = new RuntimeHostSessionObservationRegistry();
+  const firstIpc = ipcHarness();
+  const firstHost = connectionHarness('transcript-source', {
+    sessionId: 'session-1',
+  });
+  const firstCandidate = await createDesktopRuntimeHostCandidate(
+    firstHost.connection,
+    deps(firstIpc),
+    observations,
+  );
+  await firstIpc.invoke('sessions:observe', 'session-1', 'observer-1');
+  await firstIpc.invoke('sessions:observe', 'session-2', 'observer-2');
+  await firstCandidate.close();
+
+  const events: Array<{ channel: string; payload: unknown }> = [];
+  const failingHost = connectionHarness('transcript-seed-failure', {
+    sessionId: 'session-1',
+    subscribeFailure: new RuntimeHostOperationError(
+      'subscription.open',
+      'transcript_unavailable',
+      'Session transcript is unavailable',
+    ),
+    subscribeFailureSessionId: 'session-1',
+    subscriptionSnapshots: {
+      'session-2': continuitySnapshot({
+        session: { ...continuitySnapshot().session, sessionId: 'session-2' },
+        rootTurn: {
+          sessionId: 'session-2',
+          turnId: 'turn-2',
+          runId: 'run-2',
+          status: 'running',
+        },
+      }),
+    },
+  });
+  const candidate = await createDesktopRuntimeHostCandidate(
+    failingHost.connection,
+    {
+      ...deps(ipcHarness()),
+      renderer: {
+        send(channel, _host, payload) {
+          events.push({ channel, payload });
+        },
+      },
+    },
+    observations,
+  );
+  assert.deepEqual(observations.observationSessionIds(), ['session-1', 'session-2']);
+  await waitFor(() => events.some(({ channel, payload }) =>
+    channel === 'sessions:event:session-1' &&
+    (payload as { type?: string }).type === 'host_observation_error'));
+  assert.ok(events.some(({ channel, payload }) =>
+    channel === 'sessions:active-interactions-changed' &&
+    (payload as { sessionId?: string }).sessionId === 'session-2'));
+  assert.ok(!events.some(({ channel, payload }) =>
+    channel === 'sessions:event:session-2' &&
+    (payload as { type?: string }).type === 'host_observation_error'));
+  await candidate.close();
+  await observations.close();
+});
+
 test('drops a stale shared Session observation when Guest access is gone', async () => {
   const observations = new RuntimeHostSessionObservationRegistry();
   const firstIpc = ipcHarness();
@@ -1259,8 +1321,10 @@ function connectionHarness(
     sessionId?: string;
     revisionAbandon?: 'abandoned' | 'retained';
     subscriptionSnapshot?: SessionContinuitySnapshot;
+    subscriptionSnapshots?: Record<string, SessionContinuitySnapshot>;
     assistantStreams?: readonly SessionAssistantStreamIdentity[];
     subscribeFailure?: Error;
+    subscribeFailureSessionId?: string;
     runtimeResourcePty?: ReturnType<typeof ptySnapshot>;
     runtimeResourceUpdate?: ShellRunUpdate;
     sharedSessionAvailable?: boolean;
@@ -1402,7 +1466,9 @@ function connectionHarness(
       throw new Error(`Unexpected operation: ${operation}`);
     },
     openSessionSubscription: async ({ sessionId }: { sessionId: string }) => {
-      if (options.subscribeFailure) throw options.subscribeFailure;
+      if (options.subscribeFailure &&
+        (!options.subscribeFailureSessionId || options.subscribeFailureSessionId === sessionId))
+        throw options.subscribeFailure;
       const subscriptionFrames = new AsyncFrameQueue();
       activeSubscriptionFrames = subscriptionFrames;
       // The Host holds a subscription's frames until the subscriber calls
@@ -1435,7 +1501,7 @@ function connectionHarness(
           ptyListeners.add(listener);
           return () => ptyListeners.delete(listener);
         },
-        snapshot: options.subscriptionSnapshot ?? {
+        snapshot: options.subscriptionSnapshots?.[sessionId] ?? options.subscriptionSnapshot ?? {
           projectionRevision: 1,
           session: { sessionId },
         },

apps/desktop/src/main/runtime-host-desktop-candidate.ts

@@ -420,6 +420,7 @@ async function restoreSessionObservations(input: {
   sessionIds(): string[];
   announcePending(sessionId: string): void;
   attach(): Promise<string[]>;
+  isRecoverable?(sessionId: string): boolean;
 }): Promise<string[]> {
   const requested = input.sessionIds();
   for (const sessionId of requested) input.announcePending(sessionId);
@@ -428,7 +429,7 @@ async function restoreSessionObservations(input: {
   const restoredSet = new Set(restored);
   const registeredSet = new Set(input.sessionIds());
   const failed = requested.filter(
-    (sessionId) => registeredSet.has(sessionId) && !restoredSet.has(sessionId),
+    (sessionId) => registeredSet.has(sessionId) && !restoredSet.has(sessionId) && !input.isRecoverable?.(sessionId),
   );
   if (failed.length > 0) {
     throw new Error(`Failed to restore Session observations: ${failed.join(', ')}`);
@@ -768,10 +769,12 @@ export async function createDesktopRuntimeHostCandidate(
       }
     }
     observationsAttached = Boolean(sessionObserver);
+    const transcriptSeedFailures = new Set<string>();
     const restoredSessionIds = await restoreSessionObservations({
       sessionIds: () => sessionObservations.observationSessionIds(),
       announcePending: (sessionId) =>
         sendToRenderer(`sessions:event:${sessionId}`, { type: 'host_observation_pending' }),
+      isRecoverable: (sessionId) => transcriptSeedFailures.has(sessionId),
       attach: () => sessionObservations.attach(
         sessionObserver,
         (target) => ({
@@ -786,6 +789,13 @@ export async function createDesktopRuntimeHostCandidate(
           off: target.off.bind(target),
         }),
         (missingSessionId) => emitSessionsChanged("deleted", missingSessionId),
+        (failedSessionId) => {
+          transcriptSeedFailures.add(failedSessionId);
+          sendToRenderer(`sessions:event:${failedSessionId}`, {
+            type: 'host_observation_error',
+            message: 'Session transcript is unavailable',
+          });
+        },
       ),
     });
     for (const sessionId of restoredSessionIds) {

apps/desktop/src/main/runtime-host-session-observation-registry.ts

@@ -99,6 +99,14 @@ function isMissingRuntimeHostSessionError(error: unknown): boolean {
   return error.code === "not_found";
 }
 
+function isTranscriptSeedFailure(error: unknown): boolean {
+  return (
+    error instanceof RuntimeHostOperationError &&
+    error.operation === 'subscription.open' &&
+    error.code === 'transcript_unavailable'
+  );
+}
+
 interface SessionObservationRegistration {
   readonly sessionId: string;
   readonly messageAdmissions: boolean;
@@ -175,6 +183,7 @@ export class RuntimeHostSessionObservationRegistry {
     source: SessionObservationSource,
     bindTarget: ObservationTargetBinding = (target) => target,
     onSessionMissing?: (sessionId: string) => void,
+    onTranscriptSeedFailure?: (sessionId: string) => void,
   ): Promise<string[]> {
     this.#assertOpen();
     if (this.#source && this.#source !== source) {
@@ -224,6 +233,9 @@ export class RuntimeHostSessionObservationRegistry {
             if (registration.lifecycle === "pending") {
               this.#deleteRegistration(observerId, registration);
             }
+            if (isTranscriptSeedFailure(error)) {
+              onTranscriptSeedFailure?.(registration.sessionId);
+            }
             this.#onError(error);
           }
           return undefined;

packages/cli/src/pi-transcript.ts

@@ -32,7 +32,7 @@ import type {
 } from '@maka/core/events';
 import {
   deriveTurnRecords,
-  isRuntimeSystemNoteKind,
+  isUserVisibleSessionSystemNote,
   STEP_LIMIT_NOTICE_TEXT,
   type StoredMessage,
   type SystemNoteMessage,
@@ -1348,7 +1348,7 @@ function tokenDelta(before: number | undefined, after: number | undefined): numb
 function systemNoteText(message: SystemNoteMessage): string | undefined {
   // Retired kinds are still decoded off legacy transcript rows, and none of
   // them ever had a line here worth reading.
-  if (!isRuntimeSystemNoteKind(message.kind)) return undefined;
+  if (!isUserVisibleSessionSystemNote(message.kind)) return undefined;
   switch (message.kind) {
     case 'context_compacted':
       return 'Context compacted to keep this task within the model window.';
@@ -1395,6 +1395,8 @@ function systemNoteText(message: SystemNoteMessage): string | undefined {
     }
     case 'step_limit':
       return STEP_LIMIT_NOTICE_TEXT;
+    case 'transcript_omitted':
+      return 'Part of this session transcript was omitted because it exceeded the display limit.';
   }
 }
 

packages/core/src/__tests__/runtime-event.test.ts

@@ -38,7 +38,12 @@ import {
   type RuntimeEvent,
   type RuntimeEventActions,
 } from '../runtime-event.js';
-import { decodeCanonicalMessage, isRuntimeSystemNoteKind } from '../session.js';
+import {
+  decodeCanonicalMessage,
+  isRuntimeSystemNoteKind,
+  isSystemNoteKind,
+  isUserVisibleSessionSystemNote,
+} from '../session.js';
 import { decodeTurnOrigin } from '../turn-origin.js';
 
 /** Minimal valid RuntimeEvent; callers spread overrides on top. */
@@ -146,6 +151,12 @@ test('decodes a released provider dropping note that nothing writes any more', (
   assert.equal(isRuntimeSystemNoteKind('context_provider_dropping'), false);
 });
 
+test('keeps reader omission notices outside the runtime-written note kinds', () => {
+  assert.equal(isSystemNoteKind('transcript_omitted'), true);
+  assert.equal(isUserVisibleSessionSystemNote('transcript_omitted'), true);
+  assert.equal(isRuntimeSystemNoteKind('transcript_omitted'), false);
+});
+
 test('shares one decoder across all TurnOrigin variants', () => {
   const origins = [
     { kind: 'scheduled_task', scheduledTaskId: 'task-1' },

packages/core/src/session.ts

@@ -832,10 +832,10 @@ export function userFacingText(message: Pick<UserMessage, 'text' | 'displayText'
 
 /**
  * Closed policy for system notes that are part of the user-visible transcript:
- * exactly the notes the runtime writes.
+ * runtime notes plus the reader's explicit omitted-transcript placeholder.
  */
 export function isUserVisibleSessionSystemNote(kind: string): boolean {
-  return isRuntimeSystemNoteKind(kind);
+  return isRuntimeSystemNoteKind(kind) || kind === 'transcript_omitted';
 }
 
 /**
@@ -1289,6 +1289,9 @@ export const RUNTIME_SYSTEM_NOTE_KINDS = [
   'step_limit',
 ] as const;
 
+/** Reader-generated notices are never written back into a RuntimeEvent ledger. */
+export const SYNTHETIC_SYSTEM_NOTE_KINDS = ['transcript_omitted'] as const;
+
 /**
  * Notes nothing writes any more, still decoded so old transcripts and run
  * ledgers stay readable, and never shown. The session-level ones are owned by
@@ -1305,15 +1308,20 @@ export const RETIRED_SYSTEM_NOTE_KINDS = [
 ] as const;
 
 export type RuntimeSystemNoteKind = (typeof RUNTIME_SYSTEM_NOTE_KINDS)[number];
-export type SystemNoteKind = RuntimeSystemNoteKind | (typeof RETIRED_SYSTEM_NOTE_KINDS)[number];
+export type SystemNoteKind =
+  | RuntimeSystemNoteKind
+  | (typeof SYNTHETIC_SYSTEM_NOTE_KINDS)[number]
+  | (typeof RETIRED_SYSTEM_NOTE_KINDS)[number];
 
 export function isRuntimeSystemNoteKind(kind: string): kind is RuntimeSystemNoteKind {
   return (RUNTIME_SYSTEM_NOTE_KINDS as readonly string[]).includes(kind);
 }
 
 export function isSystemNoteKind(kind: string): kind is SystemNoteKind {
   return (
-    isRuntimeSystemNoteKind(kind) || (RETIRED_SYSTEM_NOTE_KINDS as readonly string[]).includes(kind)
+    isRuntimeSystemNoteKind(kind) ||
+    (SYNTHETIC_SYSTEM_NOTE_KINDS as readonly string[]).includes(kind) ||
+    (RETIRED_SYSTEM_NOTE_KINDS as readonly string[]).includes(kind)
   );
 }
 

packages/runtime-host/src/__tests__/root-turn-coordinator.test.ts

@@ -34,7 +34,10 @@ import {
 } from '@maka/core/runtime-logical-execution';
 import { runtimeInvocationFailureClass } from '@maka/runtime/runtime-event-read-model';
 import { randomUUID } from 'node:crypto';
-import { createSessionTranscriptReader } from '../server/session-transcript-reader.js';
+import {
+  createSessionTranscriptReader,
+  OmittedTurnResultError,
+} from '../server/session-transcript-reader.js';
 import { mkdtemp, rm } from 'node:fs/promises';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
@@ -1383,6 +1386,36 @@ test('a failed WorkHub final target check rejects continuation without draining
   }
 });
 
+test('an omitted delegated result remains retryable without draining the Host', async () => {
+  const fixture = await createFailureFixture({
+    registerBackend: (backends) =>
+      backends.register('ai-sdk', (context) => new FakeBackend(context)),
+  });
+  try {
+    await assert.rejects(
+      fixture.coordinator.startWorkHubResult(
+        {
+          kind: 'workhub_result',
+          eventId: 'omitted-result',
+          actionId: 'action',
+          delegationId: 'delegation',
+          targetSessionId: fixture.sessionId,
+          targetTurnId: 'delegated-turn',
+        },
+        async () => {
+          throw new OmittedTurnResultError('Delegated turn has an omitted transcript result');
+        },
+      ),
+      OmittedTurnResultError,
+    );
+    assert.equal(fixture.drainRequested(), false);
+  } finally {
+    await fixture.coordinator.close();
+    await fixture.messages.close();
+    await fixture.dispose();
+  }
+});
+
 test('resume query preserves Session-before-activation lock ordering', async () => {
   const activation = new RuntimePolicyActivationGate();
   const capabilities = new HostClientCapabilityCoordinator({

packages/runtime-host/src/__tests__/session-continuity-coordinator.test.ts

@@ -1184,7 +1184,7 @@ test('open returns a bounded immutable durable tail', async () => {
   coordinator.close();
 });
 
-test('logs a durable bootstrap failure before reporting unavailable persistence', async (t) => {
+test('isolates a durable bootstrap failure without draining the Host', async (t) => {
   const logs: string[] = [];
   t.mock.method(console, 'error', (...args: unknown[]) => {
     logs.push(args.map(String).join(' '));
@@ -1193,19 +1193,21 @@ test('logs a durable bootstrap failure before reporting unavailable persistence'
     `injected durable bootstrap failure: api_key=sk-secretvalue123\n${'细'.repeat(4096)}`,
   );
   const publicationFailures: unknown[] = [];
+  const baseReader = transcriptReader([]);
+  let bootstrapReads = 0;
   const reader: SessionTranscriptReader = {
-    ...transcriptReader([]),
+    ...baseReader,
     readDurableHighWater: async () => 0,
-    readDurablePage: async () => {
-      throw failure;
+    readDurablePage: async (sessionId, request, project) => {
+      if (bootstrapReads++ === 0) throw failure;
+      return baseReader.readDurablePage(sessionId, request, project);
     },
   };
   const coordinator = new SessionContinuityCoordinator(
     HOST_EPOCH,
     async () => canonical(),
     new SessionAdmissionGate(),
     (error) => {
-      assert.equal(logs.length, 1, 'log the cause before the publication-failure hook');
       publicationFailures.push(error);
     },
     reader,
@@ -1221,9 +1223,17 @@ test('logs a durable bootstrap failure before reporting unavailable persistence'
   );
   assert.deepEqual(outcome, {
     ok: false,
-    error: { code: 'persistence_failed', message: 'Session transcript is unavailable' },
+    error: { code: 'transcript_unavailable', message: 'Session transcript is unavailable' },
   });
-  assert.deepEqual(publicationFailures, [failure]);
+  assert.deepEqual(publicationFailures, []);
+  const retry = await coordinator.handlers['subscription.open'](
+    {
+      sessionId: SESSION_ID,
+      transcript: { kind: 'tail', maxBytes: SESSION_TRANSCRIPT_BOOTSTRAP_MAX_BYTES },
+    },
+    connectionContext('connection-failed-bootstrap'),
+  );
+  assert.equal(retry.ok, true, 'the same Host can still serve a subsequent open');
   assert.equal(logs.length, 1);
   const prefix = '[runtime-host] subscription.open transcript bootstrap failed: ';
   const diagnostic = logs[0] ?? '';