feat(miner-governor): closed-loop discovery re-entry trigger (#2338)

packages/gittensory-engine/src/index.ts

@@ -244,6 +244,15 @@ export {
   type SubmissionGateDecision,
   type SubmissionGateMode,
 } from "./miner/submission-gate.js";
+export {
+  DEFAULT_MAX_CONSECUTIVE_DISENGAGEMENTS,
+  DEFAULT_MAX_REENTRIES_PER_HOUR,
+  DEFAULT_MAX_REENTRIES_PER_SESSION,
+  shouldReenter,
+  type LoopReentryCandidate,
+  type LoopReentryDecision,
+  type LoopReentryOutcome,
+} from "./miner/loop-reentry-policy.js";
 export {
   codingAgentModeExecutes,
   isGlobalMinerCodingAgentPause,

packages/gittensory-engine/src/miner/loop-reentry-policy.ts

@@ -0,0 +1,70 @@
+// Closed-loop discovery re-entry policy (#2338): the pure decision half of "on a resolved outcome (merged, or
+// rejected-and-disengaged), automatically re-invoke discovery to select the next candidate." Deliberately split
+// from the miner-side orchestrator (packages/gittensory-miner/lib/loop-reentry.js), which owns the REAL IO --
+// reading recent event-ledger history to compute the tallies this policy consumes, dequeuing the next
+// candidate, and transitioning run-state -- mirroring this session's established engine (pure) / miner-lib
+// (stateful) split for every other governor primitive.
+//
+// TOP SLOP-AT-SCALE RISK: this issue's own framing calls out "a bug here (re-entering too fast, ignoring a
+// circuit-breaker, or looping on a permanently-rejected repo) is the top slop-at-scale risk for the whole miner
+// subsystem." Both failure modes get an INDEPENDENT hard ceiling here, neither one masking the other:
+//   - A per-repo circuit breaker: N consecutive disengaged (rejected) outcomes on the SAME repo pauses further
+//     re-entry for that repo, regardless of how much of the hour/session rate budget remains.
+//   - A hard rate/session cap: independent of any repo's own history, a conservative ceiling on how many
+//     re-entries may fire in a rolling hour or across the whole session.
+// Both reasons are collected (not short-circuited) so a caller logging the decision sees every ceiling that
+// was hit, not just the first one checked.
+
+/** The terminal outcome that just resolved for the repo the caller is considering re-entering on. */
+export type LoopReentryOutcome = "merged" | "disengaged" | "other";
+
+export const DEFAULT_MAX_CONSECUTIVE_DISENGAGEMENTS = 3;
+export const DEFAULT_MAX_REENTRIES_PER_HOUR = 4;
+export const DEFAULT_MAX_REENTRIES_PER_SESSION = 20;
+
+export type LoopReentryCandidate = {
+  repoFullName: string;
+  outcome: LoopReentryOutcome;
+  /** Caller-computed count of CONSECUTIVE `"disengaged"` outcomes for this repo, ending with (and including,
+   *  when `outcome === "disengaged"`) this one. Any non-disengaged outcome resets this to 0 -- the caller owns
+   *  that computation, this policy only consumes the resulting integer (mirrors `reputation-throttle.ts`'s
+   *  caller-supplied `RepoOutcomeHistory`). */
+  consecutiveDisengagements: number;
+  maxConsecutiveDisengagements?: number | undefined;
+  /** Caller-tracked re-entry counters for the hard rate/session cap -- independent of the per-repo circuit
+   *  breaker above. */
+  reentriesThisHour: number;
+  maxReentriesPerHour?: number | undefined;
+  reentriesThisSession: number;
+  maxReentriesPerSession?: number | undefined;
+};
+
+export type LoopReentryDecision = {
+  reenter: boolean;
+  /** Always populated when `reenter` is `false`; every ceiling that was hit, not just the first. */
+  reasons: string[];
+};
+
+/**
+ * Decide whether the loop may re-enter discovery for this repo. Pure; identical inputs always yield the
+ * identical decision. `outcome === "merged"` alone never bypasses the rate/session cap -- a healthy repo can
+ * still be rate-limited if the operator-wide ceiling is already spent.
+ */
+export function shouldReenter(candidate: LoopReentryCandidate): LoopReentryDecision {
+  const reasons: string[] = [];
+  const maxConsecutiveDisengagements = candidate.maxConsecutiveDisengagements ?? DEFAULT_MAX_CONSECUTIVE_DISENGAGEMENTS;
+  const maxReentriesPerHour = candidate.maxReentriesPerHour ?? DEFAULT_MAX_REENTRIES_PER_HOUR;
+  const maxReentriesPerSession = candidate.maxReentriesPerSession ?? DEFAULT_MAX_REENTRIES_PER_SESSION;
+
+  if (candidate.outcome === "disengaged" && candidate.consecutiveDisengagements >= maxConsecutiveDisengagements) {
+    reasons.push(`repo_paused_after_consecutive_disengagements:${candidate.consecutiveDisengagements}>=${maxConsecutiveDisengagements}`);
+  }
+  if (candidate.reentriesThisHour >= maxReentriesPerHour) {
+    reasons.push(`hourly_reentry_cap_reached:${candidate.reentriesThisHour}>=${maxReentriesPerHour}`);
+  }
+  if (candidate.reentriesThisSession >= maxReentriesPerSession) {
+    reasons.push(`session_reentry_cap_reached:${candidate.reentriesThisSession}>=${maxReentriesPerSession}`);
+  }
+
+  return { reenter: reasons.length === 0, reasons };
+}

packages/gittensory-engine/test/loop-reentry-policy.test.ts

@@ -0,0 +1,114 @@
+import assert from "node:assert/strict";
+import { test } from "node:test";
+
+import {
+  DEFAULT_MAX_CONSECUTIVE_DISENGAGEMENTS,
+  DEFAULT_MAX_REENTRIES_PER_HOUR,
+  DEFAULT_MAX_REENTRIES_PER_SESSION,
+  shouldReenter,
+  type LoopReentryCandidate,
+} from "../dist/index.js";
+
+function baseCandidate(overrides: Partial<LoopReentryCandidate> = {}): LoopReentryCandidate {
+  return {
+    repoFullName: "acme/widgets",
+    outcome: "merged",
+    consecutiveDisengagements: 0,
+    reentriesThisHour: 0,
+    reentriesThisSession: 0,
+    ...overrides,
+  };
+}
+
+test("barrel: the public entrypoint re-exports the loop-reentry policy (#2338)", () => {
+  assert.equal(typeof shouldReenter, "function");
+  assert.equal(typeof DEFAULT_MAX_CONSECUTIVE_DISENGAGEMENTS, "number");
+});
+
+test("a merged outcome with every counter well within limits re-enters cleanly", () => {
+  const decision = shouldReenter(baseCandidate({ outcome: "merged" }));
+  assert.deepEqual(decision, { reenter: true, reasons: [] });
+});
+
+test("an 'other' outcome (neither merged nor disengaged) is never subject to the per-repo circuit breaker", () => {
+  const decision = shouldReenter(baseCandidate({ outcome: "other" }));
+  assert.deepEqual(decision, { reenter: true, reasons: [] });
+});
+
+test("circuit breaker: a disengaged outcome at or beyond the consecutive-disengagement ceiling pauses the repo", () => {
+  const decision = shouldReenter(baseCandidate({ outcome: "disengaged", consecutiveDisengagements: 3, maxConsecutiveDisengagements: 3 }));
+  assert.equal(decision.reenter, false);
+  assert.deepEqual(decision.reasons, ["repo_paused_after_consecutive_disengagements:3>=3"]);
+});
+
+test("circuit breaker: a disengaged outcome below the ceiling still re-enters", () => {
+  const decision = shouldReenter(baseCandidate({ outcome: "disengaged", consecutiveDisengagements: 2, maxConsecutiveDisengagements: 3 }));
+  assert.equal(decision.reenter, true);
+});
+
+test("circuit breaker: a HIGH consecutiveDisengagements count never pauses a repo whose outcome ISN'T disengaged", () => {
+  // Exercises the && short-circuit's left-false side distinctly from the right-side threshold check -- a
+  // repo could have a high historical tally but just landed a merge, which must not be treated as a pause.
+  const decision = shouldReenter(baseCandidate({ outcome: "merged", consecutiveDisengagements: 99, maxConsecutiveDisengagements: 3 }));
+  assert.equal(decision.reenter, true);
+});
+
+test("rate cap: an hourly re-entry ceiling at or beyond the limit blocks, independent of repo history", () => {
+  const decision = shouldReenter(baseCandidate({ reentriesThisHour: 4, maxReentriesPerHour: 4 }));
+  assert.equal(decision.reenter, false);
+  assert.deepEqual(decision.reasons, ["hourly_reentry_cap_reached:4>=4"]);
+});
+
+test("rate cap: an hourly count below the limit does not block", () => {
+  const decision = shouldReenter(baseCandidate({ reentriesThisHour: 3, maxReentriesPerHour: 4 }));
+  assert.equal(decision.reenter, true);
+});
+
+test("rate cap: a session re-entry ceiling at or beyond the limit blocks, independent of the hourly cap", () => {
+  const decision = shouldReenter(baseCandidate({ reentriesThisSession: 20, maxReentriesPerSession: 20 }));
+  assert.equal(decision.reenter, false);
+  assert.deepEqual(decision.reasons, ["session_reentry_cap_reached:20>=20"]);
+});
+
+test("rate cap: a session count below the limit does not block", () => {
+  const decision = shouldReenter(baseCandidate({ reentriesThisSession: 19, maxReentriesPerSession: 20 }));
+  assert.equal(decision.reenter, true);
+});
+
+test("every ceiling that is exceeded is reported, not just the first one checked", () => {
+  const decision = shouldReenter(
+    baseCandidate({
+      outcome: "disengaged",
+      consecutiveDisengagements: 5,
+      maxConsecutiveDisengagements: 3,
+      reentriesThisHour: 10,
+      maxReentriesPerHour: 4,
+      reentriesThisSession: 30,
+      maxReentriesPerSession: 20,
+    }),
+  );
+  assert.equal(decision.reenter, false);
+  assert.equal(decision.reasons.length, 3);
+});
+
+test("default thresholds apply when the candidate omits its own overrides", () => {
+  const justUnderDefault = shouldReenter(
+    baseCandidate({ outcome: "disengaged", consecutiveDisengagements: DEFAULT_MAX_CONSECUTIVE_DISENGAGEMENTS - 1 }),
+  );
+  assert.equal(justUnderDefault.reenter, true);
+
+  const atDefault = shouldReenter(baseCandidate({ outcome: "disengaged", consecutiveDisengagements: DEFAULT_MAX_CONSECUTIVE_DISENGAGEMENTS }));
+  assert.equal(atDefault.reenter, false);
+
+  const atHourlyDefault = shouldReenter(baseCandidate({ reentriesThisHour: DEFAULT_MAX_REENTRIES_PER_HOUR }));
+  assert.equal(atHourlyDefault.reenter, false);
+
+  const atSessionDefault = shouldReenter(baseCandidate({ reentriesThisSession: DEFAULT_MAX_REENTRIES_PER_SESSION }));
+  assert.equal(atSessionDefault.reenter, false);
+});
+
+test("a caller-supplied threshold overrides the default rather than being ignored", () => {
+  // A count that would pass under the DEFAULT ceiling must still block under a stricter caller override.
+  const decision = shouldReenter(baseCandidate({ outcome: "disengaged", consecutiveDisengagements: 1, maxConsecutiveDisengagements: 1 }));
+  assert.equal(decision.reenter, false);
+});

packages/gittensory-miner/lib/loop-reentry.d.ts

@@ -0,0 +1,48 @@
+export const LOOP_REENTRY_DECISION_EVENT: "loop_reentry_decision";
+
+export type LoopReentryOutcome = "merged" | "disengaged" | "other";
+
+export type LoopReentryCandidateInput = {
+  repoFullName: string;
+  outcome: LoopReentryOutcome;
+  maxConsecutiveDisengagements?: number;
+  maxReentriesPerHour?: number;
+  maxReentriesPerSession?: number;
+};
+
+export interface LoopReentryEventLedger {
+  appendEvent(event: { type: string; repoFullName?: string; payload: Record<string, unknown> }): { id: number; seq: number; type: string; repoFullName: string | null; payload: Record<string, unknown>; createdAt: string };
+  readEvents(filter?: { since?: number; repoFullName?: string }): Array<{ type: string; repoFullName?: string | null; payload?: Record<string, unknown>; createdAt: string }>;
+}
+
+export interface LoopReentryPortfolioQueue {
+  dequeueNext(): { repoFullName: string; identifier: string; priority: number; status: string; enqueuedAt: string } | null;
+}
+
+export interface LoopReentryRunState {
+  setRunState(repoFullName: string, state: string): unknown;
+}
+
+export type LoopReentryDeps = {
+  eventLedger: LoopReentryEventLedger;
+  portfolioQueue: LoopReentryPortfolioQueue;
+  runState?: LoopReentryRunState;
+  nowMs?: number;
+  sessionStartMs?: number;
+  /** The just-completed cycle's read-only summary (loop-closure.js's `buildLoopClosureSummary`), threaded
+   *  through verbatim into the audit event's payload for traceability. Not used to compute the circuit-
+   *  breaker/rate-cap tallies -- see loop-reentry.js's own comment on why. */
+  loopSummary?: unknown;
+};
+
+export type LoopReentryResult = {
+  decision: { reenter: boolean; reasons: string[] };
+  dequeued: { repoFullName: string; identifier: string; priority: number; status: string; enqueuedAt: string } | null;
+  event: { id: number; seq: number; type: string; repoFullName: string | null; payload: Record<string, unknown>; createdAt: string };
+};
+
+export function countConsecutiveDisengagements(eventLedger: LoopReentryEventLedger, repoFullName: string): number;
+
+export function countReentriesSince(eventLedger: LoopReentryEventLedger, sinceMs: number): number;
+
+export function attemptLoopReentry(candidate: LoopReentryCandidateInput, deps: LoopReentryDeps): LoopReentryResult;

packages/gittensory-miner/lib/loop-reentry.js

@@ -0,0 +1,121 @@
+import { shouldReenter } from "@jsonbored/gittensory-engine";
+
+import { readPrOutcomes } from "./pr-outcome.js";
+
+// Closed-loop discovery re-entry orchestrator (#2338): the real-IO half of "on a resolved outcome (merged, or
+// rejected-and-disengaged), automatically re-invoke discovery to select the next candidate." The DECISION
+// itself (shouldReenter, @jsonbored/gittensory-engine) is pure; this module owns everything that decision
+// needs real state for -- reading the repo's own pr_outcome history to compute the per-repo consecutive-
+// disengagement tally, reading recent re-entry events for the hourly/session rate cap, and (only when allowed)
+// actually dequeuing the next candidate and transitioning run-state.
+//
+// NOT WIRED INTO ANY AUTOMATIC SCHEDULE: per this issue's own "manual owner sign-off before enabling by
+// default in any profile" deliverable, this is a callable function ready for that sign-off -- it is not invoked
+// by manage-poll.js or any cron/scheduler as part of this change.
+//
+// AUDITABILITY: every call appends exactly one `loop_reentry_decision` event to the ledger, whether or not the
+// decision allowed re-entry, so the full decision trail (including every suppressed re-entry and why) survives
+// independently of this function's own return value.
+
+export const LOOP_REENTRY_DECISION_EVENT = "loop_reentry_decision";
+const HOUR_MS = 60 * 60 * 1000;
+
+/** A `pr_outcome` "closed" decision is this module's practical proxy for "disengaged" -- pr-outcome.js's own
+ *  vocabulary is exactly `"merged" | "closed"` (no separate "disengaged" literal); a PR that closed without
+ *  merging IS the rejected/disengaged case rejection-state-machine.js's own `isRejectedPr` checks for. */
+function isDisengagedOutcome(outcome) {
+  return outcome?.decision === "closed";
+}
+
+/**
+ * Count a repo's CONSECUTIVE disengaged (closed-without-merge) PR outcomes, walking backward from the most
+ * recently recorded PR for that repo until a merged outcome breaks the streak (or history runs out).
+ */
+export function countConsecutiveDisengagements(eventLedger, repoFullName) {
+  const outcomes = [...readPrOutcomes(eventLedger, { repoFullName }).values()];
+  let count = 0;
+  for (let i = outcomes.length - 1; i >= 0; i -= 1) {
+    if (!isDisengagedOutcome(outcomes[i])) break;
+    count += 1;
+  }
+  return count;
+}
+
+/** Count prior re-entries (successful, i.e. `reentered: true`) recorded at or after `sinceMs`. */
+export function countReentriesSince(eventLedger, sinceMs) {
+  return eventLedger
+    .readEvents({})
+    .filter((event) => event.type === LOOP_REENTRY_DECISION_EVENT && event.payload?.reentered === true && Date.parse(event.createdAt) >= sinceMs)
+    .length;
+}
+
+/**
+ * Evaluate and (if allowed) PERFORM re-entry for one resolved outcome: reads real history to compute the
+ * circuit-breaker and rate-cap tallies, consults the pure `shouldReenter` policy, and -- only when it allows --
+ * dequeues the next candidate and transitions run-state to `"discovering"`. Always appends exactly one audit
+ * event. Fails closed (throws) on a malformed candidate or missing required dependency, mirroring
+ * `recordManagePollSnapshot`'s own validation style.
+ *
+ * @param {{ repoFullName: string, outcome: "merged"|"disengaged"|"other", maxConsecutiveDisengagements?: number, maxReentriesPerHour?: number, maxReentriesPerSession?: number }} candidate
+ * @param {{ eventLedger: object, portfolioQueue: object, runState?: object, nowMs?: number, sessionStartMs?: number }} deps
+ */
+export function attemptLoopReentry(candidate, deps) {
+  if (!candidate || typeof candidate !== "object") throw new Error("invalid_loop_reentry_candidate");
+  const repoFullName = typeof candidate.repoFullName === "string" ? candidate.repoFullName.trim() : "";
+  if (!repoFullName) throw new Error("invalid_repo_full_name");
+  if (!["merged", "disengaged", "other"].includes(candidate.outcome)) throw new Error("invalid_outcome");
+
+  if (!deps || typeof deps !== "object") throw new Error("invalid_loop_reentry_deps");
+  const { eventLedger, portfolioQueue, runState, nowMs = Date.now(), sessionStartMs = 0 } = deps;
+  if (!eventLedger || typeof eventLedger.appendEvent !== "function" || typeof eventLedger.readEvents !== "function") {
+    throw new Error("invalid_event_ledger");
+  }
+  if (!portfolioQueue || typeof portfolioQueue.dequeueNext !== "function") {
+    throw new Error("invalid_portfolio_queue");
+  }
+
+  const consecutiveDisengagements = countConsecutiveDisengagements(eventLedger, repoFullName);
+  const reentriesThisHour = countReentriesSince(eventLedger, nowMs - HOUR_MS);
+  const reentriesThisSession = countReentriesSince(eventLedger, sessionStartMs);
+
+  const decision = shouldReenter({
+    repoFullName,
+    outcome: candidate.outcome,
+    consecutiveDisengagements,
+    maxConsecutiveDisengagements: candidate.maxConsecutiveDisengagements,
+    reentriesThisHour,
+    maxReentriesPerHour: candidate.maxReentriesPerHour,
+    reentriesThisSession,
+    maxReentriesPerSession: candidate.maxReentriesPerSession,
+  });
+
+  let dequeued = null;
+  if (decision.reenter) {
+    dequeued = portfolioQueue.dequeueNext();
+    if (runState && typeof runState.setRunState === "function") {
+      runState.setRunState(repoFullName, "discovering");
+    }
+  }
+
+  const event = eventLedger.appendEvent({
+    type: LOOP_REENTRY_DECISION_EVENT,
+    repoFullName,
+    payload: {
+      outcome: candidate.outcome,
+      reentered: decision.reenter,
+      reasons: decision.reasons,
+      consecutiveDisengagements,
+      reentriesThisHour,
+      reentriesThisSession,
+      dequeuedIdentifier: dequeued ? dequeued.identifier : null,
+      // The just-completed cycle's read-only summary (loop-closure.js's buildLoopClosureSummary), when the
+      // caller supplies one -- threaded through verbatim for audit traceability. Optional: the circuit-breaker
+      // and rate-cap tallies above are computed directly from pr-outcome/event-ledger history (a
+      // LoopClosureSummary's own byType COUNTS aren't detailed enough to derive a per-repo consecutive-
+      // disengagement streak from), so this is context, not a computational input.
+      loopSummary: deps.loopSummary ?? null,
+    },
+  });
+
+  return { decision, dequeued, event };
+}

packages/gittensory-miner/lib/pr-outcome.d.ts

@@ -1,3 +1,5 @@
+import type { AppendEventInput, LedgerEntry } from "./event-ledger.js";
+
 export const MINER_PR_OUTCOME_EVENT: "pr_outcome";
 export const MINER_PR_OUTCOME_DECISIONS: readonly ["merged", "closed"];
 
@@ -20,8 +22,9 @@ export interface PrOutcomeInput {
 
 export interface RecordPrOutcomeOptions {
   /** Optional at the type level so a caller can pass an unusable ledger to exercise the fail-closed guard; the
-   *  writer throws `invalid_event_ledger` at runtime when this is absent or lacks `appendEvent`. */
-  eventLedger?: { appendEvent(event: { type: string; repoFullName: string; payload: unknown }): unknown };
+   *  writer throws `invalid_event_ledger` at runtime when this is absent or lacks `appendEvent`. Reuses the
+   *  real EventLedger#appendEvent signature so a genuine EventLedger (not just a same-shaped stub) type-checks. */
+  eventLedger?: { appendEvent(event: AppendEventInput): LedgerEntry };
 }
 
 export interface PrOutcomeLedgerReader {