diff --git a/AGENTS.md b/AGENTS.md index 3af4169bc..2dbdf982a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -386,6 +386,16 @@ exist on the post. ## CI gates +Do not classify an upstream TimeoutError as proof that a local deadline expired. +Use the owning timer's expiration state and keep shutdown cancellation distinct +from failure settlement. Test upstream failure, actual timer expiry, and shutdown +independently; exception names alone do not identify the terminating boundary. + +For optional model timeouts, verify omission, explicit null, and explicit seconds +through factories and transport. Dropping a null keyword can silently restore a +downstream default; keep null intact and distinguish remaining worker limits +from a client's default. Null transport timeouts do not prove socket cancellation. + `.github/workflows/tests.yml` runs the full suite on every PR to `main`. Do not weaken, skip, or `continue-on-error` a failing check -- fix the underlying cause or, for a genuine false positive in a third-party scanner, diff --git a/backend/app/global_ask_queue.py b/backend/app/global_ask_queue.py index 9bffd8502..f8720f715 100644 --- a/backend/app/global_ask_queue.py +++ b/backend/app/global_ask_queue.py @@ -537,6 +537,7 @@ async def process_global_ask_job( ) if row is None: return + answer_timeout: asyncio.Timeout | None = None try: async with pool.acquire() as conn: ( @@ -554,8 +555,8 @@ async def process_global_ask_job( raise _SafeJobError( "Ask Agent is unavailable: set ORCHESTRATOR_BASE_URL / ORCHESTRATOR_API_KEY" ) - payload = await asyncio.wait_for( - compute_global_ask_answer( + async with asyncio.timeout(JOB_DEADLINE_SECONDS) as answer_timeout: + payload = await compute_global_ask_answer( pool, question_text=str(row["question_text"]), corporate_entity_ids=entity_ids, @@ -567,9 +568,7 @@ async def process_global_ask_job( verify_external=bool(row["verify_external_requested"]), claim_verification_client=claim_verification_factory(), knowledge_cutoff=row["knowledge_cutoff"], - ), - timeout=JOB_DEADLINE_SECONDS, - ) + ) except asyncio.CancelledError: # Shutdown: leave the row `running`; the recovery sweep re-queues # it after the orphan window on the next process start. @@ -587,7 +586,11 @@ 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): + 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 diff --git a/backend/app/main.py b/backend/app/main.py index 122165990..56b519ef4 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -323,8 +323,8 @@ 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). - # Only this worker gets the long answer timeout; the per-post chat - # endpoint keeps the client's interactive default. + # This worker still has an explicit answer socket limit; per-post + # chat retains the default null transport timeout. global_ask_worker = asyncio.create_task( run_global_ask_worker( valkey, @@ -508,19 +508,16 @@ def _post_structure_client(): def _post_chat_client(timeout: float | None = None): """Live orchestrator client when configured; otherwise the unavailable null. - ``timeout`` overrides the client's socket timeout. Only the Ask worker - passes the long answer timeout — the synchronous per-post chat endpoint - keeps the client default so an interactive request never hangs a reader - for the worker's full budget. + Preserve an omitted or explicit null timeout through the transport boundary. + The Ask worker still supplies its separately configured answer socket limit. """ settings = load_settings() if not (settings.orchestrator_base_url and settings.orchestrator_api_key): return NullPostChatClient() - kwargs = {} if timeout is None else {"timeout": timeout} return ContextualOrchestratorPostChatClient( base_url=settings.orchestrator_base_url, api_key=settings.orchestrator_api_key, - **kwargs, + timeout=timeout, ) diff --git a/docs/adr/0083-orchestrator-runtime-commit-pin.md b/docs/adr/0083-orchestrator-runtime-commit-pin.md index 1ad14cade..142802e63 100644 --- a/docs/adr/0083-orchestrator-runtime-commit-pin.md +++ b/docs/adr/0083-orchestrator-runtime-commit-pin.md @@ -47,3 +47,25 @@ The runtime contract is: - Rebuilding the image is required after the upstream pin changes. - Protected-branch review and merge remain external gates; this pin does not bypass upstream review. + +## Proposed amendment: post-chat transport timeout (2026-09-07) + +Status: Proposed; the Accepted runtime pin decision above is unchanged. + +The post-chat client silently supplies 180 seconds when a caller omits a limit. +The factory also drops an explicit null, restoring that limit. This contradicts +the requested default-null model lifetime even when the upstream owner has no +implicit limit. Increasing the constant merely postpones the same failure; a +second per-model policy store would duplicate contextual-orchestrator. + +Use null as the post-chat transport default and pass it unchanged through the +factory and shared HTTP transport. Preserve explicit caller limits while their +separate migration is pending. This avoids client abandonment by default but can +leave a synchronous chat waiting until transport/provider termination. Do not +claim cancellation of a blocking socket merely because an async task is cancelled. + +Confirm omitted/null/explicit values at the client and factory boundaries. The +Ask worker's explicit 570-second setting, 600-second execution deadline, and +age-based recovery remain unresolved. Other model clients and upstream model +administration require separate owner-aligned verification. No runtime pin is +changed and no open upstream PR becomes a released contract. diff --git a/docs/product-technical-gap-baseline.md b/docs/product-technical-gap-baseline.md index b5d31877b..42dc5d155 100644 --- a/docs/product-technical-gap-baseline.md +++ b/docs/product-technical-gap-baseline.md @@ -932,3 +932,40 @@ The ONET rows stacked into base branches (#743/#745/#746/#740/#732) reached `main` together through the #759 promotion; their per-base merge records are historical evidence only. The job-architecture artifact ship originally via #749 is now re-verified on `main` from the promotion. + +### Ask timeout attribution repair (2026-09-07) + +The queue worker treated every `asyncio.TimeoutError` as proof its 600-second +execution deadline expired. A provider that terminated immediately therefore +produced an incorrect durable deadline explanation. A paired synthetic regression +reproduced that mismatch (provider case failed; actual zero-duration worker timer +passed). The worker now uses the standard asyncio timeout context's expiration +state to attribute only its own expiry. Other failures retain ADR 0123's existing +bounded unavailable message; provider exception content is not persisted. + +The focused queue/service suite passed 22 tests in 1.59 s, including provider +failure, actual timer expiry, and shutdown cancellation with no failed settlement. +Compilation and diff checks passed. This does not remove the execution deadline, +change model policy, or repair age-based orphan recovery: default-null execution +still requires worker liveness and claim fencing. No database migration, new +container, provider call, deployment, or protected merge was performed. + +Python Software Foundation. (2026). *Coroutines and tasks: Timeouts*. +https://docs.python.org/3/library/asyncio-task.html#timeouts + +### Post-chat null timeout propagation (2026-09-07; proposed ADR 0083 amendment) + +Three synthetic assertions reproduced an implicit 180-second limit: direct +construction with no timeout, factory construction with no timeout, and factory +construction with explicit null. The post-chat client now defaults to null and +the factory passes the value unchanged. The shared HTTP request and JSON POST +annotations accept the native transport's null timeout without a new adapter. +Explicit caller seconds remain intact. The runtime pin is unchanged. + +Focused post-chat, HTTP, queue, and service tests passed 69 cases with two +real-provider cases skipped in 13.84 s. The existing local HTTP server test covers +both null and numeric limits. Compilation and diff checks passed. These results +do not prove blocking-socket cancellation, upstream policy enforcement, or +unlimited Ask execution: the explicit 570-second Ask socket setting, 600-second +worker deadline, and age-based recovery remain unresolved. The policy amendment +is Proposed, not a protected acceptance or release claim. diff --git a/lineageweave/http_client.py b/lineageweave/http_client.py index d1791cd05..7dc0fd0be 100644 --- a/lineageweave/http_client.py +++ b/lineageweave/http_client.py @@ -148,7 +148,7 @@ def _request( *, body: bytes | None, headers: dict[str, str], - timeout: float, + timeout: float | None, maximum_response_bytes: int | None = None, expected_response_media_type: str | None = None, ) -> tuple[int, bytes]: @@ -248,7 +248,7 @@ def post_json( payload: dict, *, headers: dict[str, str], - timeout: float, + timeout: float | None, service_peer_name: str = "contextual-orchestrator", ) -> dict: """POST ``payload`` as JSON to ``url`` and return the decoded object. diff --git a/lineageweave/post_chat.py b/lineageweave/post_chat.py index 419611033..9814b8797 100644 --- a/lineageweave/post_chat.py +++ b/lineageweave/post_chat.py @@ -377,7 +377,7 @@ class ContextualOrchestratorPostChatClient: available = True def __init__( - self, base_url: str, api_key: str, *, reasoning_effort: str = "auto", timeout: float = 180.0 + self, base_url: str, api_key: str, *, reasoning_effort: str = "auto", timeout: float | None = None ) -> None: self._base_url = base_url.rstrip("/") self._api_key = api_key diff --git a/tests/test_global_ask_queue.py b/tests/test_global_ask_queue.py index fed1b90d7..453a933ef 100644 --- a/tests/test_global_ask_queue.py +++ b/tests/test_global_ask_queue.py @@ -6,6 +6,8 @@ from contextlib import asynccontextmanager from datetime import UTC, datetime +import pytest + from backend.app import global_ask_queue from backend.app.global_ask_queue import load_job_visibility from lineageweave import claim_verification as cv @@ -435,11 +437,11 @@ async def _fake_compute_global_ask_answer(*_args, **_kwargs): assert settle_args[-1] == "account lacks the post_read permission" -def test_job_deadline_timeout_settles_with_a_specific_but_still_generic_detail( - monkeypatch, +@pytest.mark.parametrize("timeout_source", ["provider", "worker", "shutdown"]) +def test_timeout_detail_identifies_only_an_expired_worker_deadline( + monkeypatch, timeout_source, ) -> None: - """A bare `asyncio.TimeoutError` (no message) still gets a useful, - non-empty detail rather than an empty string.""" + """Provider timeout is not proof that the worker deadline expired.""" connection = _Connection(_queued_row()) pool = _Pool(connection) @@ -447,13 +449,27 @@ 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): - raise asyncio.TimeoutError() - + 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 ) + if timeout_source == "shutdown": + with pytest.raises(asyncio.CancelledError): + asyncio.run(global_ask_queue.process_global_ask_job( + pool, job_id="job-1", chat_factory=_AvailableClient, + )) + assert connection.executed == [] + return + asyncio.run( global_ask_queue.process_global_ask_job( pool, @@ -463,7 +479,13 @@ async def _fake_compute_global_ask_answer(*_args, **_kwargs): ) _settle_query, settle_args = connection.executed[-1] - assert settle_args[-1] == f"job exceeded the {global_ask_queue.JOB_DEADLINE_SECONDS}s deadline" + 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" + ) def test_job_visibility_never_expands_past_queued_scope() -> None: diff --git a/tests/test_http_client.py b/tests/test_http_client.py index 546bca797..e91182fb0 100644 --- a/tests/test_http_client.py +++ b/tests/test_http_client.py @@ -130,7 +130,8 @@ def test_post_json_refuses_missing_hostname() -> None: post_json("https:///v1/embeddings", {}, headers={}, timeout=1.0) -def test_post_json_posts_json_to_http_endpoint() -> None: +@pytest.mark.parametrize("request_timeout", [None, 2.0]) +def test_post_json_posts_json_to_http_endpoint(request_timeout) -> None: _JsonHandler.received = {} server, base = _serve(_JsonHandler) try: @@ -138,7 +139,7 @@ def test_post_json_posts_json_to_http_endpoint() -> None: f"{base}/v1/embeddings", {"model": "demo", "input": "hello"}, headers={"authorization": "Bearer test-token"}, - timeout=2.0, + timeout=request_timeout, ) finally: server.shutdown() diff --git a/tests/test_post_chat.py b/tests/test_post_chat.py index 3cf4b098f..0d1c232e4 100644 --- a/tests/test_post_chat.py +++ b/tests/test_post_chat.py @@ -469,3 +469,28 @@ def fake_post_json(url, payload, *, headers, timeout): assert observed["payload"]["reasoning_effort"] == "auto" assert observed["payload"]["mode"] == "auto" assert "CITED SOURCES" in observed["payload"]["messages"][0]["content"] + + +@pytest.mark.parametrize("creation_path", ["direct", "factory"]) +@pytest.mark.parametrize("timeout_options, expected", [({}, None), ({"timeout": None}, None), ({"timeout": 7.5}, 7.5)]) +def test_post_chat_preserves_optional_transport_timeout(monkeypatch, timeout_options, expected, creation_path) -> None: + """Omitted or null limits stay null; explicit limits reach transport unchanged.""" + observed = {} + + def fake_post_json(_url, _payload, *, headers, timeout): + observed["timeout"] = timeout + return {"choices": [{"message": {"content": "Answer\nCITED SOURCES: 1"}}]} + + monkeypatch.setattr("lineageweave.post_chat.post_json", fake_post_json) + if creation_path == "factory": + from types import SimpleNamespace + from backend.app import main + + monkeypatch.setattr(main, "load_settings", lambda: SimpleNamespace( + orchestrator_base_url="https://orchestrator.test", orchestrator_api_key="synthetic-token", + )) + client = main._post_chat_client(**timeout_options) + else: + client = ContextualOrchestratorPostChatClient("https://orchestrator.test", "synthetic-token", **timeout_options) + client.answer("Question", _SOURCES) + assert observed["timeout"] == expected