diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9ad7958..ede906c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -31,6 +31,53 @@ jobs: - name: Unit and contract tests run: make test + postgres-integration: + runs-on: ubuntu-latest + timeout-minutes: 10 + env: + ACP_DATABASE_URL: postgresql+psycopg://control_plane:control_plane@127.0.0.1:5432/control_plane_test + services: + postgres: + image: postgres:17-alpine + env: + POSTGRES_DB: control_plane_test + POSTGRES_PASSWORD: control_plane + POSTGRES_USER: control_plane + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U control_plane -d control_plane_test" + --health-interval 5s + --health-timeout 3s + --health-retries 10 + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-python@v7 + with: + python-version: "3.11" + cache: pip + - name: Install + run: python -m pip install -e '.[dev]' + - name: Upgrade database + run: python -m alembic upgrade head + - name: Verify migration metadata + run: python -m alembic check + - name: PostgreSQL integration tests + run: make integration + - name: Rehearse migration rollback + run: | + python -m alembic downgrade base + python - <<'PY' + from agent_control_plane.postgres_store import PostgresControlPlaneStore + + store = PostgresControlPlaneStore( + "postgresql+psycopg://control_plane:control_plane@127.0.0.1:5432/control_plane_test" + ) + assert store.is_ready() is False + store.close() + PY + python -m alembic upgrade head + container-build: runs-on: ubuntu-latest timeout-minutes: 10 diff --git a/CHANGELOG.md b/CHANGELOG.md index 72b4338..1e652ed 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,3 +12,5 @@ for public contracts once they are declared stable. - Risk-based review and delivery policy. - Agent registration and lifecycle status APIs with optimistic revision checks. - Human approval queue with single-decision enforcement and append-only audit events. +- PostgreSQL system of record with transactional audit writes, Alembic migrations, readiness + checks, and database-level audit mutation protection. diff --git a/Dockerfile b/Dockerfile index 82c80c8..2a1f9ff 100644 --- a/Dockerfile +++ b/Dockerfile @@ -7,8 +7,9 @@ WORKDIR /app RUN groupadd --system app && useradd --system --gid app app -COPY pyproject.toml README.md ./ +COPY pyproject.toml README.md alembic.ini ./ COPY src ./src +COPY migrations ./migrations RUN python -m pip install --no-cache-dir . USER app diff --git a/Makefile b/Makefile index 65f765b..8226cbd 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: install format lint type unit smoke test audit check run +.PHONY: install format lint type unit smoke integration test audit check migrate run install: python -m pip install -e '.[dev]' @@ -20,6 +20,9 @@ unit: smoke: python -m pytest tests/smoke -m smoke +integration: + python -m pytest tests/integration -m integration + test: python -m pytest --cov=agent_control_plane --cov-report=term-missing @@ -28,5 +31,8 @@ audit: check: lint type test +migrate: + python -m alembic upgrade head + run: python -m agent_control_plane diff --git a/README.md b/README.md index 529fcbc..bede0c2 100644 --- a/README.md +++ b/README.md @@ -37,7 +37,8 @@ make check make run ``` -The API is then available at `http://127.0.0.1:8000`. Important endpoints: +Without `ACP_DATABASE_URL`, the service uses its process-local in-memory adapter. The API is +then available at `http://127.0.0.1:8000`. Important endpoints: - `GET /health/live` - `GET /health/ready` @@ -48,12 +49,21 @@ The API is then available at `http://127.0.0.1:8000`. Important endpoints: - `GET /v1/audit-events` - `GET /docs` -Container execution: +Persistent local execution starts PostgreSQL, runs migrations, and then starts the API: ```bash docker compose up --build ``` +For an externally managed PostgreSQL database, set a `postgresql+psycopg://` URL and migrate +before starting the service: + +```bash +export ACP_DATABASE_URL='postgresql+psycopg://user:password@host/database' +make migrate +make run +``` + ## Delivery policy Every change merged to `main` goes through a pull request, review, and required fast checks. @@ -67,9 +77,12 @@ The first governance loop is available: register an agent, activate or pause it revision checks, request and decide human approval, and inspect the resulting audit events. Public API compatibility starts with the `v1` schema. -The bundled store is intentionally in-memory and intended for development and evaluation. Data -does not survive a process restart and must not be treated as a production system of record. -PostgreSQL persistence, authenticated actor identity, and durable workflows remain planned. +PostgreSQL is available as the durable system of record. Agent changes, approval changes, and +their audit events commit atomically; a database trigger rejects audit updates, deletion, and +truncation. Readiness fails when the configured database is unavailable or not migrated. + +The in-memory adapter remains available for development and evaluation only. Authenticated +actor identity, request idempotency, backup automation, and durable workflows remain planned. ## License diff --git a/SECURITY.md b/SECURITY.md index c09ed88..7f7f4f8 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -10,3 +10,7 @@ changes require the high-risk review path described in `docs/QUALITY_GATES.md`. Never place API keys, customer traces, prompts, memories, production data, or credentials in the repository or test fixtures. + +The credentials in `compose.yaml` are fixed development-only values. Production deployments +must inject a separate database URL through secret management, restrict the application role, +encrypt connections, and run backup/restore exercises before storing customer data. diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..fc4ec07 --- /dev/null +++ b/alembic.ini @@ -0,0 +1,37 @@ +[alembic] +script_location = migrations +prepend_sys_path = . + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARN +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARN +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/compose.yaml b/compose.yaml index 47036c0..b27dd01 100644 --- a/compose.yaml +++ b/compose.yaml @@ -1,8 +1,38 @@ services: + postgres: + image: postgres:17-alpine + environment: + POSTGRES_DB: control_plane + POSTGRES_PASSWORD: control_plane + POSTGRES_USER: control_plane + ports: + - "${ACP_POSTGRES_PORT:-5432}:5432" + volumes: + - postgres-data:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U control_plane -d control_plane"] + interval: 5s + timeout: 3s + retries: 10 + + migrate: + build: . + command: ["alembic", "upgrade", "head"] + environment: + ACP_DATABASE_URL: postgresql+psycopg://control_plane:control_plane@postgres:5432/control_plane + depends_on: + postgres: + condition: service_healthy + control-plane: build: . + environment: + ACP_DATABASE_URL: postgresql+psycopg://control_plane:control_plane@postgres:5432/control_plane ports: - - "8000:8000" + - "${ACP_HTTP_PORT:-8000}:8000" + depends_on: + migrate: + condition: service_completed_successfully healthcheck: test: - CMD @@ -15,3 +45,6 @@ services: timeout: 3s retries: 3 start_period: 5s + +volumes: + postgres-data: diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 681339f..754819b 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -22,16 +22,21 @@ Control Plane API `- version promotion (planned) ``` -The current code implements the API shell, the first versioned contract, and an in-memory -governance loop. The storage protocol is owned by the control plane so PostgreSQL can replace -the development adapter without leaking database types into the public API. PostgreSQL becomes -the source of truth when persistence is introduced. Vector databases remain derived indexes, -not authoritative stores. +The current code implements the API shell, the first versioned contract, and a governance loop +with in-memory and PostgreSQL adapters. The storage protocol is owned by the control plane, so +database types do not leak into the public API. PostgreSQL is the durable source of truth; +vector databases remain derived indexes, not authoritative stores. State changes use an expected revision to reject stale writers. Only active agents can request approval. Approval requests are single-decision records: an approved or rejected request cannot -be overwritten. Audit events are append-only within the store and returned newest first. -Authentication and durable audit retention are required before production use. +be overwritten. State changes and their audit events share one database transaction. PostgreSQL +row locks serialize competing status and decision operations, while conditional updates provide +a second conflict check. A database trigger blocks audit mutation and removal. Events are +returned newest first. + +Alembic owns schema versioning. Deployments run migrations as a separate step before the API; +readiness stays unavailable when the schema is missing. Authentication, backup policy, and +retention enforcement are required before production use. ## Adapter policy diff --git a/docs/QUALITY_GATES.md b/docs/QUALITY_GATES.md index 140821b..39a9e2b 100644 --- a/docs/QUALITY_GATES.md +++ b/docs/QUALITY_GATES.md @@ -8,7 +8,7 @@ applies to changes merged into `main`, not to every local edit or experimental c | Lane | Target time | Required evidence | When it runs | | --- | ---: | --- | --- | | Local | under 60 seconds | focused tests, formatter | while developing | -| Pull request | under 5 minutes | lint, types, unit tests, smoke tests, container build | every PR | +| Pull request | under 5 minutes | lint, types, unit tests, PostgreSQL integration, smoke tests, container build | every PR | | Main | under 15 minutes | clean rebuild and complete current suite | every merge | | Scheduled | time-boxed | dependency audit and supported Python versions | weekly | | Release | risk-based | migration, rollback, security, and scenario tests | before release | @@ -30,6 +30,9 @@ Smoke tests prove that the packaged service starts conceptually, reports readine valid public contract, and rejects invalid input. They do not replace behavior, integration, load, recovery, or security tests. +Database changes additionally require upgrade, integration, downgrade, and re-upgrade evidence +against the supported PostgreSQL version. Migration rehearsal uses disposable data only. + ## Dependency maintenance Dependency pull requests must identify a compatibility, security, or reproducibility benefit. @@ -46,6 +49,7 @@ Configure a GitHub ruleset for `main` with: - stale approvals dismissed after new code is pushed; - conversation resolution required; - `fast-gate` and `container-build` required; +- `postgres-integration` required for database changes; - force pushes and deletion blocked; - administrators subject to the same rules. diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 169ac7e..ea93937 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -14,12 +14,12 @@ Roadmap items advance only when tied to a validated user problem and an acceptan - Run replay and failure classification. - Tool-call schema validation and risk policy. - [x] In-memory human approval queue and append-only audit contract. -- PostgreSQL-backed approval and immutable audit persistence. +- [x] PostgreSQL-backed approval and immutable audit persistence. - Offline evaluation datasets and version promotion gates. ## Durable operations -- PostgreSQL system of record. +- [x] PostgreSQL system of record with reversible migrations. - Durable workflow adapter for cross-day tasks. - Idempotency, retry, compensation, and dead-letter handling. - Backup, restore, tenant isolation, and disaster exercises. diff --git a/migrations/env.py b/migrations/env.py new file mode 100644 index 0000000..5861ee3 --- /dev/null +++ b/migrations/env.py @@ -0,0 +1,59 @@ +"""Alembic migration environment.""" + +import os +from logging.config import fileConfig + +from alembic import context +from sqlalchemy import engine_from_config, pool + +from agent_control_plane.bootstrap import DATABASE_URL_ENV +from agent_control_plane.db_schema import metadata + +config = context.config + +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +target_metadata = metadata + + +def database_url() -> str: + value = os.environ.get(DATABASE_URL_ENV) + if not value: + raise RuntimeError(f"{DATABASE_URL_ENV} is required for database migrations") + return value + + +def run_migrations_offline() -> None: + context.configure( + url=database_url(), + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + compare_type=True, + ) + + with context.begin_transaction(): + context.run_migrations() + + +def run_migrations_online() -> None: + configuration = config.get_section(config.config_ini_section) or {} + configuration["sqlalchemy.url"] = database_url() + connectable = engine_from_config( + configuration, + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + + with connectable.connect() as connection: + context.configure(connection=connection, target_metadata=target_metadata, compare_type=True) + + with context.begin_transaction(): + context.run_migrations() + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/migrations/script.py.mako b/migrations/script.py.mako new file mode 100644 index 0000000..e848d9e --- /dev/null +++ b/migrations/script.py.mako @@ -0,0 +1,25 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} +""" + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +revision: str = ${repr(up_revision)} +down_revision: str | None = ${repr(down_revision)} +branch_labels: str | Sequence[str] | None = ${repr(branch_labels)} +depends_on: str | Sequence[str] | None = ${repr(depends_on)} + + +def upgrade() -> None: + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + ${downgrades if downgrades else "pass"} diff --git a/migrations/versions/0001_initial.py b/migrations/versions/0001_initial.py new file mode 100644 index 0000000..b9bef05 --- /dev/null +++ b/migrations/versions/0001_initial.py @@ -0,0 +1,142 @@ +"""Create the durable control-plane system of record. + +Revision ID: 0001_initial +Revises: +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "0001_initial" +down_revision: str | None = None +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "agents", + sa.Column("agent_id", sa.String(length=63), nullable=False), + sa.Column("spec", postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column("status", sa.String(length=20), nullable=False), + sa.Column("revision", sa.Integer(), nullable=False), + sa.Column("registered_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.CheckConstraint("revision >= 1", name=op.f("ck_agents_positive_revision")), + sa.CheckConstraint( + "status IN ('registered', 'active', 'paused')", + name=op.f("ck_agents_valid_status"), + ), + sa.PrimaryKeyConstraint("agent_id", name=op.f("pk_agents")), + ) + op.create_table( + "approval_requests", + sa.Column("request_id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("agent_id", sa.String(length=63), nullable=False), + sa.Column("action", sa.String(length=128), nullable=False), + sa.Column("risk", sa.String(length=20), nullable=False), + sa.Column("status", sa.String(length=20), nullable=False), + sa.Column("requested_by", sa.String(length=128), nullable=False), + sa.Column("request_reason", sa.String(length=500), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("decided_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("decided_by", sa.String(length=128), nullable=True), + sa.Column("decision_reason", sa.String(length=500), nullable=True), + sa.CheckConstraint( + "(status = 'pending' AND decided_at IS NULL AND decided_by IS NULL " + "AND decision_reason IS NULL) OR " + "(status IN ('approved', 'rejected') AND decided_at IS NOT NULL " + "AND decided_by IS NOT NULL AND decision_reason IS NOT NULL)", + name=op.f("ck_approval_requests_decision_consistency"), + ), + sa.CheckConstraint( + "risk IN ('low', 'medium', 'high')", + name=op.f("ck_approval_requests_valid_risk"), + ), + sa.CheckConstraint( + "status IN ('pending', 'approved', 'rejected')", + name=op.f("ck_approval_requests_valid_status"), + ), + sa.ForeignKeyConstraint( + ["agent_id"], + ["agents.agent_id"], + name=op.f("fk_approval_requests_agent_id_agents"), + ondelete="RESTRICT", + ), + sa.PrimaryKeyConstraint("request_id", name=op.f("pk_approval_requests")), + ) + op.create_index( + op.f("ix_approval_requests_agent"), + "approval_requests", + ["agent_id"], + unique=False, + ) + op.create_index( + op.f("ix_approval_requests_queue"), + "approval_requests", + ["status", "created_at"], + unique=False, + ) + op.create_table( + "audit_events", + sa.Column("sequence", sa.BigInteger(), sa.Identity(always=False), nullable=False), + sa.Column("event_id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("agent_id", sa.String(length=63), nullable=False), + sa.Column("event_type", sa.String(length=40), nullable=False), + sa.Column("actor", sa.String(length=128), nullable=False), + sa.Column("occurred_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("resource_id", sa.String(length=128), nullable=False), + sa.Column("summary", sa.String(length=1000), nullable=False), + sa.CheckConstraint( + "event_type IN ('agent.registered', 'agent.status_changed', " + "'approval.requested', 'approval.approved', 'approval.rejected')", + name=op.f("ck_audit_events_valid_event_type"), + ), + sa.ForeignKeyConstraint( + ["agent_id"], + ["agents.agent_id"], + name=op.f("fk_audit_events_agent_id_agents"), + ondelete="RESTRICT", + ), + sa.PrimaryKeyConstraint("sequence", name=op.f("pk_audit_events")), + sa.UniqueConstraint("event_id", name=op.f("uq_audit_events_event_id")), + ) + op.create_index( + op.f("ix_audit_events_agent_sequence"), + "audit_events", + ["agent_id", "sequence"], + unique=False, + ) + op.execute( + """ + CREATE FUNCTION reject_audit_event_mutation() + RETURNS trigger + LANGUAGE plpgsql + AS $$ + BEGIN + RAISE EXCEPTION 'audit_events is append-only'; + END; + $$ + """ + ) + op.execute( + """ + CREATE TRIGGER audit_events_append_only + BEFORE UPDATE OR DELETE OR TRUNCATE ON audit_events + FOR EACH STATEMENT EXECUTE FUNCTION reject_audit_event_mutation() + """ + ) + + +def downgrade() -> None: + op.execute("DROP TRIGGER IF EXISTS audit_events_append_only ON audit_events") + op.execute("DROP FUNCTION IF EXISTS reject_audit_event_mutation()") + op.drop_index("ix_audit_events_agent_sequence", table_name="audit_events") + op.drop_table("audit_events") + op.drop_index("ix_approval_requests_queue", table_name="approval_requests") + op.drop_index("ix_approval_requests_agent", table_name="approval_requests") + op.drop_table("approval_requests") + op.drop_table("agents") diff --git a/pyproject.toml b/pyproject.toml index 6913617..7720088 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,8 +9,11 @@ description = "A reliability and governance control plane for production AI agen readme = "README.md" requires-python = ">=3.11" dependencies = [ + "alembic>=1.16,<2", "fastapi>=0.116,<1", + "psycopg[binary]>=3.2,<4", "pydantic>=2.11,<3", + "sqlalchemy>=2.0,<3", "uvicorn[standard]>=0.35,<1", ] @@ -34,6 +37,7 @@ packages = ["src/agent_control_plane"] addopts = "-ra --strict-markers" testpaths = ["tests"] markers = [ + "integration: tests that require external infrastructure", "smoke: fast end-to-end checks of the packaged service surface", ] diff --git a/src/agent_control_plane/api.py b/src/agent_control_plane/api.py index e2e33d8..6a2908c 100644 --- a/src/agent_control_plane/api.py +++ b/src/agent_control_plane/api.py @@ -1,11 +1,14 @@ """HTTP surface for the control-plane contract.""" +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager from typing import Annotated, NoReturn from uuid import UUID from fastapi import FastAPI, HTTPException, Query, status from agent_control_plane import __version__ +from agent_control_plane.bootstrap import create_store_from_environment from agent_control_plane.models import ( AgentRecord, AgentRegistrationRequest, @@ -45,10 +48,17 @@ def _raise_http_error(status_code: int, code: str, error: Exception) -> NoReturn def create_app(store: ControlPlaneStore | None = None) -> FastAPI: control_plane = store if store is not None else InMemoryControlPlaneStore() + + @asynccontextmanager + async def lifespan(_: FastAPI) -> AsyncIterator[None]: + yield + control_plane.close() + application = FastAPI( title="Agent Control Plane", description="Reliability and governance APIs for production AI agents.", version=__version__, + lifespan=lifespan, ) @application.get("/health/live", response_model=HealthResponse, tags=["health"]) @@ -56,8 +66,13 @@ async def liveness() -> HealthResponse: return HealthResponse(status=HealthStatus.OK, service=SERVICE_NAME, version=__version__) @application.get("/health/ready", response_model=HealthResponse, tags=["health"]) - async def readiness() -> HealthResponse: - # Dependency checks will be added here as durable stores are introduced. + def readiness() -> HealthResponse: + if not control_plane.is_ready(): + _raise_http_error( + status.HTTP_503_SERVICE_UNAVAILABLE, + "store_unavailable", + RuntimeError("control-plane store is unavailable or not migrated"), + ) return HealthResponse(status=HealthStatus.OK, service=SERVICE_NAME, version=__version__) @application.post( @@ -78,18 +93,18 @@ async def validate_agent_spec(spec: AgentSpec) -> AgentSpecValidationResponse: status_code=status.HTTP_201_CREATED, tags=["agents"], ) - async def register_agent(request: AgentRegistrationRequest) -> AgentRecord: + def register_agent(request: AgentRegistrationRequest) -> AgentRecord: try: return control_plane.register_agent(request) except AgentAlreadyExistsError as error: _raise_http_error(status.HTTP_409_CONFLICT, "agent_already_exists", error) @application.get("/v1/agents", response_model=list[AgentRecord], tags=["agents"]) - async def list_agents() -> tuple[AgentRecord, ...]: + def list_agents() -> tuple[AgentRecord, ...]: return control_plane.list_agents() @application.get("/v1/agents/{agent_id}", response_model=AgentRecord, tags=["agents"]) - async def get_agent(agent_id: str) -> AgentRecord: + def get_agent(agent_id: str) -> AgentRecord: try: return control_plane.get_agent(agent_id) except AgentNotFoundError as error: @@ -100,7 +115,7 @@ async def get_agent(agent_id: str) -> AgentRecord: response_model=AgentRecord, tags=["agents"], ) - async def update_agent_status(agent_id: str, update: AgentStatusUpdate) -> AgentRecord: + def update_agent_status(agent_id: str, update: AgentStatusUpdate) -> AgentRecord: try: return control_plane.update_agent_status(agent_id, update) except AgentNotFoundError as error: @@ -116,7 +131,7 @@ async def update_agent_status(agent_id: str, update: AgentStatusUpdate) -> Agent status_code=status.HTTP_201_CREATED, tags=["approvals"], ) - async def create_approval(request: ApprovalRequestCreate) -> ApprovalRecord: + def create_approval(request: ApprovalRequestCreate) -> ApprovalRecord: try: return control_plane.create_approval(request) except AgentNotFoundError as error: @@ -127,14 +142,14 @@ async def create_approval(request: ApprovalRequestCreate) -> ApprovalRecord: @application.get( "/v1/approvals/{request_id}", response_model=ApprovalRecord, tags=["approvals"] ) - async def get_approval(request_id: UUID) -> ApprovalRecord: + def get_approval(request_id: UUID) -> ApprovalRecord: try: return control_plane.get_approval(request_id) except ApprovalNotFoundError as error: _raise_http_error(status.HTTP_404_NOT_FOUND, "approval_not_found", error) @application.get("/v1/approvals", response_model=ApprovalQueueResponse, tags=["approvals"]) - async def list_approvals( + def list_approvals( approval_status: Annotated[ApprovalStatus | None, Query(alias="status")] = None, agent_id: str | None = None, ) -> ApprovalQueueResponse: @@ -146,9 +161,7 @@ async def list_approvals( response_model=ApprovalRecord, tags=["approvals"], ) - async def decide_approval( - request_id: UUID, decision: ApprovalDecisionRequest - ) -> ApprovalRecord: + def decide_approval(request_id: UUID, decision: ApprovalDecisionRequest) -> ApprovalRecord: try: return control_plane.decide_approval(request_id, decision) except ApprovalNotFoundError as error: @@ -157,7 +170,7 @@ async def decide_approval( _raise_http_error(status.HTTP_409_CONFLICT, "approval_already_decided", error) @application.get("/v1/audit-events", response_model=AuditEventPage, tags=["audit"]) - async def list_audit_events( + def list_audit_events( agent_id: str | None = None, limit: Annotated[int, Query(ge=1, le=500)] = 100, ) -> AuditEventPage: @@ -167,4 +180,4 @@ async def list_audit_events( return application -app = create_app() +app = create_app(create_store_from_environment()) diff --git a/src/agent_control_plane/bootstrap.py b/src/agent_control_plane/bootstrap.py new file mode 100644 index 0000000..b3402b7 --- /dev/null +++ b/src/agent_control_plane/bootstrap.py @@ -0,0 +1,21 @@ +"""Environment-driven adapter selection for the service process.""" + +import os +from collections.abc import Mapping + +from agent_control_plane.postgres_store import PostgresControlPlaneStore +from agent_control_plane.store import ControlPlaneStore, InMemoryControlPlaneStore + +DATABASE_URL_ENV = "ACP_DATABASE_URL" + + +def create_store_from_environment( + environment: Mapping[str, str] | None = None, +) -> ControlPlaneStore: + values = os.environ if environment is None else environment + database_url = values.get(DATABASE_URL_ENV) + if not database_url: + return InMemoryControlPlaneStore() + if not database_url.startswith(("postgresql://", "postgresql+psycopg://")): + raise ValueError(f"{DATABASE_URL_ENV} must be a PostgreSQL URL") + return PostgresControlPlaneStore(database_url) diff --git a/src/agent_control_plane/db_schema.py b/src/agent_control_plane/db_schema.py new file mode 100644 index 0000000..0dd58b8 --- /dev/null +++ b/src/agent_control_plane/db_schema.py @@ -0,0 +1,101 @@ +"""SQLAlchemy Core schema shared by the PostgreSQL adapter and migration tooling.""" + +from sqlalchemy import ( + JSON, + BigInteger, + CheckConstraint, + Column, + DateTime, + ForeignKey, + Identity, + Index, + Integer, + MetaData, + String, + Table, + Uuid, +) +from sqlalchemy.dialects import postgresql + +REQUIRED_SCHEMA_REVISION = "0001_initial" + +NAMING_CONVENTION = { + "ix": "ix_%(column_0_label)s", + "uq": "uq_%(table_name)s_%(column_0_name)s", + "ck": "ck_%(table_name)s_%(constraint_name)s", + "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s", + "pk": "pk_%(table_name)s", +} + +metadata = MetaData(naming_convention=NAMING_CONVENTION) +specification_type = JSON().with_variant(postgresql.JSONB(), "postgresql") + +agents = Table( + "agents", + metadata, + Column("agent_id", String(63), primary_key=True), + Column("spec", specification_type, nullable=False), + Column("status", String(20), nullable=False), + Column("revision", Integer, nullable=False), + Column("registered_at", DateTime(timezone=True), nullable=False), + Column("updated_at", DateTime(timezone=True), nullable=False), + CheckConstraint("revision >= 1", name="positive_revision"), + CheckConstraint("status IN ('registered', 'active', 'paused')", name="valid_status"), +) + +approval_requests = Table( + "approval_requests", + metadata, + Column("request_id", Uuid(as_uuid=True), primary_key=True), + Column( + "agent_id", + String(63), + ForeignKey("agents.agent_id", ondelete="RESTRICT"), + nullable=False, + ), + Column("action", String(128), nullable=False), + Column("risk", String(20), nullable=False), + Column("status", String(20), nullable=False), + Column("requested_by", String(128), nullable=False), + Column("request_reason", String(500), nullable=False), + Column("created_at", DateTime(timezone=True), nullable=False), + Column("decided_at", DateTime(timezone=True)), + Column("decided_by", String(128)), + Column("decision_reason", String(500)), + CheckConstraint("risk IN ('low', 'medium', 'high')", name="valid_risk"), + CheckConstraint("status IN ('pending', 'approved', 'rejected')", name="valid_status"), + CheckConstraint( + "(status = 'pending' AND decided_at IS NULL AND decided_by IS NULL " + "AND decision_reason IS NULL) OR " + "(status IN ('approved', 'rejected') AND decided_at IS NOT NULL " + "AND decided_by IS NOT NULL AND decision_reason IS NOT NULL)", + name="decision_consistency", + ), +) +Index("ix_approval_requests_queue", approval_requests.c.status, approval_requests.c.created_at) +Index("ix_approval_requests_agent", approval_requests.c.agent_id) + +audit_sequence_type = BigInteger().with_variant(Integer, "sqlite") +audit_events = Table( + "audit_events", + metadata, + Column("sequence", audit_sequence_type, Identity(), primary_key=True), + Column("event_id", Uuid(as_uuid=True), nullable=False, unique=True), + Column( + "agent_id", + String(63), + ForeignKey("agents.agent_id", ondelete="RESTRICT"), + nullable=False, + ), + Column("event_type", String(40), nullable=False), + Column("actor", String(128), nullable=False), + Column("occurred_at", DateTime(timezone=True), nullable=False), + Column("resource_id", String(128), nullable=False), + Column("summary", String(1000), nullable=False), + CheckConstraint( + "event_type IN ('agent.registered', 'agent.status_changed', " + "'approval.requested', 'approval.approved', 'approval.rejected')", + name="valid_event_type", + ), +) +Index("ix_audit_events_agent_sequence", audit_events.c.agent_id, audit_events.c.sequence) diff --git a/src/agent_control_plane/postgres_store.py b/src/agent_control_plane/postgres_store.py new file mode 100644 index 0000000..7a77cd5 --- /dev/null +++ b/src/agent_control_plane/postgres_store.py @@ -0,0 +1,395 @@ +"""Transactional SQL store used with PostgreSQL in durable deployments.""" + +from collections.abc import Callable +from datetime import UTC, datetime +from typing import overload +from uuid import UUID, uuid4 + +from sqlalchemy import Engine, create_engine, insert, inspect, select, text, update +from sqlalchemy.engine import Connection, RowMapping +from sqlalchemy.exc import IntegrityError, SQLAlchemyError + +from agent_control_plane.db_schema import ( + REQUIRED_SCHEMA_REVISION, + agents, + approval_requests, + audit_events, +) +from agent_control_plane.models import ( + AgentRecord, + AgentRegistrationRequest, + AgentRuntimeStatus, + AgentSpec, + AgentStatusUpdate, + ApprovalDecision, + ApprovalDecisionRequest, + ApprovalRecord, + ApprovalRequestCreate, + ApprovalStatus, + AuditEvent, + AuditEventType, +) +from agent_control_plane.store import ( + AgentAlreadyExistsError, + AgentNotFoundError, + ApprovalAlreadyDecidedError, + ApprovalNotFoundError, + RevisionConflictError, + utc_now, + validate_agent_is_active, + validate_status_transition, +) + + +class PostgresControlPlaneStore: + """Persist control-plane state and audit events in atomic SQL transactions.""" + + def __init__( + self, + database_url: str | None = None, + *, + engine: Engine | None = None, + clock: Callable[[], datetime] = utc_now, + id_factory: Callable[[], UUID] = uuid4, + ) -> None: + if (database_url is None) == (engine is None): + raise ValueError("provide exactly one of database_url or engine") + if engine is not None: + self._engine = engine + else: + assert database_url is not None + if database_url.startswith("postgresql://"): + database_url = database_url.replace("postgresql://", "postgresql+psycopg://", 1) + self._engine = create_engine(database_url, pool_pre_ping=True, pool_recycle=300) + self._clock = clock + self._id_factory = id_factory + + def is_ready(self) -> bool: + try: + with self._engine.connect() as connection: + inspector = inspect(connection) + required_tables = ("agents", "approval_requests", "audit_events") + if not all(inspector.has_table(name) for name in required_tables): + return False + revision = connection.execute( + text("SELECT version_num FROM alembic_version") + ).scalar_one_or_none() + return revision == REQUIRED_SCHEMA_REVISION + except SQLAlchemyError: + return False + + def close(self) -> None: + self._engine.dispose() + + def register_agent(self, request: AgentRegistrationRequest) -> AgentRecord: + timestamp = self._clock() + record = AgentRecord( + spec=request.spec, + status=AgentRuntimeStatus.REGISTERED, + revision=1, + registered_at=timestamp, + updated_at=timestamp, + ) + try: + with self._engine.begin() as connection: + connection.execute( + insert(agents).values( + agent_id=request.spec.agent_id, + spec=request.spec.model_dump(mode="json"), + status=record.status.value, + revision=record.revision, + registered_at=timestamp, + updated_at=timestamp, + ) + ) + self._insert_audit_event( + connection, + event_type=AuditEventType.AGENT_REGISTERED, + agent_id=request.spec.agent_id, + actor=request.actor, + occurred_at=timestamp, + resource_id=request.spec.agent_id, + summary=f"Registered agent version {request.spec.version}", + ) + except IntegrityError as error: + raise AgentAlreadyExistsError( + f"agent '{request.spec.agent_id}' is already registered" + ) from error + return record + + def get_agent(self, agent_id: str) -> AgentRecord: + with self._engine.connect() as connection: + row = self._get_agent_row(connection, agent_id) + if row is None: + raise AgentNotFoundError(f"agent '{agent_id}' was not found") + return self._agent_from_row(row) + + def list_agents(self) -> tuple[AgentRecord, ...]: + with self._engine.connect() as connection: + rows = connection.execute(select(agents).order_by(agents.c.agent_id)).mappings() + return tuple(self._agent_from_row(row) for row in rows) + + def update_agent_status(self, agent_id: str, update_request: AgentStatusUpdate) -> AgentRecord: + with self._engine.begin() as connection: + row = self._get_agent_row(connection, agent_id, for_update=True) + if row is None: + raise AgentNotFoundError(f"agent '{agent_id}' was not found") + current = self._agent_from_row(row) + if current.revision != update_request.expected_revision: + raise RevisionConflictError( + f"expected revision {update_request.expected_revision}, current revision is " + f"{current.revision}" + ) + validate_status_transition(current.status, update_request.status) + + timestamp = self._clock() + result = connection.execute( + update(agents) + .where( + agents.c.agent_id == agent_id, + agents.c.revision == update_request.expected_revision, + ) + .values( + status=update_request.status.value, + revision=current.revision + 1, + updated_at=timestamp, + ) + ) + if result.rowcount != 1: + raise RevisionConflictError("agent revision changed during the update") + self._insert_audit_event( + connection, + event_type=AuditEventType.AGENT_STATUS_CHANGED, + agent_id=agent_id, + actor=update_request.actor, + occurred_at=timestamp, + resource_id=agent_id, + summary=( + f"Changed status from {current.status} to {update_request.status}: " + f"{update_request.reason}" + ), + ) + return current.model_copy( + update={ + "status": update_request.status, + "revision": current.revision + 1, + "updated_at": timestamp, + } + ) + + def create_approval(self, request: ApprovalRequestCreate) -> ApprovalRecord: + with self._engine.begin() as connection: + agent_row = self._get_agent_row(connection, request.agent_id, for_update=True) + if agent_row is None: + raise AgentNotFoundError(f"agent '{request.agent_id}' was not found") + validate_agent_is_active(self._agent_from_row(agent_row)) + + timestamp = self._clock() + request_id = self._id_factory() + record = ApprovalRecord( + request_id=request_id, + agent_id=request.agent_id, + action=request.action, + risk=request.risk, + status=ApprovalStatus.PENDING, + requested_by=request.actor, + request_reason=request.reason, + created_at=timestamp, + ) + connection.execute(insert(approval_requests).values(**record.model_dump(mode="python"))) + self._insert_audit_event( + connection, + event_type=AuditEventType.APPROVAL_REQUESTED, + agent_id=request.agent_id, + actor=request.actor, + occurred_at=timestamp, + resource_id=str(request_id), + summary=f"Requested {request.risk} approval for {request.action}", + ) + return record + + def get_approval(self, request_id: UUID) -> ApprovalRecord: + with self._engine.connect() as connection: + row = self._get_approval_row(connection, request_id) + if row is None: + raise ApprovalNotFoundError(f"approval request '{request_id}' was not found") + return self._approval_from_row(row) + + def list_approvals( + self, status: ApprovalStatus | None = None, agent_id: str | None = None + ) -> tuple[ApprovalRecord, ...]: + statement = select(approval_requests) + if status is not None: + statement = statement.where(approval_requests.c.status == status.value) + if agent_id is not None: + statement = statement.where(approval_requests.c.agent_id == agent_id) + statement = statement.order_by( + approval_requests.c.created_at, approval_requests.c.request_id + ) + with self._engine.connect() as connection: + rows = connection.execute(statement).mappings() + return tuple(self._approval_from_row(row) for row in rows) + + def decide_approval( + self, request_id: UUID, decision: ApprovalDecisionRequest + ) -> ApprovalRecord: + with self._engine.begin() as connection: + row = self._get_approval_row(connection, request_id, for_update=True) + if row is None: + raise ApprovalNotFoundError(f"approval request '{request_id}' was not found") + current = self._approval_from_row(row) + if current.status is not ApprovalStatus.PENDING: + raise ApprovalAlreadyDecidedError( + f"approval request '{request_id}' has already been decided" + ) + + timestamp = self._clock() + requested_status = ( + ApprovalStatus.APPROVED + if decision.decision is ApprovalDecision.APPROVE + else ApprovalStatus.REJECTED + ) + result = connection.execute( + update(approval_requests) + .where( + approval_requests.c.request_id == request_id, + approval_requests.c.status == ApprovalStatus.PENDING.value, + ) + .values( + status=requested_status.value, + decided_at=timestamp, + decided_by=decision.actor, + decision_reason=decision.reason, + ) + ) + if result.rowcount != 1: + raise ApprovalAlreadyDecidedError( + f"approval request '{request_id}' was decided concurrently" + ) + self._insert_audit_event( + connection, + event_type=( + AuditEventType.APPROVAL_APPROVED + if requested_status is ApprovalStatus.APPROVED + else AuditEventType.APPROVAL_REJECTED + ), + agent_id=current.agent_id, + actor=decision.actor, + occurred_at=timestamp, + resource_id=str(request_id), + summary=( + f"{requested_status.value.capitalize()} {current.action}: {decision.reason}" + ), + ) + return current.model_copy( + update={ + "status": requested_status, + "decided_at": timestamp, + "decided_by": decision.actor, + "decision_reason": decision.reason, + } + ) + + def list_audit_events( + self, agent_id: str | None = None, limit: int = 100 + ) -> tuple[AuditEvent, ...]: + statement = select(audit_events) + if agent_id is not None: + statement = statement.where(audit_events.c.agent_id == agent_id) + statement = statement.order_by(audit_events.c.sequence.desc()).limit(limit) + with self._engine.connect() as connection: + rows = connection.execute(statement).mappings() + return tuple(self._audit_from_row(row) for row in rows) + + @staticmethod + def _get_agent_row( + connection: Connection, agent_id: str, *, for_update: bool = False + ) -> RowMapping | None: + statement = select(agents).where(agents.c.agent_id == agent_id) + if for_update: + statement = statement.with_for_update() + return connection.execute(statement).mappings().one_or_none() + + @staticmethod + def _get_approval_row( + connection: Connection, request_id: UUID, *, for_update: bool = False + ) -> RowMapping | None: + statement = select(approval_requests).where(approval_requests.c.request_id == request_id) + if for_update: + statement = statement.with_for_update() + return connection.execute(statement).mappings().one_or_none() + + def _insert_audit_event( + self, + connection: Connection, + *, + event_type: AuditEventType, + agent_id: str, + actor: str, + occurred_at: datetime, + resource_id: str, + summary: str, + ) -> None: + connection.execute( + insert(audit_events).values( + event_id=self._id_factory(), + event_type=event_type.value, + agent_id=agent_id, + actor=actor, + occurred_at=occurred_at, + resource_id=resource_id, + summary=summary, + ) + ) + + @staticmethod + def _agent_from_row(row: RowMapping) -> AgentRecord: + return AgentRecord( + spec=AgentSpec.model_validate(row["spec"]), + status=AgentRuntimeStatus(row["status"]), + revision=row["revision"], + registered_at=_ensure_timezone(row["registered_at"]), + updated_at=_ensure_timezone(row["updated_at"]), + ) + + @staticmethod + def _approval_from_row(row: RowMapping) -> ApprovalRecord: + return ApprovalRecord( + request_id=row["request_id"], + agent_id=row["agent_id"], + action=row["action"], + risk=row["risk"], + status=row["status"], + requested_by=row["requested_by"], + request_reason=row["request_reason"], + created_at=_ensure_timezone(row["created_at"]), + decided_at=_ensure_timezone(row["decided_at"]), + decided_by=row["decided_by"], + decision_reason=row["decision_reason"], + ) + + @staticmethod + def _audit_from_row(row: RowMapping) -> AuditEvent: + return AuditEvent( + event_id=row["event_id"], + event_type=row["event_type"], + agent_id=row["agent_id"], + actor=row["actor"], + occurred_at=_ensure_timezone(row["occurred_at"]), + resource_id=row["resource_id"], + summary=row["summary"], + ) + + +@overload +def _ensure_timezone(value: datetime) -> datetime: ... + + +@overload +def _ensure_timezone(value: None) -> None: ... + + +def _ensure_timezone(value: datetime | None) -> datetime | None: + if value is not None and value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value diff --git a/src/agent_control_plane/store.py b/src/agent_control_plane/store.py index 49f623d..7f1fd82 100644 --- a/src/agent_control_plane/store.py +++ b/src/agent_control_plane/store.py @@ -54,6 +54,10 @@ class ApprovalAlreadyDecidedError(StoreError): class ControlPlaneStore(Protocol): + def is_ready(self) -> bool: ... + + def close(self) -> None: ... + def register_agent(self, request: AgentRegistrationRequest) -> AgentRecord: ... def get_agent(self, agent_id: str) -> AgentRecord: ... @@ -83,15 +87,30 @@ def utc_now() -> datetime: return datetime.now(UTC) +ALLOWED_STATUS_TRANSITIONS = { + AgentRuntimeStatus.REGISTERED: {AgentRuntimeStatus.ACTIVE, AgentRuntimeStatus.PAUSED}, + AgentRuntimeStatus.ACTIVE: {AgentRuntimeStatus.PAUSED}, + AgentRuntimeStatus.PAUSED: {AgentRuntimeStatus.ACTIVE}, +} + + +def validate_status_transition(current: AgentRuntimeStatus, requested: AgentRuntimeStatus) -> None: + if requested not in ALLOWED_STATUS_TRANSITIONS[current]: + raise InvalidStatusTransitionError( + f"cannot change agent status from '{current}' to '{requested}'" + ) + + +def validate_agent_is_active(record: AgentRecord) -> None: + if record.status is not AgentRuntimeStatus.ACTIVE: + raise AgentNotActiveError( + f"agent '{record.spec.agent_id}' must be active to request approval" + ) + + class InMemoryControlPlaneStore: """Concurrency-safe adapter for development and single-process evaluation.""" - _allowed_status_transitions = { - AgentRuntimeStatus.REGISTERED: {AgentRuntimeStatus.ACTIVE, AgentRuntimeStatus.PAUSED}, - AgentRuntimeStatus.ACTIVE: {AgentRuntimeStatus.PAUSED}, - AgentRuntimeStatus.PAUSED: {AgentRuntimeStatus.ACTIVE}, - } - def __init__( self, *, @@ -105,6 +124,12 @@ def __init__( self._audit_events: list[AuditEvent] = [] self._lock = RLock() + def is_ready(self) -> bool: + return True + + def close(self) -> None: + return None + def register_agent(self, request: AgentRegistrationRequest) -> AgentRecord: with self._lock: agent_id = request.spec.agent_id @@ -149,10 +174,7 @@ def update_agent_status(self, agent_id: str, update: AgentStatusUpdate) -> Agent f"expected revision {update.expected_revision}, current revision is " f"{current.revision}" ) - if update.status not in self._allowed_status_transitions[current.status]: - raise InvalidStatusTransitionError( - f"cannot change agent status from '{current.status}' to '{update.status}'" - ) + validate_status_transition(current.status, update.status) timestamp = self._clock() updated = current.model_copy( @@ -176,10 +198,7 @@ def update_agent_status(self, agent_id: str, update: AgentStatusUpdate) -> Agent def create_approval(self, request: ApprovalRequestCreate) -> ApprovalRecord: with self._lock: agent = self.get_agent(request.agent_id) - if agent.status is not AgentRuntimeStatus.ACTIVE: - raise AgentNotActiveError( - f"agent '{request.agent_id}' must be active to request approval" - ) + validate_agent_is_active(agent) timestamp = self._clock() request_id = self._id_factory() record = ApprovalRecord( diff --git a/tests/integration/test_postgres_integration.py b/tests/integration/test_postgres_integration.py new file mode 100644 index 0000000..e84ca98 --- /dev/null +++ b/tests/integration/test_postgres_integration.py @@ -0,0 +1,92 @@ +import os +from uuid import uuid4 + +import pytest +from sqlalchemy import create_engine, delete, text, update +from sqlalchemy.exc import DBAPIError + +from agent_control_plane.db_schema import audit_events +from agent_control_plane.models import ( + AgentRegistrationRequest, + AgentRuntimeStatus, + AgentSpec, + AgentStatusUpdate, + ApprovalDecision, + ApprovalDecisionRequest, + ApprovalRequestCreate, + ApprovalStatus, + RiskLevel, +) +from agent_control_plane.postgres_store import PostgresControlPlaneStore + +pytestmark = pytest.mark.integration + + +def database_url() -> str: + value = os.environ.get("ACP_DATABASE_URL") + if not value: + pytest.skip("ACP_DATABASE_URL is required for PostgreSQL integration tests") + return value + + +def test_postgres_persists_governance_state_and_blocks_audit_mutation() -> None: + url = database_url() + agent_id = f"postgres-{uuid4().hex[:12]}" + first_store = PostgresControlPlaneStore(url) + assert first_store.is_ready() is True + first_store.register_agent( + AgentRegistrationRequest( + spec=AgentSpec( + agent_id=agent_id, + version="1.0.0", + display_name="PostgreSQL Agent", + description="Verifies durable control-plane behavior.", + entrypoint="https://agents.example.test/postgres", + ), + actor="operator@example.test", + ) + ) + first_store.close() + + store = PostgresControlPlaneStore(url) + assert store.get_agent(agent_id).revision == 1 + store.update_agent_status( + agent_id, + AgentStatusUpdate( + status=AgentRuntimeStatus.ACTIVE, + expected_revision=1, + actor="operator@example.test", + reason="Integration readiness passed.", + ), + ) + pending = store.create_approval( + ApprovalRequestCreate( + agent_id=agent_id, + action="deployment.promote", + risk=RiskLevel.HIGH, + actor=agent_id, + reason="Promotion requires approval.", + ) + ) + approved = store.decide_approval( + pending.request_id, + ApprovalDecisionRequest( + decision=ApprovalDecision.APPROVE, + actor="reviewer@example.test", + reason="Integration evidence passed.", + ), + ) + assert approved.status is ApprovalStatus.APPROVED + assert len(store.list_audit_events(agent_id=agent_id)) == 4 + + engine = create_engine(url) + mutation_statements = ( + update(audit_events).values(summary="tampered"), + delete(audit_events), + text("TRUNCATE audit_events"), + ) + for statement in mutation_statements: + with pytest.raises(DBAPIError, match="append-only"), engine.begin() as connection: + connection.execute(statement) + engine.dispose() + store.close() diff --git a/tests/smoke/test_api_smoke.py b/tests/smoke/test_api_smoke.py index 95f15b2..5cca05d 100644 --- a/tests/smoke/test_api_smoke.py +++ b/tests/smoke/test_api_smoke.py @@ -4,6 +4,7 @@ import pytest from agent_control_plane.api import create_app +from agent_control_plane.store import InMemoryControlPlaneStore @pytest.fixture @@ -30,6 +31,23 @@ async def test_service_is_live_and_ready(client: httpx.AsyncClient) -> None: assert ready_response.json()["status"] == "ok" +@pytest.mark.smoke +@pytest.mark.anyio +async def test_readiness_fails_when_the_store_is_unavailable() -> None: + class UnavailableStore(InMemoryControlPlaneStore): + def is_ready(self) -> bool: + return False + + transport = httpx.ASGITransport(app=create_app(UnavailableStore())) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as test_client: + live_response = await test_client.get("/health/live") + ready_response = await test_client.get("/health/ready") + + assert live_response.status_code == 200 + assert ready_response.status_code == 503 + assert ready_response.json()["detail"]["code"] == "store_unavailable" + + @pytest.mark.smoke @pytest.mark.anyio async def test_agent_spec_contract_is_reachable(client: httpx.AsyncClient) -> None: diff --git a/tests/unit/test_bootstrap.py b/tests/unit/test_bootstrap.py new file mode 100644 index 0000000..c931578 --- /dev/null +++ b/tests/unit/test_bootstrap.py @@ -0,0 +1,25 @@ +import pytest + +from agent_control_plane.bootstrap import create_store_from_environment +from agent_control_plane.postgres_store import PostgresControlPlaneStore +from agent_control_plane.store import InMemoryControlPlaneStore + + +def test_bootstrap_uses_memory_without_a_database_url() -> None: + store = create_store_from_environment({}) + + assert isinstance(store, InMemoryControlPlaneStore) + + +def test_bootstrap_builds_postgres_store_for_a_database_url() -> None: + store = create_store_from_environment( + {"ACP_DATABASE_URL": "postgresql+psycopg://user:password@localhost/database"} + ) + + assert isinstance(store, PostgresControlPlaneStore) + store.close() + + +def test_bootstrap_rejects_non_postgres_urls() -> None: + with pytest.raises(ValueError, match="must be a PostgreSQL URL"): + create_store_from_environment({"ACP_DATABASE_URL": "sqlite:///control-plane.db"}) diff --git a/tests/unit/test_postgres_store.py b/tests/unit/test_postgres_store.py new file mode 100644 index 0000000..e61b39f --- /dev/null +++ b/tests/unit/test_postgres_store.py @@ -0,0 +1,208 @@ +from datetime import UTC, datetime +from uuid import UUID + +import pytest +from sqlalchemy import Engine, create_engine, text +from sqlalchemy.pool import StaticPool + +from agent_control_plane.db_schema import REQUIRED_SCHEMA_REVISION, metadata +from agent_control_plane.models import ( + AgentRegistrationRequest, + AgentRuntimeStatus, + AgentSpec, + AgentStatusUpdate, + ApprovalDecision, + ApprovalDecisionRequest, + ApprovalRequestCreate, + ApprovalStatus, + AuditEventType, + RiskLevel, +) +from agent_control_plane.postgres_store import PostgresControlPlaneStore +from agent_control_plane.store import ( + AgentAlreadyExistsError, + AgentNotActiveError, + AgentNotFoundError, + ApprovalAlreadyDecidedError, + ApprovalNotFoundError, + InvalidStatusTransitionError, + RevisionConflictError, +) + +NOW = datetime(2026, 8, 11, 12, 0, tzinfo=UTC) + + +def build_store() -> PostgresControlPlaneStore: + engine = create_engine( + "sqlite+pysqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + metadata.create_all(engine) + stamp_test_schema(engine) + return PostgresControlPlaneStore(engine=engine, clock=lambda: NOW) + + +def stamp_test_schema(engine: Engine) -> None: + with engine.begin() as connection: + connection.execute(text("CREATE TABLE alembic_version (version_num VARCHAR(32))")) + connection.execute( + text("INSERT INTO alembic_version (version_num) VALUES (:revision)"), + {"revision": REQUIRED_SCHEMA_REVISION}, + ) + + +def registration(agent_id: str = "sql-agent") -> AgentRegistrationRequest: + return AgentRegistrationRequest( + spec=AgentSpec( + agent_id=agent_id, + version="1.0.0", + display_name="SQL Agent", + description="Exercises the transactional store.", + entrypoint="https://agents.example.test/sql", + capabilities=("ticket.read", "ticket.reply"), + ), + actor="operator@example.test", + ) + + +def status_update(status: AgentRuntimeStatus, expected_revision: int = 1) -> AgentStatusUpdate: + return AgentStatusUpdate( + status=status, + expected_revision=expected_revision, + actor="operator@example.test", + reason="Exercise status transitions.", + ) + + +def approval_request(agent_id: str = "sql-agent") -> ApprovalRequestCreate: + return ApprovalRequestCreate( + agent_id=agent_id, + action="ticket.refund", + risk=RiskLevel.HIGH, + actor=agent_id, + reason="Refund requires review.", + ) + + +def test_constructor_requires_exactly_one_connection_source() -> None: + with pytest.raises(ValueError, match="exactly one"): + PostgresControlPlaneStore() + + engine = create_engine("sqlite+pysqlite:///:memory:") + with pytest.raises(ValueError, match="exactly one"): + PostgresControlPlaneStore("sqlite+pysqlite:///:memory:", engine=engine) + engine.dispose() + + +def test_constructor_uses_psycopg_for_a_bare_postgres_url() -> None: + store = PostgresControlPlaneStore("postgresql://user:password@localhost/database") + + assert store._engine.url.drivername == "postgresql+psycopg" + store.close() + + +def test_readiness_tracks_required_schema() -> None: + engine = create_engine("sqlite+pysqlite:///:memory:", poolclass=StaticPool) + store = PostgresControlPlaneStore(engine=engine) + assert store.is_ready() is False + + metadata.create_all(engine) + assert store.is_ready() is False + stamp_test_schema(engine) + assert store.is_ready() is True + store.close() + + +def test_agent_lifecycle_is_persisted_and_conflicts_fail_closed() -> None: + store = build_store() + registered = store.register_agent(registration()) + + assert registered.status is AgentRuntimeStatus.REGISTERED + assert store.get_agent("sql-agent") == registered + assert [item.spec.agent_id for item in store.list_agents()] == ["sql-agent"] + with pytest.raises(AgentAlreadyExistsError): + store.register_agent(registration()) + with pytest.raises(AgentNotFoundError): + store.get_agent("missing-agent") + with pytest.raises(RevisionConflictError): + store.update_agent_status("sql-agent", status_update(AgentRuntimeStatus.ACTIVE, 2)) + with pytest.raises(InvalidStatusTransitionError): + store.update_agent_status("sql-agent", status_update(AgentRuntimeStatus.REGISTERED)) + + active = store.update_agent_status("sql-agent", status_update(AgentRuntimeStatus.ACTIVE)) + assert active.revision == 2 + assert active.status is AgentRuntimeStatus.ACTIVE + store.close() + + +def test_approval_lifecycle_is_transactional_and_filterable() -> None: + store = build_store() + store.register_agent(registration()) + with pytest.raises(AgentNotActiveError): + store.create_approval(approval_request()) + with pytest.raises(AgentNotFoundError): + store.create_approval(approval_request("missing-agent")) + store.update_agent_status("sql-agent", status_update(AgentRuntimeStatus.ACTIVE)) + + pending = store.create_approval(approval_request()) + assert store.get_approval(pending.request_id) == pending + assert store.list_approvals(status=ApprovalStatus.PENDING) == (pending,) + assert store.list_approvals(agent_id="missing-agent") == () + + approved = store.decide_approval( + pending.request_id, + ApprovalDecisionRequest( + decision=ApprovalDecision.APPROVE, + actor="reviewer@example.test", + reason="Evidence verified.", + ), + ) + assert approved.status is ApprovalStatus.APPROVED + with pytest.raises(ApprovalAlreadyDecidedError): + store.decide_approval( + pending.request_id, + ApprovalDecisionRequest( + decision=ApprovalDecision.REJECT, + actor="reviewer@example.test", + reason="A second decision is forbidden.", + ), + ) + with pytest.raises(ApprovalNotFoundError): + store.get_approval(UUID(int=0)) + with pytest.raises(ApprovalNotFoundError): + store.decide_approval( + UUID(int=0), + ApprovalDecisionRequest( + decision=ApprovalDecision.REJECT, + actor="reviewer@example.test", + reason="Unknown request.", + ), + ) + + events = store.list_audit_events(agent_id="sql-agent", limit=2) + assert [event.event_type for event in events] == [ + AuditEventType.APPROVAL_APPROVED, + AuditEventType.APPROVAL_REQUESTED, + ] + store.close() + + +def test_rejection_path_is_persisted() -> None: + store = build_store() + store.register_agent(registration()) + store.update_agent_status("sql-agent", status_update(AgentRuntimeStatus.ACTIVE)) + pending = store.create_approval(approval_request()) + + rejected = store.decide_approval( + pending.request_id, + ApprovalDecisionRequest( + decision=ApprovalDecision.REJECT, + actor="reviewer@example.test", + reason="Evidence missing.", + ), + ) + + assert rejected.status is ApprovalStatus.REJECTED + assert store.list_audit_events(limit=1)[0].event_type is AuditEventType.APPROVAL_REJECTED + store.close() diff --git a/tests/unit/test_store.py b/tests/unit/test_store.py index 1ab2e66..0a08a61 100644 --- a/tests/unit/test_store.py +++ b/tests/unit/test_store.py @@ -76,6 +76,7 @@ def approval_request(agent_id: str = "support-agent") -> ApprovalRequestCreate: def test_register_agent_creates_initial_state_and_audit_event() -> None: store = build_store() + assert store.is_ready() is True record = store.register_agent(registration()) assert record.status is AgentRuntimeStatus.REGISTERED @@ -85,6 +86,7 @@ def test_register_agent_creates_initial_state_and_audit_event() -> None: event = store.list_audit_events()[0] assert event.event_type is AuditEventType.AGENT_REGISTERED assert event.actor == "operator@example.test" + store.close() def test_duplicate_agent_registration_is_rejected() -> None: