fix: a paused task says what stopped it and how to fix it; an unavailable connector hands the call to the next one

core/description.md

@@ -42,7 +42,7 @@ API with web sessions and operator tokens, and the task runner (design 05a). Tea
 - `routing.py` — `RoutingPolicy` from `routing.json`: dated prices, optional tier allow-list, data-class policy
 - `gateway.py` — `Gateway.estimate()` / `.call()`: data-class guard (the operator's responsibility stands in
   for a processing agreement, a region and no training; `[policy]` blocks countries and limits personal data to
-  regions), acknowledgement and automation rules from the ledger, quota cool-down, pins, cheapest connector, contract budget, cost record with `estimated_usd`;
+  regions), acknowledgement and automation rules from the ledger, quota cool-down, pins, cheapest connector (one that cannot start the call, `UNAVAILABLE` or `QUOTA_EXHAUSTED`, hands it to the next; a pin is never bypassed, and a failure that may have reached the provider is never tried elsewhere), contract budget, cost record with `estimated_usd`;
   passes the routed model to the connector in `ModelRequest.model`. `Gateway.estimate_decision()` /
   `.decide()`: the same rules for decision connectors; every choice is asked twice with reversed options,
   answers are checked before use, and when no decision connector can answer, the `economy` text tier answers the
@@ -101,7 +101,8 @@ API with web sessions and operator tokens, and the task runner (design 05a). Tea
 - `views.py` — the API's task, timeline, HIL and rating views with a reader's data-class cap (stubs above it);
   `workspace_view()`: a task's files changed, latest diff, revert state and whether it can be reverted
 - `runner_lock.py`, `runner.py` — the OS file lock beside the ledger; the runner thread with its FIFO queue rebuilt
-  from the ledger, a 60 s tick (defaults on silence, runnable tasks) and a growing pause for paused tasks
+  from the ledger, a 60 s tick (defaults on silence, runnable tasks) and a growing pause for paused tasks;
+  `provider_failure()` / `last_failure()` say what paused one (the task summary's `paused_by`, shown in the inbox)
 - `sessions.py` — one-time sign-in codes (hash-named files in `ooat-login/` beside the ledger) and web sessions
 - `api.py` — `create_app()`: the HTTP shell of `ooat serve` (Host, Origin, CSRF, 20 MB limit, sessions and tokens,
   typed errors, security headers, a log of method, path and status); `endpoints.py` — the REST resources
@@ -137,7 +138,7 @@ API with web sessions and operator tokens, and the task runner (design 05a). Tea
   for R2 and R3, and below θ after one more try on `workhorse`), one question (after two in a row it starts on its
   best reading), an answer to a task's open question or a follow-up task, a status, a cancel, a command it says it
   cannot run (offered as an open point), a project switch, up to 3 memory notes; a failed reception sends the
-  message as a task. `tap()` applies a button at once; `report()` writes a finished task's report with numbered
+  message as a task, and its reply names the provider failure. `tap()` applies a button at once; `report()` writes a finished task's report with numbered
   next steps; `close()` writes a summary into memory. `desk.py` is its thread in `ooat serve`: the quiet window
   (`batch_seconds`), reports when the runner closes a task, a 60 s tick for missed reports and silent conversations.
   θ comes from `thresholds.reception_threshold()`: `[reception] theta` until 5 verdicts (Yes, Differently, a

core/src/ooat_core/gateway.py

@@ -34,6 +34,7 @@
 # When no decision connector can answer, the cheapest text tier answers the same typed questions (spec §6).
 FALLBACK_TIER = "economy"
 FALLBACK_TIMEOUT_S = 120
+NEVER_RAN = ("UNAVAILABLE", "QUOTA_EXHAUSTED")  # the call did not reach the provider: the next connector may serve it
 
 
 class GatewayError(Exception):
@@ -158,9 +159,7 @@ def estimate(self, request: ModelRequest) -> Estimate:
     def call(self, request: ModelRequest) -> GatewayResult:
         if request.task is None and request.conversation is None:
             raise ValueError("every gateway call belongs to a task or a conversation")
-        candidate = self._route(request)
-        self._check_budget(request, candidate.estimate)
-        response = self._invoke(request, candidate, "complete")
+        candidate, response = self._invoke_first(request, self._routes(request), "complete")
         cost = self._cost(request, candidate, response)
         self._warn_budget(request, cost["usd"])
         return GatewayResult(response, cost, candidate.estimate)
@@ -210,9 +209,7 @@ def decide(self, request: DecisionRequest) -> DecisionResult:
         return dataclasses.replace(result, fallback_from=failure)
 
     def _decide_on_connector(self, request: DecisionRequest, expanded: DecisionRequest) -> DecisionResult:
-        candidate = self._route(expanded, kind="decision")
-        self._check_budget(expanded, candidate.estimate)
-        response = self._invoke(expanded, candidate, "decide")
+        candidate, response = self._invoke_first(expanded, self._routes(expanded, kind="decision"), "decide")
         cost = self._cost(expanded, candidate, response)
         try:
             answers = checked_answers(expanded.questions, response.answers)
@@ -277,9 +274,32 @@ def _invoke(self, request, candidate: "_Candidate", method: str):
                                cost=self._failure_cost(request, candidate, True)) from None
         return self._sanitised(response)
 
+    def _invoke_first(self, request, candidates: list["_Candidate"], method: str):
+        """(candidate, response) of the cheapest candidate that runs. One that could not start the call (unavailable,
+        out of quota) hands it to the next; any other failure may have reached the provider, or changed files, so
+        it is raised. When none could start, the cheapest one's code is raised, naming each failure."""
+        failed = []
+        for candidate in candidates:
+            self._check_budget(request, candidate.estimate)
+            try:
+                return candidate, self._invoke(request, candidate, method)
+            except GatewayError as error:
+                if error.code not in NEVER_RAN:
+                    raise
+                failed.append((candidate.connector.manifest["id"], error))
+        first = failed[0][1]
+        if len(failed) == 1:
+            raise first
+        raise GatewayError(first.code, "; ".join(f"{connector_id}: {error.message}" for connector_id, error in failed),
+                           cost=first.cost)
+
     # Routing ------------------------------------------------------------------------------------------------
 
     def _route(self, request: ModelRequest | DecisionRequest, kind: str = "model") -> _Candidate:
+        return self._routes(request, kind)[0]
+
+    def _routes(self, request: ModelRequest | DecisionRequest, kind: str = "model") -> list[_Candidate]:
+        """The candidates in the order they are tried: the cheapest first, or only the pinned one."""
         if request.data_class not in DATA_CLASSES:
             raise ValueError(f"unknown data class: {request.data_class}")
         candidates, trace, cooling = [], [], set()
@@ -314,16 +334,16 @@ def _route(self, request: ModelRequest | DecisionRequest, kind: str = "model") -
             if not pinned:
                 code = "QUOTA_EXHAUSTED" if pin in cooling else "NOT_PERMITTED"
                 raise GatewayError(code, f"pinned connector {pin} cannot serve this request", trace)
-            return pinned[0]
+            return pinned[:1]
         if not candidates and getattr(request, "workspace", None) is not None and all(
                 line.endswith(("cannot work with project folders", "not acknowledged by the operator (or disabled)"))
                 for line in trace if not line.startswith(("pre-scan", "project "))):
             raise GatewayError("NOT_PERMITTED", "no enabled connector can work with project folders", trace)
         if not candidates:
             code = "QUOTA_EXHAUSTED" if cooling else "NOT_PERMITTED"
             raise GatewayError(code, f"no connector can serve tier {request.tier} for {request.data_class}", trace)
-        return min(candidates, key=lambda c: (c.estimate.usd, _ACCESS_RANK[c.connector.manifest["access"]],
-                                              c.connector.manifest["id"]))
+        return sorted(candidates, key=lambda c: (c.estimate.usd, _ACCESS_RANK[c.connector.manifest["access"]],
+                                                 c.connector.manifest["id"]))
 
     def _exclusion(self, connector, request, acknowledgement, cooldowns) -> str | None:
         manifest, connector_id, data_class = connector.manifest, connector.manifest["id"], request.data_class

core/src/ooat_core/reception/__init__.py

@@ -68,6 +68,7 @@ def __init__(self, runtime: Runtime, store: memory.MemoryStore | None,
         self.projects = projects if projects is not None else (
             runtime.workspaces.config.projects if runtime.workspaces is not None else {})
         self._deciding = threading.Lock()
+        self._failed: str | None = None  # the last decision call's provider failure, under _deciding
 
     # Messages and taps ------------------------------------------------------------------------------------------
 
@@ -339,7 +340,7 @@ def decide(self, conversation: str) -> dict | None:
             marked = self._messages(waiting, marked=True)
             called = self._ask(base, context.text, marked, data_class)
             if called is None:
-                return self._fallback(found, waiting, base, said)
+                return self._fallback(found, waiting, base, said, self._failed)
             decision = called.data
             theta = self.theta(called.engine, called.model)
             if decision["decision"] in ("task", "continue_task") and decision["confidence"] < theta:
@@ -362,7 +363,8 @@ def decide(self, conversation: str) -> dict | None:
 
     def _ask(self, base: dict, context: str, said: str, data_class: str, tier: str = model.DECISION_TIER):
         """The decision call, once more after an unusable reply; None when the reception cannot decide. Every call
-        that was charged and not used is recorded with its cost."""
+        that was charged and not used is recorded with its cost. A provider failure is kept in `_failed`."""
+        self._failed = None
         for attempt in (1, 2):
             try:
                 return model.decide_call(self.gateway, tier=tier, context=context, messages=said,
@@ -374,6 +376,7 @@ def _ask(self, base: dict, context: str, said: str, data_class: str, tier: str =
                 if error.cost is not None and error.cost.get("usd"):
                     self._record(base | {"decision": "retried", "options": [], "buttons": []}, None, None,
                                  cost=error.cost)
+                self._failed = f"{error.code}: {error.message[:300]}"
                 return None
         return None
 
@@ -402,17 +405,19 @@ def _class(self, project: str) -> str:
         settings = self.projects.get(project)
         return settings.data_class if settings is not None else "internal"
 
-    def _fallback(self, found, waiting, base, said) -> dict:
-        """The reception could not decide: the message goes as a task, as the chat bridge sent it before 08."""
+    def _fallback(self, found, waiting, base, said, failed: str | None = None) -> dict:
+        """The reception could not decide: the message goes as a task, as the chat bridge sent it before 08. The
+        reply names the provider failure, so the person can tell what to fix."""
         goal, acceptance = split_done_when(said)
+        why = f" ({failed})" if failed else ""
         operator = waiting[-1]["actor"]["id"]
         task = self._start(found, {"goal": goal or said, "acceptance": acceptance}, operator,
                            waiting[-1]["body"]["channel"], f"reception:{waiting[0]['id']}", None,
                            files_from=waiting)
         return self._record(base | {
             "decision": "task", "task": task, "fallback": True, "options": [], "buttons": [],
-            "reply": say(said, "Recepce teď není dostupná, poslal jsem zprávu jako úkol.",
-                         "The reception is not available now; I sent the message as a task.")}, None, None)
+            "reply": say(said, f"Recepce teď není dostupná{why}, poslal jsem zprávu jako úkol.",
+                         f"The reception is not available now{why}; I sent the message as a task.")}, None, None)
 
     def _act(self, found, own, waiting, said, decision, theta, data_class) -> dict:
         """The design's rules on the model's decision, then the action; returns the body of the final decision."""

core/src/ooat_core/runner.py

@@ -25,17 +25,32 @@ def _utc(text: str) -> datetime:
     return datetime.fromisoformat(text.replace("Z", "+00:00"))
 
 
+def provider_failure(event: dict) -> str | None:
+    """"CODE: message" of an event that pauses its task until the provider is back, or None: a failed worker call,
+    a paused acceptance check or a Gate that stopped before deciding."""
+    body = event["body"]
+    if event["type"] == "RESULT" and body["outcome"] == "FAILED" and body["error"]["code"] != "INVALID_OUTPUT":
+        return f"{body['error']['code']}: {body['error']['message']}"
+    if event["type"] == "DECISION" and (
+            event["actor"]["id"] == "ooat-runtime" and body["decision"].startswith("acceptance check paused")
+            or event["actor"]["id"] == "ooat-gate" and body["decision"].startswith("Gate stopped before deciding")):
+        code = body["decision"].partition("(")[2].partition(")")[0]
+        return f"{code}: {body.get('rationale') or code}"
+    return None
+
+
+def last_failure(events: list[dict]) -> str | None:
+    """The provider failure that paused the task, while it is paused; None otherwise."""
+    if resume_after(events) is None:
+        return None
+    return next(failure for failure in map(provider_failure, reversed(events)) if failure is not None)
+
+
 def resume_after(events: list[dict]) -> datetime | None:
     """When a task paused by provider failures may be tried again; None when its last step did not fail."""
     failures, last = 0, None
     for event in events:
-        body = event["body"]
-        failed = (event["type"] == "RESULT" and body["outcome"] == "FAILED"
-                  and body["error"]["code"] != "INVALID_OUTPUT") or (
-            event["type"] == "DECISION" and event["actor"]["id"] == "ooat-runtime"
-            and body["decision"].startswith("acceptance check paused")) or (
-            event["type"] == "DECISION" and event["actor"]["id"] == "ooat-gate"
-            and body["decision"].startswith("Gate stopped before deciding"))
+        failed = provider_failure(event) is not None
         if failed:
             failures, last = failures + 1, event["ts"]
         elif event["type"] in ("RESULT", "GATE_PASSED", "GATE_FAILED", "HIL_RESPONSE"):

core/src/ooat_core/views.py

@@ -12,7 +12,7 @@
 from .gate import GateSettings, gated_data_class
 from .pii import ORDER, higher_class
 from .rating import CHOICE_OPTIONS, task_decisions
-from .runner import resume_after
+from .runner import last_failure, resume_after
 from .runtime import CLOSED
 from .state import task_state
 
@@ -67,6 +67,7 @@ def task_summary(task: str, events: list[dict], data_class: str, cap: str | None
         "submitted": submitted["ts"], "updated": events[-1]["ts"],
         "closed": closed["body"]["state"] if closed else None,
         "resume_after": paused.isoformat().replace("+00:00", "Z") if paused else None,
+        "paused_by": last_failure(events) if paused and visible else None,
     }
 
 

core/src/ooat_core/web/app/i18n/cs.js

@@ -21,6 +21,7 @@ export default {
   "inbox.waiting": "Čeká na vás ({n})",
   "inbox.stuck": "Uvízlé ({n})",
   "inbox.paused": "pozastaveno po výpadku poskytovatele; další pokus v {at}",
+  "inbox.paused_by": "Důvod: {reason}. Když jde o konektor, napravíte to na stránce Konektory.",
   "inbox.tasks": "Úlohy",
   "inbox.task": "Úloha",
   "inbox.state": "Stav",

core/src/ooat_core/web/app/i18n/en.js

@@ -20,6 +20,7 @@ export default {
   "inbox.waiting": "Waiting for you ({n})",
   "inbox.stuck": "Stuck ({n})",
   "inbox.paused": "paused after a provider failure; tries again at {at}",
+  "inbox.paused_by": "Reason: {reason}. When it is a connector, fix it on the Connectors page.",
   "inbox.tasks": "Tasks",
   "inbox.task": "Task",
   "inbox.state": "State",

core/src/ooat_core/web/app/views/inbox.js

@@ -30,7 +30,8 @@ export async function taskLists(ctx, form) {
       ...ratings.tasks.map((/** @type {any} */ entry) => ratingCard(entry, ctx, {compact: true, onDone: again})),
       h("h2", {}, t("inbox.stuck", {n: stuck.length})),
       ...stuck.map((/** @type {any} */ task) => h("p", {}, "⏸ ", link(task), " ",
-        t("inbox.paused", {at: clock(task.resume_after)}))),
+        t("inbox.paused", {at: clock(task.resume_after)}),
+        ...(task.paused_by ? [h("br"), t("inbox.paused_by", {reason: task.paused_by})] : []))),
       h("h2", {}, t("inbox.tasks")), table(tasks.tasks));
   }
 

core/tests/test_gateway.py

@@ -276,12 +276,53 @@ def test_exhausted_quota_cools_down_the_connector_and_routing_moves_on():
     setup = Setup(subscription, FakeConnector(API))
     for connector in setup.connectors:
         setup.acknowledge(connector)
-    with pytest.raises(GatewayError) as info:
-        setup.gateway.call(setup.request())
-    assert info.value.code == "QUOTA_EXHAUSTED" and info.value.cost["usd"] == 0.0
+    assert setup.gateway.call(setup.request()).cost["adapter"] == "prv.fake.api"  # the same call moves on
     warning = setup.ledger.events(types=["QUOTA_WARNING"])[0]["body"]
     assert warning["window_resets_at"] == "2026-10-01T17:00:00Z"  # plan quota window of 5 hours
     assert setup.gateway.call(setup.request()).cost["adapter"] == "prv.fake.api"
+    assert len(subscription.calls) == 1  # cooling down: not tried again
+
+
+def test_an_unavailable_connector_hands_the_call_to_the_next_one():
+    hooks = FakeConnector(SUBSCRIPTION, error=ConnectorError("UNAVAILABLE", "hooks not approved"))
+    setup = Setup(hooks, FakeConnector(API))
+    for connector in setup.connectors:
+        setup.acknowledge(connector)
+    assert setup.gateway.call(setup.request()).cost["adapter"] == "prv.fake.api"
+    assert len(hooks.calls) == 1
+
+
+def test_every_connector_unavailable_names_each_one():
+    setup = Setup(FakeConnector(SUBSCRIPTION, error=ConnectorError("UNAVAILABLE", "hooks not approved")),
+                  FakeConnector(API, error=quota_error()))
+    for connector in setup.connectors:
+        setup.acknowledge(connector)
+    with pytest.raises(GatewayError) as info:
+        setup.gateway.call(setup.request())
+    assert info.value.code == "UNAVAILABLE" and info.value.cost["usd"] == 0.0
+    assert info.value.message == ("prv.fake.subscription_cli: hooks not approved; "
+                                  "prv.fake.api: usage limit reached")
+
+
+def test_a_failure_that_may_have_reached_the_provider_is_not_tried_elsewhere():
+    timed_out = FakeConnector(SUBSCRIPTION, error=ConnectorError("TIMEOUT", "no reply in 600 s"))
+    api = FakeConnector(API)
+    setup = Setup(timed_out, api)
+    for connector in setup.connectors:
+        setup.acknowledge(connector)
+    with pytest.raises(GatewayError, match="TIMEOUT"):
+        setup.gateway.call(setup.request())
+    assert api.calls == []  # it may have run, or changed files: never twice
+
+
+def test_a_pinned_connector_that_is_unavailable_is_not_bypassed():
+    setup = Setup(FakeConnector(SUBSCRIPTION, error=ConnectorError("UNAVAILABLE", "logged out")),
+                  FakeConnector(API), pins={"workhorse": "prv.fake.subscription_cli"})
+    for connector in setup.connectors:
+        setup.acknowledge(connector)
+    with pytest.raises(GatewayError, match="UNAVAILABLE"):
+        setup.gateway.call(setup.request())
+    assert setup.connectors[1].calls == []
 
 
 def test_cool_down_survives_a_restart_and_ends_at_reset():