stewards: derive liveness, idle and stall from the typed wait (BB-333)

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<NonNullable<EntitySnapshot["pull"]>> = {}) => {
+    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<AssessmentInput["binding"]["wait"]>, over: Partial<AssessmentInput> = {}) => 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" });
+  });
+});

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<string, EntitySnapshot>;
+  /** 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<string, readonly string[]> | undefined,
 const isOwn = (e: EntitySnapshot, own: ReadonlySet<string>): 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;
 }

plugins/stewards/migrations.snapshot.json

@@ -139,5 +139,6 @@
   "c1377801bd4fe44be9cc5fc10c5aca75c9455359b3d5d646d0fb23318a032124",
   "d856f2127e2c5788a01c5ca80b216d1dc9be2a67bb273ad9082bbc69a279ce9a",
   "f1636e9715f38e19e3c21cc7d3008c4f8d734ec26e03eb983961acb7d4cc8949",
-  "5aeed47ee4b133a8d26b01316c545a6f4c49b3c73fb9511cb4c2b9eae55a31fa"
+  "5aeed47ee4b133a8d26b01316c545a6f4c49b3c73fb9511cb4c2b9eae55a31fa",
+  "ded6270fedbb7d853a412b2167f164b3886f38ed975c124d83b40f6d625500ad"
 ]