diff --git a/Makefile b/Makefile index 62e1b3198..b348e7787 100644 --- a/Makefile +++ b/Makefile @@ -35,4 +35,5 @@ seed: load-http: @test -n "$${LINEAGEWEAVE_VUS:-}" || { echo "LINEAGEWEAVE_VUS is required" >&2; exit 1; } @test -n "$${LINEAGEWEAVE_DURATION:-}" || { echo "LINEAGEWEAVE_DURATION is required" >&2; exit 1; } - k6 run --vus "$${LINEAGEWEAVE_VUS}" --duration "$${LINEAGEWEAVE_DURATION}" scripts/k6_http_e2e.js + @test -n "$${LINEAGEWEAVE_REQUEST_TIMEOUT:-}" || { echo "LINEAGEWEAVE_REQUEST_TIMEOUT is required" >&2; exit 1; } + k6 run -e REQUEST_TIMEOUT="$${LINEAGEWEAVE_REQUEST_TIMEOUT}" --vus "$${LINEAGEWEAVE_VUS}" --duration "$${LINEAGEWEAVE_DURATION}" scripts/k6_http_e2e.js diff --git a/backend/app/config.py b/backend/app/config.py index 4dba383e8..827441648 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -59,6 +59,7 @@ class Settings: valkey_url: str searxng_base_url: str tepp_transport_url: str + tepp_api_key: str caldav_base_url: str naruon_calendar_base_url: str naruon_calendar_service_token: str @@ -171,6 +172,7 @@ def load_settings() -> Settings: valkey_url=os.environ.get("VALKEY_URL", "redis://localhost:16379/0"), searxng_base_url=os.environ.get("SEARXNG_BASE_URL", ""), tepp_transport_url=os.environ.get("TEPP_TRANSPORT_URL", ""), + tepp_api_key=os.environ.get("TEPP_API_KEY", ""), caldav_base_url=os.environ.get("CALDAV_BASE_URL", "").strip(), naruon_calendar_base_url=os.environ.get("NARUON_CALENDAR_BASE_URL", "").strip(), naruon_calendar_service_token=os.environ.get( diff --git a/backend/app/global_ask_queue.py b/backend/app/global_ask_queue.py index c7e570d81..0f4ab8455 100644 --- a/backend/app/global_ask_queue.py +++ b/backend/app/global_ask_queue.py @@ -46,6 +46,7 @@ _seoul_today, cited_post_images, gather_global_chat_sources, + prepare_global_question_embedding, ) GLOBAL_ASK_STREAM_KEY = "global_ask_request_stream" @@ -237,6 +238,9 @@ def can_see(row: asyncpg.Record) -> bool: today = _seoul_today() try: + question_embedding = await prepare_global_question_embedding( + question_text, embedding_client or NullEmbeddingClient() + ) async with pool.acquire() as conn: sources = await gather_global_chat_sources( conn, @@ -244,8 +248,9 @@ def can_see(row: asyncpg.Record) -> bool: corporate_entity_ids, process_unit_ids, question=question_text, + question_embedding=question_embedding, today=today, - embedding_client=embedding_client, + embedding_client=NullEmbeddingClient(), ) except Exception as exc: log_internal_fault("global_ask", exc) diff --git a/backend/app/lineage_ingestion.py b/backend/app/lineage_ingestion.py index ee240cd34..3775408a3 100644 --- a/backend/app/lineage_ingestion.py +++ b/backend/app/lineage_ingestion.py @@ -14,7 +14,7 @@ import math import re from collections import defaultdict -from collections.abc import Mapping +from collections.abc import Mapping, Sequence from datetime import datetime from typing import Any @@ -525,6 +525,40 @@ async def _fetch_visible_lineage_rows(conn: asyncpg.Connection, can_see_post): return visible_all, edge_rows +async def _fetch_lineage_landing_rows( + conn: asyncpg.Connection, + corporate_entity_ids: Sequence[str], + process_unit_ids: Sequence[str], + limit: int, +): + """Fetch only the authorized, bounded landing projection in PostgreSQL.""" + posts = await conn.fetch( + "select post_id, post_title, voc_type_code, visibility_code, " + "corporate_entity_id, process_unit_id, thread_group_key, created_at " + "from source_post where " + f"{SOURCE_POST_ELIGIBILITY_SQL.format(alias='source_post')} and " + "(visibility_code = 'public' or (corporate_entity_id::text = any($1::text[]) " + "and (cardinality($2::text[]) = 0 or process_unit_id::text = any($2::text[])))) " + "order by created_at desc, post_id desc limit $3", + list(corporate_entity_ids), + list(process_unit_ids), + limit + 1, + ) + visible = list(posts[:limit]) + visible_ids = [str(row["post_id"]) for row in visible] + edge_rows = ( + await conn.fetch( + "select parent_post_id, child_post_id, fused_score, interval_relation_code " + "from post_lineage_edge where parent_post_id = any($1::uuid[]) " + "and child_post_id = any($1::uuid[])", + visible_ids, + ) + if visible_ids + else [] + ) + return visible, edge_rows, len(posts) > limit + + def _undirected_neighbors(edge_rows) -> dict[str, set[str]]: neighbors: dict[str, set[str]] = {} for edge in edge_rows: @@ -658,6 +692,8 @@ async def visible_lineage_graph( limit: int = _LINEAGE_GRAPH_NODE_LIMIT, focus_post_id: str | None = None, include_isolated: bool = False, + corporate_entity_ids: Sequence[str] | None = None, + process_unit_ids: Sequence[str] = (), ) -> dict[str, Any]: """ABAC-filtered graph bounded for the browser's initial viewport. @@ -665,16 +701,22 @@ async def visible_lineage_graph( individual posts for complete lineage, while this landing projection keeps only the newest ``limit`` visible nodes and edges between them. """ - visible_all, edge_rows = await _fetch_visible_lineage_rows(conn, can_see_post) + if focus_post_id is None and corporate_entity_ids is not None: + visible, edge_rows, truncated = await _fetch_lineage_landing_rows( + conn, corporate_entity_ids, process_unit_ids, limit + ) + visible_all = visible + else: + visible_all, edge_rows = await _fetch_visible_lineage_rows(conn, can_see_post) - if focus_post_id is None: + if focus_post_id is None and corporate_entity_ids is None: visible = sorted( visible_all, key=lambda row: (row["created_at"], str(row["post_id"])), reverse=True, )[:limit] truncated = len(visible_all) > len(visible) - else: + elif focus_post_id is not None: focus_id = str(focus_post_id) neighbors = _undirected_neighbors(edge_rows) allowed = {str(row["post_id"]) for row in visible_all} diff --git a/backend/app/main.py b/backend/app/main.py index b53907977..bae514dc3 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -148,7 +148,7 @@ require_summary_source_body, ) from backend.app.ranking_ingestion import load_visible_ranking_posts -from backend.app.relation_verification_ingestion import verify_post_relations +from backend.app.relation_verification_ingestion import verify_post_relations_from_pool from backend.app.report_ingestion import ( GROUPING_KINDS, fetch_period_comparison, @@ -1251,6 +1251,8 @@ async def read_lineage_graph( lambda row: _can_see_post(account, row), limit=limit, focus_post_id=post_id, + corporate_entity_ids=account.corporate_entity_ids, + process_unit_ids=account.process_unit_ids, ) @@ -2310,28 +2312,25 @@ async def verify_post_entity_relationships( status.HTTP_503_SERVICE_UNAVAILABLE, "Relation verification is unavailable: set SEARXNG_BASE_URL", ) - async with pool.acquire() as conn: - try: - verified = await verify_post_relations( - conn, - client, - post_id, - visible_corporate_entity_ids=account.corporate_entity_ids, - ) - except (HttpClientError, OSError) as exc: - # verify_post_relations() deliberately raises on a failed search - # (a failed search is not "searched and found nothing" -- see - # its docstring); this is the one caller, so it is the right - # place to turn that into a clean 503 instead of a raw 500. - raise HTTPException( - status.HTTP_503_SERVICE_UNAVAILABLE, - "Relation verification is unavailable: the search provider did not respond", - ) from exc - except Exception as exc: # noqa: BLE001 - provider boundary is fail-closed. - raise HTTPException( - status.HTTP_503_SERVICE_UNAVAILABLE, - "Relation verification is unavailable: the search provider did not respond", - ) from exc + try: + verified = await verify_post_relations_from_pool( + pool, + client, + post_id, + visible_corporate_entity_ids=account.corporate_entity_ids, + ) + except (HttpClientError, OSError) as exc: + # A failed search is not "searched and found nothing"; turn the + # provider failure into a clean 503 rather than persisting a miss. + raise HTTPException( + status.HTTP_503_SERVICE_UNAVAILABLE, + "Relation verification is unavailable: the search provider did not respond", + ) from exc + except Exception as exc: # noqa: BLE001 - provider boundary is fail-closed. + raise HTTPException( + status.HTTP_503_SERVICE_UNAVAILABLE, + "Relation verification is unavailable: the search provider did not respond", + ) from exc await publish_activity_event( valkey, post_id, @@ -3386,7 +3385,12 @@ async def derive_post_commitment( # Friday" in a January post must resolve to that January, not to the # Friday after the operator clicked Derive. reference_date = post["created_at"].date().isoformat() - commitment = client.extract(post["post_title"], normalized_body, reference_date) + commitment = await asyncio.to_thread( + client.extract, + post["post_title"], + normalized_body, + reference_date, + ) except (HttpClientError, KeyError, OSError, TypeError, ValueError, RuntimeError) as exc: raise HTTPException( status.HTTP_503_SERVICE_UNAVAILABLE, @@ -3600,7 +3604,8 @@ async def read_calendar( settings = load_settings() if window_start is None or window_end is None: window_start, window_end = default_calendar_window(datetime.now(timezone.utc)) - naruon = load_observed_calendar_events( + naruon = await asyncio.to_thread( + load_observed_calendar_events, build_workspace_naruon_client( settings.naruon_calendar_base_url, settings.naruon_calendar_service_token, diff --git a/backend/app/post_chat_ingestion.py b/backend/app/post_chat_ingestion.py index 6d2c4fd6b..955d88f5a 100644 --- a/backend/app/post_chat_ingestion.py +++ b/backend/app/post_chat_ingestion.py @@ -18,6 +18,7 @@ from __future__ import annotations import asyncio +import math from dataclasses import dataclass from datetime import date, datetime from typing import Any, Callable, Iterable @@ -36,6 +37,7 @@ random_walk_with_restart, select_related_nodes, ) +from lineageweave.ontology import all_declared_lookup_codes, ontology_annotations from lineageweave.post_chat import ( CANONICAL_CHAT_QUESTION, CANONICAL_COMMITMENT_QUESTION, @@ -44,11 +46,11 @@ normalize_chat_question, ) from lineageweave.post_content_normalization import normalize_post_body +from lineageweave.rankweave_client import RankWeaveNotAvailable, build_rankweave_client from lineageweave.temporal_expressions import resolve_korean_relative_time from .knowledge_graph import hydrate_related_nodes, load_visible_subgraph from .post_eligibility import SOURCE_POST_ELIGIBILITY_SQL -from lineageweave.ontology import ontology_annotations @dataclass(frozen=True) @@ -417,6 +419,69 @@ async def gather_chat_sources( return sources +async def prepare_global_question_embedding( + question: str, + embedding_client: EmbeddingClient, +) -> tuple[list[float], str, float] | None: + """Resolve one question embedding without holding a database connection.""" + if not question.strip() or not embedding_client.available: + return None + try: + question_vector = await asyncio.to_thread(embedding_client.embed, question) + except (OSError, RuntimeError, ValueError): + return None + return _validated_question_embedding( + question_vector, embedding_client.resolved_model + ) + + +def _validated_question_embedding( + question_vector: list[float], embedding_model_code: str | None +) -> tuple[list[float], str, float] | None: + """Return a finite, non-zero embedding envelope or fail closed.""" + if ( + not question_vector + or not embedding_model_code + or any(not math.isfinite(value) for value in question_vector) + ): + return None + question_norm = math.sqrt(sum(value * value for value in question_vector)) + if not math.isfinite(question_norm) or question_norm == 0.0: + return None + return question_vector, embedding_model_code, question_norm + + +def _ontology_lookup_codes_in_question(question: str) -> list[str]: + """Return ontology lookup codes whose complete canonical IRI is cited.""" + folded_question = question.casefold() + matched: list[str] = [] + for lookup_code in sorted(all_declared_lookup_codes()): + ontology_iri = ontology_annotations(lookup_code).get("ontology_iri") + if ontology_iri and ontology_iri.casefold() in folded_question: + matched.append(lookup_code) + return matched + + +def _fuse_global_candidate_ids( + embedding_ids: list[str], evidence_ids: list[str], limit: int +) -> list[str]: + """Fuse two owned rank lists with RankWeave parameter-free RRF.""" + if not embedding_ids: + return evidence_ids[:limit] + if not evidence_ids: + return embedding_ids[:limit] + channels = {"embedding": embedding_ids, "evidence": evidence_ids} + titles_by_id = { + post_id: post_id + for post_id in dict.fromkeys([*embedding_ids, *evidence_ids]) + } + try: + fused = build_rankweave_client().fuse_rankings(channels, titles_by_id) + except RankWeaveNotAvailable: + return embedding_ids[:limit] + return [item.post_id for item in fused.items[:limit]] + + async def gather_global_chat_sources( conn: asyncpg.Connection, can_see_post: Callable[[asyncpg.Record], bool], @@ -426,6 +491,7 @@ async def gather_global_chat_sources( embedding_client: EmbeddingClient | None = None, *, question: str | None = None, + question_embedding: tuple[list[float], str, float] | None = None, limit: int = 4, today: date | None = None, ) -> list[ChatSourceDocument]: @@ -444,11 +510,9 @@ async def gather_global_chat_sources( or no expression at all applies no date filter. Cited sources name which clock matched (ADR 0202). - Candidates are ranked by the maximum cosine similarity between the - question embedding and each post's persisted semantic-unit embeddings. - The embedding model and dimension must match exactly. An unavailable - channel or incomplete persisted vectors returns no source instead of - falling back to lexical matching. + Embedding candidates use maximum cosine similarity with exact model and + dimension agreement. Persisted semantic/KG evidence remains available + when that channel is unavailable; title/body lexical fallback does not. """ if limit <= 0: return [] @@ -459,20 +523,26 @@ async def gather_global_chat_sources( resolved_time_range = resolve_korean_relative_time( question or "", today=today or _seoul_today() ) - if not (question and question.strip() and embedding_client.available): + if not (question and question.strip()): return [] - try: - question_vector = await asyncio.to_thread(embedding_client.embed, question) - except (OSError, RuntimeError, ValueError): - return [] - if not question_vector: - return [] - embedding_model_code = embedding_client.resolved_model - if not embedding_model_code: - return [] - question_norm = sum(value * value for value in question_vector) ** 0.5 - if question_norm == 0.0: + supplied_question_embedding = question_embedding is not None + if question_embedding is None: + question_embedding = await prepare_global_question_embedding( + question, embedding_client + ) + validated_embedding = ( + _validated_question_embedding(question_embedding[0], question_embedding[1]) + if question_embedding is not None + else None + ) + embedding_enabled = validated_embedding is not None + if supplied_question_embedding and not embedding_enabled: return [] + question_vector, embedding_model_code, question_norm = validated_embedding or ( + [], + "", + 1.0, + ) # Safe SQL: the only interpolation is the repository-owned eligibility # expression; all request and model values remain asyncpg parameters. candidate_rows = await conn.fetch( # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli @@ -496,7 +566,8 @@ async def gather_global_chat_sources( on value.post_content_embedding_id = embedding.post_content_embedding_id join question_vector question on question.dimension_index = value.dimension_index - where embedding.embedding_model_code = $3 + where $11::boolean + and embedding.embedding_model_code = $3 and embedding.embedding_dimension_count = cardinality($1::double precision[]) and (post.visibility_code = 'public' or (post.corporate_entity_id::text = any($4::text[]) @@ -507,14 +578,145 @@ async def gather_global_chat_sources( and ($7::date is null or (coalesce(post.event_occurred_at, post.created_at) at time zone 'Asia/Seoul')::date <= $7) group by unit.post_id, embedding.post_content_embedding_id having count(*) = cardinality($1::double precision[]) + ), embedding_candidates as ( + select similarity.post_id, + max(similarity.cosine_similarity) as semantic_score, + max(coalesce(post.event_occurred_at, post.created_at)) as event_clock + from unit_similarity similarity + join source_post post on post.post_id = similarity.post_id + group by similarity.post_id + order by semantic_score desc, event_clock desc, similarity.post_id desc + limit $8 + ), evidence_query as ( + select websearch_to_tsquery('simple', $9) as terms + ), matching_nodes as ( + select 'node_person'::text as node_type_code, person.person_id as node_id + from cataloged_person person, evidence_query query + where to_tsvector( + 'simple', + coalesce(person.person_name, '') || ' ' || + coalesce(person.last_known_job_title, '') + ) @@ query.terms + union + select 'node_corporate_entity', entity.corporate_entity_id + from corporate_entity entity, evidence_query query + where to_tsvector( + 'simple', + coalesce(entity.corporate_entity_code, '') || ' ' || + coalesce(entity.entity_name, '') + ) @@ query.terms + union + select 'node_team', team.team_id + from cataloged_team team, evidence_query query + where to_tsvector( + 'simple', + coalesce(team.team_name, '') || ' ' || + coalesce(team.affiliated_organization_name, '') + ) @@ query.terms + union + select 'node_post', endpoint.post_id + from source_post endpoint, evidence_query query + where to_tsvector('simple', coalesce(endpoint.post_title, '')) @@ query.terms + and (endpoint.visibility_code = 'public' + or (endpoint.corporate_entity_id::text = any($4::text[]) + and (cardinality($5::text[]) = 0 + or endpoint.process_unit_id::text = any($5::text[])))) + and {SOURCE_POST_ELIGIBILITY_SQL.format(alias='endpoint')} + ), matching_edges as ( + select edge.knowledge_graph_edge_id + from knowledge_graph_edge edge + join common_lookup_value lookup + on lookup.lookup_code = edge.edge_type_code + cross join evidence_query query + where to_tsvector( + 'simple', + coalesce(lookup.lookup_code, '') || ' ' || + coalesce(lookup.lookup_label, '') + ) @@ query.terms + union + select edge.knowledge_graph_edge_id + from knowledge_graph_edge edge + where edge.edge_type_code = any($10::text[]) + union + select edge.knowledge_graph_edge_id + from knowledge_graph_edge edge + join matching_nodes node + on node.node_type_code = edge.source_node_type_code + and node.node_id = edge.source_node_id + union + select edge.knowledge_graph_edge_id + from knowledge_graph_edge edge + join matching_nodes node + on node.node_type_code = edge.target_node_type_code + and node.node_id = edge.target_node_id + ), evidence_post_candidates as ( + select project.post_id + from post_project_mention project, evidence_query query + where to_tsvector( + 'simple', + coalesce(project.project_name, '') || ' ' || + coalesce(project.evidence_text, '') || ' ' || + coalesce(project.ontology_iri, '') + ) @@ query.terms + union + select role.post_id + from post_summary_role role, evidence_query query + where to_tsvector( + 'simple', + coalesce(role.actor_name, '') || ' ' || + coalesce(role.responsibility, '') || ' ' || + coalesce(role.affiliated_organization_name, '') + ) @@ query.terms + union + select mention.post_id + from combined_post_person_mention mention + join cataloged_person person on person.person_id = mention.person_id + cross join evidence_query query + where to_tsvector( + 'simple', + coalesce(person.person_name, '') || ' ' || + coalesce(person.last_known_job_title, '') + ) @@ query.terms + union + select mention.post_id + from combined_post_person_mention mention + join person_affiliation affiliation + on affiliation.person_id = mention.person_id + cross join evidence_query query + where to_tsvector( + 'simple', + coalesce(affiliation.affiliated_organization_name, '') || ' ' || + coalesce(affiliation.role_title, '') + ) @@ query.terms + union + select evidence.evidence_post_id + from matching_edges edge + join knowledge_graph_edge_evidence evidence + on evidence.knowledge_graph_edge_id = edge.knowledge_graph_edge_id + ), authorized_evidence_candidates as ( + select candidate.post_id, + max(coalesce(post.event_occurred_at, post.created_at)) as event_clock + from evidence_post_candidates candidate + join source_post post on post.post_id = candidate.post_id + where (post.visibility_code = 'public' + or (post.corporate_entity_id::text = any($4::text[]) + and (cardinality($5::text[]) = 0 + or post.process_unit_id::text = any($5::text[])))) + and {SOURCE_POST_ELIGIBILITY_SQL.format(alias='post')} + and ($6::date is null or (coalesce(post.event_occurred_at, post.created_at) at time zone 'Asia/Seoul')::date >= $6) + and ($7::date is null or (coalesce(post.event_occurred_at, post.created_at) at time zone 'Asia/Seoul')::date <= $7) + group by candidate.post_id + order by event_clock desc, candidate.post_id desc + limit $8 ) - select similarity.post_id, max(similarity.cosine_similarity) as semantic_score, - max(coalesce(post.event_occurred_at, post.created_at)) as event_clock - from unit_similarity similarity - join source_post post on post.post_id = similarity.post_id - group by similarity.post_id - order by semantic_score desc, event_clock desc, similarity.post_id desc - limit $8 + select 'embedding'::text as candidate_channel, post_id, + row_number() over (order by semantic_score desc, event_clock desc, post_id desc) as channel_rank + from embedding_candidates + union all + select 'evidence', post_id, + row_number() over (order by event_clock desc, post_id desc) as channel_rank + from authorized_evidence_candidates + order by candidate_channel, channel_rank """, question_vector, question_norm, @@ -524,8 +726,25 @@ async def gather_global_chat_sources( resolved_time_range[0] if resolved_time_range else None, resolved_time_range[1] if resolved_time_range else None, limit, + question, + _ontology_lookup_codes_in_question(question), + embedding_enabled, + ) + embedding_candidate_ids: list[str] = [] + evidence_candidate_ids: list[str] = [] + for row in candidate_rows: + channel = ( + str(row["candidate_channel"]) + if "candidate_channel" in row + else "embedding" + ) + target = ( + evidence_candidate_ids if channel == "evidence" else embedding_candidate_ids + ) + target.append(str(row["post_id"])) + candidate_ids = _fuse_global_candidate_ids( + embedding_candidate_ids, evidence_candidate_ids, limit ) - candidate_ids = [str(row["post_id"]) for row in candidate_rows] candidate_id_set = frozenset(candidate_ids) # One semantic match is still only one event snapshot. Expand the diff --git a/backend/app/relation_verification_ingestion.py b/backend/app/relation_verification_ingestion.py index ad93729a4..10dcbcf3a 100644 --- a/backend/app/relation_verification_ingestion.py +++ b/backend/app/relation_verification_ingestion.py @@ -8,6 +8,7 @@ from __future__ import annotations +import asyncio from collections.abc import Sequence from dataclasses import dataclass @@ -26,6 +27,13 @@ class VerifiedRelation: verification_evidence_post_id: str | None +@dataclass(frozen=True) +class _PendingRelation: + counterparty_entity_name: str + relationship_label: str + internal_evidence_post_id: str | None + + async def _find_internal_evidence_post( conn: asyncpg.Connection, post_id: str, @@ -124,7 +132,11 @@ async def verify_post_relations( row["relationship_label"], visible_corporate_entity_ids, ) - result = client.verify(row["counterparty_entity_name"], row["relationship_label"]) + result = await asyncio.to_thread( + client.verify, + row["counterparty_entity_name"], + row["relationship_label"], + ) await conn.execute( """ update post_counterparty_entity @@ -149,3 +161,73 @@ async def verify_post_relations( ) ) return verified + + +async def verify_post_relations_from_pool( + pool: asyncpg.Pool, + client: RelationVerificationClient, + post_id: str, + visible_corporate_entity_ids: Sequence[str] = (), +) -> list[VerifiedRelation]: + """Verify relations without reserving a DB connection during web I/O.""" + async with pool.acquire() as conn: + rows = await conn.fetch( + """ + select c.counterparty_entity_name, v.lookup_label as relationship_label + from post_counterparty_entity c + join common_lookup_value v on v.lookup_code = c.relationship_type_code + where c.post_id = $1 and c.verification_status_code = 'verify_pending' + order by c.counterparty_entity_name + """, + post_id, + ) + pending = [ + _PendingRelation( + str(row["counterparty_entity_name"]), + str(row["relationship_label"]), + await _find_internal_evidence_post( + conn, + post_id, + row["counterparty_entity_name"], + row["relationship_label"], + visible_corporate_entity_ids, + ), + ) + for row in rows + ] + + verified = [] + for relation in pending: + result = await asyncio.to_thread( + client.verify, + relation.counterparty_entity_name, + relation.relationship_label, + ) + verified.append( + VerifiedRelation( + relation.counterparty_entity_name, + result.status_code, + result.evidence_url, + relation.internal_evidence_post_id, + ) + ) + + async with pool.acquire() as conn, conn.transaction(): + for relation in verified: + await conn.execute( + """ + update post_counterparty_entity + set verification_status_code = $3, + verification_evidence_url = $4, + verification_evidence_post_id = $5, + verification_checked_at = now() + where post_id = $1 and counterparty_entity_name = $2 + and verification_status_code = 'verify_pending' + """, + post_id, + relation.counterparty_entity_name, + relation.verification_status_code, + relation.verification_evidence_url, + relation.verification_evidence_post_id, + ) + return verified diff --git a/backend/app/report_ingestion.py b/backend/app/report_ingestion.py index 4539710d6..f01c15ae0 100644 --- a/backend/app/report_ingestion.py +++ b/backend/app/report_ingestion.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import re from collections import defaultdict from datetime import datetime, timezone @@ -555,7 +556,8 @@ async def rebuild_period_reports( previous = await load_previous_group_mean(conn, kind, grouping_key, period_code) if previous is not None: previous_means[grouping_key] = previous - bank_report, scored = score_groups_on_shared_metric( + bank_report, scored = await asyncio.to_thread( + score_groups_on_shared_metric, groups, item_bank=item_bank, previous_means=previous_means, diff --git a/backend/tests/test_config.py b/backend/tests/test_config.py index 2a80f1fa4..a826fc013 100644 --- a/backend/tests/test_config.py +++ b/backend/tests/test_config.py @@ -36,12 +36,26 @@ def test_oidc_clock_skew_is_bounded(monkeypatch) -> None: raise AssertionError("clock skew above the bound must be rejected") -def test_tepp_transport_url_defaults_empty_and_is_not_a_score(monkeypatch) -> None: - """Missing TEPP_TRANSPORT_URL keeps the channel dropped.""" +def test_tepp_transport_defaults_empty_and_preserve_runtime_credentials(monkeypatch) -> None: + """Missing TEPP transport config drops the channel; a key stays runtime-only.""" monkeypatch.delenv("TEPP_TRANSPORT_URL", raising=False) - assert load_settings().tepp_transport_url == "" + monkeypatch.delenv("TEPP_API_KEY", raising=False) + settings = load_settings() + assert settings.tepp_transport_url == "" + assert settings.tepp_api_key == "" monkeypatch.setenv("TEPP_TRANSPORT_URL", "https://tepp.example/v1/analysis-runs") - assert load_settings().tepp_transport_url == "https://tepp.example/v1/analysis-runs" + monkeypatch.setenv("TEPP_API_KEY", "runtime-test-key") + settings = load_settings() + assert settings.tepp_transport_url == "https://tepp.example/v1/analysis-runs" + assert settings.tepp_api_key == "runtime-test-key" + + +def test_tepp_api_key_is_runtime_only(monkeypatch) -> None: + """TEPP authentication comes from the process boundary, never source.""" + monkeypatch.delenv("TEPP_API_KEY", raising=False) + assert load_settings().tepp_api_key == "" + monkeypatch.setenv("TEPP_API_KEY", "runtime-only-test-value") + assert load_settings().tepp_api_key == "runtime-only-test-value" def test_keyverse_issuer_overrides_local_keycloak_and_uses_oidc_discovery(monkeypatch) -> None: diff --git a/docker-compose.yml b/docker-compose.yml index e2990df32..195fb0e72 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -175,6 +175,7 @@ services: ORCHESTRATOR_API_KEY: ${ORCHESTRATOR_API_KEY:-${CONTEXTUAL_ORCHESTRATOR_TOKEN:-lineageweave-orchestrator-dev-only}} SEARXNG_BASE_URL: http://searxng:8080 TEPP_TRANSPORT_URL: ${TEPP_TRANSPORT_URL:-} + TEPP_API_KEY: ${TEPP_API_KEY:-} CALDAV_BASE_URL: ${CALDAV_BASE_URL:-} NARUON_CALENDAR_BASE_URL: ${NARUON_CALENDAR_BASE_URL:-} NARUON_CALENDAR_SERVICE_TOKEN: ${NARUON_CALENDAR_SERVICE_TOKEN:-} diff --git a/docs/adr/0047-global-ask-semantic-retrieval.md b/docs/adr/0047-global-ask-semantic-retrieval.md index d7a0954b0..5a6e03722 100644 --- a/docs/adr/0047-global-ask-semantic-retrieval.md +++ b/docs/adr/0047-global-ask-semantic-retrieval.md @@ -13,16 +13,33 @@ find a post while Ask Agent could not. Global Ask embeds the complete natural-language question once through contextual-orchestrator and ranks authorized posts by the maximum raw cosine similarity against their persisted semantic-unit embeddings. Query and unit -vectors must have the same configured embedding model and dimension. No token -extraction, keyword matching, lexical weighting, similarity threshold, or -locally invented channel weight participates in candidate selection. +vectors must have the same configured embedding model and dimension. -The retrieved posts carry their raw source fields and persisted -project/role/Keyman facts into the contextual-orchestrator prompt with -column/table provenance. These facts enrich grounded answering; they do not -become keyword retrieval signals. If the embedding channel or a complete -matching-model vector is unavailable, retrieval returns no evidence rather -than falling back to lexical search. +Persisted project, role/responsibility/affiliation, Keyman, Knowledge Graph +edge/endpoint-label, and ontology-IRI evidence is a second candidate-nomination +channel. PostgreSQL `websearch_to_tsquery('simple', ...)` runs against GIN +expression indexes on the normalized owning tables; it does not copy evidence +into a denormalized search table. A complete canonical ontology IRI in the +question maps through the published lookup-code annotation. A Knowledge Graph +match nominates only `knowledge_graph_edge_evidence.evidence_post_id`, never an +endpoint post merely because that post labels a node. + +Both owned rank lists are bounded independently after the same SQL +visibility, corporate/process scope, source-eligibility, and event-time +predicates. RankWeave combines them with Cormack, Clarke, and Buettcher's +(2009) parameter-free reciprocal rank fusion. No token extractor, similarity +threshold, hand-authored channel preference, or locally invented weight is +allowed. The existing final source-row query and `can_see_post` callback remain +a second authorization check. If RankWeave cannot combine two present +channels, the new evidence channel is dropped and the embedding ranking +remains; a sole available channel needs no fusion. + +The retrieved posts carry their raw source fields and persisted semantic/KG +facts into the contextual-orchestrator prompt with column/table provenance. +Candidate nomination does not make a fact authoritative and does not bypass +the evidence-post mapping. If the embedding channel or a complete +matching-model vector is unavailable, retrieval may use only the persisted +evidence channel; it never falls back to title/body lexical search. Raw source fields remain `hint_only`; the prompt explicitly distinguishes them from resolved ontology assertions. The existing ABAC filter is applied before @@ -31,9 +48,19 @@ semantic evidence is loaded, and the bounded source limit remains in place. ## Consequences - Ask Agent retrieves by semantic-unit meaning without a keyword rule. +- A term present only in normalized semantic, Knowledge Graph, endpoint-label, + or ontology evidence can nominate its authorized evidence post. - A source hint can retrieve a post but cannot silently bind a customer, project, PU, or Keyman. - The orchestrator receives more useful evidence while still receiving only authorized, bounded source documents. - Missing semantic measurement fails closed and cannot silently change the retrieval method. + +## References + +Cormack, G. V., Clarke, C. L. A., & Buettcher, S. (2009). Reciprocal rank +fusion outperforms Condorcet and individual rank learning methods. In +*Proceedings of the 32nd International ACM SIGIR Conference on Research and +Development in Information Retrieval* (pp. 758–759). Association for +Computing Machinery. https://doi.org/10.1145/1571941.1572114 diff --git a/docs/adr/0213-global-ask-embedding-pool-release.md b/docs/adr/0213-global-ask-embedding-pool-release.md new file mode 100644 index 000000000..3f62199c3 --- /dev/null +++ b/docs/adr/0213-global-ask-embedding-pool-release.md @@ -0,0 +1,40 @@ +# ADR 0213 — Global Ask embeds before acquiring a pooled connection + +**Decision status:** Accepted +**Date:** 2026-08-25 +**Related:** [0204](0204-analysis-run-short-transaction-delivery.md) + +## Context + +The authenticated k6 HTTP exercise found ordinary post and Event Lineage +reads waiting while Global Ask jobs called the external embedding provider. +`compute_global_ask_answer` acquired an asyncpg connection before +`gather_global_chat_sources` called that provider, so provider latency could +occupy every slot in the shared ten-connection pool. Moving the call to a +thread kept the event loop responsive but did not release the pool resource. + +## Decision + +Resolve and validate the question embedding before acquiring an asyncpg +connection. Acquire the pool only for the bounded persisted-vector query and +release it before answer generation. An unavailable, empty, unbound, or +zero-norm embedding remains a fail-closed no-source result; LineageWeave does +not substitute lexical retrieval, a local model, or an invented vector. + +The same boundary applies to future provider work: a provider call must not +run inside a pooled-connection context unless one atomic database operation +requires it and an ADR records that exception. + +## Consequences + +- Embedding latency cannot exhaust the shared HTTP database pool. +- Authorization predicates and persisted model/dimension matching remain in + the database query and are unchanged. +- A regression test observes the pool state at the embedding boundary. +- Capacity remains environment-specific; k6 observations do not create an + uncited concurrency or latency threshold. + +## References + +PostgreSQL Global Development Group. (2026). *PostgreSQL 18.6 documentation: +19.4 resource consumption*. https://www.postgresql.org/docs/18/runtime-config-resource.html diff --git a/docs/adr/README.md b/docs/adr/README.md index 88ffb3abc..c6adcfaa7 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -17,8 +17,8 @@ decision from them. | [`ONTOLOGY_NAMESPACE_INVENTORY.md`](../doctoring/ONTOLOGY_NAMESPACE_INVENTORY.md) | [0207](0207-repository-case-ontology-namespace-canonical.md), [0157](0157-public-ontology-namespace-identity.md) | | [`image-content-schema.md`](../image-content-schema.md) | [0066](0066-position-preserving-image-content.md) | | [`storybook-inventory.md`](../storybook-inventory.md) | [0118](0118-uiux-standard-guide-v3-design-overhaul.md), [0184](0184-ontology-provenance-explorer.md) | -| [`POSTGRESQL_CONCURRENCY_REFERENCES.md`](../doctoring/POSTGRESQL_CONCURRENCY_REFERENCES.md) | [0204](0204-analysis-run-short-transaction-delivery.md) | -| [`operability/http-concurrency-evidence.md`](../operability/http-concurrency-evidence.md) | [0204](0204-analysis-run-short-transaction-delivery.md), [0212](0212-single-query-authorized-post-filter-options.md) | +| [`POSTGRESQL_CONCURRENCY_REFERENCES.md`](../doctoring/POSTGRESQL_CONCURRENCY_REFERENCES.md) | [0204](0204-analysis-run-short-transaction-delivery.md), [0213](0213-global-ask-embedding-pool-release.md) | +| [`operability/http-concurrency-evidence.md`](../operability/http-concurrency-evidence.md) | [0204](0204-analysis-run-short-transaction-delivery.md), [0212](0212-single-query-authorized-post-filter-options.md), [0213](0213-global-ask-embedding-pool-release.md) | | Evidence operations Dashboard (`/`) | [0206](0206-evidence-operations-dashboard.md) | | [`temporal-topic-context-influence-research.md`](../temporal-topic-context-influence-research.md) | [0210](0210-temporal-topic-context-influence-dashboard.md) | | [`python-mathematical-compute-boundary-audit.md`](../doctoring/python-mathematical-compute-boundary-audit.md) | [0208](0208-externalize-local-mathematical-compute.md) | diff --git a/docs/operability/http-concurrency-evidence.md b/docs/operability/http-concurrency-evidence.md index ae3684142..69bbfd60f 100644 --- a/docs/operability/http-concurrency-evidence.md +++ b/docs/operability/http-concurrency-evidence.md @@ -20,7 +20,8 @@ window that match the environment under review: ```bash make up KEYCLOAK_ADMIN_PASSWORD=admin_dev_only make seed -k6 run --vus --duration \ +k6 run -e REQUEST_TIMEOUT= \ + --vus --duration \ scripts/k6_http_e2e.js ``` @@ -29,6 +30,10 @@ Pass `BACKEND_URL`, `KEYCLOAK_URL`, `KEYCLOAK_REALM`, `KEYCLOAK_CLIENT_ID`, harness at another authorized synthetic environment. Never run repository performance evidence against identifying production records. +`REQUEST_TIMEOUT` is mandatory because an unbounded request hid the first +observed saturation behind k6's graceful-stop window. It is an operator-declared +observation boundary, not a product latency threshold. + ## Interpret the output k6 reports observed request counts, failure rate, and duration distributions. @@ -58,6 +63,37 @@ Figma and screenshot review do not apply: this is a non-UI HTTP load harness. ## Current-main verification record +On 2026-08-25, the follow-up change at `a700374e` was exercised against the +authorized local Compose PostgreSQL/Keycloak/Valkey/orchestrator stack after +all schema migrations and index builds had completed. Only aggregate evidence +was retained: the database held 43,189 source posts. Ten-second authenticated +observations used the same endpoint mix and reported zero HTTP errors at 1, +10, and 25 VUs. Before the bounded-lineage query, HTTP median/p95/p99 and +throughput were 809.03 ms/6.18 s/6.22 s and 0.618 requests/s at 1 VU; +4.63 s/20.10 s/20.46 s and 1.531 requests/s at 10 VUs; and +25.35 s/33.59 s/36.30 s and 1.046 requests/s at 25 VUs. The 25-VU observation +completed six iterations. + +The same observations after moving the landing lineage ABAC, ordering, node +bound, and edge bound into PostgreSQL were 179.52 ms/3.43 s/4.17 s and 1.067 +requests/s at 1 VU; 1.88 s/20.80 s/21.04 s and 1.487 requests/s at 10 VUs; +and 22.03 s/29.78 s/31.38 s and 2.411 requests/s at 25 VUs. The 25-VU +observation completed 25 iterations. The 10-VU tail did not improve, so this +evidence does not establish a latency SLO or a product capacity ceiling. It +does establish that repeatedly loading all visible posts and all lineage edges +before applying the 500-node contract was avoidable work; the remaining tail +requires endpoint-tagged traces and database-pool telemetry before another +cause is assigned. + +An exact-code-head 4-VU, 60-second confirmation at `a700374e` completed 36 +iterations and 110 HTTP requests with zero failed checks or requests. Overall +HTTP median/p95/p99 were 392.15 ms/8.82 s/9.68 s at 1.644 requests/s. The +combined posts/lineage read median/p95/p99 were 3.21 s/9.14 s/9.89 s; Ask poll +median/p95/p99 were 41.39 ms/413.16 ms/462.65 ms. All 36 iterations observed +the Ask lifecycle state. This confirms asynchronous Ask polling remained +responsive in that observation while also preserving the remaining reader-tail +gap; it is not a deployment SLO. + On 2026-08-25, a worktree based on protected-main commit `48f013a2` passed `k6 inspect` for this script. A fresh Compose project did not reach an application-ready state: the build was stopped @@ -68,6 +104,50 @@ HTTP latency distribution was produced and no application bottleneck is claimed. This is local build-environment evidence only. Re-run the command above on an application-ready stack to obtain the product measurement. +The next application-ready exercise on protected-main `d7d5eeb3` exposed two +failures before a capacity distribution could be accepted. A clean backend +process could not start because `Settings` omitted the already-consumed +`tepp_api_key`, and the replay database lacked the non-idempotent 0203 Global +Ask scope tables. After repairing those startup and replay contracts, the k6 +setup completed, but its authenticated read batch overlapped migration replay: +PostgreSQL was still building the 0035 trigram index with a `DataFileRead` wait, +and the not-yet-reached 0140 migration meant Event Lineage correctly failed on +its absent interval column. This run therefore cannot attribute read latency to +Global Ask and is not a valid steady-state capacity exercise. + +Independent code-path diagnosis did confirm that Global Ask resolved its +external question embedding inside `pool.acquire()`. ADR 0213 moves that call +before acquisition and adds a regression check that observes zero held pool +slots during embedding. With one virtual user, a 10-second observation, and a +declared 20-second request window, the post-fix branch then observed Ask enqueue +at 3.11 seconds and Ask polling at 1.31 seconds while both reads failed under +that incomplete migration state (one reached the 20-second request boundary; +combined read duration averaged 14.13 seconds). This is replay-in-progress +failure evidence, not a steady-state capacity result or product latency claim. +Re-run only after migration replay completes. + +A subsequent exact-head run reached the 0140 interval migration but still was +not steady state: replay stopped at migration 0165 because its queue table and +indexes lacked the ADR 0166 replay guards, so migration 0174's edge-signal +table was absent. With one virtual user, a 15-second observation, and the same +20-second request window, Ask enqueue averaged 125.05 milliseconds, Ask polls +averaged 123.41 milliseconds, and posts succeeded, but all four Event Lineage +reads failed on that absent table. The branch now makes migration 0165 +idempotent and regression-checks both Global Ask migrations. These values are +diagnostic evidence only. + +After replaying the repaired 0165–0205 range to completion, a four-VU, +30-second observation with the declared 20-second request window completed 13 +iterations and all 39 endpoint checks without an HTTP failure. Ask enqueue was +57.32 milliseconds, Ask polling averaged 359.91 milliseconds (p95 969.66 +milliseconds), and the combined posts/Event-Lineage read distribution averaged +5.75 seconds (p95 11.88 seconds, maximum 12.36 seconds). A second four-VU, +15-second diagnostic run also completed every endpoint check; concurrent +`pg_stat_activity` samples repeatedly observed the authorized filter-option, +post-list, and lineage-page queries as active, including `MessageQueueSend` and +one temporary-buffer write. This identifies the measured database work to +profile next; it does not by itself assign causality or establish an SLO. + ## Older-image diagnostic observation On 2026-08-25, an application-ready local Compose stack configured with four diff --git a/docs/product-requirements.md b/docs/product-requirements.md index 75cba0410..f0bd0aebe 100644 --- a/docs/product-requirements.md +++ b/docs/product-requirements.md @@ -166,7 +166,7 @@ A release claim requires one exact protected-main head that proves: ## 7. Traceability - Product/data boundary: ADR 0001, ADR 0089. -- Asynchronous delivery and database-pool isolation: ADR 0204. +- Asynchronous delivery and database-pool isolation: ADR 0204, ADR 0213. - Knowledge Graph, ontology, and provenance: ADR 0004, ADR 0011, ADR 0065, ADR 0184, ADR 0207. - Semantic units and retrieval: ADR 0047, ADR 0062, ADR 0102. diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 01bd49319..f07bf917d 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -170,6 +170,15 @@ Recent protected-default-branch delivery evidence (squash merges onto | PR | Merged (UTC) | Delivered | | ---: | --- | --- | +| #628 | 2026-08-25 12:39 | one authorized post-filter option query per request (ADR 0212) | +| #627 | 2026-08-25 12:35 | valid k6 lifecycle evidence across VUs | +| #626 | 2026-08-25 12:25 | authenticated HTTP concurrency harness | +| #625 | 2026-08-25 12:25 | pnpm 11 esbuild build approval and repair-workflow removal | +| #624 | 2026-08-25 12:25 | asynchronous-capacity product requirement | +| #387 | 2026-08-25 12:19 | persisted and reader-explained Event Lineage channel evidence | +| #620 | 2026-08-25 12:16 | temporal-topic, Rust-boundary, and capacity gap refresh | +| #623 | 2026-08-25 11:57 | Node 24-compatible pnpm runtime path | +| #621 | 2026-08-25 11:54 | current PRD and ecosystem authority register | | #468 | 2026-08-25 08:44 | fast-mlsirm, Keyverse, contextual-orchestrator, and TEPP integration boundaries | | #493 | 2026-08-25 08:44 | evidence-grounded Event Lineage isolation reasons | | #600 | 2026-08-25 08:44 | then-current exact-head product/technical baseline | @@ -373,11 +382,12 @@ this file per §3.5 of the prior snapshot). | Shared frontend gate | The ADR 0109 login repair is on protected `main`; eight older branches carried the defect and received the same verified repair this loop (#521–#560) | Keep every future branch cut from post-repair bases; re-verify with frontend lint/test/build before push | | Identifying baseline regression | `main` gap file listed real post identifiers; separately, closed #506 and pre-existing public history contain a private runtime source-table identifier, while current `main` and #507 trees are clean | Land this non-identifying rewrite, then coordinate ADR 0001 history remediation with security/privacy owners; do not reproduce the value, force-push, or delete evidence ad hoc | | Authorized-corpus runtime | Repository tests use synthetic fixtures; private records remain outside git | Authenticated runtime validation returning only aggregate, non-identifying evidence | -| Concurrent web responsiveness | ADR 0204 releases pooled transactions during provider work, and the synthetic Compose boundary has an authenticated k6 E2E harness for Ask enqueue, concurrent reads, and job polling. An older-image local observation found repeated post-filter queries while `/api/posts` exceeded 30 seconds; ADR 0212 combines two authorized filter-option queries into one round trip without narrowing ABAC-visible options. The observation is not exact-head evidence or a product guarantee, and no physical scan reduction is claimed without an exact-head plan | Rebuild an exact-head application image, run `make load-http` with declared environment concurrency/window, and retain raw distributions and resource configuration. Compare the post-list database plan and latency with ADR 0212 while preserving the complete authorized filter set; set no SLO until representative capacity evidence is approved | +| Concurrent web responsiveness | ADR 0204 releases analysis-run transactions. ADR 0212 combines the authorized filter-option query, and ADR 0213 releases the pool before external embedding. Migration 0165 now follows ADR 0166 replay safety after it stopped replay before the 0174 edge-signal table. After repaired replay, a four-VU 30-second exact-branch observation completed all 39 endpoint checks; combined reads averaged 5.75 seconds with p95 11.88 seconds. Concurrent database samples repeatedly observed filter-option, post-list, and lineage-page work active, but do not establish causality or an SLO | Capture exact plans and resource telemetry for the three observed query families, remove measured database bottlenecks without narrowing ABAC, then repeat the declared k6 workload on representative capacity; set no SLO until that evidence is approved | | Image understanding | Region, OCR, and description work exists across active heads (#405, #419), but current runtime acceptance has not yet proved table-image structure, complete region coverage, or summary/image readiness together | Orchestrator-backed rendered workflow, original/derived asset provenance, region-before-OCR processing, and honest unsupported states; reconcile ADR 0052's image-bearing summary readiness with ADR 0098 before changing sequencing | | Semantic source rendering | Paragraph, table, list, formula, and indentation work exists across stacks (#394, #427, #448–#450); #515 adds synthetic backend/frontend parity for deterministic rows/cells, footnote boundaries, and encoded scripts | Land the #427 → #515 stack, then gather authenticated browser evidence that list nesting, continuation alignment, and formula units render without authoring-layout artifacts | | Event and project semantics | Multi-project mentions, project-bound actions, 5W1H, requester/processor, and semantic relations exist in ADR 0036/0052/0100/0111/0129 and active stacks | Aggregate authenticated evidence must show distinct projects and events, explicit requester/processor and real R&R, normalized relative time, and product/entity relations without promoting attendance or co-occurrence | | Knowledge Graph prompt provenance | PR #632 maps every ontology-annotated graph fact to the visible post recorded in `knowledge_graph_edge_evidence` and drops post endpoints outside the same authorized source window before label hydration; earlier code could attach a fact to the wrong source or reveal an out-of-window post endpoint through a visible evidence post | Exact-head tests must prove post chat and Global Ask attach each fact only to its evidencing source, never hydrate a hidden/out-of-window post endpoint, retain ABAC and prompt bounds, and merge through protected `main`; external verification remains a separate issue #272 contract | +| Semantic/KG candidate nomination | Issue #272 remains open: embedding-only nomination cannot retrieve a source when the query term exists solely in project, R&R, Keyman, Knowledge Graph endpoint/edge, or ontology evidence. The stacked implementation branch composes #629 pool discipline with #632 evidence-post provenance, adds replay-safe GIN expression indexes on normalized evidence tables, and uses parameter-free RankWeave RRF rather than a hand-authored channel preference | Live PostgreSQL and exact-head tests must prove every evidence kind nominates only its authorized evidence post, ABAC/eligibility/event-time filters run before each channel limit and again at hydration, duplicate hits deduplicate, hidden endpoint labels do not leak, missing RankWeave drops only the added channel, and protected `main` contains the merge SHA before the gap is marked delivered | | Knowledge Graph readability | The black evidence-node root cause is an undefined-token fallback; the design-token repair and long-label/evidence-table coverage remain only on closed, unmerged #490, not protected `main` | Recreate the token repair on a current base and deliver it through protected `main`, then verify light/dark contrast, keyboard graph navigation, full labels, and evidence tables in the authenticated rendered surface | | Source-code lookup UX | Source state/detail codes remain evidence-bearing machine values and current detail presentation is dense | Catalog-backed display labels with raw-code provenance, compact 5W1H/source-detail hierarchy, keyboard access, and no unsupported customer/project binding | | Calendar / Naruon | #355 delivered the projection contract; v2.17.0 wires operator consumption without forwarding the end-user token. Naruon producer, provider/consumer fixtures, and protected merge remain open (#336) | Verify observed events against the published schema without invented events; keep commitments available when the channel is unwired | diff --git a/lineageweave/rankweave_client.py b/lineageweave/rankweave_client.py index 4c157aa57..f9b091bc4 100644 --- a/lineageweave/rankweave_client.py +++ b/lineageweave/rankweave_client.py @@ -269,7 +269,7 @@ def project_ranking_list( class LibraryRankWeaveTransport: - """Call RankWeave ``weighted_reciprocal_rank_fuse`` in-process.""" + """Call RankWeave reciprocal-rank fusion in-process.""" def __call__( self, @@ -298,19 +298,32 @@ def __call__( "rankweave_not_available: no positive channel weights remain" ) try: - hits = rw.weighted_reciprocal_rank_fuse( - usable, - active_weights, - limit=DEFAULT_RANKING_LIMIT, - rank_constant_eta=DEFAULT_RANK_CONSTANT_ETA, - ) - except TypeError: - try: + if all(weight == 1.0 for weight in active_weights.values()): + hits = rw.reciprocal_rank_fuse( + usable, + limit=DEFAULT_RANKING_LIMIT, + rank_constant_eta=DEFAULT_RANK_CONSTANT_ETA, + ) + else: hits = rw.weighted_reciprocal_rank_fuse( usable, active_weights, limit=DEFAULT_RANKING_LIMIT, + rank_constant_eta=DEFAULT_RANK_CONSTANT_ETA, ) + except TypeError: + try: + if all(weight == 1.0 for weight in active_weights.values()): + hits = rw.reciprocal_rank_fuse( + usable, + limit=DEFAULT_RANKING_LIMIT, + ) + else: + hits = rw.weighted_reciprocal_rank_fuse( + usable, + active_weights, + limit=DEFAULT_RANKING_LIMIT, + ) except Exception as exc: raise RankWeaveNotAvailable( "rankweave_not_available: weighted_reciprocal_rank_fuse failed" diff --git a/migrations/0165_global_ask_job.sql b/migrations/0165_global_ask_job.sql index 266b77979..37fa4b568 100644 --- a/migrations/0165_global_ask_job.sql +++ b/migrations/0165_global_ask_job.sql @@ -7,7 +7,7 @@ -- Mirrors the durable-row-plus-stream design post_content_job already -- uses, so a lost stream entry is recovered from the queued rows. -create table global_ask_job ( +create table if not exists global_ask_job ( global_ask_job_id uuid primary key default uuid_generate_v4(), requesting_account_id uuid not null references user_account (user_account_id), question_text text not null, @@ -23,9 +23,9 @@ comment on table global_ask_job is 'One asynchronous Global Ask request: queued by POST /api/ask, ' 'processed by the Valkey-stream worker, polled by the reader.'; -create index global_ask_job_account_idx +create index if not exists global_ask_job_account_idx on global_ask_job (requesting_account_id, created_at desc); -create index global_ask_job_queued_idx +create index if not exists global_ask_job_queued_idx on global_ask_job (created_at) where job_status_code = 'queued'; diff --git a/migrations/0203_global_ask_authorization_scope.sql b/migrations/0203_global_ask_authorization_scope.sql index 17d23a4f3..078a9d822 100644 --- a/migrations/0203_global_ask_authorization_scope.sql +++ b/migrations/0203_global_ask_authorization_scope.sql @@ -1,12 +1,12 @@ -- Persist the exact authorization scope carried by the request token. -create table global_ask_job_corporate_entity_scope ( +create table if not exists global_ask_job_corporate_entity_scope ( global_ask_job_id uuid not null references global_ask_job (global_ask_job_id) on delete cascade, corporate_entity_id uuid not null references corporate_entity (corporate_entity_id), primary key (global_ask_job_id, corporate_entity_id) ); -create table global_ask_job_process_unit_scope ( +create table if not exists global_ask_job_process_unit_scope ( global_ask_job_id uuid not null references global_ask_job (global_ask_job_id) on delete cascade, process_unit_id uuid not null references process_unit (process_unit_id), primary key (global_ask_job_id, process_unit_id) diff --git a/migrations/0210_global_ask_evidence_search_indexes.sql b/migrations/0210_global_ask_evidence_search_indexes.sql new file mode 100644 index 000000000..0ed3a3934 --- /dev/null +++ b/migrations/0210_global_ask_evidence_search_indexes.sql @@ -0,0 +1,74 @@ +-- Index the normalized evidence fields used to nominate Global Ask sources. +-- Rows remain in their owning 3NF tables; these are expression indexes only. + +create index if not exists post_project_mention_evidence_search_idx + on post_project_mention using gin ( + to_tsvector( + 'simple', + coalesce(project_name, '') || ' ' || + coalesce(evidence_text, '') || ' ' || + coalesce(ontology_iri, '') + ) + ); + +create index if not exists post_summary_role_evidence_search_idx + on post_summary_role using gin ( + to_tsvector( + 'simple', + coalesce(actor_name, '') || ' ' || + coalesce(responsibility, '') || ' ' || + coalesce(affiliated_organization_name, '') + ) + ); + +create index if not exists cataloged_person_evidence_search_idx + on cataloged_person using gin ( + to_tsvector( + 'simple', + coalesce(person_name, '') || ' ' || + coalesce(last_known_job_title, '') + ) + ); + +create index if not exists person_affiliation_evidence_search_idx + on person_affiliation using gin ( + to_tsvector( + 'simple', + coalesce(affiliated_organization_name, '') || ' ' || + coalesce(role_title, '') + ) + ); + +create index if not exists corporate_entity_evidence_search_idx + on corporate_entity using gin ( + to_tsvector( + 'simple', + coalesce(corporate_entity_code, '') || ' ' || + coalesce(entity_name, '') + ) + ); + +create index if not exists cataloged_team_evidence_search_idx + on cataloged_team using gin ( + to_tsvector( + 'simple', + coalesce(team_name, '') || ' ' || + coalesce(affiliated_organization_name, '') + ) + ); + +create index if not exists source_post_title_evidence_search_idx + on source_post using gin ( + to_tsvector('simple', coalesce(post_title, '')) + ); + +create index if not exists common_lookup_value_evidence_search_idx + on common_lookup_value using gin ( + to_tsvector( + 'simple', + coalesce(lookup_code, '') || ' ' || coalesce(lookup_label, '') + ) + ); + +create index if not exists knowledge_graph_edge_type_search_idx + on knowledge_graph_edge (edge_type_code, knowledge_graph_edge_id); diff --git a/migrations/rollback/0210_global_ask_evidence_search_indexes.sql b/migrations/rollback/0210_global_ask_evidence_search_indexes.sql new file mode 100644 index 000000000..d5363b555 --- /dev/null +++ b/migrations/rollback/0210_global_ask_evidence_search_indexes.sql @@ -0,0 +1,9 @@ +drop index if exists knowledge_graph_edge_type_search_idx; +drop index if exists common_lookup_value_evidence_search_idx; +drop index if exists source_post_title_evidence_search_idx; +drop index if exists cataloged_team_evidence_search_idx; +drop index if exists corporate_entity_evidence_search_idx; +drop index if exists person_affiliation_evidence_search_idx; +drop index if exists cataloged_person_evidence_search_idx; +drop index if exists post_summary_role_evidence_search_idx; +drop index if exists post_project_mention_evidence_search_idx; diff --git a/scripts/k6_http_e2e.js b/scripts/k6_http_e2e.js index 737b2e09f..976f63c73 100644 --- a/scripts/k6_http_e2e.js +++ b/scripts/k6_http_e2e.js @@ -16,6 +16,8 @@ const realm = __ENV.KEYCLOAK_REALM || "lineageweave-demo"; const clientId = __ENV.KEYCLOAK_CLIENT_ID || "lineageweave-frontend"; const username = __ENV.K6_USERNAME || "demo.analyst"; const password = __ENV.K6_PASSWORD || "lineageweave-demo-only"; +const requestTimeout = __ENV.REQUEST_TIMEOUT; +const unitlessDuration = /^\d+(?:\.\d+)?$/; const askEnqueueDuration = new Trend("lineageweave_ask_enqueue_duration", true); const readDuration = new Trend("lineageweave_read_duration", true); @@ -33,7 +35,7 @@ function authenticate() { username, password, }, - { tags: { endpoint: "oidc_token" } }, + { tags: { endpoint: "oidc_token" }, timeout: requestTimeout }, ); if (response.status !== 200) { fail(`synthetic OIDC login failed with HTTP ${response.status}`); @@ -44,24 +46,40 @@ function authenticate() { function readBatch(token, askJobId) { const params = { headers: { Authorization: `Bearer ${token}` } }; return http.batch([ - ["GET", `${backendUrl}/api/posts`, null, { ...params, tags: { endpoint: "posts" } }], - ["GET", `${backendUrl}/api/lineage`, null, { ...params, tags: { endpoint: "lineage" } }], + [ + "GET", + `${backendUrl}/api/posts`, + null, + { ...params, tags: { endpoint: "posts" }, timeout: requestTimeout }, + ], + [ + "GET", + `${backendUrl}/api/lineage`, + null, + { ...params, tags: { endpoint: "lineage" }, timeout: requestTimeout }, + ], [ "GET", `${backendUrl}/api/ask/jobs/${askJobId}`, null, - { ...params, tags: { endpoint: "ask_poll" } }, + { ...params, tags: { endpoint: "ask_poll" }, timeout: requestTimeout }, ], ]); } export function setup() { + if (!requestTimeout) { + fail("REQUEST_TIMEOUT is required"); + } + if (unitlessDuration.test(requestTimeout)) { + fail("REQUEST_TIMEOUT must include a duration unit, for example 20s"); + } const token = authenticate(); const headers = { Authorization: `Bearer ${token}`, "Content-Type": "application/json" }; const submitted = http.post( `${backendUrl}/api/ask`, JSON.stringify({ question: "Summarize the synthetic demo lineage evidence." }), - { headers, tags: { endpoint: "ask_enqueue" } }, + { headers, tags: { endpoint: "ask_enqueue" }, timeout: requestTimeout }, ); askEnqueueDuration.add(submitted.timings.duration); if (submitted.status !== 202) { diff --git a/tests/test_global_ask_evidence_search_schema.py b/tests/test_global_ask_evidence_search_schema.py new file mode 100644 index 000000000..b9c1287bf --- /dev/null +++ b/tests/test_global_ask_evidence_search_schema.py @@ -0,0 +1,42 @@ +"""Schema contract for index-backed Global Ask evidence nomination.""" + +from pathlib import Path + + +MIGRATION = Path("migrations/0210_global_ask_evidence_search_indexes.sql") +ROLLBACK = Path("migrations/rollback/0210_global_ask_evidence_search_indexes.sql") + + +def test_evidence_search_indexes_cover_every_normalized_owner_table() -> None: + """Every searched evidence field has a replay-safe owning-table index.""" + sql = MIGRATION.read_text(encoding="utf-8") + + for table_name in ( + "post_project_mention", + "post_summary_role", + "cataloged_person", + "person_affiliation", + "corporate_entity", + "cataloged_team", + "source_post", + "common_lookup_value", + "knowledge_graph_edge", + ): + assert f"on {table_name}" in sql + assert sql.count("create index if not exists") == 9 + assert "create table" not in sql.lower() + + +def test_evidence_search_indexes_have_a_replay_safe_rollback() -> None: + """Operators can remove only this migration's indexes by exact name.""" + forward = MIGRATION.read_text(encoding="utf-8") + rollback = ROLLBACK.read_text(encoding="utf-8") + index_names = [ + line.split()[5] + for line in forward.splitlines() + if line.startswith("create index if not exists ") + ] + + assert len(index_names) == 9 + for index_name in index_names: + assert f"drop index if exists {index_name};" in rollback diff --git a/tests/test_global_ask_queue.py b/tests/test_global_ask_queue.py index 76b07170c..c502621ba 100644 --- a/tests/test_global_ask_queue.py +++ b/tests/test_global_ask_queue.py @@ -42,6 +42,94 @@ def _queued_row() -> dict[str, object]: } +def test_question_embedding_finishes_before_global_ask_acquires_a_pool_slot( + monkeypatch, +) -> None: + """Provider latency must not consume the shared database pool.""" + connection = _Connection(None) + + class TrackingPool(_Pool): + active = 0 + + @asynccontextmanager + async def acquire(self): + self.active += 1 + try: + yield self.connection + finally: + self.active -= 1 + + pool = TrackingPool(connection) + + class EmbeddingClient: + available = True + resolved_model = "synthetic-embedding" + + def embed(self, _text: str) -> list[float]: + assert pool.active == 0 + return [1.0, 0.0] + + async def fake_gather(_conn, *_args, **kwargs): + assert pool.active == 1 + assert kwargs["question_embedding"] == ( + [1.0, 0.0], + "synthetic-embedding", + 1.0, + ) + return [] + + monkeypatch.setattr(global_ask_queue, "gather_global_chat_sources", fake_gather) + + payload = asyncio.run( + global_ask_queue.compute_global_ask_answer( + pool, + question_text="What changed?", + corporate_entity_ids=set(), + process_unit_ids=set(), + process_scope_limited=False, + chat_client=_AvailableClient(), + embedding_client=EmbeddingClient(), + ) + ) + + assert payload["source_post_ids"] == [] + assert pool.active == 0 + + +def test_unavailable_question_embedding_is_not_called(monkeypatch) -> None: + """An unavailable embedding is dropped while persisted evidence still runs.""" + connection = _Connection(None) + pool = _Pool(connection) + + class UnavailableEmbedding: + available = False + resolved_model = None + + def embed(self, _text: str) -> list[float]: + raise AssertionError("unavailable embedding must not be called") + + async def fake_gather(_conn, *_args, **kwargs): + assert kwargs["question_embedding"] is None + assert kwargs["embedding_client"].available is False + return [] + + monkeypatch.setattr(global_ask_queue, "gather_global_chat_sources", fake_gather) + + payload = asyncio.run( + global_ask_queue.compute_global_ask_answer( + pool, + question_text="What changed?", + corporate_entity_ids=set(), + process_unit_ids=set(), + process_scope_limited=False, + chat_client=_AvailableClient(), + embedding_client=UnavailableEmbedding(), + ) + ) + + assert payload["source_post_ids"] == [] + + def test_unexpected_job_failure_settles_with_a_generic_detail_not_the_raw_exception( monkeypatch, ) -> None: diff --git a/tests/test_global_ask_sources.py b/tests/test_global_ask_sources.py index 0a7421277..17073b0e5 100644 --- a/tests/test_global_ask_sources.py +++ b/tests/test_global_ask_sources.py @@ -1,9 +1,15 @@ from __future__ import annotations import asyncio +import math from datetime import date, datetime, timezone -from backend.app.post_chat_ingestion import gather_global_chat_sources as _gather_global_chat_sources +from backend.app.post_chat_ingestion import ( + _fuse_global_candidate_ids, + _ontology_lookup_codes_in_question, + gather_global_chat_sources as _gather_global_chat_sources, + prepare_global_question_embedding, +) from lineageweave.ask_time_axis import TIME_AXIS_CREATED, TIME_AXIS_EVENT @@ -15,12 +21,120 @@ def embed(self, _text: str) -> list[float]: return [1.0, 0.0] +def test_prepare_global_question_embedding_rejects_blank_input_before_provider() -> None: + """A blank question must fail closed without crossing the provider boundary.""" + + class RejectCallsEmbedding: + resolved_model = "synthetic-embedding" + + def embed(self, _text: str) -> list[float]: + raise AssertionError("blank question must not call the embedding provider") + + assert ( + asyncio.run( + prepare_global_question_embedding(" \t\n", RejectCallsEmbedding()) + ) + is None + ) + + +def test_nonfinite_embeddings_fail_closed_before_database_access() -> None: + """Provider and precomputed vectors must remain finite.""" + + class NonfiniteEmbedding: + available = True + resolved_model = "synthetic-embedding" + + def embed(self, _text: str) -> list[float]: + return [math.nan, math.inf] + + class RejectDatabase: + async def fetch(self, _query: str, *_args): + raise AssertionError("nonfinite embeddings must not reach PostgreSQL") + + assert asyncio.run( + prepare_global_question_embedding("question", NonfiniteEmbedding()) + ) is None + assert asyncio.run( + _gather_global_chat_sources( + RejectDatabase(), + lambda _row: True, + question="question", + question_embedding=([math.inf, 0.0], "synthetic-embedding", math.inf), + ) + ) == [] + + def gather_global_chat_sources(*args, **kwargs): """Exercise Global Ask with an available deterministic semantic channel.""" kwargs.setdefault("embedding_client", _EmbeddingClient()) return _gather_global_chat_sources(*args, **kwargs) +def test_parameter_free_rrf_combines_embedding_and_evidence_rank_lists() -> None: + """A post supported by both owned channels outranks one-channel hits.""" + + assert _fuse_global_candidate_ids( + ["embedding-only", "shared"], ["shared", "evidence-only"], 3 + )[0] == "shared" + + +def test_complete_canonical_ontology_iri_maps_to_its_lookup_code() -> None: + """Ontology nomination uses the published full IRI, not substring guessing.""" + + codes = _ontology_lookup_codes_in_question( + "Explain https://contextualwisdomlab.github.io/LineageWeave/ontology#affiliatedWith" + ) + + assert codes == ["edge_affiliation"] + assert _ontology_lookup_codes_in_question("affiliatedWith") == [] + + +def test_evidence_only_term_nominates_its_authorized_source() -> None: + """A persisted semantic hit works even when no body embedding nominates it.""" + + source_row = { + "post_id": "semantic-only", + "post_title": "Neutral source title", + "post_body": "Neutral source body", + "visibility_code": "public", + "corporate_entity_id": None, + "process_unit_id": None, + "created_at": datetime(2026, 8, 25, tzinfo=timezone.utc), + "event_occurred_at": None, + } + + class FakeConnection: + async def fetch(self, query: str, *args): + if "unit_similarity" in query: + assert "authorized_evidence_candidates" in query + assert query.index("authorized_evidence_candidates") < query.rindex("limit $8") + assert args[8] == "exclusive responsibility" + return [ + { + "candidate_channel": "evidence", + "post_id": "semantic-only", + "channel_rank": 1, + } + ] + if "from post_lineage_edge" in query: + return [] + if "array_position($3::uuid[], post_id)" in query: + return [source_row] + return [] + + sources = asyncio.run( + gather_global_chat_sources( + FakeConnection(), + lambda row: row["visibility_code"] == "public", + question="exclusive responsibility", + limit=4, + ) + ) + + assert [source.post_id for source in sources] == ["semantic-only"] + + def test_global_sources_apply_visibility_before_normalization() -> None: rows = [ { @@ -128,14 +242,17 @@ async def fetch(self, query: str, *args): (query, args) for query, args in calls if "array_position($3::uuid[], post_id)" in query ) assert "unit_similarity" in candidate_query - assert "to_tsvector" not in candidate_query + assert "websearch_to_tsquery('simple', $9)" in candidate_query + assert "post_project_mention" in candidate_query + assert "knowledge_graph_edge_evidence" in candidate_query + assert "ilike" not in candidate_query.lower() assert candidate_args[0] == [1.0, 0.0] assert candidate_args[2] == "test-embedding" assert "array_position($3::uuid[], post_id)" in source_query assert "source_post.post_id = any($3::uuid[])" in source_query assert source_args[3] == 8 - # The database returns candidates in cosine-rank order; no local lexical - # weights or reranking may alter that order. + # A test row without a channel marker is the legacy embedding-channel + # fixture and retains its database rank order. assert list(source_args[2]) == ["newest-post", "uam-post"] assert sources[1].post_body.startswith("x" * 4000) assert "Source body truncated for Global Ask" in sources[1].post_body @@ -346,8 +463,10 @@ def embed(self, _text: str) -> list[float]: raise AssertionError("unavailable embedding must not be called") class FakeConnection: - async def fetch(self, _query: str, *_args): - raise AssertionError("lexical fallback must not query the corpus") + async def fetch(self, query: str, *args): + if "authorized_evidence_candidates" in query: + assert args[10] is False + return [] sources = asyncio.run( _gather_global_chat_sources( @@ -361,8 +480,41 @@ async def fetch(self, _query: str, *_args): assert sources == [] -def test_global_sources_fail_closed_without_a_resolved_embedding_model() -> None: - """A vector without its orchestrator-resolved model cannot match persisted rows.""" +def test_global_sources_accept_valid_precomputed_embedding_without_provider() -> None: + """A validated embedding envelope must not depend on provider availability.""" + + class UnavailableEmbedding: + available = False + resolved_model = None + + def embed(self, _text: str) -> list[float]: + raise AssertionError("precomputed embedding must not call the provider") + + calls: list[tuple[str, tuple[object, ...]]] = [] + + class FakeConnection: + async def fetch(self, query: str, *args): + calls.append((query, args)) + return [] + + sources = asyncio.run( + _gather_global_chat_sources( + FakeConnection(), + lambda _row: True, + question="semantic question", + question_embedding=([1.0, 0.0], "synthetic-embedding", 1.0), + embedding_client=UnavailableEmbedding(), + ) + ) + + assert sources == [] + candidate_calls = [(query, args) for query, args in calls if "unit_similarity" in query] + assert len(candidate_calls) == 1 + assert candidate_calls[0][1][:3] == ([1.0, 0.0], 1.0, "synthetic-embedding") + + +def test_global_sources_disable_an_embedding_without_a_resolved_model() -> None: + """An unbound vector cannot match persisted rows but evidence remains available.""" class UnboundEmbedding: available = True @@ -372,8 +524,10 @@ def embed(self, _text: str) -> list[float]: return [1.0, 0.0] class FakeConnection: - async def fetch(self, _query: str, *_args): - raise AssertionError("an unbound vector must not query persisted embeddings") + async def fetch(self, query: str, *args): + if "authorized_evidence_candidates" in query: + assert args[10] is False + return [] sources = asyncio.run( _gather_global_chat_sources( @@ -521,7 +675,7 @@ async def fetch(self, query: str, *args): ) -def test_global_sources_do_not_run_lexical_search_for_relative_time_question() -> None: +def test_global_sources_keep_body_and_title_lexical_fallback_disabled() -> None: calls: list[tuple[str, tuple[object, ...]]] = [] class FakeConnection: @@ -541,7 +695,8 @@ async def fetch(self, query: str, *args): candidate_queries = [query for query, _args in calls if "unit_similarity" in query] assert len(candidate_queries) == 1 assert "ilike" not in candidate_queries[0].lower() - assert "to_tsvector" not in candidate_queries[0].lower() + assert "source_post_search_text" not in candidate_queries[0] + assert "websearch_to_tsquery('simple', $9)" in candidate_queries[0] def test_global_sources_bind_relative_time_to_event_clock_not_ingest_cluster( diff --git a/tests/test_k6_http_e2e_contract.py b/tests/test_k6_http_e2e_contract.py index f77d8f596..3425cafa0 100644 --- a/tests/test_k6_http_e2e_contract.py +++ b/tests/test_k6_http_e2e_contract.py @@ -12,3 +12,5 @@ def test_k6_harness_renews_expired_auth_and_discloses_job_state() -> None: assert source.count("responses = readBatch(vuToken, data.askJobId)") == 2 assert "lineageweave_ask_state_observations" in source assert 'job_status: String(responses[2].json("job_status_code")' in source + assert "unitlessDuration.test(requestTimeout)" in source + assert "REQUEST_TIMEOUT must include a duration unit" in source diff --git a/tests/test_lineage_ingestion.py b/tests/test_lineage_ingestion.py index 38cf7d282..a7b19a342 100644 --- a/tests/test_lineage_ingestion.py +++ b/tests/test_lineage_ingestion.py @@ -974,6 +974,36 @@ async def fetch(self, query: str, *_args): assert focused["edges"][0]["channel_evidence"] == [] +def test_landing_lineage_applies_abac_and_limit_in_database() -> None: + class FakeConnection: + statements: list[tuple[str, tuple]] = [] + + async def fetch(self, query: str, *args): + self.statements.append((query, args)) + if "from source_post" in query: + return [] + return [] + + connection = FakeConnection() + graph = asyncio.run( + visible_lineage_graph( + connection, + lambda row: True, + limit=500, + corporate_entity_ids=("corp-a",), + process_unit_ids=("pu-a",), + ) + ) + + post_query, post_args = connection.statements[0] + assert "corporate_entity_id::text = any($1::text[])" in post_query + assert "process_unit_id::text = any($2::text[])" in post_query + assert "order by created_at desc, post_id desc limit $3" in post_query + assert post_args == (["corp-a"], ["pu-a"], 501) + assert graph["nodes"] == [] + assert graph["truncated"] is False + + class _RecordingConnection: def __init__(self) -> None: self.statements: list[tuple[str, tuple]] = [] diff --git a/tests/test_migration_replay.py b/tests/test_migration_replay.py index c170f73e8..81a98e28c 100644 --- a/tests/test_migration_replay.py +++ b/tests/test_migration_replay.py @@ -182,3 +182,16 @@ def test_topic_lineage_result_migration_is_idempotent_for_replay() -> None: assert "create table if not exists analysis_run_topic_lineage_result" in migration assert "create index if not exists" in migration + + +def test_global_ask_job_migrations_are_idempotent_for_replay() -> None: + """Existing volumes must replay the queue and authorization scope safely.""" + migrations = Path(__file__).resolve().parents[1] / "migrations" + job_sql = (migrations / "0165_global_ask_job.sql").read_text(encoding="utf-8") + scope_sql = (migrations / "0203_global_ask_authorization_scope.sql").read_text( + encoding="utf-8" + ) + + assert "create table if not exists global_ask_job" in job_sql + assert job_sql.count("create index if not exists") == 2 + assert scope_sql.count("create table if not exists") == 2 diff --git a/tests/test_rankweave_client.py b/tests/test_rankweave_client.py index f3598ca91..e9d54f101 100644 --- a/tests/test_rankweave_client.py +++ b/tests/test_rankweave_client.py @@ -121,7 +121,7 @@ def test_library_transport_fails_closed_when_fuse_raises( ) -> None: class FakeRw: @staticmethod - def weighted_reciprocal_rank_fuse(*_args: object, **_kwargs: object) -> list: + def reciprocal_rank_fuse(*_args: object, **_kwargs: object) -> list: raise RuntimeError("duplicate identifiers") monkeypatch.setattr( @@ -177,14 +177,12 @@ def test_library_transport_projects_monkeypatched_rrf( class FakeRw: @staticmethod - def weighted_reciprocal_rank_fuse( + def reciprocal_rank_fuse( channels: dict[str, list[str]], - weights: dict[str, float], limit: int = 20, rank_constant_eta: int = 60, ) -> list: captured["channels"] = channels - captured["weights"] = weights captured["limit"] = limit captured["eta"] = rank_constant_eta return [ @@ -201,7 +199,7 @@ def weighted_reciprocal_rank_fuse( ) assert captured["eta"] == 60 - assert set(captured["weights"].values()) == {1.0} + assert set(captured["channels"]) == {"temporal", "lexical"} assert payload["rankings"][0]["post_title"] == ( "Pricing renegotiation: revised quote sent" ) diff --git a/tests/test_relation_verification_internal.py b/tests/test_relation_verification_internal.py index 760d2ad45..b81ae9a7b 100644 --- a/tests/test_relation_verification_internal.py +++ b/tests/test_relation_verification_internal.py @@ -2,7 +2,10 @@ import asyncio -from backend.app.relation_verification_ingestion import verify_post_relations +from backend.app.relation_verification_ingestion import ( + verify_post_relations, + verify_post_relations_from_pool, +) from lineageweave.relation_verification import ( STATUS_CORROBORATED, RelationVerificationResult, @@ -35,9 +38,47 @@ async def execute(self, query: str, *args: object): self.execute_args = args return "UPDATE 1" + def transaction(self): + return _Transaction() + + +class _Transaction: + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, traceback): + return False + + +class _Acquire: + def __init__(self, pool: "_Pool") -> None: + self.pool = pool + + async def __aenter__(self): + assert not self.pool.acquired + self.pool.acquired = True + return self.pool.connection + + async def __aexit__(self, exc_type, exc, traceback): + self.pool.acquired = False + + +class _Pool: + def __init__(self, connection: _Connection) -> None: + self.connection = connection + self.acquired = False + + def acquire(self): + return _Acquire(self) + class _Verifier: + def __init__(self, pool: _Pool | None = None) -> None: + self.pool = pool + def verify(self, organization_name: str, relationship_label: str) -> RelationVerificationResult: + if self.pool is not None: + assert not self.pool.acquired assert (organization_name, relationship_label) == ("Example Partner", "Partner") return RelationVerificationResult(STATUS_CORROBORATED, "https://example.test/evidence") @@ -68,3 +109,20 @@ def test_relation_verification_keeps_external_result_when_internal_search_misses assert verified[0].verification_evidence_post_id is None assert conn.execute_args is not None assert conn.execute_args[-1] is None + + +def test_pool_connection_is_released_during_external_verification() -> None: + conn = _Connection("internal-post") + pool = _Pool(conn) + + verified = asyncio.run( + verify_post_relations_from_pool( + pool, + _Verifier(pool), + "origin-post", + visible_corporate_entity_ids=("corp-a",), + ) + ) + + assert verified[0].verification_status_code == STATUS_CORROBORATED + assert not pool.acquired diff --git a/tests/test_schema.py b/tests/test_schema.py index d88d07bd2..32794f34e 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -14,15 +14,19 @@ from __future__ import annotations +import asyncio import os import uuid from pathlib import Path from urllib.parse import urlsplit, urlunsplit +import asyncpg import psycopg2 import psycopg2.errors import pytest +from backend.app.post_chat_ingestion import gather_global_chat_sources + _ADMIN_DSN = os.environ.get( "LINEAGEWEAVE_TEST_POSTGRES_ADMIN_DSN", "postgresql://localhost/postgres" ) @@ -33,6 +37,27 @@ _PROJECT_MENTION_MIGRATION = ( Path(__file__).resolve().parents[1] / "migrations" / "0031_semantic_project_mentions.sql" ) +_POST_CONTENT_MIGRATION = ( + Path(__file__).resolve().parents[1] / "migrations" / "0026_post_content_artifacts.sql" +) +_SOURCE_STATE_MIGRATION = ( + Path(__file__).resolve().parents[1] / "migrations" / "0033_source_state_provenance.sql" +) +_SOURCE_CONTEXT_MIGRATION = ( + Path(__file__).resolve().parents[1] / "migrations" / "0034_source_context_provenance.sql" +) +_SOURCE_IDENTITY_MIGRATION = ( + Path(__file__).resolve().parents[1] / "migrations" / "0037_source_record_identity.sql" +) +_SOURCE_NAMED_HINTS_MIGRATION = ( + Path(__file__).resolve().parents[1] / "migrations" / "0038_source_named_hints.sql" +) +_SOURCE_ORG_HINTS_MIGRATION = ( + Path(__file__).resolve().parents[1] / "migrations" / "0039_source_org_named_hints.sql" +) +_SOURCE_EVENT_TIME_MIGRATION = ( + Path(__file__).resolve().parents[1] / "migrations" / "0183_source_post_event_occurred_at.sql" +) _PROJECT_BOUND_ACTION_MIGRATION = ( Path(__file__).resolve().parents[1] / "migrations" @@ -73,6 +98,11 @@ / "migrations" / "0169_report_leftover_map_axis.sql" ) +_GLOBAL_ASK_EVIDENCE_SEARCH_MIGRATION = ( + Path(__file__).resolve().parents[1] + / "migrations" + / "0210_global_ask_evidence_search_indexes.sql" +) _CHANNEL_EVIDENCE_MIGRATION = ( Path(__file__).resolve().parents[1] / "migrations" / "0174_post_lineage_edge_signal.sql" ) @@ -118,7 +148,14 @@ def schema_db(): try: with conn.cursor() as cur: cur.execute(_MIGRATION_PATH.read_text()) + cur.execute(_POST_CONTENT_MIGRATION.read_text()) cur.execute(_PROJECT_MENTION_MIGRATION.read_text()) + cur.execute(_SOURCE_STATE_MIGRATION.read_text()) + cur.execute(_SOURCE_CONTEXT_MIGRATION.read_text()) + cur.execute("create extension if not exists pg_trgm") + cur.execute(_SOURCE_IDENTITY_MIGRATION.read_text()) + cur.execute(_SOURCE_NAMED_HINTS_MIGRATION.read_text()) + cur.execute(_SOURCE_ORG_HINTS_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()) @@ -131,6 +168,8 @@ def schema_db(): cur.execute(_LEFTOVER_MAP_UNEXPLAINED_MIGRATION.read_text()) cur.execute(_LEFTOVER_MAP_CROSS_SHARE_MIGRATION.read_text()) cur.execute(_LEFTOVER_MAP_RECONSTRUCTION_MIGRATION.read_text()) + cur.execute(_SOURCE_EVENT_TIME_MIGRATION.read_text()) + cur.execute(_GLOBAL_ASK_EVIDENCE_SEARCH_MIGRATION.read_text()) conn.commit() yield conn finally: @@ -189,6 +228,174 @@ def test_migration_applies_cleanly(schema_db) -> None: assert expected <= tables +def test_global_ask_evidence_search_indexes_exist_on_normalized_tables(schema_db) -> None: + """The real PostgreSQL schema owns all nine evidence-search indexes.""" + with schema_db.cursor() as cur: + cur.execute( + """ + select indexname + from pg_indexes + where schemaname = 'public' + and (indexname like '%_evidence_search_idx' + or indexname = 'knowledge_graph_edge_type_search_idx') + order by indexname + """ + ) + index_names = [row[0] for row in cur.fetchall()] + + assert index_names == [ + "cataloged_person_evidence_search_idx", + "cataloged_team_evidence_search_idx", + "common_lookup_value_evidence_search_idx", + "corporate_entity_evidence_search_idx", + "knowledge_graph_edge_type_search_idx", + "person_affiliation_evidence_search_idx", + "post_project_mention_evidence_search_idx", + "post_summary_role_evidence_search_idx", + "source_post_title_evidence_search_idx", + ] + + +def test_global_ask_nominates_a_live_semantic_only_post(schema_db) -> None: + """Real PostgreSQL retrieves a post whose query term exists only in evidence.""" + post_id = "30000000-0000-0000-0000-000000000001" + with schema_db.cursor() as cur: + cur.execute( + """ + insert into common_lookup_value + (lookup_category, lookup_code, lookup_label) + values + ('corporate_entity_level', 'semantic_test_level', 'Synthetic level'), + ('voc_type', 'semantic_test_voc', 'Synthetic VOC'), + ('post_visibility', 'semantic_test_public', 'Synthetic public'), + ('person_side', 'semantic_test_person_side', 'Synthetic person side'), + ('node_type', 'node_person', 'Person'), + ('node_type', 'node_corporate_entity', 'Corporate entity'), + ('edge_type', 'edge_affiliation', 'Affiliated with') + """ + ) + cur.execute( + """ + insert into corporate_entity + (corporate_entity_id, corporate_entity_code, entity_name, + entity_level_code) + values + ('10000000-0000-0000-0000-000000000001', 'SYNTH-CORP', + 'Synthetic Corp', 'semantic_test_level') + """ + ) + cur.execute( + """ + insert into user_account + (user_account_id, external_subject_id, display_name, email_address) + values + ('20000000-0000-0000-0000-000000000001', 'synthetic-subject', + 'Synthetic User', 'synthetic@example.invalid') + """ + ) + cur.execute( + """ + insert into source_post + (post_id, author_account_id, corporate_entity_id, post_title, + post_body, voc_type_code, visibility_code) + values + (%s, '20000000-0000-0000-0000-000000000001', + '10000000-0000-0000-0000-000000000001', 'Neutral title', + 'Neutral body', 'semantic_test_voc', 'semantic_test_public') + """, + (post_id,), + ) + cur.execute( + """ + insert into post_project_mention + (post_id, project_key, project_name, evidence_text, confidence, + ontology_iri, extraction_method) + values + (%s, 'semantic-project', 'Exclusive Semantic Project', + 'Synthetic project evidence', 1.000, + 'https://contextualwisdomlab.github.io/LineageWeave/ontology#Project', + 'synthetic_test') + """, + (post_id,), + ) + cur.execute( + """ + insert into cataloged_person + (person_id, person_name, person_side_code, last_known_job_title) + values + ('40000000-0000-0000-0000-000000000001', 'Synthetic Expert', + 'semantic_test_person_side', 'Synthetic Reviewer') + """ + ) + cur.execute( + """ + insert into person_affiliation + (person_id, affiliated_organization_name, + affiliated_corporate_entity_id, role_title) + values + ('40000000-0000-0000-0000-000000000001', 'Synthetic Corp', + '10000000-0000-0000-0000-000000000001', 'Synthetic Reviewer') + """ + ) + cur.execute( + """ + insert into post_person_mention (post_id, person_id, mention_context) + values (%s, '40000000-0000-0000-0000-000000000001', 'Synthetic evidence') + """, + (post_id,), + ) + cur.execute( + """ + insert into knowledge_graph_edge + (knowledge_graph_edge_id, source_node_type_code, source_node_id, + target_node_type_code, target_node_id, edge_type_code) + values + ('50000000-0000-0000-0000-000000000001', 'node_person', + '40000000-0000-0000-0000-000000000001', + 'node_corporate_entity', + '10000000-0000-0000-0000-000000000001', 'edge_affiliation') + """ + ) + schema_db.commit() + + async def retrieve(question: str) -> list: + conn = await asyncpg.connect( + database=schema_db.info.dbname, + host=schema_db.info.host or "localhost", + port=schema_db.info.port, + user=schema_db.info.user, + ) + try: + return await gather_global_chat_sources( + conn, + lambda row: row["visibility_code"] == "semantic_test_public", + ["10000000-0000-0000-0000-000000000001"], + question=question, + question_embedding=([1.0, 0.0], "synthetic-model", 1.0), + limit=4, + ) + finally: + await conn.close() + + sources = asyncio.run(retrieve("Exclusive Semantic Project")) + + assert [source.post_id for source in sources] == [post_id] + assert any( + "Exclusive Semantic Project" in fact for fact in sources[0].evidence_facts + ) + assert [source.post_id for source in asyncio.run(retrieve("Synthetic Expert"))] == [ + post_id + ] + assert [ + source.post_id + for source in asyncio.run( + retrieve( + "https://contextualwisdomlab.github.io/LineageWeave/ontology#affiliatedWith" + ) + ) + ] == [post_id] + + def test_post_lineage_edge_requires_an_allen_interval_code(schema_db) -> None: with schema_db.cursor() as cur: cur.execute(