diff --git a/CHANGELOG.md b/CHANGELOG.md index 1daea713f..1f59a813f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,9 +8,14 @@ All notable changes to this project are documented here. Format follows ### Added -- Added a Buyer Project history destination and post-detail entry point for - bounded, authorized exact-project chronology. The release remains pending - protected-main review and Checks (ADR 0111). +- Proposed source-owned project lifecycle ingestion with explicit versioned + event mappings, evidence-bound writes, idempotent replacement, withdrawal, + and aggregate audit digests (ADR 0112). This remains pending protected + review and Checks and is not a release claim. +- Proposed Buyer Project history destination and post-detail entry point for + bounded, authorized exact-project chronology. The feature remains pending + protected-main review and Checks; it does not claim release until those + gates pass (ADR 0111). ## [2.17.0] - 2026-08-19 diff --git a/backend/app/project_lifecycle_ingestion.py b/backend/app/project_lifecycle_ingestion.py new file mode 100644 index 000000000..16fb24989 --- /dev/null +++ b/backend/app/project_lifecycle_ingestion.py @@ -0,0 +1,475 @@ +"""Atomic writer for source-owned project lifecycle evidence. + +The writer is an administrative/import boundary. It accepts explicit source +codes and evidence posts, never classifies a title or body, and replaces only +the source-system record that owns the projection. +""" + +from __future__ import annotations + +import json +from typing import Any +from unicodedata import normalize + +import asyncpg + +from lineageweave.project_lifecycle import ( + PROJECT_EVENT_TYPE_CODES, + ProjectLifecycleEventInput, + ProjectLifecycleValidationError, + project_lifecycle_digest, + require_project_lifecycle_write_permission, + validate_project_lifecycle_event, +) + + +def _text(value: str) -> str: + """Normalize a validated identifier before binding it to SQL.""" + + return normalize("NFKC", value).strip() + + +async def _assert_evidence_post( + conn: asyncpg.Connection, + evidence_post_id: str, + source_system_code: str, + source_record_key: str, +) -> None: + """Require evidence to belong to the same source record as the import.""" + + row = await conn.fetchrow( + """ + select post_id + from source_post + where post_id = $1::uuid + and source_system_code = $2 + and source_record_key = $3 + """, + evidence_post_id, + source_system_code, + source_record_key, + ) + if row is None: + raise ProjectLifecycleValidationError( + "evidence post must belong to the imported source system and record" + ) + + +async def _mapping_row( + conn: asyncpg.Connection, event: ProjectLifecycleEventInput +) -> asyncpg.Record: + """Resolve the active, versioned external code mapping or fail closed.""" + + row = await conn.fetchrow( + """ + select mapping.project_event_mapping_id as mapping_id, + mapping.project_source_system_id as source_system_id, + mapping.project_event_type_code + from project_event_mapping mapping + join project_source_system source_system + on source_system.project_source_system_id = mapping.project_source_system_id + where source_system.source_system_code = $1 + and mapping.mapping_version = $2 + and mapping.source_event_code = $3 + and mapping.is_active + """, + _text(event.source_system_code), + _text(event.mapping_version), + _text(event.source_event_code), + ) + if row is None: + raise ProjectLifecycleValidationError( + "no active project event mapping exists for the source code and version" + ) + if row["project_event_type_code"] not in PROJECT_EVENT_TYPE_CODES: + raise ProjectLifecycleValidationError("mapping resolves to an unregistered project event type") + return row + + +async def register_project_event_mapping( + conn: asyncpg.Connection, + *, + source_system_code: str, + source_system_name: str, + mapping_version: str, + source_event_code: str, + project_event_type_code: str, + administrator_key: str, + permission_codes: set[str], +) -> dict[str, str]: + """Register or activate one explicit source-system event mapping. + + This is the administrative mapping adapter. It is intentionally separate + from lifecycle ingestion so unknown source codes cannot become events by + accident. + """ + + require_project_lifecycle_write_permission(permission_codes) + if not administrator_key.strip(): + raise PermissionError("administrator_key is required") + values = { + "source_system_code": _text(source_system_code), + "source_system_name": _text(source_system_name), + "mapping_version": _text(mapping_version), + "source_event_code": _text(source_event_code), + } + if any(not value for value in values.values()): + raise ProjectLifecycleValidationError("mapping fields must be non-empty") + if project_event_type_code not in PROJECT_EVENT_TYPE_CODES: + raise ProjectLifecycleValidationError( + f"unsupported project_event_type_code: {project_event_type_code!r}" + ) + async with conn.transaction(): + system_id = await conn.fetchval( + """ + insert into project_source_system (source_system_code, source_system_name) + values ($1, $2) + on conflict (source_system_code) do update + set source_system_name = excluded.source_system_name + returning project_source_system_id + """, + values["source_system_code"], + values["source_system_name"], + ) + mapping_id = await conn.fetchval( + """ + insert into project_event_mapping + (project_source_system_id, mapping_version, source_event_code, + project_event_type_code, is_active) + values ($1, $2, $3, $4, true) + on conflict (project_source_system_id, mapping_version, source_event_code) do update + set project_event_type_code = excluded.project_event_type_code, + is_active = true + returning project_event_mapping_id + """, + system_id, + values["mapping_version"], + values["source_event_code"], + project_event_type_code, + ) + return {"project_event_mapping_id": str(mapping_id), "status_code": "registered"} + + +async def _upsert_project_identity( + conn: asyncpg.Connection, project_key: str, project_name: str +) -> str: + """Create or update the normalized project display identity.""" + + return str( + await conn.fetchval( + """ + insert into project_identity (project_key, project_name) + values ($1, $2) + on conflict (project_key) do update + set project_name = excluded.project_name + returning project_identity_id + """, + project_key, + _text(project_name), + ) + ) + + +async def ingest_project_lifecycle_event( + conn: asyncpg.Connection, + event: ProjectLifecycleEventInput, + *, + administrator_key: str, + permission_codes: set[str], +) -> dict[str, Any]: + """Atomically insert or replace one source-owned lifecycle projection. + + Repeating the same source identity is idempotent. Replacing it deletes and + recreates only its owned relations and responsibility rows; independent + source records remain untouched. The returned data is aggregate identity + and status, not source content. + """ + + require_project_lifecycle_write_permission(permission_codes) + if not administrator_key.strip(): + raise PermissionError("administrator_key is required") + project_key = validate_project_lifecycle_event(event) + digest = project_lifecycle_digest(event) + + async with conn.transaction(): + lock_key = json.dumps( + [_text(event.source_system_code), _text(event.source_record_key)], + ensure_ascii=False, + separators=(",", ":"), + ) + await conn.execute("select pg_advisory_xact_lock(hashtextextended($1, 0))", lock_key) + mapping = await _mapping_row(conn, event) + await _assert_evidence_post( + conn, + event.evidence_post_id, + _text(event.source_system_code), + _text(event.source_record_key), + ) + for relation in event.relations: + await _assert_evidence_post( + conn, + relation.evidence_post_id, + _text(event.source_system_code), + _text(event.source_record_key), + ) + for responsibility in event.responsibilities: + await _assert_evidence_post( + conn, + responsibility.evidence_post_id, + _text(event.source_system_code), + _text(event.source_record_key), + ) + + previous = await conn.fetchrow( + """ + select project_source_record_id, record_digest + from project_source_record + where project_source_system_id = $1 + and source_record_key = $2 + for update + """, + mapping["source_system_id"], + _text(event.source_record_key), + ) + identity_id = await _upsert_project_identity(conn, project_key, event.project_name) + source_record_id = await conn.fetchval( + """ + insert into project_source_record + (project_source_system_id, source_record_key, project_identity_id, + project_event_mapping_id, lifecycle_state_code, record_digest, + imported_by) + values ($1, $2, $3, $4, 'project_record_active', $5, $6) + on conflict (project_source_system_id, source_record_key) do update set + project_identity_id = excluded.project_identity_id, + project_event_mapping_id = excluded.project_event_mapping_id, + lifecycle_state_code = excluded.lifecycle_state_code, + record_digest = excluded.record_digest, + imported_by = excluded.imported_by, + imported_at = now() + returning project_source_record_id + """, + mapping["source_system_id"], + _text(event.source_record_key), + identity_id, + mapping["mapping_id"], + digest, + administrator_key.strip(), + ) + project_event_id = await conn.fetchval( + """ + insert into project_lifecycle_event + (project_source_record_id, project_event_type_code, + event_started_at, event_ended_at, event_digest) + values ($1, $2, $3, $4, $5) + on conflict (project_source_record_id) do update set + project_event_type_code = excluded.project_event_type_code, + event_started_at = excluded.event_started_at, + event_ended_at = excluded.event_ended_at, + event_digest = excluded.event_digest + returning project_lifecycle_event_id + """, + source_record_id, + mapping["project_event_type_code"], + event.event_started_at, + event.event_ended_at, + digest, + ) + await conn.execute( + "delete from project_event_evidence where project_lifecycle_event_id = $1", + project_event_id, + ) + await conn.execute( + """ + insert into project_event_evidence + (project_lifecycle_event_id, evidence_post_id, evidence_role_code) + values ($1, $2::uuid, 'project_event_primary_evidence') + """, + project_event_id, + event.evidence_post_id, + ) + await conn.execute( + "delete from project_event_relation where owner_source_record_id = $1", + source_record_id, + ) + await conn.execute( + "delete from project_event_responsibility where owner_source_record_id = $1", + source_record_id, + ) + for relation in event.relations: + target = await conn.fetchrow( + """ + select target_event.project_lifecycle_event_id, target_record.project_identity_id + from project_source_record target_record + join project_lifecycle_event target_event + on target_event.project_source_record_id = target_record.project_source_record_id + join project_source_system target_system + on target_system.project_source_system_id = target_record.project_source_system_id + where target_system.source_system_code = $1 + and target_record.source_record_key = $2 + and target_record.lifecycle_state_code = 'project_record_active' + """, + _text(relation.target_source_system_code), + _text(relation.target_source_record_key), + ) + if target is None or str(target["project_identity_id"]) != str(identity_id): + raise ProjectLifecycleValidationError( + "relation target must be an active event in the same project" + ) + await conn.execute( + """ + insert into project_event_relation + (owner_source_record_id, source_lifecycle_event_id, + target_lifecycle_event_id, relation_type_code, evidence_post_id) + values ($1, $2, $3, $4, $5::uuid) + """, + source_record_id, + project_event_id, + target["project_lifecycle_event_id"], + relation.relation_type_code, + relation.evidence_post_id, + ) + for responsibility in event.responsibilities: + actor_id = await conn.fetchval( + """ + insert into project_actor + (project_identity_id, actor_type_code, actor_key, actor_name) + values ($1, $2, $3, $4) + on conflict (project_identity_id, actor_type_code, actor_key) do update + set actor_name = excluded.actor_name + returning project_actor_id + """, + identity_id, + responsibility.actor_type_code, + _text(responsibility.actor_key), + _text(responsibility.actor_name), + ) + await conn.execute( + """ + insert into project_event_responsibility + (owner_source_record_id, project_lifecycle_event_id, + project_actor_id, responsibility_text, evidence_post_id) + values ($1, $2, $3, $4, $5::uuid) + """, + source_record_id, + project_event_id, + actor_id, + _text(responsibility.responsibility_text), + responsibility.evidence_post_id, + ) + await conn.execute( + """ + insert into project_lifecycle_audit + (project_source_record_id, action_code, actor_key, + mapping_version, before_digest, after_digest) + values ($1, 'project_lifecycle_upserted', $2, $3, $4, $5) + """, + source_record_id, + _text(administrator_key), + _text(event.mapping_version), + previous["record_digest"] if previous else None, + digest, + ) + return { + "status_code": "replaced" if previous else "inserted", + "project_source_record_id": str(source_record_id), + "project_lifecycle_event_id": str(project_event_id), + "record_digest": digest, + } + + +async def withdraw_project_lifecycle_record( + conn: asyncpg.Connection, + *, + source_system_code: str, + source_record_key: str, + administrator_key: str, + permission_codes: set[str], +) -> dict[str, Any]: + """Withdraw one source channel while preserving independent source rows.""" + + require_project_lifecycle_write_permission(permission_codes) + if not administrator_key.strip(): + raise PermissionError("administrator_key is required") + if not source_system_code.strip() or not source_record_key.strip(): + raise ProjectLifecycleValidationError("source identity must be non-empty") + + async with conn.transaction(): + await conn.execute( + "select pg_advisory_xact_lock(hashtextextended($1, 0))", + json.dumps( + [_text(source_system_code), _text(source_record_key)], + ensure_ascii=False, + separators=(",", ":"), + ), + ) + row = await conn.fetchrow( + """ + select record.project_source_record_id, record.record_digest, + mapping.mapping_version, lifecycle.project_lifecycle_event_id + from project_source_record record + join project_source_system source_system + on source_system.project_source_system_id = record.project_source_system_id + join project_event_mapping mapping + on mapping.project_event_mapping_id = record.project_event_mapping_id + left join project_lifecycle_event lifecycle + on lifecycle.project_source_record_id = record.project_source_record_id + where source_system.source_system_code = $1 + and record.source_record_key = $2 + for update of record + """, + _text(source_system_code), + _text(source_record_key), + ) + if row is None: + return {"status_code": "not_found"} + if row["project_lifecycle_event_id"] is not None: + await conn.execute( + """ + delete from project_event_relation + where source_lifecycle_event_id = $1 + or target_lifecycle_event_id = $1 + """, + row["project_lifecycle_event_id"], + ) + await conn.execute( + "delete from project_event_responsibility where owner_source_record_id = $1", + row["project_source_record_id"], + ) + await conn.execute( + "delete from project_lifecycle_event where project_lifecycle_event_id = $1", + row["project_lifecycle_event_id"], + ) + await conn.execute( + """ + update project_source_record + set lifecycle_state_code = 'project_record_withdrawn', + imported_by = $2, + imported_at = now() + where project_source_record_id = $1 + """, + row["project_source_record_id"], + _text(administrator_key), + ) + await conn.execute( + """ + insert into project_lifecycle_audit + (project_source_record_id, action_code, actor_key, + mapping_version, before_digest, after_digest) + values ($1, 'project_lifecycle_withdrawn', $2, $3, $4, null) + """, + row["project_source_record_id"], + _text(administrator_key), + row["mapping_version"], + row["record_digest"], + ) + return { + "status_code": "withdrawn", + "project_source_record_id": str(row["project_source_record_id"]), + } + + +__all__ = [ + "ingest_project_lifecycle_event", + "register_project_event_mapping", + "withdraw_project_lifecycle_record", +] diff --git a/docker/postgres-init/migrate.sh b/docker/postgres-init/migrate.sh index 122523d18..6fad01496 100644 --- a/docker/postgres-init/migrate.sh +++ b/docker/postgres-init/migrate.sh @@ -18,7 +18,7 @@ for migration in /opt/lineageweave/migrations/*.sql; do migration_name=${migration##*/} case "$migration_name" in 0012_*|0013_*|0014_*|0015_*|0016_*|0017_*|0018_*|0019_*|0020_*|0021_*|0022_*|0023_*|0024_*|0025_*|0026_*|0027_*|0028_*|0029_*|0030_*|0031_*|0032_*|0033_*|0034_*|0035_*|0036_*|0037_*|0038_*|0039_*|0040_*|0041_*|0042_*|0043_*|0044_*|0045_*|0046_*|0047_*|0048_*|0049_*|0050_*) ;; - 0051_*|0052_*) ;; + 0051_*|0052_*|0053_*|0054_*) ;; 0060_*|0100_*) ;; *) continue ;; esac diff --git a/docs/adr/0112-authoritative-project-lifecycle-ingestion.md b/docs/adr/0112-authoritative-project-lifecycle-ingestion.md new file mode 100644 index 000000000..ee699fc69 --- /dev/null +++ b/docs/adr/0112-authoritative-project-lifecycle-ingestion.md @@ -0,0 +1,64 @@ +# ADR 0112 — Authoritative project-lifecycle ingestion boundary + +- Status: Proposed on Issue #284 and stacked PR #285; not protected-main behavior +- Date: 2026-08-20 +- Figma file: `SBpgot7uTvMxEaxUwvoc0S` (reuses the project-history evidence boundary in ADR 0111; no new UI surface) + +## Context + +Project History needs a source-owned event clock and lifecycle records for +orders, specification changes, delivery, VOC, and rebids. Title/body +classification and an LLM can suggest a candidate, but neither is an +authoritative write source. Re-imports must be safe when a source changes its +record, and withdrawal must remove only the projection owned by that source +record while preserving independent evidence. + +## Decision + +1. Require an explicit project key/name, source system and source record key, + versioned source event code, offset-aware start/end, and a source-post + evidence row before writing. +2. Resolve external event codes through `project_event_mapping`. Unknown or + inactive codes fail closed; the writer does not inspect titles, bodies, or + call a model. +3. Store the project identity, source system, mapping, source record, event, + evidence, actor, relation, responsibility, and audit facts in separate + third-normal-form tables. A source-system/record advisory transaction lock + makes concurrent imports converge on one projection. +4. Re-importing the same source identity replaces its event, relations, and + responsibilities atomically and records before/after digests. Withdrawal + deletes the owned event projection and marks the source record withdrawn; + independent source records remain available. +5. Keep the writer behind the dedicated `project_lifecycle_write` permission. + This PR adds the adapter and migration only; an authenticated application + route must bind that permission to the existing administrator policy before + exposing writes to a deployment. + +## Consequences + +- Buyers can distinguish observed source lifecycle evidence from inferred + project mentions and follow a deterministic event clock. +- A bad mapping, cross-project relation, mismatched evidence post, or missing + permission fails without a partial write. +- Audit rows retain aggregate digests and actor keys, not copied source bodies. +- The current read model can adopt these records after this stacked change is + reviewed and merged through the protected order; this ADR does not claim + protected-main or production availability. + +## Verification + +- `tests/test_project_lifecycle_ingestion.py` exercises real PostgreSQL + mapping, permission, evidence, idempotent replacement, concurrent import, + cross-project rejection, and withdrawal behavior with synthetic identities. +- `tests/test_migration_replay.py` verifies that Compose replays migration + 0054 on existing volumes. +- `python -m compileall`, `git diff --check`, and the repository's backend and + frontend gates remain required before the stacked PR can be reviewed. + +## References + +World Wide Web Consortium. (2013). *PROV-O: The PROV ontology*. +https://www.w3.org/TR/prov-o/ + +World Wide Web Consortium. (2013). *PROV-DM: The PROV data model*. +https://www.w3.org/TR/prov-dm/ diff --git a/docs/adr/README.md b/docs/adr/README.md index 762f1c051..f3c901026 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -15,5 +15,7 @@ decision from them. | [`PROV_O_IMPLEMENTATION_MATRIX.md`](../PROV_O_IMPLEMENTATION_MATRIX.md) | [0065](0065-prov-o-provenance-boundary.md) | | [`image-content-schema.md`](../image-content-schema.md) | [0066](0066-position-preserving-image-content.md) | +The current project-history source-of-truth boundary is defined by [0112](0112-authoritative-project-lifecycle-ingestion.md). + Runtime evidence under `docs/doctoring/` is not converted into an ADR: it records observed results for already-decided behavior. diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index dd3cf776c..939a0c067 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -197,8 +197,9 @@ sequenceDiagram ## Current aggregate data and runtime evidence -Observed from the running local Compose stack without selecting a post title, -body, source code, person, organization, or identifier: +Observed from the running local Compose stack; the browser path used a +synthetic login and opened one authorized source post, but no title, body, +source code, person, organization, or identifier was recorded: | Evidence | Observed result | |---|---| @@ -212,13 +213,15 @@ body, source code, person, organization, or identifier: | Summary refresh | One authorized target request returned HTTP 200 with contract v5, four key events, one role, three actions, and one project | | Authentication | Real synthetic-user OIDC login, live JWKS fetch, and RS256 verification passed | | Authorization | Unauthenticated `/api/analysis-runs` and `/api/posts` returned 401; four focused live-Keycloak/PostgreSQL API tests covering authenticated account, list ABAC, direct deny, and missing token passed | +| Browser navigation | 2026-08-20 Playwright browser flow completed synthetic OIDC login, loaded the authorized board, opened one authorized post, and closed its detail view | | Focused contracts | Post-summary and transaction-contract tests: 31 passed, 1 skipped; the skip is not runtime proof for the skipped capability | These observations prove data presence and the listed boundaries only. They do -not prove a browser-clicked buyer journey, current TEPP transport success, -post-summary-action population across the corpus, or equivalence between every -running container image and the PR head. The target refresh is bounded runtime -evidence for one authorized post, not a corpus-wide acceptance claim. +prove one browser-clicked buyer journey on the local stack, but they do not +prove current TEPP transport success, post-summary-action population across the +corpus, or equivalence between every running container image and the PR head. +The target refresh and browser flow are bounded runtime evidence, not a +corpus-wide acceptance claim. ## Active PR audit @@ -233,6 +236,7 @@ Queued checks and review gates mean none of these rows is protected-main truth. | #287 | exact Event Lineage channel evidence | `#276` → feature | Ready / UNSTABLE | | #286 | exact byte-bounded MCP browser admission | `#270` → fix | Ready / UNSTABLE | | #285 | project lifecycle timeline | `#264 stack` → `v2.18.4` | Ready / UNSTABLE | +| #284 | authoritative project lifecycle ingestion | `#285 stack` → proposed 0054 writer boundary | Implementation on stacked PR; not protected-main | | #282 | TEPP project history in read/Ask | `#264 stack` → `v2.18.0` | Ready / UNSTABLE | | #276 | public verification of Global Ask claims | `#266` → `v2.20.0` | Ready / UNSTABLE | | #275 | evidence-bound Event Intelligence | `#270` → `v2.18.3` | Ready / UNSTABLE | @@ -265,7 +269,7 @@ the exact-head disposition. | Priority | Gap | Evidence | Closure criterion | |---|---|---|---| | P0 | No protected-main integrated buyer journey for the active feature stack | Main is 2.12.5; 18 open PRs span dependent and parallel bases | Establish one reviewed integration order, update each exact head, pass required checks, merge without bypass, then run login-to-source browser acceptance on main | -| P0 | Current runtime proof is incomplete | The current aggregate/OIDC/ABAC checks cover data presence and selected boundaries; 2026-08-18/19 notes cover other slices, but no evidence set proves the entire PR head or main journey | Complete the real-stack matrix on an exact revision: browser login/navigation, Ask, reports, Vision, TEPP availability, action population, and cleanup | +| P0 | Current runtime proof is incomplete | Aggregate/OIDC/ABAC checks and a 2026-08-20 local browser login/navigation flow cover selected boundaries, but no evidence set proves the entire PR head or main journey | Repeat the real-stack matrix on an exact integrated revision: browser login/navigation, Ask, reports, Vision, TEPP availability, action population, and cleanup | | P0 | PR #190's duplicate ADR identity was corrected but is not protected-main truth | Active PR head `ac1b4e17` now uses ADR 0038 and aligns the entity-resolution claims with implementation; independent review and Checks remain pending | Re-audit exact head, obtain independent approval, pass required Checks, and merge normally; never merge a duplicate ADR identity | | P0 | PR #258 is not review/CI complete at its exact current head | #258 is mergeable but BLOCKED: 14 of 22 checks queued, no approval, and four unresolved scanner threads on two SQL modules | Classify each finding against the literal SQL and bound arguments; fix a real flow or add a narrow documented suppression for a false positive, resolve threads, obtain independent approval, and re-check the exact head | | P1 | Requirements were implicit across ADRs and architecture phases | No prior PRD/TRD/requirement traceability baseline existed | Keep FR/NFR IDs in this document linked from ADR index; require new product PRs to name affected IDs and runtime evidence | diff --git a/lineageweave/project_lifecycle.py b/lineageweave/project_lifecycle.py new file mode 100644 index 000000000..e99c341b3 --- /dev/null +++ b/lineageweave/project_lifecycle.py @@ -0,0 +1,195 @@ +"""Validated input contracts for authoritative project-lifecycle imports. + +This module deliberately contains no title/body classifier. A source adapter +must provide an explicit external event code; PostgreSQL resolves that code +through the versioned mapping registry before an authoritative row is written. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +import hashlib +import json +from typing import Any +from unicodedata import normalize +from uuid import UUID + +from .project_history import normalize_project_key + +PROJECT_EVENT_TYPE_CODES = frozenset( + { + "project_event_order", + "project_event_spec_change", + "project_event_delivery", + "project_event_voc", + "project_event_rebid", + } +) +PROJECT_ACTOR_TYPE_CODES = frozenset({"prov_person", "prov_organization", "prov_team"}) +PROJECT_RELATION_TYPE_CODES = frozenset({"project_relation_precedes", "project_relation_related"}) + + +class ProjectLifecycleValidationError(ValueError): + """Raised when an import would create unsupported lifecycle evidence.""" + + +@dataclass(frozen=True) +class ProjectLifecycleRelationInput: + """One explicit relation to an already imported event in the same project.""" + + target_source_system_code: str + target_source_record_key: str + relation_type_code: str + evidence_post_id: str + + +@dataclass(frozen=True) +class ProjectLifecycleResponsibilityInput: + """One explicit responsibility assignment with its own evidence post.""" + + actor_type_code: str + actor_key: str + actor_name: str + responsibility_text: str + evidence_post_id: str + + +@dataclass(frozen=True) +class ProjectLifecycleEventInput: + """One source-owned lifecycle record accepted by the writer boundary.""" + + project_key: str + project_name: str + source_system_code: str + source_record_key: str + source_event_code: str + mapping_version: str + event_started_at: datetime + event_ended_at: datetime | None + evidence_post_id: str + relations: tuple[ProjectLifecycleRelationInput, ...] = field(default_factory=tuple) + responsibilities: tuple[ProjectLifecycleResponsibilityInput, ...] = field( + default_factory=tuple + ) + + +def _required_text(value: object, field_name: str) -> str: + """Return normalized non-empty text for a trusted application boundary.""" + + if not isinstance(value, str) or not value.strip(): + raise ProjectLifecycleValidationError(f"{field_name} must be non-empty text") + return normalize("NFKC", value).strip() + + +def _offset_aware(value: datetime, field_name: str) -> None: + """Require an offset-aware timestamp so event order is deterministic.""" + + if not isinstance(value, datetime): + raise ProjectLifecycleValidationError(f"{field_name} must be a datetime") + if value.tzinfo is None or value.utcoffset() is None: + raise ProjectLifecycleValidationError(f"{field_name} must include a timezone offset") + + +def _uuid_text(value: object, field_name: str) -> str: + """Validate an evidence identifier before the database UUID cast.""" + + text = _required_text(value, field_name) + try: + return str(UUID(text)) + except ValueError as exc: + raise ProjectLifecycleValidationError(f"{field_name} must be a UUID") from exc + + +def validate_project_lifecycle_event(event: ProjectLifecycleEventInput) -> str: + """Validate one event and return its canonical project identity key. + + The external source code is intentionally not classified here. The + mapping registry is the only authority for translating it to a registered + event type. This keeps an LLM or a title/body heuristic outside the write + path. + """ + + try: + project_key = normalize_project_key(_required_text(event.project_key, "project_key")) + except ValueError as exc: + raise ProjectLifecycleValidationError(str(exc)) from exc + _required_text(event.project_name, "project_name") + source_system_code = _required_text(event.source_system_code, "source_system_code") + source_record_key = _required_text(event.source_record_key, "source_record_key") + _required_text(event.source_event_code, "source_event_code") + _required_text(event.mapping_version, "mapping_version") + _uuid_text(event.evidence_post_id, "evidence_post_id") + _offset_aware(event.event_started_at, "event_started_at") + if event.event_ended_at is not None: + _offset_aware(event.event_ended_at, "event_ended_at") + if event.event_ended_at < event.event_started_at: + raise ProjectLifecycleValidationError("event_ended_at cannot precede event_started_at") + + for relation in event.relations: + target_source_system_code = _required_text( + relation.target_source_system_code, "relation.target_source_system_code" + ) + target_source_record_key = _required_text( + relation.target_source_record_key, "relation.target_source_record_key" + ) + if relation.relation_type_code not in PROJECT_RELATION_TYPE_CODES: + raise ProjectLifecycleValidationError( + f"unsupported relation_type_code: {relation.relation_type_code!r}" + ) + _uuid_text(relation.evidence_post_id, "relation.evidence_post_id") + if ( + target_source_system_code == source_system_code + and target_source_record_key == source_record_key + ): + raise ProjectLifecycleValidationError("an event cannot relate to itself") + + for responsibility in event.responsibilities: + if responsibility.actor_type_code not in PROJECT_ACTOR_TYPE_CODES: + raise ProjectLifecycleValidationError( + f"unsupported actor_type_code: {responsibility.actor_type_code!r}" + ) + _required_text(responsibility.actor_key, "responsibility.actor_key") + _required_text(responsibility.actor_name, "responsibility.actor_name") + _required_text(responsibility.responsibility_text, "responsibility.responsibility_text") + _uuid_text(responsibility.evidence_post_id, "responsibility.evidence_post_id") + return project_key + + +def project_lifecycle_digest(event: ProjectLifecycleEventInput) -> str: + """Return a stable, non-PII digest for audit and idempotent replacement.""" + + validate_project_lifecycle_event(event) + + def serializable(value: Any) -> Any: + if isinstance(value, datetime): + return value.isoformat() + if hasattr(value, "__dataclass_fields__"): + return {name: serializable(getattr(value, name)) for name in value.__dataclass_fields__} + if isinstance(value, tuple): + return [serializable(item) for item in value] + return value + + encoded = json.dumps(serializable(event), ensure_ascii=False, sort_keys=True).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def require_project_lifecycle_write_permission(permission_codes: set[str]) -> None: + """Require the dedicated writer permission, separate from ``post_read``.""" + + if "project_lifecycle_write" not in permission_codes: + raise PermissionError("project_lifecycle_write permission is required") + + +__all__ = [ + "PROJECT_ACTOR_TYPE_CODES", + "PROJECT_EVENT_TYPE_CODES", + "PROJECT_RELATION_TYPE_CODES", + "ProjectLifecycleEventInput", + "ProjectLifecycleRelationInput", + "ProjectLifecycleResponsibilityInput", + "ProjectLifecycleValidationError", + "project_lifecycle_digest", + "require_project_lifecycle_write_permission", + "validate_project_lifecycle_event", +] diff --git a/migrations/0054_project_lifecycle_ingestion.sql b/migrations/0054_project_lifecycle_ingestion.sql new file mode 100644 index 000000000..558897c8a --- /dev/null +++ b/migrations/0054_project_lifecycle_ingestion.sql @@ -0,0 +1,129 @@ +begin; + +insert into common_lookup_value (lookup_category, lookup_code, lookup_label, display_order) +values + ('project_event_type', 'project_event_order', 'Project order or contract award', 0), + ('project_event_type', 'project_event_spec_change', 'Project specification change', 1), + ('project_event_type', 'project_event_delivery', 'Project delivery completion', 2), + ('project_event_type', 'project_event_voc', 'Project voice of customer', 3), + ('project_event_type', 'project_event_rebid', 'Project tender or rebid', 4), + ('project_record_state', 'project_record_active', 'Active project source record', 0), + ('project_record_state', 'project_record_withdrawn', 'Withdrawn project source record', 1), + ('project_relation_type', 'project_relation_precedes', 'Explicitly precedes', 0), + ('project_relation_type', 'project_relation_related', 'Explicitly related', 1), + ('prov_agent_type', 'prov_person', 'Person', 0), + ('prov_agent_type', 'prov_organization', 'Organization', 1), + ('prov_agent_type', 'prov_team', 'Team', 2) +on conflict (lookup_code) do nothing; + +create table if not exists project_identity ( + project_identity_id uuid primary key default uuid_generate_v4(), + project_key text not null unique, + project_name text not null +); + +create table if not exists project_source_system ( + project_source_system_id uuid primary key default uuid_generate_v4(), + source_system_code text not null unique, + source_system_name text not null +); + +create table if not exists project_event_mapping ( + project_event_mapping_id uuid primary key default uuid_generate_v4(), + project_source_system_id uuid not null references project_source_system, + mapping_version text not null, + source_event_code text not null, + project_event_type_code text not null references common_lookup_value (lookup_code), + is_active boolean not null default true, + created_at timestamptz not null default now(), + unique (project_source_system_id, mapping_version, source_event_code) +); + +create table if not exists project_source_record ( + project_source_record_id uuid primary key default uuid_generate_v4(), + project_source_system_id uuid not null references project_source_system, + source_record_key text not null, + project_identity_id uuid not null references project_identity, + project_event_mapping_id uuid not null references project_event_mapping, + lifecycle_state_code text not null references common_lookup_value (lookup_code), + record_digest text not null, + imported_by text not null, + imported_at timestamptz not null default now(), + unique (project_source_system_id, source_record_key) +); + +create table if not exists project_lifecycle_event ( + project_lifecycle_event_id uuid primary key default uuid_generate_v4(), + project_source_record_id uuid not null unique references project_source_record on delete cascade, + project_event_type_code text not null references common_lookup_value (lookup_code), + event_started_at timestamptz not null, + event_ended_at timestamptz, + event_digest text not null, + created_at timestamptz not null default now(), + check (event_ended_at is null or event_ended_at >= event_started_at) +); + +create table if not exists project_event_evidence ( + project_lifecycle_event_id uuid not null references project_lifecycle_event on delete cascade, + evidence_post_id uuid not null references source_post (post_id), + evidence_role_code text not null, + primary key (project_lifecycle_event_id, evidence_post_id, evidence_role_code), + unique (project_lifecycle_event_id, evidence_role_code) +); + +create table if not exists project_actor ( + project_actor_id uuid primary key default uuid_generate_v4(), + project_identity_id uuid not null references project_identity, + actor_type_code text not null references common_lookup_value (lookup_code), + actor_key text not null, + actor_name text not null, + unique (project_identity_id, actor_type_code, actor_key) +); + +create table if not exists project_event_responsibility ( + owner_source_record_id uuid not null references project_source_record on delete cascade, + project_lifecycle_event_id uuid not null references project_lifecycle_event on delete cascade, + project_actor_id uuid not null references project_actor, + responsibility_text text not null, + evidence_post_id uuid not null references source_post (post_id), + primary key (project_lifecycle_event_id, project_actor_id, responsibility_text), + unique (owner_source_record_id, project_lifecycle_event_id, project_actor_id, responsibility_text) +); + +create table if not exists project_event_relation ( + project_event_relation_id uuid primary key default uuid_generate_v4(), + owner_source_record_id uuid not null references project_source_record on delete cascade, + source_lifecycle_event_id uuid not null references project_lifecycle_event on delete cascade, + target_lifecycle_event_id uuid not null references project_lifecycle_event, + relation_type_code text not null references common_lookup_value (lookup_code), + evidence_post_id uuid not null references source_post (post_id), + unique (owner_source_record_id, source_lifecycle_event_id, target_lifecycle_event_id, relation_type_code), + check (source_lifecycle_event_id <> target_lifecycle_event_id) +); + +create table if not exists project_lifecycle_audit ( + project_lifecycle_audit_id uuid primary key default uuid_generate_v4(), + project_source_record_id uuid not null references project_source_record, + action_code text not null, + actor_key text not null, + mapping_version text not null, + before_digest text, + after_digest text, + occurred_at timestamptz not null default now() +); + +create index if not exists project_source_record_identity_idx + on project_source_record (project_identity_id, lifecycle_state_code); +create index if not exists project_lifecycle_event_time_idx + on project_lifecycle_event (event_started_at, project_event_type_code); +create index if not exists project_event_relation_target_idx + on project_event_relation (target_lifecycle_event_id); + +comment on table project_event_mapping is + 'Versioned explicit source-code mapping; title/body classification is not authoritative.'; +comment on table project_source_record is + 'Source-owned idempotency and withdrawal boundary for lifecycle projections.'; +comment on table project_lifecycle_audit is + 'Aggregate audit digests for administrative lifecycle upsert and withdrawal actions.'; + +commit; diff --git a/migrations/rollback/0054_project_lifecycle_ingestion.sql b/migrations/rollback/0054_project_lifecycle_ingestion.sql new file mode 100644 index 000000000..41d0214ea --- /dev/null +++ b/migrations/rollback/0054_project_lifecycle_ingestion.sql @@ -0,0 +1,12 @@ +begin; +drop table if exists project_event_relation; +drop table if exists project_event_responsibility; +drop table if exists project_event_evidence; +drop table if exists project_lifecycle_event; +drop table if exists project_lifecycle_audit; +drop table if exists project_source_record; +drop table if exists project_event_mapping; +drop table if exists project_actor; +drop table if exists project_source_system; +drop table if exists project_identity; +commit; diff --git a/tests/test_migration_replay.py b/tests/test_migration_replay.py index 29098f0a5..e4f103b0b 100644 --- a/tests/test_migration_replay.py +++ b/tests/test_migration_replay.py @@ -56,3 +56,15 @@ def test_migrate_sh_replays_global_ask_context_migration() -> None: ).read_text(encoding="utf-8") assert "0052_*" in script + + +def test_migrate_sh_replays_project_lifecycle_ingestion_migration() -> None: + """Existing Compose volumes must receive authoritative lifecycle tables.""" + script = ( + Path(__file__).resolve().parents[1] + / "docker" + / "postgres-init" + / "migrate.sh" + ).read_text(encoding="utf-8") + + assert "0054_*" in script diff --git a/tests/test_project_lifecycle_ingestion.py b/tests/test_project_lifecycle_ingestion.py new file mode 100644 index 000000000..0a65efbdc --- /dev/null +++ b/tests/test_project_lifecycle_ingestion.py @@ -0,0 +1,623 @@ +"""Real-PostgreSQL tests for the authoritative project-lifecycle writer. + +The fixtures contain only synthetic source identities. These tests exercise +the transaction, identity, evidence, replacement, relation, and withdrawal +boundaries against the same SQL migrations used by the product stack. +""" + +from __future__ import annotations + +import asyncio +from dataclasses import replace +from datetime import datetime, timezone +import os +from pathlib import Path +import uuid +from urllib.parse import urlsplit, urlunsplit + +import asyncpg +import psycopg2 +from psycopg2 import sql +import pytest + +from backend.app.project_lifecycle_ingestion import ( + ingest_project_lifecycle_event, + register_project_event_mapping, + withdraw_project_lifecycle_record, +) +from lineageweave.project_lifecycle import ( + ProjectLifecycleEventInput, + ProjectLifecycleRelationInput, + ProjectLifecycleResponsibilityInput, + ProjectLifecycleValidationError, + project_lifecycle_digest, + validate_project_lifecycle_event, +) + +_ADMIN_DSN = os.environ.get( + "LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN", "postgresql://localhost/postgres" +) +_ROOT = Path(__file__).resolve().parents[1] +_INITIAL_MIGRATION = _ROOT / "migrations" / "0001_initial_schema.sql" +_SOURCE_IDENTITY_MIGRATION = _ROOT / "migrations" / "0037_source_record_identity.sql" +_LIFECYCLE_MIGRATION = _ROOT / "migrations" / "0054_project_lifecycle_ingestion.sql" +_LIFECYCLE_ROLLBACK = _ROOT / "migrations" / "rollback" / "0054_project_lifecycle_ingestion.sql" + + +def _postgres_available() -> bool: + """Return whether the real local PostgreSQL test service is reachable.""" + + try: + psycopg2.connect(_ADMIN_DSN, connect_timeout=2).close() + return True + except psycopg2.OperationalError: + return False + + +pytestmark = pytest.mark.skipif( + not _postgres_available(), + reason=f"no reachable PostgreSQL server at {_ADMIN_DSN}", +) + + +def _database_dsn(database_name: str) -> str: + """Replace only the database path while preserving DSN parameters.""" + + parsed = urlsplit(_ADMIN_DSN) + return urlunsplit(parsed._replace(path=f"/{database_name}")) + + +@pytest.fixture +def lifecycle_database() -> tuple[str, dict[str, str]]: + """Create a migrated database with two synthetic source evidence posts.""" + + database_name = f"lineageweave_lifecycle_{uuid.uuid4().hex[:12]}" + admin = psycopg2.connect(_ADMIN_DSN) + admin.autocommit = True + with admin.cursor() as cursor: + cursor.execute(sql.SQL("create database {}").format(sql.Identifier(database_name))) + admin.close() + dsn = _database_dsn(database_name) + connection = psycopg2.connect(dsn) + post_ids = {"order": str(uuid.uuid4()), "other": str(uuid.uuid4()), "target": str(uuid.uuid4())} + try: + with connection.cursor() as cursor: + cursor.execute(_INITIAL_MIGRATION.read_text(encoding="utf-8")) + cursor.execute("create extension if not exists pg_trgm") + cursor.execute(_SOURCE_IDENTITY_MIGRATION.read_text(encoding="utf-8")) + cursor.execute(_LIFECYCLE_MIGRATION.read_text(encoding="utf-8")) + cursor.execute( + """ + insert into common_lookup_value + (lookup_category, lookup_code, lookup_label) + values + ('voc_type', 'voc', 'Voice of Customer'), + ('post_visibility', 'public', 'Public'), + ('corporate_entity_level', 'company', 'Company') + """ + ) + cursor.execute( + """ + insert into corporate_entity + (corporate_entity_code, entity_name, entity_level_code) + values ('SYNTH-CORP', 'Synthetic Corporation', 'company') + returning corporate_entity_id + """ + ) + corporate_id = cursor.fetchone()[0] + cursor.execute( + """ + insert into user_account (external_subject_id, display_name, email_address) + values ('synthetic-subject', 'Synthetic Importer', 'synthetic@example.test') + returning user_account_id + """ + ) + account_id = cursor.fetchone()[0] + for record_key, post_id in ( + ("order-1", post_ids["order"]), + ("other-1", post_ids["other"]), + ("target-1", post_ids["target"]), + ): + cursor.execute( + """ + insert into source_post + (post_id, author_account_id, corporate_entity_id, post_title, + post_body, voc_type_code, visibility_code, source_system_code, + source_record_key) + values (%s, %s, %s, 'Synthetic evidence', 'Synthetic evidence body', + 'voc', 'public', 'synthetic_source', %s) + """, + (post_id, account_id, corporate_id, record_key), + ) + connection.commit() + yield dsn, post_ids + finally: + connection.close() + admin = psycopg2.connect(_ADMIN_DSN) + admin.autocommit = True + with admin.cursor() as cursor: + cursor.execute(sql.SQL("drop database {}").format(sql.Identifier(database_name))) + admin.close() + + +def _event( + post_ids: dict[str, str], + *, + record_key: str = "order-1", + project_key: str = "P-100", + source_event_code: str = "ORDER_CREATED", + started_at: datetime | None = None, + relations: tuple[ProjectLifecycleRelationInput, ...] = (), + responsibilities: tuple[ProjectLifecycleResponsibilityInput, ...] = (), +) -> ProjectLifecycleEventInput: + """Build one synthetic explicit source event.""" + + return ProjectLifecycleEventInput( + project_key=project_key, + project_name="Synthetic Project", + source_system_code="synthetic_source", + source_record_key=record_key, + source_event_code=source_event_code, + mapping_version="v1", + event_started_at=started_at or datetime(2026, 1, 1, tzinfo=timezone.utc), + event_ended_at=None, + evidence_post_id=post_ids["order" if record_key == "order-1" else "target"], + relations=relations, + responsibilities=responsibilities, + ) + + +async def _register_mapping(connection: asyncpg.Connection, source_event_code: str = "ORDER_CREATED") -> None: + """Register a deterministic synthetic mapping for a test.""" + + await register_project_event_mapping( + connection, + source_system_code="synthetic_source", + source_system_name="Synthetic Source", + mapping_version="v1", + source_event_code=source_event_code, + project_event_type_code="project_event_order", + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + + +def test_permission_is_separate_from_post_read() -> None: + """The writer cannot be authorized by the buyer read permission alone.""" + + from lineageweave.project_lifecycle import require_project_lifecycle_write_permission + + with pytest.raises(PermissionError, match="project_lifecycle_write"): + require_project_lifecycle_write_permission({"post_read"}) + + +def test_lifecycle_contract_rejects_invalid_time_uuid_project_and_relation_inputs() -> None: + """Validation fails closed before a database UUID or timestamp cast.""" + + post_ids = {"order": str(uuid.uuid4()), "target": str(uuid.uuid4())} + valid = _event(post_ids) + invalid_events = ( + (replace(valid, project_key=" "), "project_key"), + (replace(valid, project_key="x" * 257), "project key"), + (replace(valid, event_started_at="not a datetime"), "datetime"), + (replace(valid, event_started_at=datetime(2026, 1, 1)), "timezone"), + ( + replace(valid, event_ended_at=datetime(2025, 12, 31, tzinfo=timezone.utc)), + "precede", + ), + (replace(valid, evidence_post_id="not-a-uuid"), "UUID"), + ( + replace( + valid, + relations=( + ProjectLifecycleRelationInput( + "synthetic_source", + "order-1", + "unsupported_relation", + post_ids["order"], + ), + ), + ), + "relation_type_code", + ), + ( + replace( + valid, + relations=( + ProjectLifecycleRelationInput( + "synthetic_source", + " order-1 ", + "project_relation_related", + post_ids["order"], + ), + ), + ), + "itself", + ), + ( + replace( + valid, + responsibilities=( + ProjectLifecycleResponsibilityInput( + "unsupported_actor", + "actor-1", + "Synthetic Actor", + "process the order", + post_ids["order"], + ), + ), + ), + "actor_type_code", + ), + ) + + for invalid_event, message in invalid_events: + with pytest.raises(ProjectLifecycleValidationError, match=message): + validate_project_lifecycle_event(invalid_event) + + assert len(project_lifecycle_digest(valid)) == 64 + + +def test_mapping_administration_rejects_unauthorized_or_invalid_requests(lifecycle_database) -> None: + """Mapping registration has its own permission and explicit-code checks.""" + dsn, post_ids = lifecycle_database + + async def run() -> None: + connection = await asyncpg.connect(dsn) + try: + with pytest.raises(PermissionError, match="project_lifecycle_write"): + await register_project_event_mapping( + connection, + source_system_code="synthetic_source", + source_system_name="Synthetic Source", + mapping_version="v1", + source_event_code="ORDER_CREATED", + project_event_type_code="project_event_order", + administrator_key="synthetic-admin", + permission_codes={"post_read"}, + ) + with pytest.raises(PermissionError, match="administrator_key"): + await register_project_event_mapping( + connection, + source_system_code="synthetic_source", + source_system_name="Synthetic Source", + mapping_version="v1", + source_event_code="ORDER_CREATED", + project_event_type_code="project_event_order", + administrator_key=" ", + permission_codes={"project_lifecycle_write"}, + ) + with pytest.raises(ProjectLifecycleValidationError, match="mapping fields"): + await register_project_event_mapping( + connection, + source_system_code=" ", + source_system_name="Synthetic Source", + mapping_version="v1", + source_event_code="ORDER_CREATED", + project_event_type_code="project_event_order", + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + with pytest.raises(ProjectLifecycleValidationError, match="unsupported"): + await register_project_event_mapping( + connection, + source_system_code="synthetic_source", + source_system_name="Synthetic Source", + mapping_version="v1", + source_event_code="ORDER_CREATED", + project_event_type_code="unsupported_event", + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + with pytest.raises(PermissionError, match="administrator_key"): + await ingest_project_lifecycle_event( + connection, + _event(post_ids), + administrator_key=" ", + permission_codes={"project_lifecycle_write"}, + ) + with pytest.raises(PermissionError, match="administrator_key"): + await withdraw_project_lifecycle_record( + connection, + source_system_code="synthetic_source", + source_record_key="order-1", + administrator_key=" ", + permission_codes={"project_lifecycle_write"}, + ) + with pytest.raises(ProjectLifecycleValidationError, match="source identity"): + await withdraw_project_lifecycle_record( + connection, + source_system_code=" ", + source_record_key="order-1", + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + finally: + await connection.close() + + asyncio.run(run()) + + +def test_upsert_is_idempotent_and_replaces_source_owned_event(lifecycle_database) -> None: + dsn, post_ids = lifecycle_database + + async def run() -> tuple[dict[str, object], dict[str, object], list[asyncpg.Record]]: + connection = await asyncpg.connect(dsn) + try: + await _register_mapping(connection) + responsibility = ProjectLifecycleResponsibilityInput( + actor_type_code="prov_person", + actor_key="actor-1", + actor_name="Synthetic Actor", + responsibility_text="process the order", + evidence_post_id=post_ids["order"], + ) + first = await ingest_project_lifecycle_event( + connection, + _event(post_ids, responsibilities=(responsibility,)), + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + second = await ingest_project_lifecycle_event( + connection, + _event( + post_ids, + started_at=datetime(2026, 1, 2, tzinfo=timezone.utc), + responsibilities=(responsibility,), + ), + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + rows = await connection.fetch( + """ + select event.event_started_at, record.lifecycle_state_code + from project_lifecycle_event event + join project_source_record record + on record.project_source_record_id = event.project_source_record_id + """ + ) + assert await connection.fetchval( + "select count(*) from project_event_responsibility" + ) == 1 + return first, second, rows + finally: + await connection.close() + + first, second, rows = asyncio.run(run()) + assert first["status_code"] == "inserted" + assert second["status_code"] == "replaced" + assert first["project_lifecycle_event_id"] == second["project_lifecycle_event_id"] + assert len(rows) == 1 + assert rows[0]["event_started_at"].date().isoformat() == "2026-01-02" + assert rows[0]["lifecycle_state_code"] == "project_record_active" + + +def test_concurrent_duplicate_imports_converge_on_one_event(lifecycle_database) -> None: + """The source-identity advisory lock serializes duplicate imports.""" + + dsn, post_ids = lifecycle_database + + async def run() -> tuple[list[dict[str, object]], int]: + setup = await asyncpg.connect(dsn) + await _register_mapping(setup) + await setup.close() + connections = [await asyncpg.connect(dsn) for _ in range(2)] + try: + results = await asyncio.gather( + *( + ingest_project_lifecycle_event( + connection, + _event(post_ids), + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + for connection in connections + ) + ) + count = await connections[0].fetchval("select count(*) from project_lifecycle_event") + return results, count + finally: + await asyncio.gather(*(connection.close() for connection in connections)) + + results, count = asyncio.run(run()) + assert {result["status_code"] for result in results} == {"inserted", "replaced"} + assert count == 1 + + +def test_unknown_mapping_and_mismatched_evidence_fail_without_mutation(lifecycle_database) -> None: + dsn, post_ids = lifecycle_database + + async def run() -> int: + connection = await asyncpg.connect(dsn) + try: + with pytest.raises(ProjectLifecycleValidationError, match="mapping"): + await ingest_project_lifecycle_event( + connection, + _event(post_ids, source_event_code="UNREGISTERED"), + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + await _register_mapping(connection) + mismatched = ProjectLifecycleEventInput( + **{ + **_event(post_ids).__dict__, + "evidence_post_id": post_ids["other"], + } + ) + with pytest.raises(ProjectLifecycleValidationError, match="belong"): + await ingest_project_lifecycle_event( + connection, + mismatched, + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + return await connection.fetchval("select count(*) from project_lifecycle_event") + finally: + await connection.close() + + assert asyncio.run(run()) == 0 + + +def test_registered_mapping_must_resolve_to_supported_event_type(lifecycle_database) -> None: + """A direct catalog corruption cannot turn an unrelated lookup into an event.""" + dsn, post_ids = lifecycle_database + + async def run() -> None: + connection = await asyncpg.connect(dsn) + try: + source_system_id = await connection.fetchval( + """ + insert into project_source_system (source_system_code, source_system_name) + values ('invalid_mapping_source', 'Invalid Mapping Source') + returning project_source_system_id + """ + ) + await connection.execute( + """ + insert into project_event_mapping + (project_source_system_id, mapping_version, source_event_code, + project_event_type_code) + values ($1, 'v1', 'INVALID_TYPE', 'public') + """, + source_system_id, + ) + invalid_event = replace( + _event(post_ids), + source_system_code="invalid_mapping_source", + source_event_code="INVALID_TYPE", + ) + with pytest.raises(ProjectLifecycleValidationError, match="unregistered"): + await ingest_project_lifecycle_event( + connection, + invalid_event, + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + finally: + await connection.close() + + asyncio.run(run()) + + +def test_cross_project_relation_is_rejected(lifecycle_database) -> None: + dsn, post_ids = lifecycle_database + + async def run() -> int: + connection = await asyncpg.connect(dsn) + try: + await _register_mapping(connection) + await ingest_project_lifecycle_event( + connection, + _event(post_ids, record_key="target-1", project_key="P-200"), + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + relation = ProjectLifecycleRelationInput( + target_source_system_code="synthetic_source", + target_source_record_key="target-1", + relation_type_code="project_relation_precedes", + evidence_post_id=post_ids["order"], + ) + with pytest.raises(ProjectLifecycleValidationError, match="same project"): + await ingest_project_lifecycle_event( + connection, + _event(post_ids, relations=(relation,)), + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + return await connection.fetchval("select count(*) from project_lifecycle_event") + finally: + await connection.close() + + assert asyncio.run(run()) == 1 + + +def test_withdrawal_removes_owned_projection_and_keeps_independent_event(lifecycle_database) -> None: + dsn, post_ids = lifecycle_database + + async def run() -> tuple[dict[str, object], list[asyncpg.Record]]: + connection = await asyncpg.connect(dsn) + try: + await _register_mapping(connection) + await ingest_project_lifecycle_event( + connection, + _event(post_ids, record_key="target-1"), + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + relation = ProjectLifecycleRelationInput( + target_source_system_code="synthetic_source", + target_source_record_key="target-1", + relation_type_code="project_relation_precedes", + evidence_post_id=post_ids["order"], + ) + await ingest_project_lifecycle_event( + connection, + _event(post_ids, relations=(relation,)), + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + result = await withdraw_project_lifecycle_record( + connection, + source_system_code="synthetic_source", + source_record_key="order-1", + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + rows = await connection.fetch( + "select source_record_key from project_source_record where lifecycle_state_code = 'project_record_active'" + ) + return result, rows + finally: + await connection.close() + + result, active_rows = asyncio.run(run()) + assert result["status_code"] == "withdrawn" + assert [row["source_record_key"] for row in active_rows] == ["target-1"] + + +def test_withdrawal_of_unknown_source_is_idempotent_not_found(lifecycle_database) -> None: + """A repeated withdrawal returns a safe aggregate status without a write.""" + dsn, _ = lifecycle_database + + async def run() -> dict[str, object]: + connection = await asyncpg.connect(dsn) + try: + return await withdraw_project_lifecycle_record( + connection, + source_system_code="synthetic_source", + source_record_key="missing-record", + administrator_key="synthetic-admin", + permission_codes={"project_lifecycle_write"}, + ) + finally: + await connection.close() + + assert asyncio.run(run()) == {"status_code": "not_found"} + + +def test_lifecycle_rollback_removes_the_migration_owned_tables(lifecycle_database) -> None: + """A deployment rollback removes only the 0054 lifecycle schema objects.""" + dsn, _ = lifecycle_database + connection = psycopg2.connect(dsn) + try: + with connection.cursor() as cursor: + cursor.execute(_LIFECYCLE_ROLLBACK.read_text(encoding="utf-8")) + cursor.execute( + """ + select count(*) + from information_schema.tables + where table_schema = 'public' + and table_name in ( + 'project_identity', 'project_source_system', + 'project_event_mapping', 'project_source_record', + 'project_lifecycle_event', 'project_event_evidence', + 'project_actor', 'project_event_responsibility', + 'project_event_relation', 'project_lifecycle_audit' + ) + """ + ) + assert cursor.fetchone()[0] == 0 + connection.commit() + finally: + connection.close()