fix(core,storage): unify the Codex thread source gate

packages/core/src/foreign-session.ts

@@ -392,8 +392,21 @@ export function claudeToolFilePaths(record: Record<string, unknown>): string[] {
  * content.
  * ------------------------------------------------------------------ */
 
-/** Codex thread sources eligible for import (per issue #1057). */
-export const CODEX_SUPPORTED_THREAD_SOURCES = ['cli', 'vscode', 'atlas', 'chatgpt'] as const;
+/**
+ * Codex thread sources eligible for import — the single authority for both the
+ * foreign-session scanner below and the Codex Session adapter in `@maka/storage`.
+ * #1057 defined the original list; `exec` covers headless `codex exec` runs,
+ * which #2502 listed as root Sessions from the adapter's first commit. The two
+ * gates drifted apart while each owned its own token set, so the same thread was
+ * visible through one surface and invisible through the other.
+ */
+export const CODEX_SUPPORTED_THREAD_SOURCES = [
+  'cli',
+  'exec',
+  'vscode',
+  'atlas',
+  'chatgpt',
+] as const;
 
 /**
  * Timestamps below this (2020-01-01 UTC in ms) are treated as seconds and
@@ -421,34 +434,51 @@ function normalizeEpochMs(value: unknown): number | undefined {
 }
 
 /**
- * Codex persists `source` either as a bare token (`cli`, `vscode`) or as a
- * JSON object string (`{"custom":"atlas"}`, `{"custom":"chatgpt"}`). Return
- * the canonical token, or undefined when it isn't a supported source — a
- * bare-string equality check would silently drop every atlas/chatgpt thread.
+ * Codex persists `source` as a bare token (`cli`, `exec`, `vscode`), as a JSON
+ * object string (`{"custom":"atlas"}`), or — in rollout `session_meta` payloads,
+ * which arrive already parsed — as the object itself. Return the canonical
+ * token, or undefined when it isn't a supported source: a bare-string equality
+ * check would silently drop every atlas/chatgpt thread, and a token set that
+ * only knows the wrapped form would drop the bare one.
+ *
+ * Unsupported shapes (notably `{"subagent":{…}}`) resolve to undefined, which is
+ * what keeps internal subagent threads out of both surfaces.
  */
 export function codexSourceToken(value: unknown): string | undefined {
-  if (typeof value !== 'string' || value.length === 0) return undefined;
-  if ((CODEX_SUPPORTED_THREAD_SOURCES as readonly string[]).includes(value)) return value;
-  if (value.startsWith('{')) {
+  if (typeof value === 'string') {
+    if (value.length === 0) return undefined;
+    if (isSupportedCodexSourceToken(value)) return value;
+    // Only an object string can carry a `custom` token; anything else is a
+    // plain unsupported source and must not reach JSON.parse.
+    if (!value.startsWith('{')) return undefined;
     try {
-      const parsed = JSON.parse(value) as unknown;
-      const custom =
-        typeof parsed === 'object' && parsed !== null
-          ? (parsed as Record<string, unknown>).custom
-          : undefined;
-      if (
-        typeof custom === 'string' &&
-        (CODEX_SUPPORTED_THREAD_SOURCES as readonly string[]).includes(custom)
-      ) {
-        return custom;
-      }
+      return codexSourceToken(JSON.parse(value) as unknown);
     } catch {
       return undefined;
     }
   }
+  if (typeof value === 'object' && value !== null) {
+    const custom = (value as Record<string, unknown>).custom;
+    return typeof custom === 'string' && isSupportedCodexSourceToken(custom) ? custom : undefined;
+  }
   return undefined;
 }
 
+/**
+ * Whether a recorded `source` makes a Codex thread eligible. An absent source is
+ * eligible: older Codex schemas have no `source` column and older rollout
+ * `session_meta` payloads omit the field, so dropping those would hide every
+ * legacy thread. A present-but-unsupported source is a hard drop.
+ */
+export function isSupportedCodexThreadSource(value: unknown): boolean {
+  if (value === undefined || value === null) return true;
+  return codexSourceToken(value) !== undefined;
+}
+
+function isSupportedCodexSourceToken(value: string): boolean {
+  return (CODEX_SUPPORTED_THREAD_SOURCES as readonly string[]).includes(value);
+}
+
 export interface CodexThreadRow {
   id?: unknown;
   rollout_path?: unknown;
@@ -476,7 +506,7 @@ export function normalizeCodexThreadRow(
   if (row.archived === 1 || row.archived === true) return undefined;
   // A present-but-unsupported source is a hard drop; an absent source column
   // (older schema) is allowed through — the SELECT simply didn't project it.
-  if (row.source !== undefined && codexSourceToken(row.source) === undefined) return undefined;
+  if (!isSupportedCodexThreadSource(row.source)) return undefined;
   const updatedAtMs = normalizeEpochMs(row.updated_at_ms) ?? normalizeEpochMs(row.updated_at) ?? 0;
   const title =
     sanitizeForeignTitle(row.title) || sanitizeForeignTitle(row.first_user_message) || row.id;

packages/storage/src/__tests__/codex-session-adapter.test.ts

@@ -129,6 +129,64 @@ describe('CodexSessionAdapter', () => {
     });
   });
 
+  test('lists every thread source the foreign-session scanner accepts (#3693)', async () => {
+    // The adapter owned its own token set, so bare `atlas`/`chatgpt` and a
+    // wrapped `{"custom":"cli"}` were dropped here while the scanner in
+    // `@maka/core/foreign-session` listed them. Both gates now share one
+    // authority, so the catalog and the scan agree on every shape.
+    await withCodexHome(async (codexHome) => {
+      const sources = ['cli', 'exec', 'vscode', 'atlas', 'chatgpt'] as const;
+      const rows: StateRow[] = [];
+      for (const [index, source] of sources.entries()) {
+        const bareId = `codex-bare-${source}`;
+        const wrappedId = `codex-wrapped-${source}`;
+        rows.push({
+          id: bareId,
+          rolloutPath: await seedMinimalRollout(codexHome, bareId, false, '/workspace', 'Task'),
+          cwd: '/workspace',
+          name: `bare ${source}`,
+          createdAtMs: 1000 + index,
+          updatedAtMs: 3000 + index,
+          archived: false,
+          source,
+        });
+        rows.push({
+          id: wrappedId,
+          rolloutPath: await seedMinimalRollout(codexHome, wrappedId, false, '/workspace', 'Task'),
+          cwd: '/workspace',
+          name: `wrapped ${source}`,
+          createdAtMs: 1100 + index,
+          updatedAtMs: 3100 + index,
+          archived: false,
+          source: JSON.stringify({ custom: source }),
+        });
+      }
+      const subagentId = 'codex-subagent-drop';
+      rows.push({
+        id: subagentId,
+        rolloutPath: await seedMinimalRollout(codexHome, subagentId, false, '/workspace', 'Task'),
+        cwd: '/workspace',
+        name: 'internal child',
+        createdAtMs: 2000,
+        updatedAtMs: 4000,
+        archived: false,
+        source: '{"subagent":{"thread_spawn":{"parent_thread_id":"parent"}}}',
+      });
+      await seedStateDatabase(codexHome, rows);
+
+      const listed = new Set(
+        (await new CodexSessionAdapter({ codexHome }).listSessions()).map((session) => session.id),
+      );
+      for (const source of sources) {
+        assert.ok(listed.has(`codex-bare-${source}`), `bare ${source} was dropped`);
+        assert.ok(listed.has(`codex-wrapped-${source}`), `wrapped ${source} was dropped`);
+      }
+      // Internal subagent threads stay out of the catalog.
+      assert.equal(listed.has(subagentId), false);
+      assert.equal(listed.size, sources.length * 2);
+    });
+  });
+
   test('a Windows path spelling reaches the matcher instead of being lost in SQL', async () => {
     // The SQL used to prefilter with `cwd IN (<spelling variants>)`, and
     // SQLite compares those exactly — a row stored `C:\\Repo\\App` was

packages/storage/src/__tests__/foreign-session-store.test.ts

@@ -112,7 +112,7 @@ type CodexThreadSeed = {
   title?: string;
   updatedAtMs?: number;
   archived?: number;
-  source?: string;
+  source?: string | null;
   rolloutRelPath?: string;
 };
 
@@ -179,7 +179,7 @@ async function seedCodexSqliteGen(
       t.title ?? null,
       t.updatedAtMs ?? NOW - 60_000,
       t.archived ?? 0,
-      t.source ?? 'cli',
+      t.source === undefined ? 'cli' : t.source,
     );
   }
   db.close();
@@ -351,6 +351,28 @@ describe('foreign session store — Codex scan', () => {
     assert.deepEqual((await store.listSessions()).map((s) => s.id).sort(), ['atl', 'gpt']);
   });
 
+  it('routes every supported sqlite source shape through the shared gate', async () => {
+    const home = await tempHome();
+    await seedCodexSqlite(home, [
+      { id: 'bare-exec', cwd: '/repo', source: 'exec' },
+      { id: 'bare-atlas', cwd: '/repo', source: 'atlas' },
+      { id: 'bare-chatgpt', cwd: '/repo', source: 'chatgpt' },
+      { id: 'wrapped-cli', cwd: '/repo', source: '{ "custom": "cli" }' },
+      { id: 'wrapped-vscode', cwd: '/repo', source: '{"custom":"vscode"}' },
+      { id: 'legacy-null', cwd: '/repo', source: null },
+      { id: 'unsupported', cwd: '/repo', source: '{"custom":"other"}' },
+    ]);
+    const store = createForeignSessionStore({ homeDir: home, env: {} });
+    assert.deepEqual((await store.listSessions()).map((session) => session.id).sort(), [
+      'bare-atlas',
+      'bare-chatgpt',
+      'bare-exec',
+      'legacy-null',
+      'wrapped-cli',
+      'wrapped-vscode',
+    ]);
+  });
+
   it('rejects rollout paths that escape ~/.codex', async () => {
     const home = await tempHome();
     const outside = join(home, 'outside.jsonl');

packages/storage/src/codex-session-adapter.ts

@@ -22,7 +22,7 @@ import { open, readdir, realpath, stat } from 'node:fs/promises';
 import { homedir } from 'node:os';
 import { basename, join, resolve, sep } from 'node:path';
 import type { StoredMessage } from '@maka/core/session';
-import { sanitizeForeignTitle } from '@maka/core/foreign-session';
+import { isSupportedCodexThreadSource, sanitizeForeignTitle } from '@maka/core/foreign-session';
 import { externalSessionMatchesQuery } from '@maka/core/external-session';
 import type {
   ExternalMakaSession,
@@ -38,7 +38,6 @@ const CODEX_ROLLOUT_HEAD_BYTES = 512 * 1024;
 const CODEX_SESSION_ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/;
 const CODEX_UNSAFE_PATH_CHARS =
   /[\u0000-\u001F\u007F\u0080-\u009F\u061C\u200B-\u200F\u202A-\u202E\u2060-\u2064\u2066-\u2069\uFEFF]/;
-const CODEX_ROOT_SOURCE_TOKENS = new Set(['cli', 'exec', 'vscode']);
 
 export interface CodexSessionAdapterOptions {
   /** Codex's state root. Defaults to `$CODEX_HOME`, then `~/.codex`. */
@@ -156,7 +155,7 @@ export class CodexSessionAdapter implements ExternalSessionAdapter {
   private async entryFromRow(row: CodexThreadRow): Promise<CodexCatalogEntry | undefined> {
     if (!isSafeCodexSessionId(row.id)) return undefined;
     if (typeof row.rollout_path !== 'string' || row.rollout_path.length === 0) return undefined;
-    if (!isRootCodexSource(row.source)) return undefined;
+    if (!isSupportedCodexThreadSource(row.source)) return undefined;
 
     const rolloutPath = await this.resolveRolloutPath(row.rollout_path, row.id);
     if (!rolloutPath) return undefined;
@@ -575,7 +574,7 @@ function catalogEntryFromRolloutHead(
     const payload = asRecord(record.payload);
     if (!payload) continue;
     if (record.type === 'session_meta') {
-      if (!isRootCodexSource(payload.source)) return undefined;
+      if (!isSupportedCodexThreadSource(payload.source)) return undefined;
       id = stringField(payload, 'session_id') ?? stringField(payload, 'id') ?? id;
       cwd = safeCodexCwd(payload.cwd) || cwd;
       createdAt =
@@ -794,21 +793,6 @@ function firstNonEmptyTitle(...values: unknown[]): string | undefined {
   return undefined;
 }
 
-function isRootCodexSource(value: unknown): boolean {
-  if (value === undefined || value === null) return true;
-  if (typeof value === 'string') {
-    if (CODEX_ROOT_SOURCE_TOKENS.has(value)) return true;
-    if (!value.startsWith('{')) return false;
-    try {
-      return isRootCodexSource(JSON.parse(value) as unknown);
-    } catch {
-      return false;
-    }
-  }
-  if (!isRecord(value)) return false;
-  return value.custom === 'atlas' || value.custom === 'chatgpt';
-}
-
 function codexErrorAffectsTurnStatus(payload: JsonRecord): boolean {
   const info = payload.codex_error_info;
   if (info === 'thread_rollback_failed' || info === 'active_turn_not_steerable') return false;

packages/storage/src/foreign-session-store.ts

@@ -47,6 +47,7 @@ import { open, readdir, realpath, stat, type FileHandle } from 'node:fs/promises
 import { homedir } from 'node:os';
 import { basename, join, resolve, sep } from 'node:path';
 import {
+  CODEX_SUPPORTED_THREAD_SOURCES,
   FOREIGN_SESSION_DIGEST_MAX_READ_BYTES,
   FOREIGN_SESSION_HEAD_BYTES,
   FOREIGN_SESSION_SCAN_MAX_AGE_MS,
@@ -537,13 +538,6 @@ async function codexStateDbsNewestFirst(codexRoot: string): Promise<string[]> {
   }
 }
 
-/**
- * Codex source tokens as stored in the DB — bare for cli/vscode, JSON-wrapped
- * for the `custom` variants. Used as bound `source IN (…)` params so archived
- * / foreign-source rows are excluded IN SQL (before LIMIT), not after.
- */
-const CODEX_SOURCE_SQL_VALUES = ['cli', 'vscode', '{"custom":"atlas"}', '{"custom":"chatgpt"}'];
-
 /**
  * Read candidate thread rows from one state DB, filtered and ordered in SQL.
  * undefined = DB unusable (cannot open, or lacks the id/rollout_path columns)
@@ -582,8 +576,21 @@ async function readCodexThreadRows(
       const params: string[] = [];
       if (columns.has('archived')) where.push('(archived IS NULL OR archived = 0)');
       if (columns.has('source')) {
-        where.push(`source IN (${CODEX_SOURCE_SQL_VALUES.map(() => '?').join(', ')})`);
-        params.push(...CODEX_SOURCE_SQL_VALUES);
+        const sourceTokens = [...CODEX_SUPPORTED_THREAD_SOURCES];
+        const placeholders = sourceTokens.map(() => '?').join(', ');
+        // Keep unsupported rows from consuming the bounded SQL window, but
+        // derive this coarse prefilter from the same token authority as
+        // normalizeCodexThreadRow(). The JS gate remains authoritative over
+        // exact shapes after bare, wrapped-custom, and legacy NULL sources
+        // have survived the query.
+        where.push(`(
+          source IS NULL
+          OR source IN (${placeholders})
+          OR CASE WHEN json_valid(source)
+            THEN json_extract(source, '$.custom')
+          END IN (${placeholders})
+        )`);
+        params.push(...sourceTokens, ...sourceTokens);
       }
       // Filter cwd IN SQL, before LIMIT: otherwise a multi-project store with
       // many newer threads from other directories fills the LIMIT window and