tegwick
54431db583
Prefer clean short hostname temporal.coulomb.social instead of activity-temporal.coulomb.social for Ingress, CORS, and ops UI links.
532 lines
19 KiB
Python
532 lines
19 KiB
Python
"""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"""<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="utf-8"/>
|
|
<title>{html.escape(title)} · activity-core ops</title>
|
|
<style>{_CSS}</style>
|
|
</head>
|
|
<body>
|
|
<nav>
|
|
<strong>activity-core ops</strong>
|
|
<a href="/ops/ui">Inventory</a>
|
|
<a href="/ops/ui/status?since=sunday">Status</a>
|
|
<a class="external" href="{temporal_href}" target="_blank" rel="noopener noreferrer"
|
|
title="Temporal Web UI — port-forward actcore-temporal-ui :8080 until SSO ingress">Temporal UI</a>
|
|
<a href="/ops/auth/status">Auth JSON</a>
|
|
</nav>
|
|
<hr/>
|
|
{body}
|
|
<script>
|
|
function operatorHeaders() {{
|
|
const t = localStorage.getItem('actcore_operator_token') || '';
|
|
const h = {{'Content-Type': 'application/json'}};
|
|
if (t) h['{HEADER_NAME}'] = t;
|
|
return h;
|
|
}}
|
|
function saveToken() {{
|
|
const v = document.getElementById('op-token').value;
|
|
localStorage.setItem('actcore_operator_token', v);
|
|
alert('Token saved in browser localStorage (not sent to server until you act).');
|
|
}}
|
|
async function opsAction(id, action, body) {{
|
|
const res = await fetch('/ops/automations/' + id + '/' + action, {{
|
|
method: 'POST',
|
|
headers: operatorHeaders(),
|
|
body: body ? JSON.stringify(body) : '{{}}'
|
|
}});
|
|
const text = await res.text();
|
|
let data; try {{ data = JSON.parse(text); }} catch {{ data = text; }}
|
|
if (!res.ok) {{
|
|
alert(action + ' failed: ' + res.status + ' ' + JSON.stringify(data));
|
|
return;
|
|
}}
|
|
alert(action + ' ok: ' + JSON.stringify(data).slice(0, 400));
|
|
location.reload();
|
|
}}
|
|
</script>
|
|
</body>
|
|
</html>"""
|
|
return HTMLResponse(doc)
|
|
|
|
|
|
def _token_form() -> str:
|
|
configured = "yes" if operator_token_configured() else "no"
|
|
return f"""
|
|
<div class="card">
|
|
<label>Operator token (browser only):
|
|
<input id="op-token" type="password" placeholder="X-Operator-Token value" autocomplete="off"/>
|
|
</label>
|
|
<button type="button" onclick="saveToken()">Save token</button>
|
|
<p class="muted">Server token configured: <strong>{configured}</strong>.
|
|
Mutations require header <code>{html.escape(HEADER_NAME)}</code>.</p>
|
|
</div>
|
|
"""
|
|
|
|
|
|
@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"<tr><td><a href='/ops/ui/automations/{aid}'>{name}</a></td>"
|
|
f"<td>{enabled}</td><td>{html.escape(str(a.get('trigger_type')))}</td>"
|
|
f"<td><code>{cron}</code></td><td>{tz}</td><td>{pstate}</td></tr>"
|
|
)
|
|
summary = html.escape(json.dumps(inv.get("summary") or {}, indent=2))
|
|
warn = html.escape("; ".join(inv.get("warnings") or []) or "none")
|
|
err_html = f"<p class='status-bad'>Error: {html.escape(error)}</p>" if error else ""
|
|
body = f"""
|
|
<h1>Scheduled automations</h1>
|
|
{err_html}
|
|
{_token_form()}
|
|
<p class="muted">Warnings: {warn}</p>
|
|
<pre>{summary}</pre>
|
|
<table>
|
|
<thead><tr><th>Name</th><th>Enabled</th><th>Trigger</th><th>Schedule</th><th>TZ</th><th>Temporal</th></tr></thead>
|
|
<tbody>{''.join(rows) or '<tr><td colspan="6">No automations found</td></tr>'}</tbody>
|
|
</table>
|
|
<p class="muted">Schedule cron is read-only in MVP — edit definition files + sync.</p>
|
|
<p class="muted">Temporal Web UI (workflow debugger):
|
|
<a class="external" href="{html.escape(temporal_ui_url(), quote=True)}"
|
|
target="_blank" rel="noopener noreferrer">{html.escape(temporal_ui_url())}</a>
|
|
— requires port-forward of <code>svc/actcore-temporal-ui 8080:8080</code>
|
|
until SSO ingress (ACTIVITY-WP-0025).</p>
|
|
"""
|
|
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"<h1>Status</h1><p class='status-bad'>{html.escape(error or 'no report')}</p>{_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"<tr><td><a href='/ops/ui/automations/{aid}'>{html.escape(str(a.get('name')))}</a></td>"
|
|
f"<td class='{cls}'>{html.escape(st)}</td>"
|
|
f"<td title='latest fire in selected window (or Temporal last action)'>"
|
|
f"<code>{html.escape(str(last_at))}</code></td>"
|
|
f"<td>{html.escape(last_tasks_s)}</td>"
|
|
f"<td>{html.escape(str(run_count))}</td>"
|
|
f"<td>{html.escape(str(expected_n))}</td>"
|
|
f"<td>{html.escape(str(a.get('enabled')))}</td></tr>"
|
|
)
|
|
window = html.escape(json.dumps(report.get("window") or {}, indent=2))
|
|
summary = html.escape(json.dumps(report.get("summary") or {}, indent=2))
|
|
body = f"""
|
|
<h1>Automation status</h1>
|
|
{_token_form()}
|
|
<form method="get" action="/ops/ui/status">
|
|
<label>since <input type="text" name="since" value="{html.escape(since)}"/></label>
|
|
<button type="submit">Refresh</button>
|
|
</form>
|
|
<p class="muted">Last run = newest <code>fired_at</code> in the selected window
|
|
(falls back to Temporal schedule last action when no DB run is in-window).</p>
|
|
<pre>window: {window}</pre>
|
|
<pre>summary: {summary}</pre>
|
|
<table>
|
|
<thead><tr>
|
|
<th>Name</th><th>Status</th><th>Last run</th><th>Tasks</th>
|
|
<th>Runs</th><th>Expected</th><th>Enabled</th>
|
|
</tr></thead>
|
|
<tbody>{''.join(rows) or '<tr><td colspan="7">No activities</td></tr>'}</tbody>
|
|
</table>
|
|
"""
|
|
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"<tr><td><code>{html.escape(str(r.get('run_id')))}</code></td>"
|
|
f"<td>{html.escape(str(r.get('fired_at')))}</td>"
|
|
f"<td>{html.escape(str(r.get('scheduled_for')))}</td>"
|
|
f"<td>{html.escape(str(r.get('tasks_spawned')))}</td>"
|
|
f"<td>{html.escape(str(len((r.get('evidence') or {}).get('task_spawns') or [])))}</td></tr>"
|
|
)
|
|
detail_json = html.escape(json.dumps(detail, indent=2, default=str))
|
|
body = f"""
|
|
<h1>{name}</h1>
|
|
{_token_form()}
|
|
<p class="muted">id: <code>{html.escape(aid)}</code>
|
|
side_effect_risk: <strong>{side}</strong></p>
|
|
<div class="card">
|
|
<button type="button" onclick="opsAction('{html.escape(aid)}','trigger',{{confirm_side_effect:{confirm}}})">Run now</button>
|
|
<button type="button" onclick="opsAction('{html.escape(aid)}','enable')">Enable</button>
|
|
<button type="button" onclick="opsAction('{html.escape(aid)}','disable')">Disable</button>
|
|
<button type="button" onclick="opsAction('{html.escape(aid)}','pause')">Pause schedule</button>
|
|
<button type="button" onclick="opsAction('{html.escape(aid)}','unpause')">Unpause schedule</button>
|
|
</div>
|
|
<h2>Definition (read-only schedule)</h2>
|
|
<pre>{detail_json}</pre>
|
|
<h2>Recent runs</h2>
|
|
<table>
|
|
<thead><tr><th>run_id</th><th>fired_at</th><th>scheduled_for</th><th>tasks</th><th>spawn evidence</th></tr></thead>
|
|
<tbody>{''.join(run_rows) or '<tr><td colspan="5">No runs</td></tr>'}</tbody>
|
|
</table>
|
|
"""
|
|
return _page(name, body)
|