feat: add verbose/debug logging for route/conduct/provider control flow

CHANGELOG.d/verbose-debug-logging.md

@@ -0,0 +1 @@
+Added a `--verbose`/`--debug` CLI flag (also settable via `CONTEXTUAL_ORCHESTRATOR_VERBOSE=true` for an already-deployed server, without a CLI edit) that enables stdlib DEBUG-level logging of the route-vs-conduct decision, each route/conduct step's agent and latency, `ModelClient` provider request/retry/backoff attempts, `TaskOrchestrator._invoke`'s failover classification, per-agent circuit-breaker open/half-open/closed transitions, and the HTTP request/response lifecycle in `server.py`. Off by default with no behavior change; every new log record is bounded and secret-free by construction — never an API key, an `Authorization` header, a raw exception message, or prompt/response content (ADR 0125).

README.md

@@ -82,6 +82,7 @@ HTTP serving is hardened for local lab use:
 - Full orchestration traces are not returned by default. Set `include_orchestration_trace: true` per standard Chat request or start with `--expose-trace-by-default` when the caller is trusted. Requests that also use `tools` or `response_format` still fail with `unsupported_trace_disclosure`; remove the trace flag for structured or tool requests.
 - State is in-memory by default. Pass `--state-db PATH` (or `CONTEXTUAL_ORCHESTRATOR_STATE_DB`) to persist workflow runs, evaluation runs, audit, and analytics to a stdlib sqlite file so they survive a restart; without it, behavior is unchanged.
 - Response caching is off by default. Pass `--cache-ttl SECONDS` to serve identical requests (same messages + mode) from an in-memory TTL+LRU cache and skip the provider calls; `0` disables it.
+- Verbose/debug logging is off by default. Pass `--verbose`/`--debug` (or set `CONTEXTUAL_ORCHESTRATOR_VERBOSE=true` and restart, for an already-deployed server) to enable stdlib DEBUG-level logging of the route-vs-conduct decision, each route/conduct step's agent and latency, provider request/retry/backoff attempts, and circuit-breaker open/half-open/closed transitions. Log records are bounded and secret-free by construction: never an API key, an `Authorization` header, or prompt/response content — see [ADR 0125](docs/planning/adrs/0125-verbose-debug-logging.md).
 - `ModelClient.batch_chat(agent, {custom_id: messages})` runs many requests through the provider's Batch API (async, 24h completion window, typically ~50% cheaper) — suited to evaluation/benchmark workloads, not latency-sensitive chat. The mock path answers synchronously.
 
 Use real workers by replacing `mock://` agents with OpenAI-compatible endpoints. Provider secrets are resolved from a KV credential registry via `get_credential`, never from `os.getenv` at request time (see [docs/kv-credentials.md](docs/kv-credentials.md)):

contextual_orchestrator/__main__.py

@@ -4,6 +4,7 @@
 
 import argparse
 import json
+import logging
 import os
 import sys
 from dataclasses import replace
@@ -41,6 +42,32 @@
 DEFAULT_AUTH_CREDENTIAL_NAME = "CONTEXTUAL_ORCHESTRATOR_TOKEN"
 DEFAULT_ADMIN_CREDENTIAL_NAME = "CONTEXTUAL_ORCHESTRATOR_ADMIN_TOKEN"
 DEFAULT_INFERENCE_CREDENTIAL_NAME = "CONTEXTUAL_ORCHESTRATOR_INFERENCE_TOKEN"
+#: Env-var escape hatch so an already-deployed server can turn on DEBUG-level
+#: logging (env is read once at process start, same as CONTEXTUAL_ORCHESTRATOR_STATE_DB
+#: below) without editing its CLI invocation.
+VERBOSE_ENV_VAR = "CONTEXTUAL_ORCHESTRATOR_VERBOSE"
+_VERBOSE_LOG_FORMAT = "%(asctime)s %(levelname)s %(name)s: %(message)s"
+
+
+def _env_flag(name: str) -> bool:
+    """Return whether an environment variable is set to a truthy flag value."""
+    return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _configure_logging(verbose: bool) -> None:
+    """Turn on DEBUG-level logging (timestamp, level, logger name) when requested.
+
+    A no-op when ``verbose`` is false: this repository never introduces
+    unrequested log noise, and ``.debug()`` call sites across the package stay
+    silent unless a caller explicitly opts in here or via ``VERBOSE_ENV_VAR``.
+    ``force=True`` lets a later call (or a test invoking ``main()`` more than
+    once in one process) replace an earlier logging configuration instead of
+    silently no-op'ing, matching stdlib ``logging.basicConfig`` semantics for
+    a process whose configuration is decided exactly once at startup.
+    """
+    if not verbose:
+        return
+    logging.basicConfig(level=logging.DEBUG, format=_VERBOSE_LOG_FORMAT, force=True)
 
 
 def _bootstrap_telemetry_config() -> InMemoryConfigStore:
@@ -196,7 +223,15 @@ def _register_credential_command(argv: list[str]) -> None:
         metavar="VAR",
         help="Bootstrap transport: read the secret value from this env var (e.g. --from-env OPENAI_API_KEY).",
     )
+    parser.add_argument(
+        "--verbose", "--debug",
+        dest="verbose",
+        action="store_true",
+        default=_env_flag(VERBOSE_ENV_VAR),
+        help=f"Enable DEBUG-level logging (default: off; also settable via {VERBOSE_ENV_VAR}).",
+    )
     args = parser.parse_args(argv)
+    _configure_logging(args.verbose)
 
     if args.from_env:
         # Bootstrap transport only: the deploy step injects secrets.OPENAI_API_KEY
@@ -298,7 +333,15 @@ def _discover_models_command(argv: list[str]) -> None:
         default=os.environ.get("CONTEXTUAL_ORCHESTRATOR_PROVIDER_CA_BUNDLE") or None,
         help="Optional reviewed CA bundle for configured-gateway discovery TLS verification.",
     )
+    parser.add_argument(
+        "--verbose", "--debug",
+        dest="verbose",
+        action="store_true",
+        default=_env_flag(VERBOSE_ENV_VAR),
+        help=f"Enable DEBUG-level logging (default: off; also settable via {VERBOSE_ENV_VAR}).",
+    )
     args = parser.parse_args(argv)
+    _configure_logging(args.verbose)
     if args.enable_cheapest and not args.agents_db:
         parser.error("--enable-cheapest requires --agents-db")
 
@@ -553,7 +596,18 @@ def main(argv: list[str] | None = None) -> None:
         action="store_true",
         help="discover source-declared chat-capable models at startup and activate them",
     )
+    parser.add_argument(
+        "--verbose", "--debug",
+        dest="verbose",
+        action="store_true",
+        default=_env_flag(VERBOSE_ENV_VAR),
+        help="Enable DEBUG-level logging (timestamp, level, logger name) to stderr for the "
+        f"orchestrator's route/conduct/provider-retry/circuit-breaker control flow; secrets "
+        f"and full prompt/response text are never logged (default: off; also settable via "
+        f"{VERBOSE_ENV_VAR}=true so a deployed server can turn it on without a new CLI flag).",
+    )
     args = parser.parse_args(arguments)
+    _configure_logging(args.verbose)
 
     client = ModelClient(
         ca_bundle=args.provider_ca_bundle,

contextual_orchestrator/orchestrator.py

@@ -16,6 +16,7 @@
 import io
 import ipaddress
 import json
+import logging
 import math
 import os
 from pathlib import Path
@@ -86,6 +87,8 @@
 )
 
 
+_LOGGER = logging.getLogger(__name__)
+
 # content is usually str; multimodal vision messages use OpenAI content-parts lists.
 ChatMessage = dict[str, Any]
 ProviderDestination = tuple[int, tuple[Any, ...]]
@@ -1595,6 +1598,13 @@ def _send_with_retry(
         last_error: Exception | None = None
         retry_limit = self._retry_limit(agent)
         for attempt in range(retry_limit + 1):  # pragma: no branch - retry limits are validated non-negative
+            _LOGGER.debug(
+                "provider.attempt agent_id=%s model=%s attempt=%s retry_limit=%s",
+                agent.id,
+                agent.model,
+                attempt,
+                retry_limit,
+            )
             try:
                 return (
                     self._send(agent, payload, destination)
@@ -1604,8 +1614,22 @@ def _send_with_retry(
             except Exception as exc:  # noqa: BLE001 - classify then decide
                 last_error = exc
                 if attempt >= retry_limit or not is_transient_error(exc):
+                    _LOGGER.debug(
+                        "provider.attempt_exhausted agent_id=%s attempt=%s error_type=%s",
+                        agent.id,
+                        attempt,
+                        type(exc).__name__,
+                    )
                     break
-                self._sleep(self._backoff_delay(attempt))
+                delay = self._backoff_delay(attempt)
+                _LOGGER.debug(
+                    "provider.retry_scheduled agent_id=%s attempt=%s delay_seconds=%.3f error_type=%s",
+                    agent.id,
+                    attempt,
+                    delay,
+                    type(exc).__name__,
+                )
+                self._sleep(delay)
         if isinstance(last_error, urllib.error.HTTPError) and _is_tool_execution_stopped(last_error):
             raise _provider_tool_execution_stopped(agent) from None
         if isinstance(last_error, urllib.error.HTTPError) and (
@@ -1653,6 +1677,16 @@ def _send(
             headers=headers,
             method="POST",
         )
+        # Bounded metadata only: never the api_key/headers just built above, and
+        # never payload/message content -- see redact_text/redact_value for the
+        # one place raw text is intentionally allowed to reach a trace.
+        _LOGGER.debug(
+            "provider.request agent_id=%s model=%s provider=%s host=%s",
+            agent.id,
+            agent.model,
+            agent.provider_name or "",
+            urlparse(request.full_url).hostname,
+        )
         started = time.monotonic()
         opened = (
             self._open_provider(request, destination)
@@ -4385,7 +4419,9 @@ def _dispatch(
                 or not self._needs_workflow(text)
             )
         ):
+            _LOGGER.debug("dispatch.decision mode=%s model=%s path=route", mode, model_name)
             return self.route_once(messages, model_name=model_name)
+        _LOGGER.debug("dispatch.decision mode=%s model=%s path=conduct", mode, model_name)
         return self.conduct(messages, model_name=model_name)
 
     def would_route(
@@ -5217,6 +5253,12 @@ def route_once(
             if len(tried_ids) >= max_attempts:
                 break
             tried_ids.add(candidate.id)
+            _LOGGER.debug(
+                "route.attempt_started attempt=%s agent_id=%s model=%s",
+                attempt_index,
+                candidate.id,
+                candidate.model,
+            )
             start = time.perf_counter()
             attempt_answer, attempt_served_id, attempt_usage = self._invoke(
                 candidate,
@@ -5257,6 +5299,14 @@ def route_once(
                 "reason": verification["reason"],
             }
             trace_rows.append(row)
+            _LOGGER.debug(
+                "route.attempt_completed attempt=%s agent_id=%s served_id=%s accepted=%s latency_ms=%s",
+                attempt_index,
+                candidate.id,
+                served_id,
+                verification["accepted"],
+                row["latency_ms"],
+            )
             if verification["accepted"]:
                 break
             # Rejected answers already recorded a quality-ledger failure in
@@ -5454,6 +5504,13 @@ def conduct(
                     "content": user_content,
                 },
             ]
+            _LOGGER.debug(
+                "conduct.step_started step_id=%s role=%s agent_id=%s access=%s",
+                step.id,
+                step.role,
+                agent.id,
+                step.access,
+            )
             start = time.perf_counter()
             output, served_id, usage = self._invoke(
                 agent,
@@ -5463,6 +5520,15 @@ def conduct(
                 allowed_agent_ids=free_ids if model_name == self.FREE_MODEL else None,
             )
             elapsed = (time.perf_counter() - start) * 1000
+            _LOGGER.debug(
+                "conduct.step_completed step_id=%s role=%s agent_id=%s served_id=%s latency_ms=%.2f failover=%s",
+                step.id,
+                step.role,
+                agent.id,
+                served_id,
+                elapsed,
+                served_id != agent.id,
+            )
             outputs[step.id] = output
             row = step.as_dict()
             row["agent_id"] = agent.id
@@ -6547,7 +6613,14 @@ def call(agent: ModelAgent) -> tuple[str, str, dict[str, Any] | None]:
         # fully-failed pool surfaces *why* (rate limit, auth, timeout) instead of
         # one opaque collapse message.
         last_upstream_error: ProviderUpstreamError | None = None
-        for agent in candidates:
+        for candidate_index, agent in enumerate(candidates):
+            _LOGGER.debug(
+                "invoke.candidate_attempt role=%s agent_id=%s candidate_index=%s candidate_count=%s",
+                role,
+                agent.id,
+                candidate_index,
+                len(candidates),
+            )
             retry_attempt = 0
             while True:
                 try:
@@ -6601,12 +6674,32 @@ def call(agent: ModelAgent) -> tuple[str, str, dict[str, Any] | None]:
                         bounded_provider_response_failures += 1
                         last_provider_response_error = exc
                         decision = classify_tool_failure(exc)
+                        _LOGGER.debug(
+                            "invoke.failure_classified agent_id=%s role=%s kind=%s action=%s "
+                            "reason_code=%s retry_attempt=%s",
+                            agent.id,
+                            role,
+                            decision.kind,
+                            decision.action,
+                            decision.reason_code,
+                            retry_attempt,
+                        )
                         self._record_tool_fallback(agent.id, decision, retry_attempt)
                         self._record_failure(agent.id)
                         break
                     else:
                         decision = classify_tool_failure(exc)
                     action = decision.action
+                    _LOGGER.debug(
+                        "invoke.failure_classified agent_id=%s role=%s kind=%s action=%s "
+                        "reason_code=%s retry_attempt=%s",
+                        agent.id,
+                        role,
+                        decision.kind,
+                        action,
+                        decision.reason_code,
+                        retry_attempt,
+                    )
                     # A failed attempt is one Bernoulli stability observation
                     # for measured group routing regardless of what happens next.
                     if (
@@ -6629,6 +6722,12 @@ def call(agent: ModelAgent) -> tuple[str, str, dict[str, Any] | None]:
                     if action is ToolFallbackAction.RETRY_SAME_AGENT:
                         decision = downgrade_to_failover(decision)
                         action = decision.action
+                        _LOGGER.debug(
+                            "invoke.action_downgraded agent_id=%s action=%s reason_code=%s",
+                            agent.id,
+                            action,
+                            decision.reason_code,
+                        )
                     self._record_tool_fallback(agent.id, decision, retry_attempt)
                     if decision.circuit_failure:
                         self._record_failure(agent.id)
@@ -6648,6 +6747,12 @@ def call(agent: ModelAgent) -> tuple[str, str, dict[str, Any] | None]:
                         output_tokens=output_tokens,
                     )
                 self._record_success(agent.id)
+                _LOGGER.debug(
+                    "invoke.success agent_id=%s role=%s latency_ms=%.2f",
+                    agent.id,
+                    role,
+                    (time.perf_counter() - attempt_start) * 1000,
+                )
                 return output, agent.id, usage
         if (
             last_provider_response_error is not None
@@ -6742,22 +6847,29 @@ def _circuit_open(self, agent_id: str) -> bool:
             state = self._circuit.get(agent_id)
             if not state or state["failures"] < self.circuit_failure_threshold:
                 return False
-            if time.monotonic() - state["opened_at"] >= self.circuit_reset_seconds:
-                state["failures"] = 0.0
-                state["opened_at"] = 0.0
-                return False
-            return True
+            if time.monotonic() - state["opened_at"] < self.circuit_reset_seconds:
+                return True
+            state["failures"] = 0.0
+            state["opened_at"] = 0.0
+        _LOGGER.debug("circuit.half_open agent_id=%s", agent_id)
+        return False
 
     def _record_failure(self, agent_id: str) -> None:
         with self._circuit_lock:
             state = self._circuit.setdefault(agent_id, {"failures": 0.0, "opened_at": 0.0})
             state["failures"] += 1.0
-            if state["failures"] >= self.circuit_failure_threshold and not state["opened_at"]:
+            just_opened = state["failures"] >= self.circuit_failure_threshold and not state["opened_at"]
+            if just_opened:
                 state["opened_at"] = time.monotonic()
+            failure_count = int(state["failures"])
+        if just_opened:
+            _LOGGER.debug("circuit.opened agent_id=%s failures=%s", agent_id, failure_count)
 
     def _record_success(self, agent_id: str) -> None:
         with self._circuit_lock:
-            self._circuit.pop(agent_id, None)
+            previously_tracked = self._circuit.pop(agent_id, None)
+        if previously_tracked is not None:
+            _LOGGER.debug("circuit.closed agent_id=%s", agent_id)
 
     def _agent(self, agent_id: str) -> ModelAgent:
         for agent in self.candidates: