activity-core/src/activity_core/glas_evidence.py

"""Compact, non-secret Glas execution evidence for ops_run results.

Glas owns the versioned evidence envelope. Activity Core accepts that envelope
at its completion boundary, keeps only contract fields needed by operators, and
never persists direct-caller ``tool_output`` / ``tool_error`` or provider blobs.
"""

from __future__ import annotations

import math
from typing import Any

from activity_core.glas_profile import normalise_execution_refs

_LEGACY_RESULT_FIELDS = frozenset(
    {
        "ok",
        "approach",
        "path",
        "report",
        "date",
        "wrote",
        "committed",
        "pushed",
        "skipped_existing",
        "head_after",
        "target_repo",
        "collection_candidates",
        "reason",
        "error",
    }
)
_EVIDENCE_STRING_FIELDS = frozenset(
    {
        "request_id",
        "correlation_id",
        "actor",
        "project",
        "target_repo",
        "contract_version",
        "profile_ref",
        "rein_id",
        "rein_version",
        "model_route",
        "resolved_model",
        "sandbox_profile",
        "sandbox_id",
        "tool_profile",
        "outcome",
        "failure_stage",
        "error",
        "started_at",
        "finished_at",
        "commit_sha",
        "tool_events_completeness",
    }
)
_EVIDENCE_NUMBER_FIELDS = frozenset(
    {"duration_s", "tokens_spent", "token_budget", "tool_events_count"}
)
_EVIDENCE_ENUMS = {
    "outcome": frozenset({"succeeded", "failed", "refused"}),
    "failure_stage": frozenset(
        {
            "resolution",
            "sandbox_create",
            "session_start",
            "execution",
            "session_end",
            "teardown",
        }
    ),
    "tool_events_completeness": frozenset({"complete", "partial", "unavailable"}),
}


def _compact_scalar(value: Any, *, max_length: int = 2000) -> Any | None:
    if isinstance(value, bool):
        return value
    if (
        isinstance(value, (int, float))
        and not isinstance(value, bool)
        and math.isfinite(value)
    ):
        return value
    if isinstance(value, str):
        return value[:max_length]
    return None


def _compact_legacy_value(value: Any) -> Any | None:
    scalar = _compact_scalar(value)
    if scalar is not None:
        return scalar
    if isinstance(value, list):
        compact = [_compact_scalar(item, max_length=500) for item in value[:20]]
        return [item for item in compact if item is not None]
    return None


def normalise_execution_evidence(raw: Any) -> dict[str, Any]:
    """Return the safe subset of Glas ``ExecutionEvidence`` contract 1.0."""
    if not isinstance(raw, dict):
        return {}

    evidence: dict[str, Any] = {}
    for key in _EVIDENCE_STRING_FIELDS:
        value = raw.get(key)
        if isinstance(value, str):
            allowed = _EVIDENCE_ENUMS.get(key)
            if allowed is not None and value not in allowed:
                continue
            evidence[key] = value[:2000] if key == "error" else value[:500]

    for key in _EVIDENCE_NUMBER_FIELDS:
        value = raw.get(key)
        if (
            isinstance(value, (int, float))
            and not isinstance(value, bool)
            and math.isfinite(value)
            and value >= 0
        ):
            evidence[key] = value

    artifacts = raw.get("artifacts")
    if isinstance(artifacts, list):
        evidence["artifacts"] = [
            item[:1000]
            for item in artifacts[:20]
            if isinstance(item, str) and item
        ]

    refs = normalise_execution_refs(raw.get("refs"))
    if refs:
        evidence["refs"] = refs
    return evidence


def _normalise_artifact_urls(raw: Any) -> list[dict[str, str]]:
    if not isinstance(raw, list):
        return []
    urls: list[dict[str, str]] = []
    for item in raw[:10]:
        if not isinstance(item, dict):
            continue
        url = item.get("url")
        if not isinstance(url, str) or not url.startswith("https://"):
            continue
        urls.append(
            {
                "kind": str(item.get("kind") or "link")[:80],
                "label": str(item.get("label") or "artifact")[:120],
                "url": url[:2000],
            }
        )
    return urls


def _normalise_repository_baseline(raw: Any) -> dict[str, Any]:
    if not isinstance(raw, dict):
        return {}
    result: dict[str, Any] = {}
    for key in (
        "repo_id",
        "repo_name",
        "head",
        "branch",
        "status_digest",
        "index_diff_digest",
        "worktree_diff_digest",
        "upstream_ref",
        "upstream_oid",
        "remote_refs_digest",
        "protected_git_metadata_digest",
    ):
        value = raw.get(key)
        if isinstance(value, str):
            result[key] = value[:500]
        elif value is None and key in {"branch", "upstream_ref", "upstream_oid"}:
            result[key] = None
    for key in ("detached", "clean"):
        value = raw.get(key)
        if isinstance(value, bool):
            result[key] = value
    for key in ("dirty_entries", "remote_ref_count"):
        value = raw.get(key)
        if isinstance(value, int) and not isinstance(value, bool) and value >= 0:
            result[key] = min(value, 1_000_000)
    return result


def _normalise_repository_grant_evidence(raw: Any) -> dict[str, Any]:
    if not isinstance(raw, dict):
        return {}
    result: dict[str, Any] = {}
    for key in (
        "grant_id",
        "acceptance_policy_id",
        "version",
        "allowed_paths_digest",
    ):
        value = raw.get(key)
        if isinstance(value, str):
            result[key] = value[:500]
    for key in ("allowed_path_count", "min_commits", "max_commits"):
        value = raw.get(key)
        if isinstance(value, int) and not isinstance(value, bool) and value >= 0:
            result[key] = min(value, 1_000_000)
    publish = raw.get("publish")
    if isinstance(publish, bool):
        result["publish"] = publish
    return result


def _normalise_repository_acceptance(raw: Any) -> dict[str, Any]:
    if not isinstance(raw, dict):
        return {}
    result: dict[str, Any] = {}
    for key in (
        "policy_id",
        "head",
        "branch",
        "changed_paths_digest",
    ):
        value = raw.get(key)
        if isinstance(value, str):
            result[key] = value[:500]
        elif value is None and key == "branch":
            result[key] = None
    for key in (
        "accepted",
        "changed_paths_truncated",
        "clean_post_state",
        "remote_refs_unchanged",
        "protected_git_metadata_unchanged",
    ):
        value = raw.get(key)
        if isinstance(value, bool):
            result[key] = value
    for key in ("commit_count", "changed_path_count"):
        value = raw.get(key)
        if isinstance(value, int) and not isinstance(value, bool) and value >= 0:
            result[key] = min(value, 1_000_000)
    for key, limit in (("commits", 32), ("changed_paths", 100)):
        value = raw.get(key)
        if isinstance(value, list):
            result[key] = [
                item[:500] for item in value[:limit] if isinstance(item, str)
            ]
    return result


def _normalise_repository_metrics(raw: Any) -> dict[str, Any]:
    if not isinstance(raw, dict):
        return {}
    result: dict[str, Any] = {}
    for key in ("storage", "session_id"):
        value = raw.get(key)
        if isinstance(value, str):
            result[key] = value[:500]
    projection_ready = raw.get("projection_ready")
    if isinstance(projection_ready, bool):
        result["projection_ready"] = projection_ready
    return result


def normalise_repository_transaction(raw: Any) -> dict[str, Any]:
    """Allowlist bounded rein repository transaction/acceptance evidence."""
    if not isinstance(raw, dict):
        return {}

    result: dict[str, Any] = {}
    for key in ("transaction_id", "correlation_id"):
        value = raw.get(key)
        if isinstance(value, str):
            result[key] = value[:500]

    for key, normalizer in (
        ("baseline", _normalise_repository_baseline),
        ("repository_grant", _normalise_repository_grant_evidence),
        ("acceptance", _normalise_repository_acceptance),
        ("metrics", _normalise_repository_metrics),
    ):
        value = normalizer(raw.get(key))
        if value:
            result[key] = value
    return result


def normalise_ops_result(raw: Any) -> dict[str, Any]:
    """Allowlist legacy completion facts plus normalized Glas evidence."""
    if not isinstance(raw, dict):
        return {}

    result: dict[str, Any] = {}
    for key in _LEGACY_RESULT_FIELDS:
        value = _compact_legacy_value(raw.get(key))
        if value is not None:
            result[key] = value

    artifact_urls = _normalise_artifact_urls(raw.get("artifact_urls"))
    if artifact_urls:
        result["artifact_urls"] = artifact_urls

    raw_evidence = raw.get("execution_evidence")
    if not isinstance(raw_evidence, dict):
        raw_evidence = raw.get("evidence")
    evidence = normalise_execution_evidence(raw_evidence)
    if evidence:
        result["execution_evidence"] = evidence

    raw_transaction = raw.get("repository_transaction")
    if not isinstance(raw_transaction, dict):
        raw_transaction = raw.get("transaction")
    transaction = normalise_repository_transaction(raw_transaction)
    if transaction:
        result["repository_transaction"] = transaction
    return result