diff --git a/anton/chat.py b/anton/chat.py index c5b7c431..40ca528c 100644 --- a/anton/chat.py +++ b/anton/chat.py @@ -875,6 +875,8 @@ async def _handle_unpublish( import json as _json from pathlib import Path as _Path + from anton.publisher import _STATE_SNAPSHOT + removed_report_id = selected.get("report_id") or "" removed_md5 = selected.get("md5") or "" artifacts_root = _Path(settings.artifacts_dir) @@ -898,6 +900,20 @@ async def _handle_unpublish( if matches: del data[key] changed = True + # Reset the state-snapshot baseline so a later /publish with a + # changed collection set is treated as fresh (not blocked). + # Modern layout: snapshot sits next to .published.json. Legacy + # root .published.json uses "{slug}/" keys → snapshot lives in + # the per-artifact subdir named by the key. + snap_dirs = [pub_file.parent] + if key.endswith("/"): + snap_dirs.append(pub_file.parent / key.rstrip("/")) + for snap_dir in snap_dirs: + snap = snap_dir / _STATE_SNAPSHOT + try: + snap.unlink() + except OSError: + pass if changed: try: pub_file.write_text(_json.dumps(data, indent=2), encoding="utf-8") diff --git a/anton/core/artifacts/backend_launcher.py b/anton/core/artifacts/backend_launcher.py index 03735a4a..196893ba 100644 --- a/anton/core/artifacts/backend_launcher.py +++ b/anton/core/artifacts/backend_launcher.py @@ -17,15 +17,55 @@ import asyncio import os +import re +import shutil import signal import socket import sys +import tempfile import urllib.error import urllib.request from pathlib import Path from typing import Any, Protocol +def _anton_state_pythonpath_dir() -> str: + """A private directory exposing ONLY the `anton_state` package, for PYTHONPATH. + + Pointing PYTHONPATH at the package's real parent would leak the whole anton + repo (or its venv's site-packages) onto the backend's sys.path and shadow + the scratchpad venv's own deps (fastapi/pydantic) — a local != published + hazard. Instead we expose anton_state alone via a symlink (copy fallback). + In the cloud the package is vendored into the bundle instead (see publisher). + """ + import anton_state + + pkg = Path(anton_state.__file__).resolve().parent + root = Path(tempfile.gettempdir()) / "anton_state_pp" + link = root / "anton_state" + root.mkdir(parents=True, exist_ok=True) + # Always reset the entry, then (re)link — keeps it fresh and simple. + if link.is_symlink() or link.exists(): + if link.is_dir() and not link.is_symlink(): + shutil.rmtree(link) + else: + link.unlink() + try: + link.symlink_to(pkg, target_is_directory=True) + except OSError: + shutil.copytree(pkg, link) + return str(root) + + +def _build_backend_env(extra_env: dict[str, str] | None) -> dict[str, str]: + """Subprocess env: inherited environ + extra_env, with anton_state on PYTHONPATH.""" + env = {**os.environ, **(extra_env or {})} + isolated = _anton_state_pythonpath_dir() + existing = env.get("PYTHONPATH", "") + env["PYTHONPATH"] = isolated + (os.pathsep + existing if existing else "") + return env + + class ScratchpadPoolLike(Protocol): """Minimal surface the launcher needs from a scratchpad pool. @@ -101,6 +141,14 @@ async def launch_artifact_backend( line = raw_line.split("#", 1)[0].strip() if not line or line.startswith("-"): continue + # `anton_state` is the internal STATE SDK — it is provided to the + # backend at runtime via PYTHONPATH (see _anton_state_pythonpath_dir), + # not published to any package registry. Drop it if the model listed + # it in requirements.txt, otherwise the install step fails with + # "anton-state was not found in the package registry". + pkg_name = re.split(r"[<>=!~ \[]", line, maxsplit=1)[0].strip() + if pkg_name.replace("-", "_").lower() == "anton_state": + continue packages.append(line) if packages: from datetime import datetime, timezone @@ -173,7 +221,7 @@ def _set_pdeathsig() -> None: stderr=log_fd, stdin=asyncio.subprocess.DEVNULL, preexec_fn=preexec_fn, - env={**os.environ, **(extra_env or {})}, + env=_build_backend_env(extra_env), ) except OSError as exc: log_fd.close() diff --git a/anton/core/llm/prompts.py b/anton/core/llm/prompts.py index 989d3d80..ef79fb3a 100644 --- a/anton/core/llm/prompts.py +++ b/anton/core/llm/prompts.py @@ -249,7 +249,9 @@ external data source; prefer stateless when in doubt (see BACKEND & FULLSTACK \ section) → `type="fullstack-stateful-app"`, `primary="static/index.html"`. \ The frontend lives in a `static/` subfolder of the artifact, served by \ -`backend.py`. +`backend.py`. Light durable state uses the platform `STATE` store (declare \ +`state_manifest.json`); heavy/relational data uses an external connected \ +database. WHEN NOT TO REGISTER: - Pure chat answers, tables, or markdown rendered inline in the conversation \ diff --git a/anton/core/memory/builtin_skills/build-fullstack-backend/SKILL.md b/anton/core/memory/builtin_skills/build-fullstack-backend/SKILL.md index d7ba130f..6681f2c8 100644 --- a/anton/core/memory/builtin_skills/build-fullstack-backend/SKILL.md +++ b/anton/core/memory/builtin_skills/build-fullstack-backend/SKILL.md @@ -86,11 +86,35 @@ HARD CONTRACT (violating ANY of these breaks launch or deployment — full expla # "DS_POSTGRES_PROD_DB__PASSWORD": os.environ.get("DS_POSTGRES_PROD_DB__PASSWORD"), } + # === State (durable storage) — INCLUDE THIS BLOCK ONLY FOR + # `fullstack-stateful-app`; OMIT it entirely for `fullstack-stateless-app`. + # STATE mirrors SECRETS: the cloud runner overlays {url, token} (a short-lived + # capability for the trusted state broker) before each request; locally it stays + # None and the SQLite driver is used. Declare the state KEY schema in + # `state_manifest.json` next to this file — generate it from the anton_state + # model, do NOT hand-write the JSON (see STATE MANIFEST below). Build the store + # AT POINT OF USE (inside a route), reading the current STATE — never at import time. + STATE = None + + from anton_state import open_store + _STATE_DIR = Path(__file__).resolve().parent + + def get_store(): + return open_store( + state=STATE, + manifest_path=str(_STATE_DIR / "state_manifest.json"), + local_path=str(_STATE_DIR / ".anton_state.db"), + ) + # === API routes === @app.get("/api/hello") async def hello(): # Example secret use (read at point of use, not at import): # pw = SECRETS["DS_POSTGRES_PROD_DB__PASSWORD"] + # Example STATE use (stateful only; build the store at point of use): + # store = get_store() + # await store.put({"pk": user_id, "sk": "profile", "name": name}) + # item = await store.get(user_id, "profile") return {"hello": "world"} # Static mount MUST come AFTER all API routes (mount at "/" catches every @@ -123,7 +147,7 @@ HARD CONTRACT (violating ANY of these breaks launch or deployment — full expla - Prefer `async def` for I/O-bound routes (DB queries, external HTTP calls via `httpx.AsyncClient`). Sync `def` is fine for trivial CPU work, but sync blocking I/O inside an async app stalls the event loop. - LOCAL STATE (the ONE rule that differs between the two fullstack types): * `fullstack-stateless-app`: no local state of any kind survives a request. No module-level mutable caches that matter across requests (`USERS = {}`, `SESSIONS = []`) — in Lambda these globals may or may not survive between invocations, never rely on them. Treat the filesystem as read-only and non-persistent: anything written is lost between requests and may fail outright depending on the host (Linux, Windows, or a read-only cloud sandbox). NEVER write to `` at runtime, and never rely on a file surviving to a later request. If a request genuinely needs scratch space, use the OS temp dir via `tempfile` and treat it as ephemeral (gone the moment the request ends). ALL persistence goes through external data sources. - * `fullstack-stateful-app`: local on-disk state (e.g. a SQLite file) IS allowed — keep it in the artifact root (`/`, next to `backend.py`). Every other rule in this list still applies. + * `fullstack-stateful-app`: durable state goes through the platform `STATE` store (module-level `STATE`, built via `get_store()`), which is a document/key-value model — suitable for LIGHT state (counters, settings, sessions, simple documents keyed by id). Declare the state KEY schema in `state_manifest.json` next to `backend.py` (see STATE MANIFEST below for the exact format — do NOT hand-invent it). For HEAVY/relational needs (joins, transactions, analytics, large data) use an EXTERNAL database via a connected data source instead — do not force it into `STATE`. The `anton_state` SDK is injected at runtime (do NOT add it to `requirements.txt` — see REQUIREMENTS below) and needs pydantic v2, which the mandatory `fastapi` dependency provides — always keep `fastapi` in `requirements.txt`. Every other rule in this list still applies. - LOGGING: `print()` and `logging.getLogger(__name__).info(...)` both go to CloudWatch in Lambda and to `backend.log` locally — no extra setup needed. - REQUIREMENTS: always save a `/requirements.txt` with at minimum: ``` @@ -132,8 +156,35 @@ HARD CONTRACT (violating ANY of these breaks launch or deployment — full expla uvicorn ``` Add any other libraries the backend imports (one per line: `pkg` or `pkg==1.2`). `launch_backend` reads this file and installs everything into the slug-named scratchpad's venv before spawning the process. Only simple lines are supported — `-r`, `-e`, `--index-url`, blank lines and `#` comments are ignored. + NEVER list `anton_state` in `requirements.txt` — it is NOT a published package and the install will FAIL to resolve it (`anton-state was not found in the package registry`), aborting the launch. The STATE SDK is provided to the backend automatically at runtime, so `from anton_state import open_store` just works without any dependency line. This is the ONLY import you leave out of `requirements.txt`. - Do NOT start the server inside the scratchpad — use `launch_backend` in step 6. - DECLARE DATASOURCES: if `backend.py` reads any `DS____` env var, call `update_artifact(slug=, datasources=[...])` immediately after writing the file. Pass a flat list of connection slugs (e.g. `["postgres-prod_db", "hubspot-main"]`); each slug MUST match a connection from the `Connected Data Sources` section of this prompt. This records the deployable's credential dependencies in `metadata.json` so the artifact can be redeployed with the right env vars later. Skip this call only when the backend uses no `DS_*` vars at all. + - STATE MANIFEST (`fullstack-stateful-app` ONLY): `state_manifest.json` is a SINGLE universal contract read by the local SQLite driver AND (client-side) by the cloud HTTP driver — the trusted broker is schema-agnostic. GENERATE it from the `anton_state` model instead of hand-writing JSON (this makes a malformed manifest impossible): + ```python + from anton_state.schema import StateSchema, Attr + StateSchema( + pk=Attr(name="pk"), # partition key (always type "S") + sk=Attr(name="sk"), # sort key — omit entirely if unused + collections=["comments", "users"], # every Collection(store, "") you use + ).to_manifest(f"{artifact_path}/state_manifest.json") + ``` + The manifest describes ONLY the KEY schema, never data fields. The resulting JSON is a FLAT object `{version, pk, sk?, gsis?, ttl_attribute?, collections?}` where `pk`/`sk` are `{"name": ..., "type": "S"}` — string keys only in v1. Do NOT wrap it in `entities`/`attributes`/`partition_key`/`sort_key` (a DynamoDB-CreateTable-style shape) and do NOT declare non-key attributes: those fail validation (`StateSchema ... pk Field required`) at the first request. Store the actual values freely via `store.put({...})` at runtime — they need no schema entry. List every `Collection(store, "")` name in `collections` (this is NOT declaring data fields — it is the collection registry). Removing a name here when UPDATING an already-published artifact BLOCKS the publish (its stored data would be orphaned) — to change the set you must /unpublish first and publish again. + - STATE STORE API (`fullstack-stateful-app` ONLY): the `store` from `get_store()` is a key-value store keyed by `(pk, sk)`. PREFER the `Collection` helper for light state — it manages the sort key and defaults the partition: + ```python + from anton_state import Collection + todos = Collection(get_store(), "todos") + await todos.put("id1", {"text": "buy milk"}) # pk defaults to one partition + items = await todos.list() # all items in the collection + n = await Collection(get_store(), "counters").increment("visits", field="n") + ``` + Low-level `store` methods (all async; NO `scan()` / "list everything"): + * `await store.get(pk, sk=None)` → one item or `None` + * `await store.put(item)` → write (dict MUST include `pk` and, if the schema has a sort key, `sk`); `_v` is set by the store — never set it yourself + * `await store.delete(pk, sk=None)` + * `await store.query(pk, *, sk_prefix=None, filters=None, limit=None)` → items sharing partition key `pk` (NO secondary indexes in v1 — there is no `index=` argument) + * `await store.increment(pk, sk=None, *, field, by=1)` → atomic counter (use this for counters; do NOT hand-roll read-modify-write) + * `await store.update(pk, sk=None, *, set_fields=None, add_fields=None, if_version=None)` → atomic partial update + DESIGN KEYS AROUND ACCESS PATTERNS: every "list" must map to a single `query(pk=...)` (or `Collection.list()`). Do NOT call the store in a loop — collect with one `query`. Do NOT wrap a STATE mutation (`put`/`delete`/`increment`/`update`) in your own retry loop: on a timeout the outcome is unknown and a retry can double-apply — surface the error instead. 5. BUILD FRONTEND (if needed): In a separate scratchpad: - Build a single-file HTML dashboard or web interface diff --git a/anton/publisher.py b/anton/publisher.py index 6150994b..417a69fd 100644 --- a/anton/publisher.py +++ b/anton/publisher.py @@ -39,7 +39,28 @@ FULLSTACK_ARTIFACT_TYPES = frozenset({"fullstack-stateful-app", "fullstack-stateless-app"}) # Filenames inside an artifact folder that are housekeeping — never bundled. -_FULLSTACK_EXCLUDED = {"metadata.json", "README.md", "backend.log", ".published.json"} +# The .anton_state.db* entries are defensive: _zip_fullstack is allowlist-based +# so root files are not bundled anyway, but this guards against future changes +# (and covers the WAL side-files, where -wal holds the freshest data). +# Local snapshot of the last-published state key schema (for the client-side +# schema-change warning). Never bundled. +_STATE_SNAPSHOT = ".state_manifest.published.json" +_FULLSTACK_EXCLUDED = { + "metadata.json", + "README.md", + "backend.log", + ".published.json", + ".anton_state.db", + ".anton_state.db-wal", + ".anton_state.db-shm", + _STATE_SNAPSHOT, +} + + +class StatePublishBlocked(Exception): + """Publish aborted: a previously published state collection disappeared from + the new schema, which would orphan its stored data. Tooling-level error + (not an anton_state runtime error) — surfaced to the publish caller.""" DEFAULT_PUBLISH_URL = "https://view.mindshub.ai" @@ -233,9 +254,99 @@ def _zip_fullstack(artifact_dir: Path) -> tuple[bytes, list[str]]: continue _write_scrubbed(zf, f, arc_name) included.append(arc_name) + + manifest = artifact_dir / "state_manifest.json" + if manifest.is_file(): + _write_scrubbed(zf, manifest, "state_manifest.json") + included.append("state_manifest.json") + + included.extend(_vendor_anton_state(zf)) return buf.getvalue(), included +def _read_state_manifest(artifact_dir: Path) -> dict | None: + """Parse state_manifest.json from the artifact dir, or None if absent. + + Included in the publish payload so the provisioning pipeline (subplan #4) + can read the schema without unzipping the bundle. + """ + path = artifact_dir / "state_manifest.json" + if not path.is_file(): + return None + return json.loads(path.read_text(encoding="utf-8")) + + +def _state_key_shape(manifest: dict) -> tuple: + """Normalised key schema (pk/sk name+type) for compatibility comparison.""" + def attr(a): + return (a["name"], a.get("type", "S")) if a else None + return (attr(manifest.get("pk")), attr(manifest.get("sk"))) + + +def _warn_state_schema_change(artifact_dir: Path, manifest: dict) -> str | None: + """Warn (non-blocking) if the key schema changed vs the last publish — on a + shared table that orphans old data (no per-artifact key schema for AWS to + reject). Read-only: does NOT write the snapshot (see _save_state_snapshot).""" + snap = artifact_dir / _STATE_SNAPSHOT + if not snap.is_file(): + return None + try: + prev = json.loads(snap.read_text(encoding="utf-8")) + except (ValueError, OSError): + return None + if _state_key_shape(prev) == _state_key_shape(manifest): + return None + warning = ( + "WARNING: state key schema changed since the last publish — existing " + "stored data will become UNREACHABLE (its keys are encoded under the old " + "schema). Migrations are out of scope in v1." + ) + print(warning) + return warning + + +def _save_state_snapshot(artifact_dir: Path, manifest: dict) -> None: + """Record the just-published key schema for the next publish's comparison. + Called ONLY after a successful /upload, so a failed publish doesn't hide the + next real change.""" + (artifact_dir / _STATE_SNAPSHOT).write_text(json.dumps(manifest, indent=2), encoding="utf-8") + + +def _blocked_collection_drops(artifact_dir: Path, manifest: dict) -> set[str]: + """Collections present in the last published manifest but missing from the new + one — their stored data (keyed by artifact_id in the shared table) would be + orphaned. Empty when there is no snapshot (first publish) or nothing was + removed (adding a collection is safe).""" + snap = artifact_dir / _STATE_SNAPSHOT + if not snap.is_file(): + return set() + try: + prev = json.loads(snap.read_text(encoding="utf-8")) + except (ValueError, OSError): + return set() + return set(prev.get("collections", [])) - set(manifest.get("collections", [])) + + +def _vendor_anton_state(zf: zipfile.ZipFile) -> list[str]: + """Copy the `anton_state` package into the bundle under `anton_state/`. + + The published artifact imports `anton_state` from its own directory (the + runner puts the artifact dir on sys.path), so the SDK travels with the + bundle — version-locked, offline, no index needed. + """ + import anton_state + + pkg_dir = Path(anton_state.__file__).resolve().parent + added: list[str] = [] + for f in sorted(pkg_dir.rglob("*.py")): + if "__pycache__" in f.parts: + continue + arc_name = f"anton_state/{f.relative_to(pkg_dir).as_posix()}" + _write_scrubbed(zf, f, arc_name) + added.append(arc_name) + return added + + def _collect_datasource_secrets( artifact: Artifact, vault: DataVault | None = None, @@ -314,6 +425,7 @@ def publish( raise FileNotFoundError(f"Path not found: {file_path}") payload_dict: dict = {} + state_manifest = None artifact = _load_artifact_metadata(file_path) if file_path.is_dir() else None if artifact is not None and artifact.type in FULLSTACK_ARTIFACT_TYPES: @@ -323,6 +435,18 @@ def publish( payload_dict["artifact_id"] = artifact.id payload_dict["secrets"] = secrets payload_dict["python_version"] = f"{sys.version_info.major}.{sys.version_info.minor}" + state_manifest = _read_state_manifest(file_path) + if state_manifest is not None: + _warn_state_schema_change(file_path, state_manifest) # before upload + dropped = _blocked_collection_drops(file_path, state_manifest) + if dropped: + raise StatePublishBlocked( + f"Collections {sorted(dropped)} exist in the published version " + f"but are missing from the new schema — their stored data would " + f"be orphaned. Unpublish the artifact and publish again with the " + f"new schema (the old data is not migrated)." + ) + payload_dict["state_manifest"] = state_manifest if missing: payload_dict["missing_datasources"] = missing else: @@ -346,6 +470,10 @@ def publish( url = f"{publish_url.rstrip('/')}/upload" raw = minds_request(url, api_key, method="POST", payload=payload, verify=ssl_verify) + # Upload succeeded (minds_request raises on failure): record the published + # key schema so the next publish can warn on an incompatible change. + if state_manifest is not None: + _save_state_snapshot(file_path, state_manifest) return json.loads(raw) diff --git a/anton_state/__init__.py b/anton_state/__init__.py new file mode 100644 index 00000000..418ea671 --- /dev/null +++ b/anton_state/__init__.py @@ -0,0 +1,15 @@ +"""anton_state — unified STATE API for fullstack-stateful artifacts.""" + +__version__ = "0.1.0" + +from anton_state.base import Driver, Item, Store +from anton_state.errors import ( + ConditionalCheckFailed, + StateError, + StateThrottled, + StateUnavailable, + StateValidationError, +) +from anton_state.factory import from_backend_state, open_store +from anton_state.odm import Collection +from anton_state.schema import Attr, Index, StateSchema diff --git a/anton_state/base.py b/anton_state/base.py new file mode 100644 index 00000000..dad7b3d1 --- /dev/null +++ b/anton_state/base.py @@ -0,0 +1,109 @@ +"""Driver protocol and the async Store facade.""" +from __future__ import annotations + +import asyncio +from typing import Any, Protocol + +from .errors import StateValidationError +from .schema import StateSchema + +Item = dict[str, Any] + +# Reserved version attribute for optimistic locking, set by the driver/broker +# (the request body is not trusted). Single place for all drivers. +_VERSION_ATTR = "_v" + + +class Driver(Protocol): + def get(self, pk: str, sk: str | None, *, consistent: bool) -> Item | None: ... + + def put(self, item: Item, *, if_not_exists: bool, if_version: int | None) -> None: ... + + def delete(self, pk: str, sk: str | None, *, if_version: int | None) -> None: ... + + def query( + self, + pk: str, + *, + sk_prefix: str | None, + filters: dict[str, Any] | None, + consistent: bool, + limit: int | None, + ) -> list[Item]: ... + + def increment(self, pk: str, sk: str | None, *, field: str, by: int | float) -> int | float: ... + + def update( + self, + pk: str, + sk: str | None, + *, + set_fields: dict[str, Any] | None, + add_fields: dict[str, int | float] | None, + if_version: int | None, + ) -> Item: ... + + +class Store: + """Async facade: backend routes are `async def`, so sync drivers run in a thread.""" + + def __init__(self, driver: Driver, schema: StateSchema): + self._driver = driver + self.schema = schema + + async def get(self, pk: str, sk: str | None = None, *, consistent: bool = True) -> Item | None: + return await asyncio.to_thread(self._driver.get, pk, sk, consistent=consistent) + + async def put( + self, item: Item, *, if_not_exists: bool = False, if_version: int | None = None + ) -> None: + """Write an item. `_v` is managed by the driver/broker; a `_v` present in + `item` is overwritten. `if_not_exists` and `if_version` are mutually + exclusive.""" + if if_not_exists and if_version is not None: + raise StateValidationError("if_not_exists and if_version are mutually exclusive") + await asyncio.to_thread( + self._driver.put, item, if_not_exists=if_not_exists, if_version=if_version + ) + + async def delete(self, pk: str, sk: str | None = None, *, if_version: int | None = None) -> None: + await asyncio.to_thread(self._driver.delete, pk, sk, if_version=if_version) + + async def query( + self, + pk: str, + *, + sk_prefix: str | None = None, + filters: dict[str, Any] | None = None, + consistent: bool = True, + limit: int | None = None, + ) -> list[Item]: + return await asyncio.to_thread( + self._driver.query, + pk, + sk_prefix=sk_prefix, + filters=filters, + consistent=consistent, + limit=limit, + ) + + async def increment( + self, pk: str, sk: str | None = None, *, field: str, by: int | float = 1 + ) -> int | float: + """Atomically add `by` to numeric `field` of one item; returns the new value.""" + return await asyncio.to_thread(self._driver.increment, pk, sk, field=field, by=by) + + async def update( + self, + pk: str, + sk: str | None = None, + *, + set_fields: dict[str, Any] | None = None, + add_fields: dict[str, int | float] | None = None, + if_version: int | None = None, + ) -> Item: + """Atomically SET/ADD fields of one item; returns the new item.""" + return await asyncio.to_thread( + self._driver.update, pk, sk, + set_fields=set_fields, add_fields=add_fields, if_version=if_version, + ) diff --git a/anton_state/errors.py b/anton_state/errors.py new file mode 100644 index 00000000..eec4cac4 --- /dev/null +++ b/anton_state/errors.py @@ -0,0 +1,22 @@ +"""STATE API errors — identical across both drivers.""" + + +class StateError(Exception): + """Base class for all STATE errors.""" + + +class StateValidationError(StateError): + """Item/key failed validation (empty key, type, size, TTL format).""" + + +class ConditionalCheckFailed(StateError): + """Conditional write not applied (if_not_exists / if_version).""" + + +class StateThrottled(StateError): + """Operation rate limit exceeded (DynamoDB throttle / OnDemandThroughput cap).""" + + +class StateUnavailable(StateError): + """Broker unreachable / 5xx / timeout. For a mutation the outcome is unknown + (do not blindly retry — see the shared-table design spec §2.6).""" diff --git a/anton_state/factory.py b/anton_state/factory.py new file mode 100644 index 00000000..4577973c --- /dev/null +++ b/anton_state/factory.py @@ -0,0 +1,48 @@ +"""Driver selection and schema loading. + +Schema comes from the manifest file (single source, the same one the publish +pipeline reads); the driver is chosen by backend.STATE ({url, token} dict → the +HTTP broker driver, otherwise the local SQLite driver). +""" +from __future__ import annotations + +import os + +from .base import Store +from .schema import StateSchema +from .sqlite_driver import SQLiteDriver + +_DEFAULT_LOCAL = "./.anton_state.db" +_ENV_PATH = "ANTON_ARTIFACT_STATE_PATH" +_DEFAULT_MANIFEST = "./state_manifest.json" +_ENV_MANIFEST = "ANTON_STATE_MANIFEST" + + +def _resolve_schema(schema: StateSchema | None, manifest_path: str | None) -> StateSchema: + if schema is not None: + return schema + path = manifest_path or os.environ.get(_ENV_MANIFEST) or _DEFAULT_MANIFEST + return StateSchema.from_manifest(path) + + +def open_store( + schema: StateSchema | None = None, + *, + state: dict | None = None, + local_path: str | None = None, + manifest_path: str | None = None, +) -> Store: + schema = _resolve_schema(schema, manifest_path) + if state: + from .http_driver import HTTPDriver # cloud: RPC to the trusted state broker + + return Store(HTTPDriver(url=state["url"], token=state["token"], schema=schema), schema) + path = local_path or os.environ.get(_ENV_PATH) or _DEFAULT_LOCAL + return Store(SQLiteDriver(path, schema), schema) + + +def from_backend_state( + state: dict | None, schema: StateSchema | None = None, *, manifest_path: str | None = None +) -> Store: + """Entry point for the backend: pass backend.STATE here (schema from the manifest).""" + return open_store(schema, state=state, manifest_path=manifest_path) diff --git a/anton_state/http_driver.py b/anton_state/http_driver.py new file mode 100644 index 00000000..a646407f --- /dev/null +++ b/anton_state/http_driver.py @@ -0,0 +1,129 @@ +"""Cloud driver: thin HTTP RPC to the trusted state broker (plan #2). + +stdlib-only (urllib), synchronous — the async Store facade runs it in a thread. +The namespace is NEVER sent; the broker derives it from the signed token, so +untrusted code cannot address another artifact's data. Reads are retried on +transient failure; mutations are not (their outcome would be unknown). +""" +from __future__ import annotations + +import json +import urllib.error +import urllib.request +from typing import Any + +from .base import Driver # noqa: F401 (documents intent; Protocol is structural) +from .errors import ( + ConditionalCheckFailed, + StateThrottled, + StateUnavailable, + StateValidationError, +) +from .schema import StateSchema +from .validation import validate_item, validate_key + +_TIMEOUT_S = 5 +_READ_RETRIES = 2 +_READ_OPS = {"get", "query"} + + +class _WireError(Exception): + def __init__(self, status: int, code: str, message: str): + super().__init__(message) + self.status, self.code, self.message = status, code, message + + +def _map_wire(e: _WireError) -> Exception: + if e.status == 409 or e.code == "conditional": + return ConditionalCheckFailed(e.message) + if e.status == 429 or e.code == "throttled": + return StateThrottled(e.message) + if e.status == 400 or e.code == "validation": + return StateValidationError(e.message) + if e.status == 401 or e.code == "unauthorized": + # Token rejected: a config/clock problem, DEFINITELY not applied — do not + # dress it up as "outcome unknown" like a 5xx. + return StateValidationError(f"state broker rejected the token: {e.message}") + return StateUnavailable(e.message) + + +class HTTPDriver: + def __init__(self, url: str, token: str, schema: StateSchema): + self._url = url + self._token = token + self.schema = schema + self._pk = schema.pk.name + self._sk = schema.sk.name if schema.sk else None + self._ttl = schema.ttl_attribute + + # --- transport (the only network point; monkeypatched in tests) --- + def _post(self, op: str, payload: dict) -> dict: + body = json.dumps({"op": op, **payload}).encode("utf-8") + req = urllib.request.Request( + self._url, data=body, method="POST", + headers={"Authorization": f"Bearer {self._token}", "Content-Type": "application/json"}, + ) + try: + with urllib.request.urlopen(req, timeout=_TIMEOUT_S) as resp: + return json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as e: + try: + data = json.loads(e.read().decode("utf-8")) + except Exception: + data = {} + raise _WireError(e.code, data.get("error", ""), data.get("message", str(e))) + except (urllib.error.URLError, TimeoutError, OSError) as e: + raise _WireError(503, "unavailable", str(e)) + + def _call(self, op: str, payload: dict) -> dict: + attempts = _READ_RETRIES + 1 if op in _READ_OPS else 1 + last: _WireError | None = None + for _ in range(attempts): + try: + return self._post(op, payload) + except _WireError as e: + last = e + # Only retry reads on transient (5xx/unavailable); never mutations. + if op in _READ_OPS and (e.status >= 500 or e.code == "unavailable"): + continue + raise _map_wire(e) + raise _map_wire(last) # type: ignore[arg-type] + + # --- Driver protocol --- + def get(self, pk: str, sk: str | None, *, consistent: bool) -> dict | None: + validate_key(pk, sk, self.schema) + return self._call("get", {"pk": pk, "sk": sk, "consistent": consistent}).get("item") + + def put(self, item: dict, *, if_not_exists: bool, if_version: int | None) -> None: + validate_item(item, self.schema) + pk = item[self._pk] + sk = item.get(self._sk) if self._sk else None + cond: dict | None = None + if if_not_exists: + cond = {"if_not_exists": True} + elif if_version is not None: + cond = {"if_version": if_version} + ttl = item.get(self._ttl) if self._ttl else None + self._call("put", {"pk": pk, "sk": sk, "item": item, "ttl": ttl, "cond": cond}) + + def delete(self, pk: str, sk: str | None, *, if_version: int | None) -> None: + validate_key(pk, sk, self.schema) + cond = {"if_version": if_version} if if_version is not None else None + self._call("delete", {"pk": pk, "sk": sk, "cond": cond}) + + def query(self, pk, *, sk_prefix, filters, consistent, limit) -> list[dict]: + return self._call("query", { + "pk": pk, "sk_prefix": sk_prefix, "filters": filters, + "consistent": consistent, "limit": limit, + }).get("items", []) + + def increment(self, pk, sk, *, field, by) -> int | float: + validate_key(pk, sk, self.schema) + return self._call("increment", {"pk": pk, "sk": sk, "field": field, "by": by})["value"] + + def update(self, pk, sk, *, set_fields, add_fields, if_version) -> dict: + validate_key(pk, sk, self.schema) + cond = {"if_version": if_version} if if_version is not None else None + return self._call("update", { + "pk": pk, "sk": sk, "set": set_fields or {}, "add": add_fields or {}, "cond": cond, + })["item"] diff --git a/anton_state/odm.py b/anton_state/odm.py new file mode 100644 index 00000000..ba749c18 --- /dev/null +++ b/anton_state/odm.py @@ -0,0 +1,64 @@ +"""Thin single-table helper: a logical collection over (pk, sk). + +`pk` is the partition scope and defaults to a single partition — simple apps +never think about it; multi-tenant-within-an-artifact apps pass an explicit pk. +The sort key is managed here: sk = "{collection}#{key}". +""" +from __future__ import annotations + +from typing import Any + +from .base import Item, Store +from .errors import StateValidationError + +_DEFAULT_PK = "_" + + +class Collection: + def __init__(self, store: Store, name: str): + self._store = store + self._name = name + self._pk = store.schema.pk.name + self._sk = store.schema.sk.name if store.schema.sk else None + if self._sk is None: + raise ValueError("Collection requires a schema with a sort key") + self._reserved = {self._pk, self._sk, "_key"} + + def _sk_val(self, key: str) -> str: + return f"{self._name}#{key}" + + async def get(self, key: str, *, pk: str = _DEFAULT_PK) -> Item | None: + return await self._store.get(pk, self._sk_val(key)) + + async def put( + self, key: str, value: dict, *, pk: str = _DEFAULT_PK, + if_not_exists: bool = False, if_version: int | None = None, + ) -> None: + clash = self._reserved & set(value) + if clash: + raise StateValidationError(f"value must not contain reserved attrs: {sorted(clash)}") + item: dict[str, Any] = dict(value) + item[self._pk] = pk + item[self._sk] = self._sk_val(key) + item["_key"] = key + await self._store.put(item, if_not_exists=if_not_exists, if_version=if_version) + + async def delete(self, key: str, *, pk: str = _DEFAULT_PK, if_version: int | None = None) -> None: + await self._store.delete(pk, self._sk_val(key), if_version=if_version) + + async def list(self, *, pk: str = _DEFAULT_PK, limit: int | None = None) -> list[Item]: + return await self._store.query(pk, sk_prefix=f"{self._name}#", limit=limit) + + async def increment( + self, key: str, *, field: str, by: int | float = 1, pk: str = _DEFAULT_PK + ) -> int | float: + return await self._store.increment(pk, self._sk_val(key), field=field, by=by) + + async def update( + self, key: str, *, set_fields: dict | None = None, add_fields: dict | None = None, + pk: str = _DEFAULT_PK, if_version: int | None = None, + ) -> Item: + return await self._store.update( + pk, self._sk_val(key), + set_fields=set_fields, add_fields=add_fields, if_version=if_version, + ) diff --git a/anton_state/schema.py b/anton_state/schema.py new file mode 100644 index 00000000..ae2aadc5 --- /dev/null +++ b/anton_state/schema.py @@ -0,0 +1,82 @@ +"""Declarative storage schema for state — the single source of truth. + +The manifest file (JSON) is read both by artifact code (StateSchema.from_manifest) +and by the publish/provisioning pipeline (subplans #2/#4). One file means the +schema in code and the provisioned table schema cannot diverge by construction. +The file is read without executing any code. +""" +from __future__ import annotations + +from pathlib import Path +from typing import Literal + +from pydantic import BaseModel, field_validator, model_validator + +# v1: string keys only. Numeric/binary keys would require support across the +# whole chain (pk: str, validation, begins_with) — deferred. +AttrType = Literal["S"] + + +class Attr(BaseModel): + name: str + type: AttrType = "S" + + +class Index(BaseModel): + name: str + pk: Attr + sk: Attr | None = None + + +class StateSchema(BaseModel): + version: int = 1 + pk: Attr + sk: Attr | None = None + gsis: list[Index] = [] + ttl_attribute: str | None = None + collections: list[str] = [] + + @field_validator("gsis") + @classmethod + def _no_gsis_in_v1(cls, v): + # v1 on the shared table: secondary indexes are not supported (they would + # require a generic overloaded-GSI pool in the broker). Deferred. + if v: + raise ValueError("secondary indexes (gsis) are not supported in v1") + return v + + @model_validator(mode="after") + def _validate_collections(self): + if not self.collections: + return self + # Collection encodes its name into the sort key (odm.py); no sk → no collections. + if self.sk is None: + raise ValueError("collections require a schema with a sort key") + seen: set[str] = set() + for name in self.collections: + if not name: + raise ValueError("collection names must be non-empty") + if "#" in name: + raise ValueError(f"collection name must not contain '#': {name!r}") + if name in seen: + raise ValueError(f"duplicate collection name: {name!r}") + seen.add(name) + return self + + def key_attrs(self) -> set[str]: + names: set[str] = {self.pk.name} + if self.sk: + names.add(self.sk.name) + for gsi in self.gsis: + names.add(gsi.pk.name) + if gsi.sk: + names.add(gsi.sk.name) + return names + + @classmethod + def from_manifest(cls, path: str | Path) -> "StateSchema": + text = Path(path).read_text(encoding="utf-8") + return cls.model_validate_json(text) + + def to_manifest(self, path: str | Path) -> None: + Path(path).write_text(self.model_dump_json(indent=2), encoding="utf-8") diff --git a/anton_state/sqlite_driver.py b/anton_state/sqlite_driver.py new file mode 100644 index 00000000..1645cd18 --- /dev/null +++ b/anton_state/sqlite_driver.py @@ -0,0 +1,232 @@ +"""Local driver: SQLite, zero-deps, WAL, lazy TTL, conditional writes. + +Stores the whole item as JSON in the `body` column; key/index/TTL values are +extracted via json_extract for queries. Physical deletion of expired items is +lazy (a separate sweep); reads filter them out, and conditional writes still +see the "zombie" — so behavior matches DynamoDB (spec §2). +""" +from __future__ import annotations + +import json +import sqlite3 +import time +from typing import Any + +from .base import _VERSION_ATTR +from .errors import ConditionalCheckFailed +from .schema import StateSchema +from .validation import validate_item, validate_key + + +class SQLiteDriver: + def __init__(self, path: str, schema: StateSchema): + self.path = path + self.schema = schema + self._pk = schema.pk.name + self._sk = schema.sk.name if schema.sk else None + self._ttl = schema.ttl_attribute + self._init_db() + # Lazy cleanup of accumulated "zombies" on startup (dev server) — + # otherwise expired items would live forever locally. + self.sweep_expired() + + def _connect(self) -> sqlite3.Connection: + con = sqlite3.connect(self.path) + con.execute("PRAGMA journal_mode=WAL") + con.execute("PRAGMA busy_timeout=5000") + con.row_factory = sqlite3.Row + return con + + def _init_db(self) -> None: + with self._connect() as con: + con.execute( + "CREATE TABLE IF NOT EXISTS items (" + " pk TEXT NOT NULL, sk TEXT NOT NULL DEFAULT ''," + " body TEXT NOT NULL, expires_at REAL," + " PRIMARY KEY (pk, sk))" + ) + + # --- helpers --- + def _sk_value(self, sk: str | None) -> str: + return "" if self._sk is None else (sk or "") + + def _expires_of(self, item: dict) -> float | None: + if self._ttl and self._ttl in item: + return float(item[self._ttl]) + return None + + def _row_to_item(self, row: sqlite3.Row) -> dict: + return json.loads(row["body"]) + + # --- Driver protocol --- + def get(self, pk: str, sk: str | None, *, consistent: bool) -> dict | None: + validate_key(pk, sk, self.schema) + now = time.time() + with self._connect() as con: + row = con.execute( + "SELECT body, expires_at FROM items WHERE pk=? AND sk=?", + (pk, self._sk_value(sk)), + ).fetchone() + if row is None: + return None + if row["expires_at"] is not None and row["expires_at"] < now: + return None + return json.loads(row["body"]) + + def put(self, item: dict, *, if_not_exists: bool, if_version: int | None) -> None: + validate_item(item, self.schema) + pk = item[self._pk] + sk = self._sk_value(item.get(self._sk) if self._sk else None) + con = self._connect() + try: + con.execute("BEGIN IMMEDIATE") + existing = con.execute( + "SELECT body FROM items WHERE pk=? AND sk=?", (pk, sk) + ).fetchone() + cur = json.loads(existing["body"]) if existing else None + if if_not_exists and existing is not None: + raise ConditionalCheckFailed(f"item already exists: {pk}/{sk}") + if if_version is not None: + if cur is None or cur.get(_VERSION_ATTR) != if_version: + raise ConditionalCheckFailed( + f"version mismatch on {pk}/{sk}: expected {if_version}" + ) + # Server-managed version: never trust a client-supplied _v. Monotonic + # to prevent ABA lost-update: an optimistic-lock write -> if_version+1; + # any other put -> epoch-millis (practically always greater than any + # prior _v, so a stale if_version can't match a re-put generation). + # Matches the broker (plan #2). + stored = {k: v for k, v in item.items() if k != _VERSION_ATTR} + stored[_VERSION_ATTR] = (if_version + 1) if if_version is not None else int(time.time() * 1000) + body = json.dumps(stored, separators=(",", ":"), ensure_ascii=False) + expires = self._expires_of(stored) + con.execute( + "INSERT INTO items (pk, sk, body, expires_at) VALUES (?,?,?,?) " + "ON CONFLICT(pk, sk) DO UPDATE SET body=excluded.body, expires_at=excluded.expires_at", + (pk, sk, body, expires), + ) + con.commit() + except Exception: + con.rollback() + raise + finally: + con.close() + + def delete(self, pk: str, sk: str | None, *, if_version: int | None) -> None: + validate_key(pk, sk, self.schema) + skv = self._sk_value(sk) + con = self._connect() + try: + con.execute("BEGIN IMMEDIATE") + if if_version is not None: + existing = con.execute( + "SELECT body FROM items WHERE pk=? AND sk=?", (pk, skv) + ).fetchone() + cur = json.loads(existing["body"]) if existing else None + if cur is None or cur.get(_VERSION_ATTR) != if_version: + raise ConditionalCheckFailed( + f"version mismatch on {pk}/{skv}: expected {if_version}" + ) + con.execute("DELETE FROM items WHERE pk=? AND sk=?", (pk, skv)) + con.commit() + except Exception: + con.rollback() + raise + finally: + con.close() + + def query( + self, + pk: str, + *, + sk_prefix: str | None, + filters: dict[str, Any] | None, + consistent: bool, + limit: int | None, + ) -> list[dict]: + now = time.time() + where = ["(expires_at IS NULL OR expires_at >= ?)", "pk = ?"] + params: list[Any] = [now, pk] + if sk_prefix is not None: + # Literal prefix match (NOT `LIKE`): '%' and '_' in the prefix are + # LIKE wildcards — an underscore in a collection name like + # "user_settings#" would over-match — while DynamoDB begins_with is + # literal. substr keeps local and cloud identical. + where.append("substr(sk, 1, length(?)) = ?") + params += [sk_prefix, sk_prefix] + + if filters: + for k, v in filters.items(): + where.append("json_extract(body, '$.' || ?) = ?") + params += [k, v] + + sql = "SELECT body FROM items WHERE " + " AND ".join(where) + " ORDER BY pk, sk" + if limit is not None: + sql += " LIMIT ?" + params.append(limit) + + with self._connect() as con: + rows = con.execute(sql, params).fetchall() + return [json.loads(r["body"]) for r in rows] + + def _rmw(self, pk, sk, mutate): + """Read-modify-write one item atomically; `mutate(item)->item` runs inside the txn. + + Partial mutation: unlike put, this does NOT run validate_item and does + NOT synthesise the logical pk/sk attributes — mirroring the schema-agnostic + broker (plan #2), where increment/update on an absent item create a minimal + item (mutated field + _v) without the logical key attributes. + """ + skv = self._sk_value(sk) + con = self._connect() + try: + con.execute("BEGIN IMMEDIATE") + row = con.execute("SELECT body FROM items WHERE pk=? AND sk=?", (pk, skv)).fetchone() + item = json.loads(row["body"]) if row else {} + item = mutate(item) + item[_VERSION_ATTR] = item.get(_VERSION_ATTR, 0) + 1 + body = json.dumps(item, separators=(",", ":"), ensure_ascii=False) + con.execute( + "INSERT INTO items (pk, sk, body, expires_at) VALUES (?,?,?,?) " + "ON CONFLICT(pk, sk) DO UPDATE SET body=excluded.body, expires_at=excluded.expires_at", + (pk, skv, body, self._expires_of(item)), + ) + con.commit() + return item + except Exception: + con.rollback() + raise + finally: + con.close() + + def increment(self, pk: str, sk: str | None, *, field: str, by: int | float) -> int | float: + validate_key(pk, sk, self.schema) + + def mut(item): + item[field] = item.get(field, 0) + by + return item + + return self._rmw(pk, sk, mut)[field] + + def update(self, pk, sk, *, set_fields, add_fields, if_version): + validate_key(pk, sk, self.schema) + + def mut(item): + if if_version is not None and item.get(_VERSION_ATTR, 0) != if_version: + raise ConditionalCheckFailed(f"version mismatch on {pk}/{sk}: expected {if_version}") + for k, v in (set_fields or {}).items(): + item[k] = v + for k, v in (add_fields or {}).items(): + item[k] = item.get(k, 0) + v + return item + + return self._rmw(pk, sk, mut) + + def sweep_expired(self) -> int: + """Lazy physical cleanup of expired items (not called on the hot path).""" + now = time.time() + with self._connect() as con: + cur = con.execute( + "DELETE FROM items WHERE expires_at IS NOT NULL AND expires_at < ?", (now,) + ) + return cur.rowcount diff --git a/anton_state/validation.py b/anton_state/validation.py new file mode 100644 index 00000000..67d272b8 --- /dev/null +++ b/anton_state/validation.py @@ -0,0 +1,81 @@ +"""Shared validation — deliberately strict, "like the cloud" (spec §2). + +Local SQLite is by nature more permissive than DynamoDB; here we reject what +prod would reject, so invalid data is not silently accepted locally. +""" +from __future__ import annotations + +import json + +from .errors import StateValidationError +from .schema import StateSchema + +MAX_ITEM_BYTES = 400 * 1024 + +# _pk/_sk/_ttl are physical control attributes added by the broker on the shared +# table; only _v (version) and _key (Collection) are allowed reserved names. +_RESERVED_UNDERSCORE = {"_v", "_key"} + + +def _check_value(value) -> None: + if isinstance(value, bool) or isinstance(value, (int, float, str)) or value is None: + return + if isinstance(value, dict): + for k, v in value.items(): + if not isinstance(k, str): + raise StateValidationError("map keys must be strings") + _check_value(v) + return + if isinstance(value, list): + for v in value: + _check_value(v) + return + raise StateValidationError(f"unsupported value type: {type(value).__name__}") + + +def validate_key(pk: str, sk: str | None, schema: StateSchema) -> None: + if not isinstance(pk, str) or pk == "": + raise StateValidationError("partition key must be a non-empty string") + if schema.sk is not None: + if sk is None or not isinstance(sk, str) or sk == "": + raise StateValidationError("sort key must be a non-empty string") + + +def validate_item(item: dict, schema: StateSchema) -> None: + if not isinstance(item, dict): + raise StateValidationError("item must be a dict") + + pk_name = schema.pk.name + if pk_name not in item: + raise StateValidationError(f"item missing partition key '{pk_name}'") + sk_name = schema.sk.name if schema.sk else None + validate_key(item.get(pk_name), item.get(sk_name) if sk_name else None, schema) + + # Empty strings are forbidden only in key attributes (DynamoDB has allowed + # empty strings in non-key attributes since 2020). + for key_name in schema.key_attrs(): + if key_name in item and isinstance(item[key_name], str) and item[key_name] == "": + raise StateValidationError(f"key attribute '{key_name}' must not be empty") + + # Forbid user attrs with a reserved "_" prefix (collide with the broker's + # physical _pk/_sk/_ttl on the shared table). _v and _key are allowed. + for name in item: + if name.startswith("_") and name not in _RESERVED_UNDERSCORE: + raise StateValidationError(f"attribute '{name}' uses a reserved '_' prefix") + + for value in item.values(): + _check_value(value) + + if schema.ttl_attribute and schema.ttl_attribute in item: + ttl = item[schema.ttl_attribute] + if isinstance(ttl, bool) or not isinstance(ttl, (int, float)): + raise StateValidationError( + f"TTL attribute '{schema.ttl_attribute}' must be a number (epoch seconds)" + ) + + try: + size = len(json.dumps(item, separators=(",", ":"), ensure_ascii=False).encode("utf-8")) + except (TypeError, ValueError) as e: + raise StateValidationError(f"item is not JSON-serializable: {e}") + if size > MAX_ITEM_BYTES: + raise StateValidationError(f"item too large: {size} > {MAX_ITEM_BYTES} bytes") diff --git a/pyproject.toml b/pyproject.toml index 4901fe87..da9be216 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -61,4 +61,4 @@ markers = [ ] [tool.hatch.build.targets.wheel] - packages = ["anton"] + packages = ["anton", "anton_state"] diff --git a/tests/test_anton_state/__init__.py b/tests/test_anton_state/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/test_anton_state/test_errors.py b/tests/test_anton_state/test_errors.py new file mode 100644 index 00000000..b1c5c660 --- /dev/null +++ b/tests/test_anton_state/test_errors.py @@ -0,0 +1,26 @@ +import pytest + +from anton_state import StateUnavailable # exported from the package +from anton_state.errors import ( + ConditionalCheckFailed, + StateError, + StateThrottled, + StateValidationError, +) + + +def test_hierarchy(): + for cls in (StateValidationError, ConditionalCheckFailed, StateThrottled, StateUnavailable): + assert issubclass(cls, StateError) + + +def test_state_unavailable_raises_as_state_error(): + with pytest.raises(StateError): + raise StateUnavailable("broker down") + + +def test_can_raise_and_message(): + try: + raise StateThrottled("rate exceeded") + except StateError as e: + assert "rate exceeded" in str(e) diff --git a/tests/test_anton_state/test_factory.py b/tests/test_anton_state/test_factory.py new file mode 100644 index 00000000..d5a39110 --- /dev/null +++ b/tests/test_anton_state/test_factory.py @@ -0,0 +1,33 @@ +from anton_state.schema import Attr, StateSchema +from anton_state.factory import open_store +from anton_state.sqlite_driver import SQLiteDriver +from anton_state.http_driver import HTTPDriver + +M = StateSchema(pk=Attr(name="pk"), sk=Attr(name="sk")) + + +def test_none_state_selects_sqlite(tmp_path): + store = open_store(M, state=None, local_path=str(tmp_path / "s.db")) + assert isinstance(store._driver, SQLiteDriver) + + +def test_local_path_from_env(tmp_path, monkeypatch): + p = str(tmp_path / "envstate.db") + monkeypatch.setenv("ANTON_ARTIFACT_STATE_PATH", p) + store = open_store(M, state=None) + assert isinstance(store._driver, SQLiteDriver) + assert store._driver.path == p + + +def test_schema_loaded_from_manifest_file_when_omitted(tmp_path, monkeypatch): + # single source: don't pass schema — the factory reads the manifest file + man = tmp_path / "state_manifest.json" + M.to_manifest(man) + monkeypatch.setenv("ANTON_STATE_MANIFEST", str(man)) + store = open_store(state=None, local_path=str(tmp_path / "s.db")) + assert store.schema == M + + +def test_cloud_state_selects_http_driver(): + store = open_store(M, state={"url": "https://b/_state", "token": "t.sig"}) + assert isinstance(store._driver, HTTPDriver) diff --git a/tests/test_anton_state/test_http_driver.py b/tests/test_anton_state/test_http_driver.py new file mode 100644 index 00000000..3706d758 --- /dev/null +++ b/tests/test_anton_state/test_http_driver.py @@ -0,0 +1,90 @@ +import pytest +from anton_state import StateSchema, Attr, ConditionalCheckFailed, StateThrottled, StateUnavailable +from anton_state.http_driver import HTTPDriver, _WireError + +_S = StateSchema(pk=Attr(name="pk"), sk=Attr(name="sk"), ttl_attribute="exp") + + +def _drv(monkeypatch, responder): + d = HTTPDriver(url="https://broker/_state", token="t.sig", schema=_S) + monkeypatch.setattr(d, "_post", responder) + return d + + +def test_get_sends_op_and_returns_item(monkeypatch): + seen = {} + + def responder(op, payload): + seen["op"], seen["payload"] = op, payload + return {"item": {"pk": "p", "sk": "s", "n": 1}} + + d = _drv(monkeypatch, responder) + item = d.get("p", "s", consistent=True) + assert item == {"pk": "p", "sk": "s", "n": 1} + assert seen["op"] == "get" + assert seen["payload"]["pk"] == "p" and seen["payload"]["sk"] == "s" + assert "namespace" not in seen["payload"] and "artifact_id" not in seen["payload"] + + +def test_put_extracts_ttl_and_cond(monkeypatch): + seen = {} + + def responder(op, payload): + seen.update(payload) + return {"ok": True} + + d = _drv(monkeypatch, responder) + d.put({"pk": "p", "sk": "s", "exp": 1234, "n": 1}, if_not_exists=True, if_version=None) + assert seen["ttl"] == 1234 + assert seen["cond"] == {"if_not_exists": True} + + +def test_increment_returns_value(monkeypatch): + d = _drv(monkeypatch, lambda op, p: {"value": 42}) + assert d.increment("c", "g", field="n", by=2) == 42 + + +def test_conditional_error_mapped(monkeypatch): + def responder(op, payload): + raise _WireError(409, "conditional", "nope") + + d = _drv(monkeypatch, responder) + with pytest.raises(ConditionalCheckFailed): + d.put({"pk": "p", "sk": "s"}, if_not_exists=True, if_version=None) + + +def test_mutation_not_retried_on_unavailable(monkeypatch): + calls = {"n": 0} + + def responder(op, payload): + calls["n"] += 1 + raise _WireError(503, "unavailable", "down") + + d = _drv(monkeypatch, responder) + with pytest.raises(StateUnavailable): + d.put({"pk": "p", "sk": "s"}, if_not_exists=False, if_version=None) + assert calls["n"] == 1 # no retry on mutation + + +def test_read_retried_on_unavailable(monkeypatch): + calls = {"n": 0} + + def responder(op, payload): + calls["n"] += 1 + if calls["n"] < 2: + raise _WireError(503, "unavailable", "down") + return {"item": None} + + d = _drv(monkeypatch, responder) + assert d.get("p", "s", consistent=True) is None + assert calls["n"] == 2 # retried once + + +def test_unauthorized_not_treated_as_unavailable(monkeypatch): + def responder(op, payload): + raise _WireError(401, "unauthorized", "bad token") + + d = _drv(monkeypatch, responder) + with pytest.raises(Exception) as ei: + d.get("p", "s", consistent=True) + assert not isinstance(ei.value, StateUnavailable) # definitely-not-applied, clearer error diff --git a/tests/test_anton_state/test_import.py b/tests/test_anton_state/test_import.py new file mode 100644 index 00000000..1ad4e25d --- /dev/null +++ b/tests/test_anton_state/test_import.py @@ -0,0 +1,5 @@ +def test_package_imports_and_has_version(): + import anton_state + + assert isinstance(anton_state.__version__, str) + assert anton_state.__version__ diff --git a/tests/test_anton_state/test_lazy_dynamo_import.py b/tests/test_anton_state/test_lazy_dynamo_import.py new file mode 100644 index 00000000..aedb021d --- /dev/null +++ b/tests/test_anton_state/test_lazy_dynamo_import.py @@ -0,0 +1,24 @@ +import subprocess +import sys +import textwrap +from pathlib import Path + +ANTON_ROOT = Path(__file__).resolve().parents[2] # .../anton + + +def test_sqlite_path_does_not_import_dynamo_or_boto3(tmp_path): + code = textwrap.dedent( + f""" + import sys + from anton_state import open_store, StateSchema, Attr + s = open_store(StateSchema(pk=Attr(name="pk")), state=None, local_path=r"{tmp_path}/x.db") + assert "anton_state.dynamo_driver" not in sys.modules, "dynamo eagerly imported" + assert "boto3" not in sys.modules, "boto3 eagerly imported" + print("OK") + """ + ) + r = subprocess.run( + [sys.executable, "-c", code], capture_output=True, text=True, cwd=str(ANTON_ROOT) + ) + assert r.returncode == 0, r.stderr + assert "OK" in r.stdout diff --git a/tests/test_anton_state/test_odm.py b/tests/test_anton_state/test_odm.py new file mode 100644 index 00000000..2769fb51 --- /dev/null +++ b/tests/test_anton_state/test_odm.py @@ -0,0 +1,64 @@ +import pytest +from anton_state.schema import Attr, StateSchema +from anton_state.factory import open_store +from anton_state.odm import Collection + +M = StateSchema(pk=Attr(name="pk"), sk=Attr(name="sk")) + + +@pytest.fixture +def store(tmp_path): + return open_store(M, state=None, local_path=str(tmp_path / "s.db")) + + +async def test_put_get_explicit_pk(store): + c = Collection(store, "task") + await c.put("t1", {"title": "do it"}, pk="proj1") + got = await c.get("t1", pk="proj1") + assert got["title"] == "do it" and got["_key"] == "t1" + + +async def test_default_pk_roundtrip(store): + todos = Collection(store, "todos") + await todos.put("a", {"text": "buy milk"}) # pk defaulted + got = await todos.get("a") + assert got["text"] == "buy milk" and got["_key"] == "a" + lst = await todos.list() + assert len(lst) == 1 + + +async def test_list_only_collection_prefix(store): + tasks = Collection(store, "task") + notes = Collection(store, "note") + await tasks.put("a", {"n": 1}, pk="p1") + await tasks.put("b", {"n": 2}, pk="p1") + await notes.put("a", {"x": 9}, pk="p1") + items = await tasks.list(pk="p1") + assert sorted(i["_key"] for i in items) == ["a", "b"] + + +async def test_delete(store): + c = Collection(store, "task") + await c.put("t1", {"n": 1}, pk="p1") + await c.delete("t1", pk="p1") + assert await c.get("t1", pk="p1") is None + + +async def test_increment(store): + counters = Collection(store, "counters") + assert await counters.increment("visits", field="n") == 1 + assert await counters.increment("visits", field="n", by=2) == 3 + + +async def test_update(store): + c = Collection(store, "task") + await c.put("t1", {"n": 1}, pk="p1") + item = await c.update("t1", set_fields={"done": True}, add_fields={"n": 4}, pk="p1") + assert item["done"] is True and item["n"] == 5 + + +async def test_put_rejects_reserved_attrs_in_value(store): + from anton_state.errors import StateValidationError + c = Collection(store, "task") + with pytest.raises(StateValidationError): + await c.put("t1", {"pk": "hijack"}, pk="p1") diff --git a/tests/test_anton_state/test_schema.py b/tests/test_anton_state/test_schema.py new file mode 100644 index 00000000..333aa23f --- /dev/null +++ b/tests/test_anton_state/test_schema.py @@ -0,0 +1,95 @@ +import pytest +from anton_state.schema import Attr, Index, StateSchema + + +def test_minimal_schema_defaults(): + m = StateSchema(pk=Attr(name="pk")) + assert m.version == 1 + assert m.pk.type == "S" + assert m.sk is None + assert m.gsis == [] + assert m.ttl_attribute is None + + +def test_key_attrs_collects_all_keys(): + m = StateSchema( + pk=Attr(name="pk"), + sk=Attr(name="sk"), + ttl_attribute="expires_at", + ) + assert m.key_attrs() == {"pk", "sk"} + + +def test_non_empty_gsis_rejected(): + # v1 shared table: secondary indexes are not supported. + with pytest.raises(ValueError): + StateSchema(pk=Attr(name="pk"), sk=Attr(name="sk"), + gsis=[Index(name="byUser", pk=Attr(name="user"))]) + + +def test_empty_gsis_ok(): + s = StateSchema(pk=Attr(name="pk"), sk=Attr(name="sk")) + assert s.gsis == [] + + +def test_non_string_key_type_rejected(): + # v1: string keys only + with pytest.raises(Exception): + Attr(name="pk", type="N") + + +def test_manifest_roundtrip(tmp_path): + m = StateSchema(pk=Attr(name="pk"), sk=Attr(name="sk"), ttl_attribute="expires_at") + path = tmp_path / "state_manifest.json" + m.to_manifest(path) + loaded = StateSchema.from_manifest(path) + assert loaded == m + + +def test_from_manifest_missing_file(tmp_path): + with pytest.raises(FileNotFoundError): + StateSchema.from_manifest(tmp_path / "nope.json") + + +def test_collections_default_empty(): + m = StateSchema(pk=Attr(name="pk"), sk=Attr(name="sk")) + assert m.collections == [] + + +def test_collections_roundtrip(tmp_path): + m = StateSchema(pk=Attr(name="pk"), sk=Attr(name="sk"), + collections=["comments", "users"]) + path = tmp_path / "state_manifest.json" + m.to_manifest(path) + loaded = StateSchema.from_manifest(path) + assert loaded.collections == ["comments", "users"] + assert loaded == m + + +def test_collections_missing_in_manifest_defaults_empty(tmp_path): + path = tmp_path / "state_manifest.json" + path.write_text('{"pk": {"name": "pk"}, "sk": {"name": "sk"}}', encoding="utf-8") + assert StateSchema.from_manifest(path).collections == [] + + +def test_collections_hash_in_name_rejected(): + with pytest.raises(ValueError): + StateSchema(pk=Attr(name="pk"), sk=Attr(name="sk"), + collections=["comments#x"]) + + +def test_collections_duplicate_rejected(): + with pytest.raises(ValueError): + StateSchema(pk=Attr(name="pk"), sk=Attr(name="sk"), + collections=["a", "a"]) + + +def test_collections_empty_name_rejected(): + with pytest.raises(ValueError): + StateSchema(pk=Attr(name="pk"), sk=Attr(name="sk"), collections=[""]) + + +def test_collections_without_sk_rejected(): + # Collection encodes its name into the sort key; without sk it is meaningless. + with pytest.raises(ValueError): + StateSchema(pk=Attr(name="pk"), collections=["comments"]) diff --git a/tests/test_anton_state/test_single_source.py b/tests/test_anton_state/test_single_source.py new file mode 100644 index 00000000..3a1fcd34 --- /dev/null +++ b/tests/test_anton_state/test_single_source.py @@ -0,0 +1,17 @@ +from anton_state.schema import Attr, StateSchema +from anton_state.factory import open_store + +M = StateSchema( + pk=Attr(name="pk"), sk=Attr(name="sk"), + ttl_attribute="expires_at", +) + + +def test_backend_and_publish_read_same_schema(tmp_path): + manifest = tmp_path / "state_manifest.json" + M.to_manifest(manifest) + # "publish side" (without executing artifact code) + publish_schema = StateSchema.from_manifest(manifest) + # "backend side" — via the factory, from the same file + store = open_store(state=None, local_path=str(tmp_path / "s.db"), manifest_path=str(manifest)) + assert store.schema == publish_schema == M diff --git a/tests/test_anton_state/test_sqlite_driver.py b/tests/test_anton_state/test_sqlite_driver.py new file mode 100644 index 00000000..f2fc16fb --- /dev/null +++ b/tests/test_anton_state/test_sqlite_driver.py @@ -0,0 +1,116 @@ +import time +import pytest +from anton_state.schema import Attr, StateSchema +from anton_state.errors import ConditionalCheckFailed, StateValidationError +from anton_state.sqlite_driver import SQLiteDriver + +M = StateSchema( + pk=Attr(name="pk"), + sk=Attr(name="sk"), + ttl_attribute="expires_at", +) + + +@pytest.fixture +def drv(tmp_path): + return SQLiteDriver(str(tmp_path / "state.db"), M) + + +def test_put_get_roundtrip(drv): + drv.put({"pk": "u1", "sk": "profile", "name": "Alice"}, if_not_exists=False, if_version=None) + got = drv.get("u1", "profile", consistent=True) + assert got["pk"] == "u1" and got["sk"] == "profile" and got["name"] == "Alice" + assert "_v" in got # server-managed version present + + +def test_get_missing_returns_none(drv): + assert drv.get("nope", "x", consistent=True) is None + + +def test_wal_enabled(drv): + import sqlite3 + con = sqlite3.connect(drv.path) + mode = con.execute("PRAGMA journal_mode").fetchone()[0] + con.close() + assert mode.lower() == "wal" + + +def test_if_not_exists_conflict(drv): + drv.put({"pk": "u1", "sk": "s"}, if_not_exists=True, if_version=None) + with pytest.raises(ConditionalCheckFailed): + drv.put({"pk": "u1", "sk": "s"}, if_not_exists=True, if_version=None) + + +def test_put_assigns_monotonic_version(drv): + drv.put({"pk": "c", "sk": "g", "n": 1}, if_not_exists=True, if_version=None) + v0 = drv.get("c", "g", consistent=True)["_v"] + assert v0 > 1_000_000_000_000 # epoch-millis, NOT a reset-to-1 (prevents ABA) + # unconditional REPLACE assigns a fresh, non-decreasing version + drv.put({"pk": "c", "sk": "g", "n": 2}, if_not_exists=False, if_version=None) + assert drv.get("c", "g", consistent=True)["_v"] >= v0 + + +def test_put_if_version_checks_and_bumps(drv): + drv.put({"pk": "c", "sk": "g", "n": 1}, if_not_exists=True, if_version=None) + v0 = drv.get("c", "g", consistent=True)["_v"] + with pytest.raises(ConditionalCheckFailed): + drv.put({"pk": "c", "sk": "g", "n": 9}, if_not_exists=False, if_version=v0 + 999) + drv.put({"pk": "c", "sk": "g", "n": 2}, if_not_exists=False, if_version=v0) + assert drv.get("c", "g", consistent=True)["_v"] == v0 + 1 + + +def test_put_overwrites_client_supplied_version(drv): + drv.put({"pk": "c", "sk": "g", "n": 1, "_v": 999}, if_not_exists=True, if_version=None) + assert drv.get("c", "g", consistent=True)["_v"] != 999 # client _v ignored (server sets it) + + +def test_ttl_hidden_on_read_but_zombie_blocks_conditional(drv): + past = time.time() - 10 + drv.put({"pk": "u1", "sk": "s", "expires_at": past}, if_not_exists=False, if_version=None) + # expired → invisible on read + assert drv.get("u1", "s", consistent=True) is None + # but physically present → if_not_exists fails (parity with the DynamoDB zombie) + with pytest.raises(ConditionalCheckFailed): + drv.put({"pk": "u1", "sk": "s"}, if_not_exists=True, if_version=None) + + +def test_query_prefix_and_ttl_filter(drv): + drv.put({"pk": "u1", "sk": "msg#1", "t": "a"}, if_not_exists=False, if_version=None) + drv.put({"pk": "u1", "sk": "msg#2", "t": "b"}, if_not_exists=False, if_version=None) + drv.put({"pk": "u1", "sk": "note#1"}, if_not_exists=False, if_version=None) + drv.put({"pk": "u1", "sk": "msg#old", "expires_at": time.time() - 5}, if_not_exists=False, if_version=None) + rows = drv.query("u1", sk_prefix="msg#", filters=None, consistent=True, limit=None) + sks = sorted(r["sk"] for r in rows) + assert sks == ["msg#1", "msg#2"] + + +def test_query_prefix_is_literal_not_like(drv): + # An underscore in the prefix must NOT act as a LIKE wildcard. + drv.put({"pk": "u1", "sk": "user_settings#a"}, if_not_exists=False, if_version=None) + drv.put({"pk": "u1", "sk": "userXsettings#b"}, if_not_exists=False, if_version=None) + rows = drv.query("u1", sk_prefix="user_settings#", filters=None, consistent=True, limit=None) + assert [r["sk"] for r in rows] == ["user_settings#a"] + + +def test_query_filters_equality(drv): + drv.put({"pk": "u1", "sk": "a", "kind": "x"}, if_not_exists=False, if_version=None) + drv.put({"pk": "u1", "sk": "b", "kind": "y"}, if_not_exists=False, if_version=None) + rows = drv.query("u1", sk_prefix=None, filters={"kind": "x"}, consistent=True, limit=None) + assert [r["sk"] for r in rows] == ["a"] + + +def test_increment_atomic(drv): + assert drv.increment("c", "g", field="n", by=1) == 1 # creates item + assert drv.increment("c", "g", field="n", by=2) == 3 + assert drv.get("c", "g", consistent=True)["n"] == 3 + + +def test_update_set_and_add(drv): + drv.put({"pk": "c", "sk": "g", "n": 1}, if_not_exists=True, if_version=None) + item = drv.update("c", "g", set_fields={"name": "x"}, add_fields={"n": 4}, if_version=None) + assert item["name"] == "x" and item["n"] == 5 + + +def test_put_validates(drv): + with pytest.raises(StateValidationError): + drv.put({"sk": "s"}, if_not_exists=False, if_version=None) # no pk diff --git a/tests/test_anton_state/test_store_facade.py b/tests/test_anton_state/test_store_facade.py new file mode 100644 index 00000000..97332e2e --- /dev/null +++ b/tests/test_anton_state/test_store_facade.py @@ -0,0 +1,78 @@ +import pytest +from anton_state.base import Store +from anton_state.schema import Attr, StateSchema + + +class FakeDriver: + def __init__(self): + self.calls = [] + + def get(self, pk, sk, *, consistent): + self.calls.append(("get", pk, sk, consistent)) + return {"pk": pk} + + def put(self, item, *, if_not_exists, if_version): + self.calls.append(("put", item, if_not_exists, if_version)) + + def delete(self, pk, sk, *, if_version): + self.calls.append(("delete", pk, sk, if_version)) + + def query(self, pk, *, sk_prefix, filters, consistent, limit): + self.calls.append(("query", pk, sk_prefix, filters, consistent, limit)) + return [{"pk": pk}] + + def increment(self, pk, sk, *, field, by): + self.calls.append(("increment", pk, sk, field, by)) + return 7 + + def update(self, pk, sk, *, set_fields, add_fields, if_version): + self.calls.append(("update", pk, sk, set_fields, add_fields, if_version)) + return {"pk": pk} + + +M = StateSchema(pk=Attr(name="pk"), sk=Attr(name="sk")) + + +async def test_get_delegates_with_default_consistent_true(): + d = FakeDriver() + store = Store(d, M) + res = await store.get("u1", "profile") + assert res == {"pk": "u1"} + assert d.calls[0] == ("get", "u1", "profile", True) + + +async def test_query_defaults(): + d = FakeDriver() + store = Store(d, M) + res = await store.query("u1", sk_prefix="msg#") + assert res == [{"pk": "u1"}] + assert d.calls[0][0] == "query" + + +async def test_put_rejects_mutually_exclusive_conditions(): + from anton_state.errors import StateValidationError + store = Store(FakeDriver(), M) + with pytest.raises(StateValidationError): + await store.put({"pk": "u1", "sk": "s"}, if_not_exists=True, if_version=1) + + +async def test_query_has_no_index_kwarg(): + store = Store(FakeDriver(), M) + with pytest.raises(TypeError): + await store.query("u1", index="byUser") # index removed in v1 + + +async def test_increment_delegates_and_returns(): + d = FakeDriver() + store = Store(d, M) + v = await store.increment("c", "global", field="n", by=2) + assert v == 7 + assert d.calls[-1] == ("increment", "c", "global", "n", 2) + + +async def test_update_delegates(): + d = FakeDriver() + store = Store(d, M) + item = await store.update("c", "g", set_fields={"a": 1}, add_fields={"n": 1}) + assert item == {"pk": "c"} + assert d.calls[-1] == ("update", "c", "g", {"a": 1}, {"n": 1}, None) diff --git a/tests/test_anton_state/test_validation.py b/tests/test_anton_state/test_validation.py new file mode 100644 index 00000000..f14caff0 --- /dev/null +++ b/tests/test_anton_state/test_validation.py @@ -0,0 +1,52 @@ +import pytest +from anton_state.schema import Attr, StateSchema +from anton_state.errors import StateValidationError +from anton_state.validation import validate_item, validate_key, MAX_ITEM_BYTES + +M = StateSchema(pk=Attr(name="pk"), sk=Attr(name="sk"), ttl_attribute="expires_at") + + +def test_ok_item_passes(): + validate_item({"pk": "u1", "sk": "profile", "name": "Alice"}, M) + + +def test_missing_pk_rejected(): + with pytest.raises(StateValidationError): + validate_item({"sk": "profile"}, M) + + +def test_empty_string_key_rejected(): + with pytest.raises(StateValidationError): + validate_item({"pk": "", "sk": "profile"}, M) + + +def test_unsupported_type_rejected(): + with pytest.raises(StateValidationError): + validate_item({"pk": "u1", "sk": "s", "when": object()}, M) + + +def test_oversize_item_rejected(): + big = {"pk": "u1", "sk": "s", "blob": "x" * (MAX_ITEM_BYTES + 10)} + with pytest.raises(StateValidationError): + validate_item(big, M) + + +def test_ttl_must_be_number_epoch(): + with pytest.raises(StateValidationError): + validate_item({"pk": "u1", "sk": "s", "expires_at": "2026-01-01"}, M) + validate_item({"pk": "u1", "sk": "s", "expires_at": 1893456000}, M) + + +def test_validate_key_empty_rejected(): + with pytest.raises(StateValidationError): + validate_key("", None, M) + + +def test_underscore_prefixed_user_attr_rejected(): + with pytest.raises(StateValidationError): + validate_item({"pk": "p", "sk": "s", "_ttl": 1}, M) + + +def test_reserved_underscore_attrs_allowed(): + # _v (version) and _key (Collection) are the only allowed "_" names. + validate_item({"pk": "p", "sk": "s", "_v": 3, "_key": "k", "n": 1}, M) diff --git a/tests/test_backend_launcher_env.py b/tests/test_backend_launcher_env.py new file mode 100644 index 00000000..2fd1c8fb --- /dev/null +++ b/tests/test_backend_launcher_env.py @@ -0,0 +1,35 @@ +import os +from pathlib import Path + +import anton_state +from anton.core.artifacts.backend_launcher import ( + _anton_state_pythonpath_dir, + _build_backend_env, +) + + +def test_pythonpath_dir_contains_only_anton_state(): + d = _anton_state_pythonpath_dir() + assert os.listdir(d) == ["anton_state"] + # resolves to the real package + assert (Path(d) / "anton_state" / "__init__.py").resolve() == Path( + anton_state.__file__ + ).resolve() + + +def test_build_env_prepends_isolated_dir_to_pythonpath(): + env = _build_backend_env({"PYTHONPATH": "/existing"}) + parts = env["PYTHONPATH"].split(os.pathsep) + assert parts[0] == _anton_state_pythonpath_dir() + assert "/existing" in parts + + +def test_build_env_without_existing_pythonpath(): + env = _build_backend_env(None) + assert env["PYTHONPATH"] == _anton_state_pythonpath_dir() + + +def test_build_env_merges_extra_env(): + env = _build_backend_env({"DS_X__Y": "z"}) + assert env["DS_X__Y"] == "z" + assert env["PATH"] == os.environ["PATH"] # inherits parent env diff --git a/tests/test_prompts_state.py b/tests/test_prompts_state.py new file mode 100644 index 00000000..a79ce4a8 --- /dev/null +++ b/tests/test_prompts_state.py @@ -0,0 +1,64 @@ +"""State-store guidance (ENG-704). + +The backend + STATE contract was moved out of BACKEND_GENERATION_PROMPT and into +the `build-fullstack-backend` built-in skill (recalled on demand). These tests +assert the STATE guidance survives in the skill body, which is where an agent +actually reads it. +""" + +from pathlib import Path + +import pytest + +from anton.core.memory import skills as skills_mod +from anton.core.memory.skills import SkillStore + +REAL_BUILTIN_ROOT = Path(skills_mod.__file__).parent / "builtin_skills" + + +@pytest.fixture(scope="module") +def T() -> str: + """The build-fullstack-backend skill body (frontmatter stripped).""" + store = SkillStore(root=REAL_BUILTIN_ROOT.parent / "_no_user_skills", builtin_root=REAL_BUILTIN_ROOT) + skill = store.load("build-fullstack-backend") + assert skill is not None and skill.provenance == "builtin" + return skill.declarative_md + + +def test_template_declares_state_slot(T): + assert "STATE = None" in T + + +def test_template_shows_open_store_usage(T): + assert "from anton_state import open_store" in T + assert "state_manifest.json" in T + assert "Path(__file__).resolve().parent" in T + + +def test_rules_position_state_for_light_and_external_db_for_heavy(T): + low = T.lower() + assert "state_manifest.json" in T + # light state → STATE; heavy/relational → external DB + assert "external" in low and "relational" in low + + +# --- shared-table API guidance (ENG-704) --- +def test_state_overlay_is_url_token_not_creds(T): + # cloud overlay is now {url, token}, not the old STS {table, region, credentials} + assert "url, token" in T + assert "table, region" not in T + + +def test_store_api_mentions_collection_and_atomics(T): + assert "Collection" in T + assert "increment" in T + assert "update" in T + + +def test_store_api_drops_gsi_index_kwarg(T): + # the query() signature no longer offers an index= argument in v1 + assert "index=None" not in T + + +def test_rules_warn_against_manual_mutation_retries(T): + assert "retry" in T.lower() diff --git a/tests/test_publisher_state.py b/tests/test_publisher_state.py new file mode 100644 index 00000000..1fd4f1dd --- /dev/null +++ b/tests/test_publisher_state.py @@ -0,0 +1,183 @@ +import io +import json +import zipfile +from pathlib import Path + +import pytest +from anton.publisher import ( + _FULLSTACK_EXCLUDED, + _read_state_manifest, + _zip_fullstack, +) + + +def _make_fullstack_dir(tmp_path: Path) -> Path: + d = tmp_path / "art" + d.mkdir() + (d / "backend.py").write_text("STATE = None\n", encoding="utf-8") + (d / "requirements.txt").write_text("fastapi\n", encoding="utf-8") + (d / "state_manifest.json").write_text( + json.dumps({"pk": {"name": "pk"}, "sk": {"name": "sk"}}), encoding="utf-8" + ) + (d / ".anton_state.db").write_bytes(b"SQLite format 3\x00local-only") + static = d / "static" + static.mkdir() + (static / "index.html").write_text("", encoding="utf-8") + return d + + +def test_bundle_vendors_anton_state_package(tmp_path): + zbytes, included = _zip_fullstack(_make_fullstack_dir(tmp_path)) + names = zipfile.ZipFile(io.BytesIO(zbytes)).namelist() + assert "anton_state/__init__.py" in names + assert "anton_state/sqlite_driver.py" in names + assert "anton_state/factory.py" in names + + +def test_bundle_includes_manifest(tmp_path): + zbytes, included = _zip_fullstack(_make_fullstack_dir(tmp_path)) + names = zipfile.ZipFile(io.BytesIO(zbytes)).namelist() + assert "state_manifest.json" in names + + +def test_local_db_never_bundled_defensive(tmp_path): + # _zip_fullstack is allowlist-based, so root files like the SQLite db are + # never bundled regardless. The exclusion set is a defensive contract (in + # case bundling changes) and covers the WAL side-files (-wal has the freshest data). + zbytes, _ = _zip_fullstack(_make_fullstack_dir(tmp_path)) + names = zipfile.ZipFile(io.BytesIO(zbytes)).namelist() + assert ".anton_state.db" not in names + for name in (".anton_state.db", ".anton_state.db-wal", ".anton_state.db-shm"): + assert name in _FULLSTACK_EXCLUDED + + +def test_bundle_omits_manifest_when_absent(tmp_path): + d = tmp_path / "art" + d.mkdir() + (d / "backend.py").write_text("x=1\n", encoding="utf-8") + zbytes, _ = _zip_fullstack(d) + names = zipfile.ZipFile(io.BytesIO(zbytes)).namelist() + assert "state_manifest.json" not in names + # anton_state is vendored regardless (the backend may use it) + assert "anton_state/__init__.py" in names + + +def test_read_state_manifest_parses_json(tmp_path): + d = tmp_path / "art" + d.mkdir() + (d / "state_manifest.json").write_text( + json.dumps({"pk": {"name": "pk"}, "ttl_attribute": "expires_at"}), encoding="utf-8" + ) + m = _read_state_manifest(d) + assert m == {"pk": {"name": "pk"}, "ttl_attribute": "expires_at"} + + +def test_read_state_manifest_none_when_absent(tmp_path): + d = tmp_path / "art" + d.mkdir() + assert _read_state_manifest(d) is None + + +# --- schema-change warning + snapshot (ENG-704 shared table) --- +from anton import publisher as _pub + + +def test_snapshot_excluded_from_bundle(): + assert _pub._STATE_SNAPSHOT in _pub._FULLSTACK_EXCLUDED + + +def test_warns_on_key_schema_change(tmp_path): + (tmp_path / _pub._STATE_SNAPSHOT).write_text(json.dumps( + {"version": 1, "pk": {"name": "pk", "type": "S"}, "sk": {"name": "sk", "type": "S"}})) + new = {"version": 1, "pk": {"name": "userId", "type": "S"}, "sk": None} # changed pk, dropped sk + warn = _pub._warn_state_schema_change(tmp_path, new) + assert warn is not None and "key schema changed" in warn.lower() + + +def test_no_warning_when_schema_unchanged(tmp_path): + m = {"version": 1, "pk": {"name": "pk", "type": "S"}, "sk": {"name": "sk", "type": "S"}} + (tmp_path / _pub._STATE_SNAPSHOT).write_text(json.dumps(m)) + assert _pub._warn_state_schema_change(tmp_path, m) is None + + +def test_warn_does_not_write_snapshot(tmp_path): + m = {"version": 1, "pk": {"name": "pk", "type": "S"}, "sk": None} + assert _pub._warn_state_schema_change(tmp_path, m) is None # no prior snapshot -> no warning + assert not (tmp_path / _pub._STATE_SNAPSHOT).exists() # warn NEVER writes (only _save does) + + +def test_save_snapshot_writes_after_success(tmp_path): + m = {"version": 1, "pk": {"name": "pk", "type": "S"}, "sk": None} + _pub._save_state_snapshot(tmp_path, m) + assert (tmp_path / _pub._STATE_SNAPSHOT).is_file() + + +# --- collection-drop block (ENG-704 collections registry) --- + + +def test_blocked_drops_empty_without_snapshot(tmp_path): + new = {"version": 1, "pk": {"name": "pk"}, "sk": {"name": "sk"}, + "collections": ["a"]} + assert _pub._blocked_collection_drops(tmp_path, new) == set() + + +def test_blocked_drops_empty_on_addition(tmp_path): + (tmp_path / _pub._STATE_SNAPSHOT).write_text(json.dumps( + {"collections": ["a", "b"]})) + new = {"collections": ["a", "b", "c"]} + assert _pub._blocked_collection_drops(tmp_path, new) == set() + + +def test_blocked_drops_detects_removed(tmp_path): + (tmp_path / _pub._STATE_SNAPSHOT).write_text(json.dumps( + {"collections": ["a", "b"]})) + assert _pub._blocked_collection_drops(tmp_path, {"collections": ["a"]}) == {"b"} + + +def test_blocked_drops_detects_rename(tmp_path): + (tmp_path / _pub._STATE_SNAPSHOT).write_text(json.dumps( + {"collections": ["comments"]})) + assert _pub._blocked_collection_drops( + tmp_path, {"collections": ["comment"]}) == {"comments"} + + +def test_blocked_drops_tolerates_snapshot_without_collections(tmp_path): + (tmp_path / _pub._STATE_SNAPSHOT).write_text(json.dumps( + {"version": 1, "pk": {"name": "pk"}})) # pre-feature snapshot + assert _pub._blocked_collection_drops(tmp_path, {"collections": ["a"]}) == set() + + +def _make_stateful_dir(tmp_path, collections): + d = tmp_path / "art" + d.mkdir() + d.joinpath("backend.py").write_text("STATE = None\n", encoding="utf-8") + d.joinpath("requirements.txt").write_text("fastapi\n", encoding="utf-8") + d.joinpath("metadata.json").write_text(json.dumps({ + "id": "abc12345", "slug": "art", "createdAt": "2026-07-24T00:00:00Z", + "updatedAt": "2026-07-24T00:00:00Z", "name": "art", "description": "t", + "type": "fullstack-stateful-app", "primary": "static/index.html", + }), encoding="utf-8") + d.joinpath("state_manifest.json").write_text(json.dumps({ + "version": 1, "pk": {"name": "pk"}, "sk": {"name": "sk"}, + "collections": collections}), encoding="utf-8") + static = d / "static" + static.mkdir() + static.joinpath("index.html").write_text("", encoding="utf-8") + return d + + +def test_publish_blocks_on_collection_drop_without_uploading(tmp_path, monkeypatch): + d = _make_stateful_dir(tmp_path, collections=["a"]) # new schema drops "b" + (d / _pub._STATE_SNAPSHOT).write_text(json.dumps({"collections": ["a", "b"]})) + called = {"n": 0} + + def _fake_request(*a, **k): + called["n"] += 1 + return "{}" + + monkeypatch.setattr(_pub, "minds_request", _fake_request) + monkeypatch.setattr(_pub, "_collect_datasource_secrets", lambda *a, **k: ({}, [])) + + with pytest.raises(_pub.StatePublishBlocked): + _pub.publish(d, api_key="k", report_id="rid-1") + assert called["n"] == 0 # aborted before any upload diff --git a/tests/test_unpublish_clears_local.py b/tests/test_unpublish_clears_local.py index 51d47508..deaabf8a 100644 --- a/tests/test_unpublish_clears_local.py +++ b/tests/test_unpublish_clears_local.py @@ -9,6 +9,7 @@ from rich.console import Console import anton.chat as chat +from anton.publisher import _STATE_SNAPSHOT def _make_published_artifact(tmp_path: Path) -> tuple[Path, Path]: @@ -49,3 +50,27 @@ async def test_unpublish_removes_local_entry(tmp_path): # The stale entry must be gone so /publish treats it as fresh. data = json.loads(pub.read_text()) assert "report.html" not in data + + +@pytest.mark.asyncio +async def test_unpublish_removes_state_snapshot(tmp_path): + root, pub = _make_published_artifact(tmp_path) + snap = pub.parent / _STATE_SNAPSHOT + snap.write_text(json.dumps({"collections": ["comments"]}), encoding="utf-8") + + settings = mock.Mock() + settings.minds_api_key = "key" + settings.artifacts_dir = str(root) + settings.publish_url = "https://view.test" + settings.minds_ssl_verify = True + + reports = [{"title": "server-time-display-2", "report_id": "247c4983", + "md5": "m", "view_url": "https://view/a/247c4983"}] + + with mock.patch("anton.publisher.list_published", return_value=reports), \ + mock.patch("anton.publisher.unpublish", return_value={}), \ + mock.patch("anton.chat.prompt_or_cancel", + new=mock.AsyncMock(side_effect=["1", "y"])): + await chat._handle_unpublish(Console(), settings, mock.Mock()) + + assert not snap.exists() # snapshot cleared → next publish is treated as fresh