From 0ff49eb25b904a957e6e7221c8a1ae2e44c0d390 Mon Sep 17 00:00:00 2001 From: pelikhan Date: Sun, 8 Nov 2026 19:59:15 -0800 Subject: [PATCH 1/5] Cancel obsolete worker backlog during queue maintenance Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../patch-work-queue-obsolete-contracts.md | 5 + .github/aw/work-queue.md | 5 +- actions/setup/js/reconcile_work_queue_run.cjs | 5 +- .../js/reconcile_work_queue_run.test.cjs | 40 ++++++ actions/setup/js/work_queue_deployment.cjs | 6 +- .../work_queue_deployment_authority.test.cjs | 136 +++++++++++++++++- .../js/work_queue_deployment_control.cjs | 22 ++- actions/setup/js/work_queue_replay.cjs | 4 +- docs/src/content/docs/reference/work-queue.md | 2 + pkg/workqueue/deployment.go | 8 ++ pkg/workqueue/deployment_test.go | 63 ++++++++ pkg/workqueue/replay.go | 9 +- 12 files changed, 294 insertions(+), 11 deletions(-) create mode 100644 .changeset/patch-work-queue-obsolete-contracts.md diff --git a/.changeset/patch-work-queue-obsolete-contracts.md b/.changeset/patch-work-queue-obsolete-contracts.md new file mode 100644 index 00000000000..f791618da5a --- /dev/null +++ b/.changeset/patch-work-queue-obsolete-contracts.md @@ -0,0 +1,5 @@ +--- +"gh-aw": patch +--- + +Terminally cancel obsolete unpinned work during queue maintenance, preserving compatible deployments, exact execution pins, delivery barriers and ledger history. Recheck `worker_contract_obsolete` cancellations against the current approved active contract in both queue replay implementations. diff --git a/.github/aw/work-queue.md b/.github/aw/work-queue.md index 79fcec2a151..e0f1ddc0052 100644 --- a/.github/aw/work-queue.md +++ b/.github/aw/work-queue.md @@ -114,7 +114,10 @@ Lost responses, dispatcher cancellation/deadlines never prove nonlaunch/terminat Retain reservations until exact evidence. Never rerun uncertain effects after Completion; bounded verification yields Result/DeliveryFailure. Compatible deployments preserve obligations, debt, Results, frozen assignments and explicit -`execution_ref` pins. Unavailable/incompatible workers pause affected Work only. +`execution_ref` pins. Unavailable workers pause affected Work only. Scheduled +maintenance terminally cancels unclaimed, unpinned Work whose admission contract +differs from its approved active worker contract, with reason +`worker_contract_obsolete`. Work with delivery barriers is preserved. Quiesce economic Policy changes. Reject unsupported protocols/scalar assignments. Retries/checkpoints preserve submission identity; never delete queues to retry admission. diff --git a/actions/setup/js/reconcile_work_queue_run.cjs b/actions/setup/js/reconcile_work_queue_run.cjs index 4c6dec29919..5c3afc2ae6b 100644 --- a/actions/setup/js/reconcile_work_queue_run.cjs +++ b/actions/setup/js/reconcile_work_queue_run.cjs @@ -12,6 +12,7 @@ const { lifecycleEvidence, releaseAssignment, reconcileQueue } = require("./work const { isStagedMode } = require("./safe_output_helpers.cjs"); const { readQueueMetadata } = require("./work_queue_read_retry.cjs"); const { ERR_WORK_QUEUE_RECONCILIATION } = require("./error_codes.cjs"); +const { reconcileObsoleteWork } = require("./work_queue_deployment_control.cjs"); const reconciliationError = message => { const error = new Error(`${ERR_VALIDATION}: ${message}`); @@ -113,6 +114,8 @@ async function main(options = {}) { async function maintenance(options = {}) { const configured = { ...options, context: options.context || global.context, githubClient: options.githubClient || global.github, core: options.core || global.core }; + const obsolete = await reconcileObsoleteWork(configured); + log.diagnostic("maintenance.obsolete.complete", { cancelled: obsolete.cancelled }); const result = await reconcileQueue({ ...configured, requestCancellation: false, expireUnverified: true }); const released = result.results.filter(entry => entry.released).length; const unresolved = result.results.filter(entry => ["launch_unresolved", "run_binding_conflict"].includes(entry.state) || Object.values(entry.delivery?.claims || {}).some(claim => claim.reason === "verification_unresolved")).length; @@ -135,7 +138,7 @@ async function maintenance(options = {}) { log.diagnostic("maintenance.projection.complete", { pending: pending.length }); if (unresolved) throw reconciliationError("work_queue_maintenance_reconciliation_unresolved"); if (pending.length && !options.deferIssueWarning) configured.core?.warning("Work queue transitions are committed; Issue synchronization remains pending. Recover the exact terminal requests with work-queue sync-issues --request-id."); - return { ...result, issue_sync_pending: pending }; + return { ...result, obsolete_work_cancelled: obsolete.cancelled, issue_sync_pending: pending }; } module.exports = { main, maintenance }; diff --git a/actions/setup/js/reconcile_work_queue_run.test.cjs b/actions/setup/js/reconcile_work_queue_run.test.cjs index a33b96efe72..a6121dcfe60 100644 --- a/actions/setup/js/reconcile_work_queue_run.test.cjs +++ b/actions/setup/js/reconcile_work_queue_run.test.cjs @@ -286,6 +286,46 @@ describe("inline trusted worker termination", () => { }); describe("scheduled work queue recovery", () => { + it("retires obsolete backlog through the scheduled maintenance entry point without dispatching", async () => { + const contract = "a".repeat(64); + const { fixture, options } = setup({ + bound: false, + granted: false, + count: 1, + configurePolicy: policy => { + policy.authorization = "aw"; + policy.producers = {}; + const profile = policy.pools.default.profiles.default; + delete profile.principal; + profile.logical_contract = contract; + }, + }); + const worker = fixture.state.deployments.get("default").get("default"); + fixture.append("deployment", { + operations: [ + { + kind: "Deployment", + pool: "default", + worker_profile: "default", + expected_ref: worker.current_ref, + expected_contract: contract, + profile: { ...worker.revisions[REF].profile, ref: "b".repeat(40), logical_contract: "b".repeat(64) }, + available: true, + reason: "compiler_deployment", + }, + ], + }); + const publisher = fixture.nativeRun(fixture.dispatcherContext); + options.context = { ...fixture.dispatcherContext, job: "work_queue_compaction" }; + options.workflowRef = `${REPOSITORY}/${publisher.path}@${REF}`; + options.githubClient.rest.actions.getWorkflowRun.mockResolvedValue({ status: 200, data: publisher }); + expect(await maintenance(options)).toMatchObject({ obsolete_work_cancelled: 1 }); + expect([...fixture.state.works.values()][0]).toMatchObject({ state: "cancelled", cancellation_reason: "worker_contract_obsolete" }); + expect(fixture.state.dispatches.size).toBe(0); + expect(fixture.state.claims.size).toBe(0); + expect(await maintenance(options)).toMatchObject({ obsolete_work_cancelled: 0 }); + }); + it.each([ { pending: false, deferIssueWarning: false }, { pending: true, deferIssueWarning: false }, diff --git a/actions/setup/js/work_queue_deployment.cjs b/actions/setup/js/work_queue_deployment.cjs index ed80aae25ab..03bc46a63cd 100644 --- a/actions/setup/js/work_queue_deployment.cjs +++ b/actions/setup/js/work_queue_deployment.cjs @@ -74,6 +74,10 @@ function executionProfile(state, work) { return { profile: revision.profile, reason: "ready" }; } +function obsoleteWork(state, work) { + return work?.state === "available" && !work.execution_ref && !work.claim_id && work.barrier === "none" && executionProfile(state, work).reason === "worker_incompatible"; +} + function futurePolicy(state) { const policy = structuredClone(state.policy); for (const [pool, workers] of state.deployments ?? []) for (const [name, worker] of workers) policy.pools[pool].profiles[name] = structuredClone(worker.revisions[worker.current_ref].profile); @@ -102,4 +106,4 @@ function restoreDeployments(state, value) { } } -module.exports = { initializeDeployments, validateDeployment, applyDeployment, admissionProfile, executionProfile, futurePolicy, serializeDeployments, restoreDeployments }; +module.exports = { initializeDeployments, validateDeployment, applyDeployment, admissionProfile, executionProfile, obsoleteWork, futurePolicy, serializeDeployments, restoreDeployments }; diff --git a/actions/setup/js/work_queue_deployment_authority.test.cjs b/actions/setup/js/work_queue_deployment_authority.test.cjs index f804ac0363f..18dd248d4d7 100644 --- a/actions/setup/js/work_queue_deployment_authority.test.cjs +++ b/actions/setup/js/work_queue_deployment_authority.test.cjs @@ -1,10 +1,11 @@ // @ts-check import { describe, expect, it, vi } from "vitest"; import { canonical } from "./work_queue_codec.cjs"; -import { synchronizeDeployments } from "./work_queue_deployment_control.cjs"; -import { executionProfile, futurePolicy } from "./work_queue_deployment.cjs"; +import { synchronizeDeployments, reconcileObsoleteWork } from "./work_queue_deployment_control.cjs"; +import { executionProfile, obsoleteWork, futurePolicy } from "./work_queue_deployment.cjs"; import { normalizeSubmitParameters } from "./work_queue_intents.cjs"; -import { queueFixture, REF, WORKFLOW, REPOSITORY } from "./work_queue_lifecycle.test_helpers.cjs"; +import { nodeId } from "./work_queue_graph.cjs"; +import { queueFixture, REF, WORKFLOW, REPOSITORY, DISPATCHER } from "./work_queue_lifecycle.test_helpers.cjs"; const DEFAULT_REF = "b".repeat(40); const BRANCH_REF = "e".repeat(40); @@ -13,7 +14,7 @@ const NEXT_CONTRACT = "b".repeat(64); const lock = (contract = CONTRACT) => `# gh-aw-manifest: ${JSON.stringify({ work_queue: { worker: true, logical_contract: contract } })}\non:\n workflow_dispatch:\n inputs:\n work_queue_assignment:\n type: string\nenv:\n GH_AW_WORK_QUEUE_CONTRACT: "${contract}"\n`; -function fixture({ otherPool = false } = {}) { +function fixture({ otherPool = false, operationLimit = 256 } = {}) { const f = queueFixture({ count: 3, batch: 1, @@ -21,6 +22,7 @@ function fixture({ otherPool = false } = {}) { configurePolicy: policy => { policy.authorization = "aw"; policy.producers = {}; + policy.limits.operations = operationLimit; const profile = policy.pools.default.profiles.default; delete profile.principal; profile.logical_contract = CONTRACT; @@ -32,7 +34,7 @@ function fixture({ otherPool = false } = {}) { if (index === 2) work.worker_profile = "other"; }, }); - f.githubClient.rest.repos.get = vi.fn(async () => ({ data: { full_name: REPOSITORY, default_branch: "release/main" } })); + f.githubClient.rest.repos.get = vi.fn(async () => ({ status: 200, data: { full_name: REPOSITORY, id: 7, default_branch: "release/main" } })); f.githubClient.rest.git = { getRef: vi.fn(async () => ({ data: { ref: "refs/heads/release/main", object: { sha: DEFAULT_REF } } })) }; f.githubClient.rest.repos.getContent = vi.fn(async ({ path, ref }) => { expect([DEFAULT_REF, REF]).toContain(ref); @@ -71,6 +73,130 @@ describe("verified default worker deployment authority", () => { expect(canonical(f.transactions)).toBe(before); }); + describe("obsolete approved worker backlog recovery", () => { + async function outdated() { + const { f, options, trusted } = fixture(); + options.context = f.dispatcherContext; + options.workflowRef = `${REPOSITORY}/${DISPATCHER}@${REF}`; + f.githubClient.rest.repos.getContent = vi.fn(async ({ path, ref }) => ({ + data: { type: "file", path, encoding: "base64", content: Buffer.from(lock(ref === DEFAULT_REF ? NEXT_CONTRACT : CONTRACT)).toString("base64") }, + })); + await synchronizeDeployments(options, trusted); + return { f, options }; + } + + it("terminally cancels obsolete unpinned Work once, retaining pins, compatible Work, policy, debt and history", async () => { + const { f, options } = await outdated(); + const before = f.state; + const [old, pinned, other] = [...before.works.values()]; + expect(await reconcileObsoleteWork(options)).toEqual({ cancelled: 1 }); + const after = f.state; + expect(after.works.get(old.work_id)).toMatchObject({ state: "cancelled", cancellation_reason: "worker_contract_obsolete" }); + expect(after.works.get(pinned.work_id)).toEqual(pinned); + expect(after.works.get(other.work_id)).toEqual(other); + for (const key of ["policy", "policy_epoch", "claims", "dispatches", "clocks", "terminalBarriers"]) expect(after[key]).toEqual(before[key]); + expect(f.transactions.slice(0, before.transactions.length)).toEqual(before.transactions); + expect(f.transactions.at(-1).actor.role).toBe("reconciler"); + const history = canonical(f.transactions); + expect(await reconcileObsoleteWork(options)).toEqual({ cancelled: 0 }); + expect(canonical(f.transactions)).toBe(history); + }); + + it.each(["pending", "failed", "verified"])("preserves %s delivery and all owned or pinned Work", async barrier => { + const { f } = await outdated(); + const state = f.state; + const work = [...state.works.values()][0]; + expect(obsoleteWork(state, work)).toBe(true); + for (const changes of [{ barrier }, { state: "claimed" }, { claim_id: "owned" }, { execution_ref: REF }, { state: "completed" }]) { + expect(obsoleteWork(state, { ...work, ...changes })).toBe(false); + } + }); + + it("does not mistake unavailable deployments for obsolete Work or run in staged mode", async () => { + const { f, options } = await outdated(); + expect(await reconcileObsoleteWork({ ...options, staged: true })).toEqual({ cancelled: 0 }); + const worker = f.state.deployments.get("default").get("default"); + f.append("deployment", { + operations: [ + { + kind: "Deployment", + pool: "default", + worker_profile: "default", + expected_ref: worker.current_ref, + expected_contract: worker.current_contract, + profile: worker.revisions[worker.current_ref].profile, + available: false, + reason: "worker_unavailable", + }, + ], + }); + expect(await reconcileObsoleteWork(options)).toEqual({ cancelled: 0 }); + }); + + it("rejects an entire stale cancellation batch after a concurrent compatible route update", async () => { + const { f, options } = await outdated(); + options.publishWorkQueueRequest = async args => { + const worker = f.state.deployments.get("default").get("default"); + f.append("deployment", { + operations: [ + { + kind: "Deployment", + pool: "default", + worker_profile: "default", + expected_ref: worker.current_ref, + expected_contract: worker.current_contract, + profile: worker.revisions[REF].profile, + available: true, + reason: "compiler_deployment", + }, + ], + }); + return f.publishWorkQueueRequest(args); + }; + await expect(reconcileObsoleteWork(options)).rejects.toMatchObject({ code: "deployment_conflict" }); + expect([...f.state.works.values()].every(work => work.state === "available")).toBe(true); + }); + + it("fails closed on authentication or publication errors", async () => { + const { f, options } = await outdated(); + const failure = new Error("denied"); + options.publishWorkQueueRequest = vi.fn().mockRejectedValue(failure); + await expect(reconcileObsoleteWork(options)).rejects.toBe(failure); + expect([...f.state.works.values()].every(work => work.state === "available")).toBe(true); + options.publishWorkQueueRequest = vi.fn(); + f.githubClient.rest.actions.getWorkflowRun = vi.fn().mockRejectedValue(failure); + await expect(reconcileObsoleteWork(options)).rejects.toBe(failure); + expect(options.publishWorkQueueRequest).not.toHaveBeenCalled(); + }); + + it("retires a bounded deterministic prefix and leaves the rest for the next maintenance run", async () => { + const { f, options, trusted } = fixture({ operationLimit: 3 }); + const original = f.transactions[1].operations[0]; + f.append( + "submit", + { + nodes: Array.from({ length: 3 }, (_, index) => ({ ...original, graph_id: "g2", node_key: `n${index}`, work_id: nodeId("g2", `n${index}`) })), + }, + { role: "producer", principal: "11", repository: REPOSITORY } + ); + options.context = f.dispatcherContext; + options.workflowRef = `${REPOSITORY}/${DISPATCHER}@${REF}`; + f.githubClient.rest.repos.getContent = vi.fn(async ({ path, ref }) => ({ + data: { type: "file", path, encoding: "base64", content: Buffer.from(lock(ref === DEFAULT_REF ? NEXT_CONTRACT : CONTRACT)).toString("base64") }, + })); + await synchronizeDeployments(options, trusted); + const eligible = [...f.state.works.values()] + .filter(work => obsoleteWork(f.state, work)) + .map(work => work.work_id) + .sort(); + expect(eligible).toHaveLength(4); + expect(await reconcileObsoleteWork(options)).toEqual({ cancelled: 3 }); + expect(f.transactions.at(-1).operations.map(operation => operation.work_id)).toEqual(eligible.slice(0, 3)); + expect(await reconcileObsoleteWork(options)).toEqual({ cancelled: 1 }); + expect(await reconcileObsoleteWork(options)).toEqual({ cancelled: 0 }); + }); + }); + it.each([ ["identical branch contract", {}], ["altered branch authority", { workflow: ".github/workflows/branch-worker.lock.yml", max_claims: 16, logical_contract: NEXT_CONTRACT }], diff --git a/actions/setup/js/work_queue_deployment_control.cjs b/actions/setup/js/work_queue_deployment_control.cjs index 21fc7996b43..7f059d1f66b 100644 --- a/actions/setup/js/work_queue_deployment_control.cjs +++ b/actions/setup/js/work_queue_deployment_control.cjs @@ -4,6 +4,26 @@ const { canonical, digest } = require("./work_queue_codec.cjs"); const { policyProposalFor, loadQueue, publishOperations } = require("./work_queue_binding.cjs"); const { verifiedDefaultReference, configuredWorkerDeployment } = require("./work_queue_provisioning.cjs"); +const { obsoleteWork } = require("./work_queue_deployment.cjs"); +const { authenticatePublisher } = require("./work_queue_native.cjs"); +const { isStagedMode } = require("./safe_output_helpers.cjs"); + +async function reconcileObsoleteWork(options) { + if (isStagedMode(options) || isStagedMode(options.config)) return { cancelled: 0 }; + const latest = await loadQueue({ ...options, policyProposal: undefined }); + if (latest.projection.policy?.authorization !== "aw") return { cancelled: 0 }; + const operations = [...latest.projection.works.values()] + .filter(work => obsoleteWork(latest.projection, work)) + .sort((a, b) => a.work_id.localeCompare(b.work_id)) + .slice(0, Math.min(256, latest.projection.policy.limits.operations)) + .map(work => ({ kind: "WorkCancellation", work_id: work.work_id, reason: "worker_contract_obsolete" })); + if (!operations.length) return { cancelled: 0 }; + const context = await authenticatePublisher({ ...options, role: "reconciler" }); + await publishOperations(options, context, ["obsolete-work", latest.projection.tip, operations], "cancel_work", operations); + const final = await loadQueue({ ...options, policyProposal: undefined }); + if (operations.some(operation => final.projection.works.get(operation.work_id)?.state !== "cancelled")) throw new Error("work_queue_obsolete_cancellation_not_durable"); + return { cancelled: operations.length }; +} function approvedWorkerProfiles(state, pool, config) { const workflows = new Set(config.work_queue_workflows ?? []); @@ -77,4 +97,4 @@ async function synchronizeDeployments(options, context, pools = undefined) { } } -module.exports = { approvedWorkerProfiles, synchronizeDeployments }; +module.exports = { approvedWorkerProfiles, synchronizeDeployments, reconcileObsoleteWork }; diff --git a/actions/setup/js/work_queue_replay.cjs b/actions/setup/js/work_queue_replay.cjs index 1fd77f7d3d9..197ba94ccb1 100644 --- a/actions/setup/js/work_queue_replay.cjs +++ b/actions/setup/js/work_queue_replay.cjs @@ -11,7 +11,7 @@ const { indexesFor, trackState, untrackState, updateIndexes } = require("./work_ const { validateEffectResourceAuthority } = require("./work_queue_resource_scope.cjs"); const { applyIssueBinding } = require("./work_queue_issue_contract.cjs"); const { advanceExpiration, expirationView } = require("./work_queue_expiration.cjs"); -const { initializeDeployments, validateDeployment, applyDeployment, admissionProfile, serializeDeployments, restoreDeployments } = require("./work_queue_deployment.cjs"); +const { initializeDeployments, validateDeployment, applyDeployment, admissionProfile, obsoleteWork, serializeDeployments, restoreDeployments } = require("./work_queue_deployment.cjs"); const { actorFromContext, bindingAuthority, @@ -569,6 +569,8 @@ function applyLifecycle(state, operation, commit) { } case "WorkCancellation": if (!work || work.state === "completed" || work.barrier === "pending") throw queueError("work_terminal", "completed ownership cannot be reopened or cancelled"); + if (operation.reason === "worker_contract_obsolete" && work.state !== "cancelled" && !obsoleteWork(state, work)) + throw queueError("deployment_conflict", "obsolete cancellation requires available unpinned Work with no delivery barrier and an incompatible active worker contract"); if (operation.claim_id) { const selected = state.claims.get(operation.claim_id); if ( diff --git a/docs/src/content/docs/reference/work-queue.md b/docs/src/content/docs/reference/work-queue.md index 2108c66927f..9a18afb3dff 100644 --- a/docs/src/content/docs/reference/work-queue.md +++ b/docs/src/content/docs/reference/work-queue.md @@ -118,6 +118,8 @@ The immutable Work payload declares its output contract in `effect_contract`. A Worker conclusion performs inline reservation cleanup. Scheduled `agentics-maintenance.yml` reconciliation handles stranded reservations and pending delivery before compaction. Maintenance never cancels active runs or manufactures Result; disabling it leaves dispatcher/operator reconciliation as the fallback. +Maintenance also retires obsolete backlog in checked batches of at most 256 Work (further bounded by the installed operation limit). Available, unpinned Work with no delivery barrier is terminally cancelled with reason `worker_contract_obsolete` when its admission contract differs from its approved active worker contract. Missing or unavailable deployments are not proof of obsolescence. Exact `execution_ref` pins, owned Work, delivery barriers, Results, ledger history and fairness debt are preserved. Cancellation rechecks eligibility against the publication's current ledger; a concurrent Claim or compatible route change rejects the transaction. Operators can use `cancel-work --reason worker_contract_obsolete --work-id ID` for the same checked recovery without dispatching workers. + ### Claim-scoped memory `tools.work-queue.memory` generates a structured output tool and protected Git-tree publication adapter for a worker. It is not standalone `tools.repo-memory`. Configuration fields are `name` (default `persist_work_queue_memory`), `path`, `target-repo`, `base-revision`, `branch-prefix`, `schema`, and `max-bytes` (default and maximum `262144`, including the final newline). diff --git a/pkg/workqueue/deployment.go b/pkg/workqueue/deployment.go index 35aed776842..dabded8ee24 100644 --- a/pkg/workqueue/deployment.go +++ b/pkg/workqueue/deployment.go @@ -136,6 +136,14 @@ func (state Projection) executionProfile(work *WorkState) (WorkerProfile, string return revision.Profile, "ready" } +func (state Projection) obsoleteWork(work *WorkState) bool { + if work == nil || work.State != "available" || work.ExecutionRef != "" || work.ClaimID != "" || work.Barrier != "none" { + return false + } + _, reason := state.executionProfile(work) + return reason == "worker_incompatible" +} + // FuturePolicy exposes current routes for trusted admission builders only. // It does not replace the installed policy or affect existing assignments. func FuturePolicy(state Projection) Policy { diff --git a/pkg/workqueue/deployment_test.go b/pkg/workqueue/deployment_test.go index d2660ba586e..5129b71dbe2 100644 --- a/pkg/workqueue/deployment_test.go +++ b/pkg/workqueue/deployment_test.go @@ -65,6 +65,69 @@ func deployTestWorker(t *testing.T, commits []QueueCommit, id, name, ref, contra })) } +func TestObsoleteWorkerCancellationReplayParity(t *testing.T) { + commits := testGenesis(t, deploymentPolicy) + installed, err := Replay(commits) + if err != nil { + t.Fatal(err) + } + old := testNode(t, commits, "obsolete") + pinned := testNode(t, commits, "pinned") + pinned.ExecutionRef = installed.Policy.Pools["default"].Profiles["default"].Ref + commits = testSubmit(t, commits, "backlog", old, pinned) + commits = deployTestWorker(t, commits, "new-contract", "default", strings.Repeat("b", 40), strings.Repeat("b", 64), true) + before, err := Replay(commits) + if err != nil { + t.Fatal(err) + } + if !before.obsoleteWork(before.Works[old.WorkID]) || before.obsoleteWork(before.Works[pinned.WorkID]) { + t.Fatal("obsolete selection did not preserve the immutable pin") + } + operation := mustOp(t, map[string]any{"kind": "WorkCancellation", "work_id": old.WorkID, "reason": "worker_contract_obsolete"}) + commits = testOperations(t, commits, testActor("reconciler"), "retire-obsolete", "cancel_work", operation) + after, err := Replay(commits) + if err != nil || after.Works[old.WorkID].State != "cancelled" || !sameJSON(before.Works[pinned.WorkID], after.Works[pinned.WorkID]) || + !sameJSON(before.Policy, after.Policy) || !sameJSON(before.Clocks, after.Clocks) || !sameJSON(before.Claims, after.Claims) { + t.Fatalf("obsolete cancellation changed protected state: %v", err) + } + input, err := canonicalValue(commits) + if err != nil { + t.Fatal(err) + } + command := exec.Command("node", "-e", ` +const assert = require("node:assert/strict"); +const r = require("../../actions/setup/js/work_queue_replay.cjs"); +const s = r.replayTransactions(JSON.parse(require("node:fs").readFileSync(0, "utf8"))); +assert.equal([...s.works.values()].filter(w => w.state === "cancelled").length, 1); +assert.equal([...s.works.values()].filter(w => w.execution_ref && w.state === "available").length, 1); +`) + command.Stdin = bytes.NewReader(input) + if output, err := command.CombinedOutput(); err != nil { + t.Fatalf("obsolete cancellation Go/JS replay parity: %v\n%s", err, output) + } + for _, changes := range []func(*WorkState){ + func(w *WorkState) { w.State = "claimed" }, + func(w *WorkState) { w.ClaimID = "owned" }, + func(w *WorkState) { w.Barrier = "pending" }, + func(w *WorkState) { w.Barrier = "failed" }, + func(w *WorkState) { w.Barrier = "verified" }, + } { + work := *before.Works[old.WorkID] + changes(&work) + if before.obsoleteWork(&work) { + t.Fatal("owned or delivery-barrier Work was selected") + } + } + compatible := deployTestWorker(t, commits[:len(commits)-1], "compatible-again", "default", pinned.ExecutionRef, strings.Repeat("a", 64), true) + request, err := NewRequest("stale-retirement", "cancel_work", testActor("administrator"), OperationsParameters{Operations: []Operation{operation}}) + if err != nil { + t.Fatal(err) + } + if _, _, _, err := BuildCandidate(compatible, testActor("administrator"), request, 9000); err == nil { + t.Fatal("stale cancellation accepted a now-compatible contract") + } +} + func TestDeploymentCompatibleEvolutionPreservesFrozenRecoveryAndPins(t *testing.T) { commits := testGenesis(t, deploymentPolicy) root := testNode(t, commits, "root") diff --git a/pkg/workqueue/replay.go b/pkg/workqueue/replay.go index 8295ad4a9b2..335a2e312df 100644 --- a/pkg/workqueue/replay.go +++ b/pkg/workqueue/replay.go @@ -319,10 +319,17 @@ func (state *Projection) replayCancellation(operation Operation, kind string, co return nil } - var value WorkDefinition + var value struct { + WorkID string `json:"work_id"` + Reason string `json:"reason"` + } if err := json.Unmarshal(operation, &value); err != nil { return err } + work := state.Works[value.WorkID] + if value.Reason == "worker_contract_obsolete" && work != nil && work.State != "cancelled" && !state.obsoleteWork(work) { + return queueError("deployment_conflict", "obsolete cancellation requires available unpinned Work with no delivery barrier and an incompatible active worker contract") + } if err := state.cancelWork(operation, commit); err != nil { return err } From b5252c58b7f50efbf56ee9c36f3059983e02eeb3 Mon Sep 17 00:00:00 2001 From: pelikhan Date: Sun, 8 Nov 2026 20:28:51 -0800 Subject: [PATCH 2/5] fix(work-queue): isolate obsolete retirement failures Preserve core recovery after sanitized obsolete-sweep failures and use the shared deterministic comparator for bounded retirement. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- actions/setup/js/reconcile_work_queue_run.cjs | 16 +++- .../js/reconcile_work_queue_run.test.cjs | 77 ++++++++++++++++++- .../work_queue_deployment_authority.test.cjs | 21 ++++- .../js/work_queue_deployment_control.cjs | 4 +- docs/src/content/docs/reference/work-queue.md | 2 +- 5 files changed, 110 insertions(+), 10 deletions(-) diff --git a/actions/setup/js/reconcile_work_queue_run.cjs b/actions/setup/js/reconcile_work_queue_run.cjs index 5c3afc2ae6b..573a1503617 100644 --- a/actions/setup/js/reconcile_work_queue_run.cjs +++ b/actions/setup/js/reconcile_work_queue_run.cjs @@ -12,7 +12,7 @@ const { lifecycleEvidence, releaseAssignment, reconcileQueue } = require("./work const { isStagedMode } = require("./safe_output_helpers.cjs"); const { readQueueMetadata } = require("./work_queue_read_retry.cjs"); const { ERR_WORK_QUEUE_RECONCILIATION } = require("./error_codes.cjs"); -const { reconcileObsoleteWork } = require("./work_queue_deployment_control.cjs"); +const deploymentControl = require("./work_queue_deployment_control.cjs"); const reconciliationError = message => { const error = new Error(`${ERR_VALIDATION}: ${message}`); @@ -114,8 +114,16 @@ async function main(options = {}) { async function maintenance(options = {}) { const configured = { ...options, context: options.context || global.context, githubClient: options.githubClient || global.github, core: options.core || global.core }; - const obsolete = await reconcileObsoleteWork(configured); - log.diagnostic("maintenance.obsolete.complete", { cancelled: obsolete.cancelled }); + let obsolete; + let obsoleteFailed = false; + try { + obsolete = await deploymentControl.reconcileObsoleteWork(configured); + log.diagnostic("maintenance.obsolete.complete", { cancelled: obsolete.cancelled }); + } catch (error) { + obsoleteFailed = true; + log.failure("maintenance.obsolete.failed", error, true); + configured.core?.warning("Work queue obsolete-work reconciliation failed; core recovery will continue. Inspect safe API diagnostics and the checked ledger before retrying maintenance."); + } const result = await reconcileQueue({ ...configured, requestCancellation: false, expireUnverified: true }); const released = result.results.filter(entry => entry.released).length; const unresolved = result.results.filter(entry => ["launch_unresolved", "run_binding_conflict"].includes(entry.state) || Object.values(entry.delivery?.claims || {}).some(claim => claim.reason === "verification_unresolved")).length; @@ -138,7 +146,7 @@ async function maintenance(options = {}) { log.diagnostic("maintenance.projection.complete", { pending: pending.length }); if (unresolved) throw reconciliationError("work_queue_maintenance_reconciliation_unresolved"); if (pending.length && !options.deferIssueWarning) configured.core?.warning("Work queue transitions are committed; Issue synchronization remains pending. Recover the exact terminal requests with work-queue sync-issues --request-id."); - return { ...result, obsolete_work_cancelled: obsolete.cancelled, issue_sync_pending: pending }; + return { ...result, obsolete_work_cancelled: obsolete?.cancelled ?? null, obsolete_work_reconciliation_failed: obsoleteFailed, issue_sync_pending: pending }; } module.exports = { main, maintenance }; diff --git a/actions/setup/js/reconcile_work_queue_run.test.cjs b/actions/setup/js/reconcile_work_queue_run.test.cjs index a6121dcfe60..bbcdcc28a0a 100644 --- a/actions/setup/js/reconcile_work_queue_run.test.cjs +++ b/actions/setup/js/reconcile_work_queue_run.test.cjs @@ -319,13 +319,88 @@ describe("scheduled work queue recovery", () => { options.context = { ...fixture.dispatcherContext, job: "work_queue_compaction" }; options.workflowRef = `${REPOSITORY}/${publisher.path}@${REF}`; options.githubClient.rest.actions.getWorkflowRun.mockResolvedValue({ status: 200, data: publisher }); - expect(await maintenance(options)).toMatchObject({ obsolete_work_cancelled: 1 }); + expect(await maintenance(options)).toMatchObject({ obsolete_work_cancelled: 1, obsolete_work_reconciliation_failed: false }); expect([...fixture.state.works.values()][0]).toMatchObject({ state: "cancelled", cancellation_reason: "worker_contract_obsolete" }); expect(fixture.state.dispatches.size).toBe(0); expect(fixture.state.claims.size).toBe(0); expect(await maintenance(options)).toMatchObject({ obsolete_work_cancelled: 0 }); }); + it.each([ + ["deployment race", { code: "deployment_conflict" }], + ["authentication", { status: 401 }], + ["read", { status: 503 }], + ["publication", { status: 403 }], + ["durability", { code: "work_queue_obsolete_cancellation_not_durable" }], + ])("recovers a held reservation and expired pending delivery despite an obsolete sweep %s failure", async (_kind, failure) => { + const { fixture, options, nativeRun } = setup({ count: 1 }); + const member = fixture.assignment.claims[0]; + fixture.append("finish", { dispatch_id: fixture.assignment.dispatch_id, claim_handle: member.handle, outcome: "completed" }, { ...fixture.workerActor, dispatch_id: fixture.assignment.dispatch_id, claim_handle: member.handle }); + expect(fixture.state.dispatches.get(fixture.assignment.dispatch_id).released).toBe(false); + expect(fixture.state.works.get(member.work_id).barrier).toBe("pending"); + options.context = { ...fixture.dispatcherContext, job: "work_queue_compaction" }; + const publisher = fixture.nativeRun(fixture.dispatcherContext); + options.workflowRef = `${REPOSITORY}/${publisher.path}@${REF}`; + options.githubClient.rest.actions.getWorkflowRun.mockResolvedValue({ status: 200, data: publisher }); + options.now = fixture.at + fixture.policy.pools.default.reconciliation.deadline_ms; + options.publishWorkQueueRequest = vi.fn(async args => fixture.append(args.request.kind, args.request.parameters, args.actor, args.request.id, options.now)); + nativeRun.status = "completed"; + nativeRun.conclusion = "failure"; + const canary = "private-obsolete-provider-message-token-and-stack"; + const error = Object.assign(new Error(canary, { cause: new Error(canary) }), failure, { response: { data: { message: canary }, headers: { authorization: canary } } }); + const sweep = vi.spyOn(require("./work_queue_deployment_control.cjs"), "reconcileObsoleteWork").mockRejectedValue(error); + const stderr = vi.spyOn(process.stderr, "write").mockReturnValue(true); + try { + const result = await maintenance(options); + expect(sweep).toHaveBeenCalledOnce(); + expect(result).toMatchObject({ + obsolete_work_cancelled: null, + obsolete_work_reconciliation_failed: true, + reconciled: 1, + results: [{ released: true, delivery: { claims: { h1: { state: "delivery_failed", effects: "unknown" } } } }], + }); + expect(fixture.state.dispatches.get(fixture.assignment.dispatch_id).released).toBe(true); + expect(fixture.state.works.get(member.work_id)).toMatchObject({ barrier: "failed", disposition: "unknown" }); + const operations = fixture.transactions.flatMap(commit => commit.operations); + expect(operations.filter(operation => operation.kind === "Release")).toHaveLength(1); + expect(operations.filter(operation => operation.kind === "DeliveryFailure")).toHaveLength(1); + expect(operations.some(operation => operation.kind === "Result")).toBe(false); + expect(options.core.warning).toHaveBeenCalledExactlyOnceWith(expect.stringContaining("obsolete-work reconciliation failed; core recovery will continue")); + const diagnostics = stderr.mock.calls.map(call => call[0]).join(""); + expect(diagnostics).toContain("maintenance.obsolete.failed failed=true"); + expect(diagnostics).not.toContain(canary); + expect(JSON.stringify(result)).not.toContain(canary); + expect(JSON.stringify(options.core.warning.mock.calls)).not.toContain(canary); + } finally { + sweep.mockRestore(); + stderr.mockRestore(); + } + }); + + it("keeps independent delivery verification failures fatal after an obsolete sweep failure", async () => { + const { fixture, options, nativeRun } = setup({ count: 1 }); + const member = fixture.assignment.claims[0]; + fixture.append("finish", { dispatch_id: fixture.assignment.dispatch_id, claim_handle: member.handle, outcome: "completed" }, { ...fixture.workerActor, dispatch_id: fixture.assignment.dispatch_id, claim_handle: member.handle }); + options.context = { ...fixture.dispatcherContext, job: "work_queue_compaction" }; + const publisher = fixture.nativeRun(fixture.dispatcherContext); + options.workflowRef = `${REPOSITORY}/${publisher.path}@${REF}`; + options.githubClient.rest.actions.getWorkflowRun.mockResolvedValue({ status: 200, data: publisher }); + nativeRun.status = "completed"; + nativeRun.conclusion = "failure"; + const sweep = vi.spyOn(require("./work_queue_deployment_control.cjs"), "reconcileObsoleteWork").mockRejectedValue(new Error("private-sweep-error")); + const finalize = vi.spyOn(require("./finish_work_queue_claim.cjs"), "finalizeWorkerResults").mockRejectedValue(new Error("private-delivery-error")); + try { + await expect(maintenance(options)).rejects.toThrow("work_queue_maintenance_reconciliation_unresolved"); + expect(finalize).toHaveBeenCalledOnce(); + expect(fixture.state.dispatches.get(fixture.assignment.dispatch_id).released).toBe(true); + expect(fixture.state.works.get(member.work_id).barrier).toBe("pending"); + expect(options.core.warning).toHaveBeenCalledExactlyOnceWith(expect.stringContaining("obsolete-work reconciliation failed")); + } finally { + sweep.mockRestore(); + finalize.mockRestore(); + } + }); + it.each([ { pending: false, deferIssueWarning: false }, { pending: true, deferIssueWarning: false }, diff --git a/actions/setup/js/work_queue_deployment_authority.test.cjs b/actions/setup/js/work_queue_deployment_authority.test.cjs index 18dd248d4d7..bc9aab41326 100644 --- a/actions/setup/js/work_queue_deployment_authority.test.cjs +++ b/actions/setup/js/work_queue_deployment_authority.test.cjs @@ -1,6 +1,6 @@ // @ts-check import { describe, expect, it, vi } from "vitest"; -import { canonical } from "./work_queue_codec.cjs"; +import { canonical, utf8Compare } from "./work_queue_codec.cjs"; import { synchronizeDeployments, reconcileObsoleteWork } from "./work_queue_deployment_control.cjs"; import { executionProfile, obsoleteWork, futurePolicy } from "./work_queue_deployment.cjs"; import { normalizeSubmitParameters } from "./work_queue_intents.cjs"; @@ -188,13 +188,30 @@ describe("verified default worker deployment authority", () => { const eligible = [...f.state.works.values()] .filter(work => obsoleteWork(f.state, work)) .map(work => work.work_id) - .sort(); + .sort(utf8Compare); expect(eligible).toHaveLength(4); expect(await reconcileObsoleteWork(options)).toEqual({ cancelled: 3 }); expect(f.transactions.at(-1).operations.map(operation => operation.work_id)).toEqual(eligible.slice(0, 3)); expect(await reconcileObsoleteWork(options)).toEqual({ cancelled: 1 }); expect(await reconcileObsoleteWork(options)).toEqual({ cancelled: 0 }); }); + + it("orders a controlled read projection independently of locale before selecting a bounded prefix", async () => { + const { f, options } = await outdated(); + const projection = structuredClone(f.state); + const work = [...projection.works.values()].find(candidate => obsoleteWork(projection, candidate)); + projection.policy.limits.operations = 3; + // Synthetic IDs distinguish collation order without entering the replayed ledger. + projection.works = new Map(["work_a", "work-a", "Work-B", "work-B"].map(work_id => [work_id, { ...work, work_id }])); + options.readWorkQueueLog = async () => ({ ...(await f.readWorkQueueLog()), state: projection }); + const failure = new Error("stop before ledger publication"); + options.publishWorkQueueRequest = vi.fn().mockRejectedValue(failure); + const history = canonical(f.transactions); + await expect(reconcileObsoleteWork(options)).rejects.toBe(failure); + expect(options.publishWorkQueueRequest).toHaveBeenCalledOnce(); + expect(options.publishWorkQueueRequest.mock.calls[0][0].request.parameters.operations.map(operation => operation.work_id)).toEqual(["Work-B", "work-B", "work-a"]); + expect(canonical(f.transactions)).toBe(history); + }); }); it.each([ diff --git a/actions/setup/js/work_queue_deployment_control.cjs b/actions/setup/js/work_queue_deployment_control.cjs index 7f059d1f66b..b08616a7987 100644 --- a/actions/setup/js/work_queue_deployment_control.cjs +++ b/actions/setup/js/work_queue_deployment_control.cjs @@ -1,7 +1,7 @@ // @ts-check "use strict"; -const { canonical, digest } = require("./work_queue_codec.cjs"); +const { canonical, digest, utf8Compare } = require("./work_queue_codec.cjs"); const { policyProposalFor, loadQueue, publishOperations } = require("./work_queue_binding.cjs"); const { verifiedDefaultReference, configuredWorkerDeployment } = require("./work_queue_provisioning.cjs"); const { obsoleteWork } = require("./work_queue_deployment.cjs"); @@ -14,7 +14,7 @@ async function reconcileObsoleteWork(options) { if (latest.projection.policy?.authorization !== "aw") return { cancelled: 0 }; const operations = [...latest.projection.works.values()] .filter(work => obsoleteWork(latest.projection, work)) - .sort((a, b) => a.work_id.localeCompare(b.work_id)) + .sort((a, b) => utf8Compare(a.work_id, b.work_id)) .slice(0, Math.min(256, latest.projection.policy.limits.operations)) .map(work => ({ kind: "WorkCancellation", work_id: work.work_id, reason: "worker_contract_obsolete" })); if (!operations.length) return { cancelled: 0 }; diff --git a/docs/src/content/docs/reference/work-queue.md b/docs/src/content/docs/reference/work-queue.md index 9a18afb3dff..4be693daecb 100644 --- a/docs/src/content/docs/reference/work-queue.md +++ b/docs/src/content/docs/reference/work-queue.md @@ -118,7 +118,7 @@ The immutable Work payload declares its output contract in `effect_contract`. A Worker conclusion performs inline reservation cleanup. Scheduled `agentics-maintenance.yml` reconciliation handles stranded reservations and pending delivery before compaction. Maintenance never cancels active runs or manufactures Result; disabling it leaves dispatcher/operator reconciliation as the fallback. -Maintenance also retires obsolete backlog in checked batches of at most 256 Work (further bounded by the installed operation limit). Available, unpinned Work with no delivery barrier is terminally cancelled with reason `worker_contract_obsolete` when its admission contract differs from its approved active worker contract. Missing or unavailable deployments are not proof of obsolescence. Exact `execution_ref` pins, owned Work, delivery barriers, Results, ledger history and fairness debt are preserved. Cancellation rechecks eligibility against the publication's current ledger; a concurrent Claim or compatible route change rejects the transaction. Operators can use `cancel-work --reason worker_contract_obsolete --work-id ID` for the same checked recovery without dispatching workers. +Maintenance also retires obsolete backlog in checked batches of at most 256 Work (further bounded by the installed operation limit). Available, unpinned Work with no delivery barrier is terminally cancelled with reason `worker_contract_obsolete` when its admission contract differs from its approved active worker contract. Missing or unavailable deployments are not proof of obsolescence. Exact `execution_ref` pins, owned Work, delivery barriers, Results, ledger history and fairness debt are preserved. Cancellation rechecks eligibility against the publication's current ledger; a concurrent Claim or compatible route change rejects the transaction. Retirement failures emit sanitized diagnostics and a warning without skipping reservation reconciliation, pending delivery, Issue synchronization or compaction. Operators can use `cancel-work --reason worker_contract_obsolete --work-id ID` for the same checked recovery without dispatching workers. ### Claim-scoped memory From bd1ac070a641184b6ba1cb2269daa883068f2528 Mon Sep 17 00:00:00 2001 From: pelikhan Date: Sun, 8 Nov 2026 21:03:10 -0800 Subject: [PATCH 3/5] fix(work-queue): preserve legacy cancellation replay Version new checked retirements without rewriting historical receipts, preserve retry and checkpoint identities, and exclude expired Work from retirement selection. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .github/aw/work-queue.md | 7 +- .../work_queue_cancellation_version.test.cjs | 132 ++++++++++++++ .../work_queue_deployment_authority.test.cjs | 26 +++ .../js/work_queue_deployment_control.cjs | 8 +- actions/setup/js/work_queue_replay.cjs | 9 +- docs/src/content/docs/reference/work-queue.md | 4 +- .../docs/specs/work-queue-specification.md | 19 +- .../work_command_cancellation_version_test.go | 127 ++++++++++++++ pkg/cli/work_command_clear.go | 1 + pkg/cli/work_command_operations.go | 37 +++- pkg/workqueue/cancellation_version_test.go | 164 ++++++++++++++++++ pkg/workqueue/deployment_test.go | 18 +- pkg/workqueue/inspection.go | 98 ++++++----- pkg/workqueue/replay.go | 11 +- .../schema/OperationsParameters.json | 3 + pkg/workqueue/schema/QueueCommit.json | 3 + pkg/workqueue/schema/QueueOperation.json | 3 + pkg/workqueue/schema/QueueRequest.json | 3 + pkg/workqueue/schema/RequestParameters.json | 3 + .../schema/WorkCancellationOperation.json | 3 + specs/work-queue/transactions.tsp | 2 + 21 files changed, 606 insertions(+), 75 deletions(-) create mode 100644 actions/setup/js/work_queue_cancellation_version.test.cjs create mode 100644 pkg/cli/work_command_cancellation_version_test.go create mode 100644 pkg/workqueue/cancellation_version_test.go diff --git a/.github/aw/work-queue.md b/.github/aw/work-queue.md index e0f1ddc0052..6e26a003106 100644 --- a/.github/aw/work-queue.md +++ b/.github/aw/work-queue.md @@ -116,8 +116,11 @@ Completion; bounded verification yields Result/DeliveryFailure. Compatible deployments preserve obligations, debt, Results, frozen assignments and explicit `execution_ref` pins. Unavailable workers pause affected Work only. Scheduled maintenance terminally cancels unclaimed, unpinned Work whose admission contract -differs from its approved active worker contract, with reason -`worker_contract_obsolete`. Work with delivery barriers is preserved. +differs from its approved active worker contract, using `WorkCancellation` with +operation `version: 1` and reason `worker_contract_obsolete`. Current-time +expiration excludes elapsed Work; publication rechecks eligibility. Work with +delivery barriers is preserved. Version-3 history without an operation version +retains reason-only semantics, including compatible/pinned cancellations. Quiesce economic Policy changes. Reject unsupported protocols/scalar assignments. Retries/checkpoints preserve submission identity; never delete queues to retry admission. diff --git a/actions/setup/js/work_queue_cancellation_version.test.cjs b/actions/setup/js/work_queue_cancellation_version.test.cjs new file mode 100644 index 00000000000..c2e1a81d744 --- /dev/null +++ b/actions/setup/js/work_queue_cancellation_version.test.cjs @@ -0,0 +1,132 @@ +// @ts-check +import { describe, expect, it } from "vitest"; +import { canonical } from "./work_queue_codec.cjs"; +import { obsoleteWork } from "./work_queue_deployment.cjs"; +import { compactTransactions, replayTransactions, validateOperation } from "./work_queue_replay.cjs"; +import { queueFixture, REF } from "./work_queue_lifecycle.test_helpers.cjs"; + +const CONTRACT = "a".repeat(64); +const REASON = "worker_contract_obsolete"; + +function fixture(granted = false, expiresAt = undefined) { + return queueFixture({ + count: 3, + granted, + configurePolicy: policy => { + policy.authorization = "aw"; + policy.producers = {}; + const profile = policy.pools.default.profiles.default; + delete profile.principal; + profile.logical_contract = CONTRACT; + policy.pools.default.profiles.other = { ...profile, workflow: ".github/workflows/other.lock.yml" }; + }, + configureWork: (work, index) => { + if (index === 0 && expiresAt !== undefined) work.expires_at = expiresAt; + if (index === 1) work.execution_ref = REF; + if (index === 2) work.worker_profile = "other"; + }, + }); +} + +function deploy(f, contract = "b".repeat(64), available = true) { + const worker = f.state.deployments.get("default").get("default"); + f.append( + "deployment", + { + operations: [ + { + kind: "Deployment", + pool: "default", + worker_profile: "default", + expected_ref: worker.current_ref, + expected_contract: worker.current_contract, + profile: { ...worker.revisions[worker.current_ref].profile, ref: "b".repeat(40), logical_contract: contract }, + available, + reason: "compiler_deployment", + }, + ], + }, + f.dispatcher + ); +} + +describe("versioned Work cancellation compatibility", () => { + it.each([false, true])("replays legacy compatible and pinned cancellation history after incompatible deployment=%s", changed => { + const f = fixture(); + if (changed) deploy(f); + const operations = [...f.state.works.keys()].map(work_id => ({ kind: "WorkCancellation", work_id, reason: REASON })); + f.append("cancel_work", { operations }); + const original = canonical(f.transactions); + let log = f.transactions; + for (const sha of ["c".repeat(40), "d".repeat(40)]) { + log = compactTransactions(log, sha, f.administrator, f.at + 1); + const state = replayTransactions(log); + for (const operation of operations) { + expect(state.works.get(operation.work_id).state).toBe("cancelled"); + expect(state.cancellations.get(operation.work_id)).toEqual(operation); + expect(state.cancellations.get(operation.work_id)).not.toHaveProperty("version"); + } + } + expect(canonical(f.transactions)).toBe(original); + }); + + it("checks new retirement, preserves its marker across checkpoints and rejects cross-form terminal rewrites", () => { + const f = fixture(); + deploy(f); + const work_id = [...f.state.works.keys()][0]; + const operation = { kind: "WorkCancellation", version: 1, work_id, reason: REASON }; + f.append("cancel_work", { operations: [operation] }); + f.append("cancel_work", { operations: [operation] }); + const checkpoint = compactTransactions(f.transactions, "c".repeat(40), f.administrator, f.at + 1); + expect(replayTransactions(checkpoint).cancellations.get(work_id)).toEqual(operation); + expect(() => f.append("cancel_work", { operations: [{ kind: "WorkCancellation", work_id, reason: REASON }] })).toThrow(expect.objectContaining({ code: "cancellation_conflict" })); + const legacy = fixture(); + const legacyId = [...legacy.state.works.keys()][0]; + legacy.append("cancel_work", { operations: [{ kind: "WorkCancellation", work_id: legacyId, reason: REASON }] }); + expect(() => legacy.append("cancel_work", { operations: [{ ...operation, work_id: legacyId }] })).toThrow(expect.objectContaining({ code: "cancellation_conflict" })); + }); + + it.each(["compatible", "pinned", "unavailable", "claimed", "other-profile"])("rejects checked retirement of %s Work atomically", scenario => { + const f = fixture(scenario === "claimed"); + if (scenario !== "compatible") deploy(f, "b".repeat(64), scenario !== "unavailable"); + const works = [...f.state.works.values()]; + const selected = scenario === "pinned" ? works[1] : scenario === "other-profile" ? works[2] : works[0]; + const original = canonical(f.transactions); + const operation = { kind: "WorkCancellation", version: 1, work_id: selected.work_id, reason: REASON }; + expect(() => f.append("cancel_work", { operations: [operation] })).toThrow(expect.objectContaining({ code: "deployment_conflict" })); + expect(canonical(f.transactions)).toBe(original); + }); + + it("preserves all ownership/barrier guards and does not infer obsolescence from missing deployments", () => { + const f = fixture(); + deploy(f); + const state = f.state; + const work = [...state.works.values()][0]; + for (const change of [{ state: "claimed" }, { claim_id: "owned" }, ...["pending", "failed", "verified"].map(barrier => ({ barrier }))]) expect(obsoleteWork(state, { ...work, ...change })).toBe(false); + state.deployments.clear(); + expect(obsoleteWork(state, work)).toBe(false); + }); + + it("rejects checked retirement when the publication time reaches the Work deadline", () => { + const f = fixture(false, 2000); + deploy(f); + const work_id = [...f.state.works.keys()][0]; + const original = canonical(f.transactions); + expect(() => f.append("cancel_work", { operations: [{ kind: "WorkCancellation", version: 1, work_id, reason: REASON }] }, f.administrator, "expired-retirement", 2000)).toThrow(expect.objectContaining({ code: "cancellation_conflict" })); + expect(canonical(f.transactions)).toBe(original); + }); + + it.each(["dispatcher", "worker", "projector"])("does not grant %s new cancellation authority", role => { + const f = fixture(); + deploy(f); + const work_id = [...f.state.works.keys()][0]; + const original = canonical(f.transactions); + const actor = { ...f.dispatcher, role }; + expect(() => f.append("cancel_work", { operations: [{ kind: "WorkCancellation", version: 1, work_id, reason: REASON }] }, actor)).toThrow(); + expect(canonical(f.transactions)).toBe(original); + }); + + it.each([0, 2, null, "1", false])("rejects unsupported explicit cancellation version %s", version => { + expect(() => validateOperation({ kind: "WorkCancellation", version, work_id: "w", reason: REASON })).toThrow(expect.objectContaining({ code: "unsupported_protocol" })); + }); +}); diff --git a/actions/setup/js/work_queue_deployment_authority.test.cjs b/actions/setup/js/work_queue_deployment_authority.test.cjs index bc9aab41326..dd50d3ab539 100644 --- a/actions/setup/js/work_queue_deployment_authority.test.cjs +++ b/actions/setup/js/work_queue_deployment_authority.test.cjs @@ -212,6 +212,32 @@ describe("verified default worker deployment authority", () => { expect(options.publishWorkQueueRequest.mock.calls[0][0].request.parameters.operations.map(operation => operation.work_id)).toEqual(["Work-B", "work-B", "work-a"]); expect(canonical(f.transactions)).toBe(history); }); + + it.each(["wall clock", "explicit time"])("excludes elapsed deadlines using %s without mutating the read projection or ledger", async clock => { + const { f, options } = await outdated(); + const projection = structuredClone(f.state); + const work = [...projection.works.values()].find(candidate => obsoleteWork(projection, candidate)); + const now = f.at + 100; + projection.works = new Map( + [{ work_id: "work_a", expires_at: now - 1 }, { work_id: "work_b", expires_at: now }, { work_id: "work_c", expires_at: now + 1 }, { work_id: "work_d" }].map(deadline => [deadline.work_id, { ...work, ...deadline }]) + ); + options.readWorkQueueLog = async () => ({ ...(await f.readWorkQueueLog()), state: projection }); + if (clock === "explicit time") options.now = now; + const failure = new Error("stop before ledger publication"); + options.publishWorkQueueRequest = vi.fn().mockRejectedValue(failure); + const before = structuredClone(projection); + const history = canonical(f.transactions); + const time = vi.spyOn(Date, "now").mockReturnValue(clock === "explicit time" ? now + 100 : now); + try { + await expect(reconcileObsoleteWork(options)).rejects.toBe(failure); + expect(options.publishWorkQueueRequest).toHaveBeenCalledOnce(); + expect(options.publishWorkQueueRequest.mock.calls[0][0].request.parameters.operations.map(operation => operation.work_id)).toEqual(["work_c", "work_d"]); + expect(projection).toEqual(before); + expect(canonical(f.transactions)).toBe(history); + } finally { + time.mockRestore(); + } + }); }); it.each([ diff --git a/actions/setup/js/work_queue_deployment_control.cjs b/actions/setup/js/work_queue_deployment_control.cjs index b08616a7987..b63590092e4 100644 --- a/actions/setup/js/work_queue_deployment_control.cjs +++ b/actions/setup/js/work_queue_deployment_control.cjs @@ -5,6 +5,7 @@ const { canonical, digest, utf8Compare } = require("./work_queue_codec.cjs"); const { policyProposalFor, loadQueue, publishOperations } = require("./work_queue_binding.cjs"); const { verifiedDefaultReference, configuredWorkerDeployment } = require("./work_queue_provisioning.cjs"); const { obsoleteWork } = require("./work_queue_deployment.cjs"); +const { expirationView } = require("./work_queue_expiration.cjs"); const { authenticatePublisher } = require("./work_queue_native.cjs"); const { isStagedMode } = require("./safe_output_helpers.cjs"); @@ -12,11 +13,12 @@ async function reconcileObsoleteWork(options) { if (isStagedMode(options) || isStagedMode(options.config)) return { cancelled: 0 }; const latest = await loadQueue({ ...options, policyProposal: undefined }); if (latest.projection.policy?.authorization !== "aw") return { cancelled: 0 }; - const operations = [...latest.projection.works.values()] - .filter(work => obsoleteWork(latest.projection, work)) + const projection = expirationView(latest.projection, options.now ?? Date.now()); + const operations = [...projection.works.values()] + .filter(work => obsoleteWork(projection, work)) .sort((a, b) => utf8Compare(a.work_id, b.work_id)) .slice(0, Math.min(256, latest.projection.policy.limits.operations)) - .map(work => ({ kind: "WorkCancellation", work_id: work.work_id, reason: "worker_contract_obsolete" })); + .map(work => ({ kind: "WorkCancellation", version: 1, work_id: work.work_id, reason: "worker_contract_obsolete" })); if (!operations.length) return { cancelled: 0 }; const context = await authenticatePublisher({ ...options, role: "reconciler" }); await publishOperations(options, context, ["obsolete-work", latest.projection.tip, operations], "cancel_work", operations); diff --git a/actions/setup/js/work_queue_replay.cjs b/actions/setup/js/work_queue_replay.cjs index 197ba94ccb1..2aa64d4b5cb 100644 --- a/actions/setup/js/work_queue_replay.cjs +++ b/actions/setup/js/work_queue_replay.cjs @@ -62,7 +62,10 @@ const OP_FIELDS = { ["source_updated_at", "state_reason", "resource_state", "merged", "merge_commit"], ], ClaimCancellation: [["kind", "work_id", "claim_id", "reason", "retry_not_before"], []], - WorkCancellation: [["kind", "work_id", "reason"], ["claim_id"]], + WorkCancellation: [ + ["kind", "work_id", "reason"], + ["claim_id", "version"], + ], WorkPriority: [["kind", "work_id", "priority", "expected_priority", "reason"], []], Dispatch: [ ["kind", "dispatch_id", "state"], @@ -182,6 +185,7 @@ function validateOperation(operation) { for (const field of ["work_id", "claim_id", "dispatch_id", "handle", "claim_handle", "completion_id", "epoch", "observation_id", "credential_generation", "read_status"]) if (Object.hasOwn(operation, field)) identity(operation[field], field); if (Object.hasOwn(operation, "reason")) validateReason(operation.reason); + if (operation.kind === "WorkCancellation" && Object.hasOwn(operation, "version") && operation.version !== 1) throw queueError("unsupported_protocol", "checked WorkCancellation requires version 1"); if (Object.hasOwn(operation, "evidence")) validateEvidence(operation.evidence); if (Object.hasOwn(operation, "run")) validateRunBinding(operation.run); if (Object.hasOwn(operation, "sender")) validateActor(operation.sender); @@ -569,7 +573,7 @@ function applyLifecycle(state, operation, commit) { } case "WorkCancellation": if (!work || work.state === "completed" || work.barrier === "pending") throw queueError("work_terminal", "completed ownership cannot be reopened or cancelled"); - if (operation.reason === "worker_contract_obsolete" && work.state !== "cancelled" && !obsoleteWork(state, work)) + if (operation.version === 1 && operation.reason === "worker_contract_obsolete" && work.state !== "cancelled" && !obsoleteWork(state, work)) throw queueError("deployment_conflict", "obsolete cancellation requires available unpinned Work with no delivery barrier and an incompatible active worker contract"); if (operation.claim_id) { const selected = state.claims.get(operation.claim_id); @@ -802,6 +806,7 @@ function checkpointSnapshot(state, lastAt, retainPolicyHistory = true) { if (operation.kind === "Deployment" || (retainPolicyHistory && ["Policy", "PoolInstall"].includes(operation.kind))) return structuredClone(operation); const fields = [ "kind", + "version", "work_id", "logical_contract", "execution_ref", diff --git a/docs/src/content/docs/reference/work-queue.md b/docs/src/content/docs/reference/work-queue.md index 4be693daecb..eee2312c37e 100644 --- a/docs/src/content/docs/reference/work-queue.md +++ b/docs/src/content/docs/reference/work-queue.md @@ -118,7 +118,9 @@ The immutable Work payload declares its output contract in `effect_contract`. A Worker conclusion performs inline reservation cleanup. Scheduled `agentics-maintenance.yml` reconciliation handles stranded reservations and pending delivery before compaction. Maintenance never cancels active runs or manufactures Result; disabling it leaves dispatcher/operator reconciliation as the fallback. -Maintenance also retires obsolete backlog in checked batches of at most 256 Work (further bounded by the installed operation limit). Available, unpinned Work with no delivery barrier is terminally cancelled with reason `worker_contract_obsolete` when its admission contract differs from its approved active worker contract. Missing or unavailable deployments are not proof of obsolescence. Exact `execution_ref` pins, owned Work, delivery barriers, Results, ledger history and fairness debt are preserved. Cancellation rechecks eligibility against the publication's current ledger; a concurrent Claim or compatible route change rejects the transaction. Retirement failures emit sanitized diagnostics and a warning without skipping reservation reconciliation, pending delivery, Issue synchronization or compaction. Operators can use `cancel-work --reason worker_contract_obsolete --work-id ID` for the same checked recovery without dispatching workers. +Maintenance also retires obsolete backlog in checked batches of at most 256 Work (further bounded by the installed operation limit). Available, unpinned Work with no delivery barrier is terminally cancelled using `WorkCancellation` with operation `version: 1` and reason `worker_contract_obsolete` when its admission contract differs from its approved active worker contract. Selection accounts for expiration at the current time, so elapsed Work is excluded. Missing or unavailable deployments are not proof of obsolescence. Exact `execution_ref` pins, owned Work, delivery barriers, Results, ledger history and fairness debt are preserved. Cancellation rechecks eligibility against the publication's current ledger; a concurrent Claim, expiration or compatible route change rejects the transaction. Retirement failures emit sanitized diagnostics and a warning without skipping reservation reconciliation, pending delivery, Issue synchronization or compaction. Operators can use `cancel-work --reason worker_contract_obsolete --work-id ID` for the same checked recovery without dispatching workers. + +The ledger remains version 3. Historical `WorkCancellation` operations without an operation `version` retain reason-only semantics, including cancellations of compatible or pinned Work with reason `worker_contract_obsolete`; replay and checkpoints do not retrofit an obsolescence check or rewrite those receipts. New maintenance and operator retirement requests explicitly use operation `version: 1`. Retrying an accepted legacy operator request recovers its original unchecked receipt rather than creating a new cancellation. Other explicit operation versions are rejected. ### Claim-scoped memory diff --git a/docs/src/content/docs/specs/work-queue-specification.md b/docs/src/content/docs/specs/work-queue-specification.md index 3a49f53e48a..05bb4037604 100644 --- a/docs/src/content/docs/specs/work-queue-specification.md +++ b/docs/src/content/docs/specs/work-queue-specification.md @@ -995,7 +995,7 @@ Use a small operation union: | `DeliveryFailure` | Close a completed node's unresolved delivery barrier after terminal-run evidence and bounded reconciliation; never authorize effects or release successors | | `Observation` | Record normalized trusted GitHub Issue/PR condition evidence, including identity and observation time | | `ClaimCancellation` | Remove this Claim's authority; optionally record trusted retry-not-before metadata | -| `WorkCancellation` | Terminal cancellation; optional `claim_id` fences the selected current owner; does not pretend the worker has stopped using capacity | +| `WorkCancellation` | Terminal cancellation; optional `claim_id` fences the selected current owner; operation `version: 1` checks `worker_contract_obsolete` eligibility; absent version retains historical reason-only semantics; does not pretend the worker has stopped using capacity | | `WorkPriority` | Administrator-only prospective override for available Work, with `expected_priority` compare-and-set; admitted priority, FIFO position, retry boundary and all charged clocks remain unchanged | | `Dispatch` | Record launch started, bound run, definitive rejection, or uncertainty for a `dispatch_id` and its immutable Claim array | | `Release` | Release the group's native reservation once with trusted termination/nonlaunch evidence; logical Claim slots close independently | @@ -1004,6 +1004,23 @@ Do not repeat the entire Claim array in each dispatch lifecycle operation. Immutable membership is derived from the admission commit; lifecycle operations reference `dispatch_id`, its binding/evidence, and stable request identity. +The QueueCommit protocol remains version 3. `WorkCancellation` without an +operation `version` MUST retain legacy reason-only semantics, including +`worker_contract_obsolete` cancellations of compatible or explicitly pinned Work. +Operation `version: 1` adds a checked obsolescence transition for that reason; +other explicit versions MUST be rejected. New maintenance and operator retirement +producers MUST emit this checked form. Replay MUST require available, unclaimed, +unpinned Work without any delivery barrier and an incompatible, available, +approved active worker contract at the operation's committed time. Missing or +unavailable deployments MUST NOT establish obsolescence. Selection MUST exclude +Work expired at the current time, and publication MUST recheck against its own +current prefix. Historical receipts and checkpoint state MUST NOT be rewritten +to add or remove an operation version; only exact repeats of terminal +cancellation metadata are idempotent. A retry of an accepted legacy operator +request MUST recover its original request and fingerprint, not publish a checked +replacement. All existing authorization, ownership and native-reservation +requirements still apply to both forms. + `WorkPriority` is published through an administrator `control` request in this same transaction log. Replay stores `effective_priority` separately from the immutable Work definition; native Go and JavaScript selection and historical diff --git a/pkg/cli/work_command_cancellation_version_test.go b/pkg/cli/work_command_cancellation_version_test.go new file mode 100644 index 00000000000..7f257322d29 --- /dev/null +++ b/pkg/cli/work_command_cancellation_version_test.go @@ -0,0 +1,127 @@ +package cli + +import ( + "encoding/json" + "slices" + "strings" + "testing" + + "github.com/github/gh-aw/pkg/workqueue" +) + +func TestWorkOperatorCheckedCancellationIntent(t *testing.T) { + state, commits, ids := workOperatorFixture(t, false) + targets := []workOperatorTarget{{WorkID: ids[0]}} + for _, reason := range []string{"worker_contract_obsolete", "operator_cancelled"} { + params, kind, err := workOperatorIntent(state, slices.Clone(targets), reason, 0, "new-retirement") + if err != nil || kind != "cancel_work" { + t.Fatalf("cancellation intent: %s %v", kind, err) + } + var fields map[string]any + if err := json.Unmarshal(params.Operations[0], &fields); err != nil { + t.Fatal(err) + } + if reason == "worker_contract_obsolete" { + if fields["version"] != float64(1) { + t.Fatal("operator obsolete cancellation omitted its checked version") + } + actor := state.Requests["submit"].Actor + request, err := workqueue.NewRequest("new-retirement", kind, actor, params) + if err != nil { + t.Fatal(err) + } + if _, _, _, err := workqueue.BuildCandidate(commits, actor, request, 3000); err == nil || + !strings.HasPrefix(err.Error(), "deployment_conflict:") { + t.Fatalf("fresh operator intent bypassed checked eligibility: %v", err) + } + } else if _, exists := fields["version"]; exists { + t.Fatal("ordinary operator reason changed its legacy shape") + } + } + params, _, err := workOperatorIntent(state, targets, "worker_contract_obsolete", 1, "priority") + if err != nil { + t.Fatal(err) + } + var priority map[string]any + _ = json.Unmarshal(params.Operations[0], &priority) + if _, exists := priority["version"]; exists { + t.Fatal("cancellation version leaked into WorkPriority") + } +} + +func TestWorkOperatorLegacyRetirementReceiptRecovery(t *testing.T) { + state, commits, ids := workOperatorFixture(t, false) + actor := state.Requests["submit"].Actor + operation, err := workqueue.Op(map[string]any{ + "kind": "WorkCancellation", "work_id": ids[0], "reason": "worker_contract_obsolete", + }) + if err != nil { + t.Fatal(err) + } + params := workqueue.OperationsParameters{Operations: []workqueue.Operation{operation}} + request, err := workqueue.NewRequest("legacy-retirement", "cancel_work", actor, params) + if err != nil { + t.Fatal(err) + } + commits, _, _, err = workqueue.BuildCandidate(commits, actor, request, 3000) + if err != nil { + t.Fatal(err) + } + checkWorkRetirementRecovery(t, commits, ids[0], request, params) + for _, sha := range []string{strings.Repeat("c", 40), strings.Repeat("d", 40)} { + commits, err = workqueue.CompactCheckpoint(commits, sha, actor, 4000) + if err != nil { + t.Fatal(err) + } + checkWorkRetirementRecovery(t, commits, ids[0], request, params) + } +} + +func checkWorkRetirementRecovery(t *testing.T, commits []workqueue.QueueCommit, workID string, original workqueue.Request, params workqueue.OperationsParameters) { + t.Helper() + state, err := workqueue.Replay(commits) + if err != nil { + t.Fatal(err) + } + targets := []workOperatorTarget{{WorkID: workID}} + actor := state.Requests[original.ID].Actor + recovered, kind, err := workOperatorIntent(state, targets, "worker_contract_obsolete", 0, original.ID) + if err != nil { + t.Fatal(err) + } + request, err := workqueue.NewRequest(original.ID, kind, actor, recovered) + if err != nil || request.Fingerprint != original.Fingerprint { + t.Fatalf("legacy retry changed its accepted fingerprint: %v", err) + } + if string(recovered.Operations[0]) != string(params.Operations[0]) { + t.Fatal("legacy retry retrofitted checked cancellation metadata") + } + if _, _, err := workOperatorIntent(state, targets, "different_reason", 0, original.ID); err == nil { + t.Fatal("legacy retry changed its accepted reason") + } + ack, _, _, err := workqueue.BuildCandidate(commits, actor, request, 5000) + if err != nil || len(ack) != len(commits) { + t.Fatalf("legacy recovery attempted a new publication: %v", err) + } +} + +func TestWorkRecoverCheckedCancellationMarker(t *testing.T) { + state, _, ids := workOperatorFixture(t, false) + actor := state.Requests["submit"].Actor + operation, err := workqueue.Op(map[string]any{ + "kind": "WorkCancellation", "version": 1, "work_id": ids[0], "reason": "worker_contract_obsolete", + }) + if err != nil { + t.Fatal(err) + } + params := workqueue.OperationsParameters{Operations: []workqueue.Operation{operation}} + request, err := workqueue.NewRequest("checked-retirement", "cancel_work", actor, params) + if err != nil { + t.Fatal(err) + } + request.Parameters = json.RawMessage("null") + recovered, err := workRecoverClearIntent(workqueue.QueueCommit{Request: request, Actor: actor, Operations: params.Operations}, "worker_contract_obsolete") + if err != nil || string(recovered.Operations[0]) != string(operation) { + t.Fatalf("checkpoint recovery removed checked version: %v", err) + } +} diff --git a/pkg/cli/work_command_clear.go b/pkg/cli/work_command_clear.go index fa881b57209..af97f27ec94 100644 --- a/pkg/cli/work_command_clear.go +++ b/pkg/cli/work_command_clear.go @@ -229,6 +229,7 @@ func workRecoverClearIntent(accepted workqueue.QueueCommit, reason string) (work for _, operation := range accepted.Operations { var fields struct { Kind string `json:"kind"` + Version *int `json:"version,omitempty"` WorkID string `json:"work_id"` ClaimID string `json:"claim_id,omitempty"` Reason string `json:"reason"` diff --git a/pkg/cli/work_command_operations.go b/pkg/cli/work_command_operations.go index 934c8f20cb2..bb8d48f2660 100644 --- a/pkg/cli/work_command_operations.go +++ b/pkg/cli/work_command_operations.go @@ -143,6 +143,9 @@ func workOperatorIntent(state workqueue.Projection, targets []workOperatorTarget } previous = target.WorkID fields := map[string]any{"kind": "WorkCancellation", "work_id": target.WorkID, "reason": reason} + if reason == "worker_contract_obsolete" && priority == 0 { + fields["version"] = 1 + } if target.ClaimID != "" { fields["claim_id"] = target.ClaimID } @@ -167,7 +170,19 @@ func recoverWorkOperatorIntent(state workqueue.Projection, wanted workqueue.Oper return wanted, nil } var old workqueue.OperationsParameters - if err := json.Unmarshal(accepted.Request.Parameters, &old); err != nil { + var err error + if kind == "cancel_work" && len(wanted.Operations) > 0 { + var desired struct { + Reason string `json:"reason"` + } + if err := json.Unmarshal(wanted.Operations[0], &desired); err != nil { + return wanted, err + } + old, err = workRecoverClearIntent(accepted, desired.Reason) + } else { + err = json.Unmarshal(accepted.Request.Parameters, &old) + } + if err != nil { return wanted, err } if accepted.Request.Kind != kind || len(old.Operations) != len(wanted.Operations) { @@ -181,7 +196,7 @@ func recoverWorkOperatorIntent(state workqueue.Projection, wanted workqueue.Oper if err != nil { return wanted, err } - right, err := workOperatorComparison(wanted.Operations[index]) + right, err := workOperatorRetryComparison(operation, wanted.Operations[index]) if err != nil { return wanted, err } @@ -200,3 +215,21 @@ func workOperatorComparison(operation workqueue.Operation) ([]byte, error) { delete(fields, "expected_priority") return json.Marshal(fields) } + +func workOperatorRetryComparison(original, wanted workqueue.Operation) ([]byte, error) { + var old, next map[string]json.RawMessage + if err := json.Unmarshal(original, &old); err != nil { + return nil, err + } + if err := json.Unmarshal(wanted, &next); err != nil { + return nil, err + } + // Acknowledging an accepted legacy request must recover its original bytes, + // not turn its reason-only cancellation into a new checked publication. + if old["version"] == nil && string(old["kind"]) == `"WorkCancellation"` && + string(old["reason"]) == `"worker_contract_obsolete"` && string(next["version"]) == "1" { + delete(next, "version") + } + delete(next, "expected_priority") + return json.Marshal(next) +} diff --git a/pkg/workqueue/cancellation_version_test.go b/pkg/workqueue/cancellation_version_test.go new file mode 100644 index 00000000000..505b59b00a0 --- /dev/null +++ b/pkg/workqueue/cancellation_version_test.go @@ -0,0 +1,164 @@ +package workqueue + +import ( + "strconv" + "strings" + "testing" +) + +const obsoleteCancellationReason = "worker_contract_obsolete" + +func cancellationVersionFixture(t *testing.T) ([]QueueCommit, []WorkDefinition) { + t.Helper() + commits := testGenesis(t, deploymentPolicy) + nodes := []WorkDefinition{testNode(t, commits, "old"), testNode(t, commits, "pinned"), testNode(t, commits, "compatible")} + nodes[1].ExecutionRef = pinnedRef(commits, t) + nodes[2].WorkerProfile = "other" + return testSubmit(t, commits, "backlog", nodes...), nodes +} + +func TestLegacyObsoleteCancellationHistoryParity(t *testing.T) { + for _, changed := range []bool{false, true} { + t.Run(strconv.FormatBool(changed), func(t *testing.T) { + commits, nodes := cancellationVersionFixture(t) + if changed { + commits = deployTestWorker(t, commits, "incompatible", "default", strings.Repeat("b", 40), strings.Repeat("b", 64), true) + } + operations := []Operation{} + for _, node := range nodes { + operations = append(operations, mustOp(t, map[string]any{ + "kind": "WorkCancellation", "work_id": node.WorkID, "reason": obsoleteCancellationReason, + })) + } + commits = testOperations(t, commits, testActor("administrator"), "legacy-retirement", "cancel_work", operations...) + assertOperatorJSProjection(t, commits) + for _, sha := range []string{strings.Repeat("c", 40), strings.Repeat("d", 40)} { + var err error + commits, err = CompactCheckpoint(commits, sha, testActor("administrator"), 5000) + if err != nil { + t.Fatal(err) + } + state, err := Replay(commits) + if err != nil { + t.Fatal(err) + } + for index, node := range nodes { + if state.Works[node.WorkID].State != "cancelled" || !sameJSON(state.Cancellations[node.WorkID], operations[index]) { + t.Fatal("checkpoint changed historical reason-only cancellation") + } + } + assertOperatorJSProjection(t, commits) + } + }) + } +} + +func TestCheckedObsoleteCancellationGuards(t *testing.T) { + base, nodes := cancellationVersionFixture(t) + base = deployTestWorker(t, base, "incompatible", "default", strings.Repeat("b", 40), strings.Repeat("b", 64), true) + operation := mustOp(t, map[string]any{ + "kind": "WorkCancellation", "version": 1, "work_id": nodes[0].WorkID, "reason": obsoleteCancellationReason, + }) + for name, change := range map[string]func(*Projection, *WorkState){ + "claimed": func(_ *Projection, w *WorkState) { w.State = "claimed" }, + "owner": func(_ *Projection, w *WorkState) { w.ClaimID = "owned" }, + "pin": func(_ *Projection, w *WorkState) { w.ExecutionRef = nodes[1].ExecutionRef }, + "pending": func(_ *Projection, w *WorkState) { w.Barrier = "pending" }, + "failed": func(_ *Projection, w *WorkState) { w.Barrier = "failed" }, + "verified": func(_ *Projection, w *WorkState) { w.Barrier = "verified" }, + "missing": func(s *Projection, _ *WorkState) { delete(s.Deployments["default"], "default") }, + "compatible": func(s *Projection, _ *WorkState) { + worker := s.Deployments["default"]["default"] + revision := worker.Revisions[worker.CurrentRef] + revision.Profile.LogicalContract = strings.Repeat("a", 64) + worker.Revisions[worker.CurrentRef] = revision + }, + "unavailable": func(s *Projection, _ *WorkState) { + worker := s.Deployments["default"]["default"] + revision := worker.Revisions[worker.CurrentRef] + revision.Available = false + worker.Revisions[worker.CurrentRef] = revision + }, + } { + t.Run(name, func(t *testing.T) { + state, err := Replay(base) + if err != nil { + t.Fatal(err) + } + change(&state, state.Works[nodes[0].WorkID]) + before, _ := canonicalValue(state) + err = state.replayCancellation(operation, "WorkCancellation", QueueCommit{ID: "checked", Actor: testActor("administrator")}) + if err == nil || !strings.HasPrefix(err.Error(), "deployment_conflict:") { + t.Fatalf("checked cancellation accepted protected Work: %v", err) + } + after, _ := canonicalValue(state) + if string(before) != string(after) { + t.Fatal("rejected cancellation mutated protected state") + } + }) + } +} + +func TestCheckedCancellationVersionAndTerminalMetadataParity(t *testing.T) { + base, nodes := cancellationVersionFixture(t) + base = deployTestWorker(t, base, "incompatible", "default", strings.Repeat("b", 40), strings.Repeat("b", 64), true) + legacy := mustOp(t, map[string]any{"kind": "WorkCancellation", "work_id": nodes[0].WorkID, "reason": obsoleteCancellationReason}) + checked := mustOp(t, map[string]any{"kind": "WorkCancellation", "version": 1, "work_id": nodes[0].WorkID, "reason": obsoleteCancellationReason}) + for _, pair := range [][2]Operation{{legacy, checked}, {checked, legacy}} { + commits := testOperations(t, base, testActor("administrator"), "cancel", "cancel_work", pair[0]) + commits = testOperations(t, commits, testActor("administrator"), "repeat", "cancel_work", pair[0]) + assertOperatorJSProjection(t, commits) + commits, err := CompactCheckpoint(commits, strings.Repeat("c", 40), testActor("administrator"), 5000) + if err != nil { + t.Fatal(err) + } + assertOperatorJSProjection(t, commits) + request, _ := NewRequest("rewrite", "cancel_work", testActor("administrator"), OperationsParameters{Operations: []Operation{pair[1]}}) + if _, _, _, err := BuildCandidate(commits, testActor("administrator"), request, 6000); err == nil || !strings.HasPrefix(err.Error(), "cancellation_conflict:") { + t.Fatalf("terminal receipt version rewritten: %v", err) + } + } + + for _, version := range []any{0, 2, nil, "1", false} { + operation := mustOp(t, map[string]any{ + "kind": "WorkCancellation", "version": version, "work_id": nodes[0].WorkID, "reason": obsoleteCancellationReason, + }) + request, _ := NewRequest("invalid-version", "cancel_work", testActor("administrator"), OperationsParameters{Operations: []Operation{operation}}) + if _, _, _, err := BuildCandidate(base, testActor("administrator"), request, 5000); err == nil { + t.Fatalf("accepted unsupported explicit version %v", version) + } + } +} + +func TestCheckedCancellationExpirationAndAuthorization(t *testing.T) { + base := testGenesis(t, deploymentPolicy) + work := testNode(t, base, "expiring") + deadline := int64(4500) + work.ExpiresAt = &deadline + base = testSubmit(t, base, "expiring-backlog", work) + base = deployTestWorker(t, base, "incompatible", "default", strings.Repeat("b", 40), strings.Repeat("b", 64), true) + operation := mustOp(t, map[string]any{ + "kind": "WorkCancellation", "version": 1, "work_id": work.WorkID, "reason": obsoleteCancellationReason, + }) + actor := testActor("administrator") + request, err := NewRequest("expired-retirement", "cancel_work", actor, OperationsParameters{Operations: []Operation{operation}}) + if err != nil { + t.Fatal(err) + } + for _, at := range []int64{deadline, deadline + 1} { + if _, _, _, err := BuildCandidate(base, actor, request, at); err == nil || !strings.HasPrefix(err.Error(), "cancellation_conflict:") { + t.Fatalf("checked cancellation accepted elapsed Work: %v", err) + } + } + for _, role := range []string{"dispatcher", "worker", "projector"} { + actor := testActor(role) + request, err := NewRequest("unauthorized-retirement", "cancel_work", actor, OperationsParameters{Operations: []Operation{operation}}) + if err != nil { + t.Fatal(err) + } + if _, _, _, err := BuildCandidate(base, actor, request, 4001); err == nil { + t.Fatalf("checked version granted %s cancellation authority", role) + } + } + assertOperatorJSProjection(t, base) +} diff --git a/pkg/workqueue/deployment_test.go b/pkg/workqueue/deployment_test.go index 5129b71dbe2..2f6885bb288 100644 --- a/pkg/workqueue/deployment_test.go +++ b/pkg/workqueue/deployment_test.go @@ -83,28 +83,14 @@ func TestObsoleteWorkerCancellationReplayParity(t *testing.T) { if !before.obsoleteWork(before.Works[old.WorkID]) || before.obsoleteWork(before.Works[pinned.WorkID]) { t.Fatal("obsolete selection did not preserve the immutable pin") } - operation := mustOp(t, map[string]any{"kind": "WorkCancellation", "work_id": old.WorkID, "reason": "worker_contract_obsolete"}) + operation := mustOp(t, map[string]any{"kind": "WorkCancellation", "version": 1, "work_id": old.WorkID, "reason": "worker_contract_obsolete"}) commits = testOperations(t, commits, testActor("reconciler"), "retire-obsolete", "cancel_work", operation) after, err := Replay(commits) if err != nil || after.Works[old.WorkID].State != "cancelled" || !sameJSON(before.Works[pinned.WorkID], after.Works[pinned.WorkID]) || !sameJSON(before.Policy, after.Policy) || !sameJSON(before.Clocks, after.Clocks) || !sameJSON(before.Claims, after.Claims) { t.Fatalf("obsolete cancellation changed protected state: %v", err) } - input, err := canonicalValue(commits) - if err != nil { - t.Fatal(err) - } - command := exec.Command("node", "-e", ` -const assert = require("node:assert/strict"); -const r = require("../../actions/setup/js/work_queue_replay.cjs"); -const s = r.replayTransactions(JSON.parse(require("node:fs").readFileSync(0, "utf8"))); -assert.equal([...s.works.values()].filter(w => w.state === "cancelled").length, 1); -assert.equal([...s.works.values()].filter(w => w.execution_ref && w.state === "available").length, 1); -`) - command.Stdin = bytes.NewReader(input) - if output, err := command.CombinedOutput(); err != nil { - t.Fatalf("obsolete cancellation Go/JS replay parity: %v\n%s", err, output) - } + assertOperatorJSProjection(t, commits) for _, changes := range []func(*WorkState){ func(w *WorkState) { w.State = "claimed" }, func(w *WorkState) { w.ClaimID = "owned" }, diff --git a/pkg/workqueue/inspection.go b/pkg/workqueue/inspection.go index 1e041646ad7..747a752a909 100644 --- a/pkg/workqueue/inspection.go +++ b/pkg/workqueue/inspection.go @@ -225,6 +225,7 @@ type TraceEvent struct { At int64 `json:"at"` Actor Actor `json:"actor"` Kind string `json:"kind"` + Version *int `json:"version,omitempty"` Pool string `json:"pool,omitempty"` WorkerProfile string `json:"worker_profile,omitempty"` Profile *WorkerProfile `json:"profile,omitempty"` @@ -291,8 +292,10 @@ func traceEventForOperation(operation Operation) (TraceEvent, error) { if err := json.Unmarshal(operation, &fields); err != nil { return TraceEvent{}, err } - if state := fields["state"]; len(state) > 0 && (state[0] == '{' || state[0] == '[') { - delete(fields, "state") + if state := fields["state"]; len(state) > 0 { + if state[0] == '{' || state[0] == '[' { + delete(fields, "state") + } } sanitized, err := json.Marshal(fields) if err != nil { @@ -338,52 +341,19 @@ func TraceQueue(commits []QueueCommit, options TraceOptions, at int64) (TracePag } result.TotalEvents++ } - if len(ordered) > 0 && isCheckpoint(ordered[0]) { - for receiptIndex, receipt := range state.CheckpointReceipts { - commit := QueueCommit{ - ID: receipt.ID, Previous: receipt.Previous, - Request: Request{ID: receipt.RequestID, Kind: receipt.Kind}, - Actor: receipt.Actor, PolicyEpoch: receipt.PolicyEpoch, At: receipt.At, - } - for index, operation := range receipt.Events { - event, err := traceEventForOperation(operation) - if err != nil { - return TracePage{}, err - } - annotateTraceEvent(&event, state, commit, Position{Commit: receiptIndex, Operation: index}) - if scope.matches(event, commit, options) { - appendEvent(event) - } - } - } - for ordinal, commit := range ordered[1:] { - for index, operation := range commit.Operations { - event, err := traceEventForOperation(operation) - if err != nil { - return TracePage{}, err - } - if !scope.matches(event, commit, options) { - continue - } - annotateTraceEvent(&event, state, commit, Position{Commit: len(state.CheckpointReceipts) + 1 + ordinal, Operation: index}) - appendEvent(event) - } - } - } else { - for ordinal, commit := range ordered { - for index, operation := range commit.Operations { - event, err := traceEventForOperation(operation) - if err != nil { - return TracePage{}, err - } - if !scope.matches(event, commit, options) { - continue - } - annotateTraceEvent(&event, state, commit, Position{Commit: ordinal, Operation: index}) - appendEvent(event) + ordinalOffset := 0 + if len(ordered) > 0 { + if isCheckpoint(ordered[0]) { + if err := appendCheckpointTraceEvents(state, scope, options, appendEvent); err != nil { + return TracePage{}, err } + ordered = ordered[1:] + ordinalOffset = len(state.CheckpointReceipts) + 1 } } + if err := appendLiveTraceEvents(state, ordered, scope, options, ordinalOffset, appendEvent); err != nil { + return TracePage{}, err + } if options.Offset > result.TotalEvents { return TracePage{}, queueError("inspection_invalid", "trace offset exceeds matching events") } @@ -395,6 +365,44 @@ func TraceQueue(commits []QueueCommit, options TraceOptions, at int64) (TracePag return result, nil } +func appendCheckpointTraceEvents(state Projection, scope traceScope, options TraceOptions, appendEvent func(TraceEvent)) error { + for receiptIndex, receipt := range state.CheckpointReceipts { + commit := QueueCommit{ + ID: receipt.ID, Previous: receipt.Previous, + Request: Request{ID: receipt.RequestID, Kind: receipt.Kind}, + Actor: receipt.Actor, PolicyEpoch: receipt.PolicyEpoch, At: receipt.At, + } + for index, operation := range receipt.Events { + event, err := traceEventForOperation(operation) + if err != nil { + return err + } + annotateTraceEvent(&event, state, commit, Position{Commit: receiptIndex, Operation: index}) + if scope.matches(event, commit, options) { + appendEvent(event) + } + } + } + return nil +} + +func appendLiveTraceEvents(state Projection, ordered []QueueCommit, scope traceScope, options TraceOptions, ordinalOffset int, appendEvent func(TraceEvent)) error { + for ordinal, commit := range ordered { + for index, operation := range commit.Operations { + event, err := traceEventForOperation(operation) + if err != nil { + return err + } + if !scope.matches(event, commit, options) { + continue + } + annotateTraceEvent(&event, state, commit, Position{Commit: ordinalOffset + ordinal, Operation: index}) + appendEvent(event) + } + } + return nil +} + type traceScope struct { works, claims, dispatches, observations, epochs identitySet } diff --git a/pkg/workqueue/replay.go b/pkg/workqueue/replay.go index 335a2e312df..08a9bc58e79 100644 --- a/pkg/workqueue/replay.go +++ b/pkg/workqueue/replay.go @@ -320,16 +320,21 @@ func (state *Projection) replayCancellation(operation Operation, kind string, co } var value struct { - WorkID string `json:"work_id"` - Reason string `json:"reason"` + WorkID string `json:"work_id"` + Reason string `json:"reason"` + Version int `json:"version"` } if err := json.Unmarshal(operation, &value); err != nil { return err } work := state.Works[value.WorkID] - if value.Reason == "worker_contract_obsolete" && work != nil && work.State != "cancelled" && !state.obsoleteWork(work) { + if value.Version == 1 && value.Reason == "worker_contract_obsolete" && work != nil && work.State != "cancelled" && !state.obsoleteWork(work) { return queueError("deployment_conflict", "obsolete cancellation requires available unpinned Work with no delivery barrier and an incompatible active worker contract") } + if previous := state.Cancellations[value.WorkID]; work != nil && work.State == "cancelled" && + (previous != nil || value.Version == 1) && !sameJSON(previous, operation) { + return queueError("cancellation_conflict", "terminal cancellation metadata is immutable") + } if err := state.cancelWork(operation, commit); err != nil { return err } diff --git a/pkg/workqueue/schema/OperationsParameters.json b/pkg/workqueue/schema/OperationsParameters.json index e70a61c7215..f0713281d80 100644 --- a/pkg/workqueue/schema/OperationsParameters.json +++ b/pkg/workqueue/schema/OperationsParameters.json @@ -1136,6 +1136,9 @@ "reason": { "$ref": "#/$defs/ReasonCode" }, + "version": { + "const": 1 + }, "work_id": { "$ref": "#/$defs/Identity" } diff --git a/pkg/workqueue/schema/QueueCommit.json b/pkg/workqueue/schema/QueueCommit.json index b6d03c728b1..01f0acb5530 100644 --- a/pkg/workqueue/schema/QueueCommit.json +++ b/pkg/workqueue/schema/QueueCommit.json @@ -1372,6 +1372,9 @@ "reason": { "$ref": "#/$defs/ReasonCode" }, + "version": { + "const": 1 + }, "work_id": { "$ref": "#/$defs/Identity" } diff --git a/pkg/workqueue/schema/QueueOperation.json b/pkg/workqueue/schema/QueueOperation.json index 80470da7dad..4d2089d2157 100644 --- a/pkg/workqueue/schema/QueueOperation.json +++ b/pkg/workqueue/schema/QueueOperation.json @@ -1121,6 +1121,9 @@ "reason": { "$ref": "#/$defs/ReasonCode" }, + "version": { + "const": 1 + }, "work_id": { "$ref": "#/$defs/Identity" } diff --git a/pkg/workqueue/schema/QueueRequest.json b/pkg/workqueue/schema/QueueRequest.json index 1aa684f27d3..9b1c5c9cc24 100644 --- a/pkg/workqueue/schema/QueueRequest.json +++ b/pkg/workqueue/schema/QueueRequest.json @@ -1310,6 +1310,9 @@ "reason": { "$ref": "#/$defs/ReasonCode" }, + "version": { + "const": 1 + }, "work_id": { "$ref": "#/$defs/Identity" } diff --git a/pkg/workqueue/schema/RequestParameters.json b/pkg/workqueue/schema/RequestParameters.json index 4a9c301b12d..24f8deb3ec8 100644 --- a/pkg/workqueue/schema/RequestParameters.json +++ b/pkg/workqueue/schema/RequestParameters.json @@ -1242,6 +1242,9 @@ "reason": { "$ref": "#/$defs/ReasonCode" }, + "version": { + "const": 1 + }, "work_id": { "$ref": "#/$defs/Identity" } diff --git a/pkg/workqueue/schema/WorkCancellationOperation.json b/pkg/workqueue/schema/WorkCancellationOperation.json index 5ce5ed30dff..f159270e7d8 100644 --- a/pkg/workqueue/schema/WorkCancellationOperation.json +++ b/pkg/workqueue/schema/WorkCancellationOperation.json @@ -25,6 +25,9 @@ "reason": { "$ref": "#/$defs/ReasonCode" }, + "version": { + "const": 1 + }, "work_id": { "$ref": "#/$defs/Identity" } diff --git a/specs/work-queue/transactions.tsp b/specs/work-queue/transactions.tsp index 64049dc1071..5e4bc2c5352 100644 --- a/specs/work-queue/transactions.tsp +++ b/specs/work-queue/transactions.tsp @@ -354,6 +354,8 @@ model WorkCancellationOperation { kind: "WorkCancellation"; work_id: Identity; reason: ReasonCode; + // Absent preserves legacy reason-only replay; version 1 checks obsolescence. + version?: 1; // Optional compare-and-cancel fence for an operator selecting a Claim. claim_id?: Identity; } From fd183ab64630c08d569139a2feec407d687c4c04 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 9 Nov 2026 05:06:46 +0000 Subject: [PATCH 4/5] fix: keep work queue instructions within word limit Co-authored-by: gh-aw-bot <259018956+gh-aw-bot@users.noreply.github.com> --- .github/aw/work-queue.md | 63 ++++++++++++++++++++-------------------- 1 file changed, 31 insertions(+), 32 deletions(-) diff --git a/.github/aw/work-queue.md b/.github/aw/work-queue.md index e0f1ddc0052..ca576820263 100644 --- a/.github/aw/work-queue.md +++ b/.github/aw/work-queue.md @@ -11,12 +11,12 @@ For issue-backed WorkQueueOps: [WorkQueueOps](../../docs/src/content/docs/patter ## Select the role - **Observer:** explain only; never submit, dispatch or finish Work. -- **Producer/dispatcher:** submit trusted plans/request bounded pool work; - queue authority grants no arbitrary resource writes. +- **Producer/dispatcher:** submit trusted, bounded pool plans; queue authority + grants no arbitrary writes. - **Worker:** require compiler-supplied version-3 `claims`; never override reserved `work_queue_assignment` or caller context. -Trust roles, not snapshots: AW authenticates run/workflow/revision; +AW authenticates run/workflow/revision; reruns cannot inherit attempt 1. Absent branches are empty; reject malformed/unsupported logs. @@ -33,13 +33,12 @@ Dispatchers require `tools.work-queue: true`, a worker allowlist, bounded `dispatch-workflow.max`. Approved routes/credentials bind Claims to immutable native runs; ignore moved `target-ref`. -`aw.json` `work_queue`: concurrency, pending limits, retries, pools, fairness, -Issues. Defaults: weighted-priority, concurrency 16, +`aw.json` `work_queue` configures concurrency, pending limits, retries, pools, +fairness and Issues. Defaults: weighted-priority, concurrency 16, pending 4,096, singleton assignments, three attempts, 30-second backoff. -Submission initializes Policy/Work. Trusted submit/dispatch install pools/profiles; -workers, observers, previews and accepted-request recovery never install. -New workers require default-revision -AW source/contract verification and active registration. +Submission initializes Policy/Work. Only trusted submit/dispatch install +pools/profiles; other roles cannot. New workers require default-revision AW +source/contract checks and active registration. `PoolInstall.config_ref` records provenance, not authority. Preserve installed profiles/limits/epoch. Standalone seeding is unsupported; AW/Actions authorize submissions. @@ -69,24 +68,24 @@ events; disabling removes it. ## Plan and dispatch -- Read immutable snapshots; stale predictions cannot override Policy. - Submit bounded, secret-free plans within scheduling limits. +- Read immutable snapshots; predictions cannot override Policy. Submit bounded, + secret-free plans within limits. - Request a pool prefix with `work_queue_dispatch_next` (`pool`, `max_claims`, `max_dispatches`). Never choose winners/targets or use scalar dispatch. -- Trusted processing refreshes the ledger and commits a compatible fair prefix. +- Trusted processing refreshes the ledger and commits a fair compatible prefix. CAS losers discard tentative choices/charges and recompute. ESLint/daily-report/conformance agents only explain staged plans. Trusted controls validate batches; empty dispatcher intents fail. Unassigned dispatchers cannot emit `noop` or `report_incomplete`. -## Execute original Claims +## Execute Claims -- Process only the assignment's original Claims and stored plans. Scope every - safe output and finish intent to one original Claim. +- Process only assigned original Claims and stored plans. Scope each safe output + and finish intent to one original Claim. - Omit `claim_handle` only for an originally single-Claim assignment. - Multi-Claim assignments require it on every message even after others - close. Reject malformed, null, foreign or conflicting selectors. + Multi-Claim assignments require it on every message, even after others close. + Reject malformed, null, foreign or conflicting selectors. - Finish members with `work_queue_claim_finish`, using `outcome: "completed"` or `"cancelled"`. Missing finish authorizes no effects; valid siblings settle independently. @@ -96,46 +95,46 @@ Unassigned dispatchers cannot emit `noop` or `report_incomplete`. Pending/failed delivery blocks Claims; never rerun uncertain effects. - Default to one Claim; batch compatible Work with reusable setup. Native batch conclusions are not per-Work Results. -- Treat finish as an intent: trusted processing publishes Completion before - scoped effects and Result only after independently verified delivery. +- Treat finish as intent: trusted processing publishes Completion before scoped + effects and Result only after independently verified delivery. - Beyond built-in safe outputs, declare `safe-outputs.claim-adapters` ([reference](safe-outputs-runtime.md)); each adapter binds one effect-type to an independently verified result, never job success alone. `` takes one JSON argument. `uninitialized` plus null SHA is empty. -Policyless ledgers fail; workers require Policy; `status: "staged"` is no grant. +Policyless ledgers fail; workers require Policy; `status: "staged"` grants nothing. ## Dependencies and recovery Require fresh Issue completed-state/PR merge evidence. Predecessors consume no Claims; successors wait for predecessor Results. -Lost responses, dispatcher cancellation/deadlines never prove nonlaunch/termination. -Retain reservations until exact evidence. Never rerun uncertain effects after -Completion; bounded verification yields Result/DeliveryFailure. Compatible +Lost responses, cancellation or deadlines never prove nonlaunch/termination. +Retain reservations until exact evidence; never rerun uncertain effects after +Completion. Bounded verification yields Result/DeliveryFailure. Compatible deployments preserve obligations, debt, Results, frozen assignments and explicit `execution_ref` pins. Unavailable workers pause affected Work only. Scheduled -maintenance terminally cancels unclaimed, unpinned Work whose admission contract -differs from its approved active worker contract, with reason -`worker_contract_obsolete`. Work with delivery barriers is preserved. -Quiesce economic Policy changes. Reject unsupported protocols/scalar assignments. +maintenance terminally cancels unclaimed, unpinned Work when its admission +contract differs from the approved active worker contract +(`worker_contract_obsolete`). Preserve Work with delivery barriers. +Quiesce economic Policy changes; reject unsupported protocols/scalar assignments. Retries/checkpoints preserve submission identity; never delete queues to retry admission. -## Load details only when needed +## Details -- Operator commands, TUI and diagnostic artifacts: +- Operator commands, TUI, diagnostics: [queue reference](../../docs/src/content/docs/reference/work-queue.md). Cancellation terminates Work, not workers; reconcile separately. `gh aw work-queue --repo OWNER/REPO stats --json`. `dispatch-next` grants - reservations, never launches; use the authorized dispatcher. + reservations, never launches; use authorized dispatcher. - Destructive: `gh aw work-queue clear --reason CODE [--dry-run | --hard]` requires repo maintain/admin, forbids CI, and requires interactive confirmation of the exact snapshot preview; no bypass. `--hard` replaces the ledger, forgetting Work/Claim identities, retry history and fairness debt. Admission/grants must already be paused, without open Claims/pending deliveries. See [hard reset](../../docs/src/content/docs/reference/work-queue.md#hard-reset). -- Daily report rotation and Claims: +- Daily report rotation/Claims: [portfolio example](../../docs/src/content/docs/gallery/daily-report-portfolio.md) and [shared worker instructions](../workflows/shared/daily-report-worker.md). - Contracts: [specification](../../docs/src/content/docs/specs/work-queue-specification.md#91-implementation-coverage-and-remaining-requirements). -- Models/fixtures/bounded verification: [formal reference](../../specs/work-queue/README.md). +- Models/fixtures: [formal reference](../../specs/work-queue/README.md). From ea58e269e120ad3c22ee8dac99c6af9a137423d5 Mon Sep 17 00:00:00 2001 From: pelikhan Date: Sun, 8 Nov 2026 21:15:55 -0800 Subject: [PATCH 5/5] fix(work-queue): keep agent instructions within word budget Compress queue guidance while preserving cancellation semantics and safety boundaries to satisfy the existing 800-word documentation limit. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .github/aw/work-queue.md | 81 ++++++++++++++++++++-------------------- 1 file changed, 40 insertions(+), 41 deletions(-) diff --git a/.github/aw/work-queue.md b/.github/aw/work-queue.md index 6e26a003106..cb18065faa5 100644 --- a/.github/aw/work-queue.md +++ b/.github/aw/work-queue.md @@ -6,13 +6,13 @@ description: Git-backed work queue agent instructions. `tools.work-queue` schedules DAGs and Claim-scoped effects. `work-queue.jsonl` is authoritative; V3 omits storage selectors. Issues/PRs are dependencies. -For issue-backed WorkQueueOps: [WorkQueueOps](../../docs/src/content/docs/patterns/workqueue-ops.md). +For issue-backed WorkQueueOps: [deployment](../../docs/src/content/docs/patterns/workqueue-ops.md). ## Select the role - **Observer:** explain only; never submit, dispatch or finish Work. - **Producer/dispatcher:** submit trusted plans/request bounded pool work; - queue authority grants no arbitrary resource writes. + no arbitrary resource writes. - **Worker:** require compiler-supplied version-3 `claims`; never override reserved `work_queue_assignment` or caller context. @@ -34,32 +34,31 @@ Dispatchers require `tools.work-queue: true`, a worker allowlist, bounded native runs; ignore moved `target-ref`. `aw.json` `work_queue`: concurrency, pending limits, retries, pools, fairness, -Issues. Defaults: weighted-priority, concurrency 16, -pending 4,096, singleton assignments, three attempts, 30-second backoff. -Submission initializes Policy/Work. Trusted submit/dispatch install pools/profiles; +Issues. Defaults: weighted-priority, concurrency 16, pending 4,096, +singleton assignments, three attempts, 30-second backoff. +Submission initializes Policy/Work. Only trusted submit/dispatch install pools/profiles; workers, observers, previews and accepted-request recovery never install. -New workers require default-revision -AW source/contract verification and active registration. +New workers require default-revision AW source/contract verification and active registration. `PoolInstall.config_ref` records provenance, not authority. Preserve installed profiles/limits/epoch. Standalone seeding is unsupported; AW/Actions authorize submissions. -Quiesce economic/authority changes; apply with -`policy --from-config --epoch EPOCH`. Config edits never rewrite active state. +Quiesce economic/authority changes; apply: `policy --from-config --epoch EPOCH`. +Config edits never rewrite active state. Never expose queue-write credentials to agents or snapshot MCP. Repository access rules are GitHub's responsibility. See [deployment](../../docs/src/content/docs/patterns/workqueue-ops.md#deploy-a-native-queue). ## Mirror admitted Work with Issues -`aw.json` `work_queue.issues: true` mirrors admitted Work with purple `work` / -`work:` labels. `label` customizes both prefixes; `title_prefix` defaults -to `[work queue]`; `orphaned_after_days` defaults to 30. Titles/summaries show -12-character Work IDs; bodies use full IDs. -Protected AW hooks alone project admissions/Claims. Immutable `backing_issue` -binds one Work per Issue. Agents never write mirrors; human edits never establish +`aw.json` `work_queue.issues: true` mirrors admitted Work: purple `work` / +`work:` labels. `label` customizes both prefixes. Defaults: `title_prefix` = +`[work queue]`, `orphaned_after_days` = 30. Titles/summaries: 12-character Work IDs; +bodies: full IDs. +Only protected AW hooks project admissions/Claims. Immutable `backing_issue` +binds one Work/Issue. Agents never write mirrors; human edits never establish Results. Pre-existing Issues require exact installed `backing_issues` grants. Only admission-time `completion_policy` closes verified non-PR results. Maintenance reconciles at most 256 same-repository mirrors/run from the checked -ledger and flags aged orphan Issues for review, leaving them open. Keep uncertain +ledger; flag aged orphan Issues for review without closing. Keep uncertain writes pending; never recreate mirrors or repair untracked Issues. Recovery: `work-queue sync-issues --request-id ID`. Upgrade closed-schema readers. See @@ -71,7 +70,7 @@ events; disabling removes it. - Read immutable snapshots; stale predictions cannot override Policy. Submit bounded, secret-free plans within scheduling limits. -- Request a pool prefix with `work_queue_dispatch_next` (`pool`, `max_claims`, +- Request prefixes: `work_queue_dispatch_next` (`pool`, `max_claims`, `max_dispatches`). Never choose winners/targets or use scalar dispatch. - Trusted processing refreshes the ledger and commits a compatible fair prefix. CAS losers discard tentative choices/charges and recompute. @@ -82,25 +81,25 @@ Unassigned dispatchers cannot emit `noop` or `report_incomplete`. ## Execute original Claims -- Process only the assignment's original Claims and stored plans. Scope every - safe output and finish intent to one original Claim. +- Process only assigned original Claims/stored plans. Scope every safe output + and finish intent to one original Claim. - Omit `claim_handle` only for an originally single-Claim assignment. - Multi-Claim assignments require it on every message even after others + Multi-Claim assignments require it on every message even after siblings close. Reject malformed, null, foreign or conflicting selectors. - Finish members with `work_queue_claim_finish`, using `outcome: "completed"` or `"cancelled"`. Missing finish authorizes no effects; valid siblings settle independently. -- Retain Work with `outcome: "completed", keep_open: true` (default: false; +- Retain Work: `outcome: "completed", keep_open: true` (default: false; invalid with cancellation). Preserve both on retries. Verified delivery permits - another fair Claim, not successor release/backing Issue closure. + another fair Claim, never successor release/backing Issue closure. Pending/failed delivery blocks Claims; never rerun uncertain effects. -- Default to one Claim; batch compatible Work with reusable setup. Native +- Default: one Claim; batch compatible Work with reusable setup. Native batch conclusions are not per-Work Results. -- Treat finish as an intent: trusted processing publishes Completion before +- Finish is an intent: trusted processing publishes Completion before scoped effects and Result only after independently verified delivery. -- Beyond built-in safe outputs, declare `safe-outputs.claim-adapters` +- Beyond built-ins, declare `safe-outputs.claim-adapters` ([reference](safe-outputs-runtime.md)); each adapter binds one effect-type to - an independently verified result, never job success alone. + independently verified results, never job success alone. `` takes one JSON argument. `uninitialized` plus null SHA is empty. Policyless ledgers fail; workers require Policy; `status: "staged"` is no grant. @@ -108,37 +107,37 @@ Policyless ledgers fail; workers require Policy; `status: "staged"` is no grant. ## Dependencies and recovery Require fresh Issue completed-state/PR merge evidence. Predecessors consume no -Claims; successors wait for predecessor Results. +Claims; successors await predecessor Results. Lost responses, dispatcher cancellation/deadlines never prove nonlaunch/termination. Retain reservations until exact evidence. Never rerun uncertain effects after Completion; bounded verification yields Result/DeliveryFailure. Compatible deployments preserve obligations, debt, Results, frozen assignments and explicit `execution_ref` pins. Unavailable workers pause affected Work only. Scheduled -maintenance terminally cancels unclaimed, unpinned Work whose admission contract -differs from its approved active worker contract, using `WorkCancellation` with -operation `version: 1` and reason `worker_contract_obsolete`. Current-time -expiration excludes elapsed Work; publication rechecks eligibility. Work with -delivery barriers is preserved. Version-3 history without an operation version -retains reason-only semantics, including compatible/pinned cancellations. +maintenance terminally cancels unclaimed, unpinned Work with admission/approved-active-worker +contract mismatch: `WorkCancellation`, operation `version: 1`, reason +`worker_contract_obsolete`. Current-time expiration excludes elapsed Work; +publication rechecks eligibility. Preserve delivery-barrier Work. Version-3 +history lacking operation version retains reason-only semantics, including +compatible/pinned cancellations. Quiesce economic Policy changes. Reject unsupported protocols/scalar assignments. Retries/checkpoints preserve submission identity; never delete queues to retry admission. ## Load details only when needed -- Operator commands, TUI and diagnostic artifacts: +- Commands, TUI, diagnostics: [queue reference](../../docs/src/content/docs/reference/work-queue.md). Cancellation terminates Work, not workers; reconcile separately. `gh aw work-queue --repo OWNER/REPO stats --json`. `dispatch-next` grants - reservations, never launches; use the authorized dispatcher. + reservations, never launches; use authorized dispatchers. - Destructive: `gh aw work-queue clear --reason CODE [--dry-run | --hard]` - requires repo maintain/admin, forbids CI, and requires interactive confirmation - of the exact snapshot preview; no bypass. `--hard` replaces the ledger, - forgetting Work/Claim identities, retry history and fairness debt. Admission/grants - must already be paused, without open Claims/pending deliveries. + requires repo maintain/admin, forbids CI, requires interactive confirmation + of exact snapshot previews; no bypass. `--hard` replaces the ledger, + forgetting Work/Claim identities, retries and fairness debt. Pause admission/grants + first, without open Claims/pending deliveries. See [hard reset](../../docs/src/content/docs/reference/work-queue.md#hard-reset). -- Daily report rotation and Claims: +- Daily-report rotation/Claims: [portfolio example](../../docs/src/content/docs/gallery/daily-report-portfolio.md) - and [shared worker instructions](../workflows/shared/daily-report-worker.md). + and [worker instructions](../workflows/shared/daily-report-worker.md). - Contracts: [specification](../../docs/src/content/docs/specs/work-queue-specification.md#91-implementation-coverage-and-remaining-requirements). - Models/fixtures/bounded verification: [formal reference](../../specs/work-queue/README.md).