feat(decisions): degrade gracefully when the decision model is unavailable

The decision model sits in hot paths, so a failure has to reach callers as a
``DecisionModelError`` they can fall back from, and a down endpoint may not
keep costing the retry and timeout budget:

- Normalize every failure: a send error that is not an ``httpx.HTTPError``
  (for example ``httpx.InvalidURL``) and a payload that fails pydantic
  validation now surface as ``DecisionModelRequestError`` /
  ``DecisionModelResponseError`` instead of escaping past the callers'
  ``except DecisionModelError``.
- ``timeout`` is the budget of one whole judgement, retries and backoff
  included, and is capped at 5 seconds.
- After ``failure_threshold`` failures in a row the endpoint is marked down
  for ``cooldown_seconds``; judgements then raise
  ``DecisionModelUnavailableError`` without an HTTP call, and one probe after
  the cooldown decides whether to resume.
- Log one line per usable judgement at debug, retries and outages at warning;
  never log the judged state or the API key.
- Reject an unusable ``api_base`` at configuration time, and keep the
  extension disabled instead of failing startup when settings are broken.

Change-Id: I58b57993c4791c6e1c57e7c52af85af53bd05c85

Author: wujiaming-ai

tests/extensions/decisions/test_client.py

@@ -146,7 +146,7 @@ def test_rate_limit_is_retried_and_can_succeed() -> None:
 def test_retries_are_bounded() -> None:
     script = [(529, {"retry-after": "0"}, {"detail": "overloaded"})] * 3
     with fake_system_one(script) as server:
-        with pytest.raises(DecisionModelRequestError, match="after 2 retries"):
+        with pytest.raises(DecisionModelRequestError, match="after 3 attempt"):
             _client(server.base_url, max_retries=2).evaluate(
                 state="hi", questions={"q": noul_question("Is this a greeting?")}
             )

tests/extensions/decisions/test_config.py

@@ -21,7 +21,10 @@
 
 from veadk.extensions.decisions import (
     DEFAULT_API_BASE,
+    DEFAULT_COOLDOWN_SECONDS,
+    DEFAULT_FAILURE_THRESHOLD,
     DEFAULT_MODEL_NAME,
+    MAX_TIMEOUT_SECONDS,
     OPENROUTER_API_BASE,
     DecisionModelConfig,
 )
@@ -44,17 +47,21 @@ def test_from_env_reads_every_field() -> None:
             "DECISION_MODEL_NAME": "jev-1.13.0",
             "DECISION_MODEL_API_BASE": "http://localhost:9000/",
             "DECISION_MODEL_API_KEY": "secret",
-            "DECISION_MODEL_TIMEOUT": "12.5",
+            "DECISION_MODEL_TIMEOUT": "4.5",
             "DECISION_MODEL_MAX_RETRIES": "1",
+            "DECISION_MODEL_FAILURE_THRESHOLD": "5",
+            "DECISION_MODEL_COOLDOWN_SECONDS": "12.5",
         }
     )
     assert config.enabled is True
     assert config.provider == "systemone"
     assert config.name == "jev-1.13.0"
     assert config.api_base == "http://localhost:9000"
     assert config.api_key == "secret"
-    assert config.timeout == 12.5
+    assert config.timeout == 4.5
     assert config.max_retries == 1
+    assert config.failure_threshold == 5
+    assert config.cooldown_seconds == 12.5
     assert config.endpoint == "http://localhost:9000/v1/systemone"
     assert config.configured is True
 
@@ -70,8 +77,41 @@ def test_from_env_keeps_defaults_for_unusable_values() -> None:
     )
     assert config.enabled is False
     assert config.provider == "typesafe"
-    assert config.timeout == 30.0
+    assert config.timeout == MAX_TIMEOUT_SECONDS
     assert config.max_retries == 3
+    assert config.failure_threshold == DEFAULT_FAILURE_THRESHOLD
+    assert config.cooldown_seconds == DEFAULT_COOLDOWN_SECONDS
+
+
+def test_timeout_is_capped_so_one_judgement_cannot_stall_a_run() -> None:
+    """A judgement sits before and after model calls, so it has a hard ceiling."""
+    assert DecisionModelConfig().timeout == MAX_TIMEOUT_SECONDS
+    assert DecisionModelConfig(timeout=30.0).timeout == MAX_TIMEOUT_SECONDS
+    assert (
+        DecisionModelConfig.from_env({"DECISION_MODEL_TIMEOUT": "300"}).timeout
+        == MAX_TIMEOUT_SECONDS
+    )
+
+
+@pytest.mark.parametrize(
+    "api_base",
+    ["http://[::1", "not a url", "ftp://host", "http://", "https://user:pw@host"],
+)
+def test_unusable_api_base_is_rejected_at_configuration_time(api_base: str) -> None:
+    with pytest.raises(ValidationError):
+        DecisionModelConfig(api_base=api_base)
+
+
+def test_from_env_disables_itself_instead_of_breaking_startup() -> None:
+    """Broken settings degrade to "no decision model" with a warning."""
+    config = DecisionModelConfig.from_env(
+        {
+            "DECISION_MODEL_ENABLED": "true",
+            "DECISION_MODEL_API_KEY": "secret",
+            "DECISION_MODEL_API_BASE": "http://[::1",
+        }
+    )
+    assert config.configured is False
 
 
 @pytest.mark.parametrize(

tests/extensions/decisions/test_resilience.py

@@ -0,0 +1,238 @@
+# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Failure handling, degradation, and availability tests.
+
+The decision model is optional and sits in hot paths, so every failure has to
+reach callers as a ``DecisionModelError`` they can fall back from, and a
+down endpoint must stop costing the retry and timeout budget.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+
+import httpx
+import pytest
+
+from veadk.extensions.decisions import (
+    DecisionExtension,
+    DecisionModelConfig,
+    DecisionModelRequestError,
+    DecisionModelResponseError,
+    DecisionModelUnavailableError,
+    SystemOneClient,
+    noul_question,
+)
+from veadk.extensions.decisions.types import parse_answers
+
+from .fake_system_one import FakeSystemOneServer, fake_system_one
+
+QUESTION = {"q": noul_question("Is this a greeting?")}
+
+
+def _config(
+    server_url: str,
+    *,
+    timeout: float = 5.0,
+    max_retries: int = 0,
+    failure_threshold: int = 3,
+    cooldown_seconds: float = 30.0,
+) -> DecisionModelConfig:
+    return DecisionModelConfig(
+        enabled=True,
+        api_base=server_url,
+        api_key="test-key",
+        name="jev-latest",
+        timeout=timeout,
+        max_retries=max_retries,
+        failure_threshold=failure_threshold,
+        cooldown_seconds=cooldown_seconds,
+    )
+
+
+def _extension(server_url: str, **settings: float | int) -> DecisionExtension:
+    return DecisionExtension(_config(server_url, **settings))
+
+
+def _schedule_outage(server: FakeSystemOneServer, count: int = 1) -> None:
+    """Make the next ``count`` requests fail like a down endpoint."""
+    server.script.extend([(503, {}, {"detail": "unavailable"})] * count)
+
+
+def _judge(extension: DecisionExtension) -> float:
+    """Ask one noul question and return the probability."""
+    return extension.noul("state", "Is this a greeting?").noul
+
+
+# -- every failure is a decision-model error -------------------------------
+
+
+def test_a_malformed_answer_is_a_response_error() -> None:
+    """A payload that fails pydantic validation must not escape as itself."""
+    with pytest.raises(DecisionModelResponseError, match="not a valid choice"):
+        parse_answers({"q": {"type": "choice"}})
+    with pytest.raises(DecisionModelResponseError, match="not a valid noul"):
+        parse_answers({"q": {"type": "noul", "noul": "certainly"}})
+
+
+def test_a_broken_answer_payload_reaches_callers_as_a_decision_error() -> None:
+    script = [(200, {}, {"model": "fake", "answers": {"q": {"type": "choice"}}})]
+    with fake_system_one(script) as server:
+        with pytest.raises(DecisionModelResponseError):
+            _judge(_extension(server.base_url))
+
+
+def test_a_send_failure_that_is_not_an_http_error_is_normalized(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """``httpx.InvalidURL`` is not an ``httpx.HTTPError``, so it needs our own."""
+    assert not issubclass(httpx.InvalidURL, httpx.HTTPError)
+
+    def _raise_invalid_url(*_args: object, **_kwargs: object) -> None:
+        raise httpx.InvalidURL("Invalid port: ':1'")
+
+    monkeypatch.setattr(httpx.Client, "post", _raise_invalid_url)
+    client = SystemOneClient(_config("https://api.example.com"))
+    with pytest.raises(DecisionModelRequestError, match="could not be sent"):
+        client.evaluate(state="state", questions=QUESTION)
+
+
+def test_a_transport_outage_is_a_request_error() -> None:
+    with fake_system_one() as server:
+        base_url = server.base_url
+    with pytest.raises(DecisionModelRequestError):
+        SystemOneClient(_config(base_url)).evaluate(state="state", questions=QUESTION)
+
+
+# -- the endpoint stops being called once it is known to be down -----------
+
+
+def test_repeated_failures_open_the_circuit_and_skip_the_endpoint() -> None:
+    with fake_system_one() as server:
+        extension = _extension(server.base_url, failure_threshold=2)
+        _schedule_outage(server, 2)
+        for _ in range(2):
+            with pytest.raises(DecisionModelRequestError):
+                _judge(extension)
+        assert len(server.calls) == 2
+
+        with pytest.raises(DecisionModelUnavailableError, match="marked down"):
+            _judge(extension)
+    assert len(server.calls) == 2
+
+
+def test_a_successful_probe_resumes_judgements() -> None:
+    with fake_system_one() as server:
+        extension = _extension(
+            server.base_url, failure_threshold=1, cooldown_seconds=0.05
+        )
+        _schedule_outage(server)
+        with pytest.raises(DecisionModelRequestError):
+            _judge(extension)
+        with pytest.raises(DecisionModelUnavailableError):
+            _judge(extension)
+
+        time.sleep(0.06)
+        assert _judge(extension) == pytest.approx(0.9)
+        assert _judge(extension) == pytest.approx(0.9)
+
+
+def test_a_failed_probe_keeps_the_circuit_open() -> None:
+    with fake_system_one() as server:
+        extension = _extension(
+            server.base_url, failure_threshold=1, cooldown_seconds=0.05
+        )
+        _schedule_outage(server)
+        with pytest.raises(DecisionModelRequestError):
+            _judge(extension)
+
+        time.sleep(0.06)
+        _schedule_outage(server)
+        with pytest.raises(DecisionModelRequestError):
+            _judge(extension)
+        with pytest.raises(DecisionModelUnavailableError):
+            _judge(extension)
+    # 三次失败里只有两次真的打了上游：冷却期内的那次被拦下了
+    assert len(server.calls) == 2
+
+
+def test_the_circuit_closes_after_isolated_failures() -> None:
+    with fake_system_one() as server:
+        extension = _extension(server.base_url, failure_threshold=2)
+        _schedule_outage(server)
+        with pytest.raises(DecisionModelRequestError):
+            _judge(extension)
+        assert _judge(extension) == pytest.approx(0.9)
+
+        _schedule_outage(server)
+        with pytest.raises(DecisionModelRequestError):
+            _judge(extension)
+        assert _judge(extension) == pytest.approx(0.9)
+
+
+def test_the_circuit_can_be_disabled() -> None:
+    with fake_system_one() as server:
+        extension = _extension(server.base_url, failure_threshold=0)
+        _schedule_outage(server, 3)
+        for _ in range(3):
+            with pytest.raises(DecisionModelRequestError):
+                _judge(extension)
+    assert len(server.calls) == 3
+
+
+# -- the timeout is a budget for the whole judgement -----------------------
+
+
+def test_one_judgement_stays_within_its_time_budget() -> None:
+    """Retries and backoff cannot outlive the configured budget."""
+    script = [(503, {"retry-after": "5"}, {"detail": "unavailable"})] * 4
+    with fake_system_one(script) as server:
+        client = SystemOneClient(_config(server.base_url, timeout=0.3, max_retries=3))
+        started = time.perf_counter()
+        with pytest.raises(DecisionModelRequestError, match="time budget"):
+            client.evaluate(state="state", questions=QUESTION)
+        elapsed = time.perf_counter() - started
+    assert elapsed < 0.6
+    assert len(server.calls) == 1
+
+
+# -- what ends up in the log ----------------------------------------------
+
+
+def test_retries_and_outages_are_logged(caplog: pytest.LogCaptureFixture) -> None:
+    script = [(503, {"retry-after": "0"}, {"detail": "unavailable"})] * 2
+    with fake_system_one(script) as server:
+        extension = _extension(server.base_url, max_retries=1, failure_threshold=1)
+        with caplog.at_level(logging.WARNING):
+            with pytest.raises(DecisionModelRequestError):
+                _judge(extension)
+
+    messages = [record.getMessage() for record in caplog.records]
+    assert any("retrying" in message for message in messages)
+    assert any("in a row" in message for message in messages)
+    assert all("test-key" not in message for message in messages)
+
+
+def test_the_judged_state_is_never_logged(caplog: pytest.LogCaptureFixture) -> None:
+    with fake_system_one() as server:
+        extension = _extension(server.base_url)
+        with caplog.at_level(logging.DEBUG):
+            extension.noul("SECRET-USER-TEXT", "Is this a greeting?")
+
+    assert caplog.records, "a successful judgement should be observable"
+    assert all(
+        "SECRET-USER-TEXT" not in record.getMessage() for record in caplog.records
+    )

veadk/extensions/decisions/README.md

@@ -28,10 +28,17 @@ DECISION_MODEL_PROVIDER=typesafe        # typesafe | openrouter | systemone
 DECISION_MODEL_NAME=jev-latest
 DECISION_MODEL_API_BASE=https://api.typesafe.ai
 DECISION_MODEL_API_KEY=...
-DECISION_MODEL_TIMEOUT=30
+DECISION_MODEL_TIMEOUT=5                 # seconds per judgement (ceiling: 5)
 DECISION_MODEL_MAX_RETRIES=3
+DECISION_MODEL_FAILURE_THRESHOLD=3       # 0 disables the circuit breaker
+DECISION_MODEL_COOLDOWN_SECONDS=30
 ```
 
+`DECISION_MODEL_TIMEOUT` is the budget of one whole judgement, retries and
+their backoff included. Values above 5 seconds are clamped to 5 with a
+warning: a judgement runs in the agent's hot path, so a slow endpoint has to
+degrade the judgement rather than the run.
+
 Every provider speaks the same System One protocol, so switching only changes
 the API base and the API key. The provider picks the default `api_base`:
 
@@ -103,6 +110,32 @@ The tool asks the configured decision model for one judgement and returns
 decision model is unconfigured or the request fails, so a run never breaks
 because of an optional capability.
 
+## Failures and Degradation
+
+The decision model is optional, so callers only ever handle one error type:
+`DecisionModelError`. Whatever goes wrong, a judgement degrades to the caller's
+own rules instead of breaking the run.
+
+| Failure | What happens |
+| --- | --- |
+| Not configured, or no API key | `DecisionModelDisabledError` on the first call; nothing else changes. |
+| Timeout, connection error, `429`, `5xx` | Retried with exponential backoff inside the `timeout` budget, honouring `retry-after`; then `DecisionModelRequestError`. |
+| Other `4xx` | Not retried; `DecisionModelRequestError` with the status and a body snippet. |
+| `200` with unusable answers | `DecisionModelResponseError`; malformed payloads and unknown answer types are reported the same way, never as a `pydantic` or `httpx` error. |
+| `DECISION_MODEL_TIMEOUT` above the ceiling | Clamped to 5 seconds with a warning. |
+| Unusable settings at startup | The extension disables itself with a warning instead of failing startup. |
+| `DECISION_MODEL_FAILURE_THRESHOLD` failures in a row (default 3) | The endpoint is marked down for `DECISION_MODEL_COOLDOWN_SECONDS` (default 30). Further judgements raise `DecisionModelUnavailableError` immediately and make no HTTP call; one probe request after the cooldown decides whether to resume. |
+
+Logging stays quiet and carries no user data:
+
+| Level | Message |
+| --- | --- |
+| `DEBUG` | One line per usable judgement: model, latency, tokens, cost. |
+| `INFO` | Cooldown elapsed and a probe was sent; judgements resumed. |
+| `WARNING` | A retry, an outage that marked the endpoint down, settings that were clamped or are unusable. |
+
+The judged state and the API key are never logged.
+
 ## Source Layout
 
 | Path | Purpose |