Start: launch the /factory session for a claimed issue

apps/server/src/factory/index.test.ts

@@ -18,6 +18,7 @@ import { utcSecond } from "./runner/start.ts";
 import { MARKER } from "./tracker/github.ts";
 import { FactoryProcess, RunSpawnError, type RunInput, type RunResult } from "./process.ts";
 import * as Paths from "./t3/paths.ts";
+import * as Sessions from "./t3/sessions.ts";
 
 /** A process runner on a machine with no `gh`: every call fails to start. */
 const noGh = Layer.succeed(
@@ -27,6 +28,18 @@ const noGh = Layer.succeed(
   }),
 );
 
+/** A FactorySessions that finds no T3 project and dies on anything else. */
+const stubSessions = Sessions.FactorySessions.of({
+  findProject: () => Effect.succeed(null),
+  launch: () => Effect.die("not launched"),
+  status: () => Effect.die("no status"),
+  thread: () => Effect.die("no thread"),
+  send: () => Effect.die("not sent"),
+  archive: () => Effect.die("not archived"),
+  removeWorktree: () => Effect.die("not removed"),
+});
+const noSessions = Layer.succeed(Sessions.FactorySessions, stubSessions);
+
 const messageText = (message: unknown): string => {
   const first = Array.isArray(message) ? message[0] : message;
   return typeof first === "string" ? first : String(first);
@@ -76,6 +89,7 @@ const logsFor = (
               }),
             ),
             Layer.provide(gh),
+            Layer.provide(noSessions),
           ),
         ),
       ).pipe(Effect.provide(Logger.layer([capture], { mergeWithExisting: false })));
@@ -205,7 +219,7 @@ describe("liveWorld", () => {
   };
 
   const worldOf = (process: FactoryProcess["Service"]) =>
-    liveWorld(machine, "/factory-test/data", process, () => NOW, "mac-mini");
+    liveWorld(machine, "/factory-test/data", process, stubSessions, () => NOW, "mac-mini");
 
   const repo: Repo = {
     watched: machine.repos[0]!,

apps/server/src/factory/index.ts

@@ -30,8 +30,17 @@ import { runPass, type PassError, type PassWorld, type Repo } from "./pass.ts";
 import * as Process from "./process.ts";
 import { FactoryProcess } from "./process.ts";
 import { CUT_SHORT_COMMENT } from "./runner/messages.ts";
-import { start, utcSecond, type Tracker } from "./runner/start.ts";
+import {
+  isOurs,
+  resume,
+  start,
+  utcSecond,
+  type Sessions,
+  type StartContext,
+  type Tracker,
+} from "./runner/start.ts";
 import * as Paths from "./t3/paths.ts";
+import * as T3Sessions from "./t3/sessions.ts";
 import { bindProcess, makeGitHub, type GitHubError, type IssueSummary } from "./tracker/github.ts";
 
 /** The running factory, for the code that wakes it (a thread's run changing, a person's request). */
@@ -66,10 +75,7 @@ const unassignedByNumber = (items: ReadonlyArray<IssueSummary>) =>
 /** The numbers of the issues `user` is the only assignee of, by number. */
 const assignedOnlyTo = (items: ReadonlyArray<IssueSummary>, user: string) =>
   items
-    .filter(
-      (item) =>
-        item.assignees.length === 1 && item.assignees[0]!.toLowerCase() === user.toLowerCase(),
-    )
+    .filter((item) => isOurs(item.assignees, user))
     .map((item) => item.number)
     .sort(byNumber);
 
@@ -108,6 +114,7 @@ export const liveWorld = (
   machine: MachineConfig,
   dataDir: string,
   process: FactoryProcess["Service"],
+  sessions: Sessions,
   clock: () => number,
   host: string,
 ): PassWorld => {
@@ -118,6 +125,19 @@ export const liveWorld = (
   const trackerFor = (watched: WatchedRepo): Tracker => bindProcess(githubFor(watched), process);
   const tracker = (effect: Effect.Effect<unknown, GitHubError, FactoryProcess>) =>
     withProcess(effect).pipe(Effect.mapError(asPassError("tracker")));
+  const startContext = (repo: Repo): StartContext => ({
+    repo: repo.watched.repo,
+    trackerUser: repo.trackerUser,
+    machine: host,
+    project: repo.project,
+    dataDir,
+    tracker: trackerFor(repo.watched),
+    repoPath: repo.watched.path,
+    ghConfigDir: machine.ghConfigDir,
+    model: { instanceId: machine.modelInstance, model: machine.model },
+    sessions,
+    clock,
+  });
   return {
     project: (watched) => withProcess(projectFromBase(githubFor(watched))),
     trackerUser: (watched) =>
@@ -164,19 +184,8 @@ export const liveWorld = (
       withProcess(githubFor(repo.watched).textBlockers(issue)).pipe(
         Effect.mapError(asPassError("tracker")),
       ),
-    start: (repo: Repo, issue: number) =>
-      start(
-        {
-          repo: repo.watched.repo,
-          trackerUser: repo.trackerUser,
-          machine: host,
-          project: repo.project,
-          dataDir,
-          tracker: trackerFor(repo.watched),
-          clock,
-        },
-        issue,
-      ),
+    start: (repo: Repo, issue: number) => start(startContext(repo), issue),
+    resume: (repo: Repo, issue: number) => resume(startContext(repo), issue),
     gateRunning: (run) => NodeFS.existsSync(gateLockFile(dataDir, run)),
     recordError: (run, step, kind, message) =>
       ledger(() => {
@@ -231,7 +240,8 @@ const startFactory = Effect.gen(function* () {
   const machine = loaded.success.value;
   yield* Effect.logInfo("factory started");
   const process = yield* FactoryProcess;
-  const world = liveWorld(machine, factoryDir, process, () => Date.now(), hostName());
+  const sessions = yield* T3Sessions.FactorySessions;
+  const world = liveWorld(machine, factoryDir, process, sessions, () => Date.now(), hostName());
   const onePass = Effect.gen(function* () {
     // Each line is logged as it is decided, so a pass that hangs or dies has said what it did.
     const context = yield* Effect.context<never>();
@@ -242,13 +252,14 @@ const startFactory = Effect.gen(function* () {
 });
 
 /**
- * Needs FactoryPaths, FactoryProcess and FileSystem; what index.test.ts builds with a test
- * FactoryPaths and a fake FactoryProcess.
+ * Needs FactoryPaths, FactoryProcess, FactorySessions and FileSystem; what index.test.ts builds
+ * with a test FactoryPaths, a fake FactoryProcess and a stub FactorySessions.
  */
 export const FactoryRuntime = Layer.effect(FactoryLoop, startFactory);
 
 /** The factory's root layer. server.ts merges it into the runtime; nothing else imports it. */
 export const FactoryLayer = FactoryRuntime.pipe(
   Layer.provide(Paths.layer),
   Layer.provide(Process.layer),
+  Layer.provide(T3Sessions.layerLive),
 );

apps/server/src/factory/ledger/listRuns.test.ts

@@ -60,9 +60,9 @@ describe("listUnfinishedRuns", () => {
     runAt(dataDir, rid(2), T0 + 10_000, "pr-open", true);
     runAt(dataDir, rid(4), T0 + 30_000, null);
     expect(listUnfinishedRuns(dataDir)).toEqual([
-      { id: rid(1), state: "triage", unreadable: false, confirmed: true },
-      { id: rid(3), state: "building", unreadable: false, confirmed: true },
-      { id: rid(4), state: null, unreadable: false, confirmed: true },
+      { id: rid(1), state: "triage", unreadable: false, confirmed: true, session: false },
+      { id: rid(3), state: "building", unreadable: false, confirmed: true, session: false },
+      { id: rid(4), state: null, unreadable: false, confirmed: true, session: false },
     ]);
   });
 
@@ -78,8 +78,8 @@ describe("listUnfinishedRuns", () => {
     // A directory named like a run file is a run whose file cannot be read.
     NodeFS.mkdirSync(NodePath.join(dataDir, RUNS_DIR, runIdFileName(rid(6))));
     expect(listUnfinishedRuns(dataDir)).toEqual([
-      { id: rid(7), state: "building", unreadable: false, confirmed: true },
-      { id: rid(6), state: null, unreadable: true, confirmed: false },
+      { id: rid(7), state: "building", unreadable: false, confirmed: true, session: false },
+      { id: rid(6), state: null, unreadable: true, confirmed: false, session: false },
     ]);
   });
 
@@ -99,15 +99,15 @@ describe("listUnfinishedRuns", () => {
       `{"ts":"2026-10-04T10:13:16.123Z","ms":${T0},"event":"state_changed","data":{"from":"ready","to":"later"}}\n`,
     );
     expect(listUnfinishedRuns(dataDir)).toEqual([
-      { id: rid(9), state: null, unreadable: false, confirmed: false },
+      { id: rid(9), state: null, unreadable: false, confirmed: false, session: false },
     ]);
   });
 
   it("a run with no approval and no session is listed as not confirmed", () => {
     const dataDir = tempDir();
     runAt(dataDir, rid(8), T0, "triage", false, false);
     expect(listUnfinishedRuns(dataDir)).toEqual([
-      { id: rid(8), state: "triage", unreadable: false, confirmed: false },
+      { id: rid(8), state: "triage", unreadable: false, confirmed: false, session: false },
     ]);
   });
 
@@ -120,8 +120,45 @@ describe("listUnfinishedRuns", () => {
     // A file of another issue that cannot be read is not opened, so it is not listed.
     NodeFS.mkdirSync(NodePath.join(dataDir, RUNS_DIR, runIdFileName(rid(3))));
     expect(listUnfinishedRuns(dataDir, { repo: "RawPhp/Softfac", issue: 1 })).toEqual([
-      { id: rid(1), state: "triage", unreadable: false, confirmed: true },
-      { id: rid(1, "rawphp/softfac", 2), state: "building", unreadable: false, confirmed: true },
+      { id: rid(1), state: "triage", unreadable: false, confirmed: true, session: false },
+      {
+        id: rid(1, "rawphp/softfac", 2),
+        state: "building",
+        unreadable: false,
+        confirmed: true,
+        session: false,
+      },
+    ]);
+  });
+});
+
+describe("listUnfinishedRuns: whether a session may be running", () => {
+  it("is set by a launched session or a launch that was sent, and not by the claim alone", () => {
+    const dataDir = tempDir();
+    runAt(dataDir, rid(1), T0, "triage");
+    runAt(dataDir, rid(2), T0 + 10_000, "triage");
+    runAt(dataDir, rid(3), T0 + 20_000, "triage");
+    const step = (id: RunId, line: Parameters<Ledger["append"]>[0]) => {
+      const ledger = Ledger.open(dataDir, id, { clock: ticking(T0 + 30_000, 1000) });
+      handles.push(ledger);
+      ledger.append(line, cause);
+    };
+    step(rid(2), { event: "step_changed", data: { step: "launch-sending" } });
+    step(rid(3), {
+      event: "session_launched",
+      data: {
+        thread_id: "t-3",
+        worktree: "/wt",
+        branch: "b",
+        model_instance: "m",
+        model: "x",
+        resumed_from: null,
+      },
+    });
+    expect(listUnfinishedRuns(dataDir).map((r) => [r.id.issue, r.session])).toEqual([
+      ["1", false],
+      ["2", true],
+      ["3", true],
     ]);
   });
 });
@@ -132,6 +169,7 @@ describe("holdsSlot", () => {
     state,
     unreadable,
     confirmed,
+    session: false,
   });
 
   it("every confirmed unfinished run holds a slot but one waiting in pr-open", () => {

apps/server/src/factory/ledger/listRuns.ts

@@ -20,6 +20,11 @@ export interface RunInfo {
    * launch one: `start` sends it to `stuck` when it is called for the issue again.
    */
   readonly confirmed: boolean;
+  /**
+   * A session may be running for it: the run records a launched session, or a launch that was
+   * sent. False for a file that cannot be read.
+   */
+  readonly session: boolean;
 }
 
 /**
@@ -68,9 +73,10 @@ export const listUnfinishedRuns = (
       const seen = readProgress(record.entries);
       if (seen.finished) continue;
       const confirmed = seen.approvedAt !== null || seen.threadId !== null;
-      info = { id, state: seen.state, unreadable: false, confirmed };
+      const session = seen.threadId !== null || seen.launchSent;
+      info = { id, state: seen.state, unreadable: false, confirmed, session };
     } catch {
-      info = { id, state: null, unreadable: true, confirmed: false };
+      info = { id, state: null, unreadable: true, confirmed: false, session: false };
     }
     const rank = info.unreadable ? 2 : firstMs === null ? 1 : 0;
     runs.push({ firstMs, info, rank });

apps/server/src/factory/loop.test.ts

@@ -1,11 +1,12 @@
 import { assert, it } from "@effect/vitest";
 import * as Deferred from "effect/Deferred";
+import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Fiber from "effect/Fiber";
 import * as TestClock from "effect/testing/TestClock";
 import { describe } from "vite-plus/test";
 
-import { runLoop } from "./loop.ts";
+import { PASS_TIMEOUT, runLoop } from "./loop.ts";
 import type { Report } from "./pass.ts";
 
 const report = (n: number): Report => ({ runs: n, claimed: 0, errors: 0, attention: 0 });
@@ -100,6 +101,43 @@ describe("the pass loop", () => {
     ),
   );
 
+  it.effect("a pass that hangs is stopped at the pass timeout and the loop goes on", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const never = yield* Deferred.make<void>();
+        const { calls, onePass } = counting({ at: 1, on: never });
+        const loop = yield* runLoop(onePass, 60);
+        const first = yield* Effect.forkChild(loop.awaitPass, { startImmediately: true });
+        yield* TestClock.adjust(Duration.subtract(PASS_TIMEOUT, Duration.seconds(1)));
+        assert.strictEqual(calls.n, 1, "the hung pass is still running");
+        yield* TestClock.adjust("1 second");
+        assert.strictEqual(yield* Fiber.join(first), null);
+        const second = yield* Effect.forkChild(loop.awaitPass, { startImmediately: true });
+        yield* TestClock.adjust("60 seconds");
+        assert.deepStrictEqual(yield* Fiber.join(second), report(2));
+      }),
+    ),
+  );
+
+  it.effect("a long pass that finishes before the pass timeout keeps its report", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const on = yield* Deferred.make<void>();
+        const { calls, onePass } = counting({ at: 1, on });
+        const loop = yield* runLoop(onePass, 60);
+        const first = yield* Effect.forkChild(loop.awaitPass, { startImmediately: true });
+        yield* TestClock.adjust(Duration.subtract(PASS_TIMEOUT, Duration.seconds(1)));
+        yield* Deferred.succeed(on, undefined);
+        assert.deepStrictEqual(yield* Fiber.join(first), report(1));
+        // The timeout that would have fired is gone: the next pass comes on the poll alone.
+        yield* TestClock.adjust("59 seconds");
+        assert.strictEqual(calls.n, 1);
+        yield* TestClock.adjust("1 second");
+        assert.strictEqual(calls.n, 2);
+      }),
+    ),
+  );
+
   it.effect("the loop stops with its scope", () =>
     Effect.gen(function* () {
       const { calls, onePass } = counting();

apps/server/src/factory/loop.ts

@@ -2,7 +2,8 @@
  * The pass loop: one pass when the factory starts and one every `poll_seconds` after, with a
  * wake that starts the next pass now. Passes never overlap: one fiber runs them one after the
  * other, and a wake that arrives during a pass makes the next pass follow it at once, however
- * many wakes arrived.
+ * many wakes arrived. A pass that runs past `PASS_TIMEOUT` is stopped and logged, so one hung
+ * call cannot stop every later pass; the next pass goes on from the ledger.
  */
 import * as Deferred from "effect/Deferred";
 import * as Duration from "effect/Duration";
@@ -15,10 +16,18 @@ import type { Report } from "./pass.ts";
 export interface Loop {
   /** Starts the next pass now when none is running, or right after the one that is. */
   readonly wake: Effect.Effect<void>;
-  /** The report of the next pass to finish; null when that pass died. */
+  /** The report of the next pass to finish; null when that pass died or timed out. */
   readonly awaitPass: Effect.Effect<Report | null>;
 }
 
+/**
+ * Well past what a pass takes, though not a sum of every call's own limit (a `gh` call 60
+ * seconds, a worktree removal 10 minutes): a very slow pass is stopped too. That is safe, as
+ * every step goes on from the ledger on the next pass, and a claim cut short before its run
+ * file is written goes stuck, as it cannot be confirmed.
+ */
+export const PASS_TIMEOUT = Duration.minutes(30);
+
 /** Runs `onePass` on the loop's schedule in a fiber of the scope. The first pass starts at once. */
 export const runLoop = (
   onePass: Effect.Effect<Report>,
@@ -31,6 +40,12 @@ export const runLoop = (
       for (;;) {
         yield* latch.close;
         const report = yield* onePass.pipe(
+          Effect.timeoutOption(PASS_TIMEOUT),
+          Effect.flatMap((done) =>
+            done._tag === "Some"
+              ? Effect.succeed(done.value)
+              : Effect.logError("factory pass timed out and was stopped").pipe(Effect.as(null)),
+          ),
           Effect.catchCause((cause) =>
             Effect.logError("factory pass failed", cause).pipe(Effect.as(null)),
           ),