refactor(codex): split app-server root turn detection

Author: kushagrchitkar

docs/intentional-architecture-rewrite-2026-06-27/worklog.md

@@ -659,3 +659,9 @@ Ninety-seventh production slice:
 - Split sync workflow contracts into `src/services/sync/types.ts`.
 - Moved `SyncWorkflowOptions`, `SyncWorkflowSummary`, item shapes, and result union out of `src/services/sync/sync.ts`.
 - Kept `src/services/sync/sync.ts` focused on source parsing, sweep orchestration, provider selection, and Absorb handoff.
+
+Ninety-eighth production slice:
+
+- Split Codex app-server root turn detection into `src/harness/providers/codex/app-server-root-turn.ts`.
+- Moved root turn/thread completion checks and notification turn ID extraction out of the app-server process run loop.
+- Kept `src/harness/providers/codex/app-server.ts` focused on process lifecycle, RPC wiring, notification mapping, timeout handling, and result finalization.

src/harness/providers/codex/app-server-root-turn.ts

@@ -0,0 +1,44 @@
+import {
+  asRecord,
+  stringField,
+} from "./fields.js";
+import type { CodexRunState, JsonRpcNotification } from "./types.js";
+
+export function codexNotificationTurnId(
+  message: JsonRpcNotification,
+): string | undefined {
+  return stringField(asRecord(message.params), "turnId");
+}
+
+export function isCodexRootTurnCompletion(args: {
+  message: JsonRpcNotification;
+  state: CodexRunState;
+  activeTurnId: string | undefined;
+}): boolean | undefined {
+  if (args.message.method !== "turn/completed") return undefined;
+
+  const params = asRecord(args.message.params);
+  const completedTurnId = stringField(params, "turnId");
+  const completedThreadId = stringField(params, "threadId");
+  const isRootTurn =
+    (args.state.rootTurnId !== undefined &&
+      completedTurnId === args.state.rootTurnId) ||
+    (args.state.rootTurnId === undefined &&
+      args.activeTurnId !== undefined &&
+      completedTurnId === args.activeTurnId);
+  const isRootThread =
+    args.state.rootThreadId !== undefined &&
+    completedThreadId === args.state.rootThreadId;
+  return isRootTurn || isRootThread;
+}
+
+export function isCodexRootThreadNotification(args: {
+  message: JsonRpcNotification;
+  state: CodexRunState;
+}): boolean {
+  const threadId = stringField(asRecord(args.message.params), "threadId");
+  return (
+    args.state.rootThreadId !== undefined &&
+    threadId === args.state.rootThreadId
+  );
+}

src/harness/providers/codex/app-server.ts

@@ -11,12 +11,13 @@ import {
   type CodexAppServerSandboxMode,
 } from "./app-server-config.js";
 import { createCodexAppServerRpcTransport } from "./app-server-rpc.js";
+import {
+  codexNotificationTurnId,
+  isCodexRootThreadNotification,
+  isCodexRootTurnCompletion,
+} from "./app-server-root-turn.js";
 import { startCodexAppServerTurn } from "./app-server-session.js";
 import { classifyCodexFailure } from "./failures.js";
-import {
-  asRecord,
-  stringField,
-} from "./fields.js";
 import { toHarnessResult } from "./result.js";
 import { respondToCodexServerRequest } from "./server-requests.js";
 import type { CodexRunState, JsonRpcNotification } from "./types.js";
@@ -104,15 +105,19 @@ export async function runCodexAppServer(
     };
 
     const handleNotification = (message: JsonRpcNotification): void => {
-      const turnId = stringField(asRecord(message.params), "turnId");
+      const turnId = codexNotificationTurnId(message);
       if (
         activeTurnId === undefined &&
         turnId !== undefined &&
-        isRootThreadNotification(message, state)
+        isCodexRootThreadNotification({ message, state })
       ) {
         activeTurnId = turnId;
       }
-      const isRootCompletion = isRootTurnCompletion(message, state, activeTurnId);
+      const isRootCompletion = isCodexRootTurnCompletion({
+        message,
+        state,
+        activeTurnId,
+      });
       const events = mapCodexAppServerNotification(message, state, {
         activeTurnId,
         rootThreadId: state.rootThreadId,
@@ -239,33 +244,3 @@ async function terminateManagedChildSafely(
     return `Provider process cleanup failed: ${err instanceof Error ? err.message : String(err)}`;
   }
 }
-
-function isRootTurnCompletion(
-  message: JsonRpcNotification,
-  state: CodexRunState,
-  activeTurnId: string | undefined,
-): boolean | undefined {
-  if (message.method !== "turn/completed") return undefined;
-  const params = asRecord(message.params);
-  const completedTurnId = stringField(params, "turnId");
-  const completedThreadId = stringField(params, "threadId");
-  const isRootTurn =
-    (state.rootTurnId !== undefined && completedTurnId === state.rootTurnId) ||
-    (state.rootTurnId === undefined &&
-      activeTurnId !== undefined &&
-      completedTurnId === activeTurnId);
-  const isRootThread =
-    state.rootThreadId !== undefined && completedThreadId === state.rootThreadId;
-  return isRootTurn || isRootThread;
-}
-
-function isRootThreadNotification(
-  message: JsonRpcNotification,
-  state: CodexRunState,
-): boolean {
-  const completedThreadId = stringField(asRecord(message.params), "threadId");
-  return (
-    state.rootThreadId !== undefined &&
-    completedThreadId === state.rootThreadId
-  );
-}

test/architecture-boundaries.test.ts

@@ -852,11 +852,15 @@ describe("architecture boundaries", () => {
     const appServerSession = await readSource(
       "src/harness/providers/codex/app-server-session.ts",
     );
+    const appServerRootTurn = await readSource(
+      "src/harness/providers/codex/app-server-root-turn.ts",
+    );
 
     expect(existsSync(join(ROOT, "src/harness/providers/codex/app-server-config.ts"))).toBe(true);
     expect(existsSync(join(ROOT, "src/harness/providers/codex/server-requests.ts"))).toBe(true);
     expect(existsSync(join(ROOT, "src/harness/providers/codex/app-server-session.ts"))).toBe(true);
     expect(existsSync(join(ROOT, "src/harness/providers/codex/app-server-rpc.ts"))).toBe(true);
+    expect(existsSync(join(ROOT, "src/harness/providers/codex/app-server-root-turn.ts"))).toBe(true);
     expect(appServer).not.toContain("CODEALMANAC_CODEX_APP_SERVER");
     expect(appServer).not.toContain("function parsePositiveEnvInt");
     expect(appServer).not.toContain("interface PendingRequest");
@@ -871,13 +875,17 @@ describe("architecture boundaries", () => {
     expect(appServer).not.toContain("requestRpc(\"initialize\"");
     expect(appServer).not.toContain("requestRpc(\"thread/start\"");
     expect(appServer).not.toContain("requestRpc(\"turn/start\"");
+    expect(appServer).not.toContain("function isRootTurnCompletion");
+    expect(appServer).not.toContain("function isRootThreadNotification");
     expect(appServerSession).toContain("startCodexAppServerTurn");
     expect(appServerSession).toContain("requestRpc(\"initialize\"");
     expect(appServerSession).toContain("requestRpc(\"thread/start\"");
     expect(appServerSession).toContain("requestRpc(\"turn/start\"");
     expect(appServerRpc).toContain("interface PendingRequest");
     expect(appServerRpc).toContain("pending.set");
     expect(appServerRpc).toContain("function handleResponse");
+    expect(appServerRootTurn).toContain("isCodexRootTurnCompletion");
+    expect(appServerRootTurn).toContain("isCodexRootThreadNotification");
   });
 
   it("keeps migrate legacy-sources adapter out of source migration mechanics", async () => {