feat: add contracts view backend endpoints + SSE event

- POST /api/v1/bundles/evaluate — playground endpoint for testing
  contracts in dashboard (dashboard auth only, never called by agents)
- GET /api/v1/stats/contracts — per-contract coverage stats with
  time-window filtering (since/until)
- GET /api/v1/deployments — list deployments with env filter + limit
- SSE bundle_uploaded event fired on bundle upload
- Amend CLAUDE.md Principle #2 to document evaluate exception
- 32 new tests (12 evaluate, 3 stats, 4 deployments, 1 SSE,
  10 adversarial security, 2 tenant isolation)
- 140/140 total tests pass, 0 regressions

Author: Arnold Cartagena

CLAUDE.md

@@ -62,7 +62,7 @@ Svelte 5 runes are too new — LLMs still confuse Svelte 3/4/5 syntax and produc
 ## Non-Negotiable Principles
 
 1. **edictum core works without the server.** The library is standalone. Server is an optional enhancement. Never introduce a server dependency into the core library.
-2. **All governance runs in the agent process.** The server NEVER evaluates contracts. Zero latency on tool calls. Server stores events, manages approvals, pushes contract updates.
+2. **All governance runs in the agent process.** The server NEVER evaluates contracts in production. Zero latency on tool calls. Server stores events, manages approvals, pushes contract updates. **Exception:** `POST /api/v1/bundles/evaluate` is a development-time playground endpoint for testing contracts in the dashboard. It is never called by agents. Production evaluation remains agent-side only.
 3. **Fail closed.** Server unreachable → errors propagate → deny. Never fail open.
 4. **Single Docker image.** FastAPI serves the SPA at `/dashboard`, API at `/api/v1/*`, marketing/landing at `/`. One deployment.
 5. **Local-first auth.** `EDICTUM_AUTH_PROVIDER=local` is the default. No external auth dependency required.

src/edictum_server/main.py

@@ -22,6 +22,8 @@
     approvals,
     auth,
     bundles,
+    deployments,
+    evaluate,
     events,
     health,
     keys,
@@ -217,6 +219,8 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
 app.include_router(auth.router)
 app.include_router(keys.router)
 app.include_router(bundles.router)
+app.include_router(evaluate.router)
+app.include_router(deployments.router)
 app.include_router(stream.router)
 app.include_router(events.router)
 app.include_router(sessions.router)

src/edictum_server/push/manager.py

@@ -14,6 +14,7 @@
     "approval_created",
     "approval_decided",
     "approval_timeout",
+    "bundle_uploaded",
     "contract_update",
 })
 

src/edictum_server/routes/bundles.py

@@ -53,6 +53,7 @@ async def upload(
     body: BundleUploadRequest,
     auth: AuthContext = Depends(require_dashboard_auth),
     db: AsyncSession = Depends(get_db),
+    push: PushManager = Depends(get_push_manager),
 ) -> BundleResponse:
     """Upload a new contract bundle (dashboard-authenticated users)."""
     try:
@@ -68,6 +69,13 @@ async def upload(
     except ValueError as exc:
         raise HTTPException(status_code=422, detail=str(exc)) from exc
 
+    push.push_to_dashboard(auth.tenant_id, {
+        "type": "bundle_uploaded",
+        "version": bundle.version,
+        "revision_hash": bundle.revision_hash,
+        "uploaded_by": auth.user_id or "unknown",
+    })
+
     return _bundle_to_response(bundle)
 
 

src/edictum_server/routes/deployments.py

@@ -0,0 +1,35 @@
+"""Deployment listing endpoint -- ``GET /api/v1/deployments``."""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from edictum_server.auth.dependencies import AuthContext, require_dashboard_auth
+from edictum_server.db.engine import get_db
+from edictum_server.db.models import Deployment
+from edictum_server.schemas.bundles import DeploymentResponse
+
+router = APIRouter(prefix="/api/v1/deployments", tags=["deployments"])
+
+
+@router.get("", response_model=list[DeploymentResponse], summary="List deployments")
+async def list_deployments(
+    env: str | None = Query(default=None, description="Filter by environment"),
+    limit: int = Query(default=50, ge=1, le=200, description="Max results"),
+    auth: AuthContext = Depends(require_dashboard_auth),
+    db: AsyncSession = Depends(get_db),
+) -> list[DeploymentResponse]:
+    """List deployments for the tenant, newest first."""
+    stmt = (
+        select(Deployment)
+        .where(Deployment.tenant_id == auth.tenant_id)
+        .order_by(Deployment.created_at.desc())
+        .limit(limit)
+    )
+    if env is not None:
+        stmt = stmt.where(Deployment.env == env)
+
+    result = await db.execute(stmt)
+    return [DeploymentResponse.model_validate(d) for d in result.scalars().all()]

src/edictum_server/routes/evaluate.py

@@ -0,0 +1,36 @@
+"""Evaluate (playground) endpoint — stateless contract evaluation."""
+
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException
+
+from edictum_server.auth.dependencies import (
+    AuthContext,
+    require_dashboard_auth,
+)
+from edictum_server.schemas.evaluate import EvaluateRequest, EvaluateResponse
+from edictum_server.services.evaluate_service import evaluate_contracts
+
+router = APIRouter(prefix="/api/v1/bundles", tags=["evaluate"])
+
+
+@router.post("/evaluate", response_model=EvaluateResponse)
+async def evaluate(
+    body: EvaluateRequest,
+    auth: AuthContext = Depends(require_dashboard_auth),
+) -> EvaluateResponse:
+    """Evaluate a tool call against YAML contracts (dashboard playground).
+
+    This is a development-time endpoint for testing contracts in the dashboard.
+    It is never called by agents during production execution.
+    """
+    try:
+        return evaluate_contracts(
+            yaml_content=body.yaml_content,
+            tool_name=body.tool_name,
+            tool_args=body.tool_args,
+            environment=body.environment,
+            principal_input=body.principal,
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc

src/edictum_server/routes/stats.py

@@ -1,14 +1,16 @@
-"""Dashboard statistics endpoint -- ``GET /api/v1/stats/overview``."""
+"""Dashboard statistics endpoints."""
 
 from __future__ import annotations
 
-from fastapi import APIRouter, Depends
+from datetime import UTC, datetime, timedelta
+
+from fastapi import APIRouter, Depends, Query
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from edictum_server.auth.dependencies import AuthContext, get_current_tenant
+from edictum_server.auth.dependencies import AuthContext, get_current_tenant, require_dashboard_auth
 from edictum_server.db.engine import get_db
-from edictum_server.schemas.stats import StatsOverviewResponse
-from edictum_server.services.stats_service import get_overview
+from edictum_server.schemas.stats import ContractStatsResponse, StatsOverviewResponse
+from edictum_server.services.stats_service import get_contract_stats, get_overview
 
 router = APIRouter(prefix="/api/v1/stats", tags=["stats"])
 
@@ -24,3 +26,21 @@ async def stats_overview(
 ) -> StatsOverviewResponse:
     """Return aggregate stats for the dashboard home view."""
     return await get_overview(db, auth.tenant_id)
+
+
+@router.get(
+    "/contracts",
+    response_model=ContractStatsResponse,
+    summary="Get per-contract evaluation statistics",
+)
+async def contract_stats(
+    since: datetime | None = Query(default=None, description="Start of time window (ISO 8601)"),
+    until: datetime | None = Query(default=None, description="End of time window (ISO 8601)"),
+    auth: AuthContext = Depends(require_dashboard_auth),
+    db: AsyncSession = Depends(get_db),
+) -> ContractStatsResponse:
+    """Return per-contract aggregated evaluation stats."""
+    now = datetime.now(UTC)
+    effective_until = until or now
+    effective_since = since or (now - timedelta(hours=24))
+    return await get_contract_stats(db, auth.tenant_id, effective_since, effective_until)

src/edictum_server/schemas/evaluate.py

@@ -0,0 +1,51 @@
+"""Request and response schemas for the evaluate (playground) endpoint."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+class PrincipalInput(BaseModel):
+    """Optional principal context for evaluation."""
+
+    user_id: str | None = None
+    role: str | None = None
+    claims: dict[str, Any] | None = None
+
+
+class EvaluateRequest(BaseModel):
+    """Request body for contract evaluation playground."""
+
+    yaml_content: str = Field(..., description="YAML contract bundle to evaluate against")
+    tool_name: str = Field(..., description="Tool name to simulate")
+    tool_args: dict[str, Any] = Field(default_factory=dict, description="Tool arguments")
+    environment: str = Field(default="production", description="Environment context")
+    agent_id: str = Field(default="test-agent", description="Simulated agent ID")
+    principal: PrincipalInput | None = Field(
+        default=None, description="Optional principal identity"
+    )
+
+
+class ContractEvaluation(BaseModel):
+    """Result of evaluating a single contract."""
+
+    id: str
+    type: str
+    matched: bool
+    effect: str | None = None
+    message: str | None = None
+    observed: bool = False
+    tags: list[str] = []
+
+
+class EvaluateResponse(BaseModel):
+    """Response from the evaluate playground endpoint."""
+
+    verdict: str
+    mode: str
+    contracts_evaluated: list[ContractEvaluation]
+    deciding_contract: str | None = None
+    policy_version: str
+    evaluation_time_ms: float

src/edictum_server/schemas/stats.py

@@ -1,4 +1,4 @@
-"""Schemas for the stats overview endpoint."""
+"""Schemas for the stats endpoints."""
 
 from __future__ import annotations
 
@@ -19,3 +19,22 @@ class StatsOverviewResponse(BaseModel):
     contracts_triggered_24h: int = Field(
         0, description="Distinct contracts triggered in the last 24 hours"
     )
+
+
+class ContractCoverage(BaseModel):
+    """Per-contract aggregated stats."""
+
+    decision_name: str = Field(..., description="Name of the contract")
+    total_evaluations: int = Field(..., description="Total evaluation events")
+    total_denials: int = Field(..., description="Events with verdict=denied")
+    total_warnings: int = Field(..., description="Events with verdict=call_would_deny")
+    last_triggered: str | None = Field(None, description="ISO timestamp of last event")
+
+
+class ContractStatsResponse(BaseModel):
+    """Contract-level statistics for a time window."""
+
+    coverage: list[ContractCoverage] = Field(default_factory=list)
+    total_events: int = Field(..., description="Total events in the time window")
+    period_start: str = Field(..., description="ISO timestamp of window start")
+    period_end: str = Field(..., description="ISO timestamp of window end")

src/edictum_server/services/evaluate_service.py

@@ -0,0 +1,113 @@
+"""Stateless contract evaluation service (development-time playground)."""
+
+from __future__ import annotations
+
+import hashlib
+import time
+from typing import Any
+
+from edictum import Edictum, EvaluationResult, Principal
+
+from edictum_server.schemas.evaluate import (
+    ContractEvaluation,
+    EvaluateResponse,
+    PrincipalInput,
+)
+
+
+def _build_principal(inp: PrincipalInput | None) -> Principal | None:
+    """Convert API input to an edictum Principal, or None."""
+    if inp is None:
+        return None
+    return Principal(
+        user_id=inp.user_id,
+        role=inp.role,
+        claims=inp.claims or {},
+    )
+
+
+def _map_result(result: EvaluationResult, mode: str, yaml_hash: str) -> EvaluateResponse:
+    """Map an edictum EvaluationResult to the API response schema."""
+    contracts = [
+        ContractEvaluation(
+            id=cr.contract_id,
+            type=cr.contract_type,
+            matched=cr.passed,
+            effect=cr.effect,
+            message=cr.message,
+            observed=cr.observed,
+            tags=list(cr.tags),
+        )
+        for cr in result.contracts
+    ]
+
+    # The deciding contract is the first failing non-observed contract
+    deciding: str | None = None
+    for cr in result.contracts:
+        if not cr.passed and not cr.observed:
+            deciding = cr.contract_id
+            break
+
+    return EvaluateResponse(
+        verdict=result.verdict,
+        mode=mode,
+        contracts_evaluated=contracts,
+        deciding_contract=deciding,
+        policy_version=yaml_hash[:12],
+        evaluation_time_ms=0.0,  # overwritten by caller
+    )
+
+
+def evaluate_contracts(
+    *,
+    yaml_content: str,
+    tool_name: str,
+    tool_args: dict[str, Any],
+    environment: str = "production",
+    principal_input: PrincipalInput | None = None,
+) -> EvaluateResponse:
+    """Evaluate a tool call against YAML contracts.
+
+    This is a stateless, synchronous operation for the dashboard playground.
+    It never persists data or touches the database.
+
+    Args:
+        yaml_content: Raw YAML contract bundle.
+        tool_name: Tool name to evaluate.
+        tool_args: Arguments for the tool call.
+        environment: Environment context (default "production").
+        principal_input: Optional principal identity.
+
+    Returns:
+        EvaluateResponse with verdict, matched contracts, timing.
+
+    Raises:
+        ValueError: If the YAML is invalid or cannot be parsed as contracts.
+    """
+    principal = _build_principal(principal_input)
+    yaml_hash = hashlib.sha256(yaml_content.encode("utf-8")).hexdigest()
+
+    start = time.monotonic()
+
+    try:
+        edictum_instance = Edictum.from_yaml_string(
+            yaml_content,
+            environment=environment,
+        )
+    except Exception as exc:
+        raise ValueError(f"Invalid contract YAML: {exc}") from exc
+
+    result: EvaluationResult = edictum_instance.evaluate(
+        tool_name,
+        tool_args,
+        principal=principal,
+    )
+
+    elapsed_ms = (time.monotonic() - start) * 1000
+
+    # Determine mode from the edictum instance
+    mode = getattr(edictum_instance, "mode", "enforce") or "enforce"
+
+    response = _map_result(result, mode, yaml_hash)
+    response.evaluation_time_ms = round(elapsed_ms, 2)
+    return response