diff --git a/CHANGELOG.d/2.20.0-global-ask-cutoff-safety.md b/CHANGELOG.d/2.20.0-global-ask-cutoff-safety.md new file mode 100644 index 000000000..77d039572 --- /dev/null +++ b/CHANGELOG.d/2.20.0-global-ask-cutoff-safety.md @@ -0,0 +1,5 @@ +### Fixed + +- Bind the Global Ask knowledge cutoff in the final authorized-source query. +- Give the post-chat cutoff migration a unique `0054` identity and remove + self-modifying stabilization workflows. diff --git a/CHANGELOG.md b/CHANGELOG.md index 5f6086552..bce904fea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,29 @@ All notable changes to this project are documented here. Format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); versioning follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [2.20.0] - 2026-08-21 + +### Added + +- Post-scoped Ask and Global Ask now attach exact project-history links derived + only from currently authorized cited posts. Opening a link reuses the canonical + Project history timeline and its optional TEPP validation at the answer cutoff. + +### Security + +- Persisted post answers are withheld when any citation is no longer visible, and + stale Global Ask sessions are restarted before hidden prior prose can re-enter + conversation context (ADR 0113). + +## [2.19.0] - 2026-08-21 + +### Added + +- Recovered the credential-free TEPP project-history validation boundary on top of + the canonical Buyer timeline. TEPP may return only cutoff-safe temporal + associations over the exact authorized events; the timeline remains readable + when TEPP is absent, and no result is labelled as a cause (ADR 0127). + ## [2.18.0] - 2026-08-20 ### Added diff --git a/backend/app/analysis_run_start.py b/backend/app/analysis_run_start.py index 324c6cd67..fd5be4912 100644 --- a/backend/app/analysis_run_start.py +++ b/backend/app/analysis_run_start.py @@ -11,7 +11,7 @@ import hashlib import json -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import Any from uuid import UUID @@ -21,13 +21,13 @@ AnalysisRunCreateError, fetch_visible_analysis_run, ) -from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL from backend.app.analysis_run_outbox import ( latest_outbox_delivery_is_claimed, latest_outbox_delivery_is_delivered, outbox_request_digest, ) from backend.app.lineage_ingestion import records_from_source_posts +from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL from lineageweave.adjudication_client import AdjudicationClient from lineageweave.http_client import HttpClientError, post_json from lineageweave.lineage_persistence import lineage_edge_specs @@ -119,12 +119,12 @@ def tepp_run_request( """Build TEPP's published request from the frozen run, never a theta.""" cutoff = knowledge_cutoff if cutoff.tzinfo is None: - cutoff = cutoff.replace(tzinfo=timezone.utc) + cutoff = cutoff.replace(tzinfo=UTC) return AnalysisRunRequest( idempotency_key=idempotency_key, tenant_workspace_id=str(corporate_entity_id), snapshot_id=snapshot_sha256, - knowledge_cutoff=cutoff.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + knowledge_cutoff=cutoff.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ"), model_contract_version=_TEPP_MODEL_CONTRACT, output_profile=_TEPP_OUTPUT_PROFILE, ) @@ -610,7 +610,7 @@ async def deliver_queued_analysis_run( return await _visible_or_404( conn, analysis_run_id, account_id, affiliated_entity_ids ) - now = datetime.now(timezone.utc) + now = datetime.now(UTC) try: if not latest_outbox_delivery_is_claimed(latest): await _append_outbox_delivery( @@ -636,9 +636,8 @@ async def deliver_queued_analysis_run( affiliated_entity_ids=affiliated_entity_ids, adjudication_client=adjudication_client, ) - finished = datetime.now(timezone.utc) - if finished < now: - finished = now + finished = datetime.now(UTC) + finished = max(finished, now) await _append_outbox_delivery( conn, analysis_run_id, @@ -699,7 +698,7 @@ async def _deliver_lineage_reconstruction( adjudication_client: AdjudicationClient | None = None, ) -> None: """Persist ThreadWeave parent choices for the frozen bag.""" - now = datetime.now(timezone.utc) + now = datetime.now(UTC) member_rows = await _snapshot_member_posts( conn, locked["analysis_source_snapshot_id"], @@ -715,9 +714,8 @@ async def _deliver_lineage_reconstruction( ) edges = lineage_edge_specs(records_from_source_posts(rows), llm=adjudication_client) digest = reconstruction_result_digest(edges) - finished = datetime.now(timezone.utc) - if finished < now: - finished = now + finished = datetime.now(UTC) + finished = max(finished, now) await conn.execute( """ insert into analysis_run_reconstruction @@ -759,7 +757,7 @@ async def _deliver_tepp_measurement( tepp_client: TeppClient, ) -> None: """Submit the frozen snapshot through ``tepp_client``. Never persist a theta.""" - now = datetime.now(timezone.utc) + now = datetime.now(UTC) request = tepp_run_request( idempotency_key=str(locked["idempotency_key"]), snapshot_sha256=str(locked["snapshot_sha256"]), @@ -767,17 +765,15 @@ async def _deliver_tepp_measurement( corporate_entity_id=str(locked["corporate_entity_id"]), ) status_code, failure_code, envelope = _tepp_submission(tepp_client, request) - if status_code == _SUCCEEDED and envelope is not None: - if not await _persist_tepp_result( - conn, - analysis_run_id=analysis_run_id, - envelope=envelope, - ): - status_code = _FAILED - failure_code = "tepp_result_not_persisted" - finished = datetime.now(timezone.utc) - if finished < now: - finished = now + if status_code == _SUCCEEDED and envelope is not None and not await _persist_tepp_result( + conn, + analysis_run_id=analysis_run_id, + envelope=envelope, + ): + status_code = _FAILED + failure_code = "tepp_result_not_persisted" + finished = datetime.now(UTC) + finished = max(finished, now) await _append_status( conn, analysis_run_id, diff --git a/backend/app/ask_project_history.py b/backend/app/ask_project_history.py new file mode 100644 index 000000000..1bf97a57e --- /dev/null +++ b/backend/app/ask_project_history.py @@ -0,0 +1,310 @@ +"""Authorization-safe project-history links for Ask responses. + +The module accepts only citation identities already produced by post-scoped or +Global Ask. It re-applies current tenant visibility, source publication +eligibility, and the answer knowledge cutoff before returning citation labels or +project identities. A missing citation fails the whole persisted answer closed; +answer prose cannot be safely decomposed after one of its sources becomes +unauthorized. +""" + +from __future__ import annotations + +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Any, Protocol +from uuid import UUID + +from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL +from lineageweave.project_history import normalize_project_key + +ASK_CITATION_LIMIT = 64 +ASK_PROJECT_LIMIT = 8 +GLOBAL_ASK_SESSION_CITATION_LIMIT = 256 + +_ELIGIBILITY = SOURCE_POST_ELIGIBILITY_SQL.format(alias="post") +_CITATION_PROJECT_SQL = f""" +with visible_citation as materialized ( + select post.post_id::text as post_id, + post.post_title, + array_position($1::uuid[], post.post_id) as citation_ordinal, + nullif(btrim(post.source_project_code), '') as source_project_code, + nullif(btrim(post.source_project_name), '') as source_project_name + from source_post post + where post.post_id = any($1::uuid[]) + and (post.visibility_code = 'public' + or post.corporate_entity_id::text = any($2::text[])) + and post.created_at <= $3 + and {_ELIGIBILITY} +), project_evidence as ( + select visible_citation.post_id, + coalesce(visible_citation.source_project_code, + visible_citation.source_project_name) as project_key, + coalesce(visible_citation.source_project_name, + visible_citation.source_project_code) as project_name, + 'observed'::text as truth_status_code, + 0::integer as truth_order + from visible_citation + where coalesce(visible_citation.source_project_code, + visible_citation.source_project_name) is not null + union all + select visible_citation.post_id, + coalesce(nullif(btrim(mention.project_key), ''), + nullif(btrim(mention.project_name), '')) as project_key, + coalesce(nullif(btrim(mention.project_name), ''), + nullif(btrim(mention.project_key), '')) as project_name, + 'inferred'::text as truth_status_code, + 1::integer as truth_order + from visible_citation + join post_project_mention mention + on mention.post_id::text = visible_citation.post_id + where coalesce(nullif(btrim(mention.project_key), ''), + nullif(btrim(mention.project_name), '')) is not null +) +select visible_citation.post_id, + visible_citation.post_title, + visible_citation.citation_ordinal, + project_evidence.project_key, + project_evidence.project_name, + project_evidence.truth_status_code, + project_evidence.truth_order + from visible_citation + left join project_evidence + on project_evidence.post_id = visible_citation.post_id + order by visible_citation.citation_ordinal, + project_evidence.truth_order nulls last, + project_evidence.project_name nulls last, + project_evidence.project_key nulls last +""" +_SESSION_CITATION_SQL = """ +select distinct cited_post_id::text as cited_post_id + from global_ask_turn_citation + where global_ask_session_id = $1 + order by cited_post_id::text + limit $2 +""" + + +class AskEvidenceConnection(Protocol): + """Minimal async query port used by this read projection.""" + + async def fetch(self, query: str, *args: object) -> Sequence[Mapping[str, Any]]: + """Execute a bounded read query.""" + + raise NotImplementedError + + +@dataclass(frozen=True) +class AskEvidenceProjection: + """Currently authorized citation labels and exact project links.""" + + all_citations_visible: bool + cited_posts: tuple[dict[str, str], ...] + project_histories: tuple[dict[str, Any], ...] + project_histories_truncated: bool + knowledge_cutoff: str + + def response_fields(self) -> dict[str, Any]: + """Return the public response fields shared by both Ask surfaces.""" + + return { + "cited_posts": list(self.cited_posts), + "project_histories": list(self.project_histories), + "project_histories_truncated": self.project_histories_truncated, + "knowledge_cutoff": self.knowledge_cutoff, + } + + +def ask_knowledge_cutoff(value: object | None = None) -> datetime: + """Return an offset-aware UTC cutoff from a datetime or ISO text.""" + + if value is None: + return datetime.now(UTC) + if isinstance(value, datetime): + parsed = value + elif isinstance(value, str) and value.strip(): + try: + normalized = value.strip() + if normalized.endswith("Z"): + normalized = f"{normalized[:-1]}+00:00" + parsed = datetime.fromisoformat(normalized) + except ValueError as exc: + raise ValueError("knowledge cutoff must be ISO-8601") from exc + else: + raise ValueError("knowledge cutoff must be a datetime or ISO-8601 text") + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise ValueError("knowledge cutoff must include an offset") + return parsed.astimezone(UTC) + + +def _cutoff_text(value: datetime) -> str: + """Serialize one validated cutoff as canonical UTC RFC 3339 text.""" + + return value.astimezone(UTC).isoformat().replace("+00:00", "Z") + + +def _bounded_citations( + cited_post_ids: Iterable[str], *, maximum_citations: int +) -> tuple[str, ...]: + """Return unique citation IDs without silently truncating evidence.""" + + try: + citations = tuple( + dict.fromkeys( + str(UUID(str(value))) for value in cited_post_ids if str(value).strip() + ) + ) + except (AttributeError, TypeError, ValueError) as exc: + raise ValueError("citation identities must be UUIDs") from exc + if len(citations) > maximum_citations: + raise ValueError("citation count exceeds the supported bound") + return citations + + +async def read_authorized_ask_evidence( + conn: AskEvidenceConnection, + *, + cited_post_ids: Iterable[str], + corporate_entity_ids: Iterable[str], + knowledge_cutoff: datetime | str, + maximum_citations: int = ASK_CITATION_LIMIT, + maximum_projects: int = ASK_PROJECT_LIMIT, +) -> AskEvidenceProjection: + """Reauthorize citations and derive bounded exact-project history links. + + A citation is visible only when its current source row passes tenant ABAC, + publication eligibility, and the answer cutoff. If any citation is absent, + project links are withheld and callers must not reuse the persisted answer. + """ + + cutoff = ask_knowledge_cutoff(knowledge_cutoff) + cutoff_text = _cutoff_text(cutoff) + citations = _bounded_citations( + cited_post_ids, + maximum_citations=maximum_citations, + ) + if not citations: + return AskEvidenceProjection(True, (), (), False, cutoff_text) + rows = list( + await conn.fetch( + _CITATION_PROJECT_SQL, + list(citations), + list(corporate_entity_ids), + cutoff, + ) + ) + citation_order = {post_id: index for index, post_id in enumerate(citations, start=1)} + visible_titles: dict[str, str] = {} + for row in rows: + post_id = str(row["post_id"]) + if post_id in citation_order: + visible_titles.setdefault(post_id, str(row["post_title"])) + all_visible = set(visible_titles) == set(citations) + cited_posts = tuple( + {"post_id": post_id, "post_title": visible_titles[post_id]} + for post_id in citations + if post_id in visible_titles + ) + if not all_visible: + return AskEvidenceProjection(False, cited_posts, (), False, cutoff_text) + + evidence_rows = sorted( + ( + row + for row in rows + if row.get("project_key") is not None and row.get("project_name") is not None + ), + key=lambda row: ( + citation_order[str(row["post_id"])], + int(row.get("truth_order") or 0), + str(row["project_name"]), + str(row["project_key"]), + ), + ) + grouped: dict[str, dict[str, Any]] = {} + for row in evidence_rows: + project_key = str(row["project_key"]).strip() + project_name = str(row["project_name"]).strip() + try: + normalized_key = normalize_project_key(project_key) + except ValueError: + continue + post_id = str(row["post_id"]) + truth_order = int(row.get("truth_order") or 0) + group = grouped.get(normalized_key) + if group is None: + grouped[normalized_key] = { + "project_key": project_key, + "project_name": project_name, + "focus_post_id": post_id, + "source_post_ids": [post_id], + "knowledge_cutoff": cutoff_text, + "truth_status_code": str(row["truth_status_code"]), + "truth_order": truth_order, + "first_citation_ordinal": citation_order[post_id], + } + continue + if post_id not in group["source_post_ids"]: + group["source_post_ids"].append(post_id) + if truth_order < group["truth_order"]: + group["project_key"] = project_key + group["project_name"] = project_name + group["truth_status_code"] = str(row["truth_status_code"]) + group["truth_order"] = truth_order + + ordered = sorted( + grouped.values(), + key=lambda group: ( + int(group["first_citation_ordinal"]), + str(group["project_name"]), + str(group["project_key"]), + ), + ) + truncated = len(ordered) > maximum_projects + public_links: list[dict[str, Any]] = [] + for group in ordered[:maximum_projects]: + public_links.append( + { + key: value + for key, value in group.items() + if key not in {"truth_order", "first_citation_ordinal"} + } + ) + return AskEvidenceProjection( + True, + cited_posts, + tuple(public_links), + truncated, + cutoff_text, + ) + + +async def global_ask_session_citations_authorized( + conn: AskEvidenceConnection, + *, + session_id: str, + corporate_entity_ids: Iterable[str], + knowledge_cutoff: datetime | str, +) -> bool: + """Return whether every citation ever reused by a session is still visible.""" + + rows = list( + await conn.fetch( + _SESSION_CITATION_SQL, + session_id, + GLOBAL_ASK_SESSION_CITATION_LIMIT + 1, + ) + ) + if len(rows) > GLOBAL_ASK_SESSION_CITATION_LIMIT: + return False + citations = [str(row["cited_post_id"]) for row in rows] + result = await read_authorized_ask_evidence( + conn, + cited_post_ids=citations, + corporate_entity_ids=corporate_entity_ids, + knowledge_cutoff=knowledge_cutoff, + maximum_citations=GLOBAL_ASK_SESSION_CITATION_LIMIT, + maximum_projects=0, + ) + return result.all_citations_visible diff --git a/backend/app/main.py b/backend/app/main.py index 7f699da16..06f6d2457 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -186,6 +186,11 @@ persist_post_summary, require_summary_source_body, ) +from backend.app.ask_project_history import ( + ask_knowledge_cutoff, + global_ask_session_citations_authorized, + read_authorized_ask_evidence, +) from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL from backend.app.project_history import ( PROJECT_HISTORY_DEFAULT_LIMIT, @@ -196,6 +201,10 @@ fetch_project_history_index, fetch_project_history_projection, ) +from backend.app.tepp_project_history import ( + tenant_workspace_reference, + validate_project_history_with_tepp, +) from lineageweave.project_history import normalize_project_key from backend.app.demo_scope import ( fetch_demo_corporate_entity_ids, @@ -2683,9 +2692,25 @@ async def read_post_chat( an empty list, not a fabricated transcript. """ await _load_visible_post(post_id, account, pool) + authorized_exchanges: list[dict[str, Any]] = [] async with pool.acquire() as conn: exchanges = await fetch_persisted_chats(conn, post_id) - return {"post_id": post_id, "exchanges": exchanges} + for exchange in exchanges: + cutoff = ask_knowledge_cutoff(exchange.get("_knowledge_cutoff")) + evidence = await read_authorized_ask_evidence( + conn, + cited_post_ids=exchange["cited_post_ids"], + corporate_entity_ids=account.corporate_entity_ids, + knowledge_cutoff=cutoff, + ) + if not evidence.all_citations_visible: + continue + public_exchange = { + key: value for key, value in exchange.items() if not key.startswith("_") + } + public_exchange.update(evidence.response_fields()) + authorized_exchanges.append(public_exchange) + return {"post_id": post_id, "exchanges": authorized_exchanges} @app.post("/api/posts/{post_id}/chat") @@ -2711,18 +2736,28 @@ async def chat_about_post( raise HTTPException(status.HTTP_422_UNPROCESSABLE_ENTITY, "question is required") post = await _load_visible_post(post_id, account, pool) post_metadata = build_post_llm_metadata(post_id, post) + knowledge_cutoff = ask_knowledge_cutoff() async with pool.acquire() as conn: stored = await fetch_persisted_chat(conn, post_id, question) if stored is not None: - source_ids = [post_id] - source_ids.extend(cid for cid in stored["cited_post_ids"] if cid != post_id) - return { - "post_id": post_id, - "answer_text": stored["answer_text"], - "cited_post_ids": stored["cited_post_ids"], - "cited_posts": stored["cited_posts"], - "source_post_ids": source_ids, - } + stored_cutoff = ask_knowledge_cutoff(stored.get("_knowledge_cutoff")) + stored_evidence = await read_authorized_ask_evidence( + conn, + cited_post_ids=stored["cited_post_ids"], + corporate_entity_ids=account.corporate_entity_ids, + knowledge_cutoff=stored_cutoff, + ) + if stored_evidence.all_citations_visible: + source_ids = list( + dict.fromkeys([post_id, *stored["cited_post_ids"]]) + ) + return { + "post_id": post_id, + "answer_text": stored["answer_text"], + "cited_post_ids": stored["cited_post_ids"], + "source_post_ids": source_ids, + **stored_evidence.response_fields(), + } with use_llm_metadata(post_metadata): client = _post_chat_client() if not client.available: @@ -2731,7 +2766,11 @@ async def chat_about_post( "Post chat is unavailable: set ORCHESTRATOR_BASE_URL / ORCHESTRATOR_API_KEY", ) sources = await gather_chat_sources( - conn, post_id, lambda row: _can_see_post(account, row), vision_client=_vision_client() + conn, + post_id, + lambda row: _can_see_post(account, row), + vision_client=_vision_client(), + knowledge_cutoff=knowledge_cutoff, ) try: with use_llm_metadata(post_metadata): @@ -2748,7 +2787,25 @@ async def chat_about_post( ) from exc cited_ids = list(answer.cited_post_ids) async with pool.acquire() as conn: - await persist_post_chat(conn, post_id, question, answer.answer_text, cited_ids) + await persist_post_chat( + conn, + post_id, + question, + answer.answer_text, + cited_ids, + knowledge_cutoff=knowledge_cutoff, + ) + answer_evidence = await read_authorized_ask_evidence( + conn, + cited_post_ids=cited_ids, + corporate_entity_ids=account.corporate_entity_ids, + knowledge_cutoff=knowledge_cutoff, + ) + if not answer_evidence.all_citations_visible: + raise HTTPException( + status.HTTP_503_SERVICE_UNAVAILABLE, + "Post chat evidence changed before the answer could be returned", + ) await publish_activity_event( valkey, post_id, @@ -2760,8 +2817,8 @@ async def chat_about_post( "post_id": post_id, "answer_text": answer.answer_text, "cited_post_ids": cited_ids, - "cited_posts": cited_post_summaries(sources, cited_ids), "source_post_ids": [source.post_id for source in sources], + **answer_evidence.response_fields(), } @@ -2782,6 +2839,7 @@ async def ask_agent( UUID(request.session_id) except ValueError: raise HTTPException(status.HTTP_404_NOT_FOUND, "Global Ask session not found") from None + knowledge_cutoff = ask_knowledge_cutoff() client = _post_chat_client() if not client.available: raise HTTPException( @@ -2794,12 +2852,23 @@ async def ask_agent( ) if session_id is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "Global Ask session not found") + if not await global_ask_session_citations_authorized( + conn, + session_id=session_id, + corporate_entity_ids=account.corporate_entity_ids, + knowledge_cutoff=knowledge_cutoff, + ): + raise HTTPException( + status.HTTP_409_CONFLICT, + "Global Ask session evidence is no longer authorized; start a new session", + ) conversation = await load_global_ask_context(conn, session_id) sources = await gather_global_chat_sources( conn, lambda row: _can_see_post(account, row), account.corporate_entity_ids, question=question, + knowledge_cutoff=knowledge_cutoff, ) if conversation.compress_turns: compressor = getattr(client, "compress_context", None) @@ -2827,6 +2896,11 @@ async def ask_agent( status.HTTP_503_SERVICE_UNAVAILABLE, "Ask Agent conversation context compression is unavailable", ) from exc + except Exception as exc: + raise HTTPException( + status.HTTP_503_SERVICE_UNAVAILABLE, + "Ask Agent conversation context compression is unavailable", + ) from exc conversation_context = render_global_ask_context( conversation.summary, conversation.recent_turns, @@ -2848,6 +2922,9 @@ async def ask_agent( "source_post_ids": [], "cited_post_evidence": [], "timeline": [], + "project_histories": [], + "project_histories_truncated": False, + "knowledge_cutoff": knowledge_cutoff.isoformat().replace("+00:00", "Z"), "next_action": "No authorized source posts are available for this question.", } try: @@ -2876,6 +2953,17 @@ async def ask_agent( answer.answer_text, cited_ids, ) + answer_evidence = await read_authorized_ask_evidence( + conn, + cited_post_ids=cited_ids, + corporate_entity_ids=account.corporate_entity_ids, + knowledge_cutoff=knowledge_cutoff, + ) + if not answer_evidence.all_citations_visible: + raise HTTPException( + status.HTTP_503_SERVICE_UNAVAILABLE, + "Global Ask evidence changed before the answer could be returned", + ) await publish_operation_event( valkey, account.user_account_id, @@ -2886,10 +2974,10 @@ async def ask_agent( "session_id": conversation.session_id, "answer_text": answer.answer_text, "cited_post_ids": cited_ids, - "cited_posts": cited_post_summaries(sources, cited_ids), "cited_post_evidence": cited_post_evidence(sources, cited_ids), "source_post_ids": [source.post_id for source in sources], "timeline": global_ask_timeline(sources), + **answer_evidence.response_fields(), } @@ -3439,7 +3527,7 @@ async def read_project_history( ) from exc async with pool.acquire() as conn: try: - return await fetch_project_history_projection( + projection = await fetch_project_history_projection( conn, project_key=project_key, focus_post_id=focus_post_id, @@ -3449,6 +3537,13 @@ async def read_project_history( ) except ProjectHistoryNotFound as exc: raise HTTPException(status.HTTP_404_NOT_FOUND, "project history not found") from exc + projection["tepp_validation"] = await asyncio.to_thread( + validate_project_history_with_tepp, + projection=projection, + tenant_workspace_id=tenant_workspace_reference(account.corporate_entity_ids), + transport_url=load_settings().tepp_transport_url, + ) + return projection @app.get("/api/rankings") diff --git a/backend/app/post_chat_ingestion.py b/backend/app/post_chat_ingestion.py index e60b00aa5..8daf01b1f 100644 --- a/backend/app/post_chat_ingestion.py +++ b/backend/app/post_chat_ingestion.py @@ -20,6 +20,7 @@ import asyncio import re from dataclasses import dataclass +from datetime import datetime, timezone from typing import Any, Callable, Iterable from uuid import uuid4 @@ -44,6 +45,7 @@ from lineageweave.post_content_normalization import normalize_post_body from .knowledge_graph import hydrate_related_nodes, load_visible_subgraph +from .post_eligibility import SOURCE_POST_ELIGIBILITY_SQL from lineageweave.ontology import ontology_annotations @@ -340,6 +342,16 @@ async def _graph_facts_for_posts( _GLOBAL_ASK_TERM_PATTERN = re.compile(r"[^\W_]+(?:-[^\W_]+)*", re.UNICODE) _POST_CHAT_SOURCE_LIMIT = 8 _POST_CHAT_CANDIDATE_LIMIT = 32 +_SOURCE_ELIGIBILITY = SOURCE_POST_ELIGIBILITY_SQL.format(alias="source_post") + + +def _ask_cutoff(value: datetime | None) -> datetime: + """Return an aware UTC cutoff for one Ask retrieval.""" + + cutoff = value or datetime.now(timezone.utc) + if cutoff.tzinfo is None or cutoff.utcoffset() is None: + raise ValueError("knowledge_cutoff must include an offset") + return cutoff.astimezone(timezone.utc) def _source_hint_facts(row: Any) -> tuple[str, ...]: @@ -453,6 +465,8 @@ async def gather_chat_sources( post_id: str, can_see_post: Callable[[asyncpg.Record], bool], vision_client: ImageContentClient | None = None, + *, + knowledge_cutoff: datetime | None = None, ) -> list[ChatSourceDocument]: """Post `post_id` plus a bounded, deterministic linked-source window. @@ -469,15 +483,18 @@ async def gather_chat_sources( """ if vision_client is None: vision_client = NullImageContentClient() + cutoff = _ask_cutoff(knowledge_cutoff) this_post = await conn.fetchrow( - "select post_id, post_title, post_body, source_system_code, source_record_key, " + "select post_id, post_title, post_body, created_at, source_system_code, source_record_key, " "source_author_code, source_author_name, source_company_code, source_company_name, " "source_process_unit_code, source_process_unit_name, " "source_sales_pool_code, source_sales_pool_name, " "source_customer_code, source_customer_name, source_project_code, " - "source_project_name from source_post where post_id = $1", + f"source_project_name from source_post where post_id = $1 " + f"and created_at <= $2 and {_SOURCE_ELIGIBILITY}", post_id, + cutoff, ) if this_post is None: return [] @@ -510,11 +527,13 @@ async def gather_chat_sources( "source_company_code, source_company_name, source_process_unit_code, " "source_process_unit_name, source_sales_pool_code, source_sales_pool_name, " "source_customer_code, source_customer_name, " - "source_project_code, source_project_name " - "from source_post where post_id = any($1::uuid[]) " + "source_project_code, source_project_name, created_at " + f"from source_post where post_id = any($1::uuid[]) " + f"and created_at <= $3 and {_SOURCE_ELIGIBILITY} " "order by array_position($1::uuid[], post_id) limit $2", candidate_ids, _POST_CHAT_CANDIDATE_LIMIT, + cutoff, ) visible_source_ids = [post_id] visible_rows: list[asyncpg.Record] = [] @@ -558,6 +577,7 @@ async def gather_global_chat_sources( *, question: str | None = None, limit: int = 4, + knowledge_cutoff: datetime | None = None, ) -> list[ChatSourceDocument]: """Assemble a bounded, ABAC-filtered source set for Global Ask. @@ -569,6 +589,8 @@ async def gather_global_chat_sources( return [] if vision_client is None: vision_client = NullImageContentClient() + cutoff = _ask_cutoff(knowledge_cutoff) + authorized_entity_ids = list(authorized_corporate_entity_ids) search_terms = tuple( dict.fromkeys( token.casefold() @@ -612,29 +634,45 @@ async def gather_global_chat_sources( candidate_scores: dict[str, float] = {} for term in search_terms: candidate_rows = await conn.fetch( - """ + f""" select post_id, matched_in from ( (select post_id, created_at, 'title' as matched_in from source_post - where post_title ilike '%' || $1 || '%' + where (visibility_code = 'public' + or corporate_entity_id::text = any($2::text[])) + and created_at <= $3 + and {_SOURCE_ELIGIBILITY} + and post_title ilike '%' || $1 || '%' limit 32) union all (select post_id, created_at, 'body' as matched_in from source_post - where lower(left(source_post_search_text(post_body), 16384)) + where (visibility_code = 'public' + or corporate_entity_id::text = any($2::text[])) + and created_at <= $3 + and {_SOURCE_ELIGIBILITY} + and lower(left(source_post_search_text(post_body), 16384)) like '%' || lower($1) || '%' limit 32) union all (select post_id, created_at, 'body' as matched_in from source_post - where to_tsvector('simple', source_post_search_text(post_body)) + where (visibility_code = 'public' + or corporate_entity_id::text = any($2::text[])) + and created_at <= $3 + and {_SOURCE_ELIGIBILITY} + and to_tsvector('simple', source_post_search_text(post_body)) @@ plainto_tsquery('simple', $1) limit 32) union all (select post_id, created_at, 'source_field' as matched_in from source_post - where concat_ws(' ', source_system_code, source_record_key, + where (visibility_code = 'public' + or corporate_entity_id::text = any($2::text[])) + and created_at <= $3 + and {_SOURCE_ELIGIBILITY} + and concat_ws(' ', source_system_code, source_record_key, source_author_code, source_author_name, source_company_code, source_company_name, source_process_unit_code, source_process_unit_name, @@ -648,6 +686,8 @@ async def gather_global_chat_sources( limit 32 """, term, + authorized_entity_ids, + cutoff, ) for row in candidate_rows: post_id = str(row["post_id"]) @@ -686,7 +726,7 @@ async def gather_global_chat_sources( lineage_neighbor_id_set = frozenset(lineage_neighbor_ids) rows = await conn.fetch( - """ + f""" select post_id, post_title, post_body, visibility_code, corporate_entity_id, created_at, source_system_code, source_record_key, source_author_code, source_author_name, @@ -695,15 +735,18 @@ async def gather_global_chat_sources( source_customer_code, source_customer_name, source_project_code, source_project_name from source_post - where visibility_code = 'public' - or corporate_entity_id::text = any($1::text[]) + where (visibility_code = 'public' + or corporate_entity_id::text = any($1::text[])) + and created_at <= $4 + and {_SOURCE_ELIGIBILITY} order by array_position($2::uuid[], post_id) nulls last, created_at desc, post_id desc limit $3 """, - list(authorized_corporate_entity_ids), + authorized_entity_ids, candidate_ids, limit, + cutoff, ) visible_rows = [row for row in rows if can_see_post(row)][:limit] visible_ids = [str(row["post_id"]) for row in visible_rows] @@ -759,7 +802,7 @@ async def _serialize_chat( ) -> dict[str, Any] | None: """One stored exchange plus citation chips, or None when missing.""" header = await conn.fetchrow( - "select question_text, answer_text from post_chat_result " + "select question_text, answer_text, knowledge_cutoff from post_chat_result " "where post_id = $1 and question_norm = $2", post_id, question_norm, @@ -779,6 +822,7 @@ async def _serialize_chat( "question_text": header["question_text"], "answer_text": header["answer_text"], "cited_post_ids": cited_ids, + "_knowledge_cutoff": header.get("knowledge_cutoff"), "cited_posts": [ {"post_id": str(row["cited_post_id"]), "post_title": row["post_title"]} for row in cites @@ -816,23 +860,30 @@ async def persist_post_chat( question: str, answer_text: str, cited_post_ids: list[str] | tuple[str, ...], + *, + knowledge_cutoff: datetime | None = None, ) -> dict[str, Any]: """Replace the stored exchange for ``(post_id, question)`` and return it.""" norm = normalize_chat_question(question) if not norm: raise ValueError("question is empty after normalize") + cutoff = _ask_cutoff(knowledge_cutoff) + computed_at = max(datetime.now(timezone.utc), cutoff) await conn.execute( "delete from post_chat_result where post_id = $1 and question_norm = $2", post_id, norm, ) await conn.execute( - "insert into post_chat_result (post_id, question_norm, question_text, answer_text) " - "values ($1, $2, $3, $4)", + "insert into post_chat_result " + "(post_id, question_norm, question_text, answer_text, computed_at, knowledge_cutoff) " + "values ($1, $2, $3, $4, $5, $6)", post_id, norm, question.strip(), answer_text, + computed_at, + cutoff, ) seen: set[str] = set() ordinal = 0 diff --git a/backend/app/tepp_project_history.py b/backend/app/tepp_project_history.py new file mode 100644 index 000000000..e7951b520 --- /dev/null +++ b/backend/app/tepp_project_history.py @@ -0,0 +1,208 @@ +"""Map the canonical Buyer project history into TEPP's strict wire contract.""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Iterable, Mapping, Sequence +from typing import Any + +from lineageweave.tepp_project_history import ( + PROJECT_HISTORY_CONTRACT_VERSION, + TeppProjectHistoryClient, + TeppProjectHistoryInvalidResponse, + TeppProjectHistoryUnavailable, + parse_rfc3339_utc, + project_history_event_sort_key, + validate_tepp_project_history_request, +) + + +def tenant_workspace_reference(corporate_entity_ids: Iterable[str]) -> str: + """Return a deterministic opaque workspace reference for the ABAC scope.""" + + normalized = sorted({str(value).strip() for value in corporate_entity_ids if str(value).strip()}) + material = "\u001f".join(normalized) if normalized else "public-only" + digest = hashlib.sha256(material.encode("utf-8")).hexdigest() + return f"lw-workspace-{digest}" + + +def _utc_text(value: object, field_name: str) -> str: + """Return canonical UTC text from one offset-aware source timestamp.""" + + return parse_rfc3339_utc(value, field_name)[1] + + +def _opaque_actor_ids( + event: Mapping[str, Any], + *, + tenant_workspace_id: str, +) -> list[str]: + """Hash canonical actor keys so names and local identifiers do not cross.""" + + raw_roles = event.get("responsibility_evidence") + if raw_roles is None: + raw_roles = event.get("observed_responsibilities") + if not isinstance(raw_roles, Sequence) or isinstance(raw_roles, (str, bytes)): + raw_roles = () + actor_ids: set[str] = set() + for role in raw_roles: + if not isinstance(role, Mapping): + continue + actor_key = str(role.get("actor_key") or "").strip() + if not actor_key: + continue + material = f"{tenant_workspace_id}\u0000{actor_key}".encode("utf-8") + actor_ids.add(f"lw-actor-{hashlib.sha256(material).hexdigest()}") + return sorted(actor_ids) + + +def _evidence_text(event: Mapping[str, Any]) -> str: + """Build bounded source-field evidence without sending a post body.""" + + title = str(event.get("event_title") or "").strip() + event_type = str(event.get("event_type_code") or "").strip() + if not title or not event_type: + raise TeppProjectHistoryUnavailable("canonical event title and type are required") + parts = [title, f"event_type={event_type}"] + for key in ("source_stage_code", "source_detail_state_code", "voc_type_code"): + value = str(event.get(key) or "").strip() + if value: + parts.append(f"{key}={value}") + rendered = " | ".join(parts) + encoded = rendered.encode("utf-8") + if len(encoded) <= 4096: + return rendered + return encoded[:4096].decode("utf-8", errors="ignore").rstrip() + + +def _idempotency_key(request_without_key: Mapping[str, Any]) -> str: + """Hash the exact authorized evidence bundle into a stable request key.""" + + material = json.dumps( + request_without_key, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + digest = hashlib.sha256(material.encode("utf-8")).hexdigest() + return f"lineageweave-project-history-{digest}" + + +def build_tepp_project_history_request( + *, + projection: Mapping[str, Any], + tenant_workspace_id: str, +) -> dict[str, Any]: + """Build TEPP #159 input from the already-authorized canonical timeline.""" + + if projection.get("contract_version") != 1: + raise TeppProjectHistoryUnavailable("unsupported canonical project-history version") + events_value = projection.get("events") + if not isinstance(events_value, Sequence) or isinstance(events_value, (str, bytes)): + raise TeppProjectHistoryUnavailable("canonical project history has no event list") + cutoff = _utc_text(projection.get("knowledge_cutoff"), "knowledge_cutoff") + events: list[dict[str, Any]] = [] + for value in events_value: + if not isinstance(value, Mapping): + raise TeppProjectHistoryUnavailable("canonical project event must be an object") + occurred_at = _utc_text(value.get("occurred_at"), "occurred_at") + event_id = str(value.get("event_id") or "").strip() + source_post_id = str(value.get("source_post_id") or "").strip() + if not event_id or not source_post_id: + raise TeppProjectHistoryUnavailable("canonical project event identity is missing") + events.append( + { + "event_id": event_id, + "event_type_code": str(value.get("event_type_code") or "").strip(), + "event_title": str(value.get("event_title") or "").strip(), + "occurred_at": occurred_at, + # The canonical timeline explicitly declares source-post creation + # time as its fallback clock. It is therefore also the earliest + # evidence-availability instant LineageWeave can substantiate. + "available_at": occurred_at, + "source_post_id": source_post_id, + "evidence_text": _evidence_text(value), + "actor_ids": _opaque_actor_ids( + value, + tenant_workspace_id=tenant_workspace_id, + ), + } + ) + events.sort(key=project_history_event_sort_key) + request: dict[str, Any] = { + "contract_version": PROJECT_HISTORY_CONTRACT_VERSION, + "tenant_workspace_id": tenant_workspace_id, + "project_key": str(projection.get("project_key") or "").strip(), + "project_name": str(projection.get("project_name") or "").strip(), + "knowledge_cutoff": cutoff, + "focus_event_id": str(projection.get("focus_event_id") or "").strip(), + "events": events, + } + request["idempotency_key"] = _idempotency_key(request) + return validate_tepp_project_history_request(request) + + +def _buyer_metadata(projection: Mapping[str, Any]) -> dict[str, Any]: + """Strip duplicate event rows while preserving TEPP findings and evidence IDs.""" + + events = projection["events"] + return { + "contract_version": projection["contract_version"], + "project_key": projection["project_key"], + "project_name": projection["project_name"], + "focus_event_id": projection["focus_event_id"], + "knowledge_cutoff": projection["knowledge_cutoff"], + "history_span_start": projection["history_span_start"], + "history_span_end": projection["history_span_end"], + "participant_count": projection["participant_count"], + "inference_status": projection["inference_status"], + "event_count": len(events), + "findings": projection["findings"], + } + + +def validate_project_history_with_tepp( + *, + projection: Mapping[str, Any], + tenant_workspace_id: str, + transport_url: str, +) -> dict[str, Any]: + """Return optional TEPP metadata without hiding the canonical timeline.""" + + if not transport_url.strip(): + return { + "status": "not_configured", + "project_history": None, + "next_action_code": "configure_tepp_project_history", + } + try: + request = build_tepp_project_history_request( + projection=projection, + tenant_workspace_id=tenant_workspace_id, + ) + except TeppProjectHistoryUnavailable: + return { + "status": "invalid_evidence", + "project_history": None, + "next_action_code": "open_source_evidence", + } + try: + validated = TeppProjectHistoryClient(transport_url).project(request) + except TeppProjectHistoryInvalidResponse: + return { + "status": "invalid_evidence", + "project_history": None, + "next_action_code": "open_source_evidence", + } + except TeppProjectHistoryUnavailable: + return { + "status": "unavailable", + "project_history": None, + "next_action_code": "retry_tepp_project_history", + } + return { + "status": "validated", + "project_history": _buyer_metadata(validated), + "next_action_code": "open_source_evidence", + } diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index 5cc185750..12d60ca98 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -110,6 +110,11 @@ / "migrations" / "0052_global_ask_context.sql" ) +_POST_CHAT_CUTOFF_MIGRATION = ( + Path(__file__).resolve().parents[2] + / "migrations" + / "0054_post_chat_knowledge_cutoff.sql" +) _MAJOR_EVENT_ACTION_MIGRATION = ( Path(__file__).resolve().parents[2] / "migrations" / "0100_major_event_action.sql" ) @@ -235,6 +240,7 @@ def seeded_db(demo_analyst_token): cur.execute(_POST_CONTENT_QUEUE_MIGRATION.read_text()) cur.execute(_ORGANIZATION_CONTEXT_MIGRATION.read_text()) cur.execute(_GLOBAL_ASK_CONTEXT_MIGRATION.read_text()) + cur.execute(_POST_CHAT_CUTOFF_MIGRATION.read_text()) cur.execute(_MAJOR_EVENT_ACTION_MIGRATION.read_text()) cur.execute(_PROJECT_BOUND_ACTION_MIGRATION.read_text()) cur.execute(_PROJECT_BOUND_EVENT_MIGRATION.read_text()) diff --git a/docker/postgres-init/migrate.sh b/docker/postgres-init/migrate.sh index f442f628a..712107b27 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_*|0053_*) ;; + 0051_*|0052_*|0053_*|0054_*) ;; 0060_*|0100_*|0101_*|0102_*) ;; *) continue ;; esac diff --git a/docs/adr/0113-project-history-links-in-ask-surfaces.md b/docs/adr/0113-project-history-links-in-ask-surfaces.md new file mode 100644 index 000000000..de521c1c0 --- /dev/null +++ b/docs/adr/0113-project-history-links-in-ask-surfaces.md @@ -0,0 +1,56 @@ +# ADR 0113: Reuse canonical project history in Ask surfaces + +- Status: Proposed +- Date: 2026-08-21 +- Depends on: ADR 0112, ADR 0127, and the canonical Project history read model + +## Context + +Post-scoped Ask and Global Ask already cite authorized source posts, but they did not +connect those citations to the project lifecycle timeline shown in the product design. +The earlier orphaned stack attempted to solve this with another project-history flow. +That would create competing project identity, authorization, cutoff, classification, and +TEPP behavior. + +Persisted Ask prose introduces an additional security boundary: if a previously cited +post becomes hidden, deleted, draft, or otherwise ineligible, returning the old answer or +reusing it as conversation context can disclose facts no longer authorized. + +## Decision + +1. Ask responses expose structured project-history links derived only from cited post IDs. +2. Citation IDs are reauthorized with tenant ABAC, source publication eligibility, and the + answer knowledge cutoff before titles or project identities are returned. +3. Exact source project fields outrank semantic project candidates; inferred identities + remain labelled inferred. Links are bounded and deterministic. +4. Opening a link calls the canonical Project history endpoint with project key, answer + cutoff, and cited focus post. The established timeline and TEPP metadata are reused. +5. A persisted post answer is withheld in full when any citation is no longer authorized. + Its prose cannot be safely decomposed by source after access changes. +6. A Global Ask session is rejected and restarted when any citation in its persisted + continuity context is no longer authorized. Stored summaries are not reused across + that boundary. +7. Ask retrieval itself applies the same cutoff and source eligibility before an LLM sees + evidence. Prompt bodies, hidden IDs, and unauthorized project counts never enter the + project-history link response. +8. Timeline or TEPP failure does not remove the answer; the Buyer receives an actionable + error and can still open the exact cited source post. + +## Consequences + +- Document reading, post Ask, Global Ask, and the dedicated Project history destination + share one authorization-first read model and one timeline component. +- Historical answers can disappear after permission or publication changes. This is an + intentional fail-closed property, not data loss from the evidence store. +- A session restart can lose conversational convenience, but prevents a compressed + summary from carrying hidden prose forward. +- Event order remains a temporal association and is not presented as causal inference. + +## Rejected alternatives + +- Parse project identities from answer prose. This is nondeterministic and ungrounded. +- Build a second project query or timeline inside Ask. This duplicates authority. +- Return a stored answer while merely hiding its citation chips. The prose may still leak + the hidden source. +- Keep a stale Global Ask summary and filter only new citations. The summary cannot be + safely decomposed after authorization changes. diff --git a/docs/adr/0125-global-ask-cutoff-and-migration-identity.md b/docs/adr/0125-global-ask-cutoff-and-migration-identity.md new file mode 100644 index 000000000..48745b065 --- /dev/null +++ b/docs/adr/0125-global-ask-cutoff-and-migration-identity.md @@ -0,0 +1,40 @@ +# ADR 0125 — Bind Global Ask cutoffs and keep migration identities unique + +**Decision status:** Accepted on the PR #342 repair branch +**Date:** 2026-08-21 +**Figma File ID:** N/A — this is a backend, migration, and operability decision. + +## Context + +Global Ask restricts source posts by the requested knowledge cutoff. Its final +PostgreSQL query used the `$4` cutoff placeholder but supplied only three +arguments, so a real PostgreSQL execution could fail before returning any +authorized evidence. The same branch also introduced a second forward +migration with numeric prefix `0053`, colliding with an existing migration. +Temporary self-modifying workflows were compensating for both defects after a +push rather than leaving the branch itself correct. + +## Decision + +1. Bind the cutoff as the fourth argument of the final Global Ask source query. +2. Assign the cutoff schema change the next unique forward migration identity, + `0054`, and update rollback, migration dispatch, and contract tests. +3. Keep reproduction and regression checks in committed tests. Do not use a + workflow that edits, commits, pushes, or deletes product source at runtime. + +## Consequences + +- Global Ask fails neither at PostgreSQL parameter binding nor by silently + dropping the requested knowledge cutoff. +- Migration replay and rollback address one numeric identity unambiguously. +- Hosted CI evaluates the exact committed source instead of a workflow-mutated + branch state. + +## Verification + +- The synthetic query contract asserts the fourth argument is the requested + cutoff. +- The PostgreSQL integration contract executes the final query against a real + local PostgreSQL parser when `LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN` is set. +- Migration identity tests reject duplicate numeric prefixes and require the + `0054_*` dispatch path. diff --git a/docs/adr/0127-recover-tepp-validation-on-canonical-project-history.md b/docs/adr/0127-recover-tepp-validation-on-canonical-project-history.md new file mode 100644 index 000000000..5a42d964e --- /dev/null +++ b/docs/adr/0127-recover-tepp-validation-on-canonical-project-history.md @@ -0,0 +1,98 @@ +# ADR 0127: Recover TEPP validation on the canonical project history + +- Status: Proposed +- Date: 2026-08-21 +- Depends on: LineageWeave Project history stack; `ContextualWisdomLab/TEPP#159` +- Supersedes: the duplicate project-history implementation carried by LineageWeave #281/#282 + +## Context + +A Buyer project-history timeline was implemented on a canonical, authorization-first +LineageWeave read model. An earlier TEPP integration was then left behind in a closed +parent PR and an open child PR whose branch reimplemented the project query, event +classification, and timeline. The user-supplied product screen requires one project +lifecycle timeline and an optional TEPP-linked answer, not two competing histories. + +The TEPP contract in PR #159 accepts only an exact project identity, a knowledge cutoff, +a focus event, and explicit source-grounded events. It may order those events and return +coded temporal-association findings. It does not accept or return a latent score, a +probability of causation, or an authoritative assignment record. + +## Decision + +1. LineageWeave remains authoritative for RBAC/ABAC, source eligibility, exact project + identity, event classification, visible responsibility evidence, and the Buyer + timeline. +2. The TEPP request is derived from that already-authorized canonical projection. No + second database query or second timeline component is allowed. +3. Source-post creation time is sent as both `occurred_at` and `available_at` only because + the canonical timeline explicitly declares it as the current fallback clock. The UI + continues to disclose that limitation. +4. Actor names and local actor keys do not cross the service boundary. TEPP receives a + deterministic opaque SHA-256 reference scoped to the authorized workspace. This is a + data-minimizing pseudonymous reference, not a claim of irreversible anonymization. +5. Evidence text is bounded and composed from the event title and persisted source-state + fields. Post bodies, browser tokens, review credentials, provider keys, and + `TEPP_API_KEY` are not forwarded. +6. The client requires the exact versioned field set, exact event cardinality and content, + deterministic chronological ordering, unchanged project/focus/cutoff identity, and + evidence-derived participant counts. Unknown fields, changed evidence, or a response above + TEPP's published 256 KiB contract limit fail closed before JSON decoding. +7. Accepted findings are limited to the six published TEPP #159 finding codes. Duplicate + event or evidence references are rejected. Buyer UI copy is owned by LineageWeave and + keyed by those codes; provider-authored summary prose is retained for contract + validation but is not rendered as the interpretation. +8. `temporal_association_only` is the only accepted inference status. Buyer copy states + that the result does not identify a cause. +9. A transport outage is distinct from an invalid response. `not_configured`, + `unavailable`, and `invalid_evidence` states leave the canonical timeline readable and + tell the operator or Buyer what to do next. +10. Global Ask and post-scoped Ask are a subsequent stacked slice and must reuse this same + canonical projection and TEPP envelope. +11. Any unexpected TEPP transport/provider exception is converted to the stable + `TEPP transport request failed` state. Raw response bodies and exception text remain + internal chained causes and never cross the public contract. + +## Consequences + +- The previously implemented capability is recovered without reviving the orphaned + duplicate stack. +- A TEPP outage cannot remove or alter authorized LineageWeave evidence. +- TEPP findings remain inspectable through exact source-post references. +- An unrecognized finding vocabulary cannot introduce provider-authored Buyer claims. +- The product does not answer “what caused the VOC?” as a causal claim. It answers which + explicit prior records are temporally associated and provides evidence for human review. +- A future distinct event-time or available-time source can replace the current fallback + only through a versioned contract and migration. + +## Rejected alternatives + +- **Merge the old #282 branch as-is.** It is based on a closed parent and carries a second + project-history implementation with a large unrelated ancestry. +- **Let TEPP query the LineageWeave database.** This breaks authorization ownership and + modular deployment. +- **Send full post bodies or actor names.** These are unnecessary for the published + temporal contract and expand the privacy boundary. +- **Render a separate TEPP timeline.** Duplicate timelines can disagree and obscure which + system owns evidence selection. +- **Render arbitrary TEPP summary prose.** The provider may validate time, but it does not + own Buyer-facing interpretation or an open-ended claim vocabulary. +- **Describe preceding events as causes.** Event order alone does not identify causality. + +## 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/ + +World Wide Web Consortium. (2017). *Time ontology in OWL*. +https://www.w3.org/TR/owl-time/ + +MITRE. (n.d.). *CWE-209: Generation of error message containing sensitive information*. +https://cwe.mitre.org/data/definitions/209.html + +National Institute of Standards and Technology. (2020). *Security and privacy controls +for information systems and organizations: NIST SP 800-53 Rev. 5*. +https://doi.org/10.6028/NIST.SP.800-53r5 diff --git a/docs/adr/0125-customer-master-three-pane-workspace.md b/docs/adr/0129-customer-master-three-pane-workspace.md similarity index 99% rename from docs/adr/0125-customer-master-three-pane-workspace.md rename to docs/adr/0129-customer-master-three-pane-workspace.md index b6ae71a74..3f83cd7fb 100644 --- a/docs/adr/0125-customer-master-three-pane-workspace.md +++ b/docs/adr/0129-customer-master-three-pane-workspace.md @@ -1,4 +1,4 @@ -# ADR 0125: Customer-centered three-pane Customer Master workspace +# ADR 0129: Customer-centered three-pane Customer Master workspace - **Status:** Accepted - **Date:** 2026-08-21 diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 8e29fd804..1f30d0d79 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -309,10 +309,45 @@ projection: update the affected FR/NFR row and Gap closure evidence when an ADR or PR changes product behavior. Never turn a PR title, green unit test, or old runtime note into a shipped/live claim. +## Recovered TEPP project-history integration (2026-08-21) + +- The canonical Buyer project timeline remains owned by the stacked Project history PR. +- The previously implemented TEPP work had become stranded in a closed parent and an + orphaned duplicate stack. This recovery consumes the canonical timeline instead of + introducing another project query, classifier, or timeline component. +- The dependency is the exact `ContextualWisdomLab/TEPP#159` project-history contract. + Until that contract is merged and a TEPP endpoint is deployed, the UI reports an + actionable fail-closed state and keeps the authorized LineageWeave timeline readable. +- TEPP receives opaque actor references and bounded source-field evidence only. Browser, + review, provider, and `TEPP_API_KEY` credentials are not forwarded. +- `temporal_association_only` is the maximum accepted authority. Buyer copy must say + that a preceding event is related in time, not that it caused the VOC. +- The next stacked slice attaches this same canonical timeline and TEPP metadata to + Global Ask and post-scoped Ask without re-retrieving hidden evidence. + +## Ask-to-project-history integration (2026-08-21) + +- Protected-stack checkpoint: PR #342 is based on PR #339 head + `43262dc76622928fdf90b922653949b4ac7c6631`; the PR description and hosted Checks + record its exact current head. Both remain review/check gated and are not represented + as merged production behavior. +- Post-scoped Ask and Global Ask return structured project-history links only for exact + project identities on their currently authorized cited posts. +- Opening a link lazily calls the canonical Project history endpoint with the answer + knowledge cutoff and cited focus post; no second timeline, classifier, or TEPP query is + implemented in either Ask surface. +- Source publication eligibility and cutoff are applied before Ask retrieval. Persisted + answers are withheld when any citation loses visibility, and a Global Ask session with + stale citations must start a new session before prior answer prose is reused. +- The response bounds citation and project counts, discloses truncated project links, and + keeps answers readable when a timeline or TEPP validation is unavailable. +- Remaining causal-analysis work is explicitly outside this slice: temporal association + and evidence navigation do not identify why a VOC occurred. + ## Current stacked PR product-surface gaps - **Customer Master relationship composition — PR #262**: Resolved on the - current feature branch. ADR 0125 and Figma frames `313:2` / `314:2` define a + current feature branch. ADR 0129 and Figma frames `313:2` / `314:2` define a customer-centered three-pane workspace that keeps the selected customer stable while the user inspects relationships and source posts. - **Responsive Customer Master flow — PR #262**: Resolved on the current diff --git a/frontend/package.json b/frontend/package.json index f216081b9..bd8c9ff59 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,7 +1,7 @@ { "name": "frontend", "private": true, - "version": "2.18.0", + "version": "2.20.0", "type": "module", "scripts": { "dev": "vite", diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index c2d5f4ee8..9bafee3b4 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -76,6 +76,7 @@ import { type PostLineage, type PostSummary, type PostSortOrder, + type ProjectHistoryLink, type RankingList, type PersonRoleHistoryEntry, type RelatedNode, @@ -92,6 +93,7 @@ import { LineageDag } from "./LineageDag"; import { PostBody } from "./PostBody"; import { decodeHtmlEntities } from "./postBodyDisplay"; import { FiveW1H } from "./components/FiveW1H"; +import { AskProjectHistoryLinks } from "./components/AskProjectHistoryLinks"; import { ProjectHistoryTimeline } from "./components/ProjectHistoryTimeline"; import { projectHistoryText, @@ -293,6 +295,9 @@ function ChatPanel({ answer_text: result.answer_text, cited_post_ids: result.cited_post_ids, cited_posts: result.cited_posts, + knowledge_cutoff: result.knowledge_cutoff, + project_histories: result.project_histories, + project_histories_truncated: result.project_histories_truncated, }; return [...prev.filter((row) => row.question_text !== next.question_text), next]; }); @@ -335,6 +340,12 @@ function ChatPanel({ exchanges[0].cited_posts?.[0]?.post_id ?? exchanges[0].cited_post_ids[0] } /> + ) : null} {nameFirstAsk && firstCitedTitle ? ( @@ -415,6 +426,12 @@ function ChatPanel({ citedPostIds={exchange.cited_post_ids} onOpenEvidence={setEvidencePostId} /> + ))} {answer && !exchanges.some((row) => row.answer_text === answer.answer_text) && ( @@ -425,6 +442,12 @@ function ChatPanel({ citedPostIds={answer.cited_post_ids} onOpenEvidence={setEvidencePostId} /> + )} {!nameFirstAsk && evidencePostId ? ( @@ -4587,6 +4610,12 @@ function AskAgentPanel({ window.sessionStorage.getItem(GLOBAL_ASK_SESSION_STORAGE_KEY) ?? undefined, ); + function acceptAnswer(nextAnswer: AskAgentResponse) { + setAnswer(nextAnswer); + setSessionId(nextAnswer.session_id); + window.sessionStorage.setItem("lineageweave.globalAskSessionId", nextAnswer.session_id); + } + async function handleAsk() { const normalized = question.trim(); if (!normalized) return; @@ -4605,11 +4634,19 @@ function AskAgentPanel({ window.sessionStorage.removeItem(GLOBAL_ASK_SESSION_STORAGE_KEY); nextAnswer = await askAgent(accessToken, normalized); } - setAnswer(nextAnswer); - setSessionId(nextAnswer.session_id); - window.sessionStorage.setItem(GLOBAL_ASK_SESSION_STORAGE_KEY, nextAnswer.session_id); + acceptAnswer(nextAnswer); } catch (err) { - setAnswer(null); + if (err instanceof BackendError && err.status === 409 && sessionId) { + window.sessionStorage.removeItem("lineageweave.globalAskSessionId"); + setSessionId(undefined); + try { + acceptAnswer(await askAgent(accessToken, normalized)); + return; + } catch (retryError) { + setError(orchestratorUnavailableMessage(retryError, t("Ask Agent"))); + return; + } + } setError(orchestratorUnavailableMessage(err, t("Ask Agent"))); } finally { setAsking(false); @@ -4659,6 +4696,12 @@ function AskAgentPanel({ ) : null} + {answer.cited_posts && answer.cited_posts.length > 0 && ( <>

diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 1d3bb5a88..838609094 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -282,12 +282,24 @@ export interface CitedPostEvidence { facts: CitedPostEvidenceFact[]; } +export interface ProjectHistoryLink { + project_key: string; + project_name: string; + focus_post_id: string; + source_post_ids: string[]; + knowledge_cutoff: string; + truth_status_code: "observed" | "inferred"; +} + export interface ChatAnswer { post_id: string; answer_text: string; cited_post_ids: string[]; cited_posts?: CitedPostRef[]; source_post_ids: string[]; + knowledge_cutoff?: string; + project_histories?: ProjectHistoryLink[]; + project_histories_truncated?: boolean; } export interface ChatExchange { @@ -295,6 +307,9 @@ export interface ChatExchange { answer_text: string; cited_post_ids: string[]; cited_posts?: CitedPostRef[]; + knowledge_cutoff?: string; + project_histories?: ProjectHistoryLink[]; + project_histories_truncated?: boolean; } export interface ChatHistory { @@ -310,6 +325,9 @@ export interface AskAgentResponse { cited_post_evidence?: CitedPostEvidence[]; source_post_ids: string[]; timeline?: AskTimelineEntry[]; + knowledge_cutoff?: string; + project_histories?: ProjectHistoryLink[]; + project_histories_truncated?: boolean; next_action?: string; } diff --git a/frontend/src/components/AskProjectHistoryLinks.css b/frontend/src/components/AskProjectHistoryLinks.css new file mode 100644 index 000000000..8a491419d --- /dev/null +++ b/frontend/src/components/AskProjectHistoryLinks.css @@ -0,0 +1,36 @@ +.ask-project-history-links { + display: grid; + gap: 0.75rem; + margin-top: 1rem; + padding-top: 1rem; + border-top: 1px solid var(--border-color, #d7dce5); +} + +.ask-project-history-links > h4, +.ask-project-history-link p { + margin: 0; +} + +.ask-project-history-link { + display: grid; + gap: 0.625rem; + padding: 0.75rem; + border: 1px solid var(--border-color, #d7dce5); + border-radius: 0.75rem; + background: var(--surface-color, #fff); +} + +.ask-project-history-link > div:first-child { + display: flex; + flex-wrap: wrap; + align-items: center; + gap: 0.5rem; +} + +.ask-project-history-link > button { + justify-self: start; +} + +.ask-project-history-link [hidden] { + display: none; +} diff --git a/frontend/src/components/AskProjectHistoryLinks.stories.tsx b/frontend/src/components/AskProjectHistoryLinks.stories.tsx new file mode 100644 index 000000000..b3b25fd16 --- /dev/null +++ b/frontend/src/components/AskProjectHistoryLinks.stories.tsx @@ -0,0 +1,44 @@ +import type { Meta, StoryObj } from "@storybook/react"; + +import { AskProjectHistoryLinks } from "./AskProjectHistoryLinks"; + +const meta = { + title: "Buyer/Ask Project History Links", + component: AskProjectHistoryLinks, + args: { + accessToken: "storybook-token", + links: [ + { + project_key: "P-100", + project_name: "Synthetic renewal", + focus_post_id: "post-voc", + source_post_ids: ["post-spec", "post-voc"], + knowledge_cutoff: "2026-08-20T12:00:00Z", + truth_status_code: "observed", + }, + ], + truncated: false, + onOpenPost: () => undefined, + }, +} satisfies Meta; + +export default meta; +type Story = StoryObj; + +export const ObservedProject: Story = {}; + +export const InferredAndTruncated: Story = { + args: { + links: [ + { + project_key: "semantic-project", + project_name: "Semantic project candidate", + focus_post_id: "post-candidate", + source_post_ids: ["post-candidate"], + knowledge_cutoff: "2026-08-20T12:00:00Z", + truth_status_code: "inferred", + }, + ], + truncated: true, + }, +}; diff --git a/frontend/src/components/AskProjectHistoryLinks.test.tsx b/frontend/src/components/AskProjectHistoryLinks.test.tsx new file mode 100644 index 000000000..16da5ce7c --- /dev/null +++ b/frontend/src/components/AskProjectHistoryLinks.test.tsx @@ -0,0 +1,123 @@ +import { fireEvent, render, screen, waitFor } from "@testing-library/react"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { fetchProjectHistory } from "../api"; +import type { ProjectHistoryProjection } from "../projectHistory"; +import { AskProjectHistoryLinks } from "./AskProjectHistoryLinks"; + +vi.mock("../api", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + fetchProjectHistory: vi.fn(), + }; +}); + +const projection: ProjectHistoryProjection = { + contract_version: 1, + project_key: "P-100", + normalized_project_key: "p-100", + project_name: "Synthetic renewal", + focus_event_id: "post-voc", + time_basis_code: "source_post_created_at_fallback", + knowledge_cutoff: "2026-08-20T12:00:00Z", + evidence_boundary_code: "authorized_visible_source_posts", + event_count: 1, + distinct_actor_count: 0, + distinct_observed_actor_count: 0, + truncated: false, + events: [ + { + event_id: "post-voc", + source_post_id: "post-voc", + event_title: "Synthetic VOC received", + event_type_code: "voc_received", + event_type_basis_code: "display_classification", + occurred_at: "2026-02-02T09:00:00Z", + time_basis_code: "source_post_created_at_fallback", + voc_type_code: "voc", + source_stage_code: null, + source_detail_state_code: null, + project_matches: [], + responsibility_evidence: [], + observed_responsibilities: [], + responsibility_transition_code: null, + responsibility_transition_truth_status_code: null, + related_prior_paths: [], + }, + ], +}; + +const link = { + project_key: "P-100", + project_name: "Synthetic renewal", + focus_post_id: "post-voc", + source_post_ids: ["post-voc"], + knowledge_cutoff: "2026-08-20T12:00:00Z", + truth_status_code: "observed" as const, +}; + +describe("AskProjectHistoryLinks", () => { + beforeEach(() => { + vi.mocked(fetchProjectHistory).mockReset(); + }); + + it("loads the canonical timeline at the answer cutoff and preserves source navigation", async () => { + const onOpenPost = vi.fn(); + vi.mocked(fetchProjectHistory).mockResolvedValue(projection); + + render( + , + ); + + fireEvent.click(screen.getByRole("button", { name: /open project history: Synthetic renewal/i })); + + await waitFor(() => { + expect(fetchProjectHistory).toHaveBeenCalledWith( + "token", + "P-100", + "2026-08-20T12:00:00Z", + "post-voc", + ); + }); + expect(screen.getByRole("heading", { name: /project event timeline/i })).toBeInTheDocument(); + + fireEvent.click(screen.getByRole("button", { name: /open source record: Synthetic VOC received/i })); + expect(onOpenPost).toHaveBeenCalledWith("post-voc"); + }); + + it("reports truncation and leaves the answer readable when the timeline fetch fails", async () => { + vi.mocked(fetchProjectHistory).mockRejectedValue(new Error("synthetic failure")); + + render( + , + ); + + expect(screen.getByRole("status")).toHaveTextContent(/additional cited projects are not shown/i); + fireEvent.click(screen.getByRole("button", { name: /open project history: Synthetic renewal/i })); + expect(await screen.findByRole("alert")).toHaveTextContent(/project history could not be loaded/i); + expect(screen.getByText("Synthetic renewal")).toBeInTheDocument(); + }); + + it("renders nothing when the answer cites no project identity", () => { + const { container } = render( + , + ); + expect(container).toBeEmptyDOMElement(); + }); +}); diff --git a/frontend/src/components/AskProjectHistoryLinks.tsx b/frontend/src/components/AskProjectHistoryLinks.tsx new file mode 100644 index 000000000..99d144b43 --- /dev/null +++ b/frontend/src/components/AskProjectHistoryLinks.tsx @@ -0,0 +1,177 @@ +import { useEffect, useId, useState } from "react"; + +import { fetchProjectHistory, type ProjectHistoryLink } from "../api"; +import type { Locale } from "../i18n"; +import { useLocale } from "../i18n"; +import { projectHistoryText, type ProjectHistoryProjection } from "../projectHistory"; +import { ProjectHistoryTimeline } from "./ProjectHistoryTimeline"; +import "./AskProjectHistoryLinks.css"; + +interface Copy { + heading: string; + boundary: string; + open: (name: string) => string; + close: (name: string) => string; + loading: string; + truncated: string; + observed: string; + inferred: string; +} + +const COPY: Record = { + en: { + heading: "Project histories cited by this answer", + boundary: "Each timeline is rebuilt from currently authorized evidence at the answer cutoff.", + open: (name) => `Open project history: ${name}`, + close: (name) => `Close project history: ${name}`, + loading: "Loading cited project history...", + truncated: "Additional cited projects are not shown. Open the cited source records to inspect their project evidence.", + observed: "Observed project identity", + inferred: "Inferred project identity", + }, + ko: { + heading: "이 답변이 인용한 프로젝트 이력", + boundary: "각 타임라인은 답변 기준 시각과 현재 권한을 통과한 근거로 다시 구성됩니다.", + open: (name) => `프로젝트 이력 열기: ${name}`, + close: (name) => `프로젝트 이력 닫기: ${name}`, + loading: "인용된 프로젝트 이력을 불러오는 중...", + truncated: "일부 추가 프로젝트는 표시하지 않습니다. 인용된 원천 기록에서 프로젝트 근거를 확인하세요.", + observed: "관찰된 프로젝트 식별자", + inferred: "추론된 프로젝트 식별자", + }, + zh: { + heading: "此回答引用的项目历史", + boundary: "每条时间线都根据回答截止时间和当前授权证据重新构建。", + open: (name) => `打开项目历史:${name}`, + close: (name) => `关闭项目历史:${name}`, + loading: "正在加载引用的项目历史...", + truncated: "还有引用项目未显示。请打开引用的源记录检查其项目依据。", + observed: "已观察的项目身份", + inferred: "已推断的项目身份", + }, + ja: { + heading: "この回答が引用したプロジェクト履歴", + boundary: "各タイムラインは回答時点と現在の権限を通過した根拠から再構成されます。", + open: (name) => `プロジェクト履歴を開く: ${name}`, + close: (name) => `プロジェクト履歴を閉じる: ${name}`, + loading: "引用されたプロジェクト履歴を読み込み中...", + truncated: "追加の引用プロジェクトは表示されていません。引用元レコードでプロジェクト根拠を確認してください。", + observed: "観察されたプロジェクト識別子", + inferred: "推論されたプロジェクト識別子", + }, + vi: { + heading: "Lịch sử dự án được câu trả lời này trích dẫn", + boundary: "Mỗi dòng thời gian được dựng lại từ bằng chứng hiện được cấp quyền tại thời điểm cắt của câu trả lời.", + open: (name) => `Mở lịch sử dự án: ${name}`, + close: (name) => `Đóng lịch sử dự án: ${name}`, + loading: "Đang tải lịch sử dự án được trích dẫn...", + truncated: "Một số dự án được trích dẫn chưa được hiển thị. Hãy mở bản ghi nguồn để kiểm tra bằng chứng dự án.", + observed: "Danh tính dự án được quan sát", + inferred: "Danh tính dự án được suy luận", + }, +}; + +function ProjectHistoryDisclosure({ + accessToken, + link, + onOpenPost, +}: { + accessToken: string; + link: ProjectHistoryLink; + onOpenPost: (postId: string) => void; +}) { + const locale = useLocale(); + const copy = COPY[locale]; + const regionId = useId(); + const [opened, setOpened] = useState(false); + const [loading, setLoading] = useState(false); + const [projection, setProjection] = useState(null); + const [error, setError] = useState(false); + + useEffect(() => { + setOpened(false); + setLoading(false); + setProjection(null); + setError(false); + }, [link.project_key, link.focus_post_id, link.knowledge_cutoff]); + + function toggle() { + if (opened) { + setOpened(false); + return; + } + setOpened(true); + if (projection || loading) return; + setLoading(true); + setError(false); + fetchProjectHistory( + accessToken, + link.project_key, + link.knowledge_cutoff, + link.focus_post_id, + ) + .then((result) => { + setProjection(result); + setLoading(false); + }) + .catch(() => { + setError(true); + setLoading(false); + }); + } + + return ( +

+
+ {link.project_name} + + {link.truth_status_code === "observed" ? copy.observed : copy.inferred} + +
+ + +
+ ); +} + +export function AskProjectHistoryLinks({ + accessToken, + links, + truncated, + onOpenPost, +}: { + accessToken: string; + links: ProjectHistoryLink[]; + truncated: boolean; + onOpenPost: (postId: string) => void; +}) { + const locale = useLocale(); + const copy = COPY[locale]; + const headingId = useId(); + if (links.length === 0 && !truncated) return null; + return ( +
+

{copy.heading}

+

{copy.boundary}

+ {links.map((link) => ( + + ))} + {truncated ?

{copy.truncated}

: null} +
+ ); +} diff --git a/frontend/src/components/ProjectHistoryTimeline.tepp.test.tsx b/frontend/src/components/ProjectHistoryTimeline.tepp.test.tsx new file mode 100644 index 000000000..e33ad1ff8 --- /dev/null +++ b/frontend/src/components/ProjectHistoryTimeline.tepp.test.tsx @@ -0,0 +1,67 @@ +import { render, screen } from "@testing-library/react"; +import { describe, expect, it, vi } from "vitest"; + +import type { ProjectHistoryProjection } from "../projectHistory"; +import { ProjectHistoryTimeline } from "./ProjectHistoryTimeline"; + +const projection = { + contract_version: 1, + project_key: "P-100", + normalized_project_key: "p-100", + project_name: "Synthetic transformer renewal", + focus_event_id: "voc", + time_basis_code: "source_post_created_at_fallback", + knowledge_cutoff: "2026-08-20T12:00:00Z", + evidence_boundary_code: "authorized_visible_source_posts", + event_count: 1, + distinct_actor_count: 0, + distinct_observed_actor_count: 0, + truncated: false, + tepp_validation: { + status: "validated", + next_action_code: "open_source_evidence", + project_history: { + contract_version: 1, + project_key: "P-100", + project_name: "Synthetic transformer renewal", + focus_event_id: "voc", + knowledge_cutoff: "2026-08-20T12:00:00Z", + history_span_start: "2026-02-02T09:00:00Z", + history_span_end: "2026-02-02T09:00:00Z", + participant_count: 0, + inference_status: "temporal_association_only", + event_count: 1, + findings: [], + }, + }, + events: [ + { + event_id: "voc", + source_post_id: "post-voc", + event_title: "Synthetic VOC received", + event_type_code: "voc_received", + event_type_basis_code: "display_classification", + occurred_at: "2026-02-02T09:00:00Z", + time_basis_code: "source_post_created_at_fallback", + voc_type_code: "voc", + source_stage_code: null, + source_detail_state_code: null, + project_matches: [], + responsibility_evidence: [], + observed_responsibilities: [], + responsibility_transition_code: null, + responsibility_transition_truth_status_code: null, + related_prior_paths: [], + }, + ], +} as ProjectHistoryProjection; + +describe("ProjectHistoryTimeline TEPP integration", () => { + it("renders TEPP validation on the canonical timeline instead of a duplicate timeline", () => { + render(); + + expect(screen.getByRole("heading", { name: /TEPP temporal validation/i })).toBeInTheDocument(); + expect(screen.getByRole("heading", { name: /Project event timeline/i })).toBeInTheDocument(); + expect(screen.getAllByRole("tab")).toHaveLength(1); + }); +}); diff --git a/frontend/src/components/ProjectHistoryTimeline.tsx b/frontend/src/components/ProjectHistoryTimeline.tsx index a7832716a..12c7a7854 100644 --- a/frontend/src/components/ProjectHistoryTimeline.tsx +++ b/frontend/src/components/ProjectHistoryTimeline.tsx @@ -8,6 +8,7 @@ import { projectHistoryText, projectHistoryTransitionLabel, } from "../projectHistory"; +import { TeppProjectHistoryEvidence } from "./TeppProjectHistoryEvidence"; import "./ProjectHistoryTimeline.css"; function formatDate(value: string): string { @@ -124,6 +125,16 @@ export function ProjectHistoryTimeline({

) : null} + {projection.tepp_validation ? ( + [event.source_post_id, event.event_title]), + )} + /> + ) : null} +
header { + display: flex; + align-items: flex-start; + justify-content: space-between; + gap: 1rem; +} + +.tepp-project-evidence h4, +.tepp-project-evidence h5, +.tepp-project-evidence p { + margin-top: 0; +} + +.tepp-project-evidence-boundary { + font-weight: 700; +} + +.tepp-project-evidence dl { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(12rem, 1fr)); + gap: 0.75rem; + margin: 1rem 0; +} + +.tepp-project-evidence dl > div { + padding: 0.75rem; + border-radius: 0.65rem; + background: color-mix(in srgb, var(--surface-color, #ffffff) 88%, transparent); +} + +.tepp-project-evidence dt { + font-size: 0.85rem; + font-weight: 700; +} + +.tepp-project-evidence dd { + margin: 0.3rem 0 0; +} + +.tepp-project-evidence ul { + display: grid; + gap: 0.75rem; + padding-left: 1.25rem; +} + +.tepp-project-evidence-links { + display: flex; + flex-wrap: wrap; + gap: 0.5rem; +} + +.tepp-project-evidence-links button { + min-height: 2.75rem; +} + +.tepp-project-evidence-status { + border-style: dashed; +} + +@media (max-width: 42rem) { + .tepp-project-evidence > header { + flex-direction: column; + } +} + +@media print { + .tepp-project-evidence-links button { + border: 0; + padding: 0; + background: none; + } +} diff --git a/frontend/src/components/TeppProjectHistoryEvidence.stories.tsx b/frontend/src/components/TeppProjectHistoryEvidence.stories.tsx new file mode 100644 index 000000000..0a3833ddb --- /dev/null +++ b/frontend/src/components/TeppProjectHistoryEvidence.stories.tsx @@ -0,0 +1,61 @@ +import type { Meta, StoryObj } from "@storybook/react"; + +import { TeppProjectHistoryEvidence } from "./TeppProjectHistoryEvidence"; + +const meta = { + title: "Buyer/TEPP Project History Evidence", + component: TeppProjectHistoryEvidence, + args: { + validation: { + status: "validated", + next_action_code: "open_source_evidence", + project_history: { + contract_version: 1, + project_key: "P-100", + project_name: "Synthetic transformer renewal", + focus_event_id: "voc", + knowledge_cutoff: "2026-08-20T12:00:00Z", + history_span_start: "2022-03-11T09:00:00Z", + history_span_end: "2026-02-02T09:00:00Z", + participant_count: 2, + inference_status: "temporal_association_only", + event_count: 3, + findings: [ + { + finding_code: "specification_change_before_focus", + summary: "An explicit specification-change event precedes the focus event.", + related_event_ids: ["spec"], + evidence_post_ids: ["post-spec"], + }, + ], + }, + }, + sourceLabels: { "post-spec": "Synthetic specification changed" }, + onOpenPost: () => undefined, + }, +} satisfies Meta; + +export default meta; +type Story = StoryObj; + +export const Validated: Story = {}; + +export const NotConfigured: Story = { + args: { + validation: { + status: "not_configured", + project_history: null, + next_action_code: "configure_tepp_project_history", + }, + }, +}; + +export const ServiceUnavailable: Story = { + args: { + validation: { + status: "unavailable", + project_history: null, + next_action_code: "retry_tepp_project_history", + }, + }, +}; diff --git a/frontend/src/components/TeppProjectHistoryEvidence.test.tsx b/frontend/src/components/TeppProjectHistoryEvidence.test.tsx new file mode 100644 index 000000000..21cc046f6 --- /dev/null +++ b/frontend/src/components/TeppProjectHistoryEvidence.test.tsx @@ -0,0 +1,113 @@ +import { fireEvent, render, screen } from "@testing-library/react"; +import { describe, expect, it, vi } from "vitest"; + +import type { TeppProjectHistoryValidation } from "../projectHistory"; +import { TeppProjectHistoryEvidence } from "./TeppProjectHistoryEvidence"; + +const validation: TeppProjectHistoryValidation = { + status: "validated", + next_action_code: "open_source_evidence", + project_history: { + contract_version: 1, + project_key: "P-100", + project_name: "Synthetic transformer renewal", + focus_event_id: "voc", + knowledge_cutoff: "2026-08-20T12:00:00Z", + history_span_start: "2022-03-11T09:00:00Z", + history_span_end: "2026-02-02T09:00:00Z", + participant_count: 2, + inference_status: "temporal_association_only", + event_count: 3, + findings: [ + { + finding_code: "specification_change_before_focus", + summary: "An explicit specification-change event precedes the focus event.", + related_event_ids: ["spec"], + evidence_post_ids: ["post-spec"], + }, + ], + }, +}; + +describe("TeppProjectHistoryEvidence", () => { + it("shows controlled TEPP copy and opens only supplied source evidence", () => { + const onOpenPost = vi.fn(); + render( + , + ); + + expect(screen.getByRole("heading", { name: /TEPP temporal validation/i })).toBeInTheDocument(); + expect(screen.getByText(/temporal association only/i)).toBeInTheDocument(); + expect(screen.getByText(/does not identify a cause/i)).toBeInTheDocument(); + expect(screen.getByText(/participants in supplied evidence/i)).toBeInTheDocument(); + expect(screen.getByText("2", { selector: "dd" })).toBeInTheDocument(); + expect( + screen.queryByText("An explicit specification-change event precedes the focus event."), + ).not.toBeInTheDocument(); + + fireEvent.click( + screen.getByRole("button", { name: /open evidence: Synthetic specification changed/i }), + ); + expect(onOpenPost).toHaveBeenCalledWith("post-spec"); + }); + + it("gives an actionable fail-closed state without inventing a result", () => { + render( + , + ); + + expect(screen.getByRole("status")).toHaveTextContent(/configure the TEPP project-history endpoint/i); + expect(screen.queryByText(/participants in supplied evidence/i)).not.toBeInTheDocument(); + }); + + it("uses unique labelled-region ids when more than one evidence panel is present", () => { + render( + <> + + + , + ); + + const headings = screen.getAllByRole("heading", { name: /TEPP temporal validation/i }); + const regions = headings.map((heading) => heading.closest("section")); + expect(headings[0].id).not.toBe(headings[1].id); + expect(regions[0]).toHaveAttribute("aria-labelledby", headings[0].id); + expect(regions[1]).toHaveAttribute("aria-labelledby", headings[1].id); + }); + + it("fails closed when a validated response has no metadata", () => { + render( + , + ); + + expect(screen.getByRole("status")).toHaveTextContent(/open the source evidence/i); + }); +}); diff --git a/frontend/src/components/TeppProjectHistoryEvidence.tsx b/frontend/src/components/TeppProjectHistoryEvidence.tsx new file mode 100644 index 000000000..7cdddbb1d --- /dev/null +++ b/frontend/src/components/TeppProjectHistoryEvidence.tsx @@ -0,0 +1,238 @@ +import { useId } from "react"; + +import type { Locale } from "../i18n"; +import { useLocale } from "../i18n"; +import type { + TeppProjectHistoryFindingCode, + TeppProjectHistoryValidation, +} from "../projectHistory"; +import "./TeppProjectHistoryEvidence.css"; + +interface Copy { + heading: string; + eyebrow: string; + boundary: string; + participants: string; + span: string; + findings: string; + noFindings: string; + openEvidence: (label: string) => string; + unnamedEvidence: (index: number) => string; + status: Record, string>; + findingLabels: Record; +} + +const COPY: Record = { + en: { + heading: "TEPP temporal validation", + eyebrow: "TEPP-connected evidence", + boundary: "Temporal association only; this does not identify a cause.", + participants: "Participants in supplied evidence", + span: "Validated history span", + findings: "TEPP findings", + noFindings: "TEPP ordered the explicit events and returned no additional finding.", + openEvidence: (label) => `Open evidence: ${label}`, + unnamedEvidence: (index) => `Evidence record ${index}`, + status: { + not_configured: "Configure the TEPP project-history endpoint, then retry this timeline.", + unavailable: "TEPP is unavailable. Read the canonical timeline now and retry validation later.", + invalid_evidence: "Open the source evidence and correct the project-history contract before retrying TEPP.", + }, + findingLabels: { + contract_award_before_focus: "A contract-award event precedes the selected event.", + specification_change_before_focus: "A specification-change event precedes the selected event.", + delivery_before_focus: "A delivery event precedes the selected event.", + handoff_before_focus: "A handoff record precedes the selected event.", + rebid_after_focus: "A rebid event follows the selected event.", + specification_change_and_handoff_before_focus: + "Specification-change and handoff records both precede the selected event.", + }, + }, + ko: { + heading: "TEPP 시간 검증", + eyebrow: "TEPP 연계 근거", + boundary: "시간적 연관만 제시하며 원인을 식별한 결과가 아닙니다.", + participants: "제공된 근거의 참여자", + span: "검증된 이력 구간", + findings: "TEPP 검토 결과", + noFindings: "TEPP가 명시적 이벤트를 정렬했으며 추가 검토 결과는 없습니다.", + openEvidence: (label) => `근거 열기: ${label}`, + unnamedEvidence: (index) => `근거 기록 ${index}`, + status: { + not_configured: "TEPP 프로젝트 이력 엔드포인트를 설정한 뒤 이 타임라인을 다시 검증하세요.", + unavailable: "TEPP를 사용할 수 없습니다. 현재는 기준 타임라인을 읽고 나중에 검증을 다시 실행하세요.", + invalid_evidence: "원천 근거를 열어 프로젝트 이력 계약을 바로잡은 뒤 TEPP를 다시 실행하세요.", + }, + findingLabels: { + contract_award_before_focus: "선택한 이벤트보다 앞선 수주 확정 기록이 있습니다.", + specification_change_before_focus: "선택한 이벤트보다 앞선 사양 변경 기록이 있습니다.", + delivery_before_focus: "선택한 이벤트보다 앞선 납품 기록이 있습니다.", + handoff_before_focus: "선택한 이벤트보다 앞선 인수인계 기록이 있습니다.", + rebid_after_focus: "선택한 이벤트 뒤에 재입찰 기록이 있습니다.", + specification_change_and_handoff_before_focus: + "선택한 이벤트보다 앞서 사양 변경과 인수인계 기록이 모두 있습니다.", + }, + }, + zh: { + heading: "TEPP 时间验证", + eyebrow: "TEPP 关联证据", + boundary: "仅表示时间关联,不等于识别了原因。", + participants: "所提供证据中的参与者", + span: "已验证的历史区间", + findings: "TEPP 结果", + noFindings: "TEPP 已对明确事件排序,未返回其他结果。", + openEvidence: (label) => `打开证据:${label}`, + unnamedEvidence: (index) => `证据记录 ${index}`, + status: { + not_configured: "请配置 TEPP 项目历史端点,然后重新验证此时间线。", + unavailable: "TEPP 当前不可用。请先阅读标准时间线,稍后重试验证。", + invalid_evidence: "请打开源证据并修正项目历史契约,然后重试 TEPP。", + }, + findingLabels: { + contract_award_before_focus: "合同授予记录早于所选事件。", + specification_change_before_focus: "规格变更记录早于所选事件。", + delivery_before_focus: "交付记录早于所选事件。", + handoff_before_focus: "交接记录早于所选事件。", + rebid_after_focus: "重新投标记录晚于所选事件。", + specification_change_and_handoff_before_focus: "规格变更和交接记录均早于所选事件。", + }, + }, + ja: { + heading: "TEPP 時間検証", + eyebrow: "TEPP 連携根拠", + boundary: "時間的関連のみを示し、原因を特定した結果ではありません。", + participants: "提供根拠の参加者", + span: "検証済み履歴期間", + findings: "TEPP の結果", + noFindings: "TEPP は明示的イベントを並べ替え、追加の結果は返しませんでした。", + openEvidence: (label) => `根拠を開く: ${label}`, + unnamedEvidence: (index) => `根拠記録 ${index}`, + status: { + not_configured: "TEPP プロジェクト履歴エンドポイントを設定し、このタイムラインを再検証してください。", + unavailable: "TEPP は利用できません。標準タイムラインを読み、後で検証を再試行してください。", + invalid_evidence: "原典根拠を開いてプロジェクト履歴契約を修正し、TEPP を再実行してください。", + }, + findingLabels: { + contract_award_before_focus: "選択イベントより前に受注確定記録があります。", + specification_change_before_focus: "選択イベントより前に仕様変更記録があります。", + delivery_before_focus: "選択イベントより前に納品記録があります。", + handoff_before_focus: "選択イベントより前に引継ぎ記録があります。", + rebid_after_focus: "選択イベントの後に再入札記録があります。", + specification_change_and_handoff_before_focus: "仕様変更と引継ぎの記録が選択イベントより前にあります。", + }, + }, + vi: { + heading: "Xác thực thời gian TEPP", + eyebrow: "Bằng chứng liên kết TEPP", + boundary: "Chỉ thể hiện mối liên hệ theo thời gian; kết quả này không xác định nguyên nhân.", + participants: "Chủ thể trong bằng chứng đã cung cấp", + span: "Khoảng lịch sử đã xác thực", + findings: "Kết quả TEPP", + noFindings: "TEPP đã sắp xếp các sự kiện tường minh và không trả về kết quả bổ sung.", + openEvidence: (label) => `Mở bằng chứng: ${label}`, + unnamedEvidence: (index) => `Bản ghi bằng chứng ${index}`, + status: { + not_configured: "Hãy cấu hình điểm cuối lịch sử dự án TEPP rồi xác thực lại dòng thời gian này.", + unavailable: "TEPP hiện không khả dụng. Hãy đọc dòng thời gian chuẩn và thử xác thực lại sau.", + invalid_evidence: "Hãy mở bằng chứng nguồn, sửa hợp đồng lịch sử dự án rồi chạy lại TEPP.", + }, + findingLabels: { + contract_award_before_focus: "Bản ghi trao hợp đồng có trước sự kiện được chọn.", + specification_change_before_focus: "Bản ghi thay đổi đặc tả có trước sự kiện được chọn.", + delivery_before_focus: "Bản ghi bàn giao có trước sự kiện được chọn.", + handoff_before_focus: "Bản ghi chuyển giao có trước sự kiện được chọn.", + rebid_after_focus: "Bản ghi đấu thầu lại có sau sự kiện được chọn.", + specification_change_and_handoff_before_focus: + "Các bản ghi thay đổi đặc tả và chuyển giao đều có trước sự kiện được chọn.", + }, + }, +}; + +function shortDate(value: string): string { + const parsed = new Date(value); + return Number.isNaN(parsed.valueOf()) ? value : parsed.toISOString().slice(0, 10); +} + +export function TeppProjectHistoryEvidence({ + validation, + onOpenPost, + sourceLabels, +}: { + validation: TeppProjectHistoryValidation; + onOpenPost: (postId: string) => void; + sourceLabels: Record; +}) { + const locale = useLocale(); + const copy = COPY[locale]; + const headingId = useId(); + const findingsHeadingId = useId(); + + const history = validation.project_history; + if (validation.status !== "validated" || history === null) { + const statusMessage = + validation.status === "validated" + ? copy.status.invalid_evidence + : copy.status[validation.status]; + return ( +
+

{copy.heading}

+

{statusMessage}

+
+ ); + } + + return ( +
+
+
+

{copy.eyebrow}

+

{copy.heading}

+
+ TEPP · v{history.contract_version} +
+

{copy.boundary}

+
+
+
{copy.participants}
+
{history.participant_count}
+
+
+
{copy.span}
+
+ {shortDate(history.history_span_start)} – {shortDate(history.history_span_end)} +
+
+
+
+
{copy.findings}
+ {history.findings.length === 0 ?

{copy.noFindings}

: null} + {history.findings.length > 0 ? ( +
    + {history.findings.map((finding) => ( +
  • +

    {copy.findingLabels[finding.finding_code]}

    +
    + {finding.evidence_post_ids.map((postId, index) => { + const label = sourceLabels[postId] ?? copy.unnamedEvidence(index + 1); + return ( + + ); + })} +
    +
  • + ))} +
+ ) : null} +
+
+ ); +} diff --git a/frontend/src/projectHistory.ts b/frontend/src/projectHistory.ts index 208202746..6e27a317f 100644 --- a/frontend/src/projectHistory.ts +++ b/frontend/src/projectHistory.ts @@ -2,6 +2,45 @@ import type { ProjectEvidence } from "./api"; import type { Locale } from "./i18n"; export type ProjectHistoryTruthStatus = "observed" | "inferred"; + +export type TeppProjectHistoryFindingCode = + | "contract_award_before_focus" + | "specification_change_before_focus" + | "delivery_before_focus" + | "handoff_before_focus" + | "rebid_after_focus" + | "specification_change_and_handoff_before_focus"; + +export interface TeppProjectHistoryFinding { + finding_code: TeppProjectHistoryFindingCode; + summary: string; + related_event_ids: string[]; + evidence_post_ids: string[]; +} + +export interface TeppProjectHistoryMetadata { + contract_version: 1; + project_key: string; + project_name: string; + focus_event_id: string; + knowledge_cutoff: string; + history_span_start: string; + history_span_end: string; + participant_count: number; + inference_status: "temporal_association_only"; + event_count: number; + findings: TeppProjectHistoryFinding[]; +} + +export interface TeppProjectHistoryValidation { + status: "validated" | "not_configured" | "unavailable" | "invalid_evidence"; + project_history: TeppProjectHistoryMetadata | null; + next_action_code: + | "open_source_evidence" + | "configure_tepp_project_history" + | "retry_tepp_project_history"; +} + export type ResponsibilityTransitionCode = "continuous" | "handoff" | "assignment_gap"; export type ProjectHistoryTimeBasis = "source_post_created_at_fallback" | "document_time"; @@ -73,6 +112,7 @@ export interface ProjectHistoryProjection { distinct_actor_count?: number; distinct_observed_actor_count: number; truncated: boolean; + tepp_validation?: TeppProjectHistoryValidation; events: ProjectHistoryEvent[]; } diff --git a/lineageweave/__init__.py b/lineageweave/__init__.py index fc9d18b25..2cb406a3f 100644 --- a/lineageweave/__init__.py +++ b/lineageweave/__init__.py @@ -55,4 +55,4 @@ "sentence_excerpts", ] -__version__ = "2.18.0" +__version__ = "2.20.0" diff --git a/lineageweave/http_client.py b/lineageweave/http_client.py index 389b29f3e..9ef4dd40a 100644 --- a/lineageweave/http_client.py +++ b/lineageweave/http_client.py @@ -38,6 +38,7 @@ def _request( body: bytes | None, headers: dict[str, str], timeout: float, + maximum_response_bytes: int | None = None, ) -> tuple[int, bytes]: """Implement the _request operation for this channel.""" parsed = urlparse(url) @@ -68,8 +69,17 @@ def _request( ) connection.request(method, path, body=body, headers=headers) response = connection.getresponse() - length_header = response.getheader("Content-Length") - raw = response.read(int(length_header)) if length_header is not None else response.read() + if maximum_response_bytes is not None: + if maximum_response_bytes < 1: + raise ValueError("maximum_response_bytes must be positive") + raw = response.read(maximum_response_bytes + 1) + if len(raw) > maximum_response_bytes: + raise HttpClientError( + f"response exceeds {maximum_response_bytes} bytes" + ) + else: + length_header = response.getheader("Content-Length") + raw = response.read(int(length_header)) if length_header is not None else response.read() return response.status, raw finally: connection.close() @@ -105,15 +115,22 @@ def post_json( *, headers: dict[str, str], timeout: float, + include_llm_metadata: bool = True, + maximum_response_bytes: int | None = None, ) -> dict: """POST ``payload`` as JSON to ``url`` and return the decoded object. + ``include_llm_metadata`` preserves contextual-orchestrator enrichment by + default. Closed non-LLM wire contracts must set it to ``False`` so an active + LLM context cannot add an unpublished ``metadata`` member. + ``maximum_response_bytes`` bounds reads for strict remote contracts. + Raises: ValueError: ``url`` is not an ``http`` / ``https`` URL with a host. HttpClientError: the server responded with HTTP >= 400 or non-JSON. """ request_payload = payload - request_metadata = current_llm_metadata() + request_metadata = current_llm_metadata() if include_llm_metadata else None if request_metadata: request_payload = dict(payload) existing_metadata = request_payload.get("metadata") @@ -123,13 +140,14 @@ def post_json( request_payload["metadata"] = {**existing_metadata, **request_metadata} else: raise ValueError("metadata must be an object") - status, raw = _request( - "POST", - url, - body=json.dumps(request_payload).encode("utf-8"), - headers={"content-type": "application/json", **headers}, - timeout=timeout, - ) + request_options = { + "body": json.dumps(request_payload).encode("utf-8"), + "headers": {"content-type": "application/json", **headers}, + "timeout": timeout, + } + if maximum_response_bytes is not None: + request_options["maximum_response_bytes"] = maximum_response_bytes + status, raw = _request("POST", url, **request_options) hostname = urlparse(url).hostname or url if status >= 400: raise HttpClientError(f"HTTP {status} from {hostname}") diff --git a/lineageweave/tepp_client.py b/lineageweave/tepp_client.py index 7dbfd886f..70469327a 100644 --- a/lineageweave/tepp_client.py +++ b/lineageweave/tepp_client.py @@ -18,8 +18,9 @@ from __future__ import annotations +from collections.abc import Callable from dataclasses import dataclass -from typing import Any, Callable +from typing import Any class TeppNotAvailable(RuntimeError): @@ -80,4 +81,9 @@ def __init__(self, transport: Callable[[dict[str, Any]], dict[str, Any]] = _no_t def submit_analysis_run(self, request: AnalysisRunRequest) -> dict[str, Any]: """Submit a request; returns TEPP's ``AnalysisRunAccepted`` envelope.""" - return self._transport(request.to_json()) + try: + return self._transport(request.to_json()) + except TeppNotAvailable: + raise + except Exception as exc: + raise TeppNotAvailable("TEPP transport request failed") from exc diff --git a/lineageweave/tepp_project_history.py b/lineageweave/tepp_project_history.py new file mode 100644 index 000000000..03ae94552 --- /dev/null +++ b/lineageweave/tepp_project_history.py @@ -0,0 +1,450 @@ +"""Strict client for TEPP's cutoff-safe project-history projection. + +LineageWeave owns authorization, exact project identity, and source selection. +TEPP may validate ordering and return temporal-association findings over that +closed evidence bundle. This module never forwards browser credentials, never +accepts changed source evidence, and never promotes order to causality. +""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping +from datetime import datetime, timezone +import json +import re +from typing import Any +from urllib.parse import urlsplit, urlunsplit + +from .http_client import HttpClientError, post_json + +PROJECT_HISTORY_CONTRACT_VERSION = 1 +PROJECT_HISTORY_PATH = "/v1/project-histories" +PROJECT_HISTORY_INFERENCE_STATUS = "temporal_association_only" +PROJECT_HISTORY_EVENT_LIMIT = 128 +PROJECT_HISTORY_ACTOR_LIMIT = 64 +PROJECT_HISTORY_BYTE_LIMIT = 256 * 1024 +_RFC3339_PATTERN = re.compile( + r"^[0-9]{4}-[0-9]{2}-[0-9]{2}[Tt][0-9]{2}:" + r"[0-9]{2}:[0-9]{2}(?:\.[0-9]+)?(?:[Zz]|[+-][0-9]{2}:[0-9]{2})$" +) + +_REQUEST_FIELDS = frozenset( + { + "contract_version", + "idempotency_key", + "tenant_workspace_id", + "project_key", + "project_name", + "knowledge_cutoff", + "focus_event_id", + "events", + } +) +_EVENT_FIELDS = frozenset( + { + "event_id", + "event_type_code", + "event_title", + "occurred_at", + "available_at", + "source_post_id", + "evidence_text", + "actor_ids", + } +) +_PROJECTION_FIELDS = frozenset( + { + "contract_version", + "project_key", + "project_name", + "focus_event_id", + "knowledge_cutoff", + "history_span_start", + "history_span_end", + "participant_count", + "inference_status", + "events", + "findings", + } +) +_FINDING_FIELDS = frozenset( + {"finding_code", "summary", "related_event_ids", "evidence_post_ids"} +) +_ALLOWED_FINDING_CODES = frozenset( + { + "contract_award_before_focus", + "specification_change_before_focus", + "delivery_before_focus", + "handoff_before_focus", + "rebid_after_focus", + "specification_change_and_handoff_before_focus", + } +) + +Transport = Callable[[str, dict[str, Any], dict[str, str], float], Any] + + +class TeppProjectHistoryUnavailable(RuntimeError): + """TEPP was absent or returned a response outside the public contract.""" + + +class TeppProjectHistoryInvalidResponse(TeppProjectHistoryUnavailable): + """TEPP returned a response that violated the validated evidence contract.""" + + +def _exact_object(value: Any, fields: frozenset[str], name: str) -> Mapping[str, Any]: + """Return a mapping only when it has the exact versioned field set.""" + + if not isinstance(value, Mapping) or frozenset(value) != fields: + raise TeppProjectHistoryUnavailable(f"{name} has invalid fields") + return value + + +def _text(value: Any, name: str, maximum: int = 4096) -> str: + """Return bounded, non-empty text without ASCII control characters.""" + + if not isinstance(value, str): + raise TeppProjectHistoryUnavailable(f"{name} must be text") + normalized = value.strip() + if ( + not normalized + or len(normalized.encode("utf-8")) > maximum + or any(ord(character) < 0x20 or ord(character) == 0x7F for character in normalized) + ): + raise TeppProjectHistoryUnavailable(f"{name} is empty or outside its bound") + return normalized + + +def parse_rfc3339_utc(value: Any, name: str) -> tuple[datetime, str]: + """Parse an RFC 3339 timestamp and return canonical UTC text.""" + + raw = _text(value, name, 64) + if _RFC3339_PATTERN.fullmatch(raw) is None: + raise TeppProjectHistoryUnavailable(f"{name} is not RFC 3339") + normalized_text = raw[:10] + "T" + raw[11:] + try: + parsed = datetime.fromisoformat( + normalized_text[:-1] + "+00:00" + if normalized_text.endswith(("Z", "z")) + else normalized_text + ) + except ValueError as exc: + raise TeppProjectHistoryUnavailable(f"{name} is not RFC 3339") from exc + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise TeppProjectHistoryUnavailable(f"{name} must include an offset") + utc = parsed.astimezone(timezone.utc) + return utc, utc.isoformat().replace("+00:00", "Z") + + +def project_history_event_sort_key(event: Mapping[str, Any]) -> tuple[datetime, str]: + """Order project-history events by their instant, then stable identity.""" + + occurred_at, _ = parse_rfc3339_utc(event["occurred_at"], "occurred_at") + return occurred_at, str(event["event_id"]) + + +def _code(value: Any, name: str) -> str: + """Return a bounded lower-snake contract code.""" + + code = _text(value, name, 96) + if not all( + character.isascii() + and (character.islower() or character.isdigit() or character == "_") + for character in code + ): + raise TeppProjectHistoryUnavailable(f"{name} must be lower snake case") + return code + + +def _event(value: Any, *, cutoff: datetime | None = None) -> dict[str, Any]: + """Validate one exact source-grounded event.""" + + payload = _exact_object(value, _EVENT_FIELDS, "project-history event") + occurred, occurred_text = parse_rfc3339_utc(payload["occurred_at"], "occurred_at") + available, available_text = parse_rfc3339_utc(payload["available_at"], "available_at") + if cutoff is not None and (occurred > cutoff or available > cutoff): + raise TeppProjectHistoryUnavailable("event exceeds the knowledge cutoff") + raw_actors = payload["actor_ids"] + if not isinstance(raw_actors, list) or len(raw_actors) > PROJECT_HISTORY_ACTOR_LIMIT: + raise TeppProjectHistoryUnavailable("actor_ids must be a bounded list") + actors = [_text(actor, "actor_id", 256) for actor in raw_actors] + if len(actors) != len(set(actors)): + raise TeppProjectHistoryUnavailable("actor_ids must be unique within an event") + return { + "event_id": _text(payload["event_id"], "event_id", 256), + "event_type_code": _code(payload["event_type_code"], "event_type_code"), + "event_title": _text(payload["event_title"], "event_title", 512), + "occurred_at": occurred_text, + "available_at": available_text, + "source_post_id": _text(payload["source_post_id"], "source_post_id", 256), + "evidence_text": _text(payload["evidence_text"], "evidence_text", 4096), + "actor_ids": actors, + } + + +def validate_tepp_project_history_request( + value: Any, + *, + now: datetime | None = None, +) -> dict[str, Any]: + """Validate and canonicalize one TEPP project-history request.""" + + payload = _exact_object(value, _REQUEST_FIELDS, "project-history request") + if payload["contract_version"] != PROJECT_HISTORY_CONTRACT_VERSION: + raise TeppProjectHistoryUnavailable("unsupported request contract version") + receipt = now or datetime.now(timezone.utc) + if receipt.tzinfo is None or receipt.utcoffset() is None: + raise TeppProjectHistoryUnavailable("request receipt clock must be offset-aware") + cutoff, cutoff_text = parse_rfc3339_utc(payload["knowledge_cutoff"], "knowledge_cutoff") + if cutoff > receipt.astimezone(timezone.utc): + raise TeppProjectHistoryUnavailable("knowledge cutoff is after request receipt") + raw_events = payload["events"] + if ( + not isinstance(raw_events, list) + or not raw_events + or len(raw_events) > PROJECT_HISTORY_EVENT_LIMIT + ): + raise TeppProjectHistoryUnavailable("event count is outside the contract bound") + events = [_event(event, cutoff=cutoff) for event in raw_events] + event_ids = [event["event_id"] for event in events] + if len(event_ids) != len(set(event_ids)): + raise TeppProjectHistoryUnavailable("event identities must be unique") + focus_event_id = _text(payload["focus_event_id"], "focus_event_id", 256) + if focus_event_id not in set(event_ids): + raise TeppProjectHistoryUnavailable("focus event is outside the evidence bundle") + validated = { + "contract_version": PROJECT_HISTORY_CONTRACT_VERSION, + "idempotency_key": _text(payload["idempotency_key"], "idempotency_key", 256), + "tenant_workspace_id": _text( + payload["tenant_workspace_id"], "tenant_workspace_id", 256 + ), + "project_key": _text(payload["project_key"], "project_key", 256), + "project_name": _text(payload["project_name"], "project_name", 512), + "knowledge_cutoff": cutoff_text, + "focus_event_id": focus_event_id, + "events": events, + } + wire = json.dumps(validated, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + if len(wire) > PROJECT_HISTORY_BYTE_LIMIT: + raise TeppProjectHistoryUnavailable("project-history request exceeds 256 KiB") + return validated + + +def _finding( + value: Any, + *, + event_ids: set[str], + source_post_ids: set[str], +) -> dict[str, Any]: + """Validate one finding against the submitted evidence bundle.""" + + payload = _exact_object(value, _FINDING_FIELDS, "project-history finding") + related = payload["related_event_ids"] + evidence = payload["evidence_post_ids"] + if not isinstance(related, list) or not isinstance(evidence, list): + raise TeppProjectHistoryUnavailable("finding references must be lists") + related_ids = [_text(item, "related_event_id", 256) for item in related] + evidence_ids = [_text(item, "evidence_post_id", 256) for item in evidence] + if ( + not related_ids + or not evidence_ids + or not set(related_ids).issubset(event_ids) + or not set(evidence_ids).issubset(source_post_ids) + ): + raise TeppProjectHistoryUnavailable("finding cites evidence outside the bundle") + finding_code = _code(payload["finding_code"], "finding_code") + if finding_code not in _ALLOWED_FINDING_CODES: + raise TeppProjectHistoryUnavailable("finding code is not in the published vocabulary") + if len(related_ids) != len(set(related_ids)) or len(evidence_ids) != len( + set(evidence_ids) + ): + raise TeppProjectHistoryUnavailable("finding references must be unique") + return { + "finding_code": finding_code, + "summary": _text(payload["summary"], "finding summary", 4096), + "related_event_ids": related_ids, + "evidence_post_ids": evidence_ids, + } + + +def validate_tepp_project_history_projection( + value: Any, + *, + request: Any, +) -> dict[str, Any]: + """Validate TEPP output against the exact submitted events and identities.""" + + validated_request = validate_tepp_project_history_request(request) + payload = _exact_object(value, _PROJECTION_FIELDS, "project-history projection") + if payload["contract_version"] != PROJECT_HISTORY_CONTRACT_VERSION: + raise TeppProjectHistoryUnavailable("unsupported response contract version") + if payload["inference_status"] != PROJECT_HISTORY_INFERENCE_STATUS: + raise TeppProjectHistoryUnavailable("TEPP response attempted causal authority") + if ( + _text(payload["project_key"], "project_key", 256) + != validated_request["project_key"] + or _text(payload["project_name"], "project_name", 512) + != validated_request["project_name"] + or _text(payload["focus_event_id"], "focus_event_id", 256) + != validated_request["focus_event_id"] + ): + raise TeppProjectHistoryUnavailable("TEPP changed project or focus identity") + _, response_cutoff = parse_rfc3339_utc(payload["knowledge_cutoff"], "knowledge_cutoff") + if response_cutoff != validated_request["knowledge_cutoff"]: + raise TeppProjectHistoryUnavailable("TEPP changed the knowledge cutoff") + raw_events = payload["events"] + if not isinstance(raw_events, list): + raise TeppProjectHistoryUnavailable("projection events must be a list") + response_events = [_event(event) for event in raw_events] + expected_events = sorted( + validated_request["events"], + key=project_history_event_sort_key, + ) + if response_events != expected_events: + raise TeppProjectHistoryUnavailable("TEPP changed or reordered supplied evidence") + participant_count = payload["participant_count"] + expected_participants = len( + {actor for event in response_events for actor in event["actor_ids"]} + ) + if ( + isinstance(participant_count, bool) + or not isinstance(participant_count, int) + or participant_count != expected_participants + ): + raise TeppProjectHistoryUnavailable("participant count is not evidence-derived") + _, span_start = parse_rfc3339_utc(payload["history_span_start"], "history_span_start") + _, span_end = parse_rfc3339_utc(payload["history_span_end"], "history_span_end") + if ( + span_start != response_events[0]["occurred_at"] + or span_end != response_events[-1]["occurred_at"] + ): + raise TeppProjectHistoryUnavailable("history span does not match ordered events") + raw_findings = payload["findings"] + if not isinstance(raw_findings, list): + raise TeppProjectHistoryUnavailable("projection findings must be a list") + event_ids = {event["event_id"] for event in response_events} + source_post_ids = {event["source_post_id"] for event in response_events} + findings = [ + _finding( + finding, + event_ids=event_ids, + source_post_ids=source_post_ids, + ) + for finding in raw_findings + ] + return { + "contract_version": PROJECT_HISTORY_CONTRACT_VERSION, + "project_key": validated_request["project_key"], + "project_name": validated_request["project_name"], + "focus_event_id": validated_request["focus_event_id"], + "knowledge_cutoff": response_cutoff, + "history_span_start": span_start, + "history_span_end": span_end, + "participant_count": participant_count, + "inference_status": PROJECT_HISTORY_INFERENCE_STATUS, + "events": response_events, + "findings": findings, + } + + +def tepp_project_history_endpoint(transport_url: str) -> str: + """Resolve the project-history URL, allowing plain HTTP only on loopback.""" + + candidate = transport_url.strip() + if not candidate or any(ord(character) < 0x20 for character in candidate): + raise TeppProjectHistoryUnavailable("TEPP project-history transport is not configured") + parsed = urlsplit(candidate) + hostname = parsed.hostname.casefold() if parsed.hostname else "" + loopback = hostname in {"localhost", "127.0.0.1", "::1"} + if ( + not hostname + or (parsed.scheme != "https" and not (parsed.scheme == "http" and loopback)) + or parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + ): + raise TeppProjectHistoryUnavailable("TEPP URL must be HTTPS or loopback HTTP") + try: + parsed.port + except ValueError as exc: + raise TeppProjectHistoryUnavailable("TEPP URL has an invalid port") from exc + path = parsed.path.rstrip("/") + if path.endswith("/v1/analysis-runs"): + path = path[: -len("/v1/analysis-runs")] + elif path.endswith(PROJECT_HISTORY_PATH): + path = path[: -len(PROJECT_HISTORY_PATH)] + elif path not in {"", "/"}: + raise TeppProjectHistoryUnavailable("TEPP URL has an unsupported path") + return urlunsplit( + (parsed.scheme, parsed.netloc, f"{path}{PROJECT_HISTORY_PATH}", "", "") + ) + + +class TeppProjectHistoryClient: + """Submit a credential-free request and validate TEPP's exact response.""" + + def __init__( + self, + transport_url: str, + *, + transport: Transport | None = None, + timeout_seconds: float = 30.0, + ) -> None: + self._transport_url = transport_url + self._transport = transport or self._post + self._timeout_seconds = timeout_seconds + + @property + def available(self) -> bool: + """Return whether a syntactically valid endpoint is configured.""" + + try: + tepp_project_history_endpoint(self._transport_url) + except TeppProjectHistoryUnavailable: + return False + return True + + @staticmethod + def _post( + url: str, + payload: dict[str, Any], + headers: dict[str, str], + timeout: float, + ) -> Any: + """Post one bounded JSON exchange through the shared HTTP client.""" + + return post_json( + url, + payload, + headers=headers, + timeout=timeout, + include_llm_metadata=False, + maximum_response_bytes=PROJECT_HISTORY_BYTE_LIMIT, + ) + + def project(self, request: Any) -> dict[str, Any]: + """Return a validated non-causal projection or fail closed.""" + + target = tepp_project_history_endpoint(self._transport_url) + payload = validate_tepp_project_history_request(request) + headers = { + "content-type": "application/json", + "tepp-consumer": "lineageweave", + "tepp-contract-version": str(PROJECT_HISTORY_CONTRACT_VERSION), + "idempotency-key": payload["idempotency_key"], + } + try: + response = self._transport(target, payload, headers, self._timeout_seconds) + except TeppProjectHistoryUnavailable: + raise + except (HttpClientError, OSError, TypeError, ValueError) as exc: + raise TeppProjectHistoryUnavailable("TEPP project-history request failed") from exc + except Exception as exc: + raise TeppProjectHistoryUnavailable("TEPP project-history request failed") from exc + try: + return validate_tepp_project_history_projection(response, request=payload) + except TeppProjectHistoryUnavailable as exc: + raise TeppProjectHistoryInvalidResponse( + "TEPP project-history response violated its contract" + ) from exc diff --git a/migrations/0054_post_chat_knowledge_cutoff.sql b/migrations/0054_post_chat_knowledge_cutoff.sql new file mode 100644 index 000000000..00d05706d --- /dev/null +++ b/migrations/0054_post_chat_knowledge_cutoff.sql @@ -0,0 +1,28 @@ +alter table post_chat_result + add column if not exists knowledge_cutoff timestamptz; + +update post_chat_result + set knowledge_cutoff = computed_at + where knowledge_cutoff is null; + +alter table post_chat_result + alter column knowledge_cutoff set default now(), + alter column knowledge_cutoff set not null; + +do $$ +begin + if not exists ( + select 1 + from pg_constraint + where conname = 'post_chat_result_knowledge_cutoff_check' + and conrelid = 'post_chat_result'::regclass + ) then + alter table post_chat_result + add constraint post_chat_result_knowledge_cutoff_check + check (knowledge_cutoff <= computed_at); + end if; +end +$$; + +comment on column post_chat_result.knowledge_cutoff is + 'Maximum source availability time used to compute this persisted answer.'; diff --git a/migrations/rollback/0054_post_chat_knowledge_cutoff.sql b/migrations/rollback/0054_post_chat_knowledge_cutoff.sql new file mode 100644 index 000000000..8980fe69f --- /dev/null +++ b/migrations/rollback/0054_post_chat_knowledge_cutoff.sql @@ -0,0 +1,5 @@ +alter table post_chat_result + drop constraint if exists post_chat_result_knowledge_cutoff_check; + +alter table post_chat_result + drop column if exists knowledge_cutoff; diff --git a/pyproject.toml b/pyproject.toml index 5f00a4062..8d34399ec 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "lineageweave" -version = "2.18.0" +version = "2.20.0" description = "Reconstructs git-branch-style lineage DAGs from scattered short records using multi-channel score fusion and LLM adjudication." readme = "README.md" license = { text = "MIT" } diff --git a/tests/test_ask_project_history.py b/tests/test_ask_project_history.py new file mode 100644 index 000000000..c9420be90 --- /dev/null +++ b/tests/test_ask_project_history.py @@ -0,0 +1,348 @@ +"""Contracts for project histories attached to post-scoped and Global Ask.""" + +from __future__ import annotations + +import asyncio +from datetime import UTC, datetime +from types import SimpleNamespace + +import pytest +from fastapi import HTTPException + +from backend.app import main +from backend.app.ask_project_history import ( + AskEvidenceProjection, + global_ask_session_citations_authorized, + read_authorized_ask_evidence, +) +from backend.app.auth import CurrentAccount +from backend.app.post_chat_ingestion import gather_global_chat_sources + +CUTOFF = datetime(2026, 8, 20, 12, 0, tzinfo=UTC) + + +class _EvidenceConnection: + """Query-shaped double for current citation and project evidence.""" + + def __init__(self, rows: list[dict[str, object]]) -> None: + self.rows = rows + self.calls: list[tuple[str, tuple[object, ...]]] = [] + + async def fetch(self, query: str, *args: object): + self.calls.append((query, args)) + return self.rows + + +def test_authorized_ask_evidence_groups_exact_projects_and_preserves_citation_order() -> None: + conn = _EvidenceConnection( + [ + { + "post_id": "00000000-0000-4000-8000-000000000002", + "post_title": "Second evidence", + "citation_ordinal": 2, + "project_key": "P-100", + "project_name": "Synthetic renewal", + "truth_status_code": "inferred", + "truth_order": 1, + }, + { + "post_id": "00000000-0000-4000-8000-000000000001", + "post_title": "First evidence", + "citation_ordinal": 1, + "project_key": "P-100", + "project_name": "Synthetic renewal", + "truth_status_code": "observed", + "truth_order": 0, + }, + ] + ) + + result = asyncio.run( + read_authorized_ask_evidence( + conn, + cited_post_ids=[ + "00000000-0000-4000-8000-000000000001", + "00000000-0000-4000-8000-000000000002", + ], + corporate_entity_ids=["tenant-a"], + knowledge_cutoff=CUTOFF, + ) + ) + + assert result.all_citations_visible + assert [post["post_title"] for post in result.cited_posts] == [ + "First evidence", + "Second evidence", + ] + assert result.project_histories == ( + { + "project_key": "P-100", + "project_name": "Synthetic renewal", + "focus_post_id": "00000000-0000-4000-8000-000000000001", + "source_post_ids": [ + "00000000-0000-4000-8000-000000000001", + "00000000-0000-4000-8000-000000000002", + ], + "knowledge_cutoff": "2026-08-20T12:00:00Z", + "truth_status_code": "observed", + }, + ) + query, args = conn.calls[0] + assert "source_draft_code" in query + assert "source_deleted_flag" in query + assert "created_at <= $3" in query + assert args[2] == CUTOFF + + +def test_authorized_ask_evidence_fails_closed_when_any_citation_is_hidden() -> None: + conn = _EvidenceConnection( + [ + { + "post_id": "00000000-0000-4000-8000-000000000001", + "post_title": "Visible evidence", + "citation_ordinal": 1, + "project_key": None, + "project_name": None, + "truth_status_code": None, + "truth_order": None, + } + ] + ) + + result = asyncio.run( + read_authorized_ask_evidence( + conn, + cited_post_ids=[ + "00000000-0000-4000-8000-000000000001", + "00000000-0000-4000-8000-000000000099", + ], + corporate_entity_ids=["tenant-a"], + knowledge_cutoff=CUTOFF, + ) + ) + + assert not result.all_citations_visible + assert result.project_histories == () + + +def test_authorized_ask_evidence_rejects_non_uuid_citations_before_sql() -> None: + with pytest.raises(ValueError, match="UUIDs"): + asyncio.run( + read_authorized_ask_evidence( + _EvidenceConnection([]), + cited_post_ids=["not-a-uuid"], + corporate_entity_ids=["tenant-a"], + knowledge_cutoff=CUTOFF, + ) + ) + + +def test_global_ask_session_reauthorizes_every_persisted_citation() -> None: + class SessionConnection: + def __init__(self) -> None: + self.call = 0 + + async def fetch(self, query: str, *args: object): + del args + self.call += 1 + if "global_ask_turn_citation" in query: + return [ + {"cited_post_id": "00000000-0000-4000-8000-000000000001"}, + {"cited_post_id": "00000000-0000-4000-8000-000000000099"}, + ] + return [ + { + "post_id": "00000000-0000-4000-8000-000000000001", + "post_title": "Visible evidence", + "citation_ordinal": 1, + "project_key": None, + "project_name": None, + "truth_status_code": None, + "truth_order": None, + } + ] + + authorized = asyncio.run( + global_ask_session_citations_authorized( + SessionConnection(), + session_id="00000000-0000-4000-8000-000000000010", + corporate_entity_ids=["tenant-a"], + knowledge_cutoff=CUTOFF, + ) + ) + assert not authorized + + +def test_global_source_retrieval_applies_cutoff_and_publication_eligibility() -> None: + calls: list[tuple[str, tuple[object, ...]]] = [] + + class CaptureConnection: + async def fetch(self, query: str, *args: object): + calls.append((query, args)) + return [] + + asyncio.run( + gather_global_chat_sources( + CaptureConnection(), + lambda _row: True, + ["tenant-a"], + question="synthetic project", + limit=2, + knowledge_cutoff=CUTOFF, + ) + ) + + candidate_queries = [query for query, _args in calls if "matched_in" in query] + source_calls = [ + (query, args) + for query, args in calls + if "array_position($2::uuid[], post_id)" in query + ] + assert candidate_queries + assert all("source_draft_code" in query and "created_at <= $3" in query for query in candidate_queries) + assert source_calls + source_query, source_args = source_calls[0] + assert "source_deleted_flag" in source_query + assert "created_at <= $4" in source_query + assert source_args[3] == CUTOFF + + +class _Acquire: + def __init__(self, connection: object) -> None: + self.connection = connection + + async def __aenter__(self) -> object: + return self.connection + + async def __aexit__(self, exc_type, exc_value, traceback) -> None: + return None + + +class _Pool: + def __init__(self, connection: object) -> None: + self.connection = connection + + def acquire(self) -> _Acquire: + return _Acquire(self.connection) + + +def _account() -> CurrentAccount: + return CurrentAccount( + user_account_id="account-1", + external_subject_id="subject-1", + display_name="Synthetic analyst", + preferred_locale="en", + corporate_entity_ids=frozenset({"tenant-a"}), + permission_codes=frozenset({"post_read"}), + ) + + +def test_stored_post_chat_omits_an_answer_after_citation_access_is_lost(monkeypatch) -> None: + async def visible_post(*_args, **_kwargs): + return {"post_id": "post-1"} + + async def stored_chats(*_args, **_kwargs): + return [ + { + "question_text": "What happened?", + "answer_text": "A formerly authorized answer.", + "cited_post_ids": ["hidden-post"], + "cited_posts": [{"post_id": "hidden-post", "post_title": "Hidden"}], + "_knowledge_cutoff": CUTOFF, + } + ] + + async def hidden_evidence(*_args, **_kwargs): + return AskEvidenceProjection( + all_citations_visible=False, + cited_posts=(), + project_histories=(), + project_histories_truncated=False, + knowledge_cutoff="2026-08-20T12:00:00Z", + ) + + monkeypatch.setattr(main, "_load_visible_post", visible_post) + monkeypatch.setattr(main, "fetch_persisted_chats", stored_chats) + monkeypatch.setattr(main, "read_authorized_ask_evidence", hidden_evidence) + + result = asyncio.run( + main.read_post_chat( + post_id="post-1", + account=_account(), + pool=_Pool(object()), + ) + ) + assert result == {"post_id": "post-1", "exchanges": []} + + +def test_global_ask_rejects_stale_session_context_before_reusing_hidden_prose(monkeypatch) -> None: + async def ensure_session(*_args, **_kwargs): + return "00000000-0000-4000-8000-000000000010" + + async def unauthorized(*_args, **_kwargs): + return False + + monkeypatch.setattr(main, "_post_chat_client", lambda: SimpleNamespace(available=True)) + monkeypatch.setattr(main, "ensure_global_ask_session", ensure_session) + monkeypatch.setattr(main, "global_ask_session_citations_authorized", unauthorized) + + with pytest.raises(HTTPException) as exc_info: + asyncio.run( + main.ask_agent( + request=main.GlobalAskRequest( + question="Continue the prior answer", + session_id="00000000-0000-4000-8000-000000000010", + ), + account=_account(), + pool=_Pool(object()), + valkey=SimpleNamespace(), + ) + ) + + assert exc_info.value.status_code == 409 + assert "start a new session" in str(exc_info.value.detail).lower() + + +def test_global_ask_hides_unexpected_provider_errors(monkeypatch) -> None: + class ProviderFailure: + available = True + + def answer(self, *args, **kwargs): + del args, kwargs + raise RuntimeError("raw provider trace must not reach the buyer") + + async def ensure_session(*_args, **_kwargs): + return "00000000-0000-4000-8000-000000000010" + + async def authorized(*_args, **_kwargs): + return True + + async def load_context(*_args, **_kwargs): + return SimpleNamespace( + session_id="00000000-0000-4000-8000-000000000010", + summary="", + recent_turns=(), + compress_turns=(), + ) + + async def sources(*_args, **_kwargs): + return [object()] + + monkeypatch.setattr(main, "_post_chat_client", lambda: ProviderFailure()) + monkeypatch.setattr(main, "ensure_global_ask_session", ensure_session) + monkeypatch.setattr(main, "global_ask_session_citations_authorized", authorized) + monkeypatch.setattr(main, "load_global_ask_context", load_context) + monkeypatch.setattr(main, "gather_global_chat_sources", sources) + + with pytest.raises(HTTPException) as exc_info: + asyncio.run( + main.ask_agent( + request=main.GlobalAskRequest(question="What happened?"), + account=_account(), + pool=_Pool(object()), + valkey=SimpleNamespace(), + ) + ) + + assert exc_info.value.status_code == 503 + assert "raw provider trace" not in str(exc_info.value.detail) diff --git a/tests/test_ask_project_history_cutoff.py b/tests/test_ask_project_history_cutoff.py new file mode 100644 index 000000000..22c7cd059 --- /dev/null +++ b/tests/test_ask_project_history_cutoff.py @@ -0,0 +1,77 @@ +"""Contracts for persisted post-Ask knowledge cutoffs.""" + +from __future__ import annotations + +import asyncio +from datetime import datetime, timezone +from pathlib import Path + +from backend.app.post_chat_ingestion import persist_post_chat + + +ROOT = Path(__file__).resolve().parents[1] +CUTOFF = datetime(2026, 8, 20, 12, 0, tzinfo=timezone.utc) + + +class _Connection: + """Minimal chat-persistence double that records SQL parameters.""" + + def __init__(self) -> None: + self.executions: list[tuple[str, tuple[object, ...]]] = [] + + async def execute(self, query: str, *args: object) -> None: + self.executions.append((query, args)) + + async def fetchrow(self, query: str, *args: object): + del args + if "from post_chat_result" not in query: + return None + return { + "question_text": "What happened?", + "answer_text": "Synthetic answer", + "knowledge_cutoff": CUTOFF, + } + + async def fetch(self, query: str, *args: object): + del query, args + return [] + + +def test_persist_post_chat_writes_the_retrieval_cutoff_not_a_later_read_clock() -> None: + conn = _Connection() + + result = asyncio.run( + persist_post_chat( + conn, + "00000000-0000-4000-8000-000000000001", + "What happened?", + "Synthetic answer", + [], + knowledge_cutoff=CUTOFF, + ) + ) + + insert = next( + (query, args) + for query, args in conn.executions + if "insert into post_chat_result" in query + ) + assert "computed_at" in insert[0] + assert "knowledge_cutoff" in insert[0] + assert insert[1][-2] >= CUTOFF + assert insert[1][-1] == CUTOFF + assert result["_knowledge_cutoff"] == CUTOFF + + +def test_cutoff_migration_is_applied_and_fails_closed_on_inverted_clocks() -> None: + migration = ROOT / "migrations/0054_post_chat_knowledge_cutoff.sql" + rollback = ROOT / "migrations/rollback/0054_post_chat_knowledge_cutoff.sql" + migrate_script = (ROOT / "docker/postgres-init/migrate.sh").read_text(encoding="utf-8") + + assert migration.is_file() + text = migration.read_text(encoding="utf-8") + assert "knowledge_cutoff timestamptz" in text + assert "knowledge_cutoff = computed_at" in text + assert "knowledge_cutoff <= computed_at" in text + assert rollback.is_file() + assert "0054_*" in migrate_script diff --git a/tests/test_global_ask_cutoff_contract.py b/tests/test_global_ask_cutoff_contract.py new file mode 100644 index 000000000..bdcf082ad --- /dev/null +++ b/tests/test_global_ask_cutoff_contract.py @@ -0,0 +1,54 @@ +"""Regression contracts for the final Global Ask source query.""" + +from __future__ import annotations + +import asyncio +from datetime import UTC, datetime + +from backend.app.post_chat_ingestion import gather_global_chat_sources + + +CUTOFF = datetime(2026, 8, 20, 12, 0, tzinfo=UTC) +AUTHORIZED_ENTITY_ID = "00000000-0000-4000-8000-000000000001" + + +class _RecordingConnection: + """Record query arguments while returning an empty authorized corpus.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, tuple[object, ...]]] = [] + + async def fetch(self, query: str, *args: object): + """Record one query call and return no rows.""" + + self.calls.append((query, args)) + return [] + + +def test_final_global_source_query_reuses_scope_and_binds_cutoff() -> None: + """One-shot tenant scope and the cutoff survive into the final SQL call.""" + + connection = _RecordingConnection() + authorized_ids = (value for value in [AUTHORIZED_ENTITY_ID]) + + result = asyncio.run( + gather_global_chat_sources( + connection, + lambda _row: True, + authorized_ids, + question="", + limit=2, + knowledge_cutoff=CUTOFF, + ) + ) + + assert result == [] + final_calls = [ + (query, args) + for query, args in connection.calls + if "array_position($2::uuid[], post_id)" in query + ] + assert len(final_calls) == 1 + final_query, final_args = final_calls[0] + assert "created_at <= $4" in final_query + assert final_args == ([AUTHORIZED_ENTITY_ID], [], 2, CUTOFF) diff --git a/tests/test_global_ask_cutoff_postgres.py b/tests/test_global_ask_cutoff_postgres.py new file mode 100644 index 000000000..5d07797a8 --- /dev/null +++ b/tests/test_global_ask_cutoff_postgres.py @@ -0,0 +1,82 @@ +"""PostgreSQL regression for the final Global Ask cutoff boundary.""" + +from __future__ import annotations + +import asyncio +import os +from datetime import UTC, datetime + +import asyncpg +import pytest + +from backend.app.post_chat_ingestion import gather_global_chat_sources + + +CUTOFF = datetime(2026, 8, 20, 12, 0, tzinfo=UTC) +POSTGRES_DSN = os.environ.get("LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN") + + +@pytest.mark.skipif(not POSTGRES_DSN, reason="requires PostgreSQL integration DSN") +def test_final_global_source_query_binds_the_cutoff_in_real_postgresql() -> None: + """The final authorized-source SQL binds every positional parameter.""" + + async def scenario() -> None: + connection = await asyncpg.connect(POSTGRES_DSN) + try: + await connection.execute( + """ + create temporary table source_post ( + post_id uuid primary key, + post_title text, + post_body text, + visibility_code text, + corporate_entity_id uuid, + created_at timestamptz, + source_system_code text, + source_record_key text, + source_author_code text, + source_author_name text, + source_company_code text, + source_company_name text, + source_process_unit_code text, + source_process_unit_name text, + source_sales_pool_code text, + source_sales_pool_name text, + source_customer_code text, + source_customer_name text, + source_project_code text, + source_project_name text, + source_draft_code text, + source_deleted_flag text + ) + """ + ) + + class PostgresBoundary: + """Execute only the final source query against PostgreSQL.""" + + def __init__(self) -> None: + self.final_args: tuple[object, ...] | None = None + + async def fetch(self, query: str, *args: object): + if "array_position($2::uuid[], post_id)" in query: + self.final_args = args + return await connection.fetch(query, *args) + return [] + + boundary = PostgresBoundary() + result = await gather_global_chat_sources( + boundary, + lambda _row: True, + ["00000000-0000-4000-8000-000000000001"], + question="synthetic project", + limit=2, + knowledge_cutoff=CUTOFF, + ) + assert result == [] + assert boundary.final_args is not None + assert boundary.final_args[3] == CUTOFF + finally: + await connection.close() + + asyncio.run(scenario()) diff --git a/tests/test_migration_identity.py b/tests/test_migration_identity.py new file mode 100644 index 000000000..3e5c19a92 --- /dev/null +++ b/tests/test_migration_identity.py @@ -0,0 +1,30 @@ +"""Migration identity and replay-window contracts.""" + +from __future__ import annotations + +from collections import Counter +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] + + +def test_forward_migration_numeric_prefixes_are_unique() -> None: + """Every forward migration has one unambiguous numeric identity.""" + + migrations = sorted((ROOT / "migrations").glob("[0-9][0-9][0-9][0-9]_*.sql")) + counts = Counter(path.name.split("_", 1)[0] for path in migrations) + duplicates = sorted(prefix for prefix, count in counts.items() if count > 1) + assert duplicates == [] + + +def test_post_chat_cutoff_uses_the_next_unique_replayable_migration() -> None: + """The Ask cutoff migration remains independently addressable and replayed.""" + + forward = ROOT / "migrations/0054_post_chat_knowledge_cutoff.sql" + rollback = ROOT / "migrations/rollback/0054_post_chat_knowledge_cutoff.sql" + script = (ROOT / "docker/postgres-init/migrate.sh").read_text(encoding="utf-8") + assert forward.is_file() + assert rollback.is_file() + assert not (ROOT / "migrations/0053_post_chat_knowledge_cutoff.sql").exists() + assert "0054_*" in script diff --git a/tests/test_strict_http_metadata.py b/tests/test_strict_http_metadata.py new file mode 100644 index 000000000..85e87d8b3 --- /dev/null +++ b/tests/test_strict_http_metadata.py @@ -0,0 +1,205 @@ +"""Contracts for contextual metadata on open and closed HTTP payloads.""" + +from __future__ import annotations + +import json +import threading +from copy import deepcopy +from http.server import BaseHTTPRequestHandler, HTTPServer + +import pytest + +from lineageweave import tepp_project_history as tepp_transport_module +from lineageweave.http_client import HttpClientError, post_json +from lineageweave.llm_context import use_llm_metadata +from lineageweave.tepp_project_history import ( + TeppProjectHistoryClient, + TeppProjectHistoryUnavailable, + validate_tepp_project_history_request, +) + + +class _EchoHandler(BaseHTTPRequestHandler): + """Echo one JSON request for shared-client contract tests.""" + + def do_POST(self) -> None: # noqa: N802 -- stdlib callback name + length = int(self.headers.get("content-length", "0")) + payload = json.loads(self.rfile.read(length).decode("utf-8")) + body = json.dumps( + {"oversized": "x" * 512} if self.path == "/oversized" else {"echo": payload} + ).encode("utf-8") + self.send_response(200) + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, format: str, *args: object) -> None: # noqa: A002 + """Suppress test HTTP access logs.""" + + +def _serve() -> tuple[HTTPServer, str]: + """Start one local JSON echo server.""" + + server = HTTPServer(("127.0.0.1", 0), _EchoHandler) + threading.Thread(target=server.serve_forever, daemon=True).start() + host, port = server.server_address[:2] + return server, f"http://{host}:{port}" + + +def _request() -> dict[str, object]: + """Return one minimal exact TEPP project-history request.""" + + return { + "contract_version": 1, + "idempotency_key": "strict-http-metadata", + "tenant_workspace_id": "tenant-a", + "project_key": "P-100", + "project_name": "Synthetic renewal", + "knowledge_cutoff": "2026-08-20T12:00:00Z", + "focus_event_id": "event-1", + "events": [ + { + "event_id": "event-1", + "event_type_code": "voc_received", + "event_title": "Synthetic VOC received", + "occurred_at": "2026-08-20T10:00:00Z", + "available_at": "2026-08-20T10:00:00Z", + "source_post_id": "post-1", + "evidence_text": "Synthetic VOC received", + "actor_ids": ["lw-actor-1"], + } + ], + } + + +def _response(request: dict[str, object]) -> dict[str, object]: + """Return the exact successful response for ``request``.""" + + events = deepcopy(request["events"]) + return { + "contract_version": 1, + "project_key": request["project_key"], + "project_name": request["project_name"], + "focus_event_id": request["focus_event_id"], + "knowledge_cutoff": request["knowledge_cutoff"], + "history_span_start": events[0]["occurred_at"], + "history_span_end": events[-1]["occurred_at"], + "participant_count": 1, + "inference_status": "temporal_association_only", + "events": events, + "findings": [], + } + + +def test_post_json_includes_llm_metadata_by_default() -> None: + """Existing LLM clients retain contextual metadata enrichment.""" + + server, base = _serve() + try: + with use_llm_metadata({"lineageweave_post_id": "post-1"}): + body = post_json( + f"{base}/v1/chat/completions", + {"messages": []}, + headers={}, + timeout=2.0, + ) + finally: + server.shutdown() + + assert body["echo"] == { + "messages": [], + "metadata": {"lineageweave_post_id": "post-1"}, + } + + +def test_post_json_can_disable_metadata_for_a_closed_contract() -> None: + """Closed contracts remain byte-shape compatible inside an LLM context.""" + + server, base = _serve() + try: + with use_llm_metadata({"lineageweave_post_id": "post-1"}): + body = post_json( + f"{base}/v1/project-histories", + {"contract_version": 1}, + headers={}, + timeout=2.0, + include_llm_metadata=False, + ) + finally: + server.shutdown() + + assert body["echo"] == {"contract_version": 1} + + +def test_post_json_rejects_a_response_above_the_contract_byte_limit() -> None: + """A bounded wire contract never buffers an oversized remote response.""" + + server, base = _serve() + try: + with pytest.raises(HttpClientError, match="response exceeds"): + post_json( + f"{base}/oversized", + {}, + headers={}, + timeout=2.0, + maximum_response_bytes=256, + ) + finally: + server.shutdown() + + +def test_tepp_request_rejects_payload_above_the_published_byte_limit() -> None: + """LineageWeave rejects oversized evidence before TEPP returns HTTP 400.""" + + request = _request() + template = request["events"][0] + request["events"] = [ + { + **template, + "event_id": f"event-{index}", + "source_post_id": f"post-{index}", + "evidence_text": "x" * 4096, + } + for index in range(128) + ] + request["focus_event_id"] = "event-0" + + with pytest.raises(TeppProjectHistoryUnavailable, match="request exceeds"): + validate_tepp_project_history_request(request) + + +def test_default_tepp_transport_disables_context_metadata(monkeypatch) -> None: + """The strict TEPP adapter opts out even when Ask sets LLM metadata.""" + + request = _request() + captured: dict[str, object] = {} + + def fake_post_json( + url, + payload, + *, + headers, + timeout, + include_llm_metadata, + maximum_response_bytes, + ): + captured.update( + url=url, + payload=deepcopy(payload), + headers=headers, + timeout=timeout, + include_llm_metadata=include_llm_metadata, + maximum_response_bytes=maximum_response_bytes, + ) + return _response(payload) + + monkeypatch.setattr(tepp_transport_module, "post_json", fake_post_json) + with use_llm_metadata({"lineageweave_post_id": "must-not-cross"}): + result = TeppProjectHistoryClient("https://tepp.example").project(request) + + assert result["inference_status"] == "temporal_association_only" + assert captured["include_llm_metadata"] is False + assert captured["maximum_response_bytes"] == 256 * 1024 + assert captured["payload"] == request + assert "metadata" not in captured["payload"] diff --git a/tests/test_tepp_client.py b/tests/test_tepp_client.py index ea87d5558..dd79930c7 100644 --- a/tests/test_tepp_client.py +++ b/tests/test_tepp_client.py @@ -52,6 +52,18 @@ def fake_transport(payload: dict) -> dict: assert received["snapshot_id"] == "demo-snapshot-1" +def test_custom_transport_provider_errors_are_not_exposed() -> None: + """Provider response text stays behind the stable unavailable error.""" + + def broken_transport(_payload: dict) -> dict: + raise RuntimeError("provider secret response body") + + with pytest.raises(TeppNotAvailable, match="transport request failed") as error: + TeppClient(transport=broken_transport).submit_analysis_run(_sample_request()) + + assert "provider secret" not in str(error.value) + + def test_configured_transport_sends_optional_bearer_key(monkeypatch: pytest.MonkeyPatch) -> None: received = {} @@ -66,3 +78,22 @@ def fake_post_json(url: str, payload: dict, *, headers: dict, timeout: float) -> assert received["headers"] == {"authorization": "Bearer test-key"} assert received["payload"] == _sample_request().to_json() + + +def test_configured_transport_provider_errors_are_not_exposed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The configured provider boundary does not return raw transport text.""" + + def broken_post_json(*args, **kwargs): + del args, kwargs + raise RuntimeError("provider secret response body") + + monkeypatch.setattr("backend.app.analysis_run_start.post_json", broken_post_json) + + with pytest.raises(TeppNotAvailable, match="transport request failed") as error: + configured_tepp_client("https://tepp.example/v1/analysis-runs").submit_analysis_run( + _sample_request() + ) + + assert "provider secret" not in str(error.value) diff --git a/tests/test_tepp_project_history_recovery.py b/tests/test_tepp_project_history_recovery.py new file mode 100644 index 000000000..bb260517a --- /dev/null +++ b/tests/test_tepp_project_history_recovery.py @@ -0,0 +1,408 @@ +"""Regression contracts for the recovered TEPP project-history integration.""" + +from __future__ import annotations + +import asyncio +import json +from copy import deepcopy +from types import SimpleNamespace + +import pytest + +from backend.app import main +from backend.app.auth import CurrentAccount +from backend.app.tepp_project_history import ( + build_tepp_project_history_request, + tenant_workspace_reference, + validate_project_history_with_tepp, +) +from lineageweave.tepp_project_history import ( + TeppProjectHistoryClient, + TeppProjectHistoryInvalidResponse, + TeppProjectHistoryUnavailable, + parse_rfc3339_utc, +) + + +def _canonical_projection() -> dict[str, object]: + """Return one synthetic authorized LineageWeave project history.""" + + return { + "contract_version": 1, + "project_key": "P-100", + "normalized_project_key": "p-100", + "project_name": "Synthetic transformer renewal", + "focus_event_id": "00000000-0000-4000-8000-000000000003", + "time_basis_code": "source_post_created_at_fallback", + "knowledge_cutoff": "2026-08-20T12:00:00+00:00", + "evidence_boundary_code": "authorized_visible_source_posts", + "event_count": 3, + "distinct_actor_count": 2, + "distinct_observed_actor_count": 1, + "truncated": False, + "events": [ + { + "event_id": "00000000-0000-4000-8000-000000000001", + "source_post_id": "00000000-0000-4000-8000-000000000001", + "event_title": "Synthetic contract awarded", + "event_type_code": "contract_awarded", + "event_type_basis_code": "display_classification", + "occurred_at": "2022-03-11T09:00:00Z", + "time_basis_code": "source_post_created_at_fallback", + "voc_type_code": None, + "source_stage_code": "award", + "source_detail_state_code": None, + "project_matches": [], + "responsibility_evidence": [ + { + "actor_key": "text:prov_person\u001fsynthetic owner\u001fdemo org", + "actor_name": "Synthetic Owner", + "actor_type_code": "prov_person", + "affiliated_organization_name": "Demo Org", + "responsibility": "Source author", + "truth_status_code": "observed", + "provenance": "source_post.source_author", + } + ], + "observed_responsibilities": [], + "responsibility_transition_code": None, + "responsibility_transition_truth_status_code": None, + "related_prior_paths": [], + }, + { + "event_id": "00000000-0000-4000-8000-000000000002", + "source_post_id": "00000000-0000-4000-8000-000000000002", + "event_title": "Synthetic specification changed", + "event_type_code": "specification_changed", + "event_type_basis_code": "display_classification", + "occurred_at": "2023-06-15T09:00:00Z", + "time_basis_code": "source_post_created_at_fallback", + "voc_type_code": None, + "source_stage_code": "spec_change", + "source_detail_state_code": None, + "project_matches": [], + "responsibility_evidence": [ + { + "actor_key": "person:synthetic-pm", + "actor_name": "Synthetic PM", + "actor_type_code": "prov_person", + "affiliated_organization_name": "Demo Org", + "responsibility": "Coordinate change", + "truth_status_code": "inferred", + "provenance": "post_summary_role", + } + ], + "observed_responsibilities": [], + "responsibility_transition_code": "handoff", + "responsibility_transition_truth_status_code": "inferred", + "related_prior_paths": [], + }, + { + "event_id": "00000000-0000-4000-8000-000000000003", + "source_post_id": "00000000-0000-4000-8000-000000000003", + "event_title": "Synthetic VOC received", + "event_type_code": "voc_received", + "event_type_basis_code": "display_classification", + "occurred_at": "2026-02-02T09:00:00Z", + "time_basis_code": "source_post_created_at_fallback", + "voc_type_code": "voc", + "source_stage_code": None, + "source_detail_state_code": None, + "project_matches": [], + "responsibility_evidence": [], + "observed_responsibilities": [], + "responsibility_transition_code": "assignment_gap", + "responsibility_transition_truth_status_code": "inferred", + "related_prior_paths": [], + }, + ], + } + + +def _tepp_response(request: dict[str, object]) -> dict[str, object]: + """Return the exact TEPP #159 response shape for a validated request.""" + + events = sorted( + deepcopy(request["events"]), + key=lambda event: ( + parse_rfc3339_utc(event["occurred_at"], "occurred_at")[0], + event["event_id"], + ), + ) + actors = {actor for event in events for actor in event["actor_ids"]} + return { + "contract_version": 1, + "project_key": request["project_key"], + "project_name": request["project_name"], + "focus_event_id": request["focus_event_id"], + "knowledge_cutoff": request["knowledge_cutoff"], + "history_span_start": events[0]["occurred_at"], + "history_span_end": events[-1]["occurred_at"], + "participant_count": len(actors), + "inference_status": "temporal_association_only", + "events": events, + "findings": [ + { + "finding_code": "specification_change_before_focus", + "summary": "An explicit specification-change event precedes the focus event.", + "related_event_ids": [events[1]["event_id"]], + "evidence_post_ids": [events[1]["source_post_id"]], + } + ], + } + + +def test_mapper_uses_opaque_actor_references_and_bounded_source_evidence() -> None: + projection = _canonical_projection() + workspace = tenant_workspace_reference(["tenant-b", "tenant-a"]) + + request = build_tepp_project_history_request( + projection=projection, + tenant_workspace_id=workspace, + ) + encoded = json.dumps(request, ensure_ascii=False) + + assert workspace == tenant_workspace_reference(["tenant-a", "tenant-b"]) + assert "Synthetic Owner" not in encoded + assert "Synthetic PM" not in encoded + assert "Demo Org" not in encoded + assert all( + actor.startswith("lw-actor-") + for event in request["events"] + for actor in event["actor_ids"] + ) + assert request["events"][0]["available_at"] == request["events"][0]["occurred_at"] + assert request["events"][0]["evidence_text"].startswith("Synthetic contract awarded") + + +def test_mapper_and_tepp_validation_order_fractional_seconds_by_instant() -> None: + """Events in the same second retain chronological rather than text order.""" + + projection = _canonical_projection() + projection["events"][0]["occurred_at"] = "2022-03-11T09:00:00.500Z" + projection["events"][1]["occurred_at"] = "2022-03-11T09:00:00Z" + request = build_tepp_project_history_request( + projection=projection, + tenant_workspace_id=tenant_workspace_reference(["tenant-a"]), + ) + + assert [event["event_id"] for event in request["events"]] == [ + "00000000-0000-4000-8000-000000000002", + "00000000-0000-4000-8000-000000000001", + "00000000-0000-4000-8000-000000000003", + ] + + client = TeppProjectHistoryClient( + "https://tepp.example", + transport=lambda url, payload, headers, timeout: _tepp_response(payload), + ) + result = client.project(request) + assert [event["event_id"] for event in result["events"]] == [ + "00000000-0000-4000-8000-000000000002", + "00000000-0000-4000-8000-000000000001", + "00000000-0000-4000-8000-000000000003", + ] + + +@pytest.mark.parametrize("timestamp", ["2026-08-20 12:00:00Z", "2026-08-20T12:00:00+0900"]) +def test_mapper_rejects_non_rfc3339_timestamp_shapes(timestamp: str) -> None: + projection = _canonical_projection() + projection["knowledge_cutoff"] = timestamp + + with pytest.raises(TeppProjectHistoryUnavailable, match="RFC 3339"): + build_tepp_project_history_request( + projection=projection, + tenant_workspace_id=tenant_workspace_reference(["tenant-a"]), + ) + + +def test_strict_client_accepts_tepp_159_and_rejects_authority_or_evidence_drift() -> None: + request = build_tepp_project_history_request( + projection=_canonical_projection(), + tenant_workspace_id=tenant_workspace_reference(["tenant-a"]), + ) + captured: dict[str, object] = {} + + def transport(url, payload, headers, timeout): + captured.update(url=url, payload=payload, headers=headers, timeout=timeout) + return _tepp_response(payload) + + client = TeppProjectHistoryClient("https://tepp.example", transport=transport) + result = client.project(request) + + assert result["inference_status"] == "temporal_association_only" + assert captured["url"] == "https://tepp.example/v1/project-histories" + assert "authorization" not in {key.lower() for key in captured["headers"]} + assert captured["headers"]["tepp-consumer"] == "lineageweave" + + def causal_transport(url, payload, headers, timeout): + del url, headers, timeout + response = _tepp_response(payload) + response["inference_status"] = "causal" + return response + + with pytest.raises(TeppProjectHistoryInvalidResponse): + TeppProjectHistoryClient( + "https://tepp.example", transport=causal_transport + ).project(request) + + def changed_evidence_transport(url, payload, headers, timeout): + del url, headers, timeout + response = _tepp_response(payload) + response["events"][0]["evidence_text"] = "changed" + return response + + with pytest.raises(TeppProjectHistoryInvalidResponse): + TeppProjectHistoryClient( + "https://tepp.example", transport=changed_evidence_transport + ).project(request) + + +def test_strict_client_rejects_unknown_or_duplicate_finding_references() -> None: + request = build_tepp_project_history_request( + projection=_canonical_projection(), + tenant_workspace_id=tenant_workspace_reference(["tenant-a"]), + ) + + def unknown_finding_transport(url, payload, headers, timeout): + del url, headers, timeout + response = _tepp_response(payload) + response["findings"][0]["finding_code"] = "provider_authored_conclusion" + return response + + with pytest.raises(TeppProjectHistoryInvalidResponse): + TeppProjectHistoryClient( + "https://tepp.example", transport=unknown_finding_transport + ).project(request) + + def duplicate_reference_transport(url, payload, headers, timeout): + del url, headers, timeout + response = _tepp_response(payload) + event_id = response["findings"][0]["related_event_ids"][0] + response["findings"][0]["related_event_ids"] = [event_id, event_id] + return response + + with pytest.raises(TeppProjectHistoryInvalidResponse): + TeppProjectHistoryClient( + "https://tepp.example", transport=duplicate_reference_transport + ).project(request) + + +def test_strict_client_normalizes_raw_provider_errors() -> None: + request = build_tepp_project_history_request( + projection=_canonical_projection(), + tenant_workspace_id=tenant_workspace_reference(["tenant-a"]), + ) + + def provider_failure(url, payload, headers, timeout): + del url, payload, headers, timeout + raise RuntimeError("provider stack trace must not cross the boundary") + + with pytest.raises(TeppProjectHistoryUnavailable, match="request failed") as error: + TeppProjectHistoryClient( + "https://tepp.example", transport=provider_failure + ).project(request) + + assert "provider stack trace" not in str(error.value) + + +def test_validation_fails_closed_without_hiding_canonical_history(monkeypatch) -> None: + projection = _canonical_projection() + unconfigured = validate_project_history_with_tepp( + projection=projection, + tenant_workspace_id=tenant_workspace_reference([]), + transport_url="", + ) + assert unconfigured == { + "status": "not_configured", + "project_history": None, + "next_action_code": "configure_tepp_project_history", + } + + def broken_project(self, request): + del self, request + raise TeppProjectHistoryUnavailable("synthetic outage") + + monkeypatch.setattr(TeppProjectHistoryClient, "project", broken_project) + unavailable = validate_project_history_with_tepp( + projection=projection, + tenant_workspace_id=tenant_workspace_reference([]), + transport_url="https://tepp.example", + ) + assert unavailable["status"] == "unavailable" + assert projection["event_count"] == 3 + + def invalid_project(self, request): + del self, request + raise TeppProjectHistoryInvalidResponse("synthetic invalid response") + + monkeypatch.setattr(TeppProjectHistoryClient, "project", invalid_project) + invalid = validate_project_history_with_tepp( + projection=projection, + tenant_workspace_id=tenant_workspace_reference([]), + transport_url="https://tepp.example", + ) + assert invalid["status"] == "invalid_evidence" + assert projection["event_count"] == 3 + + +class _Acquire: + async def __aenter__(self) -> object: + return object() + + async def __aexit__(self, exc_type, exc_value, traceback) -> None: + return None + + +class _Pool: + def acquire(self) -> _Acquire: + return _Acquire() + + +def test_project_history_route_attaches_validation_to_the_canonical_projection(monkeypatch) -> None: + projection = _canonical_projection() + captured: dict[str, object] = {} + + async def fake_projection(connection, **kwargs): + del connection, kwargs + return deepcopy(projection) + + def fake_validate(**kwargs): + captured.update(kwargs) + return { + "status": "validated", + "project_history": {"inference_status": "temporal_association_only"}, + "next_action_code": "open_source_evidence", + } + + monkeypatch.setattr(main, "fetch_project_history_projection", fake_projection) + monkeypatch.setattr(main, "validate_project_history_with_tepp", fake_validate) + monkeypatch.setattr( + main, + "load_settings", + lambda: SimpleNamespace(tepp_transport_url="https://tepp.example"), + ) + account = CurrentAccount( + user_account_id="account-1", + external_subject_id="subject-1", + display_name="Synthetic analyst", + preferred_locale="en", + corporate_entity_ids=frozenset({"tenant-a"}), + permission_codes=frozenset({"post_read"}), + ) + + result = asyncio.run( + main.read_project_history( + project_key="P-100", + focus_post_id=None, + knowledge_cutoff="2026-08-20T12:00:00+00:00", + limit=64, + account=account, + pool=_Pool(), + ) + ) + + assert result["events"] == projection["events"] + assert result["tepp_validation"]["status"] == "validated" + assert captured["projection"]["project_key"] == "P-100" + assert captured["transport_url"] == "https://tepp.example" diff --git a/uv.lock b/uv.lock index 52a3376dd..fe7edf1ef 100644 --- a/uv.lock +++ b/uv.lock @@ -454,7 +454,7 @@ wheels = [ [[package]] name = "lineageweave" -version = "2.18.0" +version = "2.20.0" source = { editable = "." } dependencies = [ { name = "certifi" },