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 8bda198ff..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, ) @@ -1838,9 +1840,7 @@ async def read_similar_voc( "similar VOC inference is unavailable; configure contextual-orchestrator and retry", ) async with pool.acquire() as conn: - # Safe SQL: the sole interpolation is a repository-owned eligibility - # expression rendered with the fixed ``post`` alias; every request and - # identity value remains an asyncpg parameter. + # Safe SQL: the sole interpolation is the closed eligibility fragment; request and identity values remain asyncpg parameters. rows = await conn.fetch( # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli f""" select post.post_id, post.post_title, post.post_body, @@ -2312,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, @@ -3388,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, @@ -3602,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/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 3d3d963b1..a826fc013 100644 --- a/backend/tests/test_config.py +++ b/backend/tests/test_config.py @@ -50,6 +50,14 @@ def test_tepp_transport_defaults_empty_and_preserve_runtime_credentials(monkeypa 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: """Production Keyverse configuration is standard OIDC, not a local mock.""" monkeypatch.setenv("KEYVERSE_ISSUER", "https://keyverse.example/tenant/acme") diff --git a/docs/operability/http-concurrency-evidence.md b/docs/operability/http-concurrency-evidence.md index c89092e25..69bbfd60f 100644 --- a/docs/operability/http-concurrency-evidence.md +++ b/docs/operability/http-concurrency-evidence.md @@ -63,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 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_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