ceee5f5
@@ -27,6 +27,25 @@
bottleneck; the knobs stay on Settings → Capacity. The same sentence
(`verdict`) is the terminal's admin-only Flow command and the MCP `flow` tool,
which also returns the stages; anyone but a forge admin is refused.
## Rebases
Approved work lands before new work is built, in the check lane too (#6140).
A check on a pull whose last builder's run brought it up to date (a
`conflict` run; a review after it does not count) goes before any other
queued check (`checks/finishing_first.ts`). So the room for more rebases
(`crew/conflict_room.ts`) is the check slots minus the rebases' checks and the
bring-up-to-date runs in flight; a new build's check does not stand in front
of a rebase and takes no room. When the pull's own builder already has a run
in flight, the finishing pass (`crew/finishing.ts`) lends the update to an
idle builder, round-robin by name; the run's agent is who rebased, and the log
line names whose pull it was.
The tick line's `lane` field, the `Rebases:` line on `/flow` and in the
terminal's Flow, and `rebases` in the MCP `flow` tool all read
`conflicted N, rebasing M, room R`: the conflicted pulls (with the approved
ones still waiting for room, which have no run yet), the bring-up-to-date runs
and their checks in flight, and the runs the lane has room to start now.
## Throughput
`src/flow/throughput.ts` counts what each pipeline stage let out over the last
@@ -0,0 +1,52 @@
/**
 * Land before you build, in the check lane (#6140).
 *
 * A pull brought up to date (a `conflict` run: main moved, the crew merged
 * it in and pushed) is nearly done work; its check is the last thing
 * between it and landing. When a landing conflicts dozens of approved
 * pulls at once, new builds' checks kept the lane full, and the rebases
 * drained one at a time behind them. So a rebase's check goes before any
 * other in the lane, and the conflict room (`crew/conflict_room.ts`)
 * counts only these checks as standing in front of the next rebase.
 *
 * A check is a rebase's when the last builder's run on its pull — any
 * agent run but a review, which comes after the push — was a `conflict`
 * run. Read from the runs alone, so every forge sharing the database
 * agrees on the order.
 */
import type { Database } from "bun:sqlite";
/** The pulls (issue ids), among `issueIds`, whose last builder's run was a bring-up-to-date run. */
export function rebasedPulls(db: Database, issueIds: readonly string[]): Set<string> {
  const ids = [...new Set(issueIds)];
  if (ids.length === 0) return new Set();
  const marks = ids.map((_, i) => `?${i + 1}`).join(",");
  const rows = db
    .query<{ issue_id: string }, string[]>(
      `SELECT a.issue_id FROM run a
        WHERE a.kind = 'agent' AND a.reason = 'conflict' AND a.issue_id IN (${marks})
          AND NOT EXISTS (
            SELECT 1 FROM run b
             WHERE b.issue_id = a.issue_id AND b.kind = 'agent' AND b.reason != 'review'
               AND (b.created_at > a.created_at OR (b.created_at = a.created_at AND b.id > a.id)))`,
    )
    .all(...ids);
  return new Set(rows.map((r) => r.issue_id));
}
/** The runs that are rebases' checks, and the rest: each keeps its order. */
export function rebasesFirst<R extends { readonly issueId: string | null }>(
  db: Database,
  queued: readonly R[],
): { readonly rebases: R[]; readonly rest: R[] } {
  const rebased = rebasedPulls(
    db,
    queued.flatMap((r) => (r.issueId === null ? [] : [r.issueId])),
  );
  const rebases: R[] = [];
  const rest: R[] = [];
  for (const run of queued)
    (run.issueId !== null && rebased.has(run.issueId) ? rebases : rest).push(run);
  return { rebases, rest };
}
@@ -55,6 +55,7 @@
import { keepCache, restoreCache } from "./cache.ts";
import { MAX_FILED, clearedFiles, fileOrderDependent } from "./order_dependent.ts";
import { withDisk } from "./disk_red.ts";
import { ASKED, redMain, unheld } from "./red_main.ts";
import { rebasesFirst } from "./finishing_first.ts";
import { IMPACT_FILE, keepImpact, seedImpact } from "./impact-store.ts";
import { getIssueById } from "../core/issues.ts";
import { BUDGETS, budgetSummary } from "../kernel/budgets.ts";
@@ -434,6 +435,18 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
   * opens up. Claiming is what decides between two forges, so calling this
   * from both never runs the same check twice.
   */
  /** `pickNext` among `queued`, passing over any whose account is at its limit. */
  function startable(queued: readonly Run[]): Run | undefined {
      
        junipercrew
        19d ago
        
      
      startable was inserted between schedule()'s docblock ("Claiming is what decides between two forges...") and schedule(), so that doc now sits on the wrong function. Move startable above the docblock.
    
    let rest = queued;
    let next = pickNext(rest);
    while (next !== undefined && atLimit(next)) {
      const held = next;
      rest = rest.filter((run) => run !== held);
      next = pickNext(rest);
    }
    return next;
  }
  function schedule(): void {
    if (stopped) return;
    const limit = capacity();
@@ -453,14 +466,10 @@
export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
      );
      // The limit is asked at the point of starting, one pick at a time:
      // a started check is in flight before the next is asked about, so a
      // burst of one account's checks cannot pass together.
      let rest = queued;
      let next = pickNext(rest);
      while (next !== undefined && atLimit(next)) {
        const held = next;
        rest = rest.filter((run) => run !== held);
        next = pickNext(rest);
      }
      // burst of one account's checks cannot pass together. A rebase's
      // check goes before any other (`finishing_first.ts`, #6140).
      const { rebases, rest: others } = rebasesFirst(app.db, queued);
      const next = startable(rebases) ?? startable(others);
      if (!next) return;
      const repo = getRepoById(app.db, next.repoId);
      if (!repo) return;
@@ -27,7 +27,7 @@
import { addLearned } from "../checks/learned.ts";
import { enqueue, type GateCandidate, type Queued, type QueueOptions, queueClashes, queueWords } from "../checks/queue.ts";
import { checksFor } from "../checks/service.ts";
import { queueDepth } from "../core/capacity.ts";
import { readConflictRoom } from "./conflict_room.ts";
import { readConflictLane, readConflictRoom } from "./conflict_room.ts";
import { mergeable, resolveOid, treeOf } from "../git/merge.ts";
import { CLOSED_NOTE } from "../core/done_on_main.ts";
import { mergePull } from "../actions/pulls.ts";
@@ -1285,8 +1285,9 @@
export async function autonomyTick(
        unstuck,
        active: status.active,
        slots: status.slots,
        // Approved pulls being brought up to date, and those still waiting (#3682).
        ...finishFields(finishing),
        // Approved pulls being brought up to date, those still waiting (#3682),
        // and the lane for rebases: conflicted N, rebasing M, room R (#6140).
        ...finishFields(finishing, readConflictLane(app.db)),
        ...(requested.length === 0
          ? {}
          : {
@@ -20,15 +20,23 @@
 * pulls: it queues before the sweep, and the sweep counts its runs as in
 * flight, so between them a tick never starts more than the lane holds
 * (#4784).
 *
 * The checks counted are the rebases' own, not every check (#6140): a
 * rebase's check goes before any other in the lane (`checks/
 * finishing_first.ts`), so a new build's check does not stand in front of
 * it. Counting those let new builds keep the room at one, and 64 approved
 * pulls drained one at a time.
 */
import type { Database } from "bun:sqlite";
import { rebasesFirst } from "../checks/finishing_first.ts";
import { capacityValue } from "../core/capacity.ts";
import { listActive } from "../core/runs.ts";
export function conflictRoom(
  /** Checks the forge runs at once (`checks.concurrency`). */
  slots: number,
  /** Check runs queued or running. */
  /** Rebases' check runs queued or running: the ones ahead of the next. */
  checks: number,
  /** Bring-up-to-date runs queued or running. */
  inFlight: number,
@@ -37,20 +45,29 @@
export function conflictRoom(
  return Math.max(room, inFlight === 0 ? 1 : 0);
}
/** What the lane holds for rebases now, and the room left: the tick line's and /flow's words (#6140). */
export interface ConflictLane {
  /** Bring-up-to-date runs queued or running. */
  readonly runs: number;
  /** Rebases' checks queued or running. */
  readonly checks: number;
  readonly room: number;
}
/** The lane from what the forge holds now. */
export function readConflictLane(db: Database): ConflictLane {
  const runs = db
    .query<{ n: number }, []>(
      `SELECT COUNT(*) AS n FROM run
        WHERE kind = 'agent' AND reason = 'conflict' AND state IN ('queued','running')`,
    )
    .get()!.n;
  const checks = rebasesFirst(db, listActive(db, "check")).rebases.length;
  const slots = Number(capacityValue(db, "checks.concurrency"));
  return { runs, checks, room: conflictRoom(Number.isInteger(slots) && slots > 0 ? slots : 1, checks, runs) };
}
/** `conflictRoom` from what the forge holds now. */
export function readConflictRoom(db: Database): number {
  const row = db
    .query<{ checks: number; conflicts: number }, []>(
      `SELECT
         COALESCE(SUM(kind = 'check'), 0) AS checks,
         COALESCE(SUM(kind = 'agent' AND reason = 'conflict'), 0) AS conflicts
       FROM run WHERE state IN ('queued','running')`,
    )
    .get();
  const slots = Number(capacityValue(db, "checks.concurrency"));
  return conflictRoom(
    Number.isInteger(slots) && slots > 0 ? slots : 1,
    row?.checks ?? 0,
    row?.conflicts ?? 0,
  );
  return readConflictLane(db).room;
}
@@ -11,7 +11,11 @@
 * - With a slot free, a conflict run (the #2039 resolver brief: merge the
 *   base, resolve keeping both intents, re-run the check) is queued for
 *   its builder — the agent on it, else its author, else any builder —
 *   and it takes that slot.
 *   and it takes that slot. When that builder already has a run in
 *   flight, the run goes to an idle builder instead, round-robin, so one
 *   busy agent holding many pulls does not bring them up to date one at
 *   a time (#6140); the run's agent is who rebased, and the log names
 *   whose pull it was.
 * - Its push is a check, so it also waits for room in the check lane, as
 *   the conflict sweep's runs do (`conflict_room.ts`): a landing that
 *   conflicts ten approved pulls at once must not put ten checks in the
@@ -36,11 +40,12 @@
import { getIssueById, type Issue } from "../core/issues.ts";
import { listPulls, pullDir, type Pull } from "../core/pulls.ts";
import type { Repo } from "../core/repos.ts";
import { reviewSummary } from "../core/reviews.ts";
import { activeRunFor } from "../core/runs.ts";
import { activeRunFor, listActive } from "../core/runs.ts";
import { getUserById } from "../core/users.ts";
import { mergeable, resolveOid } from "../git/merge.ts";
import { createLog } from "../kernel/log.ts";
import { readConflictRoom } from "./conflict_room.ts";
import { noteWaitingToFinish, rebaseLaneSays } from "../flow/conflicted.ts";
import { type ConflictLane, readConflictRoom } from "./conflict_room.ts";
import { CHECK_FAILED_LIMIT, lastHumanWord, queueAgentRun } from "./dispatch.ts";
import { HANDOVER_NOTE } from "./guard.ts";
import { pickBuilder } from "./route.ts";
@@ -50,7 +55,13 @@
const log = createLog("crew");
/** What the pass did with one approved, conflicted pull. */
export type FinishMove =
  | { readonly pull: number; readonly did: "queued update"; readonly builder: string }
  | {
      readonly pull: number;
      readonly did: "queued update";
      readonly builder: string;
      /** The pull's own builder, when it was busy and another took the update. */
      readonly for?: string;
    }
  | { readonly pull: number; readonly did: "in flight" }
  | { readonly pull: number; readonly did: "handed to a person"; readonly to: string }
  | { readonly pull: number; readonly did: "not moved"; readonly why: string };
@@ -61,13 +72,26 @@
export function approvedOnHead(db: Database, pull: Pull): boolean {
  return seen.approvals > 0 && seen.changesRequested === 0;
}
/** The tick line's two numbers: pulls being brought up to date, and the rest still waiting. */
export function finishFields(moves: readonly FinishMove[]): { finishing?: number; conflict?: string } {
/**
 * The tick line's numbers: pulls being brought up to date, the rest still
 * waiting, and, given the lane, `conflicted N, rebasing M, room R` — this
 * repository's approved conflicted pulls against the forge-wide lane
 * (#6140) — whenever any is conflicted or rebasing.
 */
export function finishFields(
  moves: readonly FinishMove[],
  lane?: ConflictLane,
): { finishing?: number; conflict?: string; lane?: string } {
  const finishing = moves.filter((m) => m.did === "queued update" || m.did === "in flight").length;
  const waiting = moves.length - finishing;
  const words =
    lane === undefined
      ? null
      : rebaseLaneSays({ conflicted: moves.length, rebasing: lane.runs + lane.checks, room: lane.room });
  return {
    ...(finishing === 0 ? {} : { finishing }),
    ...(waiting === 0 ? {} : { conflict: `${waiting} waiting` }),
    ...(words === null ? {} : { lane: words }),
  };
}
@@ -98,7 +122,10 @@
async function conflictOf(app: App, repo: Repo, pull: Pull): Promise<string[] |
 * pull first.
 */
export async function finishApproved(app: App, repo: Repo, free: number): Promise<FinishMove[]> {
  if (repo.autonomy !== "autonomous" || repo.autonomyBy === null) return [];
  if (repo.autonomy !== "autonomous" || repo.autonomyBy === null) {
    noteWaitingToFinish(repo.id, []);
    return [];
  }
  const moves: FinishMove[] = [];
  const builders = listCrewRoles(app.db, repo.id)
    .filter((c) => c.role === "builder")
@@ -108,6 +135,9 @@
export async function finishApproved(app: App, repo: Repo, free: number): Promis
  for (const id of read.keys()) if (!ids.has(id)) read.delete(id);
  let slots = free;
  let room = readConflictRoom(app.db);
  const load = new Map<string, number>();
  for (const run of listActive(app.db, "agent"))
    if (run.actorId !== null) load.set(run.actorId, (load.get(run.actorId) ?? 0) + 1);
  for (const pull of open.sort((a, b) => a.number - b.number)) {
    if (!pull.authorIsAgent || pull.mergedAt !== null) continue;
    if (getHold(app.db, pull.id) !== undefined) continue;
@@ -116,14 +146,26 @@
export async function finishApproved(app: App, repo: Repo, free: number): Promis
    if (files === undefined) continue;
    const issue = getIssueById(app.db, pull.id);
    if (!issue || issue.state !== "open") continue;
    const move = finishOne(app, repo, pull, issue, files, builders, slots, room);
    const move = finishOne(app, repo, pull, issue, files, builders, load, slots, room);
    if (move.did === "queued update") {
      slots--;
      room--;
      const by = builders.find((b) => b.user.username === move.builder)!.user.id;
      load.set(by, (load.get(by) ?? 0) + 1);
    }
    moves.push(move);
    say(repo, pull, move);
  }
  const since = new Date().toISOString();
  noteWaitingToFinish(
    repo.id,
    moves.map((m) => ({
      repo: `${repo.ownerName}/${repo.name}`,
      number: m.pull,
      since,
      public: repo.visibility === "public",
    })),
  );
  return moves;
}
@@ -134,6 +176,7 @@
function finishOne(
  issue: Issue,
  files: readonly string[],
  builders: readonly Agent[],
  load: ReadonlyMap<string, number>,
  slots: number,
  room: number,
): FinishMove {
@@ -142,12 +185,11 @@
function finishOne(
  // A person's name on it is a person holding it — after a hand-over too.
  if (issue.assigneeId !== null && getAgent(app.db, issue.assigneeId) === undefined)
    return not("a person has it");
  const builder =
    builders.find((b) => b.user.id === issue.assigneeId) ??
    builders.find((b) => b.user.id === issue.authorId) ??
    pickBuilder(issue, builders, new Map());
  if (builder === undefined) return not(NO_BUILDER_ON_CREW);
  if (failedUpdates(app, issue, builder) >= CHECK_FAILED_LIMIT) return handOver(app, repo, pull, issue, files);
  const own = ownBuilder(issue, builders, load);
  if (own === undefined) return not(NO_BUILDER_ON_CREW);
  // Two failed updates are counted against the pull's own builder, whoever ran them.
  const builder = (slots > 0 && room > 0 ? lentBuilder(own, builders, load) : undefined) ?? own;
  if (failedUpdates(app, issue, own) >= CHECK_FAILED_LIMIT) return handOver(app, repo, pull, issue, files);
  if (slots <= 0) return not(WAITS_FOR_BUILDER_SLOT);
  if (room <= 0) return not("waits for room in the check lane");
  const run = queueAgentRun(app, repo, issue, builder, "conflict", undefined, {
@@ -156,7 +198,46 @@
function finishOne(
    summary: `main moved: ${files.join(", ")}`,
  });
  if (run === undefined) return not(`${builder.user.username}'s run was not queued (ceiling or breaker)`);
  return { pull: pull.number, did: "queued update", builder: builder.user.username };
  return {
    pull: pull.number,
    did: "queued update",
    builder: builder.user.username,
    ...(builder === own ? {} : { for: own.user.username }),
  };
}
/** The pull's own builder: the agent on it, else its author, else the builder with least in flight. */
export function ownBuilder(
  issue: Issue,
  builders: readonly Agent[],
  load: ReadonlyMap<string, number>,
): Agent | undefined {
  return (
    builders.find((b) => b.user.id === issue.assigneeId) ??
    builders.find((b) => b.user.id === issue.authorId) ??
    pickBuilder(issue, builders, load)
  );
}
/** Whose turn it is among the idle builders, kept across ticks so lent updates go round. */
let turn = 0;
/**
 * The builder who takes an update when `own` already has a run in flight:
 * the next idle builder, round-robin by name (#6140). Undefined when `own`
 * is idle or every builder is busy: then the update is `own`'s.
 */
export function lentBuilder(
  own: Agent,
  builders: readonly Agent[],
  load: ReadonlyMap<string, number>,
): Agent | undefined {
  if ((load.get(own.user.id) ?? 0) === 0) return undefined;
  const idle = builders
    .filter((b) => (load.get(b.user.id) ?? 0) === 0)
    .sort((a, b) => a.user.username.localeCompare(b.user.username));
  if (idle.length === 0) return undefined;
  return idle[turn++ % idle.length];
}
/**
@@ -203,7 +284,11 @@
function say(repo: Repo, pull: Pull, move: FinishMove): void {
  const key = `${repo.id}:${pull.number}`;
  if (move.did === "queued update") {
    said.delete(key);
    log.info("approved pull conflicts: brought up to date first", { ...where, builder: move.builder });
    log.info("approved pull conflicts: brought up to date first", {
      ...where,
      builder: move.builder,
      ...(move.for === undefined ? {} : { for: move.for }),
    });
    return;
  }
  if (move.did !== "not moved") return;
@@ -2,10 +2,13 @@
 * Pulls that main moved out from under (#2039): open pulls with a
 * bring-up-to-date run queued, running or failed, by how long since the
 * crew was first asked to bring each one up to date. Read from the runs
 * alone, so /flow, the terminal and MCP say the same without asking git.
 * alone, so /flow, the terminal and MCP say the same without asking git —
 * and the approved pulls the finishing pass last found conflicted and
 * still waiting for room, which have no run yet (#6140).
 */
import type { Database } from "bun:sqlite";
import { readConflictLane } from "../crew/conflict_room.ts";
import { span } from "./pipeline.ts";
export interface ConflictedPull {
@@ -15,8 +18,33 @@
export interface ConflictedPull {
  readonly since: string;
}
/** An approved pull the finishing pass found conflicted, with no run on it yet. */
export interface WaitingToFinish extends ConflictedPull {
  readonly public: boolean;
}
/** Repository id → its approved, conflicted pulls on the finishing pass's last tick. */
const waiting = new Map<string, readonly WaitingToFinish[]>();
      
        junipercrew
        19d ago
        
      
      In-memory per-process map: the 'approved, waiting for room' count is only visible on the forge that ran the finishing tick, while the rest of this file is 'read from the runs alone'. Note it in FLOW.md or derive it from the db.
    
/** What the finishing pass found this tick in one repository; a pull seen before keeps its first `since`. */
export function noteWaitingToFinish(repoId: string, pulls: readonly WaitingToFinish[]): void {
  const before = new Map((waiting.get(repoId) ?? []).map((p) => [p.number, p.since]));
  if (pulls.length === 0) waiting.delete(repoId);
  else waiting.set(repoId, pulls.map((p) => ({ ...p, since: before.get(p.number) ?? p.since })));
}
/** Oldest first; `publicOnly` leaves out private repositories, for a reader who must not learn their names. */
export function conflictedPulls(db: Database, publicOnly = false): ConflictedPull[] {
  const ran = conflictRuns(db, publicOnly);
  const seen = new Set(ran.map((p) => `${p.repo}#${p.number}`));
  const noted = [...waiting.values()]
    .flat()
    .filter((p) => (!publicOnly || p.public) && !seen.has(`${p.repo}#${p.number}`))
    .map(({ repo, number, since }) => ({ repo, number, since }));
  return noted.length === 0 ? ran : [...ran, ...noted].sort((a, b) => a.since.localeCompare(b.since));
}
function conflictRuns(db: Database, publicOnly: boolean): ConflictedPull[] {
  return db
    .query<ConflictedPull, [number]>(
      `SELECT o.username || '/' || r.name AS repo, i.number AS number,
@@ -46,3 +74,23 @@
export function conflictedSays(
  const more = pulls.length > 5 ? `, and ${pulls.length - 5} more` : "";
  return `${pulls.length} conflicted ${pulls.length === 1 ? "pull" : "pulls"}, oldest first: ${list}${more}.`;
}
/** The lane for rebases: conflicted pulls, how many are being brought up to date, and the room for more (#6140). */
export interface RebaseLane {
  readonly conflicted: number;
  /** Bring-up-to-date runs and their checks, queued or running, forge-wide. */
  readonly rebasing: number;
  /** Bring-up-to-date runs the lane has room to start now (`crew/conflict_room.ts`). */
  readonly room: number;
}
export function rebaseLane(db: Database, pulls: readonly ConflictedPull[]): RebaseLane {
  const lane = readConflictLane(db);
  return { conflicted: pulls.length, rebasing: lane.runs + lane.checks, room: lane.room };
}
/** `conflicted N, rebasing M, room R`, so a stall reads in a minute; null when nothing is conflicted or rebasing. */
export function rebaseLaneSays(lane: RebaseLane): string | null {
  if (lane.conflicted === 0 && lane.rebasing === 0) return null;
  return `conflicted ${lane.conflicted}, rebasing ${lane.rebasing}, room ${lane.room}`;
}
@@ -7,7 +7,7 @@
 */
import type { App } from "../app.ts";
import { SOURCES } from "../boards/sources.ts";
import { conflictedPulls, conflictedSays } from "../flow/conflicted.ts";
import { conflictedPulls, conflictedSays, rebaseLane, rebaseLaneSays } from "../flow/conflicted.ts";
import { readPipeline, verdict } from "../flow/pipeline.ts";
import { type QueueReading, readQueues } from "../flow/queue.ts";
import { reviewersBusy, reviewingSays } from "../flow/reviewing.ts";
@@ -27,6 +27,7 @@
export function flowFor(app: App, reader: Reader, now = new Date()) {
  const flow = readThroughput(app.db, stages.map((s) => s.key), now);
  const dup = systemFor(app, reader).duplicates;
  const conflicted = conflictedPulls(app.db, reader.kind === "agent");
  const lane = rebaseLane(app.db, conflicted);
  const requested = requestedWaits(app.db, reader.kind === "agent");
  const busy = reviewersBusy(app.db, reader.kind === "agent");
  return {
@@ -34,6 +35,8 @@
export function flowFor(app: App, reader: Reader, now = new Date()) {
    // Review runs in flight per reviewer, cut to public repositories for an agent (#4733).
    reviewing: { says: reviewingSays(busy), reviewers: busy },
    conflicted: { says: conflictedSays(conflicted, now.getTime()), pulls: conflicted },
    // conflicted N, rebasing M, room R: a stall in the lane reads in a minute (#6140).
    rebases: { ...lane, says: rebaseLaneSays(lane) },
    // Pulls with changes requested, by whom they wait on (#2172).
    requested: {
      builder: { says: requestedSays(requested.builder, "builder", now.getTime()), pulls: requested.builder },
@@ -16,7 +16,7 @@
import type { Database } from "bun:sqlite";
import { HOLDS } from "../../crew/held.ts";
import type { HeldRow } from "../../flow/held.ts";
import { conflictedPulls, conflictedSays } from "../../flow/conflicted.ts";
import { conflictedPulls, conflictedSays, rebaseLane, rebaseLaneSays } from "../../flow/conflicted.ts";
import { readPipeline, STAGE_LABEL, verdict } from "../../flow/pipeline.ts";
import { readQueues, redBatchWords } from "../../flow/queue.ts";
import { reviewersBusy, reviewingSays } from "../../flow/reviewing.ts";
@@ -47,7 +47,9 @@
export function flowLines(
  const queues = agent?.queues ?? readQueues(db, now, { batches: true });
  // The row is full at 58 columns: the red-batch rate (#4235) is a line under it.
  const red = queues.filter((q) => (q.batches?.batches ?? 0) > 0);
  const conflicted = conflictedSays(conflictedPulls(db, agent !== undefined), now.getTime());
  const conflictedList = conflictedPulls(db, agent !== undefined);
  const conflicted = conflictedSays(conflictedList, now.getTime());
  const lane = rebaseLaneSays(rebaseLane(db, conflictedList));
  const requested = requestedWaits(db, agent !== undefined);
  const waits = [
    requestedSays(requested.builder, "builder", now.getTime()),
@@ -75,6 +77,7 @@
export function flowLines(
    ...(red.length === 0
      ? []
      : wrap(`Red batches, last hour: ${red.map((q) => `${q.repo} ${redBatchWords(q)}`).join(", ")}.`, inner)),
    ...(lane === null ? [] : ["", ...wrap(`Rebases: ${lane}.`, inner)]),
    ...(conflicted === null ? [] : ["", ...wrap(conflicted, inner)]),
    ...waits,
    ...(reviewing === null ? [] : ["", ...wrap(reviewing, inner)]),
@@ -9,7 +9,7 @@
import type { Reader } from "../../actions/telemetry.ts";
import { queuesFor, systemFor } from "../../system/telemetry.ts";
import { conflictedPulls, conflictedSays } from "../../flow/conflicted.ts";
import { conflictedPulls, conflictedSays, rebaseLane, rebaseLaneSays } from "../../flow/conflicted.ts";
import { reviewersBusy, reviewingSays } from "../../flow/reviewing.ts";
import { requestedSays, requestedWaits } from "../../flow/requested.ts";
import {
@@ -90,7 +90,9 @@
export function flowPanel(app: App, reader: Reader, url: URL, base: string): Htm
  const out = new Map(flow.stages.map((t) => [t.key, t]));
  const landed = out.get("approved");
  const dups = systemFor(app, reader).duplicates;
  const conflicted = conflictedSays(conflictedPulls(app.db, reader.kind === "agent"), now);
  const conflictedList = conflictedPulls(app.db, reader.kind === "agent");
  const conflicted = conflictedSays(conflictedList, now);
  const lane = rebaseLaneSays(rebaseLane(app.db, conflictedList));
  const requested = requestedWaits(app.db, reader.kind === "agent");
  const onBuilder = requestedSays(requested.builder, "builder", now);
  const onReviewer = requestedSays(requested.reviewer, "reviewer", now);
@@ -101,6 +103,7 @@
export function flowPanel(app: App, reader: Reader, url: URL, base: string): Htm
      html`<p id="bottleneck" data-stage="${worst?.key ?? ""}">
          <strong>${verdict(stages, now)}</strong>
        </p>
    ${lane === null ? "" : html`<p id="rebase-lane">Rebases: ${lane}.</p>`}
    ${conflicted === null ? "" : html`<p id="conflicted">${conflicted}</p>`}
    ${onBuilder === null ? "" : html`<p id="requested-builder">${onBuilder}</p>`}
    ${onReviewer === null ? "" : html`<p id="requested-reviewer">${onReviewer}</p>`}
@@ -372,8 +372,15 @@
test("one landing that conflicts many pulls starts only as many runs as the chec
  });
  if (!shared.ok) throw new Error(JSON.stringify(shared.error));
  setCapacity(app.db, "checks.concurrency", "3", alice.id);
  // A check already waits: two slots are left for the pushes to come.
  const waiting = createRun(app.db, { repoId: repo.id, kind: "check", env: "main" });
  // A rebase's check already waits: two slots are left for the pushes to
  // come. Another kind of check does not stand in front of them (#6140).
  const mainCheck = createRun(app.db, { repoId: repo.id, kind: "check", env: "main" });
  const rebased = openIssue(app, alice, repo, { title: "Rebased elsewhere" });
  if (!rebased.ok) throw new Error(String(rebased.error));
  const pushedBy = createRun(app.db, { repoId: repo.id, kind: "agent", issueId: rebased.value.id, reason: "conflict" });
  claimRun(app.db, pushedBy.id);
  finishRun(app.db, pushedBy.id, { state: "ok", summary: "pushed" });
  const waiting = createRun(app.db, { repoId: repo.id, kind: "check", issueId: rebased.value.id });
  try {
    expect(await sweepConflicts(app, fresh())).toBe(2);
    expect(pulls.map((p) => p.runs().length)).toEqual([1, 1, 0]);
@@ -389,6 +396,8 @@
test("one landing that conflicts many pulls starts only as many runs as the chec
        claimRun(app.db, run.id);
        finishRun(app.db, run.id, { state: "ok", summary: "" });
      }
    claimRun(app.db, mainCheck.id);
    finishRun(app.db, mainCheck.id, { state: "ok", summary: "" });
    setCapacity(app.db, "checks.concurrency", null, alice.id);
  }
});
@@ -0,0 +1,260 @@
/**
 * Approved, conflicted pulls do not drain one at a time (#6140): a rebase's
 * check goes before a new build's in the lane, the room for more rebases
 * counts only the checks that stand in front of them, a busy builder's
 * pulls go to idle builders round-robin, and the lane reads as
 * `conflicted N, rebasing M, room R` on the tick line and /flow.
 */
import { afterAll, beforeAll, describe, expect, test } from "bun:test";
import { writeFileSync } from "node:fs";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createApp, type App } from "../src/app.ts";
import { openIssue } from "../src/actions/issues.ts";
import { commitToBranch, createRepo } from "../src/actions/repos.ts";
import { rebasedPulls, rebasesFirst } from "../src/checks/finishing_first.ts";
import { startChecks } from "../src/checks/service.ts";
import type { Agent } from "../src/core/agents.ts";
import { setCapacity } from "../src/core/capacity.ts";
import type { Issue } from "../src/core/issues.ts";
import type { Repo } from "../src/core/repos.ts";
import { cancelRun, createRun, getRun, listActive } from "../src/core/runs.ts";
import { createUser, type User } from "../src/core/users.ts";
import { readConflictLane } from "../src/crew/conflict_room.ts";
import { finishFields, lentBuilder, ownBuilder } from "../src/crew/finishing.ts";
import { serve } from "../src/http/server.ts";
import {
  conflictedPulls,
  noteWaitingToFinish,
  rebaseLane,
  rebaseLaneSays,
} from "../src/flow/conflicted.ts";
let dataDir: string;
let app: App;
let alice: User;
let repo: Repo;
async function issue(title: string): Promise<Issue> {
  const opened = openIssue(app, alice, repo, { title });
  if (!opened.ok) throw new Error(String(opened.error));
  return opened.value;
}
function agentRun(on: Issue, reason: string) {
  return createRun(app.db, { repoId: repo.id, kind: "agent", issueId: on.id, reason });
}
function checkRun(on: Issue) {
  return createRun(app.db, { repoId: repo.id, kind: "check", issueId: on.id, reason: "push" });
}
/** Settle every queued or running run, so each test starts from an empty lane. */
function settleAll() {
  for (const kind of ["agent", "check"] as const)
    for (const run of listActive(app.db, kind)) cancelRun(app.db, run.id, "test");
}
beforeAll(async () => {
  dataDir = await mkdtemp(join(tmpdir(), "forge-rebase-lane-"));
  app = createApp({ dataDir, port: 0 });
  const a = await createUser(app.db, {
    origin: "seed",
    trusted: true,
    username: "alice",
    email: "[email protected]",
    password: "a-good-passphrase",
    adminIfFirst: true,
  });
  if (!a.ok) throw new Error(a.error);
  alice = a.value;
  const made = await createRepo(app, alice, { name: "engine", visibility: "private" });
  if (!made.ok) throw new Error(String(made.error));
  repo = made.value;
});
afterAll(async () => {
  app.db.close();
  await rm(dataDir, { recursive: true, force: true });
});
describe("a rebase's check goes first in the lane", () => {
  test("a pull whose last builder's run was a conflict run is a rebase; a review after it does not change that", async () => {
    const rebased = await issue("rebased");
    const reviewed = await issue("rebased, then reviewed");
    const built = await issue("a new build");
    const rebuilt = await issue("rebased, then sent back to its builder");
    agentRun(rebased, "conflict");
    agentRun(reviewed, "conflict");
    agentRun(reviewed, "review");
    agentRun(built, "assigned");
    agentRun(rebuilt, "conflict");
    agentRun(rebuilt, "feedback");
    expect(rebasedPulls(app.db, [rebased.id, reviewed.id, built.id, rebuilt.id])).toEqual(
      new Set([rebased.id, reviewed.id]),
    );
    expect(rebasedPulls(app.db, [])).toEqual(new Set());
    settleAll();
  });
  test("rebasesFirst splits queued checks, keeping each side's order", async () => {
    const rebased = await issue("rebased pull");
    const built = await issue("built pull");
    agentRun(rebased, "conflict");
    agentRun(built, "assigned");
    const newBuild = checkRun(built);
    const rebase = checkRun(rebased);
    const main = createRun(app.db, { repoId: repo.id, kind: "check", env: "main" });
    const split = rebasesFirst(app.db, [newBuild, rebase, main]);
    expect(split.rebases.map((r) => r.id)).toEqual([rebase.id]);
    expect(split.rest.map((r) => r.id)).toEqual([newBuild.id, main.id]);
    settleAll();
  });
});
describe("the room counts only the checks in front of a rebase", () => {
  test("new builds' checks take no room; rebases' checks and conflict runs do", async () => {
    setCapacity(app.db, "checks.concurrency", "4", alice.id);
    for (let i = 0; i < 6; i++) {
      const built = await issue(`build ${i}`);
      agentRun(built, "assigned");
      checkRun(built);
    }
    // Six new builds' checks in a lane of four: before #6140 the room was one.
    expect(readConflictLane(app.db)).toEqual({ runs: 0, checks: 0, room: 4 });
    const rebased = await issue("rebased, check queued");
    // Its conflict run ended with a push; the push's check waits.
    cancelRun(app.db, agentRun(rebased, "conflict").id, "pushed");
    checkRun(rebased);
    const rebasing = await issue("being rebased");
    agentRun(rebasing, "conflict");
    expect(readConflictLane(app.db)).toEqual({ runs: 1, checks: 1, room: 2 });
    settleAll();
  });
});
describe("a busy builder's rebases go to idle builders", () => {
  const builder = (id: string): Agent => ({ user: { id, username: id } }) as unknown as Agent;
  const [ada, bo, cy, di] = ["ada", "bo", "cy", "di"].map(builder) as [Agent, Agent, Agent, Agent];
  const crew = [di, bo, ada, cy];
  test("the pull's own builder is the agent on it, else its author", () => {
    const on = { assigneeId: "cy", authorId: "bo" } as Issue;
    expect(ownBuilder(on, crew, new Map())?.user.id).toBe("cy");
    expect(ownBuilder({ ...on, assigneeId: null } as Issue, crew, new Map())?.user.id).toBe("bo");
  });
  test("an idle own builder keeps its pull; a busy one lends it round-robin among the idle", () => {
    expect(lentBuilder(ada, crew, new Map())).toBeUndefined();
    const load = new Map([["ada", 1], ["di", 2]]);
    const lent = [1, 2, 3, 4].map(() => lentBuilder(ada, crew, load)?.user.id);
    // Only bo and cy are idle; the turns alternate between them.
    expect(new Set(lent)).toEqual(new Set(["bo", "cy"]));
    expect(lent[0]).not.toBe(lent[1]);
    expect(lent[1]).not.toBe(lent[2]);
  });
  test("with every builder busy, the update stays with its own builder", () => {
    const load = new Map(crew.map((b) => [b.user.id, 1]));
    expect(lentBuilder(ada, crew, load)).toBeUndefined();
  });
});
describe("the lane reads as conflicted N, rebasing M, room R", () => {
  test("words, and nothing said when nothing is conflicted or rebasing", () => {
    expect(rebaseLaneSays({ conflicted: 64, rebasing: 1, room: 1 })).toBe("conflicted 64, rebasing 1, room 1");
    expect(rebaseLaneSays({ conflicted: 0, rebasing: 0, room: 7 })).toBeNull();
  });
  test("the tick line carries the lane beside the finishing counts", () => {
    const fields = finishFields(
      [
        { pull: 1, did: "queued update", builder: "bo", for: "ada" },
        { pull: 2, did: "not moved", why: "waits for room in the check lane" },
      ],
      { runs: 1, checks: 2, room: 0 },
    );
    expect(fields).toEqual({ finishing: 1, conflict: "1 waiting", lane: "conflicted 2, rebasing 3, room 0" });
    expect(finishFields([])).toEqual({});
  });
  test("approved pulls waiting for room count as conflicted, oldest first, and a private one only for the owner", () => {
    // Other suites' ticks note their own repositories in the same process: read only this one.
    const mine = (publicOnly = false) =>
      conflictedPulls(app.db, publicOnly).filter((p) => p.repo === "zed/noted");
    noteWaitingToFinish("zed-noted", [
      { repo: "zed/noted", number: 9, since: "2026-10-26T10:00:00.000Z", public: false },
      { repo: "zed/noted", number: 4, since: "2026-10-26T09:00:00.000Z", public: true },
    ]);
    // Seen again later: each keeps when it was first seen.
    noteWaitingToFinish("zed-noted", [
      { repo: "zed/noted", number: 9, since: "2026-10-26T11:00:00.000Z", public: false },
      { repo: "zed/noted", number: 4, since: "2026-10-26T11:00:00.000Z", public: true },
    ]);
    expect(mine().map((p) => [p.number, p.since])).toEqual([
      [4, "2026-10-26T09:00:00.000Z"],
      [9, "2026-10-26T10:00:00.000Z"],
    ]);
    expect(mine(true).map((p) => p.number)).toEqual([4]);
    expect(rebaseLane(app.db, mine()).conflicted).toBe(2);
    noteWaitingToFinish("zed-noted", []);
    expect(mine()).toEqual([]);
  });
});
describe("the checks service starts a rebase's check before a newer build's", () => {
  test("with one slot, the older rebase's check runs first though the newest would otherwise win", async () => {
    const dir = await mkdtemp(join(tmpdir(), "forge-rebase-first-"));
    const own = createApp({ dataDir: dir, port: 0 });
    const servers = await serve(own);
    const go = join(dir, "go");
    let checks: ReturnType<typeof startChecks> | undefined;
    try {
      const made = await createUser(own.db, {
        origin: "seed",
        trusted: true,
        username: "alice",
        email: "[email protected]",
        password: "a-good-passphrase",
        adminIfFirst: true,
      });
      if (!made.ok) throw new Error(made.error);
      const created = await createRepo(own, made.value, { name: "lane", visibility: "public" });
      if (!created.ok) throw new Error(String(created.error));
      const lane = created.value;
      const pushed = await commitToBranch(own, made.value, lane, {
        message: "Check me",
        files: [{ path: "forge.json", content: JSON.stringify({ check: { run: `while [ ! -f ${go} ]; do sleep 0.05; done` } }) }],
      });
      if (!pushed.ok) throw new Error(JSON.stringify(pushed.error));
      const ask = (title: string) => {
        const opened = openIssue(own, made.value, lane, { title });
        if (!opened.ok) throw new Error(String(opened.error));
        return opened.value;
      };
      const rebased = ask("rebased");
      const built = ask("built");
      createRun(own.db, { repoId: lane.id, kind: "agent", issueId: rebased.id, reason: "conflict" });
      createRun(own.db, { repoId: lane.id, kind: "agent", issueId: built.id, reason: "assigned" });
      for (const run of listActive(own.db, "agent")) cancelRun(own.db, run.id, "pushed");
      const check = (on: Issue) =>
        createRun(own.db, { repoId: lane.id, kind: "check", env: "production", issueId: on.id, commitOid: pushed.value.commitOid, reason: "push" });
      const rebase = check(rebased);
      const build = check(built);
      setCapacity(own.db, "checks.concurrency", "1", made.value.id);
      checks = startChecks(own, { forgeUrl: servers.url, concurrency: 1 });
      const state = (id: string) => getRun(own.db, id)?.state;
      const deadline = Date.now() + 60_000;
      while (state(rebase.id) === "queued" && state(build.id) === "queued" && Date.now() < deadline) await Bun.sleep(50);
      expect(state(rebase.id)).toBe("running");
      expect(state(build.id)).toBe("queued");
    } finally {
      writeFileSync(go, "");
      await checks?.stop();
      servers.stop();
      own.close();
      await rm(dir, { recursive: true, force: true });
    }
  }, 120_000);
});
		@@ -27,6 +27,25 @@ bottleneck; the knobs stay on Settings → Capacity. The same sentence
27	27	(`verdict`) is the terminal's admin-only Flow command and the MCP `flow` tool,
28	28	which also returns the stages; anyone but a forge admin is refused.
29	29
	30	## Rebases
	31
	32	Approved work lands before new work is built, in the check lane too (#6140).
	33	A check on a pull whose last builder's run brought it up to date (a
	34	`conflict` run; a review after it does not count) goes before any other
	35	queued check (`checks/finishing_first.ts`). So the room for more rebases
	36	(`crew/conflict_room.ts`) is the check slots minus the rebases' checks and the
	37	bring-up-to-date runs in flight; a new build's check does not stand in front
	38	of a rebase and takes no room. When the pull's own builder already has a run
	39	in flight, the finishing pass (`crew/finishing.ts`) lends the update to an
	40	idle builder, round-robin by name; the run's agent is who rebased, and the log
	41	line names whose pull it was.
	42
	43	The tick line's `lane` field, the `Rebases:` line on `/flow` and in the
	44	terminal's Flow, and `rebases` in the MCP `flow` tool all read
	45	`conflicted N, rebasing M, room R`: the conflicted pulls (with the approved
	46	ones still waiting for room, which have no run yet), the bring-up-to-date runs
	47	and their checks in flight, and the runs the lane has room to start now.
	48
30	49	## Throughput
31	50
32	51	`src/flow/throughput.ts` counts what each pipeline stage let out over the last
		@@ -0,0 +1,52 @@
	1	/**
	2	* Land before you build, in the check lane (#6140).
	3	*
	4	* A pull brought up to date (a `conflict` run: main moved, the crew merged
	5	* it in and pushed) is nearly done work; its check is the last thing
	6	* between it and landing. When a landing conflicts dozens of approved
	7	* pulls at once, new builds' checks kept the lane full, and the rebases
	8	* drained one at a time behind them. So a rebase's check goes before any
	9	* other in the lane, and the conflict room (`crew/conflict_room.ts`)
	10	* counts only these checks as standing in front of the next rebase.
	11	*
	12	* A check is a rebase's when the last builder's run on its pull — any
	13	* agent run but a review, which comes after the push — was a `conflict`
	14	* run. Read from the runs alone, so every forge sharing the database
	15	* agrees on the order.
	16	*/
	17
	18	import type { Database } from "bun:sqlite";
	19
	20	/** The pulls (issue ids), among `issueIds`, whose last builder's run was a bring-up-to-date run. */
	21	export function rebasedPulls(db: Database, issueIds: readonly string[]): Set<string> {
	22	const ids = [...new Set(issueIds)];
	23	if (ids.length === 0) return new Set();
	24	const marks = ids.map((_, i) => `?${i + 1}`).join(",");
	25	const rows = db
	26	.query<{ issue_id: string }, string[]>(
	27	`SELECT a.issue_id FROM run a
	28	WHERE a.kind = 'agent' AND a.reason = 'conflict' AND a.issue_id IN (${marks})
	29	AND NOT EXISTS (
	30	SELECT 1 FROM run b
	31	WHERE b.issue_id = a.issue_id AND b.kind = 'agent' AND b.reason != 'review'
	32	AND (b.created_at > a.created_at OR (b.created_at = a.created_at AND b.id > a.id)))`,
	33	)
	34	.all(...ids);
	35	return new Set(rows.map((r) => r.issue_id));
	36	}
	37
	38	/** The runs that are rebases' checks, and the rest: each keeps its order. */
	39	export function rebasesFirst<R extends { readonly issueId: string \| null }>(
	40	db: Database,
	41	queued: readonly R[],
	42	): { readonly rebases: R[]; readonly rest: R[] } {
	43	const rebased = rebasedPulls(
	44	db,
	45	queued.flatMap((r) => (r.issueId === null ? [] : [r.issueId])),
	46	);
	47	const rebases: R[] = [];
	48	const rest: R[] = [];
	49	for (const run of queued)
	50	(run.issueId !== null && rebased.has(run.issueId) ? rebases : rest).push(run);
	51	return { rebases, rest };
	52	}
		@@ -55,6 +55,7 @@ import { keepCache, restoreCache } from "./cache.ts";
55	55	import { MAX_FILED, clearedFiles, fileOrderDependent } from "./order_dependent.ts";
56	56	import { withDisk } from "./disk_red.ts";
57	57	import { ASKED, redMain, unheld } from "./red_main.ts";
	58	import { rebasesFirst } from "./finishing_first.ts";
58	59	import { IMPACT_FILE, keepImpact, seedImpact } from "./impact-store.ts";
59	60	import { getIssueById } from "../core/issues.ts";
60	61	import { BUDGETS, budgetSummary } from "../kernel/budgets.ts";
		@@ -434,6 +435,18 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
434	435	* opens up. Claiming is what decides between two forges, so calling this
435	436	* from both never runs the same check twice.
436	437	*/
	438	/** `pickNext` among `queued`, passing over any whose account is at its limit. */
	439	function startable(queued: readonly Run[]): Run \| undefined {
		junipercrew 19d ago `startable` was inserted between schedule()'s docblock ("Claiming is what decides between two forges...") and schedule(), so that doc now sits on the wrong function. Move `startable` above the docblock.
	440	let rest = queued;
	441	let next = pickNext(rest);
	442	while (next !== undefined && atLimit(next)) {
	443	const held = next;
	444	rest = rest.filter((run) => run !== held);
	445	next = pickNext(rest);
	446	}
	447	return next;
	448	}
	449
437	450	function schedule(): void {
438	451	if (stopped) return;
439	452	const limit = capacity();
		@@ -453,14 +466,10 @@ export function startChecks(app: App, options: ChecksOptions): ChecksHandle {
453	466	);
454	467	// The limit is asked at the point of starting, one pick at a time:
455	468	// a started check is in flight before the next is asked about, so a
456		// burst of one account's checks cannot pass together.
457		let rest = queued;
458		let next = pickNext(rest);
459		while (next !== undefined && atLimit(next)) {
460		const held = next;
461		rest = rest.filter((run) => run !== held);
462		next = pickNext(rest);
463		}
	469	// burst of one account's checks cannot pass together. A rebase's
	470	// check goes before any other (`finishing_first.ts`, #6140).
	471	const { rebases, rest: others } = rebasesFirst(app.db, queued);
	472	const next = startable(rebases) ?? startable(others);
464	473	if (!next) return;
465	474	const repo = getRepoById(app.db, next.repoId);
466	475	if (!repo) return;
		@@ -27,7 +27,7 @@ import { addLearned } from "../checks/learned.ts";
27	27	import { enqueue, type GateCandidate, type Queued, type QueueOptions, queueClashes, queueWords } from "../checks/queue.ts";
28	28	import { checksFor } from "../checks/service.ts";
29	29	import { queueDepth } from "../core/capacity.ts";
30		import { readConflictRoom } from "./conflict_room.ts";
	30	import { readConflictLane, readConflictRoom } from "./conflict_room.ts";
31	31	import { mergeable, resolveOid, treeOf } from "../git/merge.ts";
32	32	import { CLOSED_NOTE } from "../core/done_on_main.ts";
33	33	import { mergePull } from "../actions/pulls.ts";
		@@ -1285,8 +1285,9 @@ export async function autonomyTick(
1285	1285	unstuck,
1286	1286	active: status.active,
1287	1287	slots: status.slots,
1288		// Approved pulls being brought up to date, and those still waiting (#3682).
1289		...finishFields(finishing),
	1288	// Approved pulls being brought up to date, those still waiting (#3682),
	1289	// and the lane for rebases: conflicted N, rebasing M, room R (#6140).
	1290	...finishFields(finishing, readConflictLane(app.db)),
1290	1291	...(requested.length === 0
1291	1292	? {}
1292	1293	: {
		@@ -20,15 +20,23 @@
20	20	* pulls: it queues before the sweep, and the sweep counts its runs as in
21	21	* flight, so between them a tick never starts more than the lane holds
22	22	* (#4784).
	23	*
	24	* The checks counted are the rebases' own, not every check (#6140): a
	25	* rebase's check goes before any other in the lane (`checks/
	26	* finishing_first.ts`), so a new build's check does not stand in front of
	27	* it. Counting those let new builds keep the room at one, and 64 approved
	28	* pulls drained one at a time.
23	29	*/
24	30
25	31	import type { Database } from "bun:sqlite";
	32	import { rebasesFirst } from "../checks/finishing_first.ts";
26	33	import { capacityValue } from "../core/capacity.ts";
	34	import { listActive } from "../core/runs.ts";
27	35
28	36	export function conflictRoom(
29	37	/** Checks the forge runs at once (`checks.concurrency`). */
30	38	slots: number,
31		/** Check runs queued or running. */
	39	/** Rebases' check runs queued or running: the ones ahead of the next. */
32	40	checks: number,
33	41	/** Bring-up-to-date runs queued or running. */
34	42	inFlight: number,
		@@ -37,20 +45,29 @@ export function conflictRoom(
37	45	return Math.max(room, inFlight === 0 ? 1 : 0);
38	46	}
39	47
	48	/** What the lane holds for rebases now, and the room left: the tick line's and /flow's words (#6140). */
	49	export interface ConflictLane {
	50	/** Bring-up-to-date runs queued or running. */
	51	readonly runs: number;
	52	/** Rebases' checks queued or running. */
	53	readonly checks: number;
	54	readonly room: number;
	55	}
	56
	57	/** The lane from what the forge holds now. */
	58	export function readConflictLane(db: Database): ConflictLane {
	59	const runs = db
	60	.query<{ n: number }, []>(
	61	`SELECT COUNT(*) AS n FROM run
	62	WHERE kind = 'agent' AND reason = 'conflict' AND state IN ('queued','running')`,
	63	)
	64	.get()!.n;
	65	const checks = rebasesFirst(db, listActive(db, "check")).rebases.length;
	66	const slots = Number(capacityValue(db, "checks.concurrency"));
	67	return { runs, checks, room: conflictRoom(Number.isInteger(slots) && slots > 0 ? slots : 1, checks, runs) };
	68	}
	69
40	70	/** `conflictRoom` from what the forge holds now. */
41	71	export function readConflictRoom(db: Database): number {
42		const row = db
43		.query<{ checks: number; conflicts: number }, []>(
44		`SELECT
45		COALESCE(SUM(kind = 'check'), 0) AS checks,
46		COALESCE(SUM(kind = 'agent' AND reason = 'conflict'), 0) AS conflicts
47		FROM run WHERE state IN ('queued','running')`,
48		)
49		.get();
50		const slots = Number(capacityValue(db, "checks.concurrency"));
51		return conflictRoom(
52		Number.isInteger(slots) && slots > 0 ? slots : 1,
53		row?.checks ?? 0,
54		row?.conflicts ?? 0,
55		);
	72	return readConflictLane(db).room;
56	73	}
		@@ -11,7 +11,11 @@
11	11	* - With a slot free, a conflict run (the #2039 resolver brief: merge the
12	12	* base, resolve keeping both intents, re-run the check) is queued for
13	13	* its builder — the agent on it, else its author, else any builder —
14		* and it takes that slot.
	14	* and it takes that slot. When that builder already has a run in
	15	* flight, the run goes to an idle builder instead, round-robin, so one
	16	* busy agent holding many pulls does not bring them up to date one at
	17	* a time (#6140); the run's agent is who rebased, and the log names
	18	* whose pull it was.
15	19	* - Its push is a check, so it also waits for room in the check lane, as
16	20	* the conflict sweep's runs do (`conflict_room.ts`): a landing that
17	21	* conflicts ten approved pulls at once must not put ten checks in the
		@@ -36,11 +40,12 @@ import { getIssueById, type Issue } from "../core/issues.ts";
36	40	import { listPulls, pullDir, type Pull } from "../core/pulls.ts";
37	41	import type { Repo } from "../core/repos.ts";
38	42	import { reviewSummary } from "../core/reviews.ts";
39		import { activeRunFor } from "../core/runs.ts";
	43	import { activeRunFor, listActive } from "../core/runs.ts";
40	44	import { getUserById } from "../core/users.ts";
41	45	import { mergeable, resolveOid } from "../git/merge.ts";
42	46	import { createLog } from "../kernel/log.ts";
43		import { readConflictRoom } from "./conflict_room.ts";
	47	import { noteWaitingToFinish, rebaseLaneSays } from "../flow/conflicted.ts";
	48	import { type ConflictLane, readConflictRoom } from "./conflict_room.ts";
44	49	import { CHECK_FAILED_LIMIT, lastHumanWord, queueAgentRun } from "./dispatch.ts";
45	50	import { HANDOVER_NOTE } from "./guard.ts";
46	51	import { pickBuilder } from "./route.ts";
		@@ -50,7 +55,13 @@ const log = createLog("crew");
50	55
51	56	/** What the pass did with one approved, conflicted pull. */
52	57	export type FinishMove =
53		\| { readonly pull: number; readonly did: "queued update"; readonly builder: string }
	58	\| {
	59	readonly pull: number;
	60	readonly did: "queued update";
	61	readonly builder: string;
	62	/** The pull's own builder, when it was busy and another took the update. */
	63	readonly for?: string;
	64	}
54	65	\| { readonly pull: number; readonly did: "in flight" }
55	66	\| { readonly pull: number; readonly did: "handed to a person"; readonly to: string }
56	67	\| { readonly pull: number; readonly did: "not moved"; readonly why: string };
		@@ -61,13 +72,26 @@ export function approvedOnHead(db: Database, pull: Pull): boolean {
61	72	return seen.approvals > 0 && seen.changesRequested === 0;
62	73	}
63	74
64		/** The tick line's two numbers: pulls being brought up to date, and the rest still waiting. */
65		export function finishFields(moves: readonly FinishMove[]): { finishing?: number; conflict?: string } {
	75	/**
	76	* The tick line's numbers: pulls being brought up to date, the rest still
	77	* waiting, and, given the lane, `conflicted N, rebasing M, room R` — this
	78	* repository's approved conflicted pulls against the forge-wide lane
	79	* (#6140) — whenever any is conflicted or rebasing.
	80	*/
	81	export function finishFields(
	82	moves: readonly FinishMove[],
	83	lane?: ConflictLane,
	84	): { finishing?: number; conflict?: string; lane?: string } {
66	85	const finishing = moves.filter((m) => m.did === "queued update" \|\| m.did === "in flight").length;
67	86	const waiting = moves.length - finishing;
	87	const words =
	88	lane === undefined
	89	? null
	90	: rebaseLaneSays({ conflicted: moves.length, rebasing: lane.runs + lane.checks, room: lane.room });
68	91	return {
69	92	...(finishing === 0 ? {} : { finishing }),
70	93	...(waiting === 0 ? {} : { conflict: `${waiting} waiting` }),
	94	...(words === null ? {} : { lane: words }),
71	95	};
72	96	}
73	97
		@@ -98,7 +122,10 @@ async function conflictOf(app: App, repo: Repo, pull: Pull): Promise<string[] \|
98	122	* pull first.
99	123	*/
100	124	export async function finishApproved(app: App, repo: Repo, free: number): Promise<FinishMove[]> {
101		if (repo.autonomy !== "autonomous" \|\| repo.autonomyBy === null) return [];
	125	if (repo.autonomy !== "autonomous" \|\| repo.autonomyBy === null) {
	126	noteWaitingToFinish(repo.id, []);
	127	return [];
	128	}
102	129	const moves: FinishMove[] = [];
103	130	const builders = listCrewRoles(app.db, repo.id)
104	131	.filter((c) => c.role === "builder")
		@@ -108,6 +135,9 @@ export async function finishApproved(app: App, repo: Repo, free: number): Promis
108	135	for (const id of read.keys()) if (!ids.has(id)) read.delete(id);
109	136	let slots = free;
110	137	let room = readConflictRoom(app.db);
	138	const load = new Map<string, number>();
	139	for (const run of listActive(app.db, "agent"))
	140	if (run.actorId !== null) load.set(run.actorId, (load.get(run.actorId) ?? 0) + 1);
111	141	for (const pull of open.sort((a, b) => a.number - b.number)) {
112	142	if (!pull.authorIsAgent \|\| pull.mergedAt !== null) continue;
113	143	if (getHold(app.db, pull.id) !== undefined) continue;
		@@ -116,14 +146,26 @@ export async function finishApproved(app: App, repo: Repo, free: number): Promis
116	146	if (files === undefined) continue;
117	147	const issue = getIssueById(app.db, pull.id);
118	148	if (!issue \|\| issue.state !== "open") continue;
119		const move = finishOne(app, repo, pull, issue, files, builders, slots, room);
	149	const move = finishOne(app, repo, pull, issue, files, builders, load, slots, room);
120	150	if (move.did === "queued update") {
121	151	slots--;
122	152	room--;
	153	const by = builders.find((b) => b.user.username === move.builder)!.user.id;
	154	load.set(by, (load.get(by) ?? 0) + 1);
123	155	}
124	156	moves.push(move);
125	157	say(repo, pull, move);
126	158	}
	159	const since = new Date().toISOString();
	160	noteWaitingToFinish(
	161	repo.id,
	162	moves.map((m) => ({
	163	repo: `${repo.ownerName}/${repo.name}`,
	164	number: m.pull,
	165	since,
	166	public: repo.visibility === "public",
	167	})),
	168	);
127	169	return moves;
128	170	}
129	171
		@@ -134,6 +176,7 @@ function finishOne(
134	176	issue: Issue,
135	177	files: readonly string[],
136	178	builders: readonly Agent[],
	179	load: ReadonlyMap<string, number>,
137	180	slots: number,
138	181	room: number,
139	182	): FinishMove {
		@@ -142,12 +185,11 @@ function finishOne(
142	185	// A person's name on it is a person holding it — after a hand-over too.
143	186	if (issue.assigneeId !== null && getAgent(app.db, issue.assigneeId) === undefined)
144	187	return not("a person has it");
145		const builder =
146		builders.find((b) => b.user.id === issue.assigneeId) ??
147		builders.find((b) => b.user.id === issue.authorId) ??
148		pickBuilder(issue, builders, new Map());
149		if (builder === undefined) return not(NO_BUILDER_ON_CREW);
150		if (failedUpdates(app, issue, builder) >= CHECK_FAILED_LIMIT) return handOver(app, repo, pull, issue, files);
	188	const own = ownBuilder(issue, builders, load);
	189	if (own === undefined) return not(NO_BUILDER_ON_CREW);
	190	// Two failed updates are counted against the pull's own builder, whoever ran them.
	191	const builder = (slots > 0 && room > 0 ? lentBuilder(own, builders, load) : undefined) ?? own;
	192	if (failedUpdates(app, issue, own) >= CHECK_FAILED_LIMIT) return handOver(app, repo, pull, issue, files);
151	193	if (slots <= 0) return not(WAITS_FOR_BUILDER_SLOT);
152	194	if (room <= 0) return not("waits for room in the check lane");
153	195	const run = queueAgentRun(app, repo, issue, builder, "conflict", undefined, {
		@@ -156,7 +198,46 @@ function finishOne(
156	198	summary: `main moved: ${files.join(", ")}`,
157	199	});
158	200	if (run === undefined) return not(`${builder.user.username}'s run was not queued (ceiling or breaker)`);
159		return { pull: pull.number, did: "queued update", builder: builder.user.username };
	201	return {
	202	pull: pull.number,
	203	did: "queued update",
	204	builder: builder.user.username,
	205	...(builder === own ? {} : { for: own.user.username }),
	206	};
	207	}
	208
	209	/** The pull's own builder: the agent on it, else its author, else the builder with least in flight. */
	210	export function ownBuilder(
	211	issue: Issue,
	212	builders: readonly Agent[],
	213	load: ReadonlyMap<string, number>,
	214	): Agent \| undefined {
	215	return (
	216	builders.find((b) => b.user.id === issue.assigneeId) ??
	217	builders.find((b) => b.user.id === issue.authorId) ??
	218	pickBuilder(issue, builders, load)
	219	);
	220	}
	221
	222	/** Whose turn it is among the idle builders, kept across ticks so lent updates go round. */
	223	let turn = 0;
	224
	225	/**
	226	* The builder who takes an update when `own` already has a run in flight:
	227	* the next idle builder, round-robin by name (#6140). Undefined when `own`
	228	* is idle or every builder is busy: then the update is `own`'s.
	229	*/
	230	export function lentBuilder(
	231	own: Agent,
	232	builders: readonly Agent[],
	233	load: ReadonlyMap<string, number>,
	234	): Agent \| undefined {
	235	if ((load.get(own.user.id) ?? 0) === 0) return undefined;
	236	const idle = builders
	237	.filter((b) => (load.get(b.user.id) ?? 0) === 0)
	238	.sort((a, b) => a.user.username.localeCompare(b.user.username));
	239	if (idle.length === 0) return undefined;
	240	return idle[turn++ % idle.length];
160	241	}
161	242
162	243	/**
		@@ -203,7 +284,11 @@ function say(repo: Repo, pull: Pull, move: FinishMove): void {
203	284	const key = `${repo.id}:${pull.number}`;
204	285	if (move.did === "queued update") {
205	286	said.delete(key);
206		log.info("approved pull conflicts: brought up to date first", { ...where, builder: move.builder });
	287	log.info("approved pull conflicts: brought up to date first", {
	288	...where,
	289	builder: move.builder,
	290	...(move.for === undefined ? {} : { for: move.for }),
	291	});
207	292	return;
208	293	}
209	294	if (move.did !== "not moved") return;
		@@ -2,10 +2,13 @@
2	2	* Pulls that main moved out from under (#2039): open pulls with a
3	3	* bring-up-to-date run queued, running or failed, by how long since the
4	4	* crew was first asked to bring each one up to date. Read from the runs
5		* alone, so /flow, the terminal and MCP say the same without asking git.
	5	* alone, so /flow, the terminal and MCP say the same without asking git —
	6	* and the approved pulls the finishing pass last found conflicted and
	7	* still waiting for room, which have no run yet (#6140).
6	8	*/
7	9
8	10	import type { Database } from "bun:sqlite";
	11	import { readConflictLane } from "../crew/conflict_room.ts";
9	12	import { span } from "./pipeline.ts";
10	13
11	14	export interface ConflictedPull {
		@@ -15,8 +18,33 @@ export interface ConflictedPull {
15	18	readonly since: string;
16	19	}
17	20
	21	/** An approved pull the finishing pass found conflicted, with no run on it yet. */
	22	export interface WaitingToFinish extends ConflictedPull {
	23	readonly public: boolean;
	24	}
	25
	26	/** Repository id → its approved, conflicted pulls on the finishing pass's last tick. */
	27	const waiting = new Map<string, readonly WaitingToFinish[]>();
	28
		junipercrew 19d ago In-memory per-process map: the 'approved, waiting for room' count is only visible on the forge that ran the finishing tick, while the rest of this file is 'read from the runs alone'. Note it in FLOW.md or derive it from the db.
	29	/** What the finishing pass found this tick in one repository; a pull seen before keeps its first `since`. */
	30	export function noteWaitingToFinish(repoId: string, pulls: readonly WaitingToFinish[]): void {
	31	const before = new Map((waiting.get(repoId) ?? []).map((p) => [p.number, p.since]));
	32	if (pulls.length === 0) waiting.delete(repoId);
	33	else waiting.set(repoId, pulls.map((p) => ({ ...p, since: before.get(p.number) ?? p.since })));
	34	}
	35
18	36	/** Oldest first; `publicOnly` leaves out private repositories, for a reader who must not learn their names. */
19	37	export function conflictedPulls(db: Database, publicOnly = false): ConflictedPull[] {
	38	const ran = conflictRuns(db, publicOnly);
	39	const seen = new Set(ran.map((p) => `${p.repo}#${p.number}`));
	40	const noted = [...waiting.values()]
	41	.flat()
	42	.filter((p) => (!publicOnly \|\| p.public) && !seen.has(`${p.repo}#${p.number}`))
	43	.map(({ repo, number, since }) => ({ repo, number, since }));
	44	return noted.length === 0 ? ran : [...ran, ...noted].sort((a, b) => a.since.localeCompare(b.since));
	45	}
	46
	47	function conflictRuns(db: Database, publicOnly: boolean): ConflictedPull[] {
20	48	return db
21	49	.query<ConflictedPull, [number]>(
22	50	`SELECT o.username \|\| '/' \|\| r.name AS repo, i.number AS number,
		@@ -46,3 +74,23 @@ export function conflictedSays(
46	74	const more = pulls.length > 5 ? `, and ${pulls.length - 5} more` : "";
47	75	return `${pulls.length} conflicted ${pulls.length === 1 ? "pull" : "pulls"}, oldest first: ${list}${more}.`;
48	76	}
	77
	78	/** The lane for rebases: conflicted pulls, how many are being brought up to date, and the room for more (#6140). */
	79	export interface RebaseLane {
	80	readonly conflicted: number;
	81	/** Bring-up-to-date runs and their checks, queued or running, forge-wide. */
	82	readonly rebasing: number;
	83	/** Bring-up-to-date runs the lane has room to start now (`crew/conflict_room.ts`). */
	84	readonly room: number;
	85	}
	86
	87	export function rebaseLane(db: Database, pulls: readonly ConflictedPull[]): RebaseLane {
	88	const lane = readConflictLane(db);
	89	return { conflicted: pulls.length, rebasing: lane.runs + lane.checks, room: lane.room };
	90	}
	91
	92	/** `conflicted N, rebasing M, room R`, so a stall reads in a minute; null when nothing is conflicted or rebasing. */
	93	export function rebaseLaneSays(lane: RebaseLane): string \| null {
	94	if (lane.conflicted === 0 && lane.rebasing === 0) return null;
	95	return `conflicted ${lane.conflicted}, rebasing ${lane.rebasing}, room ${lane.room}`;
	96	}
		@@ -7,7 +7,7 @@
7	7	*/
8	8	import type { App } from "../app.ts";
9	9	import { SOURCES } from "../boards/sources.ts";
10		import { conflictedPulls, conflictedSays } from "../flow/conflicted.ts";
	10	import { conflictedPulls, conflictedSays, rebaseLane, rebaseLaneSays } from "../flow/conflicted.ts";
11	11	import { readPipeline, verdict } from "../flow/pipeline.ts";
12	12	import { type QueueReading, readQueues } from "../flow/queue.ts";
13	13	import { reviewersBusy, reviewingSays } from "../flow/reviewing.ts";
		@@ -27,6 +27,7 @@ export function flowFor(app: App, reader: Reader, now = new Date()) {
27	27	const flow = readThroughput(app.db, stages.map((s) => s.key), now);
28	28	const dup = systemFor(app, reader).duplicates;
29	29	const conflicted = conflictedPulls(app.db, reader.kind === "agent");
	30	const lane = rebaseLane(app.db, conflicted);
30	31	const requested = requestedWaits(app.db, reader.kind === "agent");
31	32	const busy = reviewersBusy(app.db, reader.kind === "agent");
32	33	return {
		@@ -34,6 +35,8 @@ export function flowFor(app: App, reader: Reader, now = new Date()) {
34	35	// Review runs in flight per reviewer, cut to public repositories for an agent (#4733).
35	36	reviewing: { says: reviewingSays(busy), reviewers: busy },
36	37	conflicted: { says: conflictedSays(conflicted, now.getTime()), pulls: conflicted },
	38	// conflicted N, rebasing M, room R: a stall in the lane reads in a minute (#6140).
	39	rebases: { ...lane, says: rebaseLaneSays(lane) },
37	40	// Pulls with changes requested, by whom they wait on (#2172).
38	41	requested: {
39	42	builder: { says: requestedSays(requested.builder, "builder", now.getTime()), pulls: requested.builder },
		@@ -16,7 +16,7 @@
16	16	import type { Database } from "bun:sqlite";
17	17	import { HOLDS } from "../../crew/held.ts";
18	18	import type { HeldRow } from "../../flow/held.ts";
19		import { conflictedPulls, conflictedSays } from "../../flow/conflicted.ts";
	19	import { conflictedPulls, conflictedSays, rebaseLane, rebaseLaneSays } from "../../flow/conflicted.ts";
20	20	import { readPipeline, STAGE_LABEL, verdict } from "../../flow/pipeline.ts";
21	21	import { readQueues, redBatchWords } from "../../flow/queue.ts";
22	22	import { reviewersBusy, reviewingSays } from "../../flow/reviewing.ts";
		@@ -47,7 +47,9 @@ export function flowLines(
47	47	const queues = agent?.queues ?? readQueues(db, now, { batches: true });
48	48	// The row is full at 58 columns: the red-batch rate (#4235) is a line under it.
49	49	const red = queues.filter((q) => (q.batches?.batches ?? 0) > 0);
50		const conflicted = conflictedSays(conflictedPulls(db, agent !== undefined), now.getTime());
	50	const conflictedList = conflictedPulls(db, agent !== undefined);
	51	const conflicted = conflictedSays(conflictedList, now.getTime());
	52	const lane = rebaseLaneSays(rebaseLane(db, conflictedList));
51	53	const requested = requestedWaits(db, agent !== undefined);
52	54	const waits = [
53	55	requestedSays(requested.builder, "builder", now.getTime()),
		@@ -75,6 +77,7 @@ export function flowLines(
75	77	...(red.length === 0
76	78	? []
77	79	: wrap(`Red batches, last hour: ${red.map((q) => `${q.repo} ${redBatchWords(q)}`).join(", ")}.`, inner)),
	80	...(lane === null ? [] : ["", ...wrap(`Rebases: ${lane}.`, inner)]),
78	81	...(conflicted === null ? [] : ["", ...wrap(conflicted, inner)]),
79	82	...waits,
80	83	...(reviewing === null ? [] : ["", ...wrap(reviewing, inner)]),
		@@ -9,7 +9,7 @@
9	9
10	10	import type { Reader } from "../../actions/telemetry.ts";
11	11	import { queuesFor, systemFor } from "../../system/telemetry.ts";
12		import { conflictedPulls, conflictedSays } from "../../flow/conflicted.ts";
	12	import { conflictedPulls, conflictedSays, rebaseLane, rebaseLaneSays } from "../../flow/conflicted.ts";
13	13	import { reviewersBusy, reviewingSays } from "../../flow/reviewing.ts";
14	14	import { requestedSays, requestedWaits } from "../../flow/requested.ts";
15	15	import {
		@@ -90,7 +90,9 @@ export function flowPanel(app: App, reader: Reader, url: URL, base: string): Htm
90	90	const out = new Map(flow.stages.map((t) => [t.key, t]));
91	91	const landed = out.get("approved");
92	92	const dups = systemFor(app, reader).duplicates;
93		const conflicted = conflictedSays(conflictedPulls(app.db, reader.kind === "agent"), now);
	93	const conflictedList = conflictedPulls(app.db, reader.kind === "agent");
	94	const conflicted = conflictedSays(conflictedList, now);
	95	const lane = rebaseLaneSays(rebaseLane(app.db, conflictedList));
94	96	const requested = requestedWaits(app.db, reader.kind === "agent");
95	97	const onBuilder = requestedSays(requested.builder, "builder", now);
96	98	const onReviewer = requestedSays(requested.reviewer, "reviewer", now);
		@@ -101,6 +103,7 @@ export function flowPanel(app: App, reader: Reader, url: URL, base: string): Htm
101	103	html`<p id="bottleneck" data-stage="${worst?.key ?? ""}">
102	104	<strong>${verdict(stages, now)}</strong>
103	105	</p>
	106	${lane === null ? "" : html`<p id="rebase-lane">Rebases: ${lane}.</p>`}
104	107	${conflicted === null ? "" : html`<p id="conflicted">${conflicted}</p>`}
105	108	${onBuilder === null ? "" : html`<p id="requested-builder">${onBuilder}</p>`}
106	109	${onReviewer === null ? "" : html`<p id="requested-reviewer">${onReviewer}</p>`}