fix(gateway): land unbounded timeout prerequisite without unfinished policy changes

CHANGELOG.d/unbounded_model_timeout_prerequisite.md

@@ -0,0 +1,7 @@
+# Unbounded model timeout prerequisite
+
+ModelClient and its local-provider admission default to no application timeout instead of an implicit 90-second limit. Equivalent-endpoint races preserve `None`, and synchronous embedding waits for provider completion when no finite limit is configured. Durable embedding claim leases remain positive independently of model waiting. Explicit caller timeouts and existing discovery, probe, retention and benchmark boundaries are unchanged.
+
+This is the bounded prerequisite extracted from PR #1053 through ancestor `661ce8db75460c9f5752ba1493aad026e01f5316` and integrated in PR #1118. Later model-specific administrator policy, audit/API changes and their unresolved review findings remain in #1053; this change does not claim those features are released or repaired.
+
+Conflict resolution preserves protected main's `CHANGELOG.md` verbatim at blob `5adce934a08db3199ce9ca89cbfd7e179f2569a3` and moves only this prerequisite's three-line release note into this repository's existing fragment convention. No production source was changed during that resolution and no prior main release note was removed. Owner merge, immutable release and deployment/consumer verification are separate evidence boundaries.

CHANGELOG.md

@@ -20,6 +20,36 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html)
 
 ### Fixed
 
+- Virtual `orchestrator/free` structured completions (`response_format`, no
+  tools/stream) fail over a retryable synthesizer 502/429 onto the next
+  eligible free worker and attach request-scoped eligible/attempted
+  receipts. Concrete model pins stay sticky. Default model timeout remains
+  null (issue #1045; Inkspan Noema job 101628090366 on base `414f2297`).
+- Review-sidecar `orchestrator/free` admission now treats a CI-seeded
+  `OPENCODE_ZEN_API_KEY` as an authorized free-pool source. Honest-free
+  OpenCode Zen and OpenCode Go rows can enter `G ∩ P ∩ R`; `OPENAI_API_KEY`
+  remains registered for global discovery and is still excluded from the
+  review free pool (Noema 429 on `google/gemma-4-31b-it:free` in PR #1094
+  while Zen/Go evidence was dropped before routing).
+- `OPENCODE_ZEN_API_KEY` is documented as the shared KV credential for both
+  OpenCode Zen and OpenCode Go catalogs; registering it once discovers both
+  accounts.
+- Virtual selectors (`orchestrator/free`, `orchestrator/auto`,
+  `contextual-orchestrator`) keep tools and streaming on Fugu route /
+  TRINITY-Conductor conduct. A tools array no longer ejects those calls into
+  single-agent passthrough, so a failed worker is re-selected on the control
+  plane (incident: ContextualWisdomLab/.github run 34079284863, Strix step 23).
+  A worker `tool_calls` payload is returned as Chat Completions `tool_calls`
+  instead of being treated as missing assistant text. Concrete model ids
+  remain a debug pin. Psychometric θ̂/RMSE stays an equal-budget score of
+  those paper paths, not a separate router.
+- Streamed `/v1/responses` now emits OpenAI `response.reasoning_text.*`
+  events for TRINITY thinker/worker/verifier and Conductor step outputs,
+  while `response.reasoning_summary_*` stays the paper-role stage summary.
+  The synthesizer answer remains `output_text`. Chat Completions, audio,
+  image, video, embeddings, and rerank use the same worker re-selection
+  but cannot emit those reasoning events, so only the modality result is
+  returned.
 - Workflow workers now preserve the caller message array exactly once, while
   the added envelope carries only the subtask and Conductor-style prior-step
   access list instead of duplicating the task or source attachments.

contextual_orchestrator/batch_routing.py

@@ -1111,8 +1111,8 @@ def _publish_terminal(
                 self._errors[job_id] = error
             self._states[job_id] = status
 
-    def wait(self, job: BatchJob, *, timeout: float) -> Dict[str, Any]:
-        """Wait within the caller's explicit deadline for a terminal state."""
+    def wait(self, job: BatchJob, *, timeout: float | None) -> Dict[str, Any]:
+        """Wait for a terminal state, bounded only when the caller sets a deadline."""
         event = self._terminal_events.get(job.job_id)
         if event is not None:
             event.wait(timeout=timeout)

contextual_orchestrator/cost_router.py

@@ -59,6 +59,7 @@
 _DEFAULT_EMBEDDING_MAX_TOKENS_PER_REQUEST = 280_000
 _DEFAULT_EMBEDDING_MAX_CHARS_PER_PART = 240_000
 _DEFAULT_EMBEDDING_MAX_INPUTS_PER_REQUEST = 1
+_DEFAULT_PROVIDER_EMBEDDING_CLAIM_LEASE_SECONDS = 30.0
 _BATCH_LEDGER_SETTLEMENT_TIMEOUT_SECONDS = 1.0
 _EMBEDDING_UNIT_RE = re.compile(r"\S+\s*|\s+", re.UNICODE)
 
@@ -232,15 +233,22 @@ def _run_embedding_shard(
 
     def _provider_embedding_backend(self) -> ProviderEmbeddingBatchBackend:
         client = getattr(self.orchestrator, "client", None)
-        client_timeout = float(getattr(client, "timeout", 0))
+        configured_timeout = getattr(client, "timeout", None)
+        client_timeout = (
+            float(configured_timeout) if configured_timeout is not None else 0.0
+        )
         return ProviderEmbeddingBatchBackend(
             self._run_provider_embeddings,
             job_registry=self.job_registry,
             max_concurrency=getattr(client, "local_concurrency", 1),
             claim_lease_seconds=(
                 client_timeout
                 if self.job_registry.durable and client_timeout > 0
-                else None
+                else (
+                    _DEFAULT_PROVIDER_EMBEDDING_CLAIM_LEASE_SECONDS
+                    if self.job_registry.durable
+                    else None
+                )
             ),
             execution_timeout_seconds=client_timeout if client_timeout > 0 else None,
         )
@@ -1834,9 +1842,9 @@ def complete_embeddings_batch(
     ) -> Dict[str, Any]:
         """Submit an embeddings batch and return its document (one round-trip).
 
-        Local backends complete immediately. Callers that require a synchronous
-        provider result pass ``wait_timeout``; a timed-out queued job is
-        cancelled so the synchronous surface does not leave orphaned work.
+        Local backends complete immediately. ``wait_timeout=None`` waits without
+        an application deadline; a timed-out queued job is cancelled only when
+        the caller supplied a finite deadline.
         """
         job = self.submit_embeddings_batch(
             inputs,
@@ -1848,9 +1856,13 @@ def complete_embeddings_batch(
             owner_id=owner_id,
         )
         backend = self._embedding_backend_for(job)
-        if wait_timeout is not None and hasattr(backend, "wait"):
+        if hasattr(backend, "wait"):
             status = backend.wait(job, timeout=wait_timeout)
-            if not status.get("is_complete") and hasattr(backend, "cancel"):
+            if (
+                wait_timeout is not None
+                and not status.get("is_complete")
+                and hasattr(backend, "cancel")
+            ):
                 backend.cancel(job, reason="synchronous request deadline elapsed")
         return self.embeddings_batch_document(job.job_id, owner_id=owner_id)
 

contextual_orchestrator/endpoint_race.py

@@ -79,7 +79,7 @@ def race_first_valid(
     attempts: list[EndpointAttempt[T]],
     *,
     validate: Callable[[T], bool],
-    deadline_seconds: float,
+    deadline_seconds: float | None,
     max_concurrency: int,
     on_attempt_complete: Callable[[str, T | None, BaseException | None], None] | None = None,
 ) -> RaceOutcome[T]:
@@ -94,7 +94,7 @@ def race_first_valid(
         raise ValueError("immediate_race requires concurrency capacity of at least two")
     if max_concurrency < len(attempts):
         raise ValueError("immediate_race capacity must cover every declared endpoint")
-    if deadline_seconds <= 0:
+    if deadline_seconds is not None and deadline_seconds <= 0:
         raise ValueError("deadline_seconds must be positive")
     contract = attempts[0].contract
     if any(attempt.contract != contract for attempt in attempts[1:]):
@@ -128,8 +128,12 @@ def execute(attempt: EndpointAttempt[T]) -> T:
     last_error: BaseException | None = None
     try:
         while pending:
-            remaining = deadline_seconds - (time.monotonic() - started)
-            if remaining <= 0:
+            remaining = (
+                None
+                if deadline_seconds is None
+                else deadline_seconds - (time.monotonic() - started)
+            )
+            if remaining is not None and remaining <= 0:
                 raise TimeoutError("equivalent endpoint race exceeded its deadline")
             done, pending = wait(pending, timeout=remaining, return_when=FIRST_COMPLETED)
             if not done:

contextual_orchestrator/orchestrator.py

@@ -1030,15 +1030,15 @@ def _local_provider_state(base_url: str) -> _LocalProviderState:
 def _local_provider_slot(
     agent: ModelAgent,
     capacity: int,
-    timeout: float,
+    timeout: float | None,
 ):
     """Bound local requests and serialize model switches on a shared endpoint."""
     if not _is_local_provider_url(agent.base_url):
         yield
         return
 
     state = _local_provider_state(agent.base_url)
-    deadline = time.monotonic() + max(float(timeout), 0.0)
+    deadline = None if timeout is None else time.monotonic() + max(float(timeout), 0.0)
     with state.condition:
         while True:
             if state.active == 0:
@@ -1051,8 +1051,8 @@ def _local_provider_slot(
                 state.active += 1
                 break
 
-            remaining = deadline - time.monotonic()
-            if remaining <= 0:
+            remaining = None if deadline is None else deadline - time.monotonic()
+            if remaining is not None and remaining <= 0:
                 raise TimeoutError("local provider endpoint is busy past its request deadline")
             state.condition.wait(remaining)
 
@@ -1693,7 +1693,7 @@ class ModelClient:
 
     def __init__(
         self,
-        timeout: int = 90,
+        timeout: float | None = None,
         max_output_tokens: int = 2048,
         max_retries: int = 2,
         local_max_retries: int = 0,

contextual_orchestrator/server.py

@@ -7286,25 +7286,32 @@ def register_video_job(agent: ModelAgent, provider_result: dict[str, Any]) -> di
                     if not attribution.get("service"):
                         attribution["service"] = "embeddings_api"
                     started_at = time.perf_counter()
-                    embedding_deadline = time.monotonic() + float(
-                        orchestrator.client.timeout
+                    configured_timeout = orchestrator.client.timeout
+                    embedding_deadline = (
+                        None
+                        if configured_timeout is None
+                        else time.monotonic() + float(configured_timeout)
                     )
                     document = None
                     last_embedding_error: Exception | None = None
                     for embedding_agent in embedding_agents:
-                        remaining_timeout = embedding_deadline - time.monotonic()
-                        if remaining_timeout <= 0:
+                        remaining_timeout = (
+                            None
+                            if embedding_deadline is None
+                            else embedding_deadline - time.monotonic()
+                        )
+                        if remaining_timeout is not None and remaining_timeout <= 0:
                             break
                         attempt_started_at = time.perf_counter()
                         try:
-                            document = self._run(lambda agent=embedding_agent: coordinator.complete_embeddings_batch(
+                            document = self._run(lambda agent=embedding_agent, wait_timeout=remaining_timeout: coordinator.complete_embeddings_batch(
                                 inputs,
                                 model=agent.model,
                                 attribution=attribution,
                                 metadata={"actor_scope": "inference", "endpoint_alias": "embeddings"},
                                 zdr_only=zdr_only,
                                 agent_id=agent.id,
-                                wait_timeout=remaining_timeout,
+                                wait_timeout=wait_timeout,
                                 owner_id=security.principal_id(self.headers),
                             ))
                         except Exception as exc:  # noqa: BLE001 - measured member failover

tests/test_orchestrator_client_boundaries.py

@@ -236,6 +236,22 @@ def test_batch_results_must_be_a_mapping() -> None:
 # -- local provider slot concurrency ------------------------------------------
 
 
+def test_default_model_timeout_is_unbounded() -> None:
+    """Model and repair requests inherit no application wall-clock cap."""
+    assert ModelClient().timeout is None
+
+
+def test_local_slot_accepts_unbounded_waits() -> None:
+    """The local-agent coordinator preserves the shared unbounded default."""
+    agent = ModelAgent(
+        id="unbounded_slot_agent",
+        model="unbounded-slot-model",
+        base_url="local://127.0.0.1:59343/v1",
+    )
+    with _local_provider_slot(agent, 1, None):
+        pass
+
+
 def test_slot_shrinks_capacity_for_same_model_and_resets_when_empty() -> None:
     """Concurrent same-model holders shrink capacity; last release resets."""
     url = "local://127.0.0.1:59341/v1"

tests/test_provider_embedding_batch_backend.py

@@ -5,11 +5,7 @@
 
 import pytest
 
-from contextual_orchestrator.batch_routing import (
-    EmbeddingBatchRequest,
-    ProviderEmbeddingBatchBackend,
-)
-from contextual_orchestrator.batch_job_registry import JobRegistryFactory
+import contextual_orchestrator.cost_router as cost_router_module
 from contextual_orchestrator import (
     CostRoutingCoordinator,
     InMemoryConfigStore,
@@ -18,6 +14,11 @@
     PriceEntry,
     TaskOrchestrator,
 )
+from contextual_orchestrator.batch_job_registry import JobRegistryFactory
+from contextual_orchestrator.batch_routing import (
+    EmbeddingBatchRequest,
+    ProviderEmbeddingBatchBackend,
+)
 from contextual_orchestrator.orchestrator import ModelClient
 from contextual_orchestrator.provider_errors import ProviderUpstreamError
 from contextual_orchestrator.server import SecurityConfig, build_server
@@ -46,6 +47,63 @@ def count_text(self, text, model):
         return len(text.split())
 
 
+def test_default_client_keeps_batch_lifecycle_separate_from_model_timeout() -> None:
+    """A null model timeout does not break the existing batch-retention boundary."""
+    coordinator = CostRoutingCoordinator(
+        TaskOrchestrator([], allow_empty_agents=True),
+        embedding_token_counter=_SyntheticExactCounter(),
+    )
+
+    backend = coordinator._provider_embedding_backend()
+
+    assert backend._execution_timeout_seconds == 604_800
+    assert backend._claim_lease_seconds is None
+    backend.close()
+
+
+def test_durable_claim_lease_does_not_depend_on_model_timeout(monkeypatch) -> None:
+    """A null model timeout still supplies the durable registry a positive lease."""
+    coordinator = CostRoutingCoordinator(
+        TaskOrchestrator([], allow_empty_agents=True),
+        embedding_token_counter=_SyntheticExactCounter(),
+    )
+    coordinator.job_registry._client = object()
+    captured = {}
+
+    def capture_backend(*_args, **kwargs):
+        captured.update(kwargs)
+        return object()
+
+    monkeypatch.setattr(cost_router_module, "ProviderEmbeddingBatchBackend", capture_backend)
+
+    coordinator._provider_embedding_backend()
+
+    assert captured["claim_lease_seconds"] == 30.0
+    assert captured["execution_timeout_seconds"] is None
+
+
+def test_unbounded_synchronous_embedding_waits_for_provider_completion() -> None:
+    """No application deadline means wait, rather than return an unfinished document."""
+    def delayed_runner(requests):
+        time.sleep(0.05)
+        return [[1.0] for _request in requests], len(requests)
+
+    backend = ProviderEmbeddingBatchBackend(delayed_runner)
+    coordinator = CostRoutingCoordinator(
+        TaskOrchestrator([], allow_empty_agents=True),
+        embedding_batch_backend=backend,
+        embedding_token_counter=_SyntheticExactCounter(),
+    )
+
+    document = coordinator.complete_embeddings_batch(
+        ["delayed provider input"], model="synthetic-model"
+    )
+
+    assert document["status"] == "completed"
+    assert document["embeddings"][0]["embedding"] == [1.0]
+    backend.close()
+
+
 def test_unknown_tokenizer_uses_authoritative_provider_usage() -> None:
     """A byte-safe request completes only after the provider supplies exact usage."""
     agent = ModelAgent(