diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 13f0a4599..3f3a7cc9a 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -454,6 +454,26 @@ lists the same dated tickets the period-report members already show. Re-seed is idempotent. The empty-state copy is only for accounts that truly have no dated open tickets. +## Phase 6-M2: authorized analysis-run evidence (read projection) + +Issue #79's first buyer-visible Milestone 2 slice is a source-redacting +read of the #89 registry. `GET /api/analysis-runs` and +`GET /api/analysis-runs/{id}` require `post_read` and apply the scope +in SQL: the requester always sees their own run; a corporate-entity or +process-unit scope is visible only to affiliated accounts; a +thread-group scope is visible only when the account can already see a +post in that group; `all_visible` is requester-only. Hidden runs 404. Detail also lists ABAC-visible post titles in the +run's scope so a buyer can open a post without seeing hidden rows. +The home list is clickable: `GET /api/analysis-runs/{id}` fills a +labeled detail (cutoff, requested date, counts, status history) +without exposing a DSN or raw record. Status history is detail-only +and uses lookup labels plus occurrence times; a failure event keeps +its machine `failure_code` rather than an invented caption. The +payload is lookup labels plus non-negative aggregate counts -- never +source SQL, a DSN, a raw record, or a provider body. After `make seed`, +Demo Analyst and Demo Admin see "Lineage reconstruction · Succeeded · +Demo Corp" with "3 documents" and Pending / Running / Succeeded times. + ## Phase 6a: fast-mlsirm dependency + Rust toolchain (infra only) First of three staged slices toward the brief's weekly/monthly diff --git a/CHANGELOG.d/0.79.0-analysis-run-authorized-read.md b/CHANGELOG.d/0.79.0-analysis-run-authorized-read.md new file mode 100644 index 000000000..7233020b2 --- /dev/null +++ b/CHANGELOG.d/0.79.0-analysis-run-authorized-read.md @@ -0,0 +1,9 @@ +# 0.79.0 — Authorized analysis-run read projection + +## Added + +- `GET /api/analysis-runs` and `GET /api/analysis-runs/{id}` expose + source-redacting registry evidence to `post_read` accounts. +- Home-page Analysis runs panel shows the seeded Demo Corp lineage run + after `make seed`. Hidden scopes 404. No raw source, DSN, or provider + payload is returned. diff --git a/CHANGELOG.d/0.80.0-analysis-run-detail-click.md b/CHANGELOG.d/0.80.0-analysis-run-detail-click.md new file mode 100644 index 000000000..090ebb824 --- /dev/null +++ b/CHANGELOG.d/0.80.0-analysis-run-detail-click.md @@ -0,0 +1,5 @@ +# 0.80.0 analysis-run detail click + +Home Analysis runs rows open `GET /api/analysis-runs/{id}`. The +detail shows labeled aggregates and dates only. Hidden runs stay +404 / "not visible". Synthetic Demo Corp seed only. diff --git a/CHANGELOG.d/0.81.0-analysis-run-status-history.md b/CHANGELOG.d/0.81.0-analysis-run-status-history.md new file mode 100644 index 000000000..9fa4b8706 --- /dev/null +++ b/CHANGELOG.d/0.81.0-analysis-run-status-history.md @@ -0,0 +1,5 @@ +# 0.81.0 analysis-run status history + +Detail of `GET /api/analysis-runs/{id}` shows the labeled append-only +lifecycle. The list stays latest-status only. Hidden runs 404. +Synthetic Demo Corp seed only. diff --git a/CHANGELOG.d/0.82.0-analysis-run-post-clickthrough.md b/CHANGELOG.d/0.82.0-analysis-run-post-clickthrough.md new file mode 100644 index 000000000..1802a5644 --- /dev/null +++ b/CHANGELOG.d/0.82.0-analysis-run-post-clickthrough.md @@ -0,0 +1,4 @@ +# 0.82.0 analysis-run post click-through + +Detail lists ABAC-visible post titles in the run scope. Hidden +other-corp private posts never appear. Synthetic titles only. diff --git a/CHANGELOG.d/milestone2-analysis-run-registry.md b/CHANGELOG.d/milestone2-analysis-run-registry.md new file mode 100644 index 000000000..5d7e0288f --- /dev/null +++ b/CHANGELOG.d/milestone2-analysis-run-registry.md @@ -0,0 +1,21 @@ +## Added + +- Added a normalized PostgreSQL registry for immutable source snapshots, + aggregate reconciliation counts, authenticated analysis requests, product + scopes, and append-only lifecycle evidence. +- Added a run-owned knowledge cutoff and snapshot-owned evidence-availability + clock so one capture can support multiple historically valid analyses without + future-information leakage. +- Added canonical account-scoped idempotency, immutable request and scope evidence, + deletion resistance, serialized count/run locking, scope-required request-time- + ordered lifecycle transitions, database-owned record time, and a derived + current-status view. +- Added fail-closed rollback, real-PostgreSQL contract tests, ADR 0013, and APA + 7th standards traceability. + +## Security + +- The registry deliberately excludes source SQL, DSNs, raw records, inline + images, provider payloads, credentials, private source identifiers, and raw + exceptions. Necessary PII remains in purpose-bound authorized product/source + contexts rather than being copied into audit metadata or blanket-masked. diff --git a/CHANGELOG.md b/CHANGELOG.md index dc3db16b2..6ac2df4ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,47 @@ All notable changes to this project are documented here. Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); versioning follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.82.0] - 2026-08-16 + +### Added + +- Analysis-run detail lists ABAC-visible posts in the run's scope. + After `make seed`, the Demo Corp lineage run opens the Demo public + post. Hidden other-corp private posts never appear. List payloads + stay aggregates-only. + +## [0.81.0] - 2026-08-16 + +### Added + +- Analysis-run detail shows the labeled lifecycle: Pending, Running, + then Succeeded, with occurrence times from `analysis_run_status_event`. + The list stays latest-status only. Hidden runs still 404 and never + leak events. Failure codes stay machine tokens -- no invented label. + Synthetic Demo Corp seed only. + +## [0.80.0] - 2026-08-16 + +### Added + +- Home Analysis runs rows are buttons. Clicking the seeded Demo Corp + lineage run opens `GET /api/analysis-runs/{id}` and shows cutoff, + requested date, and document count. A hidden run is "This analysis + run is not visible." -- never a raw 404 or a DSN. Still synthetic + aggregates only. + +## [0.79.0] - 2026-08-16 + +### Added + +- Authorized analysis-run evidence on the product home page. After + `make seed`, Demo Analyst sees "Lineage reconstruction · Succeeded · + Demo Corp" with the synthetic document count. `GET /api/analysis-runs` + is scoped in SQL: another tenant's run 404s and never appears in the + list. The payload is labels and aggregates -- never source SQL, a DSN, + or a raw record. TEPP stays behind `tepp_client`; Null channels are + unchanged. + ## [0.78.0] - 2026-08-15 ### Changed @@ -19,7 +60,7 @@ All notable changes to this project are documented here. Format follows and LLM-as-a-Judge consumers now request contextual-orchestrator `auto` mode so the orchestration plane can meet the quality requirement and then minimize known execution cost. Explicit checked `verify` paths remain unchanged - (ADR 0013). + (ADR 0015). ## [0.77.0] - 2026-08-14 diff --git a/backend/app/analysis_run_ingestion.py b/backend/app/analysis_run_ingestion.py new file mode 100644 index 000000000..c18ddf326 --- /dev/null +++ b/backend/app/analysis_run_ingestion.py @@ -0,0 +1,277 @@ +"""Authorized, source-redacting reads of the Milestone 2 analysis-run registry. + +The registry itself is issue #89 / migration 0018. This module is the +product projection: an account sees only runs they requested or whose +scope they already have ABAC authority to walk. Aggregate counts and +lookup labels come back; source SQL, DSNs, raw records, and provider +payloads never do. +""" + +from __future__ import annotations + +from typing import Any + +import asyncpg + +from backend.app.knowledge_graph import labels_for_codes + +_VISIBLE_RUN_SQL = """ + run.requested_by_account_id = $1 + or ( + scope.scope_kind_code = 'analysis_scope_corporate_entity' + and scope.corporate_entity_id = any($2::uuid[]) + ) + or ( + scope.scope_kind_code = 'analysis_scope_process_unit' + and exists ( + select 1 from account_affiliation aff + where aff.user_account_id = $1 + and aff.process_unit_id = scope.process_unit_id + ) + ) + or ( + scope.scope_kind_code = 'analysis_scope_thread_group' + and exists ( + select 1 from source_post p + where p.thread_group_key = scope.scope_key + and ( + p.visibility_code = 'public' + or p.corporate_entity_id = any($2::uuid[]) + ) + ) + ) +""" + +_RUN_SELECT = f""" + select + run.analysis_run_id, + run.run_kind_code, + run.knowledge_cutoff, + run.requested_at, + run.configuration_schema_version, + run.configuration_sha256, + run.code_revision_sha, + scope.scope_kind_code, + scope.corporate_entity_id, + scope.process_unit_id, + scope.scope_key, + corp.entity_name as scope_entity_name, + status.status_code, + status.failure_code + from analysis_run run + join analysis_run_scope scope on scope.analysis_run_id = run.analysis_run_id + left join analysis_run_current_status status + on status.analysis_run_id = run.analysis_run_id + left join corporate_entity corp + on corp.corporate_entity_id = scope.corporate_entity_id + where {{where}} + order by run.requested_at desc +""" + + +def _iso(value: Any) -> str: + """Serialize a timestamptz the same way post payloads do.""" + return value.isoformat() if hasattr(value, "isoformat") else str(value) + + +async def _counts_by_run( + conn: asyncpg.Connection, + run_ids: list[str], +) -> dict[str, list[asyncpg.Record]]: + """Load aggregate snapshot counts for the given runs.""" + if not run_ids: + return {} + rows = await conn.fetch( + """ + select run.analysis_run_id, counts.count_type_code, counts.count_value + from analysis_run run + join analysis_source_count counts + on counts.analysis_source_snapshot_id = run.analysis_source_snapshot_id + where run.analysis_run_id = any($1::uuid[]) + order by counts.count_type_code + """, + run_ids, + ) + grouped: dict[str, list[asyncpg.Record]] = {} + for row in rows: + grouped.setdefault(str(row["analysis_run_id"]), []).append(row) + return grouped + + +async def _status_history( + conn: asyncpg.Connection, + analysis_run_id: str, +) -> list[dict[str, Any]]: + """Labeled append-only lifecycle for one already-visible run.""" + rows = await conn.fetch( + """ + select status_ordinal, status_code, occurred_at, failure_code + from analysis_run_status_event + where analysis_run_id = $1::uuid + order by status_ordinal + """, + analysis_run_id, + ) + labels = await labels_for_codes(conn, [row["status_code"] for row in rows]) + history: list[dict[str, Any]] = [] + for row in rows: + item: dict[str, Any] = { + "status_ordinal": int(row["status_ordinal"]), + "status_code": row["status_code"], + "status_label": labels.get(row["status_code"], row["status_code"]), + "occurred_at": _iso(row["occurred_at"]), + } + if row["failure_code"]: + item["failure_code"] = row["failure_code"] + history.append(item) + return history + + +async def _serialize_runs( + conn: asyncpg.Connection, + rows: list[asyncpg.Record], +) -> list[dict[str, Any]]: + """Project registry rows into the authorized buyer-facing payload.""" + if not rows: + return [] + count_rows = await _counts_by_run(conn, [str(row["analysis_run_id"]) for row in rows]) + labels = await labels_for_codes( + conn, + [row["run_kind_code"] for row in rows] + + [row["scope_kind_code"] for row in rows] + + [row["status_code"] for row in rows if row["status_code"]] + + [ + count["count_type_code"] + for counts in count_rows.values() + for count in counts + ], + ) + payload: list[dict[str, Any]] = [] + for row in rows: + run_id = str(row["analysis_run_id"]) + kind = row["run_kind_code"] + scope = row["scope_kind_code"] + status = row["status_code"] + item: dict[str, Any] = { + "analysis_run_id": run_id, + "run_kind_code": kind, + "run_kind_label": labels.get(kind, kind), + "scope_kind_code": scope, + "scope_kind_label": labels.get(scope, scope), + "status_code": status, + "status_label": labels.get(status, status) if status else None, + "knowledge_cutoff": _iso(row["knowledge_cutoff"]), + "requested_at": _iso(row["requested_at"]), + "source_counts": [ + { + "count_type_code": count["count_type_code"], + "count_type_label": labels.get( + count["count_type_code"], count["count_type_code"] + ), + "count_value": int(count["count_value"]), + } + for count in count_rows.get(run_id, []) + ], + } + if row["scope_entity_name"]: + item["scope_entity_name"] = row["scope_entity_name"] + payload.append(item) + return payload + + +async def fetch_visible_analysis_runs( + conn: asyncpg.Connection, + account_id: str, + affiliated_entity_ids: list[str], +) -> list[dict[str, Any]]: + """Runs the account requested or whose scope they may already walk.""" + rows = await conn.fetch( + _RUN_SELECT.format(where=_VISIBLE_RUN_SQL), + account_id, + affiliated_entity_ids, + ) + return await _serialize_runs(conn, rows) + + +async def fetch_visible_analysis_run( + conn: asyncpg.Connection, + analysis_run_id: str, + account_id: str, + affiliated_entity_ids: list[str], +) -> dict[str, Any] | None: + """One visible run, or None when it is missing or hidden.""" + rows = await conn.fetch( + _RUN_SELECT.format( + where=f"run.analysis_run_id = $3 and ({_VISIBLE_RUN_SQL})" + ), + account_id, + affiliated_entity_ids, + analysis_run_id, + ) + payload = await _serialize_runs(conn, rows) + if not payload: + return None + detail = payload[0] + row = rows[0] + detail["configuration_schema_version"] = row["configuration_schema_version"] + detail["configuration_sha256"] = row["configuration_sha256"] + detail["code_revision_sha"] = row["code_revision_sha"] + if row["failure_code"]: + detail["failure_code"] = row["failure_code"] + detail["status_history"] = await _status_history(conn, analysis_run_id) + detail["visible_posts"] = await fetch_visible_scope_posts( + conn, + row["scope_kind_code"], + row["corporate_entity_id"], + row["process_unit_id"], + row["scope_key"], + affiliated_entity_ids, + ) + return detail + + +async def fetch_visible_scope_posts( + conn: asyncpg.Connection, + scope_kind_code: str, + corporate_entity_id: Any, + process_unit_id: Any, + scope_key: str | None, + affiliated_entity_ids: list[str], +) -> list[dict[str, str]]: + """ABAC-visible post titles in the run's scope -- never a hidden body.""" + if scope_kind_code == "analysis_scope_corporate_entity" and corporate_entity_id: + rows = await conn.fetch( + "select post_id, post_title, visibility_code, corporate_entity_id " + "from source_post where corporate_entity_id = $1 " + "order by created_at, post_title", + corporate_entity_id, + ) + elif scope_kind_code == "analysis_scope_process_unit" and process_unit_id: + rows = await conn.fetch( + "select post_id, post_title, visibility_code, corporate_entity_id " + "from source_post where process_unit_id = $1 " + "order by created_at, post_title", + process_unit_id, + ) + elif scope_kind_code == "analysis_scope_thread_group" and scope_key: + rows = await conn.fetch( + "select post_id, post_title, visibility_code, corporate_entity_id " + "from source_post where thread_group_key = $1 " + "order by created_at, post_title", + scope_key, + ) + elif scope_kind_code == "analysis_scope_all_visible": + rows = await conn.fetch( + "select post_id, post_title, visibility_code, corporate_entity_id " + "from source_post order by created_at, post_title" + ) + else: + return [] + affiliated = {str(entity_id) for entity_id in affiliated_entity_ids} + posts: list[dict[str, str]] = [] + for row in rows: + visible = row["visibility_code"] == "public" or str(row["corporate_entity_id"]) in affiliated + if not visible: + continue + posts.append({"post_id": str(row["post_id"]), "post_title": row["post_title"]}) + return posts diff --git a/backend/app/main.py b/backend/app/main.py index c0214f0c0..e039a2f58 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -22,6 +22,7 @@ import asyncio from contextlib import asynccontextmanager from typing import Any +from uuid import UUID import asyncpg import redis.asyncio as redis @@ -64,6 +65,10 @@ from lineageweave.post_summary import ContextualOrchestratorPostSummaryClient, NullPostSummaryClient from lineageweave.relation_verification import NullRelationVerificationClient, SearxngRelationVerificationClient +from backend.app.analysis_run_ingestion import ( + fetch_visible_analysis_run, + fetch_visible_analysis_runs, +) from backend.app.activity_stream import ( create_valkey_client, get_valkey, @@ -1148,6 +1153,53 @@ async def derive_post_commitment( return {"post_id": str(post["post_id"]), "has_commitment": True, "ticket": ticket} +@app.get("/api/analysis-runs") +async def list_analysis_runs( + account: CurrentAccount = Depends(get_current_account), + pool: asyncpg.Pool = Depends(get_pool), +) -> dict[str, Any]: + """Authorized analysis-run list: aggregates and labels only. + + Hidden scopes 404 at the item path and never appear here. The + payload has no source SQL, DSN, raw record, or provider body. + """ + _require_post_read(account) + async with pool.acquire() as conn: + runs = await fetch_visible_analysis_runs( + conn, + account.user_account_id, + list(account.corporate_entity_ids), + ) + return {"analysis_runs": runs} + + +@app.get("/api/analysis-runs/{analysis_run_id}") +async def read_analysis_run( + analysis_run_id: str, + account: CurrentAccount = Depends(get_current_account), + pool: asyncpg.Pool = Depends(get_pool), +) -> dict[str, Any]: + """One authorized analysis-run projection, or 404 when hidden. + + Detail adds the labeled status history. Hidden runs never leak events. + """ + _require_post_read(account) + try: + UUID(analysis_run_id) + except ValueError: + raise HTTPException(status.HTTP_404_NOT_FOUND, "analysis run not found") from None + async with pool.acquire() as conn: + run = await fetch_visible_analysis_run( + conn, + analysis_run_id, + account.user_account_id, + list(account.corporate_entity_ids), + ) + if run is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, "analysis run not found") + return run + + @app.get("/api/calendar") async def read_calendar( account: CurrentAccount = Depends(get_current_account), diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index 62c32bcf0..cfc2a5559 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -30,6 +30,7 @@ _VALKEY_URL = os.environ.get("LINEAGEWEAVE_TEST_VALKEY_URL", "redis://localhost:16379/0") _REALM = "lineageweave-demo" _MIGRATION_PATH = Path(__file__).resolve().parents[2] / "migrations" / "0001_initial_schema.sql" +_REGISTRY_MIGRATION = Path(__file__).resolve().parents[2] / "migrations" / "0018_analysis_run_registry.sql" def _postgres_available() -> bool: @@ -112,6 +113,7 @@ def seeded_db(demo_analyst_token): try: with conn.cursor() as cur: cur.execute(_MIGRATION_PATH.read_text()) + cur.execute(_REGISTRY_MIGRATION.read_text()) cur.execute( "insert into common_lookup_value (lookup_category, lookup_code, lookup_label) values " "('corporate_entity_level', 'group', 'Group'), " @@ -187,6 +189,108 @@ def seeded_db(demo_analyst_token): "insert into role_permission (access_role_id, permission_code) values (%s, 'post_read')", (role_id,), ) + + def _seed_analysis_run( + digest: str, + idempotency_key: str, + requester_id, + scope_kind: str, + corp_id=None, + ) -> str: + cur.execute( + """ + insert into analysis_source_snapshot + (snapshot_sha256, source_contract_version, + maximum_available_time, captured_at) + values (%s, 'source-contract-v1', + '2026-01-12T00:00:00Z', '2026-01-12T00:05:00Z') + returning analysis_source_snapshot_id + """, + (digest,), + ) + snapshot_id = cur.fetchone()[0] + cur.execute( + """ + insert into analysis_source_count + (analysis_source_snapshot_id, count_type_code, count_value) + values (%s, 'analysis_count_document', 3) + """, + (snapshot_id,), + ) + cur.execute( + """ + insert into analysis_run + (analysis_source_snapshot_id, run_kind_code, idempotency_key, + requested_by_account_id, knowledge_cutoff, + configuration_schema_version, configuration_sha256, + code_revision_sha, requested_at) + values (%s, 'analysis_run_lineage', %s, %s, + '2026-01-12T12:00:00Z', 'lineage-run-v1', %s, %s, + '2026-01-12T12:30:00Z') + returning analysis_run_id + """, + (snapshot_id, idempotency_key, requester_id, "b" * 64, "c" * 40), + ) + run_id = str(cur.fetchone()[0]) + if scope_kind == "analysis_scope_corporate_entity": + cur.execute( + """ + insert into analysis_run_scope + (analysis_run_id, scope_kind_code, corporate_entity_id) + values (%s, %s, %s) + """, + (run_id, scope_kind, corp_id), + ) + else: + cur.execute( + """ + insert into analysis_run_scope + (analysis_run_id, scope_kind_code) + values (%s, %s) + """, + (run_id, scope_kind), + ) + for ordinal, status, occurred in ( + (1, "analysis_status_pending", "2026-01-12T12:31:00Z"), + (2, "analysis_status_running", "2026-01-12T12:32:00Z"), + (3, "analysis_status_succeeded", "2026-01-12T12:33:00Z"), + ): + cur.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, %s, %s, %s) + """, + (run_id, ordinal, status, occurred), + ) + return run_id + + cur.execute( + "insert into user_account (external_subject_id, display_name, email_address) " + "values (%s, 'Other Analyst', 'other.analyst@example.test') returning user_account_id", + (f"other-{uuid.uuid4()}",), + ) + other_account_id = cur.fetchone()[0] + visible_run_id = _seed_analysis_run( + "a" * 64, + "visible-own-corp", + account_id, + "analysis_scope_corporate_entity", + own_corp_id, + ) + hidden_run_id = _seed_analysis_run( + "d" * 64, + "hidden-other-corp", + other_account_id, + "analysis_scope_corporate_entity", + other_corp_id, + ) + hidden_all_visible_id = _seed_analysis_run( + "e" * 64, + "hidden-all-visible", + other_account_id, + "analysis_scope_all_visible", + ) cur.execute( "insert into account_role_assignment (user_account_id, access_role_id) values (%s, %s)", (account_id, role_id), @@ -298,6 +402,9 @@ def _insert_post(title: str, corporate_entity_id, visibility_code: str, body: st "our_person_id": our_person_id, "counterpart_person_id": counterpart_person_id, "hidden_person_id": hidden_person_id, + "visible_run_id": visible_run_id, + "hidden_run_id": hidden_run_id, + "hidden_all_visible_id": hidden_all_visible_id, } finally: conn.close() @@ -320,6 +427,70 @@ def client(seeded_db): yield test_client +def test_analysis_runs_are_labeled_aggregates_and_hide_other_scopes( + client, demo_analyst_token, seeded_db +) -> None: + """Demo analyst sees the Test Corp run, never the Other Corp or outsider run.""" + listed = client.get("/api/analysis-runs", headers={"Authorization": f"Bearer {demo_analyst_token}"}) + assert listed.status_code == 200 + runs = listed.json()["analysis_runs"] + ids = {run["analysis_run_id"] for run in runs} + assert seeded_db["visible_run_id"] in ids + assert seeded_db["hidden_run_id"] not in ids + assert seeded_db["hidden_all_visible_id"] not in ids + visible = next(run for run in runs if run["analysis_run_id"] == seeded_db["visible_run_id"]) + assert visible["run_kind_label"] == "Lineage reconstruction" + assert visible["status_label"] == "Succeeded" + assert visible["scope_kind_label"] == "Corporate entity" + assert visible["scope_entity_name"] == "Test Corp" + assert visible["source_counts"] == [ + { + "count_type_code": "analysis_count_document", + "count_type_label": "Documents", + "count_value": 3, + } + ] + dumped = str(visible) + assert "postgresql://" not in dumped + assert "select " not in dumped.lower() + assert "status_history" not in visible + + detail = client.get( + f"/api/analysis-runs/{seeded_db['visible_run_id']}", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert detail.status_code == 200 + body = detail.json() + assert body["configuration_schema_version"] == "lineage-run-v1" + assert "snapshot_sha256" not in body + history = body["status_history"] + assert [event["status_label"] for event in history] == [ + "Pending", + "Running", + "Succeeded", + ] + assert [event["occurred_at"][:16] for event in history] == [ + "2026-01-12T12:31", + "2026-01-12T12:32", + "2026-01-12T12:33", + ] + assert all("failure_code" not in event for event in history) + titles = {post["post_title"] for post in body["visible_posts"]} + assert "Own-corp private post" in titles + assert "Other-corp private post" not in titles + assert "postgresql://" not in str(body) + assert "visible_posts" not in visible + + hidden = client.get( + f"/api/analysis-runs/{seeded_db['hidden_run_id']}", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert hidden.status_code == 404 + + unauthenticated = client.get("/api/analysis-runs") + assert unauthenticated.status_code == 401 + + def test_me_reflects_the_authenticated_account(client, demo_analyst_token) -> None: response = client.get("/api/me", headers={"Authorization": f"Bearer {demo_analyst_token}"}) assert response.status_code == 200 diff --git a/docker/postgres-init/Dockerfile b/docker/postgres-init/Dockerfile index d10dec64b..d95e2c917 100644 --- a/docker/postgres-init/Dockerfile +++ b/docker/postgres-init/Dockerfile @@ -3,10 +3,9 @@ FROM postgres:16-alpine@sha256:57c72fd2a128e416c7fcc499958864df5301e940bca0a56f5 # Keycloak-database bootstrap and the product schema can be copied from # their single sources of truth. COPY docker/postgres-init/01-create-keycloak-db.sql /docker-entrypoint-initdb.d/01-create-keycloak-db.sql -# The exact same migration files tests/test_schema.py applies -- single -# source of truth, no re-typed copy. Runs against POSTGRES_DB (the "app" -# database) because docker-entrypoint-initdb.d executes each *.sql file -# with that database already selected. +# The exact same migration files the PostgreSQL contract tests apply -- single +# source of truth, no re-typed copy. docker-entrypoint-initdb.d executes each +# file against POSTGRES_DB in lexical order. COPY migrations/0001_initial_schema.sql /docker-entrypoint-initdb.d/02-app-schema.sql COPY migrations/0002_thread_grouping_keys.sql /docker-entrypoint-initdb.d/03-thread-grouping-keys.sql COPY migrations/0003_ticket_commitment_calendar.sql /docker-entrypoint-initdb.d/04-ticket-commitment-calendar.sql @@ -24,6 +23,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_registry.sql /docker-entrypoint-initdb.d/19-analysis-run-registry.sql # Official image already drops to this account at runtime; declare it so # the Dockerfile itself satisfies DS-0002 (explicit non-root USER). USER postgres diff --git a/docs/adr/0013-normalized-analysis-run-registry.md b/docs/adr/0013-normalized-analysis-run-registry.md new file mode 100644 index 000000000..1d5a59866 --- /dev/null +++ b/docs/adr/0013-normalized-analysis-run-registry.md @@ -0,0 +1,266 @@ +# ADR 0013 — Milestone 2 uses a normalized, additive analysis-run registry + +**Decision status:** Accepted on this active PR; not protected-main truth until merge +**Date:** 2026-08-15 +**Depends on:** ADR 0011 standards-complete provenance separation and ADR 0012 corporate-entity creation locking + +## Context + +LineageWeave has a reviewed React/FastAPI/PostgreSQL product, compact lineage +navigation, normalized actor identity, report persistence, and a separate +standards-complete PROV-O layer. Milestone 2 must analyze operator-authorized +PostgreSQL evidence without replacing that product, duplicating cross-service +databases, or committing private source identity and content to a public +repository. + +A retained experiment proved that direct PostgreSQL analysis is feasible, but +its parallel application and denormalized run record cannot become product +truth. The product needs a small durable root that answers: + +- which immutable capture was used; +- which evidence was available by the run's knowledge cutoff; +- which authenticated account requested the work; +- which product scope and reproducibility digests governed the run; +- which aggregate counts reconcile the capture; +- which legal lifecycle transitions occurred. + +The registry does not store source SQL, DSNs, raw posts, inline images, provider +payloads, credentials, raw exceptions, or another service's application rows. + +## Decision + +Migration `0018_analysis_run_registry.sql` introduces five normalized relations +and one read projection. + +```mermaid +erDiagram + ANALYSIS_SOURCE_SNAPSHOT ||--o{ ANALYSIS_SOURCE_COUNT : reconciles + ANALYSIS_SOURCE_SNAPSHOT ||--o{ ANALYSIS_RUN : anchors + USER_ACCOUNT ||--o{ ANALYSIS_RUN : requests + ANALYSIS_RUN ||--o| ANALYSIS_RUN_SCOPE : limits + CORPORATE_ENTITY |o--o{ ANALYSIS_RUN_SCOPE : scopes + PROCESS_UNIT |o--o{ ANALYSIS_RUN_SCOPE : scopes + ANALYSIS_RUN ||--o{ ANALYSIS_RUN_STATUS_EVENT : records + + ANALYSIS_SOURCE_SNAPSHOT { + uuid analysis_source_snapshot_id PK + text snapshot_sha256 UK + text source_contract_version + timestamptz maximum_available_time + timestamptz captured_at + } + ANALYSIS_SOURCE_COUNT { + uuid analysis_source_snapshot_id PK,FK + text count_type_code PK,FK + bigint count_value + } + ANALYSIS_RUN { + uuid analysis_run_id PK + uuid analysis_source_snapshot_id FK + uuid requested_by_account_id FK + text idempotency_key UK + timestamptz knowledge_cutoff + text configuration_sha256 + text model_contract_sha256 + text prompt_bundle_sha256 + text code_revision_sha + } + ANALYSIS_RUN_SCOPE { + uuid analysis_run_id PK,FK + text scope_kind_code FK + uuid corporate_entity_id FK + uuid process_unit_id FK + text scope_key + } + ANALYSIS_RUN_STATUS_EVENT { + uuid analysis_run_id PK,FK + int status_ordinal PK + text status_code FK + timestamptz occurred_at + timestamptz recorded_at + text failure_code + boolean retryable + } +``` + +### Temporal ownership + +`analysis_source_snapshot.maximum_available_time` is an evidence fact: the +latest time at which any admitted fact became available. `analysis_run.knowledge_cutoff` +is an analysis fact: the latest information that this particular run may use. +A reusable capture therefore does **not** own one knowledge cutoff. + +Run creation locks the snapshot and requires: + +```text +maximum_available_time <= knowledge_cutoff <= requested_at +captured_at <= requested_at +``` + +This aggregate guard complements TEPP's finer event, assertion, document, +system, availability, and cutoff clocks. It does not replace TEPP temporal or +psychometric computation. + +### Identity and idempotency + +Every run references a real `user_account`. `requested_by_account_id` is not +nullable. Idempotency keys are trimmed, control-free canonical values and are +unique per authenticated account rather than +globally, because independent callers may legitimately choose the same opaque +client key. A later repository must compare request digests on retry and return +a conflict when the same account/key names different evidence or configuration. + +### Immutability and concurrency + +Snapshot identity and availability reject updates. Aggregate count values reject +updates. Count insert/delete and first run creation acquire the same snapshot-row +lock before checking whether a run exists. This shared lock order closes the +race in which a count set and first derivation could otherwise both commit. +After the first run, the complete count set is frozen. + +The analysis request and its authorization scope reject updates and deletes. +Lifecycle changes are represented only by append-only status events, so a cascade +cannot erase the derivation root or its access boundary. + +### Lifecycle state machine + +The parent run row serializes status appends. Events require contiguous +ordinals, monotonic occurrence time, and these transitions: + +```text +pending -> running | cancelled +running -> succeeded | failed | cancelled +succeeded | failed | cancelled -> terminal +``` + +The first event must be `pending`, requires an immutable scope, and cannot predate +the run request. Failed events require a lowercase machine-code identifier; raw +exception text is prohibited. `recorded_at` is overwritten with database system +time on every insert and cannot precede `occurred_at`. +`analysis_run_current_status` is a view, not a second mutable state authority. + +### Authorization scope + +`analysis_run_scope` stores one immutable all-visible, corporate-entity, +process-unit, or thread-group scope. Its shape is database constrained and the +first lifecycle event is rejected until it exists. The next repository/API slice +must insert run, scope, and first status in one +transaction and apply the existing RBAC/ABAC contract when listing or reading +runs. This migration does not claim that an API or UI exists. + +### Service boundaries + +- **LineageWeave** owns product run identity, authorized scope, lifecycle, + aggregate reconciliation, and product-visible derivation references. +- **TEPP** owns exact evidence spans, temporal/event measurement, + multilevel/multiple-membership psychometrics, calibration, and semantic-span + budgeting through a versioned import or REST contract. +- **contextual-orchestrator** owns provider-neutral model routing and bounded + single-model versus multi-agent test-time compute allocation through its + reviewed API. +- **fast-mlsirm** owns Rust psychometric arithmetic and calibration interfaces. +- **Valkey** remains the event queue. Durable registry truth remains in + PostgreSQL; a later outbox slice bridges the two. + +No component reads another service's private application tables. + +## Alternatives considered + +### Merge the parallel experiment unchanged + +Rejected. It replaces reviewed product history, duplicates web and identity +surfaces, and creates a second database authority. + +### Store one JSON document per run + +Rejected. Signed external manifests may be JSON artifacts, but relational +identity, scope, counts, clocks, and lifecycle need independent constraints, +authorization, and query plans. + +### Put the registry only in Valkey + +Rejected. Queue state is transient and replayable. Audit identity, +idempotency, temporal eligibility, and retention evidence require PostgreSQL. + +### Store knowledge cutoff on the snapshot + +Rejected. One immutable capture can support multiple analysis requests with +different historical cutoffs. Putting the cutoff on the snapshot violates the +functional dependency and forces duplicate snapshots. + +## Security, privacy, and compliance consequences + +- Necessary PII remains in its authorized source/product tables rather than + being blanket-masked into operational uselessness. +- This registry stores opaque UUIDs, digests, bounded machine codes, aggregate + counts, and clocks only. +- Logs and public acceptance evidence must not include SQL, DSNs, raw source + text, images, secrets, provider payloads, or private source identifiers. +- Artifact bodies remain in access-controlled deployment storage and are linked + later by content digest and policy-bound reference. +- The design supports SOC 2 and CSAP evidence collection through explicit actor, + configuration, status, retention, and rollback contracts; it does not claim + certification. +- Database RLS is deferred because the current API uses one pooled service + identity and application-level RBAC/ABAC. Adopting actor-bound RLS requires a + separate ADR and transaction-scoped identity propagation. + +## Failure and rollback + +Migration replay is idempotent and rejects lookup-category collisions. The +rollback refuses to remove non-empty registry relations. Evidence must first be +exported or explicitly deleted under an approved retention procedure. An empty +rollback removes the view, tables, functions, and lookup rows and is itself +replayable. + +## Verification + +Acceptance requires: + +- real-PostgreSQL migration and replay; +- valid snapshot, aggregate, scope, and lifecycle persistence; +- distinct cutoffs over one snapshot; +- rejection of future-information leakage; +- account-scoped idempotency; +- snapshot, count, run, and authorization-scope immutability; +- deletion resistance for request and scope audit evidence; +- scope-required lifecycle, request-time ordering, and database-owned record time; +- canonical idempotency and bounded machine-code failure identifiers; +- count/run concurrency serialization; +- pending-first, contiguous, monotonic, legal status transitions; +- append-only status evidence; +- fail-closed rollback; +- two-or-more-word `snake_case` database-object names; +- complete repository, security, SAST, documentation, and public-content gates + on the exact merge head. + +## Follow-up sequence + +1. Add a transaction repository that creates snapshot, counts, run, scope, and + first status atomically and compares request digests on idempotent retries. +2. Add RBAC/ABAC-protected run list/detail endpoints and the DB-grounded + read-only administrator surface. +3. Add a normalized PostgreSQL outbox and Valkey delivery worker. +4. Add TEPP and contextual-orchestrator adapters only after their versioned + contracts are present on reviewed main branches. +5. Execute private actual-data analysis and store only signed aggregate and + reproducibility manifests outside public source control. +6. Run browser E2E through real OIDC, product navigation, and evidence drill-down. + +## References — APA 7th + +International Organization for Standardization. (2019). *ISO 8601-1:2019: Date +and time—Representations for information interchange—Part 1: Basic rules* +(confirmed 2024; Amendment 1:2022). + +Moreau, L., & Missier, P. (Eds.). (2013). *PROV-DM: The PROV data model*. +World Wide Web Consortium. https://www.w3.org/TR/prov-dm/ + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: +5.5. Constraints*. https://www.postgresql.org/docs/current/ddl-constraints.html + +World Wide Web Consortium. (2013). *PROV-O: The PROV ontology* (W3C +Recommendation). https://www.w3.org/TR/prov-o/ + +World Wide Web Consortium. (2022). *Time ontology in OWL* (W3C Recommendation). +https://www.w3.org/TR/owl-time/ diff --git a/docs/adr/0014-authorized-analysis-run-read.md b/docs/adr/0014-authorized-analysis-run-read.md new file mode 100644 index 000000000..c0f32beef --- /dev/null +++ b/docs/adr/0014-authorized-analysis-run-read.md @@ -0,0 +1,51 @@ +# ADR 0014 — Analysis-run evidence is an authorized, source-redacting read + +**Decision status:** Accepted on this active PR; not protected-main truth until merge +**Date:** 2026-08-16 +**Depends on:** ADR 0013 normalized analysis-run registry +**Refs:** Issue #79 (Milestone 2 parent); closed PR #77 is read-only evidence + +## Context + +PR #89 persists analysis-run identity, aggregate reconciliation, scope, +and lifecycle without exposing a product API. Buyers still cannot see +whether a lineage reconstruction ran, succeeded, or reconciled how many +documents. Closed PR #77 exposed analysis records through a parallel +application that also stored raw metadata payloads -- that shape cannot +become protected product truth. + +## Decision + +LineageWeave owns a fail-closed read projection of the #89 registry: + +- `GET /api/analysis-runs` and `GET /api/analysis-runs/{id}` require + `post_read`. +- Visibility is evaluated in SQL. A run is visible when the caller + requested it, or the scope is a corporate entity / process unit / + thread group the caller may already walk. `all_visible` stays + requester-only so it cannot broaden another tenant's evidence. +- Hidden runs return 404, not 403, and never appear in the list. +- The payload carries lookup labels and non-negative aggregate counts. + It does not carry source SQL, DSNs, raw records, image bytes, provider + payloads, credentials, or another service's table names. +- `GET /api/analysis-runs/{id}` also returns the append-only labeled + `status_history`. The list does not. A failed event may include the + stored machine `failure_code`; this slice does not invent a label. +- TEPP remains a versioned `AnalysisRunRequest` consumer + (`lineageweave.tepp_client`). This slice does not fork TEPP arithmetic. +- contextual-orchestrator remains the only LLM path. This slice does not + call a raw model API. + +## Consequences + +`make seed` writes one synthetic Demo Corp lineage run so the existing +React home page can show Analysis runs without a second application. +The detail now shows the legal lifecycle the registry already stored. +Write/rebuild APIs, TEPP submission, and a fuller Analysis Run Console +remain later slices. + +## References + +Lebo, T., Sahoo, S., & McGuinness, D. (Eds.). (2013). *PROV-O: The PROV +ontology* (W3C Recommendation). World Wide Web Consortium. +https://www.w3.org/TR/2013/REC-prov-o-20130430/ diff --git a/docs/adr/0013-adaptive-contextual-orchestrator-default.md b/docs/adr/0015-adaptive-contextual-orchestrator-default.md similarity index 96% rename from docs/adr/0013-adaptive-contextual-orchestrator-default.md rename to docs/adr/0015-adaptive-contextual-orchestrator-default.md index ee0402075..433432fbb 100644 --- a/docs/adr/0013-adaptive-contextual-orchestrator-default.md +++ b/docs/adr/0015-adaptive-contextual-orchestrator-default.md @@ -1,4 +1,4 @@ -# ADR-0013: Adaptive contextual-orchestrator mode is the default +# ADR-0015: Adaptive contextual-orchestrator mode is the default - Status: Accepted - Date: 2026-08-16 diff --git a/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md b/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md new file mode 100644 index 000000000..b439cb9ca --- /dev/null +++ b/docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md @@ -0,0 +1,102 @@ +# Analysis-run registry standards and research traceability + +**Status:** Active PR evidence; not protected-main truth until merge. +**Scope:** Migration 0018, ADR 0013, rollback, and real-PostgreSQL contract tests. + +## Standards mapped to implementation + +| Source | Product implication | Implemented evidence | +|---|---|---| +| W3C PROV-DM and PROV-O | Preserve identifiable entities, activities, agents, generation/use, and derivation without flattening provenance into display-only edges. | `analysis_source_snapshot`, `analysis_run`, authenticated requester, append-only status events, immutable digests; later product bindings continue to use the separate `provenance_*` layer from ADR 0011. | +| W3C Time Ontology in OWL | Keep temporal concepts explicit and avoid collapsing distinct clocks. | Evidence availability and snapshot capture remain on `analysis_source_snapshot`; analysis knowledge cutoff and request time remain on `analysis_run`; status occurrence and database record time remain distinct. | +| ISO 8601-1:2019 | Use unambiguous timestamp representation and timezone-aware persistence. | PostgreSQL `timestamptz` for availability, capture, cutoff, request, occurrence, and record clocks; tests use explicit `Z` offsets. | +| PostgreSQL 18 constraints and trigger contracts | Put integrity close to durable truth and use constraints for row shape while triggers enforce cross-row state and serialization. | Digest/check constraints, category allowlists, account-scoped uniqueness, shape constraints, immutable-row triggers, shared snapshot-row locking, and serialized status transitions. | +| NIST SP 800-92 | Treat audit records as bounded, protected operational evidence rather than unstructured application logging. | Append-only status events, machine failure codes, actor identity, occurrence/record clocks, fail-closed rollback, and exclusion of raw source/provider payloads. | +| OpenAPI 3.2.0 | Define explicit versioned API schemas rather than exposing database rows or implementation-specific payloads. | API intentionally deferred; ADR 0013 requires a source-redacting run list/detail contract before a product surface is claimed. | + +## Temporal reasoning + +The registry applies a bitemporal discipline without claiming a complete +general-purpose bitemporal database: + +- `maximum_available_time` answers when the newest admitted evidence became + knowable; +- `captured_at` answers when the immutable source snapshot was materialized; +- `knowledge_cutoff` answers what a specific analysis was allowed to know; +- `requested_at` answers when that analysis was requested; +- `occurred_at` and `recorded_at` distinguish lifecycle occurrence from durable + database recording. + +The database requires the aggregate leakage boundary: + +```text +maximum_available_time <= knowledge_cutoff <= requested_at +captured_at <= requested_at +``` + +TEPP remains the authority for finer event/assertion/document/system/available +clocks and temporal psychometrics. The registry does not duplicate TEPP +measurement outputs. + +## Audit and privacy boundary + +The registry may store: + +- opaque product UUIDs; +- authenticated account UUIDs; +- SHA-256 digests; +- bounded configuration/version identifiers; +- aggregate counts; +- bounded status/failure codes; +- timezone-aware clocks. + +The registry must not store: + +- source SQL or source-table names; +- DSNs, credentials, or provider secrets; +- raw posts, HTML, images, base64 data, or attachments; +- model prompts/responses or raw exceptions; +- another service's application tables; +- organization-specific source identifiers in public fixtures or documentation. + +Necessary PII remains available in its purpose-bound authorized product/source +context. Auditability is achieved with actor identity, access control, +provenance, retention, and immutable evidence rather than blanket masking. + +## Verification matrix + +| Claim | Falsifiable test | +|---|---| +| One snapshot supports multiple analyses | Insert two runs over one snapshot with different valid cutoffs. | +| Future evidence is excluded | Reject a run whose cutoff precedes the snapshot's maximum availability time. | +| Evidence cannot change after derivation | Reject snapshot/count updates and count insert/delete after the first run. | +| Count/run race is serialized | Both paths acquire the snapshot row first; a later concurrency test must prove one legal winner and no lost freeze. | +| Request identity is stable | Reject analysis-run updates; scope and lifecycle live in their own relations. | +| Idempotency is actor-scoped | Permit identical opaque keys for two accounts and reject reuse by the same account. | +| Lifecycle is ordered | Require pending first, contiguous ordinals, monotonic time, legal transitions, terminal finality, and append-only rows. | +| Rollback does not erase audit data silently | Reject rollback with any registry rows and allow replay after explicit cleanup. | + +## APA 7th references + +International Organization for Standardization. (2019). *ISO 8601-1:2019: Date +and time—Representations for information interchange—Part 1: Basic rules* +(confirmed 2024; Amendment 1:2022). + +Kent, K., & Souppaya, M. (2006). *Guide to computer security log management* +(NIST Special Publication 800-92). National Institute of Standards and +Technology. https://doi.org/10.6028/NIST.SP.800-92 + +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 + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18 documentation: +5.5. Constraints*. https://www.postgresql.org/docs/current/ddl-constraints.html + +World Wide Web Consortium. (2013). *PROV-O: The PROV ontology* (W3C +Recommendation). https://www.w3.org/TR/prov-o/ + +World Wide Web Consortium. (2022). *Time ontology in OWL* (W3C Recommendation). +https://www.w3.org/TR/owl-time/ diff --git a/docs/superpowers/plans/2026-08-15-analysis-run-registry.md b/docs/superpowers/plans/2026-08-15-analysis-run-registry.md new file mode 100644 index 000000000..a3ed77d27 --- /dev/null +++ b/docs/superpowers/plans/2026-08-15-analysis-run-registry.md @@ -0,0 +1,83 @@ +# Analysis-run registry implementation plan + +> Execute test-first. Preserve the reviewed LineageWeave product and keep +> private actual-data evidence outside public source control. + +**Goal:** Establish one normalized, temporally truthful, actor-scoped registry +for Milestone 2 analysis requests and lifecycle evidence. + +## Task 1 — RED: database contract + +**File:** `tests/test_analysis_run_registry_schema.py` + +1. Require the five normalized relations, current-status view, rollback, and + fresh-install wiring. +2. Reject the retained experiment's denormalized table and JSON metadata. +3. Require evidence-owned availability/capture clocks and a run-owned cutoff. +4. Require non-null requester identity and account-scoped idempotency. +5. Require immutable snapshot, count, and run request rows. +6. Require shared row locking between count mutation and first run creation. +7. Require pending-first, contiguous, monotonic, legal status transitions and + append-only status rows. +8. Require fail-closed rollback and descriptive database-object names. + +## Task 2 — GREEN: normalized migration and rollback + +**Files:** + +- `migrations/0018_analysis_run_registry.sql` +- `migrations/rollback/0018_analysis_run_registry.sql` +- `docker/postgres-init/Dockerfile` + +1. Insert category-checked lookup values idempotently. +2. Add snapshot, count, run, scope, and status-event relations in 3NF. +3. Keep `maximum_available_time` on the snapshot and `knowledge_cutoff` on the + run. +4. Serialize count freeze and run creation through the same snapshot row lock. +5. Reject mutation of immutable evidence and request configuration. +6. Implement the lifecycle state machine as a serialized insert trigger. +7. Add the current-status read view. +8. Refuse rollback while any audit evidence exists. +9. Apply migration 0018 after the PROV-O migration on fresh PostgreSQL images. + +## Task 3 — Documentation and evidence + +**Files:** + +- `docs/adr/0013-normalized-analysis-run-registry.md` +- `docs/doctoring/ANALYSIS_RUN_REGISTRY_REFERENCES.md` +- `CHANGELOG.d/milestone2-analysis-run-registry.md` + +1. Record product/service ownership and deferred API/UI claims. +2. Trace temporal, provenance, audit, privacy, concurrency, and rollback + decisions to current authoritative sources in APA 7th form. +3. Mark active-PR decisions as non-main truth. +4. Keep public fixtures synthetic and exclude private source identifiers. + +## Task 4 — Exact-head verification + +1. Run the static test without PostgreSQL and prove it fails before migration. +2. Run all registry cases against real PostgreSQL after implementation. +3. Replay the migration and rollback. +4. Run the complete Python product suite against PostgreSQL. +5. Run frontend lint, complete tests, and production build. +6. Run `compileall`, security, SAST, documentation hygiene, and public-content + scans. +7. Inspect the exact final diff for temporary workflows/scripts. +8. Obtain independent exact-head review and merge only after the parent PR is on + protected `main` and base-sensitive evidence is regenerated. + +## Task 5 — Next bounded vertical slice + +After this registry reaches protected main: + +1. Write failing repository tests for atomic run + scope + pending-event + creation and idempotent request comparison. +2. Implement the async PostgreSQL repository with no cross-service SQL. +3. Add RBAC/ABAC-protected source-redacting list/detail endpoints. +4. Add the DB-grounded read-only administrator surface and Storybook states. +5. Add normalized outbox + Valkey delivery. +6. Integrate TEPP and contextual-orchestrator only through reviewed versioned + contracts. +7. Execute private actual-data analysis and retain signed aggregate acceptance + artifacts outside public Git history. diff --git a/frontend/package.json b/frontend/package.json index eef0c8735..5d3f2e2b8 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,7 +1,7 @@ { "name": "frontend", "private": true, - "version": "0.78.0", + "version": "0.82.0", "type": "module", "scripts": { "dev": "vite", diff --git a/frontend/src/App.test.tsx b/frontend/src/App.test.tsx index 85346bc88..4fb1b5649 100644 --- a/frontend/src/App.test.tsx +++ b/frontend/src/App.test.tsx @@ -169,6 +169,77 @@ describe("App, authenticated", () => { jsonResponse({ post_id: "post-1", has_commitment: true, ticket }), ); } + if (url.endsWith("/api/analysis-runs/run-demo-lineage")) { + return Promise.resolve( + jsonResponse({ + analysis_run_id: "run-demo-lineage", + run_kind_code: "analysis_run_lineage", + run_kind_label: "Lineage reconstruction", + scope_kind_code: "analysis_scope_corporate_entity", + scope_kind_label: "Corporate entity", + scope_entity_name: "Demo Corp", + status_code: "analysis_status_succeeded", + status_label: "Succeeded", + knowledge_cutoff: "2026-01-12T12:00:00Z", + requested_at: "2026-01-12T12:30:00Z", + source_counts: [ + { + count_type_code: "analysis_count_document", + count_type_label: "Documents", + count_value: 3, + }, + ], + visible_posts: [{ post_id: "post-1", post_title: "Public post" }], + status_history: [ + { + status_ordinal: 1, + status_code: "analysis_status_pending", + status_label: "Pending", + occurred_at: "2026-01-12T12:31:00Z", + }, + { + status_ordinal: 2, + status_code: "analysis_status_running", + status_label: "Running", + occurred_at: "2026-01-12T12:32:00Z", + }, + { + status_ordinal: 3, + status_code: "analysis_status_succeeded", + status_label: "Succeeded", + occurred_at: "2026-01-12T12:33:00Z", + }, + ], + }), + ); + } + if (url.endsWith("/api/analysis-runs")) { + return Promise.resolve( + jsonResponse({ + analysis_runs: [ + { + analysis_run_id: "run-demo-lineage", + run_kind_code: "analysis_run_lineage", + run_kind_label: "Lineage reconstruction", + scope_kind_code: "analysis_scope_corporate_entity", + scope_kind_label: "Corporate entity", + scope_entity_name: "Demo Corp", + status_code: "analysis_status_succeeded", + status_label: "Succeeded", + knowledge_cutoff: "2026-01-12T12:00:00Z", + requested_at: "2026-01-12T12:30:00Z", + source_counts: [ + { + count_type_code: "analysis_count_document", + count_type_label: "Documents", + count_value: 3, + }, + ], + }, + ], + }), + ); + } if (url.endsWith("/api/calendar")) { return Promise.resolve( jsonResponse({ @@ -1296,6 +1367,36 @@ describe("App, authenticated", () => { await waitFor(() => expect(screen.getByText("The full body text.")).toBeInTheDocument()); }); + it("shows the seeded analysis run on the home page", async () => { + stubBackend(); + render(); + + expect(await screen.findByRole("heading", { name: "Analysis runs" })).toBeInTheDocument(); + const list = screen.getByRole("list", { name: "Analysis runs" }); + expect(list).toHaveTextContent("Lineage reconstruction · Succeeded · Demo Corp"); + expect(list).toHaveTextContent("3 documents"); + expect(list).not.toHaveTextContent("postgresql://"); + expect(list).not.toHaveTextContent("select "); + + await userEvent.click( + screen.getByRole("button", { + name: "Open analysis run: Lineage reconstruction · Succeeded · Demo Corp", + }), + ); + expect(await screen.findByRole("heading", { name: "Lineage reconstruction · Succeeded · Demo Corp" })).toBeInTheDocument(); + expect(screen.getByText(/Cutoff 2026-01-12/)).toBeInTheDocument(); + expect(screen.getByText(/Requested 2026-01-12/)).toBeInTheDocument(); + const history = screen.getByRole("list", { name: "Analysis run status history" }); + expect(history).toHaveTextContent("Pending 2026-01-12 12:31"); + expect(history).toHaveTextContent("Running 2026-01-12 12:32"); + expect(history).toHaveTextContent("Succeeded 2026-01-12 12:33"); + expect(screen.getByRole("button", { name: "Open run post: Public post" })).toBeInTheDocument(); + expect(screen.queryByText(/postgresql:\/\//)).not.toBeInTheDocument(); + + await userEvent.click(screen.getByRole("button", { name: "Open run post: Public post" })); + await waitFor(() => expect(screen.getByText("The full body text.")).toBeInTheDocument()); + }); + it("shows the calibrated period-report mean theta on the home page", async () => { stubBackend(); render(); diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 5a9130f92..1f350928d 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -7,6 +7,8 @@ import { deriveCommitment, evaluatePost, extractPostKeymen, + fetchAnalysisRun, + fetchAnalysisRuns, fetchCalendar, fetchLineageGraph, fetchMe, @@ -33,6 +35,7 @@ import { verifyPostRelations, type ActivityEvent, type AffiliateNode, + type AnalysisRun, type CalendarEntry, type ChatAnswer, type ChatExchange, @@ -1344,6 +1347,128 @@ function PostDetailPopup({ ); } +function analysisRunCaption(run: AnalysisRun): string { + return [run.run_kind_label, run.status_label, run.scope_entity_name ?? run.scope_kind_label] + .filter(Boolean) + .join(" · "); +} + +function AnalysisRunsPanel({ + accessToken, + onSelectPost, +}: { + accessToken: string; + onSelectPost: (postId: string) => void; +}) { + const [runs, setRuns] = useState(null); + const [selected, setSelected] = useState(null); + const [error, setError] = useState(null); + + useEffect(() => { + fetchAnalysisRuns(accessToken) + .then((payload) => setRuns(payload.analysis_runs)) + .catch((err) => setError(String(err))); + }, [accessToken]); + + async function handleOpen(runId: string) { + setError(null); + try { + setSelected(await fetchAnalysisRun(accessToken, runId)); + } catch (err) { + setSelected(null); + if (err instanceof BackendError && err.status === 404) { + setError("This analysis run is not visible."); + return; + } + setError(String(err)); + } + } + + if (error && runs === null) return

{error}

; + if (runs === null) return

Loading analysis runs...

; + + return ( +
+
+

Analysis runs

+
+ {error &&

{error}

} + {runs.length === 0 ? ( +

+ No analysis runs visible to this account yet -- try `make seed`. +

+ ) : ( +
    + {runs.map((run) => { + const documentCount = run.source_counts.find( + (count) => count.count_type_code === "analysis_count_document", + ); + const caption = analysisRunCaption(run); + return ( +
  • + +
    • + ); + })} +
+ )} + {selected && ( +
+

{analysisRunCaption(selected)}

+

+ Cutoff {selected.knowledge_cutoff.slice(0, 10)} + {" · "} + Requested {selected.requested_at.slice(0, 10)} +

+
    + {selected.source_counts.map((count) => ( +
  • + {count.count_value} {count.count_type_label.toLowerCase()} +
    • + ))} +
+ {selected.status_history && selected.status_history.length > 0 && ( +
    + {selected.status_history.map((event) => ( +
  1. + {event.status_label} {event.occurred_at.slice(0, 16).replace("T", " ")} + {event.failure_code ? ` · ${event.failure_code}` : ""} +
    2. + ))} +
+ )} + {selected.visible_posts && selected.visible_posts.length > 0 && ( +
    + {selected.visible_posts.map((post) => ( +
  • + +
    • + ))} +
+ )} +
+ )} +
+ ); +} + function CalendarPanel({ accessToken, onSelectPost, @@ -1632,6 +1757,7 @@ function PostList({ accessToken }: { accessToken: string }) { return ( <> +
diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 2f3e3bbb6..3dacb054c 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -492,3 +492,41 @@ export function deriveCommitment(accessToken: string, postId: string): Promise { return backendFetch("/api/calendar", accessToken); } + +export interface AnalysisRunCount { + count_type_code: string; + count_type_label: string; + count_value: number; +} + +export interface AnalysisRunStatusEvent { + status_ordinal: number; + status_code: string; + status_label: string; + occurred_at: string; + failure_code?: string; +} + +export interface AnalysisRun { + analysis_run_id: string; + run_kind_code: string; + run_kind_label: string; + scope_kind_code: string; + scope_kind_label: string; + scope_entity_name?: string; + status_code: string | null; + status_label: string | null; + knowledge_cutoff: string; + requested_at: string; + source_counts: AnalysisRunCount[]; + status_history?: AnalysisRunStatusEvent[]; + visible_posts?: { post_id: string; post_title: string }[]; +} + +export function fetchAnalysisRuns(accessToken: string): Promise<{ analysis_runs: AnalysisRun[] }> { + return backendFetch("/api/analysis-runs", accessToken); +} + +export function fetchAnalysisRun(accessToken: string, analysisRunId: string): Promise { + return backendFetch(`/api/analysis-runs/${analysisRunId}`, accessToken); +} diff --git a/lineageweave/__init__.py b/lineageweave/__init__.py index be9228d75..b1f0c97b1 100644 --- a/lineageweave/__init__.py +++ b/lineageweave/__init__.py @@ -55,4 +55,4 @@ "sentence_excerpts", ] -__version__ = "0.78.0" +__version__ = "0.82.0" diff --git a/migrations/0018_analysis_run_registry.sql b/migrations/0018_analysis_run_registry.sql new file mode 100644 index 000000000..b08d80b3b --- /dev/null +++ b/migrations/0018_analysis_run_registry.sql @@ -0,0 +1,569 @@ +-- Milestone 2 additive runtime bridge: normalized analysis-run registry. +-- +-- This migration records reproducibility, authorization scope, aggregate +-- reconciliation, and lifecycle evidence without storing source SQL, DSNs, +-- raw records, image bytes, provider payloads, credentials, or free-form JSON. +-- Snapshot availability is evidence-owned; the knowledge cutoff is run-owned, +-- so one immutable capture can support multiple historically valid analyses. + +begin; + +insert into common_lookup_value + (lookup_category, lookup_code, lookup_label, display_order) +values + ('analysis_run_kind', 'analysis_run_lineage', 'Lineage reconstruction', 0), + ('analysis_run_kind', 'analysis_run_report', 'Period report', 1), + ('analysis_run_kind', 'analysis_run_tepp', 'TEPP measurement', 2), + ('analysis_run_status', 'analysis_status_pending', 'Pending', 0), + ('analysis_run_status', 'analysis_status_running', 'Running', 1), + ('analysis_run_status', 'analysis_status_succeeded', 'Succeeded', 2), + ('analysis_run_status', 'analysis_status_failed', 'Failed', 3), + ('analysis_run_status', 'analysis_status_cancelled', 'Cancelled', 4), + ('analysis_run_scope', 'analysis_scope_all_visible', 'All authorized records', 0), + ('analysis_run_scope', 'analysis_scope_corporate_entity', 'Corporate entity', 1), + ('analysis_run_scope', 'analysis_scope_process_unit', 'Process unit', 2), + ('analysis_run_scope', 'analysis_scope_thread_group', 'Thread group', 3), + ('analysis_source_count', 'analysis_count_source_row', 'Source rows', 0), + ('analysis_source_count', 'analysis_count_document', 'Documents', 1), + ('analysis_source_count', 'analysis_count_thread', 'Threads', 2), + ('analysis_source_count', 'analysis_count_lineage_node', 'Lineage nodes', 3), + ('analysis_source_count', 'analysis_count_lineage_edge', 'Lineage edges', 4) +on conflict (lookup_code) do nothing; + +-- common_lookup_value deliberately makes lookup_code globally unique. A code +-- that already exists under another category is a migration conflict rather +-- than permission to attach the wrong vocabulary to an analysis column. +do $$ +declare + lookup_mismatch_count integer; +begin + select count(*) + into lookup_mismatch_count + from common_lookup_value as actual + join (values + ('analysis_run_lineage', 'analysis_run_kind'), + ('analysis_run_report', 'analysis_run_kind'), + ('analysis_run_tepp', 'analysis_run_kind'), + ('analysis_status_pending', 'analysis_run_status'), + ('analysis_status_running', 'analysis_run_status'), + ('analysis_status_succeeded', 'analysis_run_status'), + ('analysis_status_failed', 'analysis_run_status'), + ('analysis_status_cancelled', 'analysis_run_status'), + ('analysis_scope_all_visible', 'analysis_run_scope'), + ('analysis_scope_corporate_entity', 'analysis_run_scope'), + ('analysis_scope_process_unit', 'analysis_run_scope'), + ('analysis_scope_thread_group', 'analysis_run_scope'), + ('analysis_count_source_row', 'analysis_source_count'), + ('analysis_count_document', 'analysis_source_count'), + ('analysis_count_thread', 'analysis_source_count'), + ('analysis_count_lineage_node', 'analysis_source_count'), + ('analysis_count_lineage_edge', 'analysis_source_count') + ) as expected(lookup_code, lookup_category) + on expected.lookup_code = actual.lookup_code + where actual.lookup_category <> expected.lookup_category; + + if lookup_mismatch_count <> 0 then + raise exception 'analysis_run_registry_lookup_conflict'; + end if; +end +$$; + +create table if not exists analysis_source_snapshot ( + analysis_source_snapshot_id uuid primary key default uuid_generate_v4(), + snapshot_sha256 text not null unique, + source_contract_version text not null, + maximum_available_time timestamptz not null, + captured_at timestamptz not null, + created_at timestamptz not null default now(), + constraint analysis_source_snapshot_digest_check + check (snapshot_sha256 ~ '^[0-9a-f]{64}$'), + constraint analysis_source_snapshot_contract_check + check (length(btrim(source_contract_version)) between 1 and 128), + constraint analysis_source_snapshot_capture_check + check (maximum_available_time <= captured_at), + constraint analysis_source_snapshot_created_check + check (captured_at <= created_at) +); + +comment on table analysis_source_snapshot is + 'Immutable captured-source identity and latest evidence-availability time; ' + 'knowledge cutoffs belong to analysis_run, not the reusable snapshot.'; + +create table if not exists analysis_source_count ( + analysis_source_snapshot_id uuid not null + references analysis_source_snapshot (analysis_source_snapshot_id) + on delete cascade, + count_type_code text not null + references common_lookup_value (lookup_code), + count_value bigint not null, + primary key (analysis_source_snapshot_id, count_type_code), + constraint analysis_source_count_type_check + check (count_type_code in ( + 'analysis_count_source_row', + 'analysis_count_document', + 'analysis_count_thread', + 'analysis_count_lineage_node', + 'analysis_count_lineage_edge' + )), + constraint analysis_source_count_nonnegative_check + check (count_value >= 0) +); + +comment on table analysis_source_count is + 'One normalized aggregate reconciliation count per immutable snapshot and ' + 'count vocabulary; no source record is stored.'; + +create table if not exists analysis_run ( + analysis_run_id uuid primary key default uuid_generate_v4(), + analysis_source_snapshot_id uuid not null + references analysis_source_snapshot (analysis_source_snapshot_id), + run_kind_code text not null + references common_lookup_value (lookup_code), + requested_by_account_id uuid not null + references user_account (user_account_id), + idempotency_key text not null, + knowledge_cutoff timestamptz not null, + configuration_schema_version text not null, + configuration_sha256 text not null, + model_contract_sha256 text, + prompt_bundle_sha256 text, + code_revision_sha text not null, + requested_at timestamptz not null default now(), + constraint analysis_run_kind_check + check (run_kind_code in ( + 'analysis_run_lineage', + 'analysis_run_report', + 'analysis_run_tepp' + )), + constraint analysis_run_idempotency_key_check + check ( + idempotency_key = btrim(idempotency_key) + and length(idempotency_key) between 1 and 256 + and idempotency_key !~ '[[:cntrl:]]' + ), + constraint analysis_run_configuration_version_check + check ( + configuration_schema_version = btrim(configuration_schema_version) + and length(configuration_schema_version) between 1 and 128 + ), + constraint analysis_run_configuration_digest_check + check (configuration_sha256 ~ '^[0-9a-f]{64}$'), + constraint analysis_run_model_digest_check + check ( + model_contract_sha256 is null + or model_contract_sha256 ~ '^[0-9a-f]{64}$' + ), + constraint analysis_run_prompt_digest_check + check ( + prompt_bundle_sha256 is null + or prompt_bundle_sha256 ~ '^[0-9a-f]{64}$' + ), + constraint analysis_run_code_revision_check + check (code_revision_sha ~ '^(?:[0-9a-f]{40}|[0-9a-f]{64})$'), + constraint analysis_run_request_time_check + check (knowledge_cutoff <= requested_at), + unique (requested_by_account_id, idempotency_key) +); + +create index if not exists analysis_run_snapshot_idx + on analysis_run (analysis_source_snapshot_id); +create index if not exists analysis_run_kind_requested_idx + on analysis_run (run_kind_code, requested_at desc); +create index if not exists analysis_run_requester_idx + on analysis_run (requested_by_account_id, requested_at desc); + +comment on table analysis_run is + 'Immutable account-scoped analysis request bound to one snapshot, one ' + 'knowledge cutoff, and reproducibility digests; lifecycle is event-derived.'; + +create table if not exists analysis_run_scope ( + analysis_run_id uuid primary key + references analysis_run (analysis_run_id), + scope_kind_code text not null + references common_lookup_value (lookup_code), + corporate_entity_id uuid + references corporate_entity (corporate_entity_id), + process_unit_id uuid + references process_unit (process_unit_id), + scope_key text, + constraint analysis_run_scope_kind_check + check (scope_kind_code in ( + 'analysis_scope_all_visible', + 'analysis_scope_corporate_entity', + 'analysis_scope_process_unit', + 'analysis_scope_thread_group' + )), + constraint analysis_run_scope_shape_check + check ( + (scope_kind_code = 'analysis_scope_all_visible' + and corporate_entity_id is null + and process_unit_id is null + and scope_key is null) + or + (scope_kind_code = 'analysis_scope_corporate_entity' + and corporate_entity_id is not null + and process_unit_id is null + and scope_key is null) + or + (scope_kind_code = 'analysis_scope_process_unit' + and corporate_entity_id is null + and process_unit_id is not null + and scope_key is null) + or + (scope_kind_code = 'analysis_scope_thread_group' + and corporate_entity_id is null + and process_unit_id is null + and scope_key is not null + and scope_key = btrim(scope_key) + and length(scope_key) between 1 and 256 + and scope_key !~ '[[:cntrl:]]') + ) +); + +create index if not exists analysis_run_scope_entity_idx + on analysis_run_scope (corporate_entity_id) + where corporate_entity_id is not null; +create index if not exists analysis_run_scope_unit_idx + on analysis_run_scope (process_unit_id) + where process_unit_id is not null; + +comment on table analysis_run_scope is + 'One immutable authorization-relevant scope is required before lifecycle ' + 'evidence; process-unit ownership remains derivable from process_unit.'; + +create table if not exists analysis_run_status_event ( + analysis_run_id uuid not null + references analysis_run (analysis_run_id), + status_ordinal integer not null, + status_code text not null + references common_lookup_value (lookup_code), + occurred_at timestamptz not null, + recorded_at timestamptz not null default clock_timestamp(), + failure_code text, + retryable boolean not null default false, + primary key (analysis_run_id, status_ordinal), + constraint analysis_run_status_code_check + check (status_code in ( + 'analysis_status_pending', + 'analysis_status_running', + 'analysis_status_succeeded', + 'analysis_status_failed', + 'analysis_status_cancelled' + )), + constraint analysis_run_status_ordinal_check + check (status_ordinal >= 1), + constraint analysis_run_status_time_check + check (occurred_at <= recorded_at), + constraint analysis_run_status_failure_shape_check + check ( + (status_code = 'analysis_status_failed' + and failure_code is not null + and failure_code ~ '^[a-z][a-z0-9_]{0,127}$') + or + (status_code <> 'analysis_status_failed' + and failure_code is null + and retryable = false) + ) +); + +create index if not exists analysis_run_status_current_idx + on analysis_run_status_event (analysis_run_id, status_ordinal desc); + +comment on table analysis_run_status_event is + 'Append-only, contiguous, monotonic state-machine evidence; failure_code is ' + 'a bounded machine code and never contains raw provider or source payloads.'; + +create or replace function reject_analysis_source_snapshot_update() +returns trigger +language plpgsql +as $$ +begin + raise exception 'analysis_source_snapshot_is_immutable'; +end +$$; + +comment on function reject_analysis_source_snapshot_update() is + 'Rejects mutation of captured source identity and availability evidence.'; + +drop trigger if exists analysis_source_snapshot_update_reject + on analysis_source_snapshot; +create trigger analysis_source_snapshot_update_reject +before update on analysis_source_snapshot +for each row execute function reject_analysis_source_snapshot_update(); + +create or replace function reject_analysis_source_count_update() +returns trigger +language plpgsql +as $$ +begin + raise exception 'analysis_source_count_is_immutable'; +end +$$; + +comment on function reject_analysis_source_count_update() is + 'Rejects replacement of a snapshot aggregate; delete and reinsert is only ' + 'permitted before the snapshot is attached to a run.'; + +drop trigger if exists analysis_source_count_update_reject + on analysis_source_count; +create trigger analysis_source_count_update_reject +before update on analysis_source_count +for each row execute function reject_analysis_source_count_update(); + +create or replace function enforce_analysis_source_count_freeze() +returns trigger +language plpgsql +as $$ +declare + affected_snapshot_id uuid; +begin + if tg_op = 'DELETE' then + affected_snapshot_id := old.analysis_source_snapshot_id; + else + affected_snapshot_id := new.analysis_source_snapshot_id; + end if; + + -- Both count mutation and run creation lock this row first. That common + -- lock order closes the race between the final count write and first run. + perform 1 + from analysis_source_snapshot + where analysis_source_snapshot_id = affected_snapshot_id + for update; + + if exists ( + select 1 + from analysis_run + where analysis_source_snapshot_id = affected_snapshot_id + ) then + raise exception 'analysis_source_count_frozen_after_run'; + end if; + + if tg_op = 'DELETE' then + return old; + end if; + return new; +end +$$; + +comment on function enforce_analysis_source_count_freeze() is + 'Serializes count insert/delete against first run creation and rejects ' + 'changes after any run references the snapshot.'; + +drop trigger if exists analysis_source_count_freeze_guard + on analysis_source_count; +create trigger analysis_source_count_freeze_guard +before insert or delete on analysis_source_count +for each row execute function enforce_analysis_source_count_freeze(); + +create or replace function enforce_analysis_run_knowledge_cutoff() +returns trigger +language plpgsql +as $$ +declare + snapshot_available_time timestamptz; + snapshot_capture_time timestamptz; +begin + if new.requested_at > clock_timestamp() then + raise exception 'analysis_run_request_time_in_future'; + end if; + + select maximum_available_time, captured_at + into snapshot_available_time, snapshot_capture_time + from analysis_source_snapshot + where analysis_source_snapshot_id = new.analysis_source_snapshot_id + for update; + + if not found then + raise exception 'analysis_source_snapshot_not_found'; + end if; + if snapshot_available_time > new.knowledge_cutoff then + raise exception 'analysis_run_future_information_leakage'; + end if; + if snapshot_capture_time > new.requested_at then + raise exception 'analysis_run_snapshot_captured_after_request'; + end if; + return new; +end +$$; + +comment on function enforce_analysis_run_knowledge_cutoff() is + 'Locks the immutable snapshot and rejects run cutoffs earlier than the ' + 'latest admitted evidence or requests earlier than snapshot capture.'; + +drop trigger if exists analysis_run_knowledge_cutoff_guard + on analysis_run; +create trigger analysis_run_knowledge_cutoff_guard +before insert on analysis_run +for each row execute function enforce_analysis_run_knowledge_cutoff(); + +drop trigger if exists analysis_run_update_reject + on analysis_run; +drop trigger if exists analysis_run_mutation_reject + on analysis_run; +drop function if exists reject_analysis_run_update(); + +create or replace function reject_analysis_run_mutation() +returns trigger +language plpgsql +as $$ +begin + raise exception 'analysis_run_request_is_immutable'; +end +$$; + +comment on function reject_analysis_run_mutation() is + 'Rejects update or delete of actor, cutoff, idempotency, and reproducibility ' + 'evidence; run progress belongs to append-only status events.'; + +create trigger analysis_run_mutation_reject +before update or delete on analysis_run +for each row execute function reject_analysis_run_mutation(); + +create or replace function reject_analysis_run_scope_mutation() +returns trigger +language plpgsql +as $$ +begin + raise exception 'analysis_run_scope_is_immutable'; +end +$$; + +comment on function reject_analysis_run_scope_mutation() is + 'Rejects update or delete of the authorization-relevant scope attached to ' + 'an immutable analysis request.'; + +drop trigger if exists analysis_run_scope_mutation_reject + on analysis_run_scope; +create trigger analysis_run_scope_mutation_reject +before update or delete on analysis_run_scope +for each row execute function reject_analysis_run_scope_mutation(); + +create or replace function reject_analysis_run_status_mutation() +returns trigger +language plpgsql +as $$ +begin + raise exception 'analysis_run_status_event_is_append_only'; +end +$$; + +comment on function reject_analysis_run_status_mutation() is + 'Rejects update or delete of state-machine evidence.'; + +drop trigger if exists analysis_run_status_event_update_reject + on analysis_run_status_event; +create trigger analysis_run_status_event_update_reject +before update on analysis_run_status_event +for each row execute function reject_analysis_run_status_mutation(); + +drop trigger if exists analysis_run_status_event_delete_reject + on analysis_run_status_event; +create trigger analysis_run_status_event_delete_reject +before delete on analysis_run_status_event +for each row execute function reject_analysis_run_status_mutation(); + +create or replace function enforce_analysis_run_status_transition() +returns trigger +language plpgsql +as $$ +declare + previous_ordinal integer; + previous_status_code text; + previous_occurred_at timestamptz; + run_requested_at timestamptz; +begin + -- The immutable parent row is a per-run serialization lock. It prevents + -- concurrent writers from both accepting the same next ordinal. + select requested_at + into run_requested_at + from analysis_run + where analysis_run_id = new.analysis_run_id + for update; + + if not found then + raise exception 'analysis_run_not_found'; + end if; + if not exists ( + select 1 from analysis_run_scope + where analysis_run_id = new.analysis_run_id + ) then + raise exception 'analysis_run_scope_required'; + end if; + if new.occurred_at < run_requested_at then + raise exception 'analysis_run_status_before_request'; + end if; + new.recorded_at := clock_timestamp(); + + select status_ordinal, status_code, occurred_at + into previous_ordinal, previous_status_code, previous_occurred_at + from analysis_run_status_event + where analysis_run_id = new.analysis_run_id + order by status_ordinal desc + limit 1; + + if previous_ordinal is null then + if new.status_ordinal <> 1 + or new.status_code <> 'analysis_status_pending' then + raise exception 'analysis_run_first_status_must_be_pending'; + end if; + return new; + end if; + + if new.status_ordinal <> previous_ordinal + 1 then + raise exception 'analysis_run_status_ordinal_not_contiguous'; + end if; + if new.occurred_at < previous_occurred_at then + raise exception 'analysis_run_status_time_not_monotonic'; + end if; + + if previous_status_code = 'analysis_status_pending' then + if new.status_code not in ( + 'analysis_status_running', + 'analysis_status_cancelled' + ) then + raise exception 'analysis_run_status_transition_invalid'; + end if; + elsif previous_status_code = 'analysis_status_running' then + if new.status_code not in ( + 'analysis_status_succeeded', + 'analysis_status_failed', + 'analysis_status_cancelled' + ) then + raise exception 'analysis_run_status_transition_invalid'; + end if; + else + raise exception 'analysis_run_terminal_status_has_no_successor'; + end if; + + return new; +end +$$; + +comment on function enforce_analysis_run_status_transition() is + 'Serializes status appends and requires immutable scope, request-time ' + 'ordering, database-recorded time, legal transitions, and terminal finality.'; + +drop trigger if exists analysis_run_status_transition_guard + on analysis_run_status_event; +create trigger analysis_run_status_transition_guard +before insert on analysis_run_status_event +for each row execute function enforce_analysis_run_status_transition(); + +create or replace view analysis_run_current_status as +select distinct on (status_event.analysis_run_id) + status_event.analysis_run_id, + status_event.status_code, + status_event.status_ordinal, + status_event.occurred_at, + status_event.recorded_at, + status_event.failure_code, + status_event.retryable + from analysis_run_status_event as status_event + order by status_event.analysis_run_id, + status_event.status_ordinal desc; + +comment on view analysis_run_current_status is + 'Latest append-only status projection for each run; never a second mutable ' + 'lifecycle authority.'; + +commit; diff --git a/migrations/rollback/0018_analysis_run_registry.sql b/migrations/rollback/0018_analysis_run_registry.sql new file mode 100644 index 000000000..f91abc47c --- /dev/null +++ b/migrations/rollback/0018_analysis_run_registry.sql @@ -0,0 +1,70 @@ +-- Fail-closed rollback for migration 0018. +-- +-- Registry evidence must be exported or explicitly deleted under an approved +-- retention procedure before these objects can be removed. Re-running this +-- rollback after a successful empty rollback is safe. + +begin; + +do $$ +declare + relation_name text; + relation_has_rows boolean; +begin + foreach relation_name in array array[ + 'analysis_run_status_event', + 'analysis_run_scope', + 'analysis_run', + 'analysis_source_count', + 'analysis_source_snapshot' + ] loop + if to_regclass('public.' || relation_name) is not null then + execute format('select exists (select 1 from %I)', relation_name) + into relation_has_rows; + if relation_has_rows then + raise exception 'analysis_run_registry_not_empty'; + end if; + end if; + end loop; +end +$$; + +drop view if exists analysis_run_current_status; +drop table if exists analysis_run_status_event; +drop table if exists analysis_run_scope; +drop table if exists analysis_run; +drop table if exists analysis_source_count; +drop table if exists analysis_source_snapshot; + +drop function if exists enforce_analysis_run_status_transition(); +drop function if exists reject_analysis_run_status_mutation(); +drop function if exists reject_analysis_run_scope_mutation(); +drop function if exists reject_analysis_run_mutation(); +drop function if exists reject_analysis_run_update(); +drop function if exists enforce_analysis_run_knowledge_cutoff(); +drop function if exists enforce_analysis_source_count_freeze(); +drop function if exists reject_analysis_source_count_update(); +drop function if exists reject_analysis_source_snapshot_update(); + +delete from common_lookup_value + where lookup_code in ( + 'analysis_run_lineage', + 'analysis_run_report', + 'analysis_run_tepp', + 'analysis_status_pending', + 'analysis_status_running', + 'analysis_status_succeeded', + 'analysis_status_failed', + 'analysis_status_cancelled', + 'analysis_scope_all_visible', + 'analysis_scope_corporate_entity', + 'analysis_scope_process_unit', + 'analysis_scope_thread_group', + 'analysis_count_source_row', + 'analysis_count_document', + 'analysis_count_thread', + 'analysis_count_lineage_node', + 'analysis_count_lineage_edge' + ); + +commit; diff --git a/pyproject.toml b/pyproject.toml index fe1ad488d..9238d87a2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "lineageweave" -version = "0.78.0" +version = "0.82.0" description = "Reconstructs git-branch-style lineage DAGs from scattered short records using multi-channel score fusion and LLM adjudication." readme = "README.md" license = { text = "MIT" } diff --git a/scripts/seed_demo_data.py b/scripts/seed_demo_data.py index 806014e27..dec4de9ad 100644 --- a/scripts/seed_demo_data.py +++ b/scripts/seed_demo_data.py @@ -111,6 +111,7 @@ def seed( cur.execute((migrations / "0014_role_responsibility_team_actor_type.sql").read_text()) cur.execute((migrations / "0015_organization_name_resolution.sql").read_text()) cur.execute((migrations / "0016_cross_post_actor_identity.sql").read_text()) + cur.execute((migrations / "0018_analysis_run_registry.sql").read_text()) cur.execute( """ insert into common_lookup_value (lookup_category, lookup_code, lookup_label, display_order) values @@ -322,6 +323,11 @@ def seed( corporate_entity_id, process_units["DEMO-PU-LINEAGE"], ) + _seed_demo_analysis_run( + cur, + account_ids["demo.analyst"], + corporate_entity_id, + ) conn.commit() finally: @@ -1192,6 +1198,101 @@ def _seed_demo_period_report(cur, author_account_id, corporate_entity_id, proces _persist_seed_period_report(cur, "process_unit", high_key, w03, week3[high_key]) +def _seed_demo_analysis_run(cur, requested_by_account_id, corporate_entity_id) -> None: + """Insert one Demo-Corp lineage run so Analysis runs is not empty. + + Aggregates only: three synthetic documents, one thread. The digest is + a hash of a fixed demo contract string -- never a source row or DSN. + """ + import hashlib + + digest = hashlib.sha256(b"lineageweave-synthetic-demo-snapshot-v1").hexdigest() + cur.execute( + "select analysis_source_snapshot_id from analysis_source_snapshot " + "where snapshot_sha256 = %s", + (digest,), + ) + snapshot_row = cur.fetchone() + if snapshot_row is None: + cur.execute( + """ + insert into analysis_source_snapshot + (snapshot_sha256, source_contract_version, + maximum_available_time, captured_at) + values (%s, 'demo-source-contract-v1', + '2026-01-12T00:00:00Z', '2026-01-12T00:05:00Z') + returning analysis_source_snapshot_id + """, + (digest,), + ) + snapshot_id = cur.fetchone()[0] + else: + snapshot_id = snapshot_row[0] + cur.execute( + """ + insert into analysis_source_count + (analysis_source_snapshot_id, count_type_code, count_value) + values + (%s, 'analysis_count_document', 3), + (%s, 'analysis_count_thread', 1), + (%s, 'analysis_count_lineage_node', 5), + (%s, 'analysis_count_lineage_edge', 4) + on conflict do nothing + """, + (snapshot_id, snapshot_id, snapshot_id, snapshot_id), + ) + cur.execute( + """ + select analysis_run_id from analysis_run + where requested_by_account_id = %s + and idempotency_key = 'demo-lineage-seed-2026-w02' + """, + (requested_by_account_id,), + ) + run_row = cur.fetchone() + if run_row is None: + cur.execute( + """ + insert into analysis_run + (analysis_source_snapshot_id, run_kind_code, idempotency_key, + requested_by_account_id, knowledge_cutoff, + configuration_schema_version, configuration_sha256, + code_revision_sha, requested_at) + values (%s, 'analysis_run_lineage', 'demo-lineage-seed-2026-w02', + %s, '2026-01-12T12:00:00Z', 'lineage-run-v1', %s, %s, + '2026-01-12T12:30:00Z') + returning analysis_run_id + """, + (snapshot_id, requested_by_account_id, "b" * 64, "c" * 40), + ) + run_id = cur.fetchone()[0] + else: + run_id = run_row[0] + cur.execute( + """ + insert into analysis_run_scope + (analysis_run_id, scope_kind_code, corporate_entity_id) + values (%s, 'analysis_scope_corporate_entity', %s) + on conflict (analysis_run_id) do nothing + """, + (run_id, corporate_entity_id), + ) + for ordinal, status, occurred in ( + (1, "analysis_status_pending", "2026-01-12T12:31:00Z"), + (2, "analysis_status_running", "2026-01-12T12:32:00Z"), + (3, "analysis_status_succeeded", "2026-01-12T12:33:00Z"), + ): + cur.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, %s, %s, %s) + on conflict do nothing + """, + (run_id, ordinal, status, occurred), + ) + + def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--postgres-dsn", default=DEFAULT_POSTGRES_DSN) diff --git a/tests/test_analysis_run_authorization.py b/tests/test_analysis_run_authorization.py new file mode 100644 index 000000000..730825c14 --- /dev/null +++ b/tests/test_analysis_run_authorization.py @@ -0,0 +1,254 @@ +"""SQL authorization for the Milestone 2 analysis-run read projection.""" + +from __future__ import annotations + +import os +import uuid +from pathlib import Path +from urllib.parse import urlsplit, urlunsplit + +import psycopg2 +import pytest +from psycopg2 import sql + +_ROOT = Path(__file__).resolve().parents[1] +_INITIAL_MIGRATION = _ROOT / "migrations" / "0001_initial_schema.sql" +_REGISTRY_MIGRATION = _ROOT / "migrations" / "0018_analysis_run_registry.sql" +_ADMIN_DSN = os.environ.get( + "LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN", "postgresql://localhost/postgres" +) + + +def _postgres_available() -> bool: + """Return whether the configured administrator DSN is reachable.""" + try: + psycopg2.connect(_ADMIN_DSN, connect_timeout=2).close() + return True + except psycopg2.OperationalError: + return False + + +def _database_dsn(database_name: str) -> str: + """Replace only the database path while preserving DSN query options.""" + parsed = urlsplit(_ADMIN_DSN) + return urlunsplit(parsed._replace(path=f"/{database_name}")) + + +@pytest.fixture +def authz_db(): + """Yield a throwaway database migrated through the registry schema.""" + if not _postgres_available(): + pytest.skip("a reachable PostgreSQL administrator DSN is required") + database_name = f"lineageweave_authz_{uuid.uuid4().hex[:12]}" + admin_connection = psycopg2.connect(_ADMIN_DSN) + admin_connection.autocommit = True + with admin_connection.cursor() as cursor: + cursor.execute( + sql.SQL("create database {}").format(sql.Identifier(database_name)) + ) + try: + connection = psycopg2.connect(_database_dsn(database_name)) + try: + connection.autocommit = True + with connection.cursor() as cursor: + cursor.execute(_INITIAL_MIGRATION.read_text(encoding="utf-8")) + cursor.execute(_REGISTRY_MIGRATION.read_text(encoding="utf-8")) + yield connection + finally: + connection.close() + finally: + with admin_connection.cursor() as cursor: + cursor.execute( + sql.SQL("drop database {}").format(sql.Identifier(database_name)) + ) + admin_connection.close() + + +def _insert_account(cursor, label: str) -> str: + """Insert one synthetic authenticated account and return its UUID.""" + suffix = uuid.uuid4().hex + cursor.execute( + """ + insert into user_account + (external_subject_id, display_name, email_address) + values (%s, %s, %s) + returning user_account_id + """, + (f"{label}-{suffix}", f"{label.title()} User", f"{label}-{suffix}@example.test"), + ) + return str(cursor.fetchone()[0]) + + +def _insert_corp(cursor, code: str, name: str) -> str: + """Insert one synthetic corporate entity.""" + cursor.execute( + """ + insert into common_lookup_value (lookup_category, lookup_code, lookup_label) + values ('corporate_entity_level', 'company', 'Company') + on conflict (lookup_code) do nothing + """ + ) + cursor.execute( + """ + insert into corporate_entity (corporate_entity_code, entity_name, entity_level_code) + values (%s, %s, 'company') + returning corporate_entity_id + """, + (code, name), + ) + return str(cursor.fetchone()[0]) + + +def _complete_run( + cursor, + *, + account_id: str, + digest: str, + idempotency_key: str, + scope_kind: str, + corporate_entity_id: str | None = None, +) -> str: + """Insert one succeeded run with one document-count aggregate.""" + cursor.execute( + """ + insert into analysis_source_snapshot + (snapshot_sha256, source_contract_version, + maximum_available_time, captured_at) + values (%s, 'source-contract-v1', + '2026-01-12T00:00:00Z', '2026-01-12T00:05:00Z') + returning analysis_source_snapshot_id + """, + (digest,), + ) + snapshot_id = cursor.fetchone()[0] + cursor.execute( + """ + insert into analysis_source_count + (analysis_source_snapshot_id, count_type_code, count_value) + values (%s, 'analysis_count_document', 3) + """, + (snapshot_id,), + ) + cursor.execute( + """ + insert into analysis_run + (analysis_source_snapshot_id, run_kind_code, idempotency_key, + requested_by_account_id, knowledge_cutoff, + configuration_schema_version, configuration_sha256, + code_revision_sha, requested_at) + values (%s, 'analysis_run_lineage', %s, %s, + '2026-01-12T12:00:00Z', 'lineage-run-v1', %s, %s, + '2026-01-12T12:30:00Z') + returning analysis_run_id + """, + (snapshot_id, idempotency_key, account_id, "b" * 64, "c" * 40), + ) + run_id = str(cursor.fetchone()[0]) + if scope_kind == "analysis_scope_corporate_entity": + cursor.execute( + """ + insert into analysis_run_scope + (analysis_run_id, scope_kind_code, corporate_entity_id) + values (%s, %s, %s) + """, + (run_id, scope_kind, corporate_entity_id), + ) + else: + cursor.execute( + """ + insert into analysis_run_scope + (analysis_run_id, scope_kind_code) + values (%s, %s) + """, + (run_id, scope_kind), + ) + for ordinal, status, occurred in ( + (1, "analysis_status_pending", "2026-01-12T12:31:00Z"), + (2, "analysis_status_running", "2026-01-12T12:32:00Z"), + (3, "analysis_status_succeeded", "2026-01-12T12:33:00Z"), + ): + cursor.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values (%s, %s, %s, %s) + """, + (run_id, ordinal, status, occurred), + ) + return run_id + + +def _visible_ids(cursor, account_id: str, entity_ids: list[str]) -> set[str]: + """Apply the same visibility predicate the product API uses.""" + cursor.execute( + """ + select run.analysis_run_id + from analysis_run run + join analysis_run_scope scope on scope.analysis_run_id = run.analysis_run_id + where + run.requested_by_account_id = %s + or ( + scope.scope_kind_code = 'analysis_scope_corporate_entity' + and scope.corporate_entity_id = any(%s::uuid[]) + ) + or ( + scope.scope_kind_code = 'analysis_scope_process_unit' + and exists ( + select 1 from account_affiliation aff + where aff.user_account_id = %s + and aff.process_unit_id = scope.process_unit_id + ) + ) + """, + (account_id, entity_ids, account_id), + ) + return {str(row[0]) for row in cursor.fetchall()} + + +def test_hidden_scope_does_not_leak_through_all_visible_or_other_corp(authz_db) -> None: + """A Demo-Corp viewer never sees another tenant's run or its aggregates.""" + with authz_db.cursor() as cursor: + viewer = _insert_account(cursor, "viewer") + outsider = _insert_account(cursor, "outsider") + own_corp = _insert_corp(cursor, "DEMO-CORP-AUTHZ", "Demo Corp") + other_corp = _insert_corp(cursor, "OTHER-CORP-AUTHZ", "Other Corp") + cursor.execute( + """ + insert into account_affiliation (user_account_id, corporate_entity_id) + values (%s, %s) + """, + (viewer, own_corp), + ) + own_run = _complete_run( + cursor, + account_id=viewer, + digest="a" * 64, + idempotency_key="own-corp", + scope_kind="analysis_scope_corporate_entity", + corporate_entity_id=own_corp, + ) + hidden_all_visible = _complete_run( + cursor, + account_id=outsider, + digest="d" * 64, + idempotency_key="hidden-all", + scope_kind="analysis_scope_all_visible", + ) + hidden_other_corp = _complete_run( + cursor, + account_id=outsider, + digest="e" * 64, + idempotency_key="hidden-other", + scope_kind="analysis_scope_corporate_entity", + corporate_entity_id=other_corp, + ) + + visible = _visible_ids(cursor, viewer, [own_corp]) + assert own_run in visible + assert hidden_all_visible not in visible + assert hidden_other_corp not in visible + + outsider_visible = _visible_ids(cursor, outsider, [other_corp]) + assert hidden_all_visible in outsider_visible + assert hidden_other_corp in outsider_visible + assert own_run not in outsider_visible diff --git a/tests/test_analysis_run_registry_schema.py b/tests/test_analysis_run_registry_schema.py new file mode 100644 index 000000000..f2b38badf --- /dev/null +++ b/tests/test_analysis_run_registry_schema.py @@ -0,0 +1,716 @@ +"""Real-PostgreSQL contracts for the normalized Milestone 2 run registry.""" + +from __future__ import annotations + +import os +import re +import uuid +from pathlib import Path +from urllib.parse import urlsplit, urlunsplit + +import psycopg2 +import psycopg2.errors +import pytest +from psycopg2 import sql + +_ROOT = Path(__file__).resolve().parents[1] +_INITIAL_MIGRATION = _ROOT / "migrations" / "0001_initial_schema.sql" +_REGISTRY_MIGRATION = _ROOT / "migrations" / "0018_analysis_run_registry.sql" +_REGISTRY_ROLLBACK = _ROOT / "migrations" / "rollback" / "0018_analysis_run_registry.sql" +_POSTGRES_IMAGE = _ROOT / "docker" / "postgres-init" / "Dockerfile" +_ADMIN_DSN = os.environ.get( + "LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN", "postgresql://localhost/postgres" +) +_REQUIRED_TABLES = { + "analysis_source_snapshot", + "analysis_source_count", + "analysis_run", + "analysis_run_scope", + "analysis_run_status_event", +} +_REQUIRED_LOOKUP_CODES = { + "analysis_run_lineage", + "analysis_run_report", + "analysis_run_tepp", + "analysis_status_pending", + "analysis_status_running", + "analysis_status_succeeded", + "analysis_status_failed", + "analysis_status_cancelled", + "analysis_scope_all_visible", + "analysis_scope_corporate_entity", + "analysis_scope_process_unit", + "analysis_scope_thread_group", + "analysis_count_source_row", + "analysis_count_document", + "analysis_count_thread", + "analysis_count_lineage_node", + "analysis_count_lineage_edge", +} + + +def _postgres_available() -> bool: + """Return whether the configured administrator DSN is reachable.""" + + try: + psycopg2.connect(_ADMIN_DSN, connect_timeout=2).close() + return True + except psycopg2.OperationalError: + return False + + +def _database_dsn(database_name: str) -> str: + """Replace only the database path while preserving DSN query options.""" + + parsed = urlsplit(_ADMIN_DSN) + return urlunsplit(parsed._replace(path=f"/{database_name}")) + + +def _table_definition(migration: str, table_name: str) -> str: + """Return one table definition from the deterministic migration text.""" + + match = re.search( + rf"create table if not exists {re.escape(table_name)}\s*\((.*?)\n\);", + migration, + re.IGNORECASE | re.DOTALL, + ) + assert match is not None, table_name + return match.group(1) + + +@pytest.fixture +def registry_db(): + """Yield a throwaway database migrated through the registry schema.""" + + if not _postgres_available(): + pytest.skip("a reachable PostgreSQL administrator DSN is required") + database_name = f"lineageweave_registry_{uuid.uuid4().hex[:12]}" + admin_connection = psycopg2.connect(_ADMIN_DSN) + admin_connection.autocommit = True + with admin_connection.cursor() as cursor: + cursor.execute( + sql.SQL("create database {}").format(sql.Identifier(database_name)) + ) + try: + connection = psycopg2.connect(_database_dsn(database_name)) + try: + connection.autocommit = True + with connection.cursor() as cursor: + cursor.execute(_INITIAL_MIGRATION.read_text(encoding="utf-8")) + cursor.execute(_REGISTRY_MIGRATION.read_text(encoding="utf-8")) + yield connection + finally: + connection.close() + finally: + with admin_connection.cursor() as cursor: + cursor.execute( + sql.SQL("drop database {}").format(sql.Identifier(database_name)) + ) + admin_connection.close() + + +def _insert_account(cursor, label: str = "operator") -> str: + """Insert one synthetic authenticated account and return its UUID.""" + + suffix = uuid.uuid4().hex + cursor.execute( + """ + insert into user_account + (external_subject_id, display_name, email_address) + values (%s, %s, %s) + returning user_account_id + """, + (f"{label}-{suffix}", f"{label.title()} User", f"{label}-{suffix}@example.test"), + ) + return str(cursor.fetchone()[0]) + + +def _insert_snapshot( + cursor, + *, + digest: str = "a" * 64, + maximum_available_time: str = "2026-08-15T00:00:00Z", + captured_at: str = "2026-08-15T00:05:00Z", +) -> str: + """Insert one immutable source snapshot and return its UUID.""" + + cursor.execute( + """ + insert into analysis_source_snapshot + (snapshot_sha256, source_contract_version, + maximum_available_time, captured_at) + values (%s, 'source-contract-v1', %s, %s) + returning analysis_source_snapshot_id + """, + (digest, maximum_available_time, captured_at), + ) + return str(cursor.fetchone()[0]) + + +def _insert_run( + cursor, + *, + snapshot_id: str, + account_id: str, + idempotency_key: str, + knowledge_cutoff: str = "2026-08-15T00:30:00Z", + run_kind_code: str = "analysis_run_lineage", + requested_at: str = "2026-08-15T00:45:00Z", +) -> str: + """Insert one immutable account-scoped analysis request.""" + + cursor.execute( + """ + insert into analysis_run + (analysis_source_snapshot_id, run_kind_code, idempotency_key, + requested_by_account_id, knowledge_cutoff, + configuration_schema_version, configuration_sha256, + code_revision_sha, requested_at) + values (%s, %s, %s, %s, %s, 'lineage-run-v1', %s, %s, %s) + returning analysis_run_id + """, + ( + snapshot_id, + run_kind_code, + idempotency_key, + account_id, + knowledge_cutoff, + "b" * 64, + "c" * 40, + requested_at, + ), + ) + return str(cursor.fetchone()[0]) + + +def test_registry_contract_is_normalized_and_has_one_temporal_authority() -> None: + """Static contract rejects the parallel prototype and duplicated clocks.""" + + migration = _REGISTRY_MIGRATION.read_text(encoding="utf-8") + rollback = _REGISTRY_ROLLBACK.read_text(encoding="utf-8") + dockerfile = _POSTGRES_IMAGE.read_text(encoding="utf-8") + created_tables = set( + re.findall(r"create table if not exists\s+([a-z0-9_]+)", migration, re.I) + ) + assert _REQUIRED_TABLES <= created_tables + assert "analysis_run_records" not in created_tables + assert "metadata_payload" not in migration + assert "jsonb" not in migration.casefold() + assert _REQUIRED_LOOKUP_CODES <= set( + re.findall(r"'(analysis_[a-z0-9_]+)'", migration) + ) + assert "0018_analysis_run_registry.sql" in dockerfile + assert "analysis_run_registry_not_empty" in rollback + + snapshot_definition = _table_definition(migration, "analysis_source_snapshot") + run_definition = _table_definition(migration, "analysis_run") + assert "maximum_available_time" in snapshot_definition + assert "knowledge_cutoff" not in snapshot_definition + assert "knowledge_cutoff" in run_definition + assert "requested_by_account_id uuid not null" in run_definition + assert "unique (requested_by_account_id, idempotency_key)" in run_definition + assert "enforce_analysis_run_knowledge_cutoff" in migration + assert "reject_analysis_source_snapshot_update" in migration + assert "reject_analysis_run_mutation" in migration + assert "reject_analysis_run_scope_mutation" in migration + assert "analysis_run_scope_required" in migration + assert "enforce_analysis_source_count_freeze" in migration + assert "enforce_analysis_run_status_transition" in migration + assert "analysis_run_current_status" in migration + + object_patterns = ( + r"create table if not exists\s+([a-z0-9_]+)", + r"create(?: unique)? index if not exists\s+([a-z0-9_]+)", + r"create or replace function\s+([a-z0-9_]+)", + r"create trigger\s+([a-z0-9_]+)", + r"create or replace view\s+([a-z0-9_]+)", + ) + for pattern in object_patterns: + for object_name in re.findall(pattern, migration, re.I): + assert len(object_name.split("_")) >= 2, object_name + + +def test_registry_migration_is_idempotent(registry_db) -> None: + """Sequential migration replay preserves one object set.""" + + with registry_db.cursor() as cursor: + cursor.execute(_REGISTRY_MIGRATION.read_text(encoding="utf-8")) + cursor.execute( + "select table_name from information_schema.tables " + "where table_schema = 'public'" + ) + tables = {row[0] for row in cursor.fetchall()} + cursor.execute( + "select table_name from information_schema.views " + "where table_schema = 'public'" + ) + views = {row[0] for row in cursor.fetchall()} + assert _REQUIRED_TABLES <= tables + assert "analysis_run_current_status" in views + + +def test_registry_persists_scope_counts_and_legal_status_history(registry_db) -> None: + """A valid run keeps normalized scope, counts, and current status.""" + + with registry_db.cursor() as cursor: + account_id = _insert_account(cursor) + snapshot_id = _insert_snapshot(cursor) + cursor.execute( + "insert into analysis_source_count values " + "(%s, 'analysis_count_document', 12)", + (snapshot_id,), + ) + run_id = _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=account_id, + idempotency_key="synthetic-run-1", + ) + cursor.execute( + "insert into analysis_run_scope " + "(analysis_run_id, scope_kind_code) " + "values (%s, 'analysis_scope_all_visible')", + (run_id,), + ) + cursor.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values + (%s, 1, 'analysis_status_pending', '2026-08-15T01:00:01Z'), + (%s, 2, 'analysis_status_running', '2026-08-15T01:00:02Z'), + (%s, 3, 'analysis_status_succeeded', '2026-08-15T01:00:03Z') + """, + (run_id, run_id, run_id), + ) + cursor.execute( + "select status_code, status_ordinal from analysis_run_current_status " + "where analysis_run_id = %s", + (run_id,), + ) + assert cursor.fetchone() == ("analysis_status_succeeded", 3) + + +def test_snapshot_supports_multiple_run_owned_cutoffs_and_blocks_future_evidence( + registry_db, +) -> None: + """One capture is reusable, but each run must respect its own cutoff.""" + + with registry_db.cursor() as cursor: + snapshot_id = _insert_snapshot(cursor) + first_account_id = _insert_account(cursor, "first") + second_account_id = _insert_account(cursor, "second") + first_run_id = _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=first_account_id, + idempotency_key="cutoff-one", + knowledge_cutoff="2026-08-15T00:30:00Z", + ) + second_run_id = _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=second_account_id, + idempotency_key="cutoff-two", + knowledge_cutoff="2026-08-16T00:00:00Z", + requested_at="2026-08-16T00:30:00Z", + ) + assert first_run_id != second_run_id + with pytest.raises(psycopg2.errors.RaiseException): + _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=first_account_id, + idempotency_key="future-leakage", + knowledge_cutoff="2026-08-14T23:59:59Z", + ) + + +def test_snapshot_counts_and_run_request_are_immutable(registry_db) -> None: + """Evidence and request configuration freeze before derivation starts.""" + + with registry_db.cursor() as cursor: + snapshot_id = _insert_snapshot(cursor) + cursor.execute( + "insert into analysis_source_count values " + "(%s, 'analysis_count_document', 12)", + (snapshot_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "update analysis_source_snapshot set source_contract_version = 'x' " + "where analysis_source_snapshot_id = %s", + (snapshot_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "update analysis_source_count set count_value = 13 " + "where analysis_source_snapshot_id = %s", + (snapshot_id,), + ) + account_id = _insert_account(cursor) + run_id = _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=account_id, + idempotency_key="freeze-evidence", + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "update analysis_run set knowledge_cutoff = now() " + "where analysis_run_id = %s", + (run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "insert into analysis_source_count values " + "(%s, 'analysis_count_thread', 8)", + (snapshot_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "delete from analysis_source_count " + "where analysis_source_snapshot_id = %s", + (snapshot_id,), + ) + + +def test_idempotency_is_scoped_to_the_authenticated_account(registry_db) -> None: + """Two actors may use one opaque key; one actor may not reuse it.""" + + with registry_db.cursor() as cursor: + snapshot_id = _insert_snapshot(cursor) + first_account_id = _insert_account(cursor, "first") + second_account_id = _insert_account(cursor, "second") + _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=first_account_id, + idempotency_key="shared-key", + ) + _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=second_account_id, + idempotency_key="shared-key", + ) + with pytest.raises(psycopg2.errors.UniqueViolation): + _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=first_account_id, + idempotency_key="shared-key", + ) + + +def test_registry_rejects_invalid_evidence_and_missing_actor(registry_db) -> None: + """Database constraints reject malformed audit evidence before persistence.""" + + with registry_db.cursor() as cursor: + with pytest.raises(psycopg2.errors.CheckViolation): + cursor.execute( + "insert into analysis_source_snapshot " + "(snapshot_sha256, source_contract_version, " + "maximum_available_time, captured_at) " + "values ('bad', 'source-contract-v1', now(), now())" + ) + snapshot_id = _insert_snapshot(cursor) + with pytest.raises(psycopg2.errors.CheckViolation): + cursor.execute( + "insert into analysis_source_count values " + "(%s, 'analysis_count_source_row', -1)", + (snapshot_id,), + ) + with pytest.raises(psycopg2.errors.NotNullViolation): + cursor.execute( + """ + insert into analysis_run + (analysis_source_snapshot_id, run_kind_code, idempotency_key, + knowledge_cutoff, configuration_schema_version, + configuration_sha256, code_revision_sha) + values (%s, 'analysis_run_report', 'missing-actor', now(), + 'report-run-v1', %s, %s) + """, + (snapshot_id, "d" * 64, "e" * 40), + ) + + +def test_status_history_enforces_shape_order_time_and_legal_transitions( + registry_db, +) -> None: + """Append-only status evidence is a serialized state machine.""" + + with registry_db.cursor() as cursor: + snapshot_id = _insert_snapshot(cursor) + account_id = _insert_account(cursor) + first_run_id = _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=account_id, + idempotency_key="first-status", + ) + cursor.execute( + "insert into analysis_run_scope " + "(analysis_run_id, scope_kind_code) " + "values (%s, 'analysis_scope_all_visible')", + (first_run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at) " + "values (%s, 1, 'analysis_status_running', now())", + (first_run_id,), + ) + + second_run_id = _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=account_id, + idempotency_key="second-status", + ) + cursor.execute( + "insert into analysis_run_scope " + "(analysis_run_id, scope_kind_code) " + "values (%s, 'analysis_scope_all_visible')", + (second_run_id,), + ) + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at) " + "values (%s, 1, 'analysis_status_pending', " + "'2026-08-15T01:00:00Z')", + (second_run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at) " + "values (%s, 3, 'analysis_status_running', " + "'2026-08-15T01:00:01Z')", + (second_run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at) " + "values (%s, 2, 'analysis_status_succeeded', " + "'2026-08-15T01:00:01Z')", + (second_run_id,), + ) + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at) " + "values (%s, 2, 'analysis_status_running', " + "'2026-08-15T01:00:02Z')", + (second_run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at) " + "values (%s, 3, 'analysis_status_succeeded', " + "'2026-08-15T01:00:01Z')", + (second_run_id,), + ) + with pytest.raises(psycopg2.errors.CheckViolation): + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at) " + "values (%s, 3, 'analysis_status_failed', " + "'2026-08-15T01:00:03Z')", + (second_run_id,), + ) + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at) " + "values (%s, 3, 'analysis_status_succeeded', " + "'2026-08-15T01:00:03Z')", + (second_run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at) " + "values (%s, 4, 'analysis_status_running', " + "'2026-08-15T01:00:04Z')", + (second_run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "update analysis_run_status_event set retryable = true " + "where analysis_run_id = %s and status_ordinal = 3", + (second_run_id,), + ) + + + +def test_run_scope_and_request_evidence_are_immutable(registry_db) -> None: + """Authorization scope and request identity cannot be rewritten or erased.""" + + with registry_db.cursor() as cursor: + snapshot_id = _insert_snapshot(cursor) + account_id = _insert_account(cursor) + run_id = _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=account_id, + idempotency_key="immutable-run", + ) + cursor.execute( + "insert into analysis_run_scope " + "(analysis_run_id, scope_kind_code) " + "values (%s, 'analysis_scope_all_visible')", + (run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "update analysis_run_scope set scope_kind_code = scope_kind_code " + "where analysis_run_id = %s", + (run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "delete from analysis_run_scope where analysis_run_id = %s", + (run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "delete from analysis_run where analysis_run_id = %s", + (run_id,), + ) + + +def test_status_requires_scope_and_cannot_predate_request(registry_db) -> None: + """Lifecycle evidence starts only after an immutable authorized request.""" + + with registry_db.cursor() as cursor: + snapshot_id = _insert_snapshot(cursor) + account_id = _insert_account(cursor) + run_id = _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=account_id, + idempotency_key="scoped-status", + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at) " + "values (%s, 1, 'analysis_status_pending', " + "'2026-08-15T01:00:00Z')", + (run_id,), + ) + cursor.execute( + "insert into analysis_run_scope " + "(analysis_run_id, scope_kind_code) " + "values (%s, 'analysis_scope_all_visible')", + (run_id,), + ) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at) " + "values (%s, 1, 'analysis_status_pending', " + "'2026-08-15T00:44:59Z')", + (run_id,), + ) + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at, recorded_at) " + "values (%s, 1, 'analysis_status_pending', " + "'2026-08-15T01:00:00Z', '2099-01-01T00:00:00Z') " + "returning recorded_at", + (run_id,), + ) + recorded_at = cursor.fetchone()[0] + assert recorded_at.year < 2099 + + +def test_machine_codes_and_canonical_idempotency_are_fail_closed(registry_db) -> None: + """Audit identifiers are canonical and failure details stay machine-safe.""" + + with registry_db.cursor() as cursor: + snapshot_id = _insert_snapshot(cursor) + account_id = _insert_account(cursor) + with pytest.raises(psycopg2.errors.RaiseException): + _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=account_id, + idempotency_key="future-request", + requested_at="2099-01-01T00:00:00Z", + ) + with pytest.raises(psycopg2.errors.CheckViolation): + _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=account_id, + idempotency_key=" padded-key ", + ) + run_id = _insert_run( + cursor, + snapshot_id=snapshot_id, + account_id=account_id, + idempotency_key="machine-safe", + ) + cursor.execute( + "insert into analysis_run_scope " + "(analysis_run_id, scope_kind_code) " + "values (%s, 'analysis_scope_all_visible')", + (run_id,), + ) + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at) " + "values (%s, 1, 'analysis_status_pending', " + "'2026-08-15T01:00:00Z')", + (run_id,), + ) + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at) " + "values (%s, 2, 'analysis_status_running', " + "'2026-08-15T01:00:00Z')", + (run_id,), + ) + with pytest.raises(psycopg2.errors.CheckViolation): + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at, " + "failure_code, retryable) " + "values (%s, 3, 'analysis_status_failed', " + "'2026-08-15T01:00:00Z', 'provider timeout', true)", + (run_id,), + ) + cursor.execute( + "insert into analysis_run_status_event " + "(analysis_run_id, status_ordinal, status_code, occurred_at, " + "failure_code, retryable) " + "values (%s, 3, 'analysis_status_failed', " + "'2026-08-15T01:00:00Z', 'provider_timeout', true)", + (run_id,), + ) + +def test_rollback_refuses_data_loss_then_removes_an_empty_registry(registry_db) -> None: + """Downgrade fails closed until audit evidence is explicitly removed.""" + + rollback_sql = _REGISTRY_ROLLBACK.read_text(encoding="utf-8") + with registry_db.cursor() as cursor: + snapshot_id = _insert_snapshot(cursor) + with pytest.raises(psycopg2.errors.RaiseException): + cursor.execute(rollback_sql) + # The rollback script opens an explicit transaction on this + # autocommit connection. A RAISE leaves that transaction aborted, and + # connection.rollback() is a no-op while autocommit is true. + cursor.execute("rollback") + with registry_db.cursor() as cursor: + cursor.execute( + "delete from analysis_source_snapshot " + "where analysis_source_snapshot_id = %s", + (snapshot_id,), + ) + cursor.execute(rollback_sql) + cursor.execute("select to_regclass('public.analysis_run')") + assert cursor.fetchone()[0] is None + cursor.execute(rollback_sql) diff --git a/uv.lock b/uv.lock index 6d56dde98..f3307cb32 100644 --- a/uv.lock +++ b/uv.lock @@ -454,7 +454,7 @@ wheels = [ [[package]] name = "lineageweave" -version = "0.78.0" +version = "0.82.0" source = { virtual = "." } dependencies = [ { name = "certifi" },