diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 13f0a4599..eba7dd90b 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -454,6 +454,20 @@ 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. +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". + ## 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.md b/CHANGELOG.md index 87789e2d7..43a74a4ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,18 @@ 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.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 diff --git a/backend/app/analysis_run_ingestion.py b/backend/app/analysis_run_ingestion.py new file mode 100644 index 000000000..86b75ef89 --- /dev/null +++ b/backend/app/analysis_run_ingestion.py @@ -0,0 +1,190 @@ +"""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, + 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 _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"] + return detail diff --git a/backend/app/main.py b/backend/app/main.py index c0214f0c0..81630d35b 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,50 @@ 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.""" + _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..3a6bab603 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,51 @@ 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() + + detail = client.get( + f"/api/analysis-runs/{seeded_db['visible_run_id']}", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + ) + assert detail.status_code == 200 + assert detail.json()["configuration_schema_version"] == "lineage-run-v1" + assert "snapshot_sha256" not in detail.json() + + 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/docs/adr/0014-authorized-analysis-run-read.md b/docs/adr/0014-authorized-analysis-run-read.md new file mode 100644 index 000000000..0621614d3 --- /dev/null +++ b/docs/adr/0014-authorized-analysis-run-read.md @@ -0,0 +1,47 @@ +# 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. +- 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. +Write/rebuild APIs, TEPP submission, and an 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/frontend/package.json b/frontend/package.json index eef0c8735..cde22610e 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,7 +1,7 @@ { "name": "frontend", "private": true, - "version": "0.78.0", + "version": "0.79.0", "type": "module", "scripts": { "dev": "vite", diff --git a/frontend/src/App.test.tsx b/frontend/src/App.test.tsx index 85346bc88..f7e89a3d3 100644 --- a/frontend/src/App.test.tsx +++ b/frontend/src/App.test.tsx @@ -169,6 +169,33 @@ describe("App, authenticated", () => { jsonResponse({ post_id: "post-1", has_commitment: true, ticket }), ); } + 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 +1323,18 @@ 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 "); + }); + 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..67cd99a9d 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -7,6 +7,7 @@ import { deriveCommitment, evaluatePost, extractPostKeymen, + fetchAnalysisRuns, fetchCalendar, fetchLineageGraph, fetchMe, @@ -33,6 +34,7 @@ import { verifyPostRelations, type ActivityEvent, type AffiliateNode, + type AnalysisRun, type CalendarEntry, type ChatAnswer, type ChatExchange, @@ -1344,6 +1346,58 @@ function PostDetailPopup({ ); } +function AnalysisRunsPanel({ accessToken }: { accessToken: string }) { + const [runs, setRuns] = useState(null); + const [error, setError] = useState(null); + + useEffect(() => { + fetchAnalysisRuns(accessToken) + .then((payload) => setRuns(payload.analysis_runs)) + .catch((err) => setError(String(err))); + }, [accessToken]); + + if (error) return

{error}

; + if (runs === null) return

Loading analysis runs...

; + + return ( +
+
+

Analysis runs

+
+ {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 = [ + run.run_kind_label, + run.status_label, + run.scope_entity_name ?? run.scope_kind_label, + ] + .filter(Boolean) + .join(" · "); + return ( +
  • + {caption} + {documentCount && ( + + {documentCount.count_value} {documentCount.count_type_label.toLowerCase()} + + )} +
  • + ); + })} +
+ )} +
+ ); +} + function CalendarPanel({ accessToken, onSelectPost, @@ -1632,6 +1686,7 @@ function PostList({ accessToken }: { accessToken: string }) { return ( <> +
diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 2f3e3bbb6..91e9f65e4 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -492,3 +492,27 @@ 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 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[]; +} + +export function fetchAnalysisRuns(accessToken: string): Promise<{ analysis_runs: AnalysisRun[] }> { + return backendFetch("/api/analysis-runs", accessToken); +} diff --git a/lineageweave/__init__.py b/lineageweave/__init__.py index be9228d75..a8f40a3cc 100644 --- a/lineageweave/__init__.py +++ b/lineageweave/__init__.py @@ -55,4 +55,4 @@ "sentence_excerpts", ] -__version__ = "0.78.0" +__version__ = "0.79.0" diff --git a/pyproject.toml b/pyproject.toml index fe1ad488d..9f9ed8537 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "lineageweave" -version = "0.78.0" +version = "0.79.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/uv.lock b/uv.lock index 6d56dde98..6d9094f6a 100644 --- a/uv.lock +++ b/uv.lock @@ -454,7 +454,7 @@ wheels = [ [[package]] name = "lineageweave" -version = "0.78.0" +version = "0.79.0" source = { virtual = "." } dependencies = [ { name = "certifi" },