activity-core/src/activity_core/run_artifacts.py

"""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.audit_projection import bounded_audit_projection
from activity_core.glas_evidence import normalise_ops_result
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 public_context_keys(context_snapshot: Any) -> list[str]:
    """Return safe top-level context names for the public run projection."""
    projected = bounded_audit_projection(context_snapshot or {})
    if not isinstance(projected, dict):
        return []
    return sorted(projected.keys())[:40]


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]:
    # Normalize again on read so historic rows cannot leak pre-allowlist blobs.
    result = normalise_ops_result(row.result)
    artifacts = artifacts_from_ops_result(
        result, target_repo=row.target_repo, title=row.title
    )
    execution_evidence = dict(result.get("execution_evidence") or {})
    compact_result = {
        k: result[k]
        for k in (
            "ok",
            "approach",
            "path",
            "date",
            "wrote",
            "committed",
            "pushed",
            "skipped_existing",
            "head_after",
            "target_repo",
            "collection_candidates",
            "reason",
            "error",
        )
        if k in result
    }
    if execution_evidence:
        compact_result["execution_evidence"] = execution_evidence
    repository_transaction = dict(result.get("repository_transaction") or {})
    if repository_transaction:
        compact_result["repository_transaction"] = repository_transaction
    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,
        "harness_profile_ref": row.harness_profile_ref,
        "execution_refs": dict(row.execution_refs or {}),
        "repository_grant": (
            dict(row.repository_grant) if row.repository_grant is not None else None
        ),
        "execution_evidence": execution_evidence,
        "repository_transaction": repository_transaction,
        "close_intent_digest": row.close_intent_digest,
        "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": compact_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": public_context_keys(r.context_snapshot),
                },
            }
        )
    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