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..0bf86fbf5c8 100644 --- a/.github/aw/work-queue.md +++ b/.github/aw/work-queue.md @@ -6,17 +6,17 @@ 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. +- **Producer/dispatcher:** submit trusted, bounded plans/request pool work; + no arbitrary resource 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. @@ -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,68 +70,74 @@ 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. +- 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/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. +`` takes one JSON argument. `uninitialized` + null SHA is empty. +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. +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 +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/incompatible workers pause affected Work only. +`execution_ref` pins. Unavailable workers pause affected Work only. Scheduled +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). +- [Formal reference](../../specs/work-queue/README.md). diff --git a/actions/setup/js/reconcile_work_queue_run.cjs b/actions/setup/js/reconcile_work_queue_run.cjs index 4c6dec29919..573a1503617 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 deploymentControl = require("./work_queue_deployment_control.cjs"); const reconciliationError = message => { const error = new Error(`${ERR_VALIDATION}: ${message}`); @@ -113,6 +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 }; + 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; @@ -135,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, 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 a33b96efe72..bbcdcc28a0a 100644 --- a/actions/setup/js/reconcile_work_queue_run.test.cjs +++ b/actions/setup/js/reconcile_work_queue_run.test.cjs @@ -286,6 +286,121 @@ 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, 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_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.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..dd50d3ab539 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 { 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"; -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,173 @@ 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(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(["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([ ["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..b63590092e4 100644 --- a/actions/setup/js/work_queue_deployment_control.cjs +++ b/actions/setup/js/work_queue_deployment_control.cjs @@ -1,9 +1,31 @@ // @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"); +const { expirationView } = require("./work_queue_expiration.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 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", 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); + 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 +99,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..2aa64d4b5cb 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, @@ -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,6 +573,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.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); if ( @@ -800,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 2108c66927f..eee2312c37e 100644 --- a/docs/src/content/docs/reference/work-queue.md +++ b/docs/src/content/docs/reference/work-queue.md @@ -118,6 +118,10 @@ 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 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 `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/docs/src/content/docs/specs/work-queue-specification.md b/docs/src/content/docs/specs/work-queue-specification.md index 3a49f53e48a..cbdf9e83796 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 @@ -3342,6 +3359,7 @@ unexecuted formal, supported-host, performance, or deployment-security gate. | Protected launch credentials, immutable effect targets and native delivery verification | Implemented with exact selected-client/profile proofs, immutable Work/profile/ancestor target intersection and private native readback. Both public and compiler control entry points reject missing/invalid protected launch metadata before client construction or queue publication, while previews and submit-only controls remain credential-independent. Local positive and refusal regressions pass; writer deployment automation is separate | | Trusted AW settings and first-submit bootstrap | Compiler, JavaScript and native Go share optional global scheduling/Issue settings and default-only atomic Policy-and-Work bootstrap. AW authorizes participants and approved targets; no producer enrollment, worker-principal settings or administrator seeding is required. Installed economic Policy stays authoritative. Local tests cover defaults, overrides, malformed/conflicting settings, multiple participants, concurrent bootstrap, selective route validation and duplicate-free replay. Hosted Git publication and immutable-SHA dispatch remain unverified live gates | | Worker-evolution bounded model | Trusted-AW bootstrap and prospective worker-evolution models cover contract compatibility, explicit pins, local readiness, CAS races, frozen completion/credentials and preservation of debt, Results, reservations and scheduling economics. Focused positive, exact negative and guarded witness configurations are registered in the [formal reference](https://github.com/github/gh-aw/blob/main/specs/work-queue/README.md). These bounded models are not a proof of arbitrary Go/JavaScript or hosted API refinement; earlier captures are not rewritten | +| Obsolete retirement and historical cancellation compatibility | Native Go/JavaScript and operator regressions cover checked operation versions, legacy reason-only history, immutable terminal metadata, checkpoint/retry identity, expiration-sensitive selection and independent maintenance recovery. `ObsoleteWorkRetirement` adds two bounded safety configurations, thirteen exact mutation controls and four safety-guarded reachability witnesses. Approved routes, byte ordering and timestamps remain premises; the abstract checkpoint is not a codec proof, and recovery attempts do not establish live delivery or liveness. See the [retirement model](https://github.com/github/gh-aw/blob/main/specs/work-queue/README.md#obsolete-work-retirement-model) | | Worker-evolution runtime and compiler integration | Go/JavaScript implement durable deployment CAS, frozen admission contracts, immutable execution pins, caller-local approval and readiness, historical availability updates, checkpoint replay and frozen completion/delivery/reconciliation. Compiler stamps use complete worker authority frontmatter and imports; chunked proposals reconstruct exactly below Actions' per-value limit. Actual compiled ESLint bootstrap admits three Work items and launches three mocked native workers without duplicate replay. Full native queue/setup, scoped CLI/compiler, 1,053 JavaScript tests, typecheck and 334-workflow drift checks pass. Hosted execution remains unverified; the aggregate publication gate retains independently baseline-confirmed custom Go lint findings | Authenticated ingestion, actual run/attempt binding, Claim/resource checks and 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.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..2f6885bb288 100644 --- a/pkg/workqueue/deployment_test.go +++ b/pkg/workqueue/deployment_test.go @@ -65,6 +65,55 @@ 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", "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) + } + assertOperatorJSProjection(t, commits) + 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/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 8295ad4a9b2..08a9bc58e79 100644 --- a/pkg/workqueue/replay.go +++ b/pkg/workqueue/replay.go @@ -319,10 +319,22 @@ 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"` + Version int `json:"version"` + } if err := json.Unmarshal(operation, &value); err != nil { return err } + work := state.Works[value.WorkID] + 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/BrokenRetirementAvailability.cfg b/specs/work-queue/BrokenRetirementAvailability.cfg new file mode 100644 index 00000000000..58726d4fcbe --- /dev/null +++ b/specs/work-queue/BrokenRetirementAvailability.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "unavailable" +INVARIANT CancellationAuthority +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/BrokenRetirementBarrier.cfg b/specs/work-queue/BrokenRetirementBarrier.cfg new file mode 100644 index 00000000000..7f3ca5f815d --- /dev/null +++ b/specs/work-queue/BrokenRetirementBarrier.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "barrier" +INVARIANT CancellationAuthority +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/BrokenRetirementCheckpoint.cfg b/specs/work-queue/BrokenRetirementCheckpoint.cfg new file mode 100644 index 00000000000..bb16a66d5bc --- /dev/null +++ b/specs/work-queue/BrokenRetirementCheckpoint.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "checkpoint-version" +INVARIANT CheckpointMetadata +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/BrokenRetirementDebt.cfg b/specs/work-queue/BrokenRetirementDebt.cfg new file mode 100644 index 00000000000..89020b3aaf5 --- /dev/null +++ b/specs/work-queue/BrokenRetirementDebt.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "debt" +INVARIANT ProtectedState +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/BrokenRetirementExpiry.cfg b/specs/work-queue/BrokenRetirementExpiry.cfg new file mode 100644 index 00000000000..9f587818c67 --- /dev/null +++ b/specs/work-queue/BrokenRetirementExpiry.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "selection-expiry" +INVARIANT SelectionExpiration +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/BrokenRetirementLegacy.cfg b/specs/work-queue/BrokenRetirementLegacy.cfg new file mode 100644 index 00000000000..97093514f78 --- /dev/null +++ b/specs/work-queue/BrokenRetirementLegacy.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "legacy-recheck" +INVARIANT LegacyReplayCompatibility +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/BrokenRetirementMaintenance.cfg b/specs/work-queue/BrokenRetirementMaintenance.cfg new file mode 100644 index 00000000000..7ca24f8692b --- /dev/null +++ b/specs/work-queue/BrokenRetirementMaintenance.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "maintenance-stop" +INVARIANT RecoveryIsolation +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/BrokenRetirementOwnership.cfg b/specs/work-queue/BrokenRetirementOwnership.cfg new file mode 100644 index 00000000000..2cbad6c4fe1 --- /dev/null +++ b/specs/work-queue/BrokenRetirementOwnership.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "owned" +INVARIANT CancellationAuthority +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/BrokenRetirementPin.cfg b/specs/work-queue/BrokenRetirementPin.cfg new file mode 100644 index 00000000000..d78bec7f9de --- /dev/null +++ b/specs/work-queue/BrokenRetirementPin.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "pin" +INVARIANT CancellationAuthority +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/BrokenRetirementPublication.cfg b/specs/work-queue/BrokenRetirementPublication.cfg new file mode 100644 index 00000000000..617642d08b6 --- /dev/null +++ b/specs/work-queue/BrokenRetirementPublication.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "stale-publication" +INVARIANT CancellationAuthority +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/BrokenRetirementReservation.cfg b/specs/work-queue/BrokenRetirementReservation.cfg new file mode 100644 index 00000000000..98a6bb5e67b --- /dev/null +++ b/specs/work-queue/BrokenRetirementReservation.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "reservation" +INVARIANT ProtectedState +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/BrokenRetirementResult.cfg b/specs/work-queue/BrokenRetirementResult.cfg new file mode 100644 index 00000000000..4e6429a1665 --- /dev/null +++ b/specs/work-queue/BrokenRetirementResult.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "result" +INVARIANT ProtectedState +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/BrokenRetirementTerminal.cfg b/specs/work-queue/BrokenRetirementTerminal.cfg new file mode 100644 index 00000000000..a21b115a78a --- /dev/null +++ b/specs/work-queue/BrokenRetirementTerminal.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "terminal-version" +INVARIANT TerminalMetadata +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/ObsoleteWorkRetirement.tla b/specs/work-queue/ObsoleteWorkRetirement.tla new file mode 100644 index 00000000000..3a6144e2769 --- /dev/null +++ b/specs/work-queue/ObsoleteWorkRetirement.tla @@ -0,0 +1,208 @@ +----------------------- MODULE ObsoleteWorkRetirement ----------------------- +EXTENDS Naturals, Sequences, FiniteSets, TLC + +\* Focused retirement refinement, not a Git/JSON codec or authentication proof. +\* Active revisions are already approved; original pins and admission contracts +\* are immutable. "expired" represents derived work_expired metadata, not an +\* explicit cancellation operation. Legacy/checked represent absent/version:1. +CONSTANTS BatchLimit, Broken +ASSUME /\ BatchLimit \in 1..2 + /\ Broken \in {"none", "selection-expiry", "stale-publication", "pin", + "owned", "barrier", "unavailable", "legacy-recheck", + "checkpoint-version", "terminal-version", "debt", + "reservation", "result", "maintenance-stop"} +Works == {"expired", "old", "pinned", "owned", "barrier", "other"} +\* Trusted UTF-8 order is a premise; native comparator/codec tests remain separate. +Order == <<"barrier", "expired", "old", "other", "owned", "pinned">> +None == "none" +Forms == {"legacy", "checked"} +Deadline(w) == IF w = "expired" THEN 1 ELSE 3 +Pinned(w) == w = "pinned" +SeqSet(s) == {s[i] : i \in 1..Len(s)} +Effective(q, at) == IF q.time > at THEN q.time ELSE at +Due(q, w, at) == q.work[w] = "available" /\ at >= Deadline(w) + /\ q.barrier[w] \notin {"pending", "failed"} +Expire(q, at) == + LET time == Effective(q, at) + due == {w \in Works : Due(q, w, time)} + IN [q EXCEPT !.time = time, + !.work = [w \in Works |-> IF w \in due THEN "cancelled" ELSE q.work[w]], + !.receipt = [w \in Works |-> IF w \in due THEN "expired" ELSE q.receipt[w]], + !.original = [w \in Works |-> IF w \in due THEN "expired" ELSE q.original[w]]] + +VARIABLE state +vars == <> +Mismatch(w) == w # "other" /\ state.contract = "new" +Obsolete(q, w) == + /\ q.work[w] = "available" /\ ~Pinned(w) /\ w \notin q.claims + /\ q.barrier[w] = None + /\ state.present /\ state.available /\ Mismatch(w) +MayRetire(q, w) == + /\ q.work[w] = "available" \/ (Broken = "owned" /\ q.work[w] = "claimed") + /\ ~Pinned(w) \/ Broken = "pin" + /\ w \notin q.claims \/ Broken = "owned" + /\ q.barrier[w] = None \/ Broken = "barrier" + /\ (state.present /\ state.available) \/ Broken = "unavailable" + /\ Mismatch(w) +Prefix(q, checked) == + LET eligible == SelectSeq(Order, LAMBDA w : + IF checked THEN Obsolete(q, w) ELSE MayRetire(q, w)) + IN SubSeq(eligible, 1, IF Len(eligible) > BatchLimit + THEN BatchLimit ELSE Len(eligible)) +Retire(q, batch) == + LET selected == SeqSet(batch) + IN [q EXCEPT + !.work = [w \in Works |-> IF w \in selected THEN "cancelled" ELSE q.work[w]], + !.receipt = [w \in Works |-> IF w \in selected THEN "checked" ELSE q.receipt[w]], + !.original = [w \in Works |-> IF w \in selected /\ q.original[w] = None + THEN "checked" ELSE q.original[w]], + !.charges = IF Broken = "debt" /\ selected # {} + THEN [w \in Works |-> 0] ELSE q.charges, + !.reservations = IF Broken = "reservation" /\ selected # {} + THEN {} ELSE q.reservations, + !.results = IF Broken = "result" /\ selected # {} THEN {} ELSE q.results] + +Init == + LET q == [work |-> [w \in Works |-> IF w = "owned" THEN "claimed" ELSE "available"], + barrier |-> [w \in Works |-> IF w = "barrier" THEN "failed" ELSE None], + claims |-> {"owned"}, reservations |-> {"owned"}, + charges |-> [w \in Works |-> IF w = "owned" THEN 1 ELSE 0], + results |-> {"historical"}, + receipt |-> [w \in Works |-> None], + original |-> [w \in Works |-> None], time |-> 0] + IN state = [queue |-> q, contract |-> "new", available |-> TRUE, + present |-> TRUE, updated |-> FALSE, now |-> 0, + batch |-> <<>>, expectedBatch |-> <<>>, phase |-> "idle", + failed |-> FALSE, recovered |-> FALSE, compacted |-> FALSE, + selectedExpired |-> FALSE, unsafe |-> {}, + legacyLoaded |-> FALSE, legacyRejected |-> FALSE, + snapshot |-> q, snapshotExpected |-> q, snapshotTaken |-> FALSE, + expectedCharges |-> q.charges, expectedReservations |-> q.reservations] + +Tick == + /\ state.now < 2 + /\ state' = [state EXCEPT !.now = @ + 1] + +Deploy(contract, available, present) == + /\ ~state.updated /\ contract \in {"old", "new"} + /\ available \in BOOLEAN /\ present \in BOOLEAN + /\ state' = [state EXCEPT !.contract = contract, !.available = available, + !.present = present, !.updated = TRUE] + +ClaimOld == + /\ state.queue.work["old"] = "available" /\ state.queue.barrier["old"] = None + /\ state.contract = "old" /\ state.present /\ state.available + /\ state' = [state EXCEPT !.queue.work["old"] = "claimed", + !.queue.claims = @ \cup {"old"}, !.queue.reservations = @ \cup {"old"}, + !.queue.charges["old"] = 1, !.expectedCharges["old"] = 1, + !.expectedReservations = @ \cup {"old"}] + +HoldDelivery == + \* Abstract retained completion with a closed historical Claim. + /\ state.queue.work["old"] = "available" /\ state.queue.barrier["old"] = None + /\ state' = [state EXCEPT !.queue.barrier["old"] = "pending"] + +Prepare == + /\ state.phase = "idle" + /\ LET fresh == Expire(state.queue, state.now) + view == IF Broken = "selection-expiry" THEN state.queue ELSE fresh + batch == Prefix(view, FALSE) + IN state' = [state EXCEPT !.batch = batch, !.expectedBatch = Prefix(fresh, TRUE), + !.phase = "selected", !.selectedExpired = + \E w \in SeqSet(batch) : Due(state.queue, w, Effective(state.queue, state.now))] + +Publish(infrastructureFailure) == + /\ state.phase = "selected" /\ infrastructureFailure \in BOOLEAN + /\ LET fresh == Expire(state.queue, state.now) + valid == (\A w \in SeqSet(state.batch) : MayRetire(fresh, w)) + \/ Broken = "stale-publication" + failed == infrastructureFailure \/ ~valid + IN state' = [state EXCEPT !.phase = "retired", !.failed = failed, + !.queue = IF failed THEN state.queue ELSE Retire(fresh, state.batch), + !.unsafe = IF failed THEN @ ELSE + @ \cup {w \in SeqSet(state.batch) : ~Obsolete(fresh, w)}] + +ReplayLegacy(w) == + /\ w \in {"pinned", "other"} /\ ~state.legacyLoaded + /\ state.queue.work[w] = "available" /\ state.queue.barrier[w] = None + /\ LET rejected == Broken = "legacy-recheck" /\ ~Obsolete(state.queue, w) + IN state' = [state EXCEPT !.legacyLoaded = TRUE, !.legacyRejected = rejected, + !.queue.work[w] = IF rejected THEN @ ELSE "cancelled", + !.queue.receipt[w] = IF rejected THEN @ ELSE "legacy", + !.queue.original[w] = IF rejected THEN @ ELSE "legacy"] + +Rewrite(w, form) == + /\ w \in Works /\ form \in Forms + /\ state.queue.work[w] = "cancelled" /\ state.queue.receipt[w] \in Forms + /\ form # state.queue.receipt[w] /\ Broken = "terminal-version" + /\ state' = [state EXCEPT !.queue.receipt[w] = form] + +Checkpoint == + /\ ~state.snapshotTaken + /\ \E w \in Works : state.queue.receipt[w] \in Forms + /\ state' = [state EXCEPT !.snapshotTaken = TRUE, !.snapshotExpected = state.queue, + !.snapshot = IF Broken = "checkpoint-version" + THEN [state.queue EXCEPT !.receipt = + [w \in Works |-> IF state.queue.receipt[w] = "checked" THEN "legacy" + ELSE IF state.queue.receipt[w] = "legacy" THEN "checked" + ELSE state.queue.receipt[w]]] + ELSE state.queue] + +RecoverCore == + /\ state.phase = "retired" + /\ LET stopped == Broken = "maintenance-stop" /\ state.failed + IN state' = [state EXCEPT !.phase = IF stopped THEN "done" ELSE "recovered", + !.recovered = ~stopped] +Compact == + /\ state.phase = "recovered" + /\ state' = [state EXCEPT !.phase = "done", !.compacted = TRUE] + +Next == Tick \/ ClaimOld \/ HoldDelivery \/ Prepare \/ Checkpoint \/ RecoverCore \/ Compact + \/ (\E contract \in {"old", "new"}, available \in BOOLEAN, present \in BOOLEAN : + Deploy(contract, available, present)) + \/ (\E failure \in BOOLEAN : Publish(failure)) + \/ (\E w \in Works : ReplayLegacy(w)) + \/ (\E w \in Works, form \in Forms : Rewrite(w, form)) +Spec == Init /\ [][Next]_vars + +TypeOK == + /\ state.now \in 0..2 /\ state.queue.time \in 0..2 + /\ state.contract \in {"old", "new"} + /\ state.available \in BOOLEAN /\ state.present \in BOOLEAN /\ state.updated \in BOOLEAN + /\ state.queue.work \in [Works -> {"available", "claimed", "cancelled"}] + /\ state.queue.barrier \in [Works -> {None, "pending", "failed"}] + /\ state.queue.receipt \in [Works -> Forms \cup {None, "expired"}] + /\ state.queue.original \in [Works -> Forms \cup {None, "expired"}] + /\ state.queue.claims \subseteq Works /\ state.queue.reservations \subseteq Works + /\ state.queue.charges \in [Works -> 0..1] + /\ state.queue.results = {"historical"} + /\ state.batch \in Seq(Works) /\ Len(state.batch) <= BatchLimit + /\ state.phase \in {"idle", "selected", "retired", "recovered", "done"} + /\ state.failed \in BOOLEAN /\ state.recovered \in BOOLEAN /\ state.compacted \in BOOLEAN + /\ state.selectedExpired \in BOOLEAN /\ state.unsafe \subseteq Works + /\ state.legacyLoaded \in BOOLEAN /\ state.legacyRejected \in BOOLEAN + /\ state.snapshotTaken \in BOOLEAN +SelectionExpiration == ~state.selectedExpired +DeterministicPrefix == state.batch = state.expectedBatch +CancellationAuthority == state.unsafe = {} +LegacyReplayCompatibility == ~state.legacyRejected +TerminalMetadata == state.queue.receipt = state.queue.original +CheckpointMetadata == state.snapshot = state.snapshotExpected +ProtectedState == state.queue.charges = state.expectedCharges + /\ state.queue.reservations = state.expectedReservations + /\ state.queue.results = {"historical"} +RecoveryIsolation == state.phase = "done" => state.recovered +Safety == TypeOK /\ SelectionExpiration /\ DeterministicPrefix + /\ CancellationAuthority /\ LegacyReplayCompatibility /\ TerminalMetadata + /\ CheckpointMetadata /\ ProtectedState /\ RecoveryIsolation + +\* Guarded reachability controls, not liveness or exactly-once effect proofs. +NoCheckedRetirement == ~(state.queue.receipt["old"] = "checked") +NoLegacyCheckpoint == ~(state.snapshotTaken /\ + state.snapshot.receipt["pinned"] = "legacy") +NoRetirementRecovery == ~(state.failed /\ state.phase = "done" + /\ state.recovered /\ state.compacted) +NoElapsedPrefix == ~(state.now >= 1 /\ state.phase = "done" + /\ state.queue.receipt["expired"] = "expired" + /\ state.queue.receipt["old"] = "checked") +============================================================================= diff --git a/specs/work-queue/README.md b/specs/work-queue/README.md index 8dc7306eec9..16f6e227a68 100644 --- a/specs/work-queue/README.md +++ b/specs/work-queue/README.md @@ -26,6 +26,8 @@ from the existing launch and worker models. verified Claim results and native reservation release. `ClaimDeliveryRecovery` refines released pending-delivery recovery, private receipt loss, authenticated cleanup and conditional maintenance progress. +`ObsoleteWorkRetirement` isolates checked backlog retirement, legacy +cancellation receipts, expiration-sensitive selection and failure isolation. The original `WorkQueue.tla` is retained as historical model evidence for [issue #64852](https://github.com/github/gh-aw/issues/64852). Its best-effort @@ -40,6 +42,7 @@ The design rationale and trade-offs are recorded in [ADR-64955](../../docs/adr/6 |---|---|---| | [Current operator interface](#current-operator-interface) | Point operators to the supported commands and deployment guidance. | Historical model details could be mistaken for current operator guidance. | | [Unclaimed Work expiration model](#unclaimed-work-expiration-model) | Check immutable deadlines without changing Claim authority. | Bounded deadline checks do not prove wall-clock scheduling or eventual service. | +| [Obsolete Work retirement model](#obsolete-work-retirement-model) | Check fresh retirement eligibility and preserve legacy receipts and core recovery. | Approved routes and deterministic byte ordering are premises, not authentication or codec proofs. | | [Retained Claim completion model](#retained-claim-completion-model) | Check retained completion, delivery barriers, and subsequent re-claim. | Abstract receipts do not prove external effects or runtime refinement. | | [Claim delivery health and native capacity](#claim-delivery-health-and-native-capacity) | Separate aggregate delivery reporting from per-Claim results and capacity release. | Trusted receipt and terminal evidence are model premises, not verified implementations. | | [Released Claim delivery recovery](#released-claim-delivery-recovery) | Check recovery of released Claims whose delivery remains pending. | Fair recovery depends on stable terminal evidence, writable storage, and successful retries. | @@ -109,6 +112,57 @@ physical pruning. The native conformance tests cover those runtime boundaries separately. No liveness property promises service before a finite expiration; the existing `QueueService` result remains conditional on continuous eligibility. +## Obsolete Work retirement model + +`ObsoleteWorkRetirement.tla` supplements the existing deployment and expiration +models without changing their configurations. Six Work nodes distinguish +expired, incompatible, pinned, claimed, delivery-barrier and compatible-profile +cases. The approved active route can change once; selection and publication can +observe different routes, ownership, delivery barriers or timestamps. + +Selection uses a disposable current-time expiration view and a deterministic +prefix bounded to one or two operations. Publication checks the fresh prefix +atomically. A route, Claim, barrier or expiration race rejects the whole batch; +publication/authentication failure likewise commits no retirement. Both outcomes +still permit the existing recovery and compaction phases to be attempted. +Recovery success and independently fatal core errors remain runtime concerns. + +The checked cancellation form represents operation `version: 1`. Historical +reason-only cancellations remain accepted for compatible or pinned Work. +Terminal metadata cannot change between the two forms, and an abstract checkpoint +preserves the marker. Expiration is derived `work_expired` metadata, not a +synthetic cancellation operation. Retirement preserves existing Claims, fairness +charges, reservations and historical Results; it does not authorize effect replay. + +| Configuration | Required verdict | +|---|---| +| `Retirement.cfg`, `RetirementSingle.cfg` | Exhaust bounded `Safety` for two- and one-operation prefixes. | +| `BrokenRetirementExpiry.cfg` | Reject selection from a stale, unexpired projection. | +| `BrokenRetirementPublication.cfg` | Reject publication that skips fresh eligibility checks. | +| `BrokenRetirementPin.cfg`, `BrokenRetirementOwnership.cfg`, `BrokenRetirementBarrier.cfg`, `BrokenRetirementAvailability.cfg` | Reject cancellation of protected or unavailable Work. | +| `BrokenRetirementLegacy.cfg`, `BrokenRetirementCheckpoint.cfg`, `BrokenRetirementTerminal.cfg` | Reject retroactive legacy checks, checkpoint marker changes or terminal receipt rewrites. | +| `BrokenRetirementDebt.cfg`, `BrokenRetirementReservation.cfg`, `BrokenRetirementResult.cfg` | Reject loss of charges, reservations or historical Results. | +| `BrokenRetirementMaintenance.cfg` | Reject retirement failures that skip core recovery. | +| `RetirementCheckedWitness.cfg`, `RetirementLegacyWitness.cfg`, `RetirementRecoveryWitness.cfg`, `RetirementExpirationWitness.cfg` | Reach checked retirement, legacy checkpoint preservation, recovery after failure and an elapsed-deadline prefix under `Safety`. | + +Official TLC 2.19 (pinned jar SHA below) with Java 21 exhausted **9,411 distinct +states at depth 11** for `Retirement.cfg` and **9,195 at depth 11** for +`RetirementSingle.cfg`. All thirteen mutation controls and four safety-guarded +witnesses produced their exact named counterexamples and expected exit statuses. + +```bash +TLA2TOOLS_JAR=/path/to/tla2tools.jar JAVA_BIN=/path/to/java \ + TLC_MODEL_FILTER=ObsoleteWorkRetirement bash specs/work-queue/check.sh +``` + +The runner requires exact named counterexamples and exit statuses for all +mutation controls and safety-guarded witnesses. These are bounded reachability +checks, not liveness or a Go/JavaScript refinement proof. Approved deployment +identity, immutable pins/contracts, deterministic UTF-8 ordering and trusted +timestamps are premises. The checkpoint is an abstract record copy, not a JSON +codec or Git history proof. The model does not establish actual API availability, +authentication, global scheduling fairness or successful recovery effects. + ## Retained Claim completion model `FairWorkQueue.tla` models optional `keep_open: boolean` on the finish request diff --git a/specs/work-queue/Retirement.cfg b/specs/work-queue/Retirement.cfg new file mode 100644 index 00000000000..351be337bc8 --- /dev/null +++ b/specs/work-queue/Retirement.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "none" +INVARIANT Safety +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/RetirementCheckedWitness.cfg b/specs/work-queue/RetirementCheckedWitness.cfg new file mode 100644 index 00000000000..706fb93da2e --- /dev/null +++ b/specs/work-queue/RetirementCheckedWitness.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "none" +INVARIANTS Safety NoCheckedRetirement +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/RetirementExpirationWitness.cfg b/specs/work-queue/RetirementExpirationWitness.cfg new file mode 100644 index 00000000000..232825835bd --- /dev/null +++ b/specs/work-queue/RetirementExpirationWitness.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "none" +INVARIANTS Safety NoElapsedPrefix +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/RetirementLegacyWitness.cfg b/specs/work-queue/RetirementLegacyWitness.cfg new file mode 100644 index 00000000000..fdd8b2f7560 --- /dev/null +++ b/specs/work-queue/RetirementLegacyWitness.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "none" +INVARIANTS Safety NoLegacyCheckpoint +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/RetirementRecoveryWitness.cfg b/specs/work-queue/RetirementRecoveryWitness.cfg new file mode 100644 index 00000000000..f9ee4ee1bf6 --- /dev/null +++ b/specs/work-queue/RetirementRecoveryWitness.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 2 Broken = "none" +INVARIANTS Safety NoRetirementRecovery +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/RetirementSingle.cfg b/specs/work-queue/RetirementSingle.cfg new file mode 100644 index 00000000000..0363a239f75 --- /dev/null +++ b/specs/work-queue/RetirementSingle.cfg @@ -0,0 +1,4 @@ +SPECIFICATION Spec +CONSTANTS BatchLimit = 1 Broken = "none" +INVARIANT Safety +CHECK_DEADLOCK FALSE diff --git a/specs/work-queue/check.sh b/specs/work-queue/check.sh index bb708fcd709..e225339ff1f 100644 --- a/specs/work-queue/check.sh +++ b/specs/work-queue/check.sh @@ -10,9 +10,9 @@ fi SPEC_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" case "${TLC_MODEL_FILTER:-}" in - ""|WorkQueue|FairWorkQueue|ClaimScopedWorker|ClaimDeliveryHealth|ClaimDeliveryRecovery|QueueService|QueueLifecycle|DispatchReservationRecovery|IssueProjection|IssueClosureCancellation|QueueBootstrap|QueuePoolInstall|WorkerEvolution|WorkerDeploymentBoundary|WorkExpiration) ;; + ""|WorkQueue|FairWorkQueue|ClaimScopedWorker|ClaimDeliveryHealth|ClaimDeliveryRecovery|QueueService|QueueLifecycle|DispatchReservationRecovery|IssueProjection|IssueClosureCancellation|QueueBootstrap|QueuePoolInstall|WorkerEvolution|WorkerDeploymentBoundary|WorkExpiration|ObsoleteWorkRetirement) ;; *) - echo "TLC_MODEL_FILTER must be WorkQueue, FairWorkQueue, ClaimScopedWorker, ClaimDeliveryHealth, ClaimDeliveryRecovery, QueueService, QueueLifecycle, DispatchReservationRecovery, IssueProjection, IssueClosureCancellation, QueueBootstrap, QueuePoolInstall, WorkerEvolution, WorkerDeploymentBoundary, or WorkExpiration when set." >&2 + echo "TLC_MODEL_FILTER must be WorkQueue, FairWorkQueue, ClaimScopedWorker, ClaimDeliveryHealth, ClaimDeliveryRecovery, QueueService, QueueLifecycle, DispatchReservationRecovery, IssueProjection, IssueClosureCancellation, QueueBootstrap, QueuePoolInstall, WorkerEvolution, WorkerDeploymentBoundary, WorkExpiration, or ObsoleteWorkRetirement when set." >&2 exit 1 ;; esac @@ -64,6 +64,25 @@ run_model BrokenExpirationWatermark CommittedTimeMonotonicity Invariant 12 WorkE run_model BrokenExpirationCheckpoint ProjectionSoundness Invariant 12 WorkExpiration run_model ExpirationRunningWitness NoRunningPastExpiry Invariant 12 WorkExpiration run_model ExpirationCheckpointWitness NoCheckpointExpiry Invariant 12 WorkExpiration +run_model Retirement "" Invariant 0 ObsoleteWorkRetirement +run_model RetirementSingle "" Invariant 0 ObsoleteWorkRetirement +run_model BrokenRetirementExpiry SelectionExpiration Invariant 12 ObsoleteWorkRetirement +run_model BrokenRetirementPublication CancellationAuthority Invariant 12 ObsoleteWorkRetirement +run_model BrokenRetirementPin CancellationAuthority Invariant 12 ObsoleteWorkRetirement +run_model BrokenRetirementOwnership CancellationAuthority Invariant 12 ObsoleteWorkRetirement +run_model BrokenRetirementBarrier CancellationAuthority Invariant 12 ObsoleteWorkRetirement +run_model BrokenRetirementAvailability CancellationAuthority Invariant 12 ObsoleteWorkRetirement +run_model BrokenRetirementLegacy LegacyReplayCompatibility Invariant 12 ObsoleteWorkRetirement +run_model BrokenRetirementCheckpoint CheckpointMetadata Invariant 12 ObsoleteWorkRetirement +run_model BrokenRetirementTerminal TerminalMetadata Invariant 12 ObsoleteWorkRetirement +run_model BrokenRetirementDebt ProtectedState Invariant 12 ObsoleteWorkRetirement +run_model BrokenRetirementReservation ProtectedState Invariant 12 ObsoleteWorkRetirement +run_model BrokenRetirementResult ProtectedState Invariant 12 ObsoleteWorkRetirement +run_model BrokenRetirementMaintenance RecoveryIsolation Invariant 12 ObsoleteWorkRetirement +run_model RetirementCheckedWitness NoCheckedRetirement Invariant 12 ObsoleteWorkRetirement +run_model RetirementLegacyWitness NoLegacyCheckpoint Invariant 12 ObsoleteWorkRetirement +run_model RetirementRecoveryWitness NoRetirementRecovery Invariant 12 ObsoleteWorkRetirement +run_model RetirementExpirationWitness NoElapsedPrefix Invariant 12 ObsoleteWorkRetirement run_model Recovery run_model QueueOrdering run_model BrokenCAS TerminalPersistence 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; } diff --git a/specs/work-queue/verify_models_test.py b/specs/work-queue/verify_models_test.py index 8270f38aa67..9a2a54cd3af 100644 --- a/specs/work-queue/verify_models_test.py +++ b/specs/work-queue/verify_models_test.py @@ -142,6 +142,39 @@ def test_expiration_configs_require_exact_verdicts(self): rejected = self.run_model(config, "Invariant TypeOK is violated", 12, model=model) self.assertNotEqual(rejected.returncode, 0) + def test_retirement_configs_require_exact_verdicts(self): + model = "ObsoleteWorkRetirement" + for config in ("Retirement", "RetirementSingle"): + accepted = self.run_model(config, "Model checking completed. No error", model=model) + self.assertEqual(accepted.returncode, 0, accepted.stderr) + for config, invariant in ( + ("BrokenRetirementExpiry", "SelectionExpiration"), + ("BrokenRetirementPublication", "CancellationAuthority"), + ("BrokenRetirementPin", "CancellationAuthority"), + ("BrokenRetirementOwnership", "CancellationAuthority"), + ("BrokenRetirementBarrier", "CancellationAuthority"), + ("BrokenRetirementAvailability", "CancellationAuthority"), + ("BrokenRetirementLegacy", "LegacyReplayCompatibility"), + ("BrokenRetirementCheckpoint", "CheckpointMetadata"), + ("BrokenRetirementTerminal", "TerminalMetadata"), + ("BrokenRetirementDebt", "ProtectedState"), + ("BrokenRetirementReservation", "ProtectedState"), + ("BrokenRetirementResult", "ProtectedState"), + ("BrokenRetirementMaintenance", "RecoveryIsolation"), + ("RetirementCheckedWitness", "NoCheckedRetirement"), + ("RetirementLegacyWitness", "NoLegacyCheckpoint"), + ("RetirementRecoveryWitness", "NoRetirementRecovery"), + ("RetirementExpirationWitness", "NoElapsedPrefix"), + ): + with self.subTest(config=config): + accepted = self.run_model(config, f"Invariant {invariant} is violated", 12, model=model) + self.assertEqual(accepted.returncode, 0, accepted.stderr) + for output, status in ((f"Invariant {invariant} is violated", 0), + ("Invariant TypeOK is violated", 12), + ("Parsing failed", 151)): + rejected = self.run_model(config, output, status, model=model) + self.assertNotEqual(rejected.returncode, 0) + def test_delivery_recovery_configs_require_exact_verdicts(self): model = "ClaimDeliveryRecovery" for config in ("DeliveryRecovery", "DeliveryRecoveryProgress"): @@ -199,6 +232,7 @@ def test_default_evidence_and_java_extraction_stay_repository_local(self): def test_all_new_configs_are_registered_with_real_invariants(self): runner = (SPEC / "check.sh").read_text() for model, pattern in (("ClaimDeliveryHealth", r"(?:Delivery|BrokenDelivery).*\.cfg"), + ("ObsoleteWorkRetirement", r"(?:Retirement.*|BrokenRetirement.*)\.cfg"), ("WorkExpiration", r"(?:WorkExpiration|BrokenExpiration.*|Expiration.*Witness)\.cfg"), ("ClaimDeliveryRecovery", r"(?:DeliveryRecovery.*|BrokenRecovery.*|Recovery(?:Released|Enriched|Retained|Unavailable)Witness)\.cfg"), ("QueueLifecycle", r"LifecycleReleasedPendingWitness\.cfg"),