fix(fork): backport file safety and Codex reliability updates

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts

@@ -1015,6 +1015,7 @@ describe("CodexAdapterV2 session initialize", () => {
             runtimePolicy: CODEX_TEST_RUNTIME_POLICY,
           }),
         initializeRequests: () => initializeRequests,
+        events: runtime.events,
       };
     });
 
@@ -1162,6 +1163,92 @@ describe("CodexAdapterV2 session initialize", () => {
     ),
   );
 
+  it.effect("fails the open when initialize never answers during open", () =>
+    Effect.gen(function* () {
+      const initializeAwaitingResponse = yield* Deferred.make<void>();
+      const opened = yield* openReplaySession(
+        makeCodexReplayTranscript({
+          scenario: "initialize-hung-open",
+          entries: replayPreamble("initialize-hung-open").slice(0, 2),
+        }),
+        (entry) =>
+          entry.label === "initialize"
+            ? Deferred.succeed(initializeAwaitingResponse, undefined).pipe(
+                Effect.andThen(Effect.never),
+              )
+            : Effect.void,
+        () => Effect.die("initialize never answers"),
+      ).pipe(Effect.flip, Effect.forkChild({ startImmediately: true }));
+      yield* Deferred.await(initializeAwaitingResponse);
+      yield* TestClock.adjust("30 seconds");
+
+      assert.isDefined(opened.pollUnsafe(), "initialize must fail within its deadline");
+      // The session manager discards a session whose open fails.
+      assert.instanceOf(yield* Fiber.join(opened), ProviderAdapter.ProviderAdapterOpenSessionError);
+    }).pipe(
+      Effect.scoped,
+      Effect.provide(
+        Layer.mergeAll(
+          IdAllocator.layer,
+          McpProviderSessions.layer,
+          NodeServices.layer,
+          NodeCrypto.layer,
+        ),
+      ),
+    ),
+  );
+
+  it.effect("retires the shared session when initialize never answers", () =>
+    Effect.gen(function* () {
+      const initializeAwaitingResponse = yield* Deferred.make<void>();
+      const session = yield* openReplaySession(
+        makeCodexReplayTranscript({
+          scenario: "initialize-hung",
+          entries: replayPreamble("initialize-hung").slice(0, 2),
+        }),
+        (entry) =>
+          entry.label === "initialize"
+            ? Deferred.succeed(initializeAwaitingResponse, undefined).pipe(
+                Effect.andThen(Effect.never),
+              )
+            : Effect.void,
+      );
+      const eventsExit = yield* Stream.runDrain(session.events).pipe(
+        Effect.exit,
+        Effect.forkChild({ startImmediately: true }),
+      );
+
+      const ensured = yield* session
+        .ensureThread("thread-initialize-hung")
+        .pipe(Effect.flip, Effect.forkChild({ startImmediately: true }));
+      yield* Deferred.await(initializeAwaitingResponse);
+      const waiting = yield* session
+        .ensureThread("thread-initialize-hung-waiting")
+        .pipe(Effect.flip, Effect.forkChild({ startImmediately: true }));
+      yield* TestClock.adjust("30 seconds");
+
+      assert.isDefined(ensured.pollUnsafe(), "the hung initializer must fail");
+      assert.isDefined(waiting.pollUnsafe(), "waiting threads must fail with the shared session");
+      assert.equal((yield* Fiber.join(ensured))._tag, "ProviderAdapterEnsureThreadError");
+      // A waiter fails at once instead of re-sending initialize to the hung app-server.
+      assert.equal((yield* Fiber.join(waiting))._tag, "ProviderAdapterEnsureThreadError");
+      assert.equal(session.initializeRequests(), 1);
+      // The session manager retires a session whose event stream fails.
+      const exit = yield* Fiber.join(eventsExit);
+      assert.isTrue(Exit.isFailure(exit));
+    }).pipe(
+      Effect.scoped,
+      Effect.provide(
+        Layer.mergeAll(
+          IdAllocator.layer,
+          McpProviderSessions.layer,
+          NodeServices.layer,
+          NodeCrypto.layer,
+        ),
+      ),
+    ),
+  );
+
   it.effect("completes the handshake after a caller is interrupted mid-initialize", () =>
     Effect.gen(function* () {
       const initializeAwaitingResponse = yield* Deferred.make<void>();

apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts

@@ -232,6 +232,8 @@ const decodeCodexBackgroundTerminalTerminateResponse = Schema.decodeUnknownEffec
 const decodeCodexBackgroundTerminalsListResponse = Schema.decodeUnknownEffect(
   CodexBackgroundTerminalsListResponse,
 );
+// Initialize normally answers in under a second; leave room for a slow first launch.
+const CODEX_INITIALIZE_TIMEOUT = "30 seconds";
 const CodexThreadSampleResponse = Schema.Struct({ text: Schema.NullOr(Schema.String) });
 const decodeCodexThreadSampleResponse = Schema.decodeUnknownEffect(CodexThreadSampleResponse);
 export const CODEX_PROMPT_SUGGESTION_PROMPT =
@@ -1803,7 +1805,15 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
             }),
           ),
         );
+        const events = yield* Queue.unbounded<
+          ProviderAdapter.ProviderAdapterV2Event,
+          ProviderAdapter.ProviderAdapterEventStreamError
+        >();
         const initialized = yield* Ref.make(false);
+        // After a timeout the session is retiring; waiters must not re-send to the hung app-server.
+        const initializeTimeout = yield* Ref.make<
+          ProviderAdapter.ProviderAdapterEventStreamError | undefined
+        >(undefined);
         // Codex rejects a second initialize on the shared app-server.
         const initializePermit = yield* Semaphore.make(1);
         const ensureInitialized = initializePermit.withPermit(
@@ -1812,6 +1822,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
             if (alreadyInitialized) {
               return;
             }
+            const timedOut = yield* Ref.get(initializeTimeout);
+            if (timedOut !== undefined) return yield* timedOut;
 
             const userAgent = yield* client
               .request("initialize", {
@@ -1827,6 +1839,23 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
                       ? Effect.succeed(null)
                       : Effect.fail(error),
                 }),
+                // A hung app-server never answers, and every thread on this shared
+                // session waits behind the permit. Failing the event stream retires
+                // the session, so the next turn spawns a fresh app-server.
+                Effect.timeoutOrElse({
+                  duration: CODEX_INITIALIZE_TIMEOUT,
+                  orElse: () => {
+                    const error = new ProviderAdapter.ProviderAdapterEventStreamError({
+                      driver: CODEX_PROVIDER,
+                      providerSessionId: input.providerSessionId,
+                      cause: `Codex app-server did not answer initialize within ${CODEX_INITIALIZE_TIMEOUT}.`,
+                    });
+                    return Ref.set(initializeTimeout, error).pipe(
+                      Effect.andThen(Queue.fail(events, error)),
+                      Effect.andThen(Effect.fail(error)),
+                    );
+                  },
+                }),
               );
             const version = userAgent && /\/(\S+)/u.exec(userAgent)?.[1];
             const advisory =
@@ -1856,10 +1885,6 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* (
           model: input.modelSelection.model,
           now,
         });
-        const events = yield* Queue.unbounded<
-          ProviderAdapter.ProviderAdapterV2Event,
-          ProviderAdapter.ProviderAdapterEventStreamError
-        >();
         // Drain final notifications before the session manager retires the dead process.
         // Queue.fail lets readers take items already buffered before they see the failure.
         yield* client.awaitTermination.pipe(

apps/server/src/project/ProjectService.deletion.test.ts

@@ -28,6 +28,7 @@ import * as ProjectStore from "../orchestration-v2/ProjectStore.ts";
 import * as ThreadCommandExecutor from "../orchestration-v2/ThreadCommandExecutor.ts";
 import { planThreadDeletion } from "../orchestration-v2/ThreadDeletion.ts";
 import * as SqlitePersistence from "../persistence/Sqlite.ts";
+import * as ProjectSettingsOverrideSweep from "../provider/ProjectSettingsOverrideSweep.ts";
 import * as ServerSettings from "../serverSettings.ts";
 import * as WorkspacePaths from "../workspace/WorkspacePaths.ts";
 import * as ProjectEnrichmentService from "./ProjectEnrichmentService.ts";
@@ -501,9 +502,13 @@ it.effect("deletes a project without force once its imported threads were delete
   }).pipe(Effect.provide(layerDatabase)),
 );
 
-it.effect(
-  "clears a deleted project's settings overrides so a machine-disabled instance cannot stay alive",
-  () =>
+it.effect.each([
+  { policy: "no", worktreeCleanup: undefined },
+  { policy: "its off", worktreeCleanup: { mode: "off" } as const },
+  { policy: "its legacy", worktreeCleanup: null },
+])(
+  "clears a deleted project's settings overrides, keeping $policy worktree cleanup policy, so a machine-disabled instance cannot stay alive",
+  ({ worktreeCleanup }) =>
     Effect.gen(function* () {
       const projectId = ProjectId.make("project:overrides-cleanup");
       const instanceId = ProviderInstanceId.make("codex_work");
@@ -514,7 +519,11 @@ it.effect(
         const serverSettings = yield* ServerSettings.ServerSettingsService;
         yield* serverSettings.updateSettings({
           projectSettingsOverrides: {
-            [projectId]: { providerInstances: { [instanceId]: { enabled: true } } },
+            [projectId]: {
+              providerInstances: { [instanceId]: { enabled: true } },
+              defaultAutoPull: true,
+              ...(worktreeCleanup === undefined ? {} : { worktreeCleanup }),
+            },
           },
         });
 
@@ -523,8 +532,35 @@ it.effect(
           projectId,
         });
 
-        const settingsAfterDelete = yield* serverSettings.getSettings;
-        assert.isUndefined(settingsAfterDelete.projectSettingsOverrides[projectId]);
+        const entry = (yield* serverSettings.getSettings).projectSettingsOverrides[projectId];
+        if (worktreeCleanup === undefined) {
+          assert.isUndefined(entry);
+        } else {
+          assert.deepStrictEqual(entry, { worktreeCleanup, providerInstances: {} });
+        }
+
+        const liveProjectId = ProjectId.make("project:still-live");
+        const missingProjectId = ProjectId.make("project:missing");
+        yield* seedProject(liveProjectId);
+        // Older deletions or interrupted best-effort cleanup can leave the original overrides on restart.
+        yield* serverSettings.updateSettings({
+          projectSettingsOverrides: {
+            [projectId]: {
+              providerInstances: { [instanceId]: { enabled: true } },
+              defaultAutoPull: true,
+              ...(worktreeCleanup === undefined ? {} : { worktreeCleanup }),
+            },
+            [liveProjectId]: { defaultAutoPull: true },
+            [missingProjectId]: { worktreeCleanup: { mode: "off" } },
+          },
+        });
+        yield* ProjectSettingsOverrideSweep.sweep;
+        assert.deepStrictEqual((yield* serverSettings.getSettings).projectSettingsOverrides, {
+          [liveProjectId]: { defaultAutoPull: true },
+          ...(worktreeCleanup === undefined
+            ? {}
+            : { [projectId]: { worktreeCleanup, providerInstances: {} } }),
+        });
       }).pipe(Effect.provide(layerServices));
     }).pipe(Effect.provide(layerDatabase)),
 );

apps/server/src/project/ProjectService.ts

@@ -34,6 +34,7 @@ import * as ProjectStore from "../orchestration-v2/ProjectStore.ts";
 import * as ThreadCommandExecutor from "../orchestration-v2/ThreadCommandExecutor.ts";
 import { planThreadDeletion } from "../orchestration-v2/ThreadDeletion.ts";
 import * as ServerSettings from "../serverSettings.ts";
+import { deletedProjectOverridePatch } from "../provider/ProjectSettingsOverrideSweep.ts";
 import { makeIsManagedWorktreePath } from "../worktreesDirectory.ts";
 import * as ProjectEnrichmentService from "./ProjectEnrichmentService.ts";
 import * as WorkspacePaths from "../workspace/WorkspacePaths.ts";
@@ -535,7 +536,14 @@ export const make = Effect.gen(function* () {
 
   // Best-effort cleanup releases disabled instances kept alive by overrides; crashes have no replay to revisit deletion.
   const clearProjectSettingsOverrides = (projectId: ProjectId) =>
-    serverSettings.updateSettings({ projectSettingsOverrides: { [projectId]: null } }).pipe(
+    serverSettings.getSettings.pipe(
+      Effect.flatMap(({ projectSettingsOverrides }) =>
+        serverSettings.updateSettings({
+          projectSettingsOverrides: {
+            [projectId]: deletedProjectOverridePatch(projectSettingsOverrides[projectId]),
+          },
+        }),
+      ),
       Effect.asVoid,
       Effect.catchCause((cause) =>
         Effect.logWarning(

apps/server/src/provider/CodexProvider.ts

@@ -56,7 +56,6 @@ import {
 import * as HostProcess from "@t3tools/shared/HostProcess";
 import packageJson from "../../package.json" with { type: "json" };
 const isCodexAppServerSpawnError = Schema.is(CodexErrors.CodexAppServerSpawnError);
-const RATE_LIMITS_PROBE_TIMEOUT_MS = 3_000;
 
 type CodexRateLimitsProbe =
   | {
@@ -589,7 +588,6 @@ const probeCodexAppServerProvider = Effect.fn("probeCodexAppServerProvider")(fun
         : withCodexDiscoveryDeadline(
             client.request("account/rateLimits/read", null),
             deadlineMs,
-            RATE_LIMITS_PROBE_TIMEOUT_MS,
           ).pipe(
             Effect.map((response): CodexRateLimitsProbe =>
               response

apps/server/src/provider/ProjectSettingsOverrideSweep.ts

@@ -1,32 +1,41 @@
-import { type ProjectId, type ServerSettings } from "@t3tools/contracts";
+import * as NodeUtil from "node:util";
+
+import { type ProjectSettingsOverrides } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
 
 import * as ProjectStore from "../orchestration-v2/ProjectStore.ts";
 import * as Settings from "../serverSettings.ts";
 
-const findOrphanedProjectIds = (
-  settings: Pick<ServerSettings, "projectSettingsOverrides">,
-  liveProjectIds: ReadonlySet<ProjectId>,
-): ReadonlyArray<ProjectId> =>
-  Object.keys(settings.projectSettingsOverrides)
-    .map((projectId) => projectId as ProjectId)
-    .filter((projectId) => !liveProjectIds.has(projectId));
+export function deletedProjectOverridePatch(entry: ProjectSettingsOverrides | undefined) {
+  // Keep cleanup policy for deleted threads; omitting providerInstances would preserve obsolete references in the merge.
+  return entry?.worktreeCleanup !== undefined
+    ? { worktreeCleanup: entry.worktreeCleanup, providerInstances: {} }
+    : null;
+}
 
 export const sweep = Effect.gen(function* () {
   const serverSettings = yield* Settings.ServerSettingsService;
   const projectStore = yield* ProjectStore.ProjectStoreV2;
 
   const settings = yield* serverSettings.getSettings;
-  const liveShells = yield* projectStore.listShells();
-  const liveProjectIds = new Set(liveShells.map((shell) => shell.id));
+  const projects = yield* projectStore.list({ includeDeleted: true });
+  const projectsById = new Map<string, ProjectStore.ProjectRow>(
+    projects.map((project) => [project.projectId, project]),
+  );
 
-  const orphanedProjectIds = findOrphanedProjectIds(settings, liveProjectIds);
-  if (orphanedProjectIds.length === 0) return;
+  const overrides = Object.entries(settings.projectSettingsOverrides).flatMap(
+    ([projectId, entry]) => {
+      const project = projectsById.get(projectId);
+      if (project?.deletedAt === null) return [];
+      const patch = project ? deletedProjectOverridePatch(entry) : null;
+      if (NodeUtil.isDeepStrictEqual(entry, patch)) return [];
+      return [[projectId, patch] as const];
+    },
+  );
+  if (overrides.length === 0) return;
 
   yield* serverSettings.updateSettings({
-    projectSettingsOverrides: Object.fromEntries(
-      orphanedProjectIds.map((projectId) => [projectId, null]),
-    ),
+    projectSettingsOverrides: Object.fromEntries(overrides),
   });
 }).pipe(
   Effect.asVoid,