ironclaw/scripts/import-reborn-run-artifact.py

#!/usr/bin/env python3
"""Convert a downloaded Reborn run or thread artifact into a trace candidate."""

from __future__ import annotations

import argparse
import hashlib
import json
import pathlib
import re
import sys
from collections import OrderedDict
from typing import Any

SCHEMA = "ironclaw.run_artifact.v1"
THREAD_SCHEMA = "ironclaw.thread_artifact.v1"
PROMOTION_SCHEMA_VERSION = 1
UNSAFE_PATTERNS = (
    ("API key", re.compile(r"\b(?:sk-ant|sk-proj|sk-[A-Za-z0-9_-]{24,})\b")),
    ("OAuth token", re.compile(r"\b(?:ya29\.|xox[baprs]-)[A-Za-z0-9._-]{12,}\b")),
    (
        "GitHub token",
        re.compile(r"\b(?:ghp_[A-Za-z0-9_]{20,}|github_pat_[A-Za-z0-9_]{20,})\b"),
    ),
    ("bearer token", re.compile(r"\bBearer\s+[A-Za-z0-9._-]{20,}\b", re.IGNORECASE)),
    ("private key", re.compile(r"-----BEGIN [A-Z ]+PRIVATE KEY-----")),
    ("email address", re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b")),
    ("local path", re.compile(r"/(?:Users|home)/[^\s\"']+")),
)


def verify_scrubbed(value: Any, subject: str) -> None:
    serialized = json.dumps(value, sort_keys=True, separators=(",", ":"))
    unsafe_categories = sorted(
        label for label, pattern in UNSAFE_PATTERNS if pattern.search(serialized)
    )
    if unsafe_categories:
        categories = ", ".join(unsafe_categories)
        raise ValueError(
            f"{subject} failed independent scrub verification ({categories}); "
            "raw matches are intentionally omitted"
        )


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Convert a redacted run or thread artifact into an LLM trace candidate. "
            "Review assertions and external-service determinism before committing."
        )
    )
    parser.add_argument("artifact", type=pathlib.Path)
    parser.add_argument("output", type=pathlib.Path)
    parser.add_argument(
        "--model-name",
        help="Override the fixture model name (defaults to the captured provider model).",
    )
    parser.add_argument(
        "--source-url",
        required=True,
        help="Stable issue, incident, or live-run URL proving fixture provenance.",
    )
    parser.add_argument(
        "--owning-journey",
        required=True,
        help="Canonical journey or regression identifier that owns this candidate.",
    )
    return parser.parse_args()


def load_artifact_with_sha256(path: pathlib.Path) -> tuple[dict[str, Any], str]:
    try:
        artifact_bytes = path.read_bytes()
        artifact = json.loads(artifact_bytes)
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error:
        raise ValueError(f"could not read artifact: {error}") from error
    if artifact.get("schema") not in {SCHEMA, THREAD_SCHEMA}:
        raise ValueError("unsupported artifact schema")
    if artifact.get("redaction", {}).get("pipeline") != "deterministic-trace-redactor-v1":
        raise ValueError("artifact does not declare the required deterministic redaction pipeline")
    messages = artifact.get("messages")
    if not isinstance(messages, list) or not messages:
        raise ValueError("artifact has no replayable messages")
    verify_scrubbed(artifact, "artifact")
    return artifact, hashlib.sha256(artifact_bytes).hexdigest()


def load_artifact(path: pathlib.Path) -> dict[str, Any]:
    artifact, _sha256 = load_artifact_with_sha256(path)
    return artifact


def build_turn(messages: list[dict[str, Any]]) -> tuple[dict[str, Any], list[str]]:
    user = next((item for item in messages if item.get("kind") == "user"), None)
    if not user or not str(user.get("content", "")).strip():
        raise ValueError("artifact turn has no replayable user message")

    tool_groups: OrderedDict[str, list[dict[str, Any]]] = OrderedDict()
    for message in messages:
        tool_call = message.get("tool_call")
        if not isinstance(tool_call, dict):
            continue
        provider_turn_id = str(tool_call.get("provider_turn_id") or message.get("sequence"))
        tool_groups.setdefault(provider_turn_id, []).append(message)

    assistant = next(
        (
            item
            for item in reversed(messages)
            if item.get("kind") == "assistant" and str(item.get("content", "")).strip()
        ),
        None,
    )
    steps: list[dict[str, Any]] = []
    pending_results: list[dict[str, Any]] = []
    captured_models: list[str] = []

    def replay_tool_name(call: dict[str, Any]) -> str:
        # Provider protocol identity is reconstructed only at this replay
        # boundary; product artifacts remain capability-id neutral.
        return str(call["capability_id"]).replace(".", "__")

    for group in tool_groups.values():
        calls: list[dict[str, Any]] = []
        next_results: list[dict[str, Any]] = []
        for message in group:
            call = message["tool_call"]
            model = str(call.get("provider_model_id") or "").strip()
            if model:
                captured_models.append(model)
            calls.append(
                {
                    "id": call["provider_call_id"],
                    "name": replay_tool_name(call),
                    "arguments": call.get("arguments", {}),
                }
            )
            next_results.append(
                {
                    "tool_call_id": call["provider_call_id"],
                    "name": replay_tool_name(call),
                    "content": message.get("content", ""),
                }
            )
        step: dict[str, Any] = {
            "response": {
                "type": "tool_calls",
                "tool_calls": calls,
                "input_tokens": 0,
                "output_tokens": 0,
            }
        }
        if pending_results:
            step["expected_tool_results"] = pending_results
        steps.append(step)
        pending_results = next_results

    if assistant:
        step = {
            "response": {
                "type": "text",
                "content": assistant["content"],
                "input_tokens": 0,
                "output_tokens": 0,
            }
        }
        if pending_results:
            step["expected_tool_results"] = pending_results
        steps.append(step)
    elif pending_results:
        raise ValueError("artifact ends with tool results but no finalized assistant response")

    if not steps:
        raise ValueError("artifact turn has neither tool calls nor a finalized assistant response")

    tools = list(
        OrderedDict.fromkeys(
            call["name"]
            for step in steps
            for call in step["response"].get("tool_calls", [])
        )
    )
    return (
        {
            "user_input": user["content"],
            "steps": steps,
            "expects": {"tools_used": tools} if tools else {},
        },
        captured_models,
    )


def artifact_turns(
    artifact: dict[str, Any],
) -> tuple[list[list[dict[str, Any]]], list[dict[str, Any]]]:
    messages = sorted(artifact["messages"], key=lambda item: item.get("sequence", 0))
    if artifact.get("schema") == SCHEMA:
        return [messages], []

    grouped: OrderedDict[str, list[dict[str, Any]]] = OrderedDict()
    skipped_unscoped: list[dict[str, Any]] = []
    for message in messages:
        run_id = str(message.get("run_id") or "").strip()
        if (
            not run_id
            and message.get("kind") == "user"
            and message.get("status") == "accepted"
        ):
            skipped_unscoped.append(
                {
                    "sequence": message.get("sequence"),
                    "kind": message.get("kind"),
                    "status": message.get("status"),
                }
            )
            continue
        # Persisted submitted messages carry their run id. Keep an explicit
        # fallback group so malformed/manual artifacts fail review rather than
        # silently merging unscoped records into a neighboring run.
        key = run_id or f"unscoped:{message.get('sequence', len(grouped))}"
        grouped.setdefault(key, []).append(message)
    return list(grouped.values()), skipped_unscoped


def trace_candidate(
    artifact: dict[str, Any],
    model_override: str | None,
    source_url: str = "review-required://missing-provenance",
    owning_journey: str = "review-required",
    artifact_sha256: str | None = None,
) -> dict[str, Any]:
    turns: list[dict[str, Any]] = []
    captured_models: list[str] = []
    skipped_incomplete_runs: list[dict[str, Any]] = []
    message_groups, skipped_unscoped = artifact_turns(artifact)
    if not message_groups:
        raise ValueError("thread artifact has no complete run-scoped replayable turns")
    for messages in message_groups:
        user = next(
            (
                message
                for message in messages
                if message.get("kind") == "user"
                and str(message.get("content", "")).strip()
            ),
            None,
        )
        has_assistant_response = any(
            message.get("kind") == "assistant"
            and message.get("status") == "finalized"
            and str(message.get("content", "")).strip()
            for message in messages
        )
        if artifact.get("schema") == THREAD_SCHEMA and user and not has_assistant_response:
            skipped_incomplete_runs.append(
                {
                    "run_id": messages[0].get("run_id"),
                    "sequence": user.get("sequence"),
                    "reason": "run has no finalized assistant response",
                }
            )
            continue
        turn, turn_models = build_turn(messages)
        turns.append(turn)
        captured_models.extend(turn_models)
    if not turns:
        raise ValueError("thread artifact has no complete run-scoped replayable turns")

    model_name = model_override or (captured_models[0] if captured_models else "reborn-qa-import")
    required_actions = [
        "Add scenario-specific expects and caller-level end-state assertions.",
        "Review every redaction placeholder for acceptable fixture fidelity.",
        "Record or mock external HTTP/service exchanges before enabling hermetic CI replay.",
    ]
    if skipped_unscoped:
        required_actions.insert(
            0,
            "Review skipped_unscoped_messages from accepted submissions that never received a run ID.",
        )
    if skipped_incomplete_runs:
        required_actions.insert(
            0,
            "Review skipped_incomplete_runs that had no finalized assistant response.",
        )
    review = {
        "status": "candidate",
        "source_schema": artifact.get("schema"),
        "source_run_id": artifact.get("run", {}).get("run_id"),
        "source_thread_id": artifact.get("thread_id"),
        "logs_complete": artifact.get("logs", {}).get("complete", False),
        "required_actions": required_actions,
    }
    if skipped_unscoped:
        review["skipped_unscoped_messages"] = skipped_unscoped
    if skipped_incomplete_runs:
        review["skipped_incomplete_runs"] = skipped_incomplete_runs
    provenance = {
        "source_url": source_url,
        "artifact_schema": artifact.get("schema"),
    }
    if artifact_sha256 is not None:
        provenance["artifact_sha256"] = artifact_sha256
    candidate = {
        "_review": review,
        "_promotion": {
            "schema_version": PROMOTION_SCHEMA_VERSION,
            "provenance": provenance,
            "scrub": {
                "status": "verified",
                "pipeline": artifact.get("redaction", {}).get("pipeline"),
                "independent_scan": "passed",
            },
            "owning_journey": owning_journey,
            "deterministic_test": None,
            "last_successful_replay": None,
        },
        "model_name": model_name,
        "turns": turns,
    }
    verify_scrubbed(candidate, "fixture candidate")
    return candidate


def main() -> int:
    args = parse_args()
    try:
        artifact, artifact_sha256 = load_artifact_with_sha256(args.artifact)
        candidate = trace_candidate(
            artifact,
            args.model_name,
            args.source_url,
            args.owning_journey,
            artifact_sha256,
        )
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(json.dumps(candidate, indent=2) + "\n", encoding="utf-8")
    except (ValueError, KeyError, TypeError, AttributeError) as error:
        print(error, file=sys.stderr)
        return 2
    print(f"wrote review-required fixture candidate: {args.output}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())