Normalize Glas execution evidence

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a028de-e2c8-7732-8521-46a7fc5db82f

src/activity_core/glas_evidence.py

@@ -0,0 +1,180 @@
+"""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_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
+    return result