diff --git a/plugins/stewards/assessment.test.ts b/plugins/stewards/assessment.test.ts index da3ec07c..1c637045 100644 --- a/plugins/stewards/assessment.test.ts +++ b/plugins/stewards/assessment.test.ts @@ -337,3 +337,150 @@ describe("the divergence store", () => { expect(formatDivergences([])).toBe("No divergences recorded."); }); }); + +// BB-333 (criterion 4 of BB-291): the typed-wait shadow derivation. `waitLive` +// is resolved from the stored kind on every reconcile, never from the +// report-time flag; a typed wait is judged by its own kind, a no-kind report +// keeps the free-text rule exactly. +describe("the shadow assessment: the typed wait (BB-333)", () => { + const openPull = (key: string, pullOver: Partial> = {}) => { + const [repo, number] = key.split("#") as [string, string]; + return pull({ key, url: `https://github.com/${repo}/pull/${number}`, state: "open" }, pullOver); + }; + const waited = (e: EntitySnapshot) => new Map([[e.key.toLowerCase(), e]]); + const withWait = (wait: NonNullable, over: Partial = {}) => input({ binding: binding({ wait }), ...over }); + + it("derives a live pull wait from the freshly read pull, and suppresses idle", () => { + const a = assess(withWait({ kind: "pull", ref: "acme/web#9" }, { waitedPulls: waited(openPull("acme/web#9")) })); + expect(a.liveness.waitLive).toBe(true); + expect(a.liveness.stalled).toBe(false); + // The base idle rule is true (build stage, no source), and a live typed wait suppresses it. + expect(a.liveness.idle).toBe(false); + }); + + it("derives a stalled external wait even when an unrelated source is live", () => { + const a = assess(withWait({ kind: "external", ref: "the vendor" }, { sources: { ...noSources, pullRequestWatch: true } })); + expect(a.liveness.waitLive).toBe(false); + expect(a.liveness.stalled).toBe(true); + }); + + it("derives a review wait live only until the review is given", () => { + const live = assess(withWait({ kind: "review", ref: "acme/web#9" }, { waitedPulls: waited(openPull("acme/web#9")) })); + expect(live.liveness.waitLive).toBe(true); + const reviewed = assess(withWait({ kind: "review", ref: "acme/web#9" }, { waitedPulls: waited(openPull("acme/web#9", { reviewDecision: "APPROVED" })) })); + expect(reviewed.liveness.waitLive).toBe(false); + expect(reviewed.liveness.stalled).toBe(true); + }); + + it("derives a deployment wait live while the watch is merged or deploying", () => { + const watch = { key: "acme/web#9", state: "deploying" as const, mergeSha: null, target: "", mergedAt: null, prRepo: "acme/web", files: [], repo: "acme/web", environment: "production", deploymentId: null, deploymentState: null, ref: null, sha: null, url: null, at: 1, deadlineAt: null, note: null, workflowRuns: [] }; + const a = assess(input({ binding: binding({ wait: { kind: "deployment", ref: "acme/web#9" }, deployments: { "acme/web#9": watch } }) })); + expect(a.liveness.waitLive).toBe(true); + }); + + it("derives a worker wait live only while the worker runs", () => { + const running = assess(withWait({ kind: "worker", ref: "thr_w1" }, { workers: [{ threadId: "thr_w1", status: "running" } as never] })); + expect(running.liveness.waitLive).toBe(true); + const finished = assess(withWait({ kind: "worker", ref: "thr_w1" }, { workers: [parkedWorker({ threadId: "thr_w1" })] })); + expect(finished.liveness.waitLive).toBe(false); + }); + + it("derives a binding and a person wait from the lookups the reconcile holds", () => { + const open = assess(withWait({ kind: "binding", ref: "ENG-13" }, { hasOpenBinding: () => true })); + expect(open.liveness.waitLive).toBe(true); + const closed = assess(withWait({ kind: "binding", ref: "ENG-13" }, { hasOpenBinding: () => false })); + expect(closed.liveness.waitLive).toBe(false); + const person = assess(withWait({ kind: "person", ref: "need_1" }, { hasOpenItem: () => true })); + expect(person.liveness.waitLive).toBe(true); + }); + + it("derives a date wait live while the date is ahead", () => { + const now = Date.UTC(2026, 5, 1); + const ahead = assess(withWait({ kind: "date", ref: "2026-12-01" }, { now })); + expect(ahead.liveness.waitLive).toBe(true); + expect(ahead.liveness.stalled).toBe(false); + const past = assess(withWait({ kind: "date", ref: "2000-01-01" }, { now })); + expect(past.liveness.waitLive).toBe(false); + expect(past.liveness.stalled).toBe(true); + }); + + it("keeps the free-text rule exactly for a report with no typed wait", () => { + const stalled = assess(input({ binding: binding({ status: status("the vendor"), noWakeSource: false }) })); + expect(stalled.liveness.waitLive).toBeNull(); + expect(stalled.liveness.stalled).toBe(true); + const dated = assess(input({ binding: binding({ status: status("the vendor"), requeueUntil: 2_000_000, noWakeSource: false }) })); + expect(dated.liveness.stalled).toBe(false); + }); + + it("keeps the idleInActiveStage divergence on the unadjusted idle when a typed wait is live", () => { + const base = bb266(false); + const i: AssessmentInput = { ...base, binding: { ...base.binding, wait: { kind: "pull", ref: "acme/web#9" } }, waitedPulls: waited(openPull("acme/web#9")) }; + const a = assess(i); + // The consumer's idle (idleBase) is true, the typed-wait-adjusted idle is not. + expect(a.liveness.idleBase).toBe(true); + expect(a.liveness.idle).toBe(false); + // The parked-worker surface still compares the consumer's idle. + expect(compare(i, a).find((x) => x.surface === "idleInActiveStage")).toMatchObject({ old: "idle", new: "parked" }); + }); +}); + +describe("the typed-wait divergence store (BB-333)", () => { + function store() { + const db = new Database(":memory:"); + for (const sql of MIGRATIONS) db.exec(sql); + let now = 1_000; + const s = createStore(db, () => now); + return { s, advance: (ms: number) => (now += ms) }; + } + + it("writes a noWakeSource row only when the disagreement starts", () => { + const { s } = store(); + // The first observation disagrees: a start row. + expect(s.recordWaitDivergence({ bindingId: "b1", derived: true, reported: false, waitKind: "external", waitLive: false, inputs: {} }, 1_000)).toBe(true); + // The same disagreement again: no new row, but the open row is kept fresh. + expect(s.recordWaitDivergence({ bindingId: "b1", derived: true, reported: false, waitKind: "external", waitLive: false, inputs: {} }, 2_000)).toBe(false); + let rows = s.divergences({ bindingId: "b1" }); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ old: "source-or-no-wait", new: "no-wake-source", firstSeenAt: 1_000, lastSeenAt: 2_000, count: 1 }); + expect(rows[0]!.inputs).toMatchObject({ transition: "ongoing" }); + // It stops disagreeing: noted on the same row, never a row of its own. + expect(s.recordWaitDivergence({ bindingId: "b1", derived: false, reported: false, waitKind: null, waitLive: false, inputs: {} }, 3_000)).toBe(false); + rows = s.divergences({ bindingId: "b1" }); + expect(rows).toHaveLength(1); + expect(rows[0]!.inputs).toMatchObject({ transition: "stop" }); + // It starts again: the same (old, new) row bumps its count rather than adding a row. + expect(s.recordWaitDivergence({ bindingId: "b1", derived: true, reported: false, waitKind: "external", waitLive: false, inputs: {} }, 4_000)).toBe(true); + rows = s.divergences({ bindingId: "b1" }); + expect(rows).toHaveLength(1); + expect(rows[0]!.count).toBe(2); + // The surface filter still reads it as `noWakeSource`. + expect(s.divergences({ surface: "noWakeSource" })).toHaveLength(1); + // A first observation that agrees is a baseline, not a transition. + expect(s.recordWaitDivergence({ bindingId: "b2", derived: false, reported: false, waitKind: null, waitLive: false, inputs: {} }, 5_000)).toBe(false); + expect(s.divergences({ bindingId: "b2" })).toHaveLength(0); + }); + + it("keeps the two directions' rows apart and treats a flip as a stop plus a start", () => { + const { s } = store(); + // start A, stop A, start B, ongoing B. + expect(s.recordWaitDivergence({ bindingId: "b1", derived: true, reported: false, waitKind: "external", waitLive: false, inputs: {} }, 1_000)).toBe(true); + expect(s.recordWaitDivergence({ bindingId: "b1", derived: false, reported: false, waitKind: null, waitLive: false, inputs: {} }, 2_000)).toBe(false); + expect(s.recordWaitDivergence({ bindingId: "b1", derived: false, reported: true, waitKind: "pull", waitLive: true, inputs: {} }, 3_000)).toBe(true); + expect(s.recordWaitDivergence({ bindingId: "b1", derived: false, reported: true, waitKind: "pull", waitLive: true, inputs: {} }, 4_000)).toBe(false); + let rows = s.divergences({ bindingId: "b1" }); + expect(rows).toHaveLength(2); + const a = rows.find((r) => r.new === "no-wake-source")!; + const b = rows.find((r) => r.old === "no-wake-source")!; + // The stop touched A, and ongoing B did not touch A again. + expect(a.inputs).toMatchObject({ transition: "stop" }); + expect(a.lastSeenAt).toBe(2_000); + expect(b.inputs).toMatchObject({ transition: "ongoing" }); + expect(b.lastSeenAt).toBe(4_000); + // A flip B → A is a stop on B plus a start on A (bumping A's count). + expect(s.recordWaitDivergence({ bindingId: "b1", derived: true, reported: false, waitKind: "external", waitLive: false, inputs: {} }, 5_000)).toBe(true); + rows = s.divergences({ bindingId: "b1" }); + expect(rows).toHaveLength(2); + expect(rows.find((r) => r.new === "no-wake-source")!.count).toBe(2); + expect(rows.find((r) => r.old === "no-wake-source")!.inputs).toMatchObject({ transition: "stop" }); + }); +}); diff --git a/plugins/stewards/assessment.ts b/plugins/stewards/assessment.ts index 501a3266..cf3ac5a2 100644 --- a/plugins/stewards/assessment.ts +++ b/plugins/stewards/assessment.ts @@ -11,13 +11,15 @@ // disagree, so the divergence log can be read for several days before any // consumer reads the assessment. import { assessedPullGate, gateInputKey, pullGate } from "./assessed-gate.js"; -import type { PrGate, ProjectApprovalRecord } from "./gate.js"; +import { latestReviewMarker, type PrGate, type ProjectApprovalRecord } from "./gate.js"; +import { parsePullRef } from "./github.js"; import { gateState } from "./ready-gate.js"; import { anyWorkerInFlight, idleInActiveStage, observedStall, productGateOptions, runningWorkerReason, stuckDue, type IdleBinding } from "./reconcile.js"; -import { hasLiveWakeSource, type WakeSourceInput } from "./wake-source.js"; +import { hasLiveWakeSource, parseUntil, type WakeSourceInput } from "./wake-source.js"; import type { Playbook } from "./config.js"; import type { EntitySnapshot } from "./source-contract.js"; import type { Binding, Escalation, Need, Observed, Worker } from "./store.js"; +import type { Wait } from "./wait.js"; export type DecideMark = "project" | "user"; @@ -59,6 +61,10 @@ export type AssessedWorker = { export type AssessedLiveness = { /** `hasLiveWakeSource` over the observable rows. */ live: boolean; + /** The stored typed wait's own liveness (BB-333); null when the binding has no typed wait. */ + waitLive: boolean | null; + /** `idleInActiveStage` before the typed wait is applied: what the consumer's idle wake reads (BB-333). */ + idleBase: boolean; /** `idleInActiveStage`. */ idle: boolean; /** `stuckDue`: the phase backstop. */ @@ -94,7 +100,17 @@ export type AssessmentInput = { binding: Pick< Binding, "id" | "entityKey" | "kind" | "ready" | "stage" | "ending" | "hold" | "deployments" | "requeueUntil" | "noWakeSource" | "status" | "createdAt" | "advancedAt" | "phase" - >; + > & { wait?: Binding["wait"] }; + /** + * Fresh snapshots of the binding's waited `pull`/`review` refs (BB-333), + * keyed by lowercased `owner/repo#N`. Read by the look, never stored in the + * binding's snapshot. + */ + waitedPulls?: ReadonlyMap; + /** Whether a binding with this entity key is open and unfinished, for a `binding` wait (BB-333). */ + hasOpenBinding?: (key: string) => boolean; + /** Whether a page item with this id is open, for a `person` wait (BB-333). */ + hasOpenItem?: (id: string) => boolean; /** The observation the reconcile just read. */ observed: Observed; /** The binding's own pull keys and URLs, lowercased (BB-52 rule 0). */ @@ -146,6 +162,58 @@ function rememberedFor(map: ReadonlyMap | undefined, const isOwn = (e: EntitySnapshot, own: ReadonlySet): boolean => (e.key !== "" && own.has(lc(e.key))) || (e.url !== null && e.url !== "" && own.has(lc(e.url))); +/** A pull request the plugin can see is still open (not merged or closed). */ +const pullIsOpen = (e: EntitySnapshot): boolean => e.pull !== null && e.state !== "merged" && e.state !== "closed" && e.pull.merged !== true; + +/** Whether a pull request's review has been given: an approved review or marker at its head (BB-333). */ +function reviewGiven(e: EntitySnapshot): boolean { + if (e.pull === null) return false; + if (e.pull.reviewDecision === "APPROVED") return true; + const marker = latestReviewMarker(e.comments); + if (marker !== null && marker.verdict === "approved" && (marker.sha === "" || marker.sha === e.pull.headSha)) return true; + return e.pull.reviews.some((r) => r.state.toLowerCase() === "approved" && (r.commit === null || r.commit === e.pull?.headSha)); +} + +/** The lowercased `owner/repo#N` a stored `pull`/`review` ref names, or null when it does not parse. */ +function waitPullKey(ref: string): string | null { + const parsed = parsePullRef(ref); + return parsed === null ? null : `${parsed.repo}#${parsed.number}`.toLowerCase(); +} + +/** + * Whether the stored typed wait is live, resolved against the rows the look + * holds (BB-333): the freshly read waited pull, the deployment watches, the + * workers, the other bindings, the page items and the date. The typed wait is + * judged by its own kind, never by `hasLiveWakeSource`. + */ +function typedWaitLive(wait: Wait, input: AssessmentInput): boolean { + switch (wait.kind) { + case "pull": + case "review": { + const key = waitPullKey(wait.ref); + const e = key === null ? undefined : input.waitedPulls?.get(key); + if (e === undefined || !pullIsOpen(e)) return false; + return wait.kind === "pull" ? true : !reviewGiven(e); + } + case "deployment": { + const key = waitPullKey(wait.ref); + return key !== null && Object.values(input.binding.deployments).some((w) => lc(w.key) === key && (w.state === "merged" || w.state === "deploying")); + } + case "worker": + return input.workers.some((w) => w.threadId === wait.ref && w.status === "running"); + case "binding": + return input.hasOpenBinding?.(wait.ref) ?? false; + case "date": { + const at = parseUntil(wait.ref) ?? input.binding.requeueUntil; + return at !== null && at > input.now; + } + case "person": + return input.hasOpenItem?.(wait.ref) ?? false; + case "external": + return false; + } +} + /** * Whether the project steward forwarded this pull request's approval escalation * to the owner (BB-192): an open `approval` escalation for the pull, routed to @@ -228,6 +296,13 @@ export function assess(input: AssessmentInput): Assessment { needs, }; const live = assessLiveWakeSource(sources); + // BB-333: the typed wait is judged by its own kind, on every reconcile, + // never from a flag stored at report time. A no-kind report keeps today's + // free-text rule exactly. + const wait = binding.wait ?? null; + const waitLive = wait === null ? null : typedWaitLive(wait, input); + const dateAhead = binding.requeueUntil !== null && binding.requeueUntil > now; + const idleBase = assessIdle(idleBinding, observed, now); return { bindingId: binding.id, entityKey: binding.entityKey, @@ -241,9 +316,13 @@ export function assess(input: AssessmentInput): Assessment { worker: assessWorker(workers), liveness: { live, - idle: assessIdle(idleBinding, observed, now), + waitLive, + idleBase, + // A live typed wait is something the plugin waits on, so it suppresses + // the idle wake; a stalled or external one does not (BB-333). + idle: wait !== null && waitLive === true ? false : idleBase, stuck: stuckDue(binding, playbook, [...escalations], now) !== null, - stalled: observedStall(binding, live, now), + stalled: wait === null ? observedStall(binding, live, now) : waitLive !== true && !dateAhead, reportedNoWakeSource: binding.noWakeSource, }, }; @@ -295,7 +374,7 @@ export function compare(input: AssessmentInput, assessment: Assessment): Diverge if (assessment.worker.parked) { if (anyWorkerInFlight(workers)) add("workerInFlight", "in-flight", "parked", workerInputs); if (runningWorkerReason(workers) === null) add("runningWorkerReason", "not-mid-turn", "parked", workerInputs); - if (assessment.liveness.idle) add("idleInActiveStage", "idle", "parked", workerInputs); + if (assessment.liveness.idleBase) add("idleInActiveStage", "idle", "parked", workerInputs); if (!assessment.liveness.live) add("hasLiveWakeSource", "no-live-source", "parked", workerInputs); } @@ -332,6 +411,8 @@ export function compare(input: AssessmentInput, assessment: Assessment): Diverge phase: binding.phase, // The row the predicates read, as a boolean, never the wait text. hasWait: (binding.status?.waitingOn ?? "").trim() !== "", + waitKind: binding.wait?.kind ?? null, + waitLive: assessment.liveness.waitLive === true, }); return out; } diff --git a/plugins/stewards/migrations.snapshot.json b/plugins/stewards/migrations.snapshot.json index 242e5ba5..87ba322a 100644 --- a/plugins/stewards/migrations.snapshot.json +++ b/plugins/stewards/migrations.snapshot.json @@ -139,5 +139,6 @@ "c1377801bd4fe44be9cc5fc10c5aca75c9455359b3d5d646d0fb23318a032124", "d856f2127e2c5788a01c5ca80b216d1dc9be2a67bb273ad9082bbc69a279ce9a", "f1636e9715f38e19e3c21cc7d3008c4f8d734ec26e03eb983961acb7d4cc8949", - "5aeed47ee4b133a8d26b01316c545a6f4c49b3c73fb9511cb4c2b9eae55a31fa" + "5aeed47ee4b133a8d26b01316c545a6f4c49b3c73fb9511cb4c2b9eae55a31fa", + "ded6270fedbb7d853a412b2167f164b3886f38ed975c124d83b40f6d625500ad" ] diff --git a/plugins/stewards/server.test.ts b/plugins/stewards/server.test.ts index cedf551b..96b850c6 100644 --- a/plugins/stewards/server.test.ts +++ b/plugins/stewards/server.test.ts @@ -14804,6 +14804,107 @@ describe("a typed wait (BB-333)", () => { }); }); +// BB-333 (criterion 4 of BB-291) PR 2: the typed-wait shadow derivation. The +// four fixtures assert the result derived from the stored wait kind on a +// reconcile, beside `noWakeSource`, which stays the report-time flag. +describe("the typed wait's shadow derivation (BB-333)", () => { + const base = { phase: "In Progress", summary: "Waiting.", nextAction: "Wait", actor: "other" as const }; + const noReview = comment({ id: "r1", author: "jane", body: ``, signedBy: "thr_reviewer", createdAt: "2026-10-14T15:00:00Z" }); + type World = Awaited>; + const derived = (t: World, bindingId: string) => t.hooks.store.assessment(bindingId)!.liveness as { stalled: boolean; waitLive: boolean | null }; + + async function ticket() { + const t = await setup(); + t.entities.set("uuid-12", issue({ state: "In Progress", stateType: "started" })); + await t.ring("linear", "uuid-12"); + return { t, steward: t.spawns[1]!.id, b: t.hooks.store.find("ENG-12")! }; + } + + it("derives the BB-266 fixture: the external wait stalls despite the second open own pull", async () => { + const t = await setup(); + t.entities.set("uuid-12", issue({ state: "In Progress", stateType: "started", linkedPulls: [{ key: "acme/web#7", url: "https://github.com/acme/web/pull/7" }, { key: "acme/web#8", url: "https://github.com/acme/web/pull/8" }] })); + t.entities.set("acme/web#7", pull({ comments: [noReview] }, { checks: { state: "success", failing: [] } })); + t.entities.set("acme/web#8", pull({ id: "acme/web#8", key: "acme/web#8", url: "https://github.com/acme/web/pull/8" })); + await t.ring("linear", "uuid-12"); + await t.created("thr_w1", t.spawns[1]!.id, "ENG-12: Implement"); + t.hooks.store.setWorker("thr_w1", "finished", "Waiting on the owner's approval."); + t.hooks.store.parkWorker("thr_w1", "the owner's approval"); + const b = t.hooks.store.find("ENG-12")!; + await t.tool("steward_report", { ...base, waitingOn: "the plugin auto-merge", wait: { kind: "pull", ref: "acme/web#7" } }, b.threadId!); + await t.hooks.reconcilePass("p1"); + expect(t.hooks.store.find("ENG-12")!.noWakeSource).toBe(false); + expect(derived(t, b.id)).toMatchObject({ stalled: true, waitLive: false }); + // The report-time flag agrees with the second open own pull, so the + // derived stall is a divergence and is recorded. + expect(t.hooks.store.divergences({ bindingId: b.id, surface: "noWakeSource" })).toContainEqual(expect.objectContaining({ old: "source-or-no-wait", new: "no-wake-source" })); + }); + + it("derives the bb-plugins#294 fixture: a pr binding's own external wait stalls", async () => { + const t = await setup(); + t.entities.set("acme/web#7", pull({ labels: ["bb-managed"], author: "jane" })); + await t.ring("github", "acme/web#7"); + const b = t.hooks.store.find("acme/web#7")!; + await t.tool("steward_report", { ...base, waitingOn: "the dompurify fix on main", wait: { kind: "pull", ref: "acme/web#7" } }, b.threadId!); + await t.hooks.reconcilePass("p1"); + expect(derived(t, b.id)).toMatchObject({ stalled: true, waitLive: false }); + }); + + it("derives the bc-prod#422/BCP-157 fixture: the closed person wait stalls despite the open pull watch", async () => { + const t = await setup(); + t.entities.set("uuid-12", issue({ state: "In Progress", stateType: "started", linkedPulls: [{ key: "acme/web#7", url: "https://github.com/acme/web/pull/7" }] })); + t.entities.set("acme/web#7", pull({ id: "acme/web#7", key: "acme/web#7", url: "https://github.com/acme/web/pull/7" })); + await t.ring("linear", "uuid-12"); + const b = t.hooks.store.find("ENG-12")!; + await t.tool("steward_report", { ...base, waitingOn: "GitHub approval of acme/web#7", wait: { kind: "person", ref: "need_approve" } }, b.threadId!); + await t.hooks.reconcilePass("p1"); + expect(t.hooks.store.find("ENG-12")!.noWakeSource).toBe(false); + expect(derived(t, b.id)).toMatchObject({ stalled: true, waitLive: false }); + }); + + it("derives the bc-prod#430/BCP-160 fixture: the leased foreign pull is live", async () => { + const { t, steward } = await ticket(); + t.entities.set("acme/web#430", pull({ id: "acme/web#430", key: "acme/web#430", url: "https://github.com/acme/web/pull/430", state: "open" })); + const b = t.hooks.store.find("ENG-12")!; + await t.tool("steward_report", { ...base, waitingOn: "bc-prod#430 checks", wait: { kind: "pull", ref: "acme/web#430" } }, steward); + await t.hooks.reconcilePass("p1"); + expect(derived(t, b.id)).toMatchObject({ stalled: false, waitLive: true }); + }); + + it("leases a waited pull ref on the report and drops it when the wait clears", async () => { + const { t, steward } = await ticket(); + t.entities.set("acme/web#430", pull({ id: "acme/web#430", key: "acme/web#430", url: "https://github.com/acme/web/pull/430", state: "open" })); + const github = () => t.rpcCalls.filter((c) => c.pluginId === "github-watcher" && c.method === "sourceSetInterest").at(-1)!.input as { entities: string[] }; + await t.tool("steward_report", { ...base, waitingOn: "bc-prod#430 checks", wait: { kind: "pull", ref: "acme/web#430" } }, steward); + await t.flush(); + expect(github().entities).toContain("acme/web#430"); + // A report without the wait clears it, and the lease drops the ref. + await t.tool("steward_report", { ...base, waitingOn: "the vendor" }, steward); + await t.flush(); + expect(github().entities).not.toContain("acme/web#430"); + }); + + it("a ring on a waited ref looks, sends no wake, and touches no snapshot or resource", async () => { + const { t, steward } = await ticket(); + t.entities.set("acme/web#430", pull({ id: "acme/web#430", key: "acme/web#430", url: "https://github.com/acme/web/pull/430", state: "open" })); + await t.tool("steward_report", { ...base, waitingOn: "bc-prod#430 checks", wait: { kind: "pull", ref: "acme/web#430" } }, steward); + await t.hooks.reconcilePass("p1"); + const before = t.hooks.store.find("ENG-12")!; + const wakesBefore = t.sends.filter((s) => s.threadId === steward).length; + const pullsBefore = JSON.stringify(before.lastSnapshot?.pulls ?? []); + t.rpcCalls.length = 0; + // The waited pull changes; the doorbell rings it. + t.entities.set("acme/web#430", pull({ id: "acme/web#430", key: "acme/web#430", url: "https://github.com/acme/web/pull/430", state: "open", title: "changed" })); + await t.ring("github", "acme/web#430"); + await t.flush(); + // The waiting ticket looked (it read the waited ref) but the change woke nobody. + expect(t.rpcCalls.some((c) => c.method === "sourceSnapshot" && c.input.id === "acme/web#430")).toBe(true); + expect(t.sends.filter((s) => s.threadId === steward).length).toBe(wakesBefore); + const after = t.hooks.store.find("ENG-12")!; + expect(JSON.stringify(after.lastSnapshot?.pulls ?? [])).toBe(pullsBefore); + expect(after.status?.resources.pullRequests ?? []).not.toContain("acme/web#430"); + }); +}); + // BB-203: the plugin writes the binding's status from events it observed // itself, before the steward's next turn, and quiets a running worker's own // pull request activity only while nothing the user must act on is open. diff --git a/plugins/stewards/server.ts b/plugins/stewards/server.ts index 9a409baf..f18be7cf 100644 --- a/plugins/stewards/server.ts +++ b/plugins/stewards/server.ts @@ -3150,6 +3150,45 @@ export function createStewardsPlugin(deps: StewardsDeps): (bb: BbPluginApi) => P return pulls; } + /** + * The `pull`/`review` refs the binding's stored typed wait names, normalized + * to `owner/repo#N` (BB-333). They are leased and read beside the ticket's + * linked pulls, but never join `ownPulls`, the gate loop or the stored + * snapshot. + */ + function waitedPullKeys(b: Pick): string[] { + const wait = b.wait; + if (wait === null || (wait.kind !== "pull" && wait.kind !== "review")) return []; + const parsed = parsePullRef(wait.ref); + return parsed === null ? [] : [`${parsed.repo}#${parsed.number}`]; + } + + /** + * Read the binding's waited pull/review refs fresh, one source read each, + * under the linked-pull ceiling (BB-333). A ref already in the look's linked + * pulls is reused without a read, so it never adds one. Returns them keyed + * by lowercased `owner/repo#N`; a read that fails is left out, which reads + * as not-live. + */ + async function readWaitedPulls(b: Binding, observed: Observed): Promise> { + const out = new Map(); + for (const key of waitedPullKeys(b)) { + const already = observed.pulls.find((p) => lc(p.key) === lc(key)); + if (already !== undefined) { + out.set(lc(key), already); + continue; + } + if (observed.pulls.length + out.size >= LINKED_PULLS_MAX) break; + try { + const e = await snapshot("github", key); + if (e !== null) out.set(lc(key), e); + } catch (error) { + bb.log.warn(`stewards: could not read waited ${key} of ${b.entityKey}: ${errorMessage(error)}`); + } + } + return out; + } + /** * A binding's pull requests, re-read from GitHub rather than from its last * snapshot (BB-54): a schedule request must not ask about work that has @@ -3366,6 +3405,7 @@ export function createStewardsPlugin(deps: StewardsDeps): (bb: BbPluginApi) => P ...new Set([ ...open.filter((b) => b.source === "github").map((b) => b.entityId), ...open.flatMap((b) => leasedPulls(b)), + ...open.flatMap((b) => waitedPullKeys(b)), ]), ].sort(), // Every repo the project watches for Dependabot alerts. A server @@ -4646,7 +4686,7 @@ export function createStewardsPlugin(deps: StewardsDeps): (bb: BbPluginApi) => P * in a try/catch by the caller so a failure cannot crash the reconcile; * an absent assessment fails closed everywhere it is read. */ - function assessBinding( + async function assessBinding( b: Binding, observed: Observed, workers: readonly Worker[], @@ -4654,7 +4694,7 @@ export function createStewardsPlugin(deps: StewardsDeps): (bb: BbPluginApi) => P playbook: Playbook, currentSpecHash: string, config: ProjectConfig, - ): Assessment { + ): Promise { // BB-322: gate every pull request that belongs to the binding, the same // set `syncNeeds` acts on, not only the worker-reported own ones. A // linked pull request whose head names the ticket is the binding's even @@ -4694,6 +4734,12 @@ export function createStewardsPlugin(deps: StewardsDeps): (bb: BbPluginApi) => P if (e.url !== null) reRequestedDelegates.set(lc(e.url), reRequested); } } + const waitedPulls = await readWaitedPulls(b, observed); + // The look already holds the binding's needs and escalations; a `person` + // wait resolves from those, and only a `qst_` ref reads the project's + // questions (BB-333). + const needs = store.openNeeds(b.id); + const escalations = store.openEscalations(b.id); const input: AssessmentInput = { binding: b, observed, @@ -4704,16 +4750,47 @@ export function createStewardsPlugin(deps: StewardsDeps): (bb: BbPluginApi) => P rememberedRequesters, reRequestedDelegates, workers, - escalations: store.openEscalations(b.id), - needs: store.openNeeds(b.id), + escalations, + needs, sources: wakeSourcesFor(b, ownPulls), playbook, currentSpecHash, now: now(), + waitedPulls, + hasOpenBinding: (key) => { + const other = store.find(key, b.projectId); + return other !== null && other.endedAt === null && (other.lastSnapshot === null || !finished(other, other.lastSnapshot)); + }, + hasOpenItem: (id) => needs.some((n) => n.id === id) || escalations.some((x) => x.id === id) || (id.startsWith("qst_") && store.openQuestions(b.projectId).some((q) => q.id === id)), }; const assessment = assess(input); store.recordAssessment({ bindingId: b.id, at: input.now, gate: assessment.gate, worker: assessment.worker, liveness: assessment.liveness }); - store.recordDivergences(compare(input, assessment)); + const divergences = compare(input, assessment); + // BB-333: a typed wait's divergence is written only when it starts or + // stops disagreeing, through the state table. A no-kind report keeps the + // existing deduped row, and a binding with no typed wait reads no state. + if (b.wait !== null) { + store.recordDivergences(divergences.filter((d) => d.surface !== "noWakeSource")); + store.recordWaitDivergence({ + bindingId: b.id, + derived: assessment.liveness.stalled, + reported: b.noWakeSource, + waitKind: b.wait.kind, + waitLive: assessment.liveness.waitLive === true, + inputs: { + binding: b.id, + live: assessment.liveness.live, + idle: assessment.liveness.idle, + stuck: assessment.liveness.stuck, + readyPassed: assessment.gate.readyPassed, + requeueUntil: b.requeueUntil, + phase: b.phase, + hasWait: (b.status?.waitingOn ?? "").trim() !== "", + }, + }); + } else { + store.recordDivergences(divergences); + } store.pruneDivergences(ASSESSMENT_DIVERGENCE_MAX_AGE_MS, ASSESSMENT_DIVERGENCE_MAX_ROWS, input.now); return assessment; } @@ -4730,13 +4807,16 @@ export function createStewardsPlugin(deps: StewardsDeps): (bb: BbPluginApi) => P // reconcile. Its totals land in the current pass `bb stewards stats` // reports. await readCounts.runReconcile(async () => { + // BB-333: a `pull`/`review` typed wait joins the github lease beside the + // linked pulls, so a change to either is what the interest set compares. + const interest = (b: Binding): string => [...leasedPulls(b), ...waitedPullKeys(b)].sort().join(" "); const before = store.binding(bindingId); - const leased = before === null ? "" : leasedPulls(before).join(" "); + const leased = before === null ? "" : interest(before); try { await reconcileOnce(bindingId, force); } finally { const after = store.binding(bindingId); - if (after !== null && leasedPulls(after).join(" ") !== leased) pushInterest(); + if (after !== null && interest(after) !== leased) pushInterest(); } }); } @@ -5023,7 +5103,7 @@ export function createStewardsPlugin(deps: StewardsDeps): (bb: BbPluginApi) => P // absent assessment fails closed. let assessment: Assessment | null = null; try { - assessment = assessBinding(b, observed, workers, ownPulls, playbook, currentSpec, config); + assessment = await assessBinding(b, observed, workers, ownPulls, playbook, currentSpec, config); } catch (error) { bb.log.warn(`stewards: assessment failed for ${b.entityKey}: ${errorMessage(error)}`); } @@ -7067,7 +7147,9 @@ export function createStewardsPlugin(deps: StewardsDeps): (bb: BbPluginApi) => P if (deps !== null && deps.endedAt === null) scheduleReconcile(deps.id); } for (const t of store.open()) { - if (t.kind === "ticket" && (t.lastSnapshot?.entity.linkedPulls ?? []).some((l) => lc(l.key) === lc(id))) scheduleReconcile(t.id); + const linked = t.kind === "ticket" && (t.lastSnapshot?.entity.linkedPulls ?? []).some((l) => lc(l.key) === lc(id)); + const waited = waitedPullKeys(t).some((k) => lc(k) === lc(id)); + if (linked || waited) scheduleReconcile(t.id); } } // BB-334: a Linear issue doorbell also wakes the backlog binding of the @@ -11653,6 +11735,7 @@ export function createStewardsPlugin(deps: StewardsDeps): (bb: BbPluginApi) => P const maxWaitMs = playbook?.maxWaitMs ?? DEFAULT_MAX_WAIT_MS; const noteBeyondMaxWait = untilAt !== null && untilAt - now() > maxWaitMs && b.requeueUntilNoted !== untilAt; if (noteBeyondMaxWait) untilNote += ` The date is beyond the playbook's max-wait (${formatDuration(maxWaitMs)}); the project steward is told once.`; + const waitBefore = b.wait; b = store.patch(b.id, { status, statusAt: now(), @@ -11683,6 +11766,10 @@ export function createStewardsPlugin(deps: StewardsDeps): (bb: BbPluginApi) => P ...(mergeBehind === undefined ? {} : { mergeBehind }), ...(b.ending === "terminal" ? { followThrough, followThroughUntil } : {}), }); + // BB-333: a changed typed wait changes the github lease (a `pull`/ + // `review` ref joins or leaves it). This report may not run a + // reconcile that notices, so push the interest set here. + if ((waitBefore?.kind ?? null) !== (wait?.kind ?? null) || (waitBefore?.ref ?? null) !== (wait?.ref ?? null)) pushInterest(); // BB-255: the report supersedes any retry the plugin had pending, // deleting its queued row too. if (b.threadId !== null) await clearWakeRetry(b.threadId); diff --git a/plugins/stewards/store.test.ts b/plugins/stewards/store.test.ts index fe9f3058..0f5d608a 100644 --- a/plugins/stewards/store.test.ts +++ b/plugins/stewards/store.test.ts @@ -1,7 +1,7 @@ import Database from "better-sqlite3"; import { describe, expect, it } from "vitest"; import { z } from "zod"; -import { ACTOR_MIGRATIONS, ALERTMANAGER_LOOK_MIGRATIONS, BACKLOG_LOOK_MIGRATIONS, BACKLOG_SNAPSHOT_MIGRATIONS, CHECK_FINDING_MIGRATIONS, HUDDLE_MIGRATIONS, NON_ACTION_MIGRATIONS, TYPED_WAIT_MIGRATIONS, APPROVAL_MARKER_MIGRATIONS, ASSESSMENT_MIGRATIONS, ASSUMPTION_MIGRATIONS, BASE_MIGRATIONS, CHAIN_MIGRATIONS, DATED_WAIT_MIGRATIONS, DEPS_DECISION_MIGRATIONS, ENDED_READ_MIGRATIONS, ESCALATION_ROUTE_MIGRATIONS, GUIDANCE_MIGRATIONS, IDLE_MIGRATIONS, LESSON_COUNTER_MIGRATIONS, LESSON_PR_MIGRATIONS, LESSON_TIER_MIGRATIONS, MERGE_BEHIND_MIGRATIONS, MIGRATIONS, PULL_REQUESTER_MIGRATIONS, PULL_WATCH_MIGRATIONS, RATIONALE_MIGRATIONS, RETARGET_PULLS_MIGRATIONS, REVIEWER_MODEL_MIGRATIONS, REVIEWER_REQUEST_MIGRATIONS, SERVER_SETTINGS_MIGRATIONS, STACK_MIGRATIONS, STACKED_PULLS_MIGRATIONS, VALIDATION_MIGRATIONS, WAKE_RETRY_MIGRATIONS, WAKE_SOURCE_MIGRATIONS, WORKER_CHECK_MIGRATIONS, STANDUP_MIGRATIONS, STANDUP_BINDING_MIGRATIONS, STANDUP_DELEGATION_MIGRATIONS, STANDUP_COMPLETED_MIGRATIONS, WORKER_STOP_REASON_MIGRATIONS, WORKER_PARK_MIGRATIONS, createStore, lessonScope, statusRecordSchema, type Binding } from "./store.js"; +import { ACTOR_MIGRATIONS, ALERTMANAGER_LOOK_MIGRATIONS, BACKLOG_LOOK_MIGRATIONS, BACKLOG_SNAPSHOT_MIGRATIONS, CHECK_FINDING_MIGRATIONS, HUDDLE_MIGRATIONS, NON_ACTION_MIGRATIONS, TYPED_WAIT_MIGRATIONS, WAIT_DIVERGENCE_MIGRATIONS, APPROVAL_MARKER_MIGRATIONS, ASSESSMENT_MIGRATIONS, ASSUMPTION_MIGRATIONS, BASE_MIGRATIONS, CHAIN_MIGRATIONS, DATED_WAIT_MIGRATIONS, DEPS_DECISION_MIGRATIONS, ENDED_READ_MIGRATIONS, ESCALATION_ROUTE_MIGRATIONS, GUIDANCE_MIGRATIONS, IDLE_MIGRATIONS, LESSON_COUNTER_MIGRATIONS, LESSON_PR_MIGRATIONS, LESSON_TIER_MIGRATIONS, MERGE_BEHIND_MIGRATIONS, MIGRATIONS, PULL_REQUESTER_MIGRATIONS, PULL_WATCH_MIGRATIONS, RATIONALE_MIGRATIONS, RETARGET_PULLS_MIGRATIONS, REVIEWER_MODEL_MIGRATIONS, REVIEWER_REQUEST_MIGRATIONS, SERVER_SETTINGS_MIGRATIONS, STACK_MIGRATIONS, STACKED_PULLS_MIGRATIONS, VALIDATION_MIGRATIONS, WAKE_RETRY_MIGRATIONS, WAKE_SOURCE_MIGRATIONS, WORKER_CHECK_MIGRATIONS, STANDUP_MIGRATIONS, STANDUP_BINDING_MIGRATIONS, STANDUP_DELEGATION_MIGRATIONS, STANDUP_COMPLETED_MIGRATIONS, WORKER_STOP_REASON_MIGRATIONS, WORKER_PARK_MIGRATIONS, createStore, lessonScope, statusRecordSchema, type Binding } from "./store.js"; /** MIGRATIONS without the groups appended after the ones these order tests assert. */ const beforeActorMigrations = MIGRATIONS.slice(0, MIGRATIONS.indexOf(ACTOR_MIGRATIONS[0]!)); @@ -744,9 +744,10 @@ describe("the actor rename migration (BB-97)", () => { // and reviewer-request tables after that, BB-329 appended the worker park // column after that, BB-334 appended the backlog look column after that, // and BB-328 appended the check-findings table after that. BB-334's - // snapshot column followed, and BB-333's typed wait column after that. - // BB-335's non-action column is the newest. - const groups = [NON_ACTION_MIGRATIONS, TYPED_WAIT_MIGRATIONS, BACKLOG_SNAPSHOT_MIGRATIONS, CHECK_FINDING_MIGRATIONS, BACKLOG_LOOK_MIGRATIONS, WORKER_PARK_MIGRATIONS, REVIEWER_REQUEST_MIGRATIONS, PULL_REQUESTER_MIGRATIONS, ALERTMANAGER_LOOK_MIGRATIONS, STANDUP_COMPLETED_MIGRATIONS, STANDUP_DELEGATION_MIGRATIONS, STANDUP_BINDING_MIGRATIONS, STANDUP_MIGRATIONS, HUDDLE_MIGRATIONS, ENDED_READ_MIGRATIONS, ASSESSMENT_MIGRATIONS, STACKED_PULLS_MIGRATIONS, RETARGET_PULLS_MIGRATIONS, MERGE_BEHIND_MIGRATIONS, WAKE_RETRY_MIGRATIONS, DATED_WAIT_MIGRATIONS, ASSUMPTION_MIGRATIONS, WORKER_STOP_REASON_MIGRATIONS, DEPS_DECISION_MIGRATIONS, ESCALATION_ROUTE_MIGRATIONS, STACK_MIGRATIONS, PULL_WATCH_MIGRATIONS, SERVER_SETTINGS_MIGRATIONS, WAKE_SOURCE_MIGRATIONS, REVIEWER_MODEL_MIGRATIONS, LESSON_PR_MIGRATIONS, ACTOR_MIGRATIONS, APPROVAL_MARKER_MIGRATIONS]; + // snapshot column followed, BB-333's typed wait column after that, and + // BB-335's non-action column after that. BB-333 PR 2's wait-divergence + // table is the newest. + const groups = [WAIT_DIVERGENCE_MIGRATIONS, NON_ACTION_MIGRATIONS, TYPED_WAIT_MIGRATIONS, BACKLOG_SNAPSHOT_MIGRATIONS, CHECK_FINDING_MIGRATIONS, BACKLOG_LOOK_MIGRATIONS, WORKER_PARK_MIGRATIONS, REVIEWER_REQUEST_MIGRATIONS, PULL_REQUESTER_MIGRATIONS, ALERTMANAGER_LOOK_MIGRATIONS, STANDUP_COMPLETED_MIGRATIONS, STANDUP_DELEGATION_MIGRATIONS, STANDUP_BINDING_MIGRATIONS, STANDUP_MIGRATIONS, HUDDLE_MIGRATIONS, ENDED_READ_MIGRATIONS, ASSESSMENT_MIGRATIONS, STACKED_PULLS_MIGRATIONS, RETARGET_PULLS_MIGRATIONS, MERGE_BEHIND_MIGRATIONS, WAKE_RETRY_MIGRATIONS, DATED_WAIT_MIGRATIONS, ASSUMPTION_MIGRATIONS, WORKER_STOP_REASON_MIGRATIONS, DEPS_DECISION_MIGRATIONS, ESCALATION_ROUTE_MIGRATIONS, STACK_MIGRATIONS, PULL_WATCH_MIGRATIONS, SERVER_SETTINGS_MIGRATIONS, WAKE_SOURCE_MIGRATIONS, REVIEWER_MODEL_MIGRATIONS, LESSON_PR_MIGRATIONS, ACTOR_MIGRATIONS, APPROVAL_MARKER_MIGRATIONS]; let end = MIGRATIONS.length; for (const group of groups) { expect(MIGRATIONS.slice(end - group.length, end)).toEqual([...group]); @@ -845,9 +846,11 @@ describe("the escalation route rename migration (BB-209)", () => { // worker park column after that, BB-334's backlog look column after that, // BB-328's check-findings table after that, BB-334's backlog snapshot // column after that, and BB-333's typed wait column after that. BB-335's - // non-action column is last. - const withoutNonAction = MIGRATIONS.slice(0, -NON_ACTION_MIGRATIONS.length); - expect(MIGRATIONS.slice(-NON_ACTION_MIGRATIONS.length)).toEqual([...NON_ACTION_MIGRATIONS]); + // non-action column is last but for BB-333 PR 2's wait-divergence table. + const withoutWaitDivergence = MIGRATIONS.slice(0, -WAIT_DIVERGENCE_MIGRATIONS.length); + expect(MIGRATIONS.slice(-WAIT_DIVERGENCE_MIGRATIONS.length)).toEqual([...WAIT_DIVERGENCE_MIGRATIONS]); + const withoutNonAction = withoutWaitDivergence.slice(0, -NON_ACTION_MIGRATIONS.length); + expect(withoutWaitDivergence.slice(-NON_ACTION_MIGRATIONS.length)).toEqual([...NON_ACTION_MIGRATIONS]); const withoutTypedWait = withoutNonAction.slice(0, -TYPED_WAIT_MIGRATIONS.length); expect(withoutNonAction.slice(-TYPED_WAIT_MIGRATIONS.length)).toEqual([...TYPED_WAIT_MIGRATIONS]); const withoutBacklogSnapshot = withoutTypedWait.slice(0, -BACKLOG_SNAPSHOT_MIGRATIONS.length); diff --git a/plugins/stewards/store.ts b/plugins/stewards/store.ts index b6b18a06..3cc7fcaa 100644 --- a/plugins/stewards/store.ts +++ b/plugins/stewards/store.ts @@ -769,7 +769,7 @@ export const BACKLOG_SNAPSHOT_MIGRATIONS = [`ALTER TABLE bindings ADD COLUMN bac // wait, and a pre-migration build never reads the column. Appended after every // group already shipped, including BB-334's backlog columns and BB-328's // check-findings table, so no recorded index moves. PR 1 stores and renders -// it; nothing derives from it yet. +// it; PR 2 derives from it in shadow. export const TYPED_WAIT_MIGRATIONS = [`ALTER TABLE bindings ADD COLUMN wait TEXT`]; // BB-335: the reasons a declining branch records, one JSON object keyed by the @@ -779,7 +779,20 @@ export const TYPED_WAIT_MIGRATIONS = [`ALTER TABLE bindings ADD COLUMN wait TEXT // including BB-333's typed wait column, so no recorded index moves. export const NON_ACTION_MIGRATIONS = [`ALTER TABLE assessments ADD COLUMN non_actions TEXT`]; -export const MIGRATIONS = [...BASE_MIGRATIONS, ...CHAIN_MIGRATIONS, ...LESSON_COUNTER_MIGRATIONS, ...WORKER_CHECK_MIGRATIONS, ...DECLINED_PR_MIGRATIONS, ...DEPLOYMENT_MIGRATIONS, ...FINISHED_AND_DIGEST_MIGRATIONS, ...LESSON_TIER_MIGRATIONS, ...RATIONALE_MIGRATIONS, ...VALIDATION_MIGRATIONS, ...GUIDANCE_MIGRATIONS, ...IDLE_MIGRATIONS, ...APPROVAL_MARKER_MIGRATIONS, ...ACTOR_MIGRATIONS, ...LESSON_PR_MIGRATIONS, ...REVIEWER_MODEL_MIGRATIONS, ...WAKE_SOURCE_MIGRATIONS, ...SERVER_SETTINGS_MIGRATIONS, ...PULL_WATCH_MIGRATIONS, ...STACK_MIGRATIONS, ...ESCALATION_ROUTE_MIGRATIONS, ...DEPS_DECISION_MIGRATIONS, ...WORKER_STOP_REASON_MIGRATIONS, ...ASSUMPTION_MIGRATIONS, ...DATED_WAIT_MIGRATIONS, ...WAKE_RETRY_MIGRATIONS, ...MERGE_BEHIND_MIGRATIONS, ...RETARGET_PULLS_MIGRATIONS, ...STACKED_PULLS_MIGRATIONS, ...ASSESSMENT_MIGRATIONS, ...ENDED_READ_MIGRATIONS, ...HUDDLE_MIGRATIONS, ...STANDUP_MIGRATIONS, ...STANDUP_BINDING_MIGRATIONS, ...STANDUP_DELEGATION_MIGRATIONS, ...STANDUP_COMPLETED_MIGRATIONS, ...ALERTMANAGER_LOOK_MIGRATIONS, ...PULL_REQUESTER_MIGRATIONS, ...REVIEWER_REQUEST_MIGRATIONS, ...WORKER_PARK_MIGRATIONS, ...BACKLOG_LOOK_MIGRATIONS, ...CHECK_FINDING_MIGRATIONS, ...BACKLOG_SNAPSHOT_MIGRATIONS, ...TYPED_WAIT_MIGRATIONS, ...NON_ACTION_MIGRATIONS]; +// BB-333 PR 2: the last derived-vs-flag state per binding, so the +// `noWakeSource` divergence row is written only when the disagreement starts. +// A new table appended after BB-335's non-action column, so no recorded index +// moves. +export const WAIT_DIVERGENCE_MIGRATIONS = [ + `CREATE TABLE IF NOT EXISTS wait_divergence_state ( + binding_id TEXT PRIMARY KEY, + derived INTEGER NOT NULL, + reported INTEGER NOT NULL, + at INTEGER NOT NULL + )`, +]; + +export const MIGRATIONS = [...BASE_MIGRATIONS, ...CHAIN_MIGRATIONS, ...LESSON_COUNTER_MIGRATIONS, ...WORKER_CHECK_MIGRATIONS, ...DECLINED_PR_MIGRATIONS, ...DEPLOYMENT_MIGRATIONS, ...FINISHED_AND_DIGEST_MIGRATIONS, ...LESSON_TIER_MIGRATIONS, ...RATIONALE_MIGRATIONS, ...VALIDATION_MIGRATIONS, ...GUIDANCE_MIGRATIONS, ...IDLE_MIGRATIONS, ...APPROVAL_MARKER_MIGRATIONS, ...ACTOR_MIGRATIONS, ...LESSON_PR_MIGRATIONS, ...REVIEWER_MODEL_MIGRATIONS, ...WAKE_SOURCE_MIGRATIONS, ...SERVER_SETTINGS_MIGRATIONS, ...PULL_WATCH_MIGRATIONS, ...STACK_MIGRATIONS, ...ESCALATION_ROUTE_MIGRATIONS, ...DEPS_DECISION_MIGRATIONS, ...WORKER_STOP_REASON_MIGRATIONS, ...ASSUMPTION_MIGRATIONS, ...DATED_WAIT_MIGRATIONS, ...WAKE_RETRY_MIGRATIONS, ...MERGE_BEHIND_MIGRATIONS, ...RETARGET_PULLS_MIGRATIONS, ...STACKED_PULLS_MIGRATIONS, ...ASSESSMENT_MIGRATIONS, ...ENDED_READ_MIGRATIONS, ...HUDDLE_MIGRATIONS, ...STANDUP_MIGRATIONS, ...STANDUP_BINDING_MIGRATIONS, ...STANDUP_DELEGATION_MIGRATIONS, ...STANDUP_COMPLETED_MIGRATIONS, ...ALERTMANAGER_LOOK_MIGRATIONS, ...PULL_REQUESTER_MIGRATIONS, ...REVIEWER_REQUEST_MIGRATIONS, ...WORKER_PARK_MIGRATIONS, ...BACKLOG_LOOK_MIGRATIONS, ...CHECK_FINDING_MIGRATIONS, ...BACKLOG_SNAPSHOT_MIGRATIONS, ...TYPED_WAIT_MIGRATIONS, ...NON_ACTION_MIGRATIONS, ...WAIT_DIVERGENCE_MIGRATIONS]; // ---- values kept as JSON ------------------------------------------------------------ @@ -1963,6 +1976,22 @@ export function createStore(db: Database.Database, now: () => number, options: S const pruneDivergencesByCountStmt = lazy<[number], unknown>( "DELETE FROM assessment_divergences WHERE id NOT IN (SELECT id FROM assessment_divergences ORDER BY last_seen_at DESC, id LIMIT ?)", ); + // BB-333: the last derived-vs-flag state per binding, so a typed-wait + // divergence row is written only when the disagreement starts. The + // `(binding_id, surface, old, new)` unique constraint caps the rows at two + // per binding, so no separate per-binding prune is needed. Prepared lazily: a + // store from before this group has no table yet. + const getWaitDivergenceStateStmt = lazy<[string], Record>("SELECT * FROM wait_divergence_state WHERE binding_id = ?"); + const upsertWaitDivergenceStateStmt = lazy<[string, number, number, number], unknown>( + "INSERT INTO wait_divergence_state (binding_id, derived, reported, at) VALUES (?, ?, ?, ?) ON CONFLICT (binding_id) DO UPDATE SET derived = excluded.derived, reported = excluded.reported, at = excluded.at", + ); + // Keep the open row's `last_seen_at` fresh while the disagreement stands, so + // the age prune cannot drop it and leave the state table saying "disagree" + // with no row to show for it. Scoped to the state table's pair, so a second + // (other-direction) row for the same binding is not touched. + const touchWaitDivergenceStmt = lazy<[number, string, string, string, string], unknown>( + "UPDATE assessment_divergences SET last_seen_at = ?, inputs = ? WHERE binding_id = ? AND surface = 'noWakeSource' AND old = ? AND new = ?", + ); // BB-251: assumptions, prepared lazily for the same reason: a store from // before this group has no `assumptions` table yet. @@ -3631,6 +3660,56 @@ export function createStore(db: Database.Database, now: () => number, options: S removed += pruneDivergencesByCountStmt().run(maxRows).changes; return removed; }, + /** + * Record a typed-wait divergence from the report-time `noWakeSource` flag + * (BB-333). A `noWakeSource` row is written only when the disagreement + * starts; while it stands the open row's `last_seen_at` is refreshed so the + * age prune cannot drop it, and when it stops the row is noted and the + * state cleared without adding a row that would read as a divergence. The + * `(binding_id, surface, old, new)` unique constraint caps the rows at two + * per binding. Returns whether a divergence row was written. + */ + recordWaitDivergence(fields: { bindingId: string; derived: boolean; reported: boolean; waitKind: string | null; waitLive: boolean; inputs: Record }, at = now()): boolean { + const derived = fields.derived ? 1 : 0; + const reported = fields.reported ? 1 : 0; + const disagree = derived !== reported; + const prev = getWaitDivergenceStateStmt().get(fields.bindingId); + const prevDerived = prev === undefined ? null : Number(prev.derived); + const prevReported = prev === undefined ? null : Number(prev.reported); + const prevDisagree = prevDerived === null ? null : prevDerived !== prevReported; + const old = (v: number): string => (v === 1 ? "no-wake-source" : "source-or-no-wait"); + const note = (transition: string): string => JSON.stringify({ ...fields.inputs, waitKind: fields.waitKind, waitLive: fields.waitLive, transition }); + const touch = (transition: string, d: number, r: number): void => { + touchWaitDivergenceStmt().run(at, note(transition), fields.bindingId, old(r), old(d)); + }; + const pairChanged = prevDerived !== derived || prevReported !== reported; + if (!pairChanged) { + // No transition. An ongoing disagreement keeps its open row fresh so + // the age prune cannot drop it; an agreement writes nothing. + if (disagree) touch("ongoing", prevDerived!, prevReported!); + return false; + } + // The pair changed. If the previous pair disagreed, that disagreement + // stopped: note it on the old row, never as a row of its own (its `old` + // would equal its `new`). A change from one disagreement to the other is + // that stop plus a new start. + if (prevDisagree === true) touch("stop", prevDerived!, prevReported!); + upsertWaitDivergenceStateStmt().run(fields.bindingId, derived, reported, at); + if (!disagree) return false; + // A start: the first observation of a disagreement, a move from + // agreement to disagreement, or a flip to the other direction. + insertDivergenceStmt().run( + newId("dvg"), + fields.bindingId, + "noWakeSource", + old(reported), + old(derived), + note("start"), + at, + at, + ); + return true; + }, /** * Forget a deleted project: its bindings and everything keyed by them * (escalations, workers, needs, approvals, fingerprints, queued answers), @@ -3644,7 +3723,7 @@ export function createStore(db: Database.Database, now: () => number, options: S const ids = db.prepare<[string], { id: string }>("SELECT id FROM bindings WHERE project_id = ?").all(projectId).map((r) => r.id); const byBinding = "SELECT id FROM bindings WHERE project_id = ?"; db.prepare(`DELETE FROM secret_requesters WHERE thread_id IN (SELECT thread_id FROM workers WHERE binding_id IN (${byBinding}))`).run(projectId); - for (const table of ["escalations", "workers", "needs", "project_approvals", "assessments", "assessment_divergences"]) db.prepare(`DELETE FROM ${table} WHERE binding_id IN (${byBinding})`).run(projectId); + for (const table of ["escalations", "workers", "needs", "project_approvals", "assessments", "assessment_divergences", "wait_divergence_state"]) db.prepare(`DELETE FROM ${table} WHERE binding_id IN (${byBinding})`).run(projectId); db.prepare(`DELETE FROM pending_deliveries WHERE origin_binding_id IN (${byBinding})`).run(projectId); for (const table of ["ops_fingerprints", "deps_decisions", "lessons", "guidance_citations", "assumptions", "bindings", "project_stewards"]) db.prepare(`DELETE FROM ${table} WHERE project_id = ?`).run(projectId); options.onProjectChange?.(projectId, ["purged"]);