diff --git a/applications/worker.py b/applications/worker.py index 861c6c37..53af2367 100644 --- a/applications/worker.py +++ b/applications/worker.py @@ -57,7 +57,7 @@ def complete_noop( class FileDispatchAuthority(Protocol): - """Application port for database-selected first-attempt File work.""" + """Application port for database-selected bounded-attempt File work.""" def claim(self) -> FileDispatchLease | FileDispatchNoWork: ... diff --git a/docs/decisions/0060-reclaim-expired-file-imports-with-bounded-retries.md b/docs/decisions/0060-reclaim-expired-file-imports-with-bounded-retries.md new file mode 100644 index 00000000..4c9c08cf --- /dev/null +++ b/docs/decisions/0060-reclaim-expired-file-imports-with-bounded-retries.md @@ -0,0 +1,73 @@ +--- +name: adr-0060-reclaim-expired-file-imports-with-bounded-retries +version: "1.0.0" +description: > + Reclaim expired scheduler-owned File imports with database-timed bounded + backoff while preserving the existing exact WorkerLease recovery path. +--- + +# 0060. Reclaim expired File imports with bounded retries + +- Status: accepted +- Date: 2026-07-26 +- Refines: ADR-0029, ADR-0041, ADR-0043, ADR-0055, ADR-0059 + +## Context + +Autonomous File dispatch can stop after leasing or between the acquired, +prepared, and indexed publication boundaries. The durable job retains enough +lineage to resume, but ADR-0059 deliberately leaves every expired scheduler +lease untouched. Without a bounded reclaim owner, one ordinary process stop can +leave an accepted upsert unpublished indefinitely. Letting the process choose a +tenant, job, retry time, or generation would weaken the function-only scheduler +boundary and duplicate the existing WorkerLease authority. + +## Decision + +The existing scheduler claim function first considers expired scheduler-owned +File imports, then delegates to its unchanged first-attempt selector. PostgreSQL +selects one globally eligible recovery with `FOR UPDATE SKIP LOCKED`, using the +database clock and a fixed exponential delay of 30, 60, then 120 seconds after +lease expiry. It may mint only generations two through four, so one first +attempt has at most three automatic reclaims. Generation four remains untouched +after expiry; this decision adds no dead-letter transition or operator requeue. + +Recovery revalidates the same exact current page, upsert observation, +SourceVersion/root capability, Membership version, receiver, and scan epoch as +first dispatch. It also holds the existing per-Source progress lock and advances +only the locked job to the exact next generation. The transaction preserves the +durable recovery boundary, clears prior redemption, replaces the nonce and +database timestamps, and appends one immutable digest-only `reclaimed` event. +The scheduler still has no table access and receives only the existing closed +lease/no-work result. A missing configured root or failed current authority +check yields no work and no mutation. + +The independent Supply worker consumes the replacement WorkerLease through the +existing publication-recovery state machine. Every older token is stale because +generation, nonce, and current job state are checked together. No new process, +queue, Runtime authority, or content-bearing retry path is introduced. + +## Consequences + +- Concurrent schedulers can advance an expired generation only once. +- Recovery work is preferred over fresh first attempts once its database-owned + delay has elapsed; ineligible or exhausted work does not block fresh work. +- A crash at any durable publication boundary resumes toward one active Revision + and one publication effect; the replaced lease has zero effect. +- Downgrade is refused while a scheduler-owned higher-generation lease remains, + because the previous schema cannot preserve its automatic-reclaim provenance. +- Retry exhaustion remains durable but operationally unresolved. +- The supporting partial-index build remains transactional with the migration + fences and therefore takes a relation lock. Before a production rollout, the + operator must measure the retained `file_import_job` population and index-build + lock duration against the deployment drain budget, using a maintenance window + when the measured bound is not acceptable. + +## Revisit trigger + +Revisit before adding terminal-failure retries, dead-letter state, exhaustion +transitions, operator remediation/requeue, configurable backoff, provider +polling, automatic page acceptance, or automatic upsert/delete ordering. Any +revision must preserve database-selected tenant routing, exact-generation lease +fencing, current authority revalidation, content-free scheduler results, and one +publication authority. diff --git a/docs/decisions/README.md b/docs/decisions/README.md index 4734db5e..2aab7609 100644 --- a/docs/decisions/README.md +++ b/docs/decisions/README.md @@ -49,6 +49,7 @@ kernel, capability separation, and publication visibility model. | File delete observation execution | [0057 — Execute current File deletes through tombstone authority](0057-execute-current-file-deletes-through-tombstone-authority.md) | One trusted exact locator is revalidated against the current complete v4 scan, derives effect identity server-side, calls the sole tombstone authority, and records one immutable exact binding | Provider deletion authority, caller-authored path/effect identity, stale scan deletion, a second visibility authority, or Runtime trust in Supply metadata | | Mixed File page upsert scheduling | [0058 — Schedule only upserts from mixed File pages](0058-schedule-only-upserts-from-mixed-file-pages.md) | Validate one complete current v4 page, then atomically schedule and replay its nonempty ordered upsert projection through the existing import path while preserving original ordinals | Scheduler delete/tombstone authority, implicit audience, automatic upsert/delete ordering, a second queue, or Runtime trust in Supply metadata | | Autonomous File dispatch | [0059 — Dispatch scheduled File imports through exact leases](0059-dispatch-scheduled-file-imports-through-exact-leases.md) | A function-only scheduler login atomically claims the oldest current page-scheduled upsert and mints the existing exact first-attempt WorkerLease | Caller tenant/job routing, broad Control credentials, direct scheduler table access, retry/reclaim, delete execution, or a second queue/process | +| Bounded File reclaim | [0060 — Reclaim expired File imports with bounded retries](0060-reclaim-expired-file-imports-with-bounded-retries.md) | The same function-only scheduler prefers database-timed expired work, revalidates exact current authority, and advances at most three higher WorkerLease generations through durable-boundary recovery | Caller-selected retry routing/timing/generation, terminal-failure retry, dead-letter or requeue authority, delete execution, Runtime authority, or a new process/queue | | Private delivery ingress | [0045 — Redeem private delivery evidence at ingress](0045-redeem-private-delivery-evidence-at-ingress.md) | One digest-only service/request/asker/audience/epoch-bound DeliveryEvidenceRef constructs private TrustedDeliveryContext inside the current UserActor transaction before content work | Raw trusted delivery facts on the wire, bearer persistence, application-role minting/table reads, alternate Runtime paths, or claiming later M2 carriers | | Exact Package egress | [0046 — Bind egress to one exact Package hop](0046-bind-egress-to-one-exact-package-hop.md) | One digest-only grant binds one exact audience-bound Package to one model or channel preflight hop and redeems atomically | Treating Package construction as disclosure authority, arbitrary content at egress, cross-hop reuse, or bypassing final policy | | Public OpenAPI v0 | [0047 — Freeze OpenAPI v0 through one Runtime path](0047-freeze-openapi-v0-through-one-runtime-path.md) | One public `/v0/resolve` schema and a hidden provisional v1 bridge share the same sealed Runtime; Package release lineage is read-only from the Learning-published active manifest | Two authorization compositions, caller-authored release facts, Runtime publication/fallback, or in-place mutation of historical snapshots | diff --git a/engine/persistence/__init__.py b/engine/persistence/__init__.py index ac9a7178..9ff08604 100644 --- a/engine/persistence/__init__.py +++ b/engine/persistence/__init__.py @@ -81,6 +81,7 @@ ) from engine.persistence.worker_jobs import ( DEFAULT_WORKER_LEASE_TTL_SECONDS, + FILE_DISPATCH_MAX_LEASE_GENERATION, MAX_WORKER_LEASE_TTL_SECONDS, FileDispatchLease, FileDispatchNoWork, @@ -162,6 +163,7 @@ "WorkerNoOpCompletion", "WorkerNoOpOutcome", "DEFAULT_WORKER_LEASE_TTL_SECONDS", + "FILE_DISPATCH_MAX_LEASE_GENERATION", "MAX_WORKER_LEASE_TTL_SECONDS", "assert_runtime_role", "assert_action_role", diff --git a/engine/persistence/schema_security_manifest.yaml b/engine/persistence/schema_security_manifest.yaml index d2d86abf..aef59831 100644 --- a/engine/persistence/schema_security_manifest.yaml +++ b/engine/persistence/schema_security_manifest.yaml @@ -1,5 +1,5 @@ { - "manifestVersion": "31.0.0", + "manifestVersion": "32.0.0", "controlOperations": [ { "name": "register_file_source", @@ -314,7 +314,15 @@ "serverCapabilityInput": "all-or-nothing configured logical File root assertion applied only after globally oldest eligible selection and again during final post-lock revalidation; an omitted root yields no-work and cannot redirect selection", "databaseOwnedTime": "full-precision clock_timestamp authority check refreshed after every progress-lock wait; protocol lease timestamps and immediate worker verification use the PostgreSQL clock normalized to whole seconds", "fixedTtlSeconds": 300, - "selection": "oldest current v3/v4 page-scheduled upsert ordered by database acceptance time, checkpoint sequence, page ordinal, change ordinal, Organization, Source, and job id; the claim holds the per-Source progress advisory lock, refreshes its statement snapshot, revalidates latest accepted scan epoch, and uses FOR UPDATE SKIP LOCKED authority-row fencing before leasing", + "automaticReclaimPolicy": { + "maximumReclaims": 3, + "maximumLeaseGeneration": 4, + "baseBackoffSeconds": 30, + "backoff": "30 * 2^(current lease generation - 1), measured after database-issued lease expiry", + "eligibleStates": ["leased", "running", "prepared", "ready"], + "exhaustedBehavior": "content-free no-work; later dead-letter/operator owner remains NOT_ACTIVE" + }, + "selection": "eligible expired recovery attempts are selected before first attempts; recovery orders by database-owned retry eligibility then original acceptance lineage and exact tenant/job tie-breakers, while first attempts retain the Issue #91 global order; both paths hold the per-Source progress advisory lock, refresh the statement snapshot, revalidate latest accepted scan epoch, and use FOR UPDATE SKIP LOCKED authority-row fencing before leasing", "currentAuthority": [ "active exact SourceVersion", "latest accepted scan epoch", @@ -322,7 +330,8 @@ "enabled exact File import ServicePrincipal" ], "atomicWrites": [ - "file_import_job" + "file_import_job", + "file_import_job_event" ] }, { @@ -5840,6 +5849,7 @@ "context_worker_activate_recoverable_file_publication", "context_worker_record_file_import_interruption" ,"context_scheduler_claim_file_import" + ,"context_scheduler_claim_first_file_import" ], "definerRoles": [ "context_engine_worker_lease_definer", @@ -5884,7 +5894,7 @@ ], "context_engine_file_dispatch_definer": [ "SELECT", - "UPDATE state, signing_key_version, lease_nonce_digest, lease_issued_at, lease_expires_at, lease_generation, dispatch_claimed" + "UPDATE state, signing_key_version, lease_nonce_digest, lease_issued_at, lease_expires_at, lease_generation, dispatch_claimed, recovery_from_state, lease_redeemed_at" ] }, "partitions": [], @@ -5905,6 +5915,8 @@ "PG-FILE-CHANGE-SUPERSESSION-083" ,"PG-FILE-DISPATCH-091" ,"PG-FILE-DISPATCH-CONCURRENCY-091" + ,"PG-FILE-RECLAIM-093" + ,"PG-FILE-RECLAIM-CONCURRENCY-093" ] }, { @@ -6171,6 +6183,22 @@ "context_engine_worker_lease_definer" ], "withCheck": "organization_id = NULLIF(current_setting('app.organization_id', true), '')::uuid" + }, + { + "name": "file_import_job_event_file_dispatch_definer_select", + "command": "SELECT", + "roles": [ + "context_engine_file_dispatch_definer" + ], + "using": "organization_id = NULLIF(current_setting('app.organization_id', true), '')::uuid" + }, + { + "name": "file_import_job_event_file_dispatch_definer_insert", + "command": "INSERT", + "roles": [ + "context_engine_file_dispatch_definer" + ], + "withCheck": "organization_id = NULLIF(current_setting('app.organization_id', true), '')::uuid" } ] }, @@ -6191,8 +6219,12 @@ "context_worker_index_file_publication", "context_worker_activate_recoverable_file_publication", "context_worker_record_file_import_interruption" + ,"context_scheduler_claim_file_import" + ], + "definerRoles": [ + "context_engine_worker_lease_definer", + "context_engine_file_dispatch_definer" ], - "definerRole": "context_engine_worker_lease_definer", "directTableMutationAllowed": false }, "permittedOperations": { @@ -6210,6 +6242,10 @@ "context_engine_worker_lease_definer": [ "SELECT", "INSERT" + ], + "context_engine_file_dispatch_definer": [ + "SELECT", + "INSERT" ] }, "retention": { @@ -6228,7 +6264,9 @@ "DB-001", "DB-004", "DB-008", - "PG-FILE-RECOVERY-027" + "PG-FILE-RECOVERY-027", + "PG-FILE-RECLAIM-093", + "PG-FILE-RECLAIM-CONCURRENCY-093" ] }, { diff --git a/engine/persistence/worker_jobs.py b/engine/persistence/worker_jobs.py index 979d49d4..12b609e2 100644 --- a/engine/persistence/worker_jobs.py +++ b/engine/persistence/worker_jobs.py @@ -39,6 +39,7 @@ DEFAULT_WORKER_LEASE_TTL_SECONDS: Final = 300 MAX_WORKER_LEASE_TTL_SECONDS: Final = 3600 +FILE_DISPATCH_MAX_LEASE_GENERATION: Final = 4 def _utc_now() -> datetime: @@ -165,14 +166,14 @@ class WorkerLeaseAuthorityUnavailable(RuntimeError): @dataclass(frozen=True, slots=True) class FileDispatchNoWork: - """Closed, content-free result when no first-attempt File job is eligible.""" + """Closed, content-free result when no automatic File work is eligible.""" status: Literal["no_work"] = field(default="no_work", init=False) @dataclass(frozen=True, slots=True, repr=False) class FileDispatchLease: - """Scheduler-minted exact first-attempt lease and internal routing facts.""" + """Scheduler-minted exact bounded-attempt lease and internal routing facts.""" token: WorkerLeaseToken = field(repr=False) organization_id: UUID = field(repr=False) @@ -191,8 +192,15 @@ def __post_init__(self) -> None: if type(self.source_ref) is not SourceRef: raise TypeError("File dispatch source must be SourceRef") _require_uuid("service_principal_id", self.service_principal_id) - if self.lease_generation != 1: - raise ValueError("first-attempt File dispatch requires generation one") + if ( + type(self.lease_generation) is not int + or not 1 + <= self.lease_generation + <= FILE_DISPATCH_MAX_LEASE_GENERATION + ): + raise ValueError( + "File dispatch lease exceeds the automatic generation budget" + ) issued_at = _require_utc("issued_at", self.issued_at) expires_at = _require_utc("expires_at", self.expires_at) if expires_at <= issued_at: @@ -219,7 +227,7 @@ def __repr__(self) -> str: class PostgreSQLFileDispatchAuthority: - """Claim the oldest eligible scheduled File import without tenant input.""" + """Claim the oldest eligible first or bounded recovery File attempt.""" __slots__ = ("_codec", "_configured_root_refs", "_scheduler_engine") @@ -245,7 +253,7 @@ def __init__( self._configured_root_refs = configured_root_refs def claim(self) -> FileDispatchClaim: - """Atomically fence one first attempt, then mint its exact existing token.""" + """Atomically fence one bounded attempt, then mint its exact token.""" nonce = generate_worker_lease_nonce() try: diff --git a/eval/catalogs/m0-security-evidence.yaml b/eval/catalogs/m0-security-evidence.yaml index c2662699..5e14c1b4 100644 --- a/eval/catalogs/m0-security-evidence.yaml +++ b/eval/catalogs/m0-security-evidence.yaml @@ -540,6 +540,21 @@ "layer": "runtime", "selector": "tests/integration/test_file_dispatch.py::test_independent_worker_process_dispatches_and_publishes_one_job" }, + { + "id": "PG-FILE-RECLAIM-093", + "layer": "postgres", + "selector": "tests/integration/test_file_dispatch.py::test_scheduler_reclaims_one_expired_dispatch_after_database_backoff" + }, + { + "id": "PG-FILE-RECLAIM-CONCURRENCY-093", + "layer": "postgres", + "selector": "tests/integration/test_file_dispatch.py::test_concurrent_schedulers_reclaim_one_expired_generation_once" + }, + { + "id": "PROC-FILE-RECLAIM-093", + "layer": "runtime", + "selector": "tests/integration/test_file_dispatch.py::test_scheduler_recovers_interrupted_publication_and_stales_old_lease" + }, { "id": "HTTP-FILE-MIXED-UPSERT-NO-DELETE-089", "layer": "runtime", @@ -686,12 +701,15 @@ "PG-FILE-MIXED-UPSERT-SCHEDULE-089", "PG-FILE-MIXED-UPSERT-REPLAY-089", "PG-FILE-DISPATCH-091", - "PG-FILE-DISPATCH-CONCURRENCY-091" + "PG-FILE-DISPATCH-CONCURRENCY-091", + "PG-FILE-RECLAIM-093", + "PG-FILE-RECLAIM-CONCURRENCY-093" ], "runtime": [ "RUNTIME-WORKER-LEASE-007", "HTTP-FILE-MIXED-UPSERT-NO-DELETE-089", - "PROC-FILE-DISPATCH-091" + "PROC-FILE-DISPATCH-091", + "PROC-FILE-RECLAIM-093" ] } }, diff --git a/eval/catalogs/security-catalog.schema.json b/eval/catalogs/security-catalog.schema.json index 991f7adb..6814622d 100644 --- a/eval/catalogs/security-catalog.schema.json +++ b/eval/catalogs/security-catalog.schema.json @@ -95,8 +95,8 @@ }, "activations": { "type": "array", - "minItems": 21, - "maxItems": 21, + "minItems": 22, + "maxItems": 22, "uniqueItems": true, "prefixItems": [ { @@ -1031,6 +1031,48 @@ "Runtime authorization from Supply scheduling or lease state" ] } + }, + { + "const": { + "issueRef": "#93", + "invariantRef": "WORKER-LEASE-007", + "carrier": "bounded autonomous reclaim of expired scheduled File upserts", + "status": "active_fail_closed", + "policyEpochScope": "not-runtime-authority", + "controlBoundary": "function-only scheduler login -> expired lease plus database-owned exponential backoff -> current page/acquisition/audience/receiver revalidation -> deterministic recovery-first SKIP LOCKED selector -> exact next-generation WorkerLease -> existing durable-boundary File worker resume", + "testEvidence": [ + { + "id": "PG-FILE-RECLAIM-093", + "surface": "tests/integration/test_file_dispatch.py::test_scheduler_reclaims_one_expired_dispatch_after_database_backoff tests/integration/test_file_dispatch.py::test_scheduler_does_not_reclaim_after_automatic_generation_budget tests/integration/test_file_dispatch.py::test_scheduler_reclaim_revalidates_current_membership_without_mutation", + "oracle": "The non-owner scheduler reclaims only after database expiry plus generation-derived backoff, atomically advances to the exact next generation with one digest-only event, returns content-free no-work before eligibility, after the generation-four budget, or after current Membership authority is revoked, and mutates nothing on refusal." + }, + { + "id": "PG-FILE-RECLAIM-CONCURRENCY-093", + "surface": "tests/integration/test_file_dispatch.py::test_concurrent_schedulers_reclaim_one_expired_generation_once", + "oracle": "Two scheduler-role authorities competing for one expired generation yield exactly one next-generation lease and one immutable reclaim event; the loser receives content-free no-work." + }, + { + "id": "PROC-FILE-RECLAIM-093", + "surface": "tests/integration/test_file_dispatch.py::test_scheduler_recovers_interrupted_publication_and_stales_old_lease", + "oracle": "Across acquired, prepared, and indexed durable boundaries, the scheduler and existing File worker resume with one higher-generation lease, the replaced lease has zero effect, and exactly one active Revision/publication effect remains." + } + ], + "deferredEvidence": [ + "dead-letter state, exhaustion transition, operator remediation, requeue, and retry observability", + "provider polling, automatic page acceptance, full resync, and automatic delete ordering" + ], + "futureCarriers": [ + "dead-letter and operator remediation owner", + "provider polling and full resync", + "explicit mixed-change ordering policy" + ], + "notActive": [ + "scheduler tenant, Source, job, attempt, timing, backoff, or generation choice", + "terminal failure retry, dead-letter transition, or operator requeue", + "manual import or delete recovery", + "Runtime authorization from Supply retry or lease state" + ] + } } ], "items": false @@ -1335,7 +1377,10 @@ "HTTP-FILE-MIXED-UPSERT-NO-DELETE-089", "PG-FILE-DISPATCH-091", "PG-FILE-DISPATCH-CONCURRENCY-091", - "PROC-FILE-DISPATCH-091" + "PROC-FILE-DISPATCH-091", + "PG-FILE-RECLAIM-093", + "PG-FILE-RECLAIM-CONCURRENCY-093", + "PROC-FILE-RECLAIM-093" ] }, "surface": { @@ -1384,7 +1429,8 @@ "#85", "#87", "#89", - "#91" + "#91", + "#93" ] }, "invariantRef": { @@ -1422,7 +1468,8 @@ "bounded durable File deletion observations without execution", "explicit current File delete execution through sole tombstone authority", "explicit current mixed File page upsert-projection scheduling", - "autonomous first-attempt dispatch of explicit scheduled File upserts" + "autonomous first-attempt dispatch of explicit scheduled File upserts", + "bounded autonomous reclaim of expired scheduled File upserts" ] }, "status": { @@ -1458,7 +1505,8 @@ "latest complete same-SourceVersion v4 baseline -> stable shallow File snapshot -> provider-authenticated upsert/delete ChangePage -> exact baseline-bound ContextControl PostgreSQL acceptance", "exact trusted Source/SourceVersion/page/ordinal locator -> current complete v4 scan revalidation -> server-derived Resource/event lineage -> existing tombstone authority -> immutable exact execution binding", "complete current v4 upsert/delete page validation -> trusted ContextControl explicit FileImportAudience -> nonempty original-ordinal upsert projection -> exact existing acquisition/import-job lineage -> existing WorkerLease and current-scan publication fences", - "function-only scheduler login -> current page/acquisition/audience/receiver eligibility -> deterministic SKIP LOCKED selector -> database-timed generation-one lease -> existing WorkerLease and File worker" + "function-only scheduler login -> current page/acquisition/audience/receiver eligibility -> deterministic SKIP LOCKED selector -> database-timed generation-one lease -> existing WorkerLease and File worker", + "function-only scheduler login -> expired lease plus database-owned exponential backoff -> current page/acquisition/audience/receiver revalidation -> deterministic recovery-first SKIP LOCKED selector -> exact next-generation WorkerLease -> existing durable-boundary File worker resume" ] }, "testEvidence": { diff --git a/eval/catalogs/security-invariants.yaml b/eval/catalogs/security-invariants.yaml index 8582bfb0..e6f00fb3 100644 --- a/eval/catalogs/security-invariants.yaml +++ b/eval/catalogs/security-invariants.yaml @@ -24,7 +24,8 @@ "#85", "#87", "#89", - "#91" + "#91", + "#93" ], "documentRefs": [ "README.md", @@ -53,9 +54,10 @@ "docs/decisions/0056-detect-file-deletions-without-tombstone-authority.md", "docs/decisions/0057-execute-current-file-deletes-through-tombstone-authority.md", "docs/decisions/0058-schedule-only-upserts-from-mixed-file-pages.md", - "docs/decisions/0059-dispatch-scheduled-file-imports-through-exact-leases.md" + "docs/decisions/0059-dispatch-scheduled-file-imports-through-exact-leases.md", + "docs/decisions/0060-reclaim-expired-file-imports-with-bounded-retries.md" ], - "reconciliation": "Issue #2 fixes the product and testing decisions, issue #5 requires exactly fifteen release invariants and twelve canonical acceptance fixtures, and ADR-0019 resolves the later nineteen-label prose expansion without weakening any safeguard. Issue #15 activates only Organization-level next-request resolve(Acquire) revocation evidence under REVOCATION-006: at that activation, PG-REVOCATION-006, RUN-006, and CACHE-002 are active while BLOB-002 and Continue, citation, Policy-Epoch-bound WorkerLease, production ContextAccessTicket/ActionTicket, audit, outbox, cleanup, finer-epoch, UI, and external-admin carriers remain future or NOT_ACTIVE; later issue records are authoritative for subsequently activated carriers. Issue #16 activates only the M0 refusal gate for unavailable Continue, profile-disabled OpenCitation, and server-owned unavailable Acquire plans: at that activation its real continuation, profile-enabled citation, federated/source-native, and File carriers remain future, while its Runtime and HTTP refusal surfaces prove generic outcomes before content I/O; Issue #69 later activates the private/direct File profile-enabled citation carrier. Issue #17 activates only the signed one-shot persistent no-op durable-job WorkerLease subcarrier under WORKER-LEASE-007. It binds one exact worker audience but no end-user delivery audience or Policy Epoch, and proves only LEASE-SIGNING-017, PG-WORKER-LEASE-NOOP-017, and WORKER-LEASE-REPLAY-007; Source, Resource, Revision, Policy Epoch, end-user delivery audience, idempotency, generation, business mutation, outbox, File publication, and the full ACCEPT-008 matrix remain deferred or NOT_ACTIVE. Issue #18 activates only distinct signed synthetic ContextAccessTicket Provider-read and ActionTicket no-op channel-action subcarriers under ACTION-SEPARATION-014, with current Organization-v0 Policy Epoch validation. TICKET-AUDIENCE-018 and PG-TICKET-EPOCH-018 do not activate production ContextProvider integration, ContextRuntime ticket integration, BotDelivery, full M2 ActionPlane.prepare/perform, a real Sender or external effect, payload/destination/approval/idempotency binding, durable one-shot/replay/reconciliation, or full ACCEPT-012 PASS; those remain future or NOT_ACTIVE. Issue #19 activates only the current Acquire authorized-only ContextRun and restricted delivered-empty DecisionAudit subcarrier under TRACE-REDACTION-012. DIGEST-019, RUN-LINEAGE-019, AUTHORIZED-RUN-019, and PG-TRACE-REDACTION-012 prove deterministic Package and Organization-bound query digests, retained-UserActor-transaction persistence, decisionRef resolution, redaction, and short-lived exact-Organization operator ticket reads with no application-role table access; the supported reader commits deletion before returning, while a direct caller rollback is not claimed as durable exactly-once redemption. Raw query retention, full ContextPackage body retention, unauthenticated transport failures as ContextRuns, cross-Organization analytics, and general observability redaction remain NOT_ACTIVE. Issue #48 activates only the current ACCEPT-002 authenticated HTTP Acquire Membership field-projection carrier under SCOPE-INTERSECTION-004, INDEX-NOT-AUTHORITY-005, and TRACE-REDACTION-012. PROP-FIELD-PROJECTION-048, PG-FIELD-PROJECTION-048, and HTTP-ACCEPT-002-048 bind one current Membership/version field ceiling to same-transaction FORCE-RLS reduction, the sealed AuthorizationKernel, AuthorizedProjection and Evidence integrity, and authorized-only ContextRun/audit persistence. General permission DSLs, caller-authored projection lists, CandidateRef or index field authority, production Provider/source-native ACL negotiation, Supply publication, File/Base field ACL, typed fields, Continue, and Issue #20 runner substitution remain future or NOT_ACTIVE; Issue #69 later activates private/direct File OpenCitation through the same field-projection gates. Issue #63 activates only the digest-only private authenticated HTTP Acquire DeliveryEvidenceRef carrier under TRANSPORT-UNTRUSTED-008. PROP-DELIVERY-EVIDENCE-063, PG-DELIVERY-EVIDENCE-063, HTTP-DELIVERY-EVIDENCE-063, and FILE-DELIVERY-EVIDENCE-063 prove exact service/request/Organization/asker/Membership-version/destination/consumer/purpose/audience/epoch/lifetime binding, stable identical retry identity, role isolation, expiry cleanup, pre-content generic rejection, and one File-backed sealed Runtime delivery. Group AudienceSnapshot, group/public DeliveryEvidenceRef, production ModelGateway, ActionPlane, and the BotDelivery application remain future or NOT_ACTIVE; Issues #64, #66, #69, and #70 later activate the frozen OpenAPI, generated TypeScript SDK, private/direct OpenCitation, and deterministic private model-egress carriers. Issue #65 activates only one opaque digest-only model or channel EgressGrant after final Package policy, exact atomic PostgreSQL redemption and restricted audit, nominal BotDelivery inputs, and deterministic network-free ModelGateway or Sender-preflight spies under EGRESS-011. PROP-EGRESS-011, PG-EGRESS-011, and RUNTIME-EGRESS-011 prove exact Package/Organization/purpose/audience/epoch/hop/profile/lifetime binding and zero additional bytes on replay. Real model/provider calls, a real Sender or channel write, ActionTicket effects, group AudienceSnapshot revalidation, and the BotDelivery application process remain future or NOT_ACTIVE; Issue #64 later activates the generated SDK consumer and Issue #70 later activates the deterministic private TypeScript ModelGateway. Issue #66 activates the frozen public POST /v0/resolve OpenAPI carrier under TRANSPORT-UNTRUSTED-008. OPENAPI-CONTRACT-066, OPENAPI-BREAKING-066, HTTP-V0-066, and PG-RUNTIME-RELEASE-066 prove one public closed operation, deterministic immutable snapshot and breaking-change refusal, a hidden v1 bridge through the same handler and sealed Runtime path, and exact read-only observation of the active Learning-promoted release with fail-closed missing-release behavior before content work. A production BotDelivery caller, Continue redemption, MCP, group AudienceSnapshot, and external effects remain future or NOT_ACTIVE; Issues #64 and #69 later activate the generated TypeScript SDK and private/direct OpenCitation redemption through this frozen operation. Issue #64 activates only the packaged generated TypeScript POST /v0/resolve client under TRANSPORT-UNTRUSTED-008. SDK-CONTRACT-064 and SDK-LIVE-FILE-064 prove deterministic pinned generation, strict closed types, a narrow export map and metadata-only facade, installable tarball consumption, and one real PostgreSQL/File-backed Acquire through CandidateRef, AuthorizationKernel, AuthorizedProjection, ContextPackage, and opaque model egress grant. Issue #69 later extends SDK-LIVE-FILE-064 with a successful private/direct File OpenCitation through a second request-bound DeliveryEvidenceRef; Issue #70 extends the installed SDK fixture into deterministic Package-bound model generation; generated Continue remains generic unavailable. External package publication, production provider access, MCP, group AudienceSnapshot, real Continue redemption, and external effects remain future or NOT_ACTIVE. Issue #67 activates only private ActionPlane.prepare for create-placeholder, finalize-reply, and private-follow-up operation-specific tickets under ACTION-SEPARATION-014. PG-ACTION-PREPARE-067 proves exact current delivery, Organization, destination, audience, source, payload, approval, epoch, lifetime, and idempotency binding under a dedicated non-owner PostgreSQL role with digest-only FORCE-RLS persistence and zero effects. Issue #68 activates private ActionPlane.perform only through a deterministic Sender twin. PG-ACTION-PERFORM-068 proves one pre-Sender current-authority validation, one provider-attempt identity, immutable receipt replay, zero-effect ticket/payload mutation and stale-audience refusal, same-label cross-Organization isolation, and monotonic applied/rejected reconciliation including crash interleavings. Real provider or channel network effects, group AudienceSnapshot, compensation/delete, production BotDelivery orchestration, and the full ACCEPT-012 pass remain future or NOT_ACTIVE. The canonical set is IDs 001 through 012, 014, 015, and 019: CACHE-SCOPE-013 remains a preregistered conditional extension; AUDIENCE-016 is absorbed by SCOPE-INTERSECTION-004 and EGRESS-011; ACL-PROOF-017 is absorbed by INDEX-NOT-AUTHORITY-005 and REVOCATION-006; DELIVERY-EVIDENCE-018 is absorbed by TRANSPORT-UNTRUSTED-008. ACCEPT-001 through ACCEPT-012 follow ADR-0019's category order. Protected-asset references A-01 through A-08 refer, in order, to the eight bullets in the threat model's Protected assets section. Every expectedEvidence value below is a stable planned case identifier, not a claim that the case ran or passed; only an exact activation record upgrades named evidence, while fixture carrier status and the explicit M0 oracle preserve every other accepted-versus-active distinction. Issue #69 activates private/direct File CitationOpenRef issuance and OpenCitation under CITATION-AUTH-010: digest-only multi-use locators reveal only prior Package/Evidence and Fragment location lineage, every open obtains a current UserActor and trusted delivery context then traverses CandidateRef, AuthorizationKernel, AuthorizedProjection, a replacement ContextPackage, EgressGrant, ContextRun, and restricted DecisionAudit. PG-CITATION-AUTH-010, RUNTIME-CITATION-AUTH-010, and SDK-LIVE-FILE-064 prove A/B/A reauthorization, non-consumption on denial, database-clock expiry, cross-kind and cross-Organization opacity, and the generated SDK carrier. Group/public AudienceSnapshot, non-File providers, raw source URL locators, and Continue remain future or NOT_ACTIVE. Issue #70 activates only the private deterministic TypeScript ModelGateway under EGRESS-011. TS-MODEL-EGRESS-070, SDK-MODEL-EGRESS-070, and PG-MODEL-EGRESS-070 prove one current Package, exact grant redemption, closed provider input, bounded Package-subset citations, replay zero bytes, and digest-only retained audit. Real providers, streaming, group AudienceSnapshot, model-authored ActionPlane authority, and external effects remain future or NOT_ACTIVE. Issue #71 activates the complete private File-backed deterministic-twin BotDelivery carrier under TRANSPORT-UNTRUSTED-008, CITATION-AUTH-010, EGRESS-011, and ACTION-SEPARATION-014. TS-PRIVATE-BOT-FLOW-071, SDK-PRIVATE-BOT-FLOW-071, and PG-PRIVATE-BOT-FLOW-071 prove the independent TypeScript process and import boundary, exact verified private event binding, opaque DeliveryEvidenceRef transport, installed generated-SDK HTTP resolve, sealed File Package path, controlled model generation, distinct placeholder/final ActionPlane effects, digest-only DeliveryReceipt audit, citation reopening, and the composed wrong-binding oracles. The historical FIXTURE-ACCEPT-012 continues to prove the M0 unavailable cross-capability baseline separately. Live Feishu/model/Sender network carriers, group/public delivery, compensation/delete, Continue, and MCP remain NOT_ACTIVE. Issue #81 activates deterministic shallow File readChanges and whole-page acknowledgement under WORKER-LEASE-007. PG-FILE-CHANGE-ACTIVATE-081, PG-FILE-CHANGE-PAGE-081, and PG-FILE-CHANGE-DENY-081 prove immutable v3 activation, provider-authenticated content-free paging, exact replay, predecessor ordering, post-commit cursor issuance, FORCE-RLS tenant isolation, source/version invalidation, and zero implicit job or publication effect. Automatic scheduling, deletion execution, recursive discovery, full resync, and Runtime authority from cursor or checkpoint metadata remain NOT_ACTIVE. Issue #83 activates only explicit accepted File page scheduling through the existing file.import acquisition, WorkerLease, and publication path under WORKER-LEASE-007. PG-FILE-CHANGE-SCHEDULE-083, PG-FILE-CHANGE-SCHEDULE-DENY-083, and PG-FILE-CHANGE-SUPERSESSION-083 prove whole-page atomicity, exact job replay, explicit current FileImportAudience and receiver validation, immutable raw-byte observation lineage, pre-compiler drift refusal, and scan-epoch fences before content read and visible publication with zero stale publication effect. Autonomous polling, implicit audience inheritance, deletion execution, automatic retry/reclaim, dead-letter handling, full resync, and Runtime authority from provider checkpoints remain NOT_ACTIVE. Issue #85 activates bounded File delete observations only under WORKER-LEASE-007. PG-FILE-DELETE-DETECT-085, PG-FILE-DELETE-PAGE-085, and PG-FILE-DELETE-NO-EFFECT-085 prove exact latest-complete same-SourceVersion baseline binding, stable shallow diff, immutable canonical persistence/replay, forged/incomplete/stale/cross-Organization refusal, mixed-page scheduling refusal, and zero tombstone, Policy Epoch, cleanup, watermark, or generated-SDK Runtime visibility effect. At that activation, deletion execution remains NOT_ACTIVE; autonomous polling, retry/reclaim/dead-letter, full resync, recursive scan, and Runtime authority from baseline or delete metadata remain NOT_ACTIVE. Issue #87 later activates only exact trusted current File delete execution through the existing #28 tombstone authority under REVOCATION-006. PG-FILE-DELETE-EXECUTE-087, PG-FILE-DELETE-REPLAY-087, and HTTP-FILE-DELETE-INVISIBLE-087 prove current complete-scan revalidation, server-derived effect identity, atomic tombstone/epoch/cleanup/binding, exact replay, mismatch rollback, and immediate sealed-Runtime invisibility. Issue #89 later activates only the exact upsert projection of one current mixed v4 page under WORKER-LEASE-007. PG-FILE-MIXED-UPSERT-SCHEDULE-089, PG-FILE-MIXED-UPSERT-REPLAY-089, and HTTP-FILE-MIXED-UPSERT-NO-DELETE-089 prove complete-page validation, gapped original-ordinal job binding, exact replay, partial-lineage refusal, and zero delete/tombstone authority through the generated SDK. Autonomous scheduling, automatic upsert/delete ordering, batch deletion execution, Provider deletion authority, physical cleanup, restore/recreate, retry/reclaim/dead-letter, full resync, and Runtime authority from observation, page, checkpoint, or execution metadata remain NOT_ACTIVE. Issue #91 activates only scheduler-owned first-attempt File dispatch through the existing exact WorkerLease. Automatic reclaim, retry/backoff, dead-letter handling, provider polling, and delete ordering remain NOT_ACTIVE." + "reconciliation": "Issue #2 fixes the product and testing decisions, issue #5 requires exactly fifteen release invariants and twelve canonical acceptance fixtures, and ADR-0019 resolves the later nineteen-label prose expansion without weakening any safeguard. Issue #15 activates only Organization-level next-request resolve(Acquire) revocation evidence under REVOCATION-006: at that activation, PG-REVOCATION-006, RUN-006, and CACHE-002 are active while BLOB-002 and Continue, citation, Policy-Epoch-bound WorkerLease, production ContextAccessTicket/ActionTicket, audit, outbox, cleanup, finer-epoch, UI, and external-admin carriers remain future or NOT_ACTIVE; later issue records are authoritative for subsequently activated carriers. Issue #16 activates only the M0 refusal gate for unavailable Continue, profile-disabled OpenCitation, and server-owned unavailable Acquire plans: at that activation its real continuation, profile-enabled citation, federated/source-native, and File carriers remain future, while its Runtime and HTTP refusal surfaces prove generic outcomes before content I/O; Issue #69 later activates the private/direct File profile-enabled citation carrier. Issue #17 activates only the signed one-shot persistent no-op durable-job WorkerLease subcarrier under WORKER-LEASE-007. It binds one exact worker audience but no end-user delivery audience or Policy Epoch, and proves only LEASE-SIGNING-017, PG-WORKER-LEASE-NOOP-017, and WORKER-LEASE-REPLAY-007; Source, Resource, Revision, Policy Epoch, end-user delivery audience, idempotency, generation, business mutation, outbox, File publication, and the full ACCEPT-008 matrix remain deferred or NOT_ACTIVE. Issue #18 activates only distinct signed synthetic ContextAccessTicket Provider-read and ActionTicket no-op channel-action subcarriers under ACTION-SEPARATION-014, with current Organization-v0 Policy Epoch validation. TICKET-AUDIENCE-018 and PG-TICKET-EPOCH-018 do not activate production ContextProvider integration, ContextRuntime ticket integration, BotDelivery, full M2 ActionPlane.prepare/perform, a real Sender or external effect, payload/destination/approval/idempotency binding, durable one-shot/replay/reconciliation, or full ACCEPT-012 PASS; those remain future or NOT_ACTIVE. Issue #19 activates only the current Acquire authorized-only ContextRun and restricted delivered-empty DecisionAudit subcarrier under TRACE-REDACTION-012. DIGEST-019, RUN-LINEAGE-019, AUTHORIZED-RUN-019, and PG-TRACE-REDACTION-012 prove deterministic Package and Organization-bound query digests, retained-UserActor-transaction persistence, decisionRef resolution, redaction, and short-lived exact-Organization operator ticket reads with no application-role table access; the supported reader commits deletion before returning, while a direct caller rollback is not claimed as durable exactly-once redemption. Raw query retention, full ContextPackage body retention, unauthenticated transport failures as ContextRuns, cross-Organization analytics, and general observability redaction remain NOT_ACTIVE. Issue #48 activates only the current ACCEPT-002 authenticated HTTP Acquire Membership field-projection carrier under SCOPE-INTERSECTION-004, INDEX-NOT-AUTHORITY-005, and TRACE-REDACTION-012. PROP-FIELD-PROJECTION-048, PG-FIELD-PROJECTION-048, and HTTP-ACCEPT-002-048 bind one current Membership/version field ceiling to same-transaction FORCE-RLS reduction, the sealed AuthorizationKernel, AuthorizedProjection and Evidence integrity, and authorized-only ContextRun/audit persistence. General permission DSLs, caller-authored projection lists, CandidateRef or index field authority, production Provider/source-native ACL negotiation, Supply publication, File/Base field ACL, typed fields, Continue, and Issue #20 runner substitution remain future or NOT_ACTIVE; Issue #69 later activates private/direct File OpenCitation through the same field-projection gates. Issue #63 activates only the digest-only private authenticated HTTP Acquire DeliveryEvidenceRef carrier under TRANSPORT-UNTRUSTED-008. PROP-DELIVERY-EVIDENCE-063, PG-DELIVERY-EVIDENCE-063, HTTP-DELIVERY-EVIDENCE-063, and FILE-DELIVERY-EVIDENCE-063 prove exact service/request/Organization/asker/Membership-version/destination/consumer/purpose/audience/epoch/lifetime binding, stable identical retry identity, role isolation, expiry cleanup, pre-content generic rejection, and one File-backed sealed Runtime delivery. Group AudienceSnapshot, group/public DeliveryEvidenceRef, production ModelGateway, ActionPlane, and the BotDelivery application remain future or NOT_ACTIVE; Issues #64, #66, #69, and #70 later activate the frozen OpenAPI, generated TypeScript SDK, private/direct OpenCitation, and deterministic private model-egress carriers. Issue #65 activates only one opaque digest-only model or channel EgressGrant after final Package policy, exact atomic PostgreSQL redemption and restricted audit, nominal BotDelivery inputs, and deterministic network-free ModelGateway or Sender-preflight spies under EGRESS-011. PROP-EGRESS-011, PG-EGRESS-011, and RUNTIME-EGRESS-011 prove exact Package/Organization/purpose/audience/epoch/hop/profile/lifetime binding and zero additional bytes on replay. Real model/provider calls, a real Sender or channel write, ActionTicket effects, group AudienceSnapshot revalidation, and the BotDelivery application process remain future or NOT_ACTIVE; Issue #64 later activates the generated SDK consumer and Issue #70 later activates the deterministic private TypeScript ModelGateway. Issue #66 activates the frozen public POST /v0/resolve OpenAPI carrier under TRANSPORT-UNTRUSTED-008. OPENAPI-CONTRACT-066, OPENAPI-BREAKING-066, HTTP-V0-066, and PG-RUNTIME-RELEASE-066 prove one public closed operation, deterministic immutable snapshot and breaking-change refusal, a hidden v1 bridge through the same handler and sealed Runtime path, and exact read-only observation of the active Learning-promoted release with fail-closed missing-release behavior before content work. A production BotDelivery caller, Continue redemption, MCP, group AudienceSnapshot, and external effects remain future or NOT_ACTIVE; Issues #64 and #69 later activate the generated TypeScript SDK and private/direct OpenCitation redemption through this frozen operation. Issue #64 activates only the packaged generated TypeScript POST /v0/resolve client under TRANSPORT-UNTRUSTED-008. SDK-CONTRACT-064 and SDK-LIVE-FILE-064 prove deterministic pinned generation, strict closed types, a narrow export map and metadata-only facade, installable tarball consumption, and one real PostgreSQL/File-backed Acquire through CandidateRef, AuthorizationKernel, AuthorizedProjection, ContextPackage, and opaque model egress grant. Issue #69 later extends SDK-LIVE-FILE-064 with a successful private/direct File OpenCitation through a second request-bound DeliveryEvidenceRef; Issue #70 extends the installed SDK fixture into deterministic Package-bound model generation; generated Continue remains generic unavailable. External package publication, production provider access, MCP, group AudienceSnapshot, real Continue redemption, and external effects remain future or NOT_ACTIVE. Issue #67 activates only private ActionPlane.prepare for create-placeholder, finalize-reply, and private-follow-up operation-specific tickets under ACTION-SEPARATION-014. PG-ACTION-PREPARE-067 proves exact current delivery, Organization, destination, audience, source, payload, approval, epoch, lifetime, and idempotency binding under a dedicated non-owner PostgreSQL role with digest-only FORCE-RLS persistence and zero effects. Issue #68 activates private ActionPlane.perform only through a deterministic Sender twin. PG-ACTION-PERFORM-068 proves one pre-Sender current-authority validation, one provider-attempt identity, immutable receipt replay, zero-effect ticket/payload mutation and stale-audience refusal, same-label cross-Organization isolation, and monotonic applied/rejected reconciliation including crash interleavings. Real provider or channel network effects, group AudienceSnapshot, compensation/delete, production BotDelivery orchestration, and the full ACCEPT-012 pass remain future or NOT_ACTIVE. The canonical set is IDs 001 through 012, 014, 015, and 019: CACHE-SCOPE-013 remains a preregistered conditional extension; AUDIENCE-016 is absorbed by SCOPE-INTERSECTION-004 and EGRESS-011; ACL-PROOF-017 is absorbed by INDEX-NOT-AUTHORITY-005 and REVOCATION-006; DELIVERY-EVIDENCE-018 is absorbed by TRANSPORT-UNTRUSTED-008. ACCEPT-001 through ACCEPT-012 follow ADR-0019's category order. Protected-asset references A-01 through A-08 refer, in order, to the eight bullets in the threat model's Protected assets section. Every expectedEvidence value below is a stable planned case identifier, not a claim that the case ran or passed; only an exact activation record upgrades named evidence, while fixture carrier status and the explicit M0 oracle preserve every other accepted-versus-active distinction. Issue #69 activates private/direct File CitationOpenRef issuance and OpenCitation under CITATION-AUTH-010: digest-only multi-use locators reveal only prior Package/Evidence and Fragment location lineage, every open obtains a current UserActor and trusted delivery context then traverses CandidateRef, AuthorizationKernel, AuthorizedProjection, a replacement ContextPackage, EgressGrant, ContextRun, and restricted DecisionAudit. PG-CITATION-AUTH-010, RUNTIME-CITATION-AUTH-010, and SDK-LIVE-FILE-064 prove A/B/A reauthorization, non-consumption on denial, database-clock expiry, cross-kind and cross-Organization opacity, and the generated SDK carrier. Group/public AudienceSnapshot, non-File providers, raw source URL locators, and Continue remain future or NOT_ACTIVE. Issue #70 activates only the private deterministic TypeScript ModelGateway under EGRESS-011. TS-MODEL-EGRESS-070, SDK-MODEL-EGRESS-070, and PG-MODEL-EGRESS-070 prove one current Package, exact grant redemption, closed provider input, bounded Package-subset citations, replay zero bytes, and digest-only retained audit. Real providers, streaming, group AudienceSnapshot, model-authored ActionPlane authority, and external effects remain future or NOT_ACTIVE. Issue #71 activates the complete private File-backed deterministic-twin BotDelivery carrier under TRANSPORT-UNTRUSTED-008, CITATION-AUTH-010, EGRESS-011, and ACTION-SEPARATION-014. TS-PRIVATE-BOT-FLOW-071, SDK-PRIVATE-BOT-FLOW-071, and PG-PRIVATE-BOT-FLOW-071 prove the independent TypeScript process and import boundary, exact verified private event binding, opaque DeliveryEvidenceRef transport, installed generated-SDK HTTP resolve, sealed File Package path, controlled model generation, distinct placeholder/final ActionPlane effects, digest-only DeliveryReceipt audit, citation reopening, and the composed wrong-binding oracles. The historical FIXTURE-ACCEPT-012 continues to prove the M0 unavailable cross-capability baseline separately. Live Feishu/model/Sender network carriers, group/public delivery, compensation/delete, Continue, and MCP remain NOT_ACTIVE. Issue #81 activates deterministic shallow File readChanges and whole-page acknowledgement under WORKER-LEASE-007. PG-FILE-CHANGE-ACTIVATE-081, PG-FILE-CHANGE-PAGE-081, and PG-FILE-CHANGE-DENY-081 prove immutable v3 activation, provider-authenticated content-free paging, exact replay, predecessor ordering, post-commit cursor issuance, FORCE-RLS tenant isolation, source/version invalidation, and zero implicit job or publication effect. Automatic scheduling, deletion execution, recursive discovery, full resync, and Runtime authority from cursor or checkpoint metadata remain NOT_ACTIVE. Issue #83 activates only explicit accepted File page scheduling through the existing file.import acquisition, WorkerLease, and publication path under WORKER-LEASE-007. PG-FILE-CHANGE-SCHEDULE-083, PG-FILE-CHANGE-SCHEDULE-DENY-083, and PG-FILE-CHANGE-SUPERSESSION-083 prove whole-page atomicity, exact job replay, explicit current FileImportAudience and receiver validation, immutable raw-byte observation lineage, pre-compiler drift refusal, and scan-epoch fences before content read and visible publication with zero stale publication effect. Autonomous polling, implicit audience inheritance, deletion execution, automatic retry/reclaim, dead-letter handling, full resync, and Runtime authority from provider checkpoints remain NOT_ACTIVE. Issue #85 activates bounded File delete observations only under WORKER-LEASE-007. PG-FILE-DELETE-DETECT-085, PG-FILE-DELETE-PAGE-085, and PG-FILE-DELETE-NO-EFFECT-085 prove exact latest-complete same-SourceVersion baseline binding, stable shallow diff, immutable canonical persistence/replay, forged/incomplete/stale/cross-Organization refusal, mixed-page scheduling refusal, and zero tombstone, Policy Epoch, cleanup, watermark, or generated-SDK Runtime visibility effect. At that activation, deletion execution remains NOT_ACTIVE; autonomous polling, retry/reclaim/dead-letter, full resync, recursive scan, and Runtime authority from baseline or delete metadata remain NOT_ACTIVE. Issue #87 later activates only exact trusted current File delete execution through the existing #28 tombstone authority under REVOCATION-006. PG-FILE-DELETE-EXECUTE-087, PG-FILE-DELETE-REPLAY-087, and HTTP-FILE-DELETE-INVISIBLE-087 prove current complete-scan revalidation, server-derived effect identity, atomic tombstone/epoch/cleanup/binding, exact replay, mismatch rollback, and immediate sealed-Runtime invisibility. Issue #89 later activates only the exact upsert projection of one current mixed v4 page under WORKER-LEASE-007. PG-FILE-MIXED-UPSERT-SCHEDULE-089, PG-FILE-MIXED-UPSERT-REPLAY-089, and HTTP-FILE-MIXED-UPSERT-NO-DELETE-089 prove complete-page validation, gapped original-ordinal job binding, exact replay, partial-lineage refusal, and zero delete/tombstone authority through the generated SDK. Autonomous scheduling, automatic upsert/delete ordering, batch deletion execution, Provider deletion authority, physical cleanup, restore/recreate, retry/reclaim/dead-letter, full resync, and Runtime authority from observation, page, checkpoint, or execution metadata remain NOT_ACTIVE. Issue #91 activates only scheduler-owned first-attempt File dispatch through the existing exact WorkerLease. Issue #93 extends that carrier with database-timed exponential backoff and at most three automatic higher-generation reclaims through the existing durable-boundary recovery path. Exhausted attempts remain untouched; dead-letter handling, operator remediation, provider polling, and delete ordering remain NOT_ACTIVE." }, "hardOracles": [ { @@ -965,6 +967,46 @@ "automatic retry or reclaim", "Runtime authorization from Supply scheduling or lease state" ] + }, + { + "issueRef": "#93", + "invariantRef": "WORKER-LEASE-007", + "carrier": "bounded autonomous reclaim of expired scheduled File upserts", + "status": "active_fail_closed", + "policyEpochScope": "not-runtime-authority", + "controlBoundary": "function-only scheduler login -> expired lease plus database-owned exponential backoff -> current page/acquisition/audience/receiver revalidation -> deterministic recovery-first SKIP LOCKED selector -> exact next-generation WorkerLease -> existing durable-boundary File worker resume", + "testEvidence": [ + { + "id": "PG-FILE-RECLAIM-093", + "surface": "tests/integration/test_file_dispatch.py::test_scheduler_reclaims_one_expired_dispatch_after_database_backoff tests/integration/test_file_dispatch.py::test_scheduler_does_not_reclaim_after_automatic_generation_budget tests/integration/test_file_dispatch.py::test_scheduler_reclaim_revalidates_current_membership_without_mutation", + "oracle": "The non-owner scheduler reclaims only after database expiry plus generation-derived backoff, atomically advances to the exact next generation with one digest-only event, returns content-free no-work before eligibility, after the generation-four budget, or after current Membership authority is revoked, and mutates nothing on refusal." + }, + { + "id": "PG-FILE-RECLAIM-CONCURRENCY-093", + "surface": "tests/integration/test_file_dispatch.py::test_concurrent_schedulers_reclaim_one_expired_generation_once", + "oracle": "Two scheduler-role authorities competing for one expired generation yield exactly one next-generation lease and one immutable reclaim event; the loser receives content-free no-work." + }, + { + "id": "PROC-FILE-RECLAIM-093", + "surface": "tests/integration/test_file_dispatch.py::test_scheduler_recovers_interrupted_publication_and_stales_old_lease", + "oracle": "Across acquired, prepared, and indexed durable boundaries, the scheduler and existing File worker resume with one higher-generation lease, the replaced lease has zero effect, and exactly one active Revision/publication effect remains." + } + ], + "deferredEvidence": [ + "dead-letter state, exhaustion transition, operator remediation, requeue, and retry observability", + "provider polling, automatic page acceptance, full resync, and automatic delete ordering" + ], + "futureCarriers": [ + "dead-letter and operator remediation owner", + "provider polling and full resync", + "explicit mixed-change ordering policy" + ], + "notActive": [ + "scheduler tenant, Source, job, attempt, timing, backoff, or generation choice", + "terminal failure retry, dead-letter transition, or operator requeue", + "manual import or delete recovery", + "Runtime authorization from Supply retry or lease state" + ] } ], "invariants": [ diff --git a/migrations/versions/20260726_0034_autonomous_file_reclaim.py b/migrations/versions/20260726_0034_autonomous_file_reclaim.py new file mode 100644 index 00000000..284eedbb --- /dev/null +++ b/migrations/versions/20260726_0034_autonomous_file_reclaim.py @@ -0,0 +1,464 @@ +"""Reclaim expired scheduled File imports through bounded automatic leases. + +Revision ID: 20260726_0034 +Revises: 20260725_0033 +Create Date: 2026-07-26 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "20260726_0034" +down_revision: str | None = "20260725_0033" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_SCHEDULER = "context_engine_scheduler" +_DEFINER = "context_engine_file_dispatch_definer" +_FUNCTION = "context_scheduler_claim_file_import" +_FIRST_FUNCTION = "context_scheduler_claim_first_file_import" +_SIGNATURE = "(bigint, bytea, text[])" +_TTL_SECONDS = 300 +_MAX_GENERATION = 4 +_BASE_BACKOFF_SECONDS = 30 +_MAX_BIGINT = 9_223_372_036_854_775_807 +_MIGRATION_FENCE = "context-engine.file-dispatch-migration-fence" +_SCHEDULING_MIGRATION_FENCE = "context-engine.file-change-scheduling-migration-fence" +_V3 = "file-capabilities-v3" +_V4 = "file-capabilities-v4" + + +def upgrade() -> None: + """Prefer one eligible expired attempt before delegating first dispatch.""" + + connection = op.get_bind() + for migration_fence in (_SCHEDULING_MIGRATION_FENCE, _MIGRATION_FENCE): + connection.execute( + sa.text( + "SELECT pg_catalog.pg_advisory_xact_lock(" + "pg_catalog.hashtextextended(:migration_fence, 0))" + ), + {"migration_fence": migration_fence}, + ) + op.execute(f"GRANT CREATE ON SCHEMA public TO {_DEFINER}") + op.execute(f"SET LOCAL ROLE {_DEFINER}") + op.execute( + f"ALTER FUNCTION public.{_FUNCTION}{_SIGNATURE} " + f"RENAME TO {_FIRST_FUNCTION}" + ) + op.execute( + f"REVOKE EXECUTE ON FUNCTION public.{_FIRST_FUNCTION}{_SIGNATURE} " + f"FROM {_SCHEDULER}" + ) + op.execute("RESET ROLE") + op.execute(f"REVOKE CREATE ON SCHEMA public FROM {_DEFINER}") + op.execute( + "CREATE INDEX ix_file_import_job_dispatch_expired ON file_import_job " + "(lease_expires_at, lease_generation, organization_id, source_id, job_id) " + "WHERE dispatch_claimed IS TRUE " + "AND state IN ('leased', 'running', 'prepared', 'ready') " + f"AND lease_generation BETWEEN 1 AND {_MAX_GENERATION - 1}" + ) + op.execute( + "CREATE POLICY file_import_job_event_file_dispatch_definer_select " + "ON file_import_job_event FOR SELECT " + f"TO {_DEFINER} USING (organization_id = " + "NULLIF(current_setting('app.organization_id', true), '')::uuid)" + ) + op.execute( + "CREATE POLICY file_import_job_event_file_dispatch_definer_insert " + "ON file_import_job_event FOR INSERT " + f"TO {_DEFINER} WITH CHECK (organization_id = " + "NULLIF(current_setting('app.organization_id', true), '')::uuid)" + ) + op.execute(f"GRANT SELECT, INSERT ON file_import_job_event TO {_DEFINER}") + op.execute( + "GRANT UPDATE (recovery_from_state, lease_redeemed_at) " + f"ON file_import_job TO {_DEFINER}" + ) + op.execute( + f""" + CREATE FUNCTION public.{_FUNCTION}( + requested_signing_key_version bigint, + requested_nonce bytea, + configured_root_refs text[] + ) RETURNS TABLE ( + organization_id uuid, + job_id uuid, + source_id uuid, + service_principal_id uuid, + lease_generation bigint, + issued_at timestamptz, + expires_at timestamptz + ) + LANGUAGE plpgsql SECURITY DEFINER + SET search_path = pg_catalog, pg_temp + SET row_security = on + AS $function$ + DECLARE + authority_checked_at timestamptz; + minted_at timestamptz; + selected_organization_id uuid; + selected_job_id uuid; + selected_source_id uuid; + selected_state text; + selected_resume_state text; + selected_generation bigint; + next_generation bigint; + next_ordinal bigint; + BEGIN + IF requested_signing_key_version IS NULL + OR requested_nonce IS NULL + OR configured_root_refs IS NULL + OR SESSION_USER <> '{_SCHEDULER}' + OR requested_signing_key_version NOT BETWEEN 1 AND {_MAX_BIGINT} + OR pg_catalog.octet_length(requested_nonce) <> 32 + OR pg_catalog.cardinality(configured_root_refs) < 1 + OR EXISTS ( + SELECT 1 FROM pg_catalog.unnest(configured_root_refs) AS root_ref + WHERE root_ref !~ '^[A-Za-z0-9][A-Za-z0-9._-]{{0,127}}$' + OR root_ref IN ('.', '..') + ) + OR pg_catalog.cardinality(configured_root_refs) <> ( + SELECT count(DISTINCT root_ref) + FROM pg_catalog.unnest(configured_root_refs) AS root_ref + ) + THEN RETURN; END IF; + PERFORM pg_catalog.pg_advisory_xact_lock_shared( + pg_catalog.hashtextextended('{_MIGRATION_FENCE}', 0) + ); + authority_checked_at := pg_catalog.clock_timestamp(); + SELECT job.organization_id, job.job_id, job.source_id, + job.state, job.recovery_from_state, job.lease_generation + INTO selected_organization_id, selected_job_id, selected_source_id, + selected_state, selected_resume_state, selected_generation + FROM public.file_import_job AS job + JOIN public.file_acquisition AS acquisition + ON acquisition.organization_id = job.organization_id + AND acquisition.acquisition_id = job.acquisition_id + AND acquisition.source_id = job.source_id + JOIN public.file_source_change_page AS accepted_page + ON accepted_page.organization_id = acquisition.organization_id + AND accepted_page.source_id = acquisition.source_id + AND accepted_page.source_version_id = acquisition.source_version_id + AND accepted_page.page_ref = acquisition.change_page_ref + JOIN public.file_source_change AS accepted_change + ON accepted_change.organization_id = acquisition.organization_id + AND accepted_change.source_id = acquisition.source_id + AND accepted_change.source_version_id = acquisition.source_version_id + AND accepted_change.page_ref = acquisition.change_page_ref + AND accepted_change.change_ordinal = acquisition.change_ordinal + AND accepted_change.relative_path = acquisition.relative_path + AND accepted_change.content_sha256 = acquisition.expected_content_sha256 + AND accepted_change.content_length = acquisition.expected_content_length + JOIN public.file_source_acquisition_checkpoint AS scheduled + ON scheduled.organization_id = job.organization_id + AND scheduled.source_id = job.source_id + AND scheduled.acquisition_id = job.acquisition_id + AND scheduled.job_id = job.job_id + AND scheduled.change_kind = 'file_import' + JOIN public.context_source AS source + ON source.organization_id = acquisition.organization_id + AND source.source_id = acquisition.source_id + AND source.active_version_id = acquisition.source_version_id + JOIN public.source_version AS version + ON version.organization_id = source.organization_id + AND version.source_id = source.source_id + AND version.version_id = source.active_version_id + JOIN public.membership AS audience + ON audience.organization_id = acquisition.organization_id + AND audience.membership_id = acquisition.audience_membership_id + AND audience.membership_version = acquisition.audience_membership_version + JOIN public.service_principal AS receiver + ON receiver.organization_id = job.organization_id + AND receiver.service_principal_id = job.service_principal_id + AND receiver.workload = job.workload + AND receiver.worker_audience = job.worker_audience + AND receiver.operation = job.operation + WHERE job.dispatch_claimed IS TRUE + AND job.state IN ('leased', 'running', 'prepared', 'ready') + AND job.lease_generation BETWEEN 1 AND {_MAX_GENERATION - 1} + AND job.lease_expires_at + + pg_catalog.make_interval(secs => + {_BASE_BACKOFF_SECONDS} + * (1::bigint << (job.lease_generation - 1)::integer) + ) <= authority_checked_at + AND job.workload = 'supply.file-import' + AND job.worker_audience = 'context-engine-worker' + AND job.actor_kind = 'service' + AND job.operation = 'file.import' + AND acquisition.change_page_ref IS NOT NULL + AND accepted_change.change_kind = 'upsert' + AND source.source_kind = 'file' + AND source.lifecycle_state = 'active' + AND version.capability_manifest->>'declarationVersion' + IN ('{_V3}', '{_V4}') + AND audience.status = 'active' + AND audience.valid_from <= authority_checked_at + AND (audience.valid_until IS NULL + OR audience.valid_until > authority_checked_at) + AND receiver.enabled IS TRUE + AND accepted_page.scan_epoch = ( + SELECT current_page.scan_epoch + FROM public.file_source_acquisition_checkpoint AS current_checkpoint + JOIN public.file_source_change_page AS current_page + ON current_page.organization_id = current_checkpoint.organization_id + AND current_page.source_id = current_checkpoint.source_id + AND current_page.source_version_id = + current_checkpoint.source_version_id + AND current_page.page_ref = current_checkpoint.change_page_ref + WHERE current_checkpoint.organization_id = acquisition.organization_id + AND current_checkpoint.source_id = acquisition.source_id + AND current_checkpoint.change_kind = 'file_change_page' + ORDER BY current_checkpoint.sequence DESC + LIMIT 1 + ) + ORDER BY job.lease_expires_at + + pg_catalog.make_interval(secs => + {_BASE_BACKOFF_SECONDS} + * (1::bigint << (job.lease_generation - 1)::integer) + ), + scheduled.accepted_at, scheduled.sequence, + accepted_page.page_ordinal, accepted_change.change_ordinal, + job.organization_id, job.source_id, job.job_id + FOR UPDATE OF job SKIP LOCKED + LIMIT 1; + IF NOT FOUND THEN + RETURN QUERY SELECT * + FROM public.{_FIRST_FUNCTION}( + requested_signing_key_version, + requested_nonce, + configured_root_refs + ); + RETURN; + END IF; + IF NOT EXISTS ( + SELECT 1 FROM public.source_version AS selected_version + WHERE selected_version.organization_id = selected_organization_id + AND selected_version.source_id = selected_source_id + AND selected_version.root_ref = ANY(configured_root_refs) + ) THEN RETURN; END IF; + PERFORM pg_catalog.pg_advisory_xact_lock( + pg_catalog.hashtextextended( + 'context-engine.file-source-progress:' + || selected_organization_id::text || ':' + || selected_source_id::text, 0 + ) + ); + PERFORM pg_catalog.set_config( + 'app.organization_id', selected_organization_id::text, true + ); + PERFORM pg_catalog.set_config( + 'app.worker_job_id', selected_job_id::text, true + ); + authority_checked_at := pg_catalog.clock_timestamp(); + minted_at := pg_catalog.date_trunc( + 'second', pg_catalog.clock_timestamp() + ); + IF NOT EXISTS ( + SELECT 1 + FROM public.file_import_job AS job + JOIN public.file_acquisition AS acquisition + ON acquisition.organization_id = job.organization_id + AND acquisition.acquisition_id = job.acquisition_id + AND acquisition.source_id = job.source_id + JOIN public.file_source_change_page AS accepted_page + ON accepted_page.organization_id = acquisition.organization_id + AND accepted_page.source_id = acquisition.source_id + AND accepted_page.source_version_id = acquisition.source_version_id + AND accepted_page.page_ref = acquisition.change_page_ref + JOIN public.file_source_change AS accepted_change + ON accepted_change.organization_id = acquisition.organization_id + AND accepted_change.source_id = acquisition.source_id + AND accepted_change.source_version_id = acquisition.source_version_id + AND accepted_change.page_ref = acquisition.change_page_ref + AND accepted_change.change_ordinal = acquisition.change_ordinal + AND accepted_change.relative_path = acquisition.relative_path + AND accepted_change.content_sha256 = + acquisition.expected_content_sha256 + AND accepted_change.content_length = + acquisition.expected_content_length + JOIN public.context_source AS source + ON source.organization_id = acquisition.organization_id + AND source.source_id = acquisition.source_id + AND source.active_version_id = acquisition.source_version_id + JOIN public.source_version AS version + ON version.organization_id = source.organization_id + AND version.source_id = source.source_id + AND version.version_id = source.active_version_id + JOIN public.membership AS audience + ON audience.organization_id = acquisition.organization_id + AND audience.membership_id = acquisition.audience_membership_id + AND audience.membership_version = + acquisition.audience_membership_version + JOIN public.service_principal AS receiver + ON receiver.organization_id = job.organization_id + AND receiver.service_principal_id = job.service_principal_id + AND receiver.workload = job.workload + AND receiver.worker_audience = job.worker_audience + AND receiver.operation = job.operation + WHERE job.organization_id = selected_organization_id + AND job.job_id = selected_job_id + AND job.source_id = selected_source_id + AND job.state = selected_state + AND job.recovery_from_state IS NOT DISTINCT FROM selected_resume_state + AND job.lease_generation = selected_generation + AND job.dispatch_claimed IS TRUE + AND job.lease_expires_at + + pg_catalog.make_interval(secs => + {_BASE_BACKOFF_SECONDS} + * (1::bigint << (job.lease_generation - 1)::integer) + ) <= authority_checked_at + AND accepted_change.change_kind = 'upsert' + AND source.source_kind = 'file' + AND source.lifecycle_state = 'active' + AND version.capability_manifest->>'declarationVersion' + IN ('{_V3}', '{_V4}') + AND version.root_ref = ANY(configured_root_refs) + AND audience.status = 'active' + AND audience.valid_from <= authority_checked_at + AND (audience.valid_until IS NULL + OR audience.valid_until > authority_checked_at) + AND receiver.enabled IS TRUE + AND accepted_page.scan_epoch = ( + SELECT current_page.scan_epoch + FROM public.file_source_acquisition_checkpoint + AS current_checkpoint + JOIN public.file_source_change_page AS current_page + ON current_page.organization_id = + current_checkpoint.organization_id + AND current_page.source_id = current_checkpoint.source_id + AND current_page.source_version_id = + current_checkpoint.source_version_id + AND current_page.page_ref = current_checkpoint.change_page_ref + WHERE current_checkpoint.organization_id = + acquisition.organization_id + AND current_checkpoint.source_id = acquisition.source_id + AND current_checkpoint.change_kind = 'file_change_page' + ORDER BY current_checkpoint.sequence DESC + LIMIT 1 + ) + FOR UPDATE OF source, audience, receiver SKIP LOCKED + ) THEN RETURN; END IF; + IF selected_state <> 'leased' THEN + selected_resume_state := selected_state; + END IF; + next_generation := selected_generation + 1; + SELECT COALESCE(max(event.ordinal), -1) + 1 INTO next_ordinal + FROM public.file_import_job_event AS event + WHERE event.organization_id = selected_organization_id + AND event.job_id = selected_job_id; + INSERT INTO public.file_import_job_event ( + organization_id, job_id, ordinal, event_type, boundary, + lease_generation, state_at_event, revision_id, + reason_digest, occurred_at + ) SELECT + selected_organization_id, selected_job_id, next_ordinal, + 'reclaimed', + CASE COALESCE(selected_resume_state, 'running') + WHEN 'running' THEN 'acquired' + WHEN 'prepared' THEN 'prepared' + ELSE 'indexed' + END, + next_generation, selected_state, job.revision_id, + encode(public.digest( + convert_to('context-engine.file-reclaim.v1', 'UTF8') + || decode('00', 'hex') + || uuid_send(selected_organization_id) + || uuid_send(selected_job_id) + || int8send(next_generation), + 'sha256' + ), 'hex'), minted_at + FROM public.file_import_job AS job + WHERE job.organization_id = selected_organization_id + AND job.job_id = selected_job_id; + RETURN QUERY + UPDATE public.file_import_job AS job + SET state = 'leased', + lease_generation = next_generation, + recovery_from_state = selected_resume_state, + signing_key_version = requested_signing_key_version, + lease_nonce_digest = public.digest(requested_nonce, 'sha256'), + lease_issued_at = minted_at, + lease_expires_at = minted_at + + pg_catalog.make_interval(secs => {_TTL_SECONDS}), + lease_redeemed_at = NULL + WHERE job.organization_id = selected_organization_id + AND job.job_id = selected_job_id + AND job.state = selected_state + AND job.lease_generation = selected_generation + RETURNING job.organization_id, job.job_id, job.source_id, + job.service_principal_id, job.lease_generation, + job.lease_issued_at, job.lease_expires_at; + IF NOT FOUND THEN + RAISE EXCEPTION 'File reclaim lost its exact job fence' + USING ERRCODE = '40001'; + END IF; + END; + $function$ + """ + ) + op.execute(f"REVOKE ALL ON FUNCTION public.{_FUNCTION}{_SIGNATURE} FROM PUBLIC") + op.execute(f"GRANT CREATE ON SCHEMA public TO {_DEFINER}") + op.execute(f"ALTER FUNCTION public.{_FUNCTION}{_SIGNATURE} OWNER TO {_DEFINER}") + op.execute(f"SET LOCAL ROLE {_DEFINER}") + op.execute( + f"GRANT EXECUTE ON FUNCTION public.{_FUNCTION}{_SIGNATURE} TO {_SCHEDULER}" + ) + op.execute("RESET ROLE") + op.execute(f"REVOKE CREATE ON SCHEMA public FROM {_DEFINER}") + + +def downgrade() -> None: + """Remove automatic reclaim only when it has minted no later generation.""" + + connection = op.get_bind() + for migration_fence in (_SCHEDULING_MIGRATION_FENCE, _MIGRATION_FENCE): + connection.execute( + sa.text( + "SELECT pg_catalog.pg_advisory_xact_lock(" + "pg_catalog.hashtextextended(:migration_fence, 0))" + ), + {"migration_fence": migration_fence}, + ) + reclaimed = connection.execute( + sa.text( + "SELECT EXISTS (SELECT 1 FROM public.file_import_job " + "WHERE dispatch_claimed IS TRUE AND lease_generation > 1)" + ) + ).scalar_one() + if reclaimed: + raise RuntimeError( + "automatic File reclaim downgrade requires no retained " + "higher-generation lease; use a forward fix" + ) + op.execute(f"DROP FUNCTION public.{_FUNCTION}{_SIGNATURE}") + op.execute( + "REVOKE UPDATE (recovery_from_state, lease_redeemed_at) " + f"ON file_import_job FROM {_DEFINER}" + ) + op.execute(f"REVOKE SELECT, INSERT ON file_import_job_event FROM {_DEFINER}") + op.execute( + "DROP POLICY file_import_job_event_file_dispatch_definer_insert " + "ON file_import_job_event" + ) + op.execute( + "DROP POLICY file_import_job_event_file_dispatch_definer_select " + "ON file_import_job_event" + ) + op.drop_index( + "ix_file_import_job_dispatch_expired", + table_name="file_import_job", + ) + op.execute(f"GRANT CREATE ON SCHEMA public TO {_DEFINER}") + op.execute(f"SET LOCAL ROLE {_DEFINER}") + op.execute( + f"ALTER FUNCTION public.{_FIRST_FUNCTION}{_SIGNATURE} RENAME TO {_FUNCTION}" + ) + op.execute( + f"GRANT EXECUTE ON FUNCTION public.{_FUNCTION}{_SIGNATURE} TO {_SCHEDULER}" + ) + op.execute("RESET ROLE") + op.execute(f"REVOKE CREATE ON SCHEMA public FROM {_DEFINER}") diff --git a/scripts/validate_security_catalog.py b/scripts/validate_security_catalog.py index d997e62c..42b31a68 100644 --- a/scripts/validate_security_catalog.py +++ b/scripts/validate_security_catalog.py @@ -2081,6 +2081,83 @@ ], } +CANONICAL_FILE_RECLAIM_ACTIVATION: dict[str, object] = { + "issueRef": "#93", + "invariantRef": "WORKER-LEASE-007", + "carrier": "bounded autonomous reclaim of expired scheduled File upserts", + "status": "active_fail_closed", + "policyEpochScope": "not-runtime-authority", + "controlBoundary": ( + "function-only scheduler login -> expired lease plus database-owned " + "exponential backoff -> current page/acquisition/audience/receiver " + "revalidation -> deterministic recovery-first SKIP LOCKED selector -> exact " + "next-generation WorkerLease -> existing durable-boundary File worker resume" + ), + "testEvidence": [ + { + "id": "PG-FILE-RECLAIM-093", + "surface": ( + "tests/integration/test_file_dispatch.py::" + "test_scheduler_reclaims_one_expired_dispatch_after_database_backoff " + "tests/integration/test_file_dispatch.py::" + "test_scheduler_does_not_reclaim_after_automatic_generation_budget " + "tests/integration/test_file_dispatch.py::" + "test_scheduler_reclaim_revalidates_current_membership_without_mutation" + ), + "oracle": ( + "The non-owner scheduler reclaims only after database expiry plus " + "generation-derived backoff, atomically advances to the exact next " + "generation with one digest-only event, returns content-free no-work " + "before eligibility, after the generation-four budget, or after " + "current Membership authority is revoked, and mutates nothing on " + "refusal." + ), + }, + { + "id": "PG-FILE-RECLAIM-CONCURRENCY-093", + "surface": ( + "tests/integration/test_file_dispatch.py::" + "test_concurrent_schedulers_reclaim_one_expired_generation_once" + ), + "oracle": ( + "Two scheduler-role authorities competing for one expired generation " + "yield exactly one next-generation lease and one immutable reclaim " + "event; the loser receives content-free no-work." + ), + }, + { + "id": "PROC-FILE-RECLAIM-093", + "surface": ( + "tests/integration/test_file_dispatch.py::" + "test_scheduler_recovers_interrupted_publication_and_stales_old_lease" + ), + "oracle": ( + "Across acquired, prepared, and indexed durable boundaries, the " + "scheduler and existing File worker resume with one higher-generation " + "lease, the replaced lease has zero effect, and exactly one active " + "Revision/publication effect remains." + ), + }, + ], + "deferredEvidence": [ + "dead-letter state, exhaustion transition, operator remediation, requeue, " + "and retry observability", + "provider polling, automatic page acceptance, full resync, and automatic " + "delete ordering", + ], + "futureCarriers": [ + "dead-letter and operator remediation owner", + "provider polling and full resync", + "explicit mixed-change ordering policy", + ], + "notActive": [ + "scheduler tenant, Source, job, attempt, timing, backoff, or generation choice", + "terminal failure retry, dead-letter transition, or operator requeue", + "manual import or delete recovery", + "Runtime authorization from Supply retry or lease state", + ], +} + CANONICAL_ACTIVATIONS: list[dict[str, object]] = [ CANONICAL_REVOCATION_ACTIVATION, CANONICAL_UNAVAILABLE_CAPABILITY_ACTIVATION, @@ -2103,6 +2180,7 @@ CANONICAL_FILE_DELETE_EXECUTION_ACTIVATION, CANONICAL_FILE_MIXED_UPSERT_SCHEDULING_ACTIVATION, CANONICAL_FILE_DISPATCH_ACTIVATION, + CANONICAL_FILE_RECLAIM_ACTIVATION, ] CANONICAL_ACTIVATION_ISSUE_LIST = ", ".join( f"Issue {activation['issueRef']}" for activation in CANONICAL_ACTIVATIONS diff --git a/sdk/typescript/test/live-empty-consumer.mjs b/sdk/typescript/test/live-empty-consumer.mjs index b672a54a..90408714 100644 --- a/sdk/typescript/test/live-empty-consumer.mjs +++ b/sdk/typescript/test/live-empty-consumer.mjs @@ -17,7 +17,11 @@ const client = new ContextEngineResolveClient({ const acquire = await client.resolve({ request: { kind: "acquire", - need: { query: "ContextEngine delivers context." }, + need: { + query: + process.env.CONTEXT_ENGINE_SDK_QUERY ?? + "ContextEngine delivers context.", + }, }, requestId: process.env.CONTEXT_ENGINE_SDK_REQUEST_ID, }); diff --git a/tests/catalog/test_validate_security_catalog.py b/tests/catalog/test_validate_security_catalog.py index a0f7c120..81e4f19d 100644 --- a/tests/catalog/test_validate_security_catalog.py +++ b/tests/catalog/test_validate_security_catalog.py @@ -36,6 +36,7 @@ CANONICAL_FILE_DELETE_OBSERVATION_ACTIVATION, CANONICAL_FILE_DISPATCH_ACTIVATION, CANONICAL_FILE_MIXED_UPSERT_SCHEDULING_ACTIVATION, + CANONICAL_FILE_RECLAIM_ACTIVATION, CANONICAL_INVARIANT_IDS, CANONICAL_MODEL_EGRESS_ACTIVATION, CANONICAL_OPENAPI_V0_ACTIVATION, @@ -589,6 +590,7 @@ def make_catalog() -> dict[str, object]: copy.deepcopy(CANONICAL_FILE_DELETE_EXECUTION_ACTIVATION), copy.deepcopy(CANONICAL_FILE_MIXED_UPSERT_SCHEDULING_ACTIVATION), copy.deepcopy(CANONICAL_FILE_DISPATCH_ACTIVATION), + copy.deepcopy(CANONICAL_FILE_RECLAIM_ACTIVATION), ], "invariants": invariants, "fixtures": fixtures, @@ -708,6 +710,7 @@ def make_schema() -> dict[str, object]: ) }, {"const": copy.deepcopy(CANONICAL_FILE_DISPATCH_ACTIVATION)}, + {"const": copy.deepcopy(CANONICAL_FILE_RECLAIM_ACTIVATION)}, ], "items": False, }, @@ -1109,7 +1112,7 @@ def test_issue_71_activates_private_accept_012_without_rewriting_m0_history( assert isinstance(upgrade_trigger, str) self.assertIn("Issue #71 activates", upgrade_trigger) self.assertEqual( - object_list_at(catalog, "activations")[-7], + object_list_at(catalog, "activations")[-8], CANONICAL_PRIVATE_BOT_DELIVERY_ACTIVATION, ) @@ -1355,6 +1358,55 @@ def test_activation_schema_independently_freezes_record_order_and_values( ) ) + def test_issue_93_file_reclaim_activation_is_independently_frozen(self) -> None: + catalog = load_document(DEFAULT_CATALOG_PATH) + schema = load_document(DEFAULT_SCHEMA_PATH) + activation = next( + item + for item in object_list_at(catalog, "activations") + if item["issueRef"] == "#93" + ) + schema_activations = object_at(schema, "properties", "activations") + prefix_items = schema_activations["prefixItems"] + assert isinstance(prefix_items, list) + schema_activation = object_at( + cast(dict[str, object], prefix_items[-1]), "const" + ) + + expected_boundary = ( + "function-only scheduler login -> expired lease plus database-owned " + "exponential backoff -> current page/acquisition/audience/receiver " + "revalidation -> deterministic recovery-first SKIP LOCKED selector -> " + "exact next-generation WorkerLease -> existing durable-boundary File " + "worker resume" + ) + expected_evidence_ids = [ + "PG-FILE-RECLAIM-093", + "PG-FILE-RECLAIM-CONCURRENCY-093", + "PROC-FILE-RECLAIM-093", + ] + expected_deferred = [ + "dead-letter state, exhaustion transition, operator remediation, " + "requeue, and retry observability", + "provider polling, automatic page acceptance, full resync, and " + "automatic delete ordering", + ] + expected_not_active = [ + "scheduler tenant, Source, job, attempt, timing, backoff, or generation " + "choice", + "terminal failure retry, dead-letter transition, or operator requeue", + "manual import or delete recovery", + "Runtime authorization from Supply retry or lease state", + ] + for record in (activation, schema_activation): + self.assertEqual(record["controlBoundary"], expected_boundary) + self.assertEqual( + [item["id"] for item in object_list_at(record, "testEvidence")], + expected_evidence_ids, + ) + self.assertEqual(record["deferredEvidence"], expected_deferred) + self.assertEqual(record["notActive"], expected_not_active) + def test_issue_67_action_prepare_activation_stops_before_effects(self) -> None: catalog = make_catalog() activation = object_list_at(catalog, "activations")[10] @@ -1461,7 +1513,7 @@ def test_issue_70_model_egress_activation_stops_before_real_provider(self) -> No def test_issue_71_private_bot_activation_stops_before_live_providers(self) -> None: catalog = make_catalog() - activation = object_list_at(catalog, "activations")[-7] + activation = object_list_at(catalog, "activations")[-8] self.assertEqual(activation, CANONICAL_PRIVATE_BOT_DELIVERY_ACTIVATION) self.assertEqual(activation["invariantRef"], "ACTION-SEPARATION-014") @@ -1485,7 +1537,7 @@ def test_issue_71_private_bot_activation_stops_before_live_providers(self) -> No def test_issue_81_file_change_activation_stops_before_scheduling(self) -> None: catalog = make_catalog() - activation = object_list_at(catalog, "activations")[-6] + activation = object_list_at(catalog, "activations")[-7] self.assertEqual(activation, CANONICAL_FILE_CHANGE_FEED_ACTIVATION) self.assertEqual(activation["invariantRef"], "WORKER-LEASE-007") @@ -1506,7 +1558,7 @@ def test_issue_81_file_change_activation_stops_before_scheduling(self) -> None: def test_issue_83_file_change_scheduling_stays_explicit(self) -> None: catalog = make_catalog() - activation = object_list_at(catalog, "activations")[-5] + activation = object_list_at(catalog, "activations")[-6] self.assertEqual( activation, @@ -1528,7 +1580,7 @@ def test_issue_83_file_change_scheduling_stays_explicit(self) -> None: def test_issue_85_file_delete_observation_has_no_execution_authority(self) -> None: catalog = make_catalog() - activation = object_list_at(catalog, "activations")[-4] + activation = object_list_at(catalog, "activations")[-5] self.assertEqual( activation, @@ -1552,7 +1604,7 @@ def test_issue_85_file_delete_observation_has_no_execution_authority(self) -> No def test_issue_87_executes_only_current_exact_file_deletes(self) -> None: catalog = make_catalog() - activation = object_list_at(catalog, "activations")[-3] + activation = object_list_at(catalog, "activations")[-4] self.assertEqual( activation, @@ -1573,7 +1625,7 @@ def test_issue_87_executes_only_current_exact_file_deletes(self) -> None: def test_issue_89_schedules_only_the_mixed_page_upsert_projection(self) -> None: catalog = make_catalog() - activation = object_list_at(catalog, "activations")[-2] + activation = object_list_at(catalog, "activations")[-3] self.assertEqual( activation, @@ -2014,7 +2066,7 @@ def test_tracked_catalog_freezes_issue_19_authority_and_bounded_scope( self.assertEqual(catalog["catalogVersion"], "1.3.0") self.assertEqual( - issue_refs[-21:], + issue_refs[-22:], [ "#15", "#16", @@ -2037,6 +2089,7 @@ def test_tracked_catalog_freezes_issue_19_authority_and_bounded_scope( "#87", "#89", "#91", + "#93", ], ) self.assertIn( diff --git a/tests/integration/test_file_dispatch.py b/tests/integration/test_file_dispatch.py index 56734ed0..68007eb5 100644 --- a/tests/integration/test_file_dispatch.py +++ b/tests/integration/test_file_dispatch.py @@ -14,6 +14,8 @@ from uuid import UUID, uuid4 import pytest +from alembic import command +from alembic.config import Config as AlembicConfig from sqlalchemy import Engine, create_engine, text from sqlalchemy.exc import DBAPIError from uvicorn import Config, Server @@ -37,13 +39,21 @@ DatabaseConfiguration, FileDispatchLease, FileDispatchNoWork, + FileImportInterrupted, + FilePublicationBoundary, PostgreSQLFileDispatchAuthority, + PostgreSQLFileImportWorker, PostgreSQLMembershipAuthority, create_database_engine, ) from engine.runtime.construction import Runtime, required_kernel_dependencies from engine.runtime.package_digest import QueryDigestKeyring -from engine.supply import WorkerLeaseCodec, WorkerLeaseKeyring +from engine.supply import ( + MarkdownCompilerConfig, + WorkerLeaseCodec, + WorkerLeaseKeyring, + WorkNotAvailable, +) from engine.supply.jobs import FILE_IMPORT_WORKER_LEASE_OPERATION from tests.integration.test_file_change_pages import ( _SCENARIOS, @@ -65,12 +75,14 @@ _unused_port, _wait_for_tcp, ) +from tests.support.migrations import HEAD_REVISION from tests.support.releases import ( clear_test_runtime_release, ensure_test_runtime_release, ) pytestmark = pytest.mark.integration +ROOT = Path(__file__).parents[2] SIGNING_KEY = b"issue-91-file-dispatch-key-00001" ALL_TEST_ROOTS = ("dispatch-root",) @@ -454,6 +466,59 @@ def test_scheduler_root_subset_cannot_redirect_global_oldest_selection( assert claim.job_id == scheduled[0][1] +def test_scheduler_root_subset_cannot_redirect_oldest_reclaim( + tmp_path: Path, + guarded_control_engine: Engine, + guarded_scheduler_engine: Engine, + migration_configuration: DatabaseConfiguration, +) -> None: + scheduled: list[tuple[UUID, UUID, FileRootRef]] = [] + for name in ("oldest-reclaim-root", "newer-reclaim-root"): + root = tmp_path / name + root.mkdir() + (root / "reclaim.md").write_text(f"# {name}\n", encoding="utf-8") + scheduled.append( + _schedule_one( + root=root, + guarded_control_engine=guarded_control_engine, + migration_configuration=migration_configuration, + root_ref=FileRootRef(name), + ) + ) + complete = PostgreSQLFileDispatchAuthority( + guarded_scheduler_engine, + WorkerLeaseCodec(WorkerLeaseKeyring(active_version=1, keys={1: SIGNING_KEY})), + configured_root_refs=("oldest-reclaim-root", "newer-reclaim-root"), + ) + first = complete.claim() + second = complete.claim() + assert type(first) is FileDispatchLease + assert type(second) is FileDispatchLease + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.begin() as connection: + connection.execute( + text( + "UPDATE file_import_job SET lease_issued_at = " + "clock_timestamp() - interval '10 minutes', " + "lease_expires_at = clock_timestamp() - interval '31 seconds' " + "WHERE job_id = ANY(:job_ids)" + ), + {"job_ids": [first.job_id, second.job_id]}, + ) + finally: + migration_engine.dispose() + subset = PostgreSQLFileDispatchAuthority( + guarded_scheduler_engine, + WorkerLeaseCodec(WorkerLeaseKeyring(active_version=1, keys={1: SIGNING_KEY})), + configured_root_refs=("newer-reclaim-root",), + ) + assert subset.claim() == FileDispatchNoWork() + reclaimed = complete.claim() + assert type(reclaimed) is FileDispatchLease + assert reclaimed.job_id == first.job_id + + @pytest.mark.parametrize( ("mutation", "parameters"), [ @@ -732,236 +797,1174 @@ def claim() -> FileDispatchLease | FileDispatchNoWork: assert len({claim.job_id for claim in claims}) == len(claims) == 2 -def test_dispatch_order_is_global_and_deterministic_across_organizations( +@pytest.mark.security_evidence(id="PG-FILE-RECLAIM-093", layer="postgres") +def test_scheduler_reclaims_one_expired_dispatch_after_database_backoff( tmp_path: Path, guarded_control_engine: Engine, guarded_scheduler_engine: Engine, migration_configuration: DatabaseConfiguration, ) -> None: - scheduled_jobs: list[tuple[UUID, UUID]] = [] - for name in ( - "sequence-first", - "page-first", - "change-first", - "stable-first", - "stable-last", - ): - root = tmp_path / name - root.mkdir() - (root / f"{name}.md").write_text(f"# {name}\n", encoding="utf-8") - organization_id, job_id, _root_ref = _schedule_one( - root=root, - guarded_control_engine=guarded_control_engine, - migration_configuration=migration_configuration, - ) - scheduled_jobs.append((organization_id, job_id)) + root = tmp_path / "reclaim" + root.mkdir() + (root / "reclaim.md").write_text("# Reclaim\n", encoding="utf-8") + organization_id, job_id, _root_ref = _schedule_one( + root=root, + guarded_control_engine=guarded_control_engine, + migration_configuration=migration_configuration, + ) + authority = _dispatch_authority(guarded_scheduler_engine) + first = authority.claim() + assert type(first) is FileDispatchLease migration_engine = create_database_engine(migration_configuration) try: with migration_engine.begin() as connection: connection.execute( text( - "ALTER TABLE file_acquisition DROP CONSTRAINT " - "fk_file_acquisition_change_observation_exact" - ) + "UPDATE file_import_job SET " + "lease_issued_at = clock_timestamp() - interval '10 minutes', " + "lease_expires_at = clock_timestamp() - interval '20 seconds' " + "WHERE organization_id = :organization_id AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, ) - for table in ( - "file_source_acquisition_checkpoint", - "file_source_change_page", - "file_source_change", - "file_acquisition", - ): - connection.execute( - text(f"ALTER TABLE {table} DISABLE TRIGGER {table}_immutable") - ) - positions = { - scheduled_jobs[0][1]: (10, 9, 9), - scheduled_jobs[1][1]: (20, 1, 9), - scheduled_jobs[2][1]: (20, 2, 1), - scheduled_jobs[3][1]: (20, 2, 9), - scheduled_jobs[4][1]: (20, 2, 9), - } - for job_id, (sequence, page_ordinal, change_ordinal) in positions.items(): - lineage = connection.execute( - text( - "SELECT acquisition.organization_id, acquisition.source_id, " - "acquisition.change_page_ref, acquisition.change_ordinal " - "FROM file_import_job AS job JOIN file_acquisition " - "AS acquisition " - "ON acquisition.organization_id = job.organization_id " - "AND acquisition.acquisition_id = job.acquisition_id " - "WHERE job.job_id = :job_id" - ), - {"job_id": job_id}, - ).one() - connection.execute( - text( - "UPDATE file_source_acquisition_checkpoint SET " - "accepted_at = TIMESTAMPTZ '2026-07-25 12:00:00+00', " - "sequence = :sequence WHERE job_id = :job_id" - ), - {"job_id": job_id, "sequence": sequence}, - ) - connection.execute( - text( - "UPDATE file_source_change_page SET page_ordinal = :ordinal " - "WHERE organization_id = :organization_id " - "AND source_id = :source_id AND page_ref = :page_ref" - ), - { - "organization_id": lineage.organization_id, - "source_id": lineage.source_id, - "page_ref": lineage.change_page_ref, - "ordinal": page_ordinal, - }, - ) - connection.execute( - text( - "UPDATE file_source_change SET change_ordinal = :new_ordinal " - "WHERE organization_id = :organization_id " - "AND source_id = :source_id AND page_ref = :page_ref " - "AND change_ordinal = :old_ordinal" - ), - { - "organization_id": lineage.organization_id, - "source_id": lineage.source_id, - "page_ref": lineage.change_page_ref, - "old_ordinal": lineage.change_ordinal, - "new_ordinal": change_ordinal, - }, - ) - connection.execute( - text( - "UPDATE file_acquisition SET change_ordinal = :new_ordinal " - "WHERE organization_id = :organization_id " - "AND source_id = :source_id AND change_page_ref = :page_ref" - ), - { - "organization_id": lineage.organization_id, - "source_id": lineage.source_id, - "page_ref": lineage.change_page_ref, - "new_ordinal": change_ordinal, - }, - ) + assert authority.claim() == FileDispatchNoWork() + with migration_engine.begin() as connection: connection.execute( text( - "ALTER TABLE file_acquisition ADD CONSTRAINT " - "fk_file_acquisition_change_observation_exact FOREIGN KEY (" - "organization_id, source_id, source_version_id, change_page_ref, " - "change_ordinal, relative_path, expected_content_sha256, " - "expected_content_length) REFERENCES file_source_change (" - "organization_id, source_id, source_version_id, page_ref, " - "change_ordinal, relative_path, content_sha256, content_length)" - ) + "UPDATE file_import_job SET " + "lease_expires_at = clock_timestamp() - interval '31 seconds' " + "WHERE organization_id = :organization_id AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, ) - for table in ( - "file_source_acquisition_checkpoint", - "file_source_change_page", - "file_source_change", - "file_acquisition", - ): - connection.execute( - text(f"ALTER TABLE {table} ENABLE TRIGGER {table}_immutable") - ) + reclaimed = authority.claim() + assert type(reclaimed) is FileDispatchLease + assert reclaimed.job_id == job_id + assert reclaimed.lease_generation == 2 + with migration_engine.connect() as connection: + event = connection.execute( + text( + "SELECT ordinal, event_type, boundary, lease_generation, " + "state_at_event, reason_digest ~ '^[0-9a-f]{64}$' " + "FROM file_import_job_event " + "WHERE organization_id = :organization_id AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ).one() finally: migration_engine.dispose() - authority = _dispatch_authority(guarded_scheduler_engine) - - claims = tuple(authority.claim() for _index in range(5)) - - assert all(type(claim) is FileDispatchLease for claim in claims) - stable_tie = sorted(scheduled_jobs[3:]) - expected = [ - scheduled_jobs[0][1], - scheduled_jobs[1][1], - scheduled_jobs[2][1], - *(job_id for _organization_id, job_id in stable_tie), - ] - assert [ - claim.job_id for claim in claims if type(claim) is FileDispatchLease - ] == expected + assert tuple(event) == (0, "reclaimed", "acquired", 2, "leased", True) -def test_post_claim_mint_failure_leaves_one_expiring_lease_and_zero_effect( +@pytest.mark.parametrize( + ("generation", "before_seconds", "eligible_seconds"), + [(2, 50, 61), (3, 110, 121)], +) +def test_scheduler_uses_generation_derived_database_backoff( tmp_path: Path, guarded_control_engine: Engine, guarded_scheduler_engine: Engine, migration_configuration: DatabaseConfiguration, - monkeypatch: pytest.MonkeyPatch, + generation: int, + before_seconds: int, + eligible_seconds: int, ) -> None: - root = tmp_path / "crash-root" + root = tmp_path / f"backoff-{generation}" root.mkdir() - (root / "crash.md").write_text("# Crash\n", encoding="utf-8") + (root / "backoff.md").write_text("# Backoff\n", encoding="utf-8") organization_id, job_id, _root_ref = _schedule_one( root=root, guarded_control_engine=guarded_control_engine, migration_configuration=migration_configuration, ) - codec = WorkerLeaseCodec( - WorkerLeaseKeyring(active_version=1, keys={1: SIGNING_KEY}) - ) - monkeypatch.setattr( - WorkerLeaseCodec, - "mint", - lambda _self, _claims: (_ for _ in ()).throw(RuntimeError("signing failed")), - ) - authority = _dispatch_authority(guarded_scheduler_engine, codec) - - with pytest.raises(RuntimeError, match="signing failed"): - authority.claim() - + authority = _dispatch_authority(guarded_scheduler_engine) + assert type(authority.claim()) is FileDispatchLease migration_engine = create_database_engine(migration_configuration) try: - with migration_engine.connect() as connection: - snapshot = connection.execute( + with migration_engine.begin() as connection: + connection.execute( text( - "SELECT job.state, job.lease_generation, job.dispatch_claimed, " - "job.lease_expires_at > job.lease_issued_at, " - "(SELECT count(*) FROM context_revision " - " WHERE organization_id = :organization_id), " - "(SELECT count(*) FROM exact_phrase_candidate " - " WHERE organization_id = :organization_id), " - "(SELECT count(*) FROM file_source_publish_watermark " - " WHERE organization_id = :organization_id) " - "FROM file_import_job AS job " - "WHERE job.organization_id = :organization_id " - "AND job.job_id = :job_id" + "UPDATE file_import_job SET lease_generation = :generation, " + "lease_issued_at = clock_timestamp() - interval '20 minutes', " + "lease_expires_at = clock_timestamp() - " + "make_interval(secs => :before_seconds) WHERE " + "organization_id = :organization_id AND job_id = :job_id" ), - {"organization_id": organization_id, "job_id": job_id}, - ).one() + { + "generation": generation, + "before_seconds": before_seconds, + "organization_id": organization_id, + "job_id": job_id, + }, + ) + assert authority.claim() == FileDispatchNoWork() + with migration_engine.begin() as connection: + connection.execute( + text( + "UPDATE file_import_job SET lease_expires_at = " + "clock_timestamp() - make_interval(secs => :eligible_seconds) " + "WHERE organization_id = :organization_id AND job_id = :job_id" + ), + { + "eligible_seconds": eligible_seconds, + "organization_id": organization_id, + "job_id": job_id, + }, + ) + reclaimed = authority.claim() finally: migration_engine.dispose() - assert tuple(snapshot) == ("leased", 1, True, True, 0, 0, 0) - - monkeypatch.undo() - assert ( - _dispatch_authority( - guarded_scheduler_engine, - codec, - ).claim() - == FileDispatchNoWork() - ) + assert type(reclaimed) is FileDispatchLease + assert reclaimed.lease_generation == generation + 1 -def test_claim_skips_authority_row_while_revocation_is_in_flight( +def test_scheduler_does_not_reclaim_after_automatic_generation_budget( tmp_path: Path, guarded_control_engine: Engine, guarded_scheduler_engine: Engine, migration_configuration: DatabaseConfiguration, ) -> None: - root = tmp_path / "revocation-race-root" + root = tmp_path / "exhausted" root.mkdir() - (root / "race.md").write_text("# Race\n", encoding="utf-8") + (root / "exhausted.md").write_text("# Exhausted\n", encoding="utf-8") organization_id, job_id, _root_ref = _schedule_one( root=root, guarded_control_engine=guarded_control_engine, migration_configuration=migration_configuration, ) + authority = _dispatch_authority(guarded_scheduler_engine) + assert type(authority.claim()) is FileDispatchLease migration_engine = create_database_engine(migration_configuration) try: - with migration_engine.connect() as revoker: - transaction = revoker.begin() - revoker.execute( - text( + with migration_engine.begin() as connection: + connection.execute( + text( + "UPDATE file_import_job SET lease_generation = 4, " + "lease_issued_at = clock_timestamp() - interval '20 minutes', " + "lease_expires_at = clock_timestamp() - interval '10 minutes' " + "WHERE organization_id = :organization_id AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ) + assert authority.claim() == FileDispatchNoWork() + with migration_engine.connect() as connection: + snapshot = connection.execute( + text( + "SELECT state, lease_generation, " + "(SELECT count(*) FROM file_import_job_event WHERE " + "organization_id = :organization_id AND job_id = :job_id) " + "FROM file_import_job WHERE organization_id = :organization_id " + "AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ).one() + finally: + migration_engine.dispose() + assert tuple(snapshot) == ("leased", 4, 0) + + +@pytest.mark.security_evidence(id="PG-FILE-RECLAIM-CONCURRENCY-093", layer="postgres") +def test_concurrent_schedulers_reclaim_one_expired_generation_once( + tmp_path: Path, + guarded_control_engine: Engine, + guarded_scheduler_engine: Engine, + scheduler_configuration: DatabaseConfiguration, + migration_configuration: DatabaseConfiguration, +) -> None: + root = tmp_path / "reclaim-concurrent" + root.mkdir() + (root / "concurrent.md").write_text("# Concurrent\n", encoding="utf-8") + organization_id, job_id, _root_ref = _schedule_one( + root=root, + guarded_control_engine=guarded_control_engine, + migration_configuration=migration_configuration, + ) + first = _dispatch_authority(guarded_scheduler_engine).claim() + assert type(first) is FileDispatchLease + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.begin() as connection: + connection.execute( + text( + "UPDATE file_import_job SET " + "lease_issued_at = clock_timestamp() - interval '10 minutes', " + "lease_expires_at = clock_timestamp() - interval '31 seconds' " + "WHERE organization_id = :organization_id AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ) + + def claim() -> FileDispatchLease | FileDispatchNoWork: + engine = create_database_engine(scheduler_configuration) + try: + return _dispatch_authority(engine).claim() + finally: + engine.dispose() + + with ThreadPoolExecutor(max_workers=2) as executor: + results = tuple(executor.map(lambda _index: claim(), range(2))) + assert sum(type(result) is FileDispatchLease for result in results) == 1 + assert sum(type(result) is FileDispatchNoWork for result in results) == 1 + claimed = next( + result for result in results if type(result) is FileDispatchLease + ) + assert claimed.job_id == job_id + assert claimed.lease_generation == 2 + with migration_engine.connect() as connection: + snapshot = connection.execute( + text( + "SELECT lease_generation, (SELECT count(*) FROM " + "file_import_job_event WHERE organization_id = :organization_id " + "AND job_id = :job_id AND event_type = 'reclaimed') " + "FROM file_import_job WHERE organization_id = :organization_id " + "AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ).one() + finally: + migration_engine.dispose() + assert tuple(snapshot) == (2, 1) + + +def test_scheduler_reclaim_revalidates_current_membership_without_mutation( + tmp_path: Path, + guarded_control_engine: Engine, + guarded_scheduler_engine: Engine, + migration_configuration: DatabaseConfiguration, +) -> None: + root = tmp_path / "reclaim-revoked" + root.mkdir() + (root / "revoked.md").write_text("# Revoked\n", encoding="utf-8") + organization_id, job_id, _root_ref = _schedule_one( + root=root, + guarded_control_engine=guarded_control_engine, + migration_configuration=migration_configuration, + ) + authority = _dispatch_authority(guarded_scheduler_engine) + assert type(authority.claim()) is FileDispatchLease + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.begin() as connection: + connection.execute( + text( + "UPDATE file_import_job SET " + "lease_issued_at = clock_timestamp() - interval '10 minutes', " + "lease_expires_at = clock_timestamp() - interval '31 seconds' " + "WHERE organization_id = :organization_id AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ) + connection.execute( + text( + "UPDATE membership SET status = 'revoked' " + "WHERE organization_id = :organization_id" + ), + {"organization_id": organization_id}, + ) + assert authority.claim() == FileDispatchNoWork() + with migration_engine.connect() as connection: + snapshot = connection.execute( + text( + "SELECT state, lease_generation, (SELECT count(*) FROM " + "file_import_job_event WHERE organization_id = :organization_id " + "AND job_id = :job_id) FROM file_import_job " + "WHERE organization_id = :organization_id AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ).one() + finally: + migration_engine.dispose() + assert tuple(snapshot) == ("leased", 1, 0) + + +@pytest.mark.parametrize( + "mutation", + [ + "UPDATE context_source SET lifecycle_state = 'disabled', " + "disabled_version_id = active_version_id, disabled_at = now() " + "WHERE organization_id = :organization_id", + "UPDATE service_principal SET enabled = false " + "WHERE organization_id = :organization_id", + ], +) +def test_scheduler_reclaim_revalidates_source_and_receiver_without_mutation( + tmp_path: Path, + guarded_control_engine: Engine, + guarded_scheduler_engine: Engine, + migration_configuration: DatabaseConfiguration, + mutation: str, +) -> None: + root = tmp_path / "reclaim-authority" + root.mkdir() + (root / "authority.md").write_text("# Authority\n", encoding="utf-8") + organization_id, job_id, _root_ref = _schedule_one( + root=root, + guarded_control_engine=guarded_control_engine, + migration_configuration=migration_configuration, + ) + authority = _dispatch_authority(guarded_scheduler_engine) + assert type(authority.claim()) is FileDispatchLease + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.begin() as connection: + connection.execute( + text( + "UPDATE file_import_job SET lease_issued_at = " + "clock_timestamp() - interval '10 minutes', " + "lease_expires_at = clock_timestamp() - interval '31 seconds' " + "WHERE organization_id = :organization_id AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ) + connection.execute(text(mutation), {"organization_id": organization_id}) + assert authority.claim() == FileDispatchNoWork() + with migration_engine.connect() as connection: + snapshot = connection.execute( + text( + "SELECT state, lease_generation, (SELECT count(*) FROM " + "file_import_job_event WHERE organization_id = :organization_id " + "AND job_id = :job_id) FROM file_import_job WHERE " + "organization_id = :organization_id AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ).one() + finally: + migration_engine.dispose() + assert tuple(snapshot) == ("leased", 1, 0) + + +def test_scheduler_reclaim_refuses_a_superseded_scan_without_mutation( + tmp_path: Path, + guarded_control_engine: Engine, + guarded_scheduler_engine: Engine, + migration_configuration: DatabaseConfiguration, +) -> None: + root = tmp_path / "reclaim-stale-scan" + root.mkdir() + (root / "stale.md").write_text("# Stale\n", encoding="utf-8") + organization_id, job_id, _root_ref = _schedule_one( + root=root, + guarded_control_engine=guarded_control_engine, + migration_configuration=migration_configuration, + ) + authority = _dispatch_authority(guarded_scheduler_engine) + assert type(authority.claim()) is FileDispatchLease + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.begin() as connection: + source_id, version_id, accepted_at, sequence = connection.execute( + text( + "SELECT source.source_id, source.active_version_id, " + "page.accepted_at, max(checkpoint.sequence) OVER () FROM " + "context_source AS source JOIN file_source_change_page AS page " + "ON page.organization_id = source.organization_id AND " + "page.source_id = source.source_id JOIN " + "file_source_acquisition_checkpoint AS checkpoint ON " + "checkpoint.organization_id = page.organization_id AND " + "checkpoint.source_id = page.source_id WHERE " + "source.organization_id = :organization_id ORDER BY " + "checkpoint.sequence DESC LIMIT 1" + ), + {"organization_id": organization_id}, + ).one() + connection.execute( + text( + "SELECT set_config('app.organization_id', " + "CAST(:organization_id AS text), true)" + ), + {"organization_id": organization_id}, + ) + connection.execute( + text( + "SELECT set_config('app.file_source_id', " + "CAST(:source_id AS text), true)" + ), + {"source_id": source_id}, + ) + connection.execute( + text( + "SELECT set_config('app.file_source_version_id', " + "CAST(:version_id AS text), true)" + ), + {"version_id": version_id}, + ) + page_ref = sha256(organization_id.bytes + b"reclaim-stale-page").hexdigest() + connection.execute( + text( + "INSERT INTO file_source_delete_observation_page (" + "organization_id, source_id, source_version_id, page_ref) " + "VALUES (:organization_id, :source_id, :version_id, :page_ref)" + ), + { + "organization_id": organization_id, + "source_id": source_id, + "version_id": version_id, + "page_ref": page_ref, + }, + ) + connection.execute( + text( + "INSERT INTO file_source_change_page (organization_id, " + "source_id, source_version_id, page_ref, scan_ref, scan_epoch, " + "page_limit, page_ordinal, change_count, complete, accepted_at) " + "VALUES (:organization_id, :source_id, :version_id, :page_ref, " + ":scan_ref, :scan_epoch, 1, 1, 0, true, " + ":accepted_at + interval '1 second')" + ), + { + "organization_id": organization_id, + "source_id": source_id, + "version_id": version_id, + "page_ref": page_ref, + "scan_ref": sha256( + organization_id.bytes + b"reclaim-stale-scan" + ).hexdigest(), + "scan_epoch": uuid4(), + "accepted_at": accepted_at, + }, + ) + connection.execute( + text( + "INSERT INTO file_source_acquisition_checkpoint (" + "organization_id, source_id, sequence, checkpoint_ref, " + "change_kind, accepted_at, source_version_id, change_page_ref) " + "VALUES (:organization_id, :source_id, :sequence, " + ":checkpoint_ref, 'file_change_page', " + ":accepted_at + interval '1 second', :version_id, :page_ref)" + ), + { + "organization_id": organization_id, + "source_id": source_id, + "sequence": sequence + 1, + "checkpoint_ref": "facp_" + + sha256( + organization_id.bytes + b"reclaim-stale-checkpoint" + ).hexdigest(), + "accepted_at": accepted_at, + "version_id": version_id, + "page_ref": page_ref, + }, + ) + connection.execute( + text( + "UPDATE file_import_job SET lease_issued_at = " + "clock_timestamp() - interval '10 minutes', " + "lease_expires_at = clock_timestamp() - interval '31 seconds' " + "WHERE organization_id = :organization_id AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ) + assert authority.claim() == FileDispatchNoWork() + with migration_engine.connect() as connection: + snapshot = connection.execute( + text( + "SELECT state, lease_generation, (SELECT count(*) FROM " + "file_import_job_event WHERE organization_id = :organization_id " + "AND job_id = :job_id) FROM file_import_job WHERE " + "organization_id = :organization_id AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ).one() + finally: + migration_engine.dispose() + assert tuple(snapshot) == ("leased", 1, 0) + + +@pytest.mark.security_evidence(id="PROC-FILE-RECLAIM-093", layer="runtime") +@pytest.mark.parametrize( + "boundary", + [ + None, + FilePublicationBoundary.ACQUIRED, + FilePublicationBoundary.PREPARED, + FilePublicationBoundary.INDEXED, + ], +) +def test_scheduler_recovers_interrupted_publication_and_stales_old_lease( + tmp_path: Path, + request: pytest.FixtureRequest, + guarded_control_engine: Engine, + guarded_scheduler_engine: Engine, + guarded_worker_engine: Engine, + guarded_runtime_engine: Engine, + query_digest_keyring: QueryDigestKeyring, + migration_configuration: DatabaseConfiguration, + boundary: FilePublicationBoundary | None, +) -> None: + root = tmp_path / "resume" + root.mkdir() + (root / "resume.md").write_text( + "# Resume\n\nContextEngine delivers context.\n", + encoding="utf-8", + ) + organization_id, job_id, root_ref = _schedule_one( + root=root, + guarded_control_engine=guarded_control_engine, + migration_configuration=migration_configuration, + ) + codec = WorkerLeaseCodec( + WorkerLeaseKeyring(active_version=1, keys={1: SIGNING_KEY}) + ) + authority = _dispatch_authority(guarded_scheduler_engine, codec) + first = authority.claim() + assert type(first) is FileDispatchLease + roots = FileRootRegistry( + {root_ref: root}, limits=FileReadLimits(max_file_bytes=4_096) + ) + request.addfinalizer(roots.close) + if boundary is not None: + interrupted_worker = PostgreSQLFileImportWorker( + guarded_worker_engine, + codec, + FileImportReceiver(first.service_principal_id), + roots, + MarkdownCompilerConfig("markdown-config-v1"), + clock=lambda: first.issued_at, + interrupt_after=boundary, + ) + with pytest.raises(FileImportInterrupted): + interrupted_worker.run(first.redemption) + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.begin() as connection: + connection.execute( + text( + "UPDATE file_import_job SET " + "lease_issued_at = clock_timestamp() - interval '10 minutes', " + "lease_expires_at = clock_timestamp() - interval '31 seconds' " + "WHERE organization_id = :organization_id AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ) + completed = subprocess.run( + ["context-engine-worker", "--dispatch-file-once"], + env={ + **os.environ, + "CONTEXT_ENGINE_WORKER_LEASE_SIGNING_KEY_HEX": SIGNING_KEY.hex(), + "CONTEXT_ENGINE_WORKER_FILE_ROOTS_JSON": json.dumps( + {root_ref.value: str(root)} + ), + }, + check=True, + capture_output=True, + text=True, + ) + assert json.loads(completed.stdout) == { + "dispatch": "file.import", + "outcome": "dispatched", + "service": "context-engine-worker", + "status": "complete", + } + forbidden = (str(root), str(organization_id), str(job_id), SIGNING_KEY.hex()) + assert all(value not in completed.stdout for value in forbidden) + stale_worker = PostgreSQLFileImportWorker( + guarded_worker_engine, + codec, + FileImportReceiver(first.service_principal_id), + roots, + MarkdownCompilerConfig("markdown-config-v1"), + clock=lambda: first.issued_at, + ) + with pytest.raises(WorkNotAvailable): + stale_worker.run(first.redemption) + with migration_engine.connect() as connection: + snapshot = connection.execute( + text( + "SELECT job.state, job.lease_generation, job.effect_count, " + "(SELECT count(*) FROM context_revision WHERE " + "organization_id = :organization_id), " + "(SELECT count(*) FROM file_import_job_event WHERE " + "organization_id = :organization_id AND job_id = :job_id " + "AND event_type = 'reclaimed') FROM file_import_job AS job " + "WHERE job.organization_id = :organization_id " + "AND job.job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ).one() + candidate = connection.execute( + text( + "SELECT source_ref, resource_ref, CAST(revision_id AS text) " + "AS revision_ref FROM exact_phrase_candidate WHERE " + "organization_id = :organization_id" + ), + {"organization_id": organization_id}, + ).one() + membership_id, user_id = connection.execute( + text( + "SELECT membership_id, user_id FROM membership WHERE " + "organization_id = :organization_id" + ), + {"organization_id": organization_id}, + ).one() + finally: + migration_engine.dispose() + assert tuple(snapshot) == ("completed", 2, 1, 1, 1) + if boundary in (None, FilePublicationBoundary.INDEXED): + ensure_test_runtime_release( + organization_id, + active_revision_refs=(candidate.revision_ref,), + ) + consumer_root = tmp_path / "reclaim-sdk-consumer" + consumer_root.mkdir() + _pack_and_install_resolve_sdk(consumer_root) + application = create_app( + authenticator=_RuntimeAuthenticator( + organization_id, + user_id, + membership_id, + ), + organization_authority=_OrganizationAuthority(), + membership_authority=PostgreSQLMembershipAuthority(guarded_runtime_engine), + scope_authority=_ExactScopeAuthority( + candidate.source_ref, + candidate.resource_ref, + ), + runtime=Runtime( + required_kernel_dependencies(), + candidate_index=PostgreSQLExactPhraseCandidateIndex(), + query_digest_keyring=query_digest_keyring, + ), + ) + port = _unused_port() + server = Server( + Config( + application, + host="127.0.0.1", + port=port, + log_level="warning", + lifespan="off", + ) + ) + thread = Thread(target=server.run, daemon=True) + thread.start() + try: + _wait_for_tcp(port) + resolved = _run_installed_empty_consumer( + consumer_root, + base_url=f"http://127.0.0.1:{port}", + ) + package = cast(dict[str, object], resolved["package"]) + blocks = cast(list[dict[str, object]], package["blocks"]) + evidence = cast(list[dict[str, object]], package["evidence"]) + assert [block["text"] for block in blocks] == [ + "ContextEngine delivers context." + ] + assert len(evidence) == 1 + assert evidence[0]["revisionRef"] == candidate.revision_ref + finally: + server.should_exit = True + thread.join(timeout=10) + clear_test_runtime_release(organization_id) + assert not thread.is_alive() + + +def test_autonomous_replacement_reclaim_is_all_old_then_all_new_over_sdk( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + guarded_control_engine: Engine, + guarded_scheduler_engine: Engine, + guarded_worker_engine: Engine, + guarded_runtime_engine: Engine, + query_digest_keyring: QueryDigestKeyring, + migration_configuration: DatabaseConfiguration, +) -> None: + root = tmp_path / "replacement-reclaim" + root.mkdir() + path = root / "handbook.md" + path.write_text("# Handbook\n\nOLD marker.\n", encoding="utf-8") + provider_proofs, control_proofs = _proofs() + organization_id = uuid4() + receiver = FileImportReceiver(uuid4()) + control, authority, source = _seed_file_change_source( + guarded_control_engine=guarded_control_engine, + migration_configuration=migration_configuration, + organization_id=organization_id, + receiver=receiver, + root_ref=FileRootRef("dispatch-root"), + control_proofs=control_proofs, + ) + source = _activate_delete_observations(control, authority, organization_id, source) + provider = FileChangeProvider( + FileRootRegistry( + {source.source_version.root_ref: root}, + limits=FileReadLimits(max_file_bytes=4_096), + ), + proofs=provider_proofs, + ) + initial = provider.read_changes(source, InitialScan(), ChangeLimit(1)) + assert type(initial) is ProviderOk + with _authorize( + authority, + organization_id, + ControlOperation.ACCEPT_FILE_CHANGE_PAGE, + "accept-reclaim-old", + ) as call: + accepted_old = control.accept_file_change_page(call, initial.value) + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.connect() as connection: + membership_id, user_id = connection.execute( + text( + "SELECT membership_id, user_id FROM membership WHERE " + "organization_id = :organization_id" + ), + {"organization_id": organization_id}, + ).one() + finally: + migration_engine.dispose() + audience = FileImportAudience("principal:file-reader", membership_id, 1) + with _authorize( + authority, + organization_id, + ControlOperation.SCHEDULE_FILE_CHANGE_PAGE, + "schedule-reclaim-old", + ) as call: + control.schedule_file_change_page( + call, + ScheduleFileChangePage( + accepted_old.source_ref, + accepted_old.source_version_ref, + accepted_old.page_ref, + audience, + ), + ) + codec = WorkerLeaseCodec( + WorkerLeaseKeyring(active_version=1, keys={1: SIGNING_KEY}) + ) + dispatch = _dispatch_authority(guarded_scheduler_engine, codec) + old_claim = dispatch.claim() + assert type(old_claim) is FileDispatchLease + roots = FileRootRegistry( + {source.source_version.root_ref: root}, + limits=FileReadLimits(max_file_bytes=4_096), + ) + old_result = PostgreSQLFileImportWorker( + guarded_worker_engine, + codec, + receiver, + roots, + MarkdownCompilerConfig("markdown-config-v1"), + clock=lambda: old_claim.issued_at, + ).run(old_claim.redemption) + with _authorize( + authority, + organization_id, + ControlOperation.READ_SOURCE_PROGRESS, + "read-reclaim-old", + ) as call: + progress = control.read_file_source_progress(call, accepted_old.source_ref) + assert progress.complete_change_baseline is not None + path.write_text("# Handbook\n\nNEW marker.\n", encoding="utf-8") + changed = provider.read_changes( + FileChangeSource( + organization_id, + source.source_version, + scan_head=progress.change_scan_head, + complete_baseline=progress.complete_change_baseline, + ), + InitialScan(), + ChangeLimit(1), + ) + assert type(changed) is ProviderOk + with _authorize( + authority, + organization_id, + ControlOperation.ACCEPT_FILE_CHANGE_PAGE, + "accept-reclaim-new", + ) as call: + accepted_new = control.accept_file_change_page(call, changed.value) + with _authorize( + authority, + organization_id, + ControlOperation.SCHEDULE_FILE_CHANGE_PAGE, + "schedule-reclaim-new", + ) as call: + scheduled_new = control.schedule_file_change_page( + call, + ScheduleFileChangePage( + accepted_new.source_ref, + accepted_new.source_version_ref, + accepted_new.page_ref, + audience, + ), + ) + new_claim = dispatch.claim() + assert type(new_claim) is FileDispatchLease + with pytest.raises(FileImportInterrupted): + PostgreSQLFileImportWorker( + guarded_worker_engine, + codec, + receiver, + roots, + MarkdownCompilerConfig("markdown-config-v1"), + clock=lambda: new_claim.issued_at, + interrupt_after=FilePublicationBoundary.INDEXED, + ).run(new_claim.redemption) + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.connect() as connection: + pending_revision = connection.execute( + text( + "SELECT CAST(revision_id AS text) FROM file_import_job WHERE " + "organization_id = :organization_id AND job_id = :job_id" + ), + { + "organization_id": organization_id, + "job_id": scheduled_new.changes[0].prepared_import.job_id, + }, + ).scalar_one() + candidate = connection.execute( + text( + "SELECT source_ref, resource_ref FROM exact_phrase_candidate " + "WHERE organization_id = :organization_id AND " + "revision_id = :revision_id" + ), + { + "organization_id": organization_id, + "revision_id": pending_revision, + }, + ).one() + finally: + migration_engine.dispose() + ensure_test_runtime_release( + organization_id, + active_revision_refs=( + old_result.candidate_refs[0].revision_ref, + pending_revision, + ), + ) + consumer_root = tmp_path / "replacement-sdk-consumer" + consumer_root.mkdir() + _pack_and_install_resolve_sdk(consumer_root) + application = create_app( + authenticator=_RuntimeAuthenticator( + organization_id, + user_id, + membership_id, + ), + organization_authority=_OrganizationAuthority(), + membership_authority=PostgreSQLMembershipAuthority(guarded_runtime_engine), + scope_authority=_ExactScopeAuthority( + candidate.source_ref, + candidate.resource_ref, + ), + runtime=Runtime( + required_kernel_dependencies(), + candidate_index=PostgreSQLExactPhraseCandidateIndex(), + query_digest_keyring=query_digest_keyring, + ), + ) + port = _unused_port() + server = Server( + Config( + application, + host="127.0.0.1", + port=port, + log_level="warning", + lifespan="off", + ) + ) + thread = Thread(target=server.run, daemon=True) + thread.start() + try: + _wait_for_tcp(port) + base_url = f"http://127.0.0.1:{port}" + monkeypatch.setenv("CONTEXT_ENGINE_SDK_QUERY", "OLD marker.") + old = _run_installed_empty_consumer( + consumer_root, + base_url=base_url, + ) + monkeypatch.setenv("CONTEXT_ENGINE_SDK_QUERY", "NEW marker.") + hidden_new = _run_installed_empty_consumer( + consumer_root, + base_url=base_url, + ) + old_package = cast(dict[str, object], old["package"]) + assert len(cast(list[object], old_package["evidence"])) == 1 + assert cast(dict[str, object], hidden_new["package"])["evidence"] == [] + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.begin() as connection: + connection.execute( + text( + "UPDATE file_import_job SET lease_issued_at = " + "clock_timestamp() - interval '10 minutes', " + "lease_expires_at = " + "clock_timestamp() - interval '31 seconds' WHERE " + "organization_id = :organization_id AND job_id = :job_id" + ), + { + "organization_id": organization_id, + "job_id": scheduled_new.changes[0].prepared_import.job_id, + }, + ) + finally: + migration_engine.dispose() + completed = subprocess.run( + ["context-engine-worker", "--dispatch-file-once"], + env={ + **os.environ, + "CONTEXT_ENGINE_WORKER_LEASE_SIGNING_KEY_HEX": SIGNING_KEY.hex(), + "CONTEXT_ENGINE_WORKER_FILE_ROOTS_JSON": json.dumps( + {source.source_version.root_ref.value: str(root)} + ), + }, + check=True, + capture_output=True, + text=True, + ) + assert json.loads(completed.stdout)["outcome"] == "dispatched" + monkeypatch.setenv("CONTEXT_ENGINE_SDK_QUERY", "OLD marker.") + hidden_old = _run_installed_empty_consumer( + consumer_root, + base_url=base_url, + ) + monkeypatch.setenv("CONTEXT_ENGINE_SDK_QUERY", "NEW marker.") + new = _run_installed_empty_consumer( + consumer_root, + base_url=base_url, + ) + assert cast(dict[str, object], hidden_old["package"])["evidence"] == [] + new_package = cast(dict[str, object], new["package"]) + assert len(cast(list[object], new_package["evidence"])) == 1 + finally: + server.should_exit = True + thread.join(timeout=10) + clear_test_runtime_release(organization_id) + roots.close() + assert not thread.is_alive() + + +def test_dispatch_order_is_global_and_deterministic_across_organizations( + tmp_path: Path, + guarded_control_engine: Engine, + guarded_scheduler_engine: Engine, + migration_configuration: DatabaseConfiguration, +) -> None: + scheduled_jobs: list[tuple[UUID, UUID]] = [] + for name in ( + "sequence-first", + "page-first", + "change-first", + "stable-first", + "stable-last", + ): + root = tmp_path / name + root.mkdir() + (root / f"{name}.md").write_text(f"# {name}\n", encoding="utf-8") + organization_id, job_id, _root_ref = _schedule_one( + root=root, + guarded_control_engine=guarded_control_engine, + migration_configuration=migration_configuration, + ) + scheduled_jobs.append((organization_id, job_id)) + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.begin() as connection: + connection.execute( + text( + "ALTER TABLE file_acquisition DROP CONSTRAINT " + "fk_file_acquisition_change_observation_exact" + ) + ) + for table in ( + "file_source_acquisition_checkpoint", + "file_source_change_page", + "file_source_change", + "file_acquisition", + ): + connection.execute( + text(f"ALTER TABLE {table} DISABLE TRIGGER {table}_immutable") + ) + positions = { + scheduled_jobs[0][1]: (10, 9, 9), + scheduled_jobs[1][1]: (20, 1, 9), + scheduled_jobs[2][1]: (20, 2, 1), + scheduled_jobs[3][1]: (20, 2, 9), + scheduled_jobs[4][1]: (20, 2, 9), + } + for job_id, (sequence, page_ordinal, change_ordinal) in positions.items(): + lineage = connection.execute( + text( + "SELECT acquisition.organization_id, acquisition.source_id, " + "acquisition.change_page_ref, acquisition.change_ordinal " + "FROM file_import_job AS job JOIN file_acquisition " + "AS acquisition " + "ON acquisition.organization_id = job.organization_id " + "AND acquisition.acquisition_id = job.acquisition_id " + "WHERE job.job_id = :job_id" + ), + {"job_id": job_id}, + ).one() + connection.execute( + text( + "UPDATE file_source_acquisition_checkpoint SET " + "accepted_at = TIMESTAMPTZ '2026-07-25 12:00:00+00', " + "sequence = :sequence WHERE job_id = :job_id" + ), + {"job_id": job_id, "sequence": sequence}, + ) + connection.execute( + text( + "UPDATE file_source_change_page SET page_ordinal = :ordinal " + "WHERE organization_id = :organization_id " + "AND source_id = :source_id AND page_ref = :page_ref" + ), + { + "organization_id": lineage.organization_id, + "source_id": lineage.source_id, + "page_ref": lineage.change_page_ref, + "ordinal": page_ordinal, + }, + ) + connection.execute( + text( + "UPDATE file_source_change SET change_ordinal = :new_ordinal " + "WHERE organization_id = :organization_id " + "AND source_id = :source_id AND page_ref = :page_ref " + "AND change_ordinal = :old_ordinal" + ), + { + "organization_id": lineage.organization_id, + "source_id": lineage.source_id, + "page_ref": lineage.change_page_ref, + "old_ordinal": lineage.change_ordinal, + "new_ordinal": change_ordinal, + }, + ) + connection.execute( + text( + "UPDATE file_acquisition SET change_ordinal = :new_ordinal " + "WHERE organization_id = :organization_id " + "AND source_id = :source_id AND change_page_ref = :page_ref" + ), + { + "organization_id": lineage.organization_id, + "source_id": lineage.source_id, + "page_ref": lineage.change_page_ref, + "new_ordinal": change_ordinal, + }, + ) + connection.execute( + text( + "ALTER TABLE file_acquisition ADD CONSTRAINT " + "fk_file_acquisition_change_observation_exact FOREIGN KEY (" + "organization_id, source_id, source_version_id, change_page_ref, " + "change_ordinal, relative_path, expected_content_sha256, " + "expected_content_length) REFERENCES file_source_change (" + "organization_id, source_id, source_version_id, page_ref, " + "change_ordinal, relative_path, content_sha256, content_length)" + ) + ) + for table in ( + "file_source_acquisition_checkpoint", + "file_source_change_page", + "file_source_change", + "file_acquisition", + ): + connection.execute( + text(f"ALTER TABLE {table} ENABLE TRIGGER {table}_immutable") + ) + finally: + migration_engine.dispose() + authority = _dispatch_authority(guarded_scheduler_engine) + + claims = tuple(authority.claim() for _index in range(5)) + + assert all(type(claim) is FileDispatchLease for claim in claims) + stable_tie = sorted(scheduled_jobs[3:]) + expected = [ + scheduled_jobs[0][1], + scheduled_jobs[1][1], + scheduled_jobs[2][1], + *(job_id for _organization_id, job_id in stable_tie), + ] + assert [ + claim.job_id for claim in claims if type(claim) is FileDispatchLease + ] == expected + + +def test_post_claim_mint_failure_leaves_one_expiring_lease_and_zero_effect( + tmp_path: Path, + guarded_control_engine: Engine, + guarded_scheduler_engine: Engine, + migration_configuration: DatabaseConfiguration, + monkeypatch: pytest.MonkeyPatch, +) -> None: + root = tmp_path / "crash-root" + root.mkdir() + (root / "crash.md").write_text("# Crash\n", encoding="utf-8") + organization_id, job_id, _root_ref = _schedule_one( + root=root, + guarded_control_engine=guarded_control_engine, + migration_configuration=migration_configuration, + ) + codec = WorkerLeaseCodec( + WorkerLeaseKeyring(active_version=1, keys={1: SIGNING_KEY}) + ) + monkeypatch.setattr( + WorkerLeaseCodec, + "mint", + lambda _self, _claims: (_ for _ in ()).throw(RuntimeError("signing failed")), + ) + authority = _dispatch_authority(guarded_scheduler_engine, codec) + + with pytest.raises(RuntimeError, match="signing failed"): + authority.claim() + + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.connect() as connection: + snapshot = connection.execute( + text( + "SELECT job.state, job.lease_generation, job.dispatch_claimed, " + "job.lease_expires_at > job.lease_issued_at, " + "(SELECT count(*) FROM context_revision " + " WHERE organization_id = :organization_id), " + "(SELECT count(*) FROM exact_phrase_candidate " + " WHERE organization_id = :organization_id), " + "(SELECT count(*) FROM file_source_publish_watermark " + " WHERE organization_id = :organization_id) " + "FROM file_import_job AS job " + "WHERE job.organization_id = :organization_id " + "AND job.job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ).one() + finally: + migration_engine.dispose() + assert tuple(snapshot) == ("leased", 1, True, True, 0, 0, 0) + + monkeypatch.undo() + assert ( + _dispatch_authority( + guarded_scheduler_engine, + codec, + ).claim() + == FileDispatchNoWork() + ) + + +def test_claim_skips_authority_row_while_revocation_is_in_flight( + tmp_path: Path, + guarded_control_engine: Engine, + guarded_scheduler_engine: Engine, + migration_configuration: DatabaseConfiguration, +) -> None: + root = tmp_path / "revocation-race-root" + root.mkdir() + (root / "race.md").write_text("# Race\n", encoding="utf-8") + organization_id, job_id, _root_ref = _schedule_one( + root=root, + guarded_control_engine=guarded_control_engine, + migration_configuration=migration_configuration, + ) + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.connect() as revoker: + transaction = revoker.begin() + revoker.execute( + text( "UPDATE membership SET status = 'revoked' " "WHERE organization_id = :organization_id" ), @@ -1145,50 +2148,202 @@ def test_claim_refreshes_latest_scan_after_waiting_for_source_progress( assert tuple(state) == ("available", 0, False) -def test_claim_refreshes_membership_expiry_after_waiting_for_source_progress( +def test_file_reclaim_downgrade_waits_for_real_in_flight_scheduler( tmp_path: Path, guarded_control_engine: Engine, guarded_scheduler_engine: Engine, migration_configuration: DatabaseConfiguration, ) -> None: - root = tmp_path / "expiry-race-root" + """A real scheduler claim holds the 0034 fence until it finishes.""" + + root = tmp_path / "reclaim-migration-fence" root.mkdir() - (root / "expiry.md").write_text("# Expiry\n", encoding="utf-8") + (root / "fence.md").write_text("# Fence\n", encoding="utf-8") organization_id, job_id, _root_ref = _schedule_one( root=root, guarded_control_engine=guarded_control_engine, migration_configuration=migration_configuration, ) migration_engine = create_database_engine(migration_configuration) + alembic_configuration = AlembicConfig(ROOT / "alembic.ini") + migration_fence_key = "context-engine.file-dispatch-migration-fence" try: with migration_engine.connect() as blocker: - transaction = blocker.begin() + blocker_transaction = blocker.begin() source_id = blocker.execute( text( - "SELECT source_id FROM context_source " - "WHERE organization_id = :organization_id" + "SELECT source_id FROM context_source WHERE " + "organization_id = :organization_id" ), {"organization_id": organization_id}, ).scalar_one() + source_lock_key = ( + f"context-engine.file-source-progress:{organization_id}:{source_id}" + ) blocker.execute( text( - "UPDATE membership SET valid_until = " - "pg_catalog.clock_timestamp() + interval '250 milliseconds' " - "WHERE organization_id = :organization_id" + "SELECT pg_catalog.pg_advisory_xact_lock(" + "pg_catalog.hashtextextended(:source_lock_key, 0))" ), - {"organization_id": organization_id}, + {"source_lock_key": source_lock_key}, ) - blocker.execute( + with ThreadPoolExecutor(max_workers=2) as executor: + pending_claim = executor.submit( + _dispatch_authority(guarded_scheduler_engine).claim + ) + try: + with migration_engine.connect() as observer: + deadline = monotonic() + 10 + scheduler_waiting = False + while monotonic() < deadline: + scheduler_waiting = observer.execute( + text( + """ + SELECT EXISTS ( + SELECT 1 + FROM pg_catalog.pg_locks AS waiting + JOIN pg_catalog.pg_locks AS held + ON held.locktype = waiting.locktype + AND held.database = waiting.database + AND held.classid = waiting.classid + AND held.objid = waiting.objid + AND held.objsubid = waiting.objsubid + WHERE waiting.locktype = 'advisory' + AND waiting.mode = 'ExclusiveLock' + AND waiting.granted IS FALSE + AND waiting.database = ( + SELECT database.oid + FROM pg_catalog.pg_database AS database + WHERE database.datname = + pg_catalog.current_database() + ) + AND waiting.classid = ( + (pg_catalog.hashtextextended( + :lock_key, 0 + ) >> 32) & 4294967295 + )::oid + AND waiting.objid = ( + pg_catalog.hashtextextended(:lock_key, 0) + & 4294967295 + )::oid + AND waiting.objsubid = 1 + AND held.mode = 'ExclusiveLock' + AND held.granted IS TRUE + ) + """ + ), + {"lock_key": source_lock_key}, + ).scalar_one() + if scheduler_waiting: + break + sleep(0.01) + assert scheduler_waiting + pending_downgrade = executor.submit( + command.downgrade, + alembic_configuration, + "20260725_0033", + ) + with migration_engine.connect() as observer: + deadline = monotonic() + 10 + downgrade_waiting = False + while monotonic() < deadline: + downgrade_waiting = observer.execute( + text( + """ + SELECT EXISTS ( + SELECT 1 + FROM pg_catalog.pg_locks AS waiting + JOIN pg_catalog.pg_locks AS held + ON held.locktype = waiting.locktype + AND held.database = waiting.database + AND held.classid = waiting.classid + AND held.objid = waiting.objid + AND held.objsubid = waiting.objsubid + WHERE waiting.locktype = 'advisory' + AND waiting.mode = 'ExclusiveLock' + AND waiting.granted IS FALSE + AND waiting.database = ( + SELECT database.oid + FROM pg_catalog.pg_database AS database + WHERE database.datname = + pg_catalog.current_database() + ) + AND waiting.classid = ( + (pg_catalog.hashtextextended( + :lock_key, 0 + ) >> 32) & 4294967295 + )::oid + AND waiting.objid = ( + pg_catalog.hashtextextended(:lock_key, 0) + & 4294967295 + )::oid + AND waiting.objsubid = 1 + AND held.mode = 'ShareLock' + AND held.granted IS TRUE + ) + """ + ), + {"lock_key": migration_fence_key}, + ).scalar_one() + if downgrade_waiting: + break + sleep(0.01) + assert downgrade_waiting + finally: + blocker_transaction.commit() + assert type(pending_claim.result(timeout=10)) is FileDispatchLease + pending_downgrade.result(timeout=10) + with migration_engine.connect() as connection: + assert connection.execute( text( - "SELECT pg_catalog.pg_advisory_xact_lock(" - "pg_catalog.hashtextextended(" - "'context-engine.file-source-progress:' || " - "CAST(:organization_id AS text) || ':' || " - "CAST(:source_id AS text), 0))" + "SELECT lease_generation, dispatch_claimed FROM " + "file_import_job WHERE organization_id = :organization_id " + "AND job_id = :job_id" ), - {"organization_id": organization_id, "source_id": source_id}, + {"organization_id": organization_id, "job_id": job_id}, + ).one() == (1, True) + assert ( + connection.execute( + text("SELECT version_num FROM alembic_version") + ).scalar_one() + == "20260725_0033" ) - transaction.commit() + finally: + command.upgrade(alembic_configuration, "head") + with migration_engine.connect() as connection: + assert ( + connection.execute( + text("SELECT version_num FROM alembic_version") + ).scalar_one() + == HEAD_REVISION + ) + migration_engine.dispose() + + +def test_claim_refreshes_membership_expiry_after_waiting_for_source_progress( + tmp_path: Path, + guarded_control_engine: Engine, + guarded_scheduler_engine: Engine, + migration_configuration: DatabaseConfiguration, +) -> None: + root = tmp_path / "expiry-race-root" + root.mkdir() + (root / "expiry.md").write_text("# Expiry\n", encoding="utf-8") + organization_id, job_id, _root_ref = _schedule_one( + root=root, + guarded_control_engine=guarded_control_engine, + migration_configuration=migration_configuration, + ) + migration_engine = create_database_engine(migration_configuration) + try: + with migration_engine.connect() as connection: + source_id = connection.execute( + text( + "SELECT source_id FROM context_source " + "WHERE organization_id = :organization_id" + ), + {"organization_id": organization_id}, + ).scalar_one() with migration_engine.connect() as blocker: transaction = blocker.begin() @@ -1229,7 +2384,14 @@ def test_claim_refreshes_membership_expiry_after_waiting_for_source_progress( sleep(0.01) if not waiting: pytest.fail("File dispatch did not wait for Source progress") - sleep(0.4) + blocker.execute( + text( + "UPDATE membership SET valid_until = " + "pg_catalog.clock_timestamp() - interval '1 second' " + "WHERE organization_id = :organization_id" + ), + {"organization_id": organization_id}, + ) transaction.commit() assert pending_claim.result(timeout=5) == FileDispatchNoWork() with migration_engine.connect() as connection: diff --git a/tests/integration/test_migrations.py b/tests/integration/test_migrations.py index 3ecce9fa..89be4ab9 100644 --- a/tests/integration/test_migrations.py +++ b/tests/integration/test_migrations.py @@ -1196,6 +1196,112 @@ def test_file_dispatch_revision_downgrades_and_reapplies_empty( assert _revision_rows(migration_configuration) == [HEAD_REVISION] +def test_file_reclaim_revision_downgrades_and_reapplies_empty( + migration_configuration: DatabaseConfiguration, +) -> None: + """Issue #93 removes only its unminted higher-generation capability.""" + + alembic_configuration = Config(ROOT / "alembic.ini") + try: + command.downgrade(alembic_configuration, "20260725_0033") + assert _revision_rows(migration_configuration) == ["20260725_0033"] + engine = create_database_engine(migration_configuration) + try: + with engine.connect() as connection: + definition = connection.execute( + text( + "SELECT pg_get_functiondef(" + "'public.context_scheduler_claim_file_import(" + "bigint,bytea,text[])'::regprocedure)" + ) + ).scalar_one() + assert "context_scheduler_claim_first_file_import" not in definition + assert "event_type" not in definition + assert connection.execute( + text( + "SELECT to_regprocedure(" + "'public.context_scheduler_claim_first_file_import(" + "bigint,bytea,text[])') IS NULL" + ) + ).scalar_one() + assert connection.execute( + text( + "SELECT to_regclass(" + "'public.ix_file_import_job_dispatch_expired') IS NULL" + ) + ).scalar_one() + finally: + engine.dispose() + finally: + command.upgrade(alembic_configuration, "head") + + assert _revision_rows(migration_configuration) == [HEAD_REVISION] + + +def test_file_reclaim_revision_refuses_retained_higher_generation( + tmp_path: Path, + migration_configuration: DatabaseConfiguration, + guarded_control_engine: Engine, +) -> None: + """Downgrade preserves every retained scheduler-created reclaim fact.""" + + scenario = _prepare_file_import_scenario( + tmp_path, + migration_configuration, + guarded_control_engine, + issue_lease=False, + ) + organization_id = scenario.organization_id + job_id = scenario.prepared.job_id + engine = create_database_engine(migration_configuration) + try: + with engine.begin() as connection: + connection.execute( + text( + "UPDATE file_import_job SET state = 'leased', " + "lease_generation = 2, signing_key_version = 1, " + "lease_nonce_digest = digest('reclaim-test', 'sha256'), " + "lease_issued_at = clock_timestamp(), " + "lease_expires_at = clock_timestamp() + interval '5 minutes', " + "dispatch_claimed = true WHERE " + "organization_id = :organization_id AND job_id = :job_id" + ), + { + "organization_id": organization_id, + "job_id": job_id, + }, + ) + with pytest.raises( + RuntimeError, + match="automatic File reclaim downgrade requires no retained", + ): + command.downgrade(Config(ROOT / "alembic.ini"), "20260725_0033") + assert _revision_rows(migration_configuration) == [HEAD_REVISION] + with engine.connect() as connection: + assert connection.execute( + text( + "SELECT lease_generation, dispatch_claimed FROM " + "file_import_job WHERE organization_id = :organization_id " + "AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ).one() == (2, True) + finally: + with engine.begin() as connection: + connection.execute( + text( + "UPDATE file_import_job SET state = 'available', " + "lease_generation = 0, signing_key_version = NULL, " + "lease_nonce_digest = NULL, lease_issued_at = NULL, " + "lease_expires_at = NULL, dispatch_claimed = false WHERE " + "organization_id = :organization_id AND job_id = :job_id" + ), + {"organization_id": organization_id, "job_id": job_id}, + ) + command.upgrade(Config(ROOT / "alembic.ini"), "head") + engine.dispose() + + def test_mixed_file_upsert_downgrade_waits_for_in_flight_scheduler( tmp_path: Path, migration_configuration: DatabaseConfiguration, diff --git a/tests/support/migrations.py b/tests/support/migrations.py index 76c2bf1b..d165d6d3 100644 --- a/tests/support/migrations.py +++ b/tests/support/migrations.py @@ -1,3 +1,3 @@ """Shared migration assertions for tests that require the current schema head.""" -HEAD_REVISION = "20260725_0033" +HEAD_REVISION = "20260726_0034" diff --git a/tests/unit/test_file_dispatch.py b/tests/unit/test_file_dispatch.py index cd1c5a80..c5d49e20 100644 --- a/tests/unit/test_file_dispatch.py +++ b/tests/unit/test_file_dispatch.py @@ -22,6 +22,7 @@ from engine.control import FileImportPath, FileRootRef, SourceRef from engine.persistence.file_imports import FileImportRefused, FileImportUnavailable from engine.persistence.worker_jobs import ( + FILE_DISPATCH_MAX_LEASE_GENERATION, FileDispatchLease, FileDispatchNoWork, PostgreSQLFileDispatchAuthority, @@ -65,11 +66,11 @@ def test_dispatch_lease_redacts_every_routing_and_capability_value() -> None: assert claim.redemption.expected_source_ref == claim.source_ref -@pytest.mark.parametrize("generation", [0, 2]) -def test_first_attempt_dispatch_rejects_any_other_generation(generation: int) -> None: +@pytest.mark.parametrize("generation", [0, 5]) +def test_dispatch_rejects_generation_outside_automatic_budget(generation: int) -> None: claimed_at = datetime(2026, 7, 25, 10, tzinfo=UTC) - with pytest.raises(ValueError, match="generation one"): + with pytest.raises(ValueError, match="automatic generation budget"): FileDispatchLease( token=WorkerLeaseToken("lease-token"), organization_id=uuid4(), @@ -82,6 +83,27 @@ def test_first_attempt_dispatch_rejects_any_other_generation(generation: int) -> ) +@pytest.mark.parametrize("generation", [1, 2, 3, 4]) +def test_dispatch_accepts_only_the_versioned_automatic_generations( + generation: int, +) -> None: + claimed_at = datetime(2026, 7, 25, 10, tzinfo=UTC) + + claim = FileDispatchLease( + token=WorkerLeaseToken("lease-token"), + organization_id=uuid4(), + job_id=uuid4(), + source_ref=SourceRef(uuid4()), + service_principal_id=uuid4(), + lease_generation=generation, + issued_at=claimed_at, + expires_at=claimed_at + timedelta(minutes=5), + ) + + assert claim.lease_generation == generation + assert FILE_DISPATCH_MAX_LEASE_GENERATION == 4 + + class _NoWorkAuthority: def claim(self) -> FileDispatchNoWork: return FileDispatchNoWork() diff --git a/tests/unit/test_schema_security_manifest.py b/tests/unit/test_schema_security_manifest.py index 5162a593..914b01c9 100644 --- a/tests/unit/test_schema_security_manifest.py +++ b/tests/unit/test_schema_security_manifest.py @@ -32,7 +32,7 @@ def test_manifest_classifies_the_exact_current_release_schema() -> None: document = manifest() tables = table_entries(document) - assert document["manifestVersion"] == "31.0.0" + assert document["manifestVersion"] == "32.0.0" assert set(tables) == { "active_release_manifest", "action_delivery_attempt", @@ -545,6 +545,10 @@ def test_issue_27_file_recovery_contract_is_generation_fenced_and_auditable() -> assert entry["functionOnlyMutation"]["directTableMutationAllowed"] is False assert entry["permittedOperations"]["context_engine_runtime"] == [] assert history["immutableRows"]["events"] == ["UPDATE", "DELETE"] + assert history["functionOnlyMutation"]["definerRoles"] == [ + "context_engine_worker_lease_definer", + "context_engine_file_dispatch_definer", + ] assert checkpoint["retention"]["sourceContent"] == "none" boundary_functions = {