Public: anyone can read this; the crew can change it.
Land the 3-day-conflicted crew pull sweep from branch scout/5940-a-crew-pull-that-has-stayed-conflicted-f, with its event kind classified for its audience #6562
Check · Review · Land · Live done — Landed and live in production.
scout/forge:scout/6559-land-the-3-day-conflicted-crew-pull-swee → main 7 commits fork 5 threads · 5 unresolved
4 files
changed +263
−2
src/actions/issues.ts +3
| @@ -490,6 +490,7 @@ export function setIssueState( | ||
| 490 | 490 | ? { reason } |
| 491 | 491 | : {}), |
| 492 | 492 | ...(children.length > 0 ? { into: children.map((c) => c.number) } : {}), |
| 493 | ...(closing.stale === true && issue.kind === "pull" ? { stale: true } : {}), | |
| 493 | 494 | }, |
| 494 | 495 | }); |
| 495 | 496 | // The parts serve what the whole served. |
| @@ -614,6 +615,8 @@ export interface Closing { | ||
| 614 | 615 | readonly into?: readonly number[]; |
| 615 | 616 | /** The crew's live triage closed it, not the person whose name it carries. */ |
| 616 | 617 | readonly triage?: boolean; |
| 618 | /** A crew pull closed for staying conflicted 72 hours (`crew/stale_pulls.ts`). */ | |
| 619 | readonly stale?: boolean; | |
| 617 | 620 | } |
| 618 | 621 | |
| 619 | 622 | /** |
src/crew/autonomy.ts +29 −2
| @@ -12,6 +12,7 @@ | ||
| 12 | 12 | * repository's origin. |
| 13 | 13 | */ |
| 14 | 14 | |
| 15 | import { closedToday, closeStale } from "./stale_pulls.ts"; | |
| 15 | 16 | import type { Database } from "bun:sqlite"; |
| 16 | 17 | import type { App } from "../app.ts"; |
| 17 | 18 | import { ceilingReached, forgeCeiling } from "../flow/ceiling.ts"; |
| @@ -1257,6 +1258,7 @@ export async function autonomyTick( | ||
| 1257 | 1258 | activeRuns(app, current.id), |
| 1258 | 1259 | ); |
| 1259 | 1260 | const boots = bootCounts(app.config.dataDir); |
| 1261 | const staleToday = closedToday(app, current.id, now); | |
| 1260 | 1262 | // Says when the governor lowers the slots (#6375); one feed item per change. |
| 1261 | 1263 | const governor = announceGovernor(app, current, now); |
| 1262 | 1264 | log.info("tick", { |
| @@ -1274,6 +1276,8 @@ export async function autonomyTick( | ||
| 1274 | 1276 | ...(held.by.hot_file.length === 0 ? {} : { hot: hotWords(hotFiles(current.id)) }), |
| 1275 | 1277 | // Stalled issues this tick parked and freed for another builder (#4073). |
| 1276 | 1278 | unstuck, |
| 1279 | // Crew pulls closed today after 3 days in conflict (#5940). | |
| 1280 | ...(staleToday === 0 ? {} : { staleClosed: staleToday }), | |
| 1277 | 1281 | // Open pulls, opened/landed per hour and per 6 h, conflicted (#5935). |
| 1278 | 1282 | ...flowVitals(app.db, now), |
| 1279 | 1283 | active: status.active, |
| @@ -1477,6 +1481,8 @@ const checked = new Map< | ||
| 1477 | 1481 | string, |
| 1478 | 1482 | { head: string; key: string; files: string[] | null; tree: string | null } |
| 1479 | 1483 | >(); |
| 1484 | /** Pull id → the head it conflicts at, and when the sweep first saw that (#5940). */ | |
| 1485 | const conflictedAt = new Map<string, { head: string; since: string }>(); | |
| 1480 | 1486 | /** Pull id → when the sweep last saw it merge cleanly. */ |
| 1481 | 1487 | const cleanAt = new Map<string, string>(); |
| 1482 | 1488 | |
| @@ -1607,14 +1613,19 @@ export function releaseConflicted( | ||
| 1607 | 1613 | * for (`conflict_room.ts`); the rest wait for a later tick (#4459). |
| 1608 | 1614 | * Returns how many pulls were acted on. |
| 1609 | 1615 | */ |
| 1610 | export async function sweepConflicts(app: App, repo: Repo): Promise<number> { | |
| 1616 | export async function sweepConflicts( | |
| 1617 | app: App, | |
| 1618 | repo: Repo, | |
| 1619 | now: Date = new Date(), | |
| 1620 | ): Promise<number> { | |
| 1611 | 1621 | if (repo.autonomy !== "autonomous") return 0; |
| 1612 | 1622 | let acted = 0; |
| 1613 | 1623 | let room = readConflictRoom(app.db); |
| 1614 | 1624 | // Forget pulls that are no longer open: closed, merged or released. |
| 1615 | for (const id of new Set([...checked.keys(), ...cleanAt.keys()])) | |
| 1625 | for (const id of new Set([...checked.keys(), ...cleanAt.keys(), ...conflictedAt.keys()])) | |
| 1616 | 1626 | if (getIssueById(app.db, id)?.state !== "open") { |
| 1617 | 1627 | checked.delete(id); |
| 1628 | conflictedAt.delete(id); | |
| 1618 | 1629 | cleanAt.delete(id); |
| 1619 | 1630 | asked.delete(id); |
| 1620 | 1631 | } |
| @@ -1660,15 +1671,31 @@ export async function sweepConflicts(app: App, repo: Repo): Promise<number> { | ||
| 1660 | 1671 | if (seen.files === null) { |
| 1661 | 1672 | // Merges cleanly: whatever came before was a different episode. |
| 1662 | 1673 | cleanAt.set(pull.id, new Date().toISOString()); |
| 1674 | conflictedAt.delete(pull.id); | |
| 1663 | 1675 | continue; |
| 1664 | 1676 | } |
| 1665 | 1677 | const files = seen.files; |
| 1678 | // A push moves the head and the 72-hour clock starts again. | |
| 1679 | let conflict = conflictedAt.get(pull.id); | |
| 1680 | if (conflict?.head !== pull.headOid) | |
| 1681 | conflictedAt.set( | |
| 1682 | pull.id, | |
| 1683 | (conflict = { head: pull.headOid, since: now.toISOString() }), | |
| 1684 | ); | |
| 1666 | 1685 | // An approved pull is finished first, and handed to a person rather |
| 1667 | 1686 | // than closed: the tick's finishing pass has it (`finishing.ts`). |
| 1668 | 1687 | if (approvedOnHead(app.db, pull)) continue; |
| 1669 | 1688 | const issue = getIssueById(app.db, pull.id); |
| 1670 | 1689 | if (!issue || issue.state !== "open") continue; |
| 1671 | 1690 | if (activeRunFor(app.db, "agent", issue.id) !== undefined) continue; |
| 1691 | // Past 72 hours in conflict, whatever the attempts did (#5940). | |
| 1692 | if (closeStale(app, repo, pull, conflict.since, now)) { | |
| 1693 | acted++; | |
| 1694 | checked.delete(pull.id); | |
| 1695 | cleanAt.delete(pull.id); | |
| 1696 | conflictedAt.delete(pull.id); | |
| 1697 | continue; | |
| 1698 | } | |
| 1672 | 1699 | const since = conflictEpisodeStart(app, issue, pull.id); |
| 1673 | 1700 | const runs = app.db |
| 1674 | 1701 | .query<{ n: number; first: string | null }, [string, string]>( |
src/crew/stale_pulls.ts added +60
| @@ -0,0 +1,60 @@ | ||
| 1 | /** | |
| 2 | * A crew pull that stays conflicted is only a longer rebase (#5940). One | |
| 3 | * that has been conflicted for 72 hours, with no push in that time, is | |
| 4 | * closed with one line, and the issue it closed goes back to ready, | |
| 5 | * unassigned, to be built fresh from main. | |
| 6 | * | |
| 7 | * `sweepConflicts` (`autonomy.ts`) already reads every open crew pull per | |
| 8 | * pair of tips and keeps since when each has conflicted at its present | |
| 9 | * head; it calls `closeStale` for one past `STALE_MS`. It leaves a held | |
| 10 | * pull and one approved on its head alone before it gets here, so those | |
| 11 | * are never closed. A pull that names no issue to build again is not | |
| 12 | * closed either. | |
| 13 | */ | |
| 14 | ||
| 15 | import type { App } from "../app.ts"; | |
| 16 | import { assignIssue, addComment, setIssueState } from "../actions/issues.ts"; | |
| 17 | import { listEvents } from "../core/events.ts"; | |
| 18 | import { getIssue, getIssueById } from "../core/issues.ts"; | |
| 19 | import { closesFromBody, type Pull } from "../core/pulls.ts"; | |
| 20 | import { type Repo } from "../core/repos.ts"; | |
| 21 | import { getUserById } from "../core/users.ts"; | |
| 22 | import { createLog } from "../kernel/log.ts"; | |
| 23 | ||
| 24 | const log = createLog("stale-pulls"); | |
| 25 | ||
| 26 | /** How long a pull may stay conflicted, unpushed, before it is closed. */ | |
| 27 | export const STALE_MS = 72 * 3_600_000; | |
| 28 | ||
| 29 | /** The one line the pull is closed with. */ | |
| 30 | export const closedLine = (issueNumber: number): string => | |
| 31 | `Closed after 3 days in conflict; #${issueNumber} is ready again and will be built fresh from main.`; | |
| 32 | ||
| 33 | /** | |
| 34 | * Close one pull that has conflicted since `since` and free its issue. False | |
| 35 | * when it has not been 72 hours, or there is nothing to build again. | |
| 36 | */ | |
| 37 | export function closeStale(app: App, repo: Repo, pull: Pull, since: string, now: Date): boolean { | |
| 38 | if (now.getTime() - Date.parse(since) < STALE_MS) return false; | |
| 39 | const by = repo.autonomyBy ? getUserById(app.db, repo.autonomyBy) : undefined; | |
| 40 | const issue = getIssueById(app.db, pull.id); | |
| 41 | if (!by || !issue || issue.state !== "open") return false; | |
| 42 | const about = closesFromBody(pull.body) | |
| 43 | .map((n) => getIssue(app.db, repo.id, n)) | |
| 44 | .find((i) => i !== undefined && i.kind === "issue" && i.state === "open"); | |
| 45 | if (about === undefined) return false; | |
| 46 | addComment(app, by, repo, issue, closedLine(about.number)); | |
| 47 | const closed = setIssueState(app, by, repo, issue, "closed", { reason: "parked", stale: true }); | |
| 48 | if (!closed.ok) return false; | |
| 49 | if (about.assigneeId !== null) assignIssue(app, by, repo, about, null); | |
| 50 | log.info("stale pull closed", { repo: `${repo.ownerName}/${repo.name}`, pull: pull.number }); | |
| 51 | return true; | |
| 52 | } | |
| 53 | ||
| 54 | /** How many stale pulls the crew closed in this repository today (UTC), for the tick line. */ | |
| 55 | export function closedToday(app: App, repoId: string, now: Date = new Date()): number { | |
| 56 | const from = `${now.toISOString().slice(0, 10)}T00:00:00.000Z`; | |
| 57 | return listEvents(app.db, { repoId, kind: "pull.closed", limit: 500 }).filter( | |
| 58 | (e) => e.payload.stale === true && e.createdAt >= from, | |
| 59 | ).length; | |
| 60 | } |
test/stale_pulls.test.ts added +171
| @@ -0,0 +1,171 @@ | ||
| 1 | /** | |
| 2 | * A crew pull conflicted for 72 hours with no push is closed by the conflict | |
| 3 | * sweep (#5940); a person's, a held one, and one approved on its head are | |
| 4 | * left alone. | |
| 5 | */ | |
| 6 | import { afterAll, beforeAll, expect, test } from "bun:test"; | |
| 7 | import { mkdtemp, rm } from "node:fs/promises"; | |
| 8 | import { tmpdir } from "node:os"; | |
| 9 | import { join } from "node:path"; | |
| 10 | import { createApp, type App } from "../src/app.ts"; | |
| 11 | import { addToCrew, createAgent } from "../src/actions/crew.ts"; | |
| 12 | import { openIssue } from "../src/actions/issues.ts"; | |
| 13 | import { openPull } from "../src/actions/pulls.ts"; | |
| 14 | import { commitToBranch, createRepo } from "../src/actions/repos.ts"; | |
| 15 | import { listEvents } from "../src/core/events.ts"; | |
| 16 | import { setHold } from "../src/core/holds.ts"; | |
| 17 | import { getIssueById, listComments } from "../src/core/issues.ts"; | |
| 18 | import { getPull } from "../src/core/pulls.ts"; | |
| 19 | import { getRepoById, type Repo } from "../src/core/repos.ts"; | |
| 20 | import { submitReview } from "../src/core/reviews.ts"; | |
| 21 | import { createUser, getUserByName, type User } from "../src/core/users.ts"; | |
| 22 | import { sweepConflicts } from "../src/crew/autonomy.ts"; | |
| 23 | import { closedLine, closedToday } from "../src/crew/stale_pulls.ts"; | |
| 24 | ||
| 25 | let dataDir: string; | |
| 26 | let app: App; | |
| 27 | let alice: User; | |
| 28 | let mason: User; | |
| 29 | let repo: Repo; | |
| 30 | let n = 0; | |
| 31 | ||
| 32 | const HOUR = 3_600_000; | |
| 33 | const T0 = new Date(); | |
| 34 | const at = (hours: number) => new Date(T0.getTime() + hours * HOUR); | |
| 35 | /** One sweep at `hours` after T0; queued runs are cleared so the sweep reaches the 72-hour rule. */ | |
| 36 | async function sweep(hours: number): Promise<void> { | |
| 37 | app.db.query("DELETE FROM run").run(); | |
| 38 | await sweepConflicts(app, getRepoById(app.db, repo.id)!, at(hours)); | |
| 39 | } | |
| 40 | ||
| 41 | beforeAll(async () => { | |
| 42 | dataDir = await mkdtemp(join(tmpdir(), "forge-stale-pulls-")); | |
| 43 | app = createApp({ dataDir, port: 0 }); | |
| 44 | const a = await createUser(app.db, { | |
| 45 | origin: "seed", | |
| 46 | trusted: true, | |
| 47 | username: "alice", | |
| 48 | email: "[email protected]", | |
| 49 | password: "a-good-passphrase", | |
| 50 | adminIfFirst: true, | |
| 51 | }); | |
| 52 | if (!a.ok) throw new Error(a.error); | |
| 53 | alice = a.value; | |
| 54 | const made = await createRepo(app, alice, { name: "engine", visibility: "private" }); | |
| 55 | if (!made.ok) throw new Error(String(made.error)); | |
| 56 | repo = made.value; | |
| 57 | const seeded = await commitToBranch(app, alice, repo, { | |
| 58 | message: "Start", | |
| 59 | files: [{ path: "shared.ts", content: "export const shared = 0;\n" }], | |
| 60 | }); | |
| 61 | if (!seeded.ok) throw new Error(JSON.stringify(seeded.error)); | |
| 62 | const agent = await createAgent(app, alice, { username: "mason" }); | |
| 63 | if (!agent.ok) throw new Error(String(agent.error)); | |
| 64 | expect(addToCrew(app, alice, repo, "mason").ok).toBe(true); | |
| 65 | mason = getUserByName(app.db, "mason")!; | |
| 66 | app.db | |
| 67 | .query("UPDATE repo SET autonomy = 'autonomous', autonomy_by = ?2 WHERE id = ?1") | |
| 68 | .run(repo.id, alice.id); | |
| 69 | }); | |
| 70 | ||
| 71 | afterAll(async () => { | |
| 72 | app.db.close(); | |
| 73 | await rm(dataDir, { recursive: true, force: true }); | |
| 74 | }); | |
| 75 | ||
| 76 | /** An open pull by `who` that closes a fresh issue assigned to mason. */ | |
| 77 | async function pullBy(who: User) { | |
| 78 | const k = ++n; | |
| 79 | const ask = openIssue(app, alice, repo, { title: `Ask ${k}` }); | |
| 80 | if (!ask.ok) throw new Error(String(ask.error)); | |
| 81 | app.db.query("UPDATE issue SET assignee_id = ?2 WHERE id = ?1").run(ask.value.id, mason.id); | |
| 82 | const branch = await commitToBranch(app, alice, repo, { | |
| 83 | branch: `feature/s${k}`, | |
| 84 | base: "main", | |
| 85 | message: `Shared ${k}`, | |
| 86 | files: [{ path: "shared.ts", content: `export const shared = 'branch ${k}';\n` }], | |
| 87 | }); | |
| 88 | if (!branch.ok) throw new Error(JSON.stringify(branch.error)); | |
| 89 | const opened = await openPull(app, who, repo, { | |
| 90 | title: `Shared ${k}`, | |
| 91 | body: `Closes #${ask.value.number}`, | |
| 92 | baseBranch: "main", | |
| 93 | headBranch: `feature/s${k}`, | |
| 94 | }); | |
| 95 | if (!opened.ok) throw new Error(String(opened.error)); | |
| 96 | // Main moves under it, on the same line. | |
| 97 | const moved = await commitToBranch(app, alice, repo, { | |
| 98 | branch: "main", | |
| 99 | message: `Shared ${k}, main's way`, | |
| 100 | files: [{ path: "shared.ts", content: `export const shared = 'main ${k}';\n` }], | |
| 101 | }); | |
| 102 | if (!moved.ok) throw new Error(JSON.stringify(moved.error)); | |
| 103 | return { pull: opened.value, ask: ask.value }; | |
| 104 | } | |
| 105 | ||
| 106 | const open = (id: string) => getIssueById(app.db, id)!.state === "open"; | |
| 107 | ||
| 108 | test("a 73-hour crew pull is closed with one line and its issue made ready; 71 hours is not", async () => { | |
| 109 | const { pull, ask } = await pullBy(mason); | |
| 110 | await sweep(0); | |
| 111 | await sweep(71); | |
| 112 | expect(open(pull.id)).toBe(true); | |
| 113 | expect(getIssueById(app.db, ask.id)!.assigneeId).toBe(mason.id); | |
| 114 | ||
| 115 | await sweep(73); | |
| 116 | expect(open(pull.id)).toBe(false); | |
| 117 | const freed = getIssueById(app.db, ask.id)!; | |
| 118 | expect(freed.state).toBe("open"); | |
| 119 | expect(freed.assigneeId).toBeNull(); | |
| 120 | expect(listComments(app.db, pull.id).map((c) => c.body)).toContain(closedLine(ask.number)); | |
| 121 | expect(closedLine(ask.number)).toBe( | |
| 122 | `Closed after 3 days in conflict; #${ask.number} is ready again and will be built fresh from main.`, | |
| 123 | ); | |
| 124 | // One close event, with the stale mark on it, and no second kind. | |
| 125 | const closes = listEvents(app.db, { repoId: repo.id, kind: "pull.closed" }).filter( | |
| 126 | (e) => e.payload.number === pull.number, | |
| 127 | ); | |
| 128 | expect(closes.length).toBe(1); | |
| 129 | expect(closes[0]!.payload.stale).toBe(true); | |
| 130 | expect(closedToday(app, repo.id)).toBe(1); | |
| 131 | // The branch is kept. | |
| 132 | expect(getPull(app.db, repo.id, pull.number)!.headBranch).toBe(`feature/s${n}`); | |
| 133 | }); | |
| 134 | ||
| 135 | test("a person's pull, a held pull and a crew pull approved on its head are never closed", async () => { | |
| 136 | const person = await pullBy(alice); | |
| 137 | const held = await pullBy(mason); | |
| 138 | const approved = await pullBy(mason); | |
| 139 | await sweep(1000); | |
| 140 | setHold(app.db, held.pull.id, alice.id, "wait for the design"); | |
| 141 | submitReview(app.db, { | |
| 142 | issueId: approved.pull.id, | |
| 143 | authorId: alice.id, | |
| 144 | state: "approved", | |
| 145 | commitOid: approved.pull.headOid, | |
| 146 | }); | |
| 147 | await sweep(1100); | |
| 148 | expect(open(person.pull.id)).toBe(true); | |
| 149 | expect(open(held.pull.id)).toBe(true); | |
| 150 | expect(open(approved.pull.id)).toBe(true); | |
| 151 | expect(getIssueById(app.db, approved.ask.id)!.assigneeId).toBe(mason.id); | |
| 152 | }); | |
| 153 | ||
| 154 | test("a push moves the head and the 72 hours start again", async () => { | |
| 155 | const { pull } = await pullBy(mason); | |
| 156 | const branch = `feature/s${n}`; | |
| 157 | await sweep(2000); | |
| 158 | const pushed = await commitToBranch(app, alice, repo, { | |
| 159 | branch, | |
| 160 | message: "Pushed again", | |
| 161 | files: [{ path: "shared.ts", content: "export const shared = 'pushed';\n" }], | |
| 162 | }); | |
| 163 | if (!pushed.ok) throw new Error(JSON.stringify(pushed.error)); | |
| 164 | expect(getPull(app.db, repo.id, pull.number)!.headOid).not.toBe(pull.headOid); | |
| 165 | await sweep(2060); | |
| 166 | // 73 hours after the first sighting, but only 13 after the push. | |
| 167 | await sweep(2073); | |
| 168 | expect(open(pull.id)).toBe(true); | |
| 169 | await sweep(2060 + 73); | |
| 170 | expect(open(pull.id)).toBe(false); | |
| 171 | }); |
new above
offline
In-memory only: a restart forgets the 72h clock. Persist it or derive it from the first conflict event.