diff --git a/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/REVIEW_LOG.md b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/REVIEW_LOG.md new file mode 100644 index 000000000..007734b5f --- /dev/null +++ b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/REVIEW_LOG.md @@ -0,0 +1,32 @@ +# Review Log: WS-ART-001 Immutable Artifact Storage + +## WS-ART-001-02C3 + +- Signed explicit start verified on 2026-07-21 against trusted main `f2aa57a4`. +- L1 preimplementation plan review: PASS WITH CONDITIONS. +- Required conditions cover verification-job lineage, database recovery + invariants, concurrency-safe replay, AUTH separation, atomic audits, terminal + fencing, and cumulative coverage gates. +- First internal review batch returned blocking findings: taskless guide + recovery was not representable, the Operator mutation seam was not yet + fail-closed, and required fencing/terminal/migration proofs were incomplete. +- Repair in progress: recovery task context is nullable, the exact Operator + authority seam defaults to deny until AUTH-owned activation, authorization + evidence is retained in the initiation audit, and focused guide/authority + drift tests were added. No publication claim is recorded yet. +- Re-review cleared taskless-guide and architecture blockers, then identified + an authorization-ordering bypass on exact replay. Every normal and + concurrent-winner replay now revalidates the persisted human identity and + exact recovery authority before returning identifiers; denied replay is + covered with zero-write assertions. +- Final internal review results on `841f2a38`: senior engineering PASS WITH + LOW RISKS; architecture PASS WITH LOW RISKS; QA PASS; security PASS WITH LOW + RISKS; product/ops PASS WITH LOW RISKS; reuse/dedup PASS WITH LOW RISKS; CI + integrity PASS; test delta PASS; docs PASS. +- Accepted low risks are limited to legacy terminal-audit top-level metadata, + private audit-builder ownership coupling, and duplicated human-proof shape; + none changes authorization, recovery custody, or product lifecycle state. +- GitHub Actions run `29851665477` found two valid integration issues: an + over-broad lineage trigger and an incomplete integrity-mismatch upload-item + transition. Both received bounded repairs and their three failing tests pass + locally; final internal re-review and hosted rerun remain. diff --git a/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/chunks/WS-ART-001-02D-operator-artifact-operations.md b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/chunks/WS-ART-001-02D-operator-artifact-operations.md index e58f45d71..4b3b679f9 100644 --- a/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/chunks/WS-ART-001-02D-operator-artifact-operations.md +++ b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/chunks/WS-ART-001-02D-operator-artifact-operations.md @@ -1,4 +1,4 @@ -# Chunk Contract: WS-ART-001-02D Operator Artifact Operations +# Chunk Contract: WS-ART-001-02D - Operator Artifact Operations Initiative: `WS-ART-001` | Risk: L1 | Status: Proposed after 02C3, AUTH-09E, and `WS-AUTH-001-ART-CUSTODY` diff --git a/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-02C3-external-review-response.md b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-02C3-external-review-response.md new file mode 100644 index 000000000..d3b2852a3 --- /dev/null +++ b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-02C3-external-review-response.md @@ -0,0 +1,36 @@ +# WS-ART-001-02C3 External Review Response + +## GitHub Actions run 29851665477 + +Comments addressed: + +- shard 1 exposed that the new verification-lineage trigger blocked two + existing pre-recovery drift/fence tests. The trigger now freezes lineage only + after a job is a recovery source or retry; unrecovered initial jobs retain + the drift behavior needed for fail-closed verification revalidation. +- shard 2 exposed that integrity mismatch moved a contributor upload item to + `failed` without clearing stored-result references. The terminal transition + now clears `content_id` and `provider_object_ref`, preserving the existing + database state-shape invariant. +- the fan-in `test` failure was downstream of those two shard failures and + requires no independent repair. + +Comments deferred: + +- CodeRabbit supplied no code findings because its review was rate limited. + +Human decisions needed: + +- none for these bounded CI repairs; explicit human approval remains required + before merge. + +Commands rerun: + +- Ruff on the changed artifact service, migration, and focused tests: PASS. +- Both failed shard-1 tests and the failed shard-2 integrity-mismatch test: + 3 PASS. + +Remaining risks: + +- the complete hosted shards and cumulative coverage gates must rerun on the + repaired commit. diff --git a/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-02C3-internal-review-evidence.md b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-02C3-internal-review-evidence.md new file mode 100644 index 000000000..26b1bd37a --- /dev/null +++ b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-02C3-internal-review-evidence.md @@ -0,0 +1,93 @@ +# WS-ART-001-02C3 Internal Review Evidence + +Reviewed code SHA: `dd1aa0d91ed0b0a3f67638f719e8332865c128eb` + +Reviewed at: `2026-07-21T17:43:00Z` + +Reviewer run IDs: `ci_repair_senior_arch`, `ci_repair_qa_security`, +`ci_repair_ops_ci_docs` + +## CI repair review addendum + +| Reviewer | Result | Blocking findings | Notes | +|---|---|---|---| +| senior engineering | PASS WITH LOW RISKS | none | Recovery-only lineage custody and failed-item state shape are maintainable. | +| QA/test | PASS WITH LOW RISKS | none | The three exact hosted failures pass after bounded repairs. | +| security/auth | PASS | none | Recovery participants remain immutable and stale content ownership references are cleared. | +| product/ops | PASS | none | No route, review decision, or product lifecycle behavior changed. | +| architecture | PASS WITH LOW RISKS | none | Trigger coupling is intentionally limited to recovery participation. | +| CI integrity | PASS | none | No workflow, test, threshold, or bypass change was made. | +| docs | PASS | none | External response and trust status accurately describe PR #174 and the hosted rerun gate. | +| reuse/dedup | PASS | none | The bounded state transition and custody predicate need no new helper. | +| test delta | PASS | none | Existing tests exposed both failures; no test was changed or weakened. | + +Valid findings addressed: yes + +Open sub-agent sessions: none + +--- + +Reviewed code SHA: `f302838146f39e78a15080df7231ef3904a052ed` + +Reviewed against trusted main: `1473f7a0cab6d879c7b7c049a9b94f557ad712c2` + +Reviewed at: `2026-07-21T17:03:00Z` + +Reviewer run IDs: `review_senior_arch`, `review_qa_test`, +`review_security_product`, `review_reuse_ci_docs` + +Reviewer tracks: senior engineering, QA/test, security/auth, product/ops, +architecture, CI integrity, docs, reuse/dedup, and test delta + +## Scope + +WS-ART-001-02C3 adds one durable recovery-attempt envelope, immutable +source-to-retry verification lineage, exact replay ownership, atomic terminal +finalization, and a deny-only Operator authority seam. It adds no route, +provider mutation, product lifecycle transition, dependency, or CI weakening. + +## Deterministic evidence + +- Focused artifact recovery suite: PASS before reviewer repair; repaired guide, + authority-drift, denied-replay, and sampled terminal-outcome tests: PASS. +- Alembic upgrade/downgrade and recovery-schema proof: PASS. +- Ruff for backend app, tests, and migration: PASS. +- Stale artifact contracts: PASS. +- Agent gate unit suite: 89 PASS. +- Merge-intent validation against `origin/main`: PASS. +- `git diff --check`: PASS. +- Heavy sharded backend suite and cumulative coverage remain hosted CI gates. + +## Reviewer results + +| Reviewer | Result | Blocking findings | Notes | +|---|---|---|---| +| senior engineering | PASS WITH LOW RISKS | none | Taskless guide recovery is represented end to end. | +| QA/test | PASS | none | Creation, replay authorization, terminal outcomes, fencing, and chaining are covered. | +| security/auth | PASS WITH LOW RISKS | none | Every creation and replay requires fresh exact authority; deny remains production default. | +| product/ops | PASS WITH LOW RISKS | none | Recovery remains infrastructure-only and provider-read-only. | +| architecture | PASS WITH LOW RISKS | none | One recovery chain spans guide and task-backed producers without route activation. | +| CI integrity | PASS | none | No workflow, threshold, package, runner, or bypass change. | +| docs | PASS | none | Existing ART plan/glossary cover the internal contract; memory and trust evidence are current. | +| reuse/dedup | PASS WITH LOW RISKS | none | Existing repositories, hashing, actor proof, audit, and verification fences are reused. | +| test delta | PASS | none | Tests were added and strengthened; none were removed, skipped, or weakened. | + +## Findings resolved + +Initial review blocked task-bound guide recovery, missing fresh Operator +authority, incomplete terminal/fence tests, and replay-before-authorization. +The repair made task context nullable, added the deny-only exact authority seam, +retained bounded decision evidence, covered all terminal mappings and authority +drift, and reauthorized normal and concurrent-winner replay before returning +identifiers. The post-review `WS-ART-001-02D` heading change is canonical +metadata formatting only; docs and CI reviewers confirmed their PASS results +remain valid. + +Valid findings addressed: yes + +Open sub-agent sessions: none + +## Remaining gate + +GitHub backend shards, cumulative 90 percent artifact coverage, repository-wide +78 percent coverage, external review, and explicit human merge approval remain. diff --git a/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-02C3-pr-trust-bundle.md b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-02C3-pr-trust-bundle.md new file mode 100644 index 000000000..44f2d5c5b --- /dev/null +++ b/.agent-loop/initiatives/WS-ART-001-immutable-artifact-storage/reviews/WS-ART-001-02C3-pr-trust-bundle.md @@ -0,0 +1,107 @@ +# PR Trust Bundle: WS-ART-001-02C3 + +## Chunk + +`WS-ART-001-02C3` — Recovery Attempt And Idempotency Chain (L1) + +## Goal And Human-Approved Intent + +Add the durable, read-only provider-observation recovery envelope and exact +source-job to retry-job chain. The signed loop-memory start authorized this ART +chunk to run concurrently with AUTH; it does not activate Operator routes. + +## What Changed And Why + +- added migration `0032` for recovery envelopes and immutable verification + lineage; +- added nullable task context so the same chain supports guide and task-backed + artifacts; +- added exact replay, lifetime source ownership, concurrent winner recovery, + linear retry ancestry, and atomic terminal finalization; +- added a typed Operator authority seam whose production-safe implementation + denies until the later AUTH-owned activation; +- required fresh actor and exact authority revalidation on creation and every + replay, with bounded authorization evidence in the initiation audit; +- added focused migration, concurrency, guide, authorization, fencing, and + terminal-outcome tests. + +## Design And Alternatives + +The verification job remains the sole executable lease owner. Recovery is an +envelope, not a second worker lease. PostgreSQL uniqueness and custody triggers +enforce one lifetime source owner and immediate-parent chaining. Provider +mutation replay and a task-only recovery abstraction were rejected because +both violate the initiative contract. + +## Scope And Product Behavior + +No routes, provider mutation, guide/task/submission lifecycle transitions, +review decisions, payment, reputation, dependencies, CI workflow changes, or +coverage reductions are included. Recovery remains infrastructure state and +does not introduce product decisions beyond `accept`, `needs_revision`, and +`reject`. + +## Acceptance Evidence + +- exact and concurrent replay preserve one envelope, retry job, and initiation + audit; +- changed/lifetime reuse conflicts without new recovery ownership; +- only exhausted terminal `provider_unavailable` work is recoverable; +- taskless guide recovery is representable and replayable; +- denied creation and denied replay return no identifiers and add no rows; +- success and every failed terminal outcome finalize envelope and audit under + the verification transaction; +- terminal authority drift writes no terminal recovery facts; +- exhausted retries form only the next linear chain link. + +## Tests And CI Integrity + +Local focused evidence: + +- `ruff check app tests alembic/versions/0032_artifact_recovery_attempts.py`; +- focused recovery tests, including all original five, guide/deny replay, + authority drift, and sampled failed outcome; +- migration upgrade/downgrade and recovery-schema tests; +- `python3 scripts/check_stale_artifact_contracts.py`; +- `python3 scripts/test_agent_gates.py` — 89 passed; +- `git diff --check`. + +The expensive full backend suite, remaining parameter combinations, global +78% coverage, and cumulative artifact 90% coverage are intentionally delegated +to the existing sharded GitHub Actions. No CI or package-script file changed. +No tests were removed, skipped, or weakened. + +## Reviewer Results + +- senior engineering: PASS WITH LOW RISKS; +- architecture: PASS WITH LOW RISKS; +- QA/test: PASS; +- security/auth: PASS WITH LOW RISKS; +- product/ops: PASS WITH LOW RISKS; +- reuse/dedup: PASS WITH LOW RISKS; +- CI integrity: PASS; +- test delta: PASS; +- docs: PASS. + +PR #174 is published. External review and final hosted CI remain open gates. + +The first hosted shard attempt passed agent gates, preflight, API E2E, and +shards 3-4. Shards 1-2 found an over-broad lineage trigger and an incomplete +integrity-mismatch item transition; both were repaired and their three exact +failing tests pass locally. A full hosted rerun remains required. + +## Remaining Risks And Follow-Up + +Low risks: legacy terminal audit metadata describes the verifier imperfectly, +the audit builder has private cross-class ownership, and human-proof validation +has small domain-specific duplication. The later typed Operator audit surface +may clean these up. Successor `WS-ART-001-02D` owns Operator routes and real AUTH +activation; it must not start automatically before this chunk merges and the +signed successor event is approved. + +## Human Review Focus And Merge Ownership + +Verify that no source can own two recoveries, no replay can bypass fresh +authority, and recovery stays separate from provider mutation and product +lifecycle state. A human owns the merge decision; Codex must not merge without +explicit approval for this PR. diff --git a/.agent-loop/merge-intents/WS-ART-001-02C3.json b/.agent-loop/merge-intents/WS-ART-001-02C3.json new file mode 100644 index 000000000..a5ad589f9 --- /dev/null +++ b/.agent-loop/merge-intents/WS-ART-001-02C3.json @@ -0,0 +1,9 @@ +{ + "chunk_id": "WS-ART-001-02C3", + "chunk_title": "Recovery Attempt And Idempotency Chain", + "initiative_id": "WS-ART-001", + "next_chunk_id": "WS-ART-001-02D", + "next_chunk_title": "Operator Artifact Operations", + "next_requires_explicit_start": true, + "schema_version": 2 +} diff --git a/backend/alembic/versions/0032_artifact_recovery_attempts.py b/backend/alembic/versions/0032_artifact_recovery_attempts.py new file mode 100644 index 000000000..37a16100d --- /dev/null +++ b/backend/alembic/versions/0032_artifact_recovery_attempts.py @@ -0,0 +1,290 @@ +"""add artifact recovery attempts and verification retry lineage + +Revision ID: 0032_artifact_recovery +Revises: 0031_project_role_grants +Create Date: 2026-07-21 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "0032_artifact_recovery" +down_revision = "0031_project_role_grants" +branch_labels = depends_on = None + + +def upgrade() -> None: + """Install immutable recovery envelopes and retry-job lineage.""" + op.drop_constraint( + "uq_artifact_verification_origin", "artifact_verification_jobs", type_="unique" + ) + op.add_column( + "artifact_verification_jobs", + sa.Column("parent_verification_job_id", sa.String(36), nullable=True), + ) + op.create_foreign_key( + "fk_artifact_verification_parent", + "artifact_verification_jobs", + "artifact_verification_jobs", + ["parent_verification_job_id"], + ["id"], + ondelete="RESTRICT", + ) + op.create_unique_constraint( + "uq_artifact_verification_parent", + "artifact_verification_jobs", + ["parent_verification_job_id"], + ) + op.create_index( + "ix_artifact_verification_jobs_parent_verification_job_id", + "artifact_verification_jobs", + ["parent_verification_job_id"], + ) + op.create_index( + "uq_artifact_verification_initial_origin", + "artifact_verification_jobs", + ["originating_put_attempt_id"], + unique=True, + postgresql_where=sa.text("parent_verification_job_id is null"), + ) + + op.create_table( + "artifact_recovery_attempts", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("requester_actor_profile_id", sa.String(36), nullable=False), + sa.Column("requester_identity_link_id", sa.String(36), nullable=False), + sa.Column("authorization_request_id", sa.String(36), nullable=False), + sa.Column("authorization_correlation_id", sa.String(36), nullable=False), + sa.Column("project_id", sa.String(36), nullable=False), + sa.Column("task_id", sa.String(36), nullable=True), + sa.Column("submission_id", sa.String(36), nullable=True), + sa.Column("source_verification_job_id", sa.String(36), nullable=False), + sa.Column("retry_verification_job_id", sa.String(36), nullable=False), + sa.Column("parent_recovery_attempt_id", sa.String(36), nullable=True), + sa.Column("recovery_class", sa.String(40), nullable=False), + sa.Column("reason", sa.String(1000), nullable=False), + sa.Column("client_idempotency_key", sa.String(200), nullable=False), + sa.Column("request_digest", sa.String(71), nullable=False), + sa.Column("status", sa.String(20), nullable=False, server_default="requested"), + sa.Column("terminal_result_code", sa.String(40), nullable=True), + sa.Column("initiation_audit_event_id", sa.String(36), nullable=False), + sa.Column("terminal_audit_event_id", sa.String(36), nullable=True), + sa.Column("cas_version", sa.BigInteger(), nullable=False, server_default="0"), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("terminal_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.ForeignKeyConstraint( + ["requester_actor_profile_id"], ["actor_profiles.id"], ondelete="RESTRICT" + ), + sa.ForeignKeyConstraint( + ["requester_identity_link_id"], ["actor_identity_links.id"], ondelete="RESTRICT" + ), + sa.ForeignKeyConstraint(["project_id"], ["projects.id"], ondelete="RESTRICT"), + sa.ForeignKeyConstraint(["task_id"], ["workstream_tasks.id"], ondelete="RESTRICT"), + sa.ForeignKeyConstraint(["submission_id"], ["submissions.id"], ondelete="RESTRICT"), + sa.ForeignKeyConstraint( + ["source_verification_job_id"], + ["artifact_verification_jobs.id"], + ondelete="RESTRICT", + ), + sa.ForeignKeyConstraint( + ["retry_verification_job_id"], + ["artifact_verification_jobs.id"], + ondelete="RESTRICT", + ), + sa.ForeignKeyConstraint( + ["parent_recovery_attempt_id"], + ["artifact_recovery_attempts.id"], + ondelete="RESTRICT", + ), + sa.ForeignKeyConstraint( + ["initiation_audit_event_id"], ["audit_events.id"], ondelete="RESTRICT" + ), + sa.ForeignKeyConstraint( + ["terminal_audit_event_id"], ["audit_events.id"], ondelete="RESTRICT" + ), + sa.UniqueConstraint( + "requester_actor_profile_id", + "source_verification_job_id", + "recovery_class", + "client_idempotency_key", + name="uq_artifact_recovery_idempotency", + ), + sa.UniqueConstraint( + "source_verification_job_id", name="uq_artifact_recovery_source_job" + ), + sa.UniqueConstraint( + "retry_verification_job_id", name="uq_artifact_recovery_retry_job" + ), + sa.CheckConstraint( + "source_verification_job_id <> retry_verification_job_id", name="distinct_jobs" + ), + sa.CheckConstraint("recovery_class = 'provider_observation'", name="recovery_class"), + sa.CheckConstraint("status in ('requested','succeeded','failed')", name="status"), + sa.CheckConstraint("request_digest ~ '^sha256:[0-9a-f]{64}$'", name="request_digest"), + sa.CheckConstraint("cas_version >= 0", name="cas_nonnegative"), + sa.CheckConstraint( + "(status='requested' and terminal_result_code is null and terminal_at is null " + "and terminal_audit_event_id is null) or " + "(status in ('succeeded','failed') and terminal_result_code is not null " + "and terminal_at is not null and terminal_audit_event_id is not null)", + name="terminal_shape", + ), + sa.CheckConstraint( + "(status='succeeded' and terminal_result_code='verified') or " + "(status='failed' and terminal_result_code in " + "('provider_unavailable','missing','integrity_mismatch','conflict')) or " + "status='requested'", + name="terminal_result", + ), + ) + for column in ( + "requester_actor_profile_id", + "project_id", + "task_id", + "submission_id", + "parent_recovery_attempt_id", + ): + op.create_index( + f"ix_artifact_recovery_attempts_{column}", + "artifact_recovery_attempts", + [column], + ) + op.execute( + """ + create function validate_artifact_recovery_attempt() returns trigger + language plpgsql as $$ + declare + source_row artifact_verification_jobs%rowtype; + retry_row artifact_verification_jobs%rowtype; + expected_parent text; + begin + if tg_op = 'DELETE' then + raise exception 'artifact recovery attempts are append-only' using errcode='55000'; + end if; + if tg_op = 'UPDATE' and ( + to_jsonb(new) - array['status','terminal_result_code','terminal_audit_event_id', + 'terminal_at','cas_version','updated_at'] + is distinct from + to_jsonb(old) - array['status','terminal_result_code','terminal_audit_event_id', + 'terminal_at','cas_version','updated_at'] + ) then + raise exception 'artifact recovery identity is immutable' using errcode='55000'; + end if; + select * into source_row from artifact_verification_jobs + where id=new.source_verification_job_id; + select * into retry_row from artifact_verification_jobs + where id=new.retry_verification_job_id; + if source_row.id is null or retry_row.id is null + or source_row.status <> 'provider_unavailable' + or source_row.terminal_result_code <> 'provider_unavailable' + or source_row.terminal_at is null or source_row.next_run_at is not null + or source_row.executor_id is not null + or source_row.attempt_count < source_row.maximum_attempts + or retry_row.parent_verification_job_id <> source_row.id + or retry_row.originating_put_attempt_id <> source_row.originating_put_attempt_id + or retry_row.replica_id <> source_row.replica_id then + raise exception 'invalid artifact recovery verification lineage' using errcode='23514'; + end if; + if (tg_op = 'INSERT' and (retry_row.status <> 'pending' or retry_row.attempt_count <> 0)) + or (tg_op = 'UPDATE' and ( + retry_row.status <> new.terminal_result_code or retry_row.terminal_at is null + )) then + raise exception 'invalid artifact recovery retry state' using errcode='23514'; + end if; + select id into expected_parent from artifact_recovery_attempts + where retry_verification_job_id=source_row.id; + if new.parent_recovery_attempt_id is distinct from expected_parent then + raise exception 'invalid artifact recovery parent chain' using errcode='23514'; + end if; + if not exists ( + select 1 from audit_events where id=new.initiation_audit_event_id + and entity_type='artifact_recovery_attempt' and entity_id=new.id + and event_type='ArtifactRecoveryInitiated' + ) then + raise exception 'invalid artifact recovery initiation audit' using errcode='23514'; + end if; + if new.terminal_audit_event_id is not null and not exists ( + select 1 from audit_events where id=new.terminal_audit_event_id + and entity_type='artifact_recovery_attempt' and entity_id=new.id + and event_type='ArtifactRecoveryCompleted' + ) then + raise exception 'invalid artifact recovery terminal audit' using errcode='23514'; + end if; + return new; + end $$ + """ + ) + op.execute( + """ + create trigger artifact_recovery_attempt_custody + before insert or update or delete on artifact_recovery_attempts + for each row execute function validate_artifact_recovery_attempt() + """ + ) + op.execute( + """ + create function validate_artifact_verification_lineage() returns trigger + language plpgsql as $$ + begin + if ( + old.parent_verification_job_id is not null + or exists( + select 1 from artifact_recovery_attempts + where source_verification_job_id = old.id + or retry_verification_job_id = old.id + ) + ) and ( + old.originating_put_attempt_id is distinct from new.originating_put_attempt_id + or old.replica_id is distinct from new.replica_id + or old.parent_verification_job_id is distinct from new.parent_verification_job_id + ) then + raise exception 'artifact verification lineage is immutable' using errcode='55000'; + end if; + return new; + end $$ + """ + ) + op.execute( + """ + create trigger artifact_verification_lineage_custody + before update on artifact_verification_jobs + for each row execute function validate_artifact_verification_lineage() + """ + ) + + +def downgrade() -> None: + """Remove recovery only when no durable recovery lineage exists.""" + bind = op.get_bind() + if bind.execute(sa.text("select exists(select 1 from artifact_recovery_attempts)")).scalar(): + raise RuntimeError("cannot downgrade populated artifact recovery attempts") + op.execute( + "drop trigger artifact_verification_lineage_custody on artifact_verification_jobs" + ) + op.execute("drop function validate_artifact_verification_lineage()") + op.execute("drop trigger artifact_recovery_attempt_custody on artifact_recovery_attempts") + op.execute("drop function validate_artifact_recovery_attempt()") + op.drop_table("artifact_recovery_attempts") + op.drop_index( + "uq_artifact_verification_initial_origin", table_name="artifact_verification_jobs" + ) + op.drop_index( + "ix_artifact_verification_jobs_parent_verification_job_id", + table_name="artifact_verification_jobs", + ) + op.drop_constraint( + "uq_artifact_verification_parent", "artifact_verification_jobs", type_="unique" + ) + op.drop_constraint( + "fk_artifact_verification_parent", "artifact_verification_jobs", type_="foreignkey" + ) + op.drop_column("artifact_verification_jobs", "parent_verification_job_id") + op.create_unique_constraint( + "uq_artifact_verification_origin", + "artifact_verification_jobs", + ["originating_put_attempt_id"], + ) diff --git a/backend/app/db/models.py b/backend/app/db/models.py index ed871d369..8f074df36 100644 --- a/backend/app/db/models.py +++ b/backend/app/db/models.py @@ -12,6 +12,7 @@ ArtifactContent, ArtifactOperationReceipt, ArtifactReplica, + ArtifactRecoveryAttempt, ArtifactUploadItem, ArtifactUploadSession, ) diff --git a/backend/app/interfaces/artifact_operations.py b/backend/app/interfaces/artifact_operations.py index 523095cdc..1bbb84218 100644 --- a/backend/app/interfaces/artifact_operations.py +++ b/backend/app/interfaces/artifact_operations.py @@ -114,7 +114,7 @@ class ArtifactRecoveryRequest: authorization_context: AuthorizationContext project_id: UUID - task_id: UUID + task_id: UUID | None submission_id: UUID | None source_verification_job_id: UUID reason: str diff --git a/backend/app/modules/artifacts/models.py b/backend/app/modules/artifacts/models.py index 7280a227d..f5c63b3ad 100644 --- a/backend/app/modules/artifacts/models.py +++ b/backend/app/modules/artifacts/models.py @@ -640,7 +640,7 @@ class ArtifactVerificationJob(Base): __tablename__ = "artifact_verification_jobs" __table_args__ = ( - UniqueConstraint("originating_put_attempt_id", name="uq_artifact_verification_origin"), + UniqueConstraint("parent_verification_job_id", name="uq_artifact_verification_parent"), CheckConstraint( "status in ('pending', 'running', 'verified', 'missing', " "'integrity_mismatch', 'provider_unavailable', 'conflict')", @@ -662,6 +662,9 @@ class ArtifactVerificationJob(Base): originating_put_attempt_id: Mapped[str] = mapped_column( ForeignKey("artifact_put_attempts.id", ondelete="RESTRICT"), nullable=False, index=True ) + parent_verification_job_id: Mapped[str | None] = mapped_column( + ForeignKey("artifact_verification_jobs.id", ondelete="RESTRICT"), index=True + ) replica_id: Mapped[str] = mapped_column( ForeignKey("artifact_replicas.id", ondelete="RESTRICT"), nullable=False, index=True ) @@ -681,6 +684,103 @@ class ArtifactVerificationJob(Base): ) +Index( + "uq_artifact_verification_initial_origin", + ArtifactVerificationJob.originating_put_attempt_id, + unique=True, + postgresql_where=ArtifactVerificationJob.parent_verification_job_id.is_(None), +) + + +class ArtifactRecoveryAttempt(Base): + """Reason-bound idempotent envelope around one verification retry job.""" + + __tablename__ = "artifact_recovery_attempts" + __table_args__ = ( + UniqueConstraint( + "requester_actor_profile_id", + "source_verification_job_id", + "recovery_class", + "client_idempotency_key", + name="uq_artifact_recovery_idempotency", + ), + UniqueConstraint( + "source_verification_job_id", name="uq_artifact_recovery_source_job" + ), + UniqueConstraint( + "retry_verification_job_id", name="uq_artifact_recovery_retry_job" + ), + CheckConstraint( + "source_verification_job_id <> retry_verification_job_id", + name="distinct_jobs", + ), + CheckConstraint("recovery_class = 'provider_observation'", name="recovery_class"), + CheckConstraint("status in ('requested', 'succeeded', 'failed')", name="status"), + CheckConstraint(SHA256_CHECK.format(column="request_digest"), name="request_digest"), + CheckConstraint("cas_version >= 0", name="cas_nonnegative"), + CheckConstraint( + "(status = 'requested' and terminal_result_code is null and terminal_at is null " + "and terminal_audit_event_id is null) or " + "(status in ('succeeded', 'failed') and terminal_result_code is not null " + "and terminal_at is not null and terminal_audit_event_id is not null)", + name="terminal_shape", + ), + CheckConstraint( + "(status = 'succeeded' and terminal_result_code = 'verified') or " + "(status = 'failed' and terminal_result_code in " + "('provider_unavailable', 'missing', 'integrity_mismatch', 'conflict')) or " + "status = 'requested'", + name="terminal_result", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True) + requester_actor_profile_id: Mapped[str] = mapped_column( + ForeignKey("actor_profiles.id", ondelete="RESTRICT"), nullable=False, index=True + ) + requester_identity_link_id: Mapped[str] = mapped_column( + ForeignKey("actor_identity_links.id", ondelete="RESTRICT"), nullable=False + ) + authorization_request_id: Mapped[str] = mapped_column(String(36), nullable=False) + authorization_correlation_id: Mapped[str] = mapped_column(String(36), nullable=False) + project_id: Mapped[str] = mapped_column( + ForeignKey("projects.id", ondelete="RESTRICT"), nullable=False, index=True + ) + task_id: Mapped[str | None] = mapped_column( + ForeignKey("workstream_tasks.id", ondelete="RESTRICT"), nullable=True, index=True + ) + submission_id: Mapped[str | None] = mapped_column( + ForeignKey("submissions.id", ondelete="RESTRICT"), index=True + ) + source_verification_job_id: Mapped[str] = mapped_column( + ForeignKey("artifact_verification_jobs.id", ondelete="RESTRICT"), nullable=False + ) + retry_verification_job_id: Mapped[str] = mapped_column( + ForeignKey("artifact_verification_jobs.id", ondelete="RESTRICT"), nullable=False + ) + parent_recovery_attempt_id: Mapped[str | None] = mapped_column( + ForeignKey("artifact_recovery_attempts.id", ondelete="RESTRICT"), index=True + ) + recovery_class: Mapped[str] = mapped_column(String(40), nullable=False) + reason: Mapped[str] = mapped_column(String(1000), nullable=False) + client_idempotency_key: Mapped[str] = mapped_column(String(200), nullable=False) + request_digest: Mapped[str] = mapped_column(String(71), nullable=False) + status: Mapped[str] = mapped_column(String(20), nullable=False, default="requested") + terminal_result_code: Mapped[str | None] = mapped_column(String(40)) + initiation_audit_event_id: Mapped[str] = mapped_column( + ForeignKey("audit_events.id", ondelete="RESTRICT"), nullable=False + ) + terminal_audit_event_id: Mapped[str | None] = mapped_column( + ForeignKey("audit_events.id", ondelete="RESTRICT") + ) + cas_version: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + terminal_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now() + ) + + class ArtifactVerificationReceipt(Base): """Append-only result of one fenced complete-object observation.""" diff --git a/backend/app/modules/artifacts/repository.py b/backend/app/modules/artifacts/repository.py index 8f2395aea..563bf19dd 100644 --- a/backend/app/modules/artifacts/repository.py +++ b/backend/app/modules/artifacts/repository.py @@ -23,6 +23,7 @@ ArtifactVerificationJob, ArtifactVerificationReceipt, ArtifactReplica, + ArtifactRecoveryAttempt, ArtifactStorageNamespace, ArtifactUploadItem, ArtifactUploadSession, @@ -88,6 +89,12 @@ async def lock_upload_item(self, item_id: str) -> ArtifactUploadItem | None: ) return result.scalar_one_or_none() + async def lock_checker_run(self, checker_run_id: str) -> CheckerRun | None: + """Lock one checker run for canonical recovery resource derivation.""" + return await self._session.scalar( + select(CheckerRun).where(CheckerRun.id == checker_run_id).with_for_update() + ) + async def lock_upload_session(self, session_id: str) -> ArtifactUploadSession | None: """Load one upload session with a row lock.""" result = await self._session.execute( @@ -373,6 +380,36 @@ async def add_verification_job(self, job: ArtifactVerificationJob) -> ArtifactVe await self._session.flush() return job + async def add_recovery_attempt( + self, attempt: ArtifactRecoveryAttempt + ) -> ArtifactRecoveryAttempt: + """Persist one recovery envelope inside the caller-owned transaction.""" + self._session.add(attempt) + await self._session.flush() + return attempt + + async def lock_recovery_by_source( + self, source_job_id: str + ) -> ArtifactRecoveryAttempt | None: + """Lock the lifetime recovery owner for one source verification job.""" + return await self._session.scalar( + select(ArtifactRecoveryAttempt) + .where(ArtifactRecoveryAttempt.source_verification_job_id == source_job_id) + .with_for_update() + .execution_options(populate_existing=True) + ) + + async def lock_recovery_by_retry( + self, retry_job_id: str + ) -> ArtifactRecoveryAttempt | None: + """Lock the envelope finalized by one retry verification job.""" + return await self._session.scalar( + select(ArtifactRecoveryAttempt) + .where(ArtifactRecoveryAttempt.retry_verification_job_id == retry_job_id) + .with_for_update() + .execution_options(populate_existing=True) + ) + async def lock_verification_job(self, job_id: str) -> ArtifactVerificationJob | None: return await self._session.scalar( select(ArtifactVerificationJob) diff --git a/backend/app/modules/artifacts/schemas.py b/backend/app/modules/artifacts/schemas.py index 994bd26af..383469b47 100644 --- a/backend/app/modules/artifacts/schemas.py +++ b/backend/app/modules/artifacts/schemas.py @@ -162,3 +162,71 @@ async def revalidate_terminal( ) -> None: del service_identity, action_id, facts raise ArtifactAuthorityDeniedError("artifact internal action is unavailable") + + +@dataclass(frozen=True, slots=True) +class ArtifactRecoveryResult: + """Stable identifiers returned by an exact recovery request or replay.""" + + recovery_attempt_id: UUID + source_verification_job_id: UUID + retry_verification_job_id: UUID + replayed: bool + + +class ArtifactRecoveryError(Exception): + """Base failure for an internal recovery request.""" + + +class ArtifactRecoveryConflictError(ArtifactRecoveryError): + """Raised when idempotency or lifetime source ownership conflicts.""" + + +class ArtifactRecoveryIneligibleError(ArtifactRecoveryError): + """Raised when the source job is not exhausted provider-unavailable work.""" + + +@final +@dataclass(frozen=True, slots=True) +class ArtifactRecoveryAuthorityFacts: + """Canonical facts bound to one exact Operator retry decision.""" + + project_id: UUID + task_id: UUID | None + submission_id: UUID | None + source_verification_job_id: UUID + expected_source_job_cas_version: int + + +@final +@dataclass(frozen=True, slots=True) +class ArtifactRecoveryAuthorizationEvidence: + """Privacy-bounded evidence returned by the AUTH-owned recovery seam.""" + + action_id: ActionId + permission_id: str + decision_id: UUID + + +class ArtifactRecoveryAuthority(Protocol): + """Fresh exact Operator authority seam owned by the later activation chunk.""" + + async def authorize( + self, + *, + authorization_context: AuthorizationContext, + facts: ArtifactRecoveryAuthorityFacts, + ) -> ArtifactRecoveryAuthorizationEvidence: ... + + +class DenyArtifactRecoveryAuthority: + """Production-safe recovery authority until AUTH activates the Operator action.""" + + async def authorize( + self, + *, + authorization_context: AuthorizationContext, + facts: ArtifactRecoveryAuthorityFacts, + ) -> ArtifactRecoveryAuthorizationEvidence: + del authorization_context, facts + raise ArtifactAuthorityDeniedError("artifact recovery action is unavailable") diff --git a/backend/app/modules/artifacts/service.py b/backend/app/modules/artifacts/service.py index f1830577a..38a8a7701 100644 --- a/backend/app/modules/artifacts/service.py +++ b/backend/app/modules/artifacts/service.py @@ -10,6 +10,7 @@ from uuid import UUID, uuid4 from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.exc import IntegrityError from app.core.config import Settings from app.core.hashing import canonical_json_hash @@ -27,6 +28,7 @@ artifact_store_namespace_material, ) from app.interfaces.external_services import ExternalServiceAdapterIdentity +from app.interfaces.artifact_operations import ArtifactRecoveryRequest from app.modules.actors.service import ActorService from app.modules.actors.service_identities import ServiceIdentity from app.modules.artifacts.models import ( @@ -37,6 +39,7 @@ ArtifactOperationReceipt, ArtifactPutObservationReceipt, ArtifactReplica, + ArtifactRecoveryAttempt, ArtifactStorageNamespace, ArtifactVerificationJob, ArtifactVerificationReceipt, @@ -50,6 +53,11 @@ ArtifactPutAttemptAuthorityFacts, ArtifactPendingWorkAuthorityFacts, ArtifactVerificationAuthorityFacts, + ArtifactRecoveryResult, + ArtifactRecoveryAuthority, + ArtifactRecoveryAuthorityFacts, + ArtifactRecoveryConflictError, + ArtifactRecoveryIneligibleError, DenyArtifactInternalAuthority, CheckerOutputArtifactAdmissionRequest, ContributorArtifactAdmissionRequest, @@ -64,7 +72,9 @@ IdentityLinkStatus, ServiceAuthorizationContext, ) -from app.modules.authorization.catalogue import ActionId +from app.modules.authorization.catalogue import ActionId, PermissionId +from app.modules.audit.repository import AuditRepository +from app.modules.tasks.models import AuditEvent ARTIFACT_STORAGE_NAMESPACE_ID = "primary" @@ -896,6 +906,8 @@ async def _record_verification_unavailable( job.terminal_at = now if exhausted else None job.terminal_result_code = "provider_unavailable" if exhausted else None _clear_job_fence(job) + if exhausted: + await self._finalize_recovery_for_job(job, "provider_unavailable", now) return "provider_unavailable" async def _complete_verification( @@ -993,6 +1005,8 @@ async def _complete_verification( item_changed = True elif outcome == "integrity_mismatch": item.state = "failed" + item.content_id = None + item.provider_object_ref = None item.error_code = "artifact_integrity_failure" item_changed = True if item_changed: @@ -1002,6 +1016,7 @@ async def _complete_verification( job.terminal_result_code = outcome job.terminal_at = now _clear_job_fence(job) + await self._finalize_recovery_for_job(job, outcome, now) return outcome async def _terminalize_verification_conflict( @@ -1025,6 +1040,39 @@ async def _terminalize_verification_conflict( job.terminal_result_code = "conflict" job.terminal_at = now _clear_job_fence(job) + await self._finalize_recovery_for_job(job, "conflict", now) + + async def _finalize_recovery_for_job( + self, job: ArtifactVerificationJob, outcome: str, now: datetime + ) -> None: + """Terminalize the linked recovery envelope under the job's held fence.""" + recovery = await self._repo.lock_recovery_by_retry(job.id) + if recovery is None: + return + if recovery.status != "requested" or recovery.terminal_at is not None: + raise ArtifactIngestStateError("artifact recovery envelope is already terminal") + audit_id = str(uuid4()) + await AuditRepository(self._session).add_audit_event( + ArtifactRecoveryService._audit_event( + event_id=audit_id, + event_type="ArtifactRecoveryCompleted", + recovery_id=recovery.id, + actor_id=ServiceIdentity.ARTIFACT_VERIFIER.value, + reason=recovery.reason, + payload={ + "source_verification_job_id": recovery.source_verification_job_id, + "retry_verification_job_id": job.id, + "terminal_result_code": outcome, + "execution_actor_kind": "service_identity", + "execution_service_identity": ServiceIdentity.ARTIFACT_VERIFIER.value, + }, + ) + ) + recovery.status = "succeeded" if outcome == "verified" else "failed" + recovery.terminal_result_code = outcome + recovery.terminal_at = now + recovery.terminal_audit_event_id = audit_id + recovery.cas_version += 1 async def _claim_and_validate_namespace(self) -> ArtifactStorageNamespace: """Atomically claim the singleton or reject deployment identity drift.""" @@ -1115,6 +1163,281 @@ async def scan(self) -> int: return len(put_ids) + len(job_ids) +class ArtifactRecoveryService: + """Create one idempotent read-only verification recovery chain link.""" + + _RECOVERY_CLASS = "provider_observation" + + def __init__( + self, + session: AsyncSession, + settings: Settings, + authority: ArtifactRecoveryAuthority, + ) -> None: + self._session = session + self._settings = settings + self._authority = authority + self._repo = ArtifactRepository(session) + self._actors = ActorService(session) + self._audit = AuditRepository(session) + + async def create(self, request: ArtifactRecoveryRequest) -> ArtifactRecoveryResult: + """Create or replay one envelope, retry job, and initiation audit atomically.""" + self._validate_request(request) + digest = self._request_digest(request) + source_id = str(request.source_verification_job_id) + try: + async with self._session.begin(): + return await self._create_locked(request, source_id, digest) + except IntegrityError: + # A concurrent winner may have committed either lifetime-source or + # idempotency ownership. The failed transaction is closed before + # the authoritative replay row is read. + async with self._session.begin(): + existing = await self._repo.lock_recovery_by_source(source_id) + if existing is not None and self._is_exact_replay(existing, request, digest): + await self._authorize_request(request) + return self._result(existing, replayed=True) + if existing is not None: + raise ArtifactRecoveryConflictError( + "artifact recovery source is already owned" + ) + raise + + async def _create_locked( + self, request: ArtifactRecoveryRequest, source_id: str, digest: str + ) -> ArtifactRecoveryResult: + existing = await self._repo.lock_recovery_by_source(source_id) + if existing is not None: + if self._is_exact_replay(existing, request, digest): + await self._authorize_request(request) + return self._result(existing, replayed=True) + raise ArtifactRecoveryConflictError("artifact recovery source is already owned") + source = await self._repo.lock_verification_job(source_id) + if source is None or not self._is_exhausted_unavailable(source): + raise ArtifactRecoveryIneligibleError( + "artifact verification job is not exhausted provider-unavailable work" + ) + if source.cas_version != request.expected_source_job_cas_version: + raise ArtifactRecoveryConflictError("artifact verification source changed") + put_attempt = await self._repo.lock_put_attempt(source.originating_put_attempt_id) + checker_run = ( + await self._repo.lock_checker_run(put_attempt.checker_run_id) + if put_attempt is not None and put_attempt.checker_run_id is not None + else None + ) + canonical_submission_id = checker_run.submission_id if checker_run is not None else None + if ( + put_attempt is None + or put_attempt.project_id != str(request.project_id) + or put_attempt.task_id + != (str(request.task_id) if request.task_id is not None else None) + or canonical_submission_id + != (str(request.submission_id) if request.submission_id is not None else None) + ): + raise ArtifactRecoveryConflictError("artifact recovery resource facts changed") + context = request.authorization_context + authorization = await self._authorize_request(request) + parent = await self._repo.lock_recovery_by_retry(source_id) + retry_id = str(uuid4()) + recovery_id = str(uuid4()) + audit_id = str(uuid4()) + retry = ArtifactVerificationJob( + id=retry_id, + originating_put_attempt_id=source.originating_put_attempt_id, + parent_verification_job_id=source.id, + replica_id=source.replica_id, + status="pending", + maximum_attempts=self._settings.artifact_provider_observation_maximum_attempts, + ) + await self._repo.add_verification_job(retry) + await self._audit.add_audit_event( + self._audit_event( + event_id=audit_id, + event_type="ArtifactRecoveryInitiated", + recovery_id=recovery_id, + actor_id=str(context.actor_profile_id), + reason=request.reason, + payload={ + "source_verification_job_id": source.id, + "retry_verification_job_id": retry_id, + "recovery_class": self._RECOVERY_CLASS, + "request_digest": digest, + "authorization_action_id": authorization.action_id.value, + "authorization_permission_id": authorization.permission_id, + "authorization_decision_id": str(authorization.decision_id), + "authorization_request_id": str(context.request_id), + "authorization_correlation_id": str(context.correlation_id), + }, + ) + ) + recovery = ArtifactRecoveryAttempt( + id=recovery_id, + requester_actor_profile_id=str(context.actor_profile_id), + requester_identity_link_id=str(context.identity_link_id), + authorization_request_id=str(context.request_id), + authorization_correlation_id=str(context.correlation_id), + project_id=str(request.project_id), + task_id=str(request.task_id) if request.task_id is not None else None, + submission_id=( + str(request.submission_id) if request.submission_id is not None else None + ), + source_verification_job_id=source.id, + retry_verification_job_id=retry_id, + parent_recovery_attempt_id=parent.id if parent is not None else None, + recovery_class=self._RECOVERY_CLASS, + reason=request.reason, + client_idempotency_key=request.client_idempotency_key, + request_digest=digest, + status="requested", + initiation_audit_event_id=audit_id, + ) + await self._repo.add_recovery_attempt(recovery) + return self._result(recovery, replayed=False) + + async def _authorize_request(self, request: ArtifactRecoveryRequest): + """Revalidate the human requester and exact Operator action on every call.""" + context = request.authorization_context + actor = await self._actors.lock_admission_proof( + context.actor_profile_id, context.identity_link_id + ) + if ( + context.actor_kind is not ActorKind.HUMAN + or context.actor_status is not ActorStatus.ACTIVE + or context.identity_link_status is not IdentityLinkStatus.ACTIVE + or actor is None + or actor.actor_kind != "human" + or actor.actor_status != "active" + or actor.service_identity is not None + or actor.identity_link_id != str(context.identity_link_id) + or actor.identity_link_subject_kind != "human" + or actor.identity_link_status != "active" + ): + raise ArtifactRecoveryConflictError("artifact recovery requester is unavailable") + authorization = await self._authority.authorize( + authorization_context=context, + facts=ArtifactRecoveryAuthorityFacts( + project_id=request.project_id, + task_id=request.task_id, + submission_id=request.submission_id, + source_verification_job_id=request.source_verification_job_id, + expected_source_job_cas_version=request.expected_source_job_cas_version, + ), + ) + if ( + authorization.action_id is not ActionId.ARTIFACT_VERIFICATION_JOB_RETRY + or authorization.permission_id + != PermissionId.ARTIFACT_VERIFICATION_JOB_RETRY.value + ): + raise ArtifactRecoveryConflictError( + "artifact recovery authorization evidence is invalid" + ) + return authorization + + @staticmethod + def _validate_request(request: ArtifactRecoveryRequest) -> None: + if type(request) is not ArtifactRecoveryRequest: + raise TypeError("invalid artifact recovery request") + if type(request.authorization_context) is not HumanAuthorizationContext: + raise TypeError("artifact recovery requires a human requester") + if ( + request.reason != request.reason.strip() + or not request.reason + or len(request.reason) > 1000 + or request.client_idempotency_key != request.client_idempotency_key.strip() + or not request.client_idempotency_key + or len(request.client_idempotency_key) > 200 + or request.expected_source_job_cas_version < 0 + ): + raise ValueError("invalid artifact recovery request") + + @classmethod + def _request_digest(cls, request: ArtifactRecoveryRequest) -> str: + context = request.authorization_context + return canonical_json_hash( + { + "requester_actor_profile_id": str(context.actor_profile_id), + "requester_identity_link_id": str(context.identity_link_id), + "project_id": str(request.project_id), + "task_id": str(request.task_id) if request.task_id is not None else None, + "submission_id": ( + str(request.submission_id) if request.submission_id is not None else None + ), + "source_verification_job_id": str(request.source_verification_job_id), + "recovery_class": cls._RECOVERY_CLASS, + "reason": request.reason, + "client_idempotency_key": request.client_idempotency_key, + "expected_source_job_cas_version": request.expected_source_job_cas_version, + } + ) + + @staticmethod + def _is_exhausted_unavailable(job: ArtifactVerificationJob) -> bool: + return ( + job.status == "provider_unavailable" + and job.terminal_result_code == "provider_unavailable" + and job.terminal_at is not None + and job.next_run_at is None + and job.executor_id is None + and job.lease_expires_at is None + and job.attempt_count >= job.maximum_attempts + ) + + @staticmethod + def _is_exact_replay( + attempt: ArtifactRecoveryAttempt, + request: ArtifactRecoveryRequest, + digest: str, + ) -> bool: + return ( + attempt.requester_actor_profile_id + == str(request.authorization_context.actor_profile_id) + and attempt.source_verification_job_id == str(request.source_verification_job_id) + and attempt.recovery_class == ArtifactRecoveryService._RECOVERY_CLASS + and attempt.client_idempotency_key == request.client_idempotency_key + and attempt.request_digest == digest + ) + + @staticmethod + def _result( + attempt: ArtifactRecoveryAttempt, *, replayed: bool + ) -> ArtifactRecoveryResult: + return ArtifactRecoveryResult( + recovery_attempt_id=UUID(attempt.id), + source_verification_job_id=UUID(attempt.source_verification_job_id), + retry_verification_job_id=UUID(attempt.retry_verification_job_id), + replayed=replayed, + ) + + @staticmethod + def _audit_event( + *, + event_id: str, + event_type: str, + recovery_id: str, + actor_id: str, + reason: str, + payload: dict[str, object], + ) -> AuditEvent: + """Build one privacy-bounded append-only legacy-domain artifact event.""" + return AuditEvent( + id=event_id, + entity_type="artifact_recovery_attempt", + entity_id=recovery_id, + event_type=event_type, + actor_id=actor_id, + external_subject=actor_id, + external_issuer="workstream", + actor_roles=[], + claim_snapshot={}, + auth_source="external_flow", + is_dev_auth=False, + reason=reason, + event_payload=payload, + event_domain="legacy_lifecycle", + ) + + class ArtifactAdmissionService: """Create one fully admitted put attempt without provider execution.""" diff --git a/backend/tests/test_alembic.py b/backend/tests/test_alembic.py index 4c9f50df3..6165fd29d 100644 --- a/backend/tests/test_alembic.py +++ b/backend/tests/test_alembic.py @@ -123,6 +123,70 @@ def test_alembic_upgrade_and_downgrade(isolated_database_env: str, migration_loc command.downgrade(config, "base") +def test_artifact_recovery_schema_and_empty_downgrade( + isolated_database_env: str, migration_lock +) -> None: + """Prove 0032 lineage indexes, custody triggers, and reversible empty state.""" + config = _alembic_config() + with migration_lock(): + try: + command.downgrade(config, "base") + command.upgrade(config, "head") + assert asyncio.run(_artifact_recovery_schema(isolated_database_env)) == { + "revision": "0032_artifact_recovery", + "constraints": { + "artifact_recovery_attempt_custody", + "artifact_verification_lineage_custody", + "uq_artifact_recovery_idempotency", + "uq_artifact_recovery_retry_job", + "uq_artifact_recovery_source_job", + "uq_artifact_verification_initial_origin", + "uq_artifact_verification_parent", + }, + } + command.downgrade(config, "0031_project_role_grants") + assert "artifact_recovery_attempts" not in asyncio.run( + _fetch_table_names(isolated_database_env) + ) + finally: + command.downgrade(config, "base") + + +async def _artifact_recovery_schema(database_url: str) -> dict[str, object]: + engine = create_async_engine(database_url) + try: + async with engine.connect() as connection: + revision = await connection.scalar(text("select version_num from alembic_version")) + names = set( + ( + await connection.execute( + text( + "select conname from pg_constraint where conrelid in " + "('artifact_recovery_attempts'::regclass, " + "'artifact_verification_jobs'::regclass) " + "union select tgname from pg_trigger where tgrelid in " + "('artifact_recovery_attempts'::regclass, " + "'artifact_verification_jobs'::regclass) and not tgisinternal " + "union select indexname from pg_indexes where indexname = " + "'uq_artifact_verification_initial_origin'" + ) + ) + ).scalars() + ) + expected = { + "artifact_recovery_attempt_custody", + "artifact_verification_lineage_custody", + "uq_artifact_recovery_idempotency", + "uq_artifact_recovery_retry_job", + "uq_artifact_recovery_source_job", + "uq_artifact_verification_initial_origin", + "uq_artifact_verification_parent", + } + return {"revision": revision, "constraints": names & expected} + finally: + await engine.dispose() + + def test_project_role_migration_constraints_and_immutable_history( isolated_database_env: str, migration_lock, @@ -135,7 +199,7 @@ def test_project_role_migration_constraints_and_immutable_history( command.upgrade(config, "head") result = asyncio.run(_exercise_project_role_migration(isolated_database_env)) assert result == { - "revision": "0031_project_role_grants", + "revision": "0032_artifact_recovery", "role_count": 3, "invalid_availability": "23514", "duplicate_role": "23505", @@ -343,7 +407,7 @@ def test_project_role_downgrade_refuses_each_reserved_evidence_predicate( ): command.downgrade(config, "0030_artifact_verification") assert asyncio.run(_project_role_refusal_state(isolated_database_env))[:3] == ( - "0031_project_role_grants", + "0032_artifact_recovery", True, True, ) @@ -370,7 +434,7 @@ def test_project_role_downgrade_refuses_each_reserved_evidence_predicate( ): command.downgrade(config, "0030_artifact_verification") assert asyncio.run(_project_role_refusal_state(isolated_database_env))[:3] == ( - "0031_project_role_grants", + "0032_artifact_recovery", True, True, ) @@ -398,7 +462,7 @@ def test_outbox_migration_schema_and_downgrade_writer_guard( command.upgrade(config, "head") schema = asyncio.run(_outbox_schema(isolated_database_env)) assert schema == { - "revision": "0031_project_role_grants", + "revision": "0032_artifact_recovery", "columns": { "aggregate_id", "aggregate_type", @@ -458,7 +522,7 @@ def test_outbox_migration_schema_and_downgrade_writer_guard( ) assert committed == "refused_after_commit" assert asyncio.run(_current_revision(isolated_database_env)) == ( - "0031_project_role_grants" + "0032_artifact_recovery" ) asyncio.run(_remove_outbox_migration_row(isolated_database_env, committed_project_id)) command.downgrade(config, "0028_artifact_admission") diff --git a/backend/tests/test_artifact_recovery.py b/backend/tests/test_artifact_recovery.py new file mode 100644 index 000000000..3912d2a79 --- /dev/null +++ b/backend/tests/test_artifact_recovery.py @@ -0,0 +1,739 @@ +"""PostgreSQL proofs for artifact recovery idempotency and retry lineage.""" + +from __future__ import annotations + +import asyncio +from datetime import UTC, datetime, timedelta +from pathlib import Path +from unittest.mock import AsyncMock +from uuid import UUID, uuid4 + +from alembic import command +from alembic.config import Config +import pytest +from sqlalchemy import func, select, text +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.adapters.artifacts.local import LocalStorageAdapter, LocalStorageBootstrap +from app.core.config import Settings +from app.core.hashing import canonical_json_hash +from app.interfaces.artifact_operations import ArtifactRecoveryRequest +from app.interfaces.artifacts import ( + ArtifactObjectMissingError, + ArtifactStoreError, + ArtifactStoreNamespaceClaim, + ArtifactStoreUnavailableError, +) +from app.modules.actors.models import ActorIdentityLink, ActorProfile +from app.modules.artifacts.models import ( + ArtifactPutAttempt, + ArtifactRecoveryAttempt, + ArtifactUploadItem, + ArtifactUploadSession, + ArtifactVerificationJob, +) +from app.modules.artifacts.schemas import ( + ArtifactRecoveryAuthorizationEvidence, + ArtifactRecoveryConflictError, + ArtifactRecoveryIneligibleError, + ArtifactAuthorityDeniedError, + DenyArtifactRecoveryAuthority, + ContributorArtifactAdmissionRequest, + GuideArtifactAdmissionRequest, +) +from app.modules.artifacts.service import ( + ArtifactAdmissionService, + ArtifactRecoveryService, + ArtifactStorageOrchestrator, + artifact_storage_namespace_spec, +) +from app.modules.authorization.runtime import ( + ActorKind, + ActorStatus, + HumanAuthorizationContext, + IdentityLinkStatus, +) +from app.modules.authorization.catalogue import ActionId, PermissionId +from app.modules.projects.models import ( + GuideSourceSnapshot, + GuideSourceSnapshotItem, + Project, + ProjectGuide, +) +from app.modules.tasks.models import AuditEvent, WorkstreamTask +from tests.artifact_store_helpers import artifact_admission_limit_settings, minted_source + + +class _AllowArtifactAuthority: + async def preflight(self, **_values: object) -> None: ... + + async def revalidate_terminal(self, **_values: object) -> None: ... + + +class _DenyTerminalArtifactAuthority(_AllowArtifactAuthority): + async def revalidate_terminal(self, **_values: object) -> None: + raise ArtifactAuthorityDeniedError("terminal authority changed") + + +class _AllowRecoveryAuthority: + async def authorize(self, **_values: object) -> ArtifactRecoveryAuthorizationEvidence: + return ArtifactRecoveryAuthorizationEvidence( + action_id=ActionId.ARTIFACT_VERIFICATION_JOB_RETRY, + permission_id=PermissionId.ARTIFACT_VERIFICATION_JOB_RETRY.value, + decision_id=uuid4(), + ) + + +def _alembic_config() -> Config: + root = Path(__file__).resolve().parents[1] + config = Config(str(root / "alembic.ini")) + config.set_main_option("script_location", str(root / "alembic")) + return config + + +@pytest.fixture +def recovery_database_env(isolated_database_env: str, migration_lock) -> str: + config = _alembic_config() + with migration_lock(): + asyncio.run(_reset_schema(isolated_database_env)) + command.upgrade(config, "head") + try: + yield isolated_database_env + finally: + asyncio.run(_reset_schema(isolated_database_env)) + + +async def _reset_schema(database_url: str) -> None: + engine = create_async_engine(database_url) + try: + async with engine.begin() as connection: + await connection.execute(text("drop schema if exists public cascade")) + await connection.execute(text("create schema public")) + finally: + await engine.dispose() + + +def _settings(tmp_path: Path) -> Settings: + root = tmp_path / "durable" + root.mkdir(mode=0o700, parents=True) + return Settings( + **artifact_admission_limit_settings(1024), + environment="test", + artifact_store_backend="local", + artifact_local_root=root, + artifact_scratch_root=tmp_path / "scratch", + artifact_scratch_minimum_free_bytes=0, + artifact_provider_observation_maximum_attempts=1, + ) + + +def _context() -> HumanAuthorizationContext: + return HumanAuthorizationContext( + actor_profile_id=uuid4(), + actor_kind=ActorKind.HUMAN, + actor_status=ActorStatus.ACTIVE, + identity_link_id=uuid4(), + identity_link_status=IdentityLinkStatus.ACTIVE, + request_id=uuid4(), + correlation_id=uuid4(), + ) + + +async def _seed_contributor(session, context, sha256: str, size: int) -> tuple[str, str, str]: + actor_id = str(context.actor_profile_id) + project_id, task_id, upload_id, item_id = (str(uuid4()) for _ in range(4)) + session.add( + ActorProfile( + id=actor_id, + actor_kind="human", + status="active", + provisioning_method="automatic_first_access", + created_by="test", + ) + ) + await session.flush() + session.add( + ActorIdentityLink( + id=str(context.identity_link_id), + actor_profile_id=actor_id, + issuer="https://issuer.example.test", + subject=f"human-{actor_id}", + subject_kind="human", + status="active", + linked_by="test", + last_verified_at=datetime.now(UTC), + ) + ) + session.add(Project(id=project_id, name="Recovery project", slug=f"recovery-{project_id}")) + await session.flush() + session.add( + WorkstreamTask( + id=task_id, + project_id=project_id, + title="Recovery task", + description="Prove recovery.", + status="draft", + created_by="test", + ) + ) + await session.flush() + session.add( + ArtifactUploadSession( + id=upload_id, + actor_id=actor_id, + project_id=project_id, + task_id=task_id, + permitted_roles=["submission"], + state="open", + maximum_bytes=size, + current_bytes=0, + reserved_bytes=size, + maximum_items=1, + current_items=0, + reserved_items=1, + expires_at=datetime.now(UTC) + timedelta(minutes=10), + cas_version=0, + ) + ) + await session.flush() + session.add( + ArtifactUploadItem( + id=item_id, + session_id=upload_id, + logical_role="submission", + display_name="result.bin", + media_type="application/octet-stream", + reserved_bytes=size, + expected_sha256=sha256, + expected_size=size, + idempotency_key=f"put-{item_id}", + request_digest=canonical_json_hash({"sha256": sha256, "size": size}), + state="reserved", + cas_version=0, + ) + ) + await session.commit() + return project_id, task_id, item_id + + +async def _exhausted_job(session, settings, tmp_path, context): + namespace = artifact_storage_namespace_spec( + settings, + LocalStorageBootstrap(LocalStorageAdapter(root=settings.artifact_local_root)), + ) + bootstrap = LocalStorageBootstrap(LocalStorageAdapter(root=settings.artifact_local_root)) + store = bootstrap.initialize_after_namespace_claim( + ArtifactStoreNamespaceClaim( + adapter_identity=bootstrap.identity, + namespace_identity=bootstrap.namespace_identity, + namespace_fingerprint=namespace.namespace_fingerprint, + ) + ) + source_cm = minted_source(tmp_path / "source", b"recover me") + source = await source_cm.__aenter__() + project_id, task_id, item_id = await _seed_contributor( + session, context, source.commitment.sha256, source.commitment.byte_count + ) + admission = await ArtifactAdmissionService(session, settings, namespace).admit( + ContributorArtifactAdmissionRequest( + authorization_context=context, + upload_item_id=UUID(item_id), + source=source, + ) + ) + orchestrator = ArtifactStorageOrchestrator( + session, store, namespace, settings, _AllowArtifactAuthority() + ) + await orchestrator.execute_committed_put(attempt_id=admission.attempt_id, source=source) + job = await session.scalar(select(ArtifactVerificationJob)) + assert job is not None + job_id = UUID(job.id) + await session.rollback() + orchestrator._read_complete = AsyncMock( + side_effect=ArtifactStoreUnavailableError("unavailable") + ) + await orchestrator.verify_object(job_id) + job = await session.get(ArtifactVerificationJob, str(job_id)) + assert job is not None + await session.refresh(job) + await session.commit() + await source_cm.__aexit__(None, None, None) + assert job.terminal_result_code == "provider_unavailable" + return project_id, task_id, job, orchestrator, bootstrap + + +async def _exhausted_guide_job(session, settings, tmp_path, context): + namespace = artifact_storage_namespace_spec( + settings, + LocalStorageBootstrap(LocalStorageAdapter(root=settings.artifact_local_root)), + ) + bootstrap = LocalStorageBootstrap(LocalStorageAdapter(root=settings.artifact_local_root)) + store = bootstrap.initialize_after_namespace_claim( + ArtifactStoreNamespaceClaim( + adapter_identity=bootstrap.identity, + namespace_identity=bootstrap.namespace_identity, + namespace_fingerprint=namespace.namespace_fingerprint, + ) + ) + source_cm = minted_source(tmp_path / "guide-source", b"recover guide") + source = await source_cm.__aenter__() + await _seed_contributor(session, context, source.commitment.sha256, source.commitment.byte_count) + project = await session.scalar(select(Project)) + assert project is not None + project_id = project.id + guide_id, snapshot_id, item_id = (str(uuid4()) for _ in range(3)) + session.add( + ProjectGuide( + id=guide_id, + project_id=project_id, + version="v1", + status="draft", + content_markdown="# Guide", + created_by="test", + ) + ) + await session.flush() + session.add( + GuideSourceSnapshot( + id=snapshot_id, + project_id=project_id, + guide_id=guide_id, + guide_version="v1", + manifest_schema_version="v1", + manifest_json={"items": [item_id]}, + bundle_hash=canonical_json_hash({"items": [item_id]}), + captured_by=str(context.actor_profile_id), + ) + ) + await session.flush() + session.add( + GuideSourceSnapshotItem( + id=item_id, + source_snapshot_id=snapshot_id, + item_order=0, + source_kind="inline", + durable_ref="guide.md", + ingestion_adapter="inline", + content_hash=source.commitment.sha256, + media_type=source.commitment.media_type, + ) + ) + await session.commit() + admission = await ArtifactAdmissionService(session, settings, namespace).admit( + GuideArtifactAdmissionRequest( + authorization_context=context, + guide_source_item_id=UUID(item_id), + source=source, + ) + ) + orchestrator = ArtifactStorageOrchestrator( + session, store, namespace, settings, _AllowArtifactAuthority() + ) + await orchestrator.execute_committed_put(attempt_id=admission.attempt_id, source=source) + job = await session.scalar( + select(ArtifactVerificationJob).where( + ArtifactVerificationJob.originating_put_attempt_id == str(admission.attempt_id) + ) + ) + if job is None: + attempt = await session.get(ArtifactPutAttempt, str(admission.attempt_id)) + assert attempt is not None and attempt.replica_id is not None + job = ArtifactVerificationJob( + id=str(uuid4()), + originating_put_attempt_id=attempt.id, + replica_id=attempt.replica_id, + status="pending", + maximum_attempts=1, + ) + session.add(job) + await session.commit() + job_id = job.id + await session.rollback() + orchestrator._read_complete = AsyncMock(side_effect=ArtifactStoreUnavailableError("down")) + await orchestrator.verify_object(UUID(job_id)) + job = await session.get(ArtifactVerificationJob, job_id) + assert job is not None + await session.refresh(job) + await session.commit() + await source_cm.__aexit__(None, None, None) + return project_id, job, bootstrap + + +def _request(context, project_id: str, task_id: str | None, job, **changes): + values = dict( + authorization_context=context, + project_id=UUID(project_id), + task_id=UUID(task_id) if task_id is not None else None, + submission_id=None, + source_verification_job_id=UUID(job.id), + reason="provider remained unavailable", + client_idempotency_key="recovery-1", + expected_source_job_cas_version=job.cas_version, + ) + values.update(changes) + return ArtifactRecoveryRequest(**values) + + +@pytest.mark.asyncio +async def test_exact_replay_creates_one_recovery_job_and_audit( + recovery_database_env: str, tmp_path: Path +) -> None: + engine = create_async_engine(recovery_database_env) + factory = async_sessionmaker(engine, expire_on_commit=False) + try: + async with factory() as session: + context = _context() + settings = _settings(tmp_path) + project_id, task_id, source, _orchestrator, bootstrap = await _exhausted_job( + session, settings, tmp_path, context + ) + service = ArtifactRecoveryService(session, settings, _AllowRecoveryAuthority()) + request = _request(context, project_id, task_id, source) + first = await service.create(request) + replay = await service.create(request) + assert first.retry_verification_job_id == replay.retry_verification_job_id + assert replay.replayed is True + assert await session.scalar(select(func.count(ArtifactRecoveryAttempt.id))) == 1 + assert await session.scalar(select(func.count(ArtifactVerificationJob.id))) == 2 + assert await session.scalar( + select(func.count(AuditEvent.id)).where( + AuditEvent.event_type == "ArtifactRecoveryInitiated" + ) + ) == 1 + bootstrap.close() + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_taskless_recovery_and_deny_only_authority_boundary( + recovery_database_env: str, tmp_path: Path +) -> None: + engine = create_async_engine(recovery_database_env) + factory = async_sessionmaker(engine, expire_on_commit=False) + try: + async with factory() as session: + context = _context() + settings = _settings(tmp_path) + project_id, source, bootstrap = await _exhausted_guide_job( + session, settings, tmp_path, context + ) + request = _request(context, project_id, None, source) + with pytest.raises(ArtifactAuthorityDeniedError): + await ArtifactRecoveryService( + session, settings, DenyArtifactRecoveryAuthority() + ).create(request) + assert await session.scalar(select(func.count(ArtifactRecoveryAttempt.id))) == 0 + assert await session.scalar(select(func.count(ArtifactVerificationJob.id))) == 1 + await session.rollback() + created = await ArtifactRecoveryService( + session, settings, _AllowRecoveryAuthority() + ).create(request) + with pytest.raises(ArtifactAuthorityDeniedError): + await ArtifactRecoveryService( + session, settings, DenyArtifactRecoveryAuthority() + ).create(request) + assert await session.scalar(select(func.count(ArtifactRecoveryAttempt.id))) == 1 + assert await session.scalar(select(func.count(ArtifactVerificationJob.id))) == 2 + assert await session.scalar( + select(func.count(AuditEvent.id)).where( + AuditEvent.event_type == "ArtifactRecoveryInitiated" + ) + ) == 1 + await session.rollback() + replay = await ArtifactRecoveryService( + session, settings, _AllowRecoveryAuthority() + ).create(request) + assert replay.replayed is True + assert created.retry_verification_job_id == replay.retry_verification_job_id + bootstrap.close() + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_changed_or_ineligible_recovery_has_no_side_effects( + recovery_database_env: str, tmp_path: Path +) -> None: + engine = create_async_engine(recovery_database_env) + factory = async_sessionmaker(engine, expire_on_commit=False) + try: + async with factory() as session: + context = _context() + settings = _settings(tmp_path) + project_id, task_id, source, _orchestrator, bootstrap = await _exhausted_job( + session, settings, tmp_path, context + ) + service = ArtifactRecoveryService(session, settings, _AllowRecoveryAuthority()) + created = await service.create(_request(context, project_id, task_id, source)) + with pytest.raises(ArtifactRecoveryConflictError): + await service.create( + _request( + context, + project_id, + task_id, + source, + reason="changed", + ) + ) + source.status = "verified" + source.terminal_result_code = "verified" + await session.commit() + with pytest.raises(ArtifactRecoveryConflictError): + await service.create( + _request( + context, + project_id, + task_id, + source, + client_idempotency_key="different", + ) + ) + assert await session.scalar(select(func.count(ArtifactRecoveryAttempt.id))) == 1 + + retry = await session.get( + ArtifactVerificationJob, str(created.retry_verification_job_id) + ) + assert retry is not None + retry_request = _request( + context, + project_id, + task_id, + retry, + client_idempotency_key="retry-pending", + ) + await session.rollback() + with pytest.raises(ArtifactRecoveryIneligibleError): + await service.create(retry_request) + bootstrap.close() + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_retry_terminalizes_recovery_under_verification_fence( + recovery_database_env: str, tmp_path: Path +) -> None: + engine = create_async_engine(recovery_database_env) + factory = async_sessionmaker(engine, expire_on_commit=False) + try: + async with factory() as session: + context = _context() + settings = _settings(tmp_path) + project_id, task_id, source, orchestrator, bootstrap = await _exhausted_job( + session, settings, tmp_path, context + ) + created = await ArtifactRecoveryService( + session, settings, _AllowRecoveryAuthority() + ).create( + _request(context, project_id, task_id, source) + ) + orchestrator._read_complete = ArtifactStorageOrchestrator._read_complete.__get__( + orchestrator + ) + assert ( + await orchestrator.verify_object(created.retry_verification_job_id) + == "verified" + ) + recovery = await session.get( + ArtifactRecoveryAttempt, str(created.recovery_attempt_id) + ) + assert recovery is not None + assert (recovery.status, recovery.terminal_result_code) == ( + "succeeded", + "verified", + ) + assert recovery.terminal_audit_event_id is not None + assert await session.scalar( + select(func.count(AuditEvent.id)).where( + AuditEvent.event_type == "ArtifactRecoveryCompleted" + ) + ) == 1 + bootstrap.close() + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_terminal_authority_drift_writes_no_recovery_terminal_facts( + recovery_database_env: str, tmp_path: Path +) -> None: + engine = create_async_engine(recovery_database_env) + factory = async_sessionmaker(engine, expire_on_commit=False) + try: + async with factory() as session: + context = _context() + settings = _settings(tmp_path) + project_id, task_id, source, orchestrator, bootstrap = await _exhausted_job( + session, settings, tmp_path, context + ) + created = await ArtifactRecoveryService( + session, settings, _AllowRecoveryAuthority() + ).create(_request(context, project_id, task_id, source)) + orchestrator._authority = _DenyTerminalArtifactAuthority() + orchestrator._read_complete = ArtifactStorageOrchestrator._read_complete.__get__( + orchestrator + ) + with pytest.raises(ArtifactAuthorityDeniedError): + await orchestrator.verify_object(created.retry_verification_job_id) + recovery = await session.get( + ArtifactRecoveryAttempt, str(created.recovery_attempt_id) + ) + retry = await session.get( + ArtifactVerificationJob, str(created.retry_verification_job_id) + ) + assert recovery is not None and retry is not None + assert recovery.status == "requested" + assert recovery.terminal_at is None + assert retry.status == "running" + assert retry.terminal_at is None + assert await session.scalar( + select(func.count(AuditEvent.id)).where( + AuditEvent.event_type == "ArtifactRecoveryCompleted" + ) + ) == 0 + bootstrap.close() + finally: + await engine.dispose() + + +@pytest.mark.parametrize( + ("provider_result", "expected_outcome"), + [ + (ArtifactObjectMissingError("missing"), "missing"), + (("sha256:" + "0" * 64, 1), "integrity_mismatch"), + (ArtifactStoreError("provider conflict"), "conflict"), + ], +) +@pytest.mark.asyncio +async def test_every_failed_retry_outcome_terminalizes_recovery_once( + recovery_database_env: str, + tmp_path: Path, + provider_result: object, + expected_outcome: str, +) -> None: + engine = create_async_engine(recovery_database_env) + factory = async_sessionmaker(engine, expire_on_commit=False) + try: + async with factory() as session: + context = _context() + settings = _settings(tmp_path) + project_id, task_id, source, orchestrator, bootstrap = await _exhausted_job( + session, settings, tmp_path, context + ) + created = await ArtifactRecoveryService( + session, settings, _AllowRecoveryAuthority() + ).create(_request(context, project_id, task_id, source)) + if isinstance(provider_result, Exception): + orchestrator._read_complete = AsyncMock(side_effect=provider_result) + else: + orchestrator._read_complete = AsyncMock(return_value=provider_result) + assert ( + await orchestrator.verify_object(created.retry_verification_job_id) + == expected_outcome + ) + recovery = await session.get( + ArtifactRecoveryAttempt, str(created.recovery_attempt_id) + ) + assert recovery is not None + assert (recovery.status, recovery.terminal_result_code) == ( + "failed", + expected_outcome, + ) + assert await session.scalar( + select(func.count(AuditEvent.id)).where( + AuditEvent.event_type == "ArtifactRecoveryCompleted" + ) + ) == 1 + bootstrap.close() + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_concurrent_exact_replay_has_one_envelope_and_retry_job( + recovery_database_env: str, tmp_path: Path +) -> None: + engine = create_async_engine(recovery_database_env) + factory = async_sessionmaker(engine, expire_on_commit=False) + bootstrap = None + try: + async with factory() as setup: + context = _context() + settings = _settings(tmp_path) + project_id, task_id, source, _orchestrator, bootstrap = await _exhausted_job( + setup, settings, tmp_path, context + ) + request = _request(context, project_id, task_id, source) + async with factory() as first_session, factory() as second_session: + first, second = await asyncio.gather( + ArtifactRecoveryService( + first_session, settings, _AllowRecoveryAuthority() + ).create(request), + ArtifactRecoveryService( + second_session, settings, _AllowRecoveryAuthority() + ).create(request), + ) + assert {first.replayed, second.replayed} == {False, True} + assert first.retry_verification_job_id == second.retry_verification_job_id + async with factory() as proof: + assert await proof.scalar(select(func.count(ArtifactRecoveryAttempt.id))) == 1 + assert await proof.scalar(select(func.count(ArtifactVerificationJob.id))) == 2 + finally: + if bootstrap is not None: + bootstrap.close() + await engine.dispose() + + +@pytest.mark.asyncio +async def test_exhausted_retry_can_form_only_the_next_linear_chain_link( + recovery_database_env: str, tmp_path: Path +) -> None: + engine = create_async_engine(recovery_database_env) + factory = async_sessionmaker(engine, expire_on_commit=False) + try: + async with factory() as session: + context = _context() + settings = _settings(tmp_path) + project_id, task_id, source, orchestrator, bootstrap = await _exhausted_job( + session, settings, tmp_path, context + ) + service = ArtifactRecoveryService(session, settings, _AllowRecoveryAuthority()) + first = await service.create(_request(context, project_id, task_id, source)) + orchestrator._read_complete = AsyncMock( + side_effect=ArtifactStoreUnavailableError("still unavailable") + ) + assert ( + await orchestrator.verify_object(first.retry_verification_job_id) + == "provider_unavailable" + ) + retry = await session.get( + ArtifactVerificationJob, str(first.retry_verification_job_id) + ) + first_attempt = await session.get( + ArtifactRecoveryAttempt, str(first.recovery_attempt_id) + ) + assert retry is not None and first_attempt is not None + assert (first_attempt.status, first_attempt.terminal_result_code) == ( + "failed", + "provider_unavailable", + ) + second_request = _request( + context, + project_id, + task_id, + retry, + client_idempotency_key="recovery-2", + ) + first_attempt_id = first_attempt.id + await session.rollback() + second = await service.create(second_request) + second_attempt = await session.get( + ArtifactRecoveryAttempt, str(second.recovery_attempt_id) + ) + assert second_attempt is not None + assert second_attempt.parent_recovery_attempt_id == first_attempt_id + assert second.source_verification_job_id == first.retry_verification_job_id + bootstrap.close() + finally: + await engine.dispose()