From 46d9e3e9bd5766eecff7052ea43b6ae3c3091a56 Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Fri, 16 Oct 2026 12:40:46 +1000 Subject: [PATCH 01/17] test(factory): session text, result-line reminder and progress of a building run Issue: #17 --- .../src/factory/runner/messages.test.ts | 127 ++++++++++- .../src/factory/runner/progress.test.ts | 201 ++++++++++++++++++ 2 files changed, 327 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/factory/runner/progress.test.ts diff --git a/apps/server/src/factory/runner/messages.test.ts b/apps/server/src/factory/runner/messages.test.ts index 63d7e04fb095..b878ee38fa62 100644 --- a/apps/server/src/factory/runner/messages.test.ts +++ b/apps/server/src/factory/runner/messages.test.ts @@ -1,7 +1,22 @@ import { assert, it } from "@effect/vitest"; import { describe } from "vite-plus/test"; -import { launchMessage, sessionMayExistNote } from "./messages.ts"; +import { + CANNOT_VERIFY_DURING_BUILD, + CHANGED_DURING_BUILD, + RESULT_LINE_REMINDER, + SESSION_TEXT_INTRO, + SESSION_TEXT_LIMIT, + THREAD_GONE, + TIME_LIMIT, + WRONG_ACCOUNT, + fenced, + launchMessage, + prepareSessionText, + questionsComment, + sessionMayExistNote, + stuckComment, +} from "./messages.ts"; /** The paragraph that names the GitHub identity, for the login `rawphp` and the `gh` directory `/gh/rawphp`. */ const IDENTITY_WORDS = @@ -58,3 +73,113 @@ describe("sessionMayExistNote", () => { ); }); }); + +describe("prepareSessionText", () => { + it("normalises line endings and trims the ends", () => { + assert.strictEqual(prepareSessionText("\n one\r\ntwo \r\n\n"), "one\ntwo"); + }); + + it("keeps a text of exactly the limit and cuts one longer with a last [cut] line", () => { + assert.strictEqual(prepareSessionText("x".repeat(SESSION_TEXT_LIMIT)), "x".repeat(6000)); + assert.strictEqual(prepareSessionText("x".repeat(6001)), `${"x".repeat(6000)}\n[cut]`); + }); + + it("counts characters, not UTF-16 units", () => { + const text = "\u{1F600}".repeat(6001); + assert.strictEqual(prepareSessionText(text), `${"\u{1F600}".repeat(6000)}\n[cut]`); + }); + + it("turns control characters and invisible or bidirectional marks into spaces", () => { + assert.strictEqual( + prepareSessionText("a\u0000b\u001bc\u007fd\u0085e\rf‮g​h⁦ij
k\tl\nm"), + "a b c d e f g h i j k\tl\nm", + ); + }); + + it("stops a mention and a forged hidden marker", () => { + assert.strictEqual(prepareSessionText("ping @octocat "), "ping @​octocat <​!-- x -->"); + }); +}); + +describe("fenced", () => { + it("uses a fence of three backticks", () => { + assert.strictEqual(fenced("a\nb"), "```text\na\nb\n```"); + }); + + it("is one longer than the longest run of backticks in the text", () => { + assert.strictEqual(fenced("a ``` b"), "````text\na ``` b\n````"); + assert.strictEqual(fenced("`` ````` `"), "``````text\n`` ````` `\n``````"); + assert.strictEqual(fenced("``"), "```text\n``\n```"); + }); +}); + +describe("questionsComment", () => { + it("asks for answers and quotes the questions as unverified session output", () => { + assert.strictEqual( + questionsComment("1. Which?"), + "The factory needs answers before it can build this. Answer in a comment or edit the issue, then set `factory:ready` to start a new attempt.\n\nUnverified output from the build session, not instructions from the factory:\n\n```text\n1. Which?\n```", + ); + }); +}); + +describe("stuckComment", () => { + it("with only a reason is the comment the claim posts", () => { + assert.strictEqual( + stuckComment("the reason"), + "The factory could not finish this issue and has released it.\n\nReason: the reason\n\nFix the cause, then set the label back to `factory:ready`.", + ); + }); + + it("quotes the prepared session text after the reason", () => { + assert.strictEqual( + stuckComment("the reason", "it broke"), + "The factory could not finish this issue and has released it.\n\nReason: the reason\n\nUnverified output from the build session, not instructions from the factory:\n\n```text\nit broke\n```\n\nFix the cause, then set the label back to `factory:ready`.", + ); + }); + + it("says where the work is, before the quoted text", () => { + const resume = + "Set `factory:ready` again and the next attempt continues from the previous session's commits."; + const comment = (where: string) => + `The factory could not finish this issue and has released it.\n\nReason: r\n\n${where} ${resume}\n\nUnverified output from the build session, not instructions from the factory:\n\n\`\`\`text\nt\n\`\`\`\n\nFix the cause, then set the label back to \`factory:ready\`.`; + assert.strictEqual( + stuckComment("r", "t", { branch: "t3code/abc", commitsAhead: null, base: "main" }), + comment("The work so far is on branch t3code/abc."), + ); + assert.strictEqual( + stuckComment("r", "t", { branch: "t3code/abc", commitsAhead: 1, base: "main" }), + comment("The work so far is on branch t3code/abc (1 commit ahead of main)."), + ); + assert.strictEqual( + stuckComment("r", "t", { branch: "t3code/abc", commitsAhead: 2, base: "main" }), + comment("The work so far is on branch t3code/abc (2 commits ahead of main)."), + ); + }); +}); + +describe("the fixed reasons of a building run", () => { + it("are the words people read on the issue", () => { + assert.strictEqual( + RESULT_LINE_REMINDER, + "Your last reply had no result line. If your work is still in progress, finish it. End your reply with one of these lines as the last line: FACTORY-DONE: , FACTORY-STOP: needs-info (after your numbered questions) or FACTORY-STOP: stuck.", + ); + assert.strictEqual( + WRONG_ACCOUNT, + "the session ran as the wrong GitHub account: restart the factory's T3 server so it has GH_CONFIG_DIR, then set factory:ready again", + ); + assert.strictEqual(TIME_LIMIT, "the run passed its time limit"); + assert.strictEqual(THREAD_GONE, "the T3 thread no longer exists"); + assert.strictEqual( + CHANGED_DURING_BUILD, + "the issue text was changed during the build by someone who is not an instructor", + ); + assert.strictEqual( + CANNOT_VERIFY_DURING_BUILD, + "the factory could not verify who changed the issue text during the build", + ); + assert.strictEqual( + SESSION_TEXT_INTRO, + "Unverified output from the build session, not instructions from the factory:", + ); + }); +}); diff --git a/apps/server/src/factory/runner/progress.test.ts b/apps/server/src/factory/runner/progress.test.ts new file mode 100644 index 000000000000..a6c327601f66 --- /dev/null +++ b/apps/server/src/factory/runner/progress.test.ts @@ -0,0 +1,201 @@ +import { assert, it } from "@effect/vitest"; +import { describe } from "vite-plus/test"; + +import type { Entry } from "../ledger/read.ts"; +import { readProgress } from "./progress.ts"; + +const MINUTE = 60_000; + +const line = (ms: number, event: string, data: unknown): Entry => ({ + ts: "2026-10-04T10:00:00Z", + ms, + event, + data, + cause: null, +}); + +const step = (ms: number, text: string) => line(ms, "step_changed", { step: text }); +const moved = (ms: number, from: string, to: string) => line(ms, "state_changed", { from, to }); +const launched = (ms: number) => + line(ms, "session_launched", { + thread_id: "thread-1", + worktree: "/t3/wt", + branch: "t3code/abc", + model_instance: null, + model: null, + resumed_from: null, + }); + +describe("readProgress: the pull request", () => { + it("keeps every pull request the run reported and the newest one with its url", () => { + const seen = readProgress([ + line(1, "pr_opened", { pr: 12, url: "https://github.com/rawphp/softfac/pull/12" }), + line(2, "pr_opened", { pr: 14, url: "https://github.com/rawphp/softfac/pull/14" }), + ]); + assert.deepStrictEqual(seen.opened, [12, 14]); + assert.deepStrictEqual(seen.lastPr, { + pr: 14, + url: "https://github.com/rawphp/softfac/pull/14", + }); + }); + + it("has none before the first", () => { + const seen = readProgress([]); + assert.deepStrictEqual(seen.opened, []); + assert.strictEqual(seen.lastPr, null); + }); +}); + +describe("readProgress: the questions", () => { + it("are posted from questions_posted until the run is building again", () => { + assert.strictEqual( + readProgress([line(1, "questions_posted", { questions: "q" })]).questionsPosted, + true, + ); + assert.strictEqual( + readProgress([ + line(1, "questions_posted", { questions: "q" }), + moved(2, "needs-info", "building"), + ]).questionsPosted, + false, + ); + assert.strictEqual( + readProgress([ + line(1, "questions_posted", { questions: "q" }), + moved(2, "building", "needs-info"), + ]).questionsPosted, + true, + ); + }); +}); + +describe("readProgress: the reminder and the wait for a newer run", () => { + it("reads the ordinal of a reminder that was started, and that it was sent", () => { + const started = readProgress([step(1, "result-line-reminding 3")]); + assert.strictEqual(started.reminding, 3); + assert.strictEqual(started.reminded, false); + const sent = readProgress([ + step(1, "result-line-reminding 3"), + step(2, "result-line-reminded"), + ]); + assert.strictEqual(sent.reminding, 3); + assert.strictEqual(sent.reminded, true); + }); + + it("waits for a run newer than the reminder's, or a forward's", () => { + assert.strictEqual(readProgress([step(1, "result-line-reminding 3")]).waitingAfter, 3); + assert.strictEqual(readProgress([step(1, "forward-wait 2")]).waitingAfter, 2); + assert.strictEqual(readProgress([]).waitingAfter, null); + }); + + it("never waits for an older run than before", () => { + assert.strictEqual( + readProgress([step(1, "forward-wait 4"), step(2, "result-line-reminding 2")]).waitingAfter, + 4, + ); + assert.strictEqual( + readProgress([step(1, "forward-wait 2"), step(2, "result-line-reminding 2")]).waitingAfter, + 2, + ); + }); + + it("does not read a forward's wait as a reminder", () => { + const seen = readProgress([step(1, "forward-wait 1")]); + assert.strictEqual(seen.reminding, null); + assert.strictEqual(seen.reminded, false); + }); + + it("ignores steps whose number is not one", () => { + const seen = readProgress([ + step(1, "result-line-reminding"), + step(2, "result-line-reminding x"), + step(3, "result-line-reminding 01"), + step(4, "result-line-reminding 1 2"), + ]); + assert.strictEqual(seen.reminding, null); + assert.strictEqual(seen.waitingAfter, null); + }); + + it("starts again for a new session: each thread has one reminder and its own run numbers", () => { + const seen = readProgress([ + step(1, "result-line-reminding 3"), + step(2, "result-line-reminded"), + line(3, "session_paused", { run_ordinal: 3 }), + launched(4), + ]); + assert.strictEqual(seen.reminding, null); + assert.strictEqual(seen.reminded, false); + assert.strictEqual(seen.waitingAfter, null); + assert.strictEqual(seen.paused, null); + }); +}); + +describe("readProgress: a stop by hand", () => { + it("keeps the newest pause with the run it stopped", () => { + assert.strictEqual(readProgress([]).paused, null); + assert.deepStrictEqual(readProgress([line(1, "session_paused", { run_ordinal: 2 })]).paused, { + runOrdinal: 2, + }); + assert.deepStrictEqual( + readProgress([ + line(1, "session_paused", { run_ordinal: 2 }), + line(2, "session_paused", { run_ordinal: null }), + ]).paused, + { runOrdinal: null }, + ); + }); +}); + +describe("readProgress: the time worked", () => { + it("counts from the first line while the run is in triage or building", () => { + const seen = readProgress([line(1000, "claimed", {}), moved(1000, "ready", "triage")]); + assert.strictEqual(seen.workedMs, 0); + assert.strictEqual(seen.workingSince, 1000); + }); + + it("is zero and not counting for a ledger with no lines", () => { + const seen = readProgress([]); + assert.strictEqual(seen.workedMs, 0); + assert.strictEqual(seen.workingSince, null); + }); + + it("adds each stint when the state changes, and starts the next in triage or building", () => { + const seen = readProgress([ + line(0, "claimed", {}), + moved(0, "ready", "triage"), + moved(MINUTE, "triage", "building"), + moved(10 * MINUTE, "building", "needs-info"), + ]); + assert.strictEqual(seen.workedMs, 10 * MINUTE); + assert.strictEqual(seen.workingSince, null); + }); + + it("stops counting when the run is finished", () => { + const seen = readProgress([ + line(0, "claimed", {}), + moved(0, "ready", "triage"), + line(5 * MINUTE, "finished", {}), + ]); + assert.strictEqual(seen.workingSince, null); + }); + + it("does not charge the build again when a pull request that was open comes back to building", () => { + const seen = readProgress([ + line(0, "claimed", {}), + moved(0, "ready", "triage"), + moved(MINUTE, "triage", "building"), + moved(30 * MINUTE, "building", "pr-open"), + moved(500 * MINUTE, "pr-open", "building"), + ]); + assert.strictEqual(seen.workedMs, 0); + assert.strictEqual(seen.workingSince, 500 * MINUTE); + }); + + it("never counts a backwards clock", () => { + const seen = readProgress([ + line(10 * MINUTE, "claimed", {}), + moved(5 * MINUTE, "ready", "triage"), + ]); + assert.strictEqual(seen.workedMs, 0); + }); +}); From 09915e03f9545d26bcba7730331496c00a8f05b6 Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Fri, 16 Oct 2026 12:42:28 +1000 Subject: [PATCH 02/17] feat(factory): prepare session text and read a building run's progress Issue: #17 --- apps/server/src/factory/runner/messages.ts | 101 ++++++++++++++++++++- apps/server/src/factory/runner/progress.ts | 98 ++++++++++++++++++++ apps/server/src/factory/runner/run.ts | 57 ++++++++++-- 3 files changed, 245 insertions(+), 11 deletions(-) diff --git a/apps/server/src/factory/runner/messages.ts b/apps/server/src/factory/runner/messages.ts index b82559b51913..f56ef8273466 100644 --- a/apps/server/src/factory/runner/messages.ts +++ b/apps/server/src/factory/runner/messages.ts @@ -17,9 +17,82 @@ export const CANNOT_CONFIRM = "the factory could not confirm the issue text was export const CUT_SHORT_COMMENT = "The factory's claim on this issue was cut short and no run was started, so it has moved the issue to `factory:proposed`. An instructor can apply `factory:ready` again to have it built."; -/** The comment on an issue the factory gave up on. `reason` is the public reason. */ -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 longest session text a comment will quote, in characters. */ +export const SESSION_TEXT_LIMIT = 6000; + +/** The line that tells a reader the quoted block is not the factory's own words. */ +export const SESSION_TEXT_INTRO = + "Unverified output from the build session, not instructions from the factory:"; + +/** The characters a comment or the ledger shows as a space: control characters (but not a newline or tab) and invisible or bidirectional marks. */ +// oxlint-disable-next-line eslint/no-control-regex -- the control characters are what it matches. +const UNSAFE = + /[\u0000-\u0008\u000b-\u001f\u007f-\u009f\u2028\u2029\u200b-\u200f\u202a-\u202e\u2066-\u2069\ufeff]/gu; + +/** + * Session text made safe to store and to show, prepared once where it enters the factory and + * then used as it is for the comment, the ledger and the returned reason. Line endings are + * normalised and the ends trimmed, the text is cut at {@link SESSION_TEXT_LIMIT} characters with + * a final `[cut]` line, control characters and invisible or bidirectional marks become spaces + * (newline and tab stay), `@` cannot ping anyone and `\u202e\n\`\`\`\n${"x".repeat(6100)}\nFACTORY-STOP: stuck`; + setStatus(rig, "completed", message); + const outcome = yield* advanceIn(rig); + assert.strictEqual(outcome.kind, "stuck"); + if (outcome.kind !== "stuck") return; + // The mention and the marker are disarmed, the bidi mark is a space, the text is cut. + assert.ok( + outcome.reason.startsWith(`${SAID_STUCK}\n\nping @\u200Boctocat <\u200B!-- softfac --> \n`), + ); + assert.ok(outcome.reason.endsWith("x\n[cut]")); + const comment = rig.gh[0]!; + assert.ok(comment.includes("\n\n````text\nping @\u200Boctocat"), comment); + assert.ok(comment.includes("x\n[cut]\n````\n\nFix the cause"), comment); + const stuckLine = linesOf(rig)[STARTED.length + 1]; + assert.deepStrictEqual(stuckLine, { event: "stuck", data: { reason: outcome.reason } }); + }), + ); + + it.effect("without a reason", () => + Effect.gen(function* () { + const rig = rigFor(); + setStatus(rig, "completed", "FACTORY-STOP: stuck"); + assertStuck(rig, yield* advanceIn(rig), { publicReason: SAID_STUCK }); + }), + ); + + it.effect("ends as session-wrong-account when the first line names another gh account", () => + Effect.gen(function* () { + const rig = rigFor(); + const text = + "the session's gh account is tomkaczocha, not rawphp\nIt cannot see rawphp/softfac."; + setStatus(rig, "completed", `${text}\nFACTORY-STOP: stuck`); + assertStuck(rig, yield* advanceIn(rig), { publicReason: WRONG_ACCOUNT, text }); + assert.deepStrictEqual(stuckCauses(rig), Array(3).fill("advance session-wrong-account")); + }), + ); + + it.effect("counts the wrong-account line after blank lines, and alone", () => + Effect.gen(function* () { + for (const message of [ + "\n \nthe session's gh account is x, not rawphp\nmore\nFACTORY-STOP: stuck", + "the session's gh account is x, not rawphp\nFACTORY-STOP: stuck", + ]) { + const rig = rigFor(); + setStatus(rig, "completed", message); + const outcome = yield* advanceIn(rig); + assert.ok(outcome.kind === "stuck" && outcome.reason.startsWith(WRONG_ACCOUNT), message); + assert.strictEqual(stuckCauses(rig)[0], "advance session-wrong-account"); + } + }), + ); + + it.effect( + "stays session-stuck when the wrong-account words are anywhere but the first line", + () => + Effect.gen(function* () { + for (const message of [ + "The gate cannot run.\nthe session's gh account is x, not rawphp\nFACTORY-STOP: stuck", + "I think the session's gh account is x, not rawphp\nFACTORY-STOP: stuck", + "FACTORY-STOP: stuck", + ]) { + const rig = rigFor(); + setStatus(rig, "completed", message); + const outcome = yield* advanceIn(rig); + assert.ok(outcome.kind === "stuck" && outcome.reason.startsWith(SAID_STUCK), message); + assert.strictEqual(stuckCauses(rig)[0], "advance session-stuck", message); + } + }), + ); + + it.effect("counts the wrong-account words only with the stuck line", () => + Effect.gen(function* () { + const rig = rigFor(); + setStatus( + rig, + "completed", + "the session's gh account is x, not rawphp\nFACTORY-DONE: not a url", + ); + const outcome = yield* advanceIn(rig); + assert.ok( + outcome.kind === "stuck" && + outcome.reason.startsWith("the session reported a malformed pull request URL"), + JSON.stringify(outcome), + ); + assert.strictEqual(stuckCauses(rig)[0], "advance session-stuck"); + }), + ); + + it.effect("when the message has no usable marker, after the one reminder", () => + Effect.gen(function* () { + for (const message of [ + "I changed some files.", + "FACTORY-STOP: stuck\nThanks for waiting.", + "FACTORY-STOP: stuck, sorry", + "FACTORY-STOP: needs-info\n1. Why?", + `FACTORY-DONE: ${URL12}\nbye`, + ]) { + // The first ended run gets the reminder, the second goes stuck. + const rig = yield* remindedRun(message, message); + assertStuck(rig, yield* advanceIn(rig), { + publicReason: NO_RESULT_LINE, + text: message, + extra: REMINDED_STEPS, + }); + } + }), + ); + + it.effect("when there is no final message, after the one reminder", () => + Effect.gen(function* () { + for (const message of [null, "", " \n"]) { + const rig = yield* remindedRun(message, message); + assertStuck(rig, yield* advanceIn(rig), { + publicReason: "the session ended without a final message", + extra: REMINDED_STEPS, + }); + } + }), + ); + + it.effect("when the run ended badly, with its status and no reminder", () => + Effect.gen(function* () { + const ended = ["failed", "cancelled", "rolled_back"] as const; + for (const status of ended) { + const rig = rigFor(); + setStatus(rig, status, "it crashed"); + assertStuck(rig, yield* advanceIn(rig), { + publicReason: `the session run ended as ${status}`, + text: "it crashed", + }); + assert.deepStrictEqual(stuckCauses(rig), Array(3).fill("advance session-ended-badly")); + } + const rig = rigFor(); + setStatus(rig, "failed", null); + assertStuck(rig, yield* advanceIn(rig), { publicReason: "the session run ended as failed" }); + }), + ); + + it.effect("counts only the known terminal statuses as ended", () => + Effect.gen(function* () { + const statuses = ["completed", "interrupted", "failed", "cancelled", "rolled_back"] as const; + for (const status of statuses) { + const rig = rigFor(); + setStatus(rig, status, "no marker"); + const outcome = yield* advanceIn(rig); + // A run that ended on its own with no result line is reminded; one a person stopped is + // paused; one that ended badly is stuck at once. + if (status === "completed") assert.deepStrictEqual(outcome, { kind: "reminded" }, status); + else if (status === "interrupted") + assert.deepStrictEqual(outcome, { kind: "paused" }, status); + else assert.strictEqual(outcome.kind, "stuck", status); + } + }), + ); + + it.effect("when the thread no longer exists", () => + Effect.gen(function* () { + const rig = rigFor(); + rig.world.status = new ThreadNotFoundError({ threadId: "thread-1" }); + assertStuck(rig, yield* advanceIn(rig), { publicReason: THREAD_GONE }); + assert.deepStrictEqual(stuckCauses(rig), Array(3).fill("advance thread-gone")); + }), + ); + + it.effect("when the thread is archived", () => + Effect.gen(function* () { + const rig = rigFor(); + setStatus(rig, "completed", DONE_MESSAGE, { archived: true }); + assertStuck(rig, yield* advanceIn(rig), { publicReason: THREAD_GONE }); + }), + ); + + it.effect("is never decided by a transient T3 error, which writes nothing", () => + Effect.gen(function* () { + const rig = rigFor(); + rig.world.status = sessionsFailure(); + const before = bytesOf(rig); + const error = yield* Effect.flip(advanceIn(rig)); + assert.strictEqual(error._tag, "FactorySessionsError"); + assert.deepStrictEqual(rig.gh, []); + assert.strictEqual(bytesOf(rig), before); + }), + ); +}); + +describe("advance: a run a person stopped", () => { + it.effect("is paused, once for each run, with nothing sent and nothing on the issue", () => + Effect.gen(function* () { + const rig = rigFor(); + setStatus(rig, "interrupted", "I was about to"); + assert.deepStrictEqual(yield* advanceIn(rig), { kind: "paused" }); + assert.deepStrictEqual(linesOf(rig).slice(STARTED.length), [ + { event: "session_paused", data: { run_ordinal: 1 } }, + ]); + assert.deepStrictEqual(lastCause(rig), { step: "advance", reason: "session-paused" }); + const bytes = bytesOf(rig); + assert.deepStrictEqual(yield* advanceIn(rig), { kind: "paused" }); + assert.strictEqual(bytesOf(rig), bytes); + // The person resumes the session and stops its next run. + setStatus(rig, "interrupted", "again", { runOrdinal: 2 }); + assert.deepStrictEqual(yield* advanceIn(rig), { kind: "paused" }); + assert.deepStrictEqual(linesOf(rig).slice(STARTED.length), [ + { event: "session_paused", data: { run_ordinal: 1 } }, + { event: "session_paused", data: { run_ordinal: 2 } }, + ]); + assert.deepStrictEqual(rig.gh, []); + assert.deepStrictEqual(sends(rig), []); + }), + ); + + it.effect("is read as done when its last message says so", () => + Effect.gen(function* () { + const rig = rigFor(); + setStatus(rig, "interrupted", DONE_MESSAGE); + assert.strictEqual((yield* advanceIn(rig)).kind, "ready-for-gate"); + }), + ); +}); + +describe("advance: refused before anything is read", () => { + it.effect("on a finished run", () => + Effect.gen(function* () { + const rig = rigFor([...STARTED, FINISHED_EVENT]); + const error = yield* Effect.flip(advanceIn(rig)); + assert.strictEqual(error._tag, "NotBuildingError"); + assert.strictEqual(error.message, "run rawphp__softfac__10__1 is already finished"); + assert.deepStrictEqual(rig.gh, []); + assert.deepStrictEqual(rig.t3, []); + }), + ); + + it.effect("on a run with no session", () => + Effect.gen(function* () { + const rig = rigFor([ + { event: "claimed", data: { machine: "mac-mini", github_user: "rawphp" } }, + ]); + const error = yield* Effect.flip(advanceIn(rig)); + assert.strictEqual(error._tag, "NoSessionError"); + assert.strictEqual(error.message, "run rawphp__softfac__10__1 has no launched session"); + assert.deepStrictEqual(rig.t3, []); + }), + ); + + it.effect("on a run with no file", () => + Effect.gen(function* () { + const rig = rigFor(); + const error = yield* Effect.flip(advance(rig.ctx, { ...run10, attempt: 2 })); + assert.strictEqual(error._tag, "LedgerNotFoundError"); + }), + ); + + it.effect("on a run from another repo", () => + Effect.gen(function* () { + const rig = rigFor(); + const other: RunId = { repo: "someone/else", issue: "10", attempt: 1 }; + const error = yield* Effect.flip(advance(rig.ctx, other)); + assert.strictEqual(error._tag, "WrongRepoError"); + assert.strictEqual(error.message, "run someone__else__10__1 is not a run of rawphp/softfac"); + assert.deepStrictEqual(rig.gh, []); + assert.deepStrictEqual(rig.t3, []); + }), + ); + + it.effect("compares the repo case-insensitively", () => + Effect.gen(function* () { + const rig = rigFor(); + setStatus(rig, "running", null); + const outcome = yield* advance({ ...rig.ctx, repo: "RawPHP/SoftFac" }, rig.run); + assert.deepStrictEqual(outcome, { kind: "running" }); + }), + ); +}); + +// ---- advance: the result-line reminder ----------------------------------------------------------- + +describe("advance: the result-line reminder", () => { + it.effect("is sent once, as one fixed message, to a run that ended with no result line", () => + Effect.gen(function* () { + const rig = rigFor(); + setEnded(rig, 1, WORKING_MESSAGE); + assert.deepStrictEqual(yield* advanceIn(rig), { kind: "reminded" }); + assert.deepStrictEqual(rig.t3, ["status thread-1", remindCall()]); + // Nothing on GitHub: the issue is still building and still ours. + assert.deepStrictEqual(rig.gh, []); + assert.deepStrictEqual(linesOf(rig), [...STARTED, ...REMINDED_STEPS]); + assert.deepStrictEqual( + readRun(rig.dataDir, rig.run) + .entries.slice(STARTED.length) + .map((e) => e.cause?.reason), + ["result-line-reminding", "result-line-reminded"], + ); + }), + ); + + it.effect("writes the wait marker before the send and the step after it", () => + Effect.gen(function* () { + const rig = rigFor(); + setEnded(rig, 1, WORKING_MESSAGE); + rig.world.sendFailure = { error: sessionsFailure("send"), delivered: false }; + assert.strictEqual((yield* Effect.flip(advanceIn(rig)))._tag, "FactorySessionsError"); + assert.deepStrictEqual(linesOf(rig), [...STARTED, reminding(1)]); + }), + ); + + it.effect("is not sent again while the session works on it", () => + Effect.gen(function* () { + const rig = rigFor(); + setEnded(rig, 1, WORKING_MESSAGE); + assert.deepStrictEqual(yield* advanceIn(rig), { kind: "reminded" }); + const bytes = bytesOf(rig); + // T3 has not started the run yet, then it is running. + assert.deepStrictEqual(yield* advanceIn(rig), { kind: "running" }); + setStatus(rig, "running", "working", { runOrdinal: 2 }); + assert.deepStrictEqual(yield* advanceIn(rig), { kind: "running" }); + assert.deepStrictEqual(sends(rig), [remindCall()]); + assert.strictEqual(bytesOf(rig), bytes); + }), + ); + + it.effect("is not sent for a run that is still active", () => + Effect.gen(function* () { + const actives: ReadonlyArray = [ + "preparing", + "queued", + "starting", + "running", + "waiting", + null, + ]; + for (const run of actives) { + const rig = rigFor(); + setStatus(rig, run, WORKING_MESSAGE); + assert.deepStrictEqual(yield* advanceIn(rig), { kind: "running" }, String(run)); + assert.deepStrictEqual(sends(rig), [], String(run)); + assert.deepStrictEqual(linesOf(rig), [...STARTED], String(run)); + } + }), + ); + + it.effect("is followed by a newer run that ends with done, which goes to the gate", () => + Effect.gen(function* () { + const rig = yield* remindedRun(WORKING_MESSAGE, `Done.\nFACTORY-DONE: ${URL12}`); + assert.strictEqual((yield* advanceIn(rig)).kind, "ready-for-gate"); + assert.deepStrictEqual(sends(rig), []); + assert.strictEqual(rig.world.issue.state, "building"); + assert.deepStrictEqual(linesOf(rig).at(-1), PR_OPENED); + assert.deepStrictEqual(rig.world.archived, []); + }), + ); + + it.effect("is followed by a newer run that ends with questions, which posts them", () => + Effect.gen(function* () { + const rig = yield* remindedRun(WORKING_MESSAGE, `${QUESTIONS}\nFACTORY-STOP: needs-info`); + assert.deepStrictEqual(yield* advanceIn(rig), { kind: "needs-info" }); + assert.strictEqual(rig.world.issue.state, "needs-info"); + assert.deepStrictEqual(rig.gh, needsInfoCalls(QUESTIONS)); + assert.deepStrictEqual(sends(rig), []); + }), + ); + + it.effect( + "is followed by a second run with no result line, which goes stuck, with no second reminder", + () => + Effect.gen(function* () { + const rig = yield* remindedRun(WORKING_MESSAGE, "Still working."); + assertStuck(rig, yield* advanceIn(rig), { + publicReason: NO_RESULT_LINE, + text: "Still working.", + extra: REMINDED_STEPS, + }); + assert.deepStrictEqual(sends(rig), []); + assert.strictEqual(rig.world.delivered.size, 1); + }), + ); + + it.effect("waits for the second run in the exact live sequence and releases nothing", () => + Effect.gen(function* () { + const rig = rigFor(); + setEnded(rig, 1, WORKING_MESSAGE); + assert.deepStrictEqual(yield* advanceIn(rig), { kind: "reminded" }); + // The session wakes up and finishes: running, then ended with the result line. + setStatus(rig, "running", "working", { runOrdinal: 2 }); + assert.deepStrictEqual(yield* advanceIn(rig), { kind: "running" }); + setEnded(rig, 2, `Opened it.\nFACTORY-DONE: ${URL12}`); + assert.strictEqual((yield* advanceIn(rig)).kind, "ready-for-gate"); + assert.strictEqual(rig.world.issue.state, "building"); + assert.deepStrictEqual(rig.world.issue.assignees, ["rawphp"]); + assert.deepStrictEqual( + rig.gh.filter((call) => call.startsWith("comment")), + [], + ); + assert.ok(!rig.t3.some((call) => call.startsWith("archive")), rig.t3.join("\n")); + assert.ok(!linesOf(rig).some((line) => line.event === "stuck")); + }), + ); + + it.effect("treats a missing message like one with no result line", () => + Effect.gen(function* () { + for (const message of [null, "", " \n"]) { + const rig = rigFor(); + setEnded(rig, 1, message); + assert.deepStrictEqual(yield* advanceIn(rig), { kind: "reminded" }); + assert.deepStrictEqual(sends(rig), [remindCall()]); + } + }), + ); + + it.effect("is not sent for a result line that is present but unusable", () => + Effect.gen(function* () { + for (const message of [ + "FACTORY-DONE: nonsense", + "FACTORY-DONE: https://github.com/someone/else/pull/3", + "FACTORY-STOP: needs-info", + ]) { + const rig = rigFor(); + setEnded(rig, 1, message); + assert.strictEqual((yield* advanceIn(rig)).kind, "stuck", message); + assert.deepStrictEqual(sends(rig), [], message); + } + }), + ); + + it.effect("is not sent for a run that ended badly", () => + Effect.gen(function* () { + for (const status of ["failed", "cancelled", "rolled_back"] as const) { + const rig = rigFor(); + setStatus(rig, status, WORKING_MESSAGE); + assert.strictEqual((yield* advanceIn(rig)).kind, "stuck", status); + assert.deepStrictEqual(sends(rig), [], status); + } + }), + ); + + it.effect("cut short before its send is sent on the next call, under one reminding step", () => + Effect.gen(function* () { + const rig = rigFor([...STARTED, reminding(1)]); + setEnded(rig, 1, WORKING_MESSAGE); + assert.deepStrictEqual(yield* advanceIn(rig), { kind: "reminded" }); + assert.deepStrictEqual(sends(rig), [remindCall()]); + assert.deepStrictEqual(linesOf(rig), [...STARTED, ...REMINDED_STEPS]); + }), + ); + + it.effect("whose step is missing is not sent again when the session answered it", () => + Effect.gen(function* () { + const rig = rigFor([...STARTED, reminding(1)]); + setEnded(rig, 2, `Done.\nFACTORY-DONE: ${URL12}`); + assert.strictEqual((yield* advanceIn(rig)).kind, "ready-for-gate"); + assert.deepStrictEqual(sends(rig), []); + // The send is taken to have arrived, and the step is made good. + assert.deepStrictEqual(linesOf(rig), [...STARTED, ...REMINDED_STEPS, PR_OPENED]); + }), + ); + + it.effect("sent again after a crash between the send and its step uses the same command id", () => + Effect.gen(function* () { + for (const delivered of [false, true]) { + const rig = rigFor(); + setEnded(rig, 1, WORKING_MESSAGE); + rig.world.sendFailure = { error: sessionsFailure("send"), delivered }; + yield* Effect.flip(advanceIn(rig)); + assert.deepStrictEqual(yield* advanceIn(rig), { kind: "reminded" }); + const ids = sends(rig).map((call) => call.split(" ")[2]); + assert.strictEqual(ids.length, 2, `delivered ${delivered}`); + assert.strictEqual(ids[0], ids[1], `delivered ${delivered}`); + assert.strictEqual(rig.world.delivered.size, 1, `delivered ${delivered}`); + assert.deepStrictEqual( + linesOf(rig), + [...STARTED, ...REMINDED_STEPS], + `delivered ${delivered}`, + ); + } + }), + ); + + it.effect("is not owed to a run with no ordinal and no result line, which is stuck at once", () => + Effect.gen(function* () { + const rig = rigFor(); + setStatus(rig, "completed", WORKING_MESSAGE, { runOrdinal: null }); + assert.strictEqual((yield* advanceIn(rig)).kind, "stuck"); + assert.deepStrictEqual(sends(rig), []); + assert.ok( + !linesOf(rig).some( + (line) => + JSON.stringify(line) === + JSON.stringify({ event: "step_changed", data: { step: "result-line-reminded" } }), + ), + ); + }), + ); + + it.effect("with a lost step and a newer run that has no line goes stuck and does not hang", () => + Effect.gen(function* () { + const rig = rigFor([...STARTED, reminding(1)]); + setEnded(rig, 2, "Nothing yet."); + assert.strictEqual((yield* advanceIn(rig)).kind, "stuck"); + assert.deepStrictEqual(sends(rig), []); + }), + ); + + it.effect("is the error, with the marker written, when T3 refuses the message", () => + Effect.gen(function* () { + const rig = rigFor(); + setEnded(rig, 1, WORKING_MESSAGE); + rig.world.sendFailure = { + error: new MessageRejectedError({ + threadId: "thread-1", + commandId: "softfac:rawphp__softfac__10__1:remind:1", + cause: "no", + }), + delivered: false, + }; + assert.strictEqual((yield* Effect.flip(advanceIn(rig)))._tag, "FactoryMessageRejectedError"); + assert.deepStrictEqual(linesOf(rig), [...STARTED, reminding(1)]); + }), + ); + + it.effect("goes stuck as thread-gone when T3 no longer has the thread at the send", () => + Effect.gen(function* () { + const rig = rigFor(); + setEnded(rig, 1, WORKING_MESSAGE); + rig.world.sendFailure = { + error: new ThreadNotFoundError({ threadId: "thread-1" }), + delivered: false, + }; + const outcome = yield* advanceIn(rig); + assert.deepStrictEqual(outcome, { kind: "stuck", reason: THREAD_GONE }); + assert.strictEqual(stuckCauses(rig)[0], "advance thread-gone"); + }), + ); +}); From 8d6ef7830c88366922092b7fe55faaf1f9c32155 Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Fri, 16 Oct 2026 12:49:28 +1000 Subject: [PATCH 04/17] feat(factory): advance a building run to its result line Issue: #17 --- apps/server/src/factory/runner/building.ts | 608 +++++++++++++++++++++ apps/server/src/factory/runner/messages.ts | 2 +- 2 files changed, 609 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/factory/runner/building.ts diff --git a/apps/server/src/factory/runner/building.ts b/apps/server/src/factory/runner/building.ts new file mode 100644 index 000000000000..bdc71eb7367f --- /dev/null +++ b/apps/server/src/factory/runner/building.ts @@ -0,0 +1,608 @@ +/** + * The steps that work a building run, ported from softfac's `runner/building.rs`: `advance` + * reads how the session ended and moves the issue on, `watch` stops a run whose issue text was + * changed by someone who is not an instructor, `timeLimit` stops a run that took too long. The + * rest of `building.rs` (interruptions and relaunches, forwarding comments, ending a run whose + * pull request was merged or closed) belongs to later issues. + * + * Each step reads the ledger first and does only what it does not record, so calling it again + * resumes whatever an earlier call left half done. A failure of the tracker or of T3 is an + * error, with nothing final written, and the next call decides again. + */ +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; + +import type { ProjectConfig } from "../config.ts"; +import { transition, type State } from "../domain/lifecycle.ts"; +import { + NO_MESSAGE, + NO_RESULT_LINE, + SAID_STUCK, + parseSessionResult, + type SessionResult, +} from "../domain/resultLine.ts"; +import { ledgerCall as ledger, type LedgerError } from "../ledger/errors.ts"; +import { makeCause, type CauseStep, type RunEvent } from "../ledger/events.ts"; +import { Ledger } from "../ledger/ledger.ts"; +import { readRun } from "../ledger/read.ts"; +import { asciiLower, runIdToString, type RunId } from "../ledger/runId.ts"; +import type { BoundGitHub, PullRequest } from "../tracker/github.ts"; +import { + MessageRejectedError, + ThreadNotFoundError, + type FactorySessions, + type RunStatus, + type ThreadStatus, +} from "../t3/sessions.ts"; +import { + RESULT_LINE_REMINDER, + THREAD_GONE, + WRONG_ACCOUNT, + WRONG_ACCOUNT_PREFIX, + prepareSessionText, + questionsComment, + statusBody, +} from "./messages.ts"; +import { + RESULT_LINE_REMINDED, + RESULT_LINE_REMINDING, + readProgress, + type Seen, +} from "./progress.ts"; +import { + goStuck, + recordedReason, + stopSession, + finishBuilding, + type Progress, + type RunContext, + type RunError, + type Stop, +} from "./run.ts"; + +/** What the steps of a building run ask of the factory: the run's repo and project, the tracker and the sessions. */ +export interface BuildingContext extends RunContext { + /** `owner/name` as the machine config writes it. */ + readonly repo: string; + readonly project: ProjectConfig; + readonly dataDir: string; + readonly tracker: Pick< + BoundGitHub, + | "setState" + | "unassign" + | "postComment" + | "upsertStatusComment" + | "pullRequest" + | "markPrDraft" + | "textCheck" + | "issue" + >; + readonly sessions: Pick< + FactorySessions["Service"], + "status" | "thread" | "send" | "archive" | "removeWorktree" + >; + /** Unix epoch milliseconds. */ + readonly clock: () => number; +} + +const RunRef = Schema.Struct({ repo: Schema.String, issue: Schema.String, attempt: Schema.Number }); + +/** The run has a `finished` line: it is not a building run any more. */ +export class NotBuildingError extends Schema.TaggedError()("NotBuildingError", { + run: RunRef, +}) { + override get message(): string { + return `run ${runIdToString(this.run)} is already finished`; + } +} + +/** The run has no `session_launched` line: there is no session to read. */ +export class NoSessionError extends Schema.TaggedError()("NoSessionError", { + run: RunRef, +}) { + override get message(): string { + return `run ${runIdToString(this.run)} has no launched session`; + } +} + +/** The run is another repo's. */ +export class WrongRepoError extends Schema.TaggedError()("WrongRepoError", { + run: RunRef, + repo: Schema.String, +}) { + override get message(): string { + return `run ${runIdToString(this.run)} is not a run of ${this.repo}`; + } +} + +export type BuildingError = + | RunError + | NotBuildingError + | NoSessionError + | WrongRepoError + | ThreadNotFoundError + | MessageRejectedError; + +/** What `advance` did. */ +export type AdvanceOutcome = + /** The session is working, or the factory waits for a run it asked for. Nothing was decided. */ + | { readonly kind: "running" } + /** The run ended with no result line and the one reminder was sent (or is waited for). */ + | { readonly kind: "reminded" } + /** A person stopped the session's run in T3: nothing is sent and the issue is not touched. */ + | { readonly kind: "paused" } + /** The session's draft pull request is bound to the run and waits for the gate. */ + | { readonly kind: "ready-for-gate"; readonly pr: number; readonly prUrl: string } + /** The session asked its questions: they are on the issue and the run is ended. */ + | { readonly kind: "needs-info" } + /** The issue went to `stuck`. `reason` is the one the ledger records. */ + | { readonly kind: "stuck"; readonly reason: string }; + +const issueOf = (run: RunId) => Number(run.issue); + +const checkRepo = (ctx: BuildingContext, run: RunId) => + asciiLower(run.repo) === asciiLower(ctx.repo) + ? Effect.void + : Effect.fail(new WrongRepoError({ run, repo: ctx.repo })); + +const readSeen = (ctx: BuildingContext, run: RunId) => + ledger(() => readRun(ctx.dataDir, run)).pipe( + Effect.map((record) => readProgress(record.entries)), + ); + +/** Where the issue stands, from what the ledger says. */ +const progressFrom = (step: CauseStep, seen: Seen, fallback: State): Progress => ({ + step, + state: seen.state ?? fallback, + threadId: seen.threadId, + worktree: seen.worktree, + worktreeDone: seen.worktreeDone, + pr: seen.lastPr?.pr ?? null, + seen, +}); + +/** The run file, opened for append the first time a step writes to it and closed when the step ends. */ +interface Out { + readonly open: Effect.Effect; + readonly close: () => void; +} + +const lazyLedger = (ctx: BuildingContext, run: RunId): Out => { + let handle: Ledger | null = null; + return { + open: ledger(() => (handle ??= Ledger.open(ctx.dataDir, run, { clock: ctx.clock }))), + close: () => handle?.close(), + }; +}; + +const withLedger = ( + ctx: BuildingContext, + run: RunId, + body: (out: Out) => Effect.Effect, +): Effect.Effect => + Effect.suspend(() => { + const out = lazyLedger(ctx, run); + return body(out).pipe(Effect.ensuring(Effect.sync(out.close))); + }); + +const append = (out: Out, event: RunEvent, step: CauseStep, reason: string) => + Effect.flatMap(out.open, (opened) => ledger(() => opened.append(event, makeCause(step, reason)))); + +/** A stop with the session's text, prepared once and left out when empty. */ +const stopOf = ( + reason: string, + sessionText: string | null | undefined, + extra: { readonly detail?: string } = {}, +): Stop => { + const prepared = + sessionText === null || sessionText === undefined ? "" : prepareSessionText(sessionText); + return { + reason, + ...(extra.detail !== undefined && { detail: extra.detail }), + ...(prepared !== "" && { sessionText: prepared }), + }; +}; + +/** Whether a run status is one the run has finished with. An active status is not. */ +const isTerminal = (status: RunStatus): boolean => + status === "completed" || + status === "interrupted" || + status === "failed" || + status === "cancelled" || + status === "rolled_back"; + +/** A run that ended without finishing: T3's own name for it. */ +const endedBadly = (status: RunStatus): boolean => + status === "failed" || + status === "interrupted" || + status === "cancelled" || + status === "rolled_back"; + +/** + * Whether the session ended without any result line: no message, or a last line that is none of + * the three. A malformed `FACTORY-DONE`, or questions with no questions, is not this: the + * session did write a result line. + */ +const lacksResultLine = (result: SessionResult): boolean => + result.kind === "stuck" && (result.reason === NO_MESSAGE || result.reason === NO_RESULT_LINE); + +/** + * Whether the session is still working: its latest run must have ended, and after the factory + * asked for a newer run (the reminder), that run must be a newer one than the run the marker + * names. A newer run is enough, whichever message started it. + */ +const sessionBusy = (seen: Seen, status: ThreadStatus): boolean => { + const ended = status.run !== null && isTerminal(status.run); + const newer = + seen.waitingAfter === null || + (status.runOrdinal !== null && status.runOrdinal > seen.waitingAfter); + return !ended || !newer; +}; + +/** Whether the first non-empty line of what a stuck session said is the one the launch message asks for. */ +const saysWrongAccount = (sessionText: string | undefined): boolean => { + if (sessionText === undefined) return false; + const first = sessionText + .split("\n") + .map((line) => line.trim()) + .find((line) => line !== ""); + return first !== undefined && first.startsWith(WRONG_ACCOUNT_PREFIX); +}; + +const isThreadNotFound = Schema.is(ThreadNotFoundError); + +/** The thread's status. A thread that is archived counts as one T3 no longer has: `null`. */ +const threadStatus = ( + ctx: BuildingContext, + threadId: string, +): Effect.Effect => + ctx.sessions.status(threadId).pipe( + Effect.map((status) => (status.archived ? null : status)), + Effect.catchIf(isThreadNotFound, () => Effect.succeed(null)), + ); + +/** Why a pull request is not the session's own. `public` is one of the factory's fixed lines. */ +interface Unbound { + /** What the comment says. */ + readonly public: string; + /** Set when the public line must leave out something that came from the pull request: the ledger and the caller get this instead. */ + readonly detail?: string; +} + +/** + * Binds a pull request to a run: it exists, is in this repo and not a fork, merges into `base`, + * and has as its head the session's own `branch`. The text came from a session that issue text + * can steer, so it must not be able to name someone else's pull request. Whether it is a draft + * is asked after, so only a bound one is ever put back to draft. + */ +const checkBinding = ( + pull: PullRequest | null, + pr: number, + base: string, + branch: string | null, +): Unbound | null => { + const reported = `the session reported PR #${pr} but`; + if (pull === null) return { public: `${reported} it does not exist` }; + if (pull.crossRepository) return { public: `${reported} it comes from a fork` }; + if (pull.baseRef !== base) { + return { + public: `${reported} its base is not the configured base branch`, + detail: `${reported} its base ${prepareSessionText(pull.baseRef)} is not the configured base branch`, + }; + } + if (branch === null) return { public: `${reported} the session has no branch` }; + if (pull.headRef !== branch) { + return { + public: `${reported} its branch is not this session's branch`, + detail: `${reported} its branch ${prepareSessionText(pull.headRef)} is not this session's branch`, + }; + } + return null; +}; + +const stuckOutcome = (stop: Stop): AdvanceOutcome => ({ + kind: "stuck", + reason: recordedReason(stop), +}); + +/** + * Reads how the run's session ended and moves the issue on. + * + * While the run is active this writes nothing. Once it has ended, the last line of its last + * message decides: a draft pull request that is open, on the session's own branch, is ready for + * the gate; questions go on the issue and end the run; anything else sends the issue to `stuck`, + * after the session's thread is archived (so a session that is still working stops with it; the + * draft pull request, if any, is not touched). + * + * A run that ended on its own with no result line gets one fixed reminder before the factory + * gives up: the session may have ended a turn while a subagent was still working. The step + * `result-line-reminding ` (`k` is the run's ordinal) is written first. It records that a + * send is owed and is the wait marker too: the session counts as working until a run newer than + * `k` has ended. Then the message is sent under `softfac::remind:`, then + * `result-line-reminded` is written. A run that ended as failed, interrupted or cancelled, a run + * with no ordinal, and a result line that is present but unusable get no reminder. + * + * Calling `advance` again resumes whatever an earlier call or `start` left half done, judging by + * the ledger: a launch whose `building` label is missing gets it (once `start` has looked at the + * issue text again after the launch; until then `advance` reports `running` and leaves it to + * `start`), a stuck decision already in the ledger is carried out without asking T3 and without + * a second comment, and questions already posted are not posted again. A run whose + * `state_changed` to `needs-info` is written has nothing left on GitHub: the thread is archived, + * the worktree removed and the run finished. Comments are at-least-once; everything else + * happens once. + */ +export const advance = ( + ctx: BuildingContext, + run: RunId, +): Effect.Effect => + withLedger(ctx, run, (out) => + Effect.gen(function* () { + yield* checkRepo(ctx, run); + const seen = yield* readSeen(ctx, run); + if (seen.finished) return yield* new NotBuildingError({ run }); + const issue = issueOf(run); + const at = progressFrom("advance", seen, "triage"); + const stuck = (code: string, stop: Stop) => + Effect.gen(function* () { + const opened = yield* out.open; + yield* goStuck(ctx, issue, opened, at, code, stop); + return stuckOutcome(stop); + }); + // A stuck decision an older version recorded for a run it left in `needs-info` is not + // carried out: that run is ended below and the issue stays `needs-info`. + if (seen.stuck !== null && at.state !== "needs-info") { + return yield* stuck("stuck-resumed", { reason: seen.stuck }); + } + const threadId = seen.threadId; + if (threadId === null) return yield* new NoSessionError({ run }); + // `state_changed` is the commit marker: a run in `needs-info` has had every write done, and + // a person may have changed the issue since, so it is ended and nothing is written on GitHub. + if (at.state === "needs-info") { + yield* releaseOldClaim(ctx, issue); + yield* endNeedsInfo(ctx, out, at); + return { kind: "needs-info" } as const; + } + if (at.state === "triage") { + // `start` looks at the issue text again after the launch. A launch it has not done that + // for is left to it, so this cannot give the run its `building` label around the check. + if (!seen.rechecked) return { kind: "running" } as const; + yield* finishBuilding(ctx, issue, yield* out.open, at, threadId); + } + + // A thread T3 no longer has is a stop. + const status = yield* threadStatus(ctx, threadId); + if (status === null) return yield* stuck("thread-gone", stopOf(THREAD_GONE, null)); + const message = status.lastAssistantMessage; + const result = parseSessionResult(ctx.repo, message ?? undefined); + + // A `result-line-reminding ` step with no `result-line-reminded` is a reminder cut short + // before its send or its step. If a run newer than `k` exists the send reached the session + // and the step is only written; otherwise the send is made again, under the same command id. + let alreadyReminded = seen.reminded; + if (seen.reminding !== null && !alreadyReminded) { + if (status.runOrdinal !== null && status.runOrdinal > seen.reminding) { + yield* append(out, remindedStep, "advance", "result-line-reminded"); + alreadyReminded = true; + } else { + const sent = yield* sendReminder(ctx, out, run, threadId, seen.reminding); + if (sent === "gone") return yield* stuck("thread-gone", stopOf(THREAD_GONE, null)); + return { kind: "reminded" } as const; + } + } + // The reminder is owed once per run: the latest run, which has an ordinal, ended on its own + // (not as failed or cancelled) with no result line, and none was sent. A run with no + // ordinal cannot be waited for, so it is read as it always was. + const owed = + !alreadyReminded && + lacksResultLine(result) && + status.runOrdinal !== null && + status.run !== null && + isTerminal(status.run) && + !endedBadly(status.run); + if (sessionBusy(seen, status)) return { kind: "running" } as const; + // A run a person stopped in T3 is paused, on purpose: nothing is sent to resume it and the + // issue is not touched. + if (status.run === "interrupted" && result.kind !== "done") { + if (seen.paused === null || seen.paused.runOrdinal !== status.runOrdinal) { + yield* append( + out, + { event: "session_paused", data: { run_ordinal: status.runOrdinal } }, + "advance", + "session-paused", + ); + } + return { kind: "paused" } as const; + } + if ( + status.run !== null && + result.kind !== "done" && + (status.run === "failed" || status.run === "cancelled" || status.run === "rolled_back") + ) { + return yield* stuck( + "session-ended-badly", + stopOf(`the session run ended as ${status.run}`, message), + ); + } + if (owed && status.runOrdinal !== null) { + yield* append( + out, + { + event: "step_changed", + data: { step: `${RESULT_LINE_REMINDING} ${status.runOrdinal}` }, + }, + "advance", + "result-line-reminding", + ); + const sent = yield* sendReminder(ctx, out, run, threadId, status.runOrdinal); + if (sent === "gone") return yield* stuck("thread-gone", stopOf(THREAD_GONE, null)); + return { kind: "reminded" } as const; + } + + switch (result.kind) { + case "done": { + const { pr, prUrl } = result; + const pull = yield* ctx.tracker.pullRequest(pr); + const unbound = checkBinding(pull, pr, ctx.project.base, status.branch); + if (unbound !== null) { + return yield* stuck( + "pr-not-bound", + stopOf( + unbound.public, + message, + unbound.detail === undefined ? {} : { detail: unbound.detail }, + ), + ); + } + // The session's pull request that is open and ready for review goes back to draft: + // only the gate marks it ready. A call that fails leaves it as it is, and stuck. + if (pull !== null && pull.open && !pull.isDraft) { + const drafted = yield* Effect.result(ctx.tracker.markPrDraft(pr)); + if (drafted._tag === "Failure") { + // A rate limit says nothing about the pull request: ask again next pass. + if (drafted.failure._tag === "GitHubRateLimited") return yield* drafted.failure; + return yield* stuck( + "pr-not-bound", + stopOf(`the session reported PR #${pr} but it is not a draft`, message), + ); + } + yield* append(out, { event: "pr_set_draft", data: { pr } }, "advance", "pr-set-draft"); + } + if (!seen.opened.includes(pr)) { + yield* append( + out, + { event: "pr_opened", data: { pr, url: prUrl } }, + "advance", + "pr-opened", + ); + } + // A pull request a person merged or closed before the gate looked at it is ended by + // #20; until then it goes to the gate like any other. + return { kind: "ready-for-gate", pr, prUrl } as const; + } + case "needs-info": { + const prepared = prepareSessionText(result.questions); + const raised = yield* Effect.fromResult( + transition(at.state, { kind: "questions-raised", questions: prepared }), + ); + let questionsPosted = seen.questionsPosted; + let movedTo: State | null = null; + for (const action of raised.actions) { + switch (action.kind) { + case "post-questions": + if (!questionsPosted) { + yield* ctx.tracker.postComment(issue, questionsComment(action.questions)); + yield* append( + out, + { event: "questions_posted", data: { questions: action.questions } }, + "advance", + "questions-asked", + ); + questionsPosted = 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, threadId, at.pr ?? null), + ); + break; + default: + return yield* Effect.die( + new Error(`unexpected action ${action.kind} for questions-raised`), + ); + } + } + if (movedTo !== null && movedTo !== at.state) { + yield* append( + out, + { event: "state_changed", data: { from: at.state, to: movedTo } }, + "advance", + "questions-asked", + ); + at.state = movedTo; + } + yield* endNeedsInfo(ctx, out, at); + return { kind: "needs-info" } as const; + } + case "stuck": { + if (result.reason === SAID_STUCK && saysWrongAccount(result.sessionText)) { + return yield* stuck("session-wrong-account", stopOf(WRONG_ACCOUNT, result.sessionText)); + } + return yield* stuck("session-stuck", stopOf(result.reason, result.sessionText)); + } + } + }), + ); + +const remindedStep: RunEvent = { event: "step_changed", data: { step: RESULT_LINE_REMINDED } }; + +/** + * Gives back an issue that an older version's `answer` claimed for a run now left in + * `needs-info`: that step set `factory:building` and assigned the machine's user before it wrote + * its `state_changed` marker, so a crash between the two leaves the issue claimed with no + * unfinished run. Only that exact shape is touched: the label is `factory:building` and the + * machine's user is the only assignee. Any other label or assignee is a person's, and is left + * alone. Both calls are safe to repeat. + */ +const releaseOldClaim = (ctx: BuildingContext, issue: number) => + Effect.gen(function* () { + const detail = yield* ctx.tracker.issue(issue); + const ours = + detail.assignees.length === 1 && + asciiLower(detail.assignees[0]!) === asciiLower(ctx.trackerUser); + if (detail.state === "building" && ours) { + yield* ctx.tracker.setState(issue, "needs-info"); + yield* ctx.tracker.unassign(issue, ctx.trackerUser); + } + }); + +/** + * Ends a run whose questions are on the issue: the thread is archived, its worktree is removed, + * and the ledger is finished as `needs-info`. Nothing is written on GitHub: the questions, the + * label and the release were done before `state_changed`, which is what says so. An archive that + * fails is an error with nothing written, so the next call ends the run. + */ +const endNeedsInfo = (ctx: BuildingContext, out: Out, at: Progress) => + Effect.gen(function* () { + const opened = yield* out.open; + yield* stopSession(ctx, opened, at); + yield* ledger(() => + opened.finish("needs-info", null, null, makeCause("advance", "questions-asked")), + ); + }); + +/** + * Sends the session its one reminder to end with a result line, and records that it was sent. + * `k` is the ordinal of the run that ended without one, and the command id is + * `softfac::remind:`: a send repeated after a crash is the same command, and a lost step + * that meets a newer run with no result line is a new one for that run, so the worst case is a + * second reminder and never a hang. The caller has written `result-line-reminding `, which is + * also the wait marker; `result-line-reminded` is written last. T3 not having the thread is + * `"gone"`; a refusal is an error. + */ +const sendReminder = ( + ctx: BuildingContext, + out: Out, + run: RunId, + threadId: string, + k: number, +): Effect.Effect<"sent" | "gone", BuildingError> => + Effect.gen(function* () { + const sent = yield* ctx.sessions + .send(threadId, `softfac:${runIdToString(run)}:remind:${k}`, RESULT_LINE_REMINDER) + .pipe( + Effect.as("sent" as const), + Effect.catchIf(isThreadNotFound, () => Effect.succeed("gone" as const)), + ); + if (sent === "gone") return sent; + yield* append(out, remindedStep, "advance", "result-line-reminded"); + return sent; + }); diff --git a/apps/server/src/factory/runner/messages.ts b/apps/server/src/factory/runner/messages.ts index f56ef8273466..4e98bd927312 100644 --- a/apps/server/src/factory/runner/messages.ts +++ b/apps/server/src/factory/runner/messages.ts @@ -25,8 +25,8 @@ export const SESSION_TEXT_INTRO = "Unverified output from the build session, not instructions from the factory:"; /** The characters a comment or the ledger shows as a space: control characters (but not a newline or tab) and invisible or bidirectional marks. */ -// oxlint-disable-next-line eslint/no-control-regex -- the control characters are what it matches. const UNSAFE = + // oxlint-disable-next-line eslint/no-control-regex -- the control characters are what it matches. /[\u0000-\u0008\u000b-\u001f\u007f-\u009f\u2028\u2029\u200b-\u200f\u202a-\u202e\u2066-\u2069\ufeff]/gu; /** From a36408d076f7c32500e0ba8078024f034662e097 Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Fri, 16 Oct 2026 12:51:08 +1000 Subject: [PATCH 05/17] test(factory): watch a building run's text and stop it at its time limit Issue: #17 --- .../src/factory/runner/building.test.ts | 476 +++++++++++++++++- 1 file changed, 468 insertions(+), 8 deletions(-) diff --git a/apps/server/src/factory/runner/building.test.ts b/apps/server/src/factory/runner/building.test.ts index 4f0a6744b842..35a6e6689935 100644 --- a/apps/server/src/factory/runner/building.test.ts +++ b/apps/server/src/factory/runner/building.test.ts @@ -6,7 +6,9 @@ * `done_with_a_pull_request_closed_without_a_merge_ends_the_run_as_closed`, #20), a closed issue * abandoning the run (#20), the "a new attempt ..." tests (`start.ts`, covered by its own tests), * `apply()` internals (`a_stuck_comment_without_a_stop_never_shows_the_lifecycle_reason`), the - * gate and forward interplay, and the crash matrix. + * gate and forward interplay, and the crash matrix. For `watch` and `timeLimit` also left out: a + * recorded gate pass (#19), a run whose own merge closed the issue (#20), a relaunch decided by + * an interruption, and the tests over the real `gh` client (`tracker/github.test.ts` owns them). */ import * as NodeFS from "node:fs"; @@ -38,7 +40,14 @@ import { type IssueDetail, type PullRequest, } from "../tracker/github.ts"; -import { advance, type AdvanceOutcome, type BuildingContext } from "./building.ts"; +import { + advance, + timeLimit, + watch, + type AdvanceOutcome, + type BuildingContext, +} from "./building.ts"; +import { sessionMayExistNote } from "./messages.ts"; const tempDir = useTempDirs(); @@ -227,6 +236,8 @@ interface World { delivered: boolean; } | null; archiveFails: SessionsError | null; + /** The next read of the thread fails, once. */ + threadOnce: SessionsError | null; /** The text of each message T3 took, by command id. */ delivered: Map; archived: Array; @@ -243,12 +254,18 @@ const issueDetail = (state: State | null, assignees: ReadonlyArray): Iss closed: false, }); -const projectConfig = (base = "main"): ProjectConfig => ({ - ...timeLimitOnlyProjectConfig(), - instructors: ["rawphp"], - base, - gate: ["cargo test"], -}); +const MINUTE = 60_000; + +const projectConfig = (base = "main"): ProjectConfig => { + const config = timeLimitOnlyProjectConfig(); + return { + ...config, + instructors: ["rawphp"], + base, + gate: ["cargo test"], + limits: { ...config.limits, minutes: 90 }, + }; +}; /** Seeds a run file with `events`, line `i` written at `ms(i)`. */ const seed = ( @@ -296,6 +313,7 @@ const rigFor = ( failNext: {}, sendFailure: null, archiveFails: null, + threadOnce: null, delivered: new Map(), archived: [], }; @@ -371,6 +389,9 @@ const rigFor = ( }, thread: (threadId) => { t3Call(`thread ${threadId}`); + const once = world.threadOnce; + world.threadOnce = null; + if (once !== null) return Effect.fail(once); const answer = world.thread; return isThreadNotFound(answer) || isSessionsError(answer) ? Effect.fail(answer) @@ -1619,3 +1640,442 @@ describe("advance: the result-line reminder", () => { }), ); }); + +// ---- watch ----------------------------------------------------------------------------------------- + +const CHANGED_DURING_BUILD = + "the issue text was changed during the build by someone who is not an instructor"; +const CANNOT_VERIFY_DURING_BUILD = + "the factory could not verify who changed the issue text during the build"; +const CANNOT_CONFIRM = "the factory could not confirm the issue text was approved"; +const TEXT_READ = `textCheck 10 ${LABELED} rawphp`; +/** What a stop that does not read the session does in T3: archive the thread, then remove its worktree. */ +const ARCHIVE_CALLS = ["archive thread-1", "thread thread-1", "removeWorktree /t3/wt"]; + +const approvedLine = (at: string): RunEvent => step(`approved ${at}`); + +describe("watch", () => { + it.effect("leaves a run alone when the text is unchanged since the approval", () => + Effect.gen(function* () { + const rig = rigFor(); + const bytes = bytesOf(rig); + assert.deepStrictEqual(yield* watch(rig.ctx, rig.run), { kind: "unchanged" }); + // One read, since the moment the run's first line records, with the project's instructors. + assert.deepStrictEqual(rig.gh, [TEXT_READ]); + assert.deepStrictEqual(rig.t3, []); + assert.strictEqual(bytesOf(rig), bytes); + // And again: the same read each pass, never a step that says it passed. + assert.deepStrictEqual(yield* watch(rig.ctx, rig.run), { kind: "unchanged" }); + assert.deepStrictEqual(rig.gh, [TEXT_READ, TEXT_READ]); + assert.strictEqual(bytesOf(rig), bytes); + }), + ); + + it.effect("checks since the approval the ledger records", () => + Effect.gen(function* () { + const rig = rigFor([approvedLine("2026-10-04T12:34:56Z"), ...STARTED.slice(1)]); + assert.deepStrictEqual(yield* watch(rig.ctx, rig.run), { kind: "unchanged" }); + assert.deepStrictEqual(rig.gh, ["textCheck 10 2026-10-04T12:34:56Z rawphp"]); + }), + ); + + it.effect("stops a run whose text someone else changed", () => + Effect.gen(function* () { + const rig = rigFor(); + rig.world.textCheck = "changed"; + assert.deepStrictEqual(yield* watch(rig.ctx, rig.run), { + kind: "stopped", + reason: CHANGED_DURING_BUILD, + }); + assert.deepStrictEqual(rig.gh, [TEXT_READ, ...stuckCalls(CHANGED_DURING_BUILD)]); + assert.deepStrictEqual(rig.t3, ARCHIVE_CALLS); + assert.deepStrictEqual(linesOf(rig), stuckEvents([], CHANGED_DURING_BUILD)); + assert.deepStrictEqual(stuckCauses(rig), Array(3).fill("watch text-changed-during-build")); + assert.strictEqual(rig.world.issue.state, "stuck"); + assert.deepStrictEqual(rig.world.issue.assignees, []); + }), + ); + + it.effect("stops a run whose history cannot be judged, without blaming anyone", () => + Effect.gen(function* () { + const rig = rigFor(); + rig.world.textCheck = "unjudged"; + assert.deepStrictEqual(yield* watch(rig.ctx, rig.run), { + kind: "stopped", + reason: CANNOT_VERIFY_DURING_BUILD, + }); + assert.deepStrictEqual(rig.gh, [TEXT_READ, ...stuckCalls(CANNOT_VERIFY_DURING_BUILD)]); + assert.deepStrictEqual(rig.t3, ARCHIVE_CALLS); + assert.deepStrictEqual(linesOf(rig), stuckEvents([], CANNOT_VERIFY_DURING_BUILD)); + assert.deepStrictEqual(stuckCauses(rig), Array(3).fill("watch edits-unverifiable")); + }), + ); + + it.effect("finishes a stopped run, and nothing polls it again", () => + Effect.gen(function* () { + const rig = rigFor(); + rig.world.textCheck = "changed"; + yield* watch(rig.ctx, rig.run); + clearCalls(rig); + assert.deepStrictEqual(yield* watch(rig.ctx, rig.run), { kind: "skipped", why: "finished" }); + assert.strictEqual((yield* Effect.flip(advanceIn(rig)))._tag, "NotBuildingError"); + assert.deepStrictEqual(rig.gh, []); + assert.deepStrictEqual(rig.t3, []); + }), + ); + + it.effect("leaves the draft pull request open and does not mark it ready", () => + Effect.gen(function* () { + const rig = rigFor([...STARTED, PR_OPENED]); + rig.world.textCheck = "changed"; + assert.strictEqual((yield* watch(rig.ctx, rig.run)).kind, "stopped"); + assert.ok( + !rig.gh.some((call) => call.startsWith("pullRequest") || call.startsWith("markPrDraft")), + rig.gh.join("\n"), + ); + // The status comment still names the pull request the run reported. + assert.ok(rig.gh.at(-1)!.endsWith(statusText("stuck", "#12")), rig.gh.at(-1)); + }), + ); + + it.effect("does nothing for a run that is finished, has no session or is not building", () => + Effect.gen(function* () { + const questions: RunEvent = { event: "questions_posted", data: { questions: "1. Which?" } }; + const stuck: RunEvent = { event: "stuck", data: { reason: "gave up" } }; + const cases: ReadonlyArray<[string, ReadonlyArray, string]> = [ + ["finished", [...STARTED, FINISHED_EVENT], "finished"], + [ + "no session", + [STARTED[0]!, STARTED[1]!, moved("ready", "triage"), moved("triage", "building")], + "no-session", + ], + ["triage", STARTED.slice(0, 5), "not-building"], + ["needs-info", [...STARTED, questions, moved("building", "needs-info")], "not-building"], + ["stuck", [...STARTED, stuck, moved("building", "stuck")], "not-building"], + // The decision is in the ledger and `state_changed` is not: `advance` finishes it. + ["half stuck", [...STARTED, stuck], "not-building"], + ["half needs-info", [...STARTED, questions], "not-building"], + ]; + for (const [name, events, why] of cases) { + const rig = rigFor(events); + rig.world.textCheck = "changed"; + const bytes = bytesOf(rig); + assert.deepStrictEqual(yield* watch(rig.ctx, rig.run), { kind: "skipped", why }, name); + assert.deepStrictEqual(rig.gh, [], name); + assert.deepStrictEqual(rig.t3, [], name); + assert.strictEqual(bytesOf(rig), bytes, name); + } + }), + ); + + it.effect("refuses a run of another repo before touching anything", () => + Effect.gen(function* () { + const rig = rigFor(); + const other: RunId = { repo: "rawphp/other", issue: "10", attempt: 1 }; + const error = yield* Effect.flip(watch(rig.ctx, other)); + assert.strictEqual(error._tag, "WrongRepoError"); + assert.deepStrictEqual(rig.gh, []); + assert.deepStrictEqual(rig.t3, []); + }), + ); + + it.effect("is an error, never a stop or a pass, when the text cannot be read", () => + Effect.gen(function* () { + const rig = rigFor(); + rig.world.textCheck = "changed"; + rig.world.failNext.textCheck = ghFailure(); + const bytes = bytesOf(rig); + assert.strictEqual((yield* Effect.flip(watch(rig.ctx, rig.run)))._tag, "GitHubFailed"); + assert.deepStrictEqual(rig.gh, [TEXT_READ]); + assert.deepStrictEqual(rig.t3, [], "nothing archived"); + assert.strictEqual(bytesOf(rig), bytes, "nothing written"); + assert.strictEqual(rig.world.issue.state, "building"); + // The next pass reads again and, the text being changed, stops the run. + assert.strictEqual((yield* watch(rig.ctx, rig.run)).kind, "stopped"); + }), + ); + + it.effect( + "is an error with nothing written when the archive fails, and the next pass retries", + () => + Effect.gen(function* () { + const rig = rigFor(); + rig.world.textCheck = "changed"; + rig.world.archiveFails = sessionsFailure("archive"); + const bytes = bytesOf(rig); + for (const pass of [1, 2]) { + const error = yield* Effect.flip(watch(rig.ctx, rig.run)); + assert.strictEqual(error._tag, "FactorySessionsError"); + assert.strictEqual(bytesOf(rig), bytes, `pass ${pass}`); + assert.strictEqual(rig.world.issue.state, "building"); + // Only the read: no comment, no label, no release. + assert.strictEqual(rig.gh.length, pass, rig.gh.join("\n")); + } + rig.world.archiveFails = null; + assert.strictEqual((yield* watch(rig.ctx, rig.run)).kind, "stopped"); + assert.deepStrictEqual(rig.world.archived, ["thread-1"]); + assert.deepStrictEqual(linesOf(rig), stuckEvents([], CHANGED_DURING_BUILD)); + }), + ); + + it.effect("does not let a thread T3 no longer knows stop the stop", () => + Effect.gen(function* () { + const rig = rigFor(); + rig.world.textCheck = "changed"; + rig.world.thread = new ThreadNotFoundError({ threadId: "thread-1" }); + assert.strictEqual((yield* watch(rig.ctx, rig.run)).kind, "stopped"); + assert.strictEqual(rig.world.issue.state, "stuck"); + // The worktree is removed from the path the launch recorded. + assert.deepStrictEqual(linesOf(rig), stuckEvents([], CHANGED_DURING_BUILD)); + }), + ); + + it.effect("fails closed for a run with no approval step", () => + Effect.gen(function* () { + const rig = rigFor(STARTED.slice(1)); + assert.deepStrictEqual(yield* watch(rig.ctx, rig.run), { + kind: "stopped", + reason: CANNOT_CONFIRM, + }); + // No read: there is no moment to read since. The archive, then the stuck path. + assert.deepStrictEqual(rig.gh, stuckCalls(CANNOT_CONFIRM)); + assert.deepStrictEqual(rig.t3, ARCHIVE_CALLS); + assert.deepStrictEqual(linesOf(rig), [ + ...STARTED.slice(1), + worktreeRemoved, + ...stuckTail(CANNOT_CONFIRM), + ]); + assert.deepStrictEqual(stuckCauses(rig), Array(3).fill("watch approval-unconfirmed")); + }), + ); + + it.effect( + "is an error with nothing written for a run with no approval step and a failing archive", + () => + Effect.gen(function* () { + const rig = rigFor(STARTED.slice(1)); + rig.world.archiveFails = sessionsFailure("archive"); + const bytes = bytesOf(rig); + assert.strictEqual( + (yield* Effect.flip(watch(rig.ctx, rig.run)))._tag, + "FactorySessionsError", + ); + assert.deepStrictEqual(rig.gh, []); + assert.strictEqual(bytesOf(rig), bytes); + }), + ); +}); + +// ---- the time limit ----------------------------------------------------------------------------------- + +const TIME_LIMIT = "the run passed its time limit"; +const RESUME_SENTENCE = + "Set `factory:ready` again and the next attempt continues from the previous session's commits."; +const LIMIT_MS = 90 * MINUTE; + +/** The started run, claimed at `T0` and building one minute later. */ +const startedAtT0 = (index: number) => (index === STARTED.length - 1 ? T0 + MINUTE : T0); + +/** The comment a time-limit stop posts, with the line that says where the work is. */ +const limitComment = (workLine?: string) => { + const work = workLine === undefined ? "" : `${workLine} ${RESUME_SENTENCE}\n\n`; + return `The factory could not finish this issue and has released it.\n\nReason: ${TIME_LIMIT}\n\n${work}Fix the cause, then set the label back to \`factory:ready\`.`; +}; + +/** The lines of a time-limit stop `elapsed` ms after the run began, with the branch the stop recorded. */ +const limitEvents = (elapsed: number, branch?: string): Array => [ + ...STARTED, + worktreeRemoved, + { event: "stuck", data: { reason: TIME_LIMIT, ...(branch !== undefined && { branch }) } }, + moved("building", "stuck"), + { event: "finished", data: { ...(FINISHED("stuck").data as object), total_ms: elapsed } }, +]; + +/** A started run that is past its limit and stopped by `timeLimit`; returns the elapsed time. */ +const stopAtTheLimit = (rig: Rig) => + Effect.gen(function* () { + const elapsed = LIMIT_MS + 1; + rig.clock.now = T0 + elapsed; + const outcome = yield* timeLimit(rig.ctx, rig.run); + assert.deepStrictEqual(outcome, { kind: "stopped", reason: TIME_LIMIT }); + return elapsed; + }); + +const limitedRig = () => rigFor(STARTED, { ms: startedAtT0 }); + +describe("timeLimit", () => { + it.effect("is within the limit up to and including the last millisecond of it", () => + Effect.gen(function* () { + const rig = limitedRig(); + const bytes = bytesOf(rig); + rig.clock.now = T0 + 1; + assert.deepStrictEqual(yield* timeLimit(rig.ctx, rig.run), { kind: "within" }); + rig.clock.now = T0 + LIMIT_MS; + assert.deepStrictEqual(yield* timeLimit(rig.ctx, rig.run), { kind: "within" }); + assert.deepStrictEqual(rig.gh, []); + assert.deepStrictEqual(rig.t3, []); + assert.strictEqual(bytesOf(rig), bytes); + }), + ); + + it.effect("stops a run past its limit and names the branch without a count of commits", () => + Effect.gen(function* () { + const rig = limitedRig(); + const elapsed = yield* stopAtTheLimit(rig); + assert.deepStrictEqual(linesOf(rig), limitEvents(elapsed, "t3code/abc")); + assert.deepStrictEqual(rig.gh, [ + `comment 10 ${limitComment("The work so far is on branch t3code/abc.")}`, + "setState 10 stuck", + "unassign 10 rawphp", + `status 10 rawphp ${statusText("stuck")}`, + ]); + // The branch is read before the archive, then the archive and the worktree as for any stop. + assert.deepStrictEqual(rig.t3, ["thread thread-1", ...ARCHIVE_CALLS]); + assert.deepStrictEqual(stuckCauses(rig), Array(3).fill("pass time-limit")); + }), + ); + + it.effect( + "stops a run past its limit whose thread has no branch as it would without the work", + () => + Effect.gen(function* () { + const rig = limitedRig(); + rig.world.thread = { worktreePath: "/t3/wt", branch: null, archived: false }; + const elapsed = yield* stopAtTheLimit(rig); + assert.deepStrictEqual(linesOf(rig), limitEvents(elapsed)); + assert.deepStrictEqual(rig.gh[0], `comment 10 ${limitComment()}`); + assert.strictEqual(rig.world.issue.state, "stuck"); + }), + ); + + it.effect( + "stops a run past its limit whose thread cannot be read as it would without the work", + () => + Effect.gen(function* () { + const rig = limitedRig(); + // The first read of the thread is the branch; the stop path's own read goes through. + rig.world.threadOnce = sessionsFailure(); + const elapsed = yield* stopAtTheLimit(rig); + assert.deepStrictEqual(linesOf(rig), limitEvents(elapsed)); + assert.deepStrictEqual(rig.gh[0], `comment 10 ${limitComment()}`); + assert.deepStrictEqual(rig.t3, ["thread thread-1", ...ARCHIVE_CALLS]); + }), + ); + + it.effect( + "stops a run past its limit whose thread T3 no longer knows as it would without the work", + () => + Effect.gen(function* () { + const rig = limitedRig(); + rig.world.thread = new ThreadNotFoundError({ threadId: "thread-1" }); + const elapsed = yield* stopAtTheLimit(rig); + assert.deepStrictEqual(linesOf(rig), limitEvents(elapsed)); + }), + ); + + it.effect( + "is an error with nothing written when the archive fails, and the next call decides again", + () => + Effect.gen(function* () { + const rig = limitedRig(); + rig.clock.now = T0 + LIMIT_MS + 1; + rig.world.archiveFails = sessionsFailure("archive"); + const bytes = bytesOf(rig); + assert.strictEqual( + (yield* Effect.flip(timeLimit(rig.ctx, rig.run)))._tag, + "FactorySessionsError", + ); + assert.strictEqual(bytesOf(rig), bytes); + assert.deepStrictEqual(rig.gh, []); + rig.world.archiveFails = null; + assert.strictEqual((yield* timeLimit(rig.ctx, rig.run)).kind, "stopped"); + }), + ); + + it.effect( + "skips a finished run, and a run that is waiting, stuck or in the middle of either", + () => + Effect.gen(function* () { + const questions: RunEvent = { event: "questions_posted", data: { questions: "1. Which?" } }; + const stuck: RunEvent = { event: "stuck", data: { reason: "gave up" } }; + const cases: ReadonlyArray<[string, ReadonlyArray, string]> = [ + ["finished", [...STARTED, FINISHED_EVENT], "finished"], + ["needs-info", [...STARTED, questions, moved("building", "needs-info")], "not-running"], + ["stuck", [...STARTED, stuck, moved("building", "stuck")], "not-running"], + ["pr-open", [...STARTED, moved("building", "pr-open")], "not-running"], + ["half stuck", [...STARTED, stuck], "not-running"], + ["half needs-info", [...STARTED, questions], "not-running"], + ]; + for (const [name, events, why] of cases) { + const rig = rigFor(events); + rig.clock.now = T0 + 10 * LIMIT_MS; + const bytes = bytesOf(rig); + assert.deepStrictEqual( + yield* timeLimit(rig.ctx, rig.run), + { kind: "skipped", why }, + name, + ); + assert.deepStrictEqual(rig.gh, [], name); + assert.deepStrictEqual(rig.t3, [], name); + assert.strictEqual(bytesOf(rig), bytes, name); + } + }), + ); + + it.effect( + "stops a run that never left triage, whose launch was sent, saying a session may exist", + () => + Effect.gen(function* () { + const sent = [STARTED[0]!, STARTED[1]!, moved("ready", "triage"), step("launch-sending")]; + const rig = rigFor(sent); + rig.clock.now = T0 + LIMIT_MS + 1; + const recorded = sessionMayExistNote(TIME_LIMIT, "softfac:rawphp__softfac__10__1:launch"); + assert.deepStrictEqual(yield* timeLimit(rig.ctx, rig.run), { + kind: "stopped", + reason: recorded, + }); + // The ledger says a session may exist; the comment says only the plain reason. + assert.deepStrictEqual(linesOf(rig), [ + ...sent, + { event: "stuck", data: { reason: recorded } }, + moved("triage", "stuck"), + { + event: "finished", + data: { ...(FINISHED("stuck").data as object), total_ms: LIMIT_MS + 1 }, + }, + ]); + assert.deepStrictEqual(rig.gh, [ + `comment 10 ${limitComment()}`, + "setState 10 stuck", + "unassign 10 rawphp", + "status 10 rawphp **Factory status:** stuck\n- Machine: mac-mini\n- Session: none\n- PR: none", + ]); + // There is no thread: nothing to read or archive. + assert.deepStrictEqual(rig.t3, []); + }), + ); + + it.effect("stops a run that never launched anything with the plain reason", () => + Effect.gen(function* () { + const claimed = [STARTED[0]!, STARTED[1]!, moved("ready", "triage")]; + const rig = rigFor(claimed); + rig.clock.now = T0 + LIMIT_MS + 1; + assert.deepStrictEqual(yield* timeLimit(rig.ctx, rig.run), { + kind: "stopped", + reason: TIME_LIMIT, + }); + assert.deepStrictEqual(rig.t3, []); + }), + ); + + it.effect("refuses a run of another repo before touching anything", () => + Effect.gen(function* () { + const rig = limitedRig(); + const other: RunId = { repo: "rawphp/other", issue: "10", attempt: 1 }; + const error = yield* Effect.flip(timeLimit(rig.ctx, other)); + assert.strictEqual(error._tag, "WrongRepoError"); + assert.deepStrictEqual(rig.gh, []); + assert.deepStrictEqual(rig.t3, []); + }), + ); +}); From fec84c4087882ac295eb35608401eea1afe3f754 Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Fri, 16 Oct 2026 12:52:39 +1000 Subject: [PATCH 06/17] feat(factory): watch a building run's text and stop it at its time limit Issue: #17 --- .../src/factory/runner/building.test.ts | 21 +-- apps/server/src/factory/runner/building.ts | 152 ++++++++++++++++++ 2 files changed, 164 insertions(+), 9 deletions(-) diff --git a/apps/server/src/factory/runner/building.test.ts b/apps/server/src/factory/runner/building.test.ts index 35a6e6689935..b87591cb9d13 100644 --- a/apps/server/src/factory/runner/building.test.ts +++ b/apps/server/src/factory/runner/building.test.ts @@ -1742,7 +1742,9 @@ describe("watch", () => { Effect.gen(function* () { const questions: RunEvent = { event: "questions_posted", data: { questions: "1. Which?" } }; const stuck: RunEvent = { event: "stuck", data: { reason: "gave up" } }; - const cases: ReadonlyArray<[string, ReadonlyArray, string]> = [ + const cases: ReadonlyArray< + [string, ReadonlyArray, "finished" | "no-session" | "not-building"] + > = [ ["finished", [...STARTED, FINISHED_EVENT], "finished"], [ "no session", @@ -1998,14 +2000,15 @@ describe("timeLimit", () => { Effect.gen(function* () { const questions: RunEvent = { event: "questions_posted", data: { questions: "1. Which?" } }; const stuck: RunEvent = { event: "stuck", data: { reason: "gave up" } }; - const cases: ReadonlyArray<[string, ReadonlyArray, string]> = [ - ["finished", [...STARTED, FINISHED_EVENT], "finished"], - ["needs-info", [...STARTED, questions, moved("building", "needs-info")], "not-running"], - ["stuck", [...STARTED, stuck, moved("building", "stuck")], "not-running"], - ["pr-open", [...STARTED, moved("building", "pr-open")], "not-running"], - ["half stuck", [...STARTED, stuck], "not-running"], - ["half needs-info", [...STARTED, questions], "not-running"], - ]; + const cases: ReadonlyArray<[string, ReadonlyArray, "finished" | "not-running"]> = + [ + ["finished", [...STARTED, FINISHED_EVENT], "finished"], + ["needs-info", [...STARTED, questions, moved("building", "needs-info")], "not-running"], + ["stuck", [...STARTED, stuck, moved("building", "stuck")], "not-running"], + ["pr-open", [...STARTED, moved("building", "pr-open")], "not-running"], + ["half stuck", [...STARTED, stuck], "not-running"], + ["half needs-info", [...STARTED, questions], "not-running"], + ]; for (const [name, events, why] of cases) { const rig = rigFor(events); rig.clock.now = T0 + 10 * LIMIT_MS; diff --git a/apps/server/src/factory/runner/building.ts b/apps/server/src/factory/runner/building.ts index bdc71eb7367f..5ecfc3c1386d 100644 --- a/apps/server/src/factory/runner/building.ts +++ b/apps/server/src/factory/runner/building.ts @@ -35,12 +35,17 @@ import { type ThreadStatus, } from "../t3/sessions.ts"; import { + CANNOT_CONFIRM, + CANNOT_VERIFY_DURING_BUILD, + CHANGED_DURING_BUILD, RESULT_LINE_REMINDER, THREAD_GONE, + TIME_LIMIT, WRONG_ACCOUNT, WRONG_ACCOUNT_PREFIX, prepareSessionText, questionsComment, + sessionMayExistNote, statusBody, } from "./messages.ts"; import { @@ -138,6 +143,23 @@ export type AdvanceOutcome = /** The issue went to `stuck`. `reason` is the one the ledger records. */ | { readonly kind: "stuck"; readonly reason: string }; +/** What `watch` did. */ +export type WatchOutcome = + /** The text is as the approval left it. Nothing was written. */ + | { readonly kind: "unchanged" } + /** The run went to `stuck`. `reason` is the one the ledger records. */ + | { readonly kind: "stopped"; readonly reason: string } + | { readonly kind: "skipped"; readonly why: "finished" | "no-session" | "not-building" }; + +/** What `timeLimit` did. */ +export type LimitOutcome = + | { readonly kind: "within" } + /** The run went to `stuck`. `reason` is the one the ledger records. */ + | { readonly kind: "stopped"; readonly reason: string } + | { readonly kind: "skipped"; readonly why: "finished" | "not-running" }; + +const MINUTE_MS = 60_000; + const issueOf = (run: RunId) => Number(run.issue); const checkRepo = (ctx: BuildingContext, run: RunId) => @@ -606,3 +628,133 @@ const sendReminder = ( yield* append(out, remindedStep, "advance", "result-line-reminded"); return sent; }); + +/** + * Why the issue text no longer stands for a building run, or `null` when it does. A run with no + * `approved` step has no moment to read since, so it fails closed without a read; otherwise the + * histories of every rename and body edit since the recorded approval must be an instructor's. + * A history that cannot be judged counts as changed. A read that fails is an error, never an + * answer. + */ +const textStop = ( + ctx: BuildingContext, + issue: number, + seen: Seen, +): Effect.Effect<{ readonly code: string; readonly reason: string } | null, BuildingError> => + Effect.gen(function* () { + if (seen.approvedAt === null) return { code: "approval-unconfirmed", reason: CANNOT_CONFIRM }; + const check = yield* ctx.tracker.textCheck(issue, seen.approvedAt, ctx.project.instructors); + switch (check) { + case "unchanged": + return null; + case "changed": + return { code: "text-changed-during-build", reason: CHANGED_DURING_BUILD }; + case "unjudged": + return { code: "edits-unverifiable", reason: CANNOT_VERIFY_DURING_BUILD }; + } + }); + +/** + * Stops a building run whose issue text was changed by someone who is not an instructor, or whose + * history of edits cannot be judged. The session's thread is archived first, then the run goes + * `stuck` through the stuck path (the draft pull request, if any, is not touched). + * + * A read that fails is an error, never a stop and never a pass. Nothing is written when the text + * is unchanged, and no step records that a check passed: the text is read again on every call, + * since an edit can come at any time. An archive that fails is an error with nothing written, so + * the next call decides and archives again. + * + * A run that is finished, has no session, is not `building`, or has a stuck or questions + * decision not yet committed makes no call (`skipped`: `advance` finishes those). A run of + * another repo is an error. There is no loop or timer: the caller decides how often to call it, + * and it should call `watch` before `advance` on a `building` run. + */ +export const watch = ( + ctx: BuildingContext, + run: RunId, +): Effect.Effect => + Effect.gen(function* () { + yield* checkRepo(ctx, run); + const seen = yield* readSeen(ctx, run); + if (seen.finished) return { kind: "skipped", why: "finished" } as const; + if (seen.threadId === null) return { kind: "skipped", why: "no-session" } as const; + if (seen.state !== "building" || seen.stuck !== null || seen.questionsPosted) { + return { kind: "skipped", why: "not-building" } as const; + } + const issue = issueOf(run); + const why = yield* textStop(ctx, issue, seen); + if (why === null) return { kind: "unchanged" } as const; + const at = progressFrom("watch", seen, "building"); + const stop: Stop = { reason: why.reason }; + yield* withLedger(ctx, run, (out) => + Effect.flatMap(out.open, (opened) => goStuck(ctx, issue, opened, at, why.code, stop)), + ); + return { kind: "stopped", reason: recordedReason(stop) } as const; + }); + +/** The branch the thread works on, or `undefined` when it has none or T3 cannot say: a read that fails is not a stop. */ +const branchOf = (ctx: BuildingContext, threadId: string): Effect.Effect => + ctx.sessions.thread(threadId).pipe( + Effect.map((thread) => thread.branch ?? undefined), + Effect.orElseSucceed(() => undefined), + ); + +/** + * Stops a run that has worked for longer than `limits.minutes`. + * + * The time counted is what the run spent in `triage` and `building`, from the ledger: the time a + * ready pull request waited in `pr-open` is left out (such a run is not limited at all). A run + * past the limit has its session's thread read for its branch and then archived (archiving a + * thread whose run is active stops that run), then goes `stuck` through the stuck path with the + * fixed reason `the run passed its time limit`, released and finished. When the thread has a + * branch, the stuck line and the comment name it, so the next attempt can continue from there; a + * thread that cannot be read, or has no branch, changes nothing. A run still in `triage` with no + * thread has nothing to archive; when its ledger shows a launch call was sent the recorded + * reason says a session may exist, as for any other stop from there. An archive that fails is an + * error before anything is written, so the next call decides and archives again. + * + * A run that is finished, past `building`, in `needs-info`, or has a stuck or questions decision + * not yet committed is skipped. A run of another repo is an error. There is no loop or timer: the + * caller decides how often to call it, and it should call it first for a run in `triage` or + * `building`, and skip the run's other steps when it stops. + */ +export const timeLimit = ( + ctx: BuildingContext, + run: RunId, +): Effect.Effect => + Effect.gen(function* () { + yield* checkRepo(ctx, run); + const seen = yield* readSeen(ctx, run); + if (seen.finished) return { kind: "skipped", why: "finished" } as const; + const working = + seen.state === null || + seen.state === "ready" || + seen.state === "triage" || + seen.state === "building"; + if (!working || seen.stuck !== null || seen.questionsPosted) { + return { kind: "skipped", why: "not-running" } as const; + } + const worked = + seen.workedMs + + (seen.workingSince === null ? 0 : Math.max(0, ctx.clock() - seen.workingSince)); + if (worked <= ctx.project.limits.minutes * MINUTE_MS) return { kind: "within" } as const; + // Read before the archive: the next attempt is told where the work is. + const branch = seen.threadId === null ? undefined : yield* branchOf(ctx, seen.threadId); + const prepared = branch === undefined ? "" : prepareSessionText(branch); + const stop: Stop = { + reason: TIME_LIMIT, + ...(seen.threadId === null && + seen.launchSent && { + detail: sessionMayExistNote(TIME_LIMIT, `softfac:${runIdToString(run)}:launch`), + }), + ...(prepared !== "" && { branch: prepared, commitsAhead: null }), + }; + // Called by the pass before any step, so the pass is what decided. + const at = progressFrom("pass", seen, "triage"); + yield* withLedger(ctx, run, (out) => + Effect.flatMap(out.open, (opened) => + goStuck(ctx, issueOf(run), opened, at, "time-limit", stop), + ), + ); + return { kind: "stopped", reason: recordedReason(stop) } as const; + }); From a2f42fe3176b0806fa04a800b50db7e1228782ff Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Fri, 16 Oct 2026 12:57:47 +1000 Subject: [PATCH 07/17] test(factory): the pass works a building run and the loop wakes on its thread Issue: #17 --- apps/server/src/factory/index.test.ts | 114 +++++++++++- apps/server/src/factory/pass.test.ts | 259 +++++++++++++++++++++++++- 2 files changed, 366 insertions(+), 7 deletions(-) diff --git a/apps/server/src/factory/index.test.ts b/apps/server/src/factory/index.test.ts index 012787b69d36..3cd88d0d1c7b 100644 --- a/apps/server/src/factory/index.test.ts +++ b/apps/server/src/factory/index.test.ts @@ -1,22 +1,30 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; +import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Logger from "effect/Logger"; import * as Path from "effect/Path"; import type * as PlatformError from "effect/PlatformError"; +import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; import { describe } from "vite-plus/test"; import { timeLimitOnlyProjectConfig, type MachineConfig } from "./config.ts"; -import { FactoryRuntime, liveWorld } from "./index.ts"; +import { factoryRuntime, FactoryLoop, FactoryRuntime, liveWorld, ownsThread } from "./index.ts"; +import { makeCause } from "./ledger/events.ts"; +import { Ledger } from "./ledger/ledger.ts"; import { scanMachineLog } from "./ledger/machineLog.ts"; +import { newRunId } from "./ledger/runId.ts"; import type { Repo } from "./pass.ts"; import { CUT_SHORT_COMMENT } from "./runner/messages.ts"; import { utcSecond } from "./runner/start.ts"; import { MARKER } from "./tracker/github.ts"; import { FactoryProcess, RunSpawnError, type RunInput, type RunResult } from "./process.ts"; +import { EventsError, FactoryEvents } from "./t3/events.ts"; import * as Paths from "./t3/paths.ts"; import * as Sessions from "./t3/sessions.ts"; @@ -40,6 +48,12 @@ const stubSessions = Sessions.FactorySessions.of({ }); const noSessions = Layer.succeed(Sessions.FactorySessions, stubSessions); +/** FactoryEvents that never says anything. */ +const quietEvents = Layer.succeed( + FactoryEvents, + FactoryEvents.of({ runStatusChanges: Stream.never }), +); + const messageText = (message: unknown): string => { const first = Array.isArray(message) ? message[0] : message; return typeof first === "string" ? first : String(first); @@ -90,6 +104,7 @@ const logsFor = ( ), Layer.provide(gh), Layer.provide(noSessions), + Layer.provide(quietEvents), ), ), ).pipe(Effect.provide(Logger.layer([capture], { mergeWithExisting: false }))); @@ -181,6 +196,103 @@ it.effect("logs each pass line as it is decided, so a pass that dies has said wh }), ); +// ---- the wake ---- + +describe("waking the pass when a factory thread's run changes status", () => { + /** A factory dir with a valid config and one unfinished run whose session is `thread-1`. */ + const seeded = Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const { join } = yield* Path.Path; + const baseDir = yield* fs.makeTempDirectoryScoped({ prefix: "factory-wake-" }); + const factoryDir = join(baseDir, "factory"); + yield* fs.makeDirectory(factoryDir); + const machineConfigPath = join(factoryDir, "config.yml"); + yield* fs.writeFileString(machineConfigPath, VALID); + const out = Ledger.create(factoryDir, newRunId("a/b", "1", 1)); + try { + out.append( + { event: "claimed", data: { machine: "m", github_user: "u" } }, + makeCause("start", "seed"), + ); + out.append( + { + event: "session_launched", + data: { + thread_id: "thread-1", + worktree: "/t3/wt", + branch: "t3code/abc", + model_instance: "claudeAgent", + model: "claude-sonnet-5-5", + resumed_from: null, + }, + }, + makeCause("start", "seed"), + ); + } finally { + out.close(); + } + return { baseDir, factoryDir, machineConfigPath, worktreesDir: join(baseDir, "worktrees") }; + }); + + it.effect("a thread named by an unfinished run's session is owned; any other is not", () => + Effect.scoped( + Effect.gen(function* () { + const { factoryDir } = yield* seeded; + assert.isTrue(yield* ownsThread(factoryDir, "thread-1")); + assert.isFalse(yield* ownsThread(factoryDir, "thread-other")); + }), + ).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect("an owned thread starts the next pass, and a stream that failed is run again", () => + Effect.scoped( + Effect.gen(function* () { + const { baseDir, factoryDir, machineConfigPath, worktreesDir } = yield* seeded; + const queue = yield* Queue.unbounded(); + // The first run of the stream fails; the one after it reads the queue. + const failures = yield* Ref.make(1); + const events = Layer.succeed( + FactoryEvents, + FactoryEvents.of({ + runStatusChanges: Stream.unwrap( + Ref.modify(failures, (left) => + left > 0 + ? ([ + Stream.fail(new EventsError({ cause: new Error("gone") })), + left - 1, + ] as const) + : ([Stream.fromQueue(queue), 0] as const), + ), + ), + }), + ); + const loop = yield* Layer.build( + factoryRuntime({ restartDelay: 0 }).pipe( + Layer.provide( + Layer.succeed(Paths.FactoryPaths, { + baseDir, + worktreesDir, + factoryDir, + machineConfigPath, + }), + ), + Layer.provide(noGh), + Layer.provide(noSessions), + Layer.provide(events), + ), + ).pipe(Effect.map((context) => Context.get(context, FactoryLoop))); + const started = () => + scanMachineLog(factoryDir).filter((entry) => entry.event === "pass_started").length; + yield* Queue.offer(queue, "thread-other"); + yield* Queue.offer(queue, "thread-1"); + // The first pass, then the one the owned thread woke; the poll is a minute away. + while (started() < 2) yield* loop.awaitPass; + assert.isAtLeast(started(), 2); + }), + ).pipe(Effect.provide(NodeServices.layer)), + ); +}); + // ---- the live world over gh ---- describe("liveWorld", () => { diff --git a/apps/server/src/factory/pass.test.ts b/apps/server/src/factory/pass.test.ts index 8fe0372fa1dd..5961b3778de7 100644 --- a/apps/server/src/factory/pass.test.ts +++ b/apps/server/src/factory/pass.test.ts @@ -12,7 +12,7 @@ import type { HandPause } from "./claims.ts"; import { timeLimitOnlyProjectConfig, type MachineConfig, type ProjectConfig } from "./config.ts"; import type { State } from "./domain/lifecycle.ts"; import { LedgerIoError } from "./ledger/errors.ts"; -import type { ErrorKind, MachineEvent } from "./ledger/events.ts"; +import type { CauseStep, ErrorKind, MachineEvent } from "./ledger/events.ts"; import type { RunInfo } from "./ledger/listRuns.ts"; import { newRunId, runIdToString, type RunId } from "./ledger/runId.ts"; import { formatTs } from "./ledger/time.ts"; @@ -24,6 +24,13 @@ import { type Repo, type Report, } from "./pass.ts"; +import { + NotBuildingError, + type AdvanceOutcome, + type BuildingError, + type LimitOutcome, + type WatchOutcome, +} from "./runner/building.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"; @@ -86,6 +93,10 @@ class Fake implements PassWorld { starts: Record> = {}; /** Issues whose `start` leaves a `triage` run behind, whatever it answers. */ startLeavesRun: Array = []; + /** Scripted answers per `"#"`, popped per call; an exhausted script answers the default. */ + timeLimits: Record> = {}; + watches: Record> = {}; + advances: Record> = {}; pause: HandPause | null = null; machineLogFails = false; /** What the pass wrote to a run's ledger as an `error` line. */ @@ -223,6 +234,43 @@ class Fake implements PassWorld { return Effect.fail(answer); } + private scripted( + table: Record>, + call: string, + repo: Repo, + issue: number, + fallback: T, + ) { + const key = `${repo.watched.repo}#${issue}`; + this.calls.push(`${call} ${key}`); + const answer = table[key]?.shift() ?? fallback; + return "kind" in answer ? Effect.succeed(answer as T) : Effect.fail(answer); + } + + timeLimit(repo: Repo, run: RunId) { + return this.scripted(this.timeLimits, "time_limit", repo, Number(run.issue), { + kind: "within", + }); + } + + watch(repo: Repo, run: RunId) { + return this.scripted(this.watches, "watch", repo, Number(run.issue), { kind: "unchanged" }); + } + + /** An advance that ends the run (`needs-info`, `stuck`) takes it off the list, as the ledger does. */ + advance(repo: Repo, run: RunId) { + const answered = this.scripted(this.advances, "advance", repo, Number(run.issue), { + kind: "running", + }); + return Effect.tap(answered, (outcome) => + Effect.sync(() => { + if (outcome.kind === "needs-info" || outcome.kind === "stuck") { + this.runs = this.runs.filter((r) => runIdToString(r.id) !== runIdToString(run)); + } + }), + ); + } + handPause() { return this.pause; } @@ -231,7 +279,7 @@ class Fake implements PassWorld { return this.now; } - recordError(run: RunId, step: "pass" | "start", kind: ErrorKind, message: string) { + recordError(run: RunId, step: CauseStep, kind: ErrorKind, message: string) { if (this.runErrorFails) { return Effect.fail( new LedgerIoError({ path: "/data/runs/x.jsonl", cause: new Error("finished") }), @@ -883,12 +931,17 @@ describe("runs that are confirmed and still in triage", () => { }), ); - it.effect("a building run is left alone", () => + it.effect("a building run is worked: time limit, watch, advance, in that order", () => Effect.gen(function* () { const fake = new Fake(); - fake.runIn(40, BUILDING); - yield* pass(fake, machine([REPO], 3)); - assert.deepEqual(starts(fake), []); + fake.runIn(40, BUILDING, REPO, true); + const { lines } = yield* pass(fake, machine([REPO], 3)); + const calls = fake.calls.filter((c) => /^(time_limit|watch|advance|start|resume) /.test(c)); + assert.deepEqual(calls, [`time_limit ${REPO}#40`, `watch ${REPO}#40`, `advance ${REPO}#40`]); + assert.equal( + runLine(lines, `${REPO}#40 building:`), + `${REPO}#40 building: session still running`, + ); }), ); @@ -1762,3 +1815,197 @@ describe("text that is printed", () => { const _loaded: Loaded = { kind: "broken", name: REPO }; void _loaded; void runIdToString; + +// ---- following a building run ---- + +describe("following a building run", () => { + const stopped = (reason = "the reason") => ({ kind: "stopped" as const, reason }); + const callsOf = (fake: Fake) => + fake.calls.filter((c) => /^(time_limit|watch|advance|start|resume) /.test(c)); + + it.effect("a triage run gets the time limit before resume", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(12, "triage"); + yield* pass(fake, machine([REPO], 3)); + assert.deepEqual(callsOf(fake), [`time_limit ${REPO}#12`, `resume ${REPO}#12`]); + }), + ); + + it.effect("a triage run stopped at its limit is not resumed and its line says stuck", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(12, "triage"); + fake.timeLimits[`${REPO}#12`] = [stopped()]; + const { lines } = yield* pass(fake, machine([REPO], 3)); + assert.deepEqual(callsOf(fake), [`time_limit ${REPO}#12`]); + assert.equal( + runLine(lines, `${REPO}#12 `), + `${REPO}#12 triage: stuck, the reason is in the ledger of run ${runIdToString(rid(12))}`, + ); + }), + ); + + it.effect("a building run stopped by the time limit gets no watch or advance", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(40, BUILDING, REPO, true); + fake.timeLimits[`${REPO}#40`] = [stopped()]; + const { lines } = yield* pass(fake, machine([REPO], 3)); + assert.deepEqual(callsOf(fake), [`time_limit ${REPO}#40`]); + assert.include(runLine(lines, `${REPO}#40 `), "stuck, the reason is in the ledger of run"); + }), + ); + + it.effect("a building run stopped by watch gets no advance", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(40, BUILDING, REPO, true); + fake.watches[`${REPO}#40`] = [stopped()]; + const { lines } = yield* pass(fake, machine([REPO], 3)); + assert.deepEqual(callsOf(fake), [`time_limit ${REPO}#40`, `watch ${REPO}#40`]); + assert.include(runLine(lines, `${REPO}#40 `), "stuck, the reason is in the ledger of run"); + }), + ); + + const advanceLines: ReadonlyArray = [ + [{ kind: "running" }, "session still running"], + [{ kind: "reminded" }, "session ended without a result line, sent it a reminder"], + [ + { kind: "paused" }, + "session stopped by hand, waiting for an instructor's comment on the pull request or the issue", + ], + [ + { kind: "ready-for-gate", pr: 3, prUrl: "https://github.com/rawphp/softfac/pull/3" }, + "session finished, ready for the gate", + ], + [ + { kind: "needs-info" }, + "session asked questions, the issue is now needs-info and the run is finished", + ], + [ + { kind: "stuck", reason: "r" }, + `stuck, the reason is in the ledger of run ${runIdToString(rid(40))}`, + ], + ]; + + it.effect("each advance outcome has its line", () => + Effect.gen(function* () { + for (const [outcome, words] of advanceLines) { + const fake = new Fake(); + fake.runIn(40, BUILDING, REPO, true); + fake.advances[`${REPO}#40`] = [outcome]; + const { lines } = yield* pass(fake, machine([REPO], 3)); + assert.equal(runLine(lines, `${REPO}#40 `), `${REPO}#40 building: ${words}`); + } + }), + ); + + const failing = ( + script: "advances" | "watches" | "timeLimits", + error: BuildingError, + step: string, + kind: ErrorKind, + ) => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(40, BUILDING, REPO, true); + fake[script][`${REPO}#40`] = [error]; + const { report, lines } = yield* pass(fake, machine([REPO], 3)); + assert.equal(report.errors, 1); + assert.ok( + lines.some((l) => l.startsWith("error: ")), + lines.join("\n"), + ); + assert.equal(fake.runErrors.length, 1); + assert.equal(fake.runErrors[0]?.step, step); + assert.equal(fake.runErrors[0]?.kind, kind); + return fake; + }); + + it.effect("a failing advance is an error line on the run, its kind by the error", () => + Effect.gen(function* () { + yield* failing( + "advances", + new NotBuildingError({ run: { repo: REPO, issue: "40", attempt: 1 } }), + "advance", + "ledger", + ); + yield* failing( + "advances", + new SessionsError({ operation: "status", cause: new Error("x") }), + "advance", + "t3", + ); + yield* failing( + "advances", + new GitHubFailed({ args: ["pr", "view"], status: 1, stderr: "boom" }), + "advance", + "tracker", + ); + }), + ); + + it.effect("a failing watch is recorded with step watch", () => + Effect.gen(function* () { + const fake = yield* failing( + "watches", + new GitHubFailed({ args: ["issue", "view"], status: 1, stderr: "boom" }), + "watch", + "tracker", + ); + assert.ok(!fake.calls.some((c) => c.startsWith("advance "))); + }), + ); + + it.effect( + "a failing time limit is recorded with step pass and the run is left for the pass", + () => + Effect.gen(function* () { + const fake = yield* failing( + "timeLimits", + new SessionsError({ operation: "status", cause: new Error("x") }), + "pass", + "t3", + ); + assert.deepEqual(callsOf(fake), [`time_limit ${REPO}#40`]); + }), + ); + + it.effect("a building run that goes stuck in the pass frees its slot for a claim", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(40, BUILDING, REPO, true); + fake.advances[`${REPO}#40`] = [{ kind: "stuck", reason: "r" }]; + fake.ready[REPO] = [7]; + fake.pushStart(`${REPO}#7`, started(7)); + const { report } = yield* pass(fake, machine([REPO], 1)); + assert.deepEqual(starts(fake), [`start ${REPO}#7`]); + assert.equal(report.claimed, 1); + }), + ); + + it.effect("pr-open, needs-info and stuck runs are not worked", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(41, "pr-open", REPO, true); + fake.runIn(42, "needs-info", REPO, true); + fake.runIn(43, "stuck", REPO, true); + const { lines } = yield* pass(fake, machine([REPO], 5)); + assert.deepEqual( + callsOf(fake).filter((c) => !c.startsWith("time_limit ")), + [], + ); + assert.ok(!lines.some((l) => l.startsWith(`${REPO}#4`)), lines.join("\n")); + }), + ); + + it.effect("a building run of a repo that is not loaded is left alone", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(40, BUILDING, "acme/other", true); + yield* pass(fake, machine([REPO], 3)); + assert.deepEqual(callsOf(fake), []); + }), + ); +}); From cd2b0089a12533b3003afe82659dac07e9cbb6ed Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Fri, 16 Oct 2026 13:01:35 +1000 Subject: [PATCH 08/17] feat(factory): the pass follows a building run and wakes on its thread's run changes Issue: #17 --- apps/server/src/factory/index.test.ts | 9 +- apps/server/src/factory/index.ts | 150 +++++++++++++++------ apps/server/src/factory/pass.test.ts | 16 ++- apps/server/src/factory/pass.ts | 181 +++++++++++++++++++++----- apps/server/src/factory/t3/events.ts | 9 ++ 5 files changed, 281 insertions(+), 84 deletions(-) diff --git a/apps/server/src/factory/index.test.ts b/apps/server/src/factory/index.test.ts index 3cd88d0d1c7b..543373d88e98 100644 --- a/apps/server/src/factory/index.test.ts +++ b/apps/server/src/factory/index.test.ts @@ -255,13 +255,10 @@ describe("waking the pass when a factory thread's run changes status", () => { FactoryEvents, FactoryEvents.of({ runStatusChanges: Stream.unwrap( - Ref.modify(failures, (left) => + Ref.modify(failures, (left): readonly [Stream.Stream, number] => left > 0 - ? ([ - Stream.fail(new EventsError({ cause: new Error("gone") })), - left - 1, - ] as const) - : ([Stream.fromQueue(queue), 0] as const), + ? [Stream.fail(new EventsError({ cause: new Error("gone") })), left - 1] + : [Stream.fromQueue(queue), 0], ), ), }), diff --git a/apps/server/src/factory/index.ts b/apps/server/src/factory/index.ts index ede133af9142..8dbf7e65ce89 100644 --- a/apps/server/src/factory/index.ts +++ b/apps/server/src/factory/index.ts @@ -4,10 +4,12 @@ import * as NodeFS from "node:fs"; import * as NodeOS from "node:os"; import * as Context from "effect/Context"; +import type * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Result from "effect/Result"; +import * as Stream from "effect/Stream"; import { loadHandPause } from "./claims.ts"; import { @@ -24,21 +26,17 @@ import { ledgerCall as ledger } from "./ledger/errors.ts"; import { makeCause, type MachineEvent } from "./ledger/events.ts"; import { Ledger } from "./ledger/ledger.ts"; import { listUnfinishedRuns } from "./ledger/listRuns.ts"; +import { readRun } from "./ledger/read.ts"; import { MachineLog } from "./ledger/machineLog.ts"; import { runLoop, type Loop } from "./loop.ts"; import { runPass, type PassError, type PassWorld, type Repo } from "./pass.ts"; import * as Process from "./process.ts"; import { FactoryProcess } from "./process.ts"; +import { readProgress } from "./runner/progress.ts"; +import { advance, timeLimit, watch, type BuildingContext } from "./runner/building.ts"; import { CUT_SHORT_COMMENT } from "./runner/messages.ts"; -import { - isOurs, - resume, - start, - utcSecond, - type Sessions, - type StartContext, - type Tracker, -} from "./runner/start.ts"; +import { isOurs, resume, start, utcSecond, type StartContext } from "./runner/start.ts"; +import * as Events from "./t3/events.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"; @@ -114,7 +112,7 @@ export const liveWorld = ( machine: MachineConfig, dataDir: string, process: FactoryProcess["Service"], - sessions: Sessions, + sessions: T3Sessions.FactorySessions["Service"], clock: () => number, host: string, ): PassWorld => { @@ -122,7 +120,7 @@ export const liveWorld = ( Effect.provideService(effect, FactoryProcess, process); const githubFor = (watched: WatchedRepo) => makeGitHub({ repo: watched.repo, ghConfigDir: machine.ghConfigDir }); - const trackerFor = (watched: WatchedRepo): Tracker => bindProcess(githubFor(watched), process); + const trackerFor = (watched: WatchedRepo) => bindProcess(githubFor(watched), process); const tracker = (effect: Effect.Effect) => withProcess(effect).pipe(Effect.mapError(asPassError("tracker"))); const startContext = (repo: Repo): StartContext => ({ @@ -138,6 +136,16 @@ export const liveWorld = ( sessions, clock, }); + const buildingContext = (repo: Repo): BuildingContext => ({ + repo: repo.watched.repo, + trackerUser: repo.trackerUser, + machine: host, + project: repo.project, + dataDir, + tracker: trackerFor(repo.watched), + sessions, + clock, + }); return { project: (watched) => withProcess(projectFromBase(githubFor(watched))), trackerUser: (watched) => @@ -186,6 +194,9 @@ export const liveWorld = ( ), start: (repo: Repo, issue: number) => start(startContext(repo), issue), resume: (repo: Repo, issue: number) => resume(startContext(repo), issue), + timeLimit: (repo, run) => timeLimit(buildingContext(repo), run), + watch: (repo, run) => watch(buildingContext(repo), run), + advance: (repo, run) => advance(buildingContext(repo), run), gateRunning: (run) => NodeFS.existsSync(gateLockFile(dataDir, run)), recordError: (run, step, kind, message) => ledger(() => { @@ -221,45 +232,102 @@ const hostName = (): string => { } }; -const startFactory = Effect.gen(function* () { - const { factoryDir, machineConfigPath } = yield* Paths.FactoryPaths; - // Gates run inside this server, so a lock file left from an earlier run belongs to a dead gate. - const cleared = yield* clearGateLocks(factoryDir); - if (cleared > 0) yield* Effect.logInfo(`cleared ${cleared} gate lock(s)`); - const loaded = yield* Effect.result(loadMachineConfig(machineConfigPath)); - if (Result.isFailure(loaded)) { - for (const error of loaded.failure.errors) { - yield* Effect.logError(formatConfigError(error)); +/** + * Whether an unfinished run under `dataDir` names `threadId` as its session's thread. A run whose + * file cannot be read is not counted, and a failed listing is logged and counts as not owned: the + * next scheduled pass covers it. + */ +export const ownsThread = (dataDir: string, threadId: string): Effect.Effect => + ledger(() => + listUnfinishedRuns(dataDir).some((info) => { + if (info.unreadable) return false; + try { + return readProgress(readRun(dataDir, info.id).entries).threadId === threadId; + } catch { + return false; + } + }), + ).pipe( + Effect.catch((error) => + Effect.logWarning( + `factory cannot tell whether it owns thread ${threadId}: ${error.message}`, + ).pipe(Effect.as(false)), + ), + ); + +/** + * Wakes the loop when T3 says the latest run of a thread the factory owns changed status. A + * stream that ends or fails is run again after `restartDelay`; the status changes in the gap are + * not replayed, and the next pass reads them. + */ +const wakeOnOwnedThreads = (dataDir: string, loop: Loop, restartDelay: Duration.Input) => + Effect.gen(function* () { + const events = yield* Events.FactoryEvents; + return yield* events.runStatusChanges.pipe( + Stream.runForEach((threadId) => + Effect.flatMap(ownsThread(dataDir, threadId), (owned) => (owned ? loop.wake : Effect.void)), + ), + Effect.catch((error) => Effect.logWarning("factory events stream failed, restarting", error)), + Effect.andThen(Effect.sleep(restartDelay)), + Effect.forever, + ); + }); + +/** What the factory's runtime can be told, for a test that cannot wait. */ +export interface RuntimeOptions { + /** How long to wait before the events stream is run again. Default 5 seconds. */ + readonly restartDelay?: Duration.Input; +} + +const startFactory = (options: RuntimeOptions) => + Effect.gen(function* () { + const { factoryDir, machineConfigPath } = yield* Paths.FactoryPaths; + // Gates run inside this server, so a lock file left from an earlier run belongs to a dead gate. + const cleared = yield* clearGateLocks(factoryDir); + if (cleared > 0) yield* Effect.logInfo(`cleared ${cleared} gate lock(s)`); + const loaded = yield* Effect.result(loadMachineConfig(machineConfigPath)); + if (Result.isFailure(loaded)) { + for (const error of loaded.failure.errors) { + yield* Effect.logError(formatConfigError(error)); + } + return idle; } - return idle; - } - if (Option.isNone(loaded.success)) { - yield* Effect.logInfo(`no factory config at ${machineConfigPath}`); - return idle; - } - const machine = loaded.success.value; - yield* Effect.logInfo("factory started"); - const process = yield* FactoryProcess; - 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(); - const log = Effect.runSyncWith(context); - return yield* runPass(world, machine, (line) => log(Effect.logInfo(line))); + if (Option.isNone(loaded.success)) { + yield* Effect.logInfo(`no factory config at ${machineConfigPath}`); + return idle; + } + const machine = loaded.success.value; + yield* Effect.logInfo("factory started"); + const process = yield* FactoryProcess; + 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(); + const log = Effect.runSyncWith(context); + return yield* runPass(world, machine, (line) => log(Effect.logInfo(line))); + }); + const loop = yield* runLoop(onePass, machine.pollSeconds); + yield* Effect.forkScoped( + wakeOnOwnedThreads(factoryDir, loop, options.restartDelay ?? "5 seconds"), + ); + return loop; }); - return yield* runLoop(onePass, machine.pollSeconds); -}); /** - * Needs FactoryPaths, FactoryProcess, FactorySessions and FileSystem; what index.test.ts builds - * with a test FactoryPaths, a fake FactoryProcess and a stub FactorySessions. + * Needs FactoryPaths, FactoryProcess, FactorySessions, FactoryEvents and FileSystem; what + * index.test.ts builds with a test FactoryPaths, a fake FactoryProcess, a stub FactorySessions + * and a FactoryEvents it feeds by hand. */ -export const FactoryRuntime = Layer.effect(FactoryLoop, startFactory); +export const factoryRuntime = (options: RuntimeOptions = {}) => + Layer.effect(FactoryLoop, startFactory(options)); + +export const FactoryRuntime = factoryRuntime(); /** The factory's root layer. server.ts merges it into the runtime; nothing else imports it. */ export const FactoryLayer = FactoryRuntime.pipe( Layer.provide(Paths.layer), Layer.provide(Process.layer), Layer.provide(T3Sessions.layerLive), + Layer.provide(Events.layerLive), ); diff --git a/apps/server/src/factory/pass.test.ts b/apps/server/src/factory/pass.test.ts index 5961b3778de7..61b4875a4e08 100644 --- a/apps/server/src/factory/pass.test.ts +++ b/apps/server/src/factory/pass.test.ts @@ -595,7 +595,10 @@ describe("a ready issue whose older run is gating", () => { fake.blockers[`${REPO}#14`] = blockerList([[REPO, 13]]); const { report, lines } = yield* pass(fake, machine([REPO], 1)); assert.deepEqual(starts(fake), []); - assert.equal(runLine(lines, "rawphp/softfac#14 "), "rawphp/softfac#14 ready: gate running"); + assert.equal( + runLine(lines, "rawphp/softfac#14 ready"), + "rawphp/softfac#14 ready: gate running", + ); assert.ok(!fake.calls.includes("blockers rawphp/softfac#14"), fake.calls.join("\n")); assert.deepEqual( fake.events.filter((e) => e.event === "claim_skipped"), @@ -2000,6 +2003,17 @@ describe("following a building run", () => { }), ); + it.effect("a building run whose gate is running is not worked", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(40, BUILDING, REPO, true); + fake.gateLocks = [runIdToString(rid(40))]; + const { lines } = yield* pass(fake, machine([REPO], 3)); + assert.deepEqual(callsOf(fake), []); + assert.equal(runLine(lines, `${REPO}#40 `), `${REPO}#40 building: gate running`); + }), + ); + it.effect("a building run of a repo that is not loaded is left alone", () => Effect.gen(function* () { const fake = new Fake(); diff --git a/apps/server/src/factory/pass.ts b/apps/server/src/factory/pass.ts index d0a952e9ee30..c9f0161bad41 100644 --- a/apps/server/src/factory/pass.ts +++ b/apps/server/src/factory/pass.ts @@ -15,10 +15,16 @@ import type { MachineConfig, ProjectConfig, WatchedRepo } from "./config.ts"; import { stateLabel, type State } from "./domain/lifecycle.ts"; import { sanitize } from "./gate.ts"; import type { LedgerError } from "./ledger/errors.ts"; -import type { ErrorKind, MachineEvent } from "./ledger/events.ts"; +import type { CauseStep, ErrorKind, MachineEvent } from "./ledger/events.ts"; import { holdsSlot, type RunInfo } from "./ledger/listRuns.ts"; import { asciiLower, runIdToString, type RunId } from "./ledger/runId.ts"; import { duration, formatTs } from "./ledger/time.ts"; +import type { + AdvanceOutcome, + BuildingError, + LimitOutcome, + WatchOutcome, +} from "./runner/building.ts"; import { mention, type StartError, type StartOutcome } from "./runner/start.ts"; import { LIST_LIMIT, type Blocker, type Listing, type UnassignedClaim } from "./tracker/github.ts"; @@ -76,12 +82,18 @@ export interface PassWorld { 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; + /** The time limit: `runner/building.ts` `timeLimit`. */ + readonly timeLimit: (repo: Repo, run: RunId) => Effect.Effect; + /** The text watch: `watch`. */ + readonly watch: (repo: Repo, run: RunId) => Effect.Effect; + /** The session's result: `advance`. */ + readonly advance: (repo: Repo, run: RunId) => 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. */ readonly recordError: ( run: RunId, - step: "pass" | "start", + step: CauseStep, kind: ErrorKind, message: string, ) => Effect.Effect; @@ -112,7 +124,7 @@ interface Notes { interface NoteError { /** The error line, without its `error: ` prefix. */ readonly line: string; - readonly step: "pass" | "start"; + readonly step: CauseStep; readonly kind: ErrorKind; /** What went wrong, before it is cleaned for the ledger. */ readonly message: string; @@ -130,7 +142,7 @@ const passError = (kind: ErrorKind, what: string, detail: string): NoteError => }); /** Every call `start` makes is a tracker call, whatever host the tracker runs on. */ -const errorKind = (error: StartError): ErrorKind => { +const errorKind = (error: StartError | BuildingError): ErrorKind => { switch (error._tag) { case "GitHubSpawnError": case "GitHubInvalidArgument": @@ -143,13 +155,25 @@ const errorKind = (error: StartError): ErrorKind => { case "GateRunningError": return "gate"; case "FactorySessionsError": + case "FactoryThreadNotFoundError": + case "FactoryMessageRejectedError": return "t3"; + case "WrongRepoError": + return "config"; default: // The run's own record disagrees with what the step expected. return "ledger"; } }; +/** The ledger step an error line is recorded under, by the step's name in the pass line. */ +const STEP_OF: Readonly> = { + start: "start", + watch: "watch", + advance: "advance", + "time limit": "pass", +}; + /** Repos match as run ids do: `owner/name` folded with `asciiLower`. */ const sameRepo = (a: string, b: string) => asciiLower(a) === asciiLower(b); @@ -253,10 +277,10 @@ class Pass { } /** The step `name` failed for this issue. */ - failed(notes: Notes, name: string, error: StartError): void { + failed(notes: Notes, name: string, error: StartError | BuildingError): void { notes.errors.push({ line: `${notes.label} ${name}: ${plain(error.message)}`, - step: name === "start" ? "start" : "pass", + step: STEP_OF[name] ?? "pass", kind: errorKind(error), message: error.message, }); @@ -375,8 +399,8 @@ const runs = (world: PassWorld, pass: Pass): Effect.Effect => Effect.gen(function* () { for (const info of [...known.infos]) { - if (!inTriage(info)) continue; + if (info.unreadable || !info.confirmed) continue; const loaded = repos.find( (l) => l.kind === "ready" && sameRepo(l.repo.watched.repo, info.id.repo), ); @@ -417,35 +443,118 @@ const workRuns = ( 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; - } - } + if (world.gateRunning(info.id)) { + // The gate has the run: nothing else is done for it (softfac's `work_runs`). + notes.parts.push(GATE_RUNNING); + } else if (yield* withinLimit(world, repo, info, notes, pass)) { + if (inTriage(info)) yield* resumeRun(world, repo, info, notes, pass); + else if (info.state === "building") yield* followBuilding(world, repo, info, notes, pass); + // Any other state (`needs-info`, `pr-open`, `stuck`) waits for a later step. } yield* known.refresh(world, pass, name, issue); yield* pass.flush(notes, info.id); } }); +/** The time limit for the run. False when the run was stopped or the call failed: it is done for this pass. */ +const withinLimit = ( + world: PassWorld, + repo: Repo, + info: RunInfo, + notes: Notes, + pass: Pass, +): Effect.Effect => + Effect.gen(function* () { + const outcome = yield* Effect.result(world.timeLimit(repo, info.id)); + if (outcome._tag === "Failure") { + pass.failed(notes, "time limit", outcome.failure); + return false; + } + if (outcome.success.kind === "stopped") { + pass.stuck(notes, runIdToString(info.id)); + return false; + } + return true; + }); + +/** `resume` for a run in `triage`. */ +const resumeRun = ( + world: PassWorld, + repo: Repo, + info: RunInfo, + notes: Notes, + pass: Pass, +): Effect.Effect => + Effect.gen(function* () { + const issue = Number(info.id.issue); + 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); + return; + } + 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; + } + } + }); + +/** The watch, then `advance`, for a building run: what the session did, in a line. */ +const followBuilding = ( + world: PassWorld, + repo: Repo, + info: RunInfo, + notes: Notes, + pass: Pass, +): Effect.Effect => + Effect.gen(function* () { + const run = runIdToString(info.id); + const watched = yield* Effect.result(world.watch(repo, info.id)); + if (watched._tag === "Failure") return pass.failed(notes, "watch", watched.failure); + if (watched.success.kind === "stopped") return pass.stuck(notes, run); + const advanced = yield* Effect.result(world.advance(repo, info.id)); + // A `NotBuildingError` is a run that finished meanwhile: softfac prints it as an error too. + if (advanced._tag === "Failure") return pass.failed(notes, "advance", advanced.failure); + switch (advanced.success.kind) { + case "running": + notes.parts.push("session still running"); + break; + case "reminded": + notes.parts.push("session ended without a result line, sent it a reminder"); + break; + case "paused": + notes.parts.push( + "session stopped by hand, waiting for an instructor's comment on the pull request or the issue", + ); + break; + case "ready-for-gate": + notes.parts.push("session finished, ready for the gate"); + break; + case "needs-info": + notes.parts.push( + "session asked questions, the issue is now needs-info and the run is finished", + ); + break; + case "stuck": + pass.stuck(notes, run); + break; + } + }); + /** * 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. diff --git a/apps/server/src/factory/t3/events.ts b/apps/server/src/factory/t3/events.ts index 0620b015d10f..e9d9270502a2 100644 --- a/apps/server/src/factory/t3/events.ts +++ b/apps/server/src/factory/t3/events.ts @@ -6,6 +6,7 @@ import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as EventSink from "../../orchestration-v2/EventSink.ts"; +import * as RuntimeLayer from "../../orchestration-v2/runtimeLayer.ts"; import { latestRun, type RunStatus } from "./sessions.ts"; /** @@ -119,3 +120,11 @@ export const make = Effect.gen(function* () { }); export const layer = Layer.effect(FactoryEvents, make); + +/** + * `layer` with the event sink the application runs on: its live stream is that instance's own, + * so a second sink would see nothing. `RuntimeLayer.layerEventSink` is the one layer value every + * other service takes the sink from, so a build shares one instance. `server.ts` has the stores + * it reads in scope where the factory is merged in, but does not expose the sink itself. + */ +export const layerLive = layer.pipe(Layer.provide(RuntimeLayer.layerEventSink)); From b4e6422a6ac406213cb09e380475b1697b24795a Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Fri, 16 Oct 2026 13:11:04 +1000 Subject: [PATCH 09/17] test(factory): the pass advances a run left in needs-info or stuck, and a refused reminder is final Issue: #17 --- apps/server/src/factory/pass.test.ts | 61 +++++++++++++++++-- .../src/factory/runner/building.test.ts | 47 ++++++++++---- 2 files changed, 91 insertions(+), 17 deletions(-) diff --git a/apps/server/src/factory/pass.test.ts b/apps/server/src/factory/pass.test.ts index 61b4875a4e08..578a99e9116c 100644 --- a/apps/server/src/factory/pass.test.ts +++ b/apps/server/src/factory/pass.test.ts @@ -1988,18 +1988,67 @@ describe("following a building run", () => { }), ); - it.effect("pr-open, needs-info and stuck runs are not worked", () => + it.effect("a needs-info run is advanced, with no watch and no resume", () => Effect.gen(function* () { const fake = new Fake(); - fake.runIn(41, "pr-open", REPO, true); fake.runIn(42, "needs-info", REPO, true); + fake.advances[`${REPO}#42`] = [{ kind: "needs-info" }]; + const { lines } = yield* pass(fake, machine([REPO], 5)); + assert.deepEqual(callsOf(fake), [`time_limit ${REPO}#42`, `advance ${REPO}#42`]); + assert.equal( + runLine(lines, `${REPO}#42 `), + `${REPO}#42 needs-info: session asked questions, the issue is now needs-info and the run is finished`, + ); + }), + ); + + it.effect("a stuck run is advanced, with no watch and no resume", () => + Effect.gen(function* () { + const fake = new Fake(); fake.runIn(43, "stuck", REPO, true); + fake.advances[`${REPO}#43`] = [{ kind: "stuck", reason: "r" }]; const { lines } = yield* pass(fake, machine([REPO], 5)); - assert.deepEqual( - callsOf(fake).filter((c) => !c.startsWith("time_limit ")), - [], + assert.deepEqual(callsOf(fake), [`time_limit ${REPO}#43`, `advance ${REPO}#43`]); + assert.equal( + runLine(lines, `${REPO}#43 `), + `${REPO}#43 stuck: stuck, the reason is in the ledger of run ${runIdToString(rid(43))}`, ); - assert.ok(!lines.some((l) => l.startsWith(`${REPO}#4`)), lines.join("\n")); + }), + ); + + it.effect("a failing advance of a needs-info run is recorded with step advance", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(42, "needs-info", REPO, true); + fake.advances[`${REPO}#42`] = [ + new SessionsError({ operation: "archive", cause: new Error("x") }), + ]; + const { report } = yield* pass(fake, machine([REPO], 5)); + assert.equal(report.errors, 1); + assert.equal(fake.runErrors[0]?.step, "advance"); + }), + ); + + it.effect("a pr-open run gets only the time limit", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(41, "pr-open", REPO, true); + const { lines } = yield* pass(fake, machine([REPO], 5)); + assert.deepEqual(callsOf(fake), [`time_limit ${REPO}#41`]); + assert.ok(!lines.some((l) => l.startsWith(`${REPO}#41`)), lines.join("\n")); + }), + ); + + it.effect("a needs-info run that its advance finishes frees its slot for a claim", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(42, "needs-info", REPO, true); + fake.advances[`${REPO}#42`] = [{ kind: "needs-info" }]; + fake.ready[REPO] = [7]; + fake.pushStart(`${REPO}#7`, started(7)); + const { report } = yield* pass(fake, machine([REPO], 1)); + assert.deepEqual(starts(fake), [`start ${REPO}#7`]); + assert.equal(report.claimed, 1); }), ); diff --git a/apps/server/src/factory/runner/building.test.ts b/apps/server/src/factory/runner/building.test.ts index b87591cb9d13..4bfbeeefa1d3 100644 --- a/apps/server/src/factory/runner/building.test.ts +++ b/apps/server/src/factory/runner/building.test.ts @@ -1609,20 +1609,45 @@ describe("advance: the result-line reminder", () => { }), ); - it.effect("is the error, with the marker written, when T3 refuses the message", () => + const refusal = () => + new MessageRejectedError({ + threadId: "thread-1", + commandId: "softfac:rawphp__softfac__10__1:remind:1", + cause: "no", + }); + + /** A refused reminder is final: T3 keeps its refusal under the command id, so the run goes stuck at once. */ + const assertRefusedReminderIsStuck = (rig: Rig, outcome: AdvanceOutcome) => + Effect.gen(function* () { + const recorded = `${NO_RESULT_LINE}; T3 refused the reminder: ${refusal().message}\n\n${WORKING_MESSAGE}`; + assert.deepStrictEqual(outcome, { kind: "stuck", reason: recorded }); + assert.deepStrictEqual(rig.gh, stuckCalls(NO_RESULT_LINE, WORKING_MESSAGE)); + assert.deepStrictEqual(linesOf(rig), stuckEvents([reminding(1)], recorded)); + assert.deepStrictEqual(rig.world.archived, ["thread-1"]); + assert.strictEqual(stuckCauses(rig)[0], "advance session-stuck"); + // The run is finished: a second call sends nothing. + clearCalls(rig); + assert.strictEqual((yield* Effect.flip(advanceIn(rig)))._tag, "NotBuildingError"); + assert.deepStrictEqual(sends(rig), []); + }); + + it.effect("goes stuck at once, with the marker written, when T3 refuses the message", () => Effect.gen(function* () { const rig = rigFor(); setEnded(rig, 1, WORKING_MESSAGE); - rig.world.sendFailure = { - error: new MessageRejectedError({ - threadId: "thread-1", - commandId: "softfac:rawphp__softfac__10__1:remind:1", - cause: "no", - }), - delivered: false, - }; - assert.strictEqual((yield* Effect.flip(advanceIn(rig)))._tag, "FactoryMessageRejectedError"); - assert.deepStrictEqual(linesOf(rig), [...STARTED, reminding(1)]); + rig.world.sendFailure = { error: refusal(), delivered: false }; + const outcome = yield* advanceIn(rig); + yield* assertRefusedReminderIsStuck(rig, outcome); + }), + ); + + it.effect("goes stuck at once when T3 refuses the message of a reminder cut short", () => + Effect.gen(function* () { + const rig = rigFor([...STARTED, reminding(1)]); + setEnded(rig, 1, WORKING_MESSAGE); + rig.world.sendFailure = { error: refusal(), delivered: false }; + const outcome = yield* advanceIn(rig); + yield* assertRefusedReminderIsStuck(rig, outcome); }), ); From a56eabc03657df6412e7882c7e4d63af08be6247 Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Fri, 16 Oct 2026 13:11:19 +1000 Subject: [PATCH 10/17] fix(factory): the pass advances a run left in needs-info or stuck Issue: #17 --- apps/server/src/factory/pass.ts | 25 +++++++++++++++++++++---- 1 file changed, 21 insertions(+), 4 deletions(-) diff --git a/apps/server/src/factory/pass.ts b/apps/server/src/factory/pass.ts index c9f0161bad41..2c419ac39e55 100644 --- a/apps/server/src/factory/pass.ts +++ b/apps/server/src/factory/pass.ts @@ -422,7 +422,8 @@ const runs = (world: PassWorld, pass: Pass): Effect.Effect => Effect.gen(function* () { - const run = runIdToString(info.id); const watched = yield* Effect.result(world.watch(repo, info.id)); if (watched._tag === "Failure") return pass.failed(notes, "watch", watched.failure); - if (watched.success.kind === "stopped") return pass.stuck(notes, run); + if (watched.success.kind === "stopped") return pass.stuck(notes, runIdToString(info.id)); + yield* advanceRun(world, repo, info, notes, pass); + }); + +/** `advance` for a run, and what it did in a line. */ +const advanceRun = ( + world: PassWorld, + repo: Repo, + info: RunInfo, + notes: Notes, + pass: Pass, +): Effect.Effect => + Effect.gen(function* () { + const run = runIdToString(info.id); const advanced = yield* Effect.result(world.advance(repo, info.id)); // A `NotBuildingError` is a run that finished meanwhile: softfac prints it as an error too. if (advanced._tag === "Failure") return pass.failed(notes, "advance", advanced.failure); From c4ac52d5813d81628e78854fed095e3e67f77720 Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Fri, 16 Oct 2026 13:11:48 +1000 Subject: [PATCH 11/17] fix(factory): a reminder T3 refuses ends the run as stuck Issue: #17 --- apps/server/src/factory/runner/building.ts | 31 +++++++++++++++++++--- 1 file changed, 28 insertions(+), 3 deletions(-) diff --git a/apps/server/src/factory/runner/building.ts b/apps/server/src/factory/runner/building.ts index 5ecfc3c1386d..50b6f5c666e5 100644 --- a/apps/server/src/factory/runner/building.ts +++ b/apps/server/src/factory/runner/building.ts @@ -55,6 +55,7 @@ import { type Seen, } from "./progress.ts"; import { + describe, goStuck, recordedReason, stopSession, @@ -408,6 +409,8 @@ export const advance = ( } else { const sent = yield* sendReminder(ctx, out, run, threadId, seen.reminding); if (sent === "gone") return yield* stuck("thread-gone", stopOf(THREAD_GONE, null)); + if (sent !== "sent") + return yield* stuck("session-stuck", refusedStop(result, message, sent)); return { kind: "reminded" } as const; } } @@ -457,6 +460,8 @@ export const advance = ( ); const sent = yield* sendReminder(ctx, out, run, threadId, status.runOrdinal); if (sent === "gone") return yield* stuck("thread-gone", stopOf(THREAD_GONE, null)); + if (sent !== "sent") + return yield* stuck("session-stuck", refusedStop(result, message, sent)); return { kind: "reminded" } as const; } @@ -608,7 +613,8 @@ const endNeedsInfo = (ctx: BuildingContext, out: Out, at: Progress) => * that meets a newer run with no result line is a new one for that run, so the worst case is a * second reminder and never a hang. The caller has written `result-line-reminding `, which is * also the wait marker; `result-line-reminded` is written last. T3 not having the thread is - * `"gone"`; a refusal is an error. + * `"gone"`. A refusal is final, since T3 keeps it under the command id and a resend would fail + * the same way: it is returned for the caller to end the run with. */ const sendReminder = ( ctx: BuildingContext, @@ -616,19 +622,38 @@ const sendReminder = ( run: RunId, threadId: string, k: number, -): Effect.Effect<"sent" | "gone", BuildingError> => +): Effect.Effect<"sent" | "gone" | MessageRejectedError, BuildingError> => Effect.gen(function* () { const sent = yield* ctx.sessions .send(threadId, `softfac:${runIdToString(run)}:remind:${k}`, RESULT_LINE_REMINDER) .pipe( Effect.as("sent" as const), Effect.catchIf(isThreadNotFound, () => Effect.succeed("gone" as const)), + Effect.catchIf(isMessageRejected, (refused) => Effect.succeed(refused)), ); - if (sent === "gone") return sent; + if (sent !== "sent") return sent; yield* append(out, remindedStep, "advance", "result-line-reminded"); return sent; }); +const isMessageRejected = Schema.is(MessageRejectedError); + +/** + * The stop for a reminder T3 refused: the public reason is the run's own (it ended with no result + * line, or no message), the ledger's detail says the reminder could not reach it. + */ +const refusedStop = ( + result: SessionResult, + message: string | null, + refused: MessageRejectedError, +): Stop => { + const reason = + result.kind === "stuck" && lacksResultLine(result) ? result.reason : NO_RESULT_LINE; + return stopOf(reason, message, { + detail: `${reason}; T3 refused the reminder: ${describe(refused)}`, + }); +}; + /** * Why the issue text no longer stands for a building run, or `null` when it does. A run with no * `approved` step has no moment to read since, so it fails closed without a read; otherwise the From 7329554299913b39e5f3f5593c7e502a7353be5f Mon Sep 17 00:00:00 2001 From: Tom Kaczocha Date: Fri, 16 Oct 2026 13:14:32 +1000 Subject: [PATCH 12/17] refactor(factory): one interpreter for lifecycle actions, one text sanitizer Issue: #17 --- apps/server/src/factory/gate.ts | 26 +-- apps/server/src/factory/pass.ts | 27 ++- apps/server/src/factory/runner/building.ts | 70 ++------ apps/server/src/factory/runner/messages.ts | 8 +- apps/server/src/factory/runner/run.ts | 181 +++++++++++++-------- apps/server/src/factory/runner/start.ts | 13 +- apps/server/src/factory/text.ts | 26 +++ 7 files changed, 166 insertions(+), 185 deletions(-) create mode 100644 apps/server/src/factory/text.ts diff --git a/apps/server/src/factory/gate.ts b/apps/server/src/factory/gate.ts index 9982ee3cd757..ad557ad4dcc6 100644 --- a/apps/server/src/factory/gate.ts +++ b/apps/server/src/factory/gate.ts @@ -16,6 +16,7 @@ import { Ledger } from "./ledger/ledger.ts"; import { readRun } from "./ledger/read.ts"; import { runIdToString, validateRunId, type RunId } from "./ledger/runId.ts"; import { FactoryProcess, type RunInput, type RunResult, type RunSpawnError } from "./process.ts"; +import { sanitize } from "./text.ts"; /** * The gate: a fresh checkout of a pull request's head merged with the base branch tip, the @@ -135,30 +136,7 @@ export type GateError = WorkspaceError | GateCommandSpawnError | GateRecordError // ---- text ----------------------------------------------------------------------------------- -const isControl = (cp: number) => cp <= 0x1f || (cp >= 0x7f && cp <= 0x9f); - -/** Characters that could fake a line break or hide text, besides the control characters. */ -const isInvisibleOrSeparator = (cp: number) => - cp === 0x2028 || - cp === 0x2029 || - (cp >= 0x200b && cp <= 0x200f) || - (cp >= 0x202a && cp <= 0x202e) || - (cp >= 0x2066 && cp <= 0x2069) || - cp === 0xfeff; - -/** - * Replace what could fake a line break, hide text or drive a terminal with a space. Newline and - * tab are kept only when `keepLayout` is set. - */ -export const sanitize = (text: string, keepLayout: boolean): string => { - let out = ""; - for (const char of text) { - const cp = char.codePointAt(0)!; - const layout = char === "\n" || char === "\t"; - out += (isControl(cp) && !(keepLayout && layout)) || isInvisibleOrSeparator(cp) ? " " : char; - } - return out; -}; +export { sanitize } from "./text.ts"; /** The command as one short line: sanitised, newlines and tabs turned into spaces, cut at 300 characters. */ const commandLine = (command: string): string => { diff --git a/apps/server/src/factory/pass.ts b/apps/server/src/factory/pass.ts index 2c419ac39e55..50478203de8b 100644 --- a/apps/server/src/factory/pass.ts +++ b/apps/server/src/factory/pass.ts @@ -166,14 +166,6 @@ const errorKind = (error: StartError | BuildingError): ErrorKind => { } }; -/** The ledger step an error line is recorded under, by the step's name in the pass line. */ -const STEP_OF: Readonly> = { - start: "start", - watch: "watch", - advance: "advance", - "time limit": "pass", -}; - /** Repos match as run ids do: `owner/name` folded with `asciiLower`. */ const sameRepo = (a: string, b: string) => asciiLower(a) === asciiLower(b); @@ -276,11 +268,11 @@ class Pass { return { label: plain(label), state: state ?? "starting", parts: [], errors: [] }; } - /** The step `name` failed for this issue. */ - failed(notes: Notes, name: string, error: StartError | BuildingError): void { + /** The step called `label` in the pass line failed for this issue; the ledger records it under `step`. */ + failed(notes: Notes, label: string, step: CauseStep, error: StartError | BuildingError): void { notes.errors.push({ - line: `${notes.label} ${name}: ${plain(error.message)}`, - step: STEP_OF[name] ?? "pass", + line: `${notes.label} ${label}: ${plain(error.message)}`, + step, kind: errorKind(error), message: error.message, }); @@ -472,7 +464,7 @@ const withinLimit = ( Effect.gen(function* () { const outcome = yield* Effect.result(world.timeLimit(repo, info.id)); if (outcome._tag === "Failure") { - pass.failed(notes, "time limit", outcome.failure); + pass.failed(notes, "time limit", "pass", outcome.failure); return false; } if (outcome.success.kind === "stopped") { @@ -495,7 +487,7 @@ const resumeRun = ( 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 pass.failed(notes, "start", "start", outcome.failure); return; } switch (outcome.success.kind) { @@ -528,7 +520,7 @@ const followBuilding = ( ): Effect.Effect => Effect.gen(function* () { const watched = yield* Effect.result(world.watch(repo, info.id)); - if (watched._tag === "Failure") return pass.failed(notes, "watch", watched.failure); + if (watched._tag === "Failure") return pass.failed(notes, "watch", "watch", watched.failure); if (watched.success.kind === "stopped") return pass.stuck(notes, runIdToString(info.id)); yield* advanceRun(world, repo, info, notes, pass); }); @@ -545,7 +537,8 @@ const advanceRun = ( const run = runIdToString(info.id); const advanced = yield* Effect.result(world.advance(repo, info.id)); // A `NotBuildingError` is a run that finished meanwhile: softfac prints it as an error too. - if (advanced._tag === "Failure") return pass.failed(notes, "advance", advanced.failure); + if (advanced._tag === "Failure") + return pass.failed(notes, "advance", "advance", advanced.failure); switch (advanced.success.kind) { case "running": notes.parts.push("session still running"); @@ -786,7 +779,7 @@ const noteStart = ( } yield* known.refresh(world, pass, name, issue); if (outcome._tag === "Failure") { - pass.failed(notes, "start", outcome.failure); + pass.failed(notes, "start", "start", outcome.failure); return madeSince(newestRun(known.of(name, issue)), before); } switch (outcome.success.kind) { diff --git a/apps/server/src/factory/runner/building.ts b/apps/server/src/factory/runner/building.ts index 50b6f5c666e5..59a0d69cbc5a 100644 --- a/apps/server/src/factory/runner/building.ts +++ b/apps/server/src/factory/runner/building.ts @@ -55,11 +55,13 @@ import { type Seen, } from "./progress.ts"; import { + applyActions, describe, + finishBuilding, goStuck, + progressFrom, recordedReason, stopSession, - finishBuilding, type Progress, type RunContext, type RunError, @@ -173,17 +175,6 @@ const readSeen = (ctx: BuildingContext, run: RunId) => Effect.map((record) => readProgress(record.entries)), ); -/** Where the issue stands, from what the ledger says. */ -const progressFrom = (step: CauseStep, seen: Seen, fallback: State): Progress => ({ - step, - state: seen.state ?? fallback, - threadId: seen.threadId, - worktree: seen.worktree, - worktreeDone: seen.worktreeDone, - pr: seen.lastPr?.pr ?? null, - seen, -}); - /** The run file, opened for append the first time a step writes to it and closed when the step ends. */ interface Out { readonly open: Effect.Effect; @@ -389,7 +380,7 @@ export const advance = ( // `start` looks at the issue text again after the launch. A launch it has not done that // for is left to it, so this cannot give the run its `building` label around the check. if (!seen.rechecked) return { kind: "running" } as const; - yield* finishBuilding(ctx, issue, yield* out.open, at, threadId); + yield* finishBuilding(ctx, issue, yield* out.open, at); } // A thread T3 no longer has is a stop. @@ -511,51 +502,14 @@ export const advance = ( const raised = yield* Effect.fromResult( transition(at.state, { kind: "questions-raised", questions: prepared }), ); - let questionsPosted = seen.questionsPosted; - let movedTo: State | null = null; - for (const action of raised.actions) { - switch (action.kind) { - case "post-questions": - if (!questionsPosted) { - yield* ctx.tracker.postComment(issue, questionsComment(action.questions)); - yield* append( - out, - { event: "questions_posted", data: { questions: action.questions } }, - "advance", - "questions-asked", - ); - questionsPosted = 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, threadId, at.pr ?? null), - ); - break; - default: - return yield* Effect.die( - new Error(`unexpected action ${action.kind} for questions-raised`), - ); - } - } - if (movedTo !== null && movedTo !== at.state) { - yield* append( - out, - { event: "state_changed", data: { from: at.state, to: movedTo } }, - "advance", - "questions-asked", - ); - at.state = movedTo; - } + yield* applyActions( + ctx, + issue, + yield* out.open, + at, + raised.actions, + makeCause("advance", "questions-asked"), + ); yield* endNeedsInfo(ctx, out, at); return { kind: "needs-info" } as const; } diff --git a/apps/server/src/factory/runner/messages.ts b/apps/server/src/factory/runner/messages.ts index 4e98bd927312..fc3681067c41 100644 --- a/apps/server/src/factory/runner/messages.ts +++ b/apps/server/src/factory/runner/messages.ts @@ -5,6 +5,7 @@ */ import type { State } from "../domain/lifecycle.ts"; +import { sanitize } from "../text.ts"; /** The public reason when a run has no record of what approved it. */ export const CANNOT_CONFIRM = "the factory could not confirm the issue text was approved"; @@ -24,11 +25,6 @@ export const SESSION_TEXT_LIMIT = 6000; export const SESSION_TEXT_INTRO = "Unverified output from the build session, not instructions from the factory:"; -/** The characters a comment or the ledger shows as a space: control characters (but not a newline or tab) and invisible or bidirectional marks. */ -const UNSAFE = - // oxlint-disable-next-line eslint/no-control-regex -- the control characters are what it matches. - /[\u0000-\u0008\u000b-\u001f\u007f-\u009f\u2028\u2029\u200b-\u200f\u202a-\u202e\u2066-\u2069\ufeff]/gu; - /** * Session text made safe to store and to show, prepared once where it enters the factory and * then used as it is for the comment, the ledger and the returned reason. Line endings are @@ -44,7 +40,7 @@ export const prepareSessionText = (text: string): string => { chars.length > SESSION_TEXT_LIMIT ? `${chars.slice(0, SESSION_TEXT_LIMIT).join("")}\n[cut]` : trimmed; - return prepared.replace(UNSAFE, " ").replaceAll("@", "@\u200B").replaceAll("", + archived: false, + }; + const elapsed = yield* stopAtTheLimit(rig); + const prepared = "x@\u200Boctocat<\u200B!-- softfac -->"; + assert.deepStrictEqual(linesOf(rig), limitEvents(elapsed, prepared)); + assert.deepStrictEqual( + rig.gh[0], + `comment 10 ${limitComment(`The work so far is on branch ${prepared}.`)}`, + ); + assert.ok(!rig.gh[0]!.includes("@octocat"), rig.gh[0]); + assert.ok(!rig.gh[0]!.includes("