fix(storage): stream large Codex rollout imports

CHANGELOG.md

@@ -30,6 +30,11 @@
 
 ### Fixed
 
+- Fixed Codex task imports for valid rollout JSONL files larger than 64 MiB by
+  parsing a fixed file snapshot incrementally under source, record, converted-byte,
+  and message-count limits. Source read/conversion failures now reach Desktop as
+  an actionable error instead of an unknown commit outcome; the Runtime Host
+  compatibility epoch moves to 110 (#4642).
 - Fixed a renderer crash dialog reporting React error #185 ("Maximum update depth
   exceeded") coming from the composer's prompt-history inline completion (#4117): the
   offer engine the 0.1.11 composer fed could flip-flop its announcement state on

apps/desktop/src/main/__tests__/import-tasks-settings-page.test.ts

@@ -25,6 +25,7 @@ import { createRoot, type Root } from 'react-dom/client';
 import { AstryxLocaleProvider, LocaleProvider } from '@maka/ui';
 import type { DesktopRuntimeHostRef } from '../../preload/bridge-contract.js';
 import type { DesktopExternalSessionCatalogItem } from '../../preload/external-session-catalog.js';
+import type { ExternalSessionImportFailureReason } from '../../preload/external-session-import-result.js';
 import { ImportTasksSettingsPage } from '../../renderer/settings/import-tasks-settings-page.js';
 import { RuntimeHostSettingsTarget } from '../../renderer/settings/runtime-host-settings-target.js';
 
@@ -108,6 +109,31 @@ describe('ImportTasksSettingsPage durable import state', () => {
     await act(async () => harness.root.unmount());
   });
 
+  it('shows a source-unreadable banner without starting unknown-outcome recovery', async () => {
+    const harness = await renderPage({
+      catalog: catalog(externalSession()),
+      importResult: { ok: false, reason: 'source_unreadable' },
+    });
+
+    const importButton = buttonWithText(harness.container, 'Import');
+    assert.ok(importButton);
+    await act(async () => {
+      importButton.click();
+      await Promise.resolve();
+      await Promise.resolve();
+    });
+
+    assert.match(harness.container.textContent, /could not be read or converted/);
+    assert.doesNotMatch(harness.container.textContent, /Check the import result/);
+    assert.deepEqual(harness.hostCalls(), [
+      { operation: 'listSources', host: TEST_RUNTIME_HOST },
+      { operation: 'list', host: TEST_RUNTIME_HOST },
+      { operation: 'import', host: TEST_RUNTIME_HOST },
+    ]);
+
+    await act(async () => harness.root.unmount());
+  });
+
   it('uses catalog in-flight state after remount to disable the source row', async () => {
     const harness = await renderPage({
       catalog: {
@@ -1202,14 +1228,15 @@ async function renderPage(options: {
   adapterIds?: string[];
   bySource?: Record<string, Array<CatalogResult | Error | Promise<CatalogResult>>>;
   importResult?:
-    | { ok: false; reason: 'commit_outcome_unknown' }
-    | Promise<{ ok: false; reason: 'commit_outcome_unknown' }>;
+    | { ok: false; reason: ExternalSessionImportFailureReason }
+    | Promise<{ ok: false; reason: ExternalSessionImportFailureReason }>;
   /**
    * Per-source answers for a batch: `ok` lands, `unknown` is the Host not
-   * answering, `throw` is a rejection. Keyed by source session id, because a
-   * batch is exactly the case where the ids must not share one answer.
+   * answering, `source_unreadable` is a definite pre-commit failure, and
+   * `throw` is a rejection. Keyed by source session id, because a batch is
+   * exactly the case where the ids must not share one answer.
    */
-  importBySource?: Record<string, 'ok' | 'unknown' | 'throw'>;
+  importBySource?: Record<string, 'ok' | 'unknown' | 'source_unreadable' | 'throw'>;
   onOpenImported?: (sessionId: string) => void;
   locale?: 'en' | 'zh';
 }): Promise<{
@@ -1288,6 +1315,7 @@ async function renderPage(options: {
         const perSource = options.importBySource?.[sourceSessionId];
         if (perSource === 'throw') throw new Error(`import-failed:${sourceSessionId}`);
         if (perSource === 'unknown') return { ok: false, reason: 'commit_outcome_unknown' };
+        if (perSource === 'source_unreadable') return { ok: false, reason: 'source_unreadable' };
         if (perSource === 'ok') {
           return { ok: true, session: { id: `imported-${sourceSessionId}` } };
         }
@@ -1498,6 +1526,29 @@ describe('ImportTasksSettingsPage batch import', () => {
     assert.match(text, /unconfirmed|Unconfirmed|outcome/i);
   });
 
+  it('counts an unreadable source as a definite failure without offering recovery', async () => {
+    const { container } = await renderPage({
+      catalog: { sessions: [externalSession({ id: 'unreadable' })], nextCursor: null },
+      importBySource: { unreadable: 'source_unreadable' },
+    });
+
+    await tick(masterBox(container), true);
+    const run = buttonWithText(container, 'Import selected');
+    assert.ok(run);
+    await act(async () => {
+      run.click();
+      await Promise.resolve();
+      await Promise.resolve();
+      await Promise.resolve();
+    });
+
+    const text = container.textContent ?? '';
+    assert.match(text, /No conversation was imported/);
+    assert.match(text, /1 more could not be imported/);
+    assert.doesNotMatch(text, /Check the import result/);
+    assert.equal(buttonWithText(container, 'Retry'), undefined);
+  });
+
   it('spins only the conversion in flight, not every queued row', async () => {
     // A spinner claims something is happening now. Marking every selected row
     // would put one on rows the batch has not reached, and on rows it already

apps/desktop/src/main/__tests__/runtime-host-external-sessions-ipc-main.test.ts

@@ -134,6 +134,35 @@ test('an uncertain commit still asks the shell to re-read the catalog', async ()
   assert.deepEqual(events, [{ reason: 'created', sessionId: undefined }]);
 });
 
+test('maps a pre-commit source failure to a distinct non-recovering reason', async () => {
+  const events: unknown[] = [];
+  const ipc = ipcHarness();
+  registerRuntimeHostExternalSessionsIpc(
+    {
+      client: clientFixture({
+        importExternalSession: async () => {
+          throw new RuntimeHostOperationError(
+            'external-session.import',
+            'source_unreadable',
+            'External Session could not be read or converted',
+          );
+        },
+      }),
+      emitSessionsChanged: (reason, sessionId) => events.push({ reason, sessionId }),
+    },
+    ipc,
+  );
+
+  assert.deepEqual(
+    await ipc.invoke('external-sessions:import', {
+      adapterId: 'codex',
+      sourceSessionId: 'source-1',
+    }),
+    { ok: false, reason: 'source_unreadable' },
+  );
+  assert.deepEqual(events, []);
+});
+
 test('rejects malformed renderer requests before they reach the Host client', async () => {
   let calls = 0;
   const ipc = ipcHarness();

apps/desktop/src/main/runtime-host-external-sessions-ipc-main.ts

@@ -85,20 +85,27 @@ export function registerRuntimeHostExternalSessionsIpc(
     } catch (error) {
       if (
         error instanceof RuntimeHostOperationError &&
-        error.operation === 'external-session.import' &&
-        error.code === 'commit_outcome_unknown'
+        error.operation === 'external-session.import'
       ) {
-        // "Unknown" means the task may well be in the catalog, so tell the
-        // shell to read it again. Without this, the only trace of a maybe-
-        // committed import is the banner on the page, and the page is gone the
-        // moment the user leaves Settings -- which is exactly when they come
-        // back and import the same conversation a second time. No id: the
-        // whole point is that we do not know which task, if any, landed.
-        deps.emitSessionsChanged('created');
-        return {
-          ok: false,
-          reason: 'commit_outcome_unknown',
-        } satisfies ExternalSessionImportIpcResult;
+        if (error.code === 'commit_outcome_unknown') {
+          // "Unknown" means the task may well be in the catalog, so tell the
+          // shell to read it again. Without this, the only trace of a maybe-
+          // committed import is the banner on the page, and the page is gone the
+          // moment the user leaves Settings -- which is exactly when they come
+          // back and import the same conversation a second time. No id: the
+          // whole point is that we do not know which task, if any, landed.
+          deps.emitSessionsChanged('created');
+          return {
+            ok: false,
+            reason: 'commit_outcome_unknown',
+          } satisfies ExternalSessionImportIpcResult;
+        }
+        if (error.code === 'source_unreadable') {
+          return {
+            ok: false,
+            reason: 'source_unreadable',
+          } satisfies ExternalSessionImportIpcResult;
+        }
       }
       throw error;
     }

apps/desktop/src/preload/external-session-import-result.ts

@@ -19,7 +19,11 @@
 
 import type { SessionSummary } from '@maka/core/session';
 
-/** Stable Desktop IPC result for the one import failure that must not be retried blindly. */
+export type ExternalSessionImportFailureReason =
+  | 'commit_outcome_unknown'
+  | 'source_unreadable';
+
+/** Stable Desktop IPC result for import failures that require distinct handling. */
 export type ExternalSessionImportIpcResult<T extends SessionSummary = SessionSummary> =
   | { readonly ok: true; readonly session: T }
-  | { readonly ok: false; readonly reason: 'commit_outcome_unknown' };
+  | { readonly ok: false; readonly reason: ExternalSessionImportFailureReason };

apps/desktop/src/renderer/locales/external-session-import-copy.ts

@@ -68,6 +68,7 @@ type ExternalSessionImportCopy = {
   importInProgressDescription: (name: string) => string;
   importFailedTitle: string;
   importFailedFallback: string;
+  importFailedSourceUnreadable: string;
   importRecoveredTitle: string;
   importRecoveredDescription: (name: string) => string;
   importNotRecordedTitle: string;
@@ -133,6 +134,8 @@ const COPY = {
     importInProgressDescription: (name) => `正在导入「${name}」，完成后会直接打开这个任务。`,
     importFailedTitle: '导入失败',
     importFailedFallback: '该对话无法转换或保存。请检查来源后重试。',
+    importFailedSourceUnreadable:
+      '无法读取或转换该对话。来源可能已损坏，或内容超过安全导入限制。请检查来源后重试。',
     importRecoveredTitle: '已确认导入',
     importRecoveredDescription: (name) => `「${name}」导入的任务现已可用。`,
     importNotRecordedTitle: '没有发现新任务',
@@ -186,6 +189,8 @@ const COPY = {
       `Importing “${name}”. Maka opens the task as soon as it lands.`,
     importFailedTitle: 'Import failed',
     importFailedFallback: 'This conversation could not be converted or saved. Check the source and try again.',
+    importFailedSourceUnreadable:
+      'This conversation could not be read or converted. The source may be malformed or exceed a safe import limit.',
     importRecoveredTitle: 'Import confirmed',
     importRecoveredDescription: (name) =>
       `The imported task is available now for “${name}”.`,

apps/desktop/src/renderer/settings/import-tasks-settings-page.tsx

@@ -694,7 +694,14 @@ export function ImportTasksSettingsPage(props: {
         // the user has left steering the shell somewhere they did not ask for.
         if (!mountedRef.current) return;
         if (!outcome.ok) {
-          await recoverUnknownImport(attempt);
+          if (outcome.reason === 'commit_outcome_unknown') {
+            await recoverUnknownImport(attempt);
+          } else if (outcome.reason === 'source_unreadable') {
+            setImportError(copy.importFailedSourceUnreadable);
+          } else {
+            const _exhaustive: never = outcome.reason;
+            return _exhaustive;
+          }
           return;
         }
         props.onImported(outcome.session);
@@ -773,30 +780,45 @@ export function ImportTasksSettingsPage(props: {
         try {
           const result = await requestImport(attempt.adapterId, attempt.sourceSessionId);
           if (!mountedRef.current) return;
-          outcome = recordImportBatchResult(
-            outcome,
-            session.id,
-            // Not `failed`: the call did not answer, and only a catalog read
-            // settles whether the conversion landed. Calling it a failure is
-            // what invites the retry that makes a second copy.
-            result.ok ? (wasImported ? 'duplicated' : 'imported') : 'unknown',
-          );
-          if (!result.ok) {
-            // Recorded, not recovered. A single import recovers inline, but
-            // recovery re-reads the whole catalog window per attempt, and doing
-            // that between conversions would interleave N full reads with the
-            // batch and race the writes it is making. The unconfirmed banner
-            // names every one of these and its 重试 resolves them a press at a
-            // time, removing each as it settles.
-            setUncertainImports((current) =>
-              current.some(
-                (entry) =>
-                  entry.adapterId === attempt.adapterId &&
-                  entry.sourceSessionId === attempt.sourceSessionId,
-              )
-                ? current
-                : [...current, attempt],
+          if (result.ok) {
+            outcome = recordImportBatchResult(
+              outcome,
+              session.id,
+              wasImported ? 'duplicated' : 'imported',
             );
+          } else {
+            switch (result.reason) {
+              case 'commit_outcome_unknown':
+                // Not `failed`: the call did not answer, and only a catalog read
+                // settles whether the conversion landed. Calling it a failure is
+                // what invites the retry that makes a second copy.
+                outcome = recordImportBatchResult(outcome, session.id, 'unknown');
+                // Recorded, not recovered. A single import recovers inline, but
+                // recovery re-reads the whole catalog window per attempt, and doing
+                // that between conversions would interleave N full reads with the
+                // batch and race the writes it is making. The unconfirmed banner
+                // names every one of these and its 重试 resolves them a press at a
+                // time, removing each as it settles.
+                setUncertainImports((current) =>
+                  current.some(
+                    (entry) =>
+                      entry.adapterId === attempt.adapterId &&
+                      entry.sourceSessionId === attempt.sourceSessionId,
+                  )
+                    ? current
+                    : [...current, attempt],
+                );
+                break;
+              case 'source_unreadable':
+                // Conversion failed before persistence, so this one definitely
+                // did not land and must never enter unknown-outcome recovery.
+                outcome = recordImportBatchResult(outcome, session.id, 'failed');
+                break;
+              default: {
+                const _exhaustive: never = result.reason;
+                return _exhaustive;
+              }
+            }
           }
         } catch {
           if (!mountedRef.current) return;

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

@@ -345,7 +345,10 @@ test('reports conversion errors before persistence and store uncertainty after e
     ),
     {
       ok: false,
-      error: { code: 'invalid_request', message: 'External Session could not be converted' },
+      error: {
+        code: 'source_unreadable',
+        message: 'External Session could not be read or converted',
+      },
     },
   );
   assert.equal(createAttempts, 0);

packages/runtime-host/src/__tests__/protocol.test.ts

@@ -437,6 +437,10 @@ describe('Runtime Host bootstrap protocol', () => {
     assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 99);
   });
 
+  test('publishes a new compatibility epoch for external Session source failures', () => {
+    assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 109);
+  });
+
   test('selects the highest mutually supported protocol and rejects a gap', () => {
     assert.equal(negotiateProtocol({ min: 0, max: 0 }, { min: 0, max: 0 }), 0);
     assert.equal(negotiateProtocol({ min: 1, max: 3 }, { min: 2, max: 4 }), 3);

packages/runtime-host/src/protocol/external-session.ts

@@ -57,6 +57,7 @@ const IMPORT_ERRORS = [
   'not_found',
   'operation_conflict',
   'commit_outcome_unknown',
+  'source_unreadable',
 ] as const;
 
 export type ExternalSessionSourceQueryInput = Record<never, never>;

packages/runtime-host/src/protocol/index.ts

@@ -100,7 +100,9 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const;
 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.
-export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 110 as const;
+export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 111 as const;
+// 111: external Session import can report `source_unreadable`, allowing clients
+// to distinguish a definite source conversion failure from an uncertain commit.
 // 110: Runtime Host is the sole schema-migration authority for its State Root.
 // Epoch 109 Desktop builds could migrate the event-only AgentRun schema while
 // an older service Host still held the root, leaving that Host querying a

packages/runtime-host/src/protocol/operation-spec.ts

@@ -31,6 +31,7 @@ export type HostOperationErrorCode =
   | 'operation_conflict'
   | 'capability_unavailable'
   | 'invalid_request'
+  | 'source_unreadable'
   | 'projection_incomplete'
   | 'stale_cursor'
   | 'persistence_failed'

packages/runtime-host/src/server/external-session-coordinator.ts

@@ -269,10 +269,10 @@ export class HostExternalSessionCoordinator {
     } catch (error) {
       if (!commitAttempted) {
         return importFailure(
-          isSourceSessionNotFound(error) ? 'not_found' : 'invalid_request',
+          isSourceSessionNotFound(error) ? 'not_found' : 'source_unreadable',
           isSourceSessionNotFound(error)
             ? 'External Session does not exist'
-            : 'External Session could not be converted',
+            : 'External Session could not be read or converted',
         );
       }
       this.#requestDrain();