Building: follow the session to its result line

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

@@ -24,12 +24,12 @@ import {
   gateRoot,
   layer as gateLayer,
   makeWorkspace,
-  sanitize,
   type CommandResult,
   type GateError,
   type GateInput,
   type GateOptions,
 } from "./gate.ts";
+import { sanitize } from "./text.ts";
 import type { AppendIo } from "./ledger/appendFile.ts";
 import { LedgerIoError } from "./ledger/errors.ts";
 import { Ledger } from "./ledger/ledger.ts";

apps/server/src/factory/gate.ts

@@ -16,6 +16,7 @@ import { Ledger } from "./ledger/ledger.ts";
 import { readRun } from "./ledger/read.ts";
 import { runIdToString, validateRunId, type RunId } from "./ledger/runId.ts";
 import { FactoryProcess, type RunInput, type RunResult, type RunSpawnError } from "./process.ts";
+import { sanitize } from "./text.ts";
 
 /**
  * The gate: a fresh checkout of a pull request's head merged with the base branch tip, the
@@ -135,31 +136,6 @@ export type GateError = WorkspaceError | GateCommandSpawnError | GateRecordError
 
 // ---- text -----------------------------------------------------------------------------------
 
-const isControl = (cp: number) => cp <= 0x1f || (cp >= 0x7f && cp <= 0x9f);
-
-/** Characters that could fake a line break or hide text, besides the control characters. */
-const isInvisibleOrSeparator = (cp: number) =>
-  cp === 0x2028 ||
-  cp === 0x2029 ||
-  (cp >= 0x200b && cp <= 0x200f) ||
-  (cp >= 0x202a && cp <= 0x202e) ||
-  (cp >= 0x2066 && cp <= 0x2069) ||
-  cp === 0xfeff;
-
-/**
- * Replace what could fake a line break, hide text or drive a terminal with a space. Newline and
- * tab are kept only when `keepLayout` is set.
- */
-export const sanitize = (text: string, keepLayout: boolean): string => {
-  let out = "";
-  for (const char of text) {
-    const cp = char.codePointAt(0)!;
-    const layout = char === "\n" || char === "\t";
-    out += (isControl(cp) && !(keepLayout && layout)) || isInvisibleOrSeparator(cp) ? " " : char;
-  }
-  return out;
-};
-
 /** The command as one short line: sanitised, newlines and tabs turned into spaces, cut at 300 characters. */
 const commandLine = (command: string): string => {
   const chars = Array.from(sanitize(command, false));

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

@@ -1,22 +1,38 @@
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import { assert, it } from "@effect/vitest";
+import * as Context from "effect/Context";
 import * as Effect from "effect/Effect";
 import * as FileSystem from "effect/FileSystem";
 import * as Layer from "effect/Layer";
 import * as Logger from "effect/Logger";
+import * as Option from "effect/Option";
 import * as Path from "effect/Path";
 import type * as PlatformError from "effect/PlatformError";
+import * as Queue from "effect/Queue";
+import * as Ref from "effect/Ref";
+import * as Stream from "effect/Stream";
 
 import { describe } from "vite-plus/test";
 
 import { timeLimitOnlyProjectConfig, type MachineConfig } from "./config.ts";
-import { FactoryRuntime, liveWorld } from "./index.ts";
+import {
+  factoryRuntime,
+  FactoryLoop,
+  FactoryRuntime,
+  liveWorld,
+  ownedThreads,
+  wakeOn,
+} from "./index.ts";
+import { makeCause } from "./ledger/events.ts";
+import { Ledger } from "./ledger/ledger.ts";
 import { scanMachineLog } from "./ledger/machineLog.ts";
+import { newRunId } from "./ledger/runId.ts";
 import type { Repo } from "./pass.ts";
 import { CUT_SHORT_COMMENT } from "./runner/messages.ts";
 import { utcSecond } from "./runner/start.ts";
 import { MARKER } from "./tracker/github.ts";
 import { FactoryProcess, RunSpawnError, type RunInput, type RunResult } from "./process.ts";
+import { EventsError, FactoryEvents } from "./t3/events.ts";
 import * as Paths from "./t3/paths.ts";
 import * as Sessions from "./t3/sessions.ts";
 
@@ -40,6 +56,12 @@ const stubSessions = Sessions.FactorySessions.of({
 });
 const noSessions = Layer.succeed(Sessions.FactorySessions, stubSessions);
 
+/** FactoryEvents that never says anything. */
+const quietEvents = Layer.succeed(
+  FactoryEvents,
+  FactoryEvents.of({ runStatusChanges: Stream.never }),
+);
+
 const messageText = (message: unknown): string => {
   const first = Array.isArray(message) ? message[0] : message;
   return typeof first === "string" ? first : String(first);
@@ -90,6 +112,7 @@ const logsFor = (
             ),
             Layer.provide(gh),
             Layer.provide(noSessions),
+            Layer.provide(quietEvents),
           ),
         ),
       ).pipe(Effect.provide(Logger.layer([capture], { mergeWithExisting: false })));
@@ -181,6 +204,172 @@ it.effect("logs each pass line as it is decided, so a pass that dies has said wh
   }),
 );
 
+// ---- the wake ----
+
+describe("waking the pass when a factory thread's run changes status", () => {
+  /** Writes an unfinished run of issue `issue` whose session is `threadId`. */
+  const seedRun = (factoryDir: string, issue: number, threadId: string, finished = false) => {
+    const out = Ledger.create(factoryDir, newRunId("a/b", String(issue), 1));
+    try {
+      out.append(
+        { event: "claimed", data: { machine: "m", github_user: "u" } },
+        makeCause("start", "seed"),
+      );
+      out.append(
+        {
+          event: "session_launched",
+          data: {
+            thread_id: threadId,
+            worktree: "/t3/wt",
+            branch: "t3code/abc",
+            model_instance: "claudeAgent",
+            model: "claude-sonnet-5-5",
+            resumed_from: null,
+          },
+        },
+        makeCause("start", "seed"),
+      );
+      if (finished) out.finish("stuck", null, null, makeCause("start", "seed"));
+    } finally {
+      out.close();
+    }
+  };
+
+  /** A factory dir with a valid config and one unfinished run whose session is `thread-1`. */
+  const seeded = Effect.gen(function* () {
+    const fs = yield* FileSystem.FileSystem;
+    const { join } = yield* Path.Path;
+    const baseDir = yield* fs.makeTempDirectoryScoped({ prefix: "factory-wake-" });
+    const factoryDir = join(baseDir, "factory");
+    yield* fs.makeDirectory(factoryDir);
+    const machineConfigPath = join(factoryDir, "config.yml");
+    yield* fs.writeFileString(machineConfigPath, VALID);
+    seedRun(factoryDir, 1, "thread-1");
+    return { baseDir, factoryDir, machineConfigPath, worktreesDir: join(baseDir, "worktrees") };
+  });
+
+  it.effect("owns the threads of unfinished runs' sessions, and no finished run's", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const { factoryDir } = yield* seeded;
+        seedRun(factoryDir, 2, "thread-done", true);
+        assert.deepStrictEqual([...(yield* ownedThreads(factoryDir))], ["thread-1"]);
+      }),
+    ).pipe(Effect.provide(NodeServices.layer)),
+  );
+
+  it.effect("a thread the factory does not own never wakes the pass", () =>
+    Effect.gen(function* () {
+      const owned = yield* Ref.make<ReadonlySet<string>>(new Set(["thread-1"]));
+      const wakes = yield* Ref.make(0);
+      yield* wakeOn(
+        owned,
+        Stream.fromIterable(["thread-other", "thread-1", "thread-other"]),
+        Ref.update(wakes, (n) => n + 1),
+      );
+      assert.strictEqual(yield* Ref.get(wakes), 1);
+    }),
+  );
+
+  /** The factory's loop over the seeded dir, with `events` as T3's run status changes. */
+  const loopOver = (
+    dirs: Effect.Success<typeof seeded>,
+    events: Stream.Stream<string, EventsError>,
+  ) =>
+    Layer.build(
+      factoryRuntime({ restartDelay: 0 }).pipe(
+        Layer.provide(Layer.succeed(Paths.FactoryPaths, dirs)),
+        Layer.provide(noGh),
+        Layer.provide(noSessions),
+        Layer.provide(Layer.succeed(FactoryEvents, FactoryEvents.of({ runStatusChanges: events }))),
+      ),
+    ).pipe(Effect.map((context) => Context.get(context, FactoryLoop)));
+
+  const countOf = (factoryDir: string, event: string) => () =>
+    scanMachineLog(factoryDir).filter((entry) => entry.event === event).length;
+
+  it.effect("a thread launched after the start is owned once a pass has ended", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const dirs = yield* seeded;
+        const queue = yield* Queue.unbounded<string>();
+        const loop = yield* loopOver(dirs, Stream.fromQueue(queue));
+        const started = countOf(dirs.factoryDir, "pass_started");
+        const finished = countOf(dirs.factoryDir, "pass_finished");
+        // The poll is a minute away, so every pass here is the first or one a wake started. Each
+        // wait is bounded so a wake that never comes fails the test instead of hanging it.
+        const until = (done: () => boolean, each: Effect.Effect<unknown> = Effect.void) =>
+          Effect.gen(function* () {
+            while (!done()) {
+              yield* each;
+              yield* Effect.timeoutOption(loop.awaitPass, "1 second");
+            }
+          }).pipe(Effect.timeoutOption("30 seconds"), Effect.map(Option.isSome));
+        assert.isTrue(yield* until(() => finished() >= 1), "the first pass never ended");
+        seedRun(dirs.factoryDir, 2, "thread-2");
+        const before = finished();
+        yield* loop.wake;
+        assert.isTrue(yield* until(() => finished() > before), "the woken pass never ended");
+        // That pass read the new run's thread when it ended, so an event for the thread wakes the
+        // next one. It is offered until it does: one offered before the read is not a wake.
+        const after = started();
+        const woke = yield* until(() => started() > after, Queue.offer(queue, "thread-2"));
+        assert.isTrue(woke, "the new run's thread never woke the pass");
+      }),
+    ).pipe(Effect.provide(NodeServices.layer)),
+  );
+
+  it.effect("an owned thread starts the next pass, and a stream that failed is run again", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const { baseDir, factoryDir, machineConfigPath, worktreesDir } = yield* seeded;
+        const queue = yield* Queue.unbounded<string>();
+        // The first run of the stream fails; the one after it reads the queue.
+        const failures = yield* Ref.make(1);
+        const events = Layer.succeed(
+          FactoryEvents,
+          FactoryEvents.of({
+            runStatusChanges: Stream.unwrap(
+              Ref.modify(failures, (left): readonly [Stream.Stream<string, EventsError>, number] =>
+                left > 0
+                  ? [Stream.fail(new EventsError({ cause: new Error("gone") })), left - 1]
+                  : [Stream.fromQueue(queue), 0],
+              ),
+            ),
+          }),
+        );
+        const loop = yield* Layer.build(
+          factoryRuntime({ restartDelay: 0 }).pipe(
+            Layer.provide(
+              Layer.succeed(Paths.FactoryPaths, {
+                baseDir,
+                worktreesDir,
+                factoryDir,
+                machineConfigPath,
+              }),
+            ),
+            Layer.provide(noGh),
+            Layer.provide(noSessions),
+            Layer.provide(events),
+          ),
+        ).pipe(Effect.map((context) => Context.get(context, FactoryLoop)));
+        const started = () =>
+          scanMachineLog(factoryDir).filter((entry) => entry.event === "pass_started").length;
+        yield* Queue.offer(queue, "thread-other");
+        yield* Queue.offer(queue, "thread-1");
+        // The first pass, then the one the owned thread woke; the poll is a minute away. The wait is
+        // bounded so a wake that never comes fails here. The count stays `>=`: the offer can land
+        // while the first pass still runs, and a wake then is coalesced into that pass's latch.
+        const waited = yield* Effect.gen(function* () {
+          while (started() < 2) yield* loop.awaitPass;
+        }).pipe(Effect.timeoutOption("30 seconds"));
+        assert.isTrue(Option.isSome(waited), "the owned thread never woke the pass");
+        assert.isAtLeast(started(), 2);
+      }),
+    ).pipe(Effect.provide(NodeServices.layer)),
+  );
+});
+
 // ---- the live world over gh ----
 
 describe("liveWorld", () => {