diff --git a/.claude/plans/2026-05-08-step-11-plugin-runtime-and-reconcile.md b/.claude/plans/2026-05-08-step-11-plugin-runtime-and-reconcile.md new file mode 100644 index 0000000..968cae8 --- /dev/null +++ b/.claude/plans/2026-05-08-step-11-plugin-runtime-and-reconcile.md @@ -0,0 +1,90 @@ +# Step 11 — Plugin runtime + reconcile + reftest sentinel + +## Context + +Step 10 (PR #23) shipped the SDK + discovery half of the plugin surface — `Plugin` base class, manifest parsing, leaf-class loader, `GET /v1/plugins/types`. Operators can see what plugin types are available but can't yet assign one or get a reconcile to fire. + +This slice closes the loop. After this PR: + +- The operator assigns a discovered plugin via `POST /v1/plugins` and curfew-core instantiates it. +- A user lock toggle schedules a background reconcile on every plugin governing that user. +- A safety-net resync runs every `plugin_resync_seconds` and re-calls reconcile on every governing plugin for every governed user (catches missed events through restarts). +- The reftest plugin writes a sentinel file when locked / removes it when unlocked, so the **kernel acceptance plugin path** (PLAN.md §"Acceptance: kernel done" steps 11–16) passes end-to-end. + +PLAN.md §"Plugin lifecycle" + ADR-005 + PLUGINS.md §"Lifecycle" are the load-bearing references. + +## Scope + +**In:** + +- `core/curfew/plugin_runtime.py` — new. The in-memory instance manager. Holds `{(type, instance_id): _Live}` where `_Live` carries the instantiated `Plugin`, the resolved governed-users list, and the paused flag. Mutations (`assign`, `unassign`, `update`, `set_paused`) are guarded by an `asyncio.Lock`. `dispatch_for_user(username)` finds every governing instance and schedules `_run_one_reconcile` for each. `_run_one_reconcile` wraps the plugin's `reconcile()` in `asyncio.wait_for(..., timeout=plugin_reconcile_timeout_seconds)` and audits failures (timeout, exception, error result). `safety_net_resync()` walks every governed user and dispatches. +- `core/curfew/schemas.py` — `PluginAssignmentRead`, `PluginAssignmentCreate` (`type`, optional `instance_id` defaulting to `"default"`, `config: dict`, `users: list[str]`), `PluginAssignmentUpdate` (all-optional: `config`, `users`, `paused`). +- `core/curfew/__init__.py` — re-export `PluginRuntime`, the new schemas. +- `api/curfew_api/routes/plugins.py` — extend with the four CRUD verbs: + - `GET /v1/plugins` — list assignments + paused/users (operator). + - `POST /v1/plugins` — assign. Validates the discovered type exists + config matches its Pydantic schema; inserts the row; instantiates via the runtime. 422 on bad config, 404 on unknown type, 409 on duplicate `(type, instance_id)`. Admin only. + - `PATCH /v1/plugins/{type}/{instance_id}` — partial update. If `config` changes, re-instantiate (PLAN.md option 1). If `users` changes, update the governed-users cache. If `paused` changes, flip the flag. Admin only. + - `DELETE /v1/plugins/{type}/{instance_id}` — unassign. Drops from runtime. Admin only. +- `api/curfew_api/routes/locks.py` — after the lock/unlock commit, schedule `runtime.dispatch_for_user(username)` via FastAPI `BackgroundTasks`. The originating HTTP response returns immediately; reconciles run after. +- `api/curfew_api/app.py` — switch to FastAPI lifespan context manager. At startup: build the runtime, load every existing `plugin_assignments` row into memory, start an `asyncio` task that loops every `plugin_resync_seconds` (read fresh on each iteration so settings PATCHes take effect) and calls `runtime.safety_net_resync()`. At shutdown: cancel the task cleanly. +- `plugins/reftest-plugin/plugin.py` — replace the no-op reconcile with sentinel logic. `Config` gains `sentinel_path: str`. Locked → write the user's name to that path; unlocked → remove the file (idempotent if already absent). Used by the e2e test to assert reconcile actually ran. +- `core/tests/test_plugin_runtime.py` — new. Assign/unassign mutate the in-memory map. PATCH on `config` re-instantiates (assert `__init__` ran twice). PATCH on `users` rewires governance. PATCH on `paused` skips dispatch. `dispatch_for_user` picks instances whose `users` list contains the username or `["*"]`. Paused instances skipped. Timeout cancels the call and audits a `plugin.reconcile_failed`. An exception in `reconcile` is caught + audited but doesn't propagate. `safety_net_resync` calls reconcile for every governed user × every governing plugin. +- `api/tests/test_plugins_assignments.py` — new. CRUD endpoints: list (empty + populated), create (happy + 422 bad config + 404 unknown type + 409 duplicate), patch (single field, multi-field, paused toggle), delete (200 + 404). Audit rows. Auth: 401 unauth, 403 manager. +- `api/tests/test_plugins_reconcile.py` — new. e2e: configure `CURFEW_PLUGINS_DIRS` at the repo's `plugins/`; assign reftest_plugin with `sentinel_path=/sentinel`; lock kid1 → wait for reconcile → assert sentinel exists with `kid1` content; unlock → assert removed. Same flow but kill the lock-triggered reconcile (paused at the moment of lock); safety-net resync picks it up on its next tick. Failed reconcile (point sentinel_path at an unwritable dir) → next lock toggle still returns 200; audit shows `plugin.reconcile_failed`. + +**Out (next slice — beyond the kernel):** + +- `requirements.txt` auto-install (PLAN.md ADR-013 lists this as optional / future). +- Plugin SDK helper for HTTP retry/timeout (PLAN.md §"Plugin SDK"). The first plugin (`adguard`) drives that. +- `target_apps` on `Scope` (PLAN.md per-rule user config — deferred until first non-kernel rule lands). +- Concrete plugins (adguard, smart-plug, router-acl). Each is a feature on top of the kernel. + +## Notable design choices + +- **Re-instantiate on config PATCH (PLAN.md option 1).** Cleanest semantics for the operator: PATCH is "give me this plugin running with this config." Plugin authors can keep `__init__` cheap (or expensive — their call) and don't need to remember to re-read config inside `reconcile`. `users` and `paused` changes don't re-instantiate. +- **`asyncio.Lock` for runtime mutations.** Two operators submitting concurrent assigns on the same `(type, instance_id)` lose to the DB unique constraint anyway, but the lock keeps the in-memory map consistent during the brief window between insert + instantiate. Cheap at homelab scale. +- **Reconcile dispatch via `BackgroundTasks`, not bare `asyncio.create_task`.** FastAPI runs background tasks after the response is sent, with built-in error containment. `create_task` would work but loses the framework hook (and tests would have to await the task explicitly). +- **Per-call timeout via `asyncio.wait_for`.** Cancels a hung plugin's coroutine and treats it as a failure. Plugins can still hang the entire event loop if they do blocking IO without `await` — that's a class of bug the plugin author has to avoid (called out in PLUGINS.md). +- **Reconcile failures are audited, never raised.** PLAN.md ADR-003 + PLUGINS.md §"What can go wrong": originating request must already have returned successfully. Audit kind is `plugin.reconcile_failed`; payload carries `{type, instance_id, user, error}`. +- **Safety-net loop reads `plugin_resync_seconds` fresh each iteration.** A settings PATCH that lowers the resync interval takes effect on the next tick, no restart needed. Same pattern that the agent SDK will use for `agent_tick_seconds` in the agent-surface PR. +- **Loading at startup.** All existing `plugin_assignments` rows get instantiated when the app starts. A row whose plugin type is no longer discoverable (operator removed the folder) is logged and skipped — the row stays in the DB so the operator can decide whether to delete it or restore the folder. This is the "graceful degradation" path: a missing plugin folder doesn't break startup. +- **Username vs user_id in `users` JSON column.** The column already stores usernames (per the existing model + PLAN.md). When a user is renamed, existing assignments still target the old username until PATCHed — same trade-off as audit log keeping target_id strings. Acceptable for V1; if it becomes a real problem we add a normalisation pass. +- **`POST /v1/plugins` instance_id default.** Body field is optional; `None` resolves to `"default"` server-side. Matches the CLI shape (`curfew plugin assign adguard` for the single-instance case). + +## Files to touch + +| File | Change | +|---|---| +| `core/curfew/plugin_runtime.py` | New. `PluginRuntime` class. ~250 lines. | +| `core/curfew/schemas.py` | Add `PluginAssignmentRead`, `PluginAssignmentCreate`, `PluginAssignmentUpdate`. | +| `core/curfew/__init__.py` | Export the new schemas + `PluginRuntime`. | +| `api/curfew_api/routes/plugins.py` | Add the four CRUD handlers. | +| `api/curfew_api/routes/locks.py` | Inject `BackgroundTasks` + runtime; schedule dispatch after commit. | +| `api/curfew_api/app.py` | Lifespan context manager — load assignments + start resync task. | +| `plugins/reftest-plugin/plugin.py` | Sentinel logic. | +| `core/tests/test_plugin_runtime.py` | New. | +| `api/tests/test_plugins_assignments.py` | New. | +| `api/tests/test_plugins_reconcile.py` | New (e2e + safety-net + failure path). | +| `api/tests/conftest.py` | Possibly: `client_with_plugins` fixture if needed for the e2e tests. | + +## Reused utilities (no new infrastructure) + +- `core/curfew/plugin_loader.py:PluginRegistry` — runtime asks the registry for the type → resolved class + config model. +- `core/curfew/audit.py:record_audit` — for assignment writes + reconcile-failed audits. +- `core/curfew/models.py:PluginAssignment, Settings` — DB row + runtime tunables. +- `api/curfew_api/auth.py:Operator, RequireAdmin` — auth deps. + +## Verification + +- `just check` clean. +- New test count: ~12 in `test_plugin_runtime.py`, ~12 in `test_plugins_assignments.py`, ~5 in `test_plugins_reconcile.py`. Existing 207 tests still green. +- Live smoke (post-merge of step 10): + - `just serve`, then `curl -X POST -H 'Authorization: Bearer ...' -H 'Content-Type: application/json' -d '{"type":"reftest_plugin","config":{"sentinel_path":"/tmp/sent"},"users":["*"]}' :8000/v1/plugins`. + - `just user-create kid1 kid12345 member` then lock kid1; `/tmp/sent` appears with content `kid1`. + - Unlock; `/tmp/sent` disappears. + - Pause the plugin (`PATCH ... '{"paused": true}'`); lock again; sentinel does *not* appear. + +## Branch + PR + +- Branch: `plugin-runtime-reconcile` off `main` (post-#23 merge). +- One commit, one PR. Estimated diff: ~900 lines across ~10 files (most lines in tests). diff --git a/api/curfew_api/app.py b/api/curfew_api/app.py index 4cf9ed6..b238d97 100644 --- a/api/curfew_api/app.py +++ b/api/curfew_api/app.py @@ -1,31 +1,58 @@ """FastAPI app factory for curfew-api. Builds the ``FastAPI`` instance, registers the kernel's lock-status rules, -mounts CORS, and attaches the routers. Auth is enforced via per-route -``Depends(require_operator)``, not as ASGI middleware: this keeps the OpenAPI +mounts CORS, attaches the routers, and runs the lifespan that owns the +plugin runtime + safety-net resync loop. Auth is enforced via per-route +``Depends(require_actor)``, not as ASGI middleware: this keeps the OpenAPI schema honest (the security requirement is documented on each protected -route) and lets ``/v1/health`` and ``/v1/openapi.json`` opt out by simply not -including the dependency. +route) and lets ``/v1/health`` and ``/v1/openapi.json`` opt out by simply +not including the dependency. + +Lifespan ownership (per ADR-005 + PLAN.md §"Plugin lifecycle"): + +- Build the ``PluginRuntime``, hydrate it from ``plugin_assignments`` + rows. A row whose plugin type isn't discovered (folder removed) is + logged and skipped — the row stays so the operator can decide. +- Start the safety-net resync task: every ``plugin_resync_seconds`` (read + fresh each iteration so settings PATCHes take effect), call + ``runtime.safety_net_resync()``. +- On shutdown: cancel the task and await it cleanly. """ from __future__ import annotations +import asyncio +import logging import os +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager from pathlib import Path from typing import Any from curfew.config import get_config -from curfew.plugin_loader import discover_plugins +from curfew.db import get_engine +from curfew.models import PluginAssignment +from curfew.models import Settings as SettingsRow +from curfew.plugin_loader import PluginRegistry, discover_plugins +from curfew.plugin_runtime import ( + PluginInstantiationError, + PluginRuntime, + UnknownPluginTypeError, +) from curfew.rules import register_kernel_rules from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles +from pydantic import ValidationError +from sqlmodel import Session, select from starlette.exceptions import HTTPException as StarletteHTTPException from starlette.responses import Response from curfew_api.routes import apps, auth, devices, health, locks, plugins, status, system, users from curfew_api.routes.settings import router as settings_router +logger = logging.getLogger(__name__) + class SPAStaticFiles(StaticFiles): """``StaticFiles`` that falls back to ``index.html`` on missing paths. @@ -46,9 +73,95 @@ async def get_response(self, path: str, scope: Any) -> Response: raise +async def _hydrate_runtime(runtime: PluginRuntime) -> None: + """Instantiate every existing assignment row + register it. + + Errors are logged per-row and skipped — the row stays in the DB so + the operator can fix the plugin folder or unassign explicitly. PLAN.md + "graceful degradation": one missing plugin folder doesn't stop + startup. + """ + with Session(get_engine()) as session: + rows = list(session.exec(select(PluginAssignment)).all()) + + for row in rows: + try: + instance = runtime.build_instance(row.type, dict(row.config)) + except (UnknownPluginTypeError, ValidationError, PluginInstantiationError) as exc: + logger.warning( + "skipping assignment %s/%s on startup: %s", + row.type, + row.instance_id, + exc, + ) + continue + try: + await runtime.register( + type_name=row.type, + instance_id=row.instance_id, + instance=instance, + config=dict(row.config), + users=list(row.users), + paused=row.paused, + ) + except Exception: + logger.exception( + "failed to register %s/%s during hydrate", + row.type, + row.instance_id, + ) + + +async def _safety_net_loop(runtime: PluginRuntime) -> None: + """Periodic re-dispatch of every governed user. + + Reads ``plugin_resync_seconds`` from the singleton settings row each + iteration so a settings PATCH that lowers the interval takes effect + on the next tick. Sleeping is interruptible by cancellation (FastAPI + cancels this task on shutdown). + """ + while True: + with Session(get_engine()) as session: + interval = session.exec(select(SettingsRow)).one().plugin_resync_seconds + try: + await asyncio.sleep(interval) + except asyncio.CancelledError: + return + try: + await runtime.safety_net_resync() + except asyncio.CancelledError: + return + except Exception: + logger.exception("safety-net resync raised; loop continues") + + +@asynccontextmanager +async def lifespan(app: FastAPI) -> AsyncIterator[None]: + """Owns the plugin runtime + safety-net loop for the app's lifetime.""" + config = get_config() + registry: PluginRegistry = discover_plugins(config.plugins_dirs) + app.state.plugin_registry = registry + + runtime = PluginRuntime(registry=registry, engine=get_engine()) + app.state.plugin_runtime = runtime + await _hydrate_runtime(runtime) + + resync_task = asyncio.create_task(_safety_net_loop(runtime), name="plugin-safety-net") + + try: + yield + finally: + resync_task.cancel() + try: + await resync_task + except (asyncio.CancelledError, Exception): + # Cancellation is the expected shutdown path; any other + # exception was already logged inside the loop. + pass + + def create_app() -> FastAPI: """Build and return a configured FastAPI app.""" - config = get_config() register_kernel_rules() app = FastAPI( @@ -56,15 +169,10 @@ def create_app() -> FastAPI: description="curfew API service", version="0.0.0", openapi_url="/v1/openapi.json", + lifespan=lifespan, ) - # Discover plugins once at startup. The registry is read-only at - # runtime; adding a new plugin folder requires a restart (ADR-013). - # ``discover_plugins`` is robust to missing dirs and per-plugin - # errors, so a misconfigured ``CURFEW_PLUGINS_DIRS`` doesn't stop - # startup — bad plugins surface at ``GET /v1/plugins/types``. - app.state.plugin_registry = discover_plugins(config.plugins_dirs) - + config = get_config() if config.cors_origins: app.add_middleware( CORSMiddleware, diff --git a/api/curfew_api/routes/locks.py b/api/curfew_api/routes/locks.py index 55994b3..ac66687 100644 --- a/api/curfew_api/routes/locks.py +++ b/api/curfew_api/routes/locks.py @@ -3,6 +3,13 @@ Both endpoints upsert ``user_locks`` and return the updated user-scope ``LockStatus`` (so the operator sees the immediate effect of the toggle, not an empty ``204``). Both are audited as ``user.lock`` and ``user.unlock``. + +After the DB commit, both endpoints schedule +``PluginRuntime.dispatch_for_user(...)`` via ``BackgroundTasks`` — the +HTTP response returns immediately, plugins reconcile in the background +(per ADR-005 + PLAN.md §"Plugin lifecycle"). Plugin failures are caught + +audited inside the runtime; they never propagate back to the operator's +request. """ from __future__ import annotations @@ -14,9 +21,10 @@ from curfew.auth import Actor from curfew.db import get_session from curfew.models import AuditTargetKind, User, UserLock +from curfew.plugin_runtime import PluginRuntime from curfew.rules import user_scope from curfew.schemas import LockStatus -from fastapi import APIRouter, Depends, HTTPException, status +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request, status from sqlmodel import Session, select from curfew_api.auth import RequireManager @@ -24,7 +32,15 @@ router = APIRouter(prefix="/v1/users", tags=["locks"]) -def _set_manual_lock(session: Session, *, username: str, locked: bool, actor: Actor) -> LockStatus: +def _set_manual_lock( + session: Session, + *, + username: str, + locked: bool, + actor: Actor, + background_tasks: BackgroundTasks, + runtime: PluginRuntime, +) -> LockStatus: user = session.exec(select(User).where(User.username == username)).first() if user is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="user not found") @@ -55,6 +71,11 @@ def _set_manual_lock(session: Session, *, username: str, locked: bool, actor: Ac ) session.commit() + # Fire reconcile in the background so the response returns now. The + # task runs after the response is sent; if no plugin governs this + # user the dispatch is a fast no-op. + background_tasks.add_task(runtime.dispatch_for_user, user.username) + return user_scope.evaluate(session, user.username) @@ -62,15 +83,33 @@ def _set_manual_lock(session: Session, *, username: str, locked: bool, actor: Ac def lock_user( user: str, actor: RequireManager, + background_tasks: BackgroundTasks, + request: Request, session: Annotated[Session, Depends(get_session)], ) -> LockStatus: - return _set_manual_lock(session, username=user, locked=True, actor=actor) + return _set_manual_lock( + session, + username=user, + locked=True, + actor=actor, + background_tasks=background_tasks, + runtime=request.app.state.plugin_runtime, + ) @router.post("/{user}/unlock", response_model=LockStatus) def unlock_user( user: str, actor: RequireManager, + background_tasks: BackgroundTasks, + request: Request, session: Annotated[Session, Depends(get_session)], ) -> LockStatus: - return _set_manual_lock(session, username=user, locked=False, actor=actor) + return _set_manual_lock( + session, + username=user, + locked=False, + actor=actor, + background_tasks=background_tasks, + runtime=request.app.state.plugin_runtime, + ) diff --git a/api/curfew_api/routes/plugins.py b/api/curfew_api/routes/plugins.py index 4e22d61..a0b1ee4 100644 --- a/api/curfew_api/routes/plugins.py +++ b/api/curfew_api/routes/plugins.py @@ -1,26 +1,56 @@ -"""Plugin discovery + (step 11) assignment endpoints. +"""Plugin discovery + assignment CRUD. -This slice (step 10) ships only ``GET /v1/plugins/types`` — what plugin -types curfew-core discovered at startup. The list reflects the in-memory -registry built by ``discover_plugins`` (PLUGINS.md §"What curfew-core does -at startup"), so failed plugins show up here with their error. +The router serves both halves of the plugin surface: -Step 11 will add ``GET/POST/PATCH/DELETE /v1/plugins`` (assignment CRUD) -and the in-memory instance manager that talks to the same registry. +- ``GET /v1/plugins/types`` (step 10) — what curfew-core discovered at + startup. Operators see this to know which plugin types they can assign. +- ``GET/POST/PATCH/DELETE /v1/plugins`` (step 11) — operator's assignments. + POST/PATCH/DELETE are admin-gated and audited; reads are + any-authenticated. Each write goes through the in-memory ``PluginRuntime`` + so the live instance map stays in lock-step with the + ``plugin_assignments`` table. + +Per ADR-005 + PLUGINS.md §"Lifecycle": assigning a plugin instantiates it; +unassigning drops it; PATCHing config re-instantiates it. Operators don't +have to restart curfew-core for any of this. Only adding *new* plugin +folders requires a restart (discovery runs once at startup, ADR-013). """ from __future__ import annotations +from typing import Annotated + +from curfew.audit import record_audit +from curfew.db import get_session +from curfew.models import AuditTargetKind, PluginAssignment from curfew.plugin_loader import PluginRegistry, PluginType -from curfew.schemas import PluginTypeRead -from fastapi import APIRouter, Request +from curfew.plugin_runtime import ( + AlreadyAssignedError, + NotAssignedError, + PluginInstantiationError, + PluginRuntime, + UnknownPluginTypeError, +) +from curfew.schemas import ( + PluginAssignmentCreate, + PluginAssignmentRead, + PluginAssignmentUpdate, + PluginTypeRead, +) +from fastapi import APIRouter, Depends, HTTPException, Request, status +from pydantic import ValidationError +from sqlalchemy.exc import IntegrityError +from sqlmodel import Session, select -from curfew_api.auth import Operator +from curfew_api.auth import Operator, RequireAdmin router = APIRouter(prefix="/v1/plugins", tags=["plugins"]) -def _to_read(ptype: PluginType) -> PluginTypeRead: +# --- /v1/plugins/types ------------------------------------------------------- + + +def _type_to_read(ptype: PluginType) -> PluginTypeRead: m = ptype.manifest return PluginTypeRead( type=ptype.type, @@ -42,4 +72,217 @@ def list_plugin_types(request: Request, actor: Operator) -> list[PluginTypeRead] endpoint just reports what was discovered. """ registry: PluginRegistry = request.app.state.plugin_registry - return [_to_read(p) for p in registry.all()] + return [_type_to_read(p) for p in registry.all()] + + +# --- /v1/plugins (assignment CRUD) ------------------------------------------ + + +def _row_to_read(row: PluginAssignment) -> PluginAssignmentRead: + return PluginAssignmentRead( + type=row.type, + instance_id=row.instance_id, + config=dict(row.config), + users=list(row.users), + paused=row.paused, + ) + + +def _runtime(request: Request) -> PluginRuntime: + return request.app.state.plugin_runtime # type: ignore[no-any-return] + + +def _get_assignment_or_404(session: Session, type_name: str, instance_id: str) -> PluginAssignment: + row = session.exec( + select(PluginAssignment).where( + PluginAssignment.type == type_name, + PluginAssignment.instance_id == instance_id, + ) + ).first() + if row is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"plugin {type_name}/{instance_id} not assigned", + ) + return row + + +@router.get("", response_model=list[PluginAssignmentRead]) +def list_assignments( + actor: Operator, + session: Annotated[Session, Depends(get_session)], +) -> list[PluginAssignmentRead]: + rows = session.exec( + select(PluginAssignment).order_by(PluginAssignment.type, PluginAssignment.instance_id) + ).all() + return [_row_to_read(r) for r in rows] + + +@router.post("", response_model=PluginAssignmentRead, status_code=status.HTTP_201_CREATED) +async def create_assignment( + payload: PluginAssignmentCreate, + actor: RequireAdmin, + request: Request, + session: Annotated[Session, Depends(get_session)], +) -> PluginAssignmentRead: + """Assign a plugin to a set of users. + + Validates the type exists (404), the config matches the type's Pydantic + schema (422), and the ``(type, instance_id)`` is unique (409). Builds + the live instance *before* writing the DB row so a bad config / broken + plugin returns 422 / 500 without leaving an orphan row behind. + """ + runtime = _runtime(request) + try: + instance = runtime.build_instance(payload.type, dict(payload.config)) + except UnknownPluginTypeError as exc: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc + except ValidationError as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=exc.errors(), + ) from exc + except PluginInstantiationError as exc: + # The plugin's __init__ raised. Not the operator's fault per se, + # but actionable. 500 with detail surfaces it. + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc) + ) from exc + + row = PluginAssignment( + type=payload.type, + instance_id=payload.instance_id, + config=dict(payload.config), + users=list(payload.users), + paused=False, + ) + session.add(row) + try: + session.flush() + except IntegrityError: + session.rollback() + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f"plugin {payload.type}/{payload.instance_id} already assigned", + ) from None + + record_audit( + session, + actor=actor, + action="plugin.assign", + target_kind=AuditTargetKind.PLUGIN, + target_id=f"{row.type}/{row.instance_id}", + payload={"users": list(row.users)}, + ) + session.commit() + + try: + await runtime.register( + type_name=row.type, + instance_id=row.instance_id, + instance=instance, + config=dict(row.config), + users=list(row.users), + paused=row.paused, + ) + except AlreadyAssignedError: + # Race: a concurrent request beat us to the runtime register call. + # The DB row we just inserted survived, so the existing instance + # is governing this slot — return its read shape. + pass + + return _row_to_read(row) + + +@router.patch("/{type_name}/{instance_id}", response_model=PluginAssignmentRead) +async def update_assignment( + type_name: str, + instance_id: str, + payload: PluginAssignmentUpdate, + actor: RequireAdmin, + request: Request, + session: Annotated[Session, Depends(get_session)], +) -> PluginAssignmentRead: + """Partial update. + + If ``config`` changes, the runtime re-instantiates the plugin + (PLAN.md option 1). ``users`` and ``paused`` updates don't touch + ``__init__``. + """ + runtime = _runtime(request) + row = _get_assignment_or_404(session, type_name, instance_id) + update_data = payload.model_dump(exclude_unset=True) + if not update_data: + return _row_to_read(row) + + # Apply runtime mutation first so a bad config (failed re-instantiation) + # surfaces before we touch the DB. The runtime hasn't committed yet — + # if the runtime call raises, no DB write happens. + try: + await runtime.update( + type_name=type_name, + instance_id=instance_id, + config=update_data.get("config"), + users=update_data.get("users"), + paused=update_data.get("paused"), + ) + except NotAssignedError: + # Runtime out of sync with DB — should never happen for a row that + # passed _get_assignment_or_404. Treat as 404 rather than corrupt + # the DB. + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"plugin {type_name}/{instance_id} not loaded in runtime", + ) from None + except UnknownPluginTypeError as exc: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc + except ValidationError as exc: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=exc.errors(), + ) from exc + except PluginInstantiationError as exc: + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(exc) + ) from exc + + if "config" in update_data: + row.config = dict(update_data["config"]) + if "users" in update_data: + row.users = list(update_data["users"]) + if "paused" in update_data: + row.paused = bool(update_data["paused"]) + session.add(row) + record_audit( + session, + actor=actor, + action="plugin.update", + target_kind=AuditTargetKind.PLUGIN, + target_id=f"{row.type}/{row.instance_id}", + payload=update_data, + ) + session.commit() + session.refresh(row) + return _row_to_read(row) + + +@router.delete("/{type_name}/{instance_id}", status_code=status.HTTP_204_NO_CONTENT) +async def delete_assignment( + type_name: str, + instance_id: str, + actor: RequireAdmin, + request: Request, + session: Annotated[Session, Depends(get_session)], +) -> None: + runtime = _runtime(request) + row = _get_assignment_or_404(session, type_name, instance_id) + session.delete(row) + record_audit( + session, + actor=actor, + action="plugin.unassign", + target_kind=AuditTargetKind.PLUGIN, + target_id=f"{row.type}/{row.instance_id}", + ) + session.commit() + await runtime.unassign(type_name=type_name, instance_id=instance_id) diff --git a/api/tests/conftest.py b/api/tests/conftest.py index e524da0..7dccfa8 100644 --- a/api/tests/conftest.py +++ b/api/tests/conftest.py @@ -48,8 +48,61 @@ def configured_db(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[P @pytest.fixture -def client(configured_db: Path) -> TestClient: - return TestClient(create_app()) +def client(configured_db: Path) -> Iterator[TestClient]: + """TestClient as a context manager so FastAPI lifespan fires. + + Without the ``with`` block, ``startup``/``shutdown`` don't run and + ``app.state.plugin_runtime`` is never built — every route that + schedules reconciles would AttributeError. Yielding from the + fixture makes tests use the same lifecycle production does. + """ + with TestClient(create_app()) as c: + yield c + + +@pytest.fixture +def configured_db_with_repo_plugins( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> Iterator[Path]: + """``configured_db`` variant that also points the loader at the repo plugins. + + Tests that exercise plugin assignment / reconcile dispatch need a real + discovered plugin (the shipped ``reftest_plugin``). Setting + ``CURFEW_PLUGINS_DIRS`` to the repo's ``plugins/`` directory makes + discovery find it; the rest of the setup mirrors ``configured_db``. + """ + repo_root = Path(__file__).resolve().parents[2] + monkeypatch.setenv("CURFEW_PLUGINS_DIRS", str(repo_root / "plugins")) + + db = tmp_path / "test.sqlite" + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("CURFEW_ROOT_TOKEN", "test-token") + monkeypatch.setenv("CURFEW_DB_PATH", str(db)) + reset_config_cache() + reset_engine_cache() + + api_dir = Path(__file__).resolve().parents[1] + cfg = Config(str(api_dir / "alembic.ini")) + cfg.set_main_option("script_location", str(api_dir / "migrations")) + cfg.set_main_option("sqlalchemy.url", f"sqlite:///{db}") + command.upgrade(cfg, "head") + + yield db + + user_scope.reset() + reset_config_cache() + reset_engine_cache() + + +@pytest.fixture +def client_with_repo_plugins(configured_db_with_repo_plugins: Path) -> Iterator[TestClient]: + """``client`` variant with ``reftest_plugin`` discovered. + + Used by plugin assignment / reconcile tests so they don't have to + synthesise a plugin folder per test. + """ + with TestClient(create_app()) as c: + yield c @pytest.fixture diff --git a/api/tests/test_app.py b/api/tests/test_app.py index ea34f4e..1ef8f87 100644 --- a/api/tests/test_app.py +++ b/api/tests/test_app.py @@ -44,12 +44,13 @@ def configured_db(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[P @pytest.fixture -def client(configured_db: Path) -> TestClient: - return TestClient(create_app()) +def client(configured_db: Path) -> Iterator[TestClient]: + with TestClient(create_app()) as c: + yield c @pytest.fixture -def client_with_protected_stub(configured_db: Path) -> TestClient: +def client_with_protected_stub(configured_db: Path) -> Iterator[TestClient]: """An app with an extra ``/v1/_stub`` route gated by ``Operator``.""" app = create_app() stub = APIRouter() @@ -59,7 +60,8 @@ def _stub(actor: Operator) -> dict[str, str]: return {"actor": actor.audit_str} app.include_router(stub) - return TestClient(app) + with TestClient(app) as c: + yield c def test_health_unauthenticated(client: TestClient) -> None: diff --git a/api/tests/test_plugins_assignments.py b/api/tests/test_plugins_assignments.py new file mode 100644 index 0000000..8ed181e --- /dev/null +++ b/api/tests/test_plugins_assignments.py @@ -0,0 +1,279 @@ +"""Tests for ``GET/POST/PATCH/DELETE /v1/plugins`` (assignment CRUD). + +Uses the ``client_with_repo_plugins`` fixture so the shipped +``reftest_plugin`` is discovered and we can assign it without +synthesising a plugin folder per test. +""" + +from __future__ import annotations + +from pathlib import Path + +from curfew.models import AuditLog, AuditTargetKind +from fastapi.testclient import TestClient +from sqlmodel import Session, create_engine, select + + +def _engine(db: Path): # type: ignore[no-untyped-def] + return create_engine(f"sqlite:///{db}") + + +def _ok_config(tmp_path: Path) -> dict[str, str]: + """A reftest config that works — points at a writable sentinel path.""" + return {"sentinel_path": str(tmp_path / "sentinel")} + + +# --- list / GET -------------------------------------------------------------- + + +def test_list_unauthenticated(client_with_repo_plugins: TestClient) -> None: + assert client_with_repo_plugins.get("/v1/plugins").status_code == 401 + + +def test_list_empty(client_with_repo_plugins: TestClient, auth: dict[str, str]) -> None: + r = client_with_repo_plugins.get("/v1/plugins", headers=auth) + assert r.status_code == 200 + assert r.json() == [] + + +def test_list_after_create( + client_with_repo_plugins: TestClient, auth: dict[str, str], tmp_path: Path +) -> None: + client_with_repo_plugins.post( + "/v1/plugins", + json={"type": "reftest_plugin", "config": _ok_config(tmp_path), "users": ["*"]}, + headers=auth, + ) + r = client_with_repo_plugins.get("/v1/plugins", headers=auth) + assert r.status_code == 200 + [item] = r.json() + assert item["type"] == "reftest_plugin" + assert item["instance_id"] == "default" + assert item["users"] == ["*"] + assert item["paused"] is False + + +# --- POST -------------------------------------------------------------------- + + +def test_create_returns_201( + client_with_repo_plugins: TestClient, auth: dict[str, str], tmp_path: Path +) -> None: + r = client_with_repo_plugins.post( + "/v1/plugins", + json={"type": "reftest_plugin", "config": _ok_config(tmp_path), "users": ["kid1"]}, + headers=auth, + ) + assert r.status_code == 201 + body = r.json() + assert body["type"] == "reftest_plugin" + assert body["instance_id"] == "default" + assert body["users"] == ["kid1"] + + +def test_create_unknown_type_404( + client_with_repo_plugins: TestClient, auth: dict[str, str] +) -> None: + r = client_with_repo_plugins.post( + "/v1/plugins", + json={"type": "no_such_plugin", "config": {}, "users": []}, + headers=auth, + ) + assert r.status_code == 404 + assert "no_such_plugin" in r.json()["detail"] + + +def test_create_invalid_config_422( + client_with_repo_plugins: TestClient, auth: dict[str, str] +) -> None: + """reftest_plugin requires sentinel_path; missing it is a 422.""" + r = client_with_repo_plugins.post( + "/v1/plugins", + json={"type": "reftest_plugin", "config": {}, "users": []}, + headers=auth, + ) + assert r.status_code == 422 + + +def test_create_duplicate_409( + client_with_repo_plugins: TestClient, auth: dict[str, str], tmp_path: Path +) -> None: + body = {"type": "reftest_plugin", "config": _ok_config(tmp_path), "users": []} + r = client_with_repo_plugins.post("/v1/plugins", json=body, headers=auth) + assert r.status_code == 201 + r = client_with_repo_plugins.post("/v1/plugins", json=body, headers=auth) + assert r.status_code == 409 + assert "already assigned" in r.json()["detail"] + + +def test_create_with_explicit_instance_id( + client_with_repo_plugins: TestClient, auth: dict[str, str], tmp_path: Path +) -> None: + """Two instances of the same type with different instance_ids coexist.""" + body1 = { + "type": "reftest_plugin", + "instance_id": "alpha", + "config": _ok_config(tmp_path), + "users": ["kid1"], + } + body2 = { + "type": "reftest_plugin", + "instance_id": "beta", + "config": {"sentinel_path": str(tmp_path / "sentinel-b")}, + "users": ["kid2"], + } + assert client_with_repo_plugins.post("/v1/plugins", json=body1, headers=auth).status_code == 201 + assert client_with_repo_plugins.post("/v1/plugins", json=body2, headers=auth).status_code == 201 + + r = client_with_repo_plugins.get("/v1/plugins", headers=auth) + assert {p["instance_id"] for p in r.json()} == {"alpha", "beta"} + + +def test_create_audited( + client_with_repo_plugins: TestClient, + auth: dict[str, str], + tmp_path: Path, + configured_db_with_repo_plugins: Path, +) -> None: + client_with_repo_plugins.post( + "/v1/plugins", + json={"type": "reftest_plugin", "config": _ok_config(tmp_path), "users": ["kid1"]}, + headers=auth, + ) + with Session(_engine(configured_db_with_repo_plugins)) as s: + rows = list(s.exec(select(AuditLog).where(AuditLog.action == "plugin.assign"))) + assert len(rows) == 1 + assert rows[0].target_id == "reftest_plugin/default" + assert rows[0].target_kind == AuditTargetKind.PLUGIN + assert rows[0].payload == {"users": ["kid1"]} + + +# --- PATCH ------------------------------------------------------------------- + + +def test_patch_users( + client_with_repo_plugins: TestClient, auth: dict[str, str], tmp_path: Path +) -> None: + client_with_repo_plugins.post( + "/v1/plugins", + json={"type": "reftest_plugin", "config": _ok_config(tmp_path), "users": ["kid1"]}, + headers=auth, + ) + r = client_with_repo_plugins.patch( + "/v1/plugins/reftest_plugin/default", + json={"users": ["kid1", "kid2"]}, + headers=auth, + ) + assert r.status_code == 200 + assert r.json()["users"] == ["kid1", "kid2"] + + +def test_patch_paused( + client_with_repo_plugins: TestClient, auth: dict[str, str], tmp_path: Path +) -> None: + client_with_repo_plugins.post( + "/v1/plugins", + json={"type": "reftest_plugin", "config": _ok_config(tmp_path), "users": ["*"]}, + headers=auth, + ) + r = client_with_repo_plugins.patch( + "/v1/plugins/reftest_plugin/default", + json={"paused": True}, + headers=auth, + ) + assert r.status_code == 200 + assert r.json()["paused"] is True + + +def test_patch_config_invalid_422( + client_with_repo_plugins: TestClient, auth: dict[str, str], tmp_path: Path +) -> None: + client_with_repo_plugins.post( + "/v1/plugins", + json={"type": "reftest_plugin", "config": _ok_config(tmp_path), "users": ["*"]}, + headers=auth, + ) + # sentinel_path must be a string; an int fails validation. + r = client_with_repo_plugins.patch( + "/v1/plugins/reftest_plugin/default", + json={"config": {"sentinel_path": 42}}, + headers=auth, + ) + assert r.status_code == 422 + + +def test_patch_404(client_with_repo_plugins: TestClient, auth: dict[str, str]) -> None: + r = client_with_repo_plugins.patch( + "/v1/plugins/reftest_plugin/default", + json={"users": []}, + headers=auth, + ) + assert r.status_code == 404 + + +def test_patch_empty_payload_no_op( + client_with_repo_plugins: TestClient, auth: dict[str, str], tmp_path: Path +) -> None: + client_with_repo_plugins.post( + "/v1/plugins", + json={"type": "reftest_plugin", "config": _ok_config(tmp_path), "users": ["*"]}, + headers=auth, + ) + r = client_with_repo_plugins.patch( + "/v1/plugins/reftest_plugin/default", + json={}, + headers=auth, + ) + assert r.status_code == 200 + assert r.json()["users"] == ["*"] + + +# --- DELETE ------------------------------------------------------------------ + + +def test_delete_204( + client_with_repo_plugins: TestClient, auth: dict[str, str], tmp_path: Path +) -> None: + client_with_repo_plugins.post( + "/v1/plugins", + json={"type": "reftest_plugin", "config": _ok_config(tmp_path), "users": ["*"]}, + headers=auth, + ) + r = client_with_repo_plugins.delete("/v1/plugins/reftest_plugin/default", headers=auth) + assert r.status_code == 204 + r = client_with_repo_plugins.get("/v1/plugins", headers=auth) + assert r.json() == [] + + +def test_delete_404(client_with_repo_plugins: TestClient, auth: dict[str, str]) -> None: + r = client_with_repo_plugins.delete("/v1/plugins/reftest_plugin/default", headers=auth) + assert r.status_code == 404 + + +# --- Auth gating ------------------------------------------------------------- + + +def test_create_requires_admin( + client_with_repo_plugins: TestClient, + auth: dict[str, str], + tmp_path: Path, +) -> None: + """A manager cookie can read the list but can't assign.""" + # Bootstrap a manager via root token. + r = client_with_repo_plugins.post( + "/v1/users", + json={"username": "mgr", "password": "test1234", "role": "manager"}, + headers=auth, + ) + assert r.status_code == 201 + r = client_with_repo_plugins.post( + "/v1/auth/login", json={"username": "mgr", "password": "test1234"} + ) + assert r.status_code == 204 + + # No bearer in headers; client carries the manager session cookie. + r = client_with_repo_plugins.post( + "/v1/plugins", + json={"type": "reftest_plugin", "config": _ok_config(tmp_path), "users": []}, + ) + assert r.status_code == 403 diff --git a/api/tests/test_plugins_reconcile.py b/api/tests/test_plugins_reconcile.py new file mode 100644 index 0000000..5ffd14c --- /dev/null +++ b/api/tests/test_plugins_reconcile.py @@ -0,0 +1,162 @@ +"""End-to-end tests for the plugin reconcile path. + +Exercises kernel acceptance steps 11-16 (PLAN.md §"Acceptance: kernel done"): +the operator assigns the reftest plugin, locks a user, the kernel dispatches +``reconcile`` in the background, and the plugin's sentinel file appears. + +The reftest plugin lives in the repo's ``plugins/reftest-plugin/`` and is +discovered via the ``CURFEW_PLUGINS_DIRS`` env var that the +``client_with_repo_plugins`` fixture sets. +""" + +from __future__ import annotations + +import asyncio +from pathlib import Path + +from curfew.models import AuditLog +from fastapi.testclient import TestClient +from sqlmodel import Session, create_engine, select + + +def _engine(db: Path): # type: ignore[no-untyped-def] + return create_engine(f"sqlite:///{db}") + + +def _setup_kid_and_plugin( + client: TestClient, auth: dict[str, str], sentinel: Path, *, paused: bool = False +) -> None: + """Seed: one managed kid + one reftest_plugin assignment governing them.""" + r = client.post( + "/v1/users", + json={"username": "kid1", "password": "test1234", "role": "member"}, + headers=auth, + ) + assert r.status_code == 201, r.text + r = client.post( + "/v1/plugins", + json={ + "type": "reftest_plugin", + "config": {"sentinel_path": str(sentinel)}, + "users": ["kid1"], + }, + headers=auth, + ) + assert r.status_code == 201, r.text + if paused: + r = client.patch( + "/v1/plugins/reftest_plugin/default", + json={"paused": True}, + headers=auth, + ) + assert r.status_code == 200 + + +def test_lock_dispatches_reconcile_and_sentinel_appears( + client_with_repo_plugins: TestClient, auth: dict[str, str], tmp_path: Path +) -> None: + sentinel = tmp_path / "sentinel" + _setup_kid_and_plugin(client_with_repo_plugins, auth, sentinel) + assert not sentinel.exists() + + r = client_with_repo_plugins.post("/v1/users/kid1/lock", headers=auth) + assert r.status_code == 200 + assert sentinel.read_text() == "kid1" + + +def test_unlock_clears_sentinel( + client_with_repo_plugins: TestClient, auth: dict[str, str], tmp_path: Path +) -> None: + sentinel = tmp_path / "sentinel" + _setup_kid_and_plugin(client_with_repo_plugins, auth, sentinel) + client_with_repo_plugins.post("/v1/users/kid1/lock", headers=auth) + assert sentinel.exists() + + r = client_with_repo_plugins.post("/v1/users/kid1/unlock", headers=auth) + assert r.status_code == 200 + assert not sentinel.exists() + + +def test_paused_plugin_doesnt_reconcile( + client_with_repo_plugins: TestClient, auth: dict[str, str], tmp_path: Path +) -> None: + sentinel = tmp_path / "sentinel" + _setup_kid_and_plugin(client_with_repo_plugins, auth, sentinel, paused=True) + + r = client_with_repo_plugins.post("/v1/users/kid1/lock", headers=auth) + assert r.status_code == 200 + assert not sentinel.exists() + + +def test_safety_net_picks_up_missed_lock( + client_with_repo_plugins: TestClient, auth: dict[str, str], tmp_path: Path +) -> None: + """If a lock fires while the plugin is paused, unpausing + resync writes the sentinel. + + Simulates the production case where curfew-core restarted between a + ``lock`` and the plugin's reconcile firing — the safety-net loop is + meant to catch this. + """ + sentinel = tmp_path / "sentinel" + _setup_kid_and_plugin(client_with_repo_plugins, auth, sentinel, paused=True) + + # Lock while paused: nothing reconciles. + client_with_repo_plugins.post("/v1/users/kid1/lock", headers=auth) + assert not sentinel.exists() + + # Unpause + manually trigger the safety-net path. (In production the + # background loop fires every plugin_resync_seconds; we don't sleep + # 300s for a test, we invoke it directly.) + r = client_with_repo_plugins.patch( + "/v1/plugins/reftest_plugin/default", + json={"paused": False}, + headers=auth, + ) + assert r.status_code == 200 + runtime = client_with_repo_plugins.app.state.plugin_runtime + asyncio.run(runtime.safety_net_resync()) + + assert sentinel.read_text() == "kid1" + + +def test_failed_reconcile_audited_but_lock_succeeds( + client_with_repo_plugins: TestClient, + auth: dict[str, str], + tmp_path: Path, + configured_db_with_repo_plugins: Path, +) -> None: + """Pointing sentinel at an unwritable path makes reconcile raise. + + The lock route should still 200 (the operator's request must succeed + even when downstream plugins fail); the failure is recorded as + ``plugin.reconcile_failed`` per PLUGINS.md. + """ + bad_sentinel = tmp_path / "no-such-dir" / "sentinel" + _setup_kid_and_plugin(client_with_repo_plugins, auth, bad_sentinel) + + r = client_with_repo_plugins.post("/v1/users/kid1/lock", headers=auth) + assert r.status_code == 200, r.text + assert not bad_sentinel.exists() # write would have failed + + with Session(_engine(configured_db_with_repo_plugins)) as s: + rows = list(s.exec(select(AuditLog).where(AuditLog.action == "plugin.reconcile_failed"))) + assert len(rows) == 1 + assert rows[0].target_id == "reftest_plugin/default" + assert rows[0].payload["user"] == "kid1" + + +def test_assignments_hydrate_on_startup( + client_with_repo_plugins: TestClient, auth: dict[str, str], tmp_path: Path +) -> None: + """A row that exists at startup should be live in the runtime. + + We test this by creating an assignment, locking (which uses the + runtime), and asserting the sentinel appears — proving the row was + instantiated and registered. + """ + sentinel = tmp_path / "sentinel" + _setup_kid_and_plugin(client_with_repo_plugins, auth, sentinel) + + r = client_with_repo_plugins.post("/v1/users/kid1/lock", headers=auth) + assert r.status_code == 200 + assert sentinel.exists() diff --git a/core/curfew/__init__.py b/core/curfew/__init__.py index a41e234..4fed836 100644 --- a/core/curfew/__init__.py +++ b/core/curfew/__init__.py @@ -28,6 +28,14 @@ from curfew.passwords import hash_password, verify_password from curfew.plugin import Plugin, PluginManifest, ReconcileResult, ReconcileStatus, Scope from curfew.plugin_loader import PluginRegistry, PluginType, discover_plugins +from curfew.plugin_runtime import ( + AlreadyAssignedError, + AssignedInstance, + NotAssignedError, + PluginInstantiationError, + PluginRuntime, + UnknownPluginTypeError, +) from curfew.rules import ( RulePipeline, device_scope, @@ -48,6 +56,9 @@ LoginRequest, ManifestSnapshotRow, MeResponse, + PluginAssignmentCreate, + PluginAssignmentRead, + PluginAssignmentUpdate, PluginSnapshotRow, PluginTypeRead, Reason, @@ -65,10 +76,12 @@ "Agent", "AgentSnapshotRow", "AgentToken", + "AlreadyAssignedError", "App", "AppCreate", "AppRead", "AppUpdate", + "AssignedInstance", "AuditLog", "AuditTargetKind", "ChangePasswordRequest", @@ -83,10 +96,16 @@ "Manifest", "ManifestSnapshotRow", "MeResponse", + "NotAssignedError", "Plugin", "PluginAssignment", + "PluginAssignmentCreate", + "PluginAssignmentRead", + "PluginAssignmentUpdate", + "PluginInstantiationError", "PluginManifest", "PluginRegistry", + "PluginRuntime", "PluginSnapshotRow", "PluginType", "PluginTypeRead", @@ -98,6 +117,7 @@ "SettingsRead", "SettingsUpdate", "SystemSnapshot", + "UnknownPluginTypeError", "User", "UserCreate", "UserLock", diff --git a/core/curfew/plugin_runtime.py b/core/curfew/plugin_runtime.py new file mode 100644 index 0000000..52a3017 --- /dev/null +++ b/core/curfew/plugin_runtime.py @@ -0,0 +1,311 @@ +"""Plugin runtime — instance lifecycle + reconcile dispatch + safety-net resync. + +Per PLAN.md §"Plugin lifecycle" + ADR-005 + PLUGINS.md §"Lifecycle": + +- The kernel **instantiates** assigned plugins at startup (and on assign/PATCH) + using the discovered class + the operator's config. The instance lives in + memory; its lifetime is the core's. There's no second persistent layer — + ``PluginAssignment`` rows are the desired state, instances are the runtime + projection. +- When a user's lock state changes, the kernel **dispatches** ``reconcile`` + on every plugin governing that user. Dispatch is async (FastAPI + ``BackgroundTasks``) so the originating HTTP request returns as soon as the + DB write commits — slow plugins don't delay the operator. +- A **safety-net resync** runs every ``plugin_resync_seconds`` and calls + ``reconcile`` for every managed user, on every governing plugin. Catches + missed events from a crash / restart between a state change and the + matching reconcile. + +Errors at the ``reconcile`` boundary are **caught + audited, never raised**. +The originating request has already returned by the time the background task +runs; raising would just kill the task. ``plugin.reconcile_failed`` audit +rows are how the operator confirms a specific reconcile actually ran (or +didn't). PLUGINS.md §"What can go wrong". +""" + +from __future__ import annotations + +import asyncio +import logging +from contextlib import contextmanager +from dataclasses import dataclass, replace +from typing import Any + +from sqlalchemy import Engine +from sqlmodel import Session, select + +from curfew.audit import record_audit +from curfew.models import AuditTargetKind, User +from curfew.models import Settings as SettingsRow +from curfew.plugin import Plugin, ReconcileResult, Scope +from curfew.plugin_loader import PluginRegistry +from curfew.rules import user_scope + +logger = logging.getLogger(__name__) + + +class UnknownPluginTypeError(LookupError): + """The runtime was asked to assign/update an instance whose ``type`` isn't discovered.""" + + +class PluginInstantiationError(RuntimeError): + """The plugin class raised during ``__init__`` (after config validation passed).""" + + +class AlreadyAssignedError(ValueError): + """``(type, instance_id)`` is already in the runtime.""" + + +class NotAssignedError(KeyError): + """``(type, instance_id)`` is not in the runtime.""" + + +@dataclass(frozen=True, slots=True) +class AssignedInstance: + """The runtime's view of one assigned plugin instance. + + Frozen + slot-based for cheap copying via ``dataclasses.replace`` on + PATCH. ``instance`` is the live ``Plugin`` object (callable + ``reconcile``). ``config`` is the validated dict (round-tripped through + the plugin's Pydantic config model). + """ + + type: str + instance_id: str + instance: Plugin + config: dict[str, Any] + users: list[str] + paused: bool + + +class PluginRuntime: + """In-memory ``{(type, instance_id): AssignedInstance}`` map + dispatch. + + One process-global instance, stashed on ``app.state.plugin_runtime``. + All mutations go through ``async`` methods guarded by ``_lock`` so + concurrent operator actions can't tear the map mid-update. + """ + + def __init__(self, registry: PluginRegistry, engine: Engine) -> None: + self._registry = registry + self._engine = engine + self._instances: dict[tuple[str, str], AssignedInstance] = {} + self._lock = asyncio.Lock() + + # --- Read API ------------------------------------------------------------ + + def all_instances(self) -> list[AssignedInstance]: + """Return every instance, sorted by ``(type, instance_id)``. + + Named ``all_instances`` (not ``list``) to avoid shadowing the + ``list`` builtin in mypy's view of method-parameter annotations + like ``users: list[str]`` further down. + """ + return sorted(self._instances.values(), key=lambda x: (x.type, x.instance_id)) + + def get(self, type_name: str, instance_id: str) -> AssignedInstance | None: + return self._instances.get((type_name, instance_id)) + + # --- Lifecycle (mutations) ----------------------------------------------- + + def build_instance(self, type_name: str, config: dict[str, Any]) -> Plugin: + """Validate config + construct a ``Plugin``. Raises on either step. + + Public so the API layer can pre-validate config before writing the + DB row, surfacing 422s without leaving an orphaned row behind. + """ + ptype = self._registry.get(type_name) + if ptype is None or ptype.error is not None: + raise UnknownPluginTypeError( + f"plugin type {type_name!r} is not discovered or failed to load" + ) + assert ptype.config_model is not None + assert ptype.plugin_class is not None + try: + validated = ptype.config_model(**config) + except Exception: + # Let pydantic.ValidationError propagate so callers can map it + # to a 422 with field-level detail. + raise + try: + # Plugin subclasses define their own __init__ taking the + # plugin's Pydantic config; the base `Plugin` doesn't, so + # mypy can't see the call signature. + return ptype.plugin_class(validated) # type: ignore[call-arg] + except Exception as exc: + raise PluginInstantiationError(f"plugin {type_name!r} __init__ raised: {exc}") from exc + + async def register( + self, + *, + type_name: str, + instance_id: str, + instance: Plugin, + config: dict[str, Any], + users: list[str], + paused: bool = False, + ) -> AssignedInstance: + """Register a pre-built ``Plugin`` instance. + + Split from ``build_instance`` so the API can validate + construct + first (failing fast with 422 / 500), write the DB row (failing + with 409 on duplicate), and only then commit the runtime mutation + — keeping the in-memory map and the ``plugin_assignments`` table + in lock-step. + """ + async with self._lock: + key = (type_name, instance_id) + if key in self._instances: + raise AlreadyAssignedError(f"{type_name}/{instance_id}") + live = AssignedInstance( + type=type_name, + instance_id=instance_id, + instance=instance, + config=config, + users=list(users), + paused=paused, + ) + self._instances[key] = live + return live + + async def unassign(self, *, type_name: str, instance_id: str) -> None: + """Drop the instance from the map. Idempotent — silent if absent.""" + async with self._lock: + self._instances.pop((type_name, instance_id), None) + + async def update( + self, + *, + type_name: str, + instance_id: str, + config: dict[str, Any] | None = None, + users: list[str] | None = None, + paused: bool | None = None, + ) -> AssignedInstance: + """Mutate one or more fields. Re-instantiates iff ``config`` changed.""" + async with self._lock: + key = (type_name, instance_id) + live = self._instances.get(key) + if live is None: + raise NotAssignedError(f"{type_name}/{instance_id}") + + if config is not None and config != live.config: + # PLAN.md option 1: re-instantiate. Cheaper than asking every + # plugin author to handle live config swaps inside reconcile. + new_instance = self.build_instance(type_name, config) + live = replace(live, instance=new_instance, config=config) + if users is not None: + live = replace(live, users=list(users)) + if paused is not None: + live = replace(live, paused=paused) + + self._instances[key] = live + return live + + # --- Dispatch ------------------------------------------------------------ + + async def dispatch_for_user(self, username: str) -> None: + """Reconcile every plugin governing ``username``. + + Runs reconciles concurrently. ``_run_one`` catches its own errors, + so ``gather`` never sees one. Builds the ``Scope`` by evaluating the + rule pipeline against a fresh session — the originating HTTP + session is closed by the time a background task runs. + """ + governing = [ + live + for live in self._instances.values() + if not live.paused and self._governs(live, username) + ] + if not governing: + return + + with self._session() as session: + status = user_scope.evaluate(session, username) + scope = Scope(user=username, locked=status.locked, reasons=list(status.reasons)) + + await asyncio.gather(*[self._run_one(live, scope) for live in governing]) + + async def safety_net_resync(self) -> None: + """Walk every managed user, dispatch. + + PLAN.md §"Plugin lifecycle" point 4: the slow tick that catches + missed events from a restart. Reading the timeout fresh each + invocation isn't enough — the *interval* is read by the lifespan + loop, not here. + """ + with self._session() as session: + # Filter in Python rather than SQL: SQLModel + mypy can't agree on + # what `User.managed` is typed as for a boolean WHERE clause, and + # the user table is small enough at homelab scale that the in-SQL + # filter saves nothing. + usernames = [u.username for u in session.exec(select(User)).all() if u.managed] + for username in usernames: + try: + await self.dispatch_for_user(username) + except asyncio.CancelledError: + raise + except Exception: + # ``_run_one`` already catches per-reconcile errors; the only + # thing that should escape here is a programming bug. Log + # and keep iterating so one bad user doesn't stall the loop. + logger.exception("safety-net resync failed for %r", username) + + # --- Internals ----------------------------------------------------------- + + @staticmethod + def _governs(live: AssignedInstance, username: str) -> bool: + return "*" in live.users or username in live.users + + async def _run_one(self, live: AssignedInstance, scope: Scope) -> None: + """Call one plugin's ``reconcile`` with a timeout; audit failures.""" + timeout = self._read_timeout() + try: + result: ReconcileResult = await asyncio.wait_for( + live.instance.reconcile(scope), timeout=timeout + ) + except asyncio.CancelledError: + raise + except TimeoutError: + self._audit_failure(live, scope, error=f"timed out after {timeout}s") + return + except Exception as exc: + self._audit_failure(live, scope, error=f"{type(exc).__name__}: {exc}") + return + if not result.succeeded: + self._audit_failure(live, scope, error=result.message or "reconcile returned error") + + def _audit_failure(self, live: AssignedInstance, scope: Scope, *, error: str) -> None: + try: + with self._session() as session: + record_audit( + session, + actor="system", + action="plugin.reconcile_failed", + target_kind=AuditTargetKind.PLUGIN, + target_id=f"{live.type}/{live.instance_id}", + payload={ + "user": scope.user, + "locked": scope.locked, + "error": error, + }, + ) + session.commit() + except Exception: + # Audit is best-effort; we already lost the reconcile, don't + # also lose the audit chain by raising here. + logger.exception( + "failed to audit plugin.reconcile_failed for %s/%s", + live.type, + live.instance_id, + ) + + def _read_timeout(self) -> int: + with self._session() as session: + row = session.exec(select(SettingsRow)).one() + return int(row.plugin_reconcile_timeout_seconds) + + @contextmanager + def _session(self): # type: ignore[no-untyped-def] + with Session(self._engine) as s: + yield s diff --git a/core/curfew/schemas.py b/core/curfew/schemas.py index 8d21c26..1d69567 100644 --- a/core/curfew/schemas.py +++ b/core/curfew/schemas.py @@ -196,6 +196,40 @@ class AgentSnapshotRow(BaseModel): last_seen_version: str | None +class PluginAssignmentRead(BaseModel): + """One assigned plugin instance — what ``GET /v1/plugins`` returns. + + Mirrors the ``plugin_assignments`` row + the in-memory paused flag. + Doesn't surface the live ``Plugin`` instance object; that's a runtime + concern, not part of the API contract. + """ + + model_config = ConfigDict(from_attributes=True) + + type: str + instance_id: str + config: dict[str, object] + users: list[str] + paused: bool + + +class PluginAssignmentCreate(BaseModel): + """``POST /v1/plugins`` body. ``instance_id`` defaults to ``"default"``.""" + + type: str + instance_id: str = "default" + config: dict[str, object] = Field(default_factory=dict) + users: list[str] = Field(default_factory=list) + + +class PluginAssignmentUpdate(BaseModel): + """``PATCH /v1/plugins/{type}/{instance_id}`` body. All-optional.""" + + config: dict[str, object] | None = None + users: list[str] | None = None + paused: bool | None = None + + class PluginTypeRead(BaseModel): """One discovered plugin type as returned by ``GET /v1/plugins/types``. diff --git a/core/tests/test_plugin_runtime.py b/core/tests/test_plugin_runtime.py new file mode 100644 index 0000000..29c4c18 --- /dev/null +++ b/core/tests/test_plugin_runtime.py @@ -0,0 +1,525 @@ +"""Tests for PluginRuntime — the in-memory instance manager + dispatch. + +These tests use synthetic plugin classes built directly in the test (no +discovery, no manifest parsing — that's covered in test_plugin_loader.py). +A small fake ``PluginRegistry`` lets us bind a hand-rolled plugin class to +a type name and exercise the runtime's lifecycle methods without touching +the filesystem. + +Tests are sync wrappers around ``asyncio.run`` for symmetry with +``test_plugin_sdk.py`` — pytest-asyncio isn't a project dep and the small +amount of boilerplate avoids adding one for these few tests. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Iterator +from pathlib import Path +from typing import Any, ClassVar + +import pytest +from curfew.models import AuditLog, User, UserLock, UserRole +from curfew.models import Settings as SettingsRow +from curfew.plugin import Plugin, ReconcileResult, Scope +from curfew.plugin_loader import PluginManifest, PluginRegistry, PluginType +from curfew.plugin_runtime import ( + AlreadyAssignedError, + NotAssignedError, + PluginInstantiationError, + PluginRuntime, + UnknownPluginTypeError, +) +from curfew.rules import register_kernel_rules, user_scope +from pydantic import BaseModel, ValidationError +from sqlalchemy import event +from sqlalchemy.engine import Engine +from sqlmodel import Session, SQLModel, create_engine, select + + +@event.listens_for(Engine, "connect") +def _enable_sqlite_foreign_keys(dbapi_connection: Any, _connection_record: Any) -> None: + cursor = dbapi_connection.cursor() + cursor.execute("PRAGMA foreign_keys = ON") + cursor.close() + + +@pytest.fixture +def engine(tmp_path: Path) -> Iterator[Engine]: + db = tmp_path / "test.sqlite" + eng = create_engine(f"sqlite:///{db}") + SQLModel.metadata.create_all(eng) + # Seed the singleton settings row so _read_timeout finds something. + with Session(eng) as s: + s.add(SettingsRow(id=1)) + s.commit() + yield eng + + +@pytest.fixture +def kernel_rules() -> Iterator[None]: + register_kernel_rules() + yield + user_scope.reset() + + +# --- Plugin fixtures --------------------------------------------------------- + + +class _Cfg(BaseModel): + note: str = "hello" + + +class _Recorder(Plugin): + """Plugin that records every reconcile call onto a class-level list.""" + + calls: ClassVar[list[Scope]] = [] + init_count: ClassVar[int] = 0 + + def __init__(self, config: _Cfg) -> None: + self.config = config + type(self).init_count += 1 + + async def reconcile(self, scope: Scope) -> ReconcileResult: + type(self).calls.append(scope) + return ReconcileResult.ok() + + +class _Slow(Plugin): + def __init__(self, config: _Cfg) -> None: + self.config = config + + async def reconcile(self, scope: Scope) -> ReconcileResult: + await asyncio.sleep(60) + return ReconcileResult.ok() + + +class _Boom(Plugin): + def __init__(self, config: _Cfg) -> None: + self.config = config + + async def reconcile(self, scope: Scope) -> ReconcileResult: + raise RuntimeError("kaboom") + + +class _ReturnsError(Plugin): + def __init__(self, config: _Cfg) -> None: + self.config = config + + async def reconcile(self, scope: Scope) -> ReconcileResult: + return ReconcileResult.error("upstream 500") + + +class _BadInit(Plugin): + def __init__(self, config: _Cfg) -> None: + raise ValueError("init exploded") + + async def reconcile(self, scope: Scope) -> ReconcileResult: + return ReconcileResult.ok() + + +def _ptype(name: str, cls: type[Plugin]) -> PluginType: + return PluginType( + type=name, + folder=Path(f"/fake/{name}"), + manifest=PluginManifest( + type=name, name=name, version="1.0", description=name, config_schema="Config" + ), + plugin_class=cls, + config_model=_Cfg, + error=None, + ) + + +@pytest.fixture +def registry() -> PluginRegistry: + r = PluginRegistry() + r.add(_ptype("recorder", _Recorder)) + r.add(_ptype("slow", _Slow)) + r.add(_ptype("boom", _Boom)) + r.add(_ptype("err", _ReturnsError)) + r.add(_ptype("bad_init", _BadInit)) + r.add( + PluginType( + type="halfbroken", + folder=Path("/fake/halfbroken"), + manifest=None, + plugin_class=None, + config_model=None, + error="manifest.toml: missing required field(s): config_schema", + ) + ) + return r + + +@pytest.fixture(autouse=True) +def reset_recorder_state() -> None: + _Recorder.calls = [] + _Recorder.init_count = 0 + + +@pytest.fixture +def runtime(registry: PluginRegistry, engine: Engine) -> PluginRuntime: + return PluginRuntime(registry=registry, engine=engine) + + +def _seed_user( + engine: Engine, username: str, *, locked: bool = False, managed: bool = True +) -> None: + with Session(engine) as s: + u = User(username=username, role=UserRole.MEMBER, managed=managed) + s.add(u) + s.flush() + if locked: + s.add(UserLock(user_id=u.id, manual_lock=True)) + s.commit() + + +# --- build_instance ---------------------------------------------------------- + + +def test_build_unknown_type(runtime: PluginRuntime) -> None: + with pytest.raises(UnknownPluginTypeError): + runtime.build_instance("does_not_exist", {}) + + +def test_build_failed_type(runtime: PluginRuntime) -> None: + """A discovered-but-failed type can't be built either.""" + with pytest.raises(UnknownPluginTypeError): + runtime.build_instance("halfbroken", {}) + + +def test_build_invalid_config(runtime: PluginRuntime) -> None: + """A config that doesn't match the Pydantic model raises ValidationError.""" + with pytest.raises(ValidationError): + runtime.build_instance("recorder", {"note": 123}) + + +def test_build_init_raises(runtime: PluginRuntime) -> None: + with pytest.raises(PluginInstantiationError, match="init exploded"): + runtime.build_instance("bad_init", {}) + + +# --- register / unassign / update ------------------------------------------- + + +def test_register_and_unassign(runtime: PluginRuntime) -> None: + async def go() -> None: + inst = runtime.build_instance("recorder", {}) + live = await runtime.register( + type_name="recorder", + instance_id="default", + instance=inst, + config={"note": "hello"}, + users=["kid1"], + ) + assert live.type == "recorder" + assert live.instance is inst + assert runtime.get("recorder", "default") is live + + await runtime.unassign(type_name="recorder", instance_id="default") + assert runtime.get("recorder", "default") is None + + asyncio.run(go()) + + +def test_register_duplicate_raises(runtime: PluginRuntime) -> None: + async def go() -> None: + await runtime.register( + type_name="recorder", + instance_id="default", + instance=runtime.build_instance("recorder", {}), + config={}, + users=[], + ) + with pytest.raises(AlreadyAssignedError): + await runtime.register( + type_name="recorder", + instance_id="default", + instance=runtime.build_instance("recorder", {}), + config={}, + users=[], + ) + + asyncio.run(go()) + + +def test_unassign_idempotent(runtime: PluginRuntime) -> None: + """Unassigning a slot that isn't there is a no-op (no exception).""" + asyncio.run(runtime.unassign(type_name="recorder", instance_id="default")) + + +def test_update_unknown_raises(runtime: PluginRuntime) -> None: + async def go() -> None: + with pytest.raises(NotAssignedError): + await runtime.update(type_name="recorder", instance_id="default") + + asyncio.run(go()) + + +def test_update_users_doesnt_reinstantiate(runtime: PluginRuntime) -> None: + async def go() -> None: + inst = runtime.build_instance("recorder", {}) + baseline = _Recorder.init_count + await runtime.register( + type_name="recorder", + instance_id="default", + instance=inst, + config={}, + users=["kid1"], + ) + live = await runtime.update( + type_name="recorder", instance_id="default", users=["kid1", "kid2"] + ) + assert live.users == ["kid1", "kid2"] + assert live.instance is inst + assert _Recorder.init_count == baseline + + asyncio.run(go()) + + +def test_update_paused_doesnt_reinstantiate(runtime: PluginRuntime) -> None: + async def go() -> None: + inst = runtime.build_instance("recorder", {}) + await runtime.register( + type_name="recorder", + instance_id="default", + instance=inst, + config={}, + users=["*"], + ) + baseline = _Recorder.init_count + live = await runtime.update(type_name="recorder", instance_id="default", paused=True) + assert live.paused is True + assert live.instance is inst + assert _Recorder.init_count == baseline + + asyncio.run(go()) + + +def test_update_config_reinstantiates(runtime: PluginRuntime) -> None: + async def go() -> None: + inst = runtime.build_instance("recorder", {"note": "first"}) + baseline = _Recorder.init_count + await runtime.register( + type_name="recorder", + instance_id="default", + instance=inst, + config={"note": "first"}, + users=["*"], + ) + live = await runtime.update( + type_name="recorder", instance_id="default", config={"note": "second"} + ) + assert live.config == {"note": "second"} + assert live.instance is not inst + assert _Recorder.init_count == baseline + 1 + + asyncio.run(go()) + + +def test_update_config_unchanged_skips_reinstantiate(runtime: PluginRuntime) -> None: + async def go() -> None: + inst = runtime.build_instance("recorder", {"note": "x"}) + await runtime.register( + type_name="recorder", + instance_id="default", + instance=inst, + config={"note": "x"}, + users=["*"], + ) + baseline = _Recorder.init_count + live = await runtime.update( + type_name="recorder", instance_id="default", config={"note": "x"} + ) + assert live.instance is inst + assert _Recorder.init_count == baseline + + asyncio.run(go()) + + +# --- dispatch_for_user ------------------------------------------------------- + + +def test_dispatch_picks_governing_instances( + runtime: PluginRuntime, engine: Engine, kernel_rules: None +) -> None: + _seed_user(engine, "kid1", locked=True) + + async def go() -> None: + await runtime.register( + type_name="recorder", + instance_id="default", + instance=runtime.build_instance("recorder", {}), + config={}, + users=["kid1"], + ) + await runtime.dispatch_for_user("kid1") + + asyncio.run(go()) + + assert len(_Recorder.calls) == 1 + scope = _Recorder.calls[0] + assert scope.user == "kid1" + assert scope.locked is True + assert any(r.kind == "manual_lock" for r in scope.reasons) + + +def test_dispatch_skips_non_governing( + runtime: PluginRuntime, engine: Engine, kernel_rules: None +) -> None: + _seed_user(engine, "kid1") + _seed_user(engine, "kid2") + + async def go() -> None: + await runtime.register( + type_name="recorder", + instance_id="default", + instance=runtime.build_instance("recorder", {}), + config={}, + users=["kid2"], + ) + await runtime.dispatch_for_user("kid1") + + asyncio.run(go()) + assert _Recorder.calls == [] + + +def test_dispatch_wildcard_users( + runtime: PluginRuntime, engine: Engine, kernel_rules: None +) -> None: + _seed_user(engine, "anyone") + + async def go() -> None: + await runtime.register( + type_name="recorder", + instance_id="default", + instance=runtime.build_instance("recorder", {}), + config={}, + users=["*"], + ) + await runtime.dispatch_for_user("anyone") + + asyncio.run(go()) + assert len(_Recorder.calls) == 1 + + +def test_dispatch_skips_paused(runtime: PluginRuntime, engine: Engine, kernel_rules: None) -> None: + _seed_user(engine, "kid1") + + async def go() -> None: + await runtime.register( + type_name="recorder", + instance_id="default", + instance=runtime.build_instance("recorder", {}), + config={}, + users=["*"], + paused=True, + ) + await runtime.dispatch_for_user("kid1") + + asyncio.run(go()) + assert _Recorder.calls == [] + + +def test_dispatch_audits_timeout( + runtime: PluginRuntime, engine: Engine, kernel_rules: None +) -> None: + """A reconcile that exceeds the timeout is cancelled + audited.""" + with Session(engine) as s: + row = s.exec(select(SettingsRow)).one() + row.plugin_reconcile_timeout_seconds = 1 + s.commit() + + _seed_user(engine, "kid1") + + async def go() -> None: + await runtime.register( + type_name="slow", + instance_id="default", + instance=runtime.build_instance("slow", {}), + config={}, + users=["*"], + ) + await runtime.dispatch_for_user("kid1") + + asyncio.run(go()) + + with Session(engine) as s: + rows = list(s.exec(select(AuditLog).where(AuditLog.action == "plugin.reconcile_failed"))) + assert len(rows) == 1 + assert rows[0].target_id == "slow/default" + assert "timed out" in rows[0].payload["error"] + + +def test_dispatch_audits_exception( + runtime: PluginRuntime, engine: Engine, kernel_rules: None +) -> None: + _seed_user(engine, "kid1") + + async def go() -> None: + await runtime.register( + type_name="boom", + instance_id="default", + instance=runtime.build_instance("boom", {}), + config={}, + users=["*"], + ) + await runtime.dispatch_for_user("kid1") + + asyncio.run(go()) + + with Session(engine) as s: + rows = list(s.exec(select(AuditLog).where(AuditLog.action == "plugin.reconcile_failed"))) + assert len(rows) == 1 + assert "RuntimeError" in rows[0].payload["error"] + assert "kaboom" in rows[0].payload["error"] + + +def test_dispatch_audits_error_result( + runtime: PluginRuntime, engine: Engine, kernel_rules: None +) -> None: + _seed_user(engine, "kid1") + + async def go() -> None: + await runtime.register( + type_name="err", + instance_id="default", + instance=runtime.build_instance("err", {}), + config={}, + users=["*"], + ) + await runtime.dispatch_for_user("kid1") + + asyncio.run(go()) + + with Session(engine) as s: + rows = list(s.exec(select(AuditLog).where(AuditLog.action == "plugin.reconcile_failed"))) + assert len(rows) == 1 + assert "upstream 500" in rows[0].payload["error"] + + +# --- safety_net_resync ------------------------------------------------------- + + +def test_resync_walks_managed_users( + runtime: PluginRuntime, engine: Engine, kernel_rules: None +) -> None: + _seed_user(engine, "kid1", locked=True) + _seed_user(engine, "kid2") + _seed_user(engine, "guest", managed=False) + + async def go() -> None: + await runtime.register( + type_name="recorder", + instance_id="default", + instance=runtime.build_instance("recorder", {}), + config={}, + users=["*"], + ) + await runtime.safety_net_resync() + + asyncio.run(go()) + + seen = sorted({s.user for s in _Recorder.calls}) + assert seen == ["kid1", "kid2"] diff --git a/plugins/reftest-plugin/plugin.py b/plugins/reftest-plugin/plugin.py index 52e24d3..65b9e08 100644 --- a/plugins/reftest-plugin/plugin.py +++ b/plugins/reftest-plugin/plugin.py @@ -1,32 +1,50 @@ -"""Reference test plugin (step 10 — no-op). - -This plugin exists to prove the discovery contract end-to-end (the kernel -finds it, parses ``manifest.toml``, imports this module, picks -``ReftestPlugin`` as the leaf class, validates the empty ``Config``). - -Step 11 will replace ``reconcile`` with logic that writes a sentinel file -when ``scope.locked`` is true and clears it when false — that's what the -kernel acceptance test asserts against. For now the kernel only needs to -prove that *discovery* works; ``reconcile`` returning ``ok()`` is enough. +"""Reference test plugin — writes a sentinel file when the user is locked. + +Used by the kernel acceptance test (PLAN.md §"Acceptance: kernel done" +steps 11-16) to prove the full plugin path works end-to-end: + +- Discovery picks this folder up at startup. +- Operator assigns it with a ``sentinel_path`` config and a list of + governed users. +- Operator locks a user; the lock route schedules + ``runtime.dispatch_for_user`` as a background task; the kernel calls + ``ReftestPlugin.reconcile`` with ``Scope(user, locked=True, ...)``; + this method writes ```` containing the username. +- Operator unlocks; reconcile fires again with ``locked=False``; this + method removes the sentinel file. + +Idempotent on both legs — locking a user already-locked rewrites the +file with the same content; unlocking when the file isn't there is a +no-op. That matches the safety-net resync semantics: reconcile may be +called multiple times for the same state; results should converge. """ from __future__ import annotations +from pathlib import Path + from curfew.plugin import Plugin, ReconcileResult, Scope from pydantic import BaseModel class Config(BaseModel): - """Empty config — the reftest plugin takes no operator input. + """Where to write the sentinel. - Step 11 will add a ``sentinel_path`` field so each test can point the - plugin at its own ``tmp_path``. + Tests typically point this at ``tmp_path / "sentinel"``; live smoke + tests use something like ``/tmp/curfew-reftest-sentinel``. """ + sentinel_path: str + class ReftestPlugin(Plugin): def __init__(self, config: Config) -> None: self.config = config async def reconcile(self, scope: Scope) -> ReconcileResult: + path = Path(self.config.sentinel_path) + if scope.locked: + path.write_text(scope.user) + else: + path.unlink(missing_ok=True) return ReconcileResult.ok()