Playground

AUTHORIZATION_PLAN.md

@@ -0,0 +1,101 @@
+# Scalable Object-Level Authorization — Design & Plan
+
+Status: DONE (all steps implemented, tested, documented)
+Scope: `agentflow-api` (auth layer) + `agentflow` core (`aget_thread_owner` source-of-truth lookup)
+
+## Goal
+
+Owner-only access to threads, enforced on **every** thread-touching step (invoke, stream,
+stop, fix, and all checkpointer read/write/delete), **on by default in production** when the
+developer enables auth, fully overridable, and **scalable** — no database round-trip per
+request.
+
+## Design principles
+
+1. **Ownership is immutable.** A thread's owner is set at creation and never changes; only
+   deletion removes it. Therefore the owner is safe to cache aggressively (cache-forever)
+   and we essentially never re-hit the DB for a known thread.
+2. **Single decision point.** One `AuthorizationBackend`, invoked before every handler. No
+   per-endpoint bespoke checks.
+3. **Secure by construction.** No route can ship unprotected (enforced at boot).
+4. **Graceful degradation.** No checkpointer / a checkpointer that can't resolve ownership →
+   warn + allow (nothing persisted to protect). Redis is optional.
+
+## Components
+
+### 1. Source of truth: `aget_thread_owner` (core) — DONE
+
+`BaseCheckpointer.aget_thread_owner(thread_id) -> user_id | None`:
+- base: raises `NotImplementedError` (so a backend can't silently report "no owner", which
+  would defeat owner-based authorization).
+- `PgCheckpointer`: `SELECT user_id FROM threads WHERE thread_id = $1`.
+- `InMemoryCheckpointer`, `SqliteCheckpointer`: read the stored thread record's `user_id`.
+
+This is the DB-level lookup the cache sits in front of. Already implemented + unit tested.
+
+### 2. `ThreadOwnershipResolver` (the scalable cache) — NEW (`agentflow-api`)
+
+- **L1**: in-process LRU, bounded (default 10_000), immutable owners cached with **no TTL**.
+- **L2**: Redis, reusing the app's existing Redis client when configured; key
+  `af:authz:owner:{thread_id}`. Gracefully L1-only when Redis is absent.
+- `owner_of(thread_id) -> str | None`: L1 → L2 (fill L1) → DB `aget_thread_owner` (fill L2+L1).
+  **Only positive results are cached; `None` is never cached** — a nonexistent thread can
+  become owned by the very next request, so caching `None` would open a TOCTOU hole.
+- `evict(thread_id)`: clears L1+L2. Called when a thread is deleted.
+- Cost: first touch of a thread = 1 lookup; every subsequent request = in-process hit.
+
+### 3. `OwnershipAuthorizationBackend` — rewire to use the resolver
+
+Decision table (cached; no raw DB call):
+- no `user_id` → deny
+- non-thread resource (`store`, `files`) → allow (they self-scope)
+- no `resource_id` → allow (list/create endpoints)
+- `owner = resolver.owner_of(resource_id)`:
+  - `None` → allow (brand-new session / empty read)
+  - `owner == user_id` → allow
+  - `owner != user_id` → **deny, for every action including invoke/stream**
+- resolver reports unsupported backend → warn + allow; resolver error → fail closed (deny).
+
+### 4. Secure-by-construction enforcement (fail-closed at boot)
+
+`assert_all_routes_protected(app)` runs at startup: walk every route; any non-public route
+(allowlist: `/ping`) whose dependency tree lacks `RequirePermission` makes the server
+**refuse to start** with a clear error. Keeps precise per-handler `(resource, action)`, adds
+zero per-request overhead, and turns a forgotten guard into a deploy-time failure instead of
+a silently-open production endpoint. (Delivers the "authorization on every step" guarantee
+more robustly than a literal router dependency, which cannot cleanly derive each route's
+action.)
+
+### 5. Invalidation
+
+`CheckpointerService.delete_thread` (and the `aclean_thread` path) calls
+`resolver.evict(thread_id)`.
+
+### 6. Config surface (unchanged)
+
+`"authorization"`: `"ownership"` | `"allow_all"`/`"default"`/`"none"` | `"module:attr"` |
+unset → `ownership` in production, `allow_all` in development. Developer choice always wins.
+
+## Test plan
+
+- Resolver: L1 hit, L2 promote-to-L1, DB fill, negative-not-cached, evict, LRU bound.
+- Backend: full decision table over the cached path.
+- Startup invariant: a route missing `RequirePermission` → boot raises; allowlist passes.
+- Integration: non-owner invoke/stream/read/delete → 403; owner → ok; second call served from
+  cache (asserted with a call-counting fake checkpointer — no second DB hit).
+- Core: `aget_thread_owner` for in-memory/sqlite (done).
+
+## Docs
+
+Update `docs/how-to/api-cli/add-auth.md` and `agentflow-api/CLAUDE.md`: caching model,
+immutability, Redis opt-in, boot-time guarantee.
+
+## Execution order
+
+1. ThreadOwnershipResolver + unit tests.
+2. Rewire OwnershipAuthorizationBackend onto the resolver + update tests.
+3. Wire resolver into DI (loader) and evict on delete_thread.
+4. Startup invariant `assert_all_routes_protected` + boot wiring + tests.
+5. Integration tests (owner-only + cache-hit assertion).
+6. Docs.
+7. Full suite (api + core) green with coverage.

CLAUDE.md

@@ -5,9 +5,9 @@ core framework see `agentflow/CLAUDE.md`; for the TS client, docs, or playground
 for the monorepo overview see the workspace-root `CLAUDE.md`.
 
 - Package name (PyPI): `10xscale-agentflow-cli`
-- Version: `0.3.2.9` (`pyproject.toml`). `CLI_VERSION` and `agentflow_cli.__version__` are
+- Version: `0.4.0` (`pyproject.toml`). `CLI_VERSION` and `agentflow_cli.__version__` are
   single-sourced from the installed distribution metadata (falling back to `pyproject.toml`), so
-  `agentflow version` reports `0.3.2.9` consistently. The previous `1.0.0` hardcode is gone.
+  `agentflow version` reports `0.4.0` consistently. The previous `1.0.0` hardcode is gone.
 - Requires: Python >= 3.12 · Status: `4 - Beta`
 - Console entry point: `agentflow = agentflow_cli.cli.main:main`
 - Depends on the core framework: `10xscale-agentflow>=0.7.0`.
@@ -76,8 +76,25 @@ and for redis backend a `redis` sub-object `{ "url", "prefix" }` (or shorthand U
   load if missing). JWT logic lives in `core/auth/jwt_auth.py`.
 - `"auth": {"method": "custom", "path": "module:attr"}` loads your `BaseAuth` subclass
   (`from agentflow_cli import BaseAuth`).
-- Authorization (RBAC, per-tool) is separate: `core/auth/authorization.py`
-  (`AuthorizationBackend` / `DefaultAuthorizationBackend`), wired via the `authorization` key.
+- Authorization (RBAC / object-level) is separate: `core/auth/authorization.py`
+  (`AuthorizationBackend` / `DefaultAuthorizationBackend` / `OwnershipAuthorizationBackend`),
+  wired via the `authorization` key. That key accepts `"module:attr"` (custom), a built-in
+  name (`"ownership"` = owner-only thread access; `"allow_all"`/`"default"`/`"none"`), or
+  `null`. When unset it defaults **by mode**: `ownership` in production (secure by default),
+  `allow_all` in development. Selection lives in `loader._resolve_authorization_backend`;
+  `RequirePermission` passes `resource_id` (thread_id from path or body) to `authorize`.
+- **Ownership is object-level and enforced on every thread-touching step** (invoke/stream/
+  stop/fix + all checkpointer read/write/delete): a thread is accessible only to its owner;
+  a foreign `invoke`/`stream` is rejected up front (403), never reaching the graph.
+- **Scalable, not a DB call per request.** Ownership is immutable, so it is cached by
+  `core/auth/ownership_resolver.py::ThreadOwnershipResolver`: in-process LRU (L1) + optional
+  shared Redis (L2, reuses `config.redis`/`settings.REDIS_URL`). The backing lookup is
+  `BaseCheckpointer.aget_thread_owner` (implemented in pg/in-memory/sqlite; base raises
+  `NotImplementedError`). Cache is evicted on thread delete (`CheckpointerService.delete_thread`);
+  the L2 client is closed in the lifespan shutdown.
+- **Secure by construction:** `core/auth/route_guard.py::assert_all_routes_protected` runs at
+  boot (`main.py` after `init_routes`) and refuses to start if any non-public route lacks a
+  `RequirePermission` guard (`/ping` is the only public path).
 
 ## HTTP + WebSocket surface (all under `/v1` except ping)
 
@@ -134,7 +151,7 @@ ruff check . && ruff format .
 
 - **Version is now single-sourced.** `CLI_VERSION` (and `agentflow_cli.__version__`, which aliases
   it) resolve from installed distribution metadata, falling back to `pyproject.toml`. `agentflow
-  version` reports `0.3.2.9` for both the CLI and package lines. (The old hardcoded `1.0.0` drift is
+  version` reports `0.4.0` for both the CLI and package lines. (The old hardcoded `1.0.0` drift is
   resolved.)
 - **README shows `agentflow init --prod`** — that flag does not exist. `init` is interactive and
   only accepts `--path` / `--force`.

agentflow.json

@@ -1,6 +1,7 @@
 {
   "agent": "graph.react:app",
   "thread_name_generator": "graph.thread_name_generator:MyNameGenerator",
+  "store": "graph.dummy_store:store",
   "env": ".env",
   "auth": null,
   "rate_limit": {

agentflow_cli/cli/commands/build.py

@@ -12,6 +12,8 @@
 from agentflow_cli.cli.templates.defaults import (
     generate_docker_compose_content,
     generate_dockerfile_content,
+    generate_dockerignore_content,
+    generate_k8s_manifest_content,
 )
 
 
@@ -25,6 +27,7 @@ def execute(
         python_version: str = DEFAULT_PYTHON_VERSION,
         port: int = DEFAULT_PORT,
         docker_compose: bool = False,
+        k8s: bool = False,
         service_name: str = DEFAULT_SERVICE_NAME,
         **kwargs: Any,
     ) -> int:
@@ -36,6 +39,7 @@ def execute(
             python_version: Python version to use
             port: Port to expose
             docker_compose: Generate docker-compose.yml
+            k8s: Generate a Kubernetes manifest (Deployment + Service)
             service_name: Service name for docker-compose
             **kwargs: Additional arguments
 
@@ -81,6 +85,18 @@ def execute(
             self._write_dockerfile(output_path, dockerfile_content)
             self.output.success(f"Successfully generated Dockerfile at {output_path}")
 
+            # Write .dockerignore in the same directory
+            dockerignore_path = output_path.parent / ".dockerignore"
+            if not dockerignore_path.exists() or force:
+                try:
+                    dockerignore_content = generate_dockerignore_content()
+                    dockerignore_path.write_text(dockerignore_content, encoding="utf-8")
+                    self.output.success(
+                        f"Successfully generated .dockerignore at {dockerignore_path}"
+                    )
+                except Exception as e:
+                    self.output.warning(f"Could not generate .dockerignore: {e}")
+
             # Show requirements info
             if requirements_files:
                 self.output.info(f"Using requirements file: {requirements_files[0]}")
@@ -95,6 +111,12 @@ def execute(
                     force=force, service_name=validated_service_name, port=validated_port
                 )
 
+            # Generate a Kubernetes manifest if requested
+            if k8s:
+                self._write_k8s_manifest(
+                    force=force, service_name=validated_service_name, port=validated_port
+                )
+
             # Show next steps
             self._show_next_steps(docker_compose)
 
@@ -155,6 +177,35 @@ def _write_dockerfile(self, output_path: Path, content: str) -> None:
                 f"Failed to write Dockerfile: {e}", file_path=str(output_path)
             ) from e
 
+    def _write_k8s_manifest(self, force: bool, service_name: str, port: int) -> None:
+        """Write k8s.yaml (Deployment + Service).
+
+        No manifest was generated before, so users hand-rolled one -- usually with
+        the default 30s termination grace period, which SIGKILLs an agent run
+        mid-LLM-call on every rolling deploy. The generated one sets a grace period
+        that matches how long a run can actually take, plus a preStop hook.
+
+        Raises:
+            FileOperationError: If writing fails
+        """
+        manifest_path = Path("k8s.yaml")
+
+        if manifest_path.exists() and not force:
+            raise FileOperationError(
+                f"k8s.yaml already exists at {manifest_path}. Use --force to overwrite.",
+                file_path=str(manifest_path),
+            )
+
+        content = generate_k8s_manifest_content(service_name, port)
+
+        try:
+            manifest_path.write_text(content, encoding="utf-8")
+            self.output.success(f"Generated k8s.yaml at {manifest_path}")
+        except OSError as e:
+            raise FileOperationError(
+                f"Failed to write k8s.yaml: {e}", file_path=str(manifest_path)
+            ) from e
+
     def _write_docker_compose(self, force: bool, service_name: str, port: int) -> None:
         """Write docker-compose.yml file.
 

agentflow_cli/cli/commands/init.py

@@ -72,6 +72,11 @@ def execute(self, path: str = ".", force: bool = False, **kwargs: Any) -> int:
             self._print_summary(context)
 
             base_path = Path(path)
+            # Capture this BEFORE any writes: a config file present now is the
+            # developer's own (a prior init or a hand edit), and must not be
+            # clobbered unless they passed --force.
+            config_path = base_path / "agentflow.json"
+            config_pre_existed = config_path.exists()
             base_path.mkdir(parents=True, exist_ok=True)
 
             is_prod = context["setup_type"] == "production"
@@ -82,10 +87,15 @@ def execute(self, path: str = ".", force: bool = False, **kwargs: Any) -> int:
                 template_dir, base_path, context, force=force, is_prod=is_prod
             )
 
-            # Regenerate agentflow.json from the built config (overrides template copy)
-            config_path = base_path / "agentflow.json"
+            # Regenerate agentflow.json from the built config. Force is safe only to
+            # override the copy this run just made; honor --force for a file that was
+            # already there.
             config = self._build_config(context, is_prod)
-            self._write_file(config_path, json.dumps(config, indent=2) + "\n", force=True)
+            self._write_file(
+                config_path,
+                json.dumps(config, indent=2) + "\n",
+                force=force or not config_pre_existed,
+            )
             if config_path not in created:
                 self._print_file_line(config_path, base_path)
 
@@ -309,6 +319,12 @@ def _build_config(self, context: dict, is_prod: bool) -> dict:
         elif auth == "custom":
             config["auth"] = {"method": "custom", "path": "auth.agent_auth:AgentAuth"}
 
+        if auth in ("jwt", "custom"):
+            # Secure by default in production: a thread is accessible only to its owner.
+            # Change to "allow_all" to let any authenticated user access any thread, or
+            # point at a custom AuthorizationBackend ("module:attr").
+            config["authorization"] = "ownership"
+
         config["thread_name_generator"] = "graph.thread_name_generator:MyNameGenerator"
         config["injectq"] = "graph.agent:container"
 

agentflow_cli/cli/main.py

@@ -266,10 +266,18 @@ def build(
         "--docker-compose/--no-docker-compose",
         help="Also generate docker-compose.yml and omit CMD in Dockerfile",
     ),
+    k8s: bool = typer.Option(
+        False,
+        "--k8s/--no-k8s",
+        help=(
+            "Also generate k8s.yaml (Deployment + Service) with a termination grace "
+            "period long enough that a rolling deploy does not kill in-flight agent runs"
+        ),
+    ),
     service_name: str = typer.Option(
         "agentflow-cli",
         "--service-name",
-        help="Service name to use in docker-compose.yml (if generated)",
+        help="Service name to use in docker-compose.yml / k8s.yaml (if generated)",
     ),
     verbose: bool = typer.Option(
         False,
@@ -296,6 +304,7 @@ def build(
             python_version=python_version,
             port=port,
             docker_compose=docker_compose,
+            k8s=k8s,
             service_name=service_name,
         )
         sys.exit(exit_code)