diff --git a/backend/app/config.py b/backend/app/config.py index 0fea9a591..b1199e7dc 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -8,11 +8,6 @@ import os from dataclasses import dataclass, field -# Hard ceiling on one Global Ask job's answer computation, shared with the -# worker in global_ask_queue.py so config validation and execution can never -# disagree about the bound. -GLOBAL_ASK_JOB_DEADLINE_SECONDS = 600 - @dataclass(frozen=True) class Settings: @@ -51,11 +46,10 @@ class Settings: frontend_origins: list[str] orchestrator_base_url: str orchestrator_api_key: str - # Socket timeout for one Ask answer round-trip. Must stay below the Ask - # worker's job deadline so the client, not the job reaper, ends a slow - # call — hanging up earlier discards an answer the orchestrator has - # already paid to generate (observed live as a BrokenPipe on its side). - orchestrator_answer_timeout_seconds: float + # Optional socket timeout for one Ask answer round-trip. Omitted/blank + # means no LineageWeave elapsed socket limit. Explicit values are + # deployment transport policy and remain independent of worker liveness. + orchestrator_answer_timeout_seconds: float | None valkey_url: str searxng_base_url: str tepp_transport_url: str @@ -82,24 +76,24 @@ def keycloak_jwks_uri(self) -> str: return f"{self.keycloak_base_url}/realms/{self.keycloak_realm}/protocol/openid-connect/certs" -def _validated_answer_timeout(raw: str) -> float: - """Parse the Ask answer timeout and hold it under the job deadline. +def _validated_answer_timeout(raw: str | None) -> float | None: + """Parse an optional finite-positive Ask transport timeout. - The client must hang up before the worker's deadline reaper so a slow - answer settles as a clean client timeout, never a reaped job — values - at or above the deadline (or non-finite/non-positive ones) silently - break that ordering, so they are configuration errors. + Blank or omitted leaves no LineageWeave elapsed socket limit. Worker + liveness is owned by claim heartbeats and generation fencing, not by + this optional transport policy. """ + if raw is None or not str(raw).strip(): + return None try: value = float(raw) except ValueError as exc: raise ValueError( "ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS must be a number" ) from exc - if not math.isfinite(value) or not 0 < value < GLOBAL_ASK_JOB_DEADLINE_SECONDS: + if not math.isfinite(value) or value <= 0: raise ValueError( - "ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS must be a finite number greater" - f" than 0 and less than {GLOBAL_ASK_JOB_DEADLINE_SECONDS}" + "ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS must be a finite number greater than 0" ) return value @@ -201,7 +195,7 @@ def load_settings() -> Settings: orchestrator_base_url=os.environ.get("ORCHESTRATOR_BASE_URL", ""), orchestrator_api_key=os.environ.get("ORCHESTRATOR_API_KEY", ""), orchestrator_answer_timeout_seconds=_validated_answer_timeout( - os.environ.get("ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS", "570") + os.environ.get("ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS") ), valkey_url=os.environ.get("VALKEY_URL", "redis://localhost:16379/0"), searxng_base_url=os.environ.get("SEARXNG_BASE_URL", ""), diff --git a/backend/app/global_ask_queue.py b/backend/app/global_ask_queue.py index f8720f715..e679be88a 100644 --- a/backend/app/global_ask_queue.py +++ b/backend/app/global_ask_queue.py @@ -54,7 +54,6 @@ from lineageweave.semantic_query import NullSemanticQueryClient, SemanticQueryClient from lineageweave.temporal_expressions import resolve_korean_relative_time -from .config import GLOBAL_ASK_JOB_DEADLINE_SECONDS from .lineage_ingestion import lineage_graphs_for_posts from .operability import log_internal_fault, log_provider_unavailable from .post_chat_ingestion import ( @@ -76,17 +75,13 @@ # trimmed stream) and are republished by the worker's recovery sweep. _REPUBLISH_AFTER_SECONDS = 60 _RECOVERY_INTERVAL_SECONDS = 30.0 -# Hard ceiling on one job's answer computation. Without it a hung -# orchestrator round-trip kept a job `running` indefinitely (observed: -# 17+ minutes) and, before concurrent processing, stalled every job -# behind it. Shared through config so the client-timeout validation and -# this reaper can never disagree. -JOB_DEADLINE_SECONDS = GLOBAL_ASK_JOB_DEADLINE_SECONDS -# A `running` row older than this is an orphan: a live worker's deadline -# settles every job within JOB_DEADLINE_SECONDS, so one sweep interval of -# slack past that is enough — recovering sooner shortens how long a -# crashed worker's job stays invisible to a polling reader. -_ORPHAN_RUNNING_AFTER_SECONDS = JOB_DEADLINE_SECONDS + 60 +# Live workers renew the claim generation on this interval so age-based +# orphan recovery cannot reclaim a job that is still owned. +_CLAIM_HEARTBEAT_SECONDS = _RECOVERY_INTERVAL_SECONDS +# A `running` row whose claim generation has not been renewed for this +# many seconds is an orphan. Live workers heartbeat more often, so age +# alone does not reclaim a current owner. +_ORPHAN_RUNNING_AFTER_SECONDS = 3 * _CLAIM_HEARTBEAT_SECONDS # Wake-up stream cap, mirroring the post-content stream: the durable rows # are the source of truth, so trimming old wake-ups loses nothing. _STREAM_MAX_LENGTH = 1000 @@ -102,6 +97,83 @@ class _SafeJobError(Exception): """Failure whose bounded message is safe to persist for the requester.""" +def _claim_generation_retained(command_status: object) -> bool: + """Return True when PostgreSQL reports the compare-and-set updated one row.""" + return str(command_status) == "UPDATE 1" + + +class _LostAskClaim(Exception): + """Raised when orphan recovery reclaims the running claim generation.""" + + +async def _renew_ask_claim( + pool: asyncpg.Pool, job_id: str, claimed_at: object +) -> object | None: + """Advance ``updated_at`` only for the current running claim generation.""" + async with pool.acquire() as conn: + row = await conn.fetchrow( + """ + update global_ask_job set updated_at = now() + where global_ask_job_id = $1 + and job_status_code = $2 + and updated_at = $3 + returning updated_at + """, + job_id, + RUNNING, + claimed_at, + ) + if row is None: + return None + return row["updated_at"] + + +async def _run_with_ask_claim_heartbeat( + pool: asyncpg.Pool, + job_id: str, + lease: list[object], + operation: Any, +) -> Any: + """Renew the claim generation while ``operation`` runs; abort on reclaim. + + If the heartbeat task ends while compute is still running, treat the + owner as lost instead of continuing without renewals. + """ + lost = asyncio.Event() + stop = asyncio.Event() + + async def _beat() -> None: + while not stop.is_set() and not lost.is_set(): + try: + await asyncio.wait_for(stop.wait(), timeout=_CLAIM_HEARTBEAT_SECONDS) + return + except TimeoutError: + renewed = await _renew_ask_claim(pool, job_id, lease[0]) + if renewed is None: + lost.set() + return + lease[0] = renewed + + beater = asyncio.create_task(_beat()) + worker = asyncio.create_task(operation) + try: + await asyncio.wait({worker, beater}, return_when=asyncio.FIRST_COMPLETED) + if not worker.done(): + raise _LostAskClaim() + # A renewal may already be committed while its response is in flight. + # Drain it before settlement so the lease contains the committed generation. + stop.set() + (renewal_result,) = await asyncio.gather(beater, return_exceptions=True) + if lost.is_set() or isinstance(renewal_result, BaseException): + raise _LostAskClaim() + return await worker + finally: + stop.set() + worker.cancel() + beater.cancel() + await asyncio.gather(worker, beater, return_exceptions=True) + + async def enqueue_global_ask_job( conn: asyncpg.Connection, client: redis.Redis, @@ -520,8 +592,10 @@ async def process_global_ask_job( Claiming flips ``queued`` → ``running`` atomically so a duplicate stream wake-up (recovery republish racing the original entry) is a - no-op. Every failure path settles the row as ``failed`` with a - bounded detail string rather than leaving it stuck ``running``. + no-op. Settlement is compare-and-set on that claim's ``updated_at`` + so an orphan reclaim cannot be overwritten by the previous owner. + Every failure path settles the row as ``failed`` with a bounded + detail string rather than leaving it stuck ``running``. """ async with pool.acquire() as conn: row = await conn.fetchrow( @@ -529,7 +603,7 @@ async def process_global_ask_job( update global_ask_job set job_status_code = $2, updated_at = now() where global_ask_job_id = $1 and job_status_code = $3 returning requesting_account_id, question_text, verify_external_requested, - knowledge_cutoff + knowledge_cutoff, updated_at """, job_id, RUNNING, @@ -537,7 +611,7 @@ async def process_global_ask_job( ) if row is None: return - answer_timeout: asyncio.Timeout | None = None + lease = [row["updated_at"]] try: async with pool.acquire() as conn: ( @@ -555,8 +629,11 @@ async def process_global_ask_job( raise _SafeJobError( "Ask Agent is unavailable: set ORCHESTRATOR_BASE_URL / ORCHESTRATOR_API_KEY" ) - async with asyncio.timeout(JOB_DEADLINE_SECONDS) as answer_timeout: - payload = await compute_global_ask_answer( + payload = await _run_with_ask_claim_heartbeat( + pool, + job_id, + lease, + compute_global_ask_answer( pool, question_text=str(row["question_text"]), corporate_entity_ids=entity_ids, @@ -568,7 +645,10 @@ async def process_global_ask_job( verify_external=bool(row["verify_external_requested"]), claim_verification_client=claim_verification_factory(), knowledge_cutoff=row["knowledge_cutoff"], - ) + ), + ) + except _LostAskClaim: + return except asyncio.CancelledError: # Shutdown: leave the row `running`; the recovery sweep re-queues # it after the orphan window on the next process start. @@ -586,12 +666,6 @@ async def process_global_ask_job( # Raised locally with a pre-authored, safe message (permission # state / missing config) — never a provider-boundary leak. detail = str(exc) - elif ( - isinstance(exc, asyncio.TimeoutError) - and answer_timeout is not None - and answer_timeout.expired() - ): - detail = f"job exceeded the {JOB_DEADLINE_SECONDS}s deadline" else: # Provider responses/exceptions can carry credentials, gateway # diagnostics, or model output (ADR 0123): never persist the @@ -602,28 +676,40 @@ async def process_global_ask_job( "no complete evidence object" ) async with pool.acquire() as conn: - await conn.execute( + command_status = await conn.execute( """ update global_ask_job set job_status_code = $2, failure_detail = $3, updated_at = now() where global_ask_job_id = $1 + and job_status_code = $4 + and updated_at = $5 """, job_id, FAILED, detail[:1000], + RUNNING, + lease[0], ) + if not _claim_generation_retained(command_status): + return return async with pool.acquire() as conn: - await conn.execute( + command_status = await conn.execute( """ update global_ask_job set job_status_code = $2, answer_payload = $3::jsonb, updated_at = now() where global_ask_job_id = $1 + and job_status_code = $4 + and updated_at = $5 """, job_id, SUCCEEDED, _to_json(payload), + RUNNING, + lease[0], ) + if not _claim_generation_retained(command_status): + return def _to_json(payload: dict[str, Any]) -> str: @@ -640,10 +726,9 @@ async def republish_queued_global_ask_jobs( A ``queued`` row older than the republish window lost its stream entry (crash or trim between insert and XADD). A ``running`` row - older than the orphan window belongs to a worker that died mid-job — - the per-job deadline guarantees a live worker settles sooner — so it - is flipped back to ``queued`` and re-woken for at-least-once - delivery. + whose claim generation stays stale beyond the orphan window belongs + to a worker that stopped heartbeating, so it is flipped back to + ``queued`` and re-woken for at-least-once delivery. """ async with pool.acquire() as conn: # Fully parameterized ($1..$3 with module constants); the rule diff --git a/backend/app/main.py b/backend/app/main.py index 56b519ef4..ee2075739 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -323,8 +323,7 @@ async def lifespan(app: FastAPI): app.state.post_content_worker = content_worker # Late-bound lambda so tests that monkeypatch _post_chat_client reach # the worker too (the name resolves in module globals at call time). - # This worker still has an explicit answer socket limit; per-post - # chat retains the default null transport timeout. + # Ask and per-post chat both default to a null transport timeout. global_ask_worker = asyncio.create_task( run_global_ask_worker( valkey, diff --git a/backend/tests/test_config.py b/backend/tests/test_config.py index f37d89f32..0469266cc 100644 --- a/backend/tests/test_config.py +++ b/backend/tests/test_config.py @@ -2,9 +2,36 @@ from __future__ import annotations +import pytest + from backend.app.config import load_settings +def test_ask_answer_timeout_defaults_to_no_elapsed_socket_limit(monkeypatch) -> None: + """Omitted Ask socket timeout is not a hidden 570-second hang-up.""" + monkeypatch.delenv("ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS", raising=False) + assert load_settings().orchestrator_answer_timeout_seconds is None + + +def test_ask_answer_timeout_accepts_explicit_finite_values_without_worker_deadline( + monkeypatch, +) -> None: + """An operator-set Ask socket timeout is independent of worker liveness.""" + monkeypatch.setenv("ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS", "570") + assert load_settings().orchestrator_answer_timeout_seconds == 570 + + monkeypatch.setenv("ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS", "900") + assert load_settings().orchestrator_answer_timeout_seconds == 900 + + +def test_ask_answer_timeout_rejects_non_positive_explicit_values(monkeypatch) -> None: + """Zero, negative, and non-finite Ask socket timeouts remain configuration errors.""" + for raw in ("0", "-1", "nan", "inf"): + monkeypatch.setenv("ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS", raw) + with pytest.raises(ValueError, match="finite number greater than 0"): + load_settings() + + def test_frontend_origins_are_parsed_from_comma_separated_env(monkeypatch) -> None: """CORS allow-list is an explicit env CSV, never a wildcard default.""" monkeypatch.setenv( diff --git a/docs/adr/0371-ask-claim-generation-liveness.md b/docs/adr/0371-ask-claim-generation-liveness.md new file mode 100644 index 000000000..33544bd66 --- /dev/null +++ b/docs/adr/0371-ask-claim-generation-liveness.md @@ -0,0 +1,67 @@ +# ADR 0371 — Global Ask claim-generation liveness + +**Decision status:** Proposed +**Date:** 2026-09-08 + +Amends the Global Ask worker settlement path. Independent of leftover-map +ADRs 0272–0370 on other stacks (ADR 0370 is leftover-map comparison +axis-singular on the #830 validation lane) and of the versioned +translation ledger ([ADR 0362](0362-versioned-ui-translation-ledger.md)). + +## Context + +Issue #975: a live Ask job could be terminated or reclaimed from elapsed +wall time (600 s compute deadline, 570 s invented socket hang-up) while +a provider `TimeoutError` and a worker deadline meant different things. +Orphan recovery flipped `running` → `queued` by `updated_at` age, and +the original worker could still settle by job id alone. + +## Decision + +- Claim `queued` → `running` returns a generation (`updated_at`). +- Settlement is compare-and-set on that generation; PostgreSQL + `UPDATE 0` is an unapplied settle, not a buyer-visible failure. +- While computing, the owner renews `updated_at` on the recovery + interval. A failed renew aborts without settling as failed. +- Cancelling the owner task cancels the inner compute task. +- LineageWeave does not invent an Ask socket hang-up when + `ORCHESTRATOR_ANSWER_TIMEOUT_SECONDS` is omitted or blank. An + explicit value must be finite and strictly positive; it is not + bounded by the removed 600 s worker deadline. +- Live compute is not cancelled when 600 s elapse. Age-based orphan + recovery uses three missed heartbeats, not the old 660 s reaper. +- If the heartbeat task ends while compute is still running, abort as a + lost claim rather than continuing without renewals. +- When compute finishes, stop scheduling renewals and await any renewal + already in flight before admitting either its result or its failure to + settlement. A committed renewal must update the settlement generation; + cancellation during response delivery must not discard that generation. + A failed renewal wins over a simultaneously completed answer. External + owner cancellation still cancels and joins both tasks. +- Provider `TimeoutError` stays an unavailable Ask failure. + +## Consequences + +Positive: a renewing owner can outlive the former hard deadline; a +reclaimed job cannot be overwritten by the previous owner. + +Negative: crashed workers wait three heartbeat intervals to reclaim. + +`tests/test_schema.py` proves the settlement compare-and-set against a +throwaway database that replayed the real `0001` and `0165` migrations. +In-memory queue tests remain for elapsed-time and cancellation contracts. +The completion/renewal race also uses the production queue and renewal +functions against PostgreSQL: delay delivery of an already committed renewal, +finish the answer, then require the stored job to reach `succeeded` with that +answer. This is persistence evidence, not authenticated HTTP/UI acceptance. + +The inherited three-heartbeat recovery ratio is not established as a +deployment capacity or failure-detector contract by these regressions. Its +acceptance remains unresolved; this correction does not calibrate that ratio +or establish that elapsed silence alone proves a dead owner. + +## Alternatives considered + +Keep the 600 s `asyncio.timeout` around compute: rejected because it +cancels a live heartbeat owner. The elapsed-deadline RED proves that +path. diff --git a/docs/product-requirements.md b/docs/product-requirements.md index a8b741521..2f0991b61 100644 --- a/docs/product-requirements.md +++ b/docs/product-requirements.md @@ -526,7 +526,7 @@ current boundary until that repository adopts one. | `ContextualWisdomLab/keyverse` | `docs/PRD.md` | Production OIDC/JWKS/identity control plane; local demo Keycloak is not Keyverse | | `ContextualWisdomLab/RankWeave` | No standalone PRD; `README.md`, `ARCHITECTURE.md` | Store-agnostic ranking/fusion dependency; caller owns channels and authorization | | `ContextualWisdomLab/ThreadWeave` | `docs/PRD.md` | Deterministic reference-thread assembly dependency; LineageWeave owns records and persistence | -| `ContextualWisdomLab/DiskSage` | No standalone PRD; `docs/superpowers/specs/2026-07-10-disksage-design.md` | Prospective storage-policy boundary; no current runtime integration | +| `ContextualWisdomLab/disksage` | No standalone PRD; `docs/superpowers/specs/2026-07-10-disksage-design.md` | Prospective storage-policy boundary; no current runtime integration | | `ContextualWisdomLab/wardnet` | No standalone PRD; `README.md`, `docs/architecture.md` | Prospective gateway/network-policy boundary; no current runtime integration | | `ContextualWisdomLab/naruon` | Scoped `docs/topic-intelligence/PRD.md` only | Owns observed calendar/email projections; LineageWeave owns commitments and combined display | | `ContextualWisdomLab/LineageWeave` | This PRD, with ADRs normative | Evidence BI/orchestration, lineage, semantic projection, API, and UI owner | diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index 42dc5d155..d528628d6 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -1,5 +1,164 @@ # Product & Technical Gap Baseline +## Current evidence — 2026-09-08 KST + +This section supersedes the historical snapshots below. It is a bounded audit, +not a claim that every open PR or product acceptance condition is complete. +Protected main: `83eba56149eb802cd63642c507c324c9976ec78e`. +Reviewed Ask candidate: #979 at `fef48b14f302dd40e1fa83096810cc880a5c4c66`, +stacked on #974 at `def15fc691d4442c0d82103c1642147b1528d7be`. +The completion/renewal correction is source commit +`06be667c01f930fefb41a5107bf21fceb81ba7e7`; this document is its evidence-only +follow-up. Hosted checks and reviews must be fetched for the final PR head, +never inherited from either source commit or parent. + +### Authority and implementation are separate + +- Read the current LineageWeave PRD (`docs/product-requirements.md`) before + choosing the correction. PRD-FR-5 and FR-7 require observable durable outcomes + and preserve the upstream ownership boundary. ADR 0371 is **Proposed**; + accepted ADRs 0204/0213 govern short database transactions and provider work. +- Read contextual-orchestrator `docs/product_planning.md` and + `docs/architecture.md`; its remote main was + `414f22973658c4ddc3d4320fcf7acd9b4e8ba991` at this audit. The Fugu/TRINITY/ + Conductor register supports that owner's orchestration boundary. It does + not establish a LineageWeave timeout, heartbeat ratio, workload SLO, or + model ordering. This correction adds no model-selection or numeric policy. +- Python's official [task/cancellation contract](https://docs.python.org/3/library/asyncio-task.html) + supplies the task lifecycle behavior used by the correction. A task's + completed answer and another task's completion/failure are separate facts. +- GitHub repository API returned canonical names + `ContextualWisdomLab/LineageWeave`, `ContextualWisdomLab/RankWeave`, + `ContextualWisdomLab/ThreadWeave`, `ContextualWisdomLab/disksage`, + `ContextualWisdomLab/TEPP`, and `ContextualWisdomLab/contextual-orchestrator`. + The PRD register now uses the remotely confirmed lowercase `disksage`. +- DeepWiki returned repository-not-indexed; Context7 returned quota-exceeded. + Neither response is documentation or architecture evidence. Sequential + Thinking and Memory graph tools are unavailable in this tool catalog. + +### Selected actionable buyer gap — completed Ask can remain running (#975) + +The reviewed candidate could cancel a claim renewal after PostgreSQL committed +it but before the caller received its generation. A finished answer then used +the old generation for settlement and could stay `running`. A simultaneous +renewal failure could also be ignored when the answer had finished. These are +reproduced lifecycle defects, not inferred performance bottlenecks. They are +prioritized because they prevent an already completed answer from reaching the +requester. No numerical product-gap ranking or population inference is claimed. + +The minimal correction stops scheduling renewals, drains a renewal already in +flight, rejects an unconfirmed/lost claim, and then admits answer or failure to +existing compare-and-set settlement. External cancellation still cancels and +joins both tasks. No API/schema/release-number change, new dependency, provider +call, local mathematical implementation, or UI-copy change is introduced. + +Two focused cases failed before the correction: committed renewal delivery was +cancelled, and a simultaneous renewal error did not reject the answer. After +correction, **42 local tests passed** across claim cancellation, elapsed-time +behavior, queue outcomes, transport configuration, and public docstrings. +The added PostgreSQL regression invokes production claim/renewal/settlement +functions after delaying a real committed renewal response. Initial live execution +found that the shared test fixture omitted migrations 0212 and 0218; it failed +with a missing public-verification column before entering the race. The fixture +now applies and replays both real migrations. The normal two-second connection +admission skipped two local tests; skipped tests are not persistence proof. A +separate authenticated PostgreSQL execution with the exact production migrations +passed both regressions: stale-owner settlement rejection and persisted answer +after committed-renewal delivery. Idempotent migration replay also passed. The +throwaway database was dropped afterward. Only visibility/model computation +was synthetic; claim, renewal and settlement used the production functions and +real PostgreSQL. This is database evidence, not authenticated HTTP/UI acceptance. +The inherited three-heartbeat orphan threshold is still ungrounded as a +failure-detector/capacity policy and remains an unresolved ADR acceptance gap. + +### Non-identifying runtime observation, not acceptance + +The formal Compose project remains `lineageweave`. Live Docker mappings bind +backend HTTP to host 18420, contextual-orchestrator to 18000, frontend to 15173, +and PostgreSQL to 15432. No Compose rendering or credentials were printed. + +At 11:41 KST the content-safe k6 harness from exact #964 +`1cced397600b15258b36e221a33beb62c4cca4cd` attempted two VUs for ten seconds, +with an explicit 20-second request observation limit. Synthetic OIDC succeeded +in 16.80 seconds; Ask submission timed out at 19.97 seconds. Setup stopped with +zero active VUs and zero workload iterations. The two setup requests had one +failure (50%) and observed throughput 0.0537 requests/second. These are setup +observations, **not authenticated concurrent workload latency/error/throughput**. +Do not infer whether the timed-out submission eventually persisted or succeeded. + +A nearby one-shot Docker CPU observation was PostgreSQL 175.92%, backend 4.06%, +Ask service 50.83%, Valkey 8.26%, and gateway 67.73%; memory percentages were +2.47%, 1.28%, 1.60%, 8.14%, and 1.69%, respectively. Docker CPU is not normalized +to a single host-core percentage. One snapshot does not prove saturation or a +causal bottleneck; PostgreSQL waits, service occupancy, gateway capacity and +Valkey saturation remain unavailable. No tuning is justified by these numbers. +Authenticated all-page/API acceptance and rendered desktop/mobile acceptance +remain unavailable. There is no UI change in this correction and no new +screenshot or deployed behavior claim. + +### Open queue and protected delivery + +The refreshed GitHub inventory contains **139 open PRs: 17 ready, 122 draft**, +plus **22 open issues**; zero PRs have `reviewDecision=APPROVED`. These describe +the queue, not product maturity. #979's reviewed predecessor has terminal +successful repository backend/frontend Tests, but its only commit check runs +are those two jobs; central receipts and independent approval are not inferred. +#974 has failing/cancelled central review/security receipts as well as the +independent approval requirement. Normal squash auto-merge is enabled on #974; +#780's existing squash auto-merge remains enabled. No merge SHA is claimed. + +Effective main rules require one independent approval, stale-review dismissal, +resolved review threads, and centrally supplied review/security workflows; +force pushes and deletion are prohibited. Classic branch protection returns +404, which does not negate the effective rulesets. No current main or open-PR +run was cancelled. The observed active PR runs belonged to current #983/#966; +no stale closed-PR cancellation was warranted. + +### Collision and Voice acceptance audit + +- The correction retains #979's Proposed ADR 0371. Exact #980 separately adds + ADR 0370 for leftover-map comparison axis-singular evidence; neither number + is allocated anew. #974 changes ADR 0083; #929 owns translation ADR 0362 and + migrations 0246/0247. This correction does not add a migration or release. +- #974 must merge normally before #979 is retargeted to main. Do not enable a + child merge onto its unprotected feature base or inherit parent checks. +- The wider draft report queue still includes duplicate release labels and + serialized overlapping files. A whole-queue ADR/API/schema/release collision + clearance is **not established** by this bounded audit. +- ADR 0246 retains twelve atomic Voices. In the current main, ADR 0251 is the + FJA I/O-psychology document; current #780 names the combination decision + `0256-evidence-bearing-voice-combinations.md` and primary history ADR 0252. + Track title and exact branch as well as number instead of treating the + user's historical ADR 0251 combination reference as the unrelated FJA policy. +- #780 at `1d8fa267b059289e77301a09985dfac70a439814` remains the Voice export + parent. Carrying Post and derivation evidence stay distinct; combinations + remain extensible atomic assignments, with authorized evidence, PROV-O, + truth status and cutoff. Hidden evidence is never substituted. JSON-LD page + subject-property and multi-Voice union candidates #934/#968 remain separate + unmerged evidence, with #937/#971 also pending authorization work. + No fresh authenticated PostgreSQL/API or rendered Voice acceptance was + established here. An exact-head #968 Vitest attempt with one fork could not + start its test process before the runner startup timeout (zero tests ran). + The source diff preserves singleton/array property union, but that inspection + does not replace executable validation. UI/CSV and paged JSON-LD acceptance + remain open. + +## Historical snapshots — not current authority or acceptance + +The preceding file content is retained verbatim below for provenance. Its +head hashes, counts, GREEN claims and timestamps are historical only. + +# Product & Technical Gap Baseline + +> Ask ownership-fence overlay: 2026-09-08 KST. Issue #975 / #979 exact +> head `262d496a9` Tests run `34156704752` was GREEN. Follow-up: a dead +> heartbeat while compute still ran used to leave the owner live without +> renewals (RED `TimeoutError` in +> `test_dead_heartbeat_aborts_live_ask_operation`). The wrapper now +> aborts as a lost claim. Parent #974 Tests GREEN, BLOCKED on +> independent APPROVE. #979 has no independent APPROVE. Do not merge. +> ADR 0371 Proposed. Leftover-map #980 already holds ADR 0370. +> > Exact-head loop overlay: 2026-08-29 13:20 KST. Protected `main` is > `fc13acaa20adca11968238e398d4aafcf62b6cee` (v2.23.0 leftover-map > explained leftover share, #775). Open ready PRs still lack independent diff --git a/tests/test_global_ask_claim_cancellation.py b/tests/test_global_ask_claim_cancellation.py new file mode 100644 index 000000000..f14a37c64 --- /dev/null +++ b/tests/test_global_ask_claim_cancellation.py @@ -0,0 +1,183 @@ +"""Cancellation contracts for the Global Ask claim-heartbeat boundary.""" + +from __future__ import annotations + +import asyncio + +import pytest + +from backend.app import global_ask_queue + + +def test_external_cancellation_reaches_active_ask_operation(monkeypatch) -> None: + """Cancelling the owner task must not leave its Ask operation detached.""" + operation_started = asyncio.Event() + operation_cancelled = asyncio.Event() + + async def operation() -> None: + operation_started.set() + try: + await asyncio.Event().wait() + finally: + operation_cancelled.set() + + async def exercise() -> None: + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 3600.0) + owner = asyncio.create_task( + global_ask_queue._run_with_ask_claim_heartbeat( + object(), + "job-1", + [object()], + operation(), + ) + ) + await operation_started.wait() + owner.cancel() + with pytest.raises(asyncio.CancelledError): + await owner + await asyncio.sleep(0) + assert operation_cancelled.is_set(), ( + "owner cancellation left compute_global_ask_answer detached from the " + "claim heartbeat lifecycle" + ) + + asyncio.run(exercise()) + + +def test_dead_heartbeat_aborts_live_ask_operation(monkeypatch) -> None: + """A heartbeat that ends while compute is running must not keep the owner live.""" + operation_started = asyncio.Event() + operation_cancelled = asyncio.Event() + + async def operation() -> str: + operation_started.set() + try: + await asyncio.Event().wait() + return "should-not-settle" + finally: + operation_cancelled.set() + + async def exploding_renew(*_args: object, **_kwargs: object) -> None: + raise RuntimeError("heartbeat storage unavailable") + + async def exercise() -> None: + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.01) + monkeypatch.setattr(global_ask_queue, "_renew_ask_claim", exploding_renew) + owner = asyncio.create_task( + global_ask_queue._run_with_ask_claim_heartbeat( + object(), + "job-heartbeat-dead", + [object()], + operation(), + ) + ) + await operation_started.wait() + with pytest.raises(global_ask_queue._LostAskClaim): + await asyncio.wait_for(owner, timeout=1.0) + await asyncio.sleep(0) + assert operation_cancelled.is_set(), ( + "a dead claim heartbeat left compute_global_ask_answer running " + "without renewals" + ) + + asyncio.run(exercise()) + + +def test_answer_completion_drains_inflight_claim_renewal(monkeypatch) -> None: + """A committed renewal must reach settlement even if the answer finishes first.""" + renewal_started = asyncio.Event() + answer_finished = asyncio.Event() + renewal_cancelled = asyncio.Event() + original_generation = object() + committed_generation = object() + lease = [original_generation] + + async def renew(*_args): + renewal_started.set() + try: + await answer_finished.wait() + # Model the DB driver's response arriving after the committed UPDATE. + await asyncio.sleep(0) + await asyncio.sleep(0) + return committed_generation + except asyncio.CancelledError: + renewal_cancelled.set() + raise + + async def operation(): + await renewal_started.wait() + answer_finished.set() + return "completed answer" + + async def exercise(): + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.001) + monkeypatch.setattr(global_ask_queue, "_renew_ask_claim", renew) + result = await global_ask_queue._run_with_ask_claim_heartbeat( + object(), "job-renewal-race", lease, operation() + ) + assert result == "completed answer" + assert not renewal_cancelled.is_set() + assert lease[0] is committed_generation + + asyncio.run(exercise()) + + +def test_simultaneous_answer_and_failed_renewal_rejects_answer(monkeypatch) -> None: + """An answer cannot hide a renewal failure in the same event-loop turn.""" + renewal_started = asyncio.Event() + answer_finished = asyncio.Event() + + async def renew(*_args): + renewal_started.set() + await answer_finished.wait() + raise RuntimeError("renewal result unavailable") + + async def operation(): + await renewal_started.wait() + answer_finished.set() + return "answer without confirmed ownership" + + async def exercise(): + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.001) + monkeypatch.setattr(global_ask_queue, "_renew_ask_claim", renew) + with pytest.raises(global_ask_queue._LostAskClaim): + await global_ask_queue._run_with_ask_claim_heartbeat( + object(), "job-simultaneous-completion", [object()], operation() + ) + + asyncio.run(exercise()) + + +def test_owner_cancellation_interrupts_completion_drain(monkeypatch) -> None: + """Draining a renewal never detaches it from native owner cancellation.""" + renewal_started = asyncio.Event() + renewal_cancelled = asyncio.Event() + answer_finished = asyncio.Event() + + async def renew(*_args): + renewal_started.set() + try: + await asyncio.Event().wait() + finally: + renewal_cancelled.set() + + async def operation(): + await renewal_started.wait() + answer_finished.set() + return "completed answer" + + async def exercise(): + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.001) + monkeypatch.setattr(global_ask_queue, "_renew_ask_claim", renew) + owner = asyncio.create_task(global_ask_queue._run_with_ask_claim_heartbeat( + object(), "job-cancel-drain", [object()], operation() + )) + await answer_finished.wait() + await asyncio.sleep(0) + await asyncio.sleep(0) + owner.cancel() + with pytest.raises(asyncio.CancelledError): + await owner + assert renewal_cancelled.is_set() + + asyncio.run(exercise()) diff --git a/tests/test_global_ask_elapsed_deadline.py b/tests/test_global_ask_elapsed_deadline.py new file mode 100644 index 000000000..ba826cca4 --- /dev/null +++ b/tests/test_global_ask_elapsed_deadline.py @@ -0,0 +1,81 @@ +"""Regression for Global Ask liveness versus elapsed-time cancellation.""" + +from __future__ import annotations + +import asyncio +from contextlib import asynccontextmanager +from datetime import UTC, datetime, timedelta + +from backend.app import global_ask_queue + + +_CLAIMED_AT = datetime(2026, 1, 1, 9, 0, tzinfo=UTC) + + +class _AvailableClient: + available = True + + +class _Connection: + def __init__(self) -> None: + self.generation = _CLAIMED_AT + self.executed: list[tuple[str, tuple[object, ...]]] = [] + + async def fetchrow(self, query: str, *_args: object): + if "job_status_code = $3" in query: + return { + "requesting_account_id": "00000000-0000-0000-0000-000000000001", + "question_text": "Continue the live Ask operation", + "verify_external_requested": False, + "knowledge_cutoff": None, + "updated_at": self.generation, + } + if "returning updated_at" in query: + self.generation += timedelta(milliseconds=5) + return {"updated_at": self.generation} + raise AssertionError(query) + + async def execute(self, query: str, *args: object) -> str: + self.executed.append((query, args)) + return "UPDATE 1" + + +class _Pool: + def __init__(self, connection: _Connection) -> None: + self.connection = connection + + @asynccontextmanager + async def acquire(self): + yield self.connection + + +def test_live_heartbeat_operation_is_not_cancelled_by_elapsed_time(monkeypatch) -> None: + """A renewing owner remains live until completion, cancellation, or claim loss.""" + connection = _Connection() + pool = _Pool(connection) + + async def _load_visibility(_conn, _job_id, _account_id): + return {"corp-1"}, set(), False, True + + async def _long_but_live_answer(*_args, **_kwargs): + await asyncio.sleep(0.03) + return {"answer_text": "completed by live owner"} + + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.005) + monkeypatch.setattr(global_ask_queue, "load_job_visibility", _load_visibility) + monkeypatch.setattr( + global_ask_queue, "compute_global_ask_answer", _long_but_live_answer + ) + + asyncio.run( + global_ask_queue.process_global_ask_job( + pool, + job_id="job-live", + chat_factory=_AvailableClient, + ) + ) + + settle_query, settle_args = connection.executed[-1] + assert "answer_payload" in settle_query + assert "failure_detail" not in settle_query + assert settle_args[3:] == (global_ask_queue.RUNNING, connection.generation) diff --git a/tests/test_global_ask_queue.py b/tests/test_global_ask_queue.py index 453a933ef..d0aadc393 100644 --- a/tests/test_global_ask_queue.py +++ b/tests/test_global_ask_queue.py @@ -4,7 +4,7 @@ import asyncio from contextlib import asynccontextmanager -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta import pytest @@ -28,7 +28,7 @@ async def fetchrow(self, *_args: object): async def execute(self, query: str, *args: object) -> str: self.executed.append((query, args)) - return "OK" + return "UPDATE 1" class _Pool: @@ -40,12 +40,16 @@ async def acquire(self): yield self.connection +_CLAIMED_AT = datetime(2026, 1, 1, 9, 0, tzinfo=UTC) + + def _queued_row() -> dict[str, object]: return { "requesting_account_id": "00000000-0000-0000-0000-000000000001", "question_text": "What happened last week?", "verify_external_requested": False, "knowledge_cutoff": None, + "updated_at": _CLAIMED_AT, } @@ -398,7 +402,7 @@ async def _fake_compute_global_ask_answer(*_args, **_kwargs): settle_query, settle_args = connection.executed[-1] assert "failure_detail" in settle_query - failure_detail = settle_args[-1] + failure_detail = settle_args[2] assert secret_bearing_message not in failure_detail assert failure_detail == ( "Ask Agent is unavailable: contextual-orchestrator returned no complete evidence object" @@ -434,10 +438,11 @@ async def _fake_compute_global_ask_answer(*_args, **_kwargs): ) _settle_query, settle_args = connection.executed[-1] - assert settle_args[-1] == "account lacks the post_read permission" + assert settle_args[2] == "account lacks the post_read permission" + assert settle_args[3:] == (global_ask_queue.RUNNING, _CLAIMED_AT) -@pytest.mark.parametrize("timeout_source", ["provider", "worker", "shutdown"]) +@pytest.mark.parametrize("timeout_source", ["provider", "shutdown"]) def test_timeout_detail_identifies_only_an_expired_worker_deadline( monkeypatch, timeout_source, ) -> None: @@ -451,12 +456,8 @@ async def _fake_load_job_visibility(_conn, _job_id, _account_id): async def _fake_compute_global_ask_answer(*_args, **_kwargs): if timeout_source == "shutdown": raise asyncio.CancelledError() - if timeout_source == "worker": - await asyncio.Event().wait() raise asyncio.TimeoutError("synthetic private upstream detail") - if timeout_source == "worker": - monkeypatch.setattr(global_ask_queue, "JOB_DEADLINE_SECONDS", 0) monkeypatch.setattr(global_ask_queue, "load_job_visibility", _fake_load_job_visibility) monkeypatch.setattr( global_ask_queue, "compute_global_ask_answer", _fake_compute_global_ask_answer @@ -479,13 +480,159 @@ async def _fake_compute_global_ask_answer(*_args, **_kwargs): ) _settle_query, settle_args = connection.executed[-1] - if timeout_source == "worker": - assert settle_args[-1] == "job exceeded the 0s deadline" - else: - assert settle_args[-1] == ( - "Ask Agent is unavailable: contextual-orchestrator returned " - "no complete evidence object" + assert settle_args[2] == ( + "Ask Agent is unavailable: contextual-orchestrator returned " + "no complete evidence object" + ) + assert settle_args[3:] == (global_ask_queue.RUNNING, _CLAIMED_AT) + + +def test_orphan_reclaim_prevents_stale_owner_from_settling(monkeypatch) -> None: + """After recovery reclaims a running job, the previous owner cannot settle it.""" + reclaimed_at = datetime(2026, 1, 2, 9, 0, tzinfo=UTC) + + class _RaceConnection(_Connection): + def __init__(self) -> None: + super().__init__(_queued_row()) + self.status = global_ask_queue.QUEUED + self.generation = _CLAIMED_AT + self.applied: list[str] = [] + + async def fetchrow(self, query: str, *_args: object): + if "job_status_code = $3" in query and self.status == global_ask_queue.QUEUED: + self.status = global_ask_queue.RUNNING + return self.row + return None + + async def execute(self, query: str, *args: object) -> str: + self.executed.append((query, args)) + if "answer_payload" in query: + claimed = args[4] if len(args) > 4 else None + if ( + "updated_at = $5" in query + and args[3] == global_ask_queue.RUNNING + and claimed == self.generation + and self.status == global_ask_queue.RUNNING + ): + self.status = str(args[1]) + self.applied.append("accepted") + return "UPDATE 1" + self.applied.append("rejected") + return "UPDATE 0" + return "OK" + + connection = _RaceConnection() + pool = _Pool(connection) + + async def _fake_load_job_visibility(_conn, _job_id, _account_id): + return {"corp-1"}, set(), False, True + + async def _fake_compute_global_ask_answer(*_args, **_kwargs): + connection.status = global_ask_queue.RUNNING + connection.generation = reclaimed_at + return {"answer_text": "stale-owner-payload"} + + monkeypatch.setattr(global_ask_queue, "load_job_visibility", _fake_load_job_visibility) + monkeypatch.setattr( + global_ask_queue, "compute_global_ask_answer", _fake_compute_global_ask_answer + ) + + asyncio.run( + global_ask_queue.process_global_ask_job( + pool, + job_id="job-1", + chat_factory=_AvailableClient, + ) + ) + + settle_query, settle_args = connection.executed[-1] + assert "updated_at = $5" in settle_query + assert settle_args[3:] == (global_ask_queue.RUNNING, _CLAIMED_AT) + assert connection.applied == ["rejected"] + assert connection.status == global_ask_queue.RUNNING + assert global_ask_queue._claim_generation_retained("UPDATE 0") is False + assert global_ask_queue._claim_generation_retained("UPDATE 1") is True + + +def test_claim_heartbeat_advances_generation_used_at_settle(monkeypatch) -> None: + """A live worker renews updated_at so settlement matches the current generation.""" + renewed_at = _CLAIMED_AT + timedelta(seconds=30) + + class _HeartbeatConnection(_Connection): + async def fetchrow(self, query: str, *_args: object): + if "knowledge_cutoff, updated_at" in query: + return self.row + if "returning updated_at" in query: + assert self.row is not None + self.row = {**self.row, "updated_at": renewed_at} + return {"updated_at": renewed_at} + return self.row + + connection = _HeartbeatConnection(_queued_row()) + pool = _Pool(connection) + + async def _fake_load_job_visibility(_conn, _job_id, _account_id): + return {"corp-1"}, set(), False, True + + async def _fake_compute_global_ask_answer(*_args, **_kwargs): + await asyncio.sleep(0.05) + return {"answer_text": "live-owner"} + + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.01) + monkeypatch.setattr(global_ask_queue, "load_job_visibility", _fake_load_job_visibility) + monkeypatch.setattr( + global_ask_queue, "compute_global_ask_answer", _fake_compute_global_ask_answer + ) + + asyncio.run( + global_ask_queue.process_global_ask_job( + pool, + job_id="job-1", + chat_factory=_AvailableClient, + ) + ) + + settle_query, settle_args = connection.executed[-1] + assert "answer_payload" in settle_query + assert settle_args[4] == renewed_at + + +def test_lost_heartbeat_does_not_settle_a_reclaimed_job(monkeypatch) -> None: + """A failed claim renew leaves the reclaimed owner in place.""" + + class _LostHeartbeatConnection(_Connection): + async def fetchrow(self, query: str, *_args: object): + if "knowledge_cutoff, updated_at" in query: + return self.row + if "returning updated_at" in query: + return None + return self.row + + connection = _LostHeartbeatConnection(_queued_row()) + pool = _Pool(connection) + + async def _fake_load_job_visibility(_conn, _job_id, _account_id): + return {"corp-1"}, set(), False, True + + async def _fake_compute_global_ask_answer(*_args, **_kwargs): + await asyncio.Event().wait() + return {"answer_text": "should-not-settle"} + + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.01) + monkeypatch.setattr(global_ask_queue, "load_job_visibility", _fake_load_job_visibility) + monkeypatch.setattr( + global_ask_queue, "compute_global_ask_answer", _fake_compute_global_ask_answer + ) + + asyncio.run( + global_ask_queue.process_global_ask_job( + pool, + job_id="job-1", + chat_factory=_AvailableClient, ) + ) + + assert all("answer_payload" not in query for query, _args in connection.executed) def test_job_visibility_never_expands_past_queued_scope() -> None: diff --git a/tests/test_global_ask_transport_timeout.py b/tests/test_global_ask_transport_timeout.py new file mode 100644 index 000000000..c8653e976 --- /dev/null +++ b/tests/test_global_ask_transport_timeout.py @@ -0,0 +1,26 @@ +"""Regression contracts for the optional Global Ask transport timeout.""" + +from __future__ import annotations + +import pytest + +from backend.app.config import _validated_answer_timeout + + +def test_explicit_transport_timeout_is_not_coupled_to_removed_worker_deadline() -> None: + """A finite positive transport timeout is independent of worker liveness.""" + assert _validated_answer_timeout("900") == 900.0 + + +def test_explicit_transport_timeout_rejects_non_positive_or_non_finite_values() -> None: + """Explicit transport limits must be finite and strictly positive.""" + for raw in ("0", "-1", "nan", "inf", "-inf"): + with pytest.raises(ValueError): + _validated_answer_timeout(raw) + + +def test_omitted_or_blank_transport_timeout_remains_null() -> None: + """No configured timeout means no LineageWeave elapsed transport limit.""" + assert _validated_answer_timeout(None) is None + assert _validated_answer_timeout("") is None + assert _validated_answer_timeout(" ") is None diff --git a/tests/test_schema.py b/tests/test_schema.py index 94876295f..c95d91464 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -172,6 +172,12 @@ / "migrations" / "0203_global_ask_authorization_scope.sql" ) +_GLOBAL_ASK_CUTOFF_MIGRATION = ( + Path(__file__).resolve().parents[1] / "migrations" / "0212_global_ask_knowledge_cutoff.sql" +) +_GLOBAL_ASK_VERIFICATION_MIGRATION = ( + Path(__file__).resolve().parents[1] / "migrations" / "0218_global_ask_public_verification.sql" +) def _postgres_available() -> bool: @@ -240,10 +246,14 @@ def schema_db(): cur.execute(_SOURCE_EVENT_TIME_MIGRATION.read_text()) cur.execute(_GLOBAL_ASK_JOB_MIGRATION.read_text()) cur.execute(_GLOBAL_ASK_SCOPE_MIGRATION.read_text()) + cur.execute(_GLOBAL_ASK_CUTOFF_MIGRATION.read_text()) + cur.execute(_GLOBAL_ASK_VERIFICATION_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()) + cur.execute(_GLOBAL_ASK_CUTOFF_MIGRATION.read_text()) + cur.execute(_GLOBAL_ASK_VERIFICATION_MIGRATION.read_text()) # Match ADR 0166's production migration executor instead of # maintaining a fixture-owned SQL parser. subprocess.run( @@ -325,6 +335,106 @@ def test_migration_applies_cleanly(schema_db) -> None: assert expected <= tables +def test_postgres_stale_ask_owner_cannot_settle_after_reclaim(schema_db) -> None: + """#975: real PostgreSQL compare-and-set blocks the previous Ask owner.""" + with schema_db.cursor() as cur: + cur.execute( + """ + insert into user_account (external_subject_id, display_name, email_address) + values ('ask-claim-generation', 'Ask claim generation', 'ask-claim@example.test') + returning user_account_id + """ + ) + account_id = cur.fetchone()[0] + cur.execute( + """ + insert into global_ask_job + (requesting_account_id, question_text, job_status_code) + values (%s, 'reclaim settlement', 'queued') + returning global_ask_job_id + """, + (account_id,), + ) + job_id = cur.fetchone()[0] + schema_db.commit() + + parsed_admin_dsn = urlsplit(_ADMIN_DSN) + db_dsn = urlunsplit( + parsed_admin_dsn._replace(path=f"/{schema_db.get_dsn_parameters()['dbname']}") + ) + + async def race() -> None: + owner = await asyncpg.connect(db_dsn) + reclaim = await asyncpg.connect(db_dsn) + try: + claimed = await owner.fetchrow( + """ + update global_ask_job set job_status_code = $2, updated_at = now() + where global_ask_job_id = $1 and job_status_code = $3 + returning updated_at + """, + job_id, + "running", + "queued", + ) + assert claimed is not None + await reclaim.execute( + """ + update global_ask_job set job_status_code = $1, updated_at = now() + where global_ask_job_id = $2 and job_status_code = $3 + """, + "queued", + job_id, + "running", + ) + reclaimed = await reclaim.fetchrow( + """ + update global_ask_job set job_status_code = $2, updated_at = now() + where global_ask_job_id = $1 and job_status_code = $3 + returning updated_at + """, + job_id, + "running", + "queued", + ) + assert reclaimed is not None + stale = await owner.execute( + """ + update global_ask_job set job_status_code = $2, + answer_payload = $3::jsonb, updated_at = now() + where global_ask_job_id = $1 + and job_status_code = $4 + and updated_at = $5 + """, + job_id, + "succeeded", + '{"answer_text":"stale-owner"}', + "running", + claimed["updated_at"], + ) + live = await reclaim.execute( + """ + update global_ask_job set job_status_code = $2, + answer_payload = $3::jsonb, updated_at = now() + where global_ask_job_id = $1 + and job_status_code = $4 + and updated_at = $5 + """, + job_id, + "succeeded", + '{"answer_text":"live-owner"}', + "running", + reclaimed["updated_at"], + ) + assert stale == "UPDATE 0" + assert live == "UPDATE 1" + finally: + await owner.close() + await reclaim.close() + + asyncio.run(race()) + + def test_occupational_catalog_metadata_columns_exist(schema_db) -> None: """The real schema preserves catalog descriptions and release integrity.""" with schema_db.cursor() as cur: @@ -961,3 +1071,70 @@ def test_cataloged_team_null_affiliation_is_unique(schema_db) -> None: count = cursor.fetchone()[0] assert ids[0] == ids[1] assert count == 1 + + +def test_postgres_ask_completion_retains_inflight_renewal(schema_db, monkeypatch) -> None: + """A real committed renewal cannot strand a completed Ask in running state.""" + from backend.app import global_ask_queue + + with schema_db.cursor() as cur: + cur.execute( + "insert into user_account (external_subject_id, display_name, email_address) " + "values ('ask-renewal-race', 'Synthetic renewal', 'renewal@example.test') " + "returning user_account_id" + ) + account_id = cur.fetchone()[0] + cur.execute( + "insert into global_ask_job " + "(requesting_account_id, question_text, job_status_code) " + "values (%s, 'Synthetic renewal race', 'queued') returning global_ask_job_id", + (account_id,), + ) + job_id = cur.fetchone()[0] + schema_db.commit() + db_dsn = urlunsplit(urlsplit(_ADMIN_DSN)._replace( + path=f"/{schema_db.get_dsn_parameters()['dbname']}" + )) + + async def exercise(): + renewal_committed = asyncio.Event() + answer_finished = asyncio.Event() + renew = global_ask_queue._renew_ask_claim + + async def delayed_renew(*args): + generation = await renew(*args) + assert generation is not None + renewal_committed.set() + await answer_finished.wait() + await asyncio.sleep(0) + await asyncio.sleep(0) + return generation + + async def answer(*_args, **_kwargs): + await renewal_committed.wait() + answer_finished.set() + return {"answer_text": "Synthetic completed answer"} + + async def visibility(*_args): + return set(), set(), False, True + + class _Client: + available = True + + monkeypatch.setattr(global_ask_queue, "_CLAIM_HEARTBEAT_SECONDS", 0.001) + monkeypatch.setattr(global_ask_queue, "_renew_ask_claim", delayed_renew) + monkeypatch.setattr(global_ask_queue, "compute_global_ask_answer", answer) + monkeypatch.setattr(global_ask_queue, "load_job_visibility", visibility) + async with asyncpg.create_pool(db_dsn, min_size=1, max_size=2) as pool: + await global_ask_queue.process_global_ask_job( + pool, job_id=str(job_id), chat_factory=_Client + ) + async with pool.acquire() as conn: + row = await conn.fetchrow( + "select job_status_code, answer_payload from global_ask_job " + "where global_ask_job_id = $1", job_id + ) + assert row["job_status_code"] == "succeeded" + assert 'Synthetic completed answer' in row["answer_payload"] + + asyncio.run(exercise())