diff --git a/apps/server/src/factory/attention.test.ts b/apps/server/src/factory/attention.test.ts new file mode 100644 index 000000000000..dbff184e2890 --- /dev/null +++ b/apps/server/src/factory/attention.test.ts @@ -0,0 +1,375 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +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 { + Attention, + ATTENTION_FILE, + layer, + PASS_ERRORS_ITEM_ID, + type NewItem, +} from "./attention.ts"; +import { FactoryProcess, RunSpawnError, type RunInput, type RunResult } from "./process.ts"; +import * as Paths from "./t3/paths.ts"; + +const NOW = 1_791_380_000_000; + +const item = (overrides: Partial = {}): NewItem => ({ + id: "attention:rawphp__softfac__12__1:not-progressing", + kind: "attention", + repo: "rawphp/softfac", + issue: 12, + run: "rawphp__softfac__12__1", + message: "rawphp/softfac#12 run rawphp__softfac__12__1 is not progressing", + ...overrides, +}); + +const done = (exitCode: number | null, tail = ""): RunResult => ({ + exitCode, + timedOut: false, + tail, + stderrTail: "", + truncated: false, +}); + +/** A FactoryProcess that records each call and answers with `reply`. */ +const recorder = ( + reply: (input: RunInput) => Effect.Effect = () => + Effect.succeed(done(0)), +) => { + const calls: Array = []; + const process = FactoryProcess.of({ + run: (input) => { + calls.push(input); + return reply(input); + }, + }); + return { calls, process }; +}; + +const messageText = (message: unknown): string => { + const first = Array.isArray(message) ? message[0] : message; + return typeof first === "string" ? first : String(first); +}; + +/** The service over `dir`, as a restart builds it again. The clock reads `now.ms`. */ +const serviceOver = ( + dir: string, + process: FactoryProcess["Service"], + now: { ms: number } = { ms: NOW }, +) => + layer({ clock: () => now.ms }).pipe( + Layer.provide( + Layer.succeed(Paths.FactoryPaths, { + baseDir: dir, + worktreesDir: `${dir}/worktrees`, + factoryDir: dir, + machineConfigPath: `${dir}/config.yml`, + }), + ), + Layer.provide(Layer.succeed(FactoryProcess, process)), + ); + +/** Runs `body` over a fresh factory dir, with the warnings it logs collected. */ +const withDir = ( + body: (dir: string, warnings: Array) => Effect.Effect, +) => + Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const { join } = yield* Path.Path; + const base = yield* fs.makeTempDirectoryScoped({ prefix: "factory-attention-" }); + const dir = join(base, "factory"); + const warnings: Array = []; + const capture = Logger.make(({ message, logLevel }) => { + if (logLevel === "Warn") warnings.push(messageText(message)); + }); + return yield* body(dir, warnings).pipe( + Effect.provide(Logger.layer([capture], { mergeWithExisting: false })), + ); + }), + ).pipe(Effect.provide(NodeServices.layer)); + +const fileOf = (dir: string) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + return JSON.parse(yield* fs.readFileString(`${dir}/${ATTENTION_FILE}`)) as { + items: Array>; + }; + }); + +it.effect( + "raise keeps the item with the clock's first-seen time, and a second raise changes nothing", + () => + withDir((dir) => { + const now = { ms: NOW }; + const { calls, process } = recorder(); + return Effect.gen(function* () { + const attention = yield* Attention; + yield* attention.raise(item()); + now.ms = NOW + 60_000; + yield* attention.raise(item({ message: "changed" })); + const listed = yield* attention.list(); + assert.deepStrictEqual(listed, [{ ...item(), firstSeenMs: NOW }]); + assert.strictEqual(calls.length, 1); + }).pipe(Effect.provide(serviceOver(dir, process, now))); + }), +); + +it.effect("open items are kept in attention.json and a restart notifies none of them again", () => + withDir((dir) => { + const first = recorder(); + const second = recorder(); + return Effect.gen(function* () { + yield* Effect.gen(function* () { + yield* (yield* Attention).raise(item()); + }).pipe(Effect.provide(serviceOver(dir, first.process))); + const file = yield* fileOf(dir); + assert.strictEqual(file.items.length, 1); + assert.strictEqual(file.items[0]!["id"], item().id); + assert.strictEqual(file.items[0]!["first_seen"], "2026-10-07T13:33:20.000Z"); + + yield* Effect.gen(function* () { + const attention = yield* Attention; + assert.deepStrictEqual(yield* attention.list(), [{ ...item(), firstSeenMs: NOW }]); + yield* attention.raise(item()); + }).pipe(Effect.provide(serviceOver(dir, second.process, { ms: NOW + 1000 }))); + assert.strictEqual(first.calls.length, 1); + assert.strictEqual(second.calls.length, 0); + }); + }), +); + +it.effect("a file that is not valid is an empty list and one warning", () => + withDir((dir, warnings) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + yield* fs.makeDirectory(dir, { recursive: true }); + yield* fs.writeFileString(`${dir}/${ATTENTION_FILE}`, "{not json"); + const { process } = recorder(); + yield* Effect.gen(function* () { + assert.deepStrictEqual(yield* (yield* Attention).list(), []); + }).pipe(Effect.provide(serviceOver(dir, process))); + assert.strictEqual(warnings.length, 1); + }), + ), +); + +it.effect("entries of the file that are malformed are skipped", () => + withDir((dir) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + yield* fs.makeDirectory(dir, { recursive: true }); + yield* fs.writeFileString( + `${dir}/${ATTENTION_FILE}`, + JSON.stringify({ + items: [ + { + id: "a", + kind: "config", + repo: null, + issue: null, + run: null, + message: "m", + first_seen: "nope", + }, + { + id: "b", + kind: "config", + repo: null, + issue: null, + run: null, + message: "ok", + first_seen: "2026-10-07T13:33:20.000Z", + }, + "junk", + ], + }), + ); + const { process } = recorder(); + yield* Effect.gen(function* () { + const listed = yield* (yield* Attention).list(); + assert.deepStrictEqual( + listed.map((i) => i.id), + ["b"], + ); + }).pipe(Effect.provide(serviceOver(dir, process))); + }), + ), +); + +it.effect("the text goes to osascript as argv and never into the script", () => + withDir((dir) => { + const { calls, process } = recorder(); + const message = 'say "hi"\\\\ " & do shell script "rm -rf /" & "'; + return Effect.gen(function* () { + const attention = yield* Attention; + yield* attention.raise(item({ id: "a", repo: "a/b", issue: 7, run: null, message })); + yield* attention.raise( + item({ id: "b", kind: "config", repo: null, issue: null, run: null, message }), + ); + yield* attention.raise(item({ id: "c", repo: "a/b", issue: null, run: null, message })); + assert.strictEqual(calls.length, 3); + const [issueCall, machineCall, repoCall] = calls; + assert.strictEqual(issueCall!.command, "osascript"); + assert.strictEqual(issueCall!.args![0], "-e"); + const script = issueCall!.args![1]!; + assert.include(script, "on run argv"); + assert.notInclude(script, "do shell script"); + assert.notInclude(script, message); + assert.strictEqual(issueCall!.args![2], "a/b#7"); + assert.strictEqual(issueCall!.args![3], message); + assert.strictEqual(issueCall!.args!.length, 4); + assert.strictEqual(issueCall!.timeoutMs, 15_000); + assert.strictEqual(issueCall!.termGraceMs, 1_000); + assert.strictEqual(machineCall!.args![2], "factory"); + assert.strictEqual(repoCall!.args![2], "a/b"); + }).pipe(Effect.provide(serviceOver(dir, process))); + }), +); + +it.effect("notify off keeps the list and posts nothing", () => + withDir((dir) => { + const { calls, process } = recorder(); + return Effect.gen(function* () { + const attention = yield* Attention; + yield* attention.setNotify(false); + yield* attention.raise(item()); + assert.strictEqual(calls.length, 0); + assert.strictEqual((yield* attention.list()).length, 1); + }).pipe(Effect.provide(serviceOver(dir, process))); + }), +); + +it.effect("a missing osascript is logged once and no later item runs a process", () => + withDir((dir, warnings) => { + const { calls, process } = recorder(() => Effect.succeed(done(127, "osascript: not found"))); + return Effect.gen(function* () { + const attention = yield* Attention; + yield* attention.raise(item({ id: "a" })); + yield* attention.raise(item({ id: "b" })); + assert.strictEqual(calls.length, 1); + assert.strictEqual(warnings.length, 1); + assert.include(warnings[0]!, "osascript not found"); + assert.strictEqual((yield* attention.list()).length, 2); + }).pipe(Effect.provide(serviceOver(dir, process))); + }), +); + +it.effect("an osascript that times out is logged once and no later item runs a process", () => + withDir((dir, warnings) => { + const { calls, process } = recorder(() => + Effect.succeed({ + exitCode: null, + timedOut: true, + tail: "", + stderrTail: "", + truncated: false, + }), + ); + return Effect.gen(function* () { + const attention = yield* Attention; + yield* attention.raise(item({ id: "a" })); + yield* attention.raise(item({ id: "b" })); + assert.strictEqual(calls.length, 1); + assert.strictEqual(warnings.length, 1); + assert.include(warnings[0]!, "osascript did not answer in 15 seconds"); + assert.strictEqual((yield* attention.list()).length, 2); + }).pipe(Effect.provide(serviceOver(dir, process))); + }), +); + +it.effect("a failing notifier is logged and the raise still succeeds", () => + withDir((dir, warnings) => { + let n = 0; + const { calls, process } = recorder(() => { + n += 1; + return n === 1 + ? Effect.succeed(done(1, "execution error")) + : Effect.fail(new RunSpawnError({ shell: false, cause: new Error("no spawn") })); + }); + return Effect.gen(function* () { + const attention = yield* Attention; + yield* attention.raise(item({ id: "a" })); + yield* attention.raise(item({ id: "b" })); + assert.strictEqual(calls.length, 2); + assert.strictEqual(warnings.length, 2); + assert.include(warnings[0]!, "a"); + assert.strictEqual((yield* attention.list()).length, 2); + }).pipe(Effect.provide(serviceOver(dir, process))); + }), +); + +it.effect( + "clear removes the items that match and the file follows, so a stopped run's item is gone", + () => + withDir((dir) => { + const { process } = recorder(); + return Effect.gen(function* () { + const attention = yield* Attention; + yield* attention.raise(item({ id: "a", run: "run-1" })); + yield* attention.raise(item({ id: "b", run: "run-2" })); + yield* attention.clear((i) => i.run === "run-1"); + assert.deepStrictEqual( + (yield* attention.list()).map((i) => i.id), + ["b"], + ); + assert.deepStrictEqual( + (yield* fileOf(dir)).items.map((i) => i["id"]), + ["b"], + ); + yield* attention.clear(() => false); + assert.strictEqual((yield* attention.list()).length, 1); + }).pipe(Effect.provide(serviceOver(dir, process))); + }), +); + +it.effect("an item cleared and raised again is new, so it is notified again", () => + withDir((dir) => { + const { calls, process } = recorder(); + return Effect.gen(function* () { + const attention = yield* Attention; + yield* attention.raise(item()); + yield* attention.clear(() => true); + yield* attention.raise(item()); + assert.strictEqual(calls.length, 2); + }).pipe(Effect.provide(serviceOver(dir, process))); + }), +); + +it.effect( + "the third pass in a row that errors raises pass-errors, and a success clears it and resets the count", + () => + withDir((dir) => { + const { calls, process } = recorder(); + return Effect.gen(function* () { + const attention = yield* Attention; + const fail = (message: string) => attention.passEnded({ ok: false, message }); + yield* fail("one"); + yield* fail("two"); + assert.strictEqual((yield* attention.list()).length, 0); + yield* fail("three"); + const listed = yield* attention.list(); + assert.strictEqual(listed.length, 1); + assert.strictEqual(listed[0]!.id, PASS_ERRORS_ITEM_ID); + assert.strictEqual(listed[0]!.kind, "pass-errors"); + assert.strictEqual(listed[0]!.repo, null); + assert.include(listed[0]!.message, "3 times in a row"); + assert.include(listed[0]!.message, "three"); + assert.strictEqual(calls.length, 1); + // Later failures keep the one item and notify nothing. + yield* fail("four"); + assert.strictEqual(calls.length, 1); + + yield* attention.passEnded({ ok: true }); + assert.strictEqual((yield* attention.list()).length, 0); + yield* fail("five"); + yield* fail("six"); + assert.strictEqual((yield* attention.list()).length, 0); + }).pipe(Effect.provide(serviceOver(dir, process))); + }), +); diff --git a/apps/server/src/factory/attention.ts b/apps/server/src/factory/attention.ts new file mode 100644 index 000000000000..d705ac26ae89 --- /dev/null +++ b/apps/server/src/factory/attention.ts @@ -0,0 +1,299 @@ +// @effect-diagnostics nodeBuiltinImport:off - the open items are one file, written whole as the pause file is. +// @effect-diagnostics globalDate:off - an item is stamped with the wall clock, as the ledger is. +import * as NodeFS from "node:fs"; +import * as NodePath from "node:path"; + +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import { ensureDir } from "./ledger/appendFile.ts"; +import { formatTs, parseTs } from "./ledger/time.ts"; +import { FactoryProcess } from "./process.ts"; +import * as Paths from "./t3/paths.ts"; + +/** + * What a person has to look at. `attention` is an `attention` event of the pass; `stuck` and + * `needs-info` are runs that ended there; `config` is a machine config that did not load; + * `pass-errors` is a pass that errored several times in a row. + */ +export type AttentionKind = "attention" | "stuck" | "needs-info" | "config" | "pass-errors"; + +const KINDS: ReadonlyArray = ["attention", "stuck", "needs-info", "config", "pass-errors"]; + +export interface AttentionItem { + /** What the item is about, so raising it twice is one item. */ + readonly id: string; + readonly kind: AttentionKind; + /** The watched repo, or none for an item about the machine. */ + readonly repo: string | null; + readonly issue: number | null; + /** The run id, when the item is about one run. */ + readonly run: string | null; + readonly message: string; + /** When the item was first raised, in Unix epoch milliseconds. */ + readonly firstSeenMs: number; +} + +export type NewItem = Omit; + +/** How a pass ended, for the count of passes that error in a row. */ +export type PassOutcome = { readonly ok: true } | { readonly ok: false; readonly message: string }; + +/** + * The open attention items: what the factory needs a person for. An item is raised when its + * condition starts and cleared when the condition ends; a new item posts one macOS notification, + * and one that stays open is never notified again, because the open items are kept in + * `/attention.json` and read back at start. + */ +export class Attention extends Context.Service< + Attention, + { + readonly list: () => Effect.Effect>; + /** Opens the item unless one with its id is open; a new item is persisted, then notified once. Never fails: a file or notifier problem is logged. */ + readonly raise: (item: NewItem) => Effect.Effect; + /** Removes every open item `matches` is true for, and persists. */ + readonly clear: (matches: (item: AttentionItem) => boolean) => Effect.Effect; + /** Counts erroring passes in a row; the third raises `pass-errors`; a success clears it and resets the count. */ + readonly passEnded: (outcome: PassOutcome) => Effect.Effect; + /** Whether new items post a notification. Starts true (the machine config may not have loaded). */ + readonly setNotify: (on: boolean) => Effect.Effect; + } +>()("t3/factory/attention") {} + +export const ATTENTION_FILE = "attention.json"; +export const PASS_ERRORS_IN_A_ROW = 3; +export const CONFIG_ITEM_ID = "config"; +export const PASS_ERRORS_ITEM_ID = "pass-errors"; + +/** The exit code of the `sh` wrapper that found no such command. */ +const COMMAND_NOT_FOUND = 127; +const NOTIFY_TIMEOUT_MS = 15_000; +/** A notifier that ignores SIGTERM is killed after this, not after the process runner's default. */ +const NOTIFY_TERM_GRACE_MS = 1_000; + +/** + * The text arrives as argv, so nothing a repo, an issue title or a session wrote can become + * AppleScript source. + */ +const SCRIPT = `on run argv + display notification (item 2 of argv) with title (item 1 of argv) +end run`; + +const titleOf = (item: NewItem): string => { + if (item.repo === null) return "factory"; + return item.issue === null ? item.repo : `${item.repo}#${item.issue}`; +}; + +const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); + +const stringOrNull = (value: unknown): string | null | undefined => + value === null || typeof value === "string" ? value : undefined; + +/** One entry of the file, or `null` when it is malformed. */ +const itemFromFile = (entry: unknown): AttentionItem | null => { + if (!isRecord(entry)) return null; + const { id, kind, repo, issue, run, message } = entry; + const firstSeenMs = typeof entry["first_seen"] === "string" ? parseTs(entry["first_seen"]) : null; + if (typeof id !== "string" || typeof kind !== "string" || !KINDS.includes(kind)) return null; + if (typeof message !== "string" || firstSeenMs === null) return null; + const repoOrNull = stringOrNull(repo); + const runOrNull = stringOrNull(run); + if (repoOrNull === undefined || runOrNull === undefined) return null; + if (issue !== null && typeof issue !== "number") return null; + return { + id, + kind: kind as AttentionKind, + repo: repoOrNull, + issue, + run: runOrNull, + message, + firstSeenMs, + }; +}; + +type Loaded = + | { readonly items: ReadonlyArray } + | { readonly items: ReadonlyArray; readonly warning: string }; + +/** The open items on disk. A missing file is none; a file that is not the expected shape is none, with a warning. */ +const readFile = (path: string): Loaded => { + let text: string; + try { + text = NodeFS.readFileSync(path, "utf8"); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return { items: [] }; + return { items: [], warning: `cannot read ${path}: ${String(error)}` }; + } + let parsed: unknown; + try { + parsed = JSON.parse(text); + } catch { + return { items: [], warning: `${path} is not valid JSON, so no attention item is kept` }; + } + if (!isRecord(parsed) || !Array.isArray(parsed["items"])) { + return { items: [], warning: `${path} has no items list, so no attention item is kept` }; + } + return { + items: parsed["items"].flatMap((entry) => { + const found = itemFromFile(entry); + return found === null ? [] : [found]; + }), + }; +}; + +/** Writes the items whole or not at all, private to the owner, as the pause file is. */ +const writeFile = (dir: string, items: ReadonlyArray): void => { + ensureDir(dir); + const path = NodePath.join(dir, ATTENTION_FILE); + const temp = `${path}.${process.pid}.tmp`; + const body = JSON.stringify({ + items: items.map((item) => ({ + id: item.id, + kind: item.kind, + repo: item.repo, + issue: item.issue, + run: item.run, + message: item.message, + first_seen: formatTs(item.firstSeenMs), + })), + }); + const fd = NodeFS.openSync( + temp, + NodeFS.constants.O_WRONLY | NodeFS.constants.O_CREAT | NodeFS.constants.O_TRUNC, + 0o600, + ); + let open = true; + try { + NodeFS.writeFileSync(fd, `${body}\n`); + NodeFS.fsyncSync(fd); + open = false; + NodeFS.closeSync(fd); + NodeFS.renameSync(temp, path); + } catch (error) { + // No failure leaves the temp file behind. + if (open) { + try { + NodeFS.closeSync(fd); + } catch { + // The write error is the one to report. + } + } + NodeFS.rmSync(temp, { force: true }); + throw error; + } +}; + +const make = (options: { readonly clock?: () => number }) => + Effect.gen(function* () { + const { factoryDir } = yield* Paths.FactoryPaths; + const runner = yield* FactoryProcess; + const clock = options.clock ?? Date.now; + const path = NodePath.join(factoryDir, ATTENTION_FILE); + + const loaded = readFile(path); + if ("warning" in loaded) yield* Effect.logWarning(loaded.warning); + const items = new Map(loaded.items.map((item) => [item.id, item])); + let notifying = true; + let notifierDown = false; + let failedPasses = 0; + + const persist = Effect.suspend(() => { + try { + writeFile(factoryDir, [...items.values()]); + return Effect.void; + } catch (error) { + return Effect.logWarning( + `cannot write ${path}, so open attention items may be notified again: ${String(error)}`, + ); + } + }); + + const post = (item: NewItem) => + Effect.gen(function* () { + if (!notifying || notifierDown) return; + const result = yield* Effect.result( + runner.run({ + command: "osascript", + args: ["-e", SCRIPT, titleOf(item), item.message], + timeoutMs: NOTIFY_TIMEOUT_MS, + termGraceMs: NOTIFY_TERM_GRACE_MS, + }), + ); + if (result._tag === "Failure") { + return yield* Effect.logWarning( + `cannot notify attention item ${item.id}: ${String(result.failure.cause)}`, + ); + } + const ran = result.success; + if (ran.exitCode === COMMAND_NOT_FOUND) { + notifierDown = true; + return yield* Effect.logWarning( + "osascript not found, so attention items are not notified on this machine", + ); + } + if (ran.timedOut) { + // A hung notifier would cost its timeout again on every new item. + notifierDown = true; + return yield* Effect.logWarning( + "osascript did not answer in 15 seconds, so attention items are not notified again until the server restarts", + ); + } + if (ran.exitCode !== 0) { + return yield* Effect.logWarning( + `cannot notify attention item ${item.id}: exit ${ran.exitCode}: ${ran.tail.trim()}`, + ); + } + }); + + const raise = (item: NewItem): Effect.Effect => + Effect.gen(function* () { + if (items.has(item.id)) return; + items.set(item.id, { ...item, firstSeenMs: clock() }); + // Saved before it is posted, so a crash between the two cannot notify the item twice. + yield* persist; + yield* post(item); + }); + + const clear = (matches: (item: AttentionItem) => boolean): Effect.Effect => + Effect.gen(function* () { + const gone = [...items.values()].filter(matches); + if (gone.length === 0) return; + for (const item of gone) items.delete(item.id); + yield* persist; + }); + + const passEnded = (outcome: PassOutcome): Effect.Effect => + Effect.gen(function* () { + if (outcome.ok) { + failedPasses = 0; + return yield* clear((item) => item.id === PASS_ERRORS_ITEM_ID); + } + failedPasses += 1; + if (failedPasses < PASS_ERRORS_IN_A_ROW) return; + yield* raise({ + id: PASS_ERRORS_ITEM_ID, + kind: "pass-errors", + repo: null, + issue: null, + run: null, + message: `the pass has failed ${PASS_ERRORS_IN_A_ROW} times in a row, last: ${outcome.message}`, + }); + }); + + return Attention.of({ + list: () => Effect.sync(() => [...items.values()]), + raise, + clear, + passEnded, + setNotify: (on) => + Effect.sync(() => { + notifying = on; + }), + }); + }); + +/** Needs FactoryPaths and FactoryProcess. `clock` is for a test; it reads the wall clock by default. */ +export const layer = (options: { readonly clock?: () => number } = {}) => + Layer.effect(Attention, make(options)); diff --git a/apps/server/src/factory/config.test.ts b/apps/server/src/factory/config.test.ts index e4c6d58874fb..4eb18f1b1c7a 100644 --- a/apps/server/src/factory/config.test.ts +++ b/apps/server/src/factory/config.test.ts @@ -427,10 +427,21 @@ describe("machine config", () => { maxConcurrentGates: 2, modelInstance: "claudeAgent", model: "claude-sonnet-5-5", + notify: true, repos: [{ repo: "rawphp/softfac", path: "/Users/me/EA/projects/softfac" }], }); }); + it("machine notify can be turned off", () => { + expect(machine(edit(MACHINE, (t) => set(t, "notify", "false"))).notify).toBe(false); + }); + + it("machine notify that is not a boolean is rejected", () => { + const errs = machineErr(edit(MACHINE, (t) => set(t, "notify", '"no"'))); + assertOne(errs, "notify", MFILE); + expect(errs[0]!.message).toBe("must be true or false"); + }); + it("machine every key can be set", () => { const t = edit(MACHINE, (tree) => { set(tree, "poll_seconds", "10"); diff --git a/apps/server/src/factory/config.ts b/apps/server/src/factory/config.ts index 14721f638c81..1f72ce01b7c1 100644 --- a/apps/server/src/factory/config.ts +++ b/apps/server/src/factory/config.ts @@ -87,6 +87,8 @@ export interface MachineConfig { readonly maxConcurrentGates: number; readonly modelInstance: string; readonly model: string; + /** Whether a new attention item posts a macOS notification. The list is kept either way. */ + readonly notify: boolean; readonly repos: ReadonlyArray; } @@ -162,6 +164,21 @@ const makeChecker = (file: string) => { return uint(parent[key], join(path, key), min); }; + /** A boolean under `key` of `parent`; an absent key takes `fallback`. */ + const optionalBoolean = ( + parent: Mapping | undefined, + path: string, + key: string, + fallback: boolean, + ): boolean | undefined => { + if (parent === undefined) return undefined; + if (!Object.hasOwn(parent, key)) return fallback; + const value = parent[key]; + if (typeof value === "boolean") return value; + err(join(path, key), "must be true or false"); + return undefined; + }; + const optionalNonEmptyString = ( parent: Mapping | undefined, path: string, @@ -241,6 +258,7 @@ const makeChecker = (file: string) => { uint, optionalUint, optionalNonEmptyString, + optionalBoolean, stringList, autonomy, unknownKeys, @@ -336,6 +354,7 @@ const MACHINE_KEYS = [ "max_concurrent_gates", "model_instance", "model", + "notify", "repos", ] as const; @@ -397,6 +416,7 @@ const machineFrom = (c: Checker, root: Mapping): MachineConfig | undefined => { DEFAULT_MODEL_INSTANCE, ); const model = c.nonEmptyString(c.child(root, "", "model"), "model"); + const notify = c.optionalBoolean(root, "", "notify", true); const repos = reposFrom(c, root); checkUnknownKeys(c, root); @@ -407,6 +427,7 @@ const machineFrom = (c: Checker, root: Mapping): MachineConfig | undefined => { maxConcurrentGates === undefined || modelInstance === undefined || model === undefined || + notify === undefined || repos === undefined ) { return undefined; @@ -418,6 +439,7 @@ const machineFrom = (c: Checker, root: Mapping): MachineConfig | undefined => { maxConcurrentGates, modelInstance, model, + notify, repos, }; }; diff --git a/apps/server/src/factory/index.test.ts b/apps/server/src/factory/index.test.ts index 318614629bdb..679e216eb398 100644 --- a/apps/server/src/factory/index.test.ts +++ b/apps/server/src/factory/index.test.ts @@ -14,6 +14,7 @@ import * as Stream from "effect/Stream"; import { describe } from "vite-plus/test"; +import { Attention } from "./attention.ts"; import { timeLimitOnlyProjectConfig, type MachineConfig } from "./config.ts"; import type { State } from "./domain/lifecycle.ts"; import { @@ -150,13 +151,185 @@ it.effect("logs factory started once for a valid machine file, then runs the fir it.effect("logs each config error and does not start for an invalid machine file", () => Effect.gen(function* () { + // The notification for the config item goes to a runner that accepts it, so it logs nothing. const { logs, machineConfigPath } = yield* logsFor( VALID.replace("model: claude-sonnet-5-5\n", ""), + () => Effect.void, + Layer.succeed(FactoryProcess, recordingProcess().process), ); assert.deepStrictEqual(logs, [`${machineConfigPath}: model: is missing`]); }), ); +// ---- attention ---- + +/** A process runner that records each call: osascript succeeds and every other command fails to start. */ +const recordingProcess = () => { + const calls: Array = []; + const process = FactoryProcess.of({ + run: (input) => { + calls.push(input); + return input.command === "osascript" + ? Effect.succeed({ + exitCode: 0, + timedOut: false, + tail: "", + stderrTail: "", + truncated: false, + } satisfies RunResult) + : Effect.fail(new RunSpawnError({ shell: false, cause: new Error("gh not found") })); + }, + }); + return { calls, process }; +}; + +describe("attention", () => { + /** A base dir whose factory dir holds `config`, and the paths layer over it. */ + const dirWith = (config: string, seed?: string) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const { join } = yield* Path.Path; + const baseDir = yield* fs.makeTempDirectoryScoped({ prefix: "factory-attention-index-" }); + const factoryDir = join(baseDir, "factory"); + yield* fs.makeDirectory(factoryDir); + const machineConfigPath = join(factoryDir, "config.yml"); + yield* fs.writeFileString(machineConfigPath, config); + if (seed !== undefined) yield* fs.writeFileString(join(factoryDir, "attention.json"), seed); + return { baseDir, factoryDir, machineConfigPath, worktreesDir: join(baseDir, "worktrees") }; + }); + + /** Builds the runtime over `dirs`, and reads the open items while it runs. */ + const itemsAfterStart = ( + dirs: Effect.Success>, + process: FactoryProcess["Service"], + ) => + Effect.scoped( + Layer.build( + FactoryRuntime.pipe( + Layer.provide(Layer.succeed(Paths.FactoryPaths, dirs)), + Layer.provide(Layer.succeed(FactoryProcess, process)), + Layer.provide(noSessions), + Layer.provide(quietEvents), + ), + ).pipe(Effect.flatMap((context) => Context.get(context, Attention).list())), + ); + + const INVALID = VALID.replace("model: claude-sonnet-5-5\n", ""); + + it.effect("a machine config that does not load raises one config item and one notification", () => + Effect.scoped( + Effect.gen(function* () { + const dirs = yield* dirWith(INVALID); + const { calls, process } = recordingProcess(); + const items = yield* itemsAfterStart(dirs, process); + assert.deepStrictEqual( + items.map((item) => [item.id, item.kind, item.repo]), + [["config", "config", null]], + ); + assert.include(items[0]!.message, "model: is missing"); + assert.strictEqual(calls.length, 1); + assert.strictEqual(calls[0]!.command, "osascript"); + assert.strictEqual(calls[0]!.args![0], "-e"); + assert.include(calls[0]!.args![1]!, "on run argv"); + assert.strictEqual(calls[0]!.args![2], "factory"); + assert.strictEqual(calls[0]!.args![3], items[0]!.message); + }), + ).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect("a config that loads clears the config item an earlier start left", () => + Effect.scoped( + Effect.gen(function* () { + const seed = JSON.stringify({ + items: [ + { + id: "config", + kind: "config", + repo: null, + issue: null, + run: null, + message: "machine config not loaded: x", + first_seen: "2026-10-07T13:33:20.000Z", + }, + ], + }); + const dirs = yield* dirWith(VALID, seed); + const { calls, process } = recordingProcess(); + assert.deepStrictEqual(yield* itemsAfterStart(dirs, process), []); + assert.strictEqual(calls.filter((call) => call.command === "osascript").length, 0); + }), + ).pipe(Effect.provide(NodeServices.layer)), + ); + + it.effect("a restart over the same files notifies nothing again and still lists the item", () => + Effect.scoped( + Effect.gen(function* () { + const dirs = yield* dirWith(INVALID); + const first = recordingProcess(); + const second = recordingProcess(); + yield* itemsAfterStart(dirs, first.process); + const items = yield* itemsAfterStart(dirs, second.process); + assert.strictEqual(first.calls.length, 1); + assert.strictEqual(second.calls.length, 0); + assert.deepStrictEqual( + items.map((item) => item.id), + ["config"], + ); + }), + ).pipe(Effect.provide(NodeServices.layer)), + ); + + /** Wakes the loop until a `pass-errors` item is open, for a config whose every pass errors (no gh). */ + const untilPassErrors = (config: string) => + Effect.scoped( + Effect.gen(function* () { + const dirs = yield* dirWith(config); + const { calls, process } = recordingProcess(); + const context = yield* Layer.build( + FactoryRuntime.pipe( + Layer.provide(Layer.succeed(Paths.FactoryPaths, dirs)), + Layer.provide(Layer.succeed(FactoryProcess, process)), + Layer.provide(noSessions), + Layer.provide(quietEvents), + ), + ); + const loop = Context.get(context, FactoryLoop); + const attention = Context.get(context, Attention); + const open = () => + attention + .list() + .pipe(Effect.map((items) => items.filter((item) => item.id === "pass-errors"))); + // Each wake is one more pass, and each wait is bounded so a pass that never ends fails + // the test instead of hanging it. + const reached = yield* Effect.gen(function* () { + while ((yield* open()).length === 0) { + yield* loop.wake; + yield* Effect.timeoutOption(loop.awaitPass, "1 second"); + } + }).pipe(Effect.timeoutOption("30 seconds"), Effect.map(Option.isSome)); + assert.isTrue(reached, "no pass-errors item was raised"); + return { items: yield* open(), osascript: calls.filter((c) => c.command === "osascript") }; + }), + ).pipe(Effect.provide(NodeServices.layer)); + + it.effect("passes that keep erroring raise a pass-errors item and one notification", () => + Effect.gen(function* () { + const { items, osascript } = yield* untilPassErrors(VALID); + assert.include(items[0]!.message, "3 times in a row"); + assert.include(items[0]!.message, "a/b: project config not loaded"); + assert.strictEqual(osascript.length, 1); + }), + ); + + it.effect("notify: false keeps the item and posts nothing", () => + Effect.gen(function* () { + const { items, osascript } = yield* untilPassErrors(`${VALID}notify: false\n`); + assert.strictEqual(items.length, 1); + assert.strictEqual(osascript.length, 0); + }), + ); +}); + it.effect("clears the gate locks a dead server left and nothing else under gates", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; @@ -405,11 +578,15 @@ describe("liveWorld", () => { maxConcurrentGates: 1, modelInstance: "claudeAgent", model: "claude-sonnet-5-5", + notify: true, repos: [{ repo: REPO, path: "/p" }], }; const worldOf = (process: FactoryProcess["Service"]) => - liveWorld(machine, "/factory-test/data", process, stubSessions, () => NOW, "mac-mini"); + liveWorld(machine, "/factory-test/data", process, stubSessions, () => NOW, "mac-mini", { + raise: () => Effect.void, + clear: () => Effect.void, + }); const repo: Repo = { watched: machine.repos[0]!, diff --git a/apps/server/src/factory/index.ts b/apps/server/src/factory/index.ts index b01e37e3a7e2..46288e33847c 100644 --- a/apps/server/src/factory/index.ts +++ b/apps/server/src/factory/index.ts @@ -3,15 +3,19 @@ import * as NodeFS from "node:fs"; import * as NodeOS from "node:os"; +import * as Cause from "effect/Cause"; 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 Exit from "effect/Exit"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import * as Result from "effect/Result"; import * as Stream from "effect/Stream"; +import * as AttentionService from "./attention.ts"; +import { Attention, CONFIG_ITEM_ID, type PassOutcome } from "./attention.ts"; import { loadHandPause } from "./claims.ts"; import { formatConfigError, @@ -31,7 +35,7 @@ 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 { runPass, type PassError, type PassWorld, type Repo, type Report } from "./pass.ts"; import * as Process from "./process.ts"; import { FactoryProcess } from "./process.ts"; import { readProgress } from "./runner/progress.ts"; @@ -123,6 +127,7 @@ export const liveWorld = ( sessions: T3Sessions.FactorySessions["Service"], clock: () => number, host: string, + attention: Pick, ): PassWorld => { // Built once per process, so judgements last as long as the factory runs. const claimMemo = new ClaimMemo(); @@ -241,6 +246,7 @@ export const liveWorld = ( }), handPause: () => loadHandPause(dataDir), nowMs: clock, + attention, record: (event: MachineEvent) => ledger(() => { const log = MachineLog.open(dataDir, { clock }); @@ -325,6 +331,22 @@ const wakeOnOwnedThreads = ( ); }); +/** How a pass ended, for the count of passes that error in a row. */ +const outcomeOf = (exit: Exit.Exit, lastError: () => string | null): PassOutcome => { + if (Exit.isSuccess(exit)) { + const errors = exit.value.errors; + return errors === 0 + ? { ok: true } + : { ok: false, message: lastError() ?? `${errors} error lines` }; + } + return { + ok: false, + message: Cause.hasInterruptsOnly(exit.cause) + ? "the pass was stopped before it finished" + : `the pass died: ${String(Cause.squash(exit.cause))}`, + }; +}; + /** 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. */ @@ -334,6 +356,7 @@ export interface RuntimeOptions { const startFactory = (options: RuntimeOptions) => Effect.gen(function* () { const { factoryDir, machineConfigPath } = yield* Paths.FactoryPaths; + const attention = yield* Attention; // 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)`); @@ -342,17 +365,35 @@ const startFactory = (options: RuntimeOptions) => for (const error of loaded.failure.errors) { yield* Effect.logError(formatConfigError(error)); } + yield* attention.raise({ + id: CONFIG_ITEM_ID, + kind: "config", + repo: null, + issue: null, + run: null, + message: `machine config not loaded: ${loaded.failure.errors.map(formatConfigError).join("; ")}`, + }); return idle; } + yield* attention.clear((item) => item.id === CONFIG_ITEM_ID); if (Option.isNone(loaded.success)) { yield* Effect.logInfo(`no factory config at ${machineConfigPath}`); return idle; } const machine = loaded.success.value; + yield* attention.setNotify(machine.notify); 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 world = liveWorld( + machine, + factoryDir, + process, + sessions, + () => Date.now(), + hostName(), + attention, + ); // The threads of the unfinished runs, read at start and after every pass: a pass is what // launches and finishes sessions, so between passes the set does not change. const owned = yield* Ref.make>(new Set()); @@ -361,7 +402,11 @@ const startFactory = (options: RuntimeOptions) => // 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))); + let lastError: string | null = null; + return yield* runPass(world, machine, (line) => { + if (line.startsWith("error: ")) lastError = line; + log(Effect.logInfo(line)); + }).pipe(Effect.onExit((exit) => attention.passEnded(outcomeOf(exit, () => lastError)))); }).pipe(Effect.ensuring(refreshOwned(factoryDir, owned))); const loop = yield* runLoop(onePass, machine.pollSeconds); yield* Effect.forkScoped(wakeOnOwnedThreads(owned, loop, options.restartDelay ?? "5 seconds")); @@ -369,12 +414,14 @@ const startFactory = (options: RuntimeOptions) => }); /** - * Needs FactoryPaths, FactoryProcess, FactorySessions, FactoryEvents and FileSystem; what - * index.test.ts builds with a test FactoryPaths, a fake FactoryProcess, a stub FactorySessions + * Gives FactoryLoop and Attention. 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 = (options: RuntimeOptions = {}) => - Layer.effect(FactoryLoop, startFactory(options)); + Layer.effect(FactoryLoop, startFactory(options)).pipe( + Layer.provideMerge(AttentionService.layer()), + ); export const FactoryRuntime = factoryRuntime(); diff --git a/apps/server/src/factory/pass.test.ts b/apps/server/src/factory/pass.test.ts index bd72097d27d2..db3e4f9fadf6 100644 --- a/apps/server/src/factory/pass.test.ts +++ b/apps/server/src/factory/pass.test.ts @@ -8,6 +8,7 @@ import { assert, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; import { describe } from "vite-plus/test"; +import type { Attention, AttentionItem } from "./attention.ts"; import type { HandPause } from "./claims.ts"; import { timeLimitOnlyProjectConfig, type MachineConfig, type ProjectConfig } from "./config.ts"; import type { State } from "./domain/lifecycle.ts"; @@ -50,6 +51,7 @@ const machine = (repos: ReadonlyArray, max: number): MachineConfig => ({ maxConcurrentGates: 2, modelInstance: "claudeAgent", model: "claude-sonnet-5-5", + notify: true, repos: repos.map((repo) => ({ repo, path: `/p/${repo.replace("/", "-")}` })), }); @@ -109,6 +111,21 @@ class Fake implements PassWorld { calls: Array = []; events: Array = []; + /** The open attention items, as the service keeps them: one per id. */ + items: Array = []; + attention: Pick = { + raise: (item) => + Effect.sync(() => { + if (!this.items.some((open) => open.id === item.id)) { + this.items.push({ ...item, firstSeenMs: this.now }); + } + }), + clear: (matches) => + Effect.sync(() => { + this.items = this.items.filter((item) => !matches(item)); + }), + }; + runIn(issue: number, state: State | null, repo = REPO, session = false) { this.runs.push({ id: rid(issue, repo), @@ -2187,3 +2204,169 @@ describe("following a building run", () => { }), ); }); + +// ---- attention items ---- + +describe("attention items", () => { + const open = (overrides: Partial): AttentionItem => ({ + id: "seed", + kind: "attention", + repo: REPO, + issue: 12, + run: null, + message: "seeded", + firstSeenMs: T0, + ...overrides, + }); + const ids = (fake: Fake) => fake.items.map((item) => item.id); + const RUN_40 = runIdToString(rid(40)); + + it.effect("a run that goes stuck raises a stuck item", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(40, BUILDING, REPO, true); + fake.advances[`${REPO}#40`] = [{ kind: "stuck", reason: "r" }]; + yield* pass(fake, machine([REPO], 1)); + assert.deepEqual( + fake.items.map(({ firstSeenMs: _, ...item }) => item), + [ + { + id: `run:${RUN_40}:stuck`, + kind: "stuck", + repo: REPO, + issue: 40, + run: RUN_40, + message: `${REPO}#40 stuck, the reason is in the ledger of run ${RUN_40}`, + }, + ], + ); + }), + ); + + it.effect("a run that needs info raises a needs-info item", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(42, "needs-info", REPO, true); + fake.advances[`${REPO}#42`] = [{ kind: "needs-info" }]; + yield* pass(fake, machine([REPO], 5)); + const run = runIdToString(rid(42)); + assert.deepEqual( + fake.items.map(({ firstSeenMs: _, ...item }) => item), + [ + { + id: `run:${run}:needs-info`, + kind: "needs-info", + repo: REPO, + issue: 42, + run, + message: `${REPO}#42 needs-info, the session asked questions on the issue`, + }, + ], + ); + }), + ); + + it.effect("an ATTENTION line raises an attention item with the line's text", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(12, "triage", REPO, true); + fake.ready[REPO] = [12]; + fake.pushStart(`${REPO}#12`, { kind: "skipped", reason: "ready again", moved: "ready" }); + const { lines } = yield* pass(fake, machine([REPO], 1)); + const line = runLine(lines, "ATTENTION: "); + const run = runIdToString(rid(12)); + assert.deepEqual( + fake.items.map(({ firstSeenMs: _, ...item }) => item), + [ + { + id: `attention:${run}:not-progressing`, + kind: "attention", + repo: REPO, + issue: 12, + run, + message: line.slice("ATTENTION: ".length), + }, + ], + ); + }), + ); + + it.effect( + "a claim whose launch fails clears the issue's earlier items and keeps the new one", + () => + Effect.gen(function* () { + const fake = new Fake(); + const oldRun = runIdToString(newRunId(REPO, "12", 1)); + const newRun = newRunId(REPO, "12", 2); + fake.items = [ + open({ id: `run:${oldRun}:stuck`, kind: "stuck", issue: 12, run: oldRun }), + open({ id: "other", kind: "needs-info", issue: 13 }), + ]; + fake.ready[REPO] = [12]; + fake.pushStart(`${REPO}#12`, { kind: "stuck", run: newRun, reason: "launch failed" }); + yield* pass(fake, machine([REPO], 1)); + assert.deepEqual(ids(fake), ["other", `run:${runIdToString(newRun)}:stuck`]); + }), + ); + + it.effect("claiming an issue again clears every item of that issue and no other", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.items = [ + open({ id: "a", kind: "attention", run: "r" }), + open({ id: "b", kind: "stuck" }), + open({ id: "c", kind: "needs-info" }), + open({ id: "d", kind: "config" }), + open({ id: "e", kind: "pass-errors" }), + open({ id: "other", issue: 13 }), + open({ id: "elsewhere", repo: OTHER }), + ]; + fake.ready[REPO] = [12]; + fake.pushStart(`${REPO}#12`, started(12)); + yield* pass(fake, machine([REPO, OTHER], 5)); + assert.deepEqual(ids(fake), ["other", "elsewhere"]); + }), + ); + + it.effect( + "an item of a repo that left the config is cleared at the pass start, a machine item stays", + () => + Effect.gen(function* () { + const fake = new Fake(); + fake.items = [ + open({ id: "gone", repo: "rawphp/gone" }), + open({ id: "kept", repo: REPO, issue: 99 }), + open({ id: "machine", kind: "config", repo: null, issue: null }), + ]; + yield* pass(fake, machine([REPO], 5)); + assert.deepEqual(ids(fake), ["kept", "machine"]); + }), + ); + + it.effect( + "an attention item of a run the ledger no longer lists is cleared at the pass end, a stuck item stays", + () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runIn(40, BUILDING, REPO, true); + fake.items = [ + open({ id: "live", kind: "attention", issue: 40, run: RUN_40 }), + open({ id: "finished", kind: "attention", issue: 41, run: runIdToString(rid(41)) }), + open({ id: "stuck", kind: "stuck", issue: 41, run: runIdToString(rid(41)) }), + open({ id: "needs-info", kind: "needs-info", issue: 41, run: runIdToString(rid(41)) }), + ]; + yield* pass(fake, machine([REPO], 5)); + assert.deepEqual(ids(fake), ["live", "stuck", "needs-info"]); + }), + ); + + it.effect("no item is cleared at the pass end when the ledger cannot be read", () => + Effect.gen(function* () { + const fake = new Fake(); + fake.runsError = true; + fake.items = [open({ id: "a", kind: "attention", run: RUN_40 })]; + yield* pass(fake, machine([REPO], 5)); + assert.deepEqual(ids(fake), ["a"]); + }), + ); +}); diff --git a/apps/server/src/factory/pass.ts b/apps/server/src/factory/pass.ts index d57caf132249..d9d4efb30607 100644 --- a/apps/server/src/factory/pass.ts +++ b/apps/server/src/factory/pass.ts @@ -10,6 +10,7 @@ */ import * as Effect from "effect/Effect"; +import type { Attention, NewItem } from "./attention.ts"; import type { HandPause } from "./claims.ts"; import type { MachineConfig, ProjectConfig, WatchedRepo } from "./config.ts"; import { stateLabel, type State } from "./domain/lifecycle.ts"; @@ -113,6 +114,8 @@ export interface PassWorld { readonly nowMs: () => number; /** Writes one event to the machine log. */ readonly record: (event: MachineEvent) => Effect.Effect; + /** The open attention items: what the pass raises when it needs a person, and what it clears. */ + readonly attention: Pick; } /** The counts the pass prints in its last line. */ @@ -125,10 +128,14 @@ export interface Report { /** What the pass has to say about one issue, collected as its steps go and printed at the end. */ interface Notes { + readonly repo: string; + readonly issue: number; readonly label: string; readonly state: string; readonly parts: Array; readonly errors: Array; + /** The attention items to raise when the notes are flushed. */ + readonly raises: Array; } /** An error, as printed and as recorded. */ @@ -232,13 +239,24 @@ class Pass { } /** - * A line a person has to act on about `run`, and the machine log's `attention` event for it - * with the ledger.md `reason` code. + * A line a person has to act on about `run`, the machine log's `attention` event for it with + * the ledger.md `reason` code, and an attention item whose message is the line. */ - attention(text: string, run: string, reason: string): Effect.Effect { + attention(notes: Notes, text: string, run: string, reason: string): Effect.Effect { this.report.attention += 1; this.out(`ATTENTION: ${text}`); - return this.record({ event: "attention", data: { run, reason } }); + return this.record({ event: "attention", data: { run, reason } }).pipe( + Effect.andThen( + this.world.attention.raise({ + id: `attention:${run}:${reason}`, + kind: "attention", + repo: notes.repo, + issue: notes.issue, + run, + message: text, + }), + ), + ); } /** @@ -277,8 +295,16 @@ class Pass { .pipe(Effect.catch(() => toMachineLog(runIdToString(run)))); } - notes(label: string, state: State | null): Notes { - return { label: plain(label), state: state ?? "starting", parts: [], errors: [] }; + notes(repo: string, issue: number, state: State | null): Notes { + return { + repo, + issue, + label: `${plain(repo)}${mention(issue)}`, + state: state ?? "starting", + parts: [], + errors: [], + raises: [], + }; } /** The step called `label` in the pass line failed for this issue; the ledger records it under `step`. */ @@ -300,13 +326,27 @@ class Pass { /** The run ended in `stuck`. The reason is not printed: it can hold text the session or GitHub wrote. */ stuck(notes: Notes, run: string): void { notes.parts.push(`stuck, the reason is in the ledger of run ${run}`); + notes.raises.push({ + id: `run:${run}:stuck`, + kind: "stuck", + repo: notes.repo, + issue: notes.issue, + run, + message: `${notes.label} stuck, the reason is in the ledger of run ${run}`, + }); } flush(notes: Notes, run: RunId | null): Effect.Effect { if (notes.parts.length > 0) { this.line(`${notes.label} ${notes.state}: ${notes.parts.join(", ")}`); } - return Effect.forEach(notes.errors, (error) => this.fail(run, error), { discard: true }); + return Effect.forEach(notes.raises, (item) => this.world.attention.raise(item), { + discard: true, + }).pipe( + Effect.andThen( + Effect.forEach(notes.errors, (error) => this.fail(run, error), { discard: true }), + ), + ); } sayIfCut(repo: string, state: State, cut: boolean): void { @@ -327,6 +367,10 @@ export const runPass = ( Effect.gen(function* () { const pass = new Pass(world, out); yield* pass.record({ event: "pass_started", data: {} }); + // A repo that left the config has nothing the factory still needs a person for. + yield* world.attention.clear( + (item) => item.repo !== null && !machine.repos.some((r) => sameRepo(r.repo, item.repo!)), + ); const repos = yield* loadRepos(world, machine, pass); // Without the ledger there is no telling how many runs are active, so nothing is claimed. const infos = yield* runs(world, pass); @@ -336,6 +380,16 @@ export const runPass = ( yield* untrackedClaims(world, repos, known, pass); yield* unassignedClaims(world, repos, known, pass); yield* claim(world, repos, known, machine, pass); + // A run-bound attention item whose run is no longer unfinished (stopped, abandoned, timed + // out, claimed on) is over. `stuck` and `needs-info` items are about finished runs and stay. + if (!known.lost) { + yield* world.attention.clear( + (item) => + item.kind === "attention" && + item.run !== null && + !known.infos.some((info) => runIdToString(info.id) === item.run), + ); + } } const report = pass.report; pass.line( @@ -450,7 +504,7 @@ const workRuns = ( const repo = loaded.repo; const name = repo.watched.repo; const issue = Number(info.id.issue); - const notes = pass.notes(`${name}${mention(issue)}`, info.state); + const notes = pass.notes(name, issue, info.state); 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); @@ -575,6 +629,14 @@ const advanceRun = ( notes.parts.push( "session asked questions, the issue is now needs-info and the run is finished", ); + notes.raises.push({ + id: `run:${run}:needs-info`, + kind: "needs-info", + repo: notes.repo, + issue: notes.issue, + run, + message: `${notes.label} needs-info, the session asked questions on the issue`, + }); break; case "stuck": pass.stuck(notes, run); @@ -588,6 +650,7 @@ const advanceRun = ( */ const waitsOnItself = (pass: Pass, notes: Notes, run: string): Effect.Effect => pass.attention( + notes, `${notes.label} run ${run} is not progressing and holds a slot: its issue is ready again, and the new claim waits for a slot this run holds. Put the issue back to \`factory:triage\` assigned to the factory's user so \`start\` can finish this run, or free a slot`, run, "not-progressing", @@ -617,6 +680,7 @@ const stalled = ( ? `If it was closed by mistake, reopen it and ${back}` : `If you moved the issue by hand, ${back}`; return pass.attention( + notes, `${notes.label} run ${run} is not progressing and holds a slot: ${plain(why)}. ${advice}`, run, "not-progressing", @@ -695,7 +759,7 @@ const untrackedClaims = ( pass.sayIfCut(name, "triage", listing.success.cut); for (const issue of listing.success.items) { if (known.of(name, issue).some((i) => i.unreadable || i.confirmed)) continue; - const notes = pass.notes(`${name}${mention(issue)}`, "triage"); + const notes = pass.notes(name, issue, "triage"); const run = yield* noteStart(world, pass, known, notes, repo, issue, "resume", { claimed: (run) => `no run on record, claimed in run ${run}`, notTaken: "no run on record, not ours to resume", @@ -746,7 +810,7 @@ const unassignedClaims = ( } for (const { issue, state } of issues.success.claims) { if (known.of(name, issue).length > 0) continue; - const notes = pass.notes(`${name}${mention(issue)}`, state); + const notes = pass.notes(name, issue, state); const released = yield* Effect.result(world.releaseClaim(repo, issue, state)); if (released._tag === "Success") { notes.parts.push("no run on record and no assignee, moved to proposed"); @@ -805,6 +869,10 @@ const noteStart = ( Effect.gen(function* () { const name = repo.watched.repo; const before = newestRun(known.of(name, issue)); + // The issue was claimed again, so what was open about its earlier runs is over. + const clearEarlier = world.attention.clear( + (item) => item.repo !== null && sameRepo(item.repo, name) && item.issue === issue, + ); const outcome = yield* Effect.result(world[call](repo, issue)); if (outcome._tag === "Failure" && outcome.failure._tag === "GateRunningError") { // Nothing was called or written. @@ -821,6 +889,7 @@ const noteStart = ( const run = runIdToString(outcome.success.run); notes.parts.push(words.claimed(run)); if (notes.state === "ready") pass.report.claimed += 1; + yield* clearEarlier; return outcome.success.run; } case "skipped": @@ -838,6 +907,8 @@ const noteStart = ( case "stuck": { const run = runIdToString(outcome.success.run); notes.parts.push(words.stuck); + // The new run's item is raised when the pass flushes, so it outlives this clear. + yield* clearEarlier; pass.stuck(notes, run); return outcome.success.run; } @@ -910,7 +981,7 @@ const claim = ( const repo = queue.repo; const name = repo.watched.repo; const issue = queue.issues.shift()!; - const notes = pass.notes(`${name}${mention(issue)}`, "ready"); + const notes = pass.notes(name, issue, "ready"); // An issue with a run whose gate is running waits: `start` would finish that run under // its gate. It takes no slot, no blocker read, and is not counted as left over. const gating = known.of(name, issue).some((info) => world.gateRunning(info.id));