fix(ask): fence Ask settlement on the claim generation

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", ""),

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,24 +592,26 @@ 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(
             """
             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,
             QUEUED,
         )
     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

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,

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(