diff --git a/CHANGELOG.d/0.79.0-analysis-run-registry.md b/CHANGELOG.d/0.79.0-analysis-run-registry.md new file mode 100644 index 000000000..9a99b4895 --- /dev/null +++ b/CHANGELOG.d/0.79.0-analysis-run-registry.md @@ -0,0 +1,30 @@ +# 0.79.0 — Normalized analysis-run registry + +- Adds an additive, third-normalized PostgreSQL registry for immutable source + captures, aggregate reconciliation counts, account-scoped idempotent analysis + requests, authorization-relevant run scopes, and append-only lifecycle events. +- Assigns `maximum_available_time` and `captured_at` to the reusable source + snapshot while assigning `knowledge_cutoff` to each analysis run. +- Rejects future-information leakage when a source contains evidence unavailable + at the requested run cutoff. +- Freezes source snapshots, aggregate counts, run requests, and run scopes; count + insert/delete is serialized with run creation so the evidence set cannot race + the first derivation. +- Rejects both update and deletion of a persisted run scope so an analysis cannot + lose or rewrite the authorization boundary under which it was requested. +- Enforces a contiguous, time-monotonic analysis state machine with a required + `pending` first event, legal running/terminal transitions, separate occurrence + and database record clocks, and immutable status history. +- Scopes idempotency to the real requesting account and requires every run to + reference an authenticated `user_account`. +- Derives current state through `analysis_run_current_status` instead of storing + a second mutable status authority. +- Binds runs to configuration, optional model/prompt, and code-revision digests + without storing source text, source-table names, credentials, provider + payloads, raw exceptions, or organization-specific fixtures. +- Adds real-PostgreSQL contracts for temporal eligibility, immutable evidence, + idempotency ownership, immutable authorization scope, legal state transitions, + fail-closed rollback, and reproducible fresh-container migration order. +- Does not yet claim an analysis-run API, Valkey outbox, TEPP execution adapter, + contextual-orchestrator execution, administrator screen, or actual-data + acceptance run; those remain separate Milestone 2 vertical slices. diff --git a/docker/postgres-init/Dockerfile b/docker/postgres-init/Dockerfile index d10dec64b..2160445e0 100644 --- a/docker/postgres-init/Dockerfile +++ b/docker/postgres-init/Dockerfile @@ -24,6 +24,8 @@ COPY migrations/0014_role_responsibility_team_actor_type.sql /docker-entrypoint- COPY migrations/0015_organization_name_resolution.sql /docker-entrypoint-initdb.d/16-organization-name-resolution.sql COPY migrations/0016_cross_post_actor_identity.sql /docker-entrypoint-initdb.d/17-cross-post-actor-identity.sql COPY migrations/0017_prov_o_standard_relations.sql /docker-entrypoint-initdb.d/18-prov-o-standard-relations.sql +COPY migrations/0018_analysis_run_registry.sql /docker-entrypoint-initdb.d/19-analysis-run-registry.sql +COPY migrations/0019_analysis_run_scope_immutability.sql /docker-entrypoint-initdb.d/20-analysis-run-scope-immutability.sql # Official image already drops to this account at runtime; declare it so # the Dockerfile itself satisfies DS-0002 (explicit non-root USER). USER postgres diff --git a/docs/adr/0013-normalized-analysis-run-registry.md b/docs/adr/0013-normalized-analysis-run-registry.md new file mode 100644 index 000000000..0b84c0f3a --- /dev/null +++ b/docs/adr/0013-normalized-analysis-run-registry.md @@ -0,0 +1,269 @@ +# ADR 0013 — Milestone 2 analysis runs use a normalized additive registry + +**Decision status:** Accepted on this active PR; not protected-main truth until merge +**Date:** 2026-08-15 + +## Context + +The retained Milestone 2 experiment demonstrated useful direct-PostgreSQL +analysis, but its run shape repeated aggregate counts beside a free-form +metadata object and belonged to a parallel replacement application. Merging +that branch would delete or duplicate the reviewed LineageWeave package, +migration lineage, PROV-O layer, identity boundary, and React product. + +Issue #79 therefore requires an additive bridge on the post-ADR-0012 product +line. The first bridge must preserve reproducibility, temporal eligibility, +authorization scope, and operational evidence without copying source records, +organization-specific identifiers, source-table names, provider credentials, +raw exceptions, or cross-service application tables. + +The existing product owns authenticated accounts, corporate entities, process +units, source posts, compact lineage edges, report scores, and PROV-O +persistence. TEPP owns temporal and psychometric estimation. +`contextual-orchestrator` owns model routing and provider execution. The +registry records that a product analysis was requested and which immutable +evidence, scope, and configuration it used; it does not become either service's +internal database. + +## Alternatives considered + +### Copy the experiment tables unchanged + +Rejected. Repeated counts and unstructured metadata create competing sources of +truth, weaken relational constraints, and reopen a parallel product. + +### Store one JSON document per run + +Rejected. Signed external manifests may use JSON, but identity, scope, +idempotency, aggregate counts, clocks, and lifecycle rules must remain +independently queryable and enforceable in PostgreSQL. + +### Put durable state only in Valkey + +Rejected. Valkey remains the event queue. Durable audit identity, +idempotency, scope, and reproducibility evidence require PostgreSQL; queue state +must be reconstructable from durable product state. + +### Use a normalized additive registry + +Accepted. It preserves useful experimental evidence while maintaining the +existing bounded contexts and migration lineage. + +## Decision + +Migration `0018_analysis_run_registry.sql` introduces five third-normalized +relations and one read projection. Migration +`0019_analysis_run_scope_immutability.sql` hardens the authorization boundary +without redefining the schema: + +```mermaid +erDiagram + ANALYSIS_SOURCE_SNAPSHOT ||--o{ ANALYSIS_SOURCE_COUNT : records + ANALYSIS_SOURCE_SNAPSHOT ||--o{ ANALYSIS_RUN : anchors + USER_ACCOUNT ||--o{ ANALYSIS_RUN : requests + ANALYSIS_RUN ||--o| ANALYSIS_RUN_SCOPE : limits + CORPORATE_ENTITY |o--o{ ANALYSIS_RUN_SCOPE : scopes + PROCESS_UNIT |o--o{ ANALYSIS_RUN_SCOPE : scopes + ANALYSIS_RUN ||--o{ ANALYSIS_RUN_STATUS_EVENT : records + + ANALYSIS_SOURCE_SNAPSHOT { + uuid analysis_source_snapshot_id PK + text snapshot_sha256 UK + text source_contract_version UK + timestamptz maximum_available_time + timestamptz captured_at + } + ANALYSIS_SOURCE_COUNT { + uuid analysis_source_snapshot_id PK,FK + text count_type_code PK,FK + bigint count_value + } + ANALYSIS_RUN { + uuid analysis_run_id PK + uuid analysis_source_snapshot_id FK + uuid requested_by_account_id FK + text idempotency_key UK + timestamptz knowledge_cutoff + text run_kind_code FK + text configuration_sha256 + text model_contract_sha256 + text prompt_bundle_sha256 + text code_revision_sha + } + ANALYSIS_RUN_SCOPE { + uuid analysis_run_id PK,FK + text scope_kind_code FK + uuid corporate_entity_id FK + uuid process_unit_id FK + text scope_key + } + ANALYSIS_RUN_STATUS_EVENT { + uuid analysis_run_id PK,FK + int status_ordinal PK + text status_code FK + timestamptz occurred_at + timestamptz recorded_at + text failure_code + boolean retryable + } +``` + +### Source capture and run-owned temporal cutoff + +`analysis_source_snapshot` identifies one immutable capture by exact digest and +source-contract revision. It stores: + +- `maximum_available_time`: the latest evidence-availability time represented + by the capture; +- `captured_at`: when that immutable capture was created. + +`knowledge_cutoff` belongs to `analysis_run`, not the snapshot. One immutable +capture may therefore support several analyses with different later cutoffs. +The run-creation trigger locks the snapshot and enforces: + +```text +maximum_available_time <= knowledge_cutoff +``` + +This is an aggregate product guard against future-information leakage. It does +not replace TEPP's document-, event-, assertion-, system-, availability-, and +analysis-cutoff clocks. + +### Immutable evidence and concurrency + +`analysis_source_count` stores one non-negative aggregate per snapshot and count +vocabulary. Snapshot rows and existing counts reject updates. Count insert or +delete and run creation lock the same snapshot row. This makes the boundary +race-safe: + +- a count-set change that wins the lock completes before the first run starts; +- a run that wins the lock freezes the count set, and later count insert/delete + fails closed. + +`analysis_run` is an immutable request. The account is required, and the +idempotency key is unique within that requesting account rather than globally. +Independent authenticated users may therefore choose the same opaque key +without colliding, while one user cannot reuse a key for a second request. +Request updates and deletes fail closed after registration. + +### Immutable authorization scope + +`analysis_run_scope` stores at most one authorization-relevant product scope. +Corporate-entity, process-unit, thread-group, and all-visible scopes use +mutually exclusive columns. Process-unit ownership remains derivable from +`process_unit` rather than being duplicated. The later creation repository must +insert the required scope in the same transaction as the run. + +Migration 0018 originally rejected scope updates but still allowed a direct +scope deletion. That would leave a durable run and lifecycle history after its +recorded authorization boundary had disappeared. Migration 0019 therefore +replaces the update-only guard with an update-or-delete guard. Its replay-safe +rollback restores the migration-0018 update-only policy without deleting run or +scope data. + +### Ordered lifecycle + +`analysis_run_status_event` is an append-only state machine rather than an +unordered event bag. PostgreSQL serializes status appends per run and enforces: + +```text +pending -> running -> succeeded + \-> failed + \-> cancelled + +pending -> failed | cancelled +``` + +The first event is ordinal 1 and `pending`; ordinals are contiguous; occurrence +time is nondecreasing; terminal states cannot transition; failed events require +a bounded machine failure code; non-failed events cannot carry failure or retry +metadata. `recorded_at` separately preserves the database system clock. + +`analysis_run_current_status` derives the highest ordinal event. It is a view, +not a second mutable state authority. + +### Lookup, migration, rollback, and ownership + +All enum-like values remain in `common_lookup_value`. Column checks additionally +restrict each field to its own allowed code family because the repository's +shared lookup foreign key targets globally unique codes. + +Migration 0018 is replay-safe. Its rollback refuses to drop non-empty registry +relations, so downgrade cannot silently destroy audit evidence. Migration 0019 +is also replay-safe; its rollback changes only the scope mutation policy. Any +approved retention/export process that empties append-only evidence must be +explicit and audited before the registry rollback. + +The PostgreSQL image applies migrations 0018 and 0019 after the reviewed PROV-O +migration. This PR adds no second web application, Keyverse imitation, TEPP +arithmetic, contextual-orchestrator database dependency, API, or UI. + +## Consequences + +- The product gains a durable base for analysis APIs, actual-data aggregate + reconciliation, TEPP adapters, Valkey outbox delivery, and administrator run + visibility. +- Source rows, document nodes, lineage edges, report payloads, and scientific + artifacts remain in their existing owners; this registry never duplicates + them. +- Application/API writers and row-level authorization are deliberately deferred + to the next vertical slice. A schema existing is not a claim that users can + submit or inspect runs yet. +- RLS is not enabled here because the current FastAPI application authorizes + through a pooled service identity and application-level RBAC/ABAC. A future + RLS design requires a separate ADR and transaction-scoped actor context. +- Append-only evidence, immutable requests, and immutable scopes increase + operational safety but require explicit retention/export tooling before + destructive cleanup. + +## Verification + +- Static contracts reject the legacy denormalized table, JSON metadata, + temporary repair artifacts, ambiguous clock ownership, optional requester, + globally scoped idempotency, and missing 0019 fresh-install wiring. +- Real PostgreSQL tests apply migrations 0018 and 0019 and exercise valid + snapshot/run/scope/status writes. +- Database regressions reject evidence later than the run cutoff, snapshot and + count mutation, post-run count-set changes, missing actors, same-account + idempotency reuse, malformed digests, negative counts, incoherent scopes, + scope update/delete, incomplete failure events, noncontiguous or + time-reversing histories, illegal transitions, terminal-state reuse, and + status mutation. +- The 0019 downgrade is replay-safe and restores exactly one update-only scope + trigger. +- The 0018 rollback refuses non-empty evidence and removes an explicitly emptied + registry. +- Generated database identifiers use `psycopg2.sql.Identifier`, DSN query + parameters survive throwaway-database creation, and every disposable database + connection closes in a fixture `finally` block. + +## Follow-up sequence + +1. Add a transactionally atomic repository/API for snapshot registration, run + creation, required scope, idempotent retry, and status append. +2. Add a normalized transactional Valkey outbox instead of introducing an MQ. +3. Bind TEPP through its reviewed versioned import/REST contract without + cross-service SQL. +4. Bind contextual-orchestrator through a fail-closed versioned adapter after + the canonical API and multimodal message contracts merge. +5. Add administrator and user run surfaces inside the existing React + application, following the DB-grounded Figma information architecture, + Storybook inventory, and design-token contract. +6. Execute private actual-data acceptance and retain only signed, + aggregate-only manifests outside public source control. + +## References — APA 7th + +International Organization for Standardization. (2019). *ISO 8601-1:2019: Date +and time—Representations for information interchange—Part 1: Basic rules* +(confirmed 2024; Amendment 1:2022). +https://www.iso.org/standard/70907.html + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: 5.5. +Constraints*. https://www.postgresql.org/docs/current/ddl-constraints.html + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: 37. +Triggers*. https://www.postgresql.org/docs/current/triggers.html + +World Wide Web Consortium. (2013). *PROV-O: The PROV ontology* (W3C +Recommendation). https://www.w3.org/TR/prov-o/ diff --git a/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md b/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md new file mode 100644 index 000000000..222f1aca2 --- /dev/null +++ b/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md @@ -0,0 +1,49 @@ +# Analysis-run registry research and standards doctoring + +**Capability maturity:** implemented on an active stacked PR; not protected-main +truth until merge. + +## Decision traceability + +| Source | Product decision | +|---|---| +| PostgreSQL 18 constraints and trigger documentation | Use primary/foreign/unique/check constraints for row-local invariants; use serialized trigger functions for cross-row cutoff, immutability, evidence-freeze, and lifecycle rules rather than unsupported cross-table `CHECK` constraints. | +| ISO 8601-1:2019, confirmed 2024 | Store evidence availability, capture, run cutoff, request, occurrence, and record clocks as timezone-aware PostgreSQL timestamps; do not collapse them into one ambiguous date string. | +| W3C PROV-O | Treat the registry as product execution/provenance metadata that may later bind to the standards-complete provenance layer; do not flatten source entities, activities, agents, or qualified relations into a JSON run payload. | +| Accepted TEPP temporal baseline | Keep reusable source-capture clocks separate from run-specific knowledge cutoff and enforce `maximum_available_time <= knowledge_cutoff` without claiming that this aggregate guard replaces TEPP's full multi-clock model. | + +## Current-standard note + +ISO 8601-1:2019 remains the published International Standard and was confirmed +in 2024. ISO/CD 8601-1 edition 2 is under development in 2026 and is tracked as +a draft, not used as the binding production standard. + +PostgreSQL 18 is the current supported documentation line at the time of this +decision. The shipped container remains PostgreSQL 16, so migration syntax and +runtime behavior are intentionally limited to PostgreSQL 16-compatible +features while design guidance is checked against current documentation. + +## Evidence and claim boundary + +The migration records only opaque IDs, digests, bounded code values, aggregate +counts, and timestamps. Actual source rows, SQL, DSNs, private identifiers, +model/provider payloads, and acceptance artifacts remain outside public source +control and outside this registry. Real-data acceptance is not established by +schema tests; it requires a later private execution and signed aggregate-only +manifest. + +## APA 7th references + +International Organization for Standardization. (2019). *ISO 8601-1:2019: Date +and time—Representations for information interchange—Part 1: Basic rules* +(confirmed 2024; Amendment 1:2022). +https://www.iso.org/standard/70907.html + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: 5.5. +Constraints*. https://www.postgresql.org/docs/current/ddl-constraints.html + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: 37. +Triggers*. https://www.postgresql.org/docs/current/triggers.html + +World Wide Web Consortium. (2013). *PROV-O: The PROV ontology* (W3C +Recommendation). https://www.w3.org/TR/prov-o/ diff --git a/docs/superpowers/plans/2026-08-15-analysis-run-registry.md b/docs/superpowers/plans/2026-08-15-analysis-run-registry.md new file mode 100644 index 000000000..d3d82bbf0 --- /dev/null +++ b/docs/superpowers/plans/2026-08-15-analysis-run-registry.md @@ -0,0 +1,104 @@ +# Normalized Analysis-Run Registry Implementation Plan + +> Execute test-first. Preserve the protected LineageWeave product and port only +> bounded evidence from the retained Milestone 2 experiment. + +**Goal:** Add a normalized, fail-closed PostgreSQL registry for reusable source +captures and account-owned analysis runs without copying the parallel product. + +**Architecture:** Sequential migration 0018 adds snapshot, aggregate-count, run, +scope, and append-only status relations plus a derived current-status view. +`maximum_available_time` and `captured_at` belong to the reusable snapshot; +`knowledge_cutoff` belongs to each run. Existing identity, product data, +lineage, report, provenance, TEPP, contextual-orchestrator, and Valkey boundaries +remain unchanged. + +**Tech stack:** PostgreSQL 16-compatible SQL, Python 3.12+, pytest, psycopg2, and +the digest-pinned official PostgreSQL image. + +## Global constraints + +- Start from PR #74 exact head `2ace79ea90a82d61f8467bbe644dd23b0deaa8b6`. +- Keep source data, organization-specific names, source-table identifiers, + base64 payloads, credentials, provider payloads, and raw exceptions out of + public Git and registry rows. +- Use descriptive two-or-more-word `snake_case` database objects and 3NF. +- Do not add a second React app, Keyverse-shaped identity service, TEPP + arithmetic, or cross-service SQL. +- Fail closed on PostgreSQL contract violations and refuse destructive rollback + while audit evidence exists. +- Exact-head hosted PostgreSQL, security, SAST, and review gates become + authoritative after the stack is refreshed onto protected `main`. + +## Task 1 — RED: normalized registry and temporal ownership + +**File:** `tests/test_analysis_run_registry_schema.py` + +- Require the five normalized relations and current-status view. +- Reject the legacy denormalized table and JSON metadata. +- Require reusable snapshot clocks (`maximum_available_time`, `captured_at`) and + run-owned `knowledge_cutoff`. +- Require non-null requesting account and account-scoped idempotency. +- Use quoted generated database identifiers and preserve DSN query options. +- Prove static RED before migration implementation. + +## Task 2 — RED: immutable evidence and lifecycle + +**File:** `tests/test_analysis_run_registry_schema.py` + +- Prove one snapshot supports multiple later run cutoffs. +- Reject source evidence later than the run cutoff. +- Reject snapshot/count mutation and count insert/delete after first run. +- Reject missing actor and same-account idempotency reuse while allowing the + same key for another account. +- Reject incoherent scopes and incomplete failure events. +- Reject non-pending first state, ordinal gaps, direct pending-to-success, + reversed occurrence time, and transitions after terminal state. + +## Task 3 — GREEN: migration and rollback + +**Files:** + +- `migrations/0018_analysis_run_registry.sql` +- `migrations/rollback/0018_analysis_run_registry.sql` +- `docker/postgres-init/Dockerfile` + +Implementation requirements: + +1. Register bounded lookup codes and reject lookup-category collision. +2. Add snapshot, count, run, scope, status, and current-status view. +3. Lock the snapshot on count-set changes and run creation. +4. Enforce `maximum_available_time <= knowledge_cutoff` per run. +5. Freeze snapshot/count/run/scope evidence as appropriate. +6. Serialize status appends and validate the complete history after each insert + statement, including multi-row inserts. +7. Make status updates/deletes fail closed. +8. Make migration replay-safe and rollback refuse non-empty evidence. +9. Apply migration 0018 after PROV-O migration 0017 in fresh containers. + +## Task 4 — Architecture and research truth + +**Files:** + +- `docs/adr/0013-normalized-analysis-run-registry.md` +- `docs/superpowers/specs/2026-08-15-analysis-run-registry-design.md` +- `docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md` +- `CHANGELOG.d/0.79.0-analysis-run-registry.md` + +Document exact ownership, temporal semantics, concurrency locks, lifecycle, +privacy exclusions, rollback, deferred API/UI/service adapters, maturity, and +APA 7 references. Do not present active-PR behavior as protected-main truth. + +## Task 5 — Verification and stack integration + +1. Run static contract locally; do not count skipped PostgreSQL tests as passing. +2. Run all real PostgreSQL cases on hosted CI with a required reachable service. +3. Run full Python, frontend lint/test/build, documentation hygiene, security, + SAST, and public-content scans on the exact refreshed head. +4. Keep the PR Draft while #74 is open. +5. After #74 merges, rebuild or retarget the bounded delta on exact protected + `main`; prior-base checks and reviews do not transfer. +6. Obtain independent current-head approval and merge only through the live + protected policy. +7. Continue with the atomic repository/API/outbox slice; do not duplicate + migration 0018 or ADR 0013. diff --git a/docs/superpowers/specs/2026-08-15-analysis-run-registry-design.md b/docs/superpowers/specs/2026-08-15-analysis-run-registry-design.md new file mode 100644 index 000000000..979d6813d --- /dev/null +++ b/docs/superpowers/specs/2026-08-15-analysis-run-registry-design.md @@ -0,0 +1,118 @@ +# Normalized Analysis-Run Registry Design + +**Status:** Approved implementation design on an active stacked PR +**Parent:** LineageWeave PR #74 exact head `2ace79ea90a82d61f8467bbe644dd23b0deaa8b6` +**Issue:** #79, Milestone 2 additive direct-PostgreSQL port + +## Goal + +Add the smallest durable PostgreSQL contract needed to identify and audit +LineageWeave analysis runs without importing the retained experiment's parallel +application, denormalized run row, raw data, or service-owned computation. + +## Product boundary + +LineageWeave owns product-visible run identity, source-capture identity, +authorization scope, actor-scoped idempotency, and state history. Existing +product relations remain authoritative for posts, lineage, entities, reports, +and provenance. TEPP owns temporal and psychometric estimation. +contextual-orchestrator owns model routing and provider execution. Valkey carries +events but does not become durable run truth. + +## Data design + +The design uses five relations: + +- `analysis_source_snapshot`: immutable capture digest, source-contract revision, + latest evidence-availability time, and capture time; +- `analysis_source_count`: normalized aggregate reconciliation values; +- `analysis_run`: immutable account-owned request, run-specific knowledge cutoff, + and reproducibility digests; +- `analysis_run_scope`: at most one mutually exclusive product scope; the later + creation repository inserts the required scope atomically; +- `analysis_run_status_event`: append-only, ordered lifecycle history. + +The `analysis_run_current_status` view returns the highest ordinal event for each +run. No JSON payload is persisted. Optional model and prompt hashes remain null +for deterministic runs. + +## Temporal and concurrency contract + +A source capture is reusable across analysis occasions, so `knowledge_cutoff` +belongs to the run rather than the snapshot. PostgreSQL locks the snapshot on +run creation and rejects: + +```text +maximum_available_time > knowledge_cutoff +``` + +Snapshot and count rows reject updates. Count insert/delete and run creation +lock the same snapshot row, closing the race between changing aggregate evidence +and starting the first derivation. Once a run references the snapshot, the count +set is frozen. + +## Identity and idempotency + +Every run references a real OIDC-backed `user_account`. Idempotency is unique on +`(requested_by_account_id, idempotency_key)`, allowing different authenticated +actors to use the same opaque key while preventing one actor from creating two +requests with it. + +## Lifecycle contract + +Status history is a serialized state machine: + +```text +pending -> running -> succeeded | failed | cancelled +pending -> failed | cancelled +``` + +The first status is pending at ordinal 1; ordinals are contiguous; occurrence +time is nondecreasing; terminal states cannot transition; failure metadata is +bounded and code-consistent. `recorded_at` preserves the database system clock +separately. Updates and deletes are rejected. + +## Security and privacy + +The public schema stores no raw record, source-table name, source identifier, +image, credential, provider payload, raw exception, or organization-specific +fixture. Scope references existing corporate/process-unit identities or a +bounded thread key; it does not trust token claims as database truth. Future API +access reuses current RBAC/ABAC until a separate accepted RLS decision exists. + +## Failure behavior + +- malformed digests, unsupported enum codes, negative counts, missing request + actors, same-actor duplicate idempotency keys, and incoherent scopes fail in + PostgreSQL; +- evidence unavailable at a run cutoff fails before run creation; +- immutable snapshot/count/run/scope changes fail closed; +- status histories with illegal starts, gaps, reversed time, invalid + transitions, or post-terminal appends fail closed; +- lookup-code category collisions abort migration; +- rollback refuses any non-empty registry relation. + +## Deployment + +The product PostgreSQL image applies migration 0018 after migration 0017. No new +service or container is introduced. A later API/outbox slice may use these +relations but cannot claim implementation from this schema alone. + +## Testing + +Static tests run without external services and lock file names, table names, +clock ownership, actor/idempotency scope, lookup inventory, Docker migration +order, temporary-artifact absence, and fail-closed rollback markers. Real +PostgreSQL tests use the committed migration files, generated quoted database +identifiers, preserved DSN query options, and actual database exceptions. They +exercise multiple run cutoffs over one snapshot, leakage rejection, immutable +and concurrency-frozen evidence, account-scoped idempotency, scope shape, +ordered lifecycle transitions, migration replay, and rollback. Hosted exact-head +CI remains authoritative for the PostgreSQL lane. + +## Deferred work + +API repository methods, Valkey outbox persistence, TEPP submission, +contextual-orchestrator task envelopes, React/Storybook surfaces, RLS, signed +acceptance manifests, artifact registries, and retention tooling are separate +reviewable slices. diff --git a/migrations/0018_analysis_run_registry.sql b/migrations/0018_analysis_run_registry.sql new file mode 100644 index 000000000..bd4658c55 --- /dev/null +++ b/migrations/0018_analysis_run_registry.sql @@ -0,0 +1,566 @@ +-- Milestone 2 additive runtime bridge: normalized analysis-run registry. +-- +-- The retained direct-PostgreSQL experiment stored repeated counts and an +-- unconstrained metadata JSON object in one run row. This migration preserves +-- the useful run/snapshot evidence without copying that parallel product +-- schema. Source content, source-table names, credentials, provider payloads, +-- raw exceptions, and organization-specific identifiers are not stored here. +-- +-- Database objects remain descriptive two-or-more-word snake_case and every +-- multi-valued fact is represented by a separate relation. + +begin; + +insert into common_lookup_value + (lookup_category, lookup_code, lookup_label, display_order) +values + ('analysis_run_kind', 'analysis_run_lineage', 'Lineage reconstruction', 0), + ('analysis_run_kind', 'analysis_run_report', 'Period report', 1), + ('analysis_run_kind', 'analysis_run_tepp', 'TEPP measurement', 2), + ('analysis_run_status', 'analysis_status_pending', 'Pending', 0), + ('analysis_run_status', 'analysis_status_running', 'Running', 1), + ('analysis_run_status', 'analysis_status_succeeded', 'Succeeded', 2), + ('analysis_run_status', 'analysis_status_failed', 'Failed', 3), + ('analysis_run_status', 'analysis_status_cancelled', 'Cancelled', 4), + ('analysis_run_scope', 'analysis_scope_all_visible', 'All authorized records', 0), + ('analysis_run_scope', 'analysis_scope_corporate_entity', 'Corporate entity', 1), + ('analysis_run_scope', 'analysis_scope_process_unit', 'Process unit', 2), + ('analysis_run_scope', 'analysis_scope_thread_group', 'Thread group', 3), + ('analysis_source_count', 'analysis_count_source_row', 'Source rows', 0), + ('analysis_source_count', 'analysis_count_document', 'Documents', 1), + ('analysis_source_count', 'analysis_count_thread', 'Threads', 2), + ('analysis_source_count', 'analysis_count_lineage_node', 'Lineage nodes', 3), + ('analysis_source_count', 'analysis_count_lineage_edge', 'Lineage edges', 4) +on conflict (lookup_code) do nothing; + +-- A globally unique lookup_code is already the repository-wide contract. A +-- pre-existing code under another category is a schema conflict, not a reason +-- to silently accept the wrong vocabulary. +do $$ +declare + mismatch_count integer; +begin + select count(*) + into mismatch_count + from common_lookup_value as actual + join (values + ('analysis_run_lineage', 'analysis_run_kind'), + ('analysis_run_report', 'analysis_run_kind'), + ('analysis_run_tepp', 'analysis_run_kind'), + ('analysis_status_pending', 'analysis_run_status'), + ('analysis_status_running', 'analysis_run_status'), + ('analysis_status_succeeded', 'analysis_run_status'), + ('analysis_status_failed', 'analysis_run_status'), + ('analysis_status_cancelled', 'analysis_run_status'), + ('analysis_scope_all_visible', 'analysis_run_scope'), + ('analysis_scope_corporate_entity', 'analysis_run_scope'), + ('analysis_scope_process_unit', 'analysis_run_scope'), + ('analysis_scope_thread_group', 'analysis_run_scope'), + ('analysis_count_source_row', 'analysis_source_count'), + ('analysis_count_document', 'analysis_source_count'), + ('analysis_count_thread', 'analysis_source_count'), + ('analysis_count_lineage_node', 'analysis_source_count'), + ('analysis_count_lineage_edge', 'analysis_source_count') + ) as expected(lookup_code, lookup_category) + on expected.lookup_code = actual.lookup_code + where actual.lookup_category <> expected.lookup_category; + + if mismatch_count <> 0 then + raise exception 'analysis_run_registry_lookup_conflict'; + end if; +end +$$; + +create table if not exists analysis_source_snapshot ( + analysis_source_snapshot_id uuid primary key default uuid_generate_v4(), + snapshot_sha256 text not null, + source_contract_version text not null, + maximum_available_time timestamptz not null, + captured_at timestamptz not null, + created_at timestamptz not null default now(), + constraint analysis_source_snapshot_digest_check + check (snapshot_sha256 ~ '^[0-9a-f]{64}$'), + constraint analysis_source_snapshot_contract_check + check (length(btrim(source_contract_version)) between 1 and 128), + constraint analysis_source_snapshot_capture_check + check (maximum_available_time <= captured_at), + constraint analysis_source_snapshot_identity_unique + unique (snapshot_sha256, source_contract_version) +); + +comment on table analysis_source_snapshot is + 'Immutable source-capture identity and latest evidence-availability time. ' + 'Analysis-specific knowledge cutoffs belong to analysis_run.'; + +create or replace function reject_analysis_source_snapshot_update() +returns trigger +language plpgsql +as $$ +begin + raise exception 'analysis_source_snapshot_is_immutable'; +end +$$; + +comment on function reject_analysis_source_snapshot_update() is + 'Rejects mutation of immutable source-capture identity and clocks.'; + +drop trigger if exists analysis_source_snapshot_update_reject + on analysis_source_snapshot; +create trigger analysis_source_snapshot_update_reject +before update on analysis_source_snapshot +for each row execute function reject_analysis_source_snapshot_update(); + +create table if not exists analysis_source_count ( + analysis_source_snapshot_id uuid not null + references analysis_source_snapshot (analysis_source_snapshot_id) + on delete cascade, + count_type_code text not null + references common_lookup_value (lookup_code), + count_value bigint not null, + primary key (analysis_source_snapshot_id, count_type_code), + constraint analysis_source_count_type_check + check (count_type_code in ( + 'analysis_count_source_row', + 'analysis_count_document', + 'analysis_count_thread', + 'analysis_count_lineage_node', + 'analysis_count_lineage_edge' + )), + constraint analysis_source_count_nonnegative_check + check (count_value >= 0) +); + +comment on table analysis_source_count is + 'One immutable aggregate count per source snapshot and count vocabulary.'; + +create or replace function reject_analysis_source_count_update() +returns trigger +language plpgsql +as $$ +begin + raise exception 'analysis_source_count_is_immutable'; +end +$$; + +comment on function reject_analysis_source_count_update() is + 'Rejects rewriting a persisted source-snapshot count.'; + +drop trigger if exists analysis_source_count_update_reject + on analysis_source_count; +create trigger analysis_source_count_update_reject +before update on analysis_source_count +for each row execute function reject_analysis_source_count_update(); + +create table if not exists analysis_run ( + analysis_run_id uuid primary key default uuid_generate_v4(), + analysis_source_snapshot_id uuid not null + references analysis_source_snapshot (analysis_source_snapshot_id), + run_kind_code text not null + references common_lookup_value (lookup_code), + idempotency_key text not null, + requested_by_account_id uuid not null + references user_account (user_account_id), + knowledge_cutoff timestamptz not null, + configuration_schema_version text not null, + configuration_sha256 text not null, + model_contract_sha256 text, + prompt_bundle_sha256 text, + code_revision_sha text not null, + requested_at timestamptz not null default now(), + constraint analysis_run_kind_check + check (run_kind_code in ( + 'analysis_run_lineage', + 'analysis_run_report', + 'analysis_run_tepp' + )), + constraint analysis_run_idempotency_key_check + check (length(btrim(idempotency_key)) between 1 and 256), + constraint analysis_run_configuration_version_check + check (length(btrim(configuration_schema_version)) between 1 and 128), + constraint analysis_run_configuration_digest_check + check (configuration_sha256 ~ '^[0-9a-f]{64}$'), + constraint analysis_run_model_digest_check + check ( + model_contract_sha256 is null + or model_contract_sha256 ~ '^[0-9a-f]{64}$' + ), + constraint analysis_run_prompt_digest_check + check ( + prompt_bundle_sha256 is null + or prompt_bundle_sha256 ~ '^[0-9a-f]{64}$' + ), + constraint analysis_run_code_revision_check + check (code_revision_sha ~ '^(?:[0-9a-f]{40}|[0-9a-f]{64})$'), + constraint analysis_run_requester_idempotency_unique + unique (requested_by_account_id, idempotency_key) +); + +create index if not exists analysis_run_snapshot_idx + on analysis_run (analysis_source_snapshot_id); +create index if not exists analysis_run_kind_requested_idx + on analysis_run (run_kind_code, requested_at desc); +create index if not exists analysis_run_requester_idx + on analysis_run (requested_by_account_id, requested_at desc); + +comment on table analysis_run is + 'Immutable, account-scoped analysis request bound to one source snapshot, ' + 'one knowledge cutoff, and reproducibility digests.'; + +create or replace function enforce_analysis_run_knowledge_cutoff() +returns trigger +language plpgsql +as $$ +declare + snapshot_available_time timestamptz; +begin + select maximum_available_time + into snapshot_available_time + from analysis_source_snapshot + where analysis_source_snapshot_id = new.analysis_source_snapshot_id + for update; + + if not found then + raise exception 'analysis_source_snapshot_not_found'; + end if; + + if snapshot_available_time > new.knowledge_cutoff then + raise exception 'analysis_run_future_information_leakage'; + end if; + + return new; +end +$$; + +comment on function enforce_analysis_run_knowledge_cutoff() is + 'Serializes run creation with count-set changes and rejects source evidence ' + 'that was unavailable at the run knowledge cutoff.'; + +drop trigger if exists analysis_run_knowledge_cutoff_guard + on analysis_run; +create trigger analysis_run_knowledge_cutoff_guard +before insert on analysis_run +for each row execute function enforce_analysis_run_knowledge_cutoff(); + +create or replace function reject_analysis_run_mutation() +returns trigger +language plpgsql +as $$ +begin + raise exception 'analysis_run_request_is_immutable'; +end +$$; + +comment on function reject_analysis_run_mutation() is + 'Rejects update/delete of a registered analysis request and its digests.'; + +drop trigger if exists analysis_run_mutation_reject + on analysis_run; +create trigger analysis_run_mutation_reject +before update or delete on analysis_run +for each row execute function reject_analysis_run_mutation(); + +create or replace function enforce_analysis_source_count_freeze() +returns trigger +language plpgsql +as $$ +declare + affected_snapshot_id uuid; +begin + if tg_op = 'DELETE' then + affected_snapshot_id := old.analysis_source_snapshot_id; + else + affected_snapshot_id := new.analysis_source_snapshot_id; + end if; + + -- Both count-set changes and run creation lock the same parent row. This + -- closes the race in which a run could start while another transaction was + -- still extending or deleting the aggregate evidence set. + perform 1 + from analysis_source_snapshot + where analysis_source_snapshot_id = affected_snapshot_id + for update; + + if exists ( + select 1 + from analysis_run + where analysis_source_snapshot_id = affected_snapshot_id + ) then + raise exception 'analysis_source_count_frozen_after_run'; + end if; + + if tg_op = 'DELETE' then + return old; + end if; + return new; +end +$$; + +comment on function enforce_analysis_source_count_freeze() is + 'Allows count insert/delete only before the first run references a snapshot.'; + +drop trigger if exists analysis_source_count_freeze_guard + on analysis_source_count; +create trigger analysis_source_count_freeze_guard +before insert or delete on analysis_source_count +for each row execute function enforce_analysis_source_count_freeze(); + +create table if not exists analysis_run_scope ( + analysis_run_id uuid primary key + references analysis_run (analysis_run_id) + on delete cascade, + scope_kind_code text not null + references common_lookup_value (lookup_code), + corporate_entity_id uuid + references corporate_entity (corporate_entity_id), + process_unit_id uuid + references process_unit (process_unit_id), + scope_key text, + constraint analysis_run_scope_shape_check + check ( + (scope_kind_code = 'analysis_scope_all_visible' + and corporate_entity_id is null + and process_unit_id is null + and scope_key is null) + or + (scope_kind_code = 'analysis_scope_corporate_entity' + and corporate_entity_id is not null + and process_unit_id is null + and scope_key is null) + or + (scope_kind_code = 'analysis_scope_process_unit' + and corporate_entity_id is null + and process_unit_id is not null + and scope_key is null) + or + (scope_kind_code = 'analysis_scope_thread_group' + and corporate_entity_id is null + and process_unit_id is null + and scope_key is not null + and length(btrim(scope_key)) between 1 and 256) + ) +); + +create index if not exists analysis_run_scope_entity_idx + on analysis_run_scope (corporate_entity_id) + where corporate_entity_id is not null; +create index if not exists analysis_run_scope_unit_idx + on analysis_run_scope (process_unit_id) + where process_unit_id is not null; + +comment on table analysis_run_scope is + 'At most one authorization-relevant product scope for a run; process-unit ' + 'ownership remains derivable from process_unit.'; + +create or replace function reject_analysis_run_scope_update() +returns trigger +language plpgsql +as $$ +begin + raise exception 'analysis_run_scope_is_immutable'; +end +$$; + +comment on function reject_analysis_run_scope_update() is + 'Rejects mutation of the authorization scope attached to a run.'; + +drop trigger if exists analysis_run_scope_update_reject + on analysis_run_scope; +create trigger analysis_run_scope_update_reject +before update on analysis_run_scope +for each row execute function reject_analysis_run_scope_update(); + +create table if not exists analysis_run_status_event ( + analysis_run_id uuid not null + references analysis_run (analysis_run_id) + on delete cascade, + status_ordinal integer not null, + status_code text not null + references common_lookup_value (lookup_code), + occurred_at timestamptz not null, + recorded_at timestamptz not null default clock_timestamp(), + failure_code text, + retryable boolean not null default false, + primary key (analysis_run_id, status_ordinal), + constraint analysis_run_status_code_check + check (status_code in ( + 'analysis_status_pending', + 'analysis_status_running', + 'analysis_status_succeeded', + 'analysis_status_failed', + 'analysis_status_cancelled' + )), + constraint analysis_run_status_ordinal_check + check (status_ordinal >= 1), + constraint analysis_run_status_recorded_check + check (occurred_at <= recorded_at), + constraint analysis_run_status_failure_shape_check + check ( + (status_code = 'analysis_status_failed' + and failure_code is not null + and length(btrim(failure_code)) between 1 and 128) + or + (status_code <> 'analysis_status_failed' + and failure_code is null + and retryable = false) + ) +); + +create index if not exists analysis_run_status_current_idx + on analysis_run_status_event (analysis_run_id, status_ordinal desc); + +comment on table analysis_run_status_event is + 'Append-only, contiguous state-machine evidence with separate occurrence ' + 'and database record clocks.'; + +create or replace function lock_analysis_run_status_append() +returns trigger +language plpgsql +as $$ +begin + -- Serialize all status appends for one run across concurrent transactions. + perform 1 + from analysis_run + where analysis_run_id = new.analysis_run_id + for update; + + if not found then + raise exception 'analysis_run_not_found'; + end if; + + return new; +end +$$; + +comment on function lock_analysis_run_status_append() is + 'Locks one analysis run before status rows are appended.'; + +drop trigger if exists analysis_run_status_append_lock + on analysis_run_status_event; +create trigger analysis_run_status_append_lock +before insert on analysis_run_status_event +for each row execute function lock_analysis_run_status_append(); + +create or replace function enforce_analysis_run_status_transition() +returns trigger +language plpgsql +as $$ +declare + invalid_history_exists boolean; +begin + -- AFTER-row triggers execute after the statement, so a multi-row INSERT is + -- validated as one complete history rather than depending on VALUES order. + select exists ( + select 1 + from ( + select status_ordinal, + status_code, + occurred_at, + lag(status_ordinal) over ( + order by status_ordinal + ) as previous_status_ordinal, + lag(status_code) over ( + order by status_ordinal + ) as previous_status_code, + lag(occurred_at) over ( + order by status_ordinal + ) as previous_occurred_at + from analysis_run_status_event + where analysis_run_id = new.analysis_run_id + ) as history + where ( + previous_status_ordinal is null + and ( + status_ordinal <> 1 + or status_code <> 'analysis_status_pending' + ) + ) + or ( + previous_status_ordinal is not null + and status_ordinal <> previous_status_ordinal + 1 + ) + or ( + previous_occurred_at is not null + and occurred_at < previous_occurred_at + ) + or ( + previous_status_code is not null + and not ( + ( + previous_status_code = 'analysis_status_pending' + and status_code in ( + 'analysis_status_running', + 'analysis_status_failed', + 'analysis_status_cancelled' + ) + ) + or ( + previous_status_code = 'analysis_status_running' + and status_code in ( + 'analysis_status_succeeded', + 'analysis_status_failed', + 'analysis_status_cancelled' + ) + ) + ) + ) + ) into invalid_history_exists; + + if invalid_history_exists then + raise exception 'analysis_run_status_history_invalid'; + end if; + + return new; +end +$$; + +comment on function enforce_analysis_run_status_transition() is + 'Enforces initial pending state, contiguous ordinals, monotonic occurrence ' + 'time, legal transitions, and terminal-state finality.'; + +drop trigger if exists analysis_run_status_transition_guard + on analysis_run_status_event; +create trigger analysis_run_status_transition_guard +after insert on analysis_run_status_event +for each row execute function enforce_analysis_run_status_transition(); + +create or replace function reject_analysis_run_status_mutation() +returns trigger +language plpgsql +as $$ +begin + raise exception 'analysis_run_status_event_is_append_only'; +end +$$; + +comment on function reject_analysis_run_status_mutation() is + 'Rejects update/delete of analysis_run_status_event so history stays append-only.'; + +drop trigger if exists analysis_run_status_event_update_reject + on analysis_run_status_event; +create trigger analysis_run_status_event_update_reject +before update on analysis_run_status_event +for each row execute function reject_analysis_run_status_mutation(); + +drop trigger if exists analysis_run_status_event_delete_reject + on analysis_run_status_event; +create trigger analysis_run_status_event_delete_reject +before delete on analysis_run_status_event +for each row execute function reject_analysis_run_status_mutation(); + +create or replace view analysis_run_current_status as +select distinct on (status_event.analysis_run_id) + status_event.analysis_run_id, + status_event.status_code, + status_event.status_ordinal, + status_event.occurred_at, + status_event.recorded_at, + status_event.failure_code, + status_event.retryable + from analysis_run_status_event as status_event + order by status_event.analysis_run_id, + status_event.status_ordinal desc; + +comment on view analysis_run_current_status is + 'Latest append-only status event for each run; not a second state authority.'; + +commit; diff --git a/migrations/0019_analysis_run_scope_immutability.sql b/migrations/0019_analysis_run_scope_immutability.sql new file mode 100644 index 000000000..0a61c85fb --- /dev/null +++ b/migrations/0019_analysis_run_scope_immutability.sql @@ -0,0 +1,31 @@ +-- Make the authorization scope attached to an analysis run fully immutable. +-- Migration 0018 already rejected scope updates. Deletion would still remove +-- the only persisted authorization boundary while leaving the run and status +-- history intact, so both mutation forms must fail closed. + +begin; + +drop trigger if exists analysis_run_scope_update_reject + on analysis_run_scope; +drop trigger if exists analysis_run_scope_mutation_reject + on analysis_run_scope; + +drop function if exists reject_analysis_run_scope_update(); + +create or replace function reject_analysis_run_scope_mutation() +returns trigger +language plpgsql +as $$ +begin + raise exception 'analysis_run_scope_is_immutable'; +end +$$; + +comment on function reject_analysis_run_scope_mutation() is + 'Rejects update/delete of an analysis run authorization scope.'; + +create trigger analysis_run_scope_mutation_reject +before update or delete on analysis_run_scope +for each row execute function reject_analysis_run_scope_mutation(); + +commit; diff --git a/migrations/rollback/0018_analysis_run_registry.sql b/migrations/rollback/0018_analysis_run_registry.sql new file mode 100644 index 000000000..9422c0f95 --- /dev/null +++ b/migrations/rollback/0018_analysis_run_registry.sql @@ -0,0 +1,71 @@ +-- Fail-closed rollback for migration 0018. +-- +-- Registry evidence must be explicitly exported or deleted under an approved +-- retention procedure before the schema can be removed. The rollback is +-- idempotent only when every registry table is absent or empty. + +begin; + +do $$ +declare + relation_name text; + relation_has_rows boolean; +begin + foreach relation_name in array array[ + 'analysis_run_status_event', + 'analysis_run_scope', + 'analysis_run', + 'analysis_source_count', + 'analysis_source_snapshot' + ] loop + if to_regclass('public.' || relation_name) is not null then + execute format('select exists (select 1 from %I)', relation_name) + into relation_has_rows; + if relation_has_rows then + raise exception 'analysis_run_registry_not_empty'; + end if; + end if; + end loop; +end +$$; + +drop view if exists analysis_run_current_status; + +drop table if exists analysis_run_status_event; +drop table if exists analysis_run_scope; +drop table if exists analysis_run; +drop table if exists analysis_source_count; +drop table if exists analysis_source_snapshot; + +drop function if exists reject_analysis_run_status_mutation(); +drop function if exists enforce_analysis_run_status_transition(); +drop function if exists lock_analysis_run_status_append(); +drop function if exists reject_analysis_run_scope_update(); +drop function if exists enforce_analysis_source_count_freeze(); +drop function if exists reject_analysis_run_mutation(); +drop function if exists enforce_analysis_run_knowledge_cutoff(); +drop function if exists reject_analysis_source_count_update(); +drop function if exists reject_analysis_source_snapshot_update(); + +delete from common_lookup_value + where lookup_code in ( + 'analysis_run_lineage', + 'analysis_run_report', + 'analysis_run_tepp', + 'analysis_status_pending', + 'analysis_status_running', + 'analysis_status_succeeded', + 'analysis_status_failed', + 'analysis_status_cancelled', + 'analysis_scope_all_visible', + 'analysis_scope_corporate_entity', + 'analysis_scope_process_unit', + 'analysis_scope_thread_group', + 'analysis_count_source_row', + 'analysis_count_document', + 'analysis_count_thread', + 'analysis_count_lineage_node', + 'analysis_count_lineage_edge' + ); + +commit; diff --git a/migrations/rollback/0019_analysis_run_scope_immutability.sql b/migrations/rollback/0019_analysis_run_scope_immutability.sql new file mode 100644 index 000000000..a81692889 --- /dev/null +++ b/migrations/rollback/0019_analysis_run_scope_immutability.sql @@ -0,0 +1,28 @@ +-- Restore migration 0018's update-only scope guard. +-- This rollback changes the mutation policy but does not remove registry data. + +begin; + +drop trigger if exists analysis_run_scope_mutation_reject + on analysis_run_scope; +drop trigger if exists analysis_run_scope_update_reject + on analysis_run_scope; +drop function if exists reject_analysis_run_scope_mutation(); + +create or replace function reject_analysis_run_scope_update() +returns trigger +language plpgsql +as $$ +begin + raise exception 'analysis_run_scope_is_immutable'; +end +$$; + +comment on function reject_analysis_run_scope_update() is + 'Rejects mutation of the authorization scope attached to a run.'; + +create trigger analysis_run_scope_update_reject +before update on analysis_run_scope +for each row execute function reject_analysis_run_scope_update(); + +commit; diff --git a/tests/test_analysis_run_registry_request_immutability.py b/tests/test_analysis_run_registry_request_immutability.py new file mode 100644 index 000000000..946ba1b5c --- /dev/null +++ b/tests/test_analysis_run_registry_request_immutability.py @@ -0,0 +1,154 @@ +"""Request immutability and transient-artifact contracts for analysis runs.""" + +from __future__ import annotations + +import os +import uuid +from pathlib import Path +from urllib.parse import urlsplit, urlunsplit + +import pytest + +try: + import psycopg2 + import psycopg2.errors + from psycopg2 import sql +except ModuleNotFoundError: # pragma: no cover - local static-only environments + psycopg2 = None # type: ignore[assignment] + sql = None # type: ignore[assignment] + +_ROOT = Path(__file__).resolve().parents[1] +_INITIAL_MIGRATION = _ROOT / "migrations" / "0001_initial_schema.sql" +_REGISTRY_MIGRATION = _ROOT / "migrations" / "0018_analysis_run_registry.sql" +_ADMIN_DSN = os.environ.get( + "LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN", "postgresql://localhost/postgres" +) +_TRANSIENT_PATHS = ( + _ROOT / ".github" / "workflows" / "pr83-analysis-run-registry-repair.yml", + _ROOT / ".github" / "workflows" / "pr83-analysis-run-registry-repair-v2.yml", + _ROOT / ".github" / "workflows" / "pr83-analysis-run-registry-repair-v3.yml", + _ROOT / "scripts" / "pr83_analysis_run_registry_repair.py", +) + + +def _postgres_available() -> bool: + """Return whether the configured PostgreSQL administrator is reachable.""" + + if psycopg2 is None: + return False + try: + connection = psycopg2.connect(_ADMIN_DSN, connect_timeout=2) + connection.close() + return True + except psycopg2.OperationalError: + return False + + +def _database_dsn(database_name: str) -> str: + """Return the configured DSN with only its database path replaced.""" + + parsed = urlsplit(_ADMIN_DSN) + return urlunsplit(parsed._replace(path=f"/{database_name}")) + + +@pytest.fixture +def immutable_run_db(): + """Yield a throwaway database migrated through the registry schema.""" + + if not _postgres_available(): + pytest.skip("a reachable PostgreSQL administrator DSN is required") + assert psycopg2 is not None + assert sql is not None + database_name = f"lineageweave_run_{uuid.uuid4().hex[:12]}" + admin_connection = psycopg2.connect(_ADMIN_DSN) + admin_connection.autocommit = True + with admin_connection.cursor() as cursor: + cursor.execute( + sql.SQL("create database {}").format(sql.Identifier(database_name)) + ) + try: + connection = psycopg2.connect(_database_dsn(database_name)) + try: + connection.autocommit = True + with connection.cursor() as cursor: + cursor.execute(_INITIAL_MIGRATION.read_text(encoding="utf-8")) + cursor.execute(_REGISTRY_MIGRATION.read_text(encoding="utf-8")) + yield connection + finally: + connection.close() + finally: + with admin_connection.cursor() as cursor: + cursor.execute( + sql.SQL("drop database {}").format(sql.Identifier(database_name)) + ) + admin_connection.close() + + +def _insert_run(cursor) -> str: + """Insert one account, source snapshot, and analysis request.""" + + suffix = uuid.uuid4().hex + cursor.execute( + """ + insert into user_account + (external_subject_id, display_name, email_address) + values (%s, 'Registry Operator', %s) + returning user_account_id + """, + (f"registry-{suffix}", f"registry-{suffix}@example.test"), + ) + account_id = cursor.fetchone()[0] + cursor.execute( + """ + insert into analysis_source_snapshot + (snapshot_sha256, source_contract_version, + maximum_available_time, captured_at) + values (%s, 'source-contract-v1', + '2026-08-15T00:00:00Z', '2026-08-15T00:05:00Z') + returning analysis_source_snapshot_id + """, + ("a" * 64,), + ) + snapshot_id = cursor.fetchone()[0] + cursor.execute( + """ + insert into analysis_run + (analysis_source_snapshot_id, run_kind_code, idempotency_key, + requested_by_account_id, knowledge_cutoff, + configuration_schema_version, configuration_sha256, + code_revision_sha) + values (%s, 'analysis_run_lineage', 'immutable-request', %s, + '2026-08-15T00:30:00Z', 'lineage-run-v1', %s, %s) + returning analysis_run_id + """, + (snapshot_id, account_id, "b" * 64, "c" * 40), + ) + return str(cursor.fetchone()[0]) + + +def test_transient_repair_artifacts_are_absent_from_the_product_diff() -> None: + """The final product change contains no one-shot mutation machinery.""" + + assert all(not path.exists() for path in _TRANSIENT_PATHS) + + +def test_analysis_run_request_rejects_update_and_delete(immutable_run_db) -> None: + """A registered request remains stable for its full provenance lifetime.""" + + assert psycopg2 is not None + with immutable_run_db.cursor() as cursor: + run_id = _insert_run(cursor) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + """ + update analysis_run + set knowledge_cutoff = '2026-08-16T00:00:00Z' + where analysis_run_id = %s + """, + (run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "delete from analysis_run where analysis_run_id = %s", + (run_id,), + ) diff --git a/tests/test_analysis_run_registry_schema.py b/tests/test_analysis_run_registry_schema.py new file mode 100644 index 000000000..4a445e2b0 --- /dev/null +++ b/tests/test_analysis_run_registry_schema.py @@ -0,0 +1,665 @@ +"""Contracts for the normalized Milestone 2 analysis-run registry.""" + +from __future__ import annotations + +import os +import re +import uuid +from pathlib import Path +from urllib.parse import urlsplit, urlunsplit + +import pytest + +try: + import psycopg2 + import psycopg2.errors + from psycopg2 import sql +except ModuleNotFoundError: # pragma: no cover - local static-only environments + psycopg2 = None # type: ignore[assignment] + sql = None # type: ignore[assignment] + +_ROOT = Path(__file__).resolve().parents[1] +_INITIAL_MIGRATION = _ROOT / "migrations" / "0001_initial_schema.sql" +_REGISTRY_MIGRATION = _ROOT / "migrations" / "0018_analysis_run_registry.sql" +_REGISTRY_ROLLBACK = _ROOT / "migrations" / "rollback" / "0018_analysis_run_registry.sql" +_POSTGRES_IMAGE = _ROOT / "docker" / "postgres-init" / "Dockerfile" +_REPAIR_WORKFLOW = ( + _ROOT / ".github" / "workflows" / "pr83-analysis-run-registry-repair.yml" +) +_ADMIN_DSN = os.environ.get( + "LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN", "postgresql://localhost/postgres" +) +_REQUIRED_TABLES = { + "analysis_source_snapshot", + "analysis_source_count", + "analysis_run", + "analysis_run_scope", + "analysis_run_status_event", +} +_REQUIRED_LOOKUP_CODES = { + "analysis_run_lineage", + "analysis_run_report", + "analysis_run_tepp", + "analysis_status_pending", + "analysis_status_running", + "analysis_status_succeeded", + "analysis_status_failed", + "analysis_status_cancelled", + "analysis_scope_all_visible", + "analysis_scope_corporate_entity", + "analysis_scope_process_unit", + "analysis_scope_thread_group", + "analysis_count_source_row", + "analysis_count_document", + "analysis_count_thread", + "analysis_count_lineage_node", + "analysis_count_lineage_edge", +} + + +def _postgres_available() -> bool: + """Return whether the configured PostgreSQL administrator DSN is reachable.""" + + if psycopg2 is None: + return False + try: + connection = psycopg2.connect(_ADMIN_DSN, connect_timeout=2) + connection.close() + return True + except psycopg2.OperationalError: + return False + + +def _database_dsn(database_name: str) -> str: + """Replace only the database path while preserving DSN query options.""" + + parsed = urlsplit(_ADMIN_DSN) + return urlunsplit(parsed._replace(path=f"/{database_name}")) + + +def _table_definition(migration: str, table_name: str) -> str: + """Return one table definition from the deterministic migration text.""" + + match = re.search( + rf"create table if not exists {re.escape(table_name)}\s*\((.*?)\n\);", + migration, + re.I | re.S, + ) + assert match is not None, table_name + return match.group(1) + + +@pytest.fixture +def registry_db(): + """Yield a throwaway database migrated through the analysis registry.""" + + if not _postgres_available(): + pytest.skip("a reachable PostgreSQL administrator DSN is required") + assert psycopg2 is not None + assert sql is not None + database_name = f"lineageweave_analysis_{uuid.uuid4().hex[:12]}" + admin_connection = psycopg2.connect(_ADMIN_DSN) + admin_connection.autocommit = True + with admin_connection.cursor() as cursor: + cursor.execute( + sql.SQL("create database {}").format(sql.Identifier(database_name)) + ) + try: + connection = psycopg2.connect(_database_dsn(database_name)) + try: + connection.autocommit = True + with connection.cursor() as cursor: + cursor.execute(_INITIAL_MIGRATION.read_text(encoding="utf-8")) + cursor.execute(_REGISTRY_MIGRATION.read_text(encoding="utf-8")) + yield connection + finally: + connection.close() + finally: + with admin_connection.cursor() as cursor: + cursor.execute( + sql.SQL("drop database {}").format(sql.Identifier(database_name)) + ) + admin_connection.close() + + +def _insert_account(cursor, label: str = "operator") -> str: + """Insert one synthetic real-account identity and return its UUID.""" + + suffix = uuid.uuid4().hex + cursor.execute( + """ + insert into user_account + (external_subject_id, display_name, email_address) + values (%s, %s, %s) + returning user_account_id + """, + (f"{label}-{suffix}", f"{label.title()} User", f"{label}-{suffix}@example.test"), + ) + return str(cursor.fetchone()[0]) + + +def _insert_snapshot( + cursor, + *, + digest: str = "a" * 64, + maximum_available_time: str = "2026-08-15T00:00:00Z", + captured_at: str = "2026-08-15T01:00:00Z", +) -> str: + """Insert one synthetic immutable snapshot and return its identifier.""" + + cursor.execute( + """ + insert into analysis_source_snapshot + (snapshot_sha256, source_contract_version, + maximum_available_time, captured_at) + values (%s, %s, %s, %s) + returning analysis_source_snapshot_id + """, + (digest, "source-contract-v1", maximum_available_time, captured_at), + ) + return str(cursor.fetchone()[0]) + + +def _insert_run( + cursor, + *, + snapshot_id: str, + account_id: str, + idempotency_key: str, + knowledge_cutoff: str = "2026-08-15T00:30:00Z", + run_kind_code: str = "analysis_run_lineage", +) -> str: + """Insert one synthetic account-scoped analysis request.""" + + cursor.execute( + """ + insert into analysis_run + (analysis_source_snapshot_id, run_kind_code, idempotency_key, + requested_by_account_id, knowledge_cutoff, + configuration_schema_version, configuration_sha256, + code_revision_sha) + values (%s, %s, %s, %s, %s, 'lineage-run-v1', %s, %s) + returning analysis_run_id + """, + ( + snapshot_id, + run_kind_code, + idempotency_key, + account_id, + knowledge_cutoff, + "b" * 64, + "c" * 40, + ), + ) + return str(cursor.fetchone()[0]) + + +def test_registry_contract_files_are_present_and_normalized() -> None: + """The additive bridge fixes temporal facts at their functional owners.""" + + migration = _REGISTRY_MIGRATION.read_text(encoding="utf-8") + rollback = _REGISTRY_ROLLBACK.read_text(encoding="utf-8") + dockerfile = _POSTGRES_IMAGE.read_text(encoding="utf-8") + + created_tables = set( + re.findall(r"create table if not exists\s+([a-z0-9_]+)", migration, re.I) + ) + assert _REQUIRED_TABLES <= created_tables + assert "analysis_run_records" not in created_tables + assert "metadata_payload" not in migration + assert "jsonb" not in migration.casefold() + assert "analysis_run_current_status" in migration + assert _REQUIRED_LOOKUP_CODES <= set( + re.findall(r"'(analysis_[a-z0-9_]+)'", migration) + ) + assert "0018_analysis_run_registry.sql" in dockerfile + assert "analysis_run_registry_not_empty" in rollback + assert not _REPAIR_WORKFLOW.exists() + + snapshot_definition = _table_definition(migration, "analysis_source_snapshot") + run_definition = _table_definition(migration, "analysis_run") + assert "maximum_available_time" in snapshot_definition + assert "knowledge_cutoff" not in snapshot_definition + assert "knowledge_cutoff" in run_definition + assert "requested_by_account_id uuid not null" in run_definition + assert "unique (requested_by_account_id, idempotency_key)" in run_definition + assert "enforce_analysis_run_knowledge_cutoff" in migration + assert "enforce_analysis_run_status_transition" in migration + assert "reject_analysis_source_snapshot_update" in migration + assert "enforce_analysis_source_count_freeze" in migration + for table_name in created_tables: + assert len(table_name.split("_")) >= 2 + + +def test_registry_migration_is_idempotent(registry_db) -> None: + """The sequential migration can be replayed without duplicating objects.""" + + with registry_db.cursor() as cursor: + cursor.execute(_REGISTRY_MIGRATION.read_text(encoding="utf-8")) + cursor.execute( + """ + select table_name + from information_schema.tables + where table_schema = 'public' + """ + ) + tables = {row[0] for row in cursor.fetchall()} + cursor.execute( + """ + select table_name + from information_schema.views + where table_schema = 'public' + """ + ) + views = {row[0] for row in cursor.fetchall()} + assert _REQUIRED_TABLES <= tables + assert "analysis_run_current_status" in views + + +def test_registry_persists_normalized_snapshot_scope_and_status(registry_db) -> None: + """A run references one snapshot, one scope, and a legal status history.""" + + with registry_db.cursor() as cursor: + account_id = _insert_account(cursor) + snapshot_id = _insert_snapshot(cursor) + cursor.execute( + """ + insert into analysis_source_count + (analysis_source_snapshot_id, count_type_code, count_value) + values (%s, 'analysis_count_document', 12) + """, + (snapshot_id,), + ) + run_id = _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=account_id, + idempotency_key="synthetic-run-1", + ) + cursor.execute( + """ + insert into analysis_run_scope + (analysis_run_id, scope_kind_code) + values (%s, 'analysis_scope_all_visible') + """, + (run_id,), + ) + cursor.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values + (%s, 1, 'analysis_status_pending', '2026-08-15T01:00:01Z'), + (%s, 2, 'analysis_status_running', '2026-08-15T01:00:02Z'), + (%s, 3, 'analysis_status_succeeded', '2026-08-15T01:00:03Z') + """, + (run_id, run_id, run_id), + ) + cursor.execute( + """ + select status_code, status_ordinal + from analysis_run_current_status + where analysis_run_id = %s + """, + (run_id,), + ) + current_status = cursor.fetchone() + cursor.execute( + """ + select count_value + from analysis_source_count + where analysis_source_snapshot_id = %s + and count_type_code = 'analysis_count_document' + """, + (snapshot_id,), + ) + count_value = cursor.fetchone()[0] + assert current_status == ("analysis_status_succeeded", 3) + assert count_value == 12 + + +def test_snapshot_is_reusable_across_run_owned_knowledge_cutoffs(registry_db) -> None: + """One immutable capture can support multiple later analysis cutoffs.""" + + assert psycopg2 is not None + with registry_db.cursor() as cursor: + snapshot_id = _insert_snapshot( + cursor, + maximum_available_time="2026-08-15T00:00:00Z", + captured_at="2026-08-15T00:05:00Z", + ) + first_account_id = _insert_account(cursor, "first") + second_account_id = _insert_account(cursor, "second") + first_run_id = _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=first_account_id, + idempotency_key="cutoff-one", + knowledge_cutoff="2026-08-15T00:30:00Z", + ) + second_run_id = _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=second_account_id, + idempotency_key="cutoff-two", + knowledge_cutoff="2026-08-16T00:00:00Z", + ) + assert first_run_id != second_run_id + with pytest.raises(psycopg2.errors.RaiseException): + _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=first_account_id, + idempotency_key="future-leakage", + knowledge_cutoff="2026-08-14T23:59:59Z", + ) + + +def test_snapshot_and_counts_are_immutable_and_freeze_at_first_run(registry_db) -> None: + """A run cannot derive from evidence that can still be rewritten.""" + + assert psycopg2 is not None + with registry_db.cursor() as cursor: + snapshot_id = _insert_snapshot(cursor) + cursor.execute( + """ + insert into analysis_source_count + (analysis_source_snapshot_id, count_type_code, count_value) + values (%s, 'analysis_count_document', 12) + """, + (snapshot_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + """ + update analysis_source_snapshot + set source_contract_version = 'rewritten' + where analysis_source_snapshot_id = %s + """, + (snapshot_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + """ + update analysis_source_count + set count_value = 13 + where analysis_source_snapshot_id = %s + """, + (snapshot_id,), + ) + account_id = _insert_account(cursor) + _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=account_id, + idempotency_key="freeze-evidence", + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + """ + insert into analysis_source_count + (analysis_source_snapshot_id, count_type_code, count_value) + values (%s, 'analysis_count_thread', 8) + """, + (snapshot_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + """ + delete from analysis_source_count + where analysis_source_snapshot_id = %s + and count_type_code = 'analysis_count_document' + """, + (snapshot_id,), + ) + + +def test_idempotency_keys_are_scoped_to_the_requesting_account(registry_db) -> None: + """Independent authenticated actors may choose the same opaque key.""" + + assert psycopg2 is not None + with registry_db.cursor() as cursor: + snapshot_id = _insert_snapshot(cursor) + first_account_id = _insert_account(cursor, "first") + second_account_id = _insert_account(cursor, "second") + _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=first_account_id, + idempotency_key="shared-key", + ) + _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=second_account_id, + idempotency_key="shared-key", + ) + with pytest.raises(psycopg2.errors.UniqueViolation): + _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=first_account_id, + idempotency_key="shared-key", + ) + + +def test_registry_rejects_invalid_hashes_negative_counts_and_missing_actor(registry_db) -> None: + """Integrity constraints fail closed before untrusted audit data persists.""" + + assert psycopg2 is not None + with registry_db.cursor() as cursor: + with pytest.raises(psycopg2.errors.CheckViolation): + cursor.execute( + """ + insert into analysis_source_snapshot + (snapshot_sha256, source_contract_version, + maximum_available_time, captured_at) + values ('not-a-digest', 'source-contract-v1', now(), now()) + """ + ) + snapshot_id = _insert_snapshot(cursor) + with pytest.raises(psycopg2.errors.CheckViolation): + cursor.execute( + """ + insert into analysis_source_count + (analysis_source_snapshot_id, count_type_code, count_value) + values (%s, 'analysis_count_source_row', -1) + """, + (snapshot_id,), + ) + with pytest.raises(psycopg2.errors.NotNullViolation): + cursor.execute( + """ + insert into analysis_run + (analysis_source_snapshot_id, run_kind_code, idempotency_key, + knowledge_cutoff, configuration_schema_version, + configuration_sha256, code_revision_sha) + values (%s, 'analysis_run_report', 'missing-actor', + '2026-08-15T00:30:00Z', 'report-run-v1', %s, %s) + """, + (snapshot_id, "d" * 64, "e" * 40), + ) + + +def test_registry_rejects_incoherent_scope_and_failure_events(registry_db) -> None: + """Scope discriminators and failure metadata must agree with their codes.""" + + assert psycopg2 is not None + with registry_db.cursor() as cursor: + snapshot_id = _insert_snapshot(cursor) + account_id = _insert_account(cursor) + run_id = _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=account_id, + idempotency_key="scope-check", + run_kind_code="analysis_run_tepp", + ) + with pytest.raises(psycopg2.errors.CheckViolation): + cursor.execute( + """ + insert into analysis_run_scope + (analysis_run_id, scope_kind_code, scope_key) + values (%s, 'analysis_scope_all_visible', 'unexpected') + """, + (run_id,), + ) + with pytest.raises(psycopg2.errors.CheckViolation): + cursor.execute( + """ + insert into analysis_run_scope + (analysis_run_id, scope_kind_code) + values (%s, 'analysis_scope_thread_group') + """, + (run_id,), + ) + cursor.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, 1, 'analysis_status_pending', '2026-08-15T01:00:00Z') + """, + (run_id,), + ) + with pytest.raises(psycopg2.errors.CheckViolation): + cursor.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, 2, 'analysis_status_failed', '2026-08-15T01:00:01Z') + """, + (run_id,), + ) + cursor.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at, + failure_code, retryable) + values (%s, 2, 'analysis_status_failed', '2026-08-15T01:00:01Z', + 'synthetic_failure', true) + """, + (run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + """ + update analysis_run_status_event + set retryable = false + where analysis_run_id = %s and status_ordinal = 2 + """, + (run_id,), + ) + + +def test_status_history_enforces_contiguous_monotonic_legal_transitions(registry_db) -> None: + """Run state is an ordered state machine rather than an arbitrary event bag.""" + + assert psycopg2 is not None + with registry_db.cursor() as cursor: + snapshot_id = _insert_snapshot(cursor) + account_id = _insert_account(cursor) + + first_run_id = _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=account_id, + idempotency_key="first-status", + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, 1, 'analysis_status_running', '2026-08-15T01:00:00Z') + """, + (first_run_id,), + ) + + second_run_id = _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=account_id, + idempotency_key="second-status", + ) + cursor.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, 1, 'analysis_status_pending', '2026-08-15T01:00:00Z') + """, + (second_run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, 3, 'analysis_status_running', '2026-08-15T01:00:01Z') + """, + (second_run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, 2, 'analysis_status_succeeded', '2026-08-15T01:00:01Z') + """, + (second_run_id,), + ) + cursor.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, 2, 'analysis_status_running', '2026-08-15T01:00:02Z') + """, + (second_run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, 3, 'analysis_status_succeeded', '2026-08-15T01:00:01Z') + """, + (second_run_id,), + ) + cursor.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, 3, 'analysis_status_succeeded', '2026-08-15T01:00:03Z') + """, + (second_run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, 4, 'analysis_status_running', '2026-08-15T01:00:04Z') + """, + (second_run_id,), + ) + + +def test_rollback_refuses_data_loss_and_succeeds_after_explicit_cleanup(registry_db) -> None: + """Downgrade is fail-closed while registry evidence still exists.""" + + assert psycopg2 is not None + rollback_sql = _REGISTRY_ROLLBACK.read_text(encoding="utf-8") + with registry_db.cursor() as cursor: + snapshot_id = _insert_snapshot(cursor) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute(rollback_sql) + registry_db.rollback() + with registry_db.cursor() as cursor: + cursor.execute( + "delete from analysis_source_snapshot " + "where analysis_source_snapshot_id = %s", + (snapshot_id,), + ) + cursor.execute(rollback_sql) + cursor.execute("select to_regclass('public.analysis_run')") + assert cursor.fetchone()[0] is None + cursor.execute(rollback_sql) diff --git a/tests/test_analysis_run_scope_immutability.py b/tests/test_analysis_run_scope_immutability.py new file mode 100644 index 000000000..930967967 --- /dev/null +++ b/tests/test_analysis_run_scope_immutability.py @@ -0,0 +1,186 @@ +"""Regression contracts for immutable analysis authorization scopes.""" + +from __future__ import annotations + +import os +from pathlib import Path +import uuid +from urllib.parse import urlsplit, urlunsplit + +import pytest + +try: + import psycopg2 + import psycopg2.errors + from psycopg2 import sql +except ModuleNotFoundError: # pragma: no cover - local static-only environments + psycopg2 = None # type: ignore[assignment] + sql = None # type: ignore[assignment] + +_ROOT = Path(__file__).resolve().parents[1] +_INITIAL_MIGRATION = _ROOT / "migrations" / "0001_initial_schema.sql" +_REGISTRY_MIGRATION = _ROOT / "migrations" / "0018_analysis_run_registry.sql" +_SCOPE_MIGRATION = _ROOT / "migrations" / "0019_analysis_run_scope_immutability.sql" +_SCOPE_ROLLBACK = ( + _ROOT / "migrations" / "rollback" / "0019_analysis_run_scope_immutability.sql" +) +_POSTGRES_IMAGE = _ROOT / "docker" / "postgres-init" / "Dockerfile" +_ADMIN_DSN = os.environ.get( + "LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN", "postgresql://localhost/postgres" +) + + +def _postgres_available() -> bool: + """Return whether the configured PostgreSQL administrator DSN is reachable.""" + + if psycopg2 is None: + return False + try: + connection = psycopg2.connect(_ADMIN_DSN, connect_timeout=2) + connection.close() + return True + except psycopg2.OperationalError: + return False + + +def _database_dsn(database_name: str) -> str: + """Replace only the database path while retaining connection options.""" + + parsed = urlsplit(_ADMIN_DSN) + return urlunsplit(parsed._replace(path=f"/{database_name}")) + + +def test_scope_hardening_migration_is_wired_and_reversible() -> None: + """Fresh installs and downgrades must include the scope-mutation contract.""" + + assert _SCOPE_MIGRATION.exists() + assert _SCOPE_ROLLBACK.exists() + migration = _SCOPE_MIGRATION.read_text(encoding="utf-8") + rollback = _SCOPE_ROLLBACK.read_text(encoding="utf-8") + dockerfile = _POSTGRES_IMAGE.read_text(encoding="utf-8") + assert "before update or delete on analysis_run_scope" in migration.casefold() + assert "analysis_run_scope_is_immutable" in migration + assert "reject_analysis_run_scope_mutation" in migration + assert "drop trigger if exists analysis_run_scope_update_reject" in rollback + assert "reject_analysis_run_scope_update" in rollback + assert "0019_analysis_run_scope_immutability.sql" in dockerfile + + +@pytest.fixture +def scope_database(): + """Yield a disposable database migrated through scope hardening.""" + + if not _postgres_available(): + pytest.skip("a reachable PostgreSQL administrator DSN is required") + assert psycopg2 is not None + assert sql is not None + database_name = f"lineageweave_scope_{uuid.uuid4().hex[:12]}" + admin = psycopg2.connect(_ADMIN_DSN) + admin.autocommit = True + with admin.cursor() as cursor: + cursor.execute(sql.SQL("create database {}").format(sql.Identifier(database_name))) + try: + connection = psycopg2.connect(_database_dsn(database_name)) + try: + connection.autocommit = True + with connection.cursor() as cursor: + cursor.execute(_INITIAL_MIGRATION.read_text(encoding="utf-8")) + cursor.execute(_REGISTRY_MIGRATION.read_text(encoding="utf-8")) + cursor.execute(_SCOPE_MIGRATION.read_text(encoding="utf-8")) + yield connection + finally: + connection.close() + finally: + with admin.cursor() as cursor: + cursor.execute(sql.SQL("drop database {}").format(sql.Identifier(database_name))) + admin.close() + + +def test_scope_cannot_be_updated_or_deleted_after_registration(scope_database) -> None: + """A persisted run cannot lose or rewrite its authorization boundary.""" + + assert psycopg2 is not None + with scope_database.cursor() as cursor: + suffix = uuid.uuid4().hex + cursor.execute( + """ + insert into user_account + (external_subject_id, display_name, email_address) + values (%s, 'Scope Operator', %s) + returning user_account_id + """, + (f"scope-{suffix}", f"scope-{suffix}@example.test"), + ) + account_id = cursor.fetchone()[0] + cursor.execute( + """ + insert into analysis_source_snapshot + (snapshot_sha256, source_contract_version, + maximum_available_time, captured_at) + values (%s, 'source-contract-v1', + '2026-08-15T00:00:00Z', '2026-08-15T00:05:00Z') + returning analysis_source_snapshot_id + """, + ("a" * 64,), + ) + snapshot_id = cursor.fetchone()[0] + cursor.execute( + """ + insert into analysis_run + (analysis_source_snapshot_id, run_kind_code, idempotency_key, + requested_by_account_id, knowledge_cutoff, + configuration_schema_version, configuration_sha256, + code_revision_sha) + values (%s, 'analysis_run_lineage', 'scope-immutable', %s, + '2026-08-15T00:30:00Z', 'lineage-run-v1', %s, %s) + returning analysis_run_id + """, + (snapshot_id, account_id, "b" * 64, "c" * 40), + ) + run_id = cursor.fetchone()[0] + cursor.execute( + """ + insert into analysis_run_scope (analysis_run_id, scope_kind_code) + values (%s, 'analysis_scope_all_visible') + """, + (run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + """ + update analysis_run_scope + set scope_kind_code = 'analysis_scope_thread_group', + scope_key = 'synthetic-thread' + where analysis_run_id = %s + """, + (run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "delete from analysis_run_scope where analysis_run_id = %s", + (run_id,), + ) + cursor.execute( + "select scope_kind_code from analysis_run_scope where analysis_run_id = %s", + (run_id,), + ) + assert cursor.fetchone() == ("analysis_scope_all_visible",) + + +def test_scope_hardening_rollback_is_replay_safe(scope_database) -> None: + """Repeated downgrade attempts restore exactly one update-only guard.""" + + rollback = _SCOPE_ROLLBACK.read_text(encoding="utf-8") + with scope_database.cursor() as cursor: + cursor.execute(rollback) + cursor.execute(rollback) + cursor.execute( + """ + select tgname + from pg_trigger + where tgrelid = 'analysis_run_scope'::regclass + and not tgisinternal + order by tgname + """ + ) + assert cursor.fetchall() == [("analysis_run_scope_update_reject",)]