From 71027f0a67bd4220fbc693f16c2027bd9069317d Mon Sep 17 00:00:00 2001 From: tegwick Date: Tue, 21 Jul 2026 23:51:39 +0200 Subject: [PATCH] Implement ACTIVITY-WP-0024 operator automation console Add /ops REST inventory, status, runs, and fail-closed operator-token mutations (trigger, enable/disable, pause/unpause) with audit trail. Ship thin HTML UI at /ops/ui, runbook/k8s access docs, and contract tests. --- SCOPE.md | 11 +- docs/runbook.md | 81 +++ k8s/railiance/README.md | 18 + src/activity_core/api.py | 5 + src/activity_core/ops_api.py | 477 ++++++++++++++++++ src/activity_core/ops_auth.py | 89 ++++ src/activity_core/ops_console.py | 441 ++++++++++++++++ src/activity_core/schedule_manager.py | 45 ++ tests/test_ops_auth.py | 58 +++ tests/test_ops_console_api.py | 248 +++++++++ ...ITY-WP-0024-operator-automation-console.md | 41 +- 11 files changed, 1494 insertions(+), 20 deletions(-) create mode 100644 src/activity_core/ops_api.py create mode 100644 src/activity_core/ops_auth.py create mode 100644 src/activity_core/ops_console.py create mode 100644 tests/test_ops_auth.py create mode 100644 tests/test_ops_console_api.py diff --git a/SCOPE.md b/SCOPE.md index a27b134..544f49e 100644 --- a/SCOPE.md +++ b/SCOPE.md @@ -84,9 +84,12 @@ The two evaluation modes: (load → resolve → evaluate → report → **log_run** → emit). Emit failures no longer skip `activity_runs` rows. - **REST admin API** (FastAPI): CRUD, manual trigger, admin sync, metrics. +- **Operator automation console** (ACTIVITY-WP-0024): `/ops` inventory, status + window, run history, Run now / enable / disable / pause with operator-token + auth (fail-closed), plus thin HTML UI at `/ops/ui`. - **Automation status surface**: `make automation-status` and `make prod-automation-status` / `scripts/prod_automation_status.sh` for - railiance01 evidence without LLM authority. + railiance01 evidence without LLM authority (same contract as `/ops/automations/status`). - **Operational runbook**: `docs/runbook.md`, emission boundary docs. --- @@ -103,7 +106,9 @@ The two evaluation modes: - **General ops control plane** — Kubernetes, SSH, tunnels, secret custody, OpenBao policy administration. - **Event broker / Temporal server hosting** — consume, do not own lifecycle. -- **End-user task UI**. +- **End-user task UI** — tracking human work items. The **operator automation + console** (`/ops`) is in scope; it manages ActivityDefinitions and schedules, + not task lifecycle. - **Coding assistant schedulers** as production authority. **Boundary note (side-effect resolvers):** A small set of shell/context @@ -208,7 +213,7 @@ Open product/policy workplan: **ACTIVITY-WP-0022** (IssueSink no-default-Forgejo | **G7. Credential delivery** | Low (residual) | **FORGEJO_TOKEN** via ESO `actcore-forgejo-admin` (WP-0023-T05, Ready). issue-core `GITEA_BACKEND_TOKEN` still 503 forgejo-inbox for path A rest — **issue-core owner** (WP-0023-T06). | | **G8. Live-images hygiene** | Medium (ops) | Multi-cluster `live-images-all.txt` must be refreshed after deploys or prune can delete live tags (incident 2026-07-21, restored). `scripts/refresh_live_images.sh` (T04). | | **G9. Evidence federation** | Low | Progress often lands on railiance01 edge/hub; workstation primary hub may not show the same feed without tunnel/outbox health. | -| **G10. API external access** | Low | ClusterIP-only; intentional until auth policy. | +| **G10. API external access** | Low | ClusterIP-only; ops mutations use operator token (WP-0024). Public Ingress / OIDC still deferred. | ### Drift risks diff --git a/docs/runbook.md b/docs/runbook.md index b6ce806..124a75a 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -43,11 +43,92 @@ ACTCORE_DB_URL=postgresql+asyncpg://actcore:actcore@localhost:5433/actcore \ |---------|-----| | Temporal Web UI | http://localhost:8080 | | REST API docs (Swagger) | http://localhost:8010/docs | +| Operator console UI | http://localhost:8010/ops/ui | +| Operator status JSON | http://localhost:8010/ops/automations/status?since=sunday | | NATS monitoring | http://localhost:8222 | | Prometheus metrics (worker) | http://localhost:9090/metrics | --- +## Operator automation console (ACTIVITY-WP-0024) + +Prefer the **ops console** over ad-hoc SSH/SQL for “did automations run?” and +“run this now”. + +### Auth + +| Env | Purpose | +| --- | --- | +| `ACTIVITY_CORE_OPERATOR_TOKEN` | Shared operator token; required for **mutations** | +| `ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS` | `1` only for local dev without a token | + +Mutations: `POST /ops/automations/{id}/trigger|enable|disable|pause|unpause` +Header: `X-Operator-Token: ` (or `Authorization: Bearer `). + +Fail-closed: if the token is unset and unauth is not allowed, mutations return +**403**. Reads (`GET /ops/...`) do not require the token (ClusterIP / port-forward +posture). **Do not** put the token in git, chat, or workplans. Store in +`actcore-runtime-secret` (or local `.env`) via operator custody. + +### Daily checklist + +```bash +# How did automations go since Sunday? +curl -sS "http://localhost:8010/ops/automations/status?since=sunday" | python3 -m json.tool +# or CLI equivalent: +make automation-status SINCE=sunday + +# Inventory +curl -sS "http://localhost:8010/ops/automations" | python3 -m json.tool + +# Run now (requires token) +curl -sS -X POST "http://localhost:8010/ops/automations//trigger" \ + -H "X-Operator-Token: $ACTIVITY_CORE_OPERATOR_TOKEN" \ + -H "Content-Type: application/json" -d '{}' + +# Side-effect activities (e.g. forgejo prune) need explicit confirm: +curl -sS -X POST "http://localhost:8010/ops/automations//trigger" \ + -H "X-Operator-Token: $ACTIVITY_CORE_OPERATOR_TOKEN" \ + -H "Content-Type: application/json" \ + -d '{"confirm_side_effect": true}' + +# Pause / disable schedule (token required) +curl -sS -X POST "http://localhost:8010/ops/automations//pause" \ + -H "X-Operator-Token: $ACTIVITY_CORE_OPERATOR_TOKEN" +curl -sS -X POST "http://localhost:8010/ops/automations//disable" \ + -H "X-Operator-Token: $ACTIVITY_CORE_OPERATOR_TOKEN" +``` + +Thin UI: open `/ops/ui`, paste the operator token into the browser field +(localStorage only), then use Run now / pause actions. **Cron edits are not in +the UI** — change definition files and sync. + +### Production access (railiance01) + +API remains **ClusterIP** (no public Ingress in WP-0024). + +```bash +# From a machine with kubectl to railiance01: +kubectl -n activity-core port-forward svc/actcore-api 8010:8010 +# Browser: http://127.0.0.1:8010/ops/ui +# Ensure ACTIVITY_CORE_OPERATOR_TOKEN is set on actcore-api (runtime secret key). +``` + +Bootstrap token (operator workstation; never commit the value): + +```bash +# Generate and inject (example — adjust secret key name to match cluster) +TOKEN=$(openssl rand -hex 24) +kubectl -n activity-core create secret generic actcore-runtime-secret \ + --from-literal=ACTIVITY_CORE_OPERATOR_TOKEN="$TOKEN" \ + --dry-run=client -o yaml | kubectl apply -f - # only if creating fresh; +# Prefer: kubectl patch / edit to merge the key into existing secret, then +kubectl -n activity-core rollout restart deploy/actcore-api +unset TOKEN +``` + +--- + ## REST API — common operations ```bash diff --git a/k8s/railiance/README.md b/k8s/railiance/README.md index 5ef1dda..a88ae7d 100644 --- a/k8s/railiance/README.md +++ b/k8s/railiance/README.md @@ -110,3 +110,21 @@ kubectl -n activity-core exec deploy/actcore-api -- \ kubectl -n activity-core get pods kubectl -n activity-core get svc ``` + +## Operator automation console (ACTIVITY-WP-0024) + +API Service is ClusterIP-only. From a shell with cluster access: + +```bash +kubectl -n activity-core port-forward svc/actcore-api 8010:8010 +# UI: http://127.0.0.1:8010/ops/ui +# JSON: http://127.0.0.1:8010/ops/automations/status?since=sunday +``` + +Mutations require `ACTIVITY_CORE_OPERATOR_TOKEN` in `actcore-runtime-secret` +(injected via `envFrom` on `actcore-api`). Merge the key into the existing +secret (do not replace DB URL keys). Header: `X-Operator-Token`. See +`docs/runbook.md` § Operator automation console. + +Cron/schedule expression changes remain git-owned (definition files + sync); +the console supports Run now, enable/disable, and Temporal pause/unpause only. diff --git a/src/activity_core/api.py b/src/activity_core/api.py index bc03d11..a03f99f 100644 --- a/src/activity_core/api.py +++ b/src/activity_core/api.py @@ -1,6 +1,7 @@ """FastAPI REST API for activity-core. T30: CRUD for ActivityDefinition + manual one-shot trigger. +ACTIVITY-WP-0024: operator automation console under /ops. Endpoints: GET /activity-definitions/ — list all @@ -9,6 +10,7 @@ Endpoints: PUT /activity-definitions/{id} — update DELETE /activity-definitions/{id} — delete POST /activity-definitions/{id}/trigger — manual one-shot run + GET/POST /ops/... — operator console (inventory, status, control) Schedule lifecycle: - POST/PUT with trigger_type='cron' upserts a Temporal Schedule. @@ -38,6 +40,7 @@ from temporalio.api.workflowservice.v1 import GetSystemInfoRequest from temporalio.client import Client from activity_core.models import ActivityDefinition, CronTriggerConfig +from activity_core.ops_api import bind_ops_deps, router as ops_router from activity_core.orm import ActivityDefinition as ActivityDefinitionRow, EventType as EventTypeRow from activity_core.schedule_manager import delete_schedule, upsert_schedule from activity_core.sync_service import run_sync @@ -69,6 +72,7 @@ async def lifespan(app: FastAPI): # type: ignore[type-arg] engine = create_async_engine(db_url) _session_factory = async_sessionmaker(engine, expire_on_commit=False) _temporal_client = await Client.connect(TEMPORAL_HOST, namespace=TEMPORAL_NAMESPACE) + bind_ops_deps(_get_db, _get_temporal) yield @@ -77,6 +81,7 @@ async def lifespan(app: FastAPI): # type: ignore[type-arg] app = FastAPI(title="activity-core API", lifespan=lifespan) app.include_router(webhook_router) +app.include_router(ops_router) def _get_db() -> async_sessionmaker[AsyncSession]: diff --git a/src/activity_core/ops_api.py b/src/activity_core/ops_api.py new file mode 100644 index 0000000..2d1b476 --- /dev/null +++ b/src/activity_core/ops_api.py @@ -0,0 +1,477 @@ +"""FastAPI router for operator automation console (ACTIVITY-WP-0024).""" + +from __future__ import annotations + +import html +import json +import os +import uuid +from datetime import datetime, timezone +from typing import Any, Callable + +from fastapi import APIRouter, Depends, HTTPException, Query, Request +from fastapi.responses import HTMLResponse +from pydantic import BaseModel, Field +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker +from temporalio.client import Client + +from activity_core.ops_auth import HEADER_NAME, operator_token_configured, require_operator +from activity_core.ops_console import ( + is_side_effect_definition, + ops_definition_detail, + ops_inventory, + ops_runs, + ops_status, + recent_audits, + record_ops_audit, + set_definition_enabled, + set_schedule_paused, +) +from activity_core.orm import ActivityDefinition as ActivityDefinitionRow + +router = APIRouter(prefix="/ops", tags=["ops-console"]) + +_get_db: Callable[[], async_sessionmaker[AsyncSession]] | None = None +_get_temporal: Callable[[], Client] | None = None + + +def bind_ops_deps( + get_db: Callable[[], async_sessionmaker[AsyncSession]], + get_temporal: Callable[[], Client], +) -> None: + global _get_db, _get_temporal + _get_db = get_db + _get_temporal = get_temporal + + +def _db() -> async_sessionmaker[AsyncSession]: + assert _get_db is not None + return _get_db() + + +def _temporal() -> Client: + assert _get_temporal is not None + return _get_temporal() + + +def _db_url() -> str | None: + return os.environ.get("ACTCORE_DB_URL") + + +class TriggerBody(BaseModel): + confirm_side_effect: bool = False + + +class MutateBody(BaseModel): + note: str | None = Field(default=None, max_length=500) + + +@router.get("/automations") +async def list_automations( + enabled: str = Query(default="all"), +) -> dict[str, Any]: + return await ops_inventory( + db_url=_db_url(), + temporal_host=os.environ.get("TEMPORAL_HOST"), + temporal_namespace=os.environ.get("TEMPORAL_NAMESPACE", "default"), + enabled=enabled, + ) + + +@router.get("/automations/status") +async def automations_status( + since: str = Query(default="today"), + until: str | None = Query(default=None), + timezone_name: str = Query(default="Europe/Berlin", alias="timezone"), + activity_id: str | None = Query(default=None), +) -> dict[str, Any]: + return await ops_status( + since=since, + until=until, + timezone_name=timezone_name, + db_url=_db_url(), + temporal_host=os.environ.get("TEMPORAL_HOST"), + temporal_namespace=os.environ.get("TEMPORAL_NAMESPACE", "default"), + state_hub_url=os.environ.get("STATE_HUB_URL"), + activity_id=activity_id, + ) + + +@router.get("/automations/{definition_id}") +async def get_automation(definition_id: uuid.UUID) -> dict[str, Any]: + detail = await ops_definition_detail( + _db(), + _temporal(), + definition_id, + db_url=_db_url(), + temporal_host=os.environ.get("TEMPORAL_HOST"), + temporal_namespace=os.environ.get("TEMPORAL_NAMESPACE", "default"), + ) + if not detail: + raise HTTPException(status_code=404, detail="ActivityDefinition not found") + return detail + + +@router.get("/automations/{definition_id}/runs") +async def list_runs( + definition_id: uuid.UUID, + since: str | None = Query(default=None, description="ISO datetime lower bound on fired_at"), + limit: int = Query(default=50, ge=1, le=200), +) -> dict[str, Any]: + since_dt: datetime | None = None + if since: + try: + since_dt = datetime.fromisoformat(since) + if since_dt.tzinfo is None: + since_dt = since_dt.replace(tzinfo=timezone.utc) + except ValueError as exc: + raise HTTPException(status_code=400, detail=f"invalid since: {exc}") from exc + async with _db()() as session: + if await session.get(ActivityDefinitionRow, definition_id) is None: + raise HTTPException(status_code=404, detail="ActivityDefinition not found") + return await ops_runs(_db(), definition_id, since=since_dt, limit=limit) + + +@router.post("/automations/{definition_id}/trigger") +async def trigger_automation( + definition_id: uuid.UUID, + body: TriggerBody | None = None, + principal: str = Depends(require_operator), +) -> dict[str, Any]: + body = body or TriggerBody() + Session = _db() + async with Session() as session: + row = await session.get(ActivityDefinitionRow, definition_id) + if row is None: + raise HTTPException(status_code=404, detail="ActivityDefinition not found") + if is_side_effect_definition(row) and not body.confirm_side_effect: + raise HTTPException( + status_code=400, + detail=( + "this activity may perform side effects; " + "pass confirm_side_effect=true to proceed" + ), + ) + name = row.name + + trigger_key = f"manual-{uuid.uuid4()}" + workflow_id = f"activity-{definition_id}:{trigger_key}" + handle = await _temporal().start_workflow( + "RunActivityWorkflow", + args=[str(definition_id), trigger_key, datetime.now(tz=timezone.utc).isoformat()], + id=workflow_id, + task_queue="orchestrator-tq", + ) + audit = await record_ops_audit( + action="trigger", + activity_id=str(definition_id), + activity_name=name, + principal=principal, + detail={"workflow_id": handle.id, "trigger_key": trigger_key}, + ) + return { + "workflow_id": handle.id, + "trigger_key": trigger_key, + "activity_id": str(definition_id), + "name": name, + "audit": audit, + } + + +@router.post("/automations/{definition_id}/enable") +async def enable_automation( + definition_id: uuid.UUID, + principal: str = Depends(require_operator), +) -> dict[str, Any]: + try: + return await set_definition_enabled( + _db(), _temporal(), definition_id, enabled=True, principal=principal + ) + except KeyError: + raise HTTPException(status_code=404, detail="ActivityDefinition not found") from None + + +@router.post("/automations/{definition_id}/disable") +async def disable_automation( + definition_id: uuid.UUID, + principal: str = Depends(require_operator), +) -> dict[str, Any]: + try: + return await set_definition_enabled( + _db(), _temporal(), definition_id, enabled=False, principal=principal + ) + except KeyError: + raise HTTPException(status_code=404, detail="ActivityDefinition not found") from None + + +@router.post("/automations/{definition_id}/pause") +async def pause_automation( + definition_id: uuid.UUID, + principal: str = Depends(require_operator), +) -> dict[str, Any]: + try: + return await set_schedule_paused( + _db(), _temporal(), definition_id, paused=True, principal=principal + ) + except KeyError: + raise HTTPException(status_code=404, detail="ActivityDefinition not found") from None + + +@router.post("/automations/{definition_id}/unpause") +async def unpause_automation( + definition_id: uuid.UUID, + principal: str = Depends(require_operator), +) -> dict[str, Any]: + try: + return await set_schedule_paused( + _db(), _temporal(), definition_id, paused=False, principal=principal + ) + except KeyError: + raise HTTPException(status_code=404, detail="ActivityDefinition not found") from None + + +@router.get("/audits") +async def list_audits(limit: int = Query(default=20, ge=1, le=100)) -> dict[str, Any]: + return {"audits": recent_audits(limit)} + + +@router.get("/auth/status") +async def auth_status() -> dict[str, Any]: + allow = (os.environ.get("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS") or "").lower() in { + "1", + "true", + "yes", + "on", + } + return { + "operator_token_configured": operator_token_configured(), + "mutation_header": HEADER_NAME, + "mutations_require_token": operator_token_configured() or not allow, + } + + +# --- Thin UI (no Jinja dependency) ------------------------------------------- + +_CSS = """ +:root { font-family: system-ui, sans-serif; color: #1a1a1a; } +body { margin: 1.5rem; max-width: 1100px; } +nav a { margin-right: 1rem; } +table { border-collapse: collapse; width: 100%; margin: 1rem 0; } +th, td { border: 1px solid #ccc; padding: 0.4rem 0.6rem; text-align: left; font-size: 0.9rem; } +th { background: #f4f4f4; } +.status-ok { color: #0a0; } .status-bad { color: #a00; } .status-warn { color: #a60; } +.card { border: 1px solid #ddd; border-radius: 6px; padding: 1rem; margin: 1rem 0; } +button, .btn { padding: 0.35rem 0.7rem; margin-right: 0.35rem; cursor: pointer; } +input[type=password], input[type=text] { padding: 0.3rem; min-width: 16rem; } +.muted { color: #666; font-size: 0.85rem; } +pre { background: #f8f8f8; padding: 0.75rem; overflow: auto; font-size: 0.8rem; } +""" + + +def _page(title: str, body: str) -> HTMLResponse: + doc = f""" + + + + {html.escape(title)} · activity-core ops + + + + +
+ {body} + + +""" + return HTMLResponse(doc) + + +def _token_form() -> str: + configured = "yes" if operator_token_configured() else "no" + return f""" +
+ + +

Server token configured: {configured}. + Mutations require header {html.escape(HEADER_NAME)}.

+
+""" + + +@router.get("/ui", response_class=HTMLResponse) +@router.get("/ui/", response_class=HTMLResponse) +async def ui_index() -> HTMLResponse: + try: + inv = await ops_inventory( + db_url=_db_url(), + temporal_host=os.environ.get("TEMPORAL_HOST"), + ) + error = None + except Exception as exc: # noqa: BLE001 + inv = {"automations": [], "summary": {}, "warnings": [str(exc)]} + error = str(exc) + + rows = [] + for a in inv.get("automations") or []: + aid = html.escape(str(a.get("id") or "")) + name = html.escape(str(a.get("name") or "")) + enabled = "yes" if a.get("enabled") else "no" + cron = html.escape(str(a.get("cron_expression") or a.get("at") or "")) + tz = html.escape(str(a.get("timezone") or "")) + paused = a.get("temporal") or {} + pstate = html.escape(str(paused.get("paused") if "paused" in paused else paused.get("status"))) + rows.append( + f"{name}" + f"{enabled}{html.escape(str(a.get('trigger_type')))}" + f"{cron}{tz}{pstate}" + ) + summary = html.escape(json.dumps(inv.get("summary") or {}, indent=2)) + warn = html.escape("; ".join(inv.get("warnings") or []) or "none") + err_html = f"

Error: {html.escape(error)}

" if error else "" + body = f""" +

Scheduled automations

+ {err_html} + {_token_form()} +

Warnings: {warn}

+ 
{summary}
+ + + {''.join(rows) or ''} +
NameEnabledTriggerScheduleTZTemporal
No automations found
+

Schedule cron is read-only in MVP — edit definition files + sync.

+""" + return _page("Inventory", body) + + +@router.get("/ui/status", response_class=HTMLResponse) +async def ui_status(since: str = Query(default="sunday")) -> HTMLResponse: + try: + report = await ops_status( + since=since, + db_url=_db_url(), + temporal_host=os.environ.get("TEMPORAL_HOST"), + state_hub_url=os.environ.get("STATE_HUB_URL"), + ) + error = None + except Exception as exc: # noqa: BLE001 + report = None + error = str(exc) + + if error or not report: + body = f"

Status

{html.escape(error or 'no report')}

{_token_form()}" + return _page("Status", body) + + rows = [] + for a in report.get("activities") or []: + st = str(a.get("status") or "") + cls = "status-ok" if st in {"completed", "ok", "disabled"} else ( + "status-bad" if st in {"missed", "validation_failed", "sink_failed"} else "status-warn" + ) + aid = html.escape(str(a.get("id") or "")) + rows.append( + f"{html.escape(str(a.get('name')))}" + f"{html.escape(st)}" + f"{html.escape(str(a.get('run_count', a.get('runs_count', ''))))}" + f"{html.escape(str(len(a.get('expected_fires') or a.get('expected') or [])))}" + f"{html.escape(str(a.get('enabled')))}" + ) + window = html.escape(json.dumps(report.get("window") or {}, indent=2)) + summary = html.escape(json.dumps(report.get("summary") or {}, indent=2)) + body = f""" +

Automation status

+ {_token_form()} +
+ + +
+ 
window: {window}
+ 
summary: {summary}
+ + + {''.join(rows) or ''} +
NameStatusRunsExpectedEnabled
No activities
+""" + return _page("Status", body) + + +@router.get("/ui/automations/{definition_id}", response_class=HTMLResponse) +async def ui_detail(definition_id: uuid.UUID) -> HTMLResponse: + detail = await ops_definition_detail( + _db(), + _temporal(), + definition_id, + db_url=_db_url(), + temporal_host=os.environ.get("TEMPORAL_HOST"), + ) + if not detail: + raise HTTPException(status_code=404, detail="not found") + runs = await ops_runs(_db(), definition_id, limit=25) + aid = str(definition_id) + name = html.escape(str(detail.get("name") or "")) + side = bool(detail.get("side_effect")) + confirm = "true" if side else "false" + run_rows = [] + for r in runs.get("runs") or []: + run_rows.append( + f"{html.escape(str(r.get('run_id')))}" + f"{html.escape(str(r.get('fired_at')))}" + f"{html.escape(str(r.get('scheduled_for')))}" + f"{html.escape(str(r.get('tasks_spawned')))}" + f"{html.escape(str(len((r.get('evidence') or {}).get('task_spawns') or [])))}" + ) + detail_json = html.escape(json.dumps(detail, indent=2, default=str)) + body = f""" +

{name}

+ {_token_form()} +

id: {html.escape(aid)} + side_effect_risk: {side}

+
+ + + + + +
+

Definition (read-only schedule)

+ 
{detail_json}
+

Recent runs

+ + + {''.join(run_rows) or ''} +
run_idfired_atscheduled_fortasksspawn evidence
No runs
+""" + return _page(name, body) diff --git a/src/activity_core/ops_auth.py b/src/activity_core/ops_auth.py new file mode 100644 index 0000000..14c4ef2 --- /dev/null +++ b/src/activity_core/ops_auth.py @@ -0,0 +1,89 @@ +"""Operator token auth for activity-core ops console (ACTIVITY-WP-0024). + +Mutations under ``/ops`` are fail-closed: +- If ``ACTIVITY_CORE_OPERATOR_TOKEN`` is set, requests must send matching + ``X-Operator-Token`` (or ``Authorization: Bearer ``). +- If the token is **unset**, mutations are refused unless + ``ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS`` is truthy (local dev only). + +Read endpoints do not require the token (ClusterIP / port-forward posture). +""" + +from __future__ import annotations + +import hmac +import os +from typing import Annotated + +from fastapi import Header, HTTPException, Request + +OPERATOR_TOKEN_ENV = "ACTIVITY_CORE_OPERATOR_TOKEN" +ALLOW_UNAUTH_ENV = "ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS" +HEADER_NAME = "X-Operator-Token" + + +def operator_token_configured() -> bool: + return bool((os.environ.get(OPERATOR_TOKEN_ENV) or "").strip()) + + +def allow_unauth_mutations() -> bool: + return (os.environ.get(ALLOW_UNAUTH_ENV) or "").strip().lower() in { + "1", + "true", + "yes", + "on", + } + + +def extract_operator_token( + *, + x_operator_token: str | None = None, + authorization: str | None = None, +) -> str | None: + if x_operator_token and x_operator_token.strip(): + return x_operator_token.strip() + if authorization and authorization.lower().startswith("bearer "): + return authorization[7:].strip() or None + return None + + +def verify_operator_token(provided: str | None) -> str: + """Return operator principal label or raise HTTPException.""" + expected = (os.environ.get(OPERATOR_TOKEN_ENV) or "").strip() + if not expected: + if allow_unauth_mutations(): + return "anonymous-dev" + raise HTTPException( + status_code=403, + detail=( + "operator auth not configured; set " + f"{OPERATOR_TOKEN_ENV} or enable {ALLOW_UNAUTH_ENV} for local dev" + ), + ) + if not provided: + raise HTTPException( + status_code=401, + detail=f"missing operator token ({HEADER_NAME} or Authorization Bearer)", + ) + if not hmac.compare_digest(provided, expected): + raise HTTPException(status_code=401, detail="invalid operator token") + return "operator" + + +async def require_operator( + request: Request, + x_operator_token: Annotated[str | None, Header(alias=HEADER_NAME)] = None, + authorization: Annotated[str | None, Header()] = None, +) -> str: + """FastAPI dependency: require valid operator token for mutations.""" + # Prefer dependency headers; fall back to raw request (HTML form headers rare). + provided = extract_operator_token( + x_operator_token=x_operator_token, + authorization=authorization, + ) + if provided is None: + provided = extract_operator_token( + x_operator_token=request.headers.get(HEADER_NAME), + authorization=request.headers.get("Authorization"), + ) + return verify_operator_token(provided) diff --git a/src/activity_core/ops_console.py b/src/activity_core/ops_console.py new file mode 100644 index 0000000..6aea991 --- /dev/null +++ b/src/activity_core/ops_console.py @@ -0,0 +1,441 @@ +"""Operator console services (ACTIVITY-WP-0024) — inventory, status, runs, audit.""" + +from __future__ import annotations + +import argparse +import logging +import os +import uuid +from collections import deque +from datetime import datetime, timezone +from typing import Any + +import httpx +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker +from temporalio.client import Client + +from activity_core.automation_status import ( + DEFAULT_STATE_HUB_URL, + DEFAULT_TEMPORAL_NAMESPACE, + DEFAULT_TIMEZONE, + automation_schedule_id, + build_inventory_report, + build_report, + inventory_row, + load_temporal_visibility, + public_trigger_config, +) +from activity_core.models import ActivityDefinition, CronTriggerConfig, ScheduledTriggerConfig +from activity_core.orm import ActivityDefinition as ActivityDefinitionRow +from activity_core.orm import ActivityRun, TaskSpawnLog +from activity_core.runtime_paths import custodian_working_memory_dir +from activity_core.schedule_manager import pause_schedule, unpause_schedule, upsert_schedule +from activity_core.state_hub_write import idempotency_headers, parse_state_hub_write_response + +logger = logging.getLogger(__name__) + +_AUDIT_BUFFER: deque[dict[str, Any]] = deque(maxlen=100) + +SIDE_EFFECT_MARKERS = ( + "forgejo_package_prune", + "apply: true", + '"apply": true', + "'apply': true", +) + + +def _status_namespace( + *, + since: str, + until: str | None, + timezone_name: str, + db_url: str | None, + temporal_host: str | None, + temporal_namespace: str, + state_hub_url: str | None, + timeout_seconds: float = 5.0, + activity_id: list[str] | None = None, + activity_name: list[str] | None = None, +) -> argparse.Namespace: + return argparse.Namespace( + since=since, + until=until, + timezone=timezone_name, + activity_id=activity_id or [], + activity_name=activity_name or [], + db_url=db_url, + state_hub_url=state_hub_url or os.environ.get("STATE_HUB_URL", DEFAULT_STATE_HUB_URL), + working_memory_dir=os.environ.get( + "AUTOMATION_STATUS_WORKING_MEMORY_DIR", + str(custodian_working_memory_dir()), + ), + temporal_host=temporal_host or os.environ.get("TEMPORAL_HOST"), + temporal_namespace=temporal_namespace + or os.environ.get("TEMPORAL_NAMESPACE", DEFAULT_TEMPORAL_NAMESPACE), + timeout_seconds=timeout_seconds, + progress_limit=int(os.environ.get("AUTOMATION_STATUS_PROGRESS_LIMIT", "100")), + progress_event_type=None, + format="json", + ) + + +def _inventory_namespace( + *, + db_url: str | None, + temporal_host: str | None, + temporal_namespace: str, + enabled: str = "all", + trigger: list[str] | None = None, + activity_id: list[str] | None = None, + activity_name: list[str] | None = None, + timeout_seconds: float = 5.0, +) -> argparse.Namespace: + return argparse.Namespace( + db_url=db_url, + temporal_host=temporal_host or os.environ.get("TEMPORAL_HOST"), + temporal_namespace=temporal_namespace + or os.environ.get("TEMPORAL_NAMESPACE", DEFAULT_TEMPORAL_NAMESPACE), + timeout_seconds=timeout_seconds, + enabled=enabled, + trigger_type=trigger or [], + activity_id=activity_id or [], + activity_name=activity_name or [], + format="json", + ) + + +async def ops_inventory( + *, + db_url: str | None, + temporal_host: str | None = None, + temporal_namespace: str = DEFAULT_TEMPORAL_NAMESPACE, + enabled: str = "all", +) -> dict[str, Any]: + args = _inventory_namespace( + db_url=db_url, + temporal_host=temporal_host, + temporal_namespace=temporal_namespace, + enabled=enabled, + ) + report, _exit = await build_inventory_report(args) + return report + + +async def ops_status( + *, + since: str = "today", + until: str | None = None, + timezone_name: str = DEFAULT_TIMEZONE, + db_url: str | None = None, + temporal_host: str | None = None, + temporal_namespace: str = DEFAULT_TEMPORAL_NAMESPACE, + state_hub_url: str | None = None, + activity_id: str | None = None, +) -> dict[str, Any]: + ids = [activity_id] if activity_id else [] + args = _status_namespace( + since=since, + until=until, + timezone_name=timezone_name, + db_url=db_url, + temporal_host=temporal_host, + temporal_namespace=temporal_namespace, + state_hub_url=state_hub_url, + activity_id=ids, + ) + report, exit_code = await build_report(args) + report["exit_code"] = exit_code + return report + + +async def ops_definition_detail( + session_factory: async_sessionmaker[AsyncSession], + temporal: Client | None, + definition_id: uuid.UUID, + *, + db_url: str | None, + temporal_host: str | None = None, + temporal_namespace: str = DEFAULT_TEMPORAL_NAMESPACE, +) -> dict[str, Any]: + async with session_factory() as session: + row = await session.get(ActivityDefinitionRow, definition_id) + if row is None: + return {} + definition = { + "id": str(row.id), + "name": row.name, + "enabled": row.enabled, + "trigger_type": row.trigger_type, + "trigger_config": row.trigger_config or {}, + "source": "database", + "version": row.version, + "created_at": row.created_at.isoformat() if row.created_at else None, + "updated_at": row.updated_at.isoformat() if row.updated_at else None, + "side_effect": is_side_effect_definition(row), + } + + temporal_map: dict[str, dict[str, Any]] = {} + if temporal is not None or temporal_host: + host = temporal_host or os.environ.get("TEMPORAL_HOST") + temporal_map, _src = await load_temporal_visibility( + host, + temporal_namespace, + [definition], + timeout_seconds=5.0, + ) + row_out = inventory_row(definition, temporal_map.get(definition["id"])) + row_out["version"] = definition.get("version") + row_out["created_at"] = definition.get("created_at") + row_out["updated_at"] = definition.get("updated_at") + row_out["side_effect"] = definition.get("side_effect") + row_out["trigger_config_public"] = public_trigger_config(definition.get("trigger_config") or {}) + return row_out + + +def is_side_effect_definition(row: ActivityDefinitionRow) -> bool: + blob = str(row.context_sources or []) + str(row.task_templates or []) + str(row.trigger_config or {}) + lower = blob.lower() + return any(marker in lower for marker in SIDE_EFFECT_MARKERS) + + +async def ops_runs( + session_factory: async_sessionmaker[AsyncSession], + definition_id: uuid.UUID, + *, + since: datetime | None = None, + limit: int = 50, +) -> dict[str, Any]: + limit = max(1, min(limit, 200)) + async with session_factory() as session: + stmt = ( + select(ActivityRun) + .where(ActivityRun.activity_id == definition_id) + .order_by(ActivityRun.fired_at.desc()) + .limit(limit) + ) + if since is not None: + stmt = stmt.where(ActivityRun.fired_at >= since) + runs = list((await session.scalars(stmt)).all()) + + run_ids = [r.run_id for r in runs] + spawn_by_run: dict[str, list[dict[str, Any]]] = {str(rid): [] for rid in run_ids} + if run_ids: + # Spawns keyed by triggering_event_id often equal run_id or workflow key; + # also collect by activity_def_id for recent window. + spawn_stmt = ( + select(TaskSpawnLog) + .where(TaskSpawnLog.activity_def_id == definition_id) + .order_by(TaskSpawnLog.id.desc()) + .limit(limit * 5) + ) + for log in (await session.scalars(spawn_stmt)).all(): + entry = { + "task_ref": log.task_ref, + "source_type": log.source_type, + "source_id": log.source_id, + "triggering_event_id": log.triggering_event_id, + "condition_matched": log.condition_matched, + } + # Attach if triggering_event_id matches a run_id string, else keep under activity. + tid = log.triggering_event_id or "" + matched = False + for rid in run_ids: + if tid == str(rid) or str(rid) in tid: + spawn_by_run[str(rid)].append(entry) + matched = True + break + if not matched and runs: + # bucket orphan spawns onto most recent run for operator visibility + spawn_by_run[str(runs[0].run_id)].append(entry) + + items = [] + for r in runs: + items.append( + { + "run_id": str(r.run_id), + "activity_id": str(r.activity_id), + "scheduled_for": r.scheduled_for.isoformat() if r.scheduled_for else None, + "fired_at": r.fired_at.isoformat() if r.fired_at else None, + "tasks_spawned": r.tasks_spawned, + "version_used": r.version_used, + "evidence": { + "task_spawns": spawn_by_run.get(str(r.run_id), [])[:20], + # context_snapshot may be large; only surface shallow keys + "context_keys": sorted((r.context_snapshot or {}).keys())[:40], + }, + } + ) + return { + "activity_id": str(definition_id), + "count": len(items), + "runs": items, + } + + +async def record_ops_audit( + *, + action: str, + activity_id: str, + activity_name: str | None, + principal: str, + detail: dict[str, Any] | None = None, +) -> dict[str, Any]: + event = { + "event_type": "ops_console_audit", + "action": action, + "activity_id": activity_id, + "activity_name": activity_name, + "principal": principal, + "detail": detail or {}, + "at": datetime.now(tz=timezone.utc).isoformat(), + "audit_id": str(uuid.uuid4()), + } + _AUDIT_BUFFER.appendleft(event) + + hub = (os.environ.get("STATE_HUB_URL") or "").rstrip("/") + if hub: + try: + body = { + "event_type": "ops_console_audit", + "summary": f"ops {action} {activity_name or activity_id} by {principal}", + "author": f"activity-core-ops:{principal}", + "detail": { + "action": action, + "activity_id": activity_id, + "activity_name": activity_name, + "principal": principal, + **(detail or {}), + }, + } + headers = { + "Content-Type": "application/json", + **idempotency_headers("ops_console_audit", action, activity_id, event["audit_id"]), + } + with httpx.Client(timeout=3.0) as client: + resp = client.post(f"{hub}/progress/", json=body, headers=headers) + parse_state_hub_write_response(resp) + event["hub"] = "ok" + except Exception as exc: # noqa: BLE001 — audit must not fail mutation + logger.warning("ops audit hub write failed: %s", exc) + event["hub"] = f"failed:{type(exc).__name__}" + else: + event["hub"] = "skipped" + return event + + +def recent_audits(limit: int = 20) -> list[dict[str, Any]]: + return list(_AUDIT_BUFFER)[: max(1, min(limit, 100))] + + +def clear_audit_buffer_for_tests() -> None: + _AUDIT_BUFFER.clear() + + +async def set_definition_enabled( + session_factory: async_sessionmaker[AsyncSession], + temporal: Client, + definition_id: uuid.UUID, + *, + enabled: bool, + principal: str, +) -> dict[str, Any]: + async with session_factory() as session: + row = await session.get(ActivityDefinitionRow, definition_id) + if row is None: + raise KeyError("not found") + row.enabled = enabled + async with session.begin(): + session.add(row) + name = row.name + defn = _row_to_model(row) + + schedule_result: dict[str, Any] = {} + try: + if isinstance(defn.trigger_config, (CronTriggerConfig, ScheduledTriggerConfig)): + await upsert_schedule(temporal, defn) + schedule_result = {"upserted": True, "paused": not enabled} + else: + schedule_result = {"upserted": False, "reason": "non-scheduled trigger"} + except Exception as exc: # noqa: BLE001 + schedule_result = {"upserted": False, "error": str(exc)} + + audit = await record_ops_audit( + action="enable" if enabled else "disable", + activity_id=str(definition_id), + activity_name=name, + principal=principal, + detail={"enabled": enabled, "schedule": schedule_result}, + ) + return { + "activity_id": str(definition_id), + "name": name, + "enabled": enabled, + "schedule": schedule_result, + "audit": audit, + } + + +async def set_schedule_paused( + session_factory: async_sessionmaker[AsyncSession], + temporal: Client, + definition_id: uuid.UUID, + *, + paused: bool, + principal: str, +) -> dict[str, Any]: + async with session_factory() as session: + row = await session.get(ActivityDefinitionRow, definition_id) + if row is None: + raise KeyError("not found") + name = row.name + trigger_type = row.trigger_type + onetime = trigger_type == "scheduled" + + note = f"{'paused' if paused else 'unpaused'} by ops console ({principal})" + if paused: + schedule_result = await pause_schedule(temporal, definition_id, note=note, onetime=onetime) + else: + schedule_result = await unpause_schedule(temporal, definition_id, note=note, onetime=onetime) + + audit = await record_ops_audit( + action="pause" if paused else "unpause", + activity_id=str(definition_id), + activity_name=name, + principal=principal, + detail={"schedule": schedule_result}, + ) + return { + "activity_id": str(definition_id), + "name": name, + "schedule": schedule_result, + "audit": audit, + } + + +def _row_to_model(row: ActivityDefinitionRow) -> ActivityDefinition: + return ActivityDefinition.model_validate( + { + "id": row.id, + "name": row.name, + "enabled": row.enabled, + "trigger_config": row.trigger_config, + "context_sources": row.context_sources or [], + "task_templates": row.task_templates or [], + "rules": row.rules_json or [], + "instructions": row.instructions_json or [], + "dedupe_key_strategy": row.dedupe_key_strategy, + "version": row.version, + } + ) + + +def definition_has_side_effect( + session_row: ActivityDefinitionRow | None, + *, + context_sources: Any = None, +) -> bool: + if session_row is not None: + return is_side_effect_definition(session_row) + blob = str(context_sources or "").lower() + return any(m in blob for m in SIDE_EFFECT_MARKERS) diff --git a/src/activity_core/schedule_manager.py b/src/activity_core/schedule_manager.py index 397ff4b..e1d1422 100644 --- a/src/activity_core/schedule_manager.py +++ b/src/activity_core/schedule_manager.py @@ -353,6 +353,51 @@ async def delete_schedule(client: Client, activity_id: str | UUID) -> None: pass # Not found — treat as success. +async def pause_schedule( + client: Client, + activity_id: str | UUID, + *, + note: str = "paused via ops console", + onetime: bool = False, +) -> dict: + """Pause a Temporal Schedule. Returns status dict; missing schedule is an error.""" + sid = _onetime_schedule_id(activity_id) if onetime else schedule_id(activity_id) + handle = client.get_schedule_handle(sid) + try: + await handle.pause(note=note) + except ScheduleAlreadyRunningError: + # Race with in-flight action — treat as best-effort success after retry note. + try: + await handle.pause(note=note) + except (RPCError, ScheduleAlreadyRunningError) as exc: + return {"schedule_id": sid, "paused": None, "warning": str(exc)} + except RPCError as exc: + return {"schedule_id": sid, "paused": None, "error": str(exc)} + return {"schedule_id": sid, "paused": True, "note": note} + + +async def unpause_schedule( + client: Client, + activity_id: str | UUID, + *, + note: str = "unpaused via ops console", + onetime: bool = False, +) -> dict: + """Unpause a Temporal Schedule. Returns status dict.""" + sid = _onetime_schedule_id(activity_id) if onetime else schedule_id(activity_id) + handle = client.get_schedule_handle(sid) + try: + await handle.unpause(note=note) + except ScheduleAlreadyRunningError: + try: + await handle.unpause(note=note) + except (RPCError, ScheduleAlreadyRunningError) as exc: + return {"schedule_id": sid, "paused": None, "warning": str(exc)} + except RPCError as exc: + return {"schedule_id": sid, "paused": None, "error": str(exc)} + return {"schedule_id": sid, "paused": False, "note": note} + + async def list_schedules(client: Client) -> list[dict]: """Enumerate all activity-core Temporal Schedules. diff --git a/tests/test_ops_auth.py b/tests/test_ops_auth.py new file mode 100644 index 0000000..62668e6 --- /dev/null +++ b/tests/test_ops_auth.py @@ -0,0 +1,58 @@ +"""Unit tests for operator token auth (ACTIVITY-WP-0024-T02/T06).""" + +from __future__ import annotations + +import pytest +from fastapi import HTTPException + +from activity_core.ops_auth import ( + extract_operator_token, + operator_token_configured, + verify_operator_token, +) + + +def test_extract_operator_token_prefers_header() -> None: + assert ( + extract_operator_token( + x_operator_token="abc", + authorization="Bearer other", + ) + == "abc" + ) + + +def test_extract_operator_token_bearer() -> None: + assert extract_operator_token(authorization="Bearer secret-token") == "secret-token" + + +def test_verify_requires_config_when_fail_closed(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("ACTIVITY_CORE_OPERATOR_TOKEN", raising=False) + monkeypatch.delenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", raising=False) + with pytest.raises(HTTPException) as exc: + verify_operator_token(None) + assert exc.value.status_code == 403 + + +def test_verify_allows_anonymous_dev(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("ACTIVITY_CORE_OPERATOR_TOKEN", raising=False) + monkeypatch.setenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", "1") + assert verify_operator_token(None) == "anonymous-dev" + + +def test_verify_token_match(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "correct-horse") + assert verify_operator_token("correct-horse") == "operator" + with pytest.raises(HTTPException) as exc: + verify_operator_token("wrong") + assert exc.value.status_code == 401 + with pytest.raises(HTTPException) as exc2: + verify_operator_token(None) + assert exc2.value.status_code == 401 + + +def test_operator_token_configured(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("ACTIVITY_CORE_OPERATOR_TOKEN", raising=False) + assert operator_token_configured() is False + monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "x") + assert operator_token_configured() is True diff --git a/tests/test_ops_console_api.py b/tests/test_ops_console_api.py new file mode 100644 index 0000000..d78b78b --- /dev/null +++ b/tests/test_ops_console_api.py @@ -0,0 +1,248 @@ +"""Contract tests for /ops console API (ACTIVITY-WP-0024).""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timezone +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient + +from activity_core import ops_api +from activity_core.ops_console import clear_audit_buffer_for_tests, recent_audits, record_ops_audit + + +@pytest.fixture(autouse=True) +def _clean_audit() -> None: + clear_audit_buffer_for_tests() + yield + clear_audit_buffer_for_tests() + + +@pytest.fixture +def ops_app(monkeypatch: pytest.MonkeyPatch) -> FastAPI: + monkeypatch.setenv("ACTIVITY_CORE_OPERATOR_TOKEN", "test-token") + monkeypatch.delenv("ACTIVITY_CORE_OPS_ALLOW_UNAUTH_MUTATIONS", raising=False) + monkeypatch.setenv("ACTCORE_DB_URL", "postgresql+asyncpg://unused/unused") + + app = FastAPI() + app.include_router(ops_api.router) + + session_factory = MagicMock(name="session_factory") + temporal = MagicMock(name="temporal") + ops_api.bind_ops_deps(lambda: session_factory, lambda: temporal) + app.state.session_factory = session_factory + app.state.temporal = temporal + return app + + +@pytest.mark.asyncio +async def test_auth_status(ops_app: FastAPI) -> None: + transport = ASGITransport(app=ops_app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + res = await client.get("/ops/auth/status") + assert res.status_code == 200 + body = res.json() + assert body["operator_token_configured"] is True + assert body["mutation_header"] == "X-Operator-Token" + + +@pytest.mark.asyncio +async def test_trigger_requires_token(ops_app: FastAPI, monkeypatch: pytest.MonkeyPatch) -> None: + def_id = uuid.uuid4() + + async def fake_status(*_a: Any, **_k: Any) -> dict[str, Any]: + return {"mode": "automation-status", "activities": []} + + monkeypatch.setattr(ops_api, "ops_status", fake_status) + + transport = ASGITransport(app=ops_app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + res = await client.post(f"/ops/automations/{def_id}/trigger") + assert res.status_code == 401 + + +@pytest.mark.asyncio +async def test_trigger_with_token(ops_app: FastAPI, monkeypatch: pytest.MonkeyPatch) -> None: + def_id = uuid.uuid4() + row = MagicMock() + row.name = "Weekly SBOM" + row.context_sources = [] + row.task_templates = [] + row.trigger_config = {"trigger_type": "cron", "cron_expression": "0 9 * * 1"} + + session = AsyncMock() + session.get = AsyncMock(return_value=row) + session.__aenter__ = AsyncMock(return_value=session) + session.__aexit__ = AsyncMock(return_value=None) + ops_app.state.session_factory.return_value = session + + handle = MagicMock() + handle.id = f"activity-{def_id}:manual-test" + ops_app.state.temporal.start_workflow = AsyncMock(return_value=handle) + + monkeypatch.setattr( + "activity_core.ops_api.record_ops_audit", + AsyncMock( + return_value={ + "action": "trigger", + "audit_id": "a1", + "principal": "operator", + } + ), + ) + + transport = ASGITransport(app=ops_app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + res = await client.post( + f"/ops/automations/{def_id}/trigger", + headers={"X-Operator-Token": "test-token"}, + json={}, + ) + assert res.status_code == 200 + body = res.json() + assert body["workflow_id"] == handle.id + assert body["name"] == "Weekly SBOM" + assert "token" not in str(body).lower() or "test-token" not in str(body) + + +@pytest.mark.asyncio +async def test_side_effect_requires_confirm( + ops_app: FastAPI, monkeypatch: pytest.MonkeyPatch +) -> None: + def_id = uuid.uuid4() + row = MagicMock() + row.name = "Weekly Forgejo Package Prune" + row.context_sources = [{"query": "forgejo_package_prune", "apply": True}] + row.task_templates = [] + row.trigger_config = {} + + session = AsyncMock() + session.get = AsyncMock(return_value=row) + session.__aenter__ = AsyncMock(return_value=session) + session.__aexit__ = AsyncMock(return_value=None) + ops_app.state.session_factory.return_value = session + + transport = ASGITransport(app=ops_app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + res = await client.post( + f"/ops/automations/{def_id}/trigger", + headers={"X-Operator-Token": "test-token"}, + json={}, + ) + assert res.status_code == 400 + assert "side effect" in res.json()["detail"].lower() + + +@pytest.mark.asyncio +async def test_status_endpoint_wraps_report( + ops_app: FastAPI, monkeypatch: pytest.MonkeyPatch +) -> None: + async def fake_status(**kwargs: Any) -> dict[str, Any]: + return { + "mode": "automation-status", + "window": {"since": kwargs.get("since")}, + "summary": {"total": 0}, + "activities": [], + "warnings": [], + "exit_code": 0, + } + + monkeypatch.setattr(ops_api, "ops_status", fake_status) + transport = ASGITransport(app=ops_app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + res = await client.get("/ops/automations/status", params={"since": "sunday"}) + assert res.status_code == 200 + assert res.json()["mode"] == "automation-status" + assert res.json()["window"]["since"] == "sunday" + + +@pytest.mark.asyncio +async def test_inventory_endpoint( + ops_app: FastAPI, monkeypatch: pytest.MonkeyPatch +) -> None: + async def fake_inv(**_k: Any) -> dict[str, Any]: + return { + "mode": "automation-inventory", + "automations": [ + { + "id": str(uuid.uuid4()), + "name": "Daily Triage", + "enabled": True, + "trigger_type": "cron", + "cron_expression": "20 7 * * *", + } + ], + "summary": {"total": 1}, + "warnings": [], + } + + monkeypatch.setattr(ops_api, "ops_inventory", fake_inv) + transport = ASGITransport(app=ops_app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + res = await client.get("/ops/automations") + assert res.status_code == 200 + assert res.json()["summary"]["total"] == 1 + assert res.json()["automations"][0]["name"] == "Daily Triage" + + +@pytest.mark.asyncio +async def test_ui_index_renders(ops_app: FastAPI, monkeypatch: pytest.MonkeyPatch) -> None: + async def fake_inv(**_k: Any) -> dict[str, Any]: + return { + "automations": [ + { + "id": str(uuid.uuid4()), + "name": "Daily Triage", + "enabled": True, + "trigger_type": "cron", + "cron_expression": "0 0 * * *", + "timezone": "Europe/Berlin", + "temporal": {"paused": False}, + } + ], + "summary": {"total": 1}, + "warnings": [], + } + + monkeypatch.setattr(ops_api, "ops_inventory", fake_inv) + transport = ASGITransport(app=ops_app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + res = await client.get("/ops/ui/") + assert res.status_code == 200 + assert "Daily Triage" in res.text + assert "Operator token" in res.text + + +@pytest.mark.asyncio +async def test_record_ops_audit_no_secret_leak(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("STATE_HUB_URL", raising=False) + event = await record_ops_audit( + action="trigger", + activity_id=str(uuid.uuid4()), + activity_name="X", + principal="operator", + detail={"workflow_id": "wf-1"}, + ) + assert event["action"] == "trigger" + assert "token" not in event + audits = recent_audits() + assert audits[0]["audit_id"] == event["audit_id"] + + +@pytest.mark.asyncio +async def test_runs_404(ops_app: FastAPI) -> None: + def_id = uuid.uuid4() + session = AsyncMock() + session.get = AsyncMock(return_value=None) + session.__aenter__ = AsyncMock(return_value=session) + session.__aexit__ = AsyncMock(return_value=None) + ops_app.state.session_factory.return_value = session + + transport = ASGITransport(app=ops_app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + res = await client.get(f"/ops/automations/{def_id}/runs") + assert res.status_code == 404 diff --git a/workplans/ACTIVITY-WP-0024-operator-automation-console.md b/workplans/ACTIVITY-WP-0024-operator-automation-console.md index 6161502..f59e793 100644 --- a/workplans/ACTIVITY-WP-0024-operator-automation-console.md +++ b/workplans/ACTIVITY-WP-0024-operator-automation-console.md @@ -4,7 +4,7 @@ type: workplan title: "Operator automation console (status API + control plane + thin UI)" domain: infotech repo: activity-core -status: proposed +status: finished owner: codex topic_slug: activity-core created: "2026-07-21" @@ -107,7 +107,7 @@ Deep-links into State Hub / edge when evidence lives there (G9-aware). ```task id: ACTIVITY-WP-0024-T01 -status: todo +status: done priority: high state_hub_task_id: "d8830c31-bbcd-41d4-b74b-e1632d8d0257" ``` @@ -124,7 +124,7 @@ state_hub_task_id: "d8830c31-bbcd-41d4-b74b-e1632d8d0257" ```task id: ACTIVITY-WP-0024-T02 -status: todo +status: done priority: high state_hub_task_id: "a94decc6-8ac9-4738-823a-6c7b19dade80" ``` @@ -143,7 +143,7 @@ runbook describes custody without leaking secrets. ```task id: ACTIVITY-WP-0024-T03 -status: todo +status: done priority: high state_hub_task_id: "ed665a54-0a5c-4d35-b47d-50de77ef91d6" ``` @@ -163,7 +163,7 @@ window within documented field mapping. ```task id: ACTIVITY-WP-0024-T04 -status: todo +status: done priority: high state_hub_task_id: "720e0783-93b8-480a-8bfb-00a6aae592a1" ``` @@ -179,7 +179,7 @@ state_hub_task_id: "720e0783-93b8-480a-8bfb-00a6aae592a1" ```task id: ACTIVITY-WP-0024-T05 -status: todo +status: done priority: high state_hub_task_id: "2673bd8e-f746-428a-9a20-79336dcad9d4" ``` @@ -200,7 +200,7 @@ evidence; pause/disable visible in inventory/status; unauthenticated mutate fail ```task id: ACTIVITY-WP-0024-T06 -status: todo +status: done priority: high state_hub_task_id: "8b10eb8f-90d1-498b-a0a3-5444c590a691" ``` @@ -215,7 +215,7 @@ state_hub_task_id: "8b10eb8f-90d1-498b-a0a3-5444c590a691" ```task id: ACTIVITY-WP-0024-T07 -status: todo +status: done priority: medium state_hub_task_id: "32c695ac-f208-4540-b2fa-199a3f83c4c9" ``` @@ -232,7 +232,7 @@ state_hub_task_id: "32c695ac-f208-4540-b2fa-199a3f83c4c9" ```task id: ACTIVITY-WP-0024-T08 -status: todo +status: done priority: medium state_hub_task_id: "032f7485-3cf6-4ec2-a076-cbf82312a63c" ``` @@ -250,7 +250,7 @@ still deferred with a clear follow-up note. ```task id: ACTIVITY-WP-0024-T09 -status: todo +status: done priority: low state_hub_task_id: "1a937c1a-6534-467e-94fb-8ba54e18236e" ``` @@ -263,13 +263,13 @@ state_hub_task_id: "1a937c1a-6534-467e-94fb-8ba54e18236e" ## Success criteria -- [ ] `GET /ops/automations/status?since=…` matches CLI automation-status semantics -- [ ] Operator can list last N runs for a definition without SSH/SQL -- [ ] Authenticated Run now returns workflow_id and produces `activity_runs` evidence -- [ ] Pause/disable visible in inventory/status and Temporal schedule state -- [ ] Unauthenticated mutation returns 401/403 when token is configured -- [ ] Thin UI usable via port-forward for the above flows -- [ ] Runbook documents token custody without secrets in git +- [x] `GET /ops/automations/status?since=…` matches CLI automation-status semantics +- [x] Operator can list last N runs for a definition without SSH/SQL +- [x] Authenticated Run now returns workflow_id and produces `activity_runs` evidence +- [x] Pause/disable visible in inventory/status and Temporal schedule state +- [x] Unauthenticated mutation returns 401/403 when token is configured +- [x] Thin UI usable via port-forward for the above flows +- [x] Runbook documents token custody without secrets in git ## Implementation order @@ -286,3 +286,10 @@ state_hub_task_id: "1a937c1a-6534-467e-94fb-8ba54e18236e" - `src/activity_core/automation_status.py`, `src/activity_core/api.py` - `docs/runbook.md`, `SCOPE.md` G10 - ACTIVITY-WP-0018, ACTIVITY-WP-0019, ACTIVITY-WP-0021 + +## Closeout 2026-07-21 + +Implemented `/ops` console: auth (fail-closed operator token), inventory/status/runs +REST, Run now + enable/disable/pause with audit buffer + optional State Hub progress, +thin HTML UI at `/ops/ui`, runbook + k8s access docs. Tests: `tests/test_ops_auth.py`, +`tests/test_ops_console_api.py`.