From d6ed152f295c92c4d31e73d165d8ac6506f34f55 Mon Sep 17 00:00:00 2001 From: Codex Date: Fri, 28 Aug 2026 20:47:30 +0900 Subject: [PATCH 1/2] feat(journey): admit digest-bound temporal evidence --- backend/app/project_history.py | 22 +- backend/app/project_journey_temporal.py | 117 ++++++++++ ...bound-project-journey-temporal-evidence.md | 61 +++++ docs/adr/README.md | 1 + docs/product-requirements.md | 3 + docs/product-technical-gap-baseline.md | 2 +- .../ProjectHistoryTimeline.test.tsx | 12 +- .../src/components/ProjectHistoryTimeline.tsx | 5 + frontend/src/projectHistory.ts | 11 + lineageweave/project_history.py | 11 + lineageweave/temporal_journey_artifact.py | 119 ++++++++++ ...0259_project_journey_temporal_artifact.sql | 52 +++++ tests/test_project_history.py | 15 +- tests/test_project_history_ingestion.py | 3 + tests/test_temporal_journey_artifact.py | 220 ++++++++++++++++++ 15 files changed, 650 insertions(+), 4 deletions(-) create mode 100644 backend/app/project_journey_temporal.py create mode 100644 docs/adr/0270-digest-bound-project-journey-temporal-evidence.md create mode 100644 lineageweave/temporal_journey_artifact.py create mode 100644 migrations/0259_project_journey_temporal_artifact.sql create mode 100644 tests/test_temporal_journey_artifact.py diff --git a/backend/app/project_history.py b/backend/app/project_history.py index 9389ef5fd..2cad58e64 100644 --- a/backend/app/project_history.py +++ b/backend/app/project_history.py @@ -133,8 +133,28 @@ async def fetch(self, query: str, *args: object) -> Sequence[Mapping[str, Any]]: order by role.post_id, role.actor_type_code, role.actor_name, role.responsibility """ _EDGE_SQL = """ -select edge.parent_post_id, edge.child_post_id, edge.fused_score +select edge.parent_post_id, edge.child_post_id, edge.fused_score, + temporal.observed as temporal_observed, + temporal.allen_relations, + temporal.artifact_digest_sha256 from post_lineage_edge edge + left join lateral ( + select relation.observed, + array_agg(kind.relation_code order by kind.relation_ordinal) as allen_relations, + artifact.artifact_digest_sha256 + from project_journey_temporal_relation relation + join project_journey_temporal_artifact artifact + on artifact.analysis_run_id = relation.analysis_run_id + join project_journey_temporal_relation_kind kind + on kind.analysis_run_id = relation.analysis_run_id + and kind.left_post_id = relation.left_post_id + and kind.right_post_id = relation.right_post_id + where relation.left_post_id = edge.parent_post_id + and relation.right_post_id = edge.child_post_id + group by relation.observed, artifact.artifact_digest_sha256, artifact.admitted_at + order by artifact.admitted_at desc, artifact.artifact_digest_sha256 desc + limit 1 + ) temporal on true where edge.parent_post_id = any($1::uuid[]) and edge.child_post_id = any($1::uuid[]) order by edge.child_post_id, edge.parent_post_id diff --git a/backend/app/project_journey_temporal.py b/backend/app/project_journey_temporal.py new file mode 100644 index 000000000..0d3cf214d --- /dev/null +++ b/backend/app/project_journey_temporal.py @@ -0,0 +1,117 @@ +"""Persist provider-owned temporal evidence for already-admitted journey edges.""" + +from __future__ import annotations + +from typing import Any, Protocol + +from lineageweave.temporal_journey_artifact import ( + ALLEN_RELATIONS, + TemporalJourneyArtifact, + parse_temporal_journey_artifact, +) + + +class TemporalArtifactConnection(Protocol): + """Minimal transaction-scoped database port for artifact admission.""" + + async def fetchrow(self, query: str, *args: object) -> Any: + """Read one binding row.""" + + ... + + async def execute(self, query: str, *args: object) -> Any: + """Execute one immutable persistence statement.""" + + ... + + async def executemany(self, query: str, args: list[tuple[object, ...]]) -> Any: + """Execute bounded normalized child inserts.""" + + ... + + +class TemporalArtifactAdmissionError(ValueError): + """The artifact cannot be bound to the declared persisted run.""" + + +async def persist_project_journey_temporal_artifact( + conn: TemporalArtifactConnection, + *, + analysis_run_id: str, + payload: bytes, + expected_run_id: str, + expected_snapshot_id: str, + expected_input_digest_sha256: str, + expected_artifact_digest_sha256: str, +) -> TemporalJourneyArtifact: + """Validate and immutably persist temporal evidence for existing edges. + + The foreign key to ``post_lineage_edge`` is the semantic admission gate: + interval order can corroborate an admitted predecessor, but cannot create + a predecessor, branch, responsibility handoff, or causal transition. + """ + + artifact = parse_temporal_journey_artifact( + payload, + expected_run_id=expected_run_id, + expected_snapshot_id=expected_snapshot_id, + expected_input_digest_sha256=expected_input_digest_sha256, + expected_artifact_digest_sha256=expected_artifact_digest_sha256, + ) + binding = await conn.fetchrow( + "select remote_run_id from analysis_run_tepp_result where analysis_run_id = $1::uuid", + analysis_run_id, + ) + if binding is None or str(binding["remote_run_id"]) != expected_run_id: + raise TemporalArtifactAdmissionError("artifact run does not match a persisted terminal result") + existing = await conn.fetchrow( + "select artifact_digest_sha256 from project_journey_temporal_artifact " + "where analysis_run_id = $1::uuid for update", + analysis_run_id, + ) + if existing is not None: + if str(existing["artifact_digest_sha256"]) != expected_artifact_digest_sha256: + raise TemporalArtifactAdmissionError("analysis run already has a different artifact") + return artifact + await conn.execute( + "insert into project_journey_temporal_artifact " + "(analysis_run_id, remote_run_id, schema_version, snapshot_id, input_digest_sha256, artifact_digest_sha256) " + "values ($1::uuid, $2, $3, $4, $5, $6)", + analysis_run_id, + expected_run_id, + "tepp.tdt_chronos_interval_consistency.v1", + expected_snapshot_id, + expected_input_digest_sha256, + expected_artifact_digest_sha256, + ) + relation_rows = [ + (analysis_run_id, relation.left_event_id, relation.right_event_id, relation.observed) + for relation in artifact.relations + ] + await conn.executemany( + "insert into project_journey_temporal_relation " + "(analysis_run_id, left_post_id, right_post_id, observed) " + "values ($1::uuid, $2::uuid, $3::uuid, $4)", + relation_rows, + ) + await conn.executemany( + "insert into project_journey_temporal_relation_kind " + "(analysis_run_id, left_post_id, right_post_id, relation_code, relation_ordinal) " + "values ($1::uuid, $2::uuid, $3::uuid, $4, $5)", + [ + (analysis_run_id, relation.left_event_id, relation.right_event_id, code, ALLEN_RELATIONS.index(code)) + for relation in artifact.relations + for code in relation.allen_relations + ], + ) + await conn.executemany( + "insert into project_journey_temporal_support " + "(analysis_run_id, left_post_id, right_post_id, assertion_ordinal) " + "values ($1::uuid, $2::uuid, $3::uuid, $4)", + [ + (analysis_run_id, relation.left_event_id, relation.right_event_id, ordinal) + for relation in artifact.relations + for ordinal in relation.support_assertion_ordinals + ], + ) + return artifact diff --git a/docs/adr/0270-digest-bound-project-journey-temporal-evidence.md b/docs/adr/0270-digest-bound-project-journey-temporal-evidence.md new file mode 100644 index 000000000..a95ecf03a --- /dev/null +++ b/docs/adr/0270-digest-bound-project-journey-temporal-evidence.md @@ -0,0 +1,61 @@ +# ADR 0270: Digest-bound project-journey temporal evidence + +- Status: Accepted on this stacked branch; not protected-main truth until merge +- Date: 2026-08-28 +- Depends on: ADR 0132, ADR 0231, ADR 0243; TEPP PR #291 +- Figma file ID: `SBpgot7uTvMxEaxUwvoc0S` + +## Context + +TEPP PR #291 publishes canonical JSON and GraphML for bounded Allen interval- +consistency results. The artifact binds a run, snapshot, exact input digest, +ordered event pair, observed/derived status, and supporting assertion ordinals. +It deliberately does not claim that temporal order is a causal transition, +project predecessor, or business-process branch. + +LineageWeave already admits related predecessor paths through authorized +`post_lineage_edge` evidence. Promoting every temporally ordered pair to a +project journey would contradict PRD-FR-5E and ADR 0243. + +## Decision + +LineageWeave accepts only canonical artifact bytes whose SHA-256, run, +snapshot, and exact input digest match caller-computed expected values. The +remote run must also match a persisted terminal TEPP result. Metadata, +relations, elementary Allen kinds, and support ordinals persist in normalized +tables. + +Every admitted temporal pair must already be an exact `post_lineage_edge`. +The database foreign key enforces that boundary. Temporal evidence may +corroborate the time order of an existing related-history path; it never +creates a predecessor, branch, responsibility handoff, or causal transition. +A branch is visible only when the independently admitted lineage graph already +contains that topology. A transition still requires its separately governed +observed business or responsibility evidence. + +The Project History API attaches the newest immutable temporal evidence to +the corresponding visible edge after ABAC and cutoff filtering select both +endpoints. The customer UI says what the user can do next—open the supporting +records and compare dates—and never names the calculation module. + +GraphML is an equivalent provider export, not the ingestion authority. The +canonical typed JSON is the sole admitted payload so two representations +cannot diverge inside the database. + +## Consequences + +- Exact temporal consistency becomes durable and auditable without duplicating + mathematical reasoning in Python. +- A valid artifact containing a pair absent from Event Lineage fails closed at + the foreign-key boundary and rolls back its transaction. +- A future contract that explicitly carries business predecessor or transition + semantics requires a new ADR; this artifact cannot be reinterpreted later. + +## References + +Allen, J. F. (1983). Maintaining knowledge about temporal intervals. +*Communications of the ACM, 26*(11), 832–843. +https://doi.org/10.1145/182.358434 + +World Wide Web Consortium. (2013). *PROV-O: The PROV ontology*. +https://www.w3.org/TR/prov-o/ diff --git a/docs/adr/README.md b/docs/adr/README.md index 2aedc44e0..649192595 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -28,6 +28,7 @@ decision from them. | [`operability/http-concurrency-evidence.md`](../operability/http-concurrency-evidence.md) | [0204](0204-analysis-run-short-transaction-delivery.md), [0212](0212-single-query-authorized-post-filter-options.md), [0213](0213-global-ask-embedding-pool-release.md) | | [`operability/mcp-concurrency-evidence.md`](../operability/mcp-concurrency-evidence.md) | [0218](0218-current-contract-mcp-global-ask.md) | | Evidence operations Dashboard (`/`) | [0206](0206-evidence-operations-dashboard.md) | +| Project-journey temporal evidence | [0270](0270-digest-bound-project-journey-temporal-evidence.md) | | [`temporal-topic-context-influence-research.md`](../temporal-topic-context-influence-research.md) | [0210](0210-temporal-topic-context-influence-dashboard.md) | | [`python-mathematical-compute-boundary-audit.md`](../doctoring/python-mathematical-compute-boundary-audit.md) | [0208](0208-externalize-local-mathematical-compute.md) | | [`WORKER_FUNCTION_TAXONOMY_REFERENCES.md`](../doctoring/WORKER_FUNCTION_TAXONOMY_REFERENCES.md) | [0232](0232-worker-function-taxonomy-in-the-published-ontology.md) | diff --git a/docs/product-requirements.md b/docs/product-requirements.md index 305da5704..5ee3f028c 100644 --- a/docs/product-requirements.md +++ b/docs/product-requirements.md @@ -459,6 +459,9 @@ comes from the cited answer rather than frontend inference. - Present project-specific journeys only from accepted evidence-bearing predecessor and branch relations. A timestamp sort may be labeled observed events, but never promoted to a journey. +- Attach digest-bound interval-consistency evidence only to an already + admitted predecessor edge. Temporal order alone never creates a predecessor, + branch, responsibility handoff, or causal transition (ADR 0270). Acceptance: every populated fact, lifecycle endpoint, membership, and journey event opens an authorized evidence post; an incomplete provenance chain fails diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index f4041e314..0f14ac724 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -836,7 +836,7 @@ post-merge reruns (not transferable evidence for later heads): | ---: | --- | --- | | #643 | Shared StatusNotice (ADR 0220): success/unavailable/retry states, WorkspaceCalendar auth-unavailable copy, 5-locale i18n; CI Full suite 22m54s green | ADR 0220 | | #644 | Native workspace surface split: 9 conditionally rendered components as lazy() dynamic imports behind a SurfaceBoundary error boundary; build emits 9 chunks (1.5-37 kB), main bundle 543 kB; 470 frontend tests, tsc, Storybook green | — | -| #762 | Evidence-bound project history (ADR 0243): /api/projects/{key}/history endpoint, project_history.py projection, fetchProjectHistory client, standalone ProjectHistoryTimeline component; supersedes #668 (3-way merge kept only the additive +2279/-0, dropping the branch's 8k shared-file reverts; popup UI hookup deferred as a scoped follow-up) | ADR 0243 | +| #762 | Evidence-bound project history (ADR 0243): /api/projects/{key}/history endpoint, project_history.py projection, fetchProjectHistory client, standalone ProjectHistoryTimeline component; supersedes #668 (3-way merge kept only the additive +2279/-0, dropping the branch's 8k shared-file reverts; popup UI hookup deferred as a scoped follow-up). ADR 0270 successor work admits digest-bound interval evidence only for existing lineage edges; it does not promote time order to a business transition. | ADR 0243, ADR 0270 | | #763 | Live-PostgreSQL A→B→A Voice history validation (ADR 0252) proving effective_from/effective_to interval replacement across repeated primary-Voice imports | ADR 0252 | | #764 | Test-only coverage lift: observability 78%→96%, post_summary 77%→89%, claim_verification 86%→99%; package line coverage 93.5%→95% (484→371 missing); 1651 Python tests green | — | | #761 | Temporal imported-primary Voice history (ADR 0252): migration 0243 (`effective_to` + GiST primary-period exclusion + synchronize trigger), refined 0237 `least()` effective_from backfill, `effective_from/effective_to` dataclass/export + `coalesce($2,$3)` cutoff predicate. Completes the half-shipped main layer that queried `voice.effective_to` against a missing column. CI Full suite 19m13s green | ADR 0252 | diff --git a/frontend/src/components/ProjectHistoryTimeline.test.tsx b/frontend/src/components/ProjectHistoryTimeline.test.tsx index c58d54efb..afb921cba 100644 --- a/frontend/src/components/ProjectHistoryTimeline.test.tsx +++ b/frontend/src/components/ProjectHistoryTimeline.test.tsx @@ -87,7 +87,16 @@ const projection: ProjectHistoryProjection = { target_event_id: "voc", event_ids: ["award", "spec", "voc"], edges: [ - { parent_event_id: "award", child_event_id: "spec", fused_score: 0.91 }, + { + parent_event_id: "award", + child_event_id: "spec", + fused_score: 0.91, + temporal_evidence: { + truth_status_code: "inferred", + interval_relations: ["before"], + artifact_digest_sha256: "a".repeat(64), + }, + }, { parent_event_id: "spec", child_event_id: "voc", fused_score: 0.73 }, ], minimum_fused_score: 0.73, @@ -114,6 +123,7 @@ describe("ProjectHistoryTimeline", () => { expect(screen.queryByText("document_time")).not.toBeInTheDocument(); expect(screen.getByText("delivery")).toBeInTheDocument(); expect(screen.getByText("delivered")).toBeInTheDocument(); + expect(screen.getByText(/Time order checked/)).toBeInTheDocument(); fireEvent.click(screen.getByRole("button", { name: /open source record: VOC received/i })); expect(onOpenPost).toHaveBeenCalledWith("post-voc"); diff --git a/frontend/src/components/ProjectHistoryTimeline.tsx b/frontend/src/components/ProjectHistoryTimeline.tsx index e0c2a1f2e..a9b994d09 100644 --- a/frontend/src/components/ProjectHistoryTimeline.tsx +++ b/frontend/src/components/ProjectHistoryTimeline.tsx @@ -259,6 +259,11 @@ export function ProjectHistoryTimeline({ {projectHistoryText(locale, "inferred")} + {path.edges.some((edge) => edge.temporal_evidence != null) ? ( + + {projectHistoryText(locale, "timeOrderChecked")} + + ) : null} ))} diff --git a/frontend/src/projectHistory.ts b/frontend/src/projectHistory.ts index 4dceb6fbd..d0dacb3f3 100644 --- a/frontend/src/projectHistory.ts +++ b/frontend/src/projectHistory.ts @@ -28,6 +28,11 @@ export interface ProjectHistoryPathEdge { parent_event_id: string; child_event_id: string; fused_score: number; + temporal_evidence?: { + truth_status_code: ProjectHistoryTruthStatus; + interval_relations: string[]; + artifact_digest_sha256: string; + } | null; } export interface ProjectHistoryPriorPath { @@ -113,6 +118,7 @@ const MESSAGE_KEYS = [ "priorHistory", "noPriorHistory", "inferredBoundary", + "timeOrderChecked", "projectEvidence", "sourceRecordEvidence", "supportingRecordEvidence", @@ -164,6 +170,7 @@ const EN: Record = { priorHistory: "Related prior history", noPriorHistory: "No visible prior lineage path is recorded for this event.", inferredBoundary: "This is inferred related history, not causality or an authoritative assignment record.", + timeOrderChecked: "Time order checked. Open the records above to compare the supporting dates.", projectEvidence: "Project identity evidence", sourceRecordEvidence: "Source record", supportingRecordEvidence: "Supporting record", @@ -212,6 +219,7 @@ const MESSAGES: Record> = { priorHistory: "관련 과거 이력", noPriorHistory: "이 이벤트로 이어지는 공개 가능한 이전 계보가 없습니다.", inferredBoundary: "이는 추론된 관련 이력이며 인과관계나 권위 있는 인사 배정 기록이 아닙니다.", + timeOrderChecked: "시간 순서를 확인했습니다. 위 기록을 열어 근거 날짜를 비교하세요.", projectEvidence: "프로젝트 식별 근거", sourceRecordEvidence: "원천 기록", supportingRecordEvidence: "뒷받침 기록", @@ -257,6 +265,7 @@ const MESSAGES: Record> = { priorHistory: "相关既往历史", noPriorHistory: "此事件没有可见的既往谱系路径。", inferredBoundary: "这是推断的相关历史，并非因果关系或权威任命记录。", + timeOrderChecked: "时间顺序已核验。请打开上方记录比较依据日期。", projectEvidence: "项目身份依据", sourceRecordEvidence: "来源记录", supportingRecordEvidence: "支持记录", @@ -302,6 +311,7 @@ const MESSAGES: Record> = { priorHistory: "関連する過去履歴", noPriorHistory: "このイベントに至る可視の過去系譜はありません。", inferredBoundary: "これは推論された関連履歴であり、因果関係や権威ある配属記録ではありません。", + timeOrderChecked: "時間順序を確認しました。上の記録を開いて根拠の日付を比較してください。", projectEvidence: "プロジェクト識別根拠", sourceRecordEvidence: "元レコード", supportingRecordEvidence: "根拠レコード", @@ -347,6 +357,7 @@ const MESSAGES: Record> = { priorHistory: "Lịch sử trước đó có liên quan", noPriorHistory: "Không có đường dẫn lịch sử trước đó khả kiến cho sự kiện này.", inferredBoundary: "Đây là lịch sử liên quan được suy luận, không phải quan hệ nhân quả hay hồ sơ phân công có thẩm quyền.", + timeOrderChecked: "Thứ tự thời gian đã được kiểm tra. Hãy mở các bản ghi trên để so sánh ngày làm căn cứ.", projectEvidence: "Bằng chứng nhận dạng dự án", sourceRecordEvidence: "Bản ghi nguồn", supportingRecordEvidence: "Bản ghi hỗ trợ", diff --git a/lineageweave/project_history.py b/lineageweave/project_history.py index 281586b03..04db6ed4f 100644 --- a/lineageweave/project_history.py +++ b/lineageweave/project_history.py @@ -148,6 +148,17 @@ def _prior_paths( "parent_event_id": parent, "child_event_id": child, "fused_score": _score(row["fused_score"]), + "temporal_evidence": ( + { + "truth_status_code": ( + "observed" if row.get("temporal_observed") else "inferred" + ), + "interval_relations": list(row.get("allen_relations") or ()), + "artifact_digest_sha256": row.get("artifact_digest_sha256"), + } + if row.get("artifact_digest_sha256") is not None + else None + ), } ) for edges in reverse_edges.values(): diff --git a/lineageweave/temporal_journey_artifact.py b/lineageweave/temporal_journey_artifact.py new file mode 100644 index 000000000..ef54513a1 --- /dev/null +++ b/lineageweave/temporal_journey_artifact.py @@ -0,0 +1,119 @@ +"""Validate TEPP interval-consistency artifacts without inventing journeys.""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass +from typing import Final + +SCHEMA_VERSION: Final = "tepp.tdt_chronos_interval_consistency.v1" +MAX_ARTIFACT_BYTES: Final = 4 * 1024 * 1024 +MAX_RELATIONS: Final = 100_000 +ALLEN_RELATIONS: Final = ( + "before", "after", "meets", "met_by", "overlaps", "overlapped_by", + "starts", "started_by", "during", "contains", "finishes", "finished_by", "equals", +) +_DIGEST = re.compile(r"^[0-9a-f]{64}$") + + +class TemporalJourneyArtifactError(ValueError): + """A fail-closed temporal-artifact contract violation.""" + + +@dataclass(frozen=True) +class TemporalRelation: + """One bounded observed or closure-derived interval relation.""" + + left_event_id: str + right_event_id: str + allen_relations: tuple[str, ...] + observed: bool + support_assertion_ordinals: tuple[int, ...] + + +@dataclass(frozen=True) +class TemporalJourneyArtifact: + """A canonical digest-bound interval-consistency artifact.""" + + run_id: str + snapshot_id: str + input_digest_sha256: str + relations: tuple[TemporalRelation, ...] + artifact_digest_sha256: str + + +def parse_temporal_journey_artifact( + payload: bytes, + *, + expected_run_id: str, + expected_snapshot_id: str, + expected_input_digest_sha256: str, + expected_artifact_digest_sha256: str, +) -> TemporalJourneyArtifact: + """Parse canonical provider JSON and bind every caller-owned identity.""" + + if not payload or len(payload) > MAX_ARTIFACT_BYTES: + raise TemporalJourneyArtifactError("artifact size is outside the supported bound") + if not all( + _DIGEST.fullmatch(value) + for value in (expected_input_digest_sha256, expected_artifact_digest_sha256) + ): + raise TemporalJourneyArtifactError("expected digest is not lowercase SHA-256") + if hashlib.sha256(payload).hexdigest() != expected_artifact_digest_sha256: + raise TemporalJourneyArtifactError("artifact bytes do not match the expected digest") + try: + decoded = payload.decode("utf-8") + value = json.loads(decoded) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise TemporalJourneyArtifactError("artifact is not valid UTF-8 JSON") from exc + if json.dumps(value, ensure_ascii=False, separators=(",", ":")) != decoded: + raise TemporalJourneyArtifactError("artifact JSON is not canonical") + if not isinstance(value, dict) or set(value) != { + "schema_version", "run_id", "snapshot_id", "input_digest_sha256", "relations" + }: + raise TemporalJourneyArtifactError("artifact object shape is unsupported") + if ( + value["schema_version"] != SCHEMA_VERSION + or value["run_id"] != expected_run_id + or value["snapshot_id"] != expected_snapshot_id + or value["input_digest_sha256"] != expected_input_digest_sha256 + ): + raise TemporalJourneyArtifactError("artifact identity does not match the admitted run") + raw_relations = value["relations"] + if not isinstance(raw_relations, list) or not 1 <= len(raw_relations) <= MAX_RELATIONS: + raise TemporalJourneyArtifactError("relation count is outside the supported bound") + parsed: list[TemporalRelation] = [] + previous: tuple[str, str] | None = None + for item in raw_relations: + if not isinstance(item, dict) or set(item) != { + "left_event_id", "right_event_id", "allen_relations", "observed", + "support_assertion_ordinals", + }: + raise TemporalJourneyArtifactError("relation object shape is unsupported") + left, right = item["left_event_id"], item["right_event_id"] + relations, support = item["allen_relations"], item["support_assertion_ordinals"] + key = (left, right) if isinstance(left, str) and isinstance(right, str) else ("", "") + if ( + not key[0].strip() or not key[1].strip() or key[0] == key[1] + or previous is not None and previous >= key + or not isinstance(item["observed"], bool) + or not isinstance(relations, list) or not relations + or any(relation not in ALLEN_RELATIONS for relation in relations) + or relations != sorted(set(relations), key=ALLEN_RELATIONS.index) + or len(relations) == len(ALLEN_RELATIONS) + or not isinstance(support, list) or not support + or any(isinstance(ordinal, bool) or not isinstance(ordinal, int) or ordinal < 0 for ordinal in support) + or support != sorted(set(support)) + ): + raise TemporalJourneyArtifactError("relation value is invalid or noncanonical") + parsed.append(TemporalRelation(key[0], key[1], tuple(relations), item["observed"], tuple(support))) + previous = key + return TemporalJourneyArtifact( + expected_run_id, + expected_snapshot_id, + expected_input_digest_sha256, + tuple(parsed), + expected_artifact_digest_sha256, + ) diff --git a/migrations/0259_project_journey_temporal_artifact.sql b/migrations/0259_project_journey_temporal_artifact.sql new file mode 100644 index 000000000..4529be923 --- /dev/null +++ b/migrations/0259_project_journey_temporal_artifact.sql @@ -0,0 +1,52 @@ +-- Digest-bound temporal evidence admitted only for existing Event Lineage edges. +create table if not exists project_journey_temporal_artifact ( + analysis_run_id uuid primary key references analysis_run_tepp_result(analysis_run_id) on delete cascade, + remote_run_id text not null, + schema_version text not null check (schema_version = 'tepp.tdt_chronos_interval_consistency.v1'), + snapshot_id text not null check (btrim(snapshot_id) <> ''), + input_digest_sha256 text not null check (input_digest_sha256 ~ '^[0-9a-f]{64}$'), + artifact_digest_sha256 text not null unique check (artifact_digest_sha256 ~ '^[0-9a-f]{64}$'), + admitted_at timestamptz not null default clock_timestamp(), + unique (analysis_run_id, remote_run_id) +); + +create table if not exists project_journey_temporal_relation ( + analysis_run_id uuid not null references project_journey_temporal_artifact(analysis_run_id) on delete cascade, + left_post_id uuid not null references source_post(post_id) on delete cascade, + right_post_id uuid not null references source_post(post_id) on delete cascade, + observed boolean not null, + primary key (analysis_run_id, left_post_id, right_post_id), + foreign key (left_post_id, right_post_id) + references post_lineage_edge(parent_post_id, child_post_id) on delete cascade, + check (left_post_id <> right_post_id) +); + +create table if not exists project_journey_temporal_relation_kind ( + analysis_run_id uuid not null, + left_post_id uuid not null, + right_post_id uuid not null, + relation_code text not null check (relation_code in ( + 'before', 'after', 'meets', 'met_by', 'overlaps', 'overlapped_by', + 'starts', 'started_by', 'during', 'contains', 'finishes', 'finished_by', 'equals' + )), + relation_ordinal smallint not null check (relation_ordinal between 0 and 12), + primary key (analysis_run_id, left_post_id, right_post_id, relation_code), + unique (analysis_run_id, left_post_id, right_post_id, relation_ordinal), + foreign key (analysis_run_id, left_post_id, right_post_id) + references project_journey_temporal_relation(analysis_run_id, left_post_id, right_post_id) + on delete cascade +); + +create table if not exists project_journey_temporal_support ( + analysis_run_id uuid not null, + left_post_id uuid not null, + right_post_id uuid not null, + assertion_ordinal integer not null check (assertion_ordinal >= 0), + primary key (analysis_run_id, left_post_id, right_post_id, assertion_ordinal), + foreign key (analysis_run_id, left_post_id, right_post_id) + references project_journey_temporal_relation(analysis_run_id, left_post_id, right_post_id) + on delete cascade +); + +create index if not exists project_journey_temporal_relation_right_idx + on project_journey_temporal_relation (right_post_id, left_post_id, analysis_run_id); diff --git a/tests/test_project_history.py b/tests/test_project_history.py index 8eedec3fe..8b716fd1d 100644 --- a/tests/test_project_history.py +++ b/tests/test_project_history.py @@ -57,7 +57,14 @@ def test_prior_paths_keep_the_first_deterministic_shortest_route_per_predecessor paths = _prior_paths( ["award", "spec-a", "spec-b", "delivery"], [ - {"parent_post_id": "award", "child_post_id": "spec-a", "fused_score": 0.9}, + { + "parent_post_id": "award", + "child_post_id": "spec-a", + "fused_score": 0.9, + "temporal_observed": True, + "allen_relations": ["before"], + "artifact_digest_sha256": "a" * 64, + }, {"parent_post_id": "award", "child_post_id": "spec-b", "fused_score": 0.8}, {"parent_post_id": "spec-a", "child_post_id": "delivery", "fused_score": 0.7}, {"parent_post_id": "spec-b", "child_post_id": "delivery", "fused_score": 0.6}, @@ -68,3 +75,9 @@ def test_prior_paths_keep_the_first_deterministic_shortest_route_per_predecessor award_paths = [path for path in paths["delivery"] if path["source_event_id"] == "award"] assert [path["event_ids"] for path in award_paths] == [["award", "spec-a", "delivery"]] + assert award_paths[0]["edges"][0]["temporal_evidence"] == { + "truth_status_code": "observed", + "interval_relations": ["before"], + "artifact_digest_sha256": "a" * 64, + } + assert award_paths[0]["edges"][1]["temporal_evidence"] is None diff --git a/tests/test_project_history_ingestion.py b/tests/test_project_history_ingestion.py index c4886becf..adc7e86f9 100644 --- a/tests/test_project_history_ingestion.py +++ b/tests/test_project_history_ingestion.py @@ -57,6 +57,9 @@ def test_project_history_query_binds_corporate_and_process_scopes() -> None: assert result["events"][0]["event_type_code"] == "source_recorded" assert result["events"][0]["occurred_at"] == "2025-12-20T00:00:00Z" assert result["events"][0]["time_basis_code"] == "document_time" + edge_query = next(query for query, _args in connection.calls if "from post_lineage_edge" in query) + assert "project_journey_temporal_relation" in edge_query + assert "project_journey_temporal_relation_kind" in edge_query def test_project_history_query_uses_the_same_ascii_edge_whitespace_as_python() -> None: diff --git a/tests/test_temporal_journey_artifact.py b/tests/test_temporal_journey_artifact.py new file mode 100644 index 000000000..3bd08779b --- /dev/null +++ b/tests/test_temporal_journey_artifact.py @@ -0,0 +1,220 @@ +"""Typed temporal-artifact admission tests.""" + +from __future__ import annotations + +import asyncio +import hashlib +import json + +import pytest + +from backend.app.project_journey_temporal import ( + TemporalArtifactAdmissionError, + persist_project_journey_temporal_artifact, +) +from lineageweave.temporal_journey_artifact import ( + TemporalJourneyArtifactError, + parse_temporal_journey_artifact, +) + + +def _payload(*, run_id: str = "remote-1") -> bytes: + return json.dumps( + { + "schema_version": "tepp.tdt_chronos_interval_consistency.v1", + "run_id": run_id, + "snapshot_id": "snapshot-1", + "input_digest_sha256": "a" * 64, + "relations": [{ + "left_event_id": "00000000-0000-0000-0000-000000000001", + "right_event_id": "00000000-0000-0000-0000-000000000002", + "allen_relations": ["before", "meets"], + "observed": False, + "support_assertion_ordinals": [0, 2], + }], + }, + separators=(",", ":"), + ).encode() + + +def _parse(payload: bytes): + return parse_temporal_journey_artifact( + payload, + expected_run_id="remote-1", + expected_snapshot_id="snapshot-1", + expected_input_digest_sha256="a" * 64, + expected_artifact_digest_sha256=hashlib.sha256(payload).hexdigest(), + ) + + +def test_parser_binds_canonical_bytes_and_all_identities() -> None: + """The admitted DTO retains no unbound provider field.""" + + result = _parse(_payload()) + assert result.relations[0].allen_relations == ("before", "meets") + assert result.relations[0].support_assertion_ordinals == (0, 2) + + +@pytest.mark.parametrize("mutation", ["digest", "run", "unknown", "order"]) +def test_parser_rejects_changed_or_noncanonical_artifacts(mutation: str) -> None: + """Malformed, moved, or noncanonical payloads fail closed.""" + + payload = _payload(run_id="other" if mutation == "run" else "remote-1") + if mutation == "unknown": + value = json.loads(payload) + value["extra"] = True + payload = json.dumps(value, separators=(",", ":")).encode() + if mutation == "order": + value = json.loads(payload) + value["relations"][0]["allen_relations"] = ["meets", "before"] + payload = json.dumps(value, separators=(",", ":")).encode() + digest = "b" * 64 if mutation == "digest" else hashlib.sha256(payload).hexdigest() + with pytest.raises(TemporalJourneyArtifactError): + parse_temporal_journey_artifact( + payload, + expected_run_id="remote-1", + expected_snapshot_id="snapshot-1", + expected_input_digest_sha256="a" * 64, + expected_artifact_digest_sha256=digest, + ) + + +@pytest.mark.parametrize( + ("payload", "input_digest", "artifact_digest"), + [ + (b"", "a" * 64, "0" * 64), + (b"{}", "bad", hashlib.sha256(b"{}").hexdigest()), + (b"\xff", "a" * 64, hashlib.sha256(b"\xff").hexdigest()), + (b" {\"x\":1}", "a" * 64, hashlib.sha256(b" {\"x\":1}").hexdigest()), + (b"[]", "a" * 64, hashlib.sha256(b"[]").hexdigest()), + ], +) +def test_parser_rejects_size_digest_encoding_and_top_level_shape( + payload: bytes, input_digest: str, artifact_digest: str +) -> None: + """Every outer wire boundary rejects before relation persistence.""" + + with pytest.raises(TemporalJourneyArtifactError): + parse_temporal_journey_artifact( + payload, + expected_run_id="remote-1", + expected_snapshot_id="snapshot-1", + expected_input_digest_sha256=input_digest, + expected_artifact_digest_sha256=artifact_digest, + ) + + +def test_parser_rejects_empty_and_malformed_relation_collections() -> None: + """An empty result or untyped relation is not journey evidence.""" + + for relations in ([], ["not-an-object"]): + value = json.loads(_payload()) + value["relations"] = relations + payload = json.dumps(value, separators=(",", ":")).encode() + with pytest.raises(TemporalJourneyArtifactError): + _parse(payload) + + +class _Connection: + """Capture the normalized producer statements.""" + + def __init__( + self, + remote_run_id: str = "remote-1", + existing_digest: str | None = None, + ) -> None: + self.remote_run_id = remote_run_id + self.existing_digest = existing_digest + self.execute_calls: list[tuple[str, tuple[object, ...]]] = [] + self.many_calls: list[tuple[str, list[tuple[object, ...]]]] = [] + + async def fetchrow(self, query: str, *args: object): + """Return the terminal binding and no prior artifact.""" + + if "analysis_run_tepp_result" in query: + return {"remote_run_id": self.remote_run_id} + return ( + {"artifact_digest_sha256": self.existing_digest} + if self.existing_digest is not None + else None + ) + + async def execute(self, query: str, *args: object): + """Capture artifact metadata persistence.""" + + self.execute_calls.append((query, args)) + + async def executemany(self, query: str, args: list[tuple[object, ...]]): + """Capture normalized relation children.""" + + self.many_calls.append((query, args)) + + +def test_producer_persists_relation_kinds_and_support_separately() -> None: + """One accepted artifact produces normalized, auditable rows.""" + + payload = _payload() + connection = _Connection() + asyncio.run( + persist_project_journey_temporal_artifact( + connection, + analysis_run_id="00000000-0000-0000-0000-000000000010", + payload=payload, + expected_run_id="remote-1", + expected_snapshot_id="snapshot-1", + expected_input_digest_sha256="a" * 64, + expected_artifact_digest_sha256=hashlib.sha256(payload).hexdigest(), + ) + ) + assert len(connection.execute_calls) == 1 + assert [len(rows) for _query, rows in connection.many_calls] == [1, 2, 2] + + +def test_producer_rejects_a_terminal_run_mismatch() -> None: + """A valid artifact cannot be attached to another persisted run.""" + + payload = _payload() + with pytest.raises(TemporalArtifactAdmissionError): + asyncio.run( + persist_project_journey_temporal_artifact( + _Connection("different"), + analysis_run_id="00000000-0000-0000-0000-000000000010", + payload=payload, + expected_run_id="remote-1", + expected_snapshot_id="snapshot-1", + expected_input_digest_sha256="a" * 64, + expected_artifact_digest_sha256=hashlib.sha256(payload).hexdigest(), + ) + ) + + +def test_producer_is_idempotent_and_rejects_changed_artifact() -> None: + """A run may replay identical bytes but cannot change immutable evidence.""" + + payload = _payload() + digest = hashlib.sha256(payload).hexdigest() + same = _Connection(existing_digest=digest) + asyncio.run( + persist_project_journey_temporal_artifact( + same, + analysis_run_id="00000000-0000-0000-0000-000000000010", + payload=payload, + expected_run_id="remote-1", + expected_snapshot_id="snapshot-1", + expected_input_digest_sha256="a" * 64, + expected_artifact_digest_sha256=digest, + ) + ) + assert same.execute_calls == [] + with pytest.raises(TemporalArtifactAdmissionError): + asyncio.run( + persist_project_journey_temporal_artifact( + _Connection(existing_digest="b" * 64), + analysis_run_id="00000000-0000-0000-0000-000000000010", + payload=payload, + expected_run_id="remote-1", + expected_snapshot_id="snapshot-1", + expected_input_digest_sha256="a" * 64, + expected_artifact_digest_sha256=digest, + ) + ) From f0509945192913f6d9d8b88f13704092275ed108 Mon Sep 17 00:00:00 2001 From: Codex Date: Fri, 28 Aug 2026 20:54:09 +0900 Subject: [PATCH 2/2] fix(journey): bind temporal evidence to view cutoff --- backend/app/project_history.py | 7 ++++++- .../0270-digest-bound-project-journey-temporal-evidence.md | 6 +++--- tests/test_project_history_ingestion.py | 6 +++++- 3 files changed, 14 insertions(+), 5 deletions(-) diff --git a/backend/app/project_history.py b/backend/app/project_history.py index 2cad58e64..d203fb8c4 100644 --- a/backend/app/project_history.py +++ b/backend/app/project_history.py @@ -145,12 +145,15 @@ async def fetch(self, query: str, *args: object) -> Sequence[Mapping[str, Any]]: from project_journey_temporal_relation relation join project_journey_temporal_artifact artifact on artifact.analysis_run_id = relation.analysis_run_id + join analysis_run temporal_run + on temporal_run.analysis_run_id = artifact.analysis_run_id join project_journey_temporal_relation_kind kind on kind.analysis_run_id = relation.analysis_run_id and kind.left_post_id = relation.left_post_id and kind.right_post_id = relation.right_post_id where relation.left_post_id = edge.parent_post_id and relation.right_post_id = edge.child_post_id + and temporal_run.knowledge_cutoff <= $2 group by relation.observed, artifact.artifact_digest_sha256, artifact.admitted_at order by artifact.admitted_at desc, artifact.artifact_digest_sha256 desc limit 1 @@ -239,6 +242,7 @@ async def fetch_project_history_projection( conn, visible_ids=visible_ids, normalized_key=normalized_key, + knowledge_cutoff=knowledge_cutoff, ) return build_project_history_projection( project_key=project_key, @@ -257,10 +261,11 @@ async def _fetch_project_children( *, visible_ids: Sequence[str], normalized_key: str, + knowledge_cutoff: datetime, ) -> tuple[list[Mapping[str, Any]], list[Mapping[str, Any]], list[Mapping[str, Any]]]: """Fetch only child evidence whose endpoints are already authorized.""" matches = list(await conn.fetch(_MATCH_SQL, list(visible_ids), normalized_key)) roles = list(await conn.fetch(_ROLE_SQL, list(visible_ids))) - edges = list(await conn.fetch(_EDGE_SQL, list(visible_ids))) + edges = list(await conn.fetch(_EDGE_SQL, list(visible_ids), knowledge_cutoff)) return matches, roles, edges diff --git a/docs/adr/0270-digest-bound-project-journey-temporal-evidence.md b/docs/adr/0270-digest-bound-project-journey-temporal-evidence.md index a95ecf03a..70548e445 100644 --- a/docs/adr/0270-digest-bound-project-journey-temporal-evidence.md +++ b/docs/adr/0270-digest-bound-project-journey-temporal-evidence.md @@ -33,9 +33,9 @@ A branch is visible only when the independently admitted lineage graph already contains that topology. A transition still requires its separately governed observed business or responsibility evidence. -The Project History API attaches the newest immutable temporal evidence to -the corresponding visible edge after ABAC and cutoff filtering select both -endpoints. The customer UI says what the user can do next—open the supporting +The Project History API attaches the newest immutable temporal evidence whose +analysis cutoff does not exceed the requested view cutoff to the corresponding +visible edge after ABAC selects both endpoints. The customer UI says what the user can do next—open the supporting records and compare dates—and never names the calculation module. GraphML is an equivalent provider export, not the ingestion authority. The diff --git a/tests/test_project_history_ingestion.py b/tests/test_project_history_ingestion.py index adc7e86f9..c07a595b8 100644 --- a/tests/test_project_history_ingestion.py +++ b/tests/test_project_history_ingestion.py @@ -57,9 +57,13 @@ def test_project_history_query_binds_corporate_and_process_scopes() -> None: assert result["events"][0]["event_type_code"] == "source_recorded" assert result["events"][0]["occurred_at"] == "2025-12-20T00:00:00Z" assert result["events"][0]["time_basis_code"] == "document_time" - edge_query = next(query for query, _args in connection.calls if "from post_lineage_edge" in query) + edge_query, edge_args = next( + (query, args) for query, args in connection.calls if "from post_lineage_edge" in query + ) assert "project_journey_temporal_relation" in edge_query assert "project_journey_temporal_relation_kind" in edge_query + assert "temporal_run.knowledge_cutoff <= $2" in edge_query + assert edge_args[1] == datetime(2026, 2, 1, tzinfo=timezone.utc) def test_project_history_query_uses_the_same_ascii_edge_whitespace_as_python() -> None: