diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..90d1dfc --- /dev/null +++ b/.dockerignore @@ -0,0 +1,26 @@ +.git +.github +.pytest_cache +.ruff_cache +__pycache__ +*.pyc +.venv +venv +*.egg-info +build +dist + +# Not needed to build/run the image +tests +docs +ui +CHANGELOG.md +RELEASING.md +docker-compose*.yml +Dockerfile +.dockerignore + +# Local dev artifacts — never bake a DB into the image +*.db +*.db-* +control_plane.db diff --git a/CHANGELOG.md b/CHANGELOG.md index 866665b..c3f0f9c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,59 @@ ## [Unreleased] +### Added + +- `control-plane start` / `stop` / `status` — run the plane detached from the + terminal as one managed background instance (PID + state file under a + `platformdirs` user-state dir; logs redirected to a file since the process is no + longer attached to a console). `start` waits for `/health` before returning and + reaps the process if it never comes up; `stop` escalates `terminate()` → `kill()` + on a timeout; both are idempotent (safe to call when already in the target state). + Cross-platform: POSIX detaches via `start_new_session`, Windows via + `DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP` — Windows has no SIGTERM, so `stop` + there is a hard stop, not a graceful drain. + New deps: `psutil`, `platformdirs`. +- `Dockerfile` + `docker-compose.yml` — the repo's first container image. Builds from + source (`pip install .`), runs as a non-root user, SQLite on a `/data` volume, + built-in `HEALTHCHECK` against `/health`. Entrypoint is `control-plane serve` + (foreground) — the container is its own process supervisor, so the new + `start`/`stop`/`status` background mode is for local dev only, not for images. + +## [0.2.0] — TokenOps remote-only (#10) + +**Additive on the wire and schema.** 0.1-era clients keep working against 0.2.x; the +breaking fold (drop `run_registrations`, remove `create_run`, narrow `PATCH`) is +deferred to 0.3.0 (#11), released after `tokenops `. + +Added +- Wire contract: `docs/api-contract.md`. +- `POST /v1/ledger/events:batch` — batched ledger writes (`spent_add` with multi-target + fan-out, `admit`/`complete`, `step`, `halt_mark`/`halt_clear`), applied in array order + in one transaction, idempotent per `idempotency_key` (`ledger_events` table). +- `POST /v1/ledger/precheck` — one read for a pre_call pass: halt + requested + spent / inflight / window slices (`SqliteStore.precheck`). +- `run_state` table — per-run `step_count` / bounded window / velocity, fed by `step` + events; `SqliteStore.get_run_state`. +- `policy_instances.data_scope` (`local` | `global`, default `local`) — surfaced in + `GET /v1/governance/{agent}`, `PUT /v1/policies`, and YAML seed. +- `GET /v1/ledger/runs/{run_id}/halt` (GET / POST / DELETE) — run-scoped halt; + `/v1/ledger/halt/*` kept as aliases. +- `PRAGMA user_version` schema versioning (`SqliteStore.SCHEMA_VERSION = 2`) + + forward-only `_apply_migrations()`. Opening a 0.1 DB adds the new tables/columns and + bumps the version — no data loss. The 0.3.0 destructive fold (#11) slots in as v3. + +Changed +- `POST /v1/runs` response includes `registered_at`; `register_run` returns the + persisted row so the SDK can bind without a follow-up `GET .../registration`. +- `GET /health` advertises `max_batch` / `max_body_bytes`. +- Version → `0.2.0`. + +Deprecated (removed in 0.3.0) +- `PATCH /v1/run-records` — client-sent `steps` / `cost_micros` are now **ignored** + (logged); they are derived from `run_state` / `ledger_spent`. +- `PUT /v1/run-records` (`create_run`), the `run_registrations` table, and the legacy + `/v1/ledger/halt/*` + single-op `/v1/ledger/{spent,inflight}/*` writes. + ## [0.1.0] — 2026-08-14 - First PyPI release of the shared AgentPlane control plane. diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..7f2a5b4 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,38 @@ +# syntax=docker/dockerfile:1 +FROM python:3.12-slim + +LABEL org.opencontainers.image.title="agentplane-control-plane" +LABEL org.opencontainers.image.description="Shared AgentPlane control plane: Chronicle envelopes + TokenOps governance over HTTP" +LABEL org.opencontainers.image.source="https://github.com/theagentplane/control-plane" + +WORKDIR /app + +# Only what pyproject.toml + hatchling need to build the wheel (see +# [tool.hatch.build.targets.wheel] / force-include) — keeps the build context and +# image small, and cache-friendly (dependency layer doesn't invalidate on doc edits). +COPY pyproject.toml README.md LICENSE ./ +COPY config ./config +COPY control_plane ./control_plane + +RUN pip install --no-cache-dir --upgrade pip \ + && pip install --no-cache-dir . + +# SQLite lives on a volume so it survives `docker compose down` / image upgrades. +ENV CONTROL_PLANE_DB=/data/control_plane.db +VOLUME ["/data"] + +# Least privilege: the process only ever needs to read/write /data. +RUN useradd --create-home --uid 1000 controlplane \ + && mkdir -p /data \ + && chown -R controlplane:controlplane /data +USER controlplane + +EXPOSE 8800 + +HEALTHCHECK --interval=10s --timeout=3s --start-period=5s --retries=3 \ + CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8800/health', timeout=2)" || exit 1 + +# Foreground, PID 1 — the container is the process supervisor. `control-plane start` +# (the background/detached mode for local dev) must not be used here: if PID 1 exits +# right after spawning a detached child, the container exits with it. +CMD ["control-plane", "serve", "--host", "0.0.0.0", "--port", "8800"] diff --git a/README.md b/README.md index 14be686..5358320 100644 --- a/README.md +++ b/README.md @@ -91,13 +91,55 @@ pip install -e ".[dev]" ## Quick start ```bash -control-plane serve --port 8800 --db control_plane.db +control-plane start # runs in the background; safe to re-run (no-op if already up) +control-plane status # pid, port, version, /health +control-plane stop ``` Open [http://127.0.0.1:8800/](http://127.0.0.1:8800/) — Admin, Chronicle, TokenOps. No login. +One managed instance per machine user; `start`/`stop`/`status` track it via a small +state file (`platformdirs` user-state dir) and a PID — `stop` never touches a process +it didn't start. Logs go to a file next to the state (path printed by `start`), since +the process is detached from your terminal. + +Prefer the foreground, un-managed form for scripting, containers, or when you want +Ctrl-C to stop it: + +```bash +control-plane serve --port 8800 --db control_plane.db +``` + `control-plane ui` is a pointer, not a second server: the HTML UI is served with the API. +## Docker + +For hosting (a shared team plane, a demo environment) rather than a local dev loop: + +```bash +docker compose up -d # builds the image, starts on :8800, persists SQLite in a volume +docker compose logs -f +docker compose down # add -v to also drop the data volume +``` + +Or without Compose: + +```bash +docker build -t agentplane-control-plane . +docker run -d --name control-plane -p 8800:8800 -v control-plane-data:/data agentplane-control-plane +``` + +The image's entrypoint is `control-plane serve` (foreground, PID 1) — **not** `start`. +Docker/Kubernetes is already the process supervisor here (restart policy, health +checks via the built-in `HEALTHCHECK` hitting `/health`, log collection from +stdout/stderr); `start`/`stop`/`status` are for running the plane directly on a +developer's machine, where nothing else is supervising the process. Don't run them +inside the container — a background/detached mode would exit PID 1 as soon as it +spawned its child, and the container would exit with it. + +Set `CONTROL_PLANE_API_KEYS` before exposing the container beyond localhost — the +default (empty) is anonymous, all-scopes access. + ## Sidecars Point both libraries at the same origin. Do **not** set a SQLite path on the agents. @@ -197,8 +239,28 @@ The plane is early (0.x). Near-term: Ideas welcome via GitHub issues. +## Compatibility + +| agentplane-control-plane | tokenops | agent-chronicle | notes | +|---|---|---|---| +| 0.1.x | ≤ 0.2.1 | ≥ 0.3.0 | single-op `/v1/ledger/*`, `PUT /v1/run-records` | +| **0.2.x** | ≤ 0.2.1 **and** `` | ≥ 0.3.0 | **additive** — old clients keep working; adds `precheck` / `events:batch`, `run_state`, `data_scope` | +| [0.3.x](https://github.com/theagentplane/control-plane/issues/11) | ``+ only | ≥ 0.3.0 | breaking — drops `run_registrations`, `PUT /v1/run-records`, legacy ledger wrappers | + +### Breaking changes + +- **0.2.0 → 0.3.0:** `run_registrations` folded into `runs` and dropped; + `PUT /v1/run-records` (`create_run`) removed; `PATCH /v1/run-records` rejects + `steps` / `cost_micros` (0.2.x only ignores them); legacy `/v1/ledger/halt/*` and + single-op `/v1/ledger/{spent,inflight}/*` writes removed. Runs as an automatic + `PRAGMA user_version` v3 migration. Released only after `tokenops ` stops + calling `create_run`. + +Full contract: [`docs/api-contract.md`](docs/api-contract.md). + ## Documentation +- [API contract](docs/api-contract.md) — the TokenOps ⇄ control-plane wire spec - [Design](docs/DESIGN.md) — storage, callers, scopes, keys - [Releasing](RELEASING.md) — Trusted Publishing to PyPI - [Changelog](CHANGELOG.md) diff --git a/control_plane/app.py b/control_plane/app.py index 240ebc2..2e1762f 100644 --- a/control_plane/app.py +++ b/control_plane/app.py @@ -58,8 +58,14 @@ def create_app( mount_web(app) @app.get("/health") - async def health() -> dict[str, str]: - return {"status": "ok", "service": "control-plane", "version": cfg.version} + async def health() -> dict[str, Any]: + return { + "status": "ok", + "service": "control-plane", + "version": cfg.version, + "max_batch": cfg.max_batch, + "max_body_bytes": cfg.max_body_bytes, + } @app.get("/ready") async def ready() -> dict[str, str]: @@ -186,6 +192,7 @@ async def register_run( "mode": saved.mode.value, "intent": saved.intent, "user_dims": dict(saved.user_dims), + "registered_at": saved.registered_at, }, status_code=201, ) @@ -210,7 +217,9 @@ async def governance_for_agent( return gov.governance_config_for(agent) @app.get("/v1/segments") - async def list_segments(principal: Principal = Depends(require_scopes("read"))) -> list[dict[str, Any]]: + async def list_segments( + principal: Principal = Depends(require_scopes("read")), + ) -> list[dict[str, Any]]: return [segment_to_dict(s) for s in gov.list_segments()] @app.put("/v1/segments") @@ -240,7 +249,9 @@ async def delete_segment( return {"status": "deleted"} @app.get("/v1/budgets") - async def list_budgets(principal: Principal = Depends(require_scopes("read"))) -> list[dict[str, Any]]: + async def list_budgets( + principal: Principal = Depends(require_scopes("read")), + ) -> list[dict[str, Any]]: return [budget_to_dict(b) for b in gov.list_budgets()] @app.put("/v1/budgets") @@ -270,7 +281,9 @@ async def delete_budget( return {"status": "deleted"} @app.get("/v1/policies") - async def list_policies(principal: Principal = Depends(require_scopes("read"))) -> list[dict[str, Any]]: + async def list_policies( + principal: Principal = Depends(require_scopes("read")), + ) -> list[dict[str, Any]]: return [policy_to_dict(p) for p in gov.list_policy_instances()] @app.put("/v1/policies") @@ -339,7 +352,9 @@ async def list_run_records( limit: int = 200, principal: Principal = Depends(require_scopes("read")), ) -> list[dict[str, Any]]: - return [run_to_dict(r) for r in gov.list_runs(problematic_only=problematic_only, limit=limit)] + return [ + run_to_dict(r) for r in gov.list_runs(problematic_only=problematic_only, limit=limit) + ] @app.get("/v1/run-records/tag-keys") async def run_tag_keys( @@ -357,7 +372,10 @@ async def ledger_add_spent( ) -> dict[str, int]: body = await request.json() spent = gov.ledger_add_spent( - body["budget_id"], body["segment_key"], body.get("period", "lifetime"), int(body["delta"]), + body["budget_id"], + body["segment_key"], + body.get("period", "lifetime"), + int(body["delta"]), ) return {"spent_micros": spent} @@ -421,6 +439,79 @@ async def ledger_clear_halt( gov.ledger_clear_halt(body["run_id"]) return {"status": "cleared"} + @app.post("/v1/ledger/precheck", tags=["agent-tokenops"]) + async def ledger_precheck( + request: Request, + principal: Principal = Depends(require_scopes("read")), + ) -> dict[str, Any]: + """Caller: TokenOps sidecar. One read for a pre_call pass — contract §4.""" + body = await request.json() + run_id = str(body.get("run_id") or "").strip() + if not run_id: + raise HTTPException(status_code=400, detail="run_id is required") + return gov.precheck( + principal.tenant_id, + run_id, + segment_keys=body.get("segment_keys") or [], + budgets=body.get("budgets") or [], + want=body.get("want") or ["spent", "inflight", "halt"], + ) + + @app.post("/v1/ledger/events:batch", tags=["agent-tokenops"]) + async def ledger_events_batch( + request: Request, + principal: Principal = Depends(require_scopes("ingest")), + ) -> JSONResponse: + """Caller: TokenOps sidecar. Batched ledger writes — contract §5. + + ``Durability: sync|queued`` header (queued == sync in 0.2.0). Applied in array + order, one transaction. Idempotent per ``idempotency_key`` (per tenant). + """ + if int(request.headers.get("content-length") or 0) > cfg.max_body_bytes: + raise HTTPException(status_code=413, detail="payload too large") + _durability(request) # validates the header (sync|queued) + payload = await request.json() + events = payload.get("events") if isinstance(payload, dict) else None + if not isinstance(events, list): + raise HTTPException(status_code=400, detail="expected {events: [...]}") + if len(events) > cfg.max_batch: + raise HTTPException(status_code=400, detail=f"batch max {cfg.max_batch}") + try: + result = gov.apply_events(principal.tenant_id, events) + except (ValueError, KeyError) as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + return JSONResponse(result, status_code=201) + + # ---- ledger halt (run-scoped — contract §7; /v1/ledger/halt/* stay as aliases) ---- # + + @app.get("/v1/ledger/runs/{run_id}/halt") + async def run_halt_get( + run_id: str, + principal: Principal = Depends(require_scopes("read")), + ) -> dict[str, Any]: + return { + "halted": gov.ledger_is_halted(run_id), + "halt_reason": gov.ledger_halt_reason(run_id), + } + + @app.post("/v1/ledger/runs/{run_id}/halt") + async def run_halt_set( + run_id: str, + request: Request, + principal: Principal = Depends(require_scopes("ingest")), + ) -> dict[str, bool]: + body = await request.json() + gov.ledger_mark_halted(run_id, str(body.get("reason") or "")) + return {"halted": True} + + @app.delete("/v1/ledger/runs/{run_id}/halt") + async def run_halt_clear( + run_id: str, + principal: Principal = Depends(require_scopes("admin")), + ) -> dict[str, bool]: + gov.ledger_clear_halt(run_id) + return {"halted": False} + # ---- admin ------------------------------------------------------------ # @app.post("/v1/admin/seed-if-empty") @@ -438,7 +529,9 @@ async def clear_all(principal: Principal = Depends(require_scopes("admin"))) -> return {"status": "cleared"} @app.post("/v1/admin/clear-governance") - async def clear_governance(principal: Principal = Depends(require_scopes("admin"))) -> dict[str, str]: + async def clear_governance( + principal: Principal = Depends(require_scopes("admin")), + ) -> dict[str, str]: gov.clear_governance() return {"status": "cleared"} @@ -458,7 +551,10 @@ async def list_keys(principal: Principal = Depends(require_scopes("admin"))) -> """Caller: Admin UI. Env keys include the secret (already on the host). DB keys show prefix only.""" db_keys = [k.__dict__ for k in gov.list_api_keys()] env_keys = [k.__dict__ for k in env_key_views(cfg.api_keys)] - return {"keys": env_keys + db_keys, "auth_disabled": not cfg.api_keys and not gov.list_api_keys()} + return { + "keys": env_keys + db_keys, + "auth_disabled": not cfg.api_keys and not gov.list_api_keys(), + } @app.post("/v1/admin/keys", tags=["ui"]) async def create_key( @@ -484,7 +580,9 @@ async def delete_key( ) -> dict[str, str]: """Caller: Admin UI. Env keys cannot be deleted here.""" if kid.startswith("env:"): - raise HTTPException(status_code=400, detail="env keys are removed by unsetting CONTROL_PLANE_API_KEYS") + raise HTTPException( + status_code=400, detail="env keys are removed by unsetting CONTROL_PLANE_API_KEYS" + ) gov.delete_api_key(kid) return {"status": "deleted"} diff --git a/control_plane/cli.py b/control_plane/cli.py index 1717b71..2e73dc0 100644 --- a/control_plane/cli.py +++ b/control_plane/cli.py @@ -4,18 +4,30 @@ import argparse import os +import sys def main() -> None: parser = argparse.ArgumentParser(prog="control-plane") sub = parser.add_subparsers(dest="cmd", required=True) - serve = sub.add_parser("serve", help="Run the FastAPI control plane") + serve = sub.add_parser("serve", help="Run the FastAPI control plane (foreground)") serve.add_argument("--host", default="127.0.0.1") serve.add_argument("--port", type=int, default=8800) serve.add_argument("--db", default=None, help="SQLite path (default CONTROL_PLANE_DB)") serve.add_argument("--reload", action="store_true") + start = sub.add_parser( + "start", + help="Start the control plane in the background (one instance; run again to check it)", + ) + start.add_argument("--host", default="127.0.0.1") + start.add_argument("--port", type=int, default=8800) + start.add_argument("--db", default=None, help="SQLite path (default CONTROL_PLANE_DB)") + + sub.add_parser("stop", help="Stop the background control plane started with `start`") + sub.add_parser("status", help="Show whether the background control plane is running") + ui = sub.add_parser("ui", help="Run Admin + Dashboard (Streamlit)") ui.add_argument("--port", type=int, default=8501) @@ -32,6 +44,23 @@ def main() -> None: port=args.port, reload=args.reload, ) + elif args.cmd == "start": + from control_plane import servicectl + + try: + servicectl.start(host=args.host, port=args.port, db=args.db) + except servicectl.ServiceError as exc: + print(str(exc), file=sys.stderr) + raise SystemExit(1) from exc + elif args.cmd == "stop": + from control_plane import servicectl + + servicectl.stop() # idempotent: "wasn't running" is not an error + elif args.cmd == "status": + from control_plane import servicectl + + info = servicectl.status() + raise SystemExit(0 if info.get("running") else 1) elif args.cmd == "ui": print("UI is served with the API. Run: control-plane serve --port 8800") print("Then open http://127.0.0.1:8800/") diff --git a/control_plane/models.py b/control_plane/models.py index 2b05140..b0ac43a 100644 --- a/control_plane/models.py +++ b/control_plane/models.py @@ -33,6 +33,7 @@ class RunRegistration: intent: str = "" user_dims: dict[str, str] = field(default_factory=dict) mode: GovernanceMode = GovernanceMode.ENFORCE + registered_at: float = 0.0 class RunNotRegisteredError(LookupError): @@ -70,6 +71,9 @@ class PolicyInstance: budget_id: str | None = None segment_id: str | None = None enabled: bool = True + #: Where the SDK runs this policy's detector — "local" (per-process LocalRunState) + #: or "global" (plane aggregate via /v1/ledger/precheck). Contract §10. + data_scope: str = "local" @dataclass diff --git a/control_plane/serde.py b/control_plane/serde.py index 4e7c1c4..e8d065e 100644 --- a/control_plane/serde.py +++ b/control_plane/serde.py @@ -20,6 +20,7 @@ def registration_to_dict(reg: RunRegistration) -> dict[str, Any]: "intent": reg.intent, "user_dims": dict(reg.user_dims), "mode": reg.mode.value, + "registered_at": reg.registered_at, "status": "registered", } @@ -30,6 +31,7 @@ def registration_from_dict(data: dict[str, Any]) -> RunRegistration: intent=str(data.get("intent", "")), user_dims={str(k): str(v) for k, v in (data.get("user_dims") or {}).items()}, mode=parse_governance_mode(data.get("mode")), + registered_at=float(data.get("registered_at") or 0.0), ) @@ -82,6 +84,7 @@ def policy_to_dict(pi: PolicyInstance) -> dict[str, Any]: "budget_id": pi.budget_id, "segment_id": pi.segment_id, "enabled": pi.enabled, + "data_scope": pi.data_scope, } @@ -94,6 +97,7 @@ def policy_from_dict(data: dict[str, Any]) -> PolicyInstance: budget_id=data.get("budget_id"), segment_id=data.get("segment_id"), enabled=bool(data.get("enabled", True)), + data_scope=str(data.get("data_scope") or "local"), ) diff --git a/control_plane/servicectl.py b/control_plane/servicectl.py new file mode 100644 index 0000000..aa67d0d --- /dev/null +++ b/control_plane/servicectl.py @@ -0,0 +1,254 @@ +"""start / stop / status for a background ``control-plane serve`` process. + +Singleton per state directory: one JSON state file tracks the one instance this CLI +manages (``control-plane start`` twice is a no-op, not two servers). Two OS rough +edges are called out rather than papered over: + +* **PIDs get reused.** An "is it running" check must confirm the process is actually + ours, not just that some process holds that PID. +* **Windows has no SIGTERM.** ``stop()`` calls ``psutil.Process.terminate()``, which is + a clean signal on POSIX (uvicorn's own graceful-shutdown handler) but a hard stop on + Windows — there is no portable "please drain and exit" on that platform. + +``CONTROL_PLANE_STATE_DIR`` overrides the state directory (tests use this to avoid +touching the real per-user state and to isolate concurrent runs). +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +import httpx +import psutil +from platformdirs import user_state_dir + +APP_NAME = "agentplane-control-plane" + +#: How long `start()` waits for /health before declaring failure. +START_TIMEOUT_S = 10.0 +#: How long `stop()` waits for a graceful exit before escalating to kill(). +STOP_TIMEOUT_S = 5.0 + + +class ServiceError(RuntimeError): + """start()/stop() failed in a way the caller should see (not just print).""" + + +def _state_dir() -> Path: + override = os.environ.get("CONTROL_PLANE_STATE_DIR") + d = Path(override) if override else Path(user_state_dir(APP_NAME)) + d.mkdir(parents=True, exist_ok=True) + return d + + +def _state_file() -> Path: + return _state_dir() / "control-plane.json" + + +def _log_file() -> Path: + return _state_dir() / "control-plane.log" + + +@dataclass +class ServiceState: + pid: int + host: str + port: int + db: str | None + log_file: str + started_at: float + + @property + def base_url(self) -> str: + return f"http://{self.host}:{self.port}" + + +def _read_state() -> ServiceState | None: + f = _state_file() + if not f.exists(): + return None + try: + return ServiceState(**json.loads(f.read_text())) + except (json.JSONDecodeError, TypeError, ValueError): + return None + + +def _write_state(state: ServiceState) -> None: + _state_file().write_text(json.dumps(asdict(state))) + + +def _clear_state() -> None: + _state_file().unlink(missing_ok=True) + + +def _is_ours(pid: int) -> bool: + """A PID match alone is not enough — PIDs are reused by the OS. Confirm the + process is actually a ``control_plane.cli serve`` invocation.""" + try: + cmdline = " ".join(psutil.Process(pid).cmdline()) + except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): + return False + return "control_plane.cli" in cmdline and "serve" in cmdline + + +def _running_state() -> ServiceState | None: + """The tracked state iff its PID is alive and genuinely ours; otherwise clears a + stale state file (crashed process, reused PID, hand-edited file) and returns None.""" + state = _read_state() + if state is None: + return None + if psutil.pid_exists(state.pid) and _is_ours(state.pid): + return state + _clear_state() + return None + + +def _wait_healthy(base_url: str, *, timeout: float) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + try: + if httpx.get(f"{base_url}/health", timeout=1.0).status_code == 200: + return True + except httpx.HTTPError: + pass + time.sleep(0.2) + return False + + +def _log_tail(log_path: Path, n: int = 20) -> str: + try: + lines = log_path.read_text(errors="replace").splitlines() + return "\n".join(lines[-n:]) + except OSError: + return "(no log)" + + +def start(*, host: str = "127.0.0.1", port: int = 8800, db: str | None = None) -> ServiceState: + """Start ``control-plane serve`` detached from this process's console/terminal. + + Idempotent: if a managed instance is already running, returns it unchanged + (does not restart it even if ``host``/``port``/``db`` differ from the request — + stop it first if you want different settings). + """ + existing = _running_state() + if existing is not None: + print(f"already running on {existing.base_url} (pid {existing.pid})") + return existing + + log_path = _log_file() + cmd = [ + sys.executable, + "-m", + "control_plane.cli", + "serve", + "--host", + host, + "--port", + str(port), + ] + if db: + cmd += ["--db", db] + + popen_kwargs: dict[str, Any] = {} + if sys.platform == "win32": + popen_kwargs["creationflags"] = ( + subprocess.DETACHED_PROCESS | subprocess.CREATE_NEW_PROCESS_GROUP + ) + else: + popen_kwargs["start_new_session"] = True # setsid: survives the parent shell closing + + with open(log_path, "ab") as log_fh: + proc = subprocess.Popen(cmd, stdout=log_fh, stderr=log_fh, **popen_kwargs) + + state = ServiceState( + pid=proc.pid, + host=host, + port=port, + db=db, + log_file=str(log_path), + started_at=time.time(), + ) + + if not _wait_healthy(state.base_url, timeout=START_TIMEOUT_S): + # It may still be alive but broken (e.g. bad --db path) — don't leave an + # unhealthy process untracked and orphaned; try to reap it. + if psutil.pid_exists(proc.pid): + try: + psutil.Process(proc.pid).kill() + except psutil.NoSuchProcess: + pass + raise ServiceError( + f"control-plane did not become healthy within {START_TIMEOUT_S:.0f}s; " + f"log tail ({log_path}):\n{_log_tail(log_path)}" + ) + + _write_state(state) + print(f"started on {state.base_url} (pid {state.pid}); log: {log_path}") + return state + + +def stop(*, timeout: float = STOP_TIMEOUT_S) -> bool: + """Stop the managed instance. Returns False (no-op, not an error) if nothing was + running — matches the idempotent feel of ``start()``.""" + state = _running_state() + if state is None: + print("not running") + return False + + try: + proc = psutil.Process(state.pid) + proc.terminate() # SIGTERM (POSIX, graceful) / TerminateProcess (Windows, hard) + try: + proc.wait(timeout=timeout) + except psutil.TimeoutExpired: + proc.kill() + proc.wait(timeout=timeout) + except psutil.NoSuchProcess: + pass + + _clear_state() + print(f"stopped (pid {state.pid})") + return True + + +def status() -> dict[str, Any]: + """Report on the managed instance. Always safe to call — never raises for + "not running".""" + state = _running_state() + if state is None: + print("not running") + return {"running": False} + + healthy = False + version = None + try: + r = httpx.get(f"{state.base_url}/health", timeout=1.0) + healthy = r.status_code == 200 + if healthy: + version = r.json().get("version") + except httpx.HTTPError: + pass + + info: dict[str, Any] = { + "running": True, + "healthy": healthy, + "pid": state.pid, + "host": state.host, + "port": state.port, + "db": state.db, + "version": version, + "log_file": state.log_file, + "started_at": state.started_at, + } + label = "running" if healthy else "running (not responding on /health)" + print( + f"{label} - pid {state.pid}, {state.base_url}" + (f", version {version}" if version else "") + ) + return info diff --git a/control_plane/settings.py b/control_plane/settings.py index 15c3005..4cbf4b7 100644 --- a/control_plane/settings.py +++ b/control_plane/settings.py @@ -15,7 +15,7 @@ class Settings: default_durability: str = "sync" # sync | queued max_batch: int = 100 max_body_bytes: int = 4 * 1024 * 1024 - version: str = "0.1.0" + version: str = "0.2.0" def load_settings() -> Settings: @@ -27,5 +27,5 @@ def load_settings() -> Settings: default_durability=os.environ.get("CONTROL_PLANE_DURABILITY", "sync"), max_batch=int(os.environ.get("CONTROL_PLANE_MAX_BATCH", "100")), max_body_bytes=int(os.environ.get("CONTROL_PLANE_MAX_BODY", str(4 * 1024 * 1024))), - version=os.environ.get("CONTROL_PLANE_VERSION", "0.1.0"), + version=os.environ.get("CONTROL_PLANE_VERSION", "0.2.0"), ) diff --git a/control_plane/store.py b/control_plane/store.py index 8ca24a1..e1df3b1 100644 --- a/control_plane/store.py +++ b/control_plane/store.py @@ -14,6 +14,7 @@ import hashlib import json +import logging import os import secrets import sqlite3 @@ -46,7 +47,8 @@ ); 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, @@ -89,13 +91,51 @@ halted INTEGER NOT NULL DEFAULT 0, halt_reason TEXT ); +CREATE TABLE IF NOT EXISTS ledger_events ( + tenant_id TEXT NOT NULL, + idempotency_key TEXT NOT NULL, + applied_at REAL NOT NULL, + PRIMARY KEY (tenant_id, idempotency_key) +); +CREATE TABLE IF NOT EXISTS run_state ( + tenant_id TEXT NOT NULL, + run_id TEXT NOT NULL, + step_count INTEGER NOT NULL DEFAULT 0, + window_json TEXT NOT NULL DEFAULT '[]', + velocity_micros_per_step REAL NOT NULL DEFAULT 0, + last_ts REAL, + PRIMARY KEY (tenant_id, run_id) +); """ +#: Bound on the per-run BoundaryStep ring kept in ``run_state.window_json``. +RUN_STATE_WINDOW = 64 + +#: Current schema version, tracked in ``PRAGMA user_version``. +#: 0 → pre-versioning (0.1.x). Legacy ad-hoc ALTERs run, then bumped to current. +#: 2 → 0.2.0: additive only. ``_SCHEMA`` (CREATE IF NOT EXISTS) covers the new +#: ``ledger_events`` / ``run_state`` tables on any existing DB. No destructive +#: change — ``run_registrations`` and the ``run-records`` write path stay +#: (deprecated). The 0.3.0 fold (drop ``run_registrations``, etc.) will be v3. +SCHEMA_VERSION = 2 + + +_log = logging.getLogger("control_plane.store") + def new_id(prefix: str) -> str: return f"{prefix}_{uuid.uuid4().hex[:8]}" +def _velocity_from_window(window: list[dict]) -> float: + """micros/step over the ring: (newest cum_spent - oldest cum_spent) / span.""" + if len(window) < 2: + return 0.0 + newest = window[-1].get("cum_spent_micros") or 0 + oldest = window[0].get("cum_spent_micros") or 0 + return (newest - oldest) / (len(window) - 1) + + class SqliteStore: def __init__(self, path: str = "control_plane.db", *, auto_seed: bool = True) -> None: self.path = path @@ -105,27 +145,55 @@ def __init__(self, path: str = "control_plane.db", *, auto_seed: bool = True) -> self._db.execute("PRAGMA journal_mode=WAL") self._db.execute("PRAGMA foreign_keys=ON") self._db.execute("PRAGMA busy_timeout=5000") - self._db.executescript(_SCHEMA) - self._migrate() + self._db.executescript(_SCHEMA) # CREATE IF NOT EXISTS — fresh DB + new tables + self._apply_migrations() self._db.commit() if auto_seed: self.seed_default_governance_if_empty() - def _migrate(self) -> None: - cols = {row[1] for row in self._db.execute("PRAGMA table_info(runs)")} - if "dims" not in cols: + def _apply_migrations(self) -> None: + """Forward-only schema migrations, tracked in ``PRAGMA user_version``. + + ``_SCHEMA`` already ran, so every table exists. This only adds columns / does + destructive folds that ``CREATE IF NOT EXISTS`` cannot express. + """ + version = self._db.execute("PRAGMA user_version").fetchone()[0] + + # Always run (all guarded / idempotent): covers a fresh DB (no-op — _SCHEMA + # already made the columns), a pre-versioning 0.1 DB, and any intermediate. + self._ensure_columns() + + # v2 (0.2.0) is otherwise purely additive — ledger_events / run_state come + # from _SCHEMA. + + # if version < 3: # 0.3.0 — fold run_registrations into runs, drop it, etc. + # self._migrate_v3_fold_registrations() + + if version != SCHEMA_VERSION: + self._db.execute(f"PRAGMA user_version = {SCHEMA_VERSION}") + + def _ensure_columns(self) -> None: + def _cols(table: str) -> set[str]: + return {row[1] for row in self._db.execute(f"PRAGMA table_info({table})")} + + runs = _cols("runs") + if "dims" not in runs: self._db.execute("ALTER TABLE runs ADD COLUMN dims TEXT NOT NULL DEFAULT '{}'") - if "parent_span" not in cols: + if "parent_span" not in runs: self._db.execute("ALTER TABLE runs ADD COLUMN parent_span TEXT") - if "governance_events" not in cols: + if "governance_events" not in runs: self._db.execute( "ALTER TABLE runs ADD COLUMN governance_events TEXT NOT NULL DEFAULT '[]'" ) - reg_cols = {row[1] for row in self._db.execute("PRAGMA table_info(run_registrations)")} - if "mode" not in reg_cols: + reg = _cols("run_registrations") + if reg and "mode" not in reg: self._db.execute( "ALTER TABLE run_registrations ADD COLUMN mode TEXT NOT NULL DEFAULT 'enforce'" ) + if "data_scope" not in _cols("policy_instances"): + self._db.execute( + "ALTER TABLE policy_instances ADD COLUMN data_scope TEXT NOT NULL DEFAULT 'local'" + ) def close(self) -> None: self._db.close() @@ -176,12 +244,23 @@ def delete_budget(self, bid: str) -> None: def upsert_policy_instance(self, pi: PolicyInstance) -> PolicyInstance: if pi.template not in _TEMPLATES: # fail closed — same rule as build_governor - raise ValueError(f"unknown policy template {pi.template!r}; known: {sorted(_TEMPLATES)}") + raise ValueError( + f"unknown policy template {pi.template!r}; known: {sorted(_TEMPLATES)}" + ) self._db.execute( - "REPLACE INTO policy_instances(id, template, params, agent, budget_id, segment_id, enabled) " - "VALUES (?,?,?,?,?,?,?)", - (pi.id, pi.template, json.dumps(pi.params), pi.agent, pi.budget_id, pi.segment_id, - 1 if pi.enabled else 0), + "REPLACE INTO policy_instances" + "(id, template, params, agent, budget_id, segment_id, enabled, data_scope) " + "VALUES (?,?,?,?,?,?,?,?)", + ( + pi.id, + pi.template, + json.dumps(pi.params), + pi.agent, + pi.budget_id, + pi.segment_id, + 1 if pi.enabled else 0, + pi.data_scope or "local", + ), ) self._db.commit() return pi @@ -191,7 +270,9 @@ def get_policy_instance(self, pid: str) -> PolicyInstance | None: return _policy(row) if row else None def list_policy_instances(self) -> list[PolicyInstance]: - return [_policy(r) for r in self._db.execute("SELECT * FROM policy_instances ORDER BY template")] + return [ + _policy(r) for r in self._db.execute("SELECT * FROM policy_instances ORDER BY template") + ] def delete_policy_instance(self, pid: str) -> None: self._db.execute("DELETE FROM policy_instances WHERE id=?", (pid,)) @@ -212,8 +293,14 @@ def seed_default_governance_if_empty(self, governance: dict | None = None) -> bo def clear_all(self) -> None: """Delete every row (runs, registrations, governance, ledger). Schema is preserved.""" for table in ( - "runs", "run_registrations", "policy_instances", "budgets", "segments", - "ledger_spent", "ledger_inflight", "ledger_halt", + "runs", + "run_registrations", + "policy_instances", + "budgets", + "segments", + "ledger_spent", + "ledger_inflight", + "ledger_halt", ): self._db.execute(f"DELETE FROM {table}") self._db.commit() @@ -251,12 +338,14 @@ def _apply_governance_yaml(self, governance: dict | None = None) -> bool: for template, raw_params in (governance.get("policies") or {}).items(): params = dict(raw_params or {}) budget_id = params.pop("budget", None) + data_scope = str(params.pop("data_scope", "local") or "local") self.upsert_policy_instance( PolicyInstance( id=f"seed_{template}", template=template, params=params, budget_id=budget_id, + data_scope=data_scope, ) ) return True @@ -281,7 +370,10 @@ def register_run(self, reg: RunRegistration) -> RunRegistration: task=reg.intent or None, ) ) - return reg + self._db.commit() + # Return the persisted row so callers get registered_at (contract §8 — + # the SDK binds this directly, no follow-up GET .../registration). + return self.get_run_registration(reg.run_id) or reg def resolve_run(self, run_id: str) -> RunRegistration: reg = self.get_run_registration(run_id) @@ -290,7 +382,9 @@ def resolve_run(self, run_id: str) -> RunRegistration: return reg def get_run_registration(self, run_id: str) -> RunRegistration | None: - row = self._db.execute("SELECT * FROM run_registrations WHERE run_id=?", (run_id,)).fetchone() + row = self._db.execute( + "SELECT * FROM run_registrations WHERE run_id=?", (run_id,) + ).fetchone() return _registration(row) if row else None # ---- the bridge to build_governor ------------------------------------- # @@ -303,8 +397,11 @@ def governance_config_for(self, agent: str) -> dict: for segment-scoped templates. One instance per template (last wins) — matches the Governor's name-routed registration. """ - instances = [pi for pi in self.list_policy_instances() - if pi.enabled and (pi.agent is None or pi.agent == agent)] + instances = [ + pi + for pi in self.list_policy_instances() + if pi.enabled and (pi.agent is None or pi.agent == agent) + ] budget_ids = {pi.budget_id for pi in instances if pi.budget_id} budgets = [_budget_dict(self.get_budget(bid)) for bid in budget_ids if self.get_budget(bid)] @@ -319,6 +416,7 @@ def governance_config_for(self, agent: str) -> dict: params.setdefault("dimension", seg.dimension) if seg.tag_key: params.setdefault("tag_key", seg.tag_key) + params.setdefault("data_scope", pi.data_scope or "local") policies[pi.template] = params return {"governance": {"budgets": budgets, "policies": policies}} @@ -331,14 +429,37 @@ def create_run(self, rec: RunRecord) -> RunRecord: "REPLACE INTO runs(run_id, agent, status, parent_run, parent_span, halt_reason, detector, " "cost_micros, steps, started_at, ended_at, task, dims, governance_events) " "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)", - (rec.run_id, rec.agent, rec.status, rec.parent_run, rec.parent_span, rec.halt_reason, - rec.detector, rec.cost_micros, rec.steps, rec.started_at, rec.ended_at, rec.task, - json.dumps(rec.dims), json.dumps(rec.governance_events)), + ( + rec.run_id, + rec.agent, + rec.status, + rec.parent_run, + rec.parent_span, + rec.halt_reason, + rec.detector, + rec.cost_micros, + rec.steps, + rec.started_at, + rec.ended_at, + rec.task, + json.dumps(rec.dims), + json.dumps(rec.governance_events), + ), ) self._db.commit() return rec def update_run(self, run_id: str, **fields) -> None: + # steps / cost_micros are derived (run_state / ledger_spent). A client-sent + # value is dropped in 0.2.x (logged) and rejected in 0.3.0. Contract §8. + for derived in ("steps", "cost_micros"): + if fields.pop(derived, None) is not None: + _log.warning( + "update_run(%s): ignoring client-sent %r — derived server-side " + "(deprecated in 0.2.x)", + run_id, + derived, + ) if not fields: return if "governance_events" in fields and not isinstance(fields["governance_events"], str): @@ -371,7 +492,11 @@ def run_tag_keys(self, *, limit: int = 500) -> list[str]: # ---- shared ledger (cross-process spend / inflight / halt) ------------ # def ledger_add_spent( - self, budget_id: str, segment_key: str, period: str, delta: int, + self, + budget_id: str, + segment_key: str, + period: str, + delta: int, ) -> int: """Atomically increment a budget accumulator; return the new total.""" self._db.execute( @@ -404,7 +529,8 @@ def ledger_admit(self, segment_key: str) -> int: (segment_key,), ) row = self._db.execute( - "SELECT count FROM ledger_inflight WHERE segment_key=?", (segment_key,), + "SELECT count FROM ledger_inflight WHERE segment_key=?", + (segment_key,), ).fetchone() self._db.commit() return int(row[0]) if row else 0 @@ -415,14 +541,16 @@ def ledger_complete(self, segment_key: str) -> int: (segment_key,), ) row = self._db.execute( - "SELECT count FROM ledger_inflight WHERE segment_key=?", (segment_key,), + "SELECT count FROM ledger_inflight WHERE segment_key=?", + (segment_key,), ).fetchone() self._db.commit() return int(row[0]) if row else 0 def ledger_inflight(self, segment_key: str) -> int: row = self._db.execute( - "SELECT count FROM ledger_inflight WHERE segment_key=?", (segment_key,), + "SELECT count FROM ledger_inflight WHERE segment_key=?", + (segment_key,), ).fetchone() return int(row[0]) if row else 0 @@ -441,13 +569,15 @@ def ledger_mark_halted(self, run_id: str, reason: str = "") -> None: def ledger_is_halted(self, run_id: str) -> bool: row = self._db.execute( - "SELECT halted FROM ledger_halt WHERE run_id=?", (run_id,), + "SELECT halted FROM ledger_halt WHERE run_id=?", + (run_id,), ).fetchone() return bool(row and row[0]) def ledger_halt_reason(self, run_id: str) -> str | None: row = self._db.execute( - "SELECT halt_reason FROM ledger_halt WHERE run_id=?", (run_id,), + "SELECT halt_reason FROM ledger_halt WHERE run_id=?", + (run_id,), ).fetchone() return row[0] if row else None @@ -459,6 +589,238 @@ def ledger_clear_halt(self, run_id: str) -> None: ) self._db.commit() + # ---- batched ledger events (contract §5) ----------------------------- # + + def apply_events(self, tenant_id: str, events: list[dict]) -> dict: + """Apply a batch of LedgerEvents in array order, in one transaction. + + Returns ``{accepted, deduped, atomic, totals, halted}``. ``totals`` is the + post-commit ``spent_micros`` for every ``(budget_id, segment_key, period)`` + touched by a ``spent_add`` in this batch. Idempotency: a replayed + ``idempotency_key`` (per tenant) is counted in ``deduped`` and not re-applied. + """ + accepted = 0 + deduped = 0 + touched: set[tuple[str, str, str]] = set() + run_ids: set[str] = set() + now = time.time() + + with self._lock: + # sqlite3 (isolation_level="") auto-opens a transaction on the first DML; + # commit()/rollback() below bound it. No explicit BEGIN (would nest). + try: + for i, ev in enumerate(events): + kind = ev.get("kind") + key = str(ev.get("idempotency_key") or "").strip() + if not key: + raise ValueError(f"event {i} missing idempotency_key") + # Record which accumulators this batch references *before* the dedup + # check, so a fully-deduped batch still gets current totals in the ack. + if kind == "spent_add": + for t in ev.get("targets") or []: + touched.add( + ( + str(t["budget_id"]), + str(t["segment_key"]), + str(t.get("period", "lifetime")), + ) + ) + if ev.get("run_id"): + run_ids.add(str(ev["run_id"])) + seen = self._db.execute( + "SELECT 1 FROM ledger_events WHERE tenant_id=? AND idempotency_key=?", + (tenant_id, key), + ).fetchone() + if seen: + deduped += 1 + continue + self._apply_one(tenant_id, kind, ev, touched, run_ids) + self._db.execute( + "INSERT INTO ledger_events(tenant_id, idempotency_key, applied_at) " + "VALUES (?,?,?)", + (tenant_id, key, now), + ) + accepted += 1 + self._db.commit() + + totals = { + f"{b}|{s}|{p}": self.ledger_get_spent(b, s, p) for (b, s, p) in sorted(touched) + } + halted = any(self.ledger_is_halted(r) for r in run_ids) + except Exception: + self._db.rollback() + raise + + return { + "accepted": accepted, + "deduped": deduped, + "atomic": True, + "totals": totals, + "halted": halted, + } + + def _apply_one( + self, + tenant_id: str, + kind: str | None, + ev: dict, + touched: set[tuple[str, str, str]], + run_ids: set[str], + ) -> None: + """Apply a single event inside the open transaction. Raw SQL only — the public + ``ledger_*`` helpers commit, which we must not do mid-batch.""" + run_id = str(ev.get("run_id") or "") + if run_id: + run_ids.add(run_id) + + if kind == "spent_add": + delta = int(ev.get("delta_micros", 0)) + for t in ev.get("targets") or []: + b = str(t["budget_id"]) + s = str(t["segment_key"]) + p = str(t.get("period", "lifetime")) + self._db.execute( + "INSERT INTO ledger_spent(budget_id, segment_key, period, spent_micros) " + "VALUES (?,?,?,?) ON CONFLICT(budget_id, segment_key, period) " + "DO UPDATE SET spent_micros = spent_micros + excluded.spent_micros", + (b, s, p, delta), + ) + touched.add((b, s, p)) + + elif kind == "admit": + self._db.execute( + "INSERT INTO ledger_inflight(segment_key, count) VALUES (?, 1) " + "ON CONFLICT(segment_key) DO UPDATE SET count = count + 1", + (str(ev["segment_key"]),), + ) + + elif kind == "complete": + self._db.execute( + "UPDATE ledger_inflight SET count = MAX(0, count - 1) WHERE segment_key=?", + (str(ev["segment_key"]),), + ) + + elif kind == "step": + self._apply_step(tenant_id, run_id, ev) + + elif kind == "halt_mark": + reason = ev.get("reason") or None + self._db.execute( + "INSERT INTO ledger_halt(run_id, halted, halt_reason) VALUES (?, 1, ?) " + "ON CONFLICT(run_id) DO UPDATE SET halted=1, " + "halt_reason=COALESCE(excluded.halt_reason, ledger_halt.halt_reason)", + (run_id, reason), + ) + self._db.execute( + "UPDATE runs SET status='halted', halt_reason=? WHERE run_id=?", + (reason, run_id), + ) + + elif kind == "halt_clear": + self._db.execute( + "INSERT INTO ledger_halt(run_id, halted, halt_reason) VALUES (?, 0, NULL) " + "ON CONFLICT(run_id) DO UPDATE SET halted=0, halt_reason=NULL", + (run_id,), + ) + + else: + raise ValueError(f"unknown event kind {kind!r}") + + def _apply_step(self, tenant_id: str, run_id: str, ev: dict) -> None: + row = self._db.execute( + "SELECT step_count, window_json FROM run_state WHERE tenant_id=? AND run_id=?", + (tenant_id, run_id), + ).fetchone() + window = json.loads(row["window_json"]) if row else [] + step_entry = { + k: ev.get(k) + for k in ( + "agent", + "seq", + "node_type", + "boundary_id", + "cost_micros", + "cum_spent_micros", + "usage", + "tags", + "tool_signature", + "result_hash", + "ts", + ) + if ev.get(k) is not None + } + window.append(step_entry) + window = window[-RUN_STATE_WINDOW:] + step_count = (row["step_count"] if row else 0) + 1 + velocity = _velocity_from_window(window) + self._db.execute( + "INSERT INTO run_state(tenant_id, run_id, step_count, window_json, " + "velocity_micros_per_step, last_ts) VALUES (?,?,?,?,?,?) " + "ON CONFLICT(tenant_id, run_id) DO UPDATE SET " + "step_count=excluded.step_count, window_json=excluded.window_json, " + "velocity_micros_per_step=excluded.velocity_micros_per_step, last_ts=excluded.last_ts", + (tenant_id, run_id, step_count, json.dumps(window), velocity, ev.get("ts")), + ) + + def precheck( + self, + tenant_id: str, + run_id: str, + *, + segment_keys: list[str] | None = None, + budgets: list[dict] | None = None, + want: list[str] | None = None, + ) -> dict: + """Consolidated pre_call read — contract §4. Returns halt + the requested + spend / inflight / window slices in one shot.""" + want = want or ["spent", "inflight", "halt"] + out: dict = {"server_ts": time.time()} + + if "halt" in want: + out["halted"] = self.ledger_is_halted(run_id) + out["halt_reason"] = self.ledger_halt_reason(run_id) + + if "spent" in want: + spent: dict[str, int] = {} + for b in budgets or []: + bid = str(b["budget_id"]) + seg = str(b["segment_key"]) + per = str(b.get("period", "lifetime")) + spent[f"{bid}|{seg}|{per}"] = self.ledger_get_spent(bid, seg, per) + out["spent"] = spent + + if "inflight" in want: + out["inflight"] = {seg: self.ledger_inflight(seg) for seg in (segment_keys or [])} + + if "window" in want: + st = self.get_run_state(tenant_id, run_id) + out["window"] = ( + { + "step_count": st["step_count"], + "recent": st["recent"], + "velocity_micros_per_step": st["velocity_micros_per_step"], + } + if st + else {"step_count": 0, "recent": [], "velocity_micros_per_step": 0.0} + ) + + return out + + def get_run_state(self, tenant_id: str, run_id: str) -> dict | None: + row = self._db.execute( + "SELECT step_count, window_json, velocity_micros_per_step, last_ts " + "FROM run_state WHERE tenant_id=? AND run_id=?", + (tenant_id, run_id), + ).fetchone() + if row is None: + return None + return { + "step_count": row["step_count"], + "recent": json.loads(row["window_json"]), + "velocity_micros_per_step": row["velocity_micros_per_step"], + "last_ts": row["last_ts"], + } + def create_api_key( self, *, @@ -527,23 +889,36 @@ def delete_api_key(self, kid: str) -> None: # ---- row -> model ---------------------------------------------------------- # + def _registration(r: sqlite3.Row) -> RunRegistration: + keys = r.keys() return RunRegistration( run_id=r["run_id"], intent=r["intent"] or "", user_dims=json.loads(r["user_dims"] or "{}"), - mode=parse_governance_mode(r["mode"] if "mode" in r.keys() else None), + mode=parse_governance_mode(r["mode"] if "mode" in keys else None), + registered_at=float(r["registered_at"] or 0.0) if "registered_at" in keys else 0.0, ) def _segment(r: sqlite3.Row) -> Segment: - return Segment(id=r["id"], name=r["name"], dimension=r["dimension"], - tag_key=r["tag_key"], match_value=r["match_value"]) + return Segment( + id=r["id"], + name=r["name"], + dimension=r["dimension"], + tag_key=r["tag_key"], + match_value=r["match_value"], + ) def _budget(r: sqlite3.Row) -> BudgetSpec: - return BudgetSpec(id=r["id"], limit_micros=r["limit_micros"], dimension=r["dimension"], - tag_key=r["tag_key"], period=r["period"]) + return BudgetSpec( + id=r["id"], + limit_micros=r["limit_micros"], + dimension=r["dimension"], + tag_key=r["tag_key"], + period=r["period"], + ) def _budget_dict(b: BudgetSpec) -> dict: @@ -554,9 +929,17 @@ def _budget_dict(b: BudgetSpec) -> dict: def _policy(r: sqlite3.Row) -> PolicyInstance: - return PolicyInstance(id=r["id"], template=r["template"], params=json.loads(r["params"]), - agent=r["agent"], budget_id=r["budget_id"], segment_id=r["segment_id"], - enabled=bool(r["enabled"])) + keys = r.keys() + return PolicyInstance( + id=r["id"], + template=r["template"], + params=json.loads(r["params"]), + agent=r["agent"], + 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") or "local", + ) def _run(r: sqlite3.Row) -> RunRecord: @@ -568,13 +951,19 @@ def _run(r: sqlite3.Row) -> RunRecord: except json.JSONDecodeError: governance_events = [] return RunRecord( - run_id=r["run_id"], agent=r["agent"], status=r["status"], + run_id=r["run_id"], + agent=r["agent"], + status=r["status"], parent_run=r["parent_run"], parent_span=r["parent_span"] if "parent_span" in keys else None, - halt_reason=r["halt_reason"], detector=r["detector"], - cost_micros=r["cost_micros"], steps=r["steps"], + halt_reason=r["halt_reason"], + detector=r["detector"], + cost_micros=r["cost_micros"], + steps=r["steps"], started_at=r["started_at"], - ended_at=r["ended_at"], task=r["task"], dims=dims, + ended_at=r["ended_at"], + task=r["task"], + dims=dims, governance_events=governance_events, ) diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..d41149d --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,23 @@ +services: + control-plane: + build: . + image: agentplane-control-plane:local + ports: + - "8800:8800" + volumes: + - control-plane-data:/data + environment: + # Comma-separated key:tenant:scope+scope,... — empty = local anonymous, all + # scopes (fine for a single trusted host; set real keys before exposing this + # beyond localhost). + CONTROL_PLANE_API_KEYS: "" + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8800/health', timeout=2)"] + interval: 10s + timeout: 3s + retries: 3 + start_period: 5s + restart: unless-stopped + +volumes: + control-plane-data: diff --git a/docs/DESIGN.md b/docs/DESIGN.md index c27caa3..f0bace1 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -40,13 +40,23 @@ No unbounded `GET /envelopes`. No single-envelope `POST`. ## TokenOps -Agent: +Full spec: [`api-contract.md`](api-contract.md). -- `POST /v1/runs` register (`intent`, `user_dims`, `mode`) +Agent (0.2.x): + +- `POST /v1/runs` register (`intent`, `user_dims`, `mode`) → returns the full record + incl. `registered_at` - `GET /v1/runs/{id}/registration` -- `GET /v1/governance/{agent}` -- `/v1/ledger/*` spend, inflight, halt -- `PUT/PATCH /v1/run-records` dashboard row + governance events +- `GET /v1/governance/{agent}` — policy params now carry `data_scope` (`local`|`global`) +- `POST /v1/ledger/precheck` — one read per pre_call pass (halt + spent + inflight + + optional window) +- `POST /v1/ledger/events:batch` — one write per crossing (`spent_add` / `admit` / + `complete` / `step` / `halt_mark` / `halt_clear`), idempotent +- `GET|POST|DELETE /v1/ledger/runs/{id}/halt` +- `PATCH /v1/run-records/{id}` — status / ended_at / halt_reason / detector / + governance_events (`steps` / `cost_micros` are derived, ignored) +- Legacy, deprecated → removed in 0.3.0: single-op `/v1/ledger/{spent,inflight,halt}/*`, + `PUT /v1/run-records` UI: diff --git a/docs/api-contract.md b/docs/api-contract.md new file mode 100644 index 0000000..a8caa04 --- /dev/null +++ b/docs/api-contract.md @@ -0,0 +1,320 @@ +# TokenOps ⇄ control-plane wire contract (0.2.0) + +**Status:** authoritative for the 0.2.0 build. +**Applies to:** `agentplane-control-plane` **0.2.0** ⇄ `tokenops` **``** (remote-only). +**0.2.0 is additive** — new `precheck` / `events:batch` routes, new `ledger_events` / +`run_state` tables, `data_scope` column. 0.1-era clients keep working. The breaking +fold (drop `run_registrations`, remove `create_run`, narrow `PATCH`) is **0.3.0**, +shipped after `tokenops `. See §13, §15. + +Epic: theagentplane/control-plane#10 · TokenOps epic: theagentplane/tokenops#118. + +--- + +## 1. Ratified decisions + +| # | Decision | +|---|---| +| D1 | Consolidated read: `POST /v1/ledger/precheck`. | +| D2 | Batch write: `POST /v1/ledger/events:batch`. Single-op ledger routes become thin wrappers over the same apply path. | +| D3 | Idempotency key **required** on every write event; plane dedupes. | +| D4 | Run state (`steps`, `window`, `velocity`) lives on the plane (`run_state` table), fed by `step` events. The SDK keeps a per-process `LocalRunState` cache. | +| D5 | Policies carry `data_scope ∈ {local, global}`. `local` reads the SDK cache only; `global` reads the plane via `precheck`. | +| D6 | No client-side event buffer yet. `apply_events(list)` is the only write path so a buffer can wrap it later. | +| D7 | No accuracy-loss bounding. Budgets remain a soft cap. | +| D8 | `trajectory_hint` disabled → **no `trajectory/*` routes**. | +| D10 | Budget fan-out = **one** `spent_add` event carrying all `targets`; plane applies them in one transaction. | +| D11 | Inflight counter safety (call-id set) is out of scope → tokenops#116. This contract assumes synchronous `admit`/`complete`. | +| D12 | Identity columns move to `runs`, write-once; `create_run` removed — **staged**: deprecated shims in 0.2.x, `run_registrations` dropped and `create_run` removed in 0.3.0, after `tokenops ` (tokenops#117). | +| D13 | One `runs` row per `run_id` — no per-agent participation rows. Per-agent spend = an `agent`-dimension accumulator in `ledger_spent`. | +| D14 | `span_id` removed from TokenOps. Cross-hop link = `X-TokenOps-Run-Id` only. No `X-TokenOps-Parent-Span-Id`. | + +--- + +## 2. Auth (unchanged from 0.1.x) + +Bearer token → `Principal{tenant_id, scopes ⊆ {ingest, read, admin}}`. +Empty key table ⇒ anonymous `tenant="local"`, all scopes (local/dev/test only). +Isolation is **per-tenant**; no per-run ownership check (out of scope — see #6). + +Route scopes: reads = `read`, ledger/run writes = `ingest`, halt-clear + admin = `admin`. + +--- + +## 3. Conventions + +- **Money:** micro-USD integers. `$1.00 == 1_000_000`. Field suffix `_micros`. +- **`segment_key`** (client-computed, opaque to the plane): `run:`, + `user:`, `agent:`, `tenant:`, `tag:=`. +- **`spent` map key** in responses: `"||"`, `period` + is `"lifetime"` for v1. +- **`run_id`** is the only cross-hop identifier. Header `X-TokenOps-Run-Id`. +- **`seq`** (in event keys): the emitting process's step counter at emit time — + per-`(run_id, agent)` monotonic, regenerated identically on replay. +- Timestamps: client wall-clock float (`ts`), used for ordering/display only. + +--- + +## 4. `POST /v1/ledger/precheck` *(new — scope: read)* + +The single read a `pre_call` pass needs. Request shape is fixed per governor, computed +once from the registered `global` detectors. + +### Request +```jsonc +{ + "run_id": "run_abc", + "segment_keys": ["run:run_abc", "agent:writer"], + "budgets": [ + { "budget_id": "run_llm_cap", "segment_key": "run:run_abc", "period": "lifetime" } + ], + "want": ["spent", "inflight", "halt", "window"] // window optional +} +``` + +### Response `200` +```jsonc +{ + "server_ts": 1732200000.12, + "halted": false, + "halt_reason": null, + "spent": { "run_llm_cap|run:run_abc|lifetime": 10500 }, + "inflight": { "run:run_abc": 0 }, + "window": { "step_count": 3, "recent": [ /* BoundaryStep */ ], + "velocity_micros_per_step": 3500 } // only if "window" in want +} +``` + +- Read-only, no idempotency. +- `observe`-phase reads are **not** a second `precheck` — the `events:batch` ack + returns fresh `totals`. + +--- + +## 5. `POST /v1/ledger/events:batch` *(new — scope: ingest)* + +Mirrors the `envelopes:batch` shape. All ledger writes go through here. + +### Headers +`Durability: sync | queued` — default `sync`. `queued` acks after the SQLite commit +(same as `sync` in 0.2.0; real queue is later). + +### Request +```jsonc +{ "events": [ LedgerEvent, ... ] } // applied strictly in array order +``` + +### `LedgerEvent` union +```jsonc +// common to every kind +{ "kind": "...", "idempotency_key": "", "ts": 1732200000.1, "run_id": "run_abc" } + +// spent_add — one per priced crossing, carries ALL budget targets (D10) +{ "kind": "spent_add", "delta_micros": 10500, + "targets": [ { "budget_id": "__run_total__", "segment_key": "run:run_abc", "period": "lifetime" }, + { "budget_id": "run_llm_cap", "segment_key": "run:run_abc", "period": "lifetime" } ] } + +// admit / complete — concurrency; synchronous only (D11) +{ "kind": "admit", "segment_key": "run:run_abc" } +{ "kind": "complete", "segment_key": "run:run_abc" } + +// step — the priced BoundaryStep; moves run window to the plane (D4) +{ "kind": "step", "agent": "researcher", "seq": 3, "node_type": "llm", + "boundary_id": "researcher.chat", "cost_micros": 10500, "cum_spent_micros": 10500, + "tool_signature": null, "result_hash": null, + "usage": { "input": 1000, "output": 500 }, + "tags": { "provider": "anthropic", "model": "claude-sonnet-4-6" } } + +// halt_mark / halt_clear +{ "kind": "halt_mark", "reason": "step_cap: 20 steps", "detector": "step_cap" } +{ "kind": "halt_clear" } +``` + +### Response `201` +```jsonc +{ + "accepted": 3, "deduped": 0, "atomic": true, + "totals": { "run_llm_cap|run:run_abc|lifetime": 15300 }, + "halted": false +} +``` + +### Semantics +- Events applied **in order**, in **one SQLite transaction** per batch (all-or-nothing). + `"atomic": false` in the ack if the server had to fall back to per-event apply. +- `spent_add` → upsert every `target` in `ledger_spent`. +- `step` → upsert the `(tenant_id, run_id)` row in `run_state` (`step_count`, bounded + `window_json` ring, velocity inputs, `last_ts`). +- `halt_mark` / `halt_clear` → `ledger_halt` (+ flip `runs.status` when the run row + exists). +- Zero-cost crossings emit **only** a `step` — no `spent_add` with `delta_micros: 0`. + `ledger_spent` moves on priced events only. +- `len(events) == 1` ⇒ immediate flush. +- `413` if body exceeds `max_body`; `400` if `len(events) > max_batch`, unknown `kind`, + missing `idempotency_key`, or bad `Durability`. + +--- + +## 6. Idempotency *(D3)* + +- Table `ledger_events(tenant_id, idempotency_key PRIMARY KEY, applied_at)`. +- A replayed key is counted in `deduped`, **not re-applied**; the ack still returns + current `totals`. +- Deterministic key recipe (so a retried flush regenerates identical keys): + - `spent_add` / `step`: `f"{run_id}:{agent}:{seq}:{kind}"` + - `admit` / `complete`: `f"{run_id}:{segment_key}:{kind}:{monotonic_seq}"` + - `halt_mark` / `halt_clear`: `f"{run_id}:{kind}:{detector or ''}:{monotonic_seq}"` +- Legacy single-op routes accept an optional `Idempotency-Key` header routed through + the same table. + +--- + +## 7. `/v1/ledger/runs/{run_id}/halt` *(new shape — replaces `/v1/ledger/halt/*`)* + +| Method | Scope | Body | Response | +|---|---|---|---| +| `GET` | read | — | `{ "halted": bool, "halt_reason": str\|null }` | +| `POST` | ingest | `{ "reason": str, "detector"?: str }` | `{ "halted": true }` — flips `runs.status="halted"` | +| `DELETE` | admin | — | `{ "halted": false }` — explicit resume | + +Legacy `POST /v1/ledger/halt/mark` · `/clear` · `GET /v1/ledger/halt/{run_id}` kept as +wrappers for one release, then removed. + +--- + +## 8. Run identity *(#117 / D12)* + +### `POST /v1/runs` *(changed return — scope: ingest)* +Request unchanged: `{ run_id?, intent, user_dims, mode }`. +**Response returns the full record** (kills register-then-resolve): +```jsonc +{ "run_id": "run_abc", "status": "registered", + "intent": "research-brief", "user_dims": { "user_id": "alice" }, + "mode": "enforce", "registered_at": 1732200000.0 } +``` +`409` if `run_id` already registered. The SDK binds this directly. + +### `GET /v1/runs/{run_id}/registration` *(unchanged surface — scope: read)* +Serves identity columns from `runs`. `404` `{ "error": "run '' is not registered" }`. + +### `PATCH /v1/run-records/{run_id}` *(scope: ingest)* +Preferred allow-list: `status`, `ended_at`, `halt_reason`, `detector`, +`governance_events`. In **0.2.x** `steps` / `cost_micros` are still accepted but +**ignored** (logged as deprecated) — they are derived from `run_state` / `ledger_spent`. +**0.3.0** narrows to the allow-list and rejects the rest. + +### Deprecated in 0.2.x, removed in 0.3.0 +- `PUT /v1/run-records` (`create_run`) — still works (writes `runs`) but the SDK must + stop calling it; registration is the run-row creator. +- `run_registrations` table — still present. The 0.3.0 fold drops it. +- Legacy `/v1/ledger/halt/*` and single-op `/v1/ledger/{spent,inflight}/*` writes — + wrappers over the new paths / `apply_events`. + +**0.2.0 is additive on the wire and in the schema.** A 0.1-era client keeps working; +the break is deferred to 0.3.0, which ships after `tokenops `. + +--- + +## 9. `run_state` table (Tier-2) *(new — D4)* + +| column | note | +|---|---| +| `tenant_id, run_id` | PK | +| `step_count` | monotonic; sum of `step` events across all agents on the run | +| `window_json` | bounded ring of the last N `BoundaryStep`s (N configurable, default 64) | +| `velocity_inputs` | enough to compute `velocity_micros_per_step` over the ring | +| `last_ts` | latest `step.ts` seen | + +Exposed only through `precheck want:["window"]`. `runs.steps` = read of +`run_state.step_count`; `runs.cost_micros` = read of +`ledger_spent[__run_total__|run:|lifetime]`. + +--- + +## 10. `data_scope` on policies *(D5)* + +- Column `policy_instances.data_scope TEXT NOT NULL DEFAULT 'local'` (`'local'` | `'global'`). +- Serde + included in `GET /v1/governance/{agent}` output: + ```jsonc + { "governance": { "budgets": [...], + "policies": { "cost_budget": { "budget": "run_llm_cap", "data_scope": "global" }, + "output_runaway": { "repeats": 4, "data_scope": "local" } } } } + ``` +- Code default per template lives in the SDK; admin may override per instance. + +--- + +## 11. Error model + client retry rules + +| Situation | HTTP | Client action | +|---|---|---| +| run already registered | `409` | `RunAlreadyRegisteredError` (join instead) | +| registration not found | `404` | `RunNotRegisteredError` — fail closed | +| bad event / batch too large / bad `Durability` | `400` | bug — do not retry; raise | +| auth | `401` / `403` | config error — raise, no retry | +| `5xx` / connect / timeout on `precheck` | — | retry ≤2× over ~250 ms, then trip circuit breaker | +| `5xx` / connect on `events:batch` | — | retry ≤2×, then retain + flush on breaker close; never raise (crossing already happened) | +| `5xx` / connect on `POST /v1/runs` | — | retry ≤2×, then per mode | + +`max_batch` / `max_body` advertised on `GET /health`. + +--- + +## 12. Plane-unreachable — client contract *(SDK-side, informational here)* + +`TOKENOPS_PLANE_UNREACHABLE ∈ {fail-fast, passthrough, local-policies-only}` + +`TOKENOPS_FALLBACK_RUN_CAP_MICROS`. A per-process circuit breaker in the SDK trips on +first failure (or a failed `GET /ready`), short-circuits all plane calls until a probe +closes it. Plane side: keep `GET /ready` cheap. + +--- + +## 13. Schema & migrations + +Tracked in `PRAGMA user_version` (`SqliteStore.SCHEMA_VERSION`). `_SCHEMA` +(`CREATE TABLE IF NOT EXISTS`) runs on every open; `_apply_migrations()` does the rest +forward-only. + +### v2 — 0.2.0 (additive, auto-applied) +- `ledger_events (tenant_id, idempotency_key PRIMARY KEY, applied_at)` — new table. +- `run_state (tenant_id, run_id PK, step_count, window_json, + velocity_micros_per_step, last_ts)` — new table. +- `policy_instances.data_scope TEXT NOT NULL DEFAULT 'local'` — new column (`_migrate` + `ALTER` for existing DBs). +- `runs`, `run_registrations`, `ledger_spent` / `ledger_inflight` / `ledger_halt` + **unchanged**. Opening a 0.1 DB just adds the new tables/column and bumps + `user_version` to 2 — **no data loss, no break.** + +### v3 — 0.3.0 (destructive, ships after `tokenops ` — issue #11) +- Fold `run_registrations` identity columns into `runs`; `DROP TABLE run_registrations`. +- Make `runs` identity columns write-once; `steps` / `cost_micros` derived only. +- Remove `PUT /v1/run-records` (`create_run`) and the legacy ledger wrappers. +- Historical `runs.cost_micros` / `steps` recomputed from `ledger_spent` where possible, + else nulled (pre-0.2 values are known-incoherent for multi-agent runs). +- Runs automatically as a `_apply_migrations()` step on first open of a v2 DB. + +--- + +## 14. Out of scope + +- `trajectory/*` routes (D8). +- Envelope ingest from the TokenOps SDK. +- Client-side buffering. +- Inflight-as-call-id-set (#116) — synchronous counter assumed. +- Server-side pricing / deriving the ledger from `envelopes:batch`. +- Per-run authorization (#6). +- Real `Durability: queued` queue. + +--- + +## 15. Compatibility + +| tokenops | agentplane-control-plane | agent-chronicle | notes | +|---|---|---|---| +| ≤ 0.2.1 | 0.1.x **or 0.2.x** | ≥ 0.3.0 | 0.2.0 is additive — old clients keep working against it | +| `` (remote-only) | ≥ 0.2.0 | ≥ 0.3.0 | uses `precheck` / `events:batch` | +| `` | **not** 0.3.0+ until it drops `create_run` | ≥ 0.3.0 | 0.3.0 removes the deprecated shims | + +The hard break is **control-plane 0.3.0**, released only after `tokenops ` stops +calling `PUT /v1/run-records`. Publish this table in both repo READMEs with a +**Breaking changes** subsection. diff --git a/pyproject.toml b/pyproject.toml index a87f15c..ab7b288 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "agentplane-control-plane" -version = "0.1.0" +version = "0.2.0" description = "Shared AgentPlane control plane: Chronicle envelopes + TokenOps governance over HTTP." readme = "README.md" requires-python = ">=3.10" @@ -40,6 +40,8 @@ dependencies = [ "httpx>=0.27", "pyyaml>=6", "jinja2>=3.1", + "psutil>=5.9", + "platformdirs>=4.0", ] [project.optional-dependencies] diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..e07ea7a --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,48 @@ +"""Shared test fixtures. + +``make_client`` builds a ``TestClient`` over a fresh SQLite file and — crucially on +Windows — closes the store connections on teardown before the temp dir is removed +(open SQLite handles block ``unlink`` on Windows; POSIX does not care). +""" + +from __future__ import annotations + +import gc +import shutil +import tempfile +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +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 + + +@pytest.fixture +def make_client(): + created: list[tuple] = [] + + def _make(*, auto_seed: bool = False, **settings_kw) -> TestClient: + td = tempfile.mkdtemp() + db = str(Path(td) / "t.db") + store = SqliteStore(db, auto_seed=auto_seed) + env = EnvelopeStore(db) + app = create_app(store=store, envelopes=env, settings=Settings(db_path=db, **settings_kw)) + client = TestClient(app) + created.append((client, store, env, td)) + return client + + yield _make + + for client, store, env, td in created: + try: + client.close() + except Exception: + pass + store.close() + env.close() + gc.collect() + shutil.rmtree(td, ignore_errors=True) diff --git a/tests/test_data_scope.py b/tests/test_data_scope.py new file mode 100644 index 0000000..d360e5d --- /dev/null +++ b/tests/test_data_scope.py @@ -0,0 +1,77 @@ +"""policy_instances.data_scope — contract §10.""" + +from __future__ import annotations + +import tempfile +from pathlib import Path + +from control_plane.models import PolicyInstance +from control_plane.store import SqliteStore + + +def _store(td): + return SqliteStore(str(Path(td) / "t.db"), auto_seed=False) + + +def test_default_is_local(): + with tempfile.TemporaryDirectory() as td: + s = _store(td) + try: + s.upsert_policy_instance( + PolicyInstance(id="p1", template="step_cap", params={"max_steps": 5}) + ) + assert s.get_policy_instance("p1").data_scope == "local" + finally: + s.close() + + +def test_roundtrips_global(): + with tempfile.TemporaryDirectory() as td: + s = _store(td) + try: + s.upsert_policy_instance( + PolicyInstance(id="p2", template="cost_budget", budget_id="b", data_scope="global") + ) + assert s.get_policy_instance("p2").data_scope == "global" + assert [p.data_scope for p in s.list_policy_instances()] == ["global"] + finally: + s.close() + + +def test_governance_config_for_emits_data_scope(): + with tempfile.TemporaryDirectory() as td: + s = _store(td) + try: + s.upsert_policy_instance( + PolicyInstance(id="p3", template="cost_budget", budget_id="b", data_scope="global") + ) + s.upsert_policy_instance( + PolicyInstance(id="p4", template="output_runaway", params={"repeats": 4}) + ) + cfg = s.governance_config_for("any")["governance"]["policies"] + assert cfg["cost_budget"]["data_scope"] == "global" + assert cfg["output_runaway"]["data_scope"] == "local" + finally: + s.close() + + +def test_yaml_seed_reads_data_scope(): + with tempfile.TemporaryDirectory() as td: + s = _store(td) + try: + s.reseed_governance( + { + "budgets": [{"id": "cap", "limit_micros": 1000, "dimension": "run"}], + "policies": { + "cost_budget": {"budget": "cap", "data_scope": "global"}, + "step_cap": {"max_steps": 3}, + }, + } + ) + got = {p.template: p.data_scope for p in s.list_policy_instances()} + assert got["cost_budget"] == "global" + assert got["step_cap"] == "local" + # data_scope must not leak into params + assert "data_scope" not in s.get_policy_instance("seed_cost_budget").params + finally: + s.close() diff --git a/tests/test_envelopes.py b/tests/test_envelopes.py index 39070a2..64062c7 100644 --- a/tests/test_envelopes.py +++ b/tests/test_envelopes.py @@ -1,142 +1,120 @@ from __future__ import annotations -import tempfile -from pathlib import Path -from fastapi.testclient import TestClient - -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 - - -def _client(tmp: Path, api_keys: str = "") -> TestClient: - db = str(tmp / "t.db") - settings = Settings(db_path=db, api_keys=api_keys) - app = create_app( - store=SqliteStore(db, auto_seed=False), - envelopes=EnvelopeStore(db), - settings=settings, +def test_batch_ingest_and_query_by_trace(make_client): + client = make_client() + body = { + "envelopes": [ + { + "envelope_id": "e1", + "trace_id": "tr1", + "node_id": "llm", + "dims": {"session_id": "s1", "message_id": "m1"}, + "timestamp": "2026-01-01T00:00:00Z", + } + ] + } + r = client.post("/v1/envelopes:batch", json=body) + assert r.status_code == 201 + assert r.json()["accepted"] == 1 + + r2 = client.post("/v1/envelopes:batch", json=body) + assert r2.status_code == 201 + assert r2.json()["deduped"] == 1 + + found = client.get("/v1/traces", params={"session_id": "s1", "message_id": "m1"}) + assert found.status_code == 200 + assert found.json()["traces"][0]["trace_id"] == "tr1" + + envs = client.get("/v1/traces/tr1/envelopes") + assert len(envs.json()["envelopes"]) == 1 + + assert ( + client.post("/v1/envelopes", json={"envelope_id": "x", "trace_id": "y"}).status_code == 404 ) - return TestClient(app) - - -def test_batch_ingest_and_query_by_trace(): - with tempfile.TemporaryDirectory() as td: - client = _client(Path(td)) - body = { - "envelopes": [ - { - "envelope_id": "e1", - "trace_id": "tr1", - "node_id": "llm", - "dims": {"session_id": "s1", "message_id": "m1"}, - "timestamp": "2026-01-01T00:00:00Z", - } - ] - } - r = client.post("/v1/envelopes:batch", json=body) - assert r.status_code == 201 - assert r.json()["accepted"] == 1 - - r2 = client.post("/v1/envelopes:batch", json=body) - assert r2.status_code == 201 - assert r2.json()["deduped"] == 1 - found = client.get("/v1/traces", params={"session_id": "s1", "message_id": "m1"}) - assert found.status_code == 200 - assert found.json()["traces"][0]["trace_id"] == "tr1" - - envs = client.get("/v1/traces/tr1/envelopes") - assert len(envs.json()["envelopes"]) == 1 - - assert client.post("/v1/envelopes", json={"envelope_id": "x", "trace_id": "y"}).status_code == 404 +def test_traces_listed_recent_first(make_client): + client = make_client() + older = { + "envelopes": [ + { + "envelope_id": "old", + "trace_id": "trace-old", + "node_id": "llm", + "timestamp": "2026-01-01T00:00:00Z", + } + ] + } + newer = { + "envelopes": [ + { + "envelope_id": "new", + "trace_id": "trace-new", + "node_id": "llm", + "timestamp": "2026-08-13T12:00:00Z", + } + ] + } + assert client.post("/v1/envelopes:batch", json=older).status_code == 201 + assert client.post("/v1/envelopes:batch", json=newer).status_code == 201 + traces = client.get("/v1/traces").json()["traces"] + assert [t["trace_id"] for t in traces] == ["trace-new", "trace-old"] + + +def test_batch_and_auth(make_client): + client = make_client(api_keys="agent:t1:ingest,reader:t1:read") + assert ( + client.post( + "/v1/envelopes:batch", json={"envelopes": [{"envelope_id": "e", "trace_id": "t"}]} + ).status_code + == 401 + ) -def test_traces_listed_recent_first(): - with tempfile.TemporaryDirectory() as td: - client = _client(Path(td)) - older = { + r = client.post( + "/v1/envelopes:batch", + headers={"Authorization": "Bearer agent"}, + json={ "envelopes": [ - { - "envelope_id": "old", - "trace_id": "trace-old", - "node_id": "llm", - "timestamp": "2026-01-01T00:00:00Z", - } + {"envelope_id": "a", "trace_id": "tr", "dims": {"message_id": "m"}}, + {"envelope_id": "b", "trace_id": "tr", "dims": {"message_id": "m"}}, ] - } - newer = { - "envelopes": [ - { - "envelope_id": "new", - "trace_id": "trace-new", - "node_id": "llm", - "timestamp": "2026-08-13T12:00:00Z", - } - ] - } - assert client.post("/v1/envelopes:batch", json=older).status_code == 201 - assert client.post("/v1/envelopes:batch", json=newer).status_code == 201 - traces = client.get("/v1/traces").json()["traces"] - assert [t["trace_id"] for t in traces] == ["trace-new", "trace-old"] - - -def test_batch_and_auth(): - keys = "agent:t1:ingest,reader:t1:read" - with tempfile.TemporaryDirectory() as td: - client = _client(Path(td), api_keys=keys) - assert client.post( - "/v1/envelopes:batch", json={"envelopes": [{"envelope_id": "e", "trace_id": "t"}]} - ).status_code == 401 - - r = client.post( - "/v1/envelopes:batch", - headers={"Authorization": "Bearer agent"}, - json={ - "envelopes": [ - {"envelope_id": "a", "trace_id": "tr", "dims": {"message_id": "m"}}, - {"envelope_id": "b", "trace_id": "tr", "dims": {"message_id": "m"}}, - ] - }, - ) - assert r.status_code == 201 - assert r.json()["accepted"] == 2 - - assert client.get("/v1/traces/tr", headers={"Authorization": "Bearer agent"}).status_code == 403 - ok = client.get("/v1/traces/tr", headers={"Authorization": "Bearer reader"}) - assert ok.status_code == 200 - assert ok.json()["span_count"] == 2 - - -def test_ui_and_register_run(): - with tempfile.TemporaryDirectory() as td: - client = _client(Path(td)) - home = client.get("/") - assert home.status_code == 200 - assert b"AgentPlane" in home.content - assert b"Chronicle" in home.content - assert b"TokenOps" in home.content - r = client.post("/v1/runs", json={"intent": "demo", "user_dims": {"user_id": "u"}}) - assert r.status_code == 201 - assert r.json()["status"] == "registered" - assert r.json()["mode"] == "enforce" - listed = client.get("/v1/run-records") - assert any(x["run_id"] == r.json()["run_id"] for x in listed.json()) - - -def test_governance_segments_roundtrip(): - with tempfile.TemporaryDirectory() as td: - client = _client(Path(td)) - payload = { - "id": "seg1", - "name": "default", - "dimension": "run", - "tag_key": None, - "match_value": None, - } - r = client.put("/v1/segments", json=payload) - assert r.status_code == 200 - listed = client.get("/v1/segments") - assert any(s["id"] == "seg1" for s in listed.json()) + }, + ) + assert r.status_code == 201 + assert r.json()["accepted"] == 2 + + assert client.get("/v1/traces/tr", headers={"Authorization": "Bearer agent"}).status_code == 403 + ok = client.get("/v1/traces/tr", headers={"Authorization": "Bearer reader"}) + assert ok.status_code == 200 + assert ok.json()["span_count"] == 2 + + +def test_ui_and_register_run(make_client): + client = make_client() + home = client.get("/") + assert home.status_code == 200 + assert b"AgentPlane" in home.content + assert b"Chronicle" in home.content + assert b"TokenOps" in home.content + r = client.post("/v1/runs", json={"intent": "demo", "user_dims": {"user_id": "u"}}) + assert r.status_code == 201 + assert r.json()["status"] == "registered" + assert r.json()["mode"] == "enforce" + listed = client.get("/v1/run-records") + assert any(x["run_id"] == r.json()["run_id"] for x in listed.json()) + + +def test_governance_segments_roundtrip(make_client): + client = make_client() + payload = { + "id": "seg1", + "name": "default", + "dimension": "run", + "tag_key": None, + "match_value": None, + } + r = client.put("/v1/segments", json=payload) + assert r.status_code == 200 + listed = client.get("/v1/segments") + assert any(s["id"] == "seg1" for s in listed.json()) diff --git a/tests/test_health.py b/tests/test_health.py index 32aa303..7353293 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -1,20 +1,8 @@ -from fastapi.testclient import TestClient +def test_health(make_client): + c = make_client() + assert c.get("/health").json()["status"] == "ok" -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 -import tempfile -from pathlib import Path - -def test_health(): - with tempfile.TemporaryDirectory() as td: - db = str(Path(td) / "t.db") - app = create_app( - store=SqliteStore(db, auto_seed=False), - envelopes=EnvelopeStore(db), - settings=Settings(db_path=db), - ) - c = TestClient(app) - assert c.get("/health").json()["status"] == "ok" +def test_ready(make_client): + c = make_client() + assert c.get("/ready").json()["status"] == "ready" diff --git a/tests/test_ledger_events.py b/tests/test_ledger_events.py new file mode 100644 index 0000000..6217c04 --- /dev/null +++ b/tests/test_ledger_events.py @@ -0,0 +1,157 @@ +"""POST /v1/ledger/events:batch — contract §5 / §6.""" + +from __future__ import annotations + + +def _spent_add(key: str, delta: int, *, run_id: str = "run_1") -> dict: + return { + "kind": "spent_add", + "idempotency_key": key, + "run_id": run_id, + "delta_micros": delta, + "targets": [ + {"budget_id": "__run_total__", "segment_key": f"run:{run_id}", "period": "lifetime"}, + {"budget_id": "run_llm_cap", "segment_key": f"run:{run_id}", "period": "lifetime"}, + ], + } + + +def _conc(key: str, kind: str, *, run_id: str = "run_1") -> dict: + return {"kind": kind, "idempotency_key": key, "run_id": run_id, "segment_key": f"run:{run_id}"} + + +def _step(seq: int, cum: int, *, run_id: str = "run_1") -> dict: + return { + "kind": "step", + "idempotency_key": f"{run_id}:researcher:{seq}:step", + "run_id": run_id, + "agent": "researcher", + "seq": seq, + "node_type": "llm", + "boundary_id": "researcher.chat", + "cost_micros": 10_500, + "cum_spent_micros": cum, + "ts": float(seq), + } + + +def test_spent_add_fans_out_and_returns_totals(make_client): + c = make_client() + r = c.post("/v1/ledger/events:batch", json={"events": [_spent_add("k1", 10_500)]}) + assert r.status_code == 201 + body = r.json() + assert body["accepted"] == 1 and body["deduped"] == 0 + assert body["totals"] == { + "__run_total__|run:run_1|lifetime": 10_500, + "run_llm_cap|run:run_1|lifetime": 10_500, + } + assert body["halted"] is False + + +def test_batch_applies_in_order_and_accumulates(make_client): + c = make_client() + r = c.post( + "/v1/ledger/events:batch", + json={"events": [_spent_add("a", 10_500), _spent_add("b", 4_800)]}, + ) + assert r.json()["totals"]["run_llm_cap|run:run_1|lifetime"] == 15_300 + + +def test_idempotent_replay_is_deduped_not_reapplied(make_client): + c = make_client() + c.post("/v1/ledger/events:batch", json={"events": [_spent_add("dup", 10_500)]}) + r = c.post("/v1/ledger/events:batch", json={"events": [_spent_add("dup", 10_500)]}) + body = r.json() + assert body["accepted"] == 0 and body["deduped"] == 1 + assert body["totals"]["run_llm_cap|run:run_1|lifetime"] == 10_500 + + +def test_admit_complete_counter(make_client): + c = make_client() + c.post( + "/v1/ledger/events:batch", + json={"events": [_conc("i1", "admit"), _conc("i2", "admit")]}, + ) + assert c.get("/v1/ledger/inflight", params={"segment_key": "run:run_1"}).json()["count"] == 2 + c.post("/v1/ledger/events:batch", json={"events": [_conc("i3", "complete")]}) + assert c.get("/v1/ledger/inflight", params={"segment_key": "run:run_1"}).json()["count"] == 1 + + +def test_step_accumulates_run_state(make_client): + c = make_client() + c.post("/v1/ledger/events:batch", json={"events": [_step(1, 10_500), _step(2, 21_000)]}) + st = c.app.state.store.get_run_state("local", "run_1") + assert st["step_count"] == 2 + assert len(st["recent"]) == 2 + assert st["velocity_micros_per_step"] == 10_500.0 + + +def test_halt_mark_shows_in_halt_and_batch_ack(make_client): + c = make_client() + r = c.post( + "/v1/ledger/events:batch", + json={ + "events": [ + { + "kind": "halt_mark", + "idempotency_key": "h1", + "run_id": "run_1", + "reason": "step_cap: 20", + "detector": "step_cap", + } + ] + }, + ) + assert r.json()["halted"] is True + assert c.get("/v1/ledger/halt/run_1").json()["halted"] is True + + +def test_unknown_kind_is_400_and_batch_rolls_back(make_client): + c = make_client() + r = c.post( + "/v1/ledger/events:batch", + json={ + "events": [ + _spent_add("ok", 10_500), + {"kind": "bogus", "idempotency_key": "x", "run_id": "run_1"}, + ] + }, + ) + assert r.status_code == 400 + assert ( + c.get( + "/v1/ledger/spent", + params={"budget_id": "run_llm_cap", "segment_key": "run:run_1", "period": "lifetime"}, + ).json()["spent_micros"] + == 0 + ) + + +def test_missing_idempotency_key_is_400(make_client): + c = make_client() + r = c.post( + "/v1/ledger/events:batch", + json={ + "events": [{"kind": "spent_add", "run_id": "run_1", "delta_micros": 1, "targets": []}] + }, + ) + assert r.status_code == 400 + + +def test_batch_over_max_is_400(make_client): + c = make_client(max_batch=2) + r = c.post( + "/v1/ledger/events:batch", + json={"events": [_spent_add(f"k{i}", 1) for i in range(3)]}, + ) + assert r.status_code == 400 + + +def test_bad_durability_header_is_400(make_client): + c = make_client() + r = c.post( + "/v1/ledger/events:batch", + json={"events": [_spent_add("k", 1)]}, + headers={"Durability": "whenever"}, + ) + assert r.status_code == 400 diff --git a/tests/test_ledger_precheck.py b/tests/test_ledger_precheck.py new file mode 100644 index 0000000..8eee177 --- /dev/null +++ b/tests/test_ledger_precheck.py @@ -0,0 +1,136 @@ +"""POST /v1/ledger/precheck — contract §4.""" + +from __future__ import annotations + + +def _seed_spend(c, run_id="run_1", delta=10_500): + c.post( + "/v1/ledger/events:batch", + json={ + "events": [ + { + "kind": "spent_add", + "idempotency_key": f"{run_id}:seed", + "run_id": run_id, + "delta_micros": delta, + "targets": [ + { + "budget_id": "run_llm_cap", + "segment_key": f"run:{run_id}", + "period": "lifetime", + }, + ], + } + ] + }, + ) + + +def test_precheck_returns_halt_spent_inflight(make_client): + c = make_client() + _seed_spend(c) + c.post( + "/v1/ledger/events:batch", + json={ + "events": [ + { + "kind": "admit", + "idempotency_key": "a1", + "run_id": "run_1", + "segment_key": "run:run_1", + } + ] + }, + ) + r = c.post( + "/v1/ledger/precheck", + json={ + "run_id": "run_1", + "segment_keys": ["run:run_1"], + "budgets": [ + {"budget_id": "run_llm_cap", "segment_key": "run:run_1", "period": "lifetime"} + ], + "want": ["spent", "inflight", "halt"], + }, + ) + assert r.status_code == 200 + body = r.json() + assert body["halted"] is False and body["halt_reason"] is None + assert body["spent"]["run_llm_cap|run:run_1|lifetime"] == 10_500 + assert body["inflight"]["run:run_1"] == 1 + assert "server_ts" in body + assert "window" not in body + + +def test_precheck_window_slice(make_client): + c = make_client() + c.post( + "/v1/ledger/events:batch", + json={ + "events": [ + { + "kind": "step", + "idempotency_key": "run_1:a:1:step", + "run_id": "run_1", + "agent": "a", + "seq": 1, + "node_type": "llm", + "boundary_id": "a.chat", + "cost_micros": 10_500, + "cum_spent_micros": 10_500, + "ts": 1.0, + } + ] + }, + ) + r = c.post( + "/v1/ledger/precheck", + json={"run_id": "run_1", "want": ["window"]}, + ) + body = r.json() + assert body["window"]["step_count"] == 1 + assert len(body["window"]["recent"]) == 1 + assert "spent" not in body + + +def test_precheck_reflects_halt(make_client): + c = make_client() + c.post( + "/v1/ledger/events:batch", + json={ + "events": [ + { + "kind": "halt_mark", + "idempotency_key": "h", + "run_id": "run_1", + "reason": "step_cap", + "detector": "step_cap", + } + ] + }, + ) + body = c.post("/v1/ledger/precheck", json={"run_id": "run_1", "want": ["halt"]}).json() + assert body["halted"] is True + assert body["halt_reason"] == "step_cap" + + +def test_precheck_unknown_run_is_zeros(make_client): + c = make_client() + body = c.post( + "/v1/ledger/precheck", + json={ + "run_id": "nope", + "segment_keys": ["run:nope"], + "budgets": [{"budget_id": "b", "segment_key": "run:nope", "period": "lifetime"}], + "want": ["spent", "inflight", "halt", "window"], + }, + ).json() + assert body["halted"] is False + assert body["spent"]["b|run:nope|lifetime"] == 0 + assert body["inflight"]["run:nope"] == 0 + assert body["window"] == {"step_count": 0, "recent": [], "velocity_micros_per_step": 0.0} + + +def test_precheck_missing_run_id_is_400(make_client): + c = make_client() + assert c.post("/v1/ledger/precheck", json={"want": ["halt"]}).status_code == 400 diff --git a/tests/test_migrations.py b/tests/test_migrations.py new file mode 100644 index 0000000..45312e5 --- /dev/null +++ b/tests/test_migrations.py @@ -0,0 +1,78 @@ +"""Schema versioning / forward migrations (SqliteStore._apply_migrations).""" + +from __future__ import annotations + +import sqlite3 +import tempfile +from pathlib import Path + +from control_plane.store import SCHEMA_VERSION, SqliteStore + + +def _tables(db: sqlite3.Connection) -> set[str]: + return {r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table'")} + + +def test_fresh_db_is_at_current_version(): + with tempfile.TemporaryDirectory() as td: + s = SqliteStore(str(Path(td) / "t.db"), auto_seed=False) + try: + assert s._db.execute("PRAGMA user_version").fetchone()[0] == SCHEMA_VERSION + assert {"ledger_events", "run_state"} <= _tables(s._db) + finally: + s.close() + + +def test_reopen_is_idempotent(): + with tempfile.TemporaryDirectory() as td: + p = str(Path(td) / "t.db") + SqliteStore(p, auto_seed=False).close() + s = SqliteStore(p, auto_seed=False) + try: + assert s._db.execute("PRAGMA user_version").fetchone()[0] == SCHEMA_VERSION + finally: + s.close() + + +def test_legacy_0_1_db_upgrades_additively(): + """A pre-versioning DB (user_version 0) missing the new tables gets them and is + bumped to current — without touching run_registrations / runs data.""" + with tempfile.TemporaryDirectory() as td: + p = str(Path(td) / "legacy.db") + raw = sqlite3.connect(p) + raw.executescript( + """ + CREATE TABLE run_registrations ( + run_id TEXT PRIMARY KEY, intent TEXT, user_dims TEXT, registered_at REAL + ); + CREATE TABLE runs (run_id TEXT PRIMARY KEY, agent TEXT, status TEXT); + CREATE TABLE 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 + ); + INSERT INTO run_registrations(run_id, intent, user_dims, registered_at) + VALUES ('r1', 'demo', '{}', 1.0); + INSERT INTO runs(run_id, agent, status) VALUES ('r1', 'demo', 'running'); + INSERT INTO policy_instances(id, template) VALUES ('pi1', 'step_cap'); + """ + ) + raw.commit() + assert raw.execute("PRAGMA user_version").fetchone()[0] == 0 + raw.close() + + s = SqliteStore(p, auto_seed=False) + try: + assert s._db.execute("PRAGMA user_version").fetchone()[0] == SCHEMA_VERSION + assert {"ledger_events", "run_state"} <= _tables(s._db) + assert "run_registrations" in _tables(s._db) # NOT dropped in v2 + assert s._db.execute("SELECT COUNT(*) FROM run_registrations").fetchone()[0] == 1 + assert s._db.execute("SELECT COUNT(*) FROM runs").fetchone()[0] == 1 + # legacy + v2 column adds applied + reg_cols = {r[1] for r in s._db.execute("PRAGMA table_info(run_registrations)")} + assert "mode" in reg_cols + pol_cols = {r[1] for r in s._db.execute("PRAGMA table_info(policy_instances)")} + assert "data_scope" in pol_cols + # pre-existing rows get the default via the ALTER + assert s.get_policy_instance("pi1").data_scope == "local" + finally: + s.close() diff --git a/tests/test_runs_and_halt.py b/tests/test_runs_and_halt.py new file mode 100644 index 0000000..20d22a7 --- /dev/null +++ b/tests/test_runs_and_halt.py @@ -0,0 +1,66 @@ +"""0.2.0 additive changes to run identity, run-record PATCH, and the run-scoped +halt route — contract §7 / §8 / §11.""" + +from __future__ import annotations + + +def test_register_returns_registered_at(make_client): + c = make_client() + r = c.post( + "/v1/runs", json={"run_id": "run_1", "intent": "demo", "user_dims": {"user_id": "u"}} + ) + assert r.status_code == 201 + body = r.json() + assert body["run_id"] == "run_1" + assert body["status"] == "registered" + assert isinstance(body["registered_at"], (int, float)) and body["registered_at"] > 0 + # and it round-trips through GET .../registration + reg = c.get("/v1/runs/run_1/registration").json() + assert reg["registered_at"] == body["registered_at"] + + +def test_patch_run_record_ignores_derived_fields(make_client): + c = make_client() + c.post("/v1/runs", json={"run_id": "run_1", "intent": "demo"}) + r = c.patch( + "/v1/run-records/run_1", + json={"status": "completed", "steps": 999, "cost_micros": 12345}, + ) + assert r.status_code == 200 + rec = c.get("/v1/run-records/run_1").json() + assert rec["status"] == "completed" + assert rec["steps"] == 0 # client value dropped + assert rec["cost_micros"] == 0 # client value dropped + + +def test_run_scoped_halt_get_post_delete(make_client): + c = make_client() + c.post("/v1/runs", json={"run_id": "run_1", "intent": "demo"}) + + assert c.get("/v1/ledger/runs/run_1/halt").json() == {"halted": False, "halt_reason": None} + + assert c.post("/v1/ledger/runs/run_1/halt", json={"reason": "step_cap: 20"}).json() == { + "halted": True + } + got = c.get("/v1/ledger/runs/run_1/halt").json() + assert got["halted"] is True and got["halt_reason"] == "step_cap: 20" + # run row status flipped too + assert c.get("/v1/run-records/run_1").json()["status"] == "halted" + + assert c.delete("/v1/ledger/runs/run_1/halt").json() == {"halted": False} + assert c.get("/v1/ledger/runs/run_1/halt").json()["halted"] is False + + +def test_legacy_halt_alias_still_works(make_client): + c = make_client() + c.post("/v1/ledger/halt/mark", json={"run_id": "run_1", "reason": "x"}) + assert c.get("/v1/ledger/halt/run_1").json()["halted"] is True + assert c.get("/v1/ledger/runs/run_1/halt").json()["halted"] is True + + +def test_health_advertises_limits(make_client): + c = make_client(max_batch=50) + body = c.get("/health").json() + assert body["max_batch"] == 50 + assert body["max_body_bytes"] > 0 + assert body["version"] == "0.2.0" diff --git a/tests/test_servicectl.py b/tests/test_servicectl.py new file mode 100644 index 0000000..da8485e --- /dev/null +++ b/tests/test_servicectl.py @@ -0,0 +1,205 @@ +"""Integration tests for `control-plane start|stop|status`. + +These spawn a real, detached `python -m control_plane.cli serve` subprocess and talk +to it over real HTTP — the one place this repo actually exercises the OS-specific +process-lifecycle code (Windows DETACHED_PROCESS vs POSIX start_new_session, +terminate()/kill() escalation, /health polling). Slower than a unit test; that's the +point. + +Every test gets its own CONTROL_PLANE_STATE_DIR (env, monkeypatched) and its own free +port, so tests never collide with each other or with a real user's running instance. +""" + +from __future__ import annotations + +import json +import socket +import time +from pathlib import Path + +import httpx +import psutil +import pytest + +from control_plane import servicectl + + +def _free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +@pytest.fixture +def service(tmp_path, monkeypatch): + """Isolated state dir; guarantees cleanup even if a test fails mid-assertion.""" + monkeypatch.setenv("CONTROL_PLANE_STATE_DIR", str(tmp_path / "state")) + monkeypatch.setenv("CONTROL_PLANE_API_KEYS", "") # anonymous/local, no auth friction + yield {"db": str(tmp_path / "cp.db"), "port": _free_port()} + try: + servicectl.stop() + except Exception: + pass + + +# --------------------------------------------------------------------------- # +# status / stop with nothing running # +# --------------------------------------------------------------------------- # + + +def test_status_when_never_started(service): + info = servicectl.status() + assert info == {"running": False} + + +def test_stop_when_never_started_is_a_noop(service): + assert servicectl.stop() is False + + +def test_status_and_stop_after_stale_state_file(service, tmp_path): + """A state file pointing at a PID that no longer exists must not be trusted.""" + state_file = tmp_path / "state" / "control-plane.json" + state_file.parent.mkdir(parents=True, exist_ok=True) + dead_pid = _unused_pid() + state_file.write_text( + json.dumps( + { + "pid": dead_pid, + "host": "127.0.0.1", + "port": service["port"], + "db": service["db"], + "log_file": str(tmp_path / "x.log"), + "started_at": time.time(), + } + ) + ) + assert servicectl.status() == {"running": False} + assert not state_file.exists() # cleaned up as stale + + +def _unused_pid() -> int: + """A PID guaranteed not to be a live control-plane process right now.""" + candidate = 999_999 + while psutil.pid_exists(candidate): + candidate -= 1 + return candidate + + +# --------------------------------------------------------------------------- # +# the real lifecycle: start -> status -> stop -> status # +# --------------------------------------------------------------------------- # + + +def test_full_lifecycle(service): + state = servicectl.start(host="127.0.0.1", port=service["port"], db=service["db"]) + assert state.port == service["port"] + assert psutil.pid_exists(state.pid) + + # the actual API is reachable over real HTTP, not just "a process exists" + r = httpx.get(f"{state.base_url}/health", timeout=2.0) + assert r.status_code == 200 + assert r.json()["status"] == "ok" + + info = servicectl.status() + assert info["running"] is True + assert info["healthy"] is True + assert info["pid"] == state.pid + assert info["version"] == r.json()["version"] + + assert servicectl.stop() is True + # the process is actually gone, not just untracked + deadline = time.monotonic() + 5 + while psutil.pid_exists(state.pid) and time.monotonic() < deadline: + time.sleep(0.1) + assert not psutil.pid_exists(state.pid) + + assert servicectl.status() == {"running": False} + + +def test_start_is_idempotent(service): + """A second start() while one is already running must not spawn a new process — + it should short-circuit and return the same pid untouched.""" + first = servicectl.start(host="127.0.0.1", port=service["port"], db=service["db"]) + second = servicectl.start(host="127.0.0.1", port=service["port"], db=service["db"]) + assert second.pid == first.pid + assert second.started_at == first.started_at # same ServiceState, not a fresh one + + +def test_stop_is_idempotent(service): + servicectl.start(host="127.0.0.1", port=service["port"], db=service["db"]) + assert servicectl.stop() is True + assert servicectl.stop() is False # already stopped -> no-op, not an error + + +def test_served_endpoints_reachable_through_real_lifecycle(service): + """Once started, the plane answers the actual TokenOps-facing API — not just + /health. One assertion per route family is enough here; the route behaviour + itself is covered by the app-level tests.""" + state = servicectl.start(host="127.0.0.1", port=service["port"], db=service["db"]) + base = state.base_url + + reg = httpx.post(f"{base}/v1/runs", json={"intent": "smoke", "user_dims": {}}, timeout=2.0) + assert reg.status_code == 201 + run_id = reg.json()["run_id"] + assert reg.json()["registered_at"] > 0 + + resolved = httpx.get(f"{base}/v1/runs/{run_id}/registration", timeout=2.0) + assert resolved.status_code == 200 and resolved.json()["intent"] == "smoke" + + gov = httpx.get(f"{base}/v1/governance/anyagent", timeout=2.0) + assert gov.status_code == 200 + + batch = httpx.post( + f"{base}/v1/ledger/events:batch", + json={ + "events": [ + { + "kind": "spent_add", + "idempotency_key": "smoke-1", + "run_id": run_id, + "delta_micros": 1000, + "targets": [ + { + "budget_id": "__run_total__", + "segment_key": f"run:{run_id}", + "period": "lifetime", + } + ], + } + ] + }, + timeout=2.0, + ) + assert batch.status_code == 201 and batch.json()["accepted"] == 1 + + pre = httpx.post( + f"{base}/v1/ledger/precheck", + json={ + "run_id": run_id, + "want": ["spent", "halt"], + "budgets": [ + {"budget_id": "__run_total__", "segment_key": f"run:{run_id}", "period": "lifetime"} + ], + }, + timeout=2.0, + ) + assert pre.status_code == 200 + assert pre.json()["spent"][f"__run_total__|run:{run_id}|lifetime"] == 1000 + + halt = httpx.get(f"{base}/v1/ledger/runs/{run_id}/halt", timeout=2.0) + assert halt.status_code == 200 and halt.json()["halted"] is False + + +def test_log_file_captures_server_output(service): + state = servicectl.start(host="127.0.0.1", port=service["port"], db=service["db"]) + log_path = Path(state.log_file) + assert log_path.exists() + # uvicorn logs its startup banner; give the flush a moment + deadline = time.monotonic() + 3 + content = "" + while time.monotonic() < deadline: + content = log_path.read_text(errors="replace") + if content: + break + time.sleep(0.1) + assert content # something was written, not silently discarded