From 0545560456f76da68b55af5f0e5cce71a3120917 Mon Sep 17 00:00:00 2001 From: seonghobae Date: Tue, 25 Aug 2026 21:39:32 +0900 Subject: [PATCH 01/23] fix(ask): release pool before embedding provider work --- Makefile | 3 +- backend/app/config.py | 2 + backend/app/global_ask_queue.py | 28 ++++++---- backend/app/post_chat_ingestion.py | 36 +++++++++---- backend/tests/test_config.py | 14 +++-- docker-compose.yml | 1 + .../0212-global-ask-embedding-pool-release.md | 40 ++++++++++++++ docs/adr/README.md | 3 +- docs/operability/http-concurrency-evidence.md | 29 +++++++++- docs/product-requirements.md | 2 +- docs/product-technical-gap-baseline.md | 2 +- .../0203_global_ask_authorization_scope.sql | 4 +- scripts/k6_http_e2e.js | 14 +++-- tests/test_global_ask_queue.py | 54 +++++++++++++++++++ 14 files changed, 195 insertions(+), 37 deletions(-) create mode 100644 docs/adr/0212-global-ask-embedding-pool-release.md 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..cdbcf9439 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,16 +238,23 @@ def can_see(row: asyncpg.Record) -> bool: today = _seoul_today() try: - async with pool.acquire() as conn: - sources = await gather_global_chat_sources( - conn, - can_see, - corporate_entity_ids, - process_unit_ids, - question=question_text, - today=today, - embedding_client=embedding_client, - ) + question_embedding = await prepare_global_question_embedding( + question_text, embedding_client or NullEmbeddingClient() + ) + if question_embedding is None: + sources = [] + else: + async with pool.acquire() as conn: + sources = await gather_global_chat_sources( + conn, + can_see, + corporate_entity_ids, + process_unit_ids, + question=question_text, + question_embedding=question_embedding, + today=today, + embedding_client=embedding_client, + ) except Exception as exc: log_internal_fault("global_ask", exc) record_server_failure("global_ask", exc, outcome="internal_error") diff --git a/backend/app/post_chat_ingestion.py b/backend/app/post_chat_ingestion.py index 4ca9e2f2d..89451ff27 100644 --- a/backend/app/post_chat_ingestion.py +++ b/backend/app/post_chat_ingestion.py @@ -385,6 +385,24 @@ 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.""" + try: + question_vector = await asyncio.to_thread(embedding_client.embed, question) + except (OSError, RuntimeError, ValueError): + return None + embedding_model_code = embedding_client.resolved_model + if not question_vector or not embedding_model_code: + return None + question_norm = sum(value * value for value in question_vector) ** 0.5 + if question_norm == 0.0: + return None + return question_vector, embedding_model_code, question_norm + + async def gather_global_chat_sources( conn: asyncpg.Connection, can_see_post: Callable[[asyncpg.Record], bool], @@ -394,6 +412,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]: @@ -429,18 +448,13 @@ async def gather_global_chat_sources( ) if not (question and question.strip() and embedding_client.available): 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: + if question_embedding is None: + question_embedding = await prepare_global_question_embedding( + question, embedding_client + ) + if question_embedding is None: return [] + question_vector, embedding_model_code, question_norm = question_embedding # 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 diff --git a/backend/tests/test_config.py b/backend/tests/test_config.py index 2a80f1fa4..3d3d963b1 100644 --- a/backend/tests/test_config.py +++ b/backend/tests/test_config.py @@ -36,12 +36,18 @@ 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_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/0212-global-ask-embedding-pool-release.md b/docs/adr/0212-global-ask-embedding-pool-release.md new file mode 100644 index 000000000..23be73066 --- /dev/null +++ b/docs/adr/0212-global-ask-embedding-pool-release.md @@ -0,0 +1,40 @@ +# ADR 0212 — 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 a7704da7e..1ff877dc3 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -17,7 +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) | +| [`POSTGRESQL_CONCURRENCY_REFERENCES.md`](../doctoring/POSTGRESQL_CONCURRENCY_REFERENCES.md) | [0204](0204-analysis-run-short-transaction-delivery.md), [0212](0212-global-ask-embedding-pool-release.md) | +| [`http-concurrency-evidence.md`](../operability/http-concurrency-evidence.md) | [0212](0212-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 9e846f7ba..fd4206a7f 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,3 +63,25 @@ steps ranged up to 292.3 seconds. No containers were running afterward, so no 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 0212 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. diff --git a/docs/product-requirements.md b/docs/product-requirements.md index 96f66733b..80dd5c5bf 100644 --- a/docs/product-requirements.md +++ b/docs/product-requirements.md @@ -168,7 +168,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 0212. - 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 d922aca0e..599db6051 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -369,7 +369,7 @@ 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 now has an authenticated k6 E2E harness for Ask enqueue, concurrent reads, and job polling. No environment-specific measurement is a protected-main product guarantee | Run `make load-http` with declared environment concurrency/window; retain raw distributions and resource configuration, then diagnose bottlenecks with backend, PostgreSQL, Valkey, and orchestrator evidence before setting any approved SLO | +| Concurrent web responsiveness | ADR 0204 releases analysis-run transactions; the first application-ready k6 exercise then found Global Ask embedding still inside `pool.acquire()`. ADR 0212 moves it before acquisition. The post-fix read batch ran while replay was still building migration 0035 and before schema migration 0140, so no steady-state capacity distribution is accepted | Re-run `make load-http` with declared concurrency, request window, and observation window after migration replay completes; retain raw distributions and resource configuration, then diagnose any remaining bottleneck before setting an approved SLO | | 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 | 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/scripts/k6_http_e2e.js b/scripts/k6_http_e2e.js index d9b79bf25..e38070b45 100644 --- a/scripts/k6_http_e2e.js +++ b/scripts/k6_http_e2e.js @@ -16,12 +16,16 @@ 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 askEnqueueDuration = new Trend("lineageweave_ask_enqueue_duration", true); const readDuration = new Trend("lineageweave_read_duration", true); const askPollDuration = new Trend("lineageweave_ask_poll_duration", true); export function setup() { + if (!requestTimeout) { + fail("REQUEST_TIMEOUT is required"); + } const tokenResponse = http.post( `${keycloakUrl}/realms/${realm}/protocol/openid-connect/token`, { @@ -30,7 +34,7 @@ export function setup() { username, password, }, - { tags: { endpoint: "oidc_token" } }, + { tags: { endpoint: "oidc_token" }, timeout: requestTimeout }, ); if (tokenResponse.status !== 200) { fail(`synthetic OIDC login failed with HTTP ${tokenResponse.status}`); @@ -41,7 +45,7 @@ export function setup() { 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) { @@ -53,13 +57,13 @@ export function setup() { export default function (data) { const params = { headers: { Authorization: `Bearer ${data.token}` } }; const responses = 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/${data.askJobId}`, null, - { ...params, tags: { endpoint: "ask_poll" } }, + { ...params, tags: { endpoint: "ask_poll" }, timeout: requestTimeout }, ], ]); diff --git a/tests/test_global_ask_queue.py b/tests/test_global_ask_queue.py index 76b07170c..915e0223b 100644 --- a/tests/test_global_ask_queue.py +++ b/tests/test_global_ask_queue.py @@ -42,6 +42,60 @@ 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_unexpected_job_failure_settles_with_a_generic_detail_not_the_raw_exception( monkeypatch, ) -> None: From 45b0a44b59d75ccf5c0f026b13e5348420cac4a5 Mon Sep 17 00:00:00 2001 From: seonghobae Date: Tue, 25 Aug 2026 21:39:55 +0900 Subject: [PATCH 02/23] style: keep load evidence reviewable --- docs/adr/0212-global-ask-embedding-pool-release.md | 4 ++-- scripts/k6_http_e2e.js | 14 ++++++++++++-- 2 files changed, 14 insertions(+), 4 deletions(-) diff --git a/docs/adr/0212-global-ask-embedding-pool-release.md b/docs/adr/0212-global-ask-embedding-pool-release.md index 23be73066..0c68861ec 100644 --- a/docs/adr/0212-global-ask-embedding-pool-release.md +++ b/docs/adr/0212-global-ask-embedding-pool-release.md @@ -1,7 +1,7 @@ # ADR 0212 — Global Ask embeds before acquiring a pooled connection -**Decision status:** Accepted -**Date:** 2026-08-25 +**Decision status:** Accepted +**Date:** 2026-08-25 **Related:** [0204](0204-analysis-run-short-transaction-delivery.md) ## Context diff --git a/scripts/k6_http_e2e.js b/scripts/k6_http_e2e.js index e38070b45..68eca5878 100644 --- a/scripts/k6_http_e2e.js +++ b/scripts/k6_http_e2e.js @@ -57,8 +57,18 @@ export function setup() { export default function (data) { const params = { headers: { Authorization: `Bearer ${data.token}` } }; const responses = http.batch([ - ["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/posts`, + null, + { ...params, tags: { endpoint: "posts" }, timeout: requestTimeout }, + ], + [ + "GET", + `${backendUrl}/api/lineage`, + null, + { ...params, tags: { endpoint: "lineage" }, timeout: requestTimeout }, + ], [ "GET", `${backendUrl}/api/ask/jobs/${data.askJobId}`, From 3d69ea4f5fc593a74b3be767b051d9d182fa1603 Mon Sep 17 00:00:00 2001 From: seonghobae Date: Tue, 25 Aug 2026 21:43:30 +0900 Subject: [PATCH 03/23] docs(gaps): refresh protected delivery evidence --- docs/product-technical-gap-baseline.md | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 774f79a10..a58e9d302 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 | @@ -362,7 +371,7 @@ this file per §3.5 of the prior snapshot). | Gap | Current evidence | Acceptance requirement | | --- | --- | --- | -| Protected release | 3 open PRs at snapshot: #627 and #628 are current-main performance follow-ups, while reopened #579 retains hosted and independent-review gates | Terminal exact-head checks, no unresolved threads, independent exact-head approvals, protected squash-merge SHA | +| Protected release | One open PR at snapshot: #629 carries ADR 0213 and the measured Global Ask pool-release repair; #579 closed without merge. #624–#628 are protected-main delivery, not open work | #629 needs terminal checks, no unresolved threads, independent exact-head approval, and a protected squash-merge SHA; then re-fetch the queue rather than treating this snapshot as live state | | Evidence-grounded operations workspace | Protected-main #614 delivers governed semantic Ask, live Similar VOC, disjoint pending/failed analysis metrics, full Storybook state inventory, and current desktop/mobile screenshot evidence. Authorized-corpus backfill acceptance remains unavailable | Perform authenticated authorized-corpus acceptance with aggregate evidence and retain fail-closed no-match behavior | | 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 | From 8797605b0aefbce4d9023d86267e909c90a8bc0e Mon Sep 17 00:00:00 2001 From: seonghobae Date: Tue, 25 Aug 2026 21:49:02 +0900 Subject: [PATCH 04/23] fix(ask): reject blank embedding requests --- backend/app/post_chat_ingestion.py | 2 ++ docs/product-technical-gap-baseline.md | 2 +- tests/test_global_ask_sources.py | 22 +++++++++++++++++++++- 3 files changed, 24 insertions(+), 2 deletions(-) diff --git a/backend/app/post_chat_ingestion.py b/backend/app/post_chat_ingestion.py index 89451ff27..98ad242f7 100644 --- a/backend/app/post_chat_ingestion.py +++ b/backend/app/post_chat_ingestion.py @@ -390,6 +390,8 @@ async def prepare_global_question_embedding( embedding_client: EmbeddingClient, ) -> tuple[list[float], str, float] | None: """Resolve one question embedding without holding a database connection.""" + if not question.strip(): + return None try: question_vector = await asyncio.to_thread(embedding_client.embed, question) except (OSError, RuntimeError, ValueError): diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index a58e9d302..e523f5c5a 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -371,7 +371,7 @@ this file per §3.5 of the prior snapshot). | Gap | Current evidence | Acceptance requirement | | --- | --- | --- | -| Protected release | One open PR at snapshot: #629 carries ADR 0213 and the measured Global Ask pool-release repair; #579 closed without merge. #624–#628 are protected-main delivery, not open work | #629 needs terminal checks, no unresolved threads, independent exact-head approval, and a protected squash-merge SHA; then re-fetch the queue rather than treating this snapshot as live state | +| Protected release | 3 open PRs at snapshot: #627 and #628 are current-main performance follow-ups, while reopened #579 retains hosted and independent-review gates | Terminal exact-head checks, no unresolved threads, independent exact-head approvals, protected squash-merge SHA | | Evidence-grounded operations workspace | Protected-main #614 delivers governed semantic Ask, live Similar VOC, disjoint pending/failed analysis metrics, full Storybook state inventory, and current desktop/mobile screenshot evidence. Authorized-corpus backfill acceptance remains unavailable | Perform authenticated authorized-corpus acceptance with aggregate evidence and retain fail-closed no-match behavior | | 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 | diff --git a/tests/test_global_ask_sources.py b/tests/test_global_ask_sources.py index f7fc31ae7..2550e18c9 100644 --- a/tests/test_global_ask_sources.py +++ b/tests/test_global_ask_sources.py @@ -3,7 +3,10 @@ import asyncio 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 ( + 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,6 +18,23 @@ 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 gather_global_chat_sources(*args, **kwargs): """Exercise Global Ask with an available deterministic semantic channel.""" kwargs.setdefault("embedding_client", _EmbeddingClient()) From 2e4fbc57614a2d26b464ee00ec133d733b5d25cc Mon Sep 17 00:00:00 2001 From: seonghobae Date: Tue, 25 Aug 2026 21:56:18 +0900 Subject: [PATCH 05/23] fix(ask): preserve unavailable embedding short circuit --- backend/app/post_chat_ingestion.py | 2 ++ tests/test_global_ask_queue.py | 27 +++++++++++++++++++++++++++ 2 files changed, 29 insertions(+) diff --git a/backend/app/post_chat_ingestion.py b/backend/app/post_chat_ingestion.py index 89451ff27..7b537001e 100644 --- a/backend/app/post_chat_ingestion.py +++ b/backend/app/post_chat_ingestion.py @@ -390,6 +390,8 @@ async def prepare_global_question_embedding( 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): diff --git a/tests/test_global_ask_queue.py b/tests/test_global_ask_queue.py index 915e0223b..aeeb42885 100644 --- a/tests/test_global_ask_queue.py +++ b/tests/test_global_ask_queue.py @@ -96,6 +96,33 @@ async def fake_gather(_conn, *_args, **kwargs): assert pool.active == 0 +def test_unavailable_question_embedding_is_not_called(monkeypatch) -> None: + """An unavailable channel is dropped without invoking its transport.""" + 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") + + 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: From 99f6b815fdd1b72a4ae150d718b7713f33128b3f Mon Sep 17 00:00:00 2001 From: seonghobae Date: Tue, 25 Aug 2026 22:02:20 +0900 Subject: [PATCH 06/23] fix(ask): honor validated precomputed embeddings --- backend/app/post_chat_ingestion.py | 4 +++- tests/test_global_ask_sources.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 33 insertions(+), 1 deletion(-) diff --git a/backend/app/post_chat_ingestion.py b/backend/app/post_chat_ingestion.py index 7b537001e..46fe6c890 100644 --- a/backend/app/post_chat_ingestion.py +++ b/backend/app/post_chat_ingestion.py @@ -448,7 +448,9 @@ 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 or not question.strip(): + return [] + if question_embedding is None and not embedding_client.available: return [] if question_embedding is None: question_embedding = await prepare_global_question_embedding( diff --git a/tests/test_global_ask_sources.py b/tests/test_global_ask_sources.py index 2550e18c9..61c16e3e4 100644 --- a/tests/test_global_ask_sources.py +++ b/tests/test_global_ask_sources.py @@ -345,6 +345,36 @@ async def fetch(self, _query: str, *_args): assert sources == [] +def test_global_sources_accept_a_precomputed_available_embedding() -> None: + """A validated precomputed vector does not re-enter its provider channel.""" + queries: list[str] = [] + + class UnavailableEmbedding: + available = False + resolved_model = None + + def embed(self, _text: str) -> list[float]: + raise AssertionError("precomputed embedding must not call its provider") + + class FakeConnection: + async def fetch(self, query: str, *_args): + queries.append(query) + return [] + + sources = asyncio.run( + _gather_global_chat_sources( + FakeConnection(), + lambda _row: True, + question="semantic question", + embedding_client=UnavailableEmbedding(), + question_embedding=([1.0, 0.0], "synthetic-embedding", 1.0), + ) + ) + + assert sources == [] + assert "with question_vector" in queries[0] + + def test_global_sources_fail_closed_without_a_resolved_embedding_model() -> None: """A vector without its orchestrator-resolved model cannot match persisted rows.""" From 23e3fdeb8f8aae78e3f45cb5809cb2ce424b2487 Mon Sep 17 00:00:00 2001 From: seonghobae Date: Tue, 25 Aug 2026 22:02:38 +0900 Subject: [PATCH 07/23] fix(ask): honor precomputed embedding envelope --- backend/app/post_chat_ingestion.py | 4 +++- tests/test_global_ask_sources.py | 33 ++++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/backend/app/post_chat_ingestion.py b/backend/app/post_chat_ingestion.py index 7b537001e..0d2c2241f 100644 --- a/backend/app/post_chat_ingestion.py +++ b/backend/app/post_chat_ingestion.py @@ -448,9 +448,11 @@ 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 [] if question_embedding is None: + if not embedding_client.available: + return [] question_embedding = await prepare_global_question_embedding( question, embedding_client ) diff --git a/tests/test_global_ask_sources.py b/tests/test_global_ask_sources.py index 2550e18c9..c97278e12 100644 --- a/tests/test_global_ask_sources.py +++ b/tests/test_global_ask_sources.py @@ -345,6 +345,39 @@ async def fetch(self, _query: str, *_args): assert sources == [] +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_fail_closed_without_a_resolved_embedding_model() -> None: """A vector without its orchestrator-resolved model cannot match persisted rows.""" From 445571abace1baadf01fe0f060d94dfe0d7b38e9 Mon Sep 17 00:00:00 2001 From: seonghobae Date: Tue, 25 Aug 2026 22:09:48 +0900 Subject: [PATCH 08/23] fix(k6): reject unitless request timeouts --- scripts/k6_http_e2e.js | 4 ++++ tests/test_k6_http_e2e_contract.py | 2 ++ 2 files changed, 6 insertions(+) diff --git a/scripts/k6_http_e2e.js b/scripts/k6_http_e2e.js index 111f6ba0f..976f63c73 100644 --- a/scripts/k6_http_e2e.js +++ b/scripts/k6_http_e2e.js @@ -17,6 +17,7 @@ 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); @@ -70,6 +71,9 @@ 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( 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 From ca5d304579439f30f803cd8040e273e5a2658ea9 Mon Sep 17 00:00:00 2001 From: seonghobae Date: Tue, 25 Aug 2026 22:14:13 +0900 Subject: [PATCH 09/23] fix(migrations): replay global ask queue safely --- docs/operability/http-concurrency-evidence.md | 22 +++++++++++++++++++ docs/product-technical-gap-baseline.md | 2 +- migrations/0165_global_ask_job.sql | 6 ++--- tests/test_migration_replay.py | 13 +++++++++++ 4 files changed, 39 insertions(+), 4 deletions(-) diff --git a/docs/operability/http-concurrency-evidence.md b/docs/operability/http-concurrency-evidence.md index 79ac4e64b..c89092e25 100644 --- a/docs/operability/http-concurrency-evidence.md +++ b/docs/operability/http-concurrency-evidence.md @@ -95,6 +95,28 @@ 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-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index a58e9d302..e57f9309a 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -376,7 +376,7 @@ 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 analysis-run transactions. An older-image observation found repeated post-filter queries, so ADR 0212 combines the authorized filter-option query without narrowing ABAC. The next application-ready exercise found Global Ask embedding inside `pool.acquire()`, so ADR 0213 moves it before acquisition. Its post-fix batch ran during incomplete migration replay, so no steady-state capacity result is accepted | Rebuild an exact-head image after migration replay, run `make load-http` with declared concurrency, request window, and observation window, retain raw distributions and resource configuration, and compare the post-list plan while verifying zero pool slots are held during embedding; set no SLO until representative 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 | 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/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 From 71c10ddcdea45bd4e75a773204b0774f3db1965b Mon Sep 17 00:00:00 2001 From: seonghobae Date: Tue, 25 Aug 2026 22:14:31 +0900 Subject: [PATCH 10/23] fix(ask): reject nonfinite embeddings --- backend/app/post_chat_ingestion.py | 28 +++++++++++++++++++++----- docs/product-technical-gap-baseline.md | 2 +- tests/test_global_ask_sources.py | 28 ++++++++++++++++++++++++++ 3 files changed, 52 insertions(+), 6 deletions(-) diff --git a/backend/app/post_chat_ingestion.py b/backend/app/post_chat_ingestion.py index 0d2c2241f..5ae54f9dd 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 @@ -396,11 +397,23 @@ async def prepare_global_question_embedding( question_vector = await asyncio.to_thread(embedding_client.embed, question) except (OSError, RuntimeError, ValueError): return None - embedding_model_code = embedding_client.resolved_model - if not question_vector or not embedding_model_code: + 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 = sum(value * value for value in question_vector) ** 0.5 - if question_norm == 0.0: + 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 @@ -458,7 +471,12 @@ async def gather_global_chat_sources( ) if question_embedding is None: return [] - question_vector, embedding_model_code, question_norm = question_embedding + validated_embedding = _validated_question_embedding( + question_embedding[0], question_embedding[1] + ) + if validated_embedding is None: + return [] + question_vector, embedding_model_code, question_norm = validated_embedding # 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 diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index a58e9d302..e523f5c5a 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -371,7 +371,7 @@ this file per §3.5 of the prior snapshot). | Gap | Current evidence | Acceptance requirement | | --- | --- | --- | -| Protected release | One open PR at snapshot: #629 carries ADR 0213 and the measured Global Ask pool-release repair; #579 closed without merge. #624–#628 are protected-main delivery, not open work | #629 needs terminal checks, no unresolved threads, independent exact-head approval, and a protected squash-merge SHA; then re-fetch the queue rather than treating this snapshot as live state | +| Protected release | 3 open PRs at snapshot: #627 and #628 are current-main performance follow-ups, while reopened #579 retains hosted and independent-review gates | Terminal exact-head checks, no unresolved threads, independent exact-head approvals, protected squash-merge SHA | | Evidence-grounded operations workspace | Protected-main #614 delivers governed semantic Ask, live Similar VOC, disjoint pending/failed analysis metrics, full Storybook state inventory, and current desktop/mobile screenshot evidence. Authorized-corpus backfill acceptance remains unavailable | Perform authenticated authorized-corpus acceptance with aggregate evidence and retain fail-closed no-match behavior | | 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 | diff --git a/tests/test_global_ask_sources.py b/tests/test_global_ask_sources.py index c97278e12..e35b602b2 100644 --- a/tests/test_global_ask_sources.py +++ b/tests/test_global_ask_sources.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import math from datetime import date, datetime, timezone from backend.app.post_chat_ingestion import ( @@ -35,6 +36,33 @@ def embed(self, _text: str) -> list[float]: ) +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()) From ccb4bb982b880adca9966d63b0bcfdfe0f8b4edc Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 25 Aug 2026 06:37:13 -0700 Subject: [PATCH 11/23] perf: keep authenticated web reads responsive (#633) * fix(backend): make the similar-VOC SQL audit reason adjacent (hotfix main) The similar-VOC candidate fetch already carried a suppression, but its Safe SQL reason sat three lines above the audited call while the review contract requires the immediately preceding line. Collapse the comment to one adjacent line; the counted total stays 36 because this repairs an existing site rather than adding one. * perf: keep authenticated web reads responsive * docs: record authenticated capacity comparison --------- Co-authored-by: seonghobae --- backend/app/lineage_ingestion.py | 50 ++++++++++- backend/app/main.py | 59 ++++++------- .../app/relation_verification_ingestion.py | 84 ++++++++++++++++++- backend/app/report_ingestion.py | 4 +- backend/tests/test_config.py | 8 ++ docs/operability/http-concurrency-evidence.md | 31 +++++++ tests/test_lineage_ingestion.py | 30 +++++++ tests/test_relation_verification_internal.py | 60 ++++++++++++- 8 files changed, 291 insertions(+), 35 deletions(-) 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 From 143a6a3fdad7fb25d669c98a2849cc008f968319 Mon Sep 17 00:00:00 2001 From: seonghobae Date: Tue, 25 Aug 2026 22:37:13 +0900 Subject: [PATCH 12/23] fix(verification): persist completed provider results --- .../app/relation_verification_ingestion.py | 25 +++++----- tests/test_relation_verification_internal.py | 49 +++++++++++++++++++ 2 files changed, 60 insertions(+), 14 deletions(-) diff --git a/backend/app/relation_verification_ingestion.py b/backend/app/relation_verification_ingestion.py index 10dcbcf3a..bd2b8b590 100644 --- a/backend/app/relation_verification_ingestion.py +++ b/backend/app/relation_verification_ingestion.py @@ -203,17 +203,13 @@ async def verify_post_relations_from_pool( 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, - ) + completed = 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: + async with pool.acquire() as conn: await conn.execute( """ update post_counterparty_entity @@ -225,9 +221,10 @@ async def verify_post_relations_from_pool( 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, + completed.counterparty_entity_name, + completed.verification_status_code, + completed.verification_evidence_url, + completed.verification_evidence_post_id, ) + verified.append(completed) return verified diff --git a/tests/test_relation_verification_internal.py b/tests/test_relation_verification_internal.py index b81ae9a7b..9dc66048e 100644 --- a/tests/test_relation_verification_internal.py +++ b/tests/test_relation_verification_internal.py @@ -2,6 +2,8 @@ import asyncio +import pytest + from backend.app.relation_verification_ingestion import ( verify_post_relations, verify_post_relations_from_pool, @@ -126,3 +128,50 @@ def test_pool_connection_is_released_during_external_verification() -> None: assert verified[0].verification_status_code == STATUS_CORROBORATED assert not pool.acquired + + +def test_pool_verification_persists_completed_rows_before_provider_failure() -> None: + class MultiConnection(_Connection): + def __init__(self) -> None: + super().__init__(None) + self.persisted_names: list[str] = [] + + async def fetch(self, query: str, post_id: str): + assert "verification_status_code = 'verify_pending'" in query + return [ + { + "counterparty_entity_name": "Example Partner", + "relationship_label": "Partner", + }, + { + "counterparty_entity_name": "Unavailable Partner", + "relationship_label": "Partner", + }, + ] + + async def execute(self, query: str, *args: object): + assert "verification_evidence_post_id = $5" in query + self.persisted_names.append(str(args[1])) + return "UPDATE 1" + + class FailingVerifier: + def verify( + self, organization_name: str, relationship_label: str + ) -> RelationVerificationResult: + assert relationship_label == "Partner" + if organization_name == "Unavailable Partner": + raise OSError("synthetic provider failure") + return RelationVerificationResult( + STATUS_CORROBORATED, "https://example.test/evidence" + ) + + conn = MultiConnection() + pool = _Pool(conn) + + with pytest.raises(OSError, match="synthetic provider failure"): + asyncio.run( + verify_post_relations_from_pool(pool, FailingVerifier(), "origin-post") + ) + + assert conn.persisted_names == ["Example Partner"] + assert not pool.acquired From 238a6cdb76f06d7c4ed235855e46bbdb3afc9678 Mon Sep 17 00:00:00 2001 From: seonghobae Date: Tue, 25 Aug 2026 22:48:52 +0900 Subject: [PATCH 13/23] fix: count only claimed relation verifications --- backend/app/relation_verification_ingestion.py | 5 +++-- tests/test_relation_verification_internal.py | 17 +++++++++++++++++ 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/backend/app/relation_verification_ingestion.py b/backend/app/relation_verification_ingestion.py index bd2b8b590..e4a629d4a 100644 --- a/backend/app/relation_verification_ingestion.py +++ b/backend/app/relation_verification_ingestion.py @@ -210,7 +210,7 @@ async def verify_post_relations_from_pool( relation.internal_evidence_post_id, ) async with pool.acquire() as conn: - await conn.execute( + update_status = await conn.execute( """ update post_counterparty_entity set verification_status_code = $3, @@ -226,5 +226,6 @@ async def verify_post_relations_from_pool( completed.verification_evidence_url, completed.verification_evidence_post_id, ) - verified.append(completed) + if update_status == "UPDATE 1": + verified.append(completed) return verified diff --git a/tests/test_relation_verification_internal.py b/tests/test_relation_verification_internal.py index 9dc66048e..4daa17f7f 100644 --- a/tests/test_relation_verification_internal.py +++ b/tests/test_relation_verification_internal.py @@ -175,3 +175,20 @@ def verify( assert conn.persisted_names == ["Example Partner"] assert not pool.acquired + + +def test_pool_verification_omits_relation_completed_by_concurrent_run() -> None: + class ConcurrentConnection(_Connection): + async def execute(self, query: str, *args: object): + assert "verification_status_code = 'verify_pending'" in query + return "UPDATE 0" + + conn = ConcurrentConnection(None) + pool = _Pool(conn) + + verified = asyncio.run( + verify_post_relations_from_pool(pool, _Verifier(pool), "origin-post") + ) + + assert verified == [] + assert not pool.acquired From 883d1ff72d33f43e6d485311825d7d32f4cf0a3f Mon Sep 17 00:00:00 2001 From: seonghobae Date: Tue, 25 Aug 2026 23:21:09 +0900 Subject: [PATCH 14/23] fix: fence concurrent relation verification and migration drift --- .../app/relation_verification_ingestion.py | 30 ++++++++----- migrations/0165_global_ask_job.sql | 42 +++++++++++++++++++ tests/test_migration_replay.py | 2 + tests/test_relation_verification_internal.py | 21 +++++++++- tests/test_schema.py | 35 ++++++++++++++++ 5 files changed, 118 insertions(+), 12 deletions(-) diff --git a/backend/app/relation_verification_ingestion.py b/backend/app/relation_verification_ingestion.py index e4a629d4a..ecbd7fe62 100644 --- a/backend/app/relation_verification_ingestion.py +++ b/backend/app/relation_verification_ingestion.py @@ -30,6 +30,7 @@ class VerifiedRelation: @dataclass(frozen=True) class _PendingRelation: counterparty_entity_name: str + relationship_type_code: str relationship_label: str internal_evidence_post_id: str | None @@ -114,7 +115,8 @@ async def verify_post_relations( """ rows = await conn.fetch( """ - select c.counterparty_entity_name, v.lookup_label as relationship_label + select c.counterparty_entity_name, c.relationship_type_code, + 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' @@ -137,7 +139,7 @@ async def verify_post_relations( row["counterparty_entity_name"], row["relationship_label"], ) - await conn.execute( + update_status = await conn.execute( """ update post_counterparty_entity set verification_status_code = $3, @@ -145,21 +147,25 @@ async def verify_post_relations( verification_evidence_post_id = $5, verification_checked_at = now() where post_id = $1 and counterparty_entity_name = $2 + and verification_status_code = 'verify_pending' + and relationship_type_code = $6 """, post_id, row["counterparty_entity_name"], result.status_code, result.evidence_url, internal_evidence_post_id, + row["relationship_type_code"], ) - verified.append( - VerifiedRelation( - counterparty_entity_name=row["counterparty_entity_name"], - verification_status_code=result.status_code, - verification_evidence_url=result.evidence_url, - verification_evidence_post_id=internal_evidence_post_id, + if update_status == "UPDATE 1": + verified.append( + VerifiedRelation( + counterparty_entity_name=row["counterparty_entity_name"], + verification_status_code=result.status_code, + verification_evidence_url=result.evidence_url, + verification_evidence_post_id=internal_evidence_post_id, + ) ) - ) return verified @@ -173,7 +179,8 @@ async def verify_post_relations_from_pool( async with pool.acquire() as conn: rows = await conn.fetch( """ - select c.counterparty_entity_name, v.lookup_label as relationship_label + select c.counterparty_entity_name, c.relationship_type_code, + 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' @@ -184,6 +191,7 @@ async def verify_post_relations_from_pool( pending = [ _PendingRelation( str(row["counterparty_entity_name"]), + str(row["relationship_type_code"]), str(row["relationship_label"]), await _find_internal_evidence_post( conn, @@ -219,12 +227,14 @@ async def verify_post_relations_from_pool( verification_checked_at = now() where post_id = $1 and counterparty_entity_name = $2 and verification_status_code = 'verify_pending' + and relationship_type_code = $6 """, post_id, completed.counterparty_entity_name, completed.verification_status_code, completed.verification_evidence_url, completed.verification_evidence_post_id, + relation.relationship_type_code, ) if update_status == "UPDATE 1": verified.append(completed) diff --git a/migrations/0165_global_ask_job.sql b/migrations/0165_global_ask_job.sql index 37fa4b568..44484291d 100644 --- a/migrations/0165_global_ask_job.sql +++ b/migrations/0165_global_ask_job.sql @@ -7,6 +7,48 @@ -- Mirrors the durable-row-plus-stream design post_content_job already -- uses, so a lost stream entry is recovered from the queued rows. +-- Existing volumes must not silently retain a differently-shaped queue table. +-- `IF NOT EXISTS` is idempotent only when the existing object is compatible; +-- fail before any insert path can observe a partial schema. +do $$ +begin + if to_regclass('public.global_ask_job') is not null + and exists ( + select 1 + from (values + ('global_ask_job_id', 'uuid'), + ('requesting_account_id', 'uuid'), + ('question_text', 'text'), + ('job_status_code', 'text'), + ('answer_payload', 'jsonb'), + ('failure_detail', 'text'), + ('created_at', 'timestamp with time zone'), + ('updated_at', 'timestamp with time zone') + ) as required(column_name, data_type) + where not exists ( + select 1 + from information_schema.columns column_info + where column_info.table_schema = 'public' + and column_info.table_name = 'global_ask_job' + and column_info.column_name = required.column_name + and column_info.data_type = required.data_type + ) + ) then + raise exception 'global_ask_job exists with an incompatible schema'; + end if; + if to_regclass('public.global_ask_job_account_idx') is not null + and pg_get_indexdef('public.global_ask_job_account_idx'::regclass) + not ilike '%(requesting_account_id, created_at DESC)%' then + raise exception 'global_ask_job_account_idx exists with an incompatible definition'; + end if; + if to_regclass('public.global_ask_job_queued_idx') is not null + and pg_get_indexdef('public.global_ask_job_queued_idx'::regclass) + not ilike '%(created_at)%where%job_status_code%' then + raise exception 'global_ask_job_queued_idx exists with an incompatible definition'; + end if; +end +$$; + 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), diff --git a/tests/test_migration_replay.py b/tests/test_migration_replay.py index 81a98e28c..582eced1d 100644 --- a/tests/test_migration_replay.py +++ b/tests/test_migration_replay.py @@ -194,4 +194,6 @@ def test_global_ask_job_migrations_are_idempotent_for_replay() -> None: assert "create table if not exists global_ask_job" in job_sql assert job_sql.count("create index if not exists") == 2 + assert "global_ask_job exists with an incompatible schema" in job_sql + assert "pg_get_indexdef" in job_sql assert scope_sql.count("create table if not exists") == 2 diff --git a/tests/test_relation_verification_internal.py b/tests/test_relation_verification_internal.py index 4daa17f7f..2650f6395 100644 --- a/tests/test_relation_verification_internal.py +++ b/tests/test_relation_verification_internal.py @@ -25,6 +25,7 @@ async def fetch(self, query: str, post_id: str): return [ { "counterparty_entity_name": "Example Partner", + "relationship_type_code": "partner", "relationship_label": "Partner", } ] @@ -53,7 +54,7 @@ async def __aexit__(self, exc_type, exc, traceback): class _Acquire: - def __init__(self, pool: "_Pool") -> None: + def __init__(self, pool: _Pool) -> None: self.pool = pool async def __aenter__(self): @@ -100,6 +101,7 @@ def test_relation_verification_persists_authorized_internal_evidence() -> None: STATUS_CORROBORATED, "https://example.test/evidence", "internal-post", + "partner", ) @@ -110,7 +112,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 + assert conn.execute_args[-2] is None + assert conn.execute_args[-1] == "partner" + + +def test_relation_verification_omits_row_changed_during_provider_call() -> None: + """A re-extracted relationship cannot receive an older provider result.""" + + class ChangedConnection(_Connection): + async def execute(self, query: str, *args: object): + assert "relationship_type_code = $6" in query + return "UPDATE 0" + + verified = asyncio.run(verify_post_relations(ChangedConnection(None), _Verifier(), "origin-post")) + assert verified == [] def test_pool_connection_is_released_during_external_verification() -> None: @@ -141,10 +156,12 @@ async def fetch(self, query: str, post_id: str): return [ { "counterparty_entity_name": "Example Partner", + "relationship_type_code": "partner", "relationship_label": "Partner", }, { "counterparty_entity_name": "Unavailable Partner", + "relationship_type_code": "partner", "relationship_label": "Partner", }, ] diff --git a/tests/test_schema.py b/tests/test_schema.py index d88d07bd2..d3c73e343 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -86,6 +86,14 @@ / "migrations" / "0182_report_leftover_map_unexplained.sql" ) +_GLOBAL_ASK_JOB_MIGRATION = ( + Path(__file__).resolve().parents[1] / "migrations" / "0165_global_ask_job.sql" +) +_GLOBAL_ASK_SCOPE_MIGRATION = ( + Path(__file__).resolve().parents[1] + / "migrations" + / "0203_global_ask_authorization_scope.sql" +) def _postgres_available() -> bool: @@ -189,6 +197,33 @@ def test_migration_applies_cleanly(schema_db) -> None: assert expected <= tables +def test_global_ask_migrations_replay_against_the_same_database(schema_db) -> None: + """Queue and authorization migrations execute twice on one real schema.""" + job_sql = _GLOBAL_ASK_JOB_MIGRATION.read_text(encoding="utf-8") + scope_sql = _GLOBAL_ASK_SCOPE_MIGRATION.read_text(encoding="utf-8") + with schema_db.cursor() as cur: + cur.execute(job_sql) + cur.execute(scope_sql) + cur.execute(job_sql) + cur.execute(scope_sql) + cur.execute( + """ + select column_name + from information_schema.columns + where table_schema = 'public' and table_name = 'global_ask_job' + order by ordinal_position + """ + ) + columns = {row[0] for row in cur.fetchall()} + schema_db.commit() + assert { + "global_ask_job_id", + "requesting_account_id", + "question_text", + "job_status_code", + } <= columns + + def test_post_lineage_edge_requires_an_allen_interval_code(schema_db) -> None: with schema_db.cursor() as cur: cur.execute( From ac38c65242000e80db178dbc30c8bf617655cbe7 Mon Sep 17 00:00:00 2001 From: seonghobae Date: Tue, 25 Aug 2026 23:22:29 +0900 Subject: [PATCH 15/23] fix: avoid evaluating absent migration indexes --- migrations/0165_global_ask_job.sql | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/migrations/0165_global_ask_job.sql b/migrations/0165_global_ask_job.sql index 44484291d..95d28372a 100644 --- a/migrations/0165_global_ask_job.sql +++ b/migrations/0165_global_ask_job.sql @@ -11,7 +11,12 @@ -- `IF NOT EXISTS` is idempotent only when the existing object is compatible; -- fail before any insert path can observe a partial schema. do $$ +declare + account_index regclass; + queued_index regclass; begin + account_index := to_regclass('public.global_ask_job_account_idx'); + queued_index := to_regclass('public.global_ask_job_queued_idx'); if to_regclass('public.global_ask_job') is not null and exists ( select 1 @@ -36,13 +41,13 @@ begin ) then raise exception 'global_ask_job exists with an incompatible schema'; end if; - if to_regclass('public.global_ask_job_account_idx') is not null - and pg_get_indexdef('public.global_ask_job_account_idx'::regclass) + if account_index is not null + and pg_get_indexdef(account_index) not ilike '%(requesting_account_id, created_at DESC)%' then raise exception 'global_ask_job_account_idx exists with an incompatible definition'; end if; - if to_regclass('public.global_ask_job_queued_idx') is not null - and pg_get_indexdef('public.global_ask_job_queued_idx'::regclass) + if queued_index is not null + and pg_get_indexdef(queued_index) not ilike '%(created_at)%where%job_status_code%' then raise exception 'global_ask_job_queued_idx exists with an incompatible definition'; end if; From 74823e99bda251c001e6e4e2284a8faa31c1b933 Mon Sep 17 00:00:00 2001 From: seonghobae Date: Wed, 26 Aug 2026 00:34:47 +0900 Subject: [PATCH 16/23] fix(ci): document trusted eligibility SQL --- backend/app/lineage_ingestion.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/backend/app/lineage_ingestion.py b/backend/app/lineage_ingestion.py index 3775408a3..9564d0633 100644 --- a/backend/app/lineage_ingestion.py +++ b/backend/app/lineage_ingestion.py @@ -47,7 +47,6 @@ {"tepp_lineage_criterion_v1"} ) - def estimated_weight_channels(llm: AdjudicationClient | None) -> set[str]: """Return the channels that one live reconstruction can actually use.""" channels = {"temporal", "secondary_key", "text"} @@ -532,7 +531,9 @@ async def _fetch_lineage_landing_rows( limit: int, ): """Fetch only the authorized, bounded landing projection in PostgreSQL.""" - posts = await conn.fetch( + # The only formatted value is the module-owned literal alias ``source_post``; + # every caller value remains an asyncpg bind parameter. + posts = await conn.fetch( # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli "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 " From 4b4d6707d5c30de5a332ad00707fb8fb2fa670f8 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 26 Aug 2026 01:30:24 +0900 Subject: [PATCH 17/23] test(ask): activate embedding diagnostics path --- tests/test_server_diagnostics.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/tests/test_server_diagnostics.py b/tests/test_server_diagnostics.py index a81aea95e..499858121 100644 --- a/tests/test_server_diagnostics.py +++ b/tests/test_server_diagnostics.py @@ -39,6 +39,16 @@ def answer(self, question: str, sources: object) -> object: raise self._exc +class _EmbeddingClient: + """Deterministic available embedding channel for Ask diagnostics.""" + + available = True + resolved_model = "synthetic-embedding" + + def embed(self, _text: str) -> list[float]: + return [1.0, 0.0] + + def _call_ask(monkeypatch: pytest.MonkeyPatch, exc: BaseException) -> None: async def _sources(*args: object, **kwargs: object) -> list[object]: return [SimpleNamespace(post_id="synthetic-post-1")] @@ -53,6 +63,7 @@ async def _sources(*args: object, **kwargs: object) -> list[object]: process_unit_ids=set(), process_scope_limited=False, chat_client=_FailingClient(exc), + embedding_client=_EmbeddingClient(), ) ) assert raised.value.status_code == 503 @@ -150,6 +161,7 @@ async def _sources(*args: object, **kwargs: object) -> list[object]: process_unit_ids=set(), process_scope_limited=False, chat_client=_FailingClient(RuntimeError("unused")), + embedding_client=_EmbeddingClient(), ) ) assert raised.value.status_code == 503 From c95f931daf073ba52b4b30f25d896651d1e280fe Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 26 Aug 2026 07:18:51 +0900 Subject: [PATCH 18/23] fix(verification): fence deleted evidence rows --- backend/app/relation_verification_ingestion.py | 4 ++++ tests/test_relation_verification_internal.py | 1 + 2 files changed, 5 insertions(+) diff --git a/backend/app/relation_verification_ingestion.py b/backend/app/relation_verification_ingestion.py index ecbd7fe62..a973d278a 100644 --- a/backend/app/relation_verification_ingestion.py +++ b/backend/app/relation_verification_ingestion.py @@ -228,6 +228,10 @@ async def verify_post_relations_from_pool( where post_id = $1 and counterparty_entity_name = $2 and verification_status_code = 'verify_pending' and relationship_type_code = $6 + and ($5::uuid is null or exists ( + select 1 from source_post evidence + where evidence.post_id = $5::uuid + )) """, post_id, completed.counterparty_entity_name, diff --git a/tests/test_relation_verification_internal.py b/tests/test_relation_verification_internal.py index 2650f6395..27a2a3c4b 100644 --- a/tests/test_relation_verification_internal.py +++ b/tests/test_relation_verification_internal.py @@ -198,6 +198,7 @@ def test_pool_verification_omits_relation_completed_by_concurrent_run() -> None: class ConcurrentConnection(_Connection): async def execute(self, query: str, *args: object): assert "verification_status_code = 'verify_pending'" in query + assert "where evidence.post_id = $5::uuid" in query return "UPDATE 0" conn = ConcurrentConnection(None) From 967ba246ee5f3a5f267fe19798be4220ab8320a0 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 26 Aug 2026 07:26:17 +0900 Subject: [PATCH 19/23] fix(lineage): preserve landing tie order --- backend/app/lineage_ingestion.py | 16 ++++++++++++++-- tests/test_lineage_ingestion.py | 11 ++++++++++- 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/backend/app/lineage_ingestion.py b/backend/app/lineage_ingestion.py index 9564d0633..c2c53faee 100644 --- a/backend/app/lineage_ingestion.py +++ b/backend/app/lineage_ingestion.py @@ -17,6 +17,7 @@ from collections.abc import Mapping, Sequence from datetime import datetime from typing import Any +from uuid import UUID import asyncpg @@ -42,6 +43,14 @@ ISOLATION_NO_COMPARISON_GROUP = "no_comparison_group" ISOLATION_COMPARISON_CANDIDATES_AVAILABLE = "comparison_candidates_available" + +def _post_id_sort_key(post_id: object) -> tuple[int, int | str]: + """Match PostgreSQL UUID ordering while retaining synthetic fixture IDs.""" + try: + return (0, UUID(str(post_id)).int) + except (ValueError, AttributeError): + return (1, str(post_id)) + # ADR 0205 authorizes only a completed, persisted TEPP criterion anchor. _SUPPORTED_ANCHOR_METHOD_CODES: frozenset[str] = frozenset( {"tepp_lineage_criterion_v1"} @@ -540,7 +549,7 @@ async def _fetch_lineage_landing_rows( 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", + "order by created_at desc, post_id::text desc limit $3", list(corporate_entity_ids), list(process_unit_ids), limit + 1, @@ -713,7 +722,10 @@ async def visible_lineage_graph( 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"])), + key=lambda row: ( + row["created_at"], + *_post_id_sort_key(row["post_id"]), + ), reverse=True, )[:limit] truncated = len(visible_all) > len(visible) diff --git a/tests/test_lineage_ingestion.py b/tests/test_lineage_ingestion.py index a7b19a342..3b56dc219 100644 --- a/tests/test_lineage_ingestion.py +++ b/tests/test_lineage_ingestion.py @@ -12,6 +12,7 @@ import backend.app.lineage_ingestion as ingestion from backend.app.lineage_ingestion import ( _budgeted_llm, + _post_id_sort_key, interval_relations_for_post, lineage_graphs_for_posts, persist_lineage_edges, @@ -51,6 +52,14 @@ async def fetch(self, _query: str): ) is None +def test_post_id_sort_key_matches_uuid_order_and_keeps_fixture_ids() -> None: + """Landing tie-breaks use database UUID order without rejecting fixtures.""" + assert _post_id_sort_key("00000000-0000-0000-0000-000000000002") > _post_id_sort_key( + "00000000-0000-0000-0000-000000000001" + ) + assert _post_id_sort_key("post-b") > _post_id_sort_key("post-a") + + def test_unapproved_weight_provenance_is_never_activated() -> None: class StoredWeightConnection: async def fetchval(self, _query: str): @@ -998,7 +1007,7 @@ async def fetch(self, query: str, *args): 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 "order by created_at desc, post_id::text desc limit $3" in post_query assert post_args == (["corp-a"], ["pu-a"], 501) assert graph["nodes"] == [] assert graph["truncated"] is False From 48496ff688d06129b5f2c09bb12aeef6fa618c93 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 26 Aug 2026 07:26:56 +0900 Subject: [PATCH 20/23] fix(lineage): retain string tie-break contract --- backend/app/lineage_ingestion.py | 14 +------------- tests/test_lineage_ingestion.py | 9 --------- 2 files changed, 1 insertion(+), 22 deletions(-) diff --git a/backend/app/lineage_ingestion.py b/backend/app/lineage_ingestion.py index c2c53faee..d2bd90065 100644 --- a/backend/app/lineage_ingestion.py +++ b/backend/app/lineage_ingestion.py @@ -17,7 +17,6 @@ from collections.abc import Mapping, Sequence from datetime import datetime from typing import Any -from uuid import UUID import asyncpg @@ -43,14 +42,6 @@ ISOLATION_NO_COMPARISON_GROUP = "no_comparison_group" ISOLATION_COMPARISON_CANDIDATES_AVAILABLE = "comparison_candidates_available" - -def _post_id_sort_key(post_id: object) -> tuple[int, int | str]: - """Match PostgreSQL UUID ordering while retaining synthetic fixture IDs.""" - try: - return (0, UUID(str(post_id)).int) - except (ValueError, AttributeError): - return (1, str(post_id)) - # ADR 0205 authorizes only a completed, persisted TEPP criterion anchor. _SUPPORTED_ANCHOR_METHOD_CODES: frozenset[str] = frozenset( {"tepp_lineage_criterion_v1"} @@ -722,10 +713,7 @@ async def visible_lineage_graph( if focus_post_id is None and corporate_entity_ids is None: visible = sorted( visible_all, - key=lambda row: ( - row["created_at"], - *_post_id_sort_key(row["post_id"]), - ), + key=lambda row: (row["created_at"], str(row["post_id"])), reverse=True, )[:limit] truncated = len(visible_all) > len(visible) diff --git a/tests/test_lineage_ingestion.py b/tests/test_lineage_ingestion.py index 3b56dc219..4cb385d5e 100644 --- a/tests/test_lineage_ingestion.py +++ b/tests/test_lineage_ingestion.py @@ -12,7 +12,6 @@ import backend.app.lineage_ingestion as ingestion from backend.app.lineage_ingestion import ( _budgeted_llm, - _post_id_sort_key, interval_relations_for_post, lineage_graphs_for_posts, persist_lineage_edges, @@ -52,14 +51,6 @@ async def fetch(self, _query: str): ) is None -def test_post_id_sort_key_matches_uuid_order_and_keeps_fixture_ids() -> None: - """Landing tie-breaks use database UUID order without rejecting fixtures.""" - assert _post_id_sort_key("00000000-0000-0000-0000-000000000002") > _post_id_sort_key( - "00000000-0000-0000-0000-000000000001" - ) - assert _post_id_sort_key("post-b") > _post_id_sort_key("post-a") - - def test_unapproved_weight_provenance_is_never_activated() -> None: class StoredWeightConnection: async def fetchval(self, _query: str): From b2bc72c1c8d833c8423f82f89783b0b53d3013d9 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 26 Aug 2026 17:47:36 +0900 Subject: [PATCH 21/23] fix(verification): restore incremental relation persistence --- .../app/relation_verification_ingestion.py | 76 +---------------- tests/test_relation_verification_internal.py | 82 +++++++++++-------- 2 files changed, 50 insertions(+), 108 deletions(-) diff --git a/backend/app/relation_verification_ingestion.py b/backend/app/relation_verification_ingestion.py index 348f277d0..a973d278a 100644 --- a/backend/app/relation_verification_ingestion.py +++ b/backend/app/relation_verification_ingestion.py @@ -30,6 +30,7 @@ class VerifiedRelation: @dataclass(frozen=True) class _PendingRelation: counterparty_entity_name: str + relationship_type_code: str relationship_label: str internal_evidence_post_id: str | None @@ -138,7 +139,7 @@ async def verify_post_relations( row["counterparty_entity_name"], row["relationship_label"], ) - await conn.execute( + update_status = await conn.execute( """ update post_counterparty_entity set verification_status_code = $3, @@ -242,76 +243,3 @@ async def verify_post_relations_from_pool( if update_status == "UPDATE 1": verified.append(completed) 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, - ) - ) - - persisted = [] - async with pool.acquire() as conn, conn.transaction(): - for relation in verified: - update_status = 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, - ) - if update_status == "UPDATE 1": - persisted.append(relation) - return persisted diff --git a/tests/test_relation_verification_internal.py b/tests/test_relation_verification_internal.py index 4aa3d45e3..86dc2ed0b 100644 --- a/tests/test_relation_verification_internal.py +++ b/tests/test_relation_verification_internal.py @@ -2,6 +2,8 @@ import asyncio +import pytest + from backend.app.relation_verification_ingestion import ( verify_post_relations, verify_post_relations_from_pool, @@ -18,6 +20,7 @@ def __init__(self, evidence_post_id: str | None, update_status: str = "UPDATE 1" self.update_status = update_status self.fetchrow_args: tuple[object, ...] | None = None self.execute_args: tuple[object, ...] | None = None + self.execute_calls: list[tuple[object, ...]] = [] async def fetch(self, query: str, post_id: str): assert "verification_status_code = 'verify_pending'" in query @@ -38,6 +41,7 @@ async def fetchrow(self, query: str, *args: object): async def execute(self, query: str, *args: object): assert "verification_evidence_post_id = $5" in query self.execute_args = args + self.execute_calls.append(args) return self.update_status def transaction(self): @@ -52,39 +56,6 @@ 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) - - 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 @@ -144,7 +115,8 @@ 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 + assert conn.execute_args[-2] is None + assert conn.execute_args[-1] == "partner" def test_pool_connection_is_released_during_external_verification() -> None: @@ -174,3 +146,45 @@ def test_pool_verification_counts_only_rows_settled_by_this_worker() -> None: ) assert verified == [] + + +def test_pool_verification_persists_completed_rows_before_provider_failure() -> None: + """A later provider failure does not roll back an earlier completed row.""" + + class _TwoRelationConnection(_Connection): + async def fetch(self, query: str, post_id: str): + assert "verification_status_code = 'verify_pending'" in query + return [ + { + "counterparty_entity_name": "Example Partner", + "relationship_type_code": "partner", + "relationship_label": "Partner", + }, + { + "counterparty_entity_name": "Example Supplier", + "relationship_type_code": "supplier", + "relationship_label": "Supplier", + }, + ] + + class _FailingSecondVerifier: + def verify( + self, organization_name: str, relationship_label: str + ) -> RelationVerificationResult: + if organization_name == "Example Supplier": + raise RuntimeError("synthetic provider failure") + return RelationVerificationResult( + STATUS_CORROBORATED, "https://example.test/evidence" + ) + + conn = _TwoRelationConnection(None) + + with pytest.raises(RuntimeError, match="synthetic provider failure"): + asyncio.run( + verify_post_relations_from_pool( + _Pool(conn), _FailingSecondVerifier(), "origin-post" + ) + ) + + assert len(conn.execute_calls) == 1 + assert conn.execute_calls[0][-1] == "partner" From c00b571c632b9b7881e1a85a1511da54aab24298 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 26 Aug 2026 17:55:05 +0900 Subject: [PATCH 22/23] test(db): execute Global Ask migration replay --- tests/test_schema.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/tests/test_schema.py b/tests/test_schema.py index 224d01c50..d6d4c8d0b 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -181,6 +181,12 @@ def schema_db(): 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_JOB_MIGRATION.read_text()) + cur.execute(_GLOBAL_ASK_SCOPE_MIGRATION.read_text()) + # Exercise the production replay contract against the same + # PostgreSQL objects instead of merely inspecting SQL text. + cur.execute(_GLOBAL_ASK_JOB_MIGRATION.read_text()) + cur.execute(_GLOBAL_ASK_SCOPE_MIGRATION.read_text()) # psql sends each statement independently, which is required # by CREATE INDEX CONCURRENTLY. psycopg2 treats a multi- # statement execute as one transaction even with autocommit. @@ -243,6 +249,9 @@ def test_migration_applies_cleanly(schema_db) -> None: "post_summary_action", "post_chat_result", "post_chat_citation", + "global_ask_job", + "global_ask_job_corporate_entity_scope", + "global_ask_job_process_unit_scope", } assert expected <= tables From b721b0f2ca7b9abad8ad0b6ce388bfdf90ada8d7 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 26 Aug 2026 18:03:42 +0900 Subject: [PATCH 23/23] fix(verification): fence deleted evidence --- backend/app/relation_verification_ingestion.py | 4 ++++ tests/test_relation_verification_internal.py | 1 + 2 files changed, 5 insertions(+) diff --git a/backend/app/relation_verification_ingestion.py b/backend/app/relation_verification_ingestion.py index a973d278a..f334a0c08 100644 --- a/backend/app/relation_verification_ingestion.py +++ b/backend/app/relation_verification_ingestion.py @@ -149,6 +149,10 @@ async def verify_post_relations( where post_id = $1 and counterparty_entity_name = $2 and verification_status_code = 'verify_pending' and relationship_type_code = $6 + and ($5::uuid is null or exists ( + select 1 from source_post evidence + where evidence.post_id = $5::uuid + )) """, post_id, row["counterparty_entity_name"], diff --git a/tests/test_relation_verification_internal.py b/tests/test_relation_verification_internal.py index 86dc2ed0b..9849d9326 100644 --- a/tests/test_relation_verification_internal.py +++ b/tests/test_relation_verification_internal.py @@ -40,6 +40,7 @@ async def fetchrow(self, query: str, *args: object): async def execute(self, query: str, *args: object): assert "verification_evidence_post_id = $5" in query + assert "$5::uuid is null or exists" in query self.execute_args = args self.execute_calls.append(args) return self.update_status