diff --git a/apps/server/src/factory/index.test.ts b/apps/server/src/factory/index.test.ts index b32f2bc50cf1..012787b69d36 100644 --- a/apps/server/src/factory/index.test.ts +++ b/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]!, diff --git a/apps/server/src/factory/index.ts b/apps/server/src/factory/index.ts index 482458aa7c54..ede133af9142 100644 --- a/apps/server/src/factory/index.ts +++ b/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) => /** The numbers of the issues `user` is the only assignee of, by number. */ const assignedOnlyTo = (items: ReadonlyArray, 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) => 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(); @@ -242,8 +252,8 @@ 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); @@ -251,4 +261,5 @@ export const FactoryRuntime = Layer.effect(FactoryLoop, startFactory); export const FactoryLayer = FactoryRuntime.pipe( Layer.provide(Paths.layer), Layer.provide(Process.layer), + Layer.provide(T3Sessions.layerLive), ); diff --git a/apps/server/src/factory/ledger/listRuns.test.ts b/apps/server/src/factory/ledger/listRuns.test.ts index 444d80554380..eb563d8a4063 100644 --- a/apps/server/src/factory/ledger/listRuns.test.ts +++ b/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,7 +99,7 @@ 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 }, ]); }); @@ -107,7 +107,7 @@ describe("listUnfinishedRuns", () => { 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[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", () => { diff --git a/apps/server/src/factory/ledger/listRuns.ts b/apps/server/src/factory/ledger/listRuns.ts index 5344e967dfbb..cf260579b2e6 100644 --- a/apps/server/src/factory/ledger/listRuns.ts +++ b/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 }); diff --git a/apps/server/src/factory/loop.test.ts b/apps/server/src/factory/loop.test.ts index c0ac9bc0cc48..031205503753 100644 --- a/apps/server/src/factory/loop.test.ts +++ b/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(); + 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(); + 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(); diff --git a/apps/server/src/factory/loop.ts b/apps/server/src/factory/loop.ts index 35898f785c99..235036990511 100644 --- a/apps/server/src/factory/loop.ts +++ b/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; - /** 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; } +/** + * 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, @@ -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)), ), diff --git a/apps/server/src/factory/pass.test.ts b/apps/server/src/factory/pass.test.ts index 667c24d7e5c0..8fe0372fa1dd 100644 --- a/apps/server/src/factory/pass.test.ts +++ b/apps/server/src/factory/pass.test.ts @@ -25,6 +25,7 @@ import { type Report, } from "./pass.ts"; import { GateRunningError, type StartError, type StartOutcome } from "./runner/start.ts"; +import { SessionsError } from "./t3/sessions.ts"; import { GitHubFailed, type Blocker } from "./tracker/github.ts"; const REPO = "rawphp/softfac"; @@ -61,8 +62,9 @@ const githubFailure = (): StartError => new GitHubFailed({ args: ["issue", "list"], status: 1, stderr: "boom" }); const started = (issue: number, repo = REPO): StartOutcome => ({ - kind: "claimed", + kind: "started", run: rid(issue, repo), + threadId: "thread-1", }); type Answers = Record; @@ -96,8 +98,8 @@ class Fake implements PassWorld { calls: Array = []; events: Array = []; - runIn(issue: number, state: State | null, repo = REPO) { - this.runs.push({ id: rid(issue, repo), state, unreadable: false, confirmed: true }); + runIn(issue: number, state: State | null, repo = REPO, session = false) { + this.runs.push({ id: rid(issue, repo), state, unreadable: false, confirmed: true, session }); } pushStart(key: string, answer: StartOutcome | StartError) { @@ -181,8 +183,17 @@ class Fake implements PassWorld { } start(repo: Repo, issue: number) { + return this.answerStart("start", repo, issue); + } + + resume(repo: Repo, issue: number) { + return this.answerStart("resume", repo, issue); + } + + /** Both calls share one script per issue; `calls` says which was made. */ + private answerStart(call: "start" | "resume", repo: Repo, issue: number) { const key = `${repo.watched.repo}#${issue}`; - this.calls.push(`start ${key}`); + this.calls.push(`${call} ${key}`); const answer = this.starts[key]?.shift() ?? { kind: "skipped", reason: "scripted" }; if (this.startLeavesRun.includes(issue)) { this.runs.push({ @@ -190,10 +201,22 @@ class Fake implements PassWorld { state: "triage", unreadable: false, confirmed: true, + session: false, }); } - if ("kind" in answer && answer.kind === "claimed") { - this.runs.push({ id: answer.run, state: "triage", unreadable: false, confirmed: true }); + if ("kind" in answer && answer.kind === "started") { + const existing = this.runs.find((r) => runIdToString(r.id) === runIdToString(answer.run)); + // A run the pass already knew moves on to building; a new claim leaves a triage run. + if (existing !== undefined) + this.runs[this.runs.indexOf(existing)] = { ...existing, state: "building" }; + else + this.runs.push({ + id: answer.run, + state: "triage", + unreadable: false, + confirmed: true, + session: false, + }); return Effect.succeed(answer); } if ("kind" in answer) return Effect.succeed(answer); @@ -243,7 +266,9 @@ const pass = ( return { report, lines }; }); -const starts = (fake: Fake) => fake.calls.filter((c) => c.startsWith("start ")); +/** The `start` and `resume` calls, in order. */ +const starts = (fake: Fake) => + fake.calls.filter((c) => c.startsWith("start ") || c.startsWith("resume ")); const releases = (fake: Fake) => fake.calls.filter((c) => c.startsWith("release ")); const runLine = (lines: ReadonlyArray, prefix: string): string => { @@ -337,7 +362,7 @@ describe("claiming new issues", () => { const { lines } = yield* pass(fake, machine([REPO], 1)); assert.equal( runLine(lines, "rawphp/softfac#2 "), - "rawphp/softfac#2 ready: claimed in run rawphp__softfac__2__1", + "rawphp/softfac#2 ready: claimed, session started in run rawphp__softfac__2__1", ); }), ); @@ -370,6 +395,7 @@ describe("claiming new issues", () => { assert.equal(report.claimed, 0); const line = runLine(lines, "rawphp/softfac#12 "); assert.ok(line.includes("stuck"), line); + assert.ok(line.includes("claimed, but the launch failed"), line); assert.ok(line.includes("rawphp__softfac__12__1"), line); assert.ok(!line.includes("secret detail"), line); }), @@ -482,7 +508,13 @@ describe("claiming new issues", () => { it.effect("a run whose file cannot be read is an error line and counts as active", () => Effect.gen(function* () { const fake = new Fake(); - fake.runs.push({ id: rid(40), state: null, unreadable: true, confirmed: false }); + fake.runs.push({ + id: rid(40), + state: null, + unreadable: true, + confirmed: false, + session: false, + }); fake.ready[REPO] = [1]; fake.pushStart(`${REPO}#1`, started(1)); const { report, lines } = yield* pass(fake, machine([REPO], 1)); @@ -628,16 +660,272 @@ describe("which repo gets a free slot", () => { ); }); +// ---- runs that are confirmed and still in triage ---- + +describe("runs that are confirmed and still in triage", () => { + it.effect("a confirmed triage run is resumed on the next pass, and the line says so", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(12, "triage"); + fake.pushStart(`${REPO}#12`, started(12)); + const { report, lines } = yield* pass(fake, machine([REPO], 3)); + assert.deepEqual(starts(fake), [`resume ${REPO}#12`]); + const line = runLine(lines, "rawphp/softfac#12 triage:"); + assert.ok(line.includes("session started, the issue is building"), line); + assert.equal(report.claimed, 0); + }), + ); + + it.effect("a confirmed run with no state yet is resumed too", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runs.push({ + id: rid(13), + state: null, + unreadable: false, + confirmed: true, + session: false, + }); + fake.pushStart(`${REPO}#13`, started(13)); + yield* pass(fake, machine([REPO], 3)); + assert.deepEqual(starts(fake), [`resume ${REPO}#13`]); + }), + ); + + it.effect("a triage run whose start fails records the error on that run", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(12, "triage"); + fake.pushStart(`${REPO}#12`, githubFailure()); + const { report } = yield* pass(fake, machine([REPO], 3)); + assert.equal(report.errors, 1); + assert.equal(fake.runErrors.length, 1); + assert.deepEqual(fake.runErrors[0]!.run, rid(12)); + assert.equal(fake.runErrors[0]!.step, "start"); + }), + ); + + it.effect("a triage run that went stuck says so and names the run", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(12, "triage"); + fake.pushStart(`${REPO}#12`, { kind: "stuck", run: rid(12), reason: "secret detail" }); + const { lines } = yield* pass(fake, machine([REPO], 3)); + const line = runLine(lines, "rawphp/softfac#12 "); + assert.ok(line.includes("stuck"), line); + assert.ok(line.includes("rawphp__softfac__12__1"), line); + assert.ok(!line.includes("secret detail"), line); + }), + ); + + it.effect("a triage run is worked before new issues are claimed", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(12, "triage"); + fake.ready[REPO] = [3]; + fake.pushStart(`${REPO}#12`, started(12)); + fake.pushStart(`${REPO}#3`, started(3)); + yield* pass(fake, machine([REPO], 5)); + assert.deepEqual(starts(fake), [`resume ${REPO}#12`, `start ${REPO}#3`]); + }), + ); + + it.effect( + "a ready-again issue whose own run has no session is claimed past that run's slot", + () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(12, "triage"); + fake.ready[REPO] = [12]; + fake.pushStart(`${REPO}#12`, { kind: "skipped", reason: "ready again", moved: "ready" }); + const { report, lines } = yield* pass(fake, machine([REPO], 1)); + // The claim finishes the old run, which left nothing running, before it claims. + assert.deepEqual(starts(fake), [`resume ${REPO}#12`, `start ${REPO}#12`]); + assert.ok(runLine(lines, "rawphp/softfac#12 triage:").includes("left to the claim step")); + assert.equal(report.attention, 0); + }), + ); + + it.effect("a ready-again issue's freed slot does not beat the limit other runs fill", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(40, BUILDING); + fake.runIn(12, "triage"); + fake.ready[REPO] = [12]; + fake.pushStart(`${REPO}#12`, { kind: "skipped", reason: "ready again", moved: "ready" }); + yield* pass(fake, machine([REPO], 1)); + assert.deepEqual(starts(fake), [`resume ${REPO}#12`]); + assert.deepEqual( + fake.events.filter((e) => e.event === "claim_skipped").map((e) => e.data), + [{ repo: REPO, issue: "12", reason: "limit" }], + ); + }), + ); + + it.effect("a ready-again issue claimed past its own run takes the slot from the next issue", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(12, "triage"); + fake.ready[REPO] = [12, 13]; + fake.pushStart(`${REPO}#12`, { kind: "skipped", reason: "ready again", moved: "ready" }); + fake.pushStart(`${REPO}#12`, started(12)); + yield* pass(fake, machine([REPO], 1)); + assert.deepEqual(starts(fake), [`resume ${REPO}#12`, `start ${REPO}#12`]); + assert.deepEqual( + fake.events.filter((e) => e.event === "claim_skipped").map((e) => e.data), + [{ repo: REPO, issue: "13", reason: "limit" }], + ); + }), + ); + + it.effect("a ready-again issue whose own run's session holds the only slot is attention", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(12, "triage", REPO, true); + fake.ready[REPO] = [12]; + fake.pushStart(`${REPO}#12`, { kind: "skipped", reason: "ready again", moved: "ready" }); + const { report, lines } = yield* pass(fake, machine([REPO], 1)); + // A second session would run past the limit, so the claim waits. + assert.deepEqual(starts(fake), [`resume ${REPO}#12`]); + assert.equal(report.attention, 1); + const line = runLine(lines, "ATTENTION: rawphp/softfac#12 "); + assert.ok(line.includes("the new claim waits for a slot this run holds"), line); + assert.deepEqual( + fake.events.filter((e) => e.event === "attention"), + [ + { + event: "attention", + data: { run: "rawphp__softfac__12__1", reason: "not-progressing" }, + }, + ], + ); + }), + ); + + it.effect("a ready-again issue that waits on another run's slot too raises no attention", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(40, BUILDING); + fake.runIn(12, "triage", REPO, true); + fake.ready[REPO] = [12]; + fake.pushStart(`${REPO}#12`, { kind: "skipped", reason: "ready again", moved: "ready" }); + const { report } = yield* pass(fake, machine([REPO], 2)); + assert.deepEqual(starts(fake), [`resume ${REPO}#12`]); + assert.equal(report.attention, 0); + }), + ); + + it.effect("a ready issue the limit holds back by other runs raises no attention", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(40, BUILDING); + fake.ready[REPO] = [3]; + const { report } = yield* pass(fake, machine([REPO], 1)); + assert.deepEqual(starts(fake), []); + assert.equal(report.attention, 0); + }), + ); + + it.effect("a triage run whose issue is ready again is claimed afresh when a slot is free", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(12, "triage"); + fake.ready[REPO] = [12]; + fake.pushStart(`${REPO}#12`, { kind: "skipped", reason: "ready again", moved: "ready" }); + const { report } = yield* pass(fake, machine([REPO], 2)); + assert.deepEqual(starts(fake), [`resume ${REPO}#12`, `start ${REPO}#12`]); + assert.equal(report.attention, 0); + }), + ); + + it.effect("a triage run whose issue a person moved is an attention line, logged, with why", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(12, "triage"); + fake.pushStart(`${REPO}#12`, { kind: "skipped", reason: "issue #12 is assigned to bob" }); + const { report, lines } = yield* pass(fake, machine([REPO], 1)); + assert.equal(report.attention, 1); + const line = runLine(lines, "ATTENTION: rawphp/softfac#12 "); + assert.ok( + line.includes( + "run rawphp__softfac__12__1 is not progressing and holds a slot: issue #12 is assigned to bob. If you moved the issue by hand", + ), + line, + ); + assert.deepEqual( + fake.events.filter((e) => e.event === "attention"), + [ + { + event: "attention", + data: { run: "rawphp__softfac__12__1", reason: "not-progressing" }, + }, + ], + ); + }), + ); + + it.effect("a triage run whose issue was closed says to reopen it", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(12, "triage"); + fake.pushStart(`${REPO}#12`, { + kind: "skipped", + reason: "issue #12 is closed", + moved: "closed", + }); + const { report, lines } = yield* pass(fake, machine([REPO], 1)); + assert.equal(report.attention, 1); + const line = runLine(lines, "ATTENTION: rawphp/softfac#12 "); + assert.ok( + line.includes("holds a slot: issue #12 is closed. If it was closed by mistake, reopen it"), + line, + ); + }), + ); + + it.effect("a building run is left alone", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(40, BUILDING); + yield* pass(fake, machine([REPO], 3)); + assert.deepEqual(starts(fake), []); + }), + ); + + it.effect("a run of a repo that is not loaded is left alone", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(12, "triage", "acme/other"); + const { report } = yield* pass(fake, machine([REPO], 3)); + assert.deepEqual(starts(fake), []); + assert.equal(report.errors, 0); + }), + ); + + it.effect("a start that fails at T3 is recorded as a t3 error", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(12, "triage"); + fake.pushStart( + `${REPO}#12`, + new SessionsError({ operation: "find-project", cause: new Error("down") }), + ); + yield* pass(fake, machine([REPO], 3)); + assert.equal(fake.runErrors[0]!.kind, "t3"); + }), + ); +}); + // ---- claims that have no run in the ledger ---- describe("claims that have no run in the ledger", () => { - it.effect("a triage issue of ours with no run gets start, and is not a new claim", () => + it.effect("a triage issue of ours with no run is resumed, and is not a new claim", () => Effect.gen(function* () { const fake = new Fake(); fake.triage[REPO] = [7]; fake.pushStart(`${REPO}#7`, { kind: "stuck", run: rid(7), reason: "could not confirm" }); const { report, lines } = yield* pass(fake, machine([REPO], 1)); - assert.deepEqual(starts(fake), [`start ${REPO}#7`]); + assert.deepEqual(starts(fake), [`resume ${REPO}#7`]); const line = runLine(lines, "rawphp/softfac#7 "); assert.ok(line.startsWith("rawphp/softfac#7 triage: no run on record, stuck"), line); assert.ok(line.includes("rawphp__softfac__7__1"), line); @@ -645,25 +933,31 @@ describe("claims that have no run in the ledger", () => { }), ); - it.effect("a triage issue that has an unfinished run is left to that run", () => + it.effect("a triage issue that has an unfinished run is resumed by the run step alone", () => Effect.gen(function* () { const fake = new Fake(); fake.runIn(12, "triage"); fake.triage[REPO] = [12]; yield* pass(fake, machine([REPO], 3)); - assert.deepEqual(starts(fake), []); + assert.deepEqual(starts(fake), [`resume ${REPO}#12`]); }), ); - it.effect("a triage issue whose only run is not confirmed gets start again", () => + it.effect("a triage issue whose only run is not confirmed is resumed again", () => Effect.gen(function* () { // A stuck path that failed half way leaves a run with no approval behind; start finishes it. const fake = new Fake(); - fake.runs.push({ id: rid(12), state: "triage", unreadable: false, confirmed: false }); + fake.runs.push({ + id: rid(12), + state: "triage", + unreadable: false, + confirmed: false, + session: false, + }); fake.triage[REPO] = [12]; fake.pushStart(`${REPO}#12`, { kind: "stuck", run: rid(12), reason: "could not confirm" }); const { lines } = yield* pass(fake, machine([REPO], 3)); - assert.deepEqual(starts(fake), [`start ${REPO}#12`]); + assert.deepEqual(starts(fake), [`resume ${REPO}#12`]); assert.ok(runLine(lines, "rawphp/softfac#12 ").includes("stuck"), lines.join("\n")); }), ); @@ -671,7 +965,13 @@ describe("claims that have no run in the ledger", () => { it.effect("a run that is not confirmed takes no slot", () => Effect.gen(function* () { const fake = new Fake(); - fake.runs.push({ id: rid(12), state: "triage", unreadable: false, confirmed: false }); + fake.runs.push({ + id: rid(12), + state: "triage", + unreadable: false, + confirmed: false, + session: false, + }); fake.ready[REPO] = [3]; fake.pushStart(`${REPO}#3`, started(3)); yield* pass(fake, machine([REPO], 1)); @@ -679,17 +979,17 @@ describe("claims that have no run in the ledger", () => { }), ); - it.effect("a triage issue with no run is started even when every slot is taken", () => + it.effect("a triage issue with no run is resumed even when every slot is taken", () => Effect.gen(function* () { const fake = new Fake(); fake.runIn(40, BUILDING); fake.triage[REPO] = [7]; yield* pass(fake, machine([REPO], 1)); - assert.deepEqual(starts(fake), [`start ${REPO}#7`]); + assert.deepEqual(starts(fake), [`resume ${REPO}#7`]); }), ); - it.effect("a triage issue is started before new issues are claimed", () => + it.effect("a triage issue is resumed before new issues are claimed", () => Effect.gen(function* () { const fake = new Fake(); fake.triage[REPO] = [7]; @@ -697,7 +997,7 @@ describe("claims that have no run in the ledger", () => { fake.pushStart(`${REPO}#7`, started(7)); fake.pushStart(`${REPO}#3`, started(3)); yield* pass(fake, machine([REPO], 5)); - assert.deepEqual(starts(fake), [`start ${REPO}#7`, `start ${REPO}#3`]); + assert.deepEqual(starts(fake), [`resume ${REPO}#7`, `start ${REPO}#3`]); }), ); @@ -723,7 +1023,7 @@ describe("claims that have no run in the ledger", () => { fake.triage[REPO] = "error"; fake.triage[OTHER] = [4]; const { report, lines } = yield* pass(fake, machine([REPO, OTHER], 1)); - assert.deepEqual(starts(fake), [`start ${OTHER}#4`]); + assert.deepEqual(starts(fake), [`resume ${OTHER}#4`]); assert.equal(report.errors, 1); assert.ok( lines.some((l) => l.startsWith("error: rawphp/softfac: cannot list the triage issues")), @@ -1177,7 +1477,7 @@ describe("the hand pause", () => { fake.unassigned[REPO] = [{ issue: 8, state: "triage" }]; pausedByHand(fake); yield* pass(fake, machine([REPO], 5)); - assert.deepEqual(starts(fake), [`start ${REPO}#7`]); + assert.deepEqual(starts(fake), [`resume ${REPO}#7`]); assert.deepEqual(releases(fake), [`release ${REPO}#8 triage`]); }), ); @@ -1205,7 +1505,13 @@ describe("what the pass records", () => { Effect.gen(function* () { const fake = new Fake(); fake.runIn(12, BUILDING); - fake.runs.push({ id: rid(13), state: null, unreadable: true, confirmed: false }); + fake.runs.push({ + id: rid(13), + state: null, + unreadable: true, + confirmed: false, + session: false, + }); const { report, lines } = yield* pass(fake, machine([REPO], 1)); assert.deepEqual(fake.events[0], { event: "pass_started", data: {} }); assert.deepEqual(fake.events.at(-1), { @@ -1426,7 +1732,7 @@ describe("reading the ledger", () => { return listed(only); }; const { report, lines } = yield* pass(fake, machine([REPO], 5)); - assert.deepEqual(starts(fake), [`start ${REPO}#7`, `start ${REPO}#8`]); + assert.deepEqual(starts(fake), [`resume ${REPO}#7`, `resume ${REPO}#8`]); assert.equal( lines.filter((l) => l.startsWith("error: cannot read the run ledger")).length, 1, diff --git a/apps/server/src/factory/pass.ts b/apps/server/src/factory/pass.ts index 0c44713fba79..d0a952e9ee30 100644 --- a/apps/server/src/factory/pass.ts +++ b/apps/server/src/factory/pass.ts @@ -17,9 +17,9 @@ import { sanitize } from "./gate.ts"; import type { LedgerError } from "./ledger/errors.ts"; import type { ErrorKind, MachineEvent } from "./ledger/events.ts"; import { holdsSlot, type RunInfo } from "./ledger/listRuns.ts"; -import { runIdToString, type RunId } from "./ledger/runId.ts"; +import { asciiLower, runIdToString, type RunId } from "./ledger/runId.ts"; import { duration, formatTs } from "./ledger/time.ts"; -import type { StartError, StartOutcome } from "./runner/start.ts"; +import { mention, type StartError, type StartOutcome } from "./runner/start.ts"; import { LIST_LIMIT, type Blocker, type Listing, type UnassignedClaim } from "./tracker/github.ts"; /** An error of the pass's own: what kind it is for the machine log, and what it says. */ @@ -74,6 +74,8 @@ export interface PassWorld { ) => Effect.Effect, PassError>; /** The claim: `runner/start.ts`. */ readonly start: (repo: Repo, issue: number) => Effect.Effect; + /** The claim gone on with, never made: `resume` in `runner/start.ts`. */ + readonly resume: (repo: Repo, issue: number) => Effect.Effect; /** Whether the run's gate is running now. */ readonly gateRunning: (run: RunId) => boolean; /** Appends an `error` line to the run's own ledger. */ @@ -140,15 +142,16 @@ const errorKind = (error: StartError): ErrorKind => { return "tracker"; case "GateRunningError": return "gate"; + case "FactorySessionsError": + return "t3"; default: // The run's own record disagrees with what the step expected. return "ledger"; } }; -const mention = (issue: number) => `#${issue}`; - -const sameRepo = (a: string, b: string) => a.toLowerCase() === b.toLowerCase(); +/** Repos match as run ids do: `owner/name` folded with `asciiLower`. */ +const sameRepo = (a: string, b: string) => asciiLower(a) === asciiLower(b); const GATE_RUNNING = "gate running"; @@ -199,6 +202,16 @@ class Pass { this.out(`error: ${text}`); } + /** + * A line a person has to act on about `run`, and the machine log's `attention` event for it + * with the ledger.md `reason` code. + */ + attention(text: string, run: string, reason: string): Effect.Effect { + this.report.attention += 1; + this.out(`ATTENTION: ${text}`); + return this.record({ event: "attention", data: { run, reason } }); + } + /** * Writes `event` to the machine log. A log that cannot be written is an error line once per * pass and nothing else: it is not an event, which could not be written either. @@ -287,9 +300,10 @@ export const runPass = ( yield* pass.record({ event: "pass_started", data: {} }); const repos = yield* loadRepos(world, machine, pass); // Without the ledger there is no telling how many runs are active, so nothing is claimed. - const infos = yield* runs(world, repos, pass); + const infos = yield* runs(world, pass); if (infos !== null) { const known = new Known(infos); + yield* workRuns(world, repos, known, pass); yield* untrackedClaims(world, repos, known, pass); yield* unassignedClaims(world, repos, known, pass); yield* claim(world, repos, known, machine, pass); @@ -349,11 +363,7 @@ const loadRepos = ( }); /** Every unfinished run, counted; a run whose file cannot be read is an error line and counts as active. */ -const runs = ( - world: PassWorld, - repos: ReadonlyArray, - pass: Pass, -): Effect.Effect | null> => +const runs = (world: PassWorld, pass: Pass): Effect.Effect | null> => Effect.gen(function* () { const listed = yield* Effect.result(world.unfinishedRuns()); if (listed._tag === "Failure") { @@ -379,12 +389,104 @@ const runs = ( ); continue; } - // The steps that follow a run of a loaded repo (issues #16 to #20) go here. - void repos; } return listed.success; }); +/** + * Calls `resume` for each confirmed run still in `triage` (or with no state yet), which goes on + * from its ledger, so a run cut short after the launch was sent is finished and its session + * never runs on untracked. A run of a repo that is not loaded is left alone. The other states + * are worked by later steps. It is not a new claim, so it is not counted as one. An issue put + * back to `ready` is left to the claim step, which finishes its run when it claims the issue. + */ +const workRuns = ( + world: PassWorld, + repos: ReadonlyArray, + known: Known, + pass: Pass, +): Effect.Effect => + Effect.gen(function* () { + for (const info of [...known.infos]) { + if (!inTriage(info)) continue; + const loaded = repos.find( + (l) => l.kind === "ready" && sameRepo(l.repo.watched.repo, info.id.repo), + ); + if (loaded === undefined || loaded.kind !== "ready") continue; + const repo = loaded.repo; + const name = repo.watched.repo; + const issue = Number(info.id.issue); + const notes = pass.notes(`${name}${mention(issue)}`, info.state); + const outcome = yield* Effect.result(world.resume(repo, issue)); + if (outcome._tag === "Failure") { + if (outcome.failure._tag === "GateRunningError") notes.parts.push(GATE_RUNNING); + else pass.failed(notes, "start", outcome.failure); + } else { + switch (outcome.success.kind) { + case "started": + notes.parts.push("session started, the issue is building"); + break; + case "stuck": + pass.stuck(notes, runIdToString(outcome.success.run)); + break; + case "skipped": + case "refused": { + const moved = outcome.success.kind === "skipped" ? outcome.success.moved : undefined; + if (moved === "ready") notes.parts.push(READY_AGAIN); + else { + const run = runIdToString(info.id); + yield* stalled(pass, notes, run, outcome.success.reason, moved === "closed"); + } + break; + } + } + } + yield* known.refresh(world, pass, name, issue); + yield* pass.flush(notes, info.id); + } + }); + +/** + * A ready issue the limit holds back only because of its own triage run whose session may be + * running: only the claim would finish that run (`finishOlderRuns`), so the slot never frees. + */ +const waitsOnItself = (pass: Pass, notes: Notes, run: string): Effect.Effect => + pass.attention( + `${notes.label} run ${run} is not progressing and holds a slot: its issue is ready again, and the new claim waits for a slot this run holds. Put the issue back to \`factory:triage\` assigned to the factory's user so \`start\` can finish this run, or free a slot`, + run, + "not-progressing", + ); + +const READY_AGAIN = "its issue is ready again, so it is left to the claim step"; + +/** A confirmed run still in `triage` (or before any state): what `finishOlderRuns` would end. */ +const inTriage = (run: RunInfo): boolean => + !run.unreadable && run.confirmed && (run.state === null || run.state === "triage"); + +/** + * A confirmed run in `triage` that `resume` would not take, for `why`: a person moved or closed + * its issue, so nothing in the pass will move it, and it holds a slot. softfac's `stalled` words. + * Nothing finishes a closed issue's run yet, so its line says how to let the start finish it. + */ +const stalled = ( + pass: Pass, + notes: Notes, + run: string, + why: string, + closed: boolean, +): Effect.Effect => { + const back = + "put it back to `factory:triage` assigned to the factory's user so `start` can finish it"; + const advice = closed + ? `If it was closed by mistake, reopen it and ${back}` + : `If you moved the issue by hand, ${back}`; + return pass.attention( + `${notes.label} run ${run} is not progressing and holds a slot: ${plain(why)}. ${advice}`, + run, + "not-progressing", + ); +}; + /** * The unfinished runs as this pass knows them: listed once at the start of the pass, then the * runs of an issue read again after each `start` for it, so the ledger is not listed whole again. @@ -424,7 +526,7 @@ class Known { /** * Claims `start` began that left no run it can go on from: the issue is `factory:triage` and * assigned to the factory's login alone, and no unfinished run for it is confirmed (`RunInfo`). - * `start` finishes such an issue (the design sends it to `stuck`), in the run it left when there + * `resume` finishes such an issue (the design sends it to `stuck`), in the run it left when there * is one, so a stuck path cut short is tried again on every pass. They are claims already made, * so the limit on new claims does not apply to them. */ @@ -451,7 +553,7 @@ const untrackedClaims = ( for (const issue of listing.success.items) { if (known.of(name, issue).some((i) => i.unreadable || i.confirmed)) continue; const notes = pass.notes(`${name}${mention(issue)}`, "triage"); - const run = yield* noteStart(world, pass, known, notes, repo, issue, { + const run = yield* noteStart(world, pass, known, notes, repo, issue, "resume", { claimed: (run) => `no run on record, claimed in run ${run}`, notTaken: "no run on record, not ours to resume", stuck: "no run on record", @@ -532,10 +634,10 @@ interface StartWords { } /** - * Calls `start` for the issue and puts what it did on `notes`: the run it names, when it made - * one, is the run the notes' errors belong to. The issue's runs are read again into `known` - * after the call. A refused claim is a `claim_skipped`; a claim counts only when the issue was - * `ready`. + * Calls `call` (`start`, or `resume` for a claim already made) for the issue and puts what it + * did on `notes`: the run it names, when it made one, is the run the notes' errors belong to. + * The issue's runs are read again into `known` after the call. A refused claim is a + * `claim_skipped`; a claim counts only when the issue was `ready`. */ const noteStart = ( world: PassWorld, @@ -544,12 +646,13 @@ const noteStart = ( notes: Notes, repo: Repo, issue: number, + call: "start" | "resume", words: StartWords, ): Effect.Effect => Effect.gen(function* () { const name = repo.watched.repo; const before = newestRun(known.of(name, issue)); - const outcome = yield* Effect.result(world.start(repo, issue)); + const outcome = yield* Effect.result(world[call](repo, issue)); if (outcome._tag === "Failure" && outcome.failure._tag === "GateRunningError") { // Nothing was called or written. notes.parts.push(GATE_RUNNING); @@ -561,7 +664,7 @@ const noteStart = ( return madeSince(newestRun(known.of(name, issue)), before); } switch (outcome.success.kind) { - case "claimed": { + case "started": { const run = runIdToString(outcome.success.run); notes.parts.push(words.claimed(run)); if (notes.state === "ready") pass.report.claimed += 1; @@ -667,13 +770,24 @@ const claim = ( // the machine's, said once after the loop. if (paused !== null) continue; // Before the blockers: with no slot free nothing is claimed, so they are not read. Every - // issue left is counted as left over, blocked or not. - if (active >= machine.maxConcurrentRuns) { + // issue left is counted as left over, blocked or not. The issue's own triage runs end with + // this claim (`finishOlderRuns`). One with no session leaves nothing running, so its slot is + // the claim's; one whose session may be running keeps its slot, so no second session + // starts past the limit. + const own = known.of(name, issue).filter(inTriage); + const freed = own.filter((run) => !run.session).length; + if (active - freed >= machine.maxConcurrentRuns) { left += 1; yield* pass.record({ event: "claim_skipped", data: { repo: name, issue: String(issue), reason: "limit" }, }); + // Its own running sessions fill the limit alone: when every other run frees its slot, + // the claim still waits, and only the claim would end them. + const running = own.filter((run) => run.session); + if (running.length > 0 && running.length >= machine.maxConcurrentRuns) { + yield* waitsOnItself(pass, notes, runIdToString(running[0]!.id)); + } continue; } // A blocked issue takes no slot. Nothing is written for it, and a read that fails holds it @@ -715,10 +829,10 @@ const claim = ( }); continue; } - const run = yield* noteStart(world, pass, known, notes, repo, issue, { - claimed: (run) => `claimed in run ${run}`, + const run = yield* noteStart(world, pass, known, notes, repo, issue, "start", { + claimed: (run) => `claimed, session started in run ${run}`, notTaken: "not claimed", - stuck: "claimed, then", + stuck: "claimed, but the launch failed", }); yield* pass.flush(notes, run); if (known.lost) return; diff --git a/apps/server/src/factory/runner/messages.test.ts b/apps/server/src/factory/runner/messages.test.ts new file mode 100644 index 000000000000..63d7e04fb095 --- /dev/null +++ b/apps/server/src/factory/runner/messages.test.ts @@ -0,0 +1,60 @@ +import { assert, it } from "@effect/vitest"; +import { describe } from "vite-plus/test"; + +import { launchMessage, sessionMayExistNote } from "./messages.ts"; + +/** The paragraph that names the GitHub identity, for the login `rawphp` and the `gh` directory `/gh/rawphp`. */ +const IDENTITY_WORDS = + 'You act on GitHub as rawphp. Your T3 server should have been started with GH_CONFIG_DIR=/gh/rawphp. Before anything else run gh api user --jq .login exactly as written, in the environment you were started with. Never set GH_CONFIG_DIR, GH_TOKEN or GH_HOST, and never use gh auth token or gh auth switch, on any command. If it prints any other login, end with FACTORY-STOP: stuck and make the first line of your message "the session\'s gh account is , not rawphp".'; + +describe("launchMessage", () => { + it("invokes /factory and says nobody is in the chat", () => { + assert.strictEqual( + launchMessage("rawphp/softfac", 10, "rawphp", "/gh/rawphp"), + `/factory rawphp/softfac#10\n\nThe factory daemon launched this session unattended, in a worktree it created for you. Nobody is in the chat. The daemon owns the issue's labels and comments and runs its own gate after your draft PR.\n\n${IDENTITY_WORDS}`, + ); + }); + + it("names the login and the gh config dir the session acts as", () => { + const message = launchMessage("acme/widgets", 7, "alice", "/Users/me/.config/softfac/gh-alice"); + assert.ok( + message.endsWith( + '\n\nYou act on GitHub as alice. Your T3 server should have been started with GH_CONFIG_DIR=/Users/me/.config/softfac/gh-alice. Before anything else run gh api user --jq .login exactly as written, in the environment you were started with. Never set GH_CONFIG_DIR, GH_TOKEN or GH_HOST, and never use gh auth token or gh auth switch, on any command. If it prints any other login, end with FACTORY-STOP: stuck and make the first line of your message "the session\'s gh account is , not alice".', + ), + message, + ); + assert.strictEqual(message.split("You act on GitHub as").length - 1, 1); + }); + + it("forbids every way of choosing the account and names the dir once", () => { + const message = launchMessage("acme/widgets", 42, "rawphp", "/factory/gh"); + const paragraph = message.slice(message.lastIndexOf("\n\n") + 2); + assert.ok( + paragraph.includes( + "run gh api user --jq .login exactly as written, in the environment you were started with.", + ), + paragraph, + ); + assert.ok( + paragraph.includes( + "Never set GH_CONFIG_DIR, GH_TOKEN or GH_HOST, and never use gh auth token or gh auth switch, on any command.", + ), + paragraph, + ); + assert.ok( + paragraph.includes("Your T3 server should have been started with GH_CONFIG_DIR=/factory/gh."), + paragraph, + ); + // The variable's name with a value appears only in the clause for the server. + assert.strictEqual(paragraph.split("GH_CONFIG_DIR=").length - 1, 1); + }); +}); + +describe("sessionMayExistNote", () => { + it("says a launch attempt may have started a session, and names the command id", () => { + assert.strictEqual( + sessionMayExistNote("the run passed its time limit", "softfac:rawphp__softfac__7__1:launch"), + "the run passed its time limit; a launch attempt may have started a session (launch command id softfac:rawphp__softfac__7__1:launch)", + ); + }); +}); diff --git a/apps/server/src/factory/runner/messages.ts b/apps/server/src/factory/runner/messages.ts index 6f0715c5628b..b82559b51913 100644 --- a/apps/server/src/factory/runner/messages.ts +++ b/apps/server/src/factory/runner/messages.ts @@ -21,6 +21,34 @@ export const CUT_SHORT_COMMENT = export const stuckComment = (reason: string): string => `The factory could not finish this issue and has released it.\n\nReason: ${reason}\n\nFix the cause, then set the label back to \`factory:ready\`.`; +/** The public reason when the issue text changed between the approval and the session. */ +export const TEXT_CHANGED = "the issue text was changed while the factory was starting"; + +/** The public reason when the launch was refused. The detail stays in the ledger. */ +export const LAUNCH_FAILED = "the factory could not launch a session"; + +/** The words a stuck run's ledger reason carries when a launch attempt may have started a session. */ +export const SESSION_MAY_EXIST = "a launch attempt may have started a session"; + +export const sessionMayExistNote = (what: string, commandId: string): string => + `${what}; ${SESSION_MAY_EXIST} (launch command id ${commandId})`; + +/** How the first line of a stuck session's message begins when gh acts as another account. */ +export const WRONG_ACCOUNT_PREFIX = "the session's gh account is "; + +/** + * The first message of a session for a GitHub issue: invokes `/factory`, says nobody is in the + * chat, and names the GitHub identity the session acts as (`login`, and the `ghConfigDir` its T3 + * server should have been started with) with what to do when `gh` says otherwise. + */ +export const launchMessage = ( + repo: string, + issue: number, + login: string, + ghConfigDir: string, +): string => + `/factory ${repo}#${issue}\n\nThe factory daemon launched this session unattended, in a worktree it created for you. Nobody is in the chat. The daemon owns the issue's labels and comments and runs its own gate after your draft PR.\n\nYou act on GitHub as ${login}. Your T3 server should have been started with GH_CONFIG_DIR=${ghConfigDir}. Before anything else run gh api user --jq .login exactly as written, in the environment you were started with. Never set GH_CONFIG_DIR, GH_TOKEN or GH_HOST, and never use gh auth token or gh auth switch, on any command. If it prints any other login, end with FACTORY-STOP: stuck and make the first line of your message "${WRONG_ACCOUNT_PREFIX}, not ${login}".`; + /** The issue's single status comment. */ export const statusBody = ( state: State, diff --git a/apps/server/src/factory/runner/progress.ts b/apps/server/src/factory/runner/progress.ts index c01706eb28ce..cdec8c808814 100644 --- a/apps/server/src/factory/runner/progress.ts +++ b/apps/server/src/factory/runner/progress.ts @@ -14,11 +14,24 @@ export interface Seen { readonly state: State | null; /** The `session_launched` line's thread. */ readonly threadId: string | null; + /** The `launch-sending` step: a launch call was about to be sent, so a session may exist. */ + readonly launchSent: boolean; + /** The `approval-rechecked` step: the issue text was checked again after the launch. */ + readonly rechecked: boolean; + /** The `session_launched` line's worktree path. */ + readonly worktree: string | null; + /** A `worktree_removed` or `worktree_remove_failed` line follows the last `session_launched`. */ + readonly worktreeDone: boolean; /** The reason on the `stuck` line: the comment is on the issue. */ readonly stuck: string | null; readonly finished: boolean; } +/** The ledger step that says a launch call is about to be sent. */ +export const LAUNCH_SENDING = "launch-sending"; +/** The ledger step that says the issue text was checked again after the launch. */ +export const APPROVAL_RECHECKED = "approval-rechecked"; + const field = (data: unknown, key: string): unknown => typeof data === "object" && data !== null ? (data as Record)[key] : undefined; @@ -27,6 +40,10 @@ export const readProgress = (entries: ReadonlyArray): Seen => { let approvedAt: string | null = null; let state: State | null = null; let threadId: string | null = null; + let launchSent = false; + let rechecked = false; + let worktree: string | null = null; + let worktreeDone = false; let stuck: string | null = null; let finished = false; for (const entry of entries) { @@ -39,6 +56,13 @@ export const readProgress = (entries: ReadonlyArray): Seen => { if (approvedAt === null && typeof step === "string" && step.startsWith("approved ")) { approvedAt = step.slice("approved ".length); } + if (step === APPROVAL_RECHECKED) rechecked = true; + if ( + typeof step === "string" && + (step === LAUNCH_SENDING || step.startsWith(`${LAUNCH_SENDING} resumed-from `)) + ) { + launchSent = true; + } break; } case "state_changed": { @@ -49,8 +73,15 @@ export const readProgress = (entries: ReadonlyArray): Seen => { case "session_launched": { const id = field(entry.data, "thread_id"); if (typeof id === "string") threadId = id; + const path = field(entry.data, "worktree"); + worktree = typeof path === "string" ? path : null; + worktreeDone = false; break; } + case "worktree_removed": + case "worktree_remove_failed": + worktreeDone = true; + break; case "stuck": { const reason = field(entry.data, "reason"); stuck = typeof reason === "string" ? reason : ""; @@ -61,5 +92,16 @@ export const readProgress = (entries: ReadonlyArray): Seen => { break; } } - return { claimed, approvedAt, state, threadId, stuck, finished }; + return { + claimed, + approvedAt, + state, + threadId, + launchSent, + rechecked, + worktree, + worktreeDone, + stuck, + finished, + }; }; diff --git a/apps/server/src/factory/runner/run.ts b/apps/server/src/factory/runner/run.ts new file mode 100644 index 000000000000..09c4ee156de7 --- /dev/null +++ b/apps/server/src/factory/runner/run.ts @@ -0,0 +1,202 @@ +/** + * What every step that stops or finishes a run shares, ported from softfac's `runner.rs`: the + * stuck path with its session stop, and the move to `factory:building`. Each step passes its own + * `Progress.step`, which names it in the ledger's causes. + */ +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; + +import { InvalidTransition, transition, type State } from "../domain/lifecycle.ts"; +import { ledgerCall as ledger, type LedgerError } from "../ledger/errors.ts"; +import { makeCause, type CauseStep } from "../ledger/events.ts"; +import type { Ledger } from "../ledger/ledger.ts"; +import type { BoundGitHub, GitHubError } from "../tracker/github.ts"; +import { ThreadNotFoundError } from "../t3/sessions.ts"; +import type { FactorySessions, SessionsError } from "../t3/sessions.ts"; +import { statusBody, stuckComment } from "./messages.ts"; +import type { Seen } from "./progress.ts"; + +/** What the stop and finish paths can fail with. */ +export type RunError = GitHubError | LedgerError | InvalidTransition | SessionsError; + +/** What the stop and finish paths call: the tracker and the sessions, whichever step runs them. */ +export interface RunContext { + readonly trackerUser: string; + readonly machine: string; + readonly tracker: Pick< + BoundGitHub, + "setState" | "unassign" | "postComment" | "upsertStatusComment" + >; + readonly sessions: Pick; +} + +/** + * Where the issue stands while the stuck actions are applied. `threadId`, `worktree` and + * `worktreeDone` start from the ledger and move as this call launches and stops the session. + */ +export interface Progress { + /** The step running, for the cause of every line written here. */ + readonly step: CauseStep; + state: State; + threadId: string | null; + worktree: string | null; + worktreeDone: boolean; + readonly seen: Seen; +} + +/** The words for what a failed call said: its message when it has one. */ +export const describe = (cause: unknown): string => + typeof cause === "object" && + cause !== null && + "message" in cause && + typeof cause.message === "string" + ? cause.message + : String(cause); + +/** The label, the status comment and the ledger line that follow a launch. Safe to repeat. */ +export const finishBuilding = ( + ctx: RunContext, + issue: number, + out: Ledger, + at: Progress, + threadId: string, +): Effect.Effect => + Effect.gen(function* () { + const building = yield* Effect.fromResult(transition(at.state, { kind: "session-starting" })); + for (const action of building.actions) { + switch (action.kind) { + case "set-label": + yield* ctx.tracker.setState(issue, action.state); + break; + case "launch-session": + // Launched already. + break; + case "update-status-comment": + yield* ctx.tracker.upsertStatusComment( + issue, + ctx.trackerUser, + statusBody("building", ctx.machine, threadId, null), + ); + break; + default: + return yield* new InvalidTransition({ state: at.state, event: "session-starting" }); + } + } + yield* ledger(() => + out.append( + { event: "state_changed", data: { from: at.state, to: building.next } }, + makeCause(at.step, "session-started"), + ), + ); + at.state = building.next; + }); + +/** + * Stops the session before the run goes stuck: archives the thread, then removes its worktree + * unless the ledger says that is done. An archive or a read of the thread that fails fails the + * start with nothing written, and the next pass archives again. A worktree that cannot be + * removed is written down and does not stop the run. + */ +const stopSession = (ctx: RunContext, out: Ledger, at: Progress): Effect.Effect => + Effect.gen(function* () { + if (at.threadId === null) return; + yield* ctx.sessions.archive(at.threadId); + if (at.worktreeDone) return; + const reported = yield* ctx.sessions.thread(at.threadId).pipe( + Effect.map((thread) => thread.worktreePath), + Effect.catchIf(isThreadNotFound, () => Effect.succeed(null)), + ); + const path = reported ?? at.worktree; + if (path !== null) { + const removed = yield* Effect.result(ctx.sessions.removeWorktree(path)); + if (removed._tag === "Success") { + yield* ledger(() => + out.append( + { event: "worktree_removed", data: { path } }, + makeCause(at.step, "worktree-removed"), + ), + ); + } else { + yield* ledger(() => + out.append( + { + event: "worktree_remove_failed", + data: { path, error: describe(removed.failure.cause) }, + }, + makeCause(at.step, "worktree-remove-failed"), + ), + ); + } + } + at.worktreeDone = true; + }); + +const isThreadNotFound = Schema.is(ThreadNotFoundError); + +/** Why a run goes stuck: `reason` is the public one; `detail`, when there is one, stays in the ledger. */ +export interface Stop { + readonly reason: string; + readonly detail?: string; +} + +/** + * The stuck path. The session is stopped first (see `stopSession`), so a failure of that leaves + * the issue as it was and the next pass tries again. Then the actions `lifecycle.transition` + * asks for on a failure, in order, then the `state_changed` line and the summary line. The + * comment says `stop.reason`; the ledger's `stuck` line says `stop.detail` when there is one. A + * comment is skipped when the ledger already records it; the label, the release and the status + * comment are repeated safely. `code` is the reason of every line it writes. + */ +export const goStuck = ( + ctx: RunContext, + issue: number, + out: Ledger, + at: Progress, + code: string, + stop: Stop, +): Effect.Effect => + Effect.gen(function* () { + const cause = makeCause(at.step, code); + if (at.state === "stuck") { + yield* ledger(() => out.finish("stuck", null, null, cause)); + return; + } + yield* stopSession(ctx, out, at); + const recorded = stop.detail ?? stop.reason; + const failed = yield* Effect.fromResult( + transition(at.state, { kind: "failed", reason: recorded }), + ); + let movedTo: State | null = null; + let stuckPosted = at.seen.stuck !== null; + for (const action of failed.actions) { + switch (action.kind) { + case "post-stuck-comment": + if (!stuckPosted) { + yield* ctx.tracker.postComment(issue, stuckComment(stop.reason)); + yield* ledger(() => out.append({ event: "stuck", data: { reason: recorded } }, cause)); + stuckPosted = true; + } + break; + case "set-label": + yield* ctx.tracker.setState(issue, action.state); + movedTo = action.state; + break; + case "unassign": + yield* ctx.tracker.unassign(issue, ctx.trackerUser); + break; + case "update-status-comment": + yield* ctx.tracker.upsertStatusComment( + issue, + ctx.trackerUser, + statusBody(movedTo ?? at.state, ctx.machine, at.threadId, null), + ); + break; + default: + return yield* new InvalidTransition({ state: at.state, event: "failed" }); + } + } + yield* ledger(() => + out.append({ event: "state_changed", data: { from: at.state, to: failed.next } }, cause), + ); + yield* ledger(() => out.finish("stuck", null, null, cause)); + }); diff --git a/apps/server/src/factory/runner/start.test.ts b/apps/server/src/factory/runner/start.test.ts index bd4ca8f0bad4..94b583cda8bc 100644 --- a/apps/server/src/factory/runner/start.test.ts +++ b/apps/server/src/factory/runner/start.test.ts @@ -4,20 +4,29 @@ import * as NodePath from "node:path"; import { assert, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; +import * as Logger from "effect/Logger"; import * as Schema from "effect/Schema"; import { afterEach, describe } from "vite-plus/test"; import { timeLimitOnlyProjectConfig, type ProjectConfig } from "../config.ts"; -import type { Approval } from "../domain/approval.ts"; +import type { Approval, TextCheck } from "../domain/approval.ts"; import { unapprovedComment } from "../domain/approval.ts"; import type { State } from "../domain/lifecycle.ts"; import { gateLockFile } from "../gate.ts"; -import { makeCause } from "../ledger/events.ts"; +import { makeCause, type Cause, type RunEvent } from "../ledger/events.ts"; import { Ledger } from "../ledger/ledger.ts"; -import { scanMachineLog } from "../ledger/machineLog.ts"; +import { MACHINE_LOG, scanMachineLog } from "../ledger/machineLog.ts"; import { readRun } from "../ledger/read.ts"; -import { runPath, type RunId } from "../ledger/runId.ts"; +import { parseRunId, runPath, type RunId } from "../ledger/runId.ts"; import { T0, useTempDirs } from "../ledger/testSupport.ts"; +import { + SessionsError, + ThreadNotFoundError, + type LaunchRequest, + type Launched, + type Model, + type Thread, +} from "../t3/sessions.ts"; import { GitHubFailed, type ClaimOutcome, @@ -25,8 +34,25 @@ import { type GitHubError, type IssueDetail, } from "../tracker/github.ts"; -import { CANNOT_CONFIRM, CUT_SHORT_COMMENT, statusBody, stuckComment } from "./messages.ts"; -import { GateRunningError, start, type StartOutcome, type Tracker } from "./start.ts"; +import { + CANNOT_CONFIRM, + CUT_SHORT_COMMENT, + LAUNCH_FAILED, + TEXT_CHANGED, + launchMessage, + sessionMayExistNote, + statusBody, + stuckComment, +} from "./messages.ts"; +import { + GateRunningError, + resume, + start, + type Sessions, + type StartContext, + type StartOutcome, + type Tracker, +} from "./start.ts"; const tempDir = useTempDirs(); @@ -71,11 +97,14 @@ const fakeTracker = (script: { claims?: Array; assignFails?: boolean; setStateFails?: boolean; + unassignFails?: boolean; newestComment?: FactoryComment | null; + textChecks?: Array; }) => { const calls: Array = []; const issues = [...(script.issues ?? [detail("ready")])]; const claims = [...(script.claims ?? [{ kind: "claimed" } as const])]; + const textChecks = [...(script.textChecks ?? [])]; const answer = (queue: Array, fallback: A) => { const next = queue.length > 1 ? queue.shift()! : (queue[0] ?? fallback); return next instanceof Object && "_tag" in next && typeof next._tag === "string" @@ -107,7 +136,7 @@ const fakeTracker = (script: { }, unassign: (n, user) => { calls.push(`unassign ${n} ${user}`); - return Effect.void; + return script.unassignFails ? Effect.fail(ghFailure()) : Effect.void; }, postComment: (n, body) => { calls.push(`comment ${n} ${body}`); @@ -121,26 +150,200 @@ const fakeTracker = (script: { calls.push(`newest ${n}`); return Effect.succeed(script.newestComment ?? null); }, + textCheck: (n, since, instructors) => { + calls.push(`textCheck ${n} ${since} ${instructors.join(",")}`); + return answer(textChecks, "unchanged"); + }, }; return { tracker, calls }; }; +const isSessionsError = Schema.is(SessionsError); +const isThreadNotFound = Schema.is(ThreadNotFoundError); + +const MODEL: Model = { instanceId: "claudeAgent", model: "claude-opus-5-5" }; +const LAUNCH_ID = "softfac:rawphp__softfac__7__1:launch"; +const WORKTREE = "/t3/wt"; + +/** + * Scripted answers for the T3 sessions the claim calls. Each method pops its next answer, or + * repeats the last one. `launch` fails the test when the run's ledger does not yet hold the + * `launch-sending` step: the step is written before the call, as softfac's own hook checks. + */ +const fakeSessions = ( + dataDir: string, + script: { + project?: string | null | SessionsError; + launches?: Array; + archive?: Array; + thread?: Thread | ThreadNotFoundError; + removeWorktree?: SessionsError; + }, +) => { + const calls: Array = []; + const launchRequests: Array = []; + const launches = [...(script.launches ?? [])]; + const archives = [...(script.archive ?? [])]; + const popped = (queue: Array): A | undefined => + queue.length > 1 ? queue.shift() : queue[0]; + const result = (answer: A | SessionsError | undefined, fallback: A) => + isSessionsError(answer) + ? Effect.fail(answer) + : Effect.succeed(answer === undefined ? fallback : answer); + const sessions: Sessions = { + findProject: (root) => { + calls.push(`findProject ${root}`); + return result(script.project === undefined ? "proj-1" : script.project, "proj-1"); + }, + launch: (request) => { + calls.push(`launch ${request.commandId}`); + launchRequests.push(request); + const ledgerFile = runPath( + dataDir, + parseRunId(request.commandId.slice("softfac:".length, -":launch".length)), + ); + if (!NodeFS.readFileSync(ledgerFile, "utf8").includes('"step":"launch-sending"')) { + return Effect.die(new Error("launch was called before the launch-sending step")); + } + return result(popped(launches), { + threadId: "thread-1", + worktreePath: WORKTREE, + branch: "t3code/abc", + }); + }, + thread: (threadId) => { + calls.push(`thread ${threadId}`); + return isThreadNotFound(script.thread) + ? Effect.fail(script.thread) + : Effect.succeed( + script.thread ?? { worktreePath: WORKTREE, branch: "t3code/abc", archived: false }, + ); + }, + archive: (threadId) => { + calls.push(`archive ${threadId}`); + const answer = popped(archives); + return isSessionsError(answer) ? Effect.fail(answer) : Effect.void; + }, + removeWorktree: (path) => { + calls.push(`removeWorktree ${path}`); + return script.removeWorktree === undefined ? Effect.void : Effect.fail(script.removeWorktree); + }, + }; + return { sessions, calls, launchRequests }; +}; + const run7: RunId = { repo: REPO, issue: "7", attempt: 1 }; const isGateRunning = Schema.is(GateRunningError); -const ctxFor = (dataDir: string, tracker: Tracker, clock: () => number = () => T0) => ({ +const ctxFor = ( + dataDir: string, + tracker: Tracker, + sessions: Sessions, + extra: Partial = {}, +): StartContext => ({ repo: REPO, trackerUser: USER, machine: "mac-mini", project: project(), dataDir, tracker, - clock, + repoPath: "/work/softfac", + ghConfigDir: "/gh/rawphp", + model: MODEL, + sessions, + clock: () => T0, + ...extra, }); -const startIn = (dataDir: string, tracker: Tracker, clock?: () => number) => - start(ctxFor(dataDir, tracker, clock), 7); +const startIn = ( + dataDir: string, + tracker: Tracker, + sessions: Sessions = fakeSessions(dataDir, {}).sessions, + extra: Partial = {}, +) => start(ctxFor(dataDir, tracker, sessions, extra), 7); + +const cause = (reason: string): Cause => makeCause("start", reason); +const BUILDING = statusBody("building", "mac-mini", "thread-1", null); +const FINISHED = { + outcome: "stuck", + total_ms: 0, + gate_attempts: 0, + review_cycles: null, + pr_url: null, + pr_end: null, +}; + +/** The lines of a run that was started, in order: what the ledger holds when the issue is `building`. */ +const startedLines = (): Array<{ line: RunEvent; cause: Cause }> => [ + { + line: { event: "step_changed", data: { step: `approved ${LABELED}` } }, + cause: makeCause("start", "approved", { actor: "tomkaczocha" }), + }, + { + line: { event: "claimed", data: { machine: "mac-mini", github_user: USER } }, + cause: cause("claimed"), + }, + { + line: { event: "state_changed", data: { from: "ready", to: "triage" } }, + cause: cause("claimed"), + }, + { + line: { event: "step_changed", data: { step: "launch-sending" } }, + cause: cause("launch-sending"), + }, + { + line: { + event: "session_launched", + data: { + thread_id: "thread-1", + worktree: WORKTREE, + branch: "t3code/abc", + model_instance: "claudeAgent", + model: "claude-opus-5-5", + resumed_from: null, + }, + }, + cause: cause("session-launched"), + }, + { + line: { event: "step_changed", data: { step: "approval-rechecked" } }, + cause: cause("approval-rechecked"), + }, + { + line: { event: "state_changed", data: { from: "triage", to: "building" } }, + cause: cause("session-started"), + }, +]; + +const startedEvents = () => + startedLines().map(({ line, cause: why }) => ({ + event: line.event, + data: line.data, + cause: why, + })); + +/** A run file holding the first `lines` lines of `startedLines()`, as an earlier pass left it. */ +const seedRun = (dataDir: string, lines: ReadonlyArray<{ line: RunEvent; cause: Cause }>) => { + const ledger = Ledger.create(dataDir, run7, { clock: () => T0 - 1000 }); + handles.push(ledger); + for (const { line, cause: why } of lines) ledger.append(line, why); +}; + +/** The lines a run ends with when it goes stuck from `triage` on the stuck line `reason`. */ +const stuckTail = (reason: string, code: string) => [ + { event: "stuck", data: { reason }, cause: cause(code) }, + { event: "state_changed", data: { from: "triage", to: "stuck" }, cause: cause(code) }, + { event: "finished", data: FINISHED, cause: cause(code) }, +]; + +/** What a stuck run does to the issue, after the session is stopped: the public comment, the label, the release, the status. */ +const stuckCalls = (publicReason: string, thread: string | null) => [ + `comment 7 ${stuckComment(publicReason)}`, + "setState 7 stuck", + "unassign 7 rawphp", + `status 7 rawphp ${statusBody("stuck", "mac-mini", thread, null)}`, +]; const entriesOf = (dataDir: string, id: RunId = run7) => readRun(dataDir, id).entries.map((e) => ({ event: e.event, data: e.data, cause: e.cause })); @@ -170,34 +373,76 @@ const olderRun = ( }; describe("start: a fresh ready issue", () => { - it.effect("checks the approval, claims, and records the approval, the claim and the state", () => + it.effect("claims, launches the session and moves the issue to building", () => Effect.gen(function* () { const dataDir = tempDir(); const { tracker, calls } = fakeTracker({}); - const outcome = yield* startIn(dataDir, tracker); - assert.deepStrictEqual(outcome, { kind: "claimed", run: run7 } satisfies StartOutcome); + const t3 = fakeSessions(dataDir, {}); + const outcome = yield* startIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(outcome, { + kind: "started", + run: run7, + threadId: "thread-1", + } satisfies StartOutcome); assert.deepStrictEqual(calls, [ "issue 7", "approval 7 tomkaczocha", "claim 7 rawphp ready->triage", + `textCheck 7 ${LABELED} tomkaczocha`, + "setState 7 building", + `status 7 rawphp ${BUILDING}`, ]); - assert.deepStrictEqual(entriesOf(dataDir), [ - { - event: "step_changed", - data: { step: `approved ${LABELED}` }, - cause: { step: "start", reason: "approved", actor: "tomkaczocha" }, - }, + assert.deepStrictEqual(t3.calls, ["findProject /work/softfac", `launch ${LAUNCH_ID}`]); + assert.deepStrictEqual(t3.launchRequests, [ { - event: "claimed", - data: { machine: "mac-mini", github_user: "rawphp" }, - cause: { step: "start", reason: "claimed" }, - }, - { - event: "state_changed", - data: { from: "ready", to: "triage" }, - cause: { step: "start", reason: "claimed" }, + commandId: LAUNCH_ID, + projectId: "proj-1", + title: "rawphp/softfac#7 Build step", + baseRef: "main", + startFromOrigin: false, + message: launchMessage(REPO, 7, USER, "/gh/rawphp"), + model: MODEL, }, ]); + assert.deepStrictEqual(entriesOf(dataDir), startedEvents()); + }), + ); + + it.effect("records the model the launch call sent", () => + Effect.gen(function* () { + const dataDir = tempDir(); + const { tracker } = fakeTracker({}); + const t3 = fakeSessions(dataDir, {}); + const model = { instanceId: "codex", model: "gpt-5.5" }; + yield* startIn(dataDir, tracker, t3.sessions, { model }); + assert.deepStrictEqual(t3.launchRequests[0]!.model, model); + assert.deepStrictEqual(entriesOf(dataDir)[4]!.data, { + thread_id: "thread-1", + worktree: WORKTREE, + branch: "t3code/abc", + model_instance: "codex", + model: "gpt-5.5", + resumed_from: null, + }); + }), + ); + + it.effect("records a launch that has no worktree yet", () => + Effect.gen(function* () { + const dataDir = tempDir(); + const { tracker } = fakeTracker({}); + const t3 = fakeSessions(dataDir, { + launches: [{ threadId: "thread-1", worktreePath: null, branch: null }], + }); + yield* startIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(entriesOf(dataDir)[4]!.data, { + thread_id: "thread-1", + worktree: null, + branch: null, + model_instance: "claudeAgent", + model: "claude-opus-5-5", + resumed_from: null, + }); }), ); @@ -274,18 +519,21 @@ describe("start: a fresh ready issue", () => { it.effect("an issue that is assigned, closed or not ready is skipped with nothing written", () => Effect.gen(function* () { - for (const [issue, reason] of [ - [detail("ready", ["a", "b"]), "issue #7 is already assigned to a, b"], - [detail("ready", [], true), "issue #7 is closed"], - [detail("building", [USER]), "issue #7 is not labelled factory:ready"], - [detail("triage", ["eve"]), "issue #7 is not labelled factory:ready"], - [detail(null), "issue #7 is not labelled factory:ready"], + const notOurs = "issue #7 is not factory:triage assigned to rawphp alone"; + for (const [issue, expected] of [ + [detail("ready", ["a", "b"]), { reason: "issue #7 is already assigned to a, b" }], + [detail("ready", [], true), { reason: "issue #7 is closed", moved: "closed" }], + [detail("building", [USER]), { reason: notOurs }], + [detail("triage", ["eve"]), { reason: notOurs }], + [detail(null), { reason: notOurs }], ] as const) { const dataDir = tempDir(); const { tracker, calls } = fakeTracker({ issues: [issue] }); - const outcome = yield* startIn(dataDir, tracker); - assert.deepStrictEqual(outcome, { kind: "skipped", reason }); + const t3 = fakeSessions(dataDir, {}); + const outcome = yield* startIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(outcome, { kind: "skipped", ...expected }); assert.deepStrictEqual(calls, ["issue 7"]); + assert.deepStrictEqual(t3.calls, []); assert.strictEqual(NodeFS.existsSync(NodePath.join(dataDir, "runs")), false); } }), @@ -295,17 +543,44 @@ describe("start: a fresh ready issue", () => { Effect.gen(function* () { const dataDir = tempDir(); const { tracker } = fakeTracker({ claims: [{ kind: "skipped-assigned", assignees: ["a"] }] }); - const outcome = yield* startIn(dataDir, tracker); + const t3 = fakeSessions(dataDir, {}); + const outcome = yield* startIn(dataDir, tracker, t3.sessions); assert.deepStrictEqual(outcome, { kind: "skipped", reason: "issue #7 is already assigned to a", }); + // T3 is asked before the claim; nothing is launched for a claim someone else won. + assert.deepStrictEqual(t3.calls, ["findProject /work/softfac"]); assert.strictEqual(NodeFS.existsSync(runPath(dataDir, run7)), false); }), ); }); describe("start: a claim call that fails", () => { + it.effect("is finished even when the machine log cannot be written, which is logged", () => + Effect.gen(function* () { + const dataDir = tempDir(); + // A directory where the log file goes: the machine log cannot be opened. + NodeFS.mkdirSync(NodePath.join(dataDir, MACHINE_LOG), { recursive: true }); + const { tracker } = fakeTracker({ + issues: [detail("ready"), detail("triage", [USER])], + claims: [ghFailure()], + }); + const warnings: Array = []; + const capture = Logger.make(({ logLevel, message }) => { + if (logLevel === "Warn") + warnings.push(String(Array.isArray(message) ? message[0] : message)); + }); + const outcome = yield* startIn(dataDir, tracker).pipe( + Effect.provide(Logger.layer([capture], { mergeWithExisting: false })), + ); + assert.deepStrictEqual(outcome, { kind: "started", run: run7, threadId: "thread-1" }); + assert.deepStrictEqual(warnings, [ + "claim_repaired (finished) for rawphp/softfac#7 could not be written to the machine log", + ]); + }), + ); + it.effect("is finished when GitHub applied it, and says so in the machine log", () => Effect.gen(function* () { const dataDir = tempDir(); @@ -314,8 +589,8 @@ describe("start: a claim call that fails", () => { claims: [ghFailure()], }); const outcome = yield* startIn(dataDir, tracker); - assert.deepStrictEqual(outcome, { kind: "claimed", run: run7 }); - assert.deepStrictEqual(calls, [ + assert.deepStrictEqual(outcome, { kind: "started", run: run7, threadId: "thread-1" }); + assert.deepStrictEqual(calls.slice(0, 4), [ "issue 7", "approval 7 tomkaczocha", "claim 7 rawphp ready->triage", @@ -325,7 +600,7 @@ describe("start: a claim call that fails", () => { scanMachineLog(dataDir).map((e) => [e.event, e.data]), [["claim_repaired", { repo: REPO, issue: 7, outcome: "finished" }]], ); - assert.deepStrictEqual(stepsOf(dataDir), ["step_changed", "claimed", "state_changed"]); + assert.deepStrictEqual(entriesOf(dataDir), startedEvents()); }), ); @@ -336,8 +611,12 @@ describe("start: a claim call that fails", () => { issues: [detail("ready"), detail("triage", [])], claims: [ghFailure()], }); - assert.deepStrictEqual(yield* startIn(dataDir, tracker), { kind: "claimed", run: run7 }); - assert.deepStrictEqual(calls.slice(3), ["issue 7", "assign 7 rawphp"]); + assert.deepStrictEqual(yield* startIn(dataDir, tracker), { + kind: "started", + run: run7, + threadId: "thread-1", + }); + assert.deepStrictEqual(calls.slice(3, 5), ["issue 7", "assign 7 rawphp"]); }), ); @@ -348,9 +627,12 @@ describe("start: a claim call that fails", () => { issues: [detail("ready"), detail("ready", [])], claims: [ghFailure()], }); - const error = yield* Effect.flip(startIn(dataDir, tracker)); + const t3 = fakeSessions(dataDir, {}); + const error = yield* Effect.flip(startIn(dataDir, tracker, t3.sessions)); assert.strictEqual(error._tag, "GitHubFailed"); assert.deepStrictEqual(calls.slice(3), ["issue 7"]); + // Nothing was launched, and no run file exists to hold a launch. + assert.deepStrictEqual(t3.calls, ["findProject /work/softfac"]); assert.strictEqual(NodeFS.existsSync(runPath(dataDir, run7)), false); assert.deepStrictEqual(scanMachineLog(dataDir), []); }), @@ -388,8 +670,12 @@ describe("start: a claim call that fails", () => { issues: [detail("ready"), detail("ready", [USER])], claims: [ghFailure()], }); - assert.deepStrictEqual(yield* startIn(dataDir, tracker), { kind: "claimed", run: run7 }); - assert.deepStrictEqual(calls.slice(3), ["issue 7", "setState 7 triage"]); + assert.deepStrictEqual(yield* startIn(dataDir, tracker), { + kind: "started", + run: run7, + threadId: "thread-1", + }); + assert.deepStrictEqual(calls.slice(3, 5), ["issue 7", "setState 7 triage"]); assert.deepStrictEqual( scanMachineLog(dataDir).map((e) => [e.event, e.data]), [["claim_repaired", { repo: REPO, issue: 7, outcome: "finished" }]], @@ -404,8 +690,16 @@ describe("start: a claim call that fails", () => { issues: [detail("ready"), detail(null, [])], claims: [ghFailure()], }); - assert.deepStrictEqual(yield* startIn(dataDir, tracker), { kind: "claimed", run: run7 }); - assert.deepStrictEqual(calls.slice(3), ["issue 7", "assign 7 rawphp", "setState 7 triage"]); + assert.deepStrictEqual(yield* startIn(dataDir, tracker), { + kind: "started", + run: run7, + threadId: "thread-1", + }); + assert.deepStrictEqual(calls.slice(3, 6), [ + "issue 7", + "assign 7 rawphp", + "setState 7 triage", + ]); assert.deepStrictEqual( scanMachineLog(dataDir).map((e) => [e.event, e.data]), [["claim_repaired", { repo: REPO, issue: 7, outcome: "finished" }]], @@ -446,15 +740,12 @@ describe("start: an issue in triage that is ours", () => { Effect.gen(function* () { const dataDir = tempDir(); const { tracker, calls } = fakeTracker({ issues: [detail("triage", [USER])] }); - const outcome = yield* startIn(dataDir, tracker); + const t3 = fakeSessions(dataDir, {}); + const outcome = yield* startIn(dataDir, tracker, t3.sessions); assert.deepStrictEqual(outcome, { kind: "stuck", run: run7, reason: CANNOT_CONFIRM }); - assert.deepStrictEqual(calls, [ - "issue 7", - `comment 7 ${stuckComment(CANNOT_CONFIRM)}`, - "setState 7 stuck", - "unassign 7 rawphp", - `status 7 rawphp ${statusBody("stuck", "mac-mini", null, null)}`, - ]); + assert.deepStrictEqual(calls, ["issue 7", ...stuckCalls(CANNOT_CONFIRM, null)]); + // No session was launched, so no T3 call at all. + assert.deepStrictEqual(t3.calls, []); const code = { step: "start", reason: "approval-unconfirmed" }; assert.deepStrictEqual(entriesOf(dataDir), [ { @@ -502,14 +793,16 @@ describe("start: an issue in triage that is ours", () => { }), ); - it.effect("with an unfinished run that was approved is left to that run", () => + it.effect("with an unfinished run in another state is left to the step that owns it", () => Effect.gen(function* () { const dataDir = tempDir(); - olderRun(dataDir, 1, "triage", false, true); + olderRun(dataDir, 1, "building", false, true); const { tracker, calls } = fakeTracker({ issues: [detail("triage", [USER])] }); - const outcome = yield* startIn(dataDir, tracker); + const t3 = fakeSessions(dataDir, {}); + const outcome = yield* startIn(dataDir, tracker, t3.sessions); assert.strictEqual(outcome.kind, "skipped"); assert.deepStrictEqual(calls, ["issue 7"]); + assert.deepStrictEqual(t3.calls, []); assert.deepStrictEqual(stepsOf(dataDir), ["step_changed", "claimed", "state_changed"]); }), ); @@ -576,7 +869,11 @@ describe("start: older runs of the issue", () => { olderRun(dataDir, 4, "stuck", true); const { tracker } = fakeTracker({}); const outcome = yield* startIn(dataDir, tracker); - assert.deepStrictEqual(outcome, { kind: "claimed", run: { ...run7, attempt: 5 } }); + assert.deepStrictEqual(outcome, { + kind: "started", + run: { ...run7, attempt: 5 }, + threadId: "thread-1", + }); for (const [id, expected] of [ [a, "needs-info"], [b, "pr-open"], @@ -612,3 +909,478 @@ describe("start: older runs of the issue", () => { }), ); }); + +const NO_PROJECT = "no T3 project for this repo on this machine"; +const REFUSED = "Thread launch failed during resolve-project."; +const sessionsError = (operation: "archive" | "find-project" | "launch", message: string) => + new SessionsError({ operation, cause: new Error(message) }); + +describe("start: the launch", () => { + it.effect("a repo with no T3 project goes stuck before any launch is sent", () => + Effect.gen(function* () { + const dataDir = tempDir(); + const { tracker, calls } = fakeTracker({}); + const t3 = fakeSessions(dataDir, { project: null }); + const outcome = yield* startIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(outcome, { kind: "stuck", run: run7, reason: NO_PROJECT }); + assert.deepStrictEqual(calls, [ + "issue 7", + "approval 7 tomkaczocha", + "claim 7 rawphp ready->triage", + ...stuckCalls(LAUNCH_FAILED, null), + ]); + assert.deepStrictEqual(t3.calls, ["findProject /work/softfac"]); + assert.deepStrictEqual(entriesOf(dataDir), [ + ...startedEvents().slice(0, 3), + ...stuckTail(NO_PROJECT, "launch-failed"), + ]); + }), + ); + + it.effect("a T3 that cannot be reached when the project is looked up claims nothing", () => + Effect.gen(function* () { + const dataDir = tempDir(); + const { tracker, calls } = fakeTracker({}); + const t3 = fakeSessions(dataDir, { project: sessionsError("find-project", "T3 is down") }); + const error = yield* Effect.flip(startIn(dataDir, tracker, t3.sessions)); + assert.strictEqual(error._tag, "FactorySessionsError"); + assert.deepStrictEqual(calls, ["issue 7", "approval 7 tomkaczocha"]); + assert.strictEqual(NodeFS.existsSync(runPath(dataDir, run7)), false); + }), + ); + + it.effect("a launch T3 refuses goes stuck with the detail in the ledger only", () => + Effect.gen(function* () { + const dataDir = tempDir(); + const { tracker, calls } = fakeTracker({}); + const t3 = fakeSessions(dataDir, { launches: [sessionsError("launch", REFUSED)] }); + const detailText = sessionMayExistNote(REFUSED, LAUNCH_ID); + const outcome = yield* startIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(outcome, { kind: "stuck", run: run7, reason: detailText }); + // The public comment never carries T3's words. + assert.deepStrictEqual(calls.slice(3), stuckCalls(LAUNCH_FAILED, null)); + assert.deepStrictEqual(t3.calls, ["findProject /work/softfac", `launch ${LAUNCH_ID}`]); + assert.deepStrictEqual(entriesOf(dataDir), [ + ...startedEvents().slice(0, 4), + ...stuckTail(detailText, "launch-failed"), + ]); + }), + ); + + it.effect("a second attempt gets the next number and its own command id", () => + Effect.gen(function* () { + const dataDir = tempDir(); + olderRun(dataDir, 1, "stuck", true); + const { tracker } = fakeTracker({}); + const t3 = fakeSessions(dataDir, {}); + const outcome = yield* startIn(dataDir, tracker, t3.sessions); + const second = { ...run7, attempt: 2 }; + assert.deepStrictEqual(outcome, { kind: "started", run: second, threadId: "thread-1" }); + assert.deepStrictEqual(t3.calls, [ + "findProject /work/softfac", + "launch softfac:rawphp__softfac__7__2:launch", + ]); + assert.deepStrictEqual(entriesOf(dataDir, second), startedEvents()); + assert.deepStrictEqual(stepsOf(dataDir), ["claimed", "state_changed", "finished"]); + }), + ); + + it.effect("a text that changed before the second look archives the session and goes stuck", () => + Effect.gen(function* () { + for (const check of ["changed", "unjudged"] as const) { + const dataDir = tempDir(); + const { tracker, calls } = fakeTracker({ textChecks: [check] }); + const t3 = fakeSessions(dataDir, {}); + const outcome = yield* startIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(outcome, { kind: "stuck", run: run7, reason: TEXT_CHANGED }); + assert.deepStrictEqual(calls, [ + "issue 7", + "approval 7 tomkaczocha", + "claim 7 rawphp ready->triage", + `textCheck 7 ${LABELED} tomkaczocha`, + ...stuckCalls(TEXT_CHANGED, "thread-1"), + ]); + assert.deepStrictEqual(t3.calls, [ + "findProject /work/softfac", + `launch ${LAUNCH_ID}`, + "archive thread-1", + "thread thread-1", + `removeWorktree ${WORKTREE}`, + ]); + assert.deepStrictEqual(entriesOf(dataDir), [ + ...startedEvents().slice(0, 5), + { + event: "worktree_removed", + data: { path: WORKTREE }, + cause: cause("worktree-removed"), + }, + ...stuckTail(TEXT_CHANGED, "text-changed-before-start"), + ]); + } + }), + ); + + it.effect( + "an archive that fails is an error before anything is written, and the next start retries", + () => + Effect.gen(function* () { + const dataDir = tempDir(); + const { tracker, calls } = fakeTracker({ + issues: [detail("ready"), detail("triage", [USER])], + textChecks: ["changed", "changed"], + }); + const t3 = fakeSessions(dataDir, { + archive: [sessionsError("archive", "T3 is busy"), undefined], + }); + const error = yield* Effect.flip(startIn(dataDir, tracker, t3.sessions)); + assert.strictEqual(error._tag, "FactorySessionsError"); + assert.deepStrictEqual(entriesOf(dataDir), startedEvents().slice(0, 5)); + assert.strictEqual(calls.at(-1), `textCheck 7 ${LABELED} tomkaczocha`); + const outcome = yield* startIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(outcome, { kind: "stuck", run: run7, reason: TEXT_CHANGED }); + assert.strictEqual(t3.calls.filter((c) => c.startsWith("launch ")).length, 1); + assert.strictEqual(t3.calls.filter((c) => c.startsWith("archive ")).length, 2); + }), + ); + + it.effect("the second look comes after the launch is recorded and before the label", () => + Effect.gen(function* () { + const dataDir = tempDir(); + const first = fakeTracker({ textChecks: [ghFailure()] }); + const t3 = fakeSessions(dataDir, {}); + const error = yield* Effect.flip(startIn(dataDir, first.tracker, t3.sessions)); + assert.strictEqual(error._tag, "GitHubFailed"); + assert.deepStrictEqual(entriesOf(dataDir), startedEvents().slice(0, 5)); + assert.strictEqual(first.calls.at(-1), `textCheck 7 ${LABELED} tomkaczocha`); + // The next start finds the thread in the ledger and does not launch again. + const second = fakeTracker({ issues: [detail("triage", [USER])] }); + const outcome = yield* startIn(dataDir, second.tracker, t3.sessions); + assert.deepStrictEqual(outcome, { kind: "started", run: run7, threadId: "thread-1" }); + assert.strictEqual(t3.calls.filter((c) => c.startsWith("launch ")).length, 1); + assert.deepStrictEqual(entriesOf(dataDir), startedEvents()); + }), + ); + + it.effect("a label change that fails after the launch is a runner error with no rollback", () => + Effect.gen(function* () { + const dataDir = tempDir(); + const { tracker, calls } = fakeTracker({ setStateFails: true }); + const t3 = fakeSessions(dataDir, {}); + const error = yield* Effect.flip(startIn(dataDir, tracker, t3.sessions)); + assert.strictEqual(error._tag, "GitHubFailed"); + assert.deepStrictEqual(entriesOf(dataDir), startedEvents().slice(0, 6)); + assert.strictEqual(calls.length, 5); + assert.strictEqual( + t3.calls.some((c) => c.startsWith("archive")), + false, + ); + }), + ); + + it.effect("an unassign that fails in the stuck path is a runner error", () => + Effect.gen(function* () { + const dataDir = tempDir(); + const { tracker } = fakeTracker({ unassignFails: true }); + const t3 = fakeSessions(dataDir, { project: null }); + const error = yield* Effect.flip(startIn(dataDir, tracker, t3.sessions)); + assert.strictEqual(error._tag, "GitHubFailed"); + // The comment and the label went out and are on record; the summary line is not written. + assert.deepStrictEqual(stepsOf(dataDir), [ + "step_changed", + "claimed", + "state_changed", + "stuck", + ]); + }), + ); +}); + +describe("start: a launch that an earlier pass may have sent", () => { + it.effect("writes the launch-sending step once and sends under the same command id", () => + Effect.gen(function* () { + const dataDir = tempDir(); + seedRun(dataDir, startedLines().slice(0, 4)); + const { tracker } = fakeTracker({ issues: [detail("triage", [USER])] }); + const t3 = fakeSessions(dataDir, {}); + yield* startIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(entriesOf(dataDir), startedEvents()); + assert.deepStrictEqual(t3.calls, ["findProject /work/softfac", `launch ${LAUNCH_ID}`]); + }), + ); + + it.effect( + "a launch that cannot be sent after an earlier one may have been says a session may exist", + () => + Effect.gen(function* () { + const dataDir = tempDir(); + seedRun(dataDir, startedLines().slice(0, 4)); + const { tracker, calls } = fakeTracker({ issues: [detail("triage", [USER])] }); + const t3 = fakeSessions(dataDir, { project: null }); + const outcome = yield* startIn(dataDir, tracker, t3.sessions); + assert.strictEqual(outcome.kind, "stuck"); + const reason = outcome.kind === "stuck" ? outcome.reason : ""; + assert.ok(reason.includes("may have started a session"), reason); + assert.ok(reason.includes(LAUNCH_ID), reason); + assert.strictEqual(reason, sessionMayExistNote(NO_PROJECT, LAUNCH_ID)); + assert.strictEqual(calls[1], `comment 7 ${stuckComment(LAUNCH_FAILED)}`); + }), + ); + + it.effect("a first launch that never went out has no such note", () => + Effect.gen(function* () { + const dataDir = tempDir(); + const { tracker } = fakeTracker({}); + const t3 = fakeSessions(dataDir, { project: null }); + const outcome = yield* startIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(outcome, { kind: "stuck", run: run7, reason: NO_PROJECT }); + }), + ); +}); + +describe("start: a resume", () => { + const textCheckCall = `textCheck 7 ${LABELED} tomkaczocha`; + const triage = () => fakeTracker({ issues: [detail("triage", [USER])] }); + + it.effect("with no session launches and then looks at the text once", () => + Effect.gen(function* () { + const dataDir = tempDir(); + seedRun(dataDir, startedLines().slice(0, 3)); + const { tracker, calls } = triage(); + const t3 = fakeSessions(dataDir, {}); + const outcome = yield* startIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(outcome, { kind: "started", run: run7, threadId: "thread-1" }); + assert.deepStrictEqual(t3.calls, ["findProject /work/softfac", `launch ${LAUNCH_ID}`]); + assert.deepStrictEqual(calls, [ + "issue 7", + textCheckCall, + "setState 7 building", + `status 7 rawphp ${BUILDING}`, + ]); + assert.deepStrictEqual(entriesOf(dataDir), startedEvents()); + }), + ); + + it.effect("with a session that was not looked at again does not launch again", () => + Effect.gen(function* () { + const dataDir = tempDir(); + seedRun(dataDir, startedLines().slice(0, 5)); + const { tracker, calls } = triage(); + const t3 = fakeSessions(dataDir, {}); + yield* startIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(calls, [ + "issue 7", + textCheckCall, + "setState 7 building", + `status 7 rawphp ${BUILDING}`, + ]); + assert.deepStrictEqual(t3.calls, []); + assert.deepStrictEqual(entriesOf(dataDir), startedEvents()); + }), + ); + + it.effect("with a second look that passed does not repeat it", () => + Effect.gen(function* () { + const dataDir = tempDir(); + seedRun(dataDir, startedLines().slice(0, 6)); + const { tracker, calls } = triage(); + const t3 = fakeSessions(dataDir, {}); + yield* startIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(calls, [ + "issue 7", + "setState 7 building", + `status 7 rawphp ${BUILDING}`, + ]); + assert.deepStrictEqual(t3.calls, []); + }), + ); + + it.effect("with a session and no recorded approval archives it and goes stuck", () => + Effect.gen(function* () { + const dataDir = tempDir(); + const lines = startedLines(); + seedRun(dataDir, [lines[1]!, lines[2]!, lines[4]!]); + const { tracker, calls } = triage(); + const t3 = fakeSessions(dataDir, {}); + const outcome = yield* startIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(outcome, { kind: "stuck", run: run7, reason: CANNOT_CONFIRM }); + assert.deepStrictEqual(calls, ["issue 7", ...stuckCalls(CANNOT_CONFIRM, "thread-1")]); + assert.deepStrictEqual(t3.calls, [ + "archive thread-1", + "thread thread-1", + `removeWorktree ${WORKTREE}`, + ]); + }), + ); +}); + +describe("start: stopping a session that goes stuck", () => { + it.effect("a worktree that cannot be removed is written down and the run still goes stuck", () => + Effect.gen(function* () { + const dataDir = tempDir(); + const { tracker, calls } = fakeTracker({ textChecks: ["changed"] }); + const t3 = fakeSessions(dataDir, { + removeWorktree: new SessionsError({ + operation: "remove-worktree", + cause: new Error("disk busy"), + }), + }); + const outcome = yield* startIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(outcome, { kind: "stuck", run: run7, reason: TEXT_CHANGED }); + assert.deepStrictEqual(calls.slice(4), stuckCalls(TEXT_CHANGED, "thread-1")); + assert.deepStrictEqual(entriesOf(dataDir), [ + ...startedEvents().slice(0, 5), + { + event: "worktree_remove_failed", + data: { path: WORKTREE, error: "disk busy" }, + cause: cause("worktree-remove-failed"), + }, + ...stuckTail(TEXT_CHANGED, "text-changed-before-start"), + ]); + }), + ); + + it.effect("a thread T3 no longer knows is stopped at the worktree the ledger records", () => + Effect.gen(function* () { + const dataDir = tempDir(); + seedRun(dataDir, startedLines().slice(0, 5)); + const { tracker } = fakeTracker({ + issues: [detail("triage", [USER])], + textChecks: ["changed"], + }); + const t3 = fakeSessions(dataDir, { + thread: new ThreadNotFoundError({ threadId: "thread-1" }), + }); + const outcome = yield* resume(ctxFor(dataDir, tracker, t3.sessions), 7); + assert.deepStrictEqual(outcome, { kind: "stuck", run: run7, reason: TEXT_CHANGED }); + assert.deepStrictEqual(t3.calls, [ + "archive thread-1", + "thread thread-1", + `removeWorktree ${WORKTREE}`, + ]); + assert.deepStrictEqual(entriesOf(dataDir).slice(5, 6), [ + { event: "worktree_removed", data: { path: WORKTREE }, cause: cause("worktree-removed") }, + ]); + }), + ); +}); + +describe("resume", () => { + const resumeIn = (dataDir: string, tracker: Tracker, sessions?: Sessions) => + resume(ctxFor(dataDir, tracker, sessions ?? fakeSessions(dataDir, {}).sessions), 7); + + it.effect("goes on with a triage run of ours as start does", () => + Effect.gen(function* () { + const dataDir = tempDir(); + seedRun(dataDir, startedLines().slice(0, 3)); + const { tracker } = fakeTracker({ issues: [detail("triage", [USER])] }); + const outcome = yield* resumeIn(dataDir, tracker); + assert.deepStrictEqual(outcome, { kind: "started", run: run7, threadId: "thread-1" }); + assert.deepStrictEqual(entriesOf(dataDir), startedEvents()); + }), + ); + + it.effect("never claims an issue put back to ready: that is the claim step's", () => + Effect.gen(function* () { + const dataDir = tempDir(); + seedRun(dataDir, startedLines().slice(0, 3)); + const { tracker, calls } = fakeTracker({ issues: [detail("ready")] }); + const t3 = fakeSessions(dataDir, {}); + const outcome = yield* resumeIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(outcome, { + kind: "skipped", + reason: "issue #7 is ready again", + moved: "ready", + }); + assert.deepStrictEqual(calls, ["issue 7"]); + assert.deepStrictEqual(t3.calls, []); + assert.deepStrictEqual(entriesOf(dataDir), startedEvents().slice(0, 3)); + assert.strictEqual(NodeFS.existsSync(runPath(dataDir, { ...run7, attempt: 2 })), false); + }), + ); + + it.effect("says nothing moved for a ready issue someone is assigned to", () => + Effect.gen(function* () { + const dataDir = tempDir(); + seedRun(dataDir, startedLines().slice(0, 3)); + const { tracker, calls } = fakeTracker({ issues: [detail("ready", ["bob"])] }); + const outcome = yield* resumeIn(dataDir, tracker); + // Not `moved: "ready"`: the claim step never takes an assigned issue, so the run stalls. + assert.deepStrictEqual(outcome, { + kind: "skipped", + reason: "issue #7 is already assigned to bob", + }); + assert.deepStrictEqual(calls, ["issue 7"]); + assert.deepStrictEqual(entriesOf(dataDir), startedEvents().slice(0, 3)); + }), + ); + + it.effect("says a closed issue was closed, and touches nothing", () => + Effect.gen(function* () { + const dataDir = tempDir(); + seedRun(dataDir, startedLines().slice(0, 3)); + const { tracker, calls } = fakeTracker({ issues: [detail("triage", [USER], true)] }); + const outcome = yield* resumeIn(dataDir, tracker); + assert.deepStrictEqual(outcome, { + kind: "skipped", + reason: "issue #7 is closed", + moved: "closed", + }); + assert.deepStrictEqual(calls, ["issue 7"]); + assert.deepStrictEqual(entriesOf(dataDir), startedEvents().slice(0, 3)); + }), + ); + + it.effect("finishes a stuck path cut short after its label moved", () => + Effect.gen(function* () { + const dataDir = tempDir(); + const first = fakeTracker({ unassignFails: true }); + const t3 = fakeSessions(dataDir, { project: null }); + yield* Effect.flip(startIn(dataDir, first.tracker, t3.sessions)); + // The label is `factory:stuck` and the factory is still assigned. + const { tracker, calls } = fakeTracker({ issues: [detail("stuck", [USER])] }); + const outcome = yield* resumeIn(dataDir, tracker, t3.sessions); + assert.deepStrictEqual(outcome, { kind: "stuck", run: run7, reason: NO_PROJECT }); + // The comment went out the first time and is not repeated. + assert.deepStrictEqual(calls, ["issue 7", ...stuckCalls(LAUNCH_FAILED, null).slice(1)]); + assert.deepStrictEqual(entriesOf(dataDir), [ + ...startedEvents().slice(0, 3), + ...stuckTail(NO_PROJECT, "launch-failed").slice(0, 1), + ...stuckTail(NO_PROJECT, "stuck-resumed").slice(1), + ]); + }), + ); + + it.effect("leaves alone a stuck issue whose run never went stuck: a person stopped it", () => + Effect.gen(function* () { + for (const lines of [startedLines().slice(0, 3), startedLines()]) { + const dataDir = tempDir(); + seedRun(dataDir, lines); + const before = entriesOf(dataDir); + const { tracker, calls } = fakeTracker({ issues: [detail("stuck", [USER])] }); + const t3 = fakeSessions(dataDir, {}); + const outcome = yield* resumeIn(dataDir, tracker, t3.sessions); + assert.strictEqual(outcome.kind, "skipped"); + assert.deepStrictEqual(calls, ["issue 7"]); + assert.deepStrictEqual(t3.calls, []); + assert.deepStrictEqual(entriesOf(dataDir), before); + } + }), + ); + + it.effect("leaves a stuck issue alone when its run is finished, or it has none", () => + Effect.gen(function* () { + const none = tempDir(); + const noRun = fakeTracker({ issues: [detail("stuck", [USER])] }); + assert.strictEqual((yield* resumeIn(none, noRun.tracker)).kind, "skipped"); + assert.deepStrictEqual(noRun.calls, ["issue 7"]); + assert.strictEqual(NodeFS.existsSync(runPath(none, run7)), false); + + const finished = tempDir(); + olderRun(finished, 1, "stuck", true); + const done = fakeTracker({ issues: [detail("stuck", [USER])] }); + assert.strictEqual((yield* resumeIn(finished, done.tracker)).kind, "skipped"); + assert.deepStrictEqual(done.calls, ["issue 7"]); + assert.strictEqual(NodeFS.existsSync(runPath(finished, { ...run7, attempt: 2 })), false); + }), + ); +}); diff --git a/apps/server/src/factory/runner/start.ts b/apps/server/src/factory/runner/start.ts index 25bfd7e5a257..a2596150fe35 100644 --- a/apps/server/src/factory/runner/start.ts +++ b/apps/server/src/factory/runner/start.ts @@ -1,7 +1,8 @@ // @effect-diagnostics nodeBuiltinImport:off - a gate lock is a file, checked the way gate.ts makes it. /** * The claim: takes a `ready` issue to `factory:triage` with a run file. Ported from softfac's - * `runner/start.rs` up to the launch, which issue #16 adds. + * `runner/start.rs`, through the launch of the `/factory` session and the move to + * `factory:building`. * * Before anything is claimed, a fresh start asks GitHub who applied the `factory:ready` label and * who edited the issue text since, and builds only when an instructor applied the label and @@ -14,9 +15,15 @@ * A new attempt first finishes every older unfinished run of the issue in the ledger, so a run * a person has moved on from stops being polled. * - * Calling `start` again resumes. An issue in `triage` whose only assignee is the factory's own - * user is an interrupted start: a run with no record of what approved it cannot be confirmed, so - * it fails closed and goes `stuck` (no session was launched, as the record is written first). + * The session is launched under a command id derived from the run, after the ledger says a + * launch is about to be sent, so a launch that fails on our side is found again by sending the + * same id. The issue text is checked again once the session exists: a change stops the run and + * archives the session. + * + * Calling `start` again resumes; `resume` does only that. An issue in `triage` whose only + * assignee is the factory's own user is an interrupted start, and every step goes on from what + * the ledger records. A run with no record of what approved it cannot be confirmed, so it fails + * closed and goes `stuck` (no session was launched, as the record is written first). */ import * as NodeFS from "node:fs"; @@ -25,7 +32,7 @@ import * as Schema from "effect/Schema"; import type { ProjectConfig } from "../config.ts"; import { refusalReason, unapprovedComment, type Refusal } from "../domain/approval.ts"; -import { InvalidTransition, stateLabel, transition, type State } from "../domain/lifecycle.ts"; +import { InvalidTransition, stateLabel, transition } from "../domain/lifecycle.ts"; import { gateLockFile } from "../gate.ts"; import { ledgerCall as ledger, type LedgerError } from "../ledger/errors.ts"; import { makeCause, type Cause, type ClaimRepair, type Outcome } from "../ledger/events.ts"; @@ -35,9 +42,18 @@ import { readRun } from "../ledger/read.ts"; import { newRunId, nextAttempt, runIdToString, type RunId } from "../ledger/runId.ts"; import { formatTs } from "../ledger/time.ts"; import { isUtcSecond } from "../tracker/approvalHistory.ts"; -import type { BoundGitHub, GitHubError } from "../tracker/github.ts"; -import { CANNOT_CONFIRM, CUT_SHORT_COMMENT, statusBody, stuckComment } from "./messages.ts"; -import { readProgress, type Seen } from "./progress.ts"; +import type { BoundGitHub, GitHubError, IssueDetail } from "../tracker/github.ts"; +import type { FactorySessions, Model, SessionsError } from "../t3/sessions.ts"; +import { + CANNOT_CONFIRM, + CUT_SHORT_COMMENT, + LAUNCH_FAILED, + TEXT_CHANGED, + launchMessage, + sessionMayExistNote, +} from "./messages.ts"; +import { APPROVAL_RECHECKED, LAUNCH_SENDING, readProgress, type Seen } from "./progress.ts"; +import { describe, finishBuilding, goStuck, type Progress } from "./run.ts"; /** * What the claim asks of the tracker: the subset of `makeGitHub` it calls, with the process @@ -55,6 +71,13 @@ export type Tracker = Pick< | "postComment" | "upsertStatusComment" | "newestFactoryComment" + | "textCheck" +>; + +/** The subset of `FactorySessions` the claim calls. */ +export type Sessions = Pick< + FactorySessions["Service"], + "findProject" | "launch" | "thread" | "archive" | "removeWorktree" >; export interface StartContext { @@ -67,16 +90,25 @@ export interface StartContext { readonly project: ProjectConfig; readonly dataDir: string; readonly tracker: Tracker; + /** The local clone the T3 project is rooted at: `WatchedRepo.path`. */ + readonly repoPath: string; + /** The `GH_CONFIG_DIR` the launch message names. */ + readonly ghConfigDir: string; + readonly model: Model; + readonly sessions: Sessions; /** Unix epoch milliseconds. */ readonly clock: () => number; } /** What `start` did. */ export type StartOutcome = - /** The issue is claimed and its run file records it. Issue #16 launches the session from here. */ - | { readonly kind: "claimed"; readonly run: RunId } - /** The issue was not ours to take. Nothing was written anywhere. */ - | { readonly kind: "skipped"; readonly reason: string } + /** The issue is claimed, its session is launched and the issue is `factory:building`. */ + | { readonly kind: "started"; readonly run: RunId; readonly threadId: string } + /** + * The issue was not ours to take. Nothing was written anywhere. `moved` says a person closed + * the issue, or put it back to `factory:ready` with nobody assigned, which the claim step takes. + */ + | { readonly kind: "skipped"; readonly reason: string; readonly moved?: "closed" | "ready" } /** The issue was `ready` but not approved, so it went back to `factory:proposed` with one fixed comment, and no run was made. */ | { readonly kind: "refused"; readonly reason: string } /** The issue went to `stuck`. */ @@ -91,16 +123,23 @@ export class GateRunningError extends Schema.TaggedError()("Ga } } -export type StartError = GitHubError | LedgerError | InvalidTransition | GateRunningError; +export type StartError = + | GitHubError + | LedgerError + | InvalidTransition + | GateRunningError + | SessionsError; -const mention = (issue: number) => `#${issue}`; +/** An issue as the pass's lines and the skip reasons name it. */ +export const mention = (issue: number) => `#${issue}`; const skipped = (issue: number, assignees: ReadonlyArray): StartOutcome => ({ kind: "skipped", reason: `issue ${mention(issue)} is already assigned to ${assignees.join(", ")}`, }); -const isOurs = (assignees: ReadonlyArray, user: string) => +/** Whether `user` is the only assignee. Not `asciiLower`: a Linear user can be a display name. */ +export const isOurs = (assignees: ReadonlyArray, user: string) => assignees.length === 1 && assignees[0]!.toLowerCase() === user.toLowerCase(); export const start = ( @@ -111,25 +150,90 @@ export const start = ( const detail = yield* ctx.tracker.issue(issue); // A closed issue is never claimed, whatever its label says. Issue #20 ports what softfac // does with its unfinished runs; until then they are left alone. - if (detail.closed) { - return { kind: "skipped", reason: `issue ${mention(issue)} is closed` }; + if (detail.closed) return closed(issue); + if (detail.state === "ready" && detail.assignees.length === 0) { + return yield* startFresh(ctx, issue, detail); } - const ours = isOurs(detail.assignees, ctx.trackerUser); + if (detail.state === "ready") return skipped(issue, detail.assignees); + return yield* goOn(ctx, issue, detail); + }); + +/** + * `start` for a claim already made, which never makes one: what the pass calls for a run it + * knows of and for a `triage` issue of ours with no run. An issue someone put back to + * `factory:ready` is left to the claim step, which checks the pause, the limit and the blockers + * first, and which finishes the issue's older runs when it claims it. + */ +export const resume = ( + ctx: StartContext, + issue: number, +): Effect.Effect => + Effect.gen(function* () { + const detail = yield* ctx.tracker.issue(issue); + if (detail.closed) return closed(issue); if (detail.state === "ready" && detail.assignees.length === 0) { - return yield* startFresh(ctx, issue); + return { kind: "skipped", reason: `issue ${mention(issue)} is ready again`, moved: "ready" }; } if (detail.state === "ready") return skipped(issue, detail.assignees); - if (detail.state === "triage" && ours) return yield* resumeStart(ctx, issue); - return { kind: "skipped", reason: `issue ${mention(issue)} is not labelled factory:ready` }; + return yield* goOn(ctx, issue, detail); }); -const startFresh = (ctx: StartContext, issue: number): Effect.Effect => +const closed = (issue: number): StartOutcome => ({ + kind: "skipped", + reason: `issue ${mention(issue)} is closed`, + moved: "closed", +}); + +/** + * Goes on from the ledger with an issue that is not `ready`: a `triage` issue of ours, or a + * `stuck` issue whose newest run gave up and was cut short before its summary line (the label + * moves before the release and the status comment, so a failure in between leaves the run + * unfinished in `triage`), which finishes the stuck path. + */ +const goOn = ( + ctx: StartContext, + issue: number, + detail: IssueDetail, +): Effect.Effect => + Effect.gen(function* () { + if (detail.state === "triage" && isOurs(detail.assignees, ctx.trackerUser)) { + return yield* resumeStart(ctx, issue, detail); + } + if (detail.state === "stuck" && (yield* stuckCutShort(ctx, issue))) { + return yield* resumeStart(ctx, issue, detail); + } + return { + kind: "skipped", + reason: `issue ${mention(issue)} is not factory:triage assigned to ${ctx.trackerUser} alone`, + }; + }); + +/** Whether the issue's newest run is unfinished, still in `triage`, and its ledger says it went stuck. */ +const stuckCutShort = (ctx: StartContext, issue: number): Effect.Effect => + Effect.gen(function* () { + const next = nextAttempt(ctx.dataDir, ctx.repo, String(issue)); + if (next === 1) return false; + const newest = newRunId(ctx.repo, String(issue), next - 1); + const seen = readProgress((yield* ledger(() => readRun(ctx.dataDir, newest))).entries); + return ( + !seen.finished && seen.stuck !== null && (seen.state === null || seen.state === "triage") + ); + }); + +const startFresh = ( + ctx: StartContext, + issue: number, + detail: IssueDetail, +): Effect.Effect => Effect.gen(function* () { // Before the claim: nothing is written for an issue that is not approved except the // explanation. const approval = yield* ctx.tracker.approval(issue, ctx.project.instructors); if (approval.kind === "refused") return yield* refuseStart(ctx, issue, approval.refusal); const { labeledAt, labeledBy } = approval; + // Also before the claim: with T3 out of reach nothing is claimed, so no issue is left in + // `triage` for a launch that cannot be sent. + const project = yield* ctx.sessions.findProject(ctx.repoPath); const run = newRunId( ctx.repo, String(issue), @@ -152,7 +256,7 @@ const startFresh = (ctx: StartContext, issue: number): Effect.Effect { - try { +/** + * Writes a `claim_repaired` line to the machine log. A log that cannot be written is not worth + * failing the claim for, so it is logged as a warning that names the issue. + */ +const recordClaimRepair = ( + ctx: StartContext, + issue: number, + outcome: ClaimRepair, +): Effect.Effect => + ledger(() => { const log = MachineLog.open(ctx.dataDir, { clock: ctx.clock }); try { log.append({ @@ -223,10 +334,14 @@ const recordClaimRepair = (ctx: StartContext, issue: number, outcome: ClaimRepai } finally { log.close(); } - } catch { - // Not worth failing the claim for. - } -}; + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning( + `claim_repaired (${outcome}) for ${ctx.repo}${mention(issue)} could not be written to the machine log`, + cause, + ), + ), + ); /** How long a refusal comment that has no label time to be compared with counts as recent. */ const RECENT_MS = 60 * 60 * 1000; @@ -286,23 +401,25 @@ const refuseStart = ( return { kind: "refused", reason: `issue ${mention(issue)} was not approved: ${reason}` }; }); -const resumeStart = (ctx: StartContext, issue: number): Effect.Effect => +const resumeStart = ( + ctx: StartContext, + issue: number, + detail: IssueDetail, +): Effect.Effect => Effect.gen(function* () { const next = nextAttempt(ctx.dataDir, ctx.repo, String(issue)); if (next > 1) { const newest = newRunId(ctx.repo, String(issue), next - 1); const seen = readProgress((yield* ledger(() => readRun(ctx.dataDir, newest))).entries); - // A run already in `stuck` is over; it is finished below with the others. A run cut short - // before it recorded an approval or a session never launched one, so `begin` goes on from - // it to `stuck`: a stuck path that failed half way is finished in its own run. Any other - // unfinished run is left to the step that owns its state: launching a claimed run is - // issue #16's. + // A run already in `stuck` is over; it is finished below with the others. A run still in + // `triage` (or not yet in any state) was cut short, so `begin` goes on from what its + // ledger records, whatever it was approved on. Any other unfinished run is left to the + // step that owns its state. if (!seen.finished && seen.state !== "stuck") { - const unconfirmed = seen.approvedAt === null && seen.threadId === null; - if (unconfirmed && (seen.state === null || seen.state === "triage")) { + if (seen.state === null || seen.state === "triage") { const out = yield* ledger(() => Ledger.open(ctx.dataDir, newest, { clock: ctx.clock })); try { - return yield* begin(ctx, issue, newest, out); + return yield* begin(ctx, issue, detail, newest, out, undefined); } finally { out.close(); } @@ -317,7 +434,7 @@ const resumeStart = (ctx: StartContext, issue: number): Effect.Effect Ledger.create(ctx.dataDir, run, { clock: ctx.clock })); try { - return yield* begin(ctx, issue, run, out); + return yield* begin(ctx, issue, detail, run, out, undefined); } finally { out.close(); } @@ -325,15 +442,19 @@ const resumeStart = (ctx: StartContext, issue: number): Effect.Effect => Effect.gen(function* () { const seen = readProgress((yield* ledger(() => readRun(ctx.dataDir, run))).entries); @@ -352,75 +473,146 @@ const begin = ( out.append({ event: "state_changed", data: { from: "ready", to: claimed.next } }, claim), ); } - const at: Progress = { state: seen.state ?? claimed.next, threadId: seen.threadId, seen }; + const at: Progress = { + step: "start", + state: seen.state ?? claimed.next, + threadId: seen.threadId, + worktree: seen.worktree, + worktreeDone: seen.worktreeDone, + seen, + }; + // A decision to give up that is already in the ledger stands. + if (seen.stuck !== null) { + yield* goStuck(ctx, issue, out, at, "stuck-resumed", { reason: seen.stuck }); + return { kind: "stuck", run, reason: seen.stuck }; + } + // The record is written before any launch: with no session either, nothing was sent, so + // nothing is launched here (a session on unconfirmed text, only to archive it). if (seen.approvedAt === null && seen.threadId === null) { - yield* goStuck(ctx, issue, out, at, "approval-unconfirmed", CANNOT_CONFIRM); + yield* goStuck(ctx, issue, out, at, "approval-unconfirmed", { reason: CANNOT_CONFIRM }); + return { kind: "stuck", run, reason: CANNOT_CONFIRM }; + } + // A launch that is not in the ledger may still have started a thread: the call can fail on + // our side after T3 acted. Launching again under the same command id either starts the + // session or returns the thread T3 already started. The text is looked at after the launch + // and never before it: skipping the launch for a changed text could leave such a thread + // running with nobody to stop it. + let threadId = seen.threadId; + if (threadId === null) { + const launched = yield* launchSession(ctx, issue, detail, run, out, seen, project); + if (!launched.ok) { + yield* goStuck(ctx, issue, out, at, "launch-failed", { + reason: LAUNCH_FAILED, + detail: launched.detail, + }); + return { kind: "stuck", run, reason: launched.detail }; + } + threadId = launched.threadId; + // What the ledger now records, for a stop that has to remove the worktree while T3 may not + // report its path yet. + at.worktree = launched.worktree; + } + at.threadId = threadId; + if (seen.approvedAt === null) { + yield* goStuck(ctx, issue, out, at, "approval-unconfirmed", { reason: CANNOT_CONFIRM }); return { kind: "stuck", run, reason: CANNOT_CONFIRM }; } - return { kind: "claimed", run }; + // The session reads the issue live a few seconds after the approval, and the claim has told + // anyone watching the labels that it is starting. Check the text again now that the session + // exists: a change by someone who is not an instructor stops the run. A failed read is an + // error, and the next call checks again; the step says the check passed. + if (!seen.rechecked) { + const check = yield* ctx.tracker.textCheck(issue, seen.approvedAt, ctx.project.instructors); + // A history that cannot be judged counts as changed. + if (check !== "unchanged") { + yield* goStuck(ctx, issue, out, at, "text-changed-before-start", { reason: TEXT_CHANGED }); + return { kind: "stuck", run, reason: TEXT_CHANGED }; + } + yield* ledger(() => + out.append( + { event: "step_changed", data: { step: APPROVAL_RECHECKED } }, + makeCause("start", "approval-rechecked"), + ), + ); + } + if (at.state !== "building") yield* finishBuilding(ctx, issue, out, at, threadId); + return { kind: "started", run, threadId }; }); -/** Where the issue stands while the stuck actions are applied. */ -interface Progress { - readonly state: State; - readonly threadId: string | null; - readonly seen: Seen; -} +type Launch = + | { readonly ok: true; readonly threadId: string; readonly worktree: string | null } + | { readonly ok: false; readonly detail: string }; /** - * The stuck path: the actions `lifecycle.transition` asks for on a failure, in order, then the - * `state_changed` line and the summary line. A comment is skipped when the ledger already - * records it; the label, the release and the status comment are repeated safely. `code` is the - * reason of every line it writes. + * Sends the launch: `launch-sending` is written once per run, before the first call, so a + * launch that fails on our side is known to have been sent. The command id comes from the run, + * so sending it again returns the thread T3 already made. In-process, every failure of the call + * is T3's answer after the request went out: it is final, and a session may exist. A launch that + * was never sent (no project) says so only when an earlier pass did send one. */ -const goStuck = ( +const launchSession = ( ctx: StartContext, issue: number, + detail: IssueDetail, + run: RunId, out: Ledger, - at: Progress, - code: string, - reason: string, -): Effect.Effect => + seen: Seen, + project: string | null | undefined, +): Effect.Effect => Effect.gen(function* () { - const cause = makeCause("start", code); - if (at.state === "stuck") { - yield* ledger(() => out.finish("stuck", null, null, cause)); - return; + const commandId = `softfac:${runIdToString(run)}:launch`; + const projectId = + project === undefined ? yield* ctx.sessions.findProject(ctx.repoPath) : project; + if (projectId === null) { + const missing = "no T3 project for this repo on this machine"; + return { + ok: false, + detail: seen.launchSent ? sessionMayExistNote(missing, commandId) : missing, + }; } - const failed = yield* Effect.fromResult(transition(at.state, { kind: "failed", reason })); - let movedTo: State | null = null; - let stuckPosted = at.seen.stuck !== null; - for (const action of failed.actions) { - switch (action.kind) { - case "post-stuck-comment": - if (!stuckPosted) { - yield* ctx.tracker.postComment(issue, stuckComment(reason)); - yield* ledger(() => out.append({ event: "stuck", data: { reason } }, cause)); - stuckPosted = true; - } - break; - case "set-label": - yield* ctx.tracker.setState(issue, action.state); - movedTo = action.state; - break; - case "unassign": - yield* ctx.tracker.unassign(issue, ctx.trackerUser); - break; - case "update-status-comment": - yield* ctx.tracker.upsertStatusComment( - issue, - ctx.trackerUser, - statusBody(movedTo ?? at.state, ctx.machine, at.threadId, null), - ); - break; - default: - return yield* new InvalidTransition({ state: at.state, event: "failed" }); - } + if (!seen.launchSent) { + yield* ledger(() => + out.append( + { event: "step_changed", data: { step: LAUNCH_SENDING } }, + makeCause("start", "launch-sending"), + ), + ); + } + const launched = yield* Effect.result( + ctx.sessions.launch({ + commandId, + projectId, + title: `${ctx.repo}#${issue} ${detail.title}`, + baseRef: ctx.project.base, + startFromOrigin: false, + message: launchMessage(ctx.repo, issue, ctx.trackerUser, ctx.ghConfigDir), + model: ctx.model, + }), + ); + if (launched._tag === "Failure") { + return { + ok: false, + detail: sessionMayExistNote(describe(launched.failure.cause), commandId), + }; } + const { threadId, worktreePath, branch } = launched.success; yield* ledger(() => - out.append({ event: "state_changed", data: { from: at.state, to: failed.next } }, cause), + out.append( + { + event: "session_launched", + data: { + thread_id: threadId, + worktree: worktreePath, + branch, + model_instance: ctx.model.instanceId, + model: ctx.model.model, + resumed_from: null, + }, + }, + makeCause("start", "session-launched"), + ), ); - yield* ledger(() => out.finish("stuck", null, null, cause)); + return { ok: true, threadId, worktree: worktreePath }; }); /** diff --git a/apps/server/src/factory/t3/sessions.ts b/apps/server/src/factory/t3/sessions.ts index 309e8366d400..e4ada0233130 100644 --- a/apps/server/src/factory/t3/sessions.ts +++ b/apps/server/src/factory/t3/sessions.ts @@ -492,3 +492,11 @@ export const make = Effect.gen(function* () { }); export const layer = Layer.effect(FactorySessions, make); + +/** + * `layer` with the projection store it reads provided: `server.ts` has the + * project store, the launch and management services and git in scope where + * the factory is merged in, but not the projection store, which upstream + * provides to each worker that reads it. + */ +export const layerLive = layer.pipe(Layer.provide(ProjectionStore.layer));