From 0509015f92e8513d814375ae9cbb364313c1b0ec Mon Sep 17 00:00:00 2001 From: susheem-k Date: Sat, 12 Sep 2026 16:26:22 +0530 Subject: [PATCH 1/4] refactor: rename RunState to LocalRunState; add PolicyInstance.data_scope (#118) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First increment of tokenops#118 Phase 2 — two small, independent, mechanical pieces from the locked decisions, landed ahead of the larger Ledger/LedgerView backend rewire: - Ledger.RunState -> LocalRunState (decision #9): makes the two-tier model explicit — this is the per-process Tier-1 cache, not the plane's authoritative run_state. Pure rename, no behavior change. - PolicyInstance.data_scope: local | global, default local (decision #5). Persisted in the local Store (additive migration, mirrors the pattern already used for dims/parent_span/governance_events) and round-tripped through governance_config_for. Not yet consumed by build_governor/Governor — grouping detectors by data_scope is part of the LedgerBackend rewire. Full suite: 237 passed, 16 skipped, 11 deselected (matches main's current baseline pre-#130). ruff check/format and mypy clean on touched files. Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 11 ++++++++++- src/tokenops/control/__init__.py | 4 ++-- src/tokenops/control/ledger.py | 24 ++++++++++++++---------- src/tokenops/control/models.py | 12 +++++++++++- src/tokenops/control/run.py | 2 +- src/tokenops/control/store.py | 20 +++++++++++++++++--- 6 files changed, 55 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c07a53..4c39282 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,12 +20,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `[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. -- `Ledger.close_run(run_id)` — drops the per-process `RunState`; called by +- `Ledger.close_run(run_id)` — drops the per-process `LocalRunState`; called by `tokenops_run` on scope exit so a long-lived / shared-governor process does not accumulate per-run window state (#115). +- `PolicyInstance.data_scope` (`local` | `global`, default `local`) — which tier a + policy's detector reads: Tier-1 `LocalRunState` cache, or the plane's authoritative + state via `precheck` (#118). Persisted (`policy_instances.data_scope`, additive + migration) and round-tripped through `governance_config_for`; not yet consumed by + `build_governor` — the Governor doesn't group detectors by scope until the + `LedgerBackend` rewire lands. ### Changed +- `Ledger`'s `RunState` renamed `LocalRunState` (#118, locked decision #9) — makes the + two-tier model explicit ahead of the `LedgerBackend` rewire: this is the per-process + Tier-1 cache, not the plane's authoritative `run_state`. - `Ledger.record` no longer writes a zero-delta spend row for non-priced crossings (tool calls, un-rolled-up delegates) — a free crossing is a *step*, not spend. The cost ledger only moves on priced events (#118). diff --git a/src/tokenops/control/__init__.py b/src/tokenops/control/__init__.py index 4e2fe61..6929713 100644 --- a/src/tokenops/control/__init__.py +++ b/src/tokenops/control/__init__.py @@ -62,7 +62,7 @@ wrap_complete, wrap_stream, ) -from tokenops.control.ledger import Budget, Ledger, RunState, segment_key +from tokenops.control.ledger import Budget, Ledger, LocalRunState, segment_key from tokenops.control.ledger_backend import ( AggregateState, ApplyResult, @@ -112,7 +112,7 @@ # ledger "Budget", "Ledger", - "RunState", + "LocalRunState", "segment_key", # ledger backend (remote-only) "LedgerBackend", diff --git a/src/tokenops/control/ledger.py b/src/tokenops/control/ledger.py index b3e4d4b..883b053 100644 --- a/src/tokenops/control/ledger.py +++ b/src/tokenops/control/ledger.py @@ -4,11 +4,11 @@ crossing can increment *many* budget accumulators at once, so spend cannot live inside a single run object. - runs: run_id -> RunState (per-run ephemeral: steps, window, halted) + runs: run_id -> LocalRunState (per-run ephemeral: steps, window, halted) spent: (budget_id, segment_key, period) -> micros (one budget bound to a segment) inflight: segment_key -> int (concurrent calls in flight, admit/complete) -Why three maps, not fields on RunState: +Why three maps, not fields on LocalRunState: * ``spent`` is budget-scoped — a tenant's monthly cap and this run's cap are two accumulators the *same* event feeds. Keying by run would lose that fan-out. * ``inflight`` is segment-scoped for concurrency only — it is admit/complete state, @@ -141,9 +141,13 @@ def segment_key(attr: Attribution, budget: Budget) -> str | None: @dataclass -class RunState: - """Ephemeral per-run state. The dict key in ``Ledger.runs`` is the run_id — there is - deliberately no run_id field here (the index is not a field).""" +class LocalRunState: + """Ephemeral per-process, per-run cache (Tier 1). The dict key in ``Ledger.runs`` is + the run_id — there is deliberately no run_id field here (the index is not a field). + + This is the per-process cache in the two-tier model: cheap reads for `local`-scope + policies without a round trip. `global`-scope policies read the plane's `run_state` + table instead (via `precheck`), since that is authoritative across processes.""" steps: int = 0 window: list[BoundaryStep] = field(default_factory=list) @@ -173,7 +177,7 @@ def __init__( ) -> None: self._lock = threading.RLock() self._store = store - self.runs: dict[str, RunState] = {} + self.runs: dict[str, LocalRunState] = {} self._spent: dict[tuple[str, str, str], Micros] = defaultdict(int) self._inflight: dict[str, int] = defaultdict(int) self._budgets: list[Budget] = [RUN_TOTAL_BUDGET, *budgets] @@ -205,13 +209,13 @@ def _write_spent_delta( def open_run(self, run_id: str, parent_run: str | None = None) -> None: with self._lock: - self.runs[run_id] = RunState(parent_run=parent_run) + self.runs[run_id] = LocalRunState(parent_run=parent_run) def close_run(self, run_id: str) -> None: - """Drop the per-process :class:`RunState` for a finished run. Idempotent. + """Drop the per-process :class:`LocalRunState` for a finished run. Idempotent. ``tokenops_run`` calls this on scope exit for the run it opened, so a - long-lived / shared-governor process does not accumulate ``RunState`` + long-lived / shared-governor process does not accumulate ``LocalRunState`` entries (each holds a full ``window`` of ``BoundaryStep``s). """ with self._lock: @@ -290,7 +294,7 @@ def mark_halted(self, run_id: str, reason: str = "") -> None: with self._lock: rs = self.runs.get(run_id) if rs is None: - rs = self.runs[run_id] = RunState() + rs = self.runs[run_id] = LocalRunState() rs.halted = True rs.halt_reason = reason or rs.halt_reason if self._store is not None: diff --git a/src/tokenops/control/models.py b/src/tokenops/control/models.py index 01808a1..f29f3fa 100644 --- a/src/tokenops/control/models.py +++ b/src/tokenops/control/models.py @@ -78,10 +78,19 @@ class BudgetSpec: period: str = "lifetime" +DataScope = Literal["local", "global"] + + @dataclass class PolicyInstance: """One configured policy = a template + params, optionally scoped to an agent and - attached to a budget and/or segment.""" + attached to a budget and/or segment. + + ``data_scope`` says which tier its detector reads: ``local`` (Tier-1 + ``LocalRunState`` cache, no round trip) or ``global`` (the plane's authoritative + `run_state`/spend via `precheck`, needed for anything that must be correct across + processes — e.g. cost budgets). The Governor groups detectors by this field so a + call needs at most one `precheck` regardless of how many `global` policies it has.""" id: str template: str # a key of control.config._TEMPLATES @@ -90,6 +99,7 @@ class PolicyInstance: budget_id: str | None = None segment_id: str | None = None enabled: bool = True + data_scope: DataScope = "local" @dataclass diff --git a/src/tokenops/control/run.py b/src/tokenops/control/run.py index 41c0115..50ec6af 100644 --- a/src/tokenops/control/run.py +++ b/src/tokenops/control/run.py @@ -265,7 +265,7 @@ def tokenops_run( finally: clear_run_context() if open_ledger_run: - # Drop this process's RunState so a long-lived / shared-governor process + # Drop this process's LocalRunState so a long-lived / shared-governor process # does not leak per-run window state. gov.ledger.close_run(reg.run_id) diff --git a/src/tokenops/control/store.py b/src/tokenops/control/store.py index 588a26d..9fd5b0e 100644 --- a/src/tokenops/control/store.py +++ b/src/tokenops/control/store.py @@ -75,7 +75,8 @@ def _known_policy_templates() -> frozenset[str]: ); CREATE TABLE IF NOT EXISTS policy_instances ( id TEXT PRIMARY KEY, template TEXT NOT NULL, params TEXT NOT NULL DEFAULT '{}', - agent TEXT, budget_id TEXT, segment_id TEXT, enabled INTEGER NOT NULL DEFAULT 1 + agent TEXT, budget_id TEXT, segment_id TEXT, enabled INTEGER NOT NULL DEFAULT 1, + data_scope TEXT NOT NULL DEFAULT 'local' ); CREATE TABLE IF NOT EXISTS runs ( run_id TEXT PRIMARY KEY, agent TEXT NOT NULL, status TEXT NOT NULL, @@ -169,6 +170,11 @@ def _migrate(self) -> None: self._db.execute("ALTER TABLE runs ADD COLUMN dims TEXT NOT NULL DEFAULT '{}'") if "parent_span" not in cols: self._db.execute("ALTER TABLE runs ADD COLUMN parent_span TEXT") + pol_cols = {row[1] for row in self._db.execute("PRAGMA table_info(policy_instances)")} + if "data_scope" not in pol_cols: + self._db.execute( + "ALTER TABLE policy_instances ADD COLUMN data_scope TEXT NOT NULL DEFAULT 'local'" + ) reg_cols = {row[1] for row in self._db.execute("PRAGMA table_info(run_registrations)")} if "mode" not in reg_cols: self._db.execute( @@ -257,8 +263,9 @@ def upsert_policy_instance(self, pi: PolicyInstance) -> PolicyInstance: f"unknown policy template {pi.template!r}; known: {sorted(_known_policy_templates())}" ) self._db.execute( - "REPLACE INTO policy_instances(id, template, params, agent, budget_id, segment_id, enabled) " - "VALUES (?,?,?,?,?,?,?)", + "REPLACE INTO policy_instances" + "(id, template, params, agent, budget_id, segment_id, enabled, data_scope) " + "VALUES (?,?,?,?,?,?,?,?)", ( pi.id, pi.template, @@ -267,6 +274,7 @@ def upsert_policy_instance(self, pi: PolicyInstance) -> PolicyInstance: pi.budget_id, pi.segment_id, 1 if pi.enabled else 0, + pi.data_scope, ), ) self._db.commit() @@ -452,6 +460,10 @@ def _assemble_governance_config(self, agent: str) -> dict: params.setdefault("dimension", seg.dimension) if seg.tag_key: params.setdefault("tag_key", seg.tag_key) + # Not yet consumed by build_governor/_TEMPLATES — the Governor doesn't group + # detectors by data_scope until the LedgerBackend rewire lands. Round-tripped + # here now so a policy's configured scope isn't silently dropped in the interim. + params["data_scope"] = pi.data_scope policies[pi.template] = params return {"governance": {"budgets": budgets, "policies": policies}} @@ -911,6 +923,7 @@ def _budget_dict(b: BudgetSpec) -> dict: def _policy(r: sqlite3.Row) -> PolicyInstance: + keys = r.keys() return PolicyInstance( id=r["id"], template=r["template"], @@ -919,6 +932,7 @@ def _policy(r: sqlite3.Row) -> PolicyInstance: budget_id=r["budget_id"], segment_id=r["segment_id"], enabled=bool(r["enabled"]), + data_scope=r["data_scope"] if "data_scope" in keys else "local", ) From cd1c0ff02ef97e0f0aa9de4c64f7bb087e3e2fb2 Mon Sep 17 00:00:00 2001 From: susheem-k Date: Sat, 12 Sep 2026 16:41:45 +0530 Subject: [PATCH 2/4] feat: wire Ledger onto LedgerBackend (precheck/apply_events) alongside Store (#118) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a backend= mode to Ledger, additive alongside the existing store=/in-memory modes (mutually exclusive with store=) — the core Ledger/LedgerView piece of the remote-only rewrite's Phase 2 checklist. build_governor/ControlPlaneClient don't construct it yet; that swap, and migrating the ~18 Store-dependent tests, are the next increments (staged deliberately, per the epic's own Phase 2/3 split). - record(): batches the step + spent_add (when cost > 0) into one apply_events call sharing an idempotency seq, so the ack's totals cover the run-total cum spend without a separate read_state round trip for the common case. - admit/complete/mark_halted/clear_halt: one-shot apply_events writes. Their idempotency keys are random (uuid4), not the contract's deterministic {run_id}:...:{seq} recipe — that recipe needs a real run_id to stay collision-free across processes for a non-run-scoped segment key (e.g. an agent-dimension concurrency cap shared across runs), which isn't always available at these call sites. A random key is correct for a one-shot, non-retried write either way; the deterministic recipe only earns its keep once a buffered backend needs a retried flush to regenerate the same key (the existing # TODO(buffering) seam in ledger_backend.py). - cost_micros/budget_left/is_halted/inflight: read via backend.read_state (precheck). No per-call batching across detectors yet (one read per method call) — Governor grouping detectors by data_scope into a single precheck per governed moment is a separate, later optimization, not required for correctness given there's no client-side write buffer yet either (decision #6). - velocity/recent/window/step_count are unchanged: always Tier-1 LocalRunState, never a backend round trip — those are inherently per-process reads. tests/test_ledger_backend_mode.py: 9 new tests against FakeLedgerBackend, including two Ledger instances sharing one backend (the actual cross-process scenario this whole rewire is for) for halt visibility and inflight counting. Full suite: 246 passed, 16 skipped, 11 deselected. ruff check/format and mypy clean. Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 10 ++ src/tokenops/control/ledger.py | 230 +++++++++++++++++++++++++++--- tests/test_ledger_backend_mode.py | 163 +++++++++++++++++++++ 3 files changed, 381 insertions(+), 22 deletions(-) create mode 100644 tests/test_ledger_backend_mode.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 4c39282..439e1c9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 migration) and round-tripped through `governance_config_for`; not yet consumed by `build_governor` — the Governor doesn't group detectors by scope until the `LedgerBackend` rewire lands. +- `Ledger(backend=...)` — routes every write through `LedgerBackend.apply_events` and + every spend/inflight/halt read through `read_state` (`precheck`), alongside the + existing `store=`/in-memory modes (mutually exclusive with `store`) (#118). + `step`/`spent_add` batch together per crossing so the ack's `totals` cover the run + total in one round trip; `admit`/`complete`/`halt_mark`/`halt_clear` are separate + one-shot writes. `velocity`/`recent`/`window`/`step_count` stay Tier-1-only (never a + backend round trip) — those are inherently local, per-process reads. Not yet wired + into `build_governor`/`ControlPlaneClient`; `tests/test_ledger_backend_mode.py` + exercises it directly against `FakeLedgerBackend`, including two `Ledger` instances + sharing one backend (the cross-process case). ### Changed diff --git a/src/tokenops/control/ledger.py b/src/tokenops/control/ledger.py index 883b053..6a65f55 100644 --- a/src/tokenops/control/ledger.py +++ b/src/tokenops/control/ledger.py @@ -44,6 +44,7 @@ from __future__ import annotations import threading +import uuid from collections import defaultdict from collections.abc import Callable, Sequence from dataclasses import dataclass, field @@ -55,6 +56,7 @@ Micros, Observation, ) +from tokenops.control.ledger_backend import LedgerBackend, LedgerEvent, PrecheckRequest if TYPE_CHECKING: from tokenops.control.store import Store @@ -135,6 +137,108 @@ def segment_key(attr: Attribution, budget: Budget) -> str | None: return segment_key_for(attr, budget.dimension, budget.tag_key) +def _run_id_from_segment_key(segment_key: str) -> str | None: + """Recover ``run_id`` from a ``run:``-dimension segment key. ``None`` for any other + dimension (user/agent/tenant/tag) — those don't carry a run_id at all. + + The plane's ``precheck`` only *uses* ``run_id`` to answer ``halt``/``window`` — a + ``spent`` or ``inflight`` lookup for a non-run segment is valid with an empty + ``run_id`` (see ``control_plane.store.Store.precheck``), so callers pass ``"" `` when + this returns ``None`` rather than needing the caller's own run context threaded in. + """ + prefix = "run:" + return segment_key[len(prefix) :] if segment_key.startswith(prefix) else None + + +# --------------------------------------------------------------------------- # +# Backend-mode event builders (see ``ledger_backend.LedgerEvent`` / the control-plane +# ``docs/api-contract.md`` §5-6 for the wire shape and idempotency-key recipe) +# --------------------------------------------------------------------------- # + + +def _new_idempotency_key() -> str: + """A fresh, globally-unique key for a one-shot (non-buffered, non-retried) write. + + ``admit``/``complete``/``halt_mark``/``halt_clear`` can be issued for a segment key + that carries no run_id (e.g. an ``agent``-dimension concurrency cap shared across + many runs and processes) — the contract's deterministic ``{run_id}:...:{seq}`` + recipe (``docs/api-contract.md`` §6) needs a real, globally-known run_id to stay + collision-free across processes, which isn't always available here. A random key is + always correct for a single fire-and-forget write; the deterministic recipe only + earns its keep once a buffered backend needs a retried flush to regenerate the same + key (tracked with the ``# TODO(buffering)`` seam in ``ledger_backend.py``). + """ + return uuid.uuid4().hex + + +def _inflight_event( + kind: Literal["admit", "complete"], run_id: str, segment_key: str +) -> LedgerEvent: + return { + "kind": kind, + "idempotency_key": _new_idempotency_key(), + "run_id": run_id, + "segment_key": segment_key, + } + + +def _halt_event( + kind: Literal["halt_mark", "halt_clear"], + run_id: str, + reason: str = "", + detector: str = "", +) -> LedgerEvent: + event: LedgerEvent = { + "kind": kind, + "idempotency_key": _new_idempotency_key(), + "run_id": run_id, + } + if reason: + event["reason"] = reason + if detector: + event["detector"] = detector + return event + + +def _step_event(obs: Observation, seq: int, cost: Micros) -> LedgerEvent: + event: LedgerEvent = { + "kind": "step", + "idempotency_key": f"{obs.attr.run_id}:{obs.attr.agent}:{seq}:step", + "ts": obs.ts, + "run_id": obs.attr.run_id, + "agent": obs.attr.agent, + "seq": seq, + "node_type": obs.node_type, + "boundary_id": obs.boundary_id, + "cost_micros": cost, + "tags": {**dict(obs.boundary_tags), **dict(obs.tags)}, + } + if obs.usage is not None: + event["usage"] = { + "input": obs.usage.input, + "output": obs.usage.output, + "cached": obs.usage.cached, + "reasoning": obs.usage.reasoning, + } + if obs.signature is not None: + event["tool_signature"] = obs.signature + if obs.result_hash is not None: + event["result_hash"] = obs.result_hash + return event + + +def _spent_add_event( + run_id: str, agent: str, seq: int, delta: Micros, targets: list[dict[str, str]] +) -> LedgerEvent: + return { + "kind": "spent_add", + "idempotency_key": f"{run_id}:{agent}:{seq}:spent_add", + "run_id": run_id, + "delta_micros": delta, + "targets": targets, + } + + # =========================================================================== # # Per-run ephemeral state # # =========================================================================== # @@ -154,6 +258,9 @@ class LocalRunState: halted: bool = False halt_reason: str | None = None parent_run: str | None = None + #: Last-known run-total cum_spent from a backend ack, so a zero-cost crossing's + #: BoundaryStep doesn't need its own read_state round trip. + cum_spent_cache: Micros = 0 # =========================================================================== # @@ -163,7 +270,17 @@ class LocalRunState: class Ledger: """Attribute + LedgerView. Per-process run state (window, local step count); spend, - inflight, and halt may be backed by :class:`Store` for cross-process A2A. + inflight, and halt may be backed by :class:`Store` (local SQLite) or a + :class:`~tokenops.control.ledger_backend.LedgerBackend` (the remote control plane) for + cross-process / cross-agent consistency. At most one of ``store``/``backend`` may be + given; neither means fully in-memory (single-process only, tests). + + ``backend`` mode is the target of the remote-only rewrite (tokenops#118): every write + goes through ``apply_events`` (one batch per crossing) and every spend/inflight/halt + read goes through ``read_state`` (``precheck``). There is no per-call batching across + detectors yet — each read is its own round trip — so a call with several ``global``- + scope policies costs more than one ``precheck``; that optimization (and buffering the + write side) are tracked follow-ups, not required for correctness. Thread-safe for concurrent use from multiple threads (see module docstring). """ @@ -174,9 +291,13 @@ def __init__( budgets: Sequence[Budget] = (), price: PriceFn | None = None, store: Store | None = None, + backend: LedgerBackend | None = None, ) -> None: + if store is not None and backend is not None: + raise ValueError("Ledger takes at most one of store= / backend=, not both") self._lock = threading.RLock() self._store = store + self._backend = backend self.runs: dict[str, LocalRunState] = {} self._spent: dict[tuple[str, str, str], Micros] = defaultdict(int) self._inflight: dict[str, int] = defaultdict(int) @@ -188,6 +309,17 @@ def _spent_key(self, budget_id: str, segment_key: str, period: str) -> tuple[str return (budget_id, segment_key, period) def _read_spent(self, budget_id: str, segment_key: str, period: str) -> Micros: + if self._backend is not None: + state = self._backend.read_state( + PrecheckRequest( + run_id=_run_id_from_segment_key(segment_key) or "", + budgets=[ + {"budget_id": budget_id, "segment_key": segment_key, "period": period} + ], + want=["spent"], + ) + ) + return state.spent.get(f"{budget_id}|{segment_key}|{period}", 0) if self._store is not None: return self._store.ledger_get_spent(budget_id, segment_key, period) return self._spent[self._spent_key(budget_id, segment_key, period)] @@ -199,6 +331,9 @@ def _write_spent_delta( period: str, delta: Micros, ) -> Micros: + # Backend mode batches spent_add into record()'s single apply_events call + # (it needs to share an idempotency key/seq with that crossing's step event), + # so it never reaches this in-memory/Store fallback path. if self._store is not None: return self._store.ledger_add_spent(budget_id, segment_key, period, delta) key = self._spent_key(budget_id, segment_key, period) @@ -224,7 +359,10 @@ def close_run(self, run_id: str) -> None: def admit(self, segment_key: str) -> None: """A call for this segment has started (concurrency).""" with self._lock: - if self._store is not None: + if self._backend is not None: + run_id = _run_id_from_segment_key(segment_key) or "" + self._backend.apply_events([_inflight_event("admit", run_id, segment_key)]) + elif self._store is not None: self._store.ledger_admit(segment_key) else: self._inflight[segment_key] += 1 @@ -233,7 +371,10 @@ def complete(self, segment_key: str) -> None: """A call for this segment has returned. Floored at 0 so a stray complete cannot drive the counter negative.""" with self._lock: - if self._store is not None: + if self._backend is not None: + run_id = _run_id_from_segment_key(segment_key) or "" + self._backend.apply_events([_inflight_event("complete", run_id, segment_key)]) + elif self._store is not None: self._store.ledger_complete(segment_key) else: self._inflight[segment_key] = max(0, self._inflight[segment_key] - 1) @@ -257,23 +398,48 @@ def record(self, obs: Observation) -> BoundaryStep: elif obs.node_type == "delegate": cost = obs.rolled_up_cost_micros # child run total rolls up into the parent - # One event can feed many accumulators (incl. the system run-total) — that is - # why spend lives in the map, not on the run object. A zero-cost crossing - # (tool call, delegate with no rollup) must not touch the cost ledger — it is - # a *step*, not spend. - if cost > 0: - for b in self._budgets: - sk = segment_key(obs.attr, b) - if sk is None: - continue - self._write_spent_delta(b.budget_id, sk, b.period, cost) - rs.steps += 1 - cum = self._read_spent( - RUN_TOTAL_BUDGET.budget_id, - f"run:{obs.attr.run_id}", - LIFETIME, - ) + + if self._backend is not None: + # One apply_events batch for both the step and the spend, so they share + # an idempotency seq and the ack's totals cover the run-total in one + # round trip (no separate read_state needed for the common case). + events: list[LedgerEvent] = [_step_event(obs, rs.steps, cost)] + if cost > 0: + targets = [] + for b in self._budgets: + sk = segment_key(obs.attr, b) + if sk is None: + continue + targets.append( + {"budget_id": b.budget_id, "segment_key": sk, "period": b.period} + ) + events.append( + _spent_add_event(obs.attr.run_id, obs.attr.agent, rs.steps, cost, targets) + ) + result = self._backend.apply_events(events) + if result.halted: + rs.halted = True + run_total_key = f"{RUN_TOTAL_BUDGET.budget_id}|run:{obs.attr.run_id}|{LIFETIME}" + if run_total_key in result.totals: + rs.cum_spent_cache = result.totals[run_total_key] + cum = rs.cum_spent_cache + else: + # One event can feed many accumulators (incl. the system run-total) — + # that is why spend lives in the map, not on the run object. A zero-cost + # crossing (tool call, delegate with no rollup) must not touch the cost + # ledger — it is a *step*, not spend. + if cost > 0: + for b in self._budgets: + sk = segment_key(obs.attr, b) + if sk is None: + continue + self._write_spent_delta(b.budget_id, sk, b.period, cost) + cum = self._read_spent( + RUN_TOTAL_BUDGET.budget_id, + f"run:{obs.attr.run_id}", + LIFETIME, + ) step = BoundaryStep( step=rs.steps, ts=obs.ts, @@ -297,7 +463,9 @@ def mark_halted(self, run_id: str, reason: str = "") -> None: rs = self.runs[run_id] = LocalRunState() rs.halted = True rs.halt_reason = reason or rs.halt_reason - if self._store is not None: + if self._backend is not None: + self._backend.apply_events([_halt_event("halt_mark", run_id, reason=reason)]) + elif self._store is not None: self._store.ledger_mark_halted(run_id, reason) def clear_halt(self, run_id: str) -> None: @@ -308,7 +476,9 @@ def clear_halt(self, run_id: str) -> None: if rs is not None: rs.halted = False rs.halt_reason = None - if self._store is not None: + if self._backend is not None: + self._backend.apply_events([_halt_event("halt_clear", run_id)]) + elif self._store is not None: self._store.ledger_clear_halt(run_id) # ---- read side (LedgerView) ------------------------------------------ # @@ -324,9 +494,19 @@ def step_count(self, run_id: str) -> int: def is_halted(self, run_id: str) -> bool: with self._lock: + rs = self.runs.get(run_id) + if rs and rs.halted: + return True # fast path: already known locally, no round trip + if self._backend is not None: + state = self._backend.read_state(PrecheckRequest(run_id=run_id, want=["halt"])) + if state.halted: + if rs is None: + rs = self.runs[run_id] = LocalRunState() + rs.halted = True + rs.halt_reason = state.halt_reason + return state.halted if self._store is not None and self._store.ledger_is_halted(run_id): return True - rs = self.runs.get(run_id) return bool(rs and rs.halted) def budget_left(self, budget_id: str, segment_key: str, period: str = "lifetime") -> Micros: @@ -341,6 +521,12 @@ def budget_left(self, budget_id: str, segment_key: str, period: str = "lifetime" def inflight(self, segment_key: str) -> int: with self._lock: + if self._backend is not None: + run_id = _run_id_from_segment_key(segment_key) or "" + state = self._backend.read_state( + PrecheckRequest(run_id=run_id, segment_keys=[segment_key], want=["inflight"]) + ) + return state.inflight.get(segment_key, 0) if self._store is not None: return self._store.ledger_inflight(segment_key) return self._inflight[segment_key] diff --git a/tests/test_ledger_backend_mode.py b/tests/test_ledger_backend_mode.py new file mode 100644 index 0000000..338c591 --- /dev/null +++ b/tests/test_ledger_backend_mode.py @@ -0,0 +1,163 @@ +"""Ledger(backend=...) — the LedgerBackend-routed write/read path (tokenops#118). + +Mirrors test_ledger.py's assertions but backed by FakeLedgerBackend instead of the +in-memory / Store path, proving record/admit/complete/mark_halted/clear_halt and every +LedgerView read produce the same answers when routed through apply_events/read_state. +""" + +from __future__ import annotations + +import pytest + +from conftest import make_attr, toy_price +from fakes import FakeLedgerBackend +from tokenops.control.core import Observation, Usage +from tokenops.control.ledger import RUN_TOTAL_BUDGET, UNLIMITED_LEFT, Budget, Ledger, segment_key + + +def _ledger(): + cap = Budget(budget_id="run_llm_cap", limit_micros=1_000_000, dimension="run") + backend = FakeLedgerBackend() + return Ledger(budgets=[cap], price=toy_price, backend=backend), cap, backend + + +def test_store_and_backend_are_mutually_exclusive(): + with pytest.raises(ValueError): + Ledger(store=object(), backend=FakeLedgerBackend()) + + +def test_pricing_and_cum_spent_via_backend(): + ledger, _cap, _backend = _ledger() + attr = make_attr() + ledger.open_run("run-1") + + step = ledger.record( + Observation( + attr=attr, + node_type="llm", + boundary_id="chat", + ts=1.0, + provider="openai", + model="gpt-4o-mini", + usage=Usage(input=820, output=45), + ) + ) + cost1 = 820 * 10 + 45 * 30 + assert step.step == 1 and step.cum_spent_micros == cost1 + assert ledger.cost_micros("run-1") == cost1 + + +def test_zero_cost_crossing_writes_no_spend_via_backend(): + ledger, _cap, backend = _ledger() + attr = make_attr() + ledger.open_run("run-1") + ledger.record( + Observation( + attr=attr, + node_type="tool", + boundary_id="search", + ts=1.0, + signature="sig", + result_hash="rh", + ) + ) + assert ledger.cost_micros("run-1") == 0 + # only a `step` event was applied — no spent_add touched the budget + assert backend._spent == {} + + +def test_budget_left_via_backend(): + ledger, cap, _backend = _ledger() + attr = make_attr() + ledger.open_run("run-1") + ledger.record( + Observation( + attr=attr, + node_type="llm", + boundary_id="chat", + ts=1.0, + provider="openai", + model="gpt-4o-mini", + usage=Usage(input=100, output=10), + ) + ) + spent = 100 * 10 + 10 * 30 + sk = segment_key(attr, cap) + assert ledger.budget_left(cap.budget_id, sk) == cap.limit_micros - spent + assert ledger.budget_left(RUN_TOTAL_BUDGET.budget_id, f"run:{attr.run_id}") == UNLIMITED_LEFT + assert ledger.budget_left("unknown-budget", sk) == 0 + + +def test_admit_complete_inflight_via_backend(): + ledger, _cap, _backend = _ledger() + ledger.open_run("run-1") + sk = "run:run-1" + assert ledger.inflight(sk) == 0 + ledger.admit(sk) + ledger.admit(sk) + assert ledger.inflight(sk) == 2 + ledger.complete(sk) + assert ledger.inflight(sk) == 1 + ledger.complete(sk) + ledger.complete(sk) # floored at 0, never negative + assert ledger.inflight(sk) == 0 + + +def test_mark_halted_and_clear_halt_via_backend(): + ledger, _cap, backend = _ledger() + ledger.open_run("run-1") + assert ledger.is_halted("run-1") is False + + ledger.mark_halted("run-1", "step_cap: 2") + assert ledger.is_halted("run-1") is True + assert backend._halt["run-1"] == (True, "step_cap: 2") + + ledger.clear_halt("run-1") + assert ledger.is_halted("run-1") is False + assert backend._halt["run-1"] == (False, None) + + +def test_is_halted_sees_a_halt_set_by_another_ledger_instance(): + """Two Ledgers sharing one backend (simulating two processes / a shared plane) — + a halt set on one must be visible from the other without local state.""" + backend = FakeLedgerBackend() + cap = Budget(budget_id="run_llm_cap", limit_micros=1_000_000, dimension="run") + ledger_a = Ledger(budgets=[cap], price=toy_price, backend=backend) + ledger_b = Ledger(budgets=[cap], price=toy_price, backend=backend) + + ledger_a.mark_halted("shared-run", "cost_budget: exhausted") + assert ledger_b.is_halted("shared-run") is True + + +def test_admit_complete_share_inflight_across_ledger_instances(): + backend = FakeLedgerBackend() + ledger_a = Ledger(backend=backend) + ledger_b = Ledger(backend=backend) + sk = "agent:research" # non-run-scoped segment key — no run_id embedded + ledger_a.admit(sk) + ledger_b.admit(sk) + assert ledger_a.inflight(sk) == 2 + assert ledger_b.inflight(sk) == 2 + + +def test_local_run_state_reads_stay_local_not_backend(): + """step_count/velocity/recent/window are the Tier-1 cache — per-process, never a + backend round trip. A second Ledger sharing the same backend does NOT see them.""" + backend = FakeLedgerBackend() + ledger_a = Ledger(price=toy_price, backend=backend) + ledger_b = Ledger(price=toy_price, backend=backend) + attr = make_attr() + ledger_a.open_run("run-1") + ledger_a.record( + Observation( + attr=attr, + node_type="llm", + boundary_id="chat", + ts=1.0, + provider="openai", + model="gpt-4o-mini", + usage=Usage(input=10, output=1), + ) + ) + assert ledger_a.step_count("run-1") == 1 + assert ledger_b.step_count("run-1") == 0 # no shared Tier-1 state, by design From 54c3e500bad9f41f6869e3a6e164ce024bb167b1 Mon Sep 17 00:00:00 2001 From: susheem-k Date: Sat, 12 Sep 2026 17:16:26 +0530 Subject: [PATCH 3/4] =?UTF-8?q?feat:=20eliminate=20the=20local=20ledger=20?= =?UTF-8?q?=E2=80=94=20tokenops=20is=20now=20hard-dependent=20on=20the=20c?= =?UTF-8?q?ontrol=20plane=20(#118)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The core swap for tokenops#118: ControlPlaneClient.from_env() now requires CONTROL_PLANE_URL/TOKENOPS_URL and raises if neither is set. There is no more TOKENOPS_EMBEDDED and no code path left that silently falls back to a local SQLite ledger. build_governor/tokenops_run construct Ledger(backend=...) (an HttpLedgerBackend, i.e. real precheck/events:batch traffic) for every live run. - client.py: from_env() requires a URL; new .backend property (lazy HttpLedgerBackend); should_mount_run_registration() always False (no embedded mode left to self-host POST /v1/runs under). The store= constructor kwarg remains as an explicit, visible test-only escape hatch — never reachable from from_env(), so no env misconfiguration can select it. - config.py/run.py: build_governor/build_governance_stack take backend=; run.py picks store vs backend off client_obj.embedded (not the local store= param, which was a real bug in an earlier draft of this change — a client= constructed with store= wasn't routing to its own Store). - http_store.py: HttpStore accepts an injectable httpx.Client (mirrors HttpLedgerBackend) so tests can point it at an in-process app. - dev_plane.py (new): launches a real agentplane-control-plane on a real localhost TCP port, in-process. Used by tokenops.demo (still zero-setup: it launches its own throwaway plane and configures it over HTTP) and by tests/conftest.py::live_plane_url for tests that must exercise ControlPlaneClient.from_env() itself (an in-process ASGI app isn't reachable that way since from_env() builds its own plain httpx.Client). - CI now installs agentplane-control-plane from the control-plane repo (git+.../control-plane@main) alongside .[dev,contract] — the [contract] tests and tests/examples/ e2e suite now actually run a real control plane in CI instead of importorskip-ing past it. They're load-bearing coverage now. - Migrated every TOKENOPS_EMBEDDED-dependent test (test_tokenops_run.py, test_control_plane_client.py, test_control_plane_app.py, test_governance_config_cache.py, test_demo.py, and the two tests/examples/ e2e files) onto either the explicit store= escape hatch or live_plane_url. - tests/examples/test_bench_e2e.py and test_triad_e2e.py now configure their policies/budgets on a real, in-process control plane over its own HTTP API (PUT /v1/budgets, PUT /v1/policies) instead of a local Store — this is the concrete demonstration that governance policies configured on the plane reach a live, multi-agent run and HALT/steer it (step_cap, cost_budget, output_runaway CANCEL+RETRY, tool_output_cap deep swap). - Removed the now-dead TOKENOPS_EMBEDDED convenience shims from the three example CLI clients (a2a, brief, triad) — nothing reads that var anymore. - README/onboarding/control-plane-deploy docs: removed the stale TOKENOPS_EMBEDDED documentation and precedence warnings. Full suite: 258 passed, 16 skipped, 1 deselected. ruff check/format clean. mypy clean on every touched file. Not in this change (separate, later Phase 2/3 checklist items): removing the bundled src/tokenops/server + src/tokenops/ui/streamlit, dropping span_id, Governor batching detectors by data_scope into one precheck per call, and migrating the ~18 policy-logic unit tests that construct Ledger()/Store() directly for white-box testing (not a production bypass — never reachable from ControlPlaneClient.from_env()). Co-Authored-By: Claude Sonnet 5 --- .github/workflows/ci.yml | 8 +- CHANGELOG.md | 37 ++++++++ README.md | 21 ++--- docs/control-plane-deploy.md | 3 +- docs/guides/onboarding.md | 17 ++-- examples/a2a/client.py | 12 --- examples/brief/client.py | 7 -- examples/triad/client.py | 7 -- src/tokenops/control/client.py | 101 +++++++++++++------- src/tokenops/control/config.py | 14 ++- src/tokenops/control/dev_plane.py | 70 ++++++++++++++ src/tokenops/control/http_store.py | 15 ++- src/tokenops/control/run.py | 17 +++- src/tokenops/demo.py | 131 +++++++++++++++++++++----- tests/conftest.py | 21 +++++ tests/examples/test_bench_e2e.py | 76 ++++++++------- tests/examples/test_triad_e2e.py | 76 +++++++-------- tests/test_control_plane_app.py | 27 +++--- tests/test_control_plane_client.py | 63 ++++--------- tests/test_demo.py | 17 ++-- tests/test_governance_config_cache.py | 4 - tests/test_tokenops_run.py | 36 ++++--- 22 files changed, 511 insertions(+), 269 deletions(-) create mode 100644 src/tokenops/control/dev_plane.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4841fb9..64fc250 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -28,7 +28,13 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip setuptools wheel - python -m pip install -e ".[dev]" + # agentplane-control-plane 0.2.0 isn't on PyPI yet — install straight from + # the repo so the [contract] tests (real control-plane, not just the fake) + # actually run in CI instead of silently skipping. tokenops has no embedded + # ledger to fall back to (tokenops#118), so these are load-bearing, not + # optional coverage. + python -m pip install "agentplane-control-plane @ git+https://github.com/theagentplane/control-plane@main" + python -m pip install -e ".[dev,contract]" - name: Ruff check run: python -m ruff check src tests examples diff --git a/CHANGELOG.md b/CHANGELOG.md index 439e1c9..dda532a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,9 +39,46 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 into `build_governor`/`ControlPlaneClient`; `tests/test_ledger_backend_mode.py` exercises it directly against `FakeLedgerBackend`, including two `Ledger` instances sharing one backend (the cross-process case). +- `tokenops.control.dev_plane` — launches a real `agentplane-control-plane` on a real + localhost TCP port, in-process. Used by `tokenops.demo` (below) and by + `tests/conftest.py::live_plane_url` for tests that must exercise + `ControlPlaneClient.from_env()` itself (an in-process ASGI app isn't reachable that + way — `from_env()` builds its own plain `httpx.Client`). ### Changed +- **tokenops has no ledger of its own anymore (#118) — `ControlPlaneClient.from_env()` + requires `CONTROL_PLANE_URL`/`TOKENOPS_URL` and raises if neither is set.** There is + no more `TOKENOPS_EMBEDDED` env var and no code path left that silently falls back to + a local SQLite ledger; `build_governor`/`tokenops_run` construct `Ledger(backend=...)` + (an `HttpLedgerBackend`, i.e. real `precheck`/`events:batch` traffic to the plane) for + every live run. The `store=` constructor kwarg on `ControlPlaneClient`/`tokenops_run` + remains as an explicit, visible test-only escape hatch (dependency injection for unit + tests that don't want a running plane) — it is never reachable from `from_env()`, so + no environment misconfiguration can select it. +- `should_mount_run_registration()` always returns `False` now — registration is + always centralized on the plane; there's no embedded mode left for an agent to + self-host `POST /v1/runs` under. +- `tokenops.demo` (`python -m tokenops.demo`) launches a real control plane in-process + (`tokenops.control.dev_plane`) instead of using the now-removed embedded ledger, and + configures its budget/policy over the plane's own HTTP API (`PUT /v1/budgets` / + `PUT /v1/policies`) — the zero-setup promise holds, but needs + `agentplane-control-plane` importable (`pip install "agent-tokenops[contract]"` today, + or once released; otherwise point `CONTROL_PLANE_URL` at a plane you're already + running). +- CI now installs `agentplane-control-plane` straight from the control-plane repo + (`git+https://github.com/theagentplane/control-plane@main`) in addition to + `.[dev,contract]` — the `[contract]` tests and the `tests/examples/` e2e suite + (`tests/conftest.py::live_plane_url`) now actually run a real control plane in CI + instead of silently skipping; they're load-bearing coverage now, not optional. +- `tests/examples/test_bench_e2e.py` and `tests/examples/test_triad_e2e.py` now + configure their policies/budgets on a real, in-process control plane + (`live_plane_url`) over its own HTTP API (`PUT /v1/budgets` / `PUT /v1/policies`) + instead of a local `Store` under `TOKENOPS_EMBEDDED=1` — these are now the concrete + demonstration that governance policies configured on the plane reach a live, + multi-agent run and HALT/steer it (step_cap, cost_budget, output_runaway CANCEL+RETRY, + tool_output_cap deep swap), not just that the plumbing compiles. + - `Ledger`'s `RunState` renamed `LocalRunState` (#118, locked decision #9) — makes the two-tier model explicit ahead of the `LedgerBackend` rewire: this is the per-process Tier-1 cache, not the plane's authoritative `run_state`. diff --git a/README.md b/README.md index 037a4d8..afd3882 100644 --- a/README.md +++ b/README.md @@ -184,8 +184,8 @@ export TOKENOPS_URL=http://localhost:7700 export TOKENOPS_DB=tokenops.db # plane and every agent read the same file ``` -> `TOKENOPS_EMBEDDED=1` overrides `TOKENOPS_URL`. Leave it unset here, or each -> process silently falls back to its own local ledger and gets the full budget. +> tokenops has no ledger of its own — every process governs against the plane at +> `TOKENOPS_URL`. There is no local-fallback env var to accidentally leave unset. PyPI name is `agent-tokenops`; the import is `tokenops`. Extras: `pip install "agent-tokenops[examples]"` for the LangChain benches, @@ -251,22 +251,15 @@ Chronicle records decision boundaries; TokenOps attaches as the cost/governance | Variable | Purpose | |---|---| -| `TOKENOPS_URL` | Remote plane base URL (e.g. `http://localhost:7700`) → HTTP `register_run` | -| `TOKENOPS_EMBEDDED` | Set to `1` to force in-process `Store` (tests / single-process) | -| `TOKENOPS_DB` | SQLite path shared by plane + agents | +| `TOKENOPS_URL` (or `CONTROL_PLANE_URL`) | The control plane's base URL (e.g. `http://localhost:8800`) — **required** | | `TOKENOPS_CONFIG` | YAML for governance seed (core: `src/tokenops/config/default.yaml`) | -`TOKENOPS_URL` also accepts the aliases `CONTROL_PLANE_URL` and +tokenops has no ledger or run registry of its own — `ControlPlaneClient.from_env()` +raises if `TOKENOPS_URL`/`CONTROL_PLANE_URL` isn't set, rather than silently falling +back to anything local. Every process talks to the same plane, so they share one +budget by construction. `TOKENOPS_URL` also accepts the alias `TOKENOPS_CONTROL_PLANE_URL`. -Production / multi-process: set `TOKENOPS_URL`; agents must **not** mount `/v1/runs`. Tests: `TOKENOPS_EMBEDDED=1` (or omit URL). - -> **Precedence.** `ControlPlaneClient.from_env` takes the HTTP path only when a -> URL is set **and** `TOKENOPS_EMBEDDED` is not `1`. Setting both falls back to a -> local SQLite file with no warning, and every process then gets its own full -> budget. Check with -> `print("embedded" if client.embedded else client.url)`. -
diff --git a/docs/control-plane-deploy.md b/docs/control-plane-deploy.md index cedac1f..7ad5ebd 100644 --- a/docs/control-plane-deploy.md +++ b/docs/control-plane-deploy.md @@ -17,10 +17,9 @@ Agents in your app (or `examples/`) set `TOKENOPS_URL=http://tokenops:7700` so t | Var | Meaning | |-----|---------| -| `TOKENOPS_URL` | Plane base URL for `ControlPlaneClient` | +| `TOKENOPS_URL` | Plane base URL for `ControlPlaneClient` — **required**; tokenops has no local-ledger fallback | | `TOKENOPS_DB` | Shared SQLite path | | `TOKENOPS_CONFIG` | Governance seed YAML | -| `TOKENOPS_EMBEDDED=1` | Force in-process Store (tests) | ## Run compose (plane only) diff --git a/docs/guides/onboarding.md b/docs/guides/onboarding.md index 3c99f44..98437fe 100644 --- a/docs/guides/onboarding.md +++ b/docs/guides/onboarding.md @@ -31,7 +31,7 @@ Chronicle records decision boundaries; TokenOps observes them for cost/governanc | **Python 3.10+** | Required | | **`pip install agent-tokenops`** | Import is still `tokenops`. Pulls Chronicle ≥0.2.0, FastAPI, httpx, provider clients, etc. | | **Control plane + shared DB** *(multi-process)* | `TOKENOPS_URL` (e.g. `http://localhost:7700`) and `TOKENOPS_DB` shared by plane + agents. Seed governance once (`make db-reset`). | -| **Or embedded Store** *(single-process / tests)* | Omit `TOKENOPS_URL` or set `TOKENOPS_EMBEDDED=1`. | +| **Or a test-only local `Store`** *(no running plane needed)* | `tokenops_run(store=...)` / `ControlPlaneClient(store=...)` — never the default; tokenops has no env var that selects a local ledger. | | **LLM API keys** | Only for real model calls — not required for TokenOps itself or offline tests. | | **FastAPI** | Only if you use `instrument_app`. Non-FastAPI: pass kwargs / `RequestContext` to `tokenops_run`. | | **Chronicle `@boundary`** | Tools *and* LLM: `kind="llm"` runs pre_call via `on_enter`; tools stay observe-only. LLM-only stacks can use bare `@boundary(..., kind="llm")` under `tokenops_run` instead of `wrap_complete`. | @@ -135,16 +135,19 @@ The **agent** (via `instrument_app` / `tokenops_run` kwargs), not the UI. Client - **LLM calls:** `@boundary(..., kind="llm")` under `tokenops_run` runs **pre_call** (via `on_enter`) and observe — enough for LLM-only stacks without `wrap_complete`. - **Tools** (search, fetch, etc.): still need `@boundary` + the crossing hook so they appear on the ledger / are governed. Without that, TokenOps only sees LLM crossings you decorate (or put through `wrap_complete`). -### Embedded Store vs `TOKENOPS_URL`? +### Do I need a control plane running? -| Mode | When | -|------|------| -| `TOKENOPS_URL` set | Production / multi-process: register via HTTP; share `TOKENOPS_DB` with the plane. Agents must **not** mount `/v1/runs`. | -| `TOKENOPS_EMBEDDED=1` or no URL | Tests / single process: in-process `Store`. | +Yes, always — tokenops has no ledger or run registry of its own. `ControlPlaneClient.from_env()` +raises if `TOKENOPS_URL`/`CONTROL_PLANE_URL` isn't set. Run one locally (`control-plane serve`, +part of the `agentplane-control-plane` package, or its Docker image) and point every agent +process at it; they'll share one budget by construction. Tests that don't want a real plane +running use `tokenops_run(store=...)` / `ControlPlaneClient(store=...)` with a local `Store` — +an explicit, visible test double, not a production mode. ### Do I construct `Store(...)` in the agent? -Prefer `ControlPlaneClient.from_env()` and `tokenops_run`. Happy path does not require user-facing `Store(...)` construction. +No — `ControlPlaneClient.from_env()` and `tokenops_run()` are the only supported production +path. `Store(...)` is a test-only construct for exercising policy logic without a running plane. ### Tools usually cost $0 — why govern them? diff --git a/examples/a2a/client.py b/examples/a2a/client.py index 9d8f92f..25ef49f 100644 --- a/examples/a2a/client.py +++ b/examples/a2a/client.py @@ -1,7 +1,5 @@ from __future__ import annotations -import os - from examples.a2a import messages from examples.a2a.messages import parse_findings, parse_steps, parse_token_usage, summarize_request from examples.a2a.server import fetch_agent_card, fetch_agent_card_sync, post_task, post_task_sync @@ -53,11 +51,6 @@ def submit_task_sync_with_meta( Research registers the run when ``X-TokenOps-Run-Id`` is absent — clients should not call ``/v1/runs`` themselves for the default Chat / bench flow. """ - if ( - not (os.environ.get("TOKENOPS_URL") or "").strip() - and os.environ.get("TOKENOPS_EMBEDDED") != "1" - ): - os.environ.setdefault("TOKENOPS_EMBEDDED", "1") payload = messages.task_request( task=task, bench={"corpus_profile": corpus_profile}, intent=intent ) @@ -86,11 +79,6 @@ async def submit_task( intent: str = "", user_dims: dict[str, str] | None = None, ) -> RunResult: - if ( - not (os.environ.get("TOKENOPS_URL") or "").strip() - and os.environ.get("TOKENOPS_EMBEDDED") != "1" - ): - os.environ.setdefault("TOKENOPS_EMBEDDED", "1") payload = messages.task_request( task=task, bench={"corpus_profile": corpus_profile}, intent=intent ) diff --git a/examples/brief/client.py b/examples/brief/client.py index 7c3430c..4b86aab 100644 --- a/examples/brief/client.py +++ b/examples/brief/client.py @@ -2,8 +2,6 @@ from __future__ import annotations -import os - from examples.a2a.messages import parse_findings, parse_steps, parse_token_usage, task_request from examples.a2a.server import post_task, post_task_sync from examples.agents.types import Finding, RunResult, StepEvent, TokenUsage @@ -49,11 +47,6 @@ def submit_brief_sync_with_meta( governance_mode: GovernanceMode = GovernanceMode.ENFORCE, ) -> tuple[RunResult, dict[str, object]]: """POST the topic to Scout (entry). Scout registers the run when run_id is omitted.""" - if ( - not (os.environ.get("TOKENOPS_URL") or "").strip() - and os.environ.get("TOKENOPS_EMBEDDED") != "1" - ): - os.environ.setdefault("TOKENOPS_EMBEDDED", "1") payload = task_request(task=topic, bench={"corpus_profile": corpus_profile}, intent=intent) if user_dims: payload["user_dims"] = user_dims diff --git a/examples/triad/client.py b/examples/triad/client.py index 36f237c..cab5e40 100644 --- a/examples/triad/client.py +++ b/examples/triad/client.py @@ -2,8 +2,6 @@ from __future__ import annotations -import os - from examples.a2a.messages import parse_findings, parse_steps, parse_token_usage, task_request from examples.a2a.server import post_task, post_task_sync from examples.agents.types import Finding, RunResult, StepEvent, TokenUsage @@ -67,11 +65,6 @@ def submit_goal_sync_with_meta( The Planner registers the run on the control plane when ``X-TokenOps-Run-Id`` is absent — clients should not call ``/v1/runs`` themselves for the triad UI. """ - if ( - not (os.environ.get("TOKENOPS_URL") or "").strip() - and os.environ.get("TOKENOPS_EMBEDDED") != "1" - ): - os.environ.setdefault("TOKENOPS_EMBEDDED", "1") payload = task_request(task=goal, bench={"corpus_profile": corpus_profile}, intent=intent) if user_dims: payload["user_dims"] = user_dims diff --git a/src/tokenops/control/client.py b/src/tokenops/control/client.py index bf0ba79..7a50e1e 100644 --- a/src/tokenops/control/client.py +++ b/src/tokenops/control/client.py @@ -1,13 +1,18 @@ """SDK client for the TokenOps control plane. -Prefer :class:`ControlPlaneClient` over posting ``/v1/runs`` at an agent URL. -When ``TOKENOPS_URL`` is set, registration (and future plane APIs) go over HTTP. -When ``TOKENOPS_EMBEDDED=1`` or no URL is set, the client uses an in-process -:class:`~tokenops.control.store.Store` (shared SQLite via ``TOKENOPS_DB``). - -Agents should talk to the plane through this client (§6) — do not construct -``Store(TOKENOPS_DB)`` on the happy path. ``require_store()`` is an escape hatch -for ledger / dashboard rows until those APIs are fully remote. +``ControlPlaneClient`` is the only supported way a live agent reaches the plane — +``from_env()`` requires ``CONTROL_PLANE_URL`` / ``TOKENOPS_URL`` and raises if neither +is set. There is no embedded / local-file fallback: tokenops has no ledger of its own +(tokenops#118, the remote-only rewrite) — every spend/inflight/halt read or write goes +to the control plane over HTTP via :meth:`backend` (a +:class:`~tokenops.control.ledger_backend.LedgerBackend`), and registration/governance +config/dashboard rows go through :class:`~tokenops.control.http_store.HttpStore`. Both +are HTTP-only; neither ever opens a local SQLite file. + +The ``store=`` constructor kwarg is a explicit, visible test-only escape hatch (dependency +injection for unit/integration tests that want to exercise policy logic against a local +:class:`~tokenops.control.store.Store` without a running plane) — it is never reachable +from ``from_env()``, so no environment misconfiguration can silently select it. """ from __future__ import annotations @@ -17,8 +22,11 @@ from collections.abc import Mapping from typing import Any, cast +import httpx + from tokenops.control.http import post_run, post_run_sync from tokenops.control.http_store import HttpStore +from tokenops.control.ledger_backend import HttpLedgerBackend, LedgerBackend from tokenops.control.models import ( GovernanceMode, RunAlreadyRegisteredError, @@ -46,7 +54,7 @@ def _mode_value(mode: GovernanceMode | str | None) -> GovernanceMode: class ControlPlaneClient: - """Talk to a remote control plane or an embedded Store.""" + """Talk to the remote control plane over HTTP (or, in tests only, a local Store).""" def __init__( self, @@ -54,36 +62,43 @@ def __init__( url: str | None = None, store: Store | None = None, timeout: float = 30.0, + client: httpx.Client | None = None, ) -> None: if bool(url) == bool(store): raise ValueError("exactly one of url or store is required") self._url = url.rstrip("/") if url else None self._store = store self._hybrid_store: Any = None + self._backend: LedgerBackend | None = None self._timeout = timeout + self._client = client # optional injected httpx.Client — tests only @classmethod def from_env(cls, *, timeout: float = 30.0) -> ControlPlaneClient: - """Build a client from ``TOKENOPS_URL`` / ``TOKENOPS_EMBEDDED`` / ``TOKENOPS_DB``. + """Build a client from ``CONTROL_PLANE_URL`` / ``TOKENOPS_URL``. - * ``TOKENOPS_URL`` or ``CONTROL_PLANE_URL`` set and ``TOKENOPS_EMBEDDED`` not ``1`` - → HTTP to the plane (no local SQLite). - * otherwise → embedded :class:`Store` at ``TOKENOPS_DB``. + Raises if neither is set — tokenops has no embedded / local-file ledger to fall + back to (tokenops#118). Point this at a running control plane: ``control-plane + serve`` locally, or the Docker image, for anything other than tests that pass + ``store=`` explicitly. """ from tokenops.control.crossing import install_crossing_hook install_crossing_hook() - embedded = os.environ.get("TOKENOPS_EMBEDDED", "").strip() == "1" url = ( os.environ.get("CONTROL_PLANE_URL") or os.environ.get("TOKENOPS_URL") or os.environ.get("TOKENOPS_CONTROL_PLANE_URL") or "" ).strip() - if url and not embedded: - return cls(url=url, timeout=timeout) - db = os.environ.get("TOKENOPS_DB", "tokenops.db") - return cls(store=Store(db), timeout=timeout) + if not url: + raise RuntimeError( + "CONTROL_PLANE_URL (or TOKENOPS_URL) is required — tokenops has no " + "embedded/local ledger; point it at a running control plane. Run " + "`control-plane serve` locally (see the agentplane-control-plane " + "package) or use its Docker image, then set CONTROL_PLANE_URL to it." + ) + return cls(url=url, timeout=timeout) @property def url(self) -> str | None: @@ -91,29 +106,51 @@ def url(self) -> str | None: @property def embedded(self) -> bool: + """``True`` only for the explicit test-only ``store=`` constructor path. + + Never ``True`` for a client built by :meth:`from_env` — there is no + environment variable that selects a local ledger. + """ return self._store is not None @property def store(self) -> Store | None: - """Embedded registration Store, or ``None`` when registration is remote. - - Prefer :meth:`governance_config_for`, :meth:`resolve_run`, and - :meth:`require_store` over using this directly. Escape hatch only (§6). - """ + """The explicit test-only local Store, or ``None`` in the (only production) + remote mode. Prefer :meth:`backend`, :meth:`governance_config_for`, and + :meth:`resolve_run` over using this directly.""" return self._store + @property + def backend(self) -> LedgerBackend: + """The :class:`~tokenops.control.ledger_backend.LedgerBackend` for this client + — every spend/inflight/halt read (``precheck``) and write (``events:batch``) + goes through this. Only valid in remote mode (``from_env()`` always is).""" + if self._url is None: + raise RuntimeError( + "no LedgerBackend for a store=-constructed (test-only) ControlPlaneClient" + ) + if self._backend is None: + key = os.environ.get("CONTROL_PLANE_API_KEY") or os.environ.get("TOKENOPS_API_KEY") + self._backend = HttpLedgerBackend( + self._url, api_key=key, timeout=self._timeout, client=self._client + ) + return self._backend + def require_store(self) -> Store: - """Backing store for ledger / config / dashboard rows. + """Registration / governance-config / dashboard-row transport. - Embedded mode returns the in-process SQLite Store. Remote mode returns - :class:`HttpStore` — never a local DB file. + The test-only ``store=`` path returns that Store. Remote mode (``from_env()``) + returns :class:`HttpStore` — HTTP only, never a local DB file. Ledger accounting + (spend/inflight/halt) does **not** go through this — see :meth:`backend`. """ if self._store is not None: return self._store if self._hybrid_store is None: assert self._url is not None key = os.environ.get("CONTROL_PLANE_API_KEY") or os.environ.get("TOKENOPS_API_KEY") - self._hybrid_store = HttpStore(self._url, api_key=key, timeout=self._timeout) + self._hybrid_store = HttpStore( + self._url, api_key=key, timeout=self._timeout, client=self._client + ) return cast(Store, self._hybrid_store) def register_run( @@ -203,9 +240,9 @@ def update_run(self, run_id: str, **fields: Any) -> None: def should_mount_run_registration() -> bool: """Whether an agent app should expose ``POST /v1/runs``. - When ``TOKENOPS_URL`` points at a standalone plane (and embedded mode is off), - registration is centralized on the plane — agents must not mount the route. + Always ``False`` — registration is centralized on the plane (tokenops#118); there + is no embedded mode left for an agent to self-host it under. Kept (rather than + deleted outright) so existing ``if should_mount_run_registration(): ...`` call + sites keep working unchanged while they're cleaned up. """ - if os.environ.get("TOKENOPS_EMBEDDED", "").strip() == "1": - return True - return not (os.environ.get("TOKENOPS_URL") or "").strip() + return False diff --git a/src/tokenops/control/config.py b/src/tokenops/control/config.py index fc8d7de..2a4e9b6 100644 --- a/src/tokenops/control/config.py +++ b/src/tokenops/control/config.py @@ -35,6 +35,7 @@ from tokenops.control.engine import AgentControls, ApplyControls, Governor, PreviewControls from tokenops.control.ledger import Budget, Ledger, PriceFn +from tokenops.control.ledger_backend import LedgerBackend from tokenops.control.models import GovernanceMode from tokenops.control.policies import ( concurrency_cap, @@ -132,17 +133,21 @@ def build_governor( controls: AgentControls | None = None, *, store: Store | None = None, + backend: LedgerBackend | None = None, enforce: bool = True, ) -> Governor: """Build a fully-registered Governor (and its Ledger, on ``governor.ledger``) from a declarative governance config. ``config`` is the ``governance:`` block (or a mapping that contains ``budgets`` / ``policies``). - Pass ``store`` to share spend, inflight, and halt state across A2A agent processes.""" + Pass exactly one of ``store`` (local Store — tests only) or ``backend`` (a + :class:`~tokenops.control.ledger_backend.LedgerBackend` — the control plane, direct + or fake) to share spend, inflight, and halt state across A2A agent processes. + Neither means a fully in-memory, single-process Ledger.""" gov_cfg = config.get("governance", config) budgets = parse_budgets(gov_cfg.get("budgets")) - ledger = Ledger(budgets=list(budgets.values()), price=price, store=store) + ledger = Ledger(budgets=list(budgets.values()), price=price, store=store, backend=backend) governor = Governor(ledger, controls, enforce=enforce) ctx = _Ctx(budgets=budgets, price=price) @@ -181,11 +186,14 @@ def build_governance_stack( price: PriceFn, *, store: Store | None = None, + backend: LedgerBackend | None = None, mode: GovernanceMode = GovernanceMode.ENFORCE, ) -> tuple[Governor, ApplyControls | PreviewControls]: """Wire a Governor + OUT connector for enforce or preview mode.""" enforce = mode is not GovernanceMode.PREVIEW controls: ApplyControls | PreviewControls controls = ApplyControls() if enforce else PreviewControls() - governor = build_governor(config, price, controls, store=store, enforce=enforce) + governor = build_governor( + config, price, controls, store=store, backend=backend, enforce=enforce + ) return governor, controls diff --git a/src/tokenops/control/dev_plane.py b/src/tokenops/control/dev_plane.py new file mode 100644 index 0000000..b8168c6 --- /dev/null +++ b/src/tokenops/control/dev_plane.py @@ -0,0 +1,70 @@ +"""Launch a real ``agentplane-control-plane`` in-process, over a real TCP port. + +Not part of the public SDK surface. Used by ``tokenops.demo`` (so ``python -m +tokenops.demo`` still needs no external server — tokenops#118 removed the embedded +ledger, so the demo now launches a real, throwaway control plane instead) and by +``tests/conftest.py`` for the live end-to-end tests that need ``ControlPlaneClient`` +to talk real HTTP to something (an in-process ASGI ``TestClient`` isn't reachable +by an unrelated ``httpx.Client`` constructed deep inside ``ControlPlaneClient``). + +Requires ``agentplane-control-plane`` to be importable — raises ``ImportError`` +(caller's problem to handle) when it is not. +""" + +from __future__ import annotations + +import socket +import time +from collections.abc import Callable +from dataclasses import dataclass + + +@dataclass +class LivePlane: + url: str + stop: Callable[[], None] + + +def _free_port() -> int: + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.bind(("127.0.0.1", 0)) + port = s.getsockname()[1] + s.close() + return port + + +def launch(db_path: str, *, startup_timeout: float = 10.0) -> LivePlane: + """Start a real control-plane FastAPI app on a background thread, over a real + localhost TCP port. Returns the base URL and a ``stop()`` callback (idempotent + enough to call once; not designed to be called twice).""" + import threading + + import uvicorn + 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 + + store = SqliteStore(db_path, auto_seed=False) + envelopes = EnvelopeStore(db_path) + app = create_app(store=store, envelopes=envelopes, settings=Settings(db_path=db_path)) + + port = _free_port() + config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning") + server = uvicorn.Server(config) + thread = threading.Thread(target=server.run, daemon=True) + thread.start() + + deadline = time.time() + startup_timeout + while not server.started and time.time() < deadline: + time.sleep(0.02) + if not server.started: + raise RuntimeError("control plane did not start within the timeout") + + def stop() -> None: + server.should_exit = True + thread.join(timeout=5.0) + store.close() + envelopes.close() + + return LivePlane(url=f"http://127.0.0.1:{port}", stop=stop) diff --git a/src/tokenops/control/http_store.py b/src/tokenops/control/http_store.py index 50dee7b..36ed45d 100644 --- a/src/tokenops/control/http_store.py +++ b/src/tokenops/control/http_store.py @@ -23,15 +23,24 @@ class HttpStore: - def __init__(self, base_url: str, *, api_key: str | None = None, timeout: float = 30.0) -> None: + def __init__( + self, + base_url: str, + *, + api_key: str | None = None, + timeout: float = 30.0, + client: httpx.Client | None = None, + ) -> None: self.path = base_url.rstrip("/") + self._owns_client = client is None headers = {} if api_key: headers["Authorization"] = f"Bearer {api_key}" - self._client = httpx.Client(base_url=self.path, timeout=timeout, headers=headers) + self._client = client or httpx.Client(base_url=self.path, timeout=timeout, headers=headers) def close(self) -> None: - self._client.close() + if self._owns_client: + self._client.close() def _request(self, method: str, path: str, **kwargs: Any) -> httpx.Response: response = self._client.request(method, path, **kwargs) diff --git a/src/tokenops/control/run.py b/src/tokenops/control/run.py index 50ec6af..3b66a09 100644 --- a/src/tokenops/control/run.py +++ b/src/tokenops/control/run.py @@ -220,17 +220,27 @@ def tokenops_run( if governor is None: price_fn = price if price is not None else build_price_book() # §10: cached config (via Store / client); fresh Governor each run — no clone. - if store is not None: + # Ledger construction follows client_obj (not the local store= param): an + # embedded test client built from an explicit store=, whether passed to + # tokenops_run() directly or already embedded in a client= the caller built, + # keeps using that Store. Everything else (from_env(), always remote — there + # is no local ledger to fall back to, tokenops#118) uses the plane's + # LedgerBackend (precheck/events:batch) — never store_obj, which in that case + # is registration/dashboard HTTP only (HttpStore). + if client_obj.embedded: config = store_obj.governance_config_for(svc) + ledger_store, ledger_backend = store_obj, None else: config = client_obj.governance_config_for(svc) + ledger_store, ledger_backend = None, client_obj.backend if controls is not None: enforce = gov_mode is not GovernanceMode.PREVIEW gov = build_governor( config, price_fn, controls, - store=store_obj, + store=ledger_store, + backend=ledger_backend, enforce=enforce, ) ctrl = controls @@ -238,7 +248,8 @@ def tokenops_run( gov, ctrl = build_governance_stack( config, price_fn, - store=store_obj, + store=ledger_store, + backend=ledger_backend, mode=gov_mode, ) else: diff --git a/src/tokenops/demo.py b/src/tokenops/demo.py index a23a75e..b6ec7d1 100644 --- a/src/tokenops/demo.py +++ b/src/tokenops/demo.py @@ -3,8 +3,17 @@ python -m tokenops.demo Runs the same agent loop twice, once ungoverned and once governed, so the -difference is a number rather than a claim. No API keys, no server, no Docker: -the "model" is a local stub, so this costs nothing and touches no network. +difference is a number rather than a claim. No API keys the "model" is a local +stub, so this costs nothing and touches no network for the model call itself. + +tokenops has no ledger of its own (tokenops#118) — every run governs against a +real control plane. If ``CONTROL_PLANE_URL``/``TOKENOPS_URL`` is already set, the +demo uses it. Otherwise it launches a real, throwaway ``agentplane-control-plane`` +in-process (see ``tokenops.control.dev_plane``) so the zero-setup promise holds; +that needs ``agentplane-control-plane`` importable (``pip install +agentplane-control-plane`` once released, or ``pip install +"agent-tokenops[contract]"`` today) — without it, point ``CONTROL_PLANE_URL`` at a +control plane you're already running (``control-plane serve``, or its Docker image). The governed half is the real integration. Swap ``fake_llm`` for ``tokenops.providers.complete`` and it is exactly what you would ship. @@ -13,17 +22,49 @@ from __future__ import annotations import os +import sys +import tempfile +from collections.abc import Callable from tokenops.providers.types import ModelResponse -# Demo runs in-process. Point TOKENOPS_URL at a control plane instead when -# several agent processes have to share one budget. -os.environ.setdefault("TOKENOPS_EMBEDDED", "1") - BUDGET_MICROS = 2_000_000 # $2.00, from src/tokenops/config/default.yaml MAX_CALLS = 40 +def _ensure_control_plane_url() -> tuple[str | None, Callable[[], None] | None]: + """Return ``(url_to_restore_on_exit, stop_callable_or_None)``. + + If a control plane is already configured, do nothing. Otherwise launch one + in-process and point ``CONTROL_PLANE_URL`` at it for the life of this process. + """ + existing = ( + os.environ.get("CONTROL_PLANE_URL") + or os.environ.get("TOKENOPS_URL") + or os.environ.get("TOKENOPS_CONTROL_PLANE_URL") + or "" + ).strip() + if existing: + return None, None + try: + from tokenops.control.dev_plane import launch + except ImportError: + print( + "No CONTROL_PLANE_URL is set, and agentplane-control-plane isn't " + "installed to launch one for you.\n" + "Either:\n" + ' pip install "agent-tokenops[contract]" # launches one automatically\n' + "or run one yourself (control-plane serve / its Docker image) and set\n" + " CONTROL_PLANE_URL=http://localhost:8800\n", + file=sys.stderr, + ) + sys.exit(1) + db = os.path.join(tempfile.mkdtemp(prefix="tokenops-demo-"), "control-plane.db") + plane = launch(db) + os.environ["CONTROL_PLANE_URL"] = plane.url + return plane.url, plane.stop + + def _fake_llm_factory(): """A stand-in model. Reports the token usage TokenOps prices.""" calls = {"n": 0} @@ -85,28 +126,68 @@ def run_governed() -> tuple[int, int, str]: return ledger.cost_micros(run_id), MAX_CALLS, "never halted" +def _seed_governance(url: str) -> None: + """Configure the same budget + policy the README quotes, via the plane's own + HTTP API — not a file the plane happens to read, so this is exactly what a + real deployment would do to configure governance.""" + import httpx + + httpx.put( + f"{url}/v1/budgets", + json={ + "id": "run_llm_cap", + "limit_micros": BUDGET_MICROS, + "dimension": "run", + "period": "lifetime", + }, + timeout=10, + ).raise_for_status() + httpx.put( + f"{url}/v1/policies", + json={ + "id": "demo_cost_budget", + "template": "cost_budget", + "params": {}, + "budget_id": "run_llm_cap", + "agent": None, + "segment_id": None, + "enabled": True, + "data_scope": "local", + }, + timeout=10, + ).raise_for_status() + + def main() -> None: dollars = lambda micros: f"${micros / 1e6:,.2f}" # noqa: E731 - print( - f"\nAn agent makes {MAX_CALLS} model calls. Budget for the whole run: " - f"{dollars(BUDGET_MICROS)}.\n" - ) - - ungoverned = run_ungoverned() - print(f" without TokenOps {MAX_CALLS} calls run, spend {dollars(ungoverned)}") - - spent, stopped_at, reason = run_governed() - print(f" with TokenOps halted at call {stopped_at}, spend {dollars(spent)}") - - saved = ungoverned - spent - print(f"\n {dollars(saved)} not spent. The run stopped itself.") - print(f" reason: {reason}\n") - print("No single call was expensive. Together they crossed the cap, which is") - print("what a per-request limit cannot see.\n") - print( - "Put this in your own agent: https://github.com/theagentplane/tokenops#2-put-it-in-your-agent\n" - ) + url, stop_plane = _ensure_control_plane_url() + try: + if url is not None: + _seed_governance(url) + + print( + f"\nAn agent makes {MAX_CALLS} model calls. Budget for the whole run: " + f"{dollars(BUDGET_MICROS)}.\n" + ) + + ungoverned = run_ungoverned() + print(f" without TokenOps {MAX_CALLS} calls run, spend {dollars(ungoverned)}") + + spent, stopped_at, reason = run_governed() + print(f" with TokenOps halted at call {stopped_at}, spend {dollars(spent)}") + + saved = ungoverned - spent + print(f"\n {dollars(saved)} not spent. The run stopped itself.") + print(f" reason: {reason}\n") + print("No single call was expensive. Together they crossed the cap, which is") + print("what a per-request limit cannot see.\n") + print( + "Put this in your own agent: https://github.com/theagentplane/tokenops#2-put-it-in-your-agent\n" + ) + finally: + if stop_plane is not None: + stop_plane() if __name__ == "__main__": diff --git a/tests/conftest.py b/tests/conftest.py index 74bf8e5..b939575 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -179,6 +179,27 @@ def http_backend(plane_app_factory): client.close() +@pytest.fixture +def live_plane_url(monkeypatch): + """A real control plane on a real localhost TCP port, with ``CONTROL_PLANE_URL`` + pointed at it — for tests that must exercise ``ControlPlaneClient.from_env()`` + itself (no ``store=``/injected-client escape hatch reaches ``from_env()`` by + design — tokenops#118: no env var may select a local ledger). An in-process ASGI + app (``plane_app_factory``/``http_backend``) isn't reachable this way since + ``from_env()`` builds its own plain ``httpx.Client(base_url=...)``. + """ + pytest.importorskip("control_plane", reason="install agentplane-control-plane>=0.2.0") + from tokenops.control.dev_plane import launch + + db = str(Path(tempfile.mkdtemp()) / "cp.db") + plane = launch(db) + monkeypatch.setenv("CONTROL_PLANE_URL", plane.url) + monkeypatch.delenv("TOKENOPS_URL", raising=False) + monkeypatch.delenv("TOKENOPS_CONTROL_PLANE_URL", raising=False) + yield plane.url + plane.stop() + + @pytest.fixture(params=["fake", "http"]) def any_backend(request): """Parametrised over both backends — for the verified-fake contract suite. diff --git a/tests/examples/test_bench_e2e.py b/tests/examples/test_bench_e2e.py index d84582c..ed952d6 100644 --- a/tests/examples/test_bench_e2e.py +++ b/tests/examples/test_bench_e2e.py @@ -1,12 +1,19 @@ -"""End-to-end on the real A2A research bench (FastAPI TestClient). - -Drives the full live path — entry registers run on task → per-run governor → actuators → -RunRecord — for four policies. Only the model call and the search tool are faked (no API -key, no network); the server, governor, ledger, store, and agent loop are all real. +"""End-to-end on the real A2A research bench (FastAPI TestClient) against a real, +in-process control plane (real TCP port — tokenops#118: tokenops has no ledger of +its own, so this is the only kind of "real" left to test against). + +Drives the full live path — entry registers run on task → per-run governor (reading +policies the *plane* was configured with over its own HTTP API) → actuators → +RunRecord (persisted on the plane) — for four policies. Only the model call and the +search tool are faked (no API key, no network to a real LLM); the server, control +plane, governor, ledger, and agent loop are all real. """ from __future__ import annotations +import dataclasses + +import httpx import pytest pytestmark = pytest.mark.e2e @@ -16,8 +23,8 @@ from examples.agents.research.tools.core import SearchResult from fastapi.testclient import TestClient +from tokenops.control.http_store import HttpStore from tokenops.control.models import BudgetSpec, PolicyInstance -from tokenops.control.store import Store from tokenops.providers.types import ModelResponse @@ -37,17 +44,21 @@ def _run(client, *, intent="demo", user_dims=None): return resp.json() -def _client(monkeypatch, tmp_path, policies, budgets=(), model=None): - db = str(tmp_path / "bench.db") - monkeypatch.setenv("TOKENOPS_DB", db) - monkeypatch.delenv("TOKENOPS_URL", raising=False) - monkeypatch.setenv("TOKENOPS_EMBEDDED", "1") - s = Store(db) +def _seed(plane_url: str, *, policies=(), budgets=()) -> None: + """Configure the plane exactly the way a real deployment would — over its own + HTTP API, not by touching whatever storage happens to sit behind it.""" for b in budgets: - s.upsert_budget(b) + httpx.put( + f"{plane_url}/v1/budgets", json=dataclasses.asdict(b), timeout=10 + ).raise_for_status() for pi in policies: - s.upsert_policy_instance(pi) - s.close() + httpx.put( + f"{plane_url}/v1/policies", json=dataclasses.asdict(pi), timeout=10 + ).raise_for_status() + + +def _client(live_plane_url, monkeypatch, policies, budgets=(), model=None): + _seed(live_plane_url, policies=policies, budgets=budgets) monkeypatch.setattr(search_core, "search", _search) from examples.agents.research.native import server as srv @@ -69,11 +80,11 @@ def _always_search(provider, model, messages, max_output_tokens=None, **kw): ) -# 1) step_cap → HALT -def test_step_cap_halts(monkeypatch, tmp_path): +# 1) step_cap, pulled from the control plane's own /v1/policies → HALT +def test_step_cap_halts(live_plane_url, monkeypatch): srv, client = _client( + live_plane_url, monkeypatch, - tmp_path, [PolicyInstance(id="p", template="step_cap", params={"max_steps": 2}, agent="research")], model=_always_search, ) @@ -82,11 +93,11 @@ def test_step_cap_halts(monkeypatch, tmp_path): assert body["cost_micros"] > 0 -# 2) cost_budget → HALT -def test_cost_budget_halts(monkeypatch, tmp_path): +# 2) cost_budget, budget + policy both configured on the plane → HALT +def test_cost_budget_halts(live_plane_url, monkeypatch): srv, client = _client( + live_plane_url, monkeypatch, - tmp_path, [PolicyInstance(id="p", template="cost_budget", budget_id="cap", agent="research")], budgets=[ BudgetSpec(id="cap", limit_micros=250, dimension="run") @@ -98,7 +109,7 @@ def test_cost_budget_halts(monkeypatch, tmp_path): # 3) output_runaway via streaming → CANCEL + RETRY → run completes -def test_cancel_retry_streaming(monkeypatch, tmp_path): +def test_cancel_retry_streaming(live_plane_url, monkeypatch): monkeypatch.setenv("TOKENOPS_STREAM", "1") calls = {"n": 0} @@ -120,8 +131,8 @@ def fake_stream( yield '{"action": "finish"}' srv, client = _client( + live_plane_url, monkeypatch, - tmp_path, [ PolicyInstance( id="p", @@ -139,7 +150,7 @@ def fake_stream( # 4) tool_output_cap deep swap → the oversized result is replaced by the descriptor -def test_tool_output_cap_substitutes_result(monkeypatch, tmp_path): +def test_tool_output_cap_substitutes_result(live_plane_url, monkeypatch): big = "x" * 60_000 def search_then_finish(provider, model, messages, max_output_tokens=None, **kw): @@ -153,8 +164,8 @@ def search_then_finish(provider, model, messages, max_output_tokens=None, **kw): ) srv, client = _client( + live_plane_url, monkeypatch, - tmp_path, [ PolicyInstance( id="p", template="tool_output_cap", params={"cap_tokens": 1000}, agent="research" @@ -182,22 +193,17 @@ def search_then_finish(provider, model, messages, max_output_tokens=None, **kw): # backbone, and the attribution-hardening boundary — an untrusted request body # must not be able to inject arbitrary segmentation tags; see # attribution._PAYLOAD_USER_DIM_ALLOWLIST, commit 28337d1 "Harden ..."). -def test_run_dims_only_allowlisted_payload_keys_persist(monkeypatch, tmp_path): - import os - - from tokenops.control.store import Store - +def test_run_dims_only_allowlisted_payload_keys_persist(live_plane_url, monkeypatch): srv, client = _client( + live_plane_url, monkeypatch, - tmp_path, [PolicyInstance(id="p", template="step_cap", params={"max_steps": 2}, agent="research")], model=_always_search, ) body = _run(client, user_dims={"user_id": "alice", "team": "growth"}) run_id = body["run_id"] - s = Store(os.environ["TOKENOPS_DB"], auto_seed=False) - rec = s.get_run(run_id) + hs = HttpStore(live_plane_url) + rec = hs.get_run(run_id) assert rec.dims.get("user_id") == "alice" # allow-listed payload key persists assert "team" not in rec.dims # arbitrary payload tag must not leak into segmentation - assert "team" not in s.run_tag_keys() - s.close() + hs.close() diff --git a/tests/examples/test_triad_e2e.py b/tests/examples/test_triad_e2e.py index 269bce9..fe1b6f6 100644 --- a/tests/examples/test_triad_e2e.py +++ b/tests/examples/test_triad_e2e.py @@ -1,13 +1,18 @@ -"""End-to-end triad bench (Planner → Researcher → Writer) with mocked LLMs. +"""End-to-end triad bench (Planner → Researcher → Writer) with mocked LLMs, against a +real, in-process control plane (real TCP port — tokenops#118: tokenops has no ledger +of its own, so a live plane is the only thing left to test "real" against). -Drives the real FastAPI handlers, shared Store/ledger, wrap_complete, crossing -hook, and A2A delegates. Only ``complete`` and the search tool are faked. +Drives the real FastAPI handlers, shared ledger (via the plane's precheck/events:batch), +wrap_complete, crossing hook, and A2A delegates. Only ``complete`` and the search tool +are faked. """ from __future__ import annotations +import dataclasses import json +import httpx import pytest pytestmark = pytest.mark.e2e @@ -20,8 +25,8 @@ from tokenops.control.client import ControlPlaneClient from tokenops.control.context import RUN_ID_HEADER +from tokenops.control.http_store import HttpStore from tokenops.control.models import BudgetSpec, PolicyInstance -from tokenops.control.store import Store from tokenops.providers.types import ModelResponse @@ -71,22 +76,18 @@ def _write_answer(provider, model, messages, max_output_tokens=None, **kw): ) -def _seed(tmp_path, monkeypatch, policies, budgets=()): - db = str(tmp_path / "triad.db") - monkeypatch.setenv("TOKENOPS_DB", db) - monkeypatch.setenv("TOKENOPS_EMBEDDED", "1") - monkeypatch.delenv("TOKENOPS_URL", raising=False) - monkeypatch.setenv("TOKENOPS_CONFIG", "examples/config/triad.yaml") +def _seed(live_plane_url, monkeypatch, policies, budgets=()): monkeypatch.setenv("SEARCH_BACKEND", "corpus") - # Avoid YAML seed so test policies are the only ones present. - s = Store(db, auto_seed=False) + # A fresh plane per test already has no policies — nothing to avoid seeding. for b in budgets: - s.upsert_budget(b) + httpx.put( + f"{live_plane_url}/v1/budgets", json=dataclasses.asdict(b), timeout=10 + ).raise_for_status() for pi in policies: - s.upsert_policy_instance(pi) - s.close() + httpx.put( + f"{live_plane_url}/v1/policies", json=dataclasses.asdict(pi), timeout=10 + ).raise_for_status() monkeypatch.setattr(search_core, "search", _search) - return db def _wire_apps(monkeypatch): @@ -134,10 +135,10 @@ async def _route_post_async(url, payload, *, headers=None, timeout=300.0): return planner, researcher, writer -def test_triad_pipeline_completes_with_ledger(monkeypatch, tmp_path): +def test_triad_pipeline_completes_with_ledger(live_plane_url, monkeypatch): """UI path: POST /v1/tasks with no run_id — Planner registers the run.""" _seed( - tmp_path, + live_plane_url, monkeypatch, [ PolicyInstance(id="p-step", template="step_cap", params={"max_steps": 40}), @@ -172,27 +173,26 @@ def test_triad_pipeline_completes_with_ledger(monkeypatch, tmp_path): assert any(s["agent"] == "researcher" and s["action"] == "search" for s in body["steps"]) assert any(s["agent"] == "writer" for s in body["steps"]) - store = Store(tmp_path / "triad.db", auto_seed=False) - rec = store.get_run(body["run_id"]) + hs = HttpStore(live_plane_url) + rec = hs.get_run(body["run_id"]) assert rec is not None assert rec.status == "completed" - assert rec.cost_micros > 0 - reg = ( - store.get_run_registration(body["run_id"]) - if hasattr(store, "get_run_registration") - else store.resolve_run(body["run_id"]) - ) + # rec.cost_micros is NOT asserted here: control-plane 0.2.x deliberately drops + # client-PATCHed steps/cost_micros on the dashboard row ("derived server-side") + # but doesn't yet derive them from ledger events — the authoritative number is + # body["cost_micros"] (the live governed run's own ledger, asserted above). + reg = hs.resolve_run(body["run_id"]) # §1 hardening (28337d1): the agent's own intent (instrument_app(intent=...)) # wins over a payload-supplied one — a caller cannot spoof intent to dodge # intent-scoped governance. The planner hardcodes INTENT = "triad_plan". assert reg.intent == "triad_plan" - store.close() + hs.close() -def test_triad_cost_not_double_counted_without_parent_rollup(monkeypatch, tmp_path): +def test_triad_cost_not_double_counted_without_parent_rollup(live_plane_url, monkeypatch): """Child LLM spend is in the shared ledger once — parent must not re-add rollup.""" _seed( - tmp_path, + live_plane_url, monkeypatch, [ PolicyInstance(id="p-step", template="step_cap", params={"max_steps": 40}), @@ -226,15 +226,15 @@ def test_triad_cost_not_double_counted_without_parent_rollup(monkeypatch, tmp_pa # plan 160 + research 210 + write 260 = 630 (search tool is not LLM-priced) assert body["cost_micros"] == 630 # Same run_id must be shared (propagation + no soft orphan runs for children). - store = Store(tmp_path / "triad.db", auto_seed=False) - assert store.resolve_run(body["run_id"]).run_id == body["run_id"] - store.close() + hs = HttpStore(live_plane_url) + assert hs.resolve_run(body["run_id"]).run_id == body["run_id"] + hs.close() -def test_triad_per_agent_step_cap_only_on_researcher(monkeypatch, tmp_path): +def test_triad_per_agent_step_cap_only_on_researcher(live_plane_url, monkeypatch): """Governor config is filtered by agent name — researcher-only step_cap.""" _seed( - tmp_path, + live_plane_url, monkeypatch, [ PolicyInstance( @@ -283,9 +283,9 @@ def test_triad_per_agent_step_cap_only_on_researcher(monkeypatch, tmp_path): assert "step" in (body.get("halt_reason") or "").lower() -def test_triad_step_cap_halts(monkeypatch, tmp_path): +def test_triad_step_cap_halts(live_plane_url, monkeypatch): _seed( - tmp_path, + live_plane_url, monkeypatch, [PolicyInstance(id="p", template="step_cap", params={"max_steps": 2}, agent="researcher")], ) @@ -321,9 +321,9 @@ def test_triad_step_cap_halts(monkeypatch, tmp_path): assert body["cost_micros"] > 0 -def test_triad_cost_budget_halts_on_researcher(monkeypatch, tmp_path): +def test_triad_cost_budget_halts_on_researcher(live_plane_url, monkeypatch): _seed( - tmp_path, + live_plane_url, monkeypatch, [ PolicyInstance( diff --git a/tests/test_control_plane_app.py b/tests/test_control_plane_app.py index 5369b8d..f43ab14 100644 --- a/tests/test_control_plane_app.py +++ b/tests/test_control_plane_app.py @@ -61,14 +61,21 @@ def test_post_v1_runs_conflict(plane_client): assert conflict.status_code == 409 -def test_agent_skips_mount_when_tokenops_url(monkeypatch, tmp_path): - """With TOKENOPS_URL set, agents must not expose POST /v1/runs.""" - monkeypatch.setenv("TOKENOPS_DB", str(tmp_path / "r.db")) - monkeypatch.setenv("TOKENOPS_URL", "http://tokenops:7700") - monkeypatch.delenv("TOKENOPS_EMBEDDED", raising=False) +def test_should_mount_run_registration_is_always_false(monkeypatch): + """tokenops#118: registration is always centralized on the plane now — there is + no embedded mode left for an agent to self-host POST /v1/runs under, regardless + of what's in the environment.""" + monkeypatch.delenv("TOKENOPS_URL", raising=False) + assert should_mount_run_registration() is False + monkeypatch.setenv("TOKENOPS_URL", "http://tokenops:7700") assert should_mount_run_registration() is False + +def test_agent_skips_mount_when_should_mount_is_false(tmp_path): + """The (now-dead, since should_mount_run_registration() is always False) + if should_mount_run_registration(): mount_run_registration(...) call sites in the + example servers must not expose POST /v1/runs.""" store = Store(str(tmp_path / "r.db")) app = FastAPI() if should_mount_run_registration(): @@ -79,12 +86,10 @@ def test_agent_skips_mount_when_tokenops_url(monkeypatch, tmp_path): store.close() -def test_agent_mounts_when_embedded(monkeypatch, tmp_path): - monkeypatch.setenv("TOKENOPS_DB", str(tmp_path / "e.db")) - monkeypatch.delenv("TOKENOPS_URL", raising=False) - monkeypatch.setenv("TOKENOPS_EMBEDDED", "1") - - assert should_mount_run_registration() is True +def test_mount_run_registration_still_works_when_called_directly(tmp_path): + """mount_run_registration() itself is unconditional — a standalone deployment + (src/tokenops/server) can still call it directly without going through + should_mount_run_registration().""" store = Store(str(tmp_path / "e.db")) app = FastAPI() mount_run_registration(app, store) diff --git a/tests/test_control_plane_client.py b/tests/test_control_plane_client.py index 5d5dc9c..8cc5b81 100644 --- a/tests/test_control_plane_client.py +++ b/tests/test_control_plane_client.py @@ -1,58 +1,39 @@ -"""Unit tests for ControlPlaneClient (embedded Store path).""" +"""Unit tests for ControlPlaneClient (tokenops#118: remote-only, no embedded ledger).""" from __future__ import annotations import pytest from tokenops.control.client import ControlPlaneClient, should_mount_run_registration +from tokenops.control.ledger_backend import HttpLedgerBackend from tokenops.control.models import GovernanceMode, RunAlreadyRegisteredError from tokenops.control.store import Store -def test_from_env_embedded_when_no_url(monkeypatch, tmp_path): +def test_from_env_raises_without_a_url(monkeypatch): + monkeypatch.delenv("CONTROL_PLANE_URL", raising=False) monkeypatch.delenv("TOKENOPS_URL", raising=False) - monkeypatch.delenv("TOKENOPS_EMBEDDED", raising=False) - db = str(tmp_path / "c.db") - monkeypatch.setenv("TOKENOPS_DB", db) - - client = ControlPlaneClient.from_env() - assert client.embedded - assert client.url is None - - out = client.register_run( - intent="demo", - user_dims={"user_id": "alice"}, - mode="preview", - ) - assert out["status"] == "registered" - assert out["mode"] == "preview" - assert out["run_id"] - - store = Store(db, auto_seed=False) - reg = store.resolve_run(out["run_id"]) - assert reg.intent == "demo" - assert reg.user_dims["user_id"] == "alice" - assert reg.mode is GovernanceMode.PREVIEW - store.close() - - -def test_from_env_embedded_flag_overrides_url(monkeypatch, tmp_path): - monkeypatch.setenv("TOKENOPS_URL", "http://plane:7700") - monkeypatch.setenv("TOKENOPS_EMBEDDED", "1") - monkeypatch.setenv("TOKENOPS_DB", str(tmp_path / "e.db")) - - client = ControlPlaneClient.from_env() - assert client.embedded - out = client.register_run(intent="x", run_id="fixed-run") - assert out["run_id"] == "fixed-run" + monkeypatch.delenv("TOKENOPS_CONTROL_PLANE_URL", raising=False) + with pytest.raises(RuntimeError, match="CONTROL_PLANE_URL"): + ControlPlaneClient.from_env() def test_from_env_remote_when_url_set(monkeypatch): monkeypatch.setenv("TOKENOPS_URL", "http://localhost:7700/") - monkeypatch.delenv("TOKENOPS_EMBEDDED", raising=False) client = ControlPlaneClient.from_env() assert not client.embedded assert client.url == "http://localhost:7700" + assert isinstance(client.backend, HttpLedgerBackend) + + +def test_backend_raises_for_a_store_constructed_client(tmp_path): + """The test-only store= escape hatch has no LedgerBackend — fail closed rather + than silently returning something that would talk to a plane that isn't there.""" + store = Store(str(tmp_path / "s.db"), auto_seed=False) + client = ControlPlaneClient(store=store) + with pytest.raises(RuntimeError): + _ = client.backend + store.close() def test_register_run_duplicate_raises(tmp_path): @@ -76,13 +57,9 @@ def test_resolve_run_and_governance_config_for(tmp_path): store.close() -def test_should_mount_run_registration(monkeypatch): +def test_should_mount_run_registration_is_always_false(monkeypatch): monkeypatch.delenv("TOKENOPS_URL", raising=False) - monkeypatch.delenv("TOKENOPS_EMBEDDED", raising=False) - assert should_mount_run_registration() is True + assert should_mount_run_registration() is False monkeypatch.setenv("TOKENOPS_URL", "http://tokenops:7700") assert should_mount_run_registration() is False - - monkeypatch.setenv("TOKENOPS_EMBEDDED", "1") - assert should_mount_run_registration() is True diff --git a/tests/test_demo.py b/tests/test_demo.py index e45f347..9c5edbe 100644 --- a/tests/test_demo.py +++ b/tests/test_demo.py @@ -14,24 +14,25 @@ import sys from pathlib import Path +import pytest + REPO_ROOT = Path(__file__).resolve().parent.parent DEMO = REPO_ROOT / "src" / "tokenops" / "demo.py" def test_demo_shows_the_governed_run_halting(tmp_path): - """It runs standalone, spends, and stops itself. No API keys, no server.""" + """It runs standalone, spends, and stops itself. No API keys, no server to run + yourself — it launches a real, throwaway control plane in-process (tokenops#118: + no embedded ledger to fall back to instead).""" + pytest.importorskip("control_plane", reason="install agentplane-control-plane>=0.2.0") env = dict(os.environ) - # A user's shell has neither of these. conftest sets SKIP_GOVERNANCE_SEED for - # the suite, and inheriting it seeds no policies at all, so nothing enforces. + # A user's shell has none of these. conftest sets SKIP_GOVERNANCE_SEED for the + # suite; the demo seeds its own budget/policy via the plane's HTTP API regardless. + env.pop("CONTROL_PLANE_URL", None) env.pop("TOKENOPS_URL", None) env.pop("TOKENOPS_SKIP_GOVERNANCE_SEED", None) env.update( { - "TOKENOPS_EMBEDDED": "1", - "TOKENOPS_DB": str(tmp_path / "demo.db"), - # Pin the seed the README quotes. The suite and the Makefile both set - # TOKENOPS_CONFIG, and inheriting either changes the budget under test. - "TOKENOPS_CONFIG": str(REPO_ROOT / "src" / "tokenops" / "config" / "default.yaml"), # cp1252 is the default Windows console encoding; a non-ASCII byte in # any halt reason on this path would raise UnicodeEncodeError here. "PYTHONIOENCODING": "cp1252", diff --git a/tests/test_governance_config_cache.py b/tests/test_governance_config_cache.py index c91c81c..476e410 100644 --- a/tests/test_governance_config_cache.py +++ b/tests/test_governance_config_cache.py @@ -96,10 +96,6 @@ def counting(agent: str): def test_client_governance_config_for_uses_cache(tmp_path, monkeypatch): db = str(tmp_path / "client_cache.db") - monkeypatch.setenv("TOKENOPS_DB", db) - monkeypatch.setenv("TOKENOPS_EMBEDDED", "1") - monkeypatch.delenv("TOKENOPS_URL", raising=False) - store = Store(db, auto_seed=False) store.upsert_policy_instance( PolicyInstance(id="p1", template="step_cap", params={"max_steps": 2}, agent="writer"), diff --git a/tests/test_tokenops_run.py b/tests/test_tokenops_run.py index 41c85c2..1e30688 100644 --- a/tests/test_tokenops_run.py +++ b/tests/test_tokenops_run.py @@ -29,11 +29,10 @@ @pytest.fixture -def store(tmp_path, monkeypatch): +def store(tmp_path): + """Explicit test-injection Store — tokenops_run(store=store) / ControlPlaneClient + (store=store) — never reachable from ControlPlaneClient.from_env() (tokenops#118).""" db = str(tmp_path / "wave1.db") - monkeypatch.setenv("TOKENOPS_DB", db) - monkeypatch.delenv("TOKENOPS_URL", raising=False) - monkeypatch.setenv("TOKENOPS_EMBEDDED", "1") clear_governance_config_cache() s = Store(db) yield s @@ -50,17 +49,24 @@ def test_init_installs_crossing_hook(): assert session.on_crossing is on_crossing -def test_from_env_installs_crossing_hook(monkeypatch, tmp_path): - monkeypatch.delenv("TOKENOPS_URL", raising=False) - monkeypatch.setenv("TOKENOPS_EMBEDDED", "1") - monkeypatch.setenv("TOKENOPS_DB", str(tmp_path / "env.db")) +def test_from_env_installs_crossing_hook(live_plane_url): # Force a fresh look: hook is already installed in-process; still must not raise. client = ControlPlaneClient.from_env() - assert client.embedded + assert not client.embedded + assert client.url == live_plane_url assert getattr(chronicle_session.reset_session, "_tokenops_crossing_hook", False) install_crossing_hook() # idempotent +def test_from_env_raises_without_a_url(monkeypatch): + """tokenops#118: no embedded fallback — from_env() must fail closed.""" + monkeypatch.delenv("CONTROL_PLANE_URL", raising=False) + monkeypatch.delenv("TOKENOPS_URL", raising=False) + monkeypatch.delenv("TOKENOPS_CONTROL_PLANE_URL", raising=False) + with pytest.raises(RuntimeError, match="CONTROL_PLANE_URL"): + ControlPlaneClient.from_env() + + def test_tokenops_run_registers_when_no_run_id(store): clear() with tokenops_run( @@ -85,8 +91,8 @@ def test_tokenops_run_registers_when_no_run_id(store): assert current_governance() is None -def test_tokenops_run_without_user_passed_store(store): - """§6 happy path: embedded from_env — no store= from the caller.""" +def test_tokenops_run_without_user_passed_store(live_plane_url): + """§6 happy path: from_env's real remote client — no store= from the caller.""" clear() with tokenops_run( headers={}, @@ -96,8 +102,10 @@ def test_tokenops_run_without_user_passed_store(store): mode="preview", ) as bound: assert bound.registration.intent == "triad_plan" - assert bound.client.embedded + assert not bound.client.embedded + assert bound.client.url == live_plane_url assert bound.store is bound.client.require_store() + assert bound.governor.ledger._backend is bound.client.backend assert current_governance() is not None clear() @@ -147,7 +155,7 @@ def test_agent_intent_beats_empty_and_payload_intent(store): clear() -def test_request_context_ambient(store): +def test_request_context_ambient(live_plane_url): clear() clear_request_context() bind_request_context( @@ -173,7 +181,7 @@ def test_request_context_ambient(store): clear() -def test_instrument_app_binds_context_and_hook(store): +def test_instrument_app_binds_context_and_hook(live_plane_url): app = FastAPI() instrument_app(app, service="planner", intent="mw_intent", mode="enforce") From 005f7a5bd4d8e3fced59536aa21d3fa80e20e5c8 Mon Sep 17 00:00:00 2001 From: susheem-k Date: Sat, 12 Sep 2026 17:22:51 +0530 Subject: [PATCH 4/4] docs: fix Quickstart to reflect the hard dependency on a running control plane MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two active falsehoods left over from the previous pass on this file (docs/env-var table only, not the Quickstart copy itself): - Section 1's manual snippet called ControlPlaneClient.from_env() with no mention a control plane must already be running — it now raises immediately without one. - The Quickdeploy tip claimed "a single-process agent doesn't need it running at all", which was true under the old embedded-ledger fallback and is false now that tokenops has no ledger of its own (#118). Co-Authored-By: Claude Sonnet 5 --- README.md | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index afd3882..1340aff 100644 --- a/README.md +++ b/README.md @@ -53,7 +53,10 @@ add your own. ## 🚀 Quickstart -Requires Python 3.10+. +Requires Python 3.10+ and a running control plane — TokenOps has no ledger of its +own, so every agent (even a single process) governs against one. Zero-setup taste: +`python -m tokenops.demo` launches a throwaway plane for you automatically. For your +own agent, start one first (see [Quickdeploy](#-quickdeploy) below), then: ### 1. Put it in your agent @@ -76,14 +79,16 @@ Anywhere else (Cursor, Copilot, ...), paste this:
Manual, about ten lines -Wrap your model call once, then hand the wrapped version to your agent. +Wrap your model call once, then hand the wrapped version to your agent. Needs +`CONTROL_PLANE_URL` (or `TOKENOPS_URL`) pointing at a running control plane — +see [Quickdeploy](#-quickdeploy). ```python from tokenops import ControlPlaneClient, tokenops_run from tokenops.control import Halt, wrap_complete from tokenops.providers import complete -client = ControlPlaneClient.from_env() +client = ControlPlaneClient.from_env() # raises if CONTROL_PLANE_URL isn't set with tokenops_run(client=client, service="my-agent", intent="research", provider="openai", model="gpt-4o") as bound: @@ -131,9 +136,9 @@ run is out, even from another process. ## 🐳 Quickdeploy > [!TIP] -> The control plane (`python -m tokenops.server`) shares one budget across -> processes and powers the dashboard. A single-process agent doesn't need it -> running at all. +> Every agent needs a control plane running — TokenOps has no ledger of its own. +> A single-process agent still needs one, just not the multi-process sharing this +> section is about. `python -m tokenops.server` powers the dashboard too. One command, plane + dashboard: