From f8fa1924889e2c134b8e7ee46cfdd4150a7e30ec Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:07:45 +0900 Subject: [PATCH 01/20] docs(design): define Milestone 2 analysis-run contract --- ...milestone2-analysis-run-contract-design.md | 113 ++++++++++++++++++ 1 file changed, 113 insertions(+) create mode 100644 docs/superpowers/specs/2026-08-15-milestone2-analysis-run-contract-design.md diff --git a/docs/superpowers/specs/2026-08-15-milestone2-analysis-run-contract-design.md b/docs/superpowers/specs/2026-08-15-milestone2-analysis-run-contract-design.md new file mode 100644 index 00000000..2b16f5ea --- /dev/null +++ b/docs/superpowers/specs/2026-08-15-milestone2-analysis-run-contract-design.md @@ -0,0 +1,113 @@ +# Milestone 2 normalized analysis-run contract design + +## Goal + +Establish the smallest protected vertical slice that lets LineageWeave record an +actual private-PostgreSQL analysis as one auditable derivation without copying +private source identity or raw data into public source control, product APIs, or +routine logs. + +## Product gap + +The protected product already reconstructs direct and indirect lineage, exposes +React and FastAPI surfaces, persists summaries, Keymen, relationship evidence, +issues, reports, and PROV-O. The retained Milestone 2 experiment also proves +that direct private-source analysis can run. The missing bridge is a normalized, +source-redacting contract that can attach that execution evidence to the +protected product instead of replacing it with a parallel application. + +## Selected approach + +```mermaid +flowchart LR + Q[Operator-owned SQL profile] -->|exact digest only| P[analysis_source_profile] + S[Private source snapshot] -->|digest, clocks, aggregate counts| N[analysis_source_snapshot] + P --> N + N --> R[analysis_run_record] + C[bounded configuration] --> R + R --> E[analysis_run_event] + R --> T[analysis_service_run] + R --> A[analysis_artifact_record] + T --> T1[TEPP] + T --> T2[contextual-orchestrator] + T --> T3[fast-mlsirm] + A --> X[private signed acceptance artifacts] +``` + +The SQL profile and credentials stay outside the product database. A deployment +may use a secret file, secret manager, or protected operator configuration. The +runtime database stores only an opaque key, revision, and exact digest. + +## Temporal contract + +The snapshot stores the latest evidence-availability time represented in the +run. The database and Python contract both require: + +```text +maximum_available_time <= knowledge_cutoff +``` + +This aggregate constraint does not replace TEPP's per-document event, +assertion, document, system, available, and cutoff clocks. It is a product +acceptance guard proving that a run did not knowingly include evidence that was +unavailable at its stated historical cutoff. + +## Data contract + +Core data is normalized instead of stored in JSON: + +- a profile revision determines one source kind and query digest; +- a snapshot determines one profile, source digest, cutoff, availability bound, + and aggregate counts; +- a run determines one snapshot, actor, lifecycle, idempotency key, request + digest, and timestamps; +- a configuration determines bounded execution choices for one run; +- events, service calls, and artifacts depend on their parent run. + +Service- and artifact-specific payloads stay in versioned external artifacts and +are linked by digest and URI. This preserves extensibility without turning a +JSON blob into an ungoverned second database. + +## API-safe projection + +The first slice exposes a Python read projection used by the subsequent admin UI +and API slice. It contains: + +- run ID and status; +- opaque profile key and revision; +- request and source digests; +- cutoff and maximum availability time; +- row, document, and thread counts; +- bounded configuration; +- start and completion timestamps. + +It excludes SQL, DSN, source table, raw content, image bytes, provider secrets, +artifact bytes, and private source identifiers. + +## Test-first sequence + +1. Add failing unit contracts for exact hashing, canonical request hashing, + temporal leakage, aggregate-count order, configuration bounds, lifecycle, + and source-safe serialization. +2. Add failing repository tests for transactional registration, immutable + profile/snapshot/idempotency conflicts, terminal status, and read + serialization. +3. Add failing real-PostgreSQL tests for the `0018` schema and constraints. +4. Implement the pure evidence contracts. +5. Implement the transaction repository. +6. Add the normalized migration and Docker fresh-install wiring. +7. Run focused branch coverage and require 100% for both new production modules. +8. Run the complete Python, PostgreSQL, frontend, build, security, and exact-head + review gates after the stack reaches protected `main`. + +## Deferred slices + +- authenticated `GET /api/analysis-runs` and the read-only System Policy UI; +- source-profile secret-manager adapter; +- bounded direct-source execution and ingestion worker; +- TEPP semantic-span import/REST adapter after TEPP ADR 0017 implementation; +- contextual-orchestrator workflow adapter after canonical API and multimodal + message PRs merge; +- private actual-data execution, browser E2E, and signed aggregate acceptance + manifest; +- reconciliation of retained experimental run evidence into this schema. From 35aeaf7f8d16d5f6431617488d52241e5cf4d154 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:08:07 +0900 Subject: [PATCH 02/20] docs(plan): sequence Milestone 2 run-contract TDD --- ...-08-15-milestone2-analysis-run-contract.md | 100 ++++++++++++++++++ 1 file changed, 100 insertions(+) create mode 100644 docs/superpowers/plans/2026-08-15-milestone2-analysis-run-contract.md diff --git a/docs/superpowers/plans/2026-08-15-milestone2-analysis-run-contract.md b/docs/superpowers/plans/2026-08-15-milestone2-analysis-run-contract.md new file mode 100644 index 00000000..7a3cd2ba --- /dev/null +++ b/docs/superpowers/plans/2026-08-15-milestone2-analysis-run-contract.md @@ -0,0 +1,100 @@ +# Milestone 2 analysis-run contract implementation plan + +> Execute with test-driven development. Do not copy the retained parallel +> product branch wholesale. Preserve private actual-data evidence outside +> public source control. + +**Goal:** Add a normalized, source-redacting provenance root for direct +PostgreSQL product analysis and downstream TEPP/orchestrator/fast-mlsirm calls. + +**Base:** Stack additively after the protected PROV-O and related-person label +line. Retarget to `main` only after its ancestors merge. + +## Task 1 — Pure evidence contracts + +**Files** + +- Create `lineageweave/analysis_run.py` +- Create `tests/test_analysis_run.py` + +**Steps** + +1. Write failing tests for exact UTF-8 hashing and deterministic canonical JSON + hashing. +2. Write failing profile validation tests for opaque keys, revisions, source + kinds, and lowercase SHA-256. +3. Write failing snapshot tests for aware timestamps, + `maximum_available_time <= knowledge_cutoff`, nonnegative counts, and + `thread_count <= document_count <= row_count`. +4. Write failing configuration and lifecycle tests. +5. Implement the smallest contract that passes. +6. Require 100% statement and branch coverage. + +## Task 2 — Transaction repository + +**Files** + +- Create `backend/app/analysis_run_ingestion.py` +- Create `backend/tests/test_analysis_run_ingestion.py` + +**Steps** + +1. Write a deterministic asynchronous fake connection and transaction. +2. Prove registration writes profile, snapshot, run, configuration, and start + event inside one transaction. +3. Prove profile, snapshot, and idempotency conflicts fail closed. +4. Prove successful and failed terminal transitions append the correct event. +5. Prove the list projection omits private source channels and bounds its limit. +6. Implement the repository through a minimal structural connection protocol. +7. Require 100% statement and branch coverage. + +## Task 3 — PostgreSQL migration + +**Files** + +- Create `migrations/0018_analysis_run_provenance.sql` +- Create `tests/test_analysis_run_schema.py` +- Modify `docker/postgres-init/Dockerfile` + +**Steps** + +1. Write a real-PostgreSQL test that applies `0001` then `0018`. +2. Prove a normalized source profile, snapshot, run, configuration, service + call, and artifact can be inserted. +3. Prove future-information leakage is rejected by a check constraint. +4. Prove duplicate profile revisions and malformed digests are rejected. +5. Prove every created table has two or more `snake_case` words. +6. Add lookup values, tables, FKs, checks, unique constraints, and indexes. +7. Wire `0018` into fresh Docker initialization. + +## Task 4 — Architecture and research traceability + +**Files** + +- Create `docs/adr/0013-analysis-run-provenance-boundary.md` +- Create this plan and the paired design specification +- Create `docs/doctoring/ANALYSIS_RUN_PROVENANCE_REFERENCES.md` +- Create `CHANGELOG.d/milestone2-analysis-run-contract.md` + +**Steps** + +1. Record LineageWeave, TEPP, contextual-orchestrator, and fast-mlsirm ownership. +2. Document why SQL, DSNs, raw content, and private source identifiers are never + persisted in the run contract. +3. Trace PROV-O, OWL-Time, OpenAPI, and the accepted TEPP baseline in APA 7th + form. +4. Document private acceptance-manifest handling and public-content scanning. + +## Task 5 — Verification and merge sequencing + +1. Run focused tests and 100% branch coverage for both new production modules. +2. Run `compileall` and `git diff --check`. +3. Scan the complete diff for prohibited private source identifiers, SQL, DSNs, + credentials, temporary workflows, and repair scripts. +4. Open a Draft stacked PR; do not mark ready while an ancestor PR is open. +5. After ancestors merge, retarget or rebuild onto current `main`. +6. Run complete Python/PostgreSQL/frontend/build/security/SAST gates on the + exact head. +7. Obtain independent current-head approval and merge only through protected + auto-merge. +8. Continue with the authenticated run-status API/UI and actual execution slice. From 9f67432eb989d5c0cd1b52392eed050b788ba8e2 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:09:00 +0900 Subject: [PATCH 03/20] test(red): define leakage-safe analysis evidence --- tests/test_analysis_run.py | 311 +++++++++++++++++++++++++++++++++++++ 1 file changed, 311 insertions(+) create mode 100644 tests/test_analysis_run.py diff --git a/tests/test_analysis_run.py b/tests/test_analysis_run.py new file mode 100644 index 00000000..f6a28d95 --- /dev/null +++ b/tests/test_analysis_run.py @@ -0,0 +1,311 @@ +"""Unit contracts for source-redacting, leakage-safe analysis evidence.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +import math + +import pytest + +from lineageweave.analysis_run import ( + AnalysisRunConfiguration, + AnalysisRunContractError, + AnalysisRunRegistration, + AnalysisRunSummary, + SourceProfileReference, + SourceSnapshotEvidence, + canonical_json_sha256, + exact_text_sha256, +) + +UTC = timezone.utc +DIGEST_A = "a" * 64 +DIGEST_B = "b" * 64 + + +def _profile() -> SourceProfileReference: + return SourceProfileReference("configured-primary", 1, DIGEST_A) + + +def _snapshot() -> SourceSnapshotEvidence: + return SourceSnapshotEvidence( + source_digest_sha256=DIGEST_B, + knowledge_cutoff=datetime(2026, 8, 15, 0, 0, tzinfo=UTC), + maximum_available_time=datetime(2026, 8, 14, 23, 59, tzinfo=UTC), + row_count=12, + document_count=10, + thread_count=8, + ) + + +def _configuration() -> AnalysisRunConfiguration: + return AnalysisRunConfiguration( + row_limit=0, + write_reports=True, + inspect_inline_images=True, + validate_runtime_schema=True, + model_contract_version="tepp-v1", + output_profile="aggregate-and-product", + ) + + +def test_exact_text_digest_does_not_normalize_source_text() -> None: + assert exact_text_sha256("select 1") != exact_text_sha256("select 1") + with pytest.raises(TypeError, match="must be text"): + exact_text_sha256(1) # type: ignore[arg-type] + + +def test_canonical_json_digest_is_order_independent_and_rejects_nan() -> None: + assert canonical_json_sha256({"b": 2, "a": 1}) == canonical_json_sha256( + {"a": 1, "b": 2} + ) + with pytest.raises(ValueError): + canonical_json_sha256({"metric": math.nan}) + + +@pytest.mark.parametrize( + ("key", "revision", "digest", "message"), + [ + ("Configured Primary", 1, DIGEST_A, "profile_key"), + ("configured-primary", True, DIGEST_A, "profile_revision"), + ("configured-primary", 0, DIGEST_A, "at least 1"), + ("configured-primary", 1, "A" * 64, "lowercase SHA-256"), + ("configured-primary", 1, DIGEST_A, None), + ], +) +def test_source_profile_validation( + key: str, revision: int, digest: str, message: str | None +) -> None: + if message is None: + assert SourceProfileReference(key, revision, digest).public_json()["profile_key"] == key + else: + with pytest.raises(AnalysisRunContractError, match=message): + SourceProfileReference(key, revision, digest) + + +def test_source_kind_must_be_bounded_nonempty_text() -> None: + with pytest.raises(AnalysisRunContractError, match="source_kind_code"): + SourceProfileReference("configured-primary", 1, DIGEST_A, " ") + with pytest.raises(AnalysisRunContractError, match="exceeds 64"): + SourceProfileReference("configured-primary", 1, DIGEST_A, "x" * 65) + + +def test_snapshot_enforces_temporal_leakage_and_count_order() -> None: + snapshot = _snapshot() + assert snapshot.public_json()["row_count"] == 12 + with pytest.raises(AnalysisRunContractError, match="timezone-aware"): + SourceSnapshotEvidence( + DIGEST_B, + datetime(2026, 8, 15), + snapshot.maximum_available_time, + 12, + 10, + 8, + ) + with pytest.raises(AnalysisRunContractError, match="must not exceed"): + SourceSnapshotEvidence( + DIGEST_B, + snapshot.knowledge_cutoff, + snapshot.knowledge_cutoff + timedelta(seconds=1), + 12, + 10, + 8, + ) + with pytest.raises(AnalysisRunContractError, match="must be an integer"): + SourceSnapshotEvidence( + DIGEST_B, + snapshot.knowledge_cutoff, + snapshot.maximum_available_time, + True, + 1, + 1, + ) + with pytest.raises(AnalysisRunContractError, match="non-negative"): + SourceSnapshotEvidence( + DIGEST_B, + snapshot.knowledge_cutoff, + snapshot.maximum_available_time, + -1, + 0, + 0, + ) + with pytest.raises(AnalysisRunContractError, match="document_count"): + SourceSnapshotEvidence( + DIGEST_B, + snapshot.knowledge_cutoff, + snapshot.maximum_available_time, + 1, + 2, + 1, + ) + with pytest.raises(AnalysisRunContractError, match="thread_count"): + SourceSnapshotEvidence( + DIGEST_B, + snapshot.knowledge_cutoff, + snapshot.maximum_available_time, + 2, + 1, + 2, + ) + + +def test_snapshot_rejects_invalid_digest_and_naive_available_time() -> None: + snapshot = _snapshot() + with pytest.raises(AnalysisRunContractError, match="source_digest_sha256"): + SourceSnapshotEvidence( + "bad", + snapshot.knowledge_cutoff, + snapshot.maximum_available_time, + 1, + 1, + 1, + ) + with pytest.raises(AnalysisRunContractError, match="maximum_available_time"): + SourceSnapshotEvidence( + DIGEST_B, + snapshot.knowledge_cutoff, + datetime(2026, 8, 14), + 1, + 1, + 1, + ) + + +@pytest.mark.parametrize( + ("changes", "message"), + [ + ({"row_limit": True}, "row_limit"), + ({"row_limit": -1}, "non-negative"), + ({"write_reports": 1}, "write_reports"), + ({"inspect_inline_images": 1}, "inspect_inline_images"), + ({"validate_runtime_schema": 1}, "validate_runtime_schema"), + ({"model_contract_version": " "}, "model_contract_version"), + ({"output_profile": "x" * 129}, "output_profile"), + ], +) +def test_configuration_validation(changes: dict[str, object], message: str) -> None: + values: dict[str, object] = { + "row_limit": 0, + "write_reports": True, + "inspect_inline_images": True, + "validate_runtime_schema": True, + "model_contract_version": "tepp-v1", + "output_profile": "aggregate-and-product", + } + values.update(changes) + with pytest.raises(AnalysisRunContractError, match=message): + AnalysisRunConfiguration(**values) # type: ignore[arg-type] + + +def test_registration_requires_safe_identifiers_and_aware_start_time() -> None: + started = datetime(2026, 8, 15, 1, 0, tzinfo=UTC) + registration = AnalysisRunRegistration("account-id", "run-key", started) + assert registration.started_at == started + for values, message in ( + ((" ", "run-key", started), "requested_by_account_id"), + (("account-id", " ", started), "idempotency_key"), + (("account-id", "x" * 256, started), "exceeds 255"), + (("account-id", "run-key", datetime(2026, 8, 15)), "started_at"), + ): + with pytest.raises(AnalysisRunContractError, match=message): + AnalysisRunRegistration(*values) + + +def test_request_digest_binds_profile_snapshot_and_configuration() -> None: + configuration = _configuration() + digest = configuration.request_digest(_profile(), _snapshot()) + assert len(digest) == 64 + changed = AnalysisRunConfiguration( + **{**configuration.__dict__, "row_limit": 10} + ) + assert changed.request_digest(_profile(), _snapshot()) != digest + + +def test_public_summary_contains_only_aggregate_safe_fields() -> None: + started = datetime(2026, 8, 15, 1, 0, tzinfo=UTC) + summary = AnalysisRunSummary( + analysis_run_id="019-analysis-run", + profile_key="configured-primary", + profile_revision=1, + run_status_code="analysis_run_succeeded", + request_digest_sha256=DIGEST_A, + source_digest_sha256=DIGEST_B, + knowledge_cutoff=_snapshot().knowledge_cutoff, + maximum_available_time=_snapshot().maximum_available_time, + row_count=12, + document_count=10, + thread_count=8, + started_at=started, + completed_at=started + timedelta(minutes=2), + configuration=_configuration(), + ) + payload = summary.public_json() + serialized = str(payload).lower() + for forbidden in ("dsn", "sql", "query_text", "source_table", "raw_content"): + assert forbidden not in serialized + assert payload["source_snapshot"]["document_count"] == 10 + + running = AnalysisRunSummary( + analysis_run_id="019-running-run", + profile_key="configured-primary", + profile_revision=1, + run_status_code="analysis_run_running", + request_digest_sha256=DIGEST_A, + source_digest_sha256=DIGEST_B, + knowledge_cutoff=_snapshot().knowledge_cutoff, + maximum_available_time=_snapshot().maximum_available_time, + row_count=12, + document_count=10, + thread_count=8, + started_at=started, + completed_at=None, + configuration=_configuration(), + ) + assert running.public_json()["completed_at"] is None + + +def test_summary_rejects_invalid_lifecycle_and_identifiers() -> None: + snapshot = _snapshot() + config = _configuration() + started = datetime(2026, 8, 15, 1, 0, tzinfo=UTC) + base = dict( + analysis_run_id="run-1", + profile_key="configured-primary", + profile_revision=1, + run_status_code="analysis_run_running", + request_digest_sha256=DIGEST_A, + source_digest_sha256=DIGEST_B, + knowledge_cutoff=snapshot.knowledge_cutoff, + maximum_available_time=snapshot.maximum_available_time, + row_count=12, + document_count=10, + thread_count=8, + started_at=started, + completed_at=None, + configuration=config, + ) + for field_name, value, message in ( + ("analysis_run_id", " ", "analysis_run_id"), + ("profile_revision", True, "profile_revision"), + ("profile_revision", 0, "at least 1"), + ("run_status_code", " ", "run_status_code"), + ("request_digest_sha256", "bad", "request_digest_sha256"), + ("started_at", datetime(2026, 8, 15), "started_at"), + ): + values = {**base, field_name: value} + with pytest.raises(AnalysisRunContractError, match=message): + AnalysisRunSummary(**values) + with pytest.raises(AnalysisRunContractError, match="completed_at"): + AnalysisRunSummary( + **{ + **base, + "completed_at": datetime(2026, 8, 15), + } + ) + with pytest.raises(AnalysisRunContractError, match="must not precede"): + AnalysisRunSummary( + **{ + **base, + "completed_at": started - timedelta(seconds=1), + } + ) From 324a483293c383b15d8cddab4176120df98b5c36 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:10:59 +0900 Subject: [PATCH 04/20] test(red): define transactional analysis provenance --- backend/tests/test_analysis_run_ingestion.py | 247 +++++++++++++++++++ 1 file changed, 247 insertions(+) create mode 100644 backend/tests/test_analysis_run_ingestion.py diff --git a/backend/tests/test_analysis_run_ingestion.py b/backend/tests/test_analysis_run_ingestion.py new file mode 100644 index 00000000..915b2b7d --- /dev/null +++ b/backend/tests/test_analysis_run_ingestion.py @@ -0,0 +1,247 @@ +"""Repository tests that exercise every provenance transaction branch.""" + +from __future__ import annotations + +import asyncio +from collections import deque +from datetime import datetime, timedelta, timezone +from typing import Any + +import pytest + +from backend.app.analysis_run_ingestion import ( + AnalysisRunConflict, + complete_analysis_run, + list_analysis_run_summaries, + register_analysis_run, +) +from lineageweave.analysis_run import ( + AnalysisRunConfiguration, + AnalysisRunRegistration, + SourceProfileReference, + SourceSnapshotEvidence, +) + +UTC = timezone.utc +DIGEST_A = "a" * 64 +DIGEST_B = "b" * 64 + + +class FakeTransaction: + """Record transaction entry/exit without a database dependency.""" + + def __init__(self, owner: "FakeConnection") -> None: + self.owner = owner + + async def __aenter__(self) -> "FakeTransaction": + self.owner.transaction_entries += 1 + return self + + async def __aexit__(self, exc_type: Any, exc: Any, traceback: Any) -> bool: + self.owner.transaction_exits.append(exc_type) + return False + + +class FakeConnection: + """Queue deterministic rows and retain every SQL call for assertions.""" + + def __init__( + self, + *, + fetchrows: list[dict[str, Any] | None] | None = None, + fetches: list[list[dict[str, Any]]] | None = None, + ) -> None: + self.fetchrow_results = deque(fetchrows or []) + self.fetch_results = deque(fetches or []) + self.calls: list[tuple[str, str, tuple[Any, ...]]] = [] + self.transaction_entries = 0 + self.transaction_exits: list[Any] = [] + + def transaction(self) -> FakeTransaction: + return FakeTransaction(self) + + async def fetchrow(self, query: str, *arguments: Any) -> dict[str, Any] | None: + self.calls.append(("fetchrow", query, arguments)) + return self.fetchrow_results.popleft() + + async def fetch(self, query: str, *arguments: Any) -> list[dict[str, Any]]: + self.calls.append(("fetch", query, arguments)) + return self.fetch_results.popleft() + + async def execute(self, query: str, *arguments: Any) -> str: + self.calls.append(("execute", query, arguments)) + return "INSERT 0 1" + + +def _inputs() -> tuple[ + SourceProfileReference, + SourceSnapshotEvidence, + AnalysisRunConfiguration, + datetime, +]: + started = datetime(2026, 8, 15, 1, 0, tzinfo=UTC) + return ( + SourceProfileReference("configured-primary", 1, DIGEST_A), + SourceSnapshotEvidence( + DIGEST_B, + started, + started - timedelta(minutes=1), + 12, + 10, + 8, + ), + AnalysisRunConfiguration(0, True, True, True, "tepp-v1", "aggregate"), + started, + ) + + +def test_register_analysis_run_is_one_idempotent_transaction() -> None: + profile, snapshot, configuration, started = _inputs() + conn = FakeConnection( + fetchrows=[ + {"source_profile_id": "profile-id"}, + {"source_snapshot_id": "snapshot-id"}, + {"analysis_run_id": "run-id"}, + ] + ) + result = asyncio.run( + register_analysis_run( + conn, + registration=AnalysisRunRegistration("account-id", "run-key", started), + profile=profile, + snapshot=snapshot, + configuration=configuration, + ) + ) + assert result == "run-id" + assert conn.transaction_entries == 1 + assert conn.transaction_exits == [None] + assert [call[0] for call in conn.calls] == [ + "fetchrow", + "fetchrow", + "fetchrow", + "execute", + "execute", + ] + all_sql = "\n".join(call[1] for call in conn.calls).lower() + assert "source sql" not in all_sql + assert "dsn" not in all_sql + + +@pytest.mark.parametrize( + ("rows", "message", "expected_calls"), + [ + ([None], "source profile", 1), + ([{"source_profile_id": "p"}, None], "source snapshot", 2), + ( + [ + {"source_profile_id": "p"}, + {"source_snapshot_id": "s"}, + None, + ], + "idempotency key", + 3, + ), + ], +) +def test_register_analysis_run_fails_closed_on_immutable_conflicts( + rows: list[dict[str, Any] | None], message: str, expected_calls: int +) -> None: + profile, snapshot, configuration, started = _inputs() + conn = FakeConnection(fetchrows=rows) + with pytest.raises(AnalysisRunConflict, match=message): + asyncio.run( + register_analysis_run( + conn, + registration=AnalysisRunRegistration("account-id", "run-key", started), + profile=profile, + snapshot=snapshot, + configuration=configuration, + ) + ) + assert len(conn.calls) == expected_calls + assert conn.transaction_exits == [AnalysisRunConflict] + + +@pytest.mark.parametrize( + ("succeeded", "status_code", "event_code"), + [ + (True, "analysis_run_succeeded", "analysis_run_completed_event"), + (False, "analysis_run_failed", "analysis_run_failed_event"), + ], +) +def test_complete_analysis_run_records_terminal_status_and_event( + succeeded: bool, status_code: str, event_code: str +) -> None: + conn = FakeConnection(fetchrows=[{"request_digest_sha256": DIGEST_A}]) + completed = datetime(2026, 8, 15, 2, 0, tzinfo=UTC) + asyncio.run( + complete_analysis_run( + conn, + analysis_run_id="run-id", + actor_account_id="account-id", + succeeded=succeeded, + completed_at=completed, + ) + ) + assert conn.calls[0][2][1] == status_code + assert conn.calls[1][2][1] == event_code + assert conn.transaction_exits == [None] + + +def test_complete_analysis_run_rejects_missing_or_completed_run() -> None: + conn = FakeConnection(fetchrows=[None]) + with pytest.raises(AnalysisRunConflict, match="missing or already completed"): + asyncio.run( + complete_analysis_run( + conn, + analysis_run_id="run-id", + actor_account_id="account-id", + succeeded=True, + completed_at=datetime(2026, 8, 15, 2, 0, tzinfo=UTC), + ) + ) + assert conn.transaction_exits == [AnalysisRunConflict] + + +def test_list_analysis_runs_serializes_only_safe_aggregate_fields() -> None: + started = datetime(2026, 8, 15, 1, 0, tzinfo=UTC) + conn = FakeConnection( + fetches=[ + [ + { + "analysis_run_id": "run-id", + "source_profile_key": "configured-primary", + "profile_revision": 1, + "run_status_code": "analysis_run_succeeded", + "request_digest_sha256": DIGEST_A, + "source_digest_sha256": DIGEST_B, + "knowledge_cutoff": started, + "maximum_available_time": started - timedelta(minutes=1), + "row_count": 12, + "document_count": 10, + "thread_count": 8, + "started_at": started, + "completed_at": started + timedelta(minutes=2), + "row_limit": 0, + "write_reports": True, + "inspect_inline_images": True, + "validate_runtime_schema": True, + "model_contract_version": "tepp-v1", + "output_profile": "aggregate", + } + ] + ] + ) + result = asyncio.run(list_analysis_run_summaries(conn, limit=25)) + assert result[0]["source_snapshot"]["row_count"] == 12 + assert conn.calls[0][2] == (25,) + serialized = str(result).lower() + for forbidden in ("dsn", "source_table", "query_text", "raw_content"): + assert forbidden not in serialized + + +@pytest.mark.parametrize("limit", [0, 101, True, 1.5]) +def test_list_analysis_runs_rejects_invalid_limits(limit: object) -> None: + with pytest.raises(ValueError, match="1 through 100"): + asyncio.run(list_analysis_run_summaries(FakeConnection(), limit=limit)) # type: ignore[arg-type] From f505031450110a74a2aebacfc3de9b0a4816b673 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:11:31 +0900 Subject: [PATCH 05/20] test(red): define PostgreSQL analysis provenance schema --- tests/test_analysis_run_schema.py | 188 ++++++++++++++++++++++++++++++ 1 file changed, 188 insertions(+) create mode 100644 tests/test_analysis_run_schema.py diff --git a/tests/test_analysis_run_schema.py b/tests/test_analysis_run_schema.py new file mode 100644 index 00000000..4413119e --- /dev/null +++ b/tests/test_analysis_run_schema.py @@ -0,0 +1,188 @@ +"""Real-PostgreSQL tests for migration 0018's 3NF provenance contract.""" + +from __future__ import annotations + +import os +from pathlib import Path +import re +import uuid +from urllib.parse import urlsplit, urlunsplit + +import psycopg2 +import psycopg2.errors +import pytest + +_ADMIN_DSN = os.environ.get( + "LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN", "postgresql://localhost/postgres" +) +_ROOT = Path(__file__).resolve().parents[1] +_INITIAL = _ROOT / "migrations" / "0001_initial_schema.sql" +_MIGRATION = _ROOT / "migrations" / "0018_analysis_run_provenance.sql" + + +def _postgres_available() -> bool: + try: + psycopg2.connect(_ADMIN_DSN, connect_timeout=2).close() + return True + except psycopg2.OperationalError: + return False + + +pytestmark = pytest.mark.skipif( + not _postgres_available(), + reason=f"no reachable PostgreSQL server at {_ADMIN_DSN}", +) + + +@pytest.fixture +def analysis_db(): + """Create a database containing the initial and analysis-run schemas.""" + + database_name = f"lineageweave_analysis_{uuid.uuid4().hex[:12]}" + admin = psycopg2.connect(_ADMIN_DSN) + admin.autocommit = True + with admin.cursor() as cursor: + cursor.execute(f'create database "{database_name}"') + try: + parsed = urlsplit(_ADMIN_DSN) + database_dsn = urlunsplit(parsed._replace(path=f"/{database_name}")) + connection = psycopg2.connect(database_dsn) + try: + with connection.cursor() as cursor: + cursor.execute(_INITIAL.read_text(encoding="utf-8")) + cursor.execute(_MIGRATION.read_text(encoding="utf-8")) + connection.commit() + yield connection + finally: + connection.close() + finally: + with admin.cursor() as cursor: + cursor.execute(f'drop database "{database_name}"') + admin.close() + + +def _seed_account(connection) -> str: + with connection.cursor() as cursor: + cursor.execute( + "insert into user_account (external_subject_id, display_name, email_address) " + "values ('analysis-subject', 'Analysis Operator', 'analysis@example.test') " + "returning user_account_id" + ) + return str(cursor.fetchone()[0]) + + +def test_analysis_schema_applies_and_persists_normalized_run(analysis_db) -> None: + account_id = _seed_account(analysis_db) + digest_a = "a" * 64 + digest_b = "b" * 64 + with analysis_db.cursor() as cursor: + cursor.execute( + "insert into analysis_source_profile (" + "source_profile_key, profile_revision, source_kind_code, query_digest_sha256" + ") values ('configured-primary', 1, 'postgresql_query_profile', %s) " + "returning source_profile_id", + (digest_a,), + ) + profile_id = cursor.fetchone()[0] + cursor.execute( + "insert into analysis_source_snapshot (" + "source_profile_id, source_digest_sha256, knowledge_cutoff, " + "maximum_available_time, row_count, document_count, thread_count" + ") values (%s, %s, '2026-08-15T00:00:00Z', " + "'2026-08-14T23:59:00Z', 12, 10, 8) returning source_snapshot_id", + (profile_id, digest_b), + ) + snapshot_id = cursor.fetchone()[0] + cursor.execute( + "insert into analysis_run_record (" + "source_snapshot_id, requested_by_account_id, run_status_code, " + "idempotency_key, request_digest_sha256, started_at" + ") values (%s, %s, 'analysis_run_running', 'run-key', %s, now()) " + "returning analysis_run_id", + (snapshot_id, account_id, digest_a), + ) + run_id = cursor.fetchone()[0] + cursor.execute( + "insert into analysis_run_configuration values " + "(%s, 0, true, true, true, 'tepp-v1', 'aggregate')", + (run_id,), + ) + cursor.execute( + "insert into analysis_service_run (" + "analysis_run_id, service_kind_code, service_status_code, " + "remote_run_identifier, idempotency_key, request_digest_sha256, " + "started_at" + ") values (%s, 'analysis_service_tepp', 'analysis_service_running', " + "'remote-run', 'service-key', %s, now())", + (run_id, digest_b), + ) + cursor.execute( + "insert into analysis_artifact_record (" + "analysis_run_id, artifact_kind_code, artifact_reference_uri, " + "content_digest_sha256, byte_count" + ") values (%s, 'analysis_aggregate_manifest', " + "'urn:lineageweave:artifact:aggregate', %s, 512)", + (run_id, digest_a), + ) + cursor.execute( + "select count(*) from analysis_run_record where analysis_run_id = %s", + (run_id,), + ) + assert cursor.fetchone()[0] == 1 + analysis_db.commit() + + +def test_snapshot_rejects_future_information_leakage(analysis_db) -> None: + with analysis_db.cursor() as cursor: + cursor.execute( + "insert into analysis_source_profile (" + "source_profile_key, profile_revision, source_kind_code, query_digest_sha256" + ") values ('configured-secondary', 1, 'postgresql_query_profile', %s) " + "returning source_profile_id", + ("a" * 64,), + ) + profile_id = cursor.fetchone()[0] + with pytest.raises(psycopg2.errors.CheckViolation): + cursor.execute( + "insert into analysis_source_snapshot (" + "source_profile_id, source_digest_sha256, knowledge_cutoff, " + "maximum_available_time, row_count, document_count, thread_count" + ") values (%s, %s, '2026-08-15T00:00:00Z', " + "'2026-08-15T00:00:01Z', 1, 1, 1)", + (profile_id, "b" * 64), + ) + analysis_db.rollback() + + +def test_profile_revision_and_digest_contracts_are_database_enforced(analysis_db) -> None: + with analysis_db.cursor() as cursor: + cursor.execute( + "insert into analysis_source_profile (" + "source_profile_key, profile_revision, source_kind_code, query_digest_sha256" + ") values ('configured-unique', 1, 'postgresql_query_profile', %s)", + ("a" * 64,), + ) + with pytest.raises(psycopg2.errors.UniqueViolation): + cursor.execute( + "insert into analysis_source_profile (" + "source_profile_key, profile_revision, source_kind_code, query_digest_sha256" + ") values ('configured-unique', 1, 'postgresql_query_profile', %s)", + ("b" * 64,), + ) + analysis_db.rollback() + with analysis_db.cursor() as cursor: + with pytest.raises(psycopg2.errors.CheckViolation): + cursor.execute( + "insert into analysis_source_profile (" + "source_profile_key, profile_revision, source_kind_code, query_digest_sha256" + ") values ('configured-bad', 1, 'postgresql_query_profile', 'NOT-A-DIGEST')" + ) + analysis_db.rollback() + + +def test_new_database_object_names_follow_two_word_snake_case_rule() -> None: + sql = _MIGRATION.read_text(encoding="utf-8") + names = re.findall(r"create table (\w+)", sql) + assert names + for name in names: + assert re.fullmatch(r"[a-z][a-z0-9]*(?:_[a-z0-9]+)+", name), name From 505ee8e6dd27a40b5f500da880b614ed4c85f97c Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:14:03 +0900 Subject: [PATCH 06/20] test(red): preserve one idempotent start event --- backend/tests/test_analysis_run_ingestion.py | 27 +++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/backend/tests/test_analysis_run_ingestion.py b/backend/tests/test_analysis_run_ingestion.py index 915b2b7d..35eda955 100644 --- a/backend/tests/test_analysis_run_ingestion.py +++ b/backend/tests/test_analysis_run_ingestion.py @@ -101,7 +101,7 @@ def test_register_analysis_run_is_one_idempotent_transaction() -> None: fetchrows=[ {"source_profile_id": "profile-id"}, {"source_snapshot_id": "snapshot-id"}, - {"analysis_run_id": "run-id"}, + {"analysis_run_id": "run-id", "started_at": started}, ] ) result = asyncio.run( @@ -128,6 +128,31 @@ def test_register_analysis_run_is_one_idempotent_transaction() -> None: assert "dsn" not in all_sql +def test_idempotent_retry_reuses_the_persisted_start_event_time() -> None: + profile, snapshot, configuration, started = _inputs() + persisted_started = started - timedelta(minutes=5) + conn = FakeConnection( + fetchrows=[ + {"source_profile_id": "profile-id"}, + {"source_snapshot_id": "snapshot-id"}, + {"analysis_run_id": "run-id", "started_at": persisted_started}, + ] + ) + result = asyncio.run( + register_analysis_run( + conn, + registration=AnalysisRunRegistration( + "account-id", "run-key", started + timedelta(minutes=1) + ), + profile=profile, + snapshot=snapshot, + configuration=configuration, + ) + ) + assert result == "run-id" + assert conn.calls[-1][2][2] == persisted_started + + @pytest.mark.parametrize( ("rows", "message", "expected_calls"), [ From e122189979b3e34401342794b9b0b45b57198e90 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:15:36 +0900 Subject: [PATCH 07/20] test(red): reject category drift and invalid lifecycle --- tests/test_analysis_run_schema.py | 69 +++++++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) diff --git a/tests/test_analysis_run_schema.py b/tests/test_analysis_run_schema.py index 4413119e..71b4e358 100644 --- a/tests/test_analysis_run_schema.py +++ b/tests/test_analysis_run_schema.py @@ -180,6 +180,75 @@ def test_profile_revision_and_digest_contracts_are_database_enforced(analysis_db analysis_db.rollback() +def test_run_status_rejects_an_unrelated_common_lookup_code(analysis_db) -> None: + account_id = _seed_account(analysis_db) + with analysis_db.cursor() as cursor: + cursor.execute( + "insert into common_lookup_value " + "(lookup_category, lookup_code, lookup_label) " + "values ('permission', 'analysis_unrelated_lookup', 'Unrelated')" + ) + cursor.execute( + "insert into analysis_source_profile (" + "source_profile_key, profile_revision, source_kind_code, query_digest_sha256" + ") values ('configured-status', 1, 'postgresql_query_profile', %s) " + "returning source_profile_id", + ("a" * 64,), + ) + profile_id = cursor.fetchone()[0] + cursor.execute( + "insert into analysis_source_snapshot (" + "source_profile_id, source_digest_sha256, knowledge_cutoff, " + "maximum_available_time, row_count, document_count, thread_count" + ") values (%s, %s, '2026-08-15T00:00:00Z', " + "'2026-08-14T23:59:00Z', 1, 1, 1) returning source_snapshot_id", + (profile_id, "b" * 64), + ) + snapshot_id = cursor.fetchone()[0] + with pytest.raises(psycopg2.errors.CheckViolation): + cursor.execute( + "insert into analysis_run_record (" + "source_snapshot_id, requested_by_account_id, run_status_code, " + "idempotency_key, request_digest_sha256, started_at" + ") values (%s, %s, 'analysis_unrelated_lookup', " + "'bad-status-key', %s, now())", + (snapshot_id, account_id, "c" * 64), + ) + analysis_db.rollback() + + +def test_terminal_run_requires_a_completion_timestamp(analysis_db) -> None: + account_id = _seed_account(analysis_db) + with analysis_db.cursor() as cursor: + cursor.execute( + "insert into analysis_source_profile (" + "source_profile_key, profile_revision, source_kind_code, query_digest_sha256" + ") values ('configured-terminal', 1, 'postgresql_query_profile', %s) " + "returning source_profile_id", + ("a" * 64,), + ) + profile_id = cursor.fetchone()[0] + cursor.execute( + "insert into analysis_source_snapshot (" + "source_profile_id, source_digest_sha256, knowledge_cutoff, " + "maximum_available_time, row_count, document_count, thread_count" + ") values (%s, %s, '2026-08-15T00:00:00Z', " + "'2026-08-14T23:59:00Z', 1, 1, 1) returning source_snapshot_id", + (profile_id, "b" * 64), + ) + snapshot_id = cursor.fetchone()[0] + with pytest.raises(psycopg2.errors.CheckViolation): + cursor.execute( + "insert into analysis_run_record (" + "source_snapshot_id, requested_by_account_id, run_status_code, " + "idempotency_key, request_digest_sha256, started_at" + ") values (%s, %s, 'analysis_run_succeeded', " + "'terminal-key', %s, now())", + (snapshot_id, account_id, "c" * 64), + ) + analysis_db.rollback() + + def test_new_database_object_names_follow_two_word_snake_case_rule() -> None: sql = _MIGRATION.read_text(encoding="utf-8") names = re.findall(r"create table (\w+)", sql) From 967aa7ec7da83431b6b49068eac4a2fe841cbac5 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:16:18 +0900 Subject: [PATCH 08/20] feat: implement leakage-safe analysis evidence --- lineageweave/analysis_run.py | 327 +++++++++++++++++++++++++++++++++++ 1 file changed, 327 insertions(+) create mode 100644 lineageweave/analysis_run.py diff --git a/lineageweave/analysis_run.py b/lineageweave/analysis_run.py new file mode 100644 index 00000000..f070baa6 --- /dev/null +++ b/lineageweave/analysis_run.py @@ -0,0 +1,327 @@ +"""Leakage-safe evidence contracts for direct-PostgreSQL analysis runs. + +The public product must prove which immutable source snapshot and model +configuration produced a derived LineageWeave result without copying SQL, +DSNs, raw source text, image bytes, or private identifiers into logs or API +responses. This module owns that source-redacting contract. Persistence is +implemented by :mod:`backend.app.analysis_run_ingestion`. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from hashlib import sha256 +import json +import re +from typing import Any, Mapping + +_SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") +_PROFILE_KEY_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]{0,127}$") + + +class AnalysisRunContractError(ValueError): + """Raised when provenance evidence would be ambiguous or unsafe.""" + + +def exact_text_sha256(value: str) -> str: + """Return the SHA-256 of the exact UTF-8 text without normalization. + + Exact hashing is intentional: normalizing operator-supplied SQL or model + configuration could make two semantically different source definitions + appear identical. + """ + + if not isinstance(value, str): + raise TypeError("value must be text") + return sha256(value.encode("utf-8")).hexdigest() + + +def canonical_json_sha256(value: Mapping[str, Any]) -> str: + """Hash a deterministic UTF-8 JSON representation of ``value``.""" + + encoded = json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + return sha256(encoded).hexdigest() + + +def _require_sha256(field_name: str, value: str) -> None: + """Reject non-lowercase SHA-256 values with a content-redacting error.""" + + if not isinstance(value, str) or _SHA256_PATTERN.fullmatch(value) is None: + raise AnalysisRunContractError( + f"{field_name} must be a lowercase SHA-256 hex digest" + ) + + +def _require_nonempty(field_name: str, value: str, *, maximum: int = 255) -> None: + """Require bounded, non-whitespace text without echoing private content.""" + + if not isinstance(value, str) or not value.strip(): + raise AnalysisRunContractError(f"{field_name} must be non-empty text") + if len(value) > maximum: + raise AnalysisRunContractError(f"{field_name} exceeds {maximum} characters") + + +def _require_aware(field_name: str, value: datetime) -> None: + """Require a timezone-aware timestamp so temporal comparisons are valid.""" + + if ( + not isinstance(value, datetime) + or value.tzinfo is None + or value.utcoffset() is None + ): + raise AnalysisRunContractError(f"{field_name} must be timezone-aware") + + +@dataclass(frozen=True) +class SourceProfileReference: + """Opaque reference to an operator-owned source query revision. + + ``profile_key`` is a non-sensitive deployment label. The query itself is + never persisted by this contract; only its exact digest is retained. + """ + + profile_key: str + profile_revision: int + query_digest_sha256: str + source_kind_code: str = "postgresql_query_profile" + + def __post_init__(self) -> None: + """Validate the opaque identifier, immutable revision, and digest.""" + + if not isinstance(self.profile_key, str) or _PROFILE_KEY_PATTERN.fullmatch( + self.profile_key + ) is None: + raise AnalysisRunContractError( + "profile_key must be a lowercase opaque identifier of at most " + "128 characters" + ) + if not isinstance(self.profile_revision, int) or isinstance( + self.profile_revision, bool + ): + raise AnalysisRunContractError("profile_revision must be an integer") + if self.profile_revision < 1: + raise AnalysisRunContractError("profile_revision must be at least 1") + _require_sha256("query_digest_sha256", self.query_digest_sha256) + _require_nonempty("source_kind_code", self.source_kind_code, maximum=64) + + def public_json(self) -> dict[str, Any]: + """Return the source-safe profile projection exposed to operators.""" + + return { + "profile_key": self.profile_key, + "profile_revision": self.profile_revision, + "query_digest_sha256": self.query_digest_sha256, + "source_kind_code": self.source_kind_code, + } + + +@dataclass(frozen=True) +class SourceSnapshotEvidence: + """Aggregate-only evidence for one immutable source snapshot. + + ``maximum_available_time`` is the latest evidence-availability time in the + snapshot. Requiring it not to exceed ``knowledge_cutoff`` prevents future + information from entering a historical analysis run. + """ + + source_digest_sha256: str + knowledge_cutoff: datetime + maximum_available_time: datetime + row_count: int + document_count: int + thread_count: int + + def __post_init__(self) -> None: + """Validate digests, clocks, leakage boundary, and aggregate counts.""" + + _require_sha256("source_digest_sha256", self.source_digest_sha256) + _require_aware("knowledge_cutoff", self.knowledge_cutoff) + _require_aware("maximum_available_time", self.maximum_available_time) + if self.maximum_available_time > self.knowledge_cutoff: + raise AnalysisRunContractError( + "maximum_available_time must not exceed knowledge_cutoff" + ) + for field_name in ("row_count", "document_count", "thread_count"): + value = getattr(self, field_name) + if not isinstance(value, int) or isinstance(value, bool): + raise AnalysisRunContractError(f"{field_name} must be an integer") + if value < 0: + raise AnalysisRunContractError(f"{field_name} must be non-negative") + if self.document_count > self.row_count: + raise AnalysisRunContractError( + "document_count must not exceed row_count" + ) + if self.thread_count > self.document_count: + raise AnalysisRunContractError( + "thread_count must not exceed document_count" + ) + + def public_json(self) -> dict[str, Any]: + """Return aggregate evidence without source rows or identifiers.""" + + return { + "source_digest_sha256": self.source_digest_sha256, + "knowledge_cutoff": self.knowledge_cutoff.isoformat(), + "maximum_available_time": self.maximum_available_time.isoformat(), + "row_count": self.row_count, + "document_count": self.document_count, + "thread_count": self.thread_count, + } + + +@dataclass(frozen=True) +class AnalysisRunRegistration: + """Actor, idempotency, and start time for one transactional run.""" + + requested_by_account_id: str + idempotency_key: str + started_at: datetime + + def __post_init__(self) -> None: + """Require bounded opaque identifiers and an aware start timestamp.""" + + _require_nonempty( + "requested_by_account_id", self.requested_by_account_id, maximum=128 + ) + _require_nonempty("idempotency_key", self.idempotency_key, maximum=255) + _require_aware("started_at", self.started_at) + + +@dataclass(frozen=True) +class AnalysisRunConfiguration: + """Versioned, bounded configuration that controls one analysis run.""" + + row_limit: int + write_reports: bool + inspect_inline_images: bool + validate_runtime_schema: bool + model_contract_version: str + output_profile: str + + def __post_init__(self) -> None: + """Reject unbounded or ambiguous run configuration.""" + + if not isinstance(self.row_limit, int) or isinstance(self.row_limit, bool): + raise AnalysisRunContractError("row_limit must be an integer") + if self.row_limit < 0: + raise AnalysisRunContractError("row_limit must be non-negative") + for field_name in ( + "write_reports", + "inspect_inline_images", + "validate_runtime_schema", + ): + if not isinstance(getattr(self, field_name), bool): + raise AnalysisRunContractError(f"{field_name} must be a boolean") + _require_nonempty( + "model_contract_version", self.model_contract_version, maximum=128 + ) + _require_nonempty("output_profile", self.output_profile, maximum=128) + + def public_json(self) -> dict[str, Any]: + """Return the bounded configuration safe for audit and API output.""" + + return { + "row_limit": self.row_limit, + "write_reports": self.write_reports, + "inspect_inline_images": self.inspect_inline_images, + "validate_runtime_schema": self.validate_runtime_schema, + "model_contract_version": self.model_contract_version, + "output_profile": self.output_profile, + } + + def request_digest( + self, + profile: SourceProfileReference, + snapshot: SourceSnapshotEvidence, + ) -> str: + """Hash the complete source-safe request contract deterministically.""" + + return canonical_json_sha256( + { + "source_profile": profile.public_json(), + "source_snapshot": snapshot.public_json(), + "configuration": self.public_json(), + } + ) + + +@dataclass(frozen=True) +class AnalysisRunSummary: + """Read-only, source-redacting operator projection for a persisted run.""" + + analysis_run_id: str + profile_key: str + profile_revision: int + run_status_code: str + request_digest_sha256: str + source_digest_sha256: str + knowledge_cutoff: datetime + maximum_available_time: datetime + row_count: int + document_count: int + thread_count: int + started_at: datetime + completed_at: datetime | None + configuration: AnalysisRunConfiguration + + def __post_init__(self) -> None: + """Reuse evidence contracts and validate lifecycle timestamps.""" + + _require_nonempty("analysis_run_id", self.analysis_run_id, maximum=128) + _require_nonempty("profile_key", self.profile_key, maximum=128) + if not isinstance(self.profile_revision, int) or isinstance( + self.profile_revision, bool + ): + raise AnalysisRunContractError("profile_revision must be an integer") + if self.profile_revision < 1: + raise AnalysisRunContractError("profile_revision must be at least 1") + _require_nonempty("run_status_code", self.run_status_code, maximum=64) + _require_sha256("request_digest_sha256", self.request_digest_sha256) + SourceSnapshotEvidence( + source_digest_sha256=self.source_digest_sha256, + knowledge_cutoff=self.knowledge_cutoff, + maximum_available_time=self.maximum_available_time, + row_count=self.row_count, + document_count=self.document_count, + thread_count=self.thread_count, + ) + _require_aware("started_at", self.started_at) + if self.completed_at is not None: + _require_aware("completed_at", self.completed_at) + if self.completed_at < self.started_at: + raise AnalysisRunContractError( + "completed_at must not precede started_at" + ) + + def public_json(self) -> dict[str, Any]: + """Return an API-safe summary with no DSN, SQL, raw content, or URI.""" + + return { + "analysis_run_id": self.analysis_run_id, + "source_profile": { + "profile_key": self.profile_key, + "profile_revision": self.profile_revision, + }, + "run_status_code": self.run_status_code, + "request_digest_sha256": self.request_digest_sha256, + "source_snapshot": { + "source_digest_sha256": self.source_digest_sha256, + "knowledge_cutoff": self.knowledge_cutoff.isoformat(), + "maximum_available_time": self.maximum_available_time.isoformat(), + "row_count": self.row_count, + "document_count": self.document_count, + "thread_count": self.thread_count, + }, + "started_at": self.started_at.isoformat(), + "completed_at": ( + self.completed_at.isoformat() if self.completed_at else None + ), + "configuration": self.configuration.public_json(), + } From ab78bdc9894428656df481b11c8cfde36b39d00a Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:16:53 +0900 Subject: [PATCH 09/20] feat: persist normalized analysis-run provenance --- backend/app/analysis_run_ingestion.py | 311 ++++++++++++++++++++++++++ 1 file changed, 311 insertions(+) create mode 100644 backend/app/analysis_run_ingestion.py diff --git a/backend/app/analysis_run_ingestion.py b/backend/app/analysis_run_ingestion.py new file mode 100644 index 00000000..f6fad803 --- /dev/null +++ b/backend/app/analysis_run_ingestion.py @@ -0,0 +1,311 @@ +"""Transactional PostgreSQL repository for normalized analysis-run evidence. + +The repository writes only opaque source-profile references, hashes, aggregate +counts, bounded configuration, status transitions, and service identifiers. +It never persists source SQL, DSNs, raw post content, image bytes, or provider +credentials. Any async PostgreSQL connection that implements the small +protocols below can be used; no ORM or file-backed database is involved. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any, Mapping, Protocol, Sequence + +from lineageweave.analysis_run import ( + AnalysisRunConfiguration, + AnalysisRunRegistration, + AnalysisRunSummary, + SourceProfileReference, + SourceSnapshotEvidence, +) + + +class AsyncTransaction(Protocol): + """Async context-manager contract returned by PostgreSQL connections.""" + + async def __aenter__(self) -> Any: + """Enter a database transaction.""" + + async def __aexit__( + self, exc_type: Any, exc: Any, traceback: Any + ) -> bool | None: + """Commit on success or roll back on failure.""" + + +class AnalysisRunConnection(Protocol): + """Minimal async PostgreSQL connection surface used by this repository.""" + + def transaction(self) -> AsyncTransaction: + """Return an async transaction context manager.""" + + async def fetchrow( + self, query: str, *arguments: Any + ) -> Mapping[str, Any] | None: + """Execute ``query`` and return at most one row.""" + + async def fetch( + self, query: str, *arguments: Any + ) -> Sequence[Mapping[str, Any]]: + """Execute ``query`` and return all rows.""" + + async def execute(self, query: str, *arguments: Any) -> str: + """Execute a statement and return its command tag.""" + + +class AnalysisRunConflict(RuntimeError): + """An immutable profile, snapshot, or idempotency key was reused differently.""" + + +async def register_analysis_run( + conn: AnalysisRunConnection, + *, + registration: AnalysisRunRegistration, + profile: SourceProfileReference, + snapshot: SourceSnapshotEvidence, + configuration: AnalysisRunConfiguration, +) -> str: + """Create or idempotently reuse one running analysis record. + + The entire profile → snapshot → run → configuration → event write occurs + in one transaction. Reusing an immutable key with different evidence + fails closed instead of silently replacing provenance. + """ + + request_digest = configuration.request_digest(profile, snapshot) + async with conn.transaction(): + profile_row = await conn.fetchrow( + """ + insert into analysis_source_profile ( + source_profile_key, profile_revision, source_kind_code, + query_digest_sha256 + ) values ($1, $2, $3, $4) + on conflict (source_profile_key, profile_revision) do update + set source_profile_key = excluded.source_profile_key + where analysis_source_profile.source_kind_code = excluded.source_kind_code + and analysis_source_profile.query_digest_sha256 = excluded.query_digest_sha256 + returning source_profile_id + """, + profile.profile_key, + profile.profile_revision, + profile.source_kind_code, + profile.query_digest_sha256, + ) + if profile_row is None: + raise AnalysisRunConflict( + "source profile revision already exists with different evidence" + ) + + snapshot_row = await conn.fetchrow( + """ + insert into analysis_source_snapshot ( + source_profile_id, source_digest_sha256, knowledge_cutoff, + maximum_available_time, row_count, document_count, thread_count + ) values ($1, $2, $3, $4, $5, $6, $7) + on conflict ( + source_profile_id, source_digest_sha256, knowledge_cutoff + ) do update + set source_digest_sha256 = excluded.source_digest_sha256 + where analysis_source_snapshot.maximum_available_time = + excluded.maximum_available_time + and analysis_source_snapshot.row_count = excluded.row_count + and analysis_source_snapshot.document_count = excluded.document_count + and analysis_source_snapshot.thread_count = excluded.thread_count + returning source_snapshot_id + """, + profile_row["source_profile_id"], + snapshot.source_digest_sha256, + snapshot.knowledge_cutoff, + snapshot.maximum_available_time, + snapshot.row_count, + snapshot.document_count, + snapshot.thread_count, + ) + if snapshot_row is None: + raise AnalysisRunConflict( + "source snapshot already exists with different aggregate evidence" + ) + + run_row = await conn.fetchrow( + """ + insert into analysis_run_record ( + source_snapshot_id, requested_by_account_id, run_status_code, + idempotency_key, request_digest_sha256, started_at + ) values ($1, $2::uuid, 'analysis_run_running', $3, $4, $5) + on conflict (idempotency_key) do update + set idempotency_key = excluded.idempotency_key + where analysis_run_record.request_digest_sha256 = + excluded.request_digest_sha256 + and analysis_run_record.requested_by_account_id = + excluded.requested_by_account_id + returning analysis_run_id, started_at + """, + snapshot_row["source_snapshot_id"], + registration.requested_by_account_id, + registration.idempotency_key, + request_digest, + registration.started_at, + ) + if run_row is None: + raise AnalysisRunConflict( + "idempotency key already exists for a different request" + ) + + analysis_run_id = run_row["analysis_run_id"] + await conn.execute( + """ + insert into analysis_run_configuration ( + analysis_run_id, row_limit, write_reports, inspect_inline_images, + validate_runtime_schema, model_contract_version, output_profile + ) values ($1, $2, $3, $4, $5, $6, $7) + on conflict (analysis_run_id) do nothing + """, + analysis_run_id, + configuration.row_limit, + configuration.write_reports, + configuration.inspect_inline_images, + configuration.validate_runtime_schema, + configuration.model_contract_version, + configuration.output_profile, + ) + await conn.execute( + """ + insert into analysis_run_event ( + analysis_run_id, event_type_code, actor_account_id, + occurred_at, payload_digest_sha256 + ) values ($1, 'analysis_run_started_event', $2::uuid, $3, $4) + on conflict ( + analysis_run_id, event_type_code, occurred_at, + payload_digest_sha256 + ) do nothing + """, + analysis_run_id, + registration.requested_by_account_id, + run_row["started_at"], + request_digest, + ) + return str(analysis_run_id) + + +async def complete_analysis_run( + conn: AnalysisRunConnection, + *, + analysis_run_id: str, + actor_account_id: str, + succeeded: bool, + completed_at: datetime, +) -> None: + """Complete a running record exactly once and append its audit event.""" + + status_code = "analysis_run_succeeded" if succeeded else "analysis_run_failed" + event_type_code = ( + "analysis_run_completed_event" + if succeeded + else "analysis_run_failed_event" + ) + async with conn.transaction(): + run_row = await conn.fetchrow( + """ + update analysis_run_record + set run_status_code = $2, completed_at = $3 + where analysis_run_id = $1::uuid + and run_status_code = 'analysis_run_running' + and completed_at is null + returning request_digest_sha256 + """, + analysis_run_id, + status_code, + completed_at, + ) + if run_row is None: + raise AnalysisRunConflict("analysis run is missing or already completed") + await conn.execute( + """ + insert into analysis_run_event ( + analysis_run_id, event_type_code, actor_account_id, + occurred_at, payload_digest_sha256 + ) values ($1::uuid, $2, $3::uuid, $4, $5) + """, + analysis_run_id, + event_type_code, + actor_account_id, + completed_at, + run_row["request_digest_sha256"], + ) + + +async def list_analysis_run_summaries( + conn: AnalysisRunConnection, + *, + limit: int = 50, +) -> list[dict[str, Any]]: + """Return the newest source-redacting run summaries for an admin surface.""" + + if ( + not isinstance(limit, int) + or isinstance(limit, bool) + or not 1 <= limit <= 100 + ): + raise ValueError("limit must be an integer from 1 through 100") + rows = await conn.fetch( + """ + select + ar.analysis_run_id, + asp.source_profile_key, + asp.profile_revision, + ar.run_status_code, + ar.request_digest_sha256, + ass.source_digest_sha256, + ass.knowledge_cutoff, + ass.maximum_available_time, + ass.row_count, + ass.document_count, + ass.thread_count, + ar.started_at, + ar.completed_at, + arc.row_limit, + arc.write_reports, + arc.inspect_inline_images, + arc.validate_runtime_schema, + arc.model_contract_version, + arc.output_profile + from analysis_run_record ar + join analysis_source_snapshot ass + on ass.source_snapshot_id = ar.source_snapshot_id + join analysis_source_profile asp + on asp.source_profile_id = ass.source_profile_id + join analysis_run_configuration arc + on arc.analysis_run_id = ar.analysis_run_id + order by ar.started_at desc, ar.analysis_run_id desc + limit $1 + """, + limit, + ) + summaries: list[dict[str, Any]] = [] + for row in rows: + configuration = AnalysisRunConfiguration( + row_limit=row["row_limit"], + write_reports=row["write_reports"], + inspect_inline_images=row["inspect_inline_images"], + validate_runtime_schema=row["validate_runtime_schema"], + model_contract_version=row["model_contract_version"], + output_profile=row["output_profile"], + ) + summary = AnalysisRunSummary( + analysis_run_id=str(row["analysis_run_id"]), + profile_key=row["source_profile_key"], + profile_revision=row["profile_revision"], + run_status_code=row["run_status_code"], + request_digest_sha256=row["request_digest_sha256"], + source_digest_sha256=row["source_digest_sha256"], + knowledge_cutoff=row["knowledge_cutoff"], + maximum_available_time=row["maximum_available_time"], + row_count=row["row_count"], + document_count=row["document_count"], + thread_count=row["thread_count"], + started_at=row["started_at"], + completed_at=row["completed_at"], + configuration=configuration, + ) + summaries.append(summary.public_json()) + return summaries From 7408b297eeddf3a6c1255a4721fa8dde09920ea1 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:17:48 +0900 Subject: [PATCH 10/20] feat(db): add normalized analysis-run provenance --- migrations/0018_analysis_run_provenance.sql | 208 ++++++++++++++++++++ 1 file changed, 208 insertions(+) create mode 100644 migrations/0018_analysis_run_provenance.sql diff --git a/migrations/0018_analysis_run_provenance.sql b/migrations/0018_analysis_run_provenance.sql new file mode 100644 index 00000000..7df8fb25 --- /dev/null +++ b/migrations/0018_analysis_run_provenance.sql @@ -0,0 +1,208 @@ +-- Normalized, aggregate-only provenance for Milestone 2 analysis runs. +-- +-- The source SQL, DSN, raw post content, image bytes, provider credentials, +-- and private source identifiers remain operator-owned and are never stored in +-- these tables. Only opaque profile keys, immutable digests, bounded +-- configuration, aggregate counts, service identifiers, and status events are +-- persisted. This keeps actual-data acceptance evidence outside public source +-- control while making runtime derivations auditable. + +begin; + +insert into common_lookup_value ( + lookup_category, lookup_code, lookup_label, display_order +) values + ('analysis_source_kind', 'postgresql_query_profile', 'PostgreSQL query profile', 10), + ('analysis_run_status', 'analysis_run_running', 'Running', 10), + ('analysis_run_status', 'analysis_run_succeeded', 'Succeeded', 20), + ('analysis_run_status', 'analysis_run_failed', 'Failed', 30), + ('analysis_run_event_type', 'analysis_run_started_event', 'Analysis run started', 10), + ('analysis_run_event_type', 'analysis_run_completed_event', 'Analysis run completed', 20), + ('analysis_run_event_type', 'analysis_run_failed_event', 'Analysis run failed', 30), + ('analysis_service_kind', 'analysis_service_tepp', 'TEPP', 10), + ('analysis_service_kind', 'analysis_service_orchestrator', 'Contextual Orchestrator', 20), + ('analysis_service_kind', 'analysis_service_fast_mlsirm', 'fast-mlsirm', 30), + ('analysis_service_status', 'analysis_service_running', 'Service run running', 10), + ('analysis_service_status', 'analysis_service_succeeded', 'Service run succeeded', 20), + ('analysis_service_status', 'analysis_service_failed', 'Service run failed', 30), + ('analysis_artifact_kind', 'analysis_aggregate_manifest', 'Aggregate acceptance manifest', 10), + ('analysis_artifact_kind', 'analysis_reproducibility_manifest', 'Reproducibility manifest', 20), + ('analysis_artifact_kind', 'analysis_browser_evidence', 'Browser acceptance evidence', 30); + +create table analysis_source_profile ( + source_profile_id uuid primary key default uuid_generate_v4(), + source_profile_key text not null + check (source_profile_key ~ '^[a-z0-9][a-z0-9._-]{0,127}$'), + profile_revision integer not null check (profile_revision >= 1), + source_kind_code text not null references common_lookup_value (lookup_code) + check (source_kind_code in ('postgresql_query_profile')), + query_digest_sha256 text not null + check (query_digest_sha256 ~ '^[0-9a-f]{64}$'), + created_at timestamptz not null default now(), + unique (source_profile_key, profile_revision) +); + +comment on table analysis_source_profile is + 'Opaque, immutable source-query revision. The SQL and DSN remain outside ' + 'the product database; only a profile key and exact query digest are stored.'; + +create table analysis_source_snapshot ( + source_snapshot_id uuid primary key default uuid_generate_v4(), + source_profile_id uuid not null + references analysis_source_profile (source_profile_id), + source_digest_sha256 text not null + check (source_digest_sha256 ~ '^[0-9a-f]{64}$'), + knowledge_cutoff timestamptz not null, + maximum_available_time timestamptz not null, + row_count bigint not null check (row_count >= 0), + document_count bigint not null check ( + document_count >= 0 and document_count <= row_count + ), + thread_count bigint not null check ( + thread_count >= 0 and thread_count <= document_count + ), + observed_at timestamptz not null default now(), + check (maximum_available_time <= knowledge_cutoff), + check (observed_at >= maximum_available_time), + unique (source_profile_id, source_digest_sha256, knowledge_cutoff) +); + +comment on table analysis_source_snapshot is + 'Aggregate-only immutable snapshot evidence. maximum_available_time must ' + 'not exceed knowledge_cutoff, preventing future-information leakage.'; + +create table analysis_run_record ( + analysis_run_id uuid primary key default uuid_generate_v4(), + source_snapshot_id uuid not null + references analysis_source_snapshot (source_snapshot_id), + requested_by_account_id uuid not null + references user_account (user_account_id), + run_status_code text not null references common_lookup_value (lookup_code) + check (run_status_code in ( + 'analysis_run_running', + 'analysis_run_succeeded', + 'analysis_run_failed' + )), + idempotency_key text not null unique + check (length(btrim(idempotency_key)) between 1 and 255), + request_digest_sha256 text not null + check (request_digest_sha256 ~ '^[0-9a-f]{64}$'), + started_at timestamptz not null, + completed_at timestamptz, + created_at timestamptz not null default now(), + check (completed_at is null or completed_at >= started_at), + check ( + (run_status_code = 'analysis_run_running' and completed_at is null) + or ( + run_status_code in ('analysis_run_succeeded', 'analysis_run_failed') + and completed_at is not null + ) + ), + check (created_at >= started_at) +); + +create index analysis_run_started_idx + on analysis_run_record (started_at desc, analysis_run_id desc); + +create table analysis_run_configuration ( + analysis_run_id uuid primary key + references analysis_run_record (analysis_run_id) on delete cascade, + row_limit bigint not null check (row_limit >= 0), + write_reports boolean not null, + inspect_inline_images boolean not null, + validate_runtime_schema boolean not null, + model_contract_version text not null, + output_profile text not null, + check (length(btrim(model_contract_version)) between 1 and 128), + check (length(btrim(output_profile)) between 1 and 128) +); + +create table analysis_run_event ( + analysis_run_event_id uuid primary key default uuid_generate_v4(), + analysis_run_id uuid not null + references analysis_run_record (analysis_run_id) on delete cascade, + event_type_code text not null references common_lookup_value (lookup_code) + check (event_type_code in ( + 'analysis_run_started_event', + 'analysis_run_completed_event', + 'analysis_run_failed_event' + )), + actor_account_id uuid not null references user_account (user_account_id), + occurred_at timestamptz not null, + recorded_at timestamptz not null default now(), + payload_digest_sha256 text not null + check (payload_digest_sha256 ~ '^[0-9a-f]{64}$'), + check (recorded_at >= occurred_at), + unique ( + analysis_run_id, event_type_code, occurred_at, payload_digest_sha256 + ) +); + +create index analysis_run_event_run_idx + on analysis_run_event (analysis_run_id, occurred_at); + +create table analysis_service_run ( + analysis_service_run_id uuid primary key default uuid_generate_v4(), + analysis_run_id uuid not null + references analysis_run_record (analysis_run_id) on delete cascade, + service_kind_code text not null references common_lookup_value (lookup_code) + check (service_kind_code in ( + 'analysis_service_tepp', + 'analysis_service_orchestrator', + 'analysis_service_fast_mlsirm' + )), + service_status_code text not null references common_lookup_value (lookup_code) + check (service_status_code in ( + 'analysis_service_running', + 'analysis_service_succeeded', + 'analysis_service_failed' + )), + remote_run_identifier text not null, + idempotency_key text not null, + request_digest_sha256 text not null + check (request_digest_sha256 ~ '^[0-9a-f]{64}$'), + retryable boolean not null default false, + started_at timestamptz not null, + completed_at timestamptz, + check (length(btrim(remote_run_identifier)) between 1 and 255), + check (length(btrim(idempotency_key)) between 1 and 255), + check (completed_at is null or completed_at >= started_at), + check ( + (service_status_code = 'analysis_service_running' and completed_at is null) + or ( + service_status_code in ( + 'analysis_service_succeeded', 'analysis_service_failed' + ) + and completed_at is not null + ) + ), + unique (service_kind_code, remote_run_identifier), + unique (service_kind_code, idempotency_key) +); + +create index analysis_service_run_parent_idx + on analysis_service_run (analysis_run_id, started_at); + +create table analysis_artifact_record ( + analysis_artifact_id uuid primary key default uuid_generate_v4(), + analysis_run_id uuid not null + references analysis_run_record (analysis_run_id) on delete cascade, + artifact_kind_code text not null references common_lookup_value (lookup_code) + check (artifact_kind_code in ( + 'analysis_aggregate_manifest', + 'analysis_reproducibility_manifest', + 'analysis_browser_evidence' + )), + artifact_reference_uri text not null, + content_digest_sha256 text not null + check (content_digest_sha256 ~ '^[0-9a-f]{64}$'), + byte_count bigint not null check (byte_count >= 0), + created_at timestamptz not null default now(), + check (length(btrim(artifact_reference_uri)) between 1 and 2048), + unique (analysis_run_id, artifact_kind_code, content_digest_sha256) +); + +create index analysis_artifact_run_idx + on analysis_artifact_record (analysis_run_id, created_at); + +commit; From b427c985ddb374b45013163463c86205b2b7b986 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:18:17 +0900 Subject: [PATCH 11/20] chore(db): apply analysis provenance on fresh install --- docker/postgres-init/Dockerfile | 1 + 1 file changed, 1 insertion(+) diff --git a/docker/postgres-init/Dockerfile b/docker/postgres-init/Dockerfile index d10dec64..6863bc8a 100644 --- a/docker/postgres-init/Dockerfile +++ b/docker/postgres-init/Dockerfile @@ -24,6 +24,7 @@ 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_provenance.sql /docker-entrypoint-initdb.d/19-analysis-run-provenance.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 From 0217d61dfb98f470361b0d5a830c252756d52d99 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:20:46 +0900 Subject: [PATCH 12/20] docs(adr): establish analysis provenance boundary --- .../0013-analysis-run-provenance-boundary.md | 160 ++++++++++++++++++ 1 file changed, 160 insertions(+) create mode 100644 docs/adr/0013-analysis-run-provenance-boundary.md diff --git a/docs/adr/0013-analysis-run-provenance-boundary.md b/docs/adr/0013-analysis-run-provenance-boundary.md new file mode 100644 index 00000000..4a41b760 --- /dev/null +++ b/docs/adr/0013-analysis-run-provenance-boundary.md @@ -0,0 +1,160 @@ +# ADR 0013 — Normalized analysis-run provenance boundary + +**Status:** Proposed +**Date:** 2026-08-15 +**Decision owners:** LineageWeave product and data architecture +**Depends on:** ADR 0001, ADR 0003, ADR 0004, ADR 0011; TEPP ADR 0011 and proposed ADR 0017 + +## Context + +Milestone 2 executes LineageWeave against an operator-controlled PostgreSQL +source and combines product reconstruction with TEPP, contextual-orchestrator, +and fast-mlsirm. The product must prove which source snapshot, temporal cutoff, +configuration, service calls, and artifacts produced a result. At the same time, +the public repository and routine operational telemetry must never reveal the +private SQL, DSN, raw source rows, image bytes, provider credentials, or private +source identifiers. + +A prior experimental implementation stored multiple service-specific run tables +with JSON-shaped metadata. Copying that schema into the protected product would +create duplicated status vocabularies, weak referential integrity, and an +unbounded channel for source content. It would also make the same analysis look +like unrelated TEPP, orchestrator, and report runs instead of one provenance +trace. + +TEPP's accepted temporal baseline distinguishes evidence availability from the +analysis knowledge cutoff. Proposed TEPP ADR 0017 further assigns exact evidence +identity, model profiles, and downstream service boundaries to TEPP; LineageWeave +may align with that boundary but cannot claim its implementation until that ADR +and its production slice are accepted. A LineageWeave run therefore needs a +normalized source snapshot that rejects future-information leakage before any +derived product claim is accepted. + +## Decision + +LineageWeave adds seven third-normal-form tables: + +- `analysis_source_profile` — immutable opaque query profile revision and exact + query digest; never the query text or DSN; +- `analysis_source_snapshot` — aggregate counts, source digest, + `maximum_available_time`, and `knowledge_cutoff`; +- `analysis_run_record` — one idempotent product run and lifecycle status; +- `analysis_run_configuration` — the bounded configuration that depends only on + the run; +- `analysis_run_event` — append-only status evidence with actor, occurrence + time, recording time, and payload digest; +- `analysis_service_run` — version-neutral child calls to TEPP, + contextual-orchestrator, or fast-mlsirm; +- `analysis_artifact_record` — digest and external reference for aggregate, + reproducibility, or browser evidence. + +`analysis_source_snapshot` enforces: + +```text +maximum_available_time <= knowledge_cutoff +``` + +This is the aggregate acceptance gate for TEPP's no-future-information +contract. Per-document clocks remain TEPP evidence authority; LineageWeave does +not collapse TEPP's event, assertion, document, system, available, and cutoff +clocks into one date. + +All run, service, event, and artifact kinds live in `common_lookup_value`. +Database object names contain at least two `snake_case` words. The product +repository exposes only source-safe summaries: opaque profile key and revision, +digests, counts, clocks, configuration, and status. It never returns source SQL, +DSNs, raw rows, source table names, content, artifact bytes, credentials, or +private file paths. + +The Python contract in `lineageweave.analysis_run` validates the same digest, +clock, count, and output rules before persistence. The repository in +`backend.app.analysis_run_ingestion` writes profile → snapshot → run → +configuration → event in one PostgreSQL transaction and fails closed when an +immutable key is reused with different evidence. + +## Ownership and service boundaries + +- LineageWeave owns product-run provenance and the buyer-facing read projection. +- TEPP owns exact evidence spans, multilingual semantic measurement, temporal + validity, model-profile budgeting, and calibrated measurement artifacts. +- contextual-orchestrator owns provider routing and workflow traces, not source + evidence or LineageWeave authorization. +- fast-mlsirm owns numerical psychometric artifacts, not product-run state. +- Private source SQL and credentials remain deployment secrets outside the + product database and outside public source control. +- Aggregate actual-data acceptance manifests remain deployment artifacts; the + public repository stores only their schema and digest contract. + +## Alternatives considered + +1. **Store source SQL and DSN in the run table.** Rejected because telemetry, + support exports, or database access could disclose infrastructure and private + source identity. +2. **One JSON metadata column per service.** Rejected because core fields lose + foreign keys, temporal constraints, and stable query semantics. +3. **Separate TEPP, orchestrator, and report run roots.** Rejected because the + product cannot reconstruct one end-to-end derivation or enforce a shared + idempotency boundary. +4. **Store only logs.** Rejected because logs are not a normalized source of + truth, are often sampled or deleted, and cannot enforce referential or + temporal constraints. +5. **Normalized aggregate provenance with external artifact references.** + Selected. + +## Consequences + +### Positive + +- private-source analysis can be audited without publishing source identity; +- a historical run fails before persistence when evidence became available + after its knowledge cutoff; +- retries are idempotent and evidence conflicts are explicit; +- TEPP, orchestrator, and fast-mlsirm calls share one parent lineage; +- operators can inspect run state through a stable, content-redacting API in a + later UI slice; +- aggregate acceptance evidence can be retained outside source control and + verified by digest. + +### Costs + +- operators must manage an external source-profile/query registry; +- actual data counts and digests must be computed before registration; +- service-specific diagnostic detail belongs in signed artifacts or service + traces, not arbitrary run-table JSON; +- a subsequent migration and UI slice must attach existing experimental run + evidence to this normalized model. + +## Security and privacy + +The design preserves authorized PII in the private source and product tables +where the business requires it; it does not blanket-mask operational data. +Instead, it minimizes broadcast and provenance surfaces. Audit output contains +only opaque IDs, hashes, aggregate counts, policy versions, clocks, and status. + +An idempotency, profile, or snapshot collision with different evidence raises a +content-redacting conflict. Source content never appears in exception text. The +product API must apply authenticated administrator authorization before exposing +run summaries. + +## Verification + +Acceptance requires: + +- production statement and branch coverage of the new Python modules at 100%; +- real PostgreSQL migration tests for normal insertion, temporal leakage + rejection, unique profile revisions, digest validation, and foreign keys; +- deterministic request digests across JSON key order; +- no source SQL, DSN, table name, raw content, image bytes, or artifact bytes in + public summaries; +- fresh-install Docker coverage of migration `0018`; +- exact-head product, security, SAST, and independent review gates; +- a private operator run that records only aggregate evidence and keeps its + acceptance manifest outside public source control. + +## Rollback + +Rollback stops new run registration and removes `0018` tables only after all +external artifacts are retained and no downstream foreign key depends on them. +It never copies private source SQL into an older log-based path. Supersession +requires another ADR preserving normalized provenance, no-future-information +enforcement, source redaction, and service ownership boundaries. From 121783b895f1fdabd9d47efe9d5bf99004ebe506 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:21:08 +0900 Subject: [PATCH 13/20] docs(doctoring): trace analysis provenance standards --- .../ANALYSIS_RUN_PROVENANCE_REFERENCES.md | 56 +++++++++++++++++++ 1 file changed, 56 insertions(+) create mode 100644 docs/doctoring/ANALYSIS_RUN_PROVENANCE_REFERENCES.md diff --git a/docs/doctoring/ANALYSIS_RUN_PROVENANCE_REFERENCES.md b/docs/doctoring/ANALYSIS_RUN_PROVENANCE_REFERENCES.md new file mode 100644 index 00000000..0422e094 --- /dev/null +++ b/docs/doctoring/ANALYSIS_RUN_PROVENANCE_REFERENCES.md @@ -0,0 +1,56 @@ +# Analysis-run provenance and temporal evidence references + +This doctoring note records the sources used by ADR 0013 and the Milestone 2 +analysis-run contract. Product code implements only the bounded provenance and +no-future-information gates described in the ADR; it does not claim to +implement every construct in these standards or the complete TEPP research +program. + +## Product and ecosystem sources + +ContextualWisdomLab. (2026a). *Temporal Event Psychometrics Platform: Product +requirements document v0.4 approved baseline* [Product requirements document]. +https://github.com/ContextualWisdomLab/TEPP + +ContextualWisdomLab. (2026b). *ADR 0017: Language-agnostic semantic-span and +embedding-budget authority* [Architecture decision record]. +https://github.com/ContextualWisdomLab/TEPP/blob/2294451d7827c3da47099f2593614c4964ec2e41/docs/adr/0017-language-agnostic-semantic-span-budgeting.md + +ContextualWisdomLab. (2026c). *Contextual Orchestrator architecture and product +requirements* [Software documentation]. +https://github.com/ContextualWisdomLab/contextual-orchestrator + +## Standards + +Cox, S., & Little, C. (Eds.). (2022). *Time ontology in OWL* (Candidate +Recommendation Draft). World Wide Web Consortium. +https://www.w3.org/TR/owl-time/ + +Lebo, T., Sahoo, S., & McGuinness, D. (Eds.). (2013). *PROV-O: The PROV +ontology*. World Wide Web Consortium. https://www.w3.org/TR/prov-o/ + +Moreau, L., & Missier, P. (Eds.). (2013). *PROV-DM: The PROV data model*. +World Wide Web Consortium. https://www.w3.org/TR/prov-dm/ + +OpenAPI Initiative. (2025). *OpenAPI specification (Version 3.2.0)*. +https://spec.openapis.org/oas/v3.2.0.html + +## Decision traceability + +| Contract decision | Source basis | +|---|---| +| One product derivation links source, run, service calls, and artifacts | PROV-DM; PROV-O | +| Occurrence and recording timestamps remain distinct | PROV-O generation/activity timing; TEPP multi-clock baseline | +| `maximum_available_time <= knowledge_cutoff` | Accepted TEPP temporal leakage contract | +| TEPP evidence authority is separate from LineageWeave product state | TEPP modular MSA and semantic-span ADRs | +| API output is a versioned source-redacting projection | OpenAPI 3.2.0; ADR 0013 security boundary | +| Exact source and request digests identify immutable evidence | PROV-O entity specialization; TEPP immutable evidence contract | + +## Scope note + +The product uses OWL-Time's instant/interval vocabulary and alignment with +PROV-O as a documented semantic reference. Database acceptance relies on +explicit PostgreSQL timestamps and constraints, not on an unsupported claim +that the complete OWL-Time ontology is implemented here. TEPP ADR 0017 remains +proposed at the cited exact commit; this LineageWeave slice records a compatible +boundary without presenting the future TEPP implementation as available. From c6a601972f2333d17c6e999d49e8cb51f7d1d9d5 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:21:22 +0900 Subject: [PATCH 14/20] docs(changelog): record Milestone 2 run provenance --- .../milestone2-analysis-run-contract.md | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) create mode 100644 CHANGELOG.d/milestone2-analysis-run-contract.md diff --git a/CHANGELOG.d/milestone2-analysis-run-contract.md b/CHANGELOG.d/milestone2-analysis-run-contract.md new file mode 100644 index 00000000..d14b6386 --- /dev/null +++ b/CHANGELOG.d/milestone2-analysis-run-contract.md @@ -0,0 +1,19 @@ +## Added + +- Added a normalized, aggregate-only provenance root for private PostgreSQL + analysis runs, downstream TEPP/contextual-orchestrator/fast-mlsirm service + calls, status events, and external acceptance artifacts. +- Added a no-future-information database and Python contract requiring the + source snapshot's latest evidence-availability time not to exceed the run's + knowledge cutoff. +- Added deterministic, source-redacting request digests and transactional + idempotency/conflict handling without storing SQL, DSNs, raw content, image + bytes, provider credentials, or private source identifiers. +- Added real-PostgreSQL schema tests and 100% statement/branch coverage for the + new Python contracts and repository. + +## Security + +- Private actual-data acceptance manifests remain external deployment + artifacts referenced only by digest and URI; public API projections contain + aggregate counts, clocks, opaque IDs, configuration, and status only. From 62a7943548b513414a4535449c1ce356d8c517a4 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:24:52 +0900 Subject: [PATCH 15/20] test(red): bound analysis completion lifecycle --- tests/test_analysis_run_completion.py | 72 +++++++++++++++++++++++++++ 1 file changed, 72 insertions(+) create mode 100644 tests/test_analysis_run_completion.py diff --git a/tests/test_analysis_run_completion.py b/tests/test_analysis_run_completion.py new file mode 100644 index 00000000..7aa6550e --- /dev/null +++ b/tests/test_analysis_run_completion.py @@ -0,0 +1,72 @@ +"""Lifecycle regressions for safe analysis-run completion and summaries.""" + +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest + +from lineageweave.analysis_run import ( + AnalysisRunCompletion, + AnalysisRunConfiguration, + AnalysisRunContractError, + AnalysisRunSummary, +) + +UTC = timezone.utc +DIGEST_A = "a" * 64 +DIGEST_B = "b" * 64 +STARTED = datetime(2026, 8, 15, 1, 0, tzinfo=UTC) +COMPLETED = datetime(2026, 8, 15, 2, 0, tzinfo=UTC) + + +def _configuration() -> AnalysisRunConfiguration: + return AnalysisRunConfiguration(0, True, True, True, "tepp-v1", "aggregate") + + +def _summary(**changes: object) -> AnalysisRunSummary: + values: dict[str, object] = { + "analysis_run_id": "run-id", + "profile_key": "configured-primary", + "profile_revision": 1, + "run_status_code": "analysis_run_running", + "request_digest_sha256": DIGEST_A, + "source_digest_sha256": DIGEST_B, + "knowledge_cutoff": STARTED, + "maximum_available_time": STARTED, + "row_count": 1, + "document_count": 1, + "thread_count": 1, + "started_at": STARTED, + "completed_at": None, + "configuration": _configuration(), + } + values.update(changes) + return AnalysisRunSummary(**values) # type: ignore[arg-type] + + +def test_completion_requires_safe_identifiers_boolean_and_aware_time() -> None: + completion = AnalysisRunCompletion("run-id", "account-id", True, COMPLETED) + assert completion.succeeded is True + for values, message in ( + ((" ", "account-id", True, COMPLETED), "analysis_run_id"), + (("run-id", " ", True, COMPLETED), "actor_account_id"), + (("run-id", "account-id", 1, COMPLETED), "succeeded"), + (("run-id", "account-id", True, datetime(2026, 8, 15)), "completed_at"), + ): + with pytest.raises(AnalysisRunContractError, match=message): + AnalysisRunCompletion(*values) # type: ignore[arg-type] + + +def test_summary_lifecycle_is_bounded_and_completion_consistent() -> None: + assert _summary().public_json()["completed_at"] is None + assert _summary( + run_status_code="analysis_run_succeeded", + completed_at=COMPLETED, + ).public_json()["run_status_code"] == "analysis_run_succeeded" + with pytest.raises(AnalysisRunContractError, match="supported run status"): + _summary(run_status_code="analysis_run_unknown") + with pytest.raises(AnalysisRunContractError, match="running run cannot"): + _summary(completed_at=COMPLETED) + with pytest.raises(AnalysisRunContractError, match="terminal run requires"): + _summary(run_status_code="analysis_run_failed") From 4fb249b5358c8794893d3fafb4f03be1b96377f0 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:25:12 +0900 Subject: [PATCH 16/20] test(red): reject unsafe completion before SQL --- backend/tests/test_analysis_run_completion.py | 52 +++++++++++++++++++ 1 file changed, 52 insertions(+) create mode 100644 backend/tests/test_analysis_run_completion.py diff --git a/backend/tests/test_analysis_run_completion.py b/backend/tests/test_analysis_run_completion.py new file mode 100644 index 00000000..15db5a69 --- /dev/null +++ b/backend/tests/test_analysis_run_completion.py @@ -0,0 +1,52 @@ +"""Repository input guards for analysis-run completion.""" + +from __future__ import annotations + +import asyncio +from datetime import datetime +from typing import Any + +import pytest + +from backend.app.analysis_run_ingestion import complete_analysis_run +from lineageweave.analysis_run import AnalysisRunContractError + + +class RejectTransactionConnection: + """Fail if invalid completion input reaches the database boundary.""" + + transaction_requested = False + + def transaction(self) -> Any: + self.transaction_requested = True + raise AssertionError("invalid completion must not open a transaction") + + +@pytest.mark.parametrize( + ("changes", "message"), + [ + ({"analysis_run_id": " "}, "analysis_run_id"), + ({"actor_account_id": " "}, "actor_account_id"), + ({"succeeded": 1}, "succeeded"), + ({"completed_at": datetime(2026, 8, 15)}, "completed_at"), + ], +) +def test_invalid_completion_fails_before_database_access( + changes: dict[str, object], message: str +) -> None: + values: dict[str, object] = { + "analysis_run_id": "run-id", + "actor_account_id": "account-id", + "succeeded": True, + "completed_at": datetime.fromisoformat("2026-08-15T02:00:00+00:00"), + } + values.update(changes) + connection = RejectTransactionConnection() + with pytest.raises(AnalysisRunContractError, match=message): + asyncio.run( + complete_analysis_run( + connection, # type: ignore[arg-type] + **values, # type: ignore[arg-type] + ) + ) + assert connection.transaction_requested is False From c159bd52f9cb2d3a871ce584907e32ed76392b84 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:26:03 +0900 Subject: [PATCH 17/20] test(red): reject unregistered source kinds --- tests/test_analysis_run_completion.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/tests/test_analysis_run_completion.py b/tests/test_analysis_run_completion.py index 7aa6550e..be20c977 100644 --- a/tests/test_analysis_run_completion.py +++ b/tests/test_analysis_run_completion.py @@ -11,6 +11,7 @@ AnalysisRunConfiguration, AnalysisRunContractError, AnalysisRunSummary, + SourceProfileReference, ) UTC = timezone.utc @@ -45,6 +46,16 @@ def _summary(**changes: object) -> AnalysisRunSummary: return AnalysisRunSummary(**values) # type: ignore[arg-type] +def test_source_profile_rejects_unknown_source_kind() -> None: + with pytest.raises(AnalysisRunContractError, match="supported source kind"): + SourceProfileReference( + "configured-primary", + 1, + DIGEST_A, + "filesystem_source_profile", + ) + + def test_completion_requires_safe_identifiers_boolean_and_aware_time() -> None: completion = AnalysisRunCompletion("run-id", "account-id", True, COMPLETED) assert completion.succeeded is True From e56bdf55c620c38c6fd29c480fa9a79c42cd74e2 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:26:50 +0900 Subject: [PATCH 18/20] feat: validate analysis completion lifecycle --- lineageweave/analysis_run.py | 48 ++++++++++++++++++++++++++++++++++++ 1 file changed, 48 insertions(+) diff --git a/lineageweave/analysis_run.py b/lineageweave/analysis_run.py index f070baa6..cdcf3817 100644 --- a/lineageweave/analysis_run.py +++ b/lineageweave/analysis_run.py @@ -18,6 +18,17 @@ _SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$") _PROFILE_KEY_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]{0,127}$") +_SOURCE_KIND_CODES = frozenset({"postgresql_query_profile"}) +_RUN_STATUS_CODES = frozenset( + { + "analysis_run_running", + "analysis_run_succeeded", + "analysis_run_failed", + } +) +_TERMINAL_RUN_STATUS_CODES = frozenset( + {"analysis_run_succeeded", "analysis_run_failed"} +) class AnalysisRunContractError(ValueError): @@ -110,6 +121,10 @@ def __post_init__(self) -> None: raise AnalysisRunContractError("profile_revision must be at least 1") _require_sha256("query_digest_sha256", self.query_digest_sha256) _require_nonempty("source_kind_code", self.source_kind_code, maximum=64) + if self.source_kind_code not in _SOURCE_KIND_CODES: + raise AnalysisRunContractError( + "source_kind_code must be a supported source kind" + ) def public_json(self) -> dict[str, Any]: """Return the source-safe profile projection exposed to operators.""" @@ -194,6 +209,25 @@ def __post_init__(self) -> None: _require_aware("started_at", self.started_at) +@dataclass(frozen=True) +class AnalysisRunCompletion: + """Validated terminal transition request for one running analysis.""" + + analysis_run_id: str + actor_account_id: str + succeeded: bool + completed_at: datetime + + def __post_init__(self) -> None: + """Reject unsafe identifiers, ambiguous booleans, and naïve clocks.""" + + _require_nonempty("analysis_run_id", self.analysis_run_id, maximum=128) + _require_nonempty("actor_account_id", self.actor_account_id, maximum=128) + if not isinstance(self.succeeded, bool): + raise AnalysisRunContractError("succeeded must be a boolean") + _require_aware("completed_at", self.completed_at) + + @dataclass(frozen=True) class AnalysisRunConfiguration: """Versioned, bounded configuration that controls one analysis run.""" @@ -283,6 +317,10 @@ def __post_init__(self) -> None: if self.profile_revision < 1: raise AnalysisRunContractError("profile_revision must be at least 1") _require_nonempty("run_status_code", self.run_status_code, maximum=64) + if self.run_status_code not in _RUN_STATUS_CODES: + raise AnalysisRunContractError( + "run_status_code must be a supported run status" + ) _require_sha256("request_digest_sha256", self.request_digest_sha256) SourceSnapshotEvidence( source_digest_sha256=self.source_digest_sha256, @@ -299,6 +337,16 @@ def __post_init__(self) -> None: raise AnalysisRunContractError( "completed_at must not precede started_at" ) + if self.run_status_code == "analysis_run_running": + if self.completed_at is not None: + raise AnalysisRunContractError( + "a running run cannot have completed_at" + ) + elif self.run_status_code in _TERMINAL_RUN_STATUS_CODES: + if self.completed_at is None: + raise AnalysisRunContractError( + "a terminal run requires completed_at" + ) def public_json(self) -> dict[str, Any]: """Return an API-safe summary with no DSN, SQL, raw content, or URI.""" From 8cdef93fe50d1467f6decaad5be029db127286f0 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:27:42 +0900 Subject: [PATCH 19/20] feat: guard completion input before PostgreSQL --- backend/app/analysis_run_ingestion.py | 25 ++++++++++++++++++------- 1 file changed, 18 insertions(+), 7 deletions(-) diff --git a/backend/app/analysis_run_ingestion.py b/backend/app/analysis_run_ingestion.py index f6fad803..92a6abbf 100644 --- a/backend/app/analysis_run_ingestion.py +++ b/backend/app/analysis_run_ingestion.py @@ -13,6 +13,7 @@ from typing import Any, Mapping, Protocol, Sequence from lineageweave.analysis_run import ( + AnalysisRunCompletion, AnalysisRunConfiguration, AnalysisRunRegistration, AnalysisRunSummary, @@ -197,10 +198,20 @@ async def complete_analysis_run( ) -> None: """Complete a running record exactly once and append its audit event.""" - status_code = "analysis_run_succeeded" if succeeded else "analysis_run_failed" + completion = AnalysisRunCompletion( + analysis_run_id=analysis_run_id, + actor_account_id=actor_account_id, + succeeded=succeeded, + completed_at=completed_at, + ) + status_code = ( + "analysis_run_succeeded" + if completion.succeeded + else "analysis_run_failed" + ) event_type_code = ( "analysis_run_completed_event" - if succeeded + if completion.succeeded else "analysis_run_failed_event" ) async with conn.transaction(): @@ -213,9 +224,9 @@ async def complete_analysis_run( and completed_at is null returning request_digest_sha256 """, - analysis_run_id, + completion.analysis_run_id, status_code, - completed_at, + completion.completed_at, ) if run_row is None: raise AnalysisRunConflict("analysis run is missing or already completed") @@ -226,10 +237,10 @@ async def complete_analysis_run( occurred_at, payload_digest_sha256 ) values ($1::uuid, $2, $3::uuid, $4, $5) """, - analysis_run_id, + completion.analysis_run_id, event_type_code, - actor_account_id, - completed_at, + completion.actor_account_id, + completion.completed_at, run_row["request_digest_sha256"], ) From 30b67c3fdeb932591ec4a743f9f6e0c6986341d4 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Sat, 15 Aug 2026 20:34:04 +0900 Subject: [PATCH 20/20] refactor: remove unreachable lifecycle branch --- lineageweave/analysis_run.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/lineageweave/analysis_run.py b/lineageweave/analysis_run.py index cdcf3817..469b3326 100644 --- a/lineageweave/analysis_run.py +++ b/lineageweave/analysis_run.py @@ -26,9 +26,6 @@ "analysis_run_failed", } ) -_TERMINAL_RUN_STATUS_CODES = frozenset( - {"analysis_run_succeeded", "analysis_run_failed"} -) class AnalysisRunContractError(ValueError): @@ -342,7 +339,7 @@ def __post_init__(self) -> None: raise AnalysisRunContractError( "a running run cannot have completed_at" ) - elif self.run_status_code in _TERMINAL_RUN_STATUS_CODES: + else: if self.completed_at is None: raise AnalysisRunContractError( "a terminal run requires completed_at"