From e3c9188fe2b9796918497ca831dd91b5ebd2003a Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 16 Aug 2026 15:53:53 +0000 Subject: [PATCH] feat: let operators request a pending analysis run (v0.85.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /api/analysis-runs records snapshot, counts, run, scope, and Pending in one transaction. The home button opens that row so a buyer can confirm the cutoff corpus. Reconstruction and TEPP stay later slices — this write never invents a theta. Rebased onto the live #74 head (includes #118, #121, and #124). Failed lineage copy stays kind-specific and does not mention TEPP; only Failed TEPP mentions the measurement service. Seed insert now asserts Failed / tepp_not_available. Co-authored-by: Seongho Bae --- ARCHITECTURE.md | 5 + CHANGELOG.d/0.85.0-analysis-run-create.md | 5 + CHANGELOG.md | 13 + CLAUDE.md | 4 +- backend/app/analysis_run_ingestion.py | 358 ++++++++++++++++++ backend/app/main.py | 49 +++ backend/tests/test_api.py | 66 ++++ .../0013-normalized-analysis-run-registry.md | 2 + docs/adr/0014-authorized-analysis-run-read.md | 6 +- .../0017-authorized-analysis-run-create.md | 56 +++ frontend/package.json | 2 +- frontend/src/App.test.tsx | 47 +++ frontend/src/App.tsx | 34 +- frontend/src/api.ts | 18 + lineageweave/__init__.py | 2 +- pyproject.toml | 2 +- tests/test_analysis_run_create.py | 134 +++++++ tests/test_seed_tepp_run.py | 47 +++ uv.lock | 2 +- 19 files changed, 844 insertions(+), 8 deletions(-) create mode 100644 CHANGELOG.d/0.85.0-analysis-run-create.md create mode 100644 docs/adr/0017-authorized-analysis-run-create.md create mode 100644 tests/test_analysis_run_create.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 063b7a196..da35cc779 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -467,6 +467,11 @@ run's scope whose `created_at` is at or before `knowledge_cutoff` (ADR 0016) so a buyer can open a post the run was allowed to know without seeing later live rows or hidden bodies. Detail also returns revision and configuration digest prefixes. +`POST /api/analysis-runs` records a Pending run on a new authorized +cutoff capture (ADR 0017): snapshot, counts, run, scope, and the first +status in one transaction. It does not reconstruct lineage and does not +invent a TEPP score. Request a lineage reconstruction from the home +list, then open the Pending row to confirm the cutoff corpus. `make seed` also records a TEPP measurement run through `tepp_client` on that same snapshot; the default transport is unavailable, so that run is Failed rather than a fabricated score. diff --git a/CHANGELOG.d/0.85.0-analysis-run-create.md b/CHANGELOG.d/0.85.0-analysis-run-create.md new file mode 100644 index 000000000..505320a19 --- /dev/null +++ b/CHANGELOG.d/0.85.0-analysis-run-create.md @@ -0,0 +1,5 @@ +# 0.85.0 authorized analysis-run create + +`POST /api/analysis-runs` records Pending on an authorized cutoff +capture. Request a lineage reconstruction from the home list. This +write does not invent a measurement. diff --git a/CHANGELOG.md b/CHANGELOG.md index 22b372c35..fec2c509b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,19 @@ 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.85.0] - 2026-08-16 + +### Added + +- `POST /api/analysis-runs` records a Pending lineage or TEPP run on an + authorized cutoff capture (ADR 0017). The home panel's **Request a + lineage reconstruction** button writes that row so an operator can + confirm the cutoff corpus immediately. Reconstruction and live TEPP + execution stay later slices — this write never invents a theta. +- Failed lineage rows tell the operator to retry reconstruction; only + Failed TEPP rows mention the measurement service. Pending rows say + reconstruction has not started yet. + ## [0.84.1] - 2026-08-16 ### Fixed diff --git a/CLAUDE.md b/CLAUDE.md index 71e671038..2f89bada7 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -3,7 +3,7 @@ Tool-specific pointer. Policy lives in [AGENTS.md](AGENTS.md) and the ADRs under `docs/adr/`. Do not fork those rules here. -## Analysis-run seed (v0.84.1) +## Analysis-run seed (v0.85.0) `make seed` writes a Demo Corp lineage run and a TEPP run on the same snapshot (ADR 0013). The TEPP path goes through `tepp_client`. A missing @@ -17,3 +17,5 @@ mention TEPP. Digest prefixes stay audible; hover a prefix to read the full digest. Opening a cutoff title shows the live post -- compare it with the cutoff before treating the body as reconstructed evidence (ADR 0016). +`POST /api/analysis-runs` records Pending on an authorized +cutoff capture (ADR 0017) and does not reconstruct lineage. diff --git a/backend/app/analysis_run_ingestion.py b/backend/app/analysis_run_ingestion.py index e96c2b7c4..6a1449818 100644 --- a/backend/app/analysis_run_ingestion.py +++ b/backend/app/analysis_run_ingestion.py @@ -5,15 +5,33 @@ 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. + +``create_pending_analysis_run`` (ADR 0017) writes snapshot, counts, run, +scope, and the first Pending event atomically. It does not reconstruct +lineage or invent a TEPP score. """ from __future__ import annotations +import hashlib +import json +from dataclasses import dataclass +from datetime import datetime, timezone from typing import Any +from uuid import UUID import asyncpg from backend.app.knowledge_graph import labels_for_codes +from lineageweave import __version__ as PACKAGE_VERSION + +_ALLOWED_CREATE_KINDS = frozenset({"analysis_run_lineage", "analysis_run_tepp"}) +_CORPORATE_SCOPE = "analysis_scope_corporate_entity" +_CAPTURE_CONTRACT_VERSION = "analysis-run-capture-v1" +_KIND_SCHEMA_VERSION = { + "analysis_run_lineage": "lineage-run-v1", + "analysis_run_tepp": "tepp-run-v1", +} _VISIBLE_RUN_SQL = """ run.requested_by_account_id = $1 @@ -289,3 +307,343 @@ async def fetch_visible_scope_posts( continue posts.append({"post_id": str(row["post_id"]), "post_title": row["post_title"]}) return posts + + +class AnalysisRunCreateError(Exception): + """Fail-closed create: HTTP status plus a next-action detail string.""" + + def __init__(self, status_code: int, detail: str) -> None: + super().__init__(detail) + self.status_code = status_code + self.detail = detail + + +@dataclass(frozen=True) +class AnalysisRunCapture: + """Immutable capture plan for one authorized create (no source rows).""" + + snapshot_sha256: str + maximum_available_time: datetime + document_count: int + thread_count: int + configuration_sha256: str + configuration_schema_version: str + code_revision_sha: str + + +def utc_iso(value: datetime) -> str: + """Normalize a timestamp to UTC ISO-8601 for digest stability.""" + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc).isoformat() + + +def plan_analysis_run_capture( + *, + run_kind_code: str, + scope_kind_code: str, + corporate_entity_id: str, + knowledge_cutoff: datetime, + idempotency_key: str, + post_ids: list[str], + thread_keys: list[str], + latest_post_created_at: datetime | None, + cutoff_explicit: bool = True, +) -> AnalysisRunCapture: + """Hash the authorized cutoff bag. Never stores a post body or DSN. + + An omitted cutoff is hashed as ``unspecified`` so a retry of the same + client key does not 409 just because the clock moved. + """ + cutoff_token = utc_iso(knowledge_cutoff) if cutoff_explicit else "unspecified" + snapshot_material = json.dumps( + { + "scope_kind_code": scope_kind_code, + "corporate_entity_id": corporate_entity_id, + "knowledge_cutoff": cutoff_token, + "post_ids": sorted(post_ids), + }, + separators=(",", ":"), + sort_keys=True, + ) + configuration_material = json.dumps( + { + "run_kind_code": run_kind_code, + "scope_kind_code": scope_kind_code, + "corporate_entity_id": corporate_entity_id, + "knowledge_cutoff": cutoff_token, + "idempotency_key": idempotency_key, + "configuration_schema_version": _KIND_SCHEMA_VERSION[run_kind_code], + }, + separators=(",", ":"), + sort_keys=True, + ) + available = latest_post_created_at if latest_post_created_at is not None else knowledge_cutoff + return AnalysisRunCapture( + snapshot_sha256=hashlib.sha256(snapshot_material.encode()).hexdigest(), + maximum_available_time=available, + document_count=len(post_ids), + thread_count=len(set(thread_keys)), + configuration_sha256=hashlib.sha256(configuration_material.encode()).hexdigest(), + configuration_schema_version=_KIND_SCHEMA_VERSION[run_kind_code], + code_revision_sha=hashlib.sha256(f"lineageweave-{PACKAGE_VERSION}".encode()).hexdigest(), + ) + + +def _canonical_idempotency_key(raw: str) -> str: + """Trim and reject empty or control-bearing client keys.""" + key = raw.strip() + if not key or len(key) > 256 or any(ord(char) < 32 for char in key): + raise AnalysisRunCreateError( + 422, + "Use a 1–256 character idempotency key without control characters, then retry.", + ) + return key + + +def _resolve_corporate_entity_id( + corporate_entity_id: str | None, + affiliated_entity_ids: list[str], +) -> str: + """Return the affiliated corp this run may cover, or a next-action error.""" + affiliated = [entity_id for entity_id in affiliated_entity_ids if entity_id] + if corporate_entity_id: + try: + UUID(corporate_entity_id) + except ValueError as exc: + raise AnalysisRunCreateError( + 404, + "This corporate entity is not visible to this account.", + ) from exc + if corporate_entity_id not in affiliated: + raise AnalysisRunCreateError( + 404, + "This corporate entity is not visible to this account.", + ) + return corporate_entity_id + if len(affiliated) != 1: + raise AnalysisRunCreateError( + 422, + "Choose the corporate entity this run should cover.", + ) + return affiliated[0] + + +async def create_pending_analysis_run( + conn: asyncpg.Connection, + *, + account_id: str, + affiliated_entity_ids: list[str], + run_kind_code: str, + scope_kind_code: str, + corporate_entity_id: str | None, + knowledge_cutoff: datetime | None, + idempotency_key: str, +) -> dict[str, Any]: + """Insert snapshot, counts, run, scope, and Pending in one transaction. + + Does not reconstruct lineage and does not call TEPP. A missing + measurement stays a later worker slice; this write only records the + request. Idempotent retries compare ``configuration_sha256``. + """ + if run_kind_code not in _ALLOWED_CREATE_KINDS: + raise AnalysisRunCreateError( + 422, + "Request a lineage reconstruction or a TEPP measurement. Other kinds are not available yet.", + ) + if scope_kind_code != _CORPORATE_SCOPE: + raise AnalysisRunCreateError( + 422, + "Request a corporate-entity run. Other scopes are not available yet.", + ) + cutoff_explicit = knowledge_cutoff is not None + if knowledge_cutoff is None: + knowledge_cutoff = datetime.now(timezone.utc) + elif knowledge_cutoff.tzinfo is None: + knowledge_cutoff = knowledge_cutoff.replace(tzinfo=timezone.utc) + now = datetime.now(timezone.utc) + if knowledge_cutoff > now: + raise AnalysisRunCreateError( + 422, + "Choose a knowledge cutoff at or before now, then request the run again.", + ) + key = _canonical_idempotency_key(idempotency_key) + corp_id = _resolve_corporate_entity_id(corporate_entity_id, affiliated_entity_ids) + + existing = await conn.fetchrow( + """ + select analysis_run_id, configuration_sha256 + from analysis_run + where requested_by_account_id = $1 and idempotency_key = $2 + """, + account_id, + key, + ) + + rows = await conn.fetch( + """ + select post_id, post_title, thread_group_key, created_at, + visibility_code, corporate_entity_id + from source_post + where corporate_entity_id = $1 and created_at <= $2 + order by created_at, post_title + """, + corp_id, + knowledge_cutoff, + ) + affiliated = {str(entity_id) for entity_id in affiliated_entity_ids} + visible_rows = [ + row + for row in rows + if row["visibility_code"] == "public" or str(row["corporate_entity_id"]) in affiliated + ] + post_ids = [str(row["post_id"]) for row in visible_rows] + thread_keys = [row["thread_group_key"] for row in visible_rows] + latest = max((row["created_at"] for row in visible_rows), default=None) + capture = plan_analysis_run_capture( + run_kind_code=run_kind_code, + scope_kind_code=scope_kind_code, + corporate_entity_id=corp_id, + knowledge_cutoff=knowledge_cutoff, + idempotency_key=key, + post_ids=post_ids, + thread_keys=thread_keys, + latest_post_created_at=latest, + cutoff_explicit=cutoff_explicit, + ) + if existing is not None: + if existing["configuration_sha256"] != capture.configuration_sha256: + raise AnalysisRunCreateError( + 409, + "This request does not match the earlier run with the same key. " + "Open that run, or retry with a new idempotency key.", + ) + replayed = await fetch_visible_analysis_run( + conn, + str(existing["analysis_run_id"]), + account_id, + affiliated_entity_ids, + ) + if replayed is None: + raise AnalysisRunCreateError(404, "This analysis run is not visible.") + return replayed + + snapshot_id = await conn.fetchval( + """ + insert into analysis_source_snapshot + (snapshot_sha256, source_contract_version, + maximum_available_time, captured_at, created_at) + values ($1, $2, $3, $4, $4) + on conflict (snapshot_sha256) do nothing + returning analysis_source_snapshot_id + """, + capture.snapshot_sha256, + _CAPTURE_CONTRACT_VERSION, + capture.maximum_available_time, + now, + ) + if snapshot_id is None: + snapshot_id = await conn.fetchval( + """ + select analysis_source_snapshot_id + from analysis_source_snapshot + where snapshot_sha256 = $1 + for update + """, + capture.snapshot_sha256, + ) + count_exists = await conn.fetchval( + """ + select 1 from analysis_source_count + where analysis_source_snapshot_id = $1 + limit 1 + """, + snapshot_id, + ) + if count_exists is None: + await conn.execute( + """ + insert into analysis_source_count + (analysis_source_snapshot_id, count_type_code, count_value) + values + ($1, 'analysis_count_document', $2), + ($1, 'analysis_count_thread', $3) + """, + snapshot_id, + capture.document_count, + capture.thread_count, + ) + try: + run_id = await conn.fetchval( + """ + 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 ($1, $2, $3, $4, $5, $6, $7, $8, $9) + returning analysis_run_id + """, + snapshot_id, + run_kind_code, + key, + account_id, + knowledge_cutoff, + capture.configuration_schema_version, + capture.configuration_sha256, + capture.code_revision_sha, + now, + ) + except asyncpg.UniqueViolationError: + raced = await conn.fetchrow( + """ + select analysis_run_id, configuration_sha256 + from analysis_run + where requested_by_account_id = $1 and idempotency_key = $2 + """, + account_id, + key, + ) + if raced is None or raced["configuration_sha256"] != capture.configuration_sha256: + raise AnalysisRunCreateError( + 409, + "This request does not match the earlier run with the same key. " + "Open that run, or retry with a new idempotency key.", + ) from None + replayed = await fetch_visible_analysis_run( + conn, + str(raced["analysis_run_id"]), + account_id, + affiliated_entity_ids, + ) + if replayed is None: + raise AnalysisRunCreateError(404, "This analysis run is not visible.") + return replayed + await conn.execute( + """ + insert into analysis_run_scope + (analysis_run_id, scope_kind_code, corporate_entity_id) + values ($1, $2, $3) + """, + run_id, + scope_kind_code, + corp_id, + ) + await conn.execute( + """ + insert into analysis_run_status_event + (analysis_run_id, status_ordinal, status_code, occurred_at) + values ($1, 1, 'analysis_status_pending', $2) + """, + run_id, + now, + ) + created = await fetch_visible_analysis_run( + conn, + str(run_id), + account_id, + affiliated_entity_ids, + ) + if created is None: + raise AnalysisRunCreateError(404, "This analysis run is not visible.") + return created diff --git a/backend/app/main.py b/backend/app/main.py index e77b173bc..de06a1167 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -21,6 +21,7 @@ import asyncio from contextlib import asynccontextmanager +from datetime import datetime from typing import Any from uuid import UUID @@ -66,6 +67,8 @@ from lineageweave.relation_verification import NullRelationVerificationClient, SearxngRelationVerificationClient from backend.app.analysis_run_ingestion import ( + AnalysisRunCreateError, + create_pending_analysis_run, fetch_visible_analysis_run, fetch_visible_analysis_runs, ) @@ -1173,6 +1176,52 @@ async def list_analysis_runs( return {"analysis_runs": runs} +class CreateAnalysisRunRequest(BaseModel): + """JSON body for ``POST /api/analysis-runs``. + + Omitting ``corporate_entity_id`` uses the account's sole affiliation. + Reconstruction and TEPP execution stay later slices; this write + records Pending only. + """ + + run_kind_code: str = "analysis_run_lineage" + scope_kind_code: str = "analysis_scope_corporate_entity" + corporate_entity_id: str | None = None + knowledge_cutoff: datetime | None = None + idempotency_key: str + + +@app.post("/api/analysis-runs", status_code=status.HTTP_201_CREATED) +async def create_analysis_run( + request: CreateAnalysisRunRequest, + account: CurrentAccount = Depends(get_current_account), + pool: asyncpg.Pool = Depends(get_pool), +) -> dict[str, Any]: + """Record a Pending analysis run on an authorized cutoff capture. + + post_read is enough: the caller requests a run of a corp they + already walk. The payload is the same authorized detail as GET. + Hidden scopes 404. A matching idempotent retry returns the same run. + """ + _require_post_read(account) + async with pool.acquire() as conn: + async with conn.transaction(): + try: + created = await create_pending_analysis_run( + conn, + account_id=account.user_account_id, + affiliated_entity_ids=list(account.corporate_entity_ids), + run_kind_code=request.run_kind_code, + scope_kind_code=request.scope_kind_code, + corporate_entity_id=request.corporate_entity_id, + knowledge_cutoff=request.knowledge_cutoff, + idempotency_key=request.idempotency_key, + ) + except AnalysisRunCreateError as exc: + raise HTTPException(exc.status_code, exc.detail) from exc + return created + + @app.get("/api/analysis-runs/{analysis_run_id}") async def read_analysis_run( analysis_run_id: str, diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index df1dfb4a1..e62ff0fa1 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -506,6 +506,72 @@ def test_analysis_runs_are_labeled_aggregates_and_hide_other_scopes( assert unauthenticated.status_code == 401 +def test_create_analysis_run_records_pending_without_inventing_a_score( + client, demo_analyst_token, seeded_db +) -> None: + """POST /api/analysis-runs writes Pending on the authorized cutoff bag.""" + created = client.post( + "/api/analysis-runs", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + json={ + "run_kind_code": "analysis_run_lineage", + "corporate_entity_id": seeded_db["own_corp_id"], + "idempotency_key": "buyer-create-2026-w02", + }, + ) + assert created.status_code == 201 + body = created.json() + assert body["run_kind_label"] == "Lineage reconstruction" + assert body["status_label"] == "Pending" + assert body["status_history"][0]["status_label"] == "Pending" + assert all(event["status_label"] != "Succeeded" for event in body["status_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 "theta" not in str(body).lower() + assert "postgresql://" not in str(body) + + replay = client.post( + "/api/analysis-runs", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + json={ + "run_kind_code": "analysis_run_lineage", + "corporate_entity_id": seeded_db["own_corp_id"], + "idempotency_key": "buyer-create-2026-w02", + }, + ) + assert replay.status_code == 201 + assert replay.json()["analysis_run_id"] == body["analysis_run_id"] + + conflict = client.post( + "/api/analysis-runs", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + json={ + "run_kind_code": "analysis_run_tepp", + "corporate_entity_id": seeded_db["own_corp_id"], + "idempotency_key": "buyer-create-2026-w02", + }, + ) + assert conflict.status_code == 409 + + hidden = client.post( + "/api/analysis-runs", + headers={"Authorization": f"Bearer {demo_analyst_token}"}, + json={ + "run_kind_code": "analysis_run_lineage", + "corporate_entity_id": seeded_db["other_corp_id"], + "idempotency_key": "buyer-create-hidden-corp", + }, + ) + assert hidden.status_code == 404 + + unauthenticated = client.post( + "/api/analysis-runs", + json={"idempotency_key": "buyer-create-unauthenticated"}, + ) + 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/0013-normalized-analysis-run-registry.md b/docs/adr/0013-normalized-analysis-run-registry.md index 631c07cd2..b60164dc6 100644 --- a/docs/adr/0013-normalized-analysis-run-registry.md +++ b/docs/adr/0013-normalized-analysis-run-registry.md @@ -238,6 +238,8 @@ Acceptance requires: 1. Add a transaction repository that creates snapshot, counts, run, scope, and first status atomically and compares request digests on idempotent retries. + `POST /api/analysis-runs` now records that Pending write (ADR 0017); + reconstruction and live TEPP execution remain later slices. 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. diff --git a/docs/adr/0014-authorized-analysis-run-read.md b/docs/adr/0014-authorized-analysis-run-read.md index 841dfb383..9188187fb 100644 --- a/docs/adr/0014-authorized-analysis-run-read.md +++ b/docs/adr/0014-authorized-analysis-run-read.md @@ -46,8 +46,10 @@ keeps that machine code off the caption (this decision) and instead tells the operator to open the TEPP run, then connect the measurement service. A failed lineage row tells the operator to retry reconstruction, not to connect TEPP. The detail now shows the legal -lifecycle the registry already stored. Write/rebuild APIs, a live TEPP -transport, and a fuller Analysis Run Console remain later slices. +lifecycle the registry already stored. `POST /api/analysis-runs` now +records a Pending run on an authorized cutoff capture (ADR 0017). +Reconstruction, a live TEPP transport, and a fuller Analysis Run +Console remain later slices. ## References diff --git a/docs/adr/0017-authorized-analysis-run-create.md b/docs/adr/0017-authorized-analysis-run-create.md new file mode 100644 index 000000000..e3a535a18 --- /dev/null +++ b/docs/adr/0017-authorized-analysis-run-create.md @@ -0,0 +1,56 @@ +# ADR 0017 — Operators request an analysis run through the product API + +**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; ADR 0014 authorized +analysis-run read; ADR 0016 knowledge-cutoff posts +**Refs:** Issue #79 (Milestone 2 parent); ADR 0013 follow-up 1 + +## Context + +Home Analysis runs could list seeded lineage and TEPP rows, but a buyer +could not request a new run. Seed-only evidence is a demo, not a product. +ADR 0013 already required a transaction that creates snapshot, counts, +run, scope, and the first status atomically. Follow-up 3 (outbox / worker) +still owns reconstruction and live TEPP execution. + +## Decision + +`POST /api/analysis-runs` is the authorized write: + +- `post_read` is enough. The caller may only cover a corporate entity + they already walk. An unaffiliated corp is 404, not 403. +- The capture digest hashes scope, entity, cutoff, and authorized post + ids — never a post body, DSN, or source SQL. +- The write inserts snapshot, aggregate counts, `analysis_run`, + `analysis_run_scope`, and `analysis_status_pending` in one transaction. +- The first status is Pending. This slice does not reconstruct lineage + and does not call TEPP. A missing measurement stays Failed only on the + seed path that already goes through `tepp_client`. +- Account-scoped idempotency compares `configuration_sha256`. An omitted + cutoff is hashed as `unspecified` so a retry of the same client key + does not conflict because the clock moved. +- The response is the same authorized detail as `GET /api/analysis-runs/{id}`. + +## Consequences + +The home panel's **Request a lineage reconstruction** button records a +Pending row the operator can open immediately. Reconstruction, TEPP +transport, and the outbox worker remain later slices. Do not stamp +Succeeded or invent a theta from this write. + +## 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). + +Jensen, C. S., & Snodgrass, R. T. (1999). Temporal data management. +*IEEE Transactions on Knowledge and Data Engineering, 11*(1), 36–44. +https://doi.org/10.1109/69.755613 + +Moreau, L., & Missier, P. (Eds.). (2013). *PROV-DM: The PROV data model*. +World Wide Web Consortium. https://www.w3.org/TR/prov-dm/ + +World Wide Web Consortium. (2022). *Time ontology in OWL* (W3C +Recommendation). https://www.w3.org/TR/owl-time/ diff --git a/frontend/package.json b/frontend/package.json index 8ce5f334b..c8f67bc8d 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,7 +1,7 @@ { "name": "frontend", "private": true, - "version": "0.84.1", + "version": "0.85.0", "type": "module", "scripts": { "dev": "vite", diff --git a/frontend/src/App.test.tsx b/frontend/src/App.test.tsx index c77d965b5..dae8e1674 100644 --- a/frontend/src/App.test.tsx +++ b/frontend/src/App.test.tsx @@ -269,6 +269,31 @@ describe("App, authenticated", () => { }), ); } + if (url.endsWith("/api/analysis-runs") && method === "POST") { + const created = { + analysis_run_id: "run-demo-lineage-pending", + 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_pending", + status_label: "Pending", + knowledge_cutoff: "2026-01-12T12:00:00Z", + requested_at: "2026-01-12T12:35:00Z", + source_counts: [], + 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:35:00Z", + }, + ], + }; + return Promise.resolve(new Response(JSON.stringify(created), { status: 201 })); + } if (url.endsWith("/api/analysis-runs")) { return Promise.resolve( jsonResponse({ @@ -1561,6 +1586,28 @@ describe("App, authenticated", () => { expect(screen.queryByText(/replace Failed/i)).not.toBeInTheDocument(); }); + it("records a pending lineage run and opens the authorized detail", async () => { + const fetchMock = stubBackend(); + render(); + + await userEvent.click( + await screen.findByRole("button", { name: "Request a lineage reconstruction" }), + ); + expect( + await screen.findByRole("heading", { name: "Lineage reconstruction · Pending · Demo Corp" }), + ).toBeInTheDocument(); + expect(screen.getByText(/has not started yet/)).toBeInTheDocument(); + const postCall = fetchMock.mock.calls.find( + (call) => String(call[0]).endsWith("/api/analysis-runs") && call[1]?.method === "POST", + ); + expect(postCall).toBeDefined(); + const body = JSON.parse(String(postCall?.[1]?.body)); + expect(body.run_kind_code).toBe("analysis_run_lineage"); + expect(body.idempotency_key).toMatch( + /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i, + ); + }); + 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 948035430..e3ddce1ac 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -3,6 +3,7 @@ import { useAuth } from "react-oidc-context"; import { askPostChat, BackendError, + createAnalysisRun, createPostTicket, deriveCommitment, evaluatePost, @@ -1361,6 +1362,9 @@ function analysisRunCaption(run: AnalysisRun): string { * for a missing TEPP transport. */ function analysisRunNextAction(run: AnalysisRun): string | null { + if (run.status_code === "analysis_status_pending") { + return "Open this run to confirm which posts it will use. Reconstruction has not started yet."; + } if (run.status_code !== "analysis_status_failed") { return null; } @@ -1472,6 +1476,7 @@ function AnalysisRunsPanel({ const [runs, setRuns] = useState(null); const [selected, setSelected] = useState(null); const [error, setError] = useState(null); + const [requesting, setRequesting] = useState(false); useEffect(() => { fetchAnalysisRuns(accessToken) @@ -1479,6 +1484,24 @@ function AnalysisRunsPanel({ .catch((err) => setError(String(err))); }, [accessToken]); + async function handleRequestLineage() { + setError(null); + setRequesting(true); + try { + const created = await createAnalysisRun(accessToken, { + run_kind_code: "analysis_run_lineage", + idempotency_key: crypto.randomUUID(), + }); + const listed = await fetchAnalysisRuns(accessToken); + setRuns(listed.analysis_runs); + setSelected(created); + } catch (err) { + setError(err instanceof BackendError ? err.message : String(err)); + } finally { + setRequesting(false); + } + } + async function handleOpen(runId: string) { setError(null); try { @@ -1502,11 +1525,20 @@ function AnalysisRunsPanel({

Analysis runs

+
{error &&

{error}

} {runs.length === 0 ? (

- No analysis runs visible to this account yet -- try `make seed`. + No analysis runs visible to this account yet. Request a lineage + reconstruction, or ask an administrator to run make seed.

) : (
    diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 740529cba..e35bcf3ed 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -532,3 +532,21 @@ export function fetchAnalysisRuns(accessToken: string): Promise<{ analysis_runs: export function fetchAnalysisRun(accessToken: string, analysisRunId: string): Promise { return backendFetch(`/api/analysis-runs/${analysisRunId}`, accessToken); } + +export interface CreateAnalysisRunRequest { + run_kind_code?: string; + scope_kind_code?: string; + corporate_entity_id?: string; + knowledge_cutoff?: string; + idempotency_key: string; +} + +export function createAnalysisRun( + accessToken: string, + request: CreateAnalysisRunRequest, +): Promise { + return backendFetch("/api/analysis-runs", accessToken, { + method: "POST", + body: JSON.stringify(request), + }); +} diff --git a/lineageweave/__init__.py b/lineageweave/__init__.py index 65fa5d182..5e05ef4ff 100644 --- a/lineageweave/__init__.py +++ b/lineageweave/__init__.py @@ -55,4 +55,4 @@ "sentence_excerpts", ] -__version__ = "0.84.1" +__version__ = "0.85.0" diff --git a/pyproject.toml b/pyproject.toml index b27655c2b..8750ae3e3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "lineageweave" -version = "0.84.1" +version = "0.85.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/tests/test_analysis_run_create.py b/tests/test_analysis_run_create.py new file mode 100644 index 000000000..4e24a4228 --- /dev/null +++ b/tests/test_analysis_run_create.py @@ -0,0 +1,134 @@ +"""Authorized analysis-run create hashes the cutoff bag, never a score.""" + +from datetime import datetime, timezone + +from backend.app.analysis_run_ingestion import ( + AnalysisRunCreateError, + _resolve_corporate_entity_id, + plan_analysis_run_capture, +) +import pytest + + +_CUTOFF = datetime(2026, 1, 12, 12, 0, tzinfo=timezone.utc) +_EARLIER = datetime(2026, 1, 10, 9, 0, tzinfo=timezone.utc) + + +def test_capture_digest_is_stable_for_the_same_authorized_bag() -> None: + first = plan_analysis_run_capture( + run_kind_code="analysis_run_lineage", + scope_kind_code="analysis_scope_corporate_entity", + corporate_entity_id="corp-1", + knowledge_cutoff=_CUTOFF, + idempotency_key="client-key-1", + post_ids=["post-b", "post-a"], + thread_keys=["thread-a", "thread-a"], + latest_post_created_at=_EARLIER, + ) + second = plan_analysis_run_capture( + run_kind_code="analysis_run_lineage", + scope_kind_code="analysis_scope_corporate_entity", + corporate_entity_id="corp-1", + knowledge_cutoff=_CUTOFF, + idempotency_key="client-key-1", + post_ids=["post-a", "post-b"], + thread_keys=["thread-a", "thread-a"], + latest_post_created_at=_EARLIER, + ) + assert first.snapshot_sha256 == second.snapshot_sha256 + assert first.configuration_sha256 == second.configuration_sha256 + assert first.document_count == 2 + assert first.thread_count == 1 + assert first.maximum_available_time == _EARLIER + assert "theta" not in first.snapshot_sha256 + assert first.configuration_schema_version == "lineage-run-v1" + + +def test_later_cutoff_or_other_kind_does_not_reuse_the_wrong_digest() -> None: + lineage = plan_analysis_run_capture( + run_kind_code="analysis_run_lineage", + scope_kind_code="analysis_scope_corporate_entity", + corporate_entity_id="corp-1", + knowledge_cutoff=_CUTOFF, + idempotency_key="client-key-1", + post_ids=["post-a"], + thread_keys=["thread-a"], + latest_post_created_at=_EARLIER, + ) + later = plan_analysis_run_capture( + run_kind_code="analysis_run_lineage", + scope_kind_code="analysis_scope_corporate_entity", + corporate_entity_id="corp-1", + knowledge_cutoff=datetime(2026, 1, 13, 12, 0, tzinfo=timezone.utc), + idempotency_key="client-key-1", + post_ids=["post-a"], + thread_keys=["thread-a"], + latest_post_created_at=_EARLIER, + ) + tepp = plan_analysis_run_capture( + run_kind_code="analysis_run_tepp", + scope_kind_code="analysis_scope_corporate_entity", + corporate_entity_id="corp-1", + knowledge_cutoff=_CUTOFF, + idempotency_key="client-key-1", + post_ids=["post-a"], + thread_keys=["thread-a"], + latest_post_created_at=_EARLIER, + ) + assert lineage.snapshot_sha256 != later.snapshot_sha256 + assert lineage.snapshot_sha256 == tepp.snapshot_sha256 + assert lineage.configuration_sha256 != tepp.configuration_sha256 + assert tepp.configuration_schema_version == "tepp-run-v1" + + +def test_omitted_cutoff_keeps_the_same_client_key_stable() -> None: + first = plan_analysis_run_capture( + run_kind_code="analysis_run_lineage", + scope_kind_code="analysis_scope_corporate_entity", + corporate_entity_id="corp-1", + knowledge_cutoff=_CUTOFF, + idempotency_key="client-key-1", + post_ids=["post-a"], + thread_keys=["thread-a"], + latest_post_created_at=_EARLIER, + cutoff_explicit=False, + ) + later_clock = plan_analysis_run_capture( + run_kind_code="analysis_run_lineage", + scope_kind_code="analysis_scope_corporate_entity", + corporate_entity_id="corp-1", + knowledge_cutoff=datetime(2026, 1, 13, 12, 0, tzinfo=timezone.utc), + idempotency_key="client-key-1", + post_ids=["post-a"], + thread_keys=["thread-a"], + latest_post_created_at=_EARLIER, + cutoff_explicit=False, + ) + assert first.configuration_sha256 == later_clock.configuration_sha256 + assert first.snapshot_sha256 == later_clock.snapshot_sha256 + + +def test_empty_corpus_uses_the_cutoff_as_latest_available_time() -> None: + capture = plan_analysis_run_capture( + run_kind_code="analysis_run_lineage", + scope_kind_code="analysis_scope_corporate_entity", + corporate_entity_id="corp-1", + knowledge_cutoff=_CUTOFF, + idempotency_key="client-key-1", + post_ids=[], + thread_keys=[], + latest_post_created_at=None, + ) + assert capture.document_count == 0 + assert capture.thread_count == 0 + assert capture.maximum_available_time == _CUTOFF + + +def test_create_rejects_an_unaffiliated_or_ambiguous_corporate_entity() -> None: + with pytest.raises(AnalysisRunCreateError) as hidden: + _resolve_corporate_entity_id("corp-other", ["corp-1"]) + assert hidden.value.status_code == 404 + with pytest.raises(AnalysisRunCreateError) as ambiguous: + _resolve_corporate_entity_id(None, ["corp-1", "corp-2"]) + assert ambiguous.value.status_code == 422 + assert _resolve_corporate_entity_id(None, ["corp-1"]) == "corp-1" diff --git a/tests/test_seed_tepp_run.py b/tests/test_seed_tepp_run.py index c865c475c..b25908cbe 100644 --- a/tests/test_seed_tepp_run.py +++ b/tests/test_seed_tepp_run.py @@ -3,6 +3,7 @@ from lineageweave.tepp_client import AnalysisRunRequest, TeppClient, TeppNotAvailable from scripts.seed_demo_data import ( _ensure_demo_source_counts, + _seed_demo_tepp_run, demo_source_snapshot_sha256, tepp_seed_outcome, tepp_seed_request, @@ -83,3 +84,49 @@ def test_ensure_demo_source_counts_inserts_when_the_snapshot_is_empty() -> None: cursor = _CountCursor(existing_counts=False) _ensure_demo_source_counts(cursor, "snapshot-1") assert any(sql.lstrip().startswith("insert into analysis_source_count") for sql in cursor.statements) + + +class _TeppSeedCursor: + """Drive `_seed_demo_tepp_run` without a live database.""" + + def __init__(self) -> None: + self.statements: list[str] = [] + self.params: list[object] = [] + + def execute(self, sql: str, params=None) -> None: + self.statements.append(" ".join(sql.split())) + self.params.append(params) + + def fetchone(self): + last = self.statements[-1] + if last.lstrip().startswith("select") and "from analysis_source_snapshot" in last: + return None + if "insert into analysis_source_snapshot" in last: + return ("snapshot-demo",) + if last.lstrip().startswith("select") and "from analysis_source_count" in last: + return None + if last.lstrip().startswith("select") and "from analysis_run" in last: + return None + if "insert into analysis_run" in last: + return ("run-demo-tepp",) + return None + + +def test_seed_demo_tepp_run_inserts_failed_tepp_not_available() -> None: + cursor = _TeppSeedCursor() + _seed_demo_tepp_run(cursor, "account-1", "corp-1") + run_inserts = [sql for sql in cursor.statements if "insert into analysis_run" in sql] + assert run_inserts, "seed must insert the TEPP analysis_run row" + assert any("analysis_run_tepp" in sql for sql in run_inserts) + status_params = [ + params + for sql, params in zip(cursor.statements, cursor.params, strict=True) + if "insert into analysis_run_status_event" in sql + ] + assert any( + params is not None and "analysis_status_failed" in params and "tepp_not_available" in params + for params in status_params + ) + assert not any( + params is not None and "analysis_status_succeeded" in params for params in status_params + ) diff --git a/uv.lock b/uv.lock index 156b0813d..20f9a8793 100644 --- a/uv.lock +++ b/uv.lock @@ -454,7 +454,7 @@ wheels = [ [[package]] name = "lineageweave" -version = "0.84.1" +version = "0.85.0" source = { virtual = "." } dependencies = [ { name = "certifi" },