Implement best-effort oldest-first work queue selection

Add matching enqueue-time selection in JavaScript and Go, preserve first-submission age and legacy records, expose snapshot recommendations, and default operator claims to the oldest available Work without reselecting on publication retry. Update schemas and exercise both implementations against shared fixtures. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
pelikhan · Oct 3, 2026 · Oct 3, 2026 · Oct 3, 2026 · Oct 3, 2026 · Oct 3, 2026
commit c118da75ba16f3c8e8fe8f42bb83a427f4d1e1ea
diff --git a/actions/setup/js/work_queue_mcp_server.cjs b/actions/setup/js/work_queue_mcp_server.cjs
@@ -40,7 +40,7 @@ function readWorkQueueState(snapshot, args = {}) {
     throw new TypeError("work must be a non-empty string when provided");
   }
 
-  const workIds = args.work === undefined ? Object.keys(snapshot.projection.work) : [args.work];
+  const workIds = args.work === undefined ? [...snapshot.projection.available, ...Object.keys(snapshot.projection.work).filter(work => !snapshot.projection.available.includes(work))] : [args.work];
   const works = workIds.map(work => {
     if (!Object.hasOwn(snapshot.projection.work, work)) {
       return { id: work, state: "absent", winner: null, claims: [] };
@@ -54,18 +54,20 @@ function readWorkQueueState(snapshot, args = {}) {
     return {
       id: work,
       state: snapshot.projection.work[work],
+      enqueued: snapshot.projection.transactions.find(transaction => transaction.kind === "Work" && transaction.work === work)?.enqueued ?? 0,
       winner: snapshot.projection.winner[work],
       claims,
     };
   });
 
-  return { snapshot_sha: snapshot.sha, works };
+  return { snapshot_sha: snapshot.sha, next_work: snapshot.projection.available[0] ?? null, works };
 }
 
 function createWorkQueueStateTool(snapshot) {
   return {
     name: "work_queue_read",
-    description: "Read the immutable work queue snapshot captured during workflow activation. This view may be stale during agent execution; safe-output processing rechecks authority before publishing changes.",
+    description:
+      "Read the immutable work queue snapshot captured during workflow activation. Available Work is ordered oldest-first; next_work recommends the oldest available item. This view may be stale during agent execution; safe-output processing rechecks authority before publishing changes without enforcing FIFO.",
     inputSchema: {
       type: "object",
       properties: {

diff --git a/actions/setup/js/work_queue_mcp_server.test.cjs b/actions/setup/js/work_queue_mcp_server.test.cjs
@@ -4,7 +4,7 @@ import fs from "fs";
 import os from "os";
 import path from "path";
 import { createWorkQueueFinishTool, createWorkQueueStateTool, loadWorkQueueSnapshot, readWorkQueueState } from "./work_queue_mcp_server.cjs";
-import { serializeTransactionLog } from "./work_queue_replay.cjs";
+import { createWorkTransaction, serializeTransactionLog } from "./work_queue_replay.cjs";
 
 const work = id => ({ version: 1, kind: "Work", work: id, claim: null, attempt: null });
 const claim = (workId, id) => ({ version: 1, kind: "Claim", work: workId, claim: id, attempt: null });
@@ -35,12 +35,14 @@ describe("work queue MCP snapshot", () => {
 
     expect(readWorkQueueState(snapshot)).toEqual({
       snapshot_sha: "activation-head",
-      works: [{ id: "w", state: "claimed", winner: "c", claims: [{ id: "c", state: "effective" }] }],
+      next_work: null,
+      works: [{ id: "w", state: "claimed", enqueued: 0, winner: "c", claims: [{ id: "c", state: "effective" }] }],
     });
     const tool = createWorkQueueStateTool(snapshot);
     expect(tool.name).toBe("work_queue_read");
     expect(tool.handler({ work: "missing" })).toEqual({
       snapshot_sha: "activation-head",
+      next_work: null,
       works: [{ id: "missing", state: "absent", winner: null, claims: [] }],
     });
   });
@@ -55,10 +57,26 @@ describe("work queue MCP snapshot", () => {
 
     expect(readWorkQueueState(loadWorkQueueSnapshot(snapshotPath), { work: "constructor" })).toEqual({
       snapshot_sha: null,
-      works: [{ id: "constructor", state: "claimed", winner: "toString", claims: [{ id: "toString", state: "effective" }] }],
+      next_work: null,
+      works: [{ id: "constructor", state: "claimed", enqueued: 0, winner: "toString", claims: [{ id: "toString", state: "effective" }] }],
     });
   });
 
+  it("recommends and lists available work oldest-first from the immutable snapshot", () => {
+    const snapshotPath = writeSnapshot({
+      version: 2,
+      sha: "old-head",
+      transactionLog: serializeTransactionLog([createWorkTransaction("a-new", 20), createWorkTransaction("z-old", 10), createWorkTransaction("claimed", 1), claim("claimed", "c")]),
+      worker: null,
+    });
+    const snapshot = loadWorkQueueSnapshot(snapshotPath);
+    const result = readWorkQueueState(snapshot);
+    expect(result.next_work).toBe("z-old");
+    expect(result.works.map(item => item.id)).toEqual(["z-old", "a-new", "claimed"]);
+    expect(result.works.map(item => item.enqueued)).toEqual([10, 20, 1]);
+    expect(readWorkQueueState(snapshot, { work: "a-new" }).next_work).toBe("z-old");
+  });
+
   it("rejects snapshots with an unsupported shape or invalid transaction log", () => {
     expect(() => loadWorkQueueSnapshot(writeSnapshot({ version: 1, sha: null, transactionLog: "" }))).toThrow(/invalid shape/);
     expect(() => loadWorkQueueSnapshot(writeSnapshot({ version: 2, sha: null, transactionLog: "{}\n", worker: null }))).toThrow(/invalid transaction log/);

diff --git a/actions/setup/js/work_queue_replay.cjs b/actions/setup/js/work_queue_replay.cjs
@@ -20,7 +20,7 @@ function debugLog(message, details = {}) {
 
 /**
  * @typedef {
- *   | {version: number, kind: "Work", work: string, claim: null, attempt: null}
+ *   | {version: number, kind: "Work", work: string, claim: null, attempt: null, enqueued?: number}
  *   | {version: number, kind: "Claim", work: string, claim: string, attempt: null}
  *   | {version: number, kind: "ClaimCancellation", work: string, claim: string, attempt: null}
  *   | {version: number, kind: "Completion", work: string, claim: string, attempt: string}
@@ -40,8 +40,12 @@ function validateTransaction(transaction) {
   /** @type {Record<string, unknown>} */
   const candidate = Object.assign(Object.create(null), transaction);
   const fields = Object.keys(candidate).sort();
-  if (fields.length !== SORTED_TRANSACTION_FIELDS.length || fields.some((field, index) => field !== SORTED_TRANSACTION_FIELDS[index])) {
-    throw new TypeError("transaction must contain exactly version, kind, work, claim, and attempt");
+  const expectedFields = candidate.kind === "Work" && Object.hasOwn(candidate, "enqueued") ? [...SORTED_TRANSACTION_FIELDS, "enqueued"].sort() : SORTED_TRANSACTION_FIELDS;
+  if (fields.length !== expectedFields.length || fields.some((field, index) => field !== expectedFields[index])) {
+    throw new TypeError("transaction must contain exactly version, kind, work, claim, and attempt, with optional enqueued only on Work");
+  }
+  if (Object.hasOwn(candidate, "enqueued") && (typeof candidate.enqueued !== "number" || !Number.isSafeInteger(candidate.enqueued) || candidate.enqueued < 0)) {
+    throw new TypeError("transaction enqueued must be a non-negative safe integer");
   }
   if (candidate.version !== CURRENT_VERSION) {
     throw new TypeError("unsupported work queue transaction version");
@@ -87,7 +91,7 @@ function validateTransaction(transaction) {
  * @returns {string}
  */
 function transactionKey(transaction) {
-  return JSON.stringify([transaction.version, transaction.kind, transaction.work, transaction.claim, transaction.attempt]);
+  return JSON.stringify([transaction.version, transaction.kind, transaction.work, transaction.claim, transaction.attempt, transaction.kind === "Work" ? (transaction.enqueued ?? 0) : 0]);
 }
 
 /**
@@ -101,6 +105,7 @@ function copyTransaction(transaction) {
     work: transaction.work,
     claim: transaction.claim,
     attempt: transaction.attempt,
+    ...(transaction.kind === "Work" && transaction.enqueued ? { enqueued: transaction.enqueued } : {}),
   });
 }
 
@@ -135,6 +140,9 @@ function collectFacts(transactions) {
   for (const transaction of facts.values()) {
     switch (transaction.kind) {
       case "Work":
+        if (works.has(transaction.work)) {
+          throw new TypeError(`work ${transaction.work} has conflicting enqueue metadata`);
+        }
         works.add(transaction.work);
         break;
       case "Claim":
@@ -208,11 +216,13 @@ function replayTransactions(transactions) {
   const cancellations = new Set();
   const completions = new Map();
   const workCancellations = new Set();
+  const enqueueTimes = new Map();
 
   for (const transaction of facts.values()) {
     switch (transaction.kind) {
       case "Work":
         works.add(transaction.work);
+        enqueueTimes.set(transaction.work, transaction.enqueued ?? 0);
         break;
       case "Claim":
         claims.set(transaction.claim, transaction.work);
@@ -256,12 +266,14 @@ function replayTransactions(transactions) {
       return [claim, state];
     })
   );
+  const available = sortedWorks.filter(work => getWorkState(work).state === "available").sort((left, right) => enqueueTimes.get(left) - enqueueTimes.get(right) || Buffer.compare(Buffer.from(left, "utf8"), Buffer.from(right, "utf8")));
 
   debugLog("replay completed", { transactions: facts.size, works: works.size, claims: claims.size });
   return {
     work: Object.fromEntries(sortedWorks.map(work => [work, getWorkState(work).state])),
     winner: Object.fromEntries(sortedWorks.map(work => [work, getWorkState(work).winner])),
     claim: claimState,
+    available,
     transactions: [...facts.values()].sort((left, right) => {
       const leftKey = transactionKey(left);
       const rightKey = transactionKey(right);
@@ -308,7 +320,10 @@ function applyTransactions(transactions, intents) {
     let reason = "";
     switch (transaction.kind) {
       case "Work":
-        if (hasWork) reason = "work already exists";
+        if (hasWork) {
+          debugLog("work already submitted", { kind: transaction.kind });
+          return;
+        }
         break;
       case "Claim":
         if (!hasWork) reason = "work does not exist";
@@ -359,6 +374,41 @@ function compactTransactions(transactions) {
   return replayTransactions(transactions).transactions;
 }
 
+/**
+ * Capture enqueue time once, before staging or publication retries.
+ * @param {string} work
+ * @param {number} [enqueued]
+ * @returns {WorkQueueTransaction}
+ */
+function createWorkTransaction(work, enqueued = Date.now()) {
+  const transaction = { version: CURRENT_VERSION, kind: "Work", work, claim: null, attempt: null, enqueued };
+  validateTransaction(transaction);
+  return copyTransaction(transaction);
+}
+
+/**
+ * @param {WorkQueueTransaction[]} transactions
+ * @returns {string | null}
+ */
+function oldestAvailableWork(transactions) {
+  return replayTransactions(transactions).available[0] ?? null;
+}
+
+/**
+ * Select from the local view, including previously staged intents. Publication
+ * revalidates this fixed Claim for safety, not FIFO.
+ * @param {WorkQueueTransaction[]} transactions
+ * @param {string} claim
+ * @returns {WorkQueueTransaction}
+ */
+function claimOldestAvailableWork(transactions, claim) {
+  const work = oldestAvailableWork(transactions);
+  if (work === null) throw new Error("no available Work to claim");
+  const transaction = { version: CURRENT_VERSION, kind: "Claim", work, claim, attempt: null };
+  validateTransaction(transaction);
+  return copyTransaction(transaction);
+}
+
 /**
  * @param {string} contents
  * @returns {WorkQueueTransaction[]}
@@ -409,6 +459,9 @@ function serializeTransactionLog(transactions) {
 module.exports = {
   applyTransactions,
   compactTransactions,
+  createWorkTransaction,
+  claimOldestAvailableWork,
+  oldestAvailableWork,
   parseTransactionLog,
   replayTransactions,
   serializeTransactionLog,

diff --git a/actions/setup/js/work_queue_replay.test.cjs b/actions/setup/js/work_queue_replay.test.cjs
@@ -1,6 +1,8 @@
 // @ts-check
 import { describe, expect, it, vi } from "vitest";
-import { applyTransactions, compactTransactions, parseTransactionLog, replayTransactions, serializeTransactionLog, validateTransaction } from "./work_queue_replay.cjs";
+import { createHash } from "node:crypto";
+import selectionFixtures from "../../../specs/work-queue/selection-fixtures.json";
+import { applyTransactions, claimOldestAvailableWork, compactTransactions, createWorkTransaction, oldestAvailableWork, parseTransactionLog, replayTransactions, serializeTransactionLog, validateTransaction } from "./work_queue_replay.cjs";
 
 const work = id => ({ version: 1, kind: "Work", work: id, claim: null, attempt: null });
 const claim = (workId, id) => ({ version: 1, kind: "Claim", work: workId, claim: id, attempt: null });
@@ -35,10 +37,74 @@ describe("work queue replay", () => {
       work: { w: "claimed" },
       winner: { w: "c" },
       claim: { c: "effective" },
+      available: [],
       transactions: [claim("w", "c"), work("w")],
     });
   });
 
+  describe("oldest-available work selection", () => {
+    for (const fixture of selectionFixtures) {
+      it(fixture.name, () => {
+        const id = task => createHash("sha256").update(JSON.stringify({ task })).digest("hex");
+        const transactions = fixture.works.flatMap(item => {
+          const workId = id(item.task);
+          const result = [item.enqueued === undefined ? work(workId) : createWorkTransaction(workId, item.enqueued)];
+          const claimId = `claim-${item.task}`;
+          if (["claimed", "completed", "released"].includes(item.state)) result.push(claim(workId, claimId));
+          if (item.state === "completed") result.push(complete(workId, claimId, `attempt-${item.task}`));
+          if (item.state === "cancelled") result.push(cancelWork(workId));
+          if (item.state === "released") result.push(cancelClaim(workId, claimId));
+          return result;
+        });
+        const expected = fixture.expected.map(id);
+        for (const log of [transactions, [...transactions].reverse(), [...transactions, ...transactions], compactTransactions(transactions), parseTransactionLog(serializeTransactionLog(transactions))]) {
+          expect(replayTransactions(log).available).toEqual(expected);
+          expect(oldestAvailableWork(log)).toBe(expected[0] ?? null);
+        }
+      });
+    }
+
+    it("captures enqueue time once and keeps original age on resubmission", () => {
+      const now = vi.spyOn(Date, "now").mockReturnValue(100);
+      try {
+        const first = createWorkTransaction("w");
+        expect(first.enqueued).toBe(100);
+        expect(Object.isFrozen(first)).toBe(true);
+        now.mockReturnValue(200);
+        const result = applyTransactions([first], [createWorkTransaction("w")]);
+        expect(result.transactions).toEqual([first]);
+        expect(result.rejected).toEqual([]);
+      } finally {
+        now.mockRestore();
+      }
+    });
+
+    it("stages successive claims from the local view without head-of-line blocking", () => {
+      let log = [createWorkTransaction("old", 1), createWorkTransaction("new", 2)];
+      const first = claimOldestAvailableWork(log, "first");
+      expect(first.work).toBe("old");
+      log = applyTransactions(log, [first]).transactions;
+      expect(claimOldestAvailableWork(log, "second").work).toBe("new");
+      expect(() => claimOldestAvailableWork([], "none")).toThrow("no available Work");
+      expect(() => claimOldestAvailableWork(log, "")).toThrow("claim must be");
+    });
+
+    it("uses UTF-8 identity order consistently with Go and handles numeric/prototype keys", () => {
+      expect(replayTransactions(["\u{10000}", "\uE000", "2", "10", "__proto__"].map(id => createWorkTransaction(id, 1))).available).toEqual(["10", "2", "__proto__", "\uE000", "\u{10000}"]);
+    });
+
+    it("rejects invalid or conflicting enqueue metadata instead of rewriting history", () => {
+      for (const enqueued of [-1, 0.5, Number.MAX_SAFE_INTEGER + 1, Infinity, "1", null, undefined]) {
+        expect(() => validateTransaction({ ...work("w"), enqueued })).toThrow("enqueued");
+      }
+      expect(() => validateTransaction({ ...claim("w", "c"), enqueued: 1 })).toThrow("optional enqueued only on Work");
+      expect(() => replayTransactions([createWorkTransaction("w", 1), createWorkTransaction("w", 2)])).toThrow("conflicting enqueue metadata");
+      expect(() => replayTransactions([work("w"), createWorkTransaction("w", 1)])).toThrow("conflicting enqueue metadata");
+      expect(replayTransactions([work("w"), { ...work("w"), enqueued: 0 }]).transactions).toEqual([work("w")]);
+      expect(createWorkTransaction("maximum", Number.MAX_SAFE_INTEGER).enqueued).toBe(Number.MAX_SAFE_INTEGER);
+    });
+  });
+
   it("produces the same projection for every ordering of a valid fact set", () => {
     const transactions = [work("w"), claim("w", "a"), claim("w", "b"), cancelClaim("w", "a"), work("other")];
     const expected = replayTransactions(transactions);

diff --git a/actions/setup/js/work_queue_store.cjs b/actions/setup/js/work_queue_store.cjs
@@ -11,7 +11,7 @@ const DEFAULT_MAX_RETRIES = 5;
 
 /**
  * @typedef {
- *   | {version: number, kind: "Work", work: string, claim: null, attempt: null}
+ *   | {version: number, kind: "Work", work: string, claim: null, attempt: null, enqueued?: number}
  *   | {version: number, kind: "Claim", work: string, claim: string, attempt: null}
  *   | {version: number, kind: "ClaimCancellation", work: string, claim: string, attempt: null}
  *   | {version: number, kind: "Completion", work: string, claim: string, attempt: string}

diff --git a/actions/setup/js/work_queue_store.test.cjs b/actions/setup/js/work_queue_store.test.cjs
@@ -1,7 +1,7 @@
 // @ts-check
 import { describe, expect, it, vi } from "vitest";
 import { applyAndPublishWorkQueueTransactions, WORK_QUEUE_BRANCH, WORK_QUEUE_LOG_PATH, readWorkQueueLog } from "./work_queue_store.cjs";
-import { parseTransactionLog, serializeTransactionLog } from "./work_queue_replay.cjs";
+import { claimOldestAvailableWork, createWorkTransaction, parseTransactionLog, replayTransactions, serializeTransactionLog } from "./work_queue_replay.cjs";
 
 const work = id => ({ version: 1, kind: "Work", work: id, claim: null, attempt: null });
 const claim = (workId, id) => ({ version: 1, kind: "Claim", work: workId, claim: id, attempt: null });
@@ -85,6 +85,28 @@ function createFakeGitHub() {
 }
 
 describe("work queue Git store", () => {
+  it("preserves enqueue age and the selected Work when an older submission arrives during retry", async () => {
+    const fake = createFakeGitHub();
+    const queued = createWorkTransaction("selected", 100);
+    await applyAndPublishWorkQueueTransactions({ githubClient: fake.githubClient, owner: "owner", repo: "repo", intents: [queued] });
+    const selected = claimOldestAvailableWork(fake.state.transactions, "c");
+    const later = createWorkTransaction("later", 200);
+    fake.state.conflictOnce = true;
+    const result = await applyAndPublishWorkQueueTransactions({
+      githubClient: fake.githubClient,
+      owner: "owner",
+      repo: "repo",
+      intents: [selected, later],
+      sleepFn: async () => {},
+    });
+    expect(result.rejected).toEqual([]);
+    expect(result.transactions).toContainEqual(queued);
+    expect(result.transactions).toContainEqual(later);
+    expect(result.transactions).toContainEqual(selected);
+    expect(replayTransactions(result.transactions).available).toEqual(["remote-work", "later"]);
+    expect(fake.state.updateCalls).toBe(2);
+  });
+
   it("creates the dedicated branch and writes the canonical transaction log", async () => {
     const fake = createFakeGitHub();
     const result = await applyAndPublishWorkQueueTransactions({