Public: anyone can read this; nobody can change it.
A crew reviewer's changes-requested queues the builder's follow-up: the reviewer's own run is not the run that hears it #31
Check · Review · Land done · Live — Landed; nothing deploys from this branch, so this is the end of its path.
reviewer-follow-up → main 1 commit
3 files
changed +158
−10
src/core/runs.ts +9 −7
| @@ -582,29 +582,31 @@ export function activeRunFor( | ||
| 582 | 582 | db: Database, |
| 583 | 583 | kind: RunKind, |
| 584 | 584 | issueId: string, |
| 585 | actorId?: string, | |
| 585 | 586 | ): Run | undefined { |
| 586 | 587 | const row = db |
| 587 | 588 | .query< |
| 588 | 589 | RunRow, |
| 589 | [RunKind, string] | |
| 590 | >("SELECT * FROM run WHERE state IN ('queued','running') AND kind = ?1 AND issue_id = ?2 ORDER BY id LIMIT 1") | |
| 591 | .get(kind, issueId); | |
| 590 | [RunKind, string, string | null] | |
| 591 | >("SELECT * FROM run WHERE state IN ('queued','running') AND kind = ?1 AND issue_id = ?2 AND (?3 IS NULL OR actor_id = ?3) ORDER BY id LIMIT 1") | |
| 592 | .get(kind, issueId, actorId ?? null); | |
| 592 | 593 | return row ? toRun(row) : undefined; |
| 593 | 594 | } |
| 594 | 595 | |
| 595 | /** Whether a run of this kind already waits its turn for the issue. */ | |
| 596 | /** Whether a run of this kind already waits its turn for the issue (for `actorId`'s agent, when given). */ | |
| 596 | 597 | export function hasQueuedRun( |
| 597 | 598 | db: Database, |
| 598 | 599 | kind: RunKind, |
| 599 | 600 | issueId: string, |
| 601 | actorId?: string, | |
| 600 | 602 | ): boolean { |
| 601 | 603 | return ( |
| 602 | 604 | db |
| 603 | 605 | .query< |
| 604 | 606 | { n: number }, |
| 605 | [RunKind, string] | |
| 606 | >("SELECT COUNT(*) AS n FROM run WHERE state = 'queued' AND kind = ?1 AND issue_id = ?2") | |
| 607 | .get(kind, issueId)!.n > 0 | |
| 607 | [RunKind, string, string | null] | |
| 608 | >("SELECT COUNT(*) AS n FROM run WHERE state = 'queued' AND kind = ?1 AND issue_id = ?2 AND (?3 IS NULL OR actor_id = ?3)") | |
| 609 | .get(kind, issueId, actorId ?? null)!.n > 0 | |
| 608 | 610 | ); |
| 609 | 611 | } |
| 610 | 612 |
src/crew/dispatch.ts +9 −3
| @@ -282,7 +282,10 @@ export function queueAgentRun( | ||
| 282 | 282 | // thing — a run checked for by one and inserted by the other would be the |
| 283 | 283 | // same work twice (docs/ARCHITECTURE.md, "Two processes at once"). |
| 284 | 284 | const run = transact(app.db, () => { |
| 285 | const active = activeRunFor(app.db, "agent", issue.id); | |
| 285 | // A follow-up is the builder's: a reviewer's own run on the pull, still | |
| 286 | // running as it posts its verdict, is not the run that hears it. | |
| 287 | const mine = reason === "feedback" ? agent.user.id : undefined; | |
| 288 | const active = activeRunFor(app.db, "agent", issue.id, mine); | |
| 286 | 289 | // A follow-up with nothing new would only read the agent's own reply |
| 287 | 290 | // and decline. With a run in flight the answer waits for it to end: the |
| 288 | 291 | // queued run asks again as it starts (`runner.ts`). |
| @@ -298,7 +301,8 @@ export function queueAgentRun( | ||
| 298 | 301 | // queued one may sit behind a running one, so it is looked for by itself. |
| 299 | 302 | if ( |
| 300 | 303 | active !== undefined && |
| 301 | (reason === "assigned" || hasQueuedRun(app.db, "agent", issue.id)) | |
| 304 | (reason === "assigned" || | |
| 305 | hasQueuedRun(app.db, "agent", issue.id, mine)) | |
| 302 | 306 | ) |
| 303 | 307 | return undefined; |
| 304 | 308 | return createRun(app.db, { |
| @@ -427,13 +431,15 @@ function route(app: App, event: ForgeEvent): void { | ||
| 427 | 431 | ); |
| 428 | 432 | return; |
| 429 | 433 | } |
| 434 | // The crew is off: a verdict is read by people, and wakes no one. | |
| 435 | if (reviewerVerdict && repo.autonomy === "off") return; | |
| 430 | 436 | const agent = agentFor(app, issue); |
| 431 | 437 | if (!agent) return; |
| 432 | 438 | // Said while the agent works: it hears the words now, as its next |
| 433 | 439 | // turn, instead of a second run queuing up behind the first. A |
| 434 | 440 | // session nobody can reach any more falls through to the queue. |
| 435 | 441 | const said = event.payload["body"]; |
| 436 | const active = activeRunFor(app.db, "agent", issue.id); | |
| 442 | const active = activeRunFor(app.db, "agent", issue.id, agent.user.id); | |
| 437 | 443 | if ( |
| 438 | 444 | active?.state === "running" && |
| 439 | 445 | typeof said === "string" && |
test/crew-loop.test.ts +140
| @@ -11,6 +11,9 @@ import { join } from "node:path"; | ||
| 11 | 11 | import { createApp, type App } from "../src/app.ts"; |
| 12 | 12 | import { addToCrew, createAgent } from "../src/actions/crew.ts"; |
| 13 | 13 | import { addComment, openIssue } from "../src/actions/issues.ts"; |
| 14 | import { commitToBranch } from "../src/actions/repos.ts"; | |
| 15 | import { openPull, submitReview } from "../src/actions/pulls.ts"; | |
| 16 | import type { Pull } from "../src/core/pulls.ts"; | |
| 14 | 17 | import { createRepo, updateRepoSettings } from "../src/actions/repos.ts"; |
| 15 | 18 | import { setCrewRole } from "../src/core/agents.ts"; |
| 16 | 19 | import { getIssueById, listComments, type Issue } from "../src/core/issues.ts"; |
| @@ -191,3 +194,140 @@ describe("an agent's own comment never starts its next run", () => { | ||
| 191 | 194 | expect(breakerTripped(app.db, fresh(), issued.value.id)).toBe(false); |
| 192 | 195 | }); |
| 193 | 196 | }); |
| 197 | ||
| 198 | describe("a crew reviewer's changes-requested wakes the builder", () => { | |
| 199 | let dataDir: string; | |
| 200 | let app: App; | |
| 201 | let alice: User; | |
| 202 | let mason: User; | |
| 203 | let sentinel: User; | |
| 204 | let repo: Repo; | |
| 205 | let crew: CrewHandle; | |
| 206 | const fresh = () => getRepoById(app.db, repo.id)!; | |
| 207 | ||
| 208 | beforeAll(async () => { | |
| 209 | dataDir = await mkdtemp(join(tmpdir(), "forge-verdict-")); | |
| 210 | app = createApp({ dataDir, port: 0 }); | |
| 211 | const a = await createUser(app.db, { | |
| 212 | trusted: true, | |
| 213 | username: "alice", | |
| 214 | email: "[email protected]", | |
| 215 | password: "a-good-passphrase", | |
| 216 | adminIfFirst: true, | |
| 217 | }); | |
| 218 | if (!a.ok) throw new Error(a.error); | |
| 219 | alice = a.value; | |
| 220 | const made = await createRepo(app, alice, { | |
| 221 | name: "engine", | |
| 222 | visibility: "private", | |
| 223 | }); | |
| 224 | if (!made.ok) throw new Error(String(made.error)); | |
| 225 | repo = made.value; | |
| 226 | await commitToBranch(app, alice, repo, { | |
| 227 | message: "Start", | |
| 228 | files: [{ path: "README.md", content: "# engine\n" }], | |
| 229 | }); | |
| 230 | for (const name of ["mason", "sentinel"]) { | |
| 231 | const agent = await createAgent(app, alice, { username: name }); | |
| 232 | if (!agent.ok) throw new Error(String(agent.error)); | |
| 233 | expect(addToCrew(app, alice, repo, name).ok).toBe(true); | |
| 234 | } | |
| 235 | mason = getUserByName(app.db, "mason")!; | |
| 236 | sentinel = getUserByName(app.db, "sentinel")!; | |
| 237 | setCrewRole(app.db, repo.id, sentinel.id, "reviewer"); | |
| 238 | crew = await startCrew(app, { | |
| 239 | forgeUrl: "http://127.0.0.1:1", | |
| 240 | localUrl: "http://127.0.0.1:1", | |
| 241 | binary: "/nonexistent/claude", | |
| 242 | concurrency: 0, | |
| 243 | budgetUsd: 1, | |
| 244 | runTimeoutMs: 60_000, | |
| 245 | }); | |
| 246 | }); | |
| 247 | ||
| 248 | afterAll(async () => { | |
| 249 | await crew.stop(); | |
| 250 | app.close(); | |
| 251 | await rm(dataDir, { recursive: true, force: true }); | |
| 252 | }); | |
| 253 | ||
| 254 | async function masonPull(name: string) { | |
| 255 | await commitToBranch(app, alice, repo, { | |
| 256 | branch: `feature/${name}`, | |
| 257 | base: "main", | |
| 258 | message: name, | |
| 259 | files: [{ path: `${name}.txt`, content: `${name}\n` }], | |
| 260 | }); | |
| 261 | const opened = await openPull(app, alice, repo, { | |
| 262 | title: name, | |
| 263 | baseBranch: "main", | |
| 264 | headBranch: `feature/${name}`, | |
| 265 | }); | |
| 266 | if (!opened.ok) throw new Error(String(opened.error)); | |
| 267 | app.db | |
| 268 | .query("UPDATE issue SET author_id = ?2 WHERE id = ?1") | |
| 269 | .run(opened.value.id, mason.id); | |
| 270 | return opened.value; | |
| 271 | } | |
| 272 | const feedback = (pull: Pull) => | |
| 273 | listRuns(app.db, repo.id, { issueId: pull.id }).filter( | |
| 274 | (r) => r.kind === "agent" && r.reason === "feedback", | |
| 275 | ); | |
| 276 | /** The reviewer's own runs on the pull, one running as its verdict lands. */ | |
| 277 | const reviewing = (pull: Pull, retry = true) => { | |
| 278 | const run = createRun(app.db, { | |
| 279 | repoId: repo.id, | |
| 280 | kind: "agent", | |
| 281 | issueId: pull.id, | |
| 282 | actorId: sentinel.id, | |
| 283 | reason: "review", | |
| 284 | }); | |
| 285 | claimRun(app.db, run.id); | |
| 286 | // And a retry of the review waiting behind it: also not the builder's. | |
| 287 | if (retry) | |
| 288 | createRun(app.db, { | |
| 289 | repoId: repo.id, | |
| 290 | kind: "agent", | |
| 291 | issueId: pull.id, | |
| 292 | actorId: sentinel.id, | |
| 293 | reason: "review", | |
| 294 | }); | |
| 295 | return run; | |
| 296 | }; | |
| 297 | const settle = () => new Promise((r) => setTimeout(r, 80)); | |
| 298 | ||
| 299 | for (const autonomy of ["autonomous", "assisted"] as const) { | |
| 300 | test(`in ${autonomy}, the verdict queues the builder's follow-up with no person speaking`, async () => { | |
| 301 | const set = await updateRepoSettings(app, alice, fresh(), { | |
| 302 | autonomy, | |
| 303 | autonomyCeilingUsd: 100, | |
| 304 | autonomyBy: alice.id, | |
| 305 | }); | |
| 306 | if (!set.ok) throw new Error(JSON.stringify(set.error)); | |
| 307 | const pull = await masonPull(`verdict-${autonomy}`); | |
| 308 | reviewing(pull); | |
| 309 | const sent = await submitReview(app, sentinel, fresh(), pull, { | |
| 310 | state: "changes_requested", | |
| 311 | body: "The token is logged in clear.", | |
| 312 | }); | |
| 313 | expect(sent.ok).toBe(true); | |
| 314 | await settle(); | |
| 315 | expect(feedback(pull).map((r) => r.actorId)).toEqual([mason.id]); | |
| 316 | }); | |
| 317 | } | |
| 318 | ||
| 319 | test("in off, the verdict queues nothing", async () => { | |
| 320 | const set = await updateRepoSettings(app, alice, fresh(), { | |
| 321 | autonomy: "off", | |
| 322 | }); | |
| 323 | if (!set.ok) throw new Error(JSON.stringify(set.error)); | |
| 324 | const pull = await masonPull("verdict-off"); | |
| 325 | reviewing(pull, false); | |
| 326 | await submitReview(app, sentinel, fresh(), pull, { | |
| 327 | state: "changes_requested", | |
| 328 | body: "Rename it.", | |
| 329 | }); | |
| 330 | await settle(); | |
| 331 | expect(feedback(pull).length).toBe(0); | |
| 332 | }); | |
| 333 | }); |
new above
offline