pelikhan · Oct 2, 2026 · Oct 3, 2026 · Oct 3, 2026 · Oct 3, 2026 · Oct 3, 2026
diff --git a/.github/workflows/smoke-dispatch-work-coordinator.lock.yml b/.github/workflows/smoke-dispatch-work-coordinator.lock.yml
diff --git a/actions/setup/js/dispatch_work_coordinator_codemods.cjs b/actions/setup/js/dispatch_work_coordinator_codemods.cjs
@@ -1,11 +1,29 @@
 // @ts-check
 "use strict";
 
-const CURRENT_VERSION = 1;
+const CURRENT_VERSION = 3;
 
 // Keep each successive protocol transformation as data so historical upgrades
 // remain inspectable and deterministic. An absent version is the original log.
-const CODEMODS = Object.freeze([Object.freeze({ from: 0, to: 1, set: Object.freeze({ version: 1 }) })]);
+const CODEMODS = Object.freeze([
+  Object.freeze({ from: 0, to: 1, rename: Object.freeze({}), set: Object.freeze({ version: 1 }) }),
+  Object.freeze({ from: 1, to: 2, rename: Object.freeze({ work: "work_id", claim: "claim_id", attempt: "attempt_id" }), set: Object.freeze({ version: 2 }) }),
+  Object.freeze({ from: 2, to: 3, rename: Object.freeze({}), set: Object.freeze({ version: 3 }) }),
+]);
+
+function applyCodemods(message, version) {
+  let upgraded = { ...message };
+  for (let next = version; next < CURRENT_VERSION; next++) {
+    const codemod = CODEMODS.find(item => item.from === next);
+    if (!codemod || codemod.to !== next + 1) throw new TypeError("missing dispatch coordinator transaction codemod");
+    for (const [from, to] of Object.entries(codemod.rename)) {
+      if (upgraded[from] !== null) upgraded[to] = upgraded[from];
+      delete upgraded[from];
+    }
+    upgraded = { ...upgraded, ...codemod.set };
+  }
+  return upgraded;
+}
 
 function upgradeTransaction(message) {
   if (!message || typeof message !== "object" || Array.isArray(message)) {
@@ -15,15 +33,56 @@ function upgradeTransaction(message) {
   if (!Number.isSafeInteger(version) || version < 0 || version > CURRENT_VERSION) {
     throw new TypeError("unsupported dispatch coordinator transaction version");
   }
-  let upgraded = message;
-  for (let next = version; next < CURRENT_VERSION; next++) {
-    const codemod = CODEMODS.find(item => item.from === next);
-    if (!codemod || codemod.to !== next + 1) {
-      throw new TypeError("missing dispatch coordinator transaction codemod");
-    }
-    upgraded = { ...upgraded, ...codemod.set };
+  if (version === CURRENT_VERSION) return message;
+  if (Object.hasOwn(message, "work_id")) {
+    if (version === 0) return { ...message, version: CURRENT_VERSION };
+    if (version === 2) return applyCodemods(message, version);
+    throw new TypeError("unsupported dispatch coordinator transaction version");
+  }
+  if (version > 1) throw new TypeError("versioned dispatch coordinator transactions require work_id");
+  const fields = Object.keys(message)
+    .filter(field => field !== "version")
+    .sort();
+  const expected = ["attempt", "claim", "kind", "work"];
+  if (fields.length !== expected.length || fields.some((field, index) => field !== expected[index])) {
+    throw new TypeError("legacy transaction must contain exactly version, kind, work, claim, and attempt");
+  }
+  if (["Work", "WorkCancellation"].includes(message.kind) && (message.claim !== null || message.attempt !== null)) {
+    throw new TypeError("legacy Work must not include a claim or attempt");
+  }
+  if (["Claim", "ClaimCancellation"].includes(message.kind) && message.attempt !== null) {
+    throw new TypeError("legacy Claim must not include an attempt");
   }
+  const upgraded = applyCodemods(message, version);
+  if (upgraded.kind === "Work") upgraded.work = { legacy_work_id: upgraded.work_id };
+  if (upgraded.kind === "Claim") upgraded.run_id = `legacy:${upgraded.claim_id}`;
   return upgraded;
 }
 
-module.exports = { CURRENT_VERSION, CODEMODS, upgradeTransaction };
+function upgradeTransactions(messages) {
+  const upgraded = messages.map(upgradeTransaction);
+  const sequences = new Map();
+  const sequenceOwners = new Map();
+  let maximum = 0;
+  for (const message of upgraded) {
+    if (message.kind === "Work" && message.sequence !== undefined) {
+      if (!Number.isSafeInteger(message.sequence) || message.sequence < 1) throw new TypeError("work sequence must be a positive safe integer");
+      const owner = sequenceOwners.get(message.sequence);
+      if (sequenceOwners.has(message.sequence) && owner !== message.work_id) throw new TypeError(`work sequence ${message.sequence} is shared by Work ${owner} and ${message.work_id}`);
+      sequenceOwners.set(message.sequence, message.work_id);
+      maximum = Math.max(maximum, message.sequence);
+      sequences.set(message.work_id, message.sequence);
+    }
+  }
+  return upgraded.map((message, index) => {
+    if (message.kind !== "Work" || message.sequence !== undefined) return message;
+    if (messages[index].version >= 2) throw new TypeError("work sequence is required");
+    if (!sequences.has(message.work_id)) {
+      if (maximum === Number.MAX_SAFE_INTEGER) throw new RangeError("work queue sequence exhausted");
+      sequences.set(message.work_id, ++maximum);
+    }
+    return { ...message, sequence: sequences.get(message.work_id) };
+  });
+}
+
+module.exports = { CURRENT_VERSION, CODEMODS, upgradeTransaction, upgradeTransactions };
diff --git a/actions/setup/js/dispatch_work_coordinator_codemods.test.cjs b/actions/setup/js/dispatch_work_coordinator_codemods.test.cjs
@@ -0,0 +1,55 @@
+// @ts-check
+import { describe, expect, it } from "vitest";
+import { CODEMODS, CURRENT_VERSION, upgradeTransaction, upgradeTransactions } from "./dispatch_work_coordinator_codemods.cjs";
+import { applyTransactions, parseTransactionLog, replayTransactions, serializeTransactionLog } from "./dispatch_work_coordinator_replay.cjs";
+import { selectNext } from "./dispatch_work_coordinator_selection.cjs";
+
+const versionTwo = [
+  { version: 2, kind: "Work", work_id: "z", work: { priority: 3, group: "a", nested: { value: true } }, sequence: 42 },
+  { version: 2, kind: "Work", work_id: "a", work: { priority: 1, group: "b" }, sequence: 43 },
+  { version: 2, kind: "Claim", work_id: "a", claim_id: "old", run_id: "trusted-run" },
+  { version: 2, kind: "ClaimCancellation", work_id: "a", claim_id: "old" },
+];
+
+describe("version 3 dispatch coordinator codemod", () => {
+  it("declares an ordered, header-only version-2 upgrade", () => {
+    expect(CURRENT_VERSION).toBe(3);
+    expect(CODEMODS.find(codemod => codemod.from === 2)).toEqual({ from: 2, to: 3, rename: {}, set: { version: 3 } });
+    const original = structuredClone(versionTwo);
+    expect(upgradeTransactions(versionTwo)).toEqual(versionTwo.map(transaction => ({ ...transaction, version: 3 })));
+    expect(versionTwo).toEqual(original);
+    for (const transaction of upgradeTransactions(versionTwo)) expect(upgradeTransaction(transaction)).toBe(transaction);
+  });
+
+  it("preserves every message kind, identities, payloads, sequence, and provenance", () => {
+    for (const transaction of [...versionTwo, { version: 2, kind: "Completion", work_id: "a", claim_id: "old", attempt_id: "attempt", outcome: "" }, { version: 2, kind: "WorkCancellation", work_id: "z" }]) {
+      expect(upgradeTransaction(transaction)).toEqual({ ...transaction, version: 3 });
+    }
+  });
+
+  it("preserves replay, selection, compaction, and future mutation decisions", () => {
+    const upgraded = parseTransactionLog(versionTwo.map(transaction => JSON.stringify(transaction)).join("\n") + "\n");
+    expect(replayTransactions(upgraded)).toEqual(replayTransactions(versionTwo));
+    expect(selectNext(upgraded).work_id).toBe("z");
+    expect(selectNext(parseTransactionLog(serializeTransactionLog(upgraded))).work_id).toBe("z");
+    const claim = { version: 3, kind: "Claim", work_id: "z", claim_id: "new", run_id: "new-run" };
+    expect(applyTransactions(upgraded, [claim])).toEqual(applyTransactions(versionTwo, [claim]));
+  });
+
+  it("does not invent missing version-2 metadata or accept future versions", () => {
+    const invalid = [
+      { version: 2, kind: "Work", work_id: "w", work: {} },
+      { version: 2, kind: "Claim", work_id: "a", claim_id: "c" },
+      { ...versionTwo[0], version: 4 },
+    ];
+    for (const transaction of invalid) {
+      expect(() => parseTransactionLog(`${JSON.stringify(versionTwo[0])}\n${JSON.stringify(transaction)}\n`)).toThrow();
+    }
+  });
+
+  it("rejects sequence exhaustion while assigning historical FIFO ranks", () => {
+    const maximum = { version: 3, kind: "Work", work_id: "max", work: { task: "max" }, sequence: Number.MAX_SAFE_INTEGER };
+    const historical = { version: 1, kind: "Work", work: "legacy", claim: null, attempt: null };
+    expect(() => upgradeTransactions([maximum, historical])).toThrow("sequence exhausted");
+  });
+});