diff --git a/CHANGELOG.md b/CHANGELOG.md index dca3f79..94116a6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- `tokenops.control.ledger_backend` — the `LedgerBackend` protocol and + `HttpLedgerBackend` for the remote-only rewrite (#118). One interface for every + control-plane read/write; `apply_events(list[LedgerEvent])` is the only write path + (a future buffered backend wraps it). Targets `agentplane-control-plane` 0.2.0 + (`precheck` / `events:batch`). Not yet wired into `Ledger` / `ControlPlaneClient`. +- `tests/fakes.py::FakeLedgerBackend` (in-memory) + `tests/test_ledger_backend_contract.py` + — parametrised over the fake and a real `control_plane.app` (in-process ASGI) so the + fake can't drift from the plane. +- `[contract]` optional-dependency group (`agentplane-control-plane>=0.2.0`). Kept out + of `[dev]` while the 0.2.0 line is unreleased; the plane-backed tests + `importorskip("control_plane")`, so `[dev]`-only CI stays green. + ## [0.2.1] - 2026-09-04 ### Changed diff --git a/pyproject.toml b/pyproject.toml index 3526421..bb5c831 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -56,6 +56,14 @@ dev = [ "pre-commit>=4.0", "types-PyYAML>=6.0", ] +# The verified-fake contract suite + e2e run against a real control plane in-process. +# Not in `dev` because the 0.2.0 line is not on PyPI yet — the tests that need it +# `pytest.importorskip("control_plane")`, so `[dev]` alone stays green. To run them: +# pip install -e ".[dev,contract]" # once 0.2.0 is released, or +# pip install "agentplane-control-plane @ git+https://github.com/theagentplane/control-plane@main" +contract = [ + "agentplane-control-plane>=0.2.0", +] examples = [ "ddgs>=9.0", "langchain-core>=0.3", diff --git a/src/tokenops/control/__init__.py b/src/tokenops/control/__init__.py index 40c27a0..4e2fe61 100644 --- a/src/tokenops/control/__init__.py +++ b/src/tokenops/control/__init__.py @@ -63,6 +63,14 @@ wrap_stream, ) from tokenops.control.ledger import Budget, Ledger, RunState, segment_key +from tokenops.control.ledger_backend import ( + AggregateState, + ApplyResult, + HttpLedgerBackend, + LedgerBackend, + LedgerEvent, + PrecheckRequest, +) # Process-wide: re-attach after every reset_session (Chronicle clears on_crossing). install_crossing_hook() @@ -106,6 +114,13 @@ "Ledger", "RunState", "segment_key", + # ledger backend (remote-only) + "LedgerBackend", + "HttpLedgerBackend", + "LedgerEvent", + "PrecheckRequest", + "AggregateState", + "ApplyResult", # harness "Governor", "RaiseControls", diff --git a/src/tokenops/control/ledger_backend.py b/src/tokenops/control/ledger_backend.py new file mode 100644 index 0000000..664f257 --- /dev/null +++ b/src/tokenops/control/ledger_backend.py @@ -0,0 +1,277 @@ +"""LedgerBackend — the single interface the SDK uses to reach the control plane. + +The remote-only rewrite (see the ``remote-only control plane`` epic) routes every +ledger read/write through this protocol. Two implementations: + +* :class:`HttpLedgerBackend` — direct ``httpx`` to a running plane. Production. +* ``FakeLedgerBackend`` (``tests/fakes.py``) — in-memory, tests only. Kept honest by + ``tests/test_ledger_backend_contract.py`` (parametrised against the real plane over + ``httpx.ASGITransport``). + +**The only write path is :meth:`apply_events`** — a list of :class:`LedgerEvent`. A +future ``BufferedLedgerBackend`` wraps an inner backend and coalesces those calls +without any call site changing. + +Wire contract: ``agentplane-control-plane`` ``docs/api-contract.md`` (0.2.0). +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from typing import Any, Protocol, TypedDict, runtime_checkable + +import httpx + +from tokenops.control.models import ( + GovernanceMode, + RunAlreadyRegisteredError, + RunNotRegisteredError, + RunRegistration, + parse_governance_mode, +) + +# --------------------------------------------------------------------------- # +# Event + result shapes (JSON-aligned with the wire contract) # +# --------------------------------------------------------------------------- # + + +class LedgerEvent(TypedDict, total=False): + """One ledger mutation. ``kind`` + ``idempotency_key`` are always required. + + kinds: ``spent_add`` · ``admit`` · ``complete`` · ``step`` · ``halt_mark`` · + ``halt_clear``. See contract §5. + """ + + kind: str + idempotency_key: str + ts: float + run_id: str + # spent_add + delta_micros: int + targets: list[dict[str, Any]] # [{budget_id, segment_key, period}] + # admit / complete + segment_key: str + # step + agent: str + seq: int + node_type: str + boundary_id: str + cost_micros: int + cum_spent_micros: int + usage: dict[str, int] + tags: dict[str, str] + tool_signature: str + result_hash: str + # halt_mark + reason: str + detector: str + + +@dataclass(kw_only=True) +class PrecheckRequest: + run_id: str + segment_keys: list[str] = field(default_factory=list) + budgets: list[dict[str, str]] = field( + default_factory=list + ) # [{budget_id, segment_key, period}] + want: list[str] = field(default_factory=lambda: ["spent", "inflight", "halt"]) + + +@dataclass +class AggregateState: + server_ts: float = 0.0 + halted: bool = False + halt_reason: str | None = None + spent: dict[str, int] = field( + default_factory=dict + ) # "||" -> micros + inflight: dict[str, int] = field(default_factory=dict) # segment_key -> count + window: dict[str, Any] | None = None # {step_count, recent, velocity_micros_per_step} + + +@dataclass +class ApplyResult: + accepted: int = 0 + deduped: int = 0 + totals: dict[str, int] = field(default_factory=dict) + halted: bool = False + + +# --------------------------------------------------------------------------- # +# Protocol # +# --------------------------------------------------------------------------- # + + +@runtime_checkable +class LedgerBackend(Protocol): + """Everything the SDK needs from the control plane.""" + + def read_state(self, req: PrecheckRequest) -> AggregateState: ... + + def apply_events( + self, events: list[LedgerEvent], *, durability: str = "sync" + ) -> ApplyResult: ... + + def register_run( + self, + *, + intent: str = "", + user_dims: dict[str, str] | None = None, + mode: GovernanceMode | str | None = None, + run_id: str | None = None, + ) -> RunRegistration: ... + + def resolve_run(self, run_id: str) -> RunRegistration: ... + + def governance_config_for(self, agent: str) -> dict[str, Any]: ... + + def patch_run_record(self, run_id: str, **fields: Any) -> None: ... + + def close(self) -> None: ... + + +def _mode_value(mode: GovernanceMode | str | None) -> GovernanceMode: + if mode is None or mode == "": + return GovernanceMode.ENFORCE + if isinstance(mode, GovernanceMode): + return mode + return parse_governance_mode(mode) + + +def _reg_from_body(body: dict[str, Any], *, fallback_intent: str = "") -> RunRegistration: + return RunRegistration( + run_id=str(body["run_id"]), + intent=str(body.get("intent") or fallback_intent), + user_dims={str(k): str(v) for k, v in (body.get("user_dims") or {}).items()}, + mode=parse_governance_mode(body.get("mode")), + ) + + +# --------------------------------------------------------------------------- # +# HTTP implementation # +# --------------------------------------------------------------------------- # + + +class HttpLedgerBackend: + """:class:`LedgerBackend` over the control-plane HTTP API.""" + + def __init__( + self, + base_url: str, + *, + api_key: str | None = None, + client: httpx.Client | None = None, + timeout: float = 30.0, + ) -> None: + self._base_url = base_url.rstrip("/") + self._owns_client = client is None + key = ( + api_key or os.environ.get("CONTROL_PLANE_API_KEY") or os.environ.get("TOKENOPS_API_KEY") + ) + headers = {"Authorization": f"Bearer {key}"} if key else {} + self._client = client or httpx.Client( + base_url=self._base_url, timeout=timeout, headers=headers + ) + + # ---- lifecycle ---------------------------------------------------------- + + def close(self) -> None: + if self._owns_client: + self._client.close() + + def _req(self, method: str, path: str, **kw: Any) -> httpx.Response: + r = self._client.request(method, path, **kw) + if r.status_code == 409: + raise RunAlreadyRegisteredError(_err(r, "run already registered")) + r.raise_for_status() + return r + + # ---- reads ------------------------------------------------------------- # + + def read_state(self, req: PrecheckRequest) -> AggregateState: + r = self._req( + "POST", + "/v1/ledger/precheck", + json={ + "run_id": req.run_id, + "segment_keys": req.segment_keys, + "budgets": req.budgets, + "want": req.want, + }, + ) + b = r.json() + return AggregateState( + server_ts=float(b.get("server_ts") or 0.0), + halted=bool(b.get("halted")), + halt_reason=b.get("halt_reason"), + spent={str(k): int(v) for k, v in (b.get("spent") or {}).items()}, + inflight={str(k): int(v) for k, v in (b.get("inflight") or {}).items()}, + window=b.get("window"), + ) + + def resolve_run(self, run_id: str) -> RunRegistration: + r = self._client.get(f"/v1/runs/{run_id}/registration") + if r.status_code == 404: + raise RunNotRegisteredError(f"run {run_id!r} is not registered") + r.raise_for_status() + return _reg_from_body(r.json()) + + def governance_config_for(self, agent: str) -> dict[str, Any]: + return self._req("GET", f"/v1/governance/{agent}").json() + + # ---- writes ---------------------------------------------------------- # + + def apply_events(self, events: list[LedgerEvent], *, durability: str = "sync") -> ApplyResult: + # TODO(buffering): a BufferedLedgerBackend wraps this backend and coalesces + # apply_events into a background flush so a governed LLM call costs 1 sync + # round trip instead of 2. See the remote-only plan Part 7 / Phase 4. + if not events: + return ApplyResult() + r = self._req( + "POST", + "/v1/ledger/events:batch", + json={"events": events}, + headers={"Durability": durability}, + ) + b = r.json() + return ApplyResult( + accepted=int(b.get("accepted") or 0), + deduped=int(b.get("deduped") or 0), + totals={str(k): int(v) for k, v in (b.get("totals") or {}).items()}, + halted=bool(b.get("halted")), + ) + + def register_run( + self, + *, + intent: str = "", + user_dims: dict[str, str] | None = None, + mode: GovernanceMode | str | None = None, + run_id: str | None = None, + ) -> RunRegistration: + payload: dict[str, Any] = { + "intent": intent, + "user_dims": {str(k): str(v) for k, v in (user_dims or {}).items()}, + "mode": _mode_value(mode).value, + } + if run_id: + payload["run_id"] = run_id + r = self._req("POST", "/v1/runs", json=payload) + return _reg_from_body(r.json(), fallback_intent=intent) + + def patch_run_record(self, run_id: str, **fields: Any) -> None: + # steps / cost_micros are derived server-side; the plane ignores them (0.2.x) + # and rejects them (0.3.0). Don't send them. + clean = {k: v for k, v in fields.items() if k not in ("steps", "cost_micros")} + if not clean: + return + self._req("PATCH", f"/v1/run-records/{run_id}", json=clean) + + +def _err(r: httpx.Response, default: str) -> str: + try: + body = r.json() + return str(body.get("error") or body.get("detail") or default) + except Exception: + return default diff --git a/tests/conftest.py b/tests/conftest.py index 8d07c0a..74bf8e5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -99,3 +99,101 @@ def apply(self, action: Action) -> None: def of_kind(self, kind: ActionKind): return [a for a in self.actions if a.kind is kind] + + +# --------------------------------------------------------------------------- # +# LedgerBackend fixtures (remote-only rewrite) # +# --------------------------------------------------------------------------- # + +import gc +import shutil +import tempfile +from pathlib import Path + +import pytest + + +@pytest.fixture +def fake_backend(): + """In-memory FakeLedgerBackend — the unit-test backend.""" + from fakes import FakeLedgerBackend + + return FakeLedgerBackend() + + +@pytest.fixture +def plane_app_factory(): + """Build a real control_plane.app over a throwaway SQLite file. + + Skips the whole test when ``agentplane-control-plane`` (>= 0.2.0) is not installed + — it is not a hard dev dep (not on PyPI yet); ``pip install -e ".[dev,contract]"`` + to run these. Yields a callable; every app it makes is torn down (connections + closed before the temp dir is removed — open SQLite handles block unlink on Windows). + """ + pytest.importorskip("control_plane", reason="install agentplane-control-plane>=0.2.0") + made: list[tuple] = [] + + def _make(**settings_kw): + from control_plane.app import create_app + from control_plane.envelope_store import EnvelopeStore + from control_plane.settings import Settings + from control_plane.store import SqliteStore + + td = tempfile.mkdtemp() + db = str(Path(td) / "cp.db") + store = SqliteStore(db, auto_seed=False) + env = EnvelopeStore(db) + app = create_app(store=store, envelopes=env, settings=Settings(db_path=db, **settings_kw)) + made.append((store, env, td)) + return app + + yield _make + + for store, env, td in made: + store.close() + env.close() + gc.collect() + shutil.rmtree(td, ignore_errors=True) + + +def _asgi_backend(app): + """HttpLedgerBackend over an in-process control_plane.app. + + ``httpx.ASGITransport`` is async-only, so we drive the app through FastAPI's + ``TestClient`` (a sync ``httpx.Client`` subclass backed by a portal). + """ + from fastapi.testclient import TestClient + + from tokenops.control.ledger_backend import HttpLedgerBackend + + client = TestClient(app) + return HttpLedgerBackend("http://testserver", client=client), client + + +@pytest.fixture +def http_backend(plane_app_factory): + backend, client = _asgi_backend(plane_app_factory()) + try: + yield backend + finally: + client.close() + + +@pytest.fixture(params=["fake", "http"]) +def any_backend(request): + """Parametrised over both backends — for the verified-fake contract suite. + + The ``fake`` param always runs; the ``http`` param resolves ``plane_app_factory`` + lazily, so it skips (not errors) when the real plane is not installed. + """ + if request.param == "fake": + from fakes import FakeLedgerBackend + + yield FakeLedgerBackend() + return + factory = request.getfixturevalue("plane_app_factory") + backend, client = _asgi_backend(factory()) + try: + yield backend + finally: + client.close() diff --git a/tests/fakes.py b/tests/fakes.py new file mode 100644 index 0000000..c734917 --- /dev/null +++ b/tests/fakes.py @@ -0,0 +1,227 @@ +"""In-memory :class:`~tokenops.control.ledger_backend.LedgerBackend` for unit tests. + +Mirrors the control-plane semantics (in-order apply, per-tenant idempotency dedup, +``totals`` in the ack). Kept honest by ``test_ledger_backend_contract.py``, which runs +the same assertions against this and the real plane over ``httpx.ASGITransport``. + +``.trip()`` / ``.heal()`` / ``.fail_next()`` drive the plane-unreachable tests. +""" + +from __future__ import annotations + +import time +import uuid +from typing import Any + +from tokenops.control.ledger_backend import ( + AggregateState, + ApplyResult, + LedgerEvent, + PrecheckRequest, +) +from tokenops.control.models import ( + GovernanceMode, + RunAlreadyRegisteredError, + RunNotRegisteredError, + RunRegistration, + parse_governance_mode, +) + +_RUN_STATE_WINDOW = 64 + + +class PlaneUnreachable(RuntimeError): + """Raised by a tripped FakeLedgerBackend, mimicking an httpx transport error.""" + + +class FakeLedgerBackend: + def __init__(self, *, governance: dict[str, Any] | None = None) -> None: + self._spent: dict[tuple[str, str, str], int] = {} + self._inflight: dict[str, int] = {} + self._halt: dict[str, tuple[bool, str | None]] = {} + self._regs: dict[str, RunRegistration] = {} + self._reg_at: dict[str, float] = {} + self._run_state: dict[str, dict[str, Any]] = {} + self._seen: set[str] = set() + self._records: dict[str, dict[str, Any]] = {} + self._governance = governance or {"governance": {"budgets": [], "policies": {}}} + # fault injection + self._tripped = False + self._fail_next = 0 + + # ---- fault injection -------------------------------------------------- # + + def trip(self) -> None: + self._tripped = True + + def heal(self) -> None: + self._tripped = False + self._fail_next = 0 + + def fail_next(self, n: int) -> None: + self._fail_next = n + + def _guard(self) -> None: + if self._tripped: + raise PlaneUnreachable("plane unreachable (tripped)") + if self._fail_next > 0: + self._fail_next -= 1 + raise PlaneUnreachable("plane unreachable (transient)") + + # ---- LedgerBackend ------------------------------------------------- # + + def close(self) -> None: # noqa: D401 - parity with HttpLedgerBackend + return None + + def read_state(self, req: PrecheckRequest) -> AggregateState: + self._guard() + out = AggregateState(server_ts=time.time()) + if "halt" in req.want: + halted, reason = self._halt.get(req.run_id, (False, None)) + out.halted, out.halt_reason = halted, reason + if "spent" in req.want: + out.spent = { + f"{b['budget_id']}|{b['segment_key']}|{b.get('period', 'lifetime')}": self._spent.get( + (b["budget_id"], b["segment_key"], b.get("period", "lifetime")), 0 + ) + for b in req.budgets + } + if "inflight" in req.want: + out.inflight = {s: self._inflight.get(s, 0) for s in req.segment_keys} + if "window" in req.want: + st = self._run_state.get(req.run_id) + out.window = ( + { + "step_count": st["step_count"], + "recent": st["recent"], + "velocity_micros_per_step": st["velocity_micros_per_step"], + } + if st + else {"step_count": 0, "recent": [], "velocity_micros_per_step": 0.0} + ) + return out + + def apply_events(self, events: list[LedgerEvent], *, durability: str = "sync") -> ApplyResult: + self._guard() + if not events: + return ApplyResult() + accepted = deduped = 0 + touched: set[tuple[str, str, str]] = set() + run_ids: set[str] = set() + for ev in events: + key = str(ev.get("idempotency_key") or "").strip() + if not key: + raise ValueError("event missing idempotency_key") + if ev.get("kind") == "spent_add": + for t in ev.get("targets") or []: + touched.add((t["budget_id"], t["segment_key"], t.get("period", "lifetime"))) + if ev.get("run_id"): + run_ids.add(str(ev["run_id"])) + if key in self._seen: + deduped += 1 + continue + self._apply_one(ev) + self._seen.add(key) + accepted += 1 + totals = {f"{b}|{s}|{p}": self._spent.get((b, s, p), 0) for (b, s, p) in touched} + halted = any(self._halt.get(r, (False, None))[0] for r in run_ids) + return ApplyResult(accepted=accepted, deduped=deduped, totals=totals, halted=halted) + + def _apply_one(self, ev: LedgerEvent) -> None: + kind = ev.get("kind") + run_id = str(ev.get("run_id") or "") + if kind == "spent_add": + delta = int(ev.get("delta_micros", 0)) + for t in ev.get("targets") or []: + k = (t["budget_id"], t["segment_key"], t.get("period", "lifetime")) + self._spent[k] = self._spent.get(k, 0) + delta + elif kind == "admit": + s = ev["segment_key"] + self._inflight[s] = self._inflight.get(s, 0) + 1 + elif kind == "complete": + s = ev["segment_key"] + self._inflight[s] = max(0, self._inflight.get(s, 0) - 1) + elif kind == "step": + st = self._run_state.setdefault( + run_id, {"step_count": 0, "recent": [], "velocity_micros_per_step": 0.0} + ) + st["recent"] = ( + st["recent"] + + [ + { + k: ev.get(k) + for k in ( + "agent", + "seq", + "node_type", + "boundary_id", + "cost_micros", + "cum_spent_micros", + "usage", + "tags", + "tool_signature", + "result_hash", + "ts", + ) + if ev.get(k) is not None + } + ] + )[-_RUN_STATE_WINDOW:] + st["step_count"] += 1 + win = st["recent"] + st["velocity_micros_per_step"] = ( + (win[-1].get("cum_spent_micros", 0) - win[0].get("cum_spent_micros", 0)) + / (len(win) - 1) + if len(win) >= 2 + else 0.0 + ) + elif kind == "halt_mark": + self._halt[run_id] = (True, ev.get("reason") or None) + if run_id in self._records: + self._records[run_id]["status"] = "halted" + elif kind == "halt_clear": + self._halt[run_id] = (False, None) + else: + raise ValueError(f"unknown event kind {kind!r}") + + def register_run( + self, + *, + intent: str = "", + user_dims: dict[str, str] | None = None, + mode: GovernanceMode | str | None = None, + run_id: str | None = None, + ) -> RunRegistration: + self._guard() + rid = (run_id or "").strip() or f"run_{uuid.uuid4().hex[:8]}" + if rid in self._regs: + raise RunAlreadyRegisteredError(f"run {rid!r} is already registered") + reg = RunRegistration( + run_id=rid, + intent=intent, + user_dims={str(k): str(v) for k, v in (user_dims or {}).items()}, + mode=parse_governance_mode(mode) if mode not in (None, "") else GovernanceMode.ENFORCE, + ) + self._regs[rid] = reg + self._reg_at[rid] = time.time() + self._records[rid] = {"run_id": rid, "agent": intent or "agent", "status": "running"} + return reg + + def resolve_run(self, run_id: str) -> RunRegistration: + self._guard() + try: + return self._regs[run_id] + except KeyError: + raise RunNotRegisteredError(f"run {run_id!r} is not registered") from None + + def governance_config_for(self, agent: str) -> dict[str, Any]: + self._guard() + return self._governance + + def patch_run_record(self, run_id: str, **fields: Any) -> None: + self._guard() + rec = self._records.setdefault(run_id, {"run_id": run_id}) + for k, v in fields.items(): + if k in ("steps", "cost_micros"): + continue + rec[k] = v diff --git a/tests/test_ledger_backend_contract.py b/tests/test_ledger_backend_contract.py new file mode 100644 index 0000000..e5cd3c8 --- /dev/null +++ b/tests/test_ledger_backend_contract.py @@ -0,0 +1,142 @@ +"""Verified-fake contract suite. + +Every assertion runs against BOTH backends (``any_backend`` is parametrised over the +in-memory ``FakeLedgerBackend`` and a real ``control_plane.app`` over ASGI). If the fake +drifts from the plane, this file fails. +""" + +from __future__ import annotations + +import pytest + +from tokenops.control.ledger_backend import PrecheckRequest +from tokenops.control.models import RunAlreadyRegisteredError, RunNotRegisteredError + +RUN = "run_1" +SEG = f"run:{RUN}" + + +def _spent_add(key: str, delta: int) -> dict: + return { + "kind": "spent_add", + "idempotency_key": key, + "run_id": RUN, + "delta_micros": delta, + "targets": [ + {"budget_id": "__run_total__", "segment_key": SEG, "period": "lifetime"}, + {"budget_id": "run_llm_cap", "segment_key": SEG, "period": "lifetime"}, + ], + } + + +def _precheck() -> PrecheckRequest: + return PrecheckRequest( + run_id=RUN, + segment_keys=[SEG], + budgets=[{"budget_id": "run_llm_cap", "segment_key": SEG, "period": "lifetime"}], + want=["spent", "inflight", "halt"], + ) + + +def test_spent_add_fan_out_and_totals(any_backend): + res = any_backend.apply_events([_spent_add("k1", 10_500)]) + assert res.accepted == 1 and res.deduped == 0 + assert res.totals["run_llm_cap|run:run_1|lifetime"] == 10_500 + assert res.totals["__run_total__|run:run_1|lifetime"] == 10_500 + + +def test_in_order_accumulation(any_backend): + any_backend.apply_events([_spent_add("a", 10_500), _spent_add("b", 4_800)]) + st = any_backend.read_state(_precheck()) + assert st.spent["run_llm_cap|run:run_1|lifetime"] == 15_300 + + +def test_idempotent_replay(any_backend): + any_backend.apply_events([_spent_add("dup", 10_500)]) + res = any_backend.apply_events([_spent_add("dup", 10_500)]) + assert res.accepted == 0 and res.deduped == 1 + assert res.totals["run_llm_cap|run:run_1|lifetime"] == 10_500 + + +def test_admit_complete(any_backend): + ev = lambda k, kind: { # noqa: E731 + "kind": kind, + "idempotency_key": k, + "run_id": RUN, + "segment_key": SEG, + } + any_backend.apply_events([ev("i1", "admit"), ev("i2", "admit")]) + assert any_backend.read_state(_precheck()).inflight[SEG] == 2 + any_backend.apply_events([ev("i3", "complete")]) + assert any_backend.read_state(_precheck()).inflight[SEG] == 1 + + +def test_step_window(any_backend): + step = lambda seq, cum: { # noqa: E731 + "kind": "step", + "idempotency_key": f"{RUN}:a:{seq}:step", + "run_id": RUN, + "agent": "a", + "seq": seq, + "node_type": "llm", + "boundary_id": "a.chat", + "cost_micros": 10_500, + "cum_spent_micros": cum, + "ts": float(seq), + } + any_backend.apply_events([step(1, 10_500), step(2, 21_000)]) + win = any_backend.read_state(PrecheckRequest(run_id=RUN, want=["window"])).window + assert win["step_count"] == 2 + assert len(win["recent"]) == 2 + assert win["velocity_micros_per_step"] == 10_500.0 + + +def test_halt_mark_visible(any_backend): + any_backend.apply_events( + [ + { + "kind": "halt_mark", + "idempotency_key": "h", + "run_id": RUN, + "reason": "step_cap: 20", + "detector": "step_cap", + } + ] + ) + st = any_backend.read_state(PrecheckRequest(run_id=RUN, want=["halt"])) + assert st.halted is True and st.halt_reason == "step_cap: 20" + + +def test_unknown_kind_raises(any_backend): + with pytest.raises(Exception): # noqa: B017 - Fake: ValueError, Http: HTTPStatusError + any_backend.apply_events([{"kind": "bogus", "idempotency_key": "x", "run_id": RUN}]) + + +def test_missing_idempotency_key_raises(any_backend): + with pytest.raises(Exception): # noqa: B017 + any_backend.apply_events( + [{"kind": "spent_add", "run_id": RUN, "delta_micros": 1, "targets": []}] + ) + + +def test_register_resolve_roundtrip(any_backend): + reg = any_backend.register_run(intent="demo", user_dims={"user_id": "alice"}, run_id=RUN) + assert reg.run_id == RUN and reg.user_dims == {"user_id": "alice"} + assert any_backend.resolve_run(RUN).intent == "demo" + + +def test_register_twice_conflicts(any_backend): + any_backend.register_run(intent="demo", run_id=RUN) + with pytest.raises(RunAlreadyRegisteredError): + any_backend.register_run(intent="demo", run_id=RUN) + + +def test_resolve_unknown_raises(any_backend): + with pytest.raises(RunNotRegisteredError): + any_backend.resolve_run("nope") + + +def test_patch_run_record_drops_derived(any_backend): + any_backend.register_run(intent="demo", run_id=RUN) + # must not raise even though steps/cost_micros are passed + any_backend.patch_run_record(RUN, status="completed", steps=99, cost_micros=123)