diff --git a/core/description.md b/core/description.md index 3ae991b..7895d25 100644 --- a/core/description.md +++ b/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 diff --git a/core/src/ooat_core/gateway.py b/core/src/ooat_core/gateway.py index fc673b9..099e122 100644 --- a/core/src/ooat_core/gateway.py +++ b/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,7 +334,7 @@ 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 "))): @@ -322,8 +342,8 @@ def _route(self, request: ModelRequest | DecisionRequest, kind: str = "model") - 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 diff --git a/core/src/ooat_core/reception/__init__.py b/core/src/ooat_core/reception/__init__.py index 5654619..8be47ef 100644 --- a/core/src/ooat_core/reception/__init__.py +++ b/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.""" diff --git a/core/src/ooat_core/runner.py b/core/src/ooat_core/runner.py index 1e94090..473442a 100644 --- a/core/src/ooat_core/runner.py +++ b/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"): diff --git a/core/src/ooat_core/views.py b/core/src/ooat_core/views.py index c453872..3b0263c 100644 --- a/core/src/ooat_core/views.py +++ b/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, } diff --git a/core/src/ooat_core/web/app/i18n/cs.js b/core/src/ooat_core/web/app/i18n/cs.js index 8c1c1c2..f24e3b6 100644 --- a/core/src/ooat_core/web/app/i18n/cs.js +++ b/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", diff --git a/core/src/ooat_core/web/app/i18n/en.js b/core/src/ooat_core/web/app/i18n/en.js index 2d23f32..10fd2eb 100644 --- a/core/src/ooat_core/web/app/i18n/en.js +++ b/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", diff --git a/core/src/ooat_core/web/app/views/inbox.js b/core/src/ooat_core/web/app/views/inbox.js index 37e038a..a23f499 100644 --- a/core/src/ooat_core/web/app/views/inbox.js +++ b/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)); } diff --git a/core/tests/test_gateway.py b/core/tests/test_gateway.py index 9a698b2..4e95390 100644 --- a/core/tests/test_gateway.py +++ b/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(): diff --git a/core/tests/test_reception.py b/core/tests/test_reception.py index 6c8e3ec..08a08d9 100644 --- a/core/tests/test_reception.py +++ b/core/tests/test_reception.py @@ -212,7 +212,8 @@ def test_an_unknown_decision_is_unusable(tmp_path): def test_a_provider_failure_sends_the_message_as_a_task(tmp_path): desk = Desk(tmp_path, [ConnectorError("UNAVAILABLE", "down")]) body = desk.ask("Change the title.\nDone when: the title reads Family tests")["body"] - assert body["fallback"] is True and body["reply"].startswith("The reception is not available") + assert body["fallback"] is True + assert body["reply"] == "The reception is not available now (UNAVAILABLE: down); I sent the message as a task." assert desk.submitted()[0]["acceptance"] == ["the title reads Family tests"] diff --git a/core/tests/test_views.py b/core/tests/test_views.py index e8878d9..4554632 100644 --- a/core/tests/test_views.py +++ b/core/tests/test_views.py @@ -93,9 +93,12 @@ def test_a_task_paused_by_a_provider_failure_says_when_it_is_tried_again(tmp_pat tasks, found = classes(setup) summary = task_summary(task, tasks[task], found[task], None) assert summary["state"] == "RUNNING" and summary["resume_after"].endswith("Z") + assert summary["paused_by"] == "UNAVAILABLE: HTTP 529" + assert task_summary(task, tasks[task], found[task], "public")["paused_by"] is None # above the reader's cap setup.runtime.run(task) # the provider is back tasks, found = classes(setup) - assert task_summary(task, tasks[task], found[task], None)["resume_after"] is None + summary = task_summary(task, tasks[task], found[task], None) + assert summary["resume_after"] is None and summary["paused_by"] is None def test_an_attachment_is_listed_with_its_file_name_within_the_cap(tmp_path): diff --git a/docs/chat-bridge.md b/docs/chat-bridge.md index 746a1f8..5260b70 100644 --- a/docs/chat-bridge.md +++ b/docs/chat-bridge.md @@ -20,7 +20,7 @@ token at once, also for a timeline the bot is following. | Chat action | `Bridge` | API | |---|---|---| | A message (with files) in a project's chat | `on_message(text, message_id, files)` → task id | `POST /tasks` with the chat's project; the text before a `Done when:` (or `Hotovo když:`) line is the `goal`, each line after it a criterion; files as attachments; `Idempotency-Key: :`, so a repeated delivery returns the first task | -| Progress | `watch(task)` | follows `GET /tasks/{id}/events` (Server-Sent Events) and posts one line for the Gate's topology, each attempt and each answer | +| Progress | `watch(task)` | follows `GET /tasks/{id}/events` (Server-Sent Events) and posts one line for the Gate's topology, each attempt and each answer; when a provider failure pauses the task (a failed call, a paused acceptance check, a Gate stopped before deciding), once per pause: what stopped it, the reason as the server gives it, for an unavailable connector where to fix it (the Connectors page or `ooat connectors`), and that OOAT tries again by itself | | A question | posted by `watch`; `on_button(data)`, `on_reply(message_id, text)` | `GET /hil`: one button per option (★ the recommended one, its cost when it has one), the default on silence and the deadline (UTC) in the text; a button calls `POST /hil/{evt}/answer`; the options answered with text (clarify, narrow the scope) ask for a reply, and a reply to the question answers with its text; an R3 question has no buttons | | Result | posted by `watch` | the final document as text in up to 3 messages (`limit`, default 4096 characters), else as a file `.md` (`` when it is not text); a footer: state, topology, cost, tokens in/out, time | | Project files | posted by `watch` | for a task on a project's folders (ADR 0017): the diff as `.diff` after the document; a "Revert the changes" row in the footer while a revert is possible; a tap asks "Yes, revert", and only that second tap calls `POST /tasks/{id}/revert`; skipped files are listed with their reason. A limit question is preceded by the task's latest document and its diff, and its `revert` button asks for a second tap too. Nothing of this is posted above the token's data class | diff --git a/docs/description.md b/docs/description.md index 4796a81..cb6ceee 100644 --- a/docs/description.md +++ b/docs/description.md @@ -13,7 +13,7 @@ with model and decision connectors, the Topology Gate for T0–T2 and the task r `ooat.toml` and those already used, from `GET /projects`; the last one is remembered). The conversation with the reception runs below it: replies with numbered options and buttons, a link to each task it started, and the task's report. Then the project's earlier conversations, a folded form for a task with exact criteria, and the inbox's lists -(questions and ratings waiting, tasks paused by a provider failure, all tasks). A conversation has its own page, +(questions and ratings waiting, tasks paused by a provider failure with its reason, all tasks). A conversation has its own page, read only unless it is open on the web. Memory lists a project's items and the operator's preferences, with add (by hand, `POST /memory/{scope}`, web session only), edit, pin, done and forget. The app also has a task page (live timeline, the document of every attempt rendered, as source or as a diff, diff --git a/sdk/python/src/ooat_client/bridge.py b/sdk/python/src/ooat_client/bridge.py index 4357db1..5a9e014 100644 --- a/sdk/python/src/ooat_client/bridge.py +++ b/sdk/python/src/ooat_client/bridge.py @@ -7,6 +7,7 @@ """ import http.client +import re import threading from collections.abc import Callable, Sequence from dataclasses import dataclass @@ -23,6 +24,7 @@ CODES = {tap: code for code, tap in TAPS.items()} REPLACED = frozenset({"escalated", "retried"}) # decisions the next call replaced: never posted OWN_KEPT = 500 # how many own message and task ids a bridge keeps (and a bot stores across restarts) +_PAUSED = re.compile(r"(?:acceptance check paused|Gate stopped before deciding) \((\w+)\)") @dataclass(frozen=True) @@ -79,6 +81,16 @@ def _trim(kept: dict) -> None: del kept[key] +def stopped_by(event: dict) -> tuple[str, str] | None: + """(code, reason) of an event that pauses a task until its provider is back (ADR 0014), 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.get("outcome") == "FAILED" and body["error"]["code"] != "INVALID_OUTPUT": + return body["error"]["code"], body["error"]["message"] + paused = _PAUSED.match(body.get("decision") or "") if event["type"] == "DECISION" else None + return (paused.group(1), body.get("rationale") or paused.group(1)) if paused else None + + def _utc(ts: str) -> datetime: return datetime.fromisoformat(ts.replace("Z", "+00:00")) @@ -108,6 +120,7 @@ def __init__(self, client: OoatClient, chat: Chat, project: str, *, language: st self._texts: dict[str, str] = {} # own message id → its text, for a handoff self._said: dict[str, str] = {} # a command decision → the text a handoff passes on self._scope: dict[str, str] = {} # memory item → its scope, from the post that showed its button + self._told_paused: dict[str, None] = {} # tasks whose pause was told: once until they move again def _t(self, name: str, **values) -> str: return self.texts[name].format(**values) @@ -443,6 +456,12 @@ def _post(self, task: str, event: dict, attempts: int) -> None: kind, body = event["type"], event["body"] if "redacted" in body and kind not in ("HIL_REQUEST", "TASK_CLOSED"): return # above the token's data class: only the question and the result are posted + stopped = stopped_by(event) + if stopped is not None: + self._paused(task, *stopped) + else: + with self._guard: + self._told_paused.pop(task, None) if kind == "TOPOLOGY_DECIDED": self._send(self._t("gate", topology=body["topology"])) elif kind == "CONTRACT_ISSUED": @@ -455,6 +474,19 @@ def _post(self, task: str, event: dict, attempts: int) -> None: elif kind == "TASK_CLOSED": self._result(task, event) + def _paused(self, task: str, code: str, reason: str) -> None: + """What stopped the task, why, how to fix it and what OOAT does meanwhile: once per pause, not per retry.""" + with self._guard: + if task in self._told_paused: + return + self._told_paused[task] = None + _trim(self._told_paused) + why = self.texts.get(f"stop_{code}", self.texts["stop_other"]) + lines = [self._t("stopped", why=why), self._t("stop_reason", reason=self._shown(reason))] + if code == "UNAVAILABLE": + lines.append(self._t("stop_fix_connector")) + self._send("\n".join([*lines, self._t("stop_next")])) + def _question(self, request_id: str) -> None: request = next((r for r in self.client.questions() if r["request"] == request_id), None) if request is None: diff --git a/sdk/python/src/ooat_client/texts.py b/sdk/python/src/ooat_client/texts.py index 678fa00..85b0424 100644 --- a/sdk/python/src/ooat_client/texts.py +++ b/sdk/python/src/ooat_client/texts.py @@ -5,6 +5,16 @@ "en": { "gate": "Gate: {topology}", "working": "Working on it (attempt {n}).", + "stopped": "⏸ The task is stopped: {why}", + "stop_UNAVAILABLE": "the connector cannot be used now.", + "stop_QUOTA_EXHAUSTED": "the subscription's quota is used up.", + "stop_TIMEOUT": "the provider did not answer in time.", + "stop_API_ERROR": "the provider returned an error.", + "stop_other": "the provider failed.", + "stop_reason": "Reason: {reason}", + "stop_fix_connector": "Fix it on the Connectors page of the web app or with `ooat connectors`.", + "stop_next": ("I try again by myself in a minute, then less often (at most every 30 minutes); once the cause " + "is fixed, it runs on. If you no longer want it, cancel the task."), "answered": "Answered: {answer}", "defaulted": "No answer in time: {answer} applies.", "default": "If nobody answers by {deadline}: {option}.", @@ -64,6 +74,16 @@ "cs": { "gate": "Gate: {topology}", "working": "Pracuji na tom (pokus {n}).", + "stopped": "⏸ Úkol stojí: {why}", + "stop_UNAVAILABLE": "konektor teď nejde použít.", + "stop_QUOTA_EXHAUSTED": "předplatné vyčerpalo kvótu.", + "stop_TIMEOUT": "poskytovatel neodpověděl včas.", + "stop_API_ERROR": "poskytovatel vrátil chybu.", + "stop_other": "poskytovatel selhal.", + "stop_reason": "Důvod: {reason}", + "stop_fix_connector": "Napravíte to ve webové aplikaci na stránce Konektory nebo příkazem `ooat connectors`.", + "stop_next": ("Za minutu to zkusím znovu sám, pak v delších odstupech (nejvýš po 30 minutách); jakmile bude " + "příčina napravená, úkol poběží dál. Pokud ho už nechcete, zrušte ho."), "answered": "Odpověď: {answer}", "defaulted": "Bez odpovědi včas: platí {answer}.", "default": "Když nikdo neodpoví do {deadline}: {option}.", diff --git a/sdk/python/tests/test_ooat_bridge.py b/sdk/python/tests/test_ooat_bridge.py index c966fd6..be2d232 100644 --- a/sdk/python/tests/test_ooat_bridge.py +++ b/sdk/python/tests/test_ooat_bridge.py @@ -427,3 +427,50 @@ def post(n): for thread in threads: thread.join() assert len(bridge._sent) <= 500 and len(bridge._posted) <= 500 and len(bridge._scope) <= 500 + + +def task_event(evt, kind, actor="ooat-runtime", **body): + return {"id": evt, "ts": "2026-10-19T19:13:06Z", "type": kind, "actor": {"kind": "system", "id": actor}, + "refs": [], "cost": None, "body": body} + + +def failed(evt, code="UNAVAILABLE", message="hooks.json defines hooks the operator has not approved"): + return task_event(evt, "RESULT", outcome="FAILED", error={"code": code, "message": message}) + + +def test_a_paused_task_says_once_what_stopped_it_and_what_happens_next(): + chat = Kept() + bridge = Bridge(OoatClient("http://127.0.0.1:9", "secret"), chat, "docs", language="cs") + for event in (failed("e1"), failed("e2"), task_event("e3", "RESULT", outcome="DONE"), failed("e4", "TIMEOUT")): + bridge._post("tsk_1", event, 1) + first, again = (text for text, _ in chat.sent) + assert first.startswith("⏸ Úkol stojí: konektor teď nejde použít.") + assert "Důvod: hooks.json defines hooks the operator has not approved" in first + assert "Konektory" in first and "zkusím znovu sám" in first + assert again.startswith("⏸ Úkol stojí: poskytovatel neodpověděl včas.") and "Konektory" not in again + + +def test_an_invalid_output_is_a_new_attempt_not_a_pause(): + chat = Kept() + Bridge(OoatClient("http://127.0.0.1:9", "secret"), chat, "docs")._post("tsk_1", failed("e1", "INVALID_OUTPUT"), 1) + assert chat.sent == [] + + +def test_a_paused_acceptance_check_or_gate_is_told_too(): + chat = Kept() + bridge = Bridge(OoatClient("http://127.0.0.1:9", "secret"), chat, "docs") + bridge._post("tsk_1", task_event("e1", "DECISION", rationale="usage limit reached", decision=( + "acceptance check paused (QUOTA_EXHAUSTED); the document is checked again later")), 1) + bridge._post("tsk_2", task_event("e2", "DECISION", actor="ooat-gate", rationale="HTTP 529", decision=( + "Gate stopped before deciding (API_ERROR); it runs again later.")), 1) + bridge._post("tsk_3", task_event("e3", "DECISION", decision="Worker attempt 1 accepted."), 1) + assert [text.split("\n")[:2] for text, _ in chat.sent] == [ + ["⏸ The task is stopped: the subscription's quota is used up.", "Reason: usage limit reached"], + ["⏸ The task is stopped: the provider returned an error.", "Reason: HTTP 529"]] + + +def test_a_pause_above_the_cap_is_told_without_its_reason(): + chat = Kept() + hidden = task_event("e1", "RESULT", redacted="Above this token's data class.", link="/tasks/tsk_1") + Bridge(OoatClient("http://127.0.0.1:9", "secret"), chat, "docs")._post("tsk_1", hidden, 1) + assert chat.sent == [] # a redacted RESULT may be any outcome: only its question and result are posted