fix(scheduler): isolate persisted state scopes

loopx/control_plane/scheduler/state_store.ts

@@ -1,5 +1,5 @@
 import { createHash } from "node:crypto";
-import { readFile } from "node:fs/promises";
+import { readFile, unlink } from "node:fs/promises";
 import { homedir } from "node:os";
 import { join } from "node:path";
 
@@ -34,6 +34,8 @@ export const CODEX_APP_SURFACE = "codex_app";
 
 const HOST_UPDATE_FAILURE_CACHE_LIMIT = 4;
 const HOST_UPDATE_FAILURE_TTL_MS = 24 * 60 * 60 * 1_000;
+const SCOPE_SEGMENT_LABEL_LIMIT = 47;
+const SCOPE_SEGMENT_HASH_LENGTH = 16;
 
 export const SCHEDULER_STATE_OPERATIONS = [
   "rrule_for_minutes",
@@ -348,6 +350,19 @@ function safeSegment(value: unknown): string {
   return safe || "default";
 }
 
+function stableHash(value: string, length: number): string {
+  return createHash("sha256")
+    .update(value, "utf8")
+    .digest("hex")
+    .slice(0, length);
+}
+
+function scopedSegment(value: unknown): string {
+  const raw = trimmed(value);
+  const label = safeSegment(raw).slice(0, SCOPE_SEGMENT_LABEL_LIMIT);
+  return `${label}-${stableHash(raw, SCOPE_SEGMENT_HASH_LENGTH)}`;
+}
+
 function expandUser(path: string): string {
   if (path === "~") return homedir();
   if (path.startsWith("~/") || path.startsWith("~\\")) {
@@ -360,18 +375,30 @@ export function schedulerStatePath(
   runtimeRoot: string,
   scope: SchedulerScope,
 ): string {
-  const stateHash = createHash("sha256")
-    .update(scope.stateKey, "utf8")
-    .digest("hex")
-    .slice(0, 16);
+  const stateHash = stableHash(scope.stateKey, 16);
+  return join(
+    expandUser(runtimeRoot),
+    "goals",
+    scopedSegment(scope.goalId),
+    "scheduler-state",
+    scopedSegment(scope.agentId),
+    scopedSegment(scope.surface),
+    `${stateHash}.json`,
+  );
+}
+
+function legacySchedulerStatePath(
+  runtimeRoot: string,
+  scope: SchedulerScope,
+): string {
   return join(
     expandUser(runtimeRoot),
     "goals",
     safeSegment(scope.goalId),
     "scheduler-state",
     safeSegment(scope.agentId),
     safeSegment(scope.surface),
-    `${stateHash}.json`,
+    `${stableHash(scope.stateKey, 16)}.json`,
   );
 }
 
@@ -453,28 +480,80 @@ function storeRequest(value: unknown, operation: "load" | "write"): {
   request: JsonObject;
   scope: SchedulerScope;
   path: string;
+  legacyPath: string;
 } {
   const request = requiredObject(value, `scheduler.state.${operation} params`);
   if (request.schema_version !== SCHEDULER_STATE_STORE_REQUEST_SCHEMA) {
     throw new Error("Scheduler state store request schema mismatch");
   }
   const scope = schedulerScope(request);
-  const path = schedulerStatePath(
-    requiredString(request.runtime_root, "runtime_root"),
-    scope,
+  const runtimeRoot = requiredString(request.runtime_root, "runtime_root");
+  const path = schedulerStatePath(runtimeRoot, scope);
+  const legacyPath = legacySchedulerStatePath(runtimeRoot, scope);
+  return { request, scope, path, legacyPath };
+}
+
+async function migrateLegacySchedulerState(
+  scope: SchedulerScope,
+  path: string,
+  legacyPath: string,
+): Promise<JsonObject | null> {
+  // Probe the legacy file before creating any directory or taking the legacy
+  // lock. The legacy layout uses unbounded safeSegment labels, so an overlong
+  // scope component can make the legacy path unaddressable (ENAMETOOLONG) or
+  // simply absent (ENOENT); either way there is nothing to migrate and the
+  // canonical path must be reported as missing rather than raising an effect
+  // rejection from mkdir on the unbounded legacy directory.
+  try {
+    await readFile(legacyPath, "utf8");
+  } catch (error) {
+    const code = (error as NodeJS.ErrnoException).code;
+    if (code === "ENOENT" || code === "ENAMETOOLONG") return null;
+    throw error;
+  }
+  return await withFileMutationLock(legacyPath, async () =>
+    await withFileMutationLock(path, async () => {
+      try {
+        const current = normalizeSchedulerState(
+          JSON.parse(await readFile(path, "utf8")),
+          scope,
+        );
+        if (current) return current;
+      } catch {
+        // The canonical path is still absent or unusable; inspect legacy state.
+      }
+      let legacyState: JsonObject | null = null;
+      try {
+        legacyState = normalizeSchedulerState(
+          JSON.parse(await readFile(legacyPath, "utf8")),
+          scope,
+        );
+      } catch {
+        return null;
+      }
+      if (!legacyState) return null;
+      await atomicWriteJson(path, stableValue(legacyState) as JsonObject);
+      try {
+        await unlink(legacyPath);
+      } catch (error) {
+        if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error;
+      }
+      return legacyState;
+    })
   );
-  return { request, scope, path };
 }
 
 export async function loadSchedulerState(
   value: unknown,
 ): Promise<SchedulerStateLoadResult> {
-  const { scope, path } = storeRequest(value, "load");
+  const { scope, path, legacyPath } = storeRequest(value, "load");
   let state: JsonObject | null = null;
   try {
     state = normalizeSchedulerState(JSON.parse(await readFile(path, "utf8")), scope);
-  } catch {
-    state = null;
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code === "ENOENT") {
+      state = await migrateLegacySchedulerState(scope, path, legacyPath);
+    }
   }
   return {
     schema_version: SCHEDULER_STATE_STORE_RESULT_SCHEMA,

tests/control_plane_ts/scheduler_state_store.test.ts

@@ -1,7 +1,7 @@
 import assert from "node:assert/strict";
-import { mkdtemp, readFile, readdir, rm, writeFile } from "node:fs/promises";
+import { mkdir, mkdtemp, readFile, readdir, rm, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
-import { join } from "node:path";
+import { dirname, join } from "node:path";
 import test from "node:test";
 
 import {
@@ -182,15 +182,115 @@ test("state path is scoped, sanitized, and stable", () => {
     join(
       "/runtime",
       "goals",
-      "goal-with-spaces",
+      "goal-with-spaces-116e9296329bcdc8",
       "scheduler-state",
-      "agent-..-..-other",
-      "codex_app",
+      "agent-..-..-other-55571d8866830ec6",
+      "codex_app-b32e6f37f5dad64e",
       "d9ad89d416adc9b2.json",
     ),
   );
 });
 
+test("sanitized-equivalent scheduler scopes have distinct bounded paths", () => {
+  const goalPaths = ["a b", "a-b", "非 ASCII", "---"].map((goalId) =>
+    schedulerStatePath("/runtime", { ...scope, goalId })
+  );
+  const agentPaths = ["agent/name", "agent-name"].map((agentId) =>
+    schedulerStatePath("/runtime", { ...scope, agentId })
+  );
+  const surfacePaths = ["codex app", "codex-app"].map((surface) =>
+    schedulerStatePath("/runtime", { ...scope, surface })
+  );
+  assert.equal(new Set(goalPaths).size, goalPaths.length);
+  assert.equal(new Set(agentPaths).size, agentPaths.length);
+  assert.equal(new Set(surfacePaths).size, surfacePaths.length);
+  for (const path of [...goalPaths, ...agentPaths, ...surfacePaths]) {
+    for (const segment of path.split("/")) assert.ok(segment.length <= 64);
+  }
+});
+
+test("legacy scheduler state is migrated once into the collision-safe layout", async (t) => {
+  const runtimeRoot = await mkdtemp(join(tmpdir(), "loopx-scheduler-state-"));
+  t.after(() => rm(runtimeRoot, { recursive: true, force: true }));
+  const legacyPath = join(
+    runtimeRoot,
+    "goals",
+    scope.goalId,
+    "scheduler-state",
+    scope.agentId,
+    scope.surface,
+    "d9ad89d416adc9b2.json",
+  );
+  await writeSchedulerState(storeRequest(runtimeRoot, { state: state() }));
+  const canonicalPath = schedulerStatePath(runtimeRoot, scope);
+  const persisted = await readFile(canonicalPath, "utf8");
+  await rm(canonicalPath);
+  await mkdir(dirname(legacyPath), { recursive: true });
+  await writeFile(legacyPath, persisted, "utf8");
+
+  const migrated = await loadSchedulerState(storeRequest(runtimeRoot));
+
+  assert.deepEqual(migrated.state, state());
+  assert.equal(migrated.path, canonicalPath);
+  assert.deepEqual(JSON.parse(await readFile(canonicalPath, "utf8")), state());
+  await assert.rejects(readFile(legacyPath, "utf8"), { code: "ENOENT" });
+});
+
+test("overlong legacy scope loads as missing state, not an effect rejection", async (t) => {
+  const runtimeRoot = await mkdtemp(join(tmpdir(), "loopx-scheduler-state-"));
+  t.after(() => rm(runtimeRoot, { recursive: true, force: true }));
+  // A legacy scope component beyond the filesystem path-component limit cannot
+  // exist in the legacy layout; the canonical bounded path has no state yet,
+  // so the first load must return missing state instead of ENAMETOOLONG.
+  const overlongGoalId = "g".repeat(300);
+  const request = storeRequest(runtimeRoot, { goal_id: overlongGoalId });
+
+  const result = await loadSchedulerState(request);
+
+  assert.equal(result.state, null);
+  const canonical = schedulerStatePath(runtimeRoot, {
+    ...scope,
+    goalId: overlongGoalId,
+  });
+  assert.ok(
+    !canonical.includes("g".repeat(256)),
+    "canonical path components must stay bounded",
+  );
+  assert.ok(
+    Math.max(...canonical.split("/").map((part) => part.length)) <= 64,
+    "canonical path component must not exceed the scoped-segment bound",
+  );
+});
+
+test("concurrent writes for sanitized-equivalent scopes stay isolated", async (t) => {
+  const runtimeRoot = await mkdtemp(join(tmpdir(), "loopx-scheduler-state-"));
+  t.after(() => rm(runtimeRoot, { recursive: true, force: true }));
+  const firstGoalId = "a b";
+  const secondGoalId = "a-b";
+  const firstState = { ...state(), goal_id: firstGoalId };
+  const secondState = { ...state(), goal_id: secondGoalId };
+
+  await Promise.all([
+    writeSchedulerState(storeRequest(runtimeRoot, {
+      goal_id: firstGoalId,
+      state: firstState,
+    })),
+    writeSchedulerState(storeRequest(runtimeRoot, {
+      goal_id: secondGoalId,
+      state: secondState,
+    })),
+  ]);
+
+  assert.deepEqual(
+    (await loadSchedulerState(storeRequest(runtimeRoot, { goal_id: firstGoalId }))).state,
+    firstState,
+  );
+  assert.deepEqual(
+    (await loadSchedulerState(storeRequest(runtimeRoot, { goal_id: secondGoalId }))).state,
+    secondState,
+  );
+});
+
 test("state store writes atomically and recognizes replay", async (t) => {
   const runtimeRoot = await mkdtemp(join(tmpdir(), "loopx-scheduler-state-"));
   t.after(() => rm(runtimeRoot, { recursive: true, force: true }));