"""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, extract_sso_principal, 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, "temporal_ui_url": temporal_ui_url(), "sso_docs": "/docs not required — see docs/ops-sso-access.md", "public_hosts": { "ops": "https://activity.coulomb.social", "temporal_ui": "https://temporal.coulomb.social", }, } # --- 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; } nav a.external::after { content: " ↗"; font-size: 0.75em; opacity: 0.7; } 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 temporal_ui_url() -> str: """Browser URL for Temporal Web UI (SSO ingress or local port-forward). Override with ACTIVITY_CORE_TEMPORAL_UI_URL. Default prefers the public SSO hostname when set via env; otherwise local port-forward on :8080. """ raw = ( os.environ.get("ACTIVITY_CORE_TEMPORAL_UI_URL") or os.environ.get("TEMPORAL_UI_URL") or "https://temporal.coulomb.social" ).strip() return raw.rstrip("/") or "https://temporal.coulomb.social" def _page(title: str, body: str) -> HTMLResponse: temporal_href = html.escape(temporal_ui_url(), quote=True) doc = f"""
Server token configured: {configured}.
Mutations require header {html.escape(HEADER_NAME)}.
{cron}Error: {html.escape(error)}
" if error else "" body = f"""Warnings: {warn}
{summary}
| Name | Enabled | Trigger | Schedule | TZ | Temporal |
|---|---|---|---|---|---|
| No automations found | |||||
Schedule cron is read-only in MVP — edit definition files + sync.
Temporal Web UI (workflow debugger):
{html.escape(temporal_ui_url())}
— requires port-forward of svc/actcore-temporal-ui 8080:8080
until SSO ingress (ACTIVITY-WP-0025).
{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 "")) last_run = a.get("last_run") or {} last_at = ( a.get("last_run_at") or last_run.get("fired_at") or a.get("temporal_last_fired_at") or "—" ) last_tasks = last_run.get("tasks_spawned") last_tasks_s = "" if last_tasks is None else str(last_tasks) run_count = a.get("observed_run_count", a.get("run_count", a.get("runs_count", ""))) expected_n = a.get("expected_fire_count") if expected_n is None: expected_n = len(a.get("expected_fires") or a.get("expected") or []) rows.append( f"{html.escape(str(last_at))}Last run = newest fired_at in the selected window
(falls back to Temporal schedule last action when no DB run is in-window).
window: {window}
summary: {summary}
| Name | Status | Last run | Tasks | Runs | Expected | Enabled |
|---|---|---|---|---|---|---|
| No activities | ||||||
{html.escape(str(r.get('run_id')))}id: {html.escape(aid)}
side_effect_risk: {side}
{detail_json}
| run_id | fired_at | scheduled_for | tasks | spawn evidence |
|---|---|---|---|---|
| No runs | ||||