fix(sdk): report observed worker sessions without status markers

sdk/typescript/README.md

@@ -159,11 +159,40 @@ Options for `security.run(repository, options)` and
 | `expectedPluginVersion`                     | Required original plugin version when replaying a scan.                             |
 | `signal`                                    | `AbortSignal` to cancel a scan.                                                     |
 
-Follow scans with `onWorkerStatus` and `onReconnect`. `onSessionEvent` receives
-saved events with thread IDs and worker numbers. Deep scans can additionally use
-`onDeepProgress` for durable independent-review counts: `completed`, `active`,
-and `maximum`. The maximum is a configured cap, not a percentage denominator.
-`ScanOptions` lists all callbacks.
+Follow scans with `onWorkerEvent` and `onReconnect`. `onWorkerEvent` reports
+persisted worker sessions discovered by the SDK's existing session tracker,
+independently of model-emitted status markers:
+
+```ts
+await security.run(repository, {
+  onWorkerEvent(event) {
+    console.log(`Worker ${event.worker} observed`);
+  },
+});
+```
+
+The callback contains only `{ kind: "observed", worker: number }`. The scan-local
+worker number matches `onActivity` and `onSessionEvent`; no prompts, raw thread
+IDs, or session contents are exposed. Each session is reported once per run,
+including nested workers and scan-associated validation or Deep Scan sessions.
+On resume, already persisted workers can be reported again. Observation ends
+with scan cost tracking, before `postScanPrompt`.
+
+This works with the bundled Codex version. Persistence and polling can delay
+notification, and missing notifications do not prove that delegation was skipped.
+An observed session does not establish that a worker just started or that file
+review has begun. The callback does not report failed spawn attempts, phase names,
+or planned counts, and cannot act as a pre-dispatch gate. Use `maxCostUsd` or
+`signal` for cancellation. Observer failures go to `onObserverError` without
+stopping the scan.
+
+`onWorkerStatus` remains available for tool-derived preflight status and
+**best-effort** phase dispatch counts from model-emitted text markers. A missing
+dispatch status does not mean delegation was skipped. `onSessionEvent` receives
+saved events with thread IDs and worker numbers and can contain source code or
+credentials. Deep scans additionally expose durable independent-review counts
+through `onDeepProgress`: `completed`, `active`, and `maximum`. The maximum is a
+configured cap, not a percentage denominator. `ScanOptions` lists all callbacks.
 
 `preflight` and CLI `--dry-run` check local inputs without starting Codex or
 using the network. They don't authenticate, verify model access, resolve Python,

sdk/typescript/scripts/fixtures/package-consumer.ts

@@ -26,6 +26,7 @@ import {
   type ScanComparisonResult,
   type ScanOptions,
   type ScanProgress,
+  type ScanWorkerEvent,
   type ScanResult,
   type ScanSettings,
   type ValidationOptions,
@@ -111,6 +112,10 @@ const options: ScanOptions = {
   onProgress(progress: ScanProgress) {
     progress.filesCompleted satisfies number;
   },
+  onWorkerEvent(event: ScanWorkerEvent) {
+    event.kind satisfies "observed";
+    event.worker satisfies number;
+  },
 };
 
 export async function scan(repository: string): Promise<ScanResult> {

sdk/typescript/src/api.ts

@@ -67,6 +67,7 @@ import {
   ScanCostTracker,
   type ScanCost,
   type ScanSessionEvent,
+  type ScanWorkerEvent,
 } from "./cost.js";
 import {
   DeepScanProgressTracker,
@@ -298,7 +299,16 @@ export interface ScanOptions extends ScanSettings {
   onSessionEvent?: (event: ScanSessionEvent) => void;
   onProgress?: (progress: ScanProgress) => void;
   onDeepProgress?: (progress: DeepScanProgress) => void;
+  /** Preflight status and best-effort, model-reported phase dispatch counts. */
   onWorkerStatus?: (status: ScanWorkerStatus) => void;
+  /**
+   * Reports each persisted worker session once when discovered during this run.
+   * Worker numbers match onActivity/onSessionEvent. Includes saved workers on
+   * resume; observation ends before postScanPrompt. Persistence and polling can
+   * delay delivery. Does not report failed spawns, phase, or planned counts and
+   * cannot gate dispatch; use maxCostUsd or signal for cancellation.
+   */
+  onWorkerEvent?: (event: ScanWorkerEvent) => void;
   onWarning?: (warning: string, details?: ScanWarningDetails) => void;
   onObserverError?: (observer: ScanObserverName, error: unknown) => void;
   signal?: AbortSignal;
@@ -393,6 +403,7 @@ type ScanObserverName =
   | "onProgress"
   | "onDeepProgress"
   | "onWorkerStatus"
+  | "onWorkerEvent"
   | "onStage"
   | "onWarning";
 
@@ -1453,6 +1464,16 @@ export class CodexSecurity {
                   options.onObserverError,
                   activity,
                 ),
+        onWorkerEvent:
+          options.onWorkerEvent === undefined
+            ? undefined
+            : (event) =>
+                notifyObserver(
+                  "onWorkerEvent",
+                  options.onWorkerEvent,
+                  options.onObserverError,
+                  event,
+                ),
         onSessionEvent:
           options.onSessionEvent === undefined
             ? undefined

sdk/typescript/src/cost.ts

@@ -22,6 +22,13 @@ import {
 
 export { estimateScanCost, formatUsd, type ScanCost } from "./cost-model.js";
 
+/** A persisted worker session discovered during this run. Contains no model text. */
+export interface ScanWorkerEvent {
+  kind: "observed";
+  /** Scan-local number shared with activity and session observers. */
+  worker: number;
+}
+
 export interface ScanSessionEvent {
   threadId: string;
   parentThreadId: string | null;
@@ -70,6 +77,7 @@ interface ScanCostTrackerOptions {
   onActivity?: (activity: ScanActivity) => void;
   onProgress?: (progress: ScanProgress) => void;
   onSessionEvent?: (event: ScanSessionEvent) => void;
+  onWorkerEvent?: (event: ScanWorkerEvent) => void;
   onError?: (error: unknown) => void;
 }
 
@@ -113,6 +121,7 @@ export class ScanCostTracker {
   readonly #workerProgress = new Map<string, number>();
   readonly #reportedProgress = new Set<string>();
   #threadId: string | null = null;
+  #observingWorkers = true;
   #timer: NodeJS.Timeout | null = null;
   #pending: Promise<void> = Promise.resolve();
   #snapshot: ScanCostSnapshot = { usage: null, cost: null };
@@ -144,7 +153,8 @@ export class ScanCostTracker {
       this.#options.onCost === undefined &&
       this.#options.onActivity === undefined &&
       this.#options.onProgress === undefined &&
-      this.#options.onSessionEvent === undefined
+      this.#options.onSessionEvent === undefined &&
+      this.#options.onWorkerEvent === undefined
     ) {
       return;
     }
@@ -188,7 +198,11 @@ export class ScanCostTracker {
       this.#timer = null;
     }
     if (fallbackUsage !== undefined) this.recordUsage(fallbackUsage);
-    await this.refresh();
+    try {
+      await this.refresh();
+    } finally {
+      this.#observingWorkers = false;
+    }
     if (this.#receipts.size > 0 || this.#snapshot.usage !== null)
       return this.#snapshot;
     const cost = estimateScanCost(this.#options.model, fallbackUsage);
@@ -279,8 +293,13 @@ export class ScanCostTracker {
       }
       let worker: number | undefined;
       if (threadId !== this.#threadId) {
-        worker = this.#workers.get(threadId) ?? this.#workers.size + 1;
-        this.#workers.set(threadId, worker);
+        worker = this.#workers.get(threadId);
+        if (worker === undefined) {
+          worker = this.#workers.size + 1;
+          this.#workers.set(threadId, worker);
+          if (this.#observingWorkers)
+            this.#options.onWorkerEvent?.({ kind: "observed", worker });
+        }
       }
       for (const event of session.events?.splice(0) ?? []) {
         this.#options.onSessionEvent?.({

sdk/typescript/src/index.ts

@@ -38,7 +38,7 @@ export type {
   ComponentPlanningOptions,
 } from "./component-plan.js";
 export { estimateScanCost } from "./cost.js";
-export type { ScanCost, ScanSessionEvent } from "./cost.js";
+export type { ScanCost, ScanSessionEvent, ScanWorkerEvent } from "./cost.js";
 export type { DeepScanProgress } from "./deep-progress.js";
 export type { CustomValidationResult } from "./custom-validation.js";
 export type { ScanActivity, ScanActivityStatus } from "./scan-activity.js";

sdk/typescript/src/worker-progress.ts

@@ -38,6 +38,7 @@ export function scanPhaseLabel(value: ScanWorkerPhase | ScanPhase): string {
   }[value];
 }
 
+/** Preflight is tool-derived; dispatch counts depend on model-emitted markers. */
 export type ScanWorkerStatus =
   | {
       kind: "preflight";

sdk/typescript/tests-ts/api.test.ts

@@ -38,6 +38,7 @@ import {
   type ScanOptions,
   type ScanProgress,
   type ScanSessionEvent,
+  type ScanWorkerEvent,
   ScanInterruptedError,
 } from "../src/index.js";
 import {
@@ -3604,6 +3605,7 @@ describe("CodexSecurity orchestration", () => {
     await mkdir(scanDir, { mode: 0o700 });
     const updates: ScanProgress[] = [];
     const sessionEvents: ScanSessionEvent[] = [];
+    const workers: ScanWorkerEvent[] = [];
     const observerErrors: ScanObserverName[] = [];
     const usage = { input_tokens: 100, output_tokens: 10 };
     const client = new TestClient(
@@ -3687,6 +3689,7 @@ describe("CodexSecurity orchestration", () => {
 
     const result = await client.run(repository, {
       onProgress: (progress) => updates.push(progress),
+      onWorkerEvent: (event) => workers.push(event),
       onSessionEvent: (event) => {
         sessionEvents.push(event);
         if (sessionEvents.length === 1) {
@@ -3711,10 +3714,85 @@ describe("CodexSecurity orchestration", () => {
         ),
       ),
     ).toEqual(new Set(["thread-1:null", "worker-thread:thread-1"]));
+    expect(workers).toEqual([{ kind: "observed", worker: 1 }]);
+    expect(
+      new Set(
+        sessionEvents
+          .filter((event) => event.threadId === "worker-thread")
+          .map((event) => event.worker),
+      ),
+    ).toEqual(new Set([1]));
     expect(observerErrors).toEqual(["onSessionEvent"]);
     await client.close();
   });
 
+  test.each(["none", "sync", "async"] as const)(
+    "observes workers before scan completion with %s observer errors",
+    async (failure) => {
+      const root = await temporaryDirectory();
+      const repository = join(root, "repository");
+      const codexHome = join(root, "codex-home");
+      const scanDir = join(root, "scan");
+      await mkdir(repository);
+      await mkdir(codexHome);
+      await mkdir(scanDir, { mode: 0o700 });
+      const observed = Promise.withResolvers<void>();
+      const workers: ScanWorkerEvent[] = [];
+      const errors: ScanObserverName[] = [];
+      const client = new TestClient(
+        {},
+        {
+          environment: {},
+          prepareRuntime: async () => preparedRuntime(codexHome),
+          resolvePluginPython: async () => "/managed/python",
+          prepareOutputDir: async () => scanDir,
+          repositoryRevision: async () => "deadbeef",
+          createCodex: () => ({
+            startThread: () => ({
+              id: "thread-1",
+              async runStreamed() {
+                await copyCompletedScan(root);
+                await writeUsageSession(codexHome, "thread-1", {});
+                async function* events(): AsyncGenerator<ThreadEvent> {
+                  for await (const event of completedEvents()) {
+                    yield event;
+                    if (event.type === "turn.started") {
+                      await writeUsageSession(
+                        codexHome,
+                        "worker-thread",
+                        {},
+                        "thread-1",
+                      );
+                      await observed.promise;
+                    }
+                  }
+                }
+                return { events: events() };
+              },
+            }),
+          }),
+        },
+      );
+      try {
+        const result = await client.run(repository, {
+          onWorkerEvent: (event) => {
+            workers.push(event);
+            observed.resolve();
+            if (failure === "sync") throw new Error("Optional observer failed");
+            if (failure === "async")
+              return Promise.reject(new Error("Optional observer failed"));
+          },
+          onObserverError: (observer) => errors.push(observer),
+        });
+        expect(result.threadId).toBe("thread-1");
+        expect(workers).toEqual([{ kind: "observed", worker: 1 }]);
+        expect(errors).toEqual(failure === "none" ? [] : ["onWorkerEvent"]);
+      } finally {
+        await client.close();
+      }
+    },
+  );
+
   test("provides only reviewed false positives to validation as a scan artifact", async () => {
     const root = await temporaryDirectory();
     const repository = join(root, "repository");
@@ -4415,13 +4493,14 @@ describe("CodexSecurity orchestration", () => {
   );
 
   test.each([
-    ["partial coverage", "partial", false],
-    ["unknown coverage", "unknown", false],
-    ["a failed scan", "failed", false],
-    ["a failed scan and follow-up", "failed", true],
+    ["partial coverage", "partial", false, false],
+    ["unknown coverage", "unknown", false, false],
+    ["a failed scan", "failed", false, false],
+    ["a failed scan and follow-up", "failed", true, false],
+    ["an interrupted follow-up", "partial", false, true],
   ] as const)(
     "runs post-scan instructions after %s",
-    async (_scenario, outcome, followUpFails) => {
+    async (_scenario, outcome, followUpFails, cancelFollowUp) => {
       const root = await temporaryDirectory();
       const repository = join(root, "repository");
       const codexHome = join(root, "codex-home");
@@ -4431,7 +4510,10 @@ describe("CodexSecurity orchestration", () => {
       await mkdir(scanDir, { mode: 0o700 });
       const prompts: string[] = [];
       const warnings: string[] = [];
+      const workers: ScanWorkerEvent[] = [];
+      const observerErrors: string[] = [];
       const scanFails = outcome === "failed";
+      const controller = new AbortController();
 
       const client = new TestClient(
         {},
@@ -4446,6 +4528,13 @@ describe("CodexSecurity orchestration", () => {
               id: "thread-1",
               async runStreamed(prompt: string) {
                 prompts.push(prompt);
+                await writeUsageSession(codexHome, "thread-1", {});
+                await writeUsageSession(
+                  codexHome,
+                  `worker-${prompts.length}`,
+                  {},
+                  "thread-1",
+                );
                 if (prompts.length === 1 && !scanFails) {
                   await copyCompletedScan(root);
                   const coveragePath = join(scanDir, "coverage.json");
@@ -4465,10 +4554,12 @@ describe("CodexSecurity orchestration", () => {
                   );
                   return { events: completedEvents() };
                 }
+                if (prompts.length === 2 && cancelFollowUp) controller.abort();
                 if (prompts.length === 2 && !followUpFails) {
                   return { events: completedEvents() };
                 }
                 async function* failedEvents(): AsyncGenerator<ThreadEvent> {
+                  yield { type: "thread.started", thread_id: "thread-1" };
                   yield {
                     type: "turn.failed",
                     error: {
@@ -4488,15 +4579,25 @@ describe("CodexSecurity orchestration", () => {
 
       const result = client.run(repository, {
         postScanPrompt: "Record the scan cost.",
+        signal: controller.signal,
         onWarning: (warning) => warnings.push(warning),
+        onWorkerEvent: (event) => {
+          workers.push(event);
+          throw new Error("optional worker observer failed");
+        },
+        onObserverError: (observer) => observerErrors.push(observer),
       });
-      if (scanFails) {
+      if (cancelFollowUp) {
+        await expect(result).rejects.toBeInstanceOf(ScanInterruptedError);
+      } else if (scanFails) {
         await expect(result).rejects.toThrow("The scan failed.");
       } else {
         expect((await result).coverage.completeness).toBe(outcome);
       }
       expect(prompts.at(-1)).toBe("Record the scan cost.");
       expect(prompts).toHaveLength(2);
+      expect(workers).toEqual([{ kind: "observed", worker: 1 }]);
+      expect(observerErrors).toEqual(["onWorkerEvent"]);
       expect(warnings).toEqual(
         followUpFails
           ? [