packages/forkhorse-core/src/race-entrants.ts

1import type { Claim, Packet, RaceEntrant, RaceRecord, RaceState, Review } from "./workflow-types";
2import { tokenBudgetState, usageSources } from "./usage";
3
4export type { RaceEntrant } from "./workflow-types";
5
6type Storage = DurableObjectStorage | DurableObjectTransaction;
7export const MAX_RACE_ENTRANTS = 3;
8export const entrantKey = (packet: string, agent: string) => `entrant:${packet}:${agent}`;
9export const publicPacket = (packet: Packet) => ({ ...packet, mode: packet.mode ?? "standard" });
10
11export async function entrants(storage: Storage, packet?: string): Promise<RaceEntrant[]> {
12  return [...(await storage.list<RaceEntrant>({
13    prefix: packet === undefined ? "entrant:" : `entrant:${packet}:`,
14  })).entries()]
15    .filter(([key, entrant]) => key === entrantKey(entrant.packet, entrant.agent))
16    .map(([, entrant]) => entrant);
17}
18
19/** Shared archive/claim boundary: only closedAt === null is open. Never deletes history. */
20export async function openEntrants(storage: Storage, packet?: string): Promise<RaceEntrant[]> {
21  return (await entrants(storage, packet)).filter(entrant => entrant.closedAt === null);
22}
23
24/** Standard records retain their existing key and shape. Race lookup is caller-scoped. */
25export async function claimForAgent(storage: Storage, packet: string, agent: string): Promise<Claim | undefined> {
26  if ((await storage.get<Packet>(`packet:${packet}`))?.mode !== "race")
27    return storage.get<Claim>(`claim:${packet}`);
28  const entrant = await storage.get<RaceEntrant>(entrantKey(packet, agent));
29  return entrant?.closedAt === null ? entrant : undefined;
30}
31
32export async function liveClaims(storage: Storage): Promise<Claim[]> {
33  return [...(await storage.list<Claim>({ prefix: "claim:" })).values(), ...await openEntrants(storage)];
34}
35
36type Milestone = "firstPushAt" | "reviewRequestedAt" | "receiptPassedAt" | "approvedAt" | "promotedAt";
37/** Call inside the same transaction as the milestone's authoritative event. */
38export async function markEntrant(
39  storage: Storage, claim: Pick<Claim, "packet" | "agent" | "repoName">,
40  milestone: Milestone, at: string,
41): Promise<void> {
42  const key = entrantKey(claim.packet, claim.agent);
43  const entrant = await storage.get<RaceEntrant>(key);
44  if (entrant?.closedAt === null && entrant.repoName === claim.repoName && entrant[milestone] === null)
45    await storage.put(key, { ...entrant, [milestone]: at });
46}
47
48const milestones = ["createdAt", "firstPushAt", "reviewRequestedAt", "receiptPassedAt", "approvedAt", "promotedAt"] as const;
49function progress(entrant: RaceEntrant): { rank: number; at: string } {
50  for (let rank = milestones.length - 1; rank >= 0; rank--) {
51    const at = entrant[milestones[rank]!];
52    if (at !== null) return { rank, at };
53  }
54  throw new Error("Race entrant has no claim time");
55}
56
57/** Rank the finish snapshot before closing entrants; withdrawals always follow runners. */
58export function placedEntrants(records: RaceEntrant[], winner: string): RaceEntrant[] {
59  return records.toSorted((a, b) => {
60    if (a.agent === b.agent) return 0;
61    if (a.agent === winner) return -1;
62    if (b.agent === winner) return 1;
63    const closed = Number(a.closedAt !== null) - Number(b.closedAt !== null);
64    if (closed) return closed;
65    const left = progress(a), right = progress(b);
66    return right.rank - left.rank || Date.parse(left.at) - Date.parse(right.at) || a.agent.localeCompare(b.agent);
67  });
68}
69
70/** Separate from the refused promotion's transaction; never overwrites a first refusal. */
71export async function recordPromoteRejected(storage: DurableObjectStorage, review: Review, found: string): Promise<void> {
72  const at = new Date().toISOString();
73  await storage.transaction(async txn => {
74    const key = entrantKey(review.packet, review.agent);
75    const entrant = await txn.get<RaceEntrant>(key);
76    if (entrant && entrant.repoName === review.repoName && !entrant.promoteRejected)
77      await txn.put(key, { ...entrant, promoteRejected: { at, expected: review.base, found } });
78  });
79}
80
81/** Finish metadata and entrants are committed together by the successful promote. */
82export async function raceStates(storage: Storage, packets: Packet[]): Promise<RaceState[]> {
83  const records = await entrants(storage);
84  const results = await storage.list<RaceRecord>({ prefix: "race:" });
85  return packets.filter(packet => packet.mode === "race").map(packet => {
86    const result = results.get(`race:${packet.id}`);
87    return {
88      packet: packet.id, status: result?.finishedAt ? "finished" : "open",
89      judge: packet.judge ?? null, verdict: result?.verdict ?? null,
90      tokenBudget: packet.tokenBudget ?? null,
91      finishedAt: result?.finishedAt ?? null, winner: result?.winner ?? null,
92      ledgerEntryId: result?.ledgerEntryId ?? null, places: result?.places ?? [],
93      entrants: records.filter(entrant => entrant.packet === packet.id)
94        .toSorted((a, b) => a.createdAt.localeCompare(b.createdAt) || a.agent.localeCompare(b.agent))
95        .map(({ packet: _packet, paths: _paths, createdAt, ...entrant }) => ({
96          ...entrant, claimedAt: createdAt, promoteRejected: entrant.promoteRejected ?? null,
97          usage: usageSources(entrant.usage), budget: tokenBudgetState(packet.tokenBudget, entrant.usage),
98        })),
99    };
100  });
101}
102