From 7711bf9c707669692fa79f37db57080f6f6e0dde Mon Sep 17 00:00:00 2001 From: tegwick Date: Wed, 5 Aug 2026 17:23:02 +0200 Subject: [PATCH] feat(ops): run artefacts in API/UI + llm-connect host access design ACTIVITY-WP-0027: document durable ClusterIP host access for llm-connect; join activity_runs to ops_runs; surface Forgejo artefact links on /ops/ui automation detail and new run detail pages. --- docs/llm-connect-host-access.md | 73 +++++++ docs/ops-run-queue.md | 17 ++ src/activity_core/ops_api.py | 91 +++++++- src/activity_core/ops_console.py | 66 ++---- src/activity_core/run_artifacts.py | 322 +++++++++++++++++++++++++++++ tests/test_run_artifacts.py | 90 ++++++++ 6 files changed, 604 insertions(+), 55 deletions(-) create mode 100644 docs/llm-connect-host-access.md create mode 100644 src/activity_core/run_artifacts.py create mode 100644 tests/test_run_artifacts.py diff --git a/docs/llm-connect-host-access.md b/docs/llm-connect-host-access.md new file mode 100644 index 0000000..5c658fb --- /dev/null +++ b/docs/llm-connect-host-access.md @@ -0,0 +1,73 @@ +# llm-connect host access (ACTIVITY-WP-0027) + +**Audience:** operators on railiance01 +**Status:** decision locked 2026-08-05 + +## Problem + +rein-aharness runs on the **host** (user systemd claim loop) and must call: + +| Service | In-cluster DNS | Used for | +| ------- | -------------- | -------- | +| `llm-connect` | `llm-connect.activity-core.svc.cluster.local:8080` | FI / Binky LLM | +| `actcore-api` | `actcore-api.activity-core.svc.cluster.local:8010` | claim / complete | +| `actcore-statehub-edge-relay` | `…:8000` | progress / completion events | + +Historically the claim-loop env pointed at `http://127.0.0.1:8080` and +`http://127.0.0.1:8010` via ad-hoc **`kubectl port-forward`**. That dies on +reboot or PF process exit while k3s and the claim loop keep running. + +## Decision (H2 — ClusterIP via kube-proxy) + +On single-node k3s (railiance01), the **host can reach Service ClusterIPs** +directly through kube-proxy. No port-forward required. + +```text +rein-aharness@host + → http://:8080 llm-connect + → http://:8010 actcore-api + → http://:8000 statehub edge relay +``` + +**Not chosen** + +| Option | Why not | +| ------ | ------- | +| Long-lived kubectl port-forward | Not production; fails silently | +| Public Ingress / NodePort for llm-connect | Exposes LLM proxy surface | +| Move claim loop fully in-cluster | Domain checkouts live on host disk today (repos under `$HOME`) | + +## How URLs are set + +1. **Preferred:** claim-loop env uses `k8s://activity-core/:` pseudo-URLs. + The `rein-aharness-claim` wrapper resolves them at process start via + `kubectl get svc` (requires `KUBECONFIG` readable by the user). +2. **Fallback:** concrete `http://10.43.x.x:port` ClusterIPs (stable until the + Service object is recreated). +3. **Refresh:** `deploy/scripts/refresh-claim-loop-k8s-urls.sh` on rein-aharness + rewrites concrete IPs from live Services. + +## Health checks (host) + +```bash +# After claim wrapper resolves k8s:// URLs into the process env: +curl -sS "$LLM_CONNECT_URL/health" +curl -sS "$ACTIVITY_CORE_URL/health" +curl -sS "${STATE_HUB_URL%/}/edge/health" # if using edge relay +``` + +## Security + +- ClusterIP ranges are not published on the Internet; they are cluster-local. +- Do not change `llm-connect` Service type to `LoadBalancer` / public NodePort + without a separate security review. +- Worker token for actcore-api stays in `~/.config/rein-aharness/claim-loop.env` + (mode 600), never git. + +## Ownership + +| Piece | Owner | +| ----- | ----- | +| In-cluster llm-connect Deployment | activity-core / railiance k8s | +| Host claim loop + URL resolution | rein-aharness | +| This decision doc | activity-core (ACTIVITY-WP-0027-T01) | diff --git a/docs/ops-run-queue.md b/docs/ops-run-queue.md index 6d1ee57..492af0c 100644 --- a/docs/ops-run-queue.md +++ b/docs/ops-run-queue.md @@ -89,6 +89,23 @@ queue empty. Never requires Forgejo or issue-core for the claim path. +## Run artefacts (ACTIVITY-WP-0027) + +`GET /ops/automations/{id}/runs` joins each `activity_run` to related +`ops_runs` and exposes deliverable links for the ops UI: + +| Field | Source | +| ----- | ------ | +| `ops_runs[]` | matched by `triggering_event_id == run_id` (or contains), else time window | +| `artifacts[]` | from `ops_runs.result` (`path`, `head_after`, `target_repo`) | +| Forgejo URL | `FORGEJO_WEB_BASE` (default `https://forgejo.coulomb.social`) + org + repo + path + ref | + +Run detail: `GET /ops/automations/{id}/runs/{run_id}` and +`/ops/ui/automations/{id}/runs/{run_id}`. + +Executor `result` should include at least: `ok`, `path`, `head_after`, +`target_repo`, `committed`. No prompts or raw model output. + ## Auth - **Worker:** `ACTIVITY_CORE_WORKER_TOKEN` via `X-Worker-Token` or diff --git a/src/activity_core/ops_api.py b/src/activity_core/ops_api.py index 456953a..1e664b7 100644 --- a/src/activity_core/ops_api.py +++ b/src/activity_core/ops_api.py @@ -25,6 +25,7 @@ from activity_core.ops_console import ( is_side_effect_definition, ops_definition_detail, ops_inventory, + ops_run_detail, ops_runs, ops_status, recent_audits, @@ -168,6 +169,17 @@ async def list_runs( return await ops_runs(_db(), definition_id, since=since_dt, limit=limit) +@router.get("/automations/{definition_id}/runs/{run_id}") +async def get_run( + definition_id: uuid.UUID, + run_id: uuid.UUID, +) -> dict[str, Any]: + detail = await ops_run_detail(_db(), definition_id, run_id) + if not detail: + raise HTTPException(status_code=404, detail="run not found") + return detail + + @router.post("/automations/{definition_id}/trigger") async def trigger_automation( definition_id: uuid.UUID, @@ -539,12 +551,31 @@ async def ui_detail(definition_id: uuid.UUID) -> HTMLResponse: confirm = "true" if side else "false" run_rows = [] for r in runs.get("runs") or []: + rid = str(r.get("run_id") or "") + arts = r.get("artifacts") or [] + if arts: + art_html = " ".join( + f'' + f'{html.escape(str(a.get("label") or a.get("kind") or "artifact"))}' + for a in arts + if a.get("url") and str(a.get("url")).startswith("https://") + ) or 'none' + else: + ops_states = [o.get("state") for o in (r.get("ops_runs") or [])] + if any(s == "open" for s in ops_states): + art_html = 'pending' + elif any(s in ("failed", "expired") for s in ops_states): + art_html = 'failed' + else: + art_html = 'none' 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"" + f"{html.escape(rid[:13])}…" + f"{html.escape(str(r.get('fired_at') or ''))}" + f"{html.escape(str(r.get('scheduled_for') or ''))}" f"{html.escape(str(r.get('tasks_spawned')))}" - f"{html.escape(str(len((r.get('evidence') or {}).get('task_spawns') or [])))}" + f"{html.escape(str(len((r.get('evidence') or {}).get('task_spawns') or [])))}" + f"{art_html}" ) detail_json = html.escape(json.dumps(detail, indent=2, default=str)) body = f""" @@ -563,8 +594,56 @@ async def ui_detail(definition_id: uuid.UUID) -> HTMLResponse: 
{detail_json}

Recent runs

- - {''.join(run_rows) or ''} + + {''.join(run_rows) or ''}
run_idfired_atscheduled_fortasksspawn evidence
No runs
run_idfired_atscheduled_fortasksspawnsArtifacts
No runs
""" return _page(name, body) + + +@router.get("/ui/automations/{definition_id}/runs/{run_id}", response_class=HTMLResponse) +async def ui_run_detail(definition_id: uuid.UUID, run_id: uuid.UUID) -> HTMLResponse: + run = await ops_run_detail(_db(), definition_id, run_id) + if not run: + raise HTTPException(status_code=404, detail="run not found") + aid = str(definition_id) + rid = str(run_id) + arts = run.get("artifacts") or [] + art_lis = [] + for a in arts: + url = a.get("url") or "" + label = html.escape(str(a.get("label") or a.get("kind") or "artifact")) + if isinstance(url, str) and url.startswith("https://"): + art_lis.append( + f'
  • {label}' + f' ({html.escape(str(a.get("kind") or ""))})
    • ' + ) + else: + art_lis.append(f"
  • {label} {html.escape(str(url))}
    • ") + ops_blocks = [] + for op in run.get("ops_runs") or []: + ops_blocks.append( + "
    " + f"

    ops_run {html.escape(str(op.get('id')))} " + f"state={html.escape(str(op.get('state')))}

    " + f"

    {html.escape(str(op.get('title') or ''))}

    " + f"
    {html.escape(json.dumps(op.get('result') or {{}}, indent=2, default=str))}
    " + "
    " + ) + body = f""" +

    ← back to automation

    +

    Run {html.escape(rid)}

    +
    +

    fired_at: {html.escape(str(run.get('fired_at') or ''))}

    +

    scheduled_for: {html.escape(str(run.get('scheduled_for') or ''))}

    +

    tasks_spawned: {html.escape(str(run.get('tasks_spawned')))} + version: {html.escape(str(run.get('version_used') or ''))}

    +
    +

    Artifacts

    +
      {''.join(art_lis) or '
    • No deliverable links for this run
      • '}
    +

    Ops runs

    + {''.join(ops_blocks) or '

    No linked ops_runs

    '} +

    Evidence (compact)

    + 
    {html.escape(json.dumps(run.get('evidence') or {{}}, indent=2, default=str))}
    +""" + return _page(f"Run {rid[:8]}", body) diff --git a/src/activity_core/ops_console.py b/src/activity_core/ops_console.py index 48f5e6b..7a3418d 100644 --- a/src/activity_core/ops_console.py +++ b/src/activity_core/ops_console.py @@ -28,7 +28,7 @@ from activity_core.automation_status import ( ) 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.orm import ActivityRun 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 @@ -207,6 +207,9 @@ async def ops_runs( since: datetime | None = None, limit: int = 50, ) -> dict[str, Any]: + """Recent activity_runs enriched with ops_runs + artefact links (WP-0027).""" + from activity_core.run_artifacts import enrich_activity_runs + limit = max(1, min(limit, 200)) async with session_factory() as session: stmt = ( @@ -218,54 +221,8 @@ async def ops_runs( 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], - }, - } + items = await enrich_activity_runs( + session, definition_id, runs, since=since ) return { "activity_id": str(definition_id), @@ -274,6 +231,17 @@ async def ops_runs( } +async def ops_run_detail( + session_factory: async_sessionmaker[AsyncSession], + definition_id: uuid.UUID, + run_id: uuid.UUID, +) -> dict[str, Any] | None: + from activity_core.run_artifacts import get_enriched_run + + async with session_factory() as session: + return await get_enriched_run(session, definition_id, run_id) + + async def record_ops_audit( *, action: str, diff --git a/src/activity_core/run_artifacts.py b/src/activity_core/run_artifacts.py new file mode 100644 index 0000000..6663487 --- /dev/null +++ b/src/activity_core/run_artifacts.py @@ -0,0 +1,322 @@ +"""Join activity_runs to ops_runs and build deliverable artefact links. + +ACTIVITY-WP-0027: operators review results from the ops UI without SSH. +""" + +from __future__ import annotations + +import os +import re +import uuid +from datetime import datetime, timedelta +from typing import Any +from urllib.parse import quote + +from sqlalchemy import Select, select +from sqlalchemy.ext.asyncio import AsyncSession + +from activity_core.orm import ActivityRun, OpsRun, TaskSpawnLog + +# Default public Forgejo web base (no trailing slash). Override with FORGEJO_WEB_BASE. +_DEFAULT_FORGEJO_WEB_BASE = "https://forgejo.coulomb.social" +_DEFAULT_FORGEJO_ORG = "coulomb" + +_SAFE_PATH = re.compile(r"^[A-Za-z0-9_./@+-]+$") + + +def forgejo_web_base() -> str: + return ( + os.environ.get("FORGEJO_WEB_BASE", _DEFAULT_FORGEJO_WEB_BASE).rstrip("/") + ) + + +def forgejo_org() -> str: + return (os.environ.get("FORGEJO_ORG", _DEFAULT_FORGEJO_ORG) or _DEFAULT_FORGEJO_ORG).strip( + "/" + ) + + +def build_forgejo_blob_url( + *, + target_repo: str | None, + path: str | None, + ref: str | None = None, + web_base: str | None = None, + org: str | None = None, +) -> str | None: + """Return a browseable Forgejo URL for a repo-relative path, or None.""" + if not target_repo or not path: + return None + repo = target_repo.strip().strip("/") + rel = path.strip().lstrip("/") + if not repo or not rel: + return None + # Reject path traversal / odd characters for safety in UI links + if ".." in rel.split("/") or not _SAFE_PATH.match(rel): + return None + if not _SAFE_PATH.match(repo): + return None + base = (web_base or forgejo_web_base()).rstrip("/") + org_part = org if org is not None else forgejo_org() + ref_part = (ref or "main").strip() + if not ref_part or not _SAFE_PATH.match(ref_part.replace("/", "")): + # allow full SHAs + if not re.fullmatch(r"[0-9a-fA-F]{7,40}", ref_part or ""): + ref_part = "main" + # Forgejo: /{owner}/{repo}/src/commit/{sha}/path or /src/branch/{branch}/path + if re.fullmatch(r"[0-9a-fA-F]{7,40}", ref_part): + kind = "commit" + else: + kind = "branch" + encoded_path = "/".join(quote(seg, safe="") for seg in rel.split("/")) + return f"{base}/{org_part}/{repo}/src/{kind}/{quote(ref_part, safe='')}/{encoded_path}" + + +def artifacts_from_ops_result( + result: dict[str, Any] | None, + *, + target_repo: str | None = None, + title: str | None = None, +) -> list[dict[str, str]]: + """Build artefact link list from ops_run.result (+ optional row target_repo).""" + result = dict(result or {}) + repo = ( + (result.get("target_repo") if isinstance(result.get("target_repo"), str) else None) + or target_repo + ) + path = result.get("path") if isinstance(result.get("path"), str) else None + if not path: + # some approaches use report key + path = result.get("report") if isinstance(result.get("report"), str) else None + head = result.get("head_after") if isinstance(result.get("head_after"), str) else None + out: list[dict[str, str]] = [] + + # Executor may already supply artifact_urls + raw_urls = result.get("artifact_urls") + if isinstance(raw_urls, list): + for item in raw_urls[:10]: + if not isinstance(item, dict): + continue + url = item.get("url") + if not isinstance(url, str) or not url.startswith("https://"): + continue + out.append( + { + "kind": str(item.get("kind") or "link"), + "label": str(item.get("label") or "artifact")[:120], + "url": url, + } + ) + + forgejo = build_forgejo_blob_url(target_repo=repo, path=path, ref=head or "main") + if forgejo and not any(a.get("url") == forgejo for a in out): + label = title or (f"{repo}: {path}" if repo and path else path or "artifact") + out.insert( + 0, + { + "kind": "forgejo_blob", + "label": str(label)[:120], + "url": forgejo, + }, + ) + + if path and repo: + out.append( + { + "kind": "repo_path", + "label": f"{repo}/{path}"[:120], + "url": forgejo or f"repo://{repo}/{path}", + } + ) + # Prefer not to show repo_path when forgejo exists and uses same path + if forgejo: + out = [a for a in out if a.get("kind") != "repo_path"] + + return out[:12] + + +def _ops_summary(row: OpsRun) -> dict[str, Any]: + result = dict(row.result or {}) + # Drop bulky keys if ever present + for bad in ("prompt", "raw_output", "messages", "token"): + result.pop(bad, None) + artifacts = artifacts_from_ops_result( + result, target_repo=row.target_repo, title=row.title + ) + return { + "id": str(row.id), + "state": row.state, + "title": row.title, + "target_repo": row.target_repo, + "claim_owner": row.claim_owner, + "attempt": row.attempt, + "triggering_event_id": row.triggering_event_id, + "source_id": row.source_id, + "approach_hint": row.approach_hint, + "created_at": row.created_at.isoformat() if row.created_at else None, + "updated_at": row.updated_at.isoformat() if row.updated_at else None, + "result": { + k: result[k] + for k in ( + "ok", + "approach", + "path", + "date", + "wrote", + "committed", + "pushed", + "skipped_existing", + "head_after", + "target_repo", + "collection_candidates", + "reason", + ) + if k in result + }, + "artifacts": artifacts, + } + + +def match_ops_runs_to_activity_run( + activity_run: ActivityRun, + ops_rows: list[OpsRun], + *, + window: timedelta = timedelta(minutes=30), +) -> list[OpsRun]: + """Pick ops_runs that belong to this activity_run fire. + + Matching order: + 1. triggering_event_id == run_id + 2. triggering_event_id contains run_id + 3. same activity_definition_id and created_at within window of fired_at + """ + rid = str(activity_run.run_id) + exact: list[OpsRun] = [] + contains: list[OpsRun] = [] + windowed: list[OpsRun] = [] + fired = activity_run.fired_at + for row in ops_rows: + tid = row.triggering_event_id or "" + if tid == rid: + exact.append(row) + continue + if rid in tid: + contains.append(row) + continue + if fired and row.created_at: + delta = abs((row.created_at - fired).total_seconds()) + if delta <= window.total_seconds(): + windowed.append(row) + if exact: + return exact + if contains: + return contains + # Prefer closest created_at when multiple in window + if windowed and fired: + windowed.sort(key=lambda r: abs((r.created_at - fired).total_seconds())) + return windowed[:3] + return windowed + + +async def load_ops_runs_for_definition( + session: AsyncSession, + definition_id: uuid.UUID, + *, + since: datetime | None = None, + limit: int = 200, +) -> list[OpsRun]: + stmt: Select[tuple[OpsRun]] = ( + select(OpsRun) + .where(OpsRun.activity_definition_id == definition_id) + .order_by(OpsRun.created_at.desc()) + .limit(limit) + ) + if since is not None: + stmt = stmt.where(OpsRun.created_at >= since) + return list((await session.scalars(stmt)).all()) + + +async def enrich_activity_runs( + session: AsyncSession, + definition_id: uuid.UUID, + runs: list[ActivityRun], + *, + since: datetime | None = None, +) -> list[dict[str, Any]]: + """Return run dicts with ops_runs + artifacts for the ops API/UI.""" + if not runs: + return [] + + # Load a wider ops window so joins succeed + oldest = since + if runs: + candidates = [r.fired_at for r in runs if r.fired_at] + if candidates: + floor = min(candidates) - timedelta(hours=2) + oldest = floor if oldest is None else min(oldest, floor) + + ops_rows = await load_ops_runs_for_definition( + session, definition_id, since=oldest, limit=max(200, len(runs) * 5) + ) + + # Spawn evidence (existing behaviour, kept compact) + spawn_stmt = ( + select(TaskSpawnLog) + .where(TaskSpawnLog.activity_def_id == definition_id) + .order_by(TaskSpawnLog.id.desc()) + .limit(len(runs) * 5) + ) + spawn_logs = list((await session.scalars(spawn_stmt)).all()) + + items: list[dict[str, Any]] = [] + for r in runs: + matched = match_ops_runs_to_activity_run(r, ops_rows) + ops_payload = [_ops_summary(o) for o in matched] + artifacts: list[dict[str, str]] = [] + for op in ops_payload: + for art in op.get("artifacts") or []: + if art not in artifacts: + artifacts.append(art) + + spawns: list[dict[str, Any]] = [] + rid = str(r.run_id) + for log in spawn_logs: + tid = log.triggering_event_id or "" + if tid == rid or rid in tid: + spawns.append( + { + "task_ref": log.task_ref, + "source_type": log.source_type, + "source_id": log.source_id, + "triggering_event_id": log.triggering_event_id, + } + ) + items.append( + { + "run_id": rid, + "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, + "ops_runs": ops_payload, + "artifacts": artifacts[:12], + "evidence": { + "task_spawns": spawns[:20], + "context_keys": sorted((r.context_snapshot or {}).keys())[:40], + }, + } + ) + return items + + +async def get_enriched_run( + session: AsyncSession, + definition_id: uuid.UUID, + run_id: uuid.UUID, +) -> dict[str, Any] | None: + row = await session.get(ActivityRun, run_id) + if row is None or row.activity_id != definition_id: + return None + items = await enrich_activity_runs(session, definition_id, [row]) + return items[0] if items else None diff --git a/tests/test_run_artifacts.py b/tests/test_run_artifacts.py new file mode 100644 index 0000000..a6dd839 --- /dev/null +++ b/tests/test_run_artifacts.py @@ -0,0 +1,90 @@ +"""Tests for run↔ops_run join and Forgejo artefact URLs (ACTIVITY-WP-0027).""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock + +from activity_core.run_artifacts import ( + artifacts_from_ops_result, + build_forgejo_blob_url, + match_ops_runs_to_activity_run, +) + + +def test_build_forgejo_blob_url_commit() -> None: + url = build_forgejo_blob_url( + target_repo="freedom-intelligence", + path="briefs/2026/08/2026-08-05.md", + ref="33a5affbee7dafe40eb7b20a53e18b6efad60e3b", + web_base="https://forgejo.coulomb.social", + org="coulomb", + ) + assert url is not None + assert url.startswith("https://forgejo.coulomb.social/coulomb/freedom-intelligence/") + assert "src/commit/" in url + assert "briefs/2026/08/2026-08-05.md" in url + + +def test_build_forgejo_blob_url_rejects_traversal() -> None: + assert ( + build_forgejo_blob_url( + target_repo="freedom-intelligence", + path="../etc/passwd", + ref="main", + ) + is None + ) + + +def test_artifacts_from_ops_result() -> None: + arts = artifacts_from_ops_result( + { + "ok": True, + "path": "briefs/2026/08/2026-08-05.md", + "head_after": "33a5affbee7dafe40eb7b20a53e18b6efad60e3b", + "target_repo": "freedom-intelligence", + }, + title="FI brief", + ) + assert arts + assert arts[0]["kind"] == "forgejo_blob" + assert "freedom-intelligence" in arts[0]["url"] + + +def test_match_ops_runs_prefers_run_id() -> None: + run_id = uuid.uuid4() + ar = MagicMock() + ar.run_id = run_id + ar.fired_at = datetime(2026, 8, 5, 5, 30, tzinfo=timezone.utc) + + exact = MagicMock() + exact.triggering_event_id = str(run_id) + exact.created_at = ar.fired_at + + other = MagicMock() + other.triggering_event_id = "scheduled" + other.created_at = ar.fired_at + + matched = match_ops_runs_to_activity_run(ar, [other, exact]) + assert matched == [exact] + + +def test_match_ops_runs_window_fallback() -> None: + run_id = uuid.uuid4() + ar = MagicMock() + ar.run_id = run_id + ar.fired_at = datetime(2026, 8, 5, 5, 30, tzinfo=timezone.utc) + + near = MagicMock() + near.triggering_event_id = "scheduled" + near.created_at = ar.fired_at + timedelta(seconds=5) + + far = MagicMock() + far.triggering_event_id = "other" + far.created_at = ar.fired_at + timedelta(hours=2) + + matched = match_ops_runs_to_activity_run(ar, [far, near]) + assert near in matched + assert far not in matched