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

@@ -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://<clusterIP>:8080   llm-connect
+  → http://<clusterIP>:8010   actcore-api
+  → http://<clusterIP>: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/<svc>:<port>` 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) |

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

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'<a href="{html.escape(str(a.get("url")))}" target="_blank" rel="noopener">'
+                f'{html.escape(str(a.get("label") or a.get("kind") or "artifact"))}</a>'
+                for a in arts
+                if a.get("url") and str(a.get("url")).startswith("https://")
+            ) or '<span class="muted">none</span>'
+        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 = '<span class="muted">pending</span>'
+            elif any(s in ("failed", "expired") for s in ops_states):
+                art_html = '<span class="muted">failed</span>'
+            else:
+                art_html = '<span class="muted">none</span>'
         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"<tr><td><a href='/ops/ui/automations/{html.escape(aid)}/runs/{html.escape(rid)}'>"
+            f"<code>{html.escape(rid[:13])}…</code></a></td>"
+            f"<td>{html.escape(str(r.get('fired_at') or ''))}</td>"
+            f"<td>{html.escape(str(r.get('scheduled_for') or ''))}</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>"
+            f"<td>{html.escape(str(len((r.get('evidence') or {}).get('task_spawns') or [])))}</td>"
+            f"<td>{art_html}</td></tr>"
         )
     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:
   <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>
+    <thead><tr><th>run_id</th><th>fired_at</th><th>scheduled_for</th><th>tasks</th><th>spawns</th><th>Artifacts</th></tr></thead>
+    <tbody>{''.join(run_rows) or '<tr><td colspan="6">No runs</td></tr>'}</tbody>
   </table>
 """
     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'<li><a href="{html.escape(url)}" target="_blank" rel="noopener">{label}</a>'
+                f' <span class="muted">({html.escape(str(a.get("kind") or ""))})</span></li>'
+            )
+        else:
+            art_lis.append(f"<li>{label} <code>{html.escape(str(url))}</code></li>")
+    ops_blocks = []
+    for op in run.get("ops_runs") or []:
+        ops_blocks.append(
+            "<div class='card'>"
+            f"<p><strong>ops_run</strong> <code>{html.escape(str(op.get('id')))}</code> "
+            f"state=<strong>{html.escape(str(op.get('state')))}</strong></p>"
+            f"<p>{html.escape(str(op.get('title') or ''))}</p>"
+            f"<pre>{html.escape(json.dumps(op.get('result') or {{}}, indent=2, default=str))}</pre>"
+            "</div>"
+        )
+    body = f"""
+  <p class="muted"><a href="/ops/ui/automations/{html.escape(aid)}">← back to automation</a></p>
+  <h1>Run <code>{html.escape(rid)}</code></h1>
+  <div class="card">
+    <p>fired_at: <code>{html.escape(str(run.get('fired_at') or ''))}</code></p>
+    <p>scheduled_for: <code>{html.escape(str(run.get('scheduled_for') or ''))}</code></p>
+    <p>tasks_spawned: <strong>{html.escape(str(run.get('tasks_spawned')))}</strong>
+       version: <code>{html.escape(str(run.get('version_used') or ''))}</code></p>
+  </div>
+  <h2>Artifacts</h2>
+  <ul>{''.join(art_lis) or '<li class="muted">No deliverable links for this run</li>'}</ul>
+  <h2>Ops runs</h2>
+  {''.join(ops_blocks) or '<p class="muted">No linked ops_runs</p>'}
+  <h2>Evidence (compact)</h2>
+  <pre>{html.escape(json.dumps(run.get('evidence') or {{}}, indent=2, default=str))}</pre>
+"""
+    return _page(f"Run {rid[:8]}", body)

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,

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

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