from __future__ import annotations

import asyncio
import json
import logging
from pathlib import Path

import pytest

from agent import runner
from agent.models import AgentResult, TerminateReason
from agent.prompts import DESIGNER_PROMPT_NAME
from storage.library_manifest import manifest_by_id
from storage.repo import StorageRepo
from tests.helpers import FakeAgent


@pytest.fixture
def fake_agent(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(runner, "ArticraftAgent", FakeAgent)


def _active_revision_dir(record_dir: Path) -> Path:
    record = json.loads((record_dir / "record.json").read_text(encoding="utf-8"))
    return record_dir / "revisions" / record.get("active_revision_id", "rev_000001")


def _artifact_path(record_dir: Path, filename: str) -> Path:
    return _active_revision_dir(record_dir) / filename


class DeletingImageAgent(FakeAgent):
    async def run(self, user_content: object):  # type: ignore[override]
        if isinstance(user_content, list):
            for item in user_content:
                if not isinstance(item, dict):
                    continue
                image_path = item.get("image_path")
                if isinstance(image_path, str):
                    Path(image_path).unlink()
                    break
        return await super().run(user_content)


class MeshVisualAgent(FakeAgent):
    async def run(self, user_content: object):  # type: ignore[override]
        self.file_path.parent.mkdir(parents=True, exist_ok=True)
        self.file_path.write_text(
            "from __future__ import annotations\n\nobject_model = None\n",
            encoding="utf-8",
        )
        meshes_dir = self.file_path.parent / "assets" / "meshes"
        meshes_dir.mkdir(parents=True, exist_ok=True)
        (meshes_dir / "part.obj").write_text(
            "\n".join(
                [
                    "o part",
                    "v 0 0 0",
                    "v 1 0 0",
                    "v 0 1 0",
                    "f 1 2 3",
                    "",
                ]
            ),
            encoding="utf-8",
        )
        cost_path = self.file_path.parent / "cost.json"
        cost_path.write_text(
            json.dumps({"total": {"costs_usd": {"total": 0.123456}}}, indent=2),
            encoding="utf-8",
        )
        if self.trace_dir is not None:
            self.trace_dir.mkdir(parents=True, exist_ok=True)
            (self.trace_dir / "trajectory.jsonl").write_text(
                '{"type":"message","message":{"role":"assistant","content":"done"}}\n',
                encoding="utf-8",
            )
        return AgentResult(
            success=True,
            reason=TerminateReason.CODE_VALID,
            message="done",
            conversation=[{"role": "user", "content": user_content}],
            final_code=self.file_path.read_text(encoding="utf-8"),
            urdf_xml=(
                "<robot name='mesh_visual'>"
                "<link name='base'>"
                "<visual><geometry><mesh filename='assets/meshes/part.obj'/></geometry></visual>"
                "</link>"
                "</robot>"
            ),
            compile_warnings=[],
            turn_count=3,
            tool_call_count=5,
            compile_attempt_count=2,
            usage={"prompt_tokens": 10, "candidates_tokens": 5, "total_tokens": 15},
        )


class MeshVisualWithUnusedAssetsAgent(MeshVisualAgent):
    async def run(self, user_content: object):  # type: ignore[override]
        result = await super().run(user_content)
        meshes_dir = self.file_path.parent / "assets" / "meshes"
        (meshes_dir / "orphan.obj").write_text("# orphan\n", encoding="utf-8")
        glb_dir = self.file_path.parent / "assets" / "glb"
        glb_dir.mkdir(parents=True, exist_ok=True)
        (glb_dir / "orphan.glb").write_bytes(b"orphan")
        return result


class ContextResetAgent(FakeAgent):
    async def run(self, user_content: object):  # type: ignore[override]
        result = await super().run(user_content)
        result.context_reset_count = 2
        return result


class OverBudgetAgent(FakeAgent):
    async def run(self, user_content: object):  # type: ignore[override]
        self.file_path.parent.mkdir(parents=True, exist_ok=True)
        self.file_path.write_text("# over budget\n", encoding="utf-8")
        (self.file_path.parent / "cost.json").write_text(
            json.dumps({"total": {"costs_usd": {"total": 0.75}}}, indent=2),
            encoding="utf-8",
        )
        return AgentResult(
            success=False,
            reason=TerminateReason.COST_LIMIT,
            message="Cost limit exceeded after turn 2: cumulative $0.750000 exceeded limit $0.500000",
            conversation=[{"role": "user", "content": user_content}],
            final_code=self.file_path.read_text(encoding="utf-8"),
            urdf_xml=None,
            compile_warnings=[],
            turn_count=2,
            tool_call_count=1,
            compile_attempt_count=0,
            usage={"prompt_tokens": 10, "candidates_tokens": 5, "total_tokens": 15},
        )


class AllInTotalsAgent(FakeAgent):
    async def run(self, user_content: object):  # type: ignore[override]
        self.file_path.parent.mkdir(parents=True, exist_ok=True)
        self.file_path.write_text(
            "from __future__ import annotations\n\nobject_model = None\n",
            encoding="utf-8",
        )
        (self.file_path.parent / "cost.json").write_text(
            json.dumps(
                {
                    "total": {
                        "tokens": {
                            "prompt_tokens": 100,
                            "candidates_tokens": 20,
                            "total_tokens": 120,
                            "cached_tokens": 0,
                        },
                        "costs_usd": {"total": 0.75},
                    },
                    "all_in_total": {
                        "tokens": {
                            "prompt_tokens": 140,
                            "candidates_tokens": 25,
                            "total_tokens": 165,
                            "cached_tokens": 10,
                            "uncached_prompt_tokens": 130,
                        },
                        "costs_usd": {"total": 0.91},
                    },
                },
                indent=2,
            ),
            encoding="utf-8",
        )
        return AgentResult(
            success=True,
            reason=TerminateReason.CODE_VALID,
            message="done",
            conversation=[{"role": "user", "content": user_content}],
            final_code=self.file_path.read_text(encoding="utf-8"),
            urdf_xml="<robot name='test'/>",
            compile_warnings=[],
            turn_count=1,
            tool_call_count=0,
            compile_attempt_count=0,
            usage={"prompt_tokens": 100, "candidates_tokens": 20, "total_tokens": 120},
        )


def test_library_run_and_rerun_persist_runtime_artifacts(
    fake_agent: None,
    tmp_path: Path,
) -> None:
    repo_root = tmp_path
    exit_code = asyncio.run(
        runner.run_from_input(
            "make a gearbox",
            prompt_text="make a gearbox",
            display_prompt="make a gearbox",
            repo_root=repo_root,
            image_path=None,
            provider="openai",
            thinking_level="high",
            max_turns=30,
            system_prompt_path=DESIGNER_PROMPT_NAME,
            sdk_package="sdk",
            label="gearbox try",
            tags=["gear", "test"],
        )
    )
    assert exit_code == 0

    records_root = repo_root / "data" / "records"
    record_dirs = [path for path in records_root.iterdir() if path.is_dir()]
    assert len(record_dirs) == 1
    record_dir = record_dirs[0]
    materialization_dir = repo_root / "data" / "cache" / "record_materialization" / record_dir.name

    revision_dir = _active_revision_dir(record_dir)
    assert _artifact_path(record_dir, "prompt.txt").read_text(encoding="utf-8") == "make a gearbox"
    assert _artifact_path(record_dir, "model.py").exists()
    assert _artifact_path(record_dir, "provenance.json").exists()
    assert _artifact_path(record_dir, "cost.json").exists()
    assert (revision_dir / "traces").exists()
    assert (revision_dir / "traces" / "trajectory.jsonl.zst").exists()
    assert not (revision_dir / "traces" / "trajectory.jsonl").exists()
    assert (materialization_dir / "model.urdf").read_text(
        encoding="utf-8"
    ) == "<robot name='test'/>"
    assert (materialization_dir / "compile_report.json").exists()
    compile_report = json.loads(
        (materialization_dir / "compile_report.json").read_text(encoding="utf-8")
    )
    assert compile_report["metrics"]["compile_level"] == "full"
    assert compile_report["metrics"]["fingerprint_inputs"]["model_py_sha256"]
    assert compile_report["metrics"]["materialization_fingerprint"]
    assert not (materialization_dir / "assets" / "meshes").exists()

    record = json.loads((record_dir / "record.json").read_text(encoding="utf-8"))
    assert record["record_id"] == record_dir.name
    assert record["label"] == "gearbox try"
    assert record["tags"] == ["gear", "test"]
    manifest_row = manifest_by_id(StorageRepo(repo_root))[record_dir.name]
    assert manifest_row["label"] == "gearbox try"
    assert manifest_row["tags"] == ["gear", "test"]

    runs_root = repo_root / "data" / "cache" / "runs"
    run_dirs = [path for path in runs_root.iterdir() if path.is_dir()]
    assert len(run_dirs) == 1
    run_metadata = json.loads((run_dirs[0] / "run.json").read_text(encoding="utf-8"))
    assert run_metadata["status"] == "success"
    results_lines = (run_dirs[0] / "results.jsonl").read_text(encoding="utf-8").splitlines()
    assert len(results_lines) == 1
    assert json.loads(results_lines[0])["record_id"] == record_dir.name
    assert not (run_dirs[0] / "staging" / record_dir.name).exists()
    assert not (repo_root / "outputs").exists()

    original_record = json.loads((record_dir / "record.json").read_text(encoding="utf-8"))
    original_run_id = original_record["source"]["run_id"]
    original_created_at = original_record["created_at"]
    original_provenance = json.loads(
        _artifact_path(record_dir, "provenance.json").read_text(encoding="utf-8")
    )
    system_prompt_sha = original_provenance["prompting"]["system_prompt_sha256"]
    assert (repo_root / "data" / "system_prompts" / f"{system_prompt_sha}.txt").exists()
    _artifact_path(record_dir, "provenance.json").write_text(
        json.dumps(original_provenance, indent=2) + "\n",
        encoding="utf-8",
    )

    original_record["rating"] = 4
    original_record["rated_by"] = "mattzh72"
    (record_dir / "record.json").write_text(
        json.dumps(original_record, indent=2) + "\n",
        encoding="utf-8",
    )

    _artifact_path(record_dir, "model.py").write_text("# stale\n", encoding="utf-8")
    _artifact_path(record_dir, "cost.json").write_text('{"stale": true}\n', encoding="utf-8")
    (revision_dir / "traces" / "stale.txt").write_text("stale\n", encoding="utf-8")
    stale_glb_dir = record_dir / "assets" / "glb"
    stale_glb_dir.mkdir(parents=True, exist_ok=True)
    (stale_glb_dir / "stale.glb").write_text("stale\n", encoding="utf-8")
    stale_viewer_dir = record_dir / "assets" / "viewer"
    stale_viewer_dir.mkdir(parents=True, exist_ok=True)
    (stale_viewer_dir / "index.html").write_text("stale\n", encoding="utf-8")

    rerun_exit_code = asyncio.run(
        runner.rerun_record_in_place(
            repo_root=repo_root,
            record_id=record_dir.name,
        )
    )
    assert rerun_exit_code == 0

    record_dirs = [path for path in records_root.iterdir() if path.is_dir()]
    assert [path.name for path in record_dirs] == [record_dir.name]
    updated_record = json.loads((record_dir / "record.json").read_text(encoding="utf-8"))
    assert updated_record["record_id"] == record_dir.name
    assert updated_record["created_at"] == original_created_at
    assert updated_record["source"]["run_id"] != original_run_id
    assert (
        _artifact_path(record_dir, "model.py")
        .read_text(encoding="utf-8")
        .startswith("from __future__ import annotations")
    )
    assert (
        json.loads(_artifact_path(record_dir, "cost.json").read_text(encoding="utf-8"))["total"][
            "costs_usd"
        ]["total"]
        == 0.123456
    )
    updated_revision_dir = _active_revision_dir(record_dir)
    assert (updated_revision_dir / "traces" / "trajectory.jsonl.zst").exists()
    assert not (updated_revision_dir / "traces" / "trajectory.jsonl").exists()
    assert not (updated_revision_dir / "traces" / "stale.txt").exists()
    assert not (record_dir / "assets" / "glb").exists()
    assert not (record_dir / "assets" / "viewer").exists()
    assert not (materialization_dir / "assets" / "meshes").exists()

    assert updated_record["label"] == "gearbox try"
    assert updated_record["tags"] == ["gear", "test"]
    assert updated_record["rating"] == 4
    assert updated_record["rated_by"] == "mattzh72"
    manifest_row = manifest_by_id(StorageRepo(repo_root))[record_dir.name]
    assert manifest_row["label"] == "gearbox try"
    assert manifest_row["tags"] == ["gear", "test"]
    assert manifest_row["rating"] == 4

    run_dirs = sorted(path for path in runs_root.iterdir() if path.is_dir())
    assert len(run_dirs) == 2
    latest_run_metadata = json.loads((run_dirs[-1] / "run.json").read_text(encoding="utf-8"))
    assert latest_run_metadata["status"] == "success"
    latest_results = (run_dirs[-1] / "results.jsonl").read_text(encoding="utf-8").splitlines()
    assert len(latest_results) == 1
    assert json.loads(latest_results[0])["record_id"] == record_dir.name
    assert not (run_dirs[-1] / "staging" / record_dir.name).exists()


def test_run_from_input_stamps_author_and_rerun_preserves_existing_author(
    fake_agent: None,
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(runner, "_resolve_runtime_record_author", lambda repo_root: "mattzh72")

    exit_code = asyncio.run(
        runner.run_from_input(
            "make a hinge",
            prompt_text="make a hinge",
            display_prompt="make a hinge",
            repo_root=tmp_path,
            image_path=None,
            provider="openai",
            thinking_level="high",
            max_turns=10,
            system_prompt_path=DESIGNER_PROMPT_NAME,
            sdk_package="sdk",
        )
    )

    assert exit_code == 0
    record_dir = next((tmp_path / "data" / "records").iterdir())
    original_record = json.loads((record_dir / "record.json").read_text(encoding="utf-8"))
    assert original_record["author"] == "mattzh72"

    monkeypatch.setattr(runner, "_resolve_runtime_record_author", lambda repo_root: "RuiningLi")
    rerun_exit_code = asyncio.run(
        runner.rerun_record_in_place(
            repo_root=tmp_path,
            record_id=record_dir.name,
        )
    )

    assert rerun_exit_code == 0
    updated_record = json.loads((record_dir / "record.json").read_text(encoding="utf-8"))
    assert updated_record["author"] == "mattzh72"


def test_create_draft_record_stamps_author_and_manifest_immediately(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(runner, "_resolve_runtime_record_author", lambda repo_root: "mattzh72")

    record_dir = runner.create_draft_record(
        repo_root=tmp_path,
        prompt_text="make a draft hinge",
        provider="openai",
        thinking_level="high",
        max_turns=10,
        system_prompt_path=DESIGNER_PROMPT_NAME,
        sdk_package="sdk",
    )

    record = json.loads((record_dir / "record.json").read_text(encoding="utf-8"))
    assert record["author"] == "mattzh72"
    assert manifest_by_id(StorageRepo(tmp_path))[record_dir.name]["record_id"] == record_dir.name


def test_over_budget_run_persists_failed_run_metadata_without_record(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    monkeypatch.setattr(runner, "ArticraftAgent", OverBudgetAgent)

    exit_code = asyncio.run(
        runner.run_from_input(
            "make an expensive model",
            prompt_text="make an expensive model",
            display_prompt="make an expensive model",
            repo_root=tmp_path,
            image_path=None,
            provider="openai",
            thinking_level="high",
            max_turns=30,
            max_cost_usd=0.5,
            system_prompt_path=DESIGNER_PROMPT_NAME,
            sdk_package="sdk",
        )
    )

    assert exit_code == 2
    records_root = tmp_path / "data" / "records"
    assert not records_root.exists() or list(records_root.iterdir()) == []

    run_dir = next(
        path for path in (tmp_path / "data" / "cache" / "runs").iterdir() if path.is_dir()
    )
    run_metadata = json.loads((run_dir / "run.json").read_text(encoding="utf-8"))
    assert run_metadata["status"] == "failed"
    assert run_metadata["settings_summary"]["max_cost_usd"] == 0.5

    result_rows = [
        json.loads(line)
        for line in (run_dir / "results.jsonl").read_text(encoding="utf-8").splitlines()
    ]
    assert len(result_rows) == 1
    assert "Cost limit exceeded" in result_rows[0]["message"]

    staging_dirs = [path for path in (run_dir / "staging").iterdir() if path.is_dir()]
    assert len(staging_dirs) == 1
    assert (
        json.loads((staging_dirs[0] / "cost.json").read_text(encoding="utf-8"))["total"][
            "costs_usd"
        ]["total"]
        == 0.75
    )


def test_successful_run_logs_all_in_cost_totals(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    caplog: pytest.LogCaptureFixture,
) -> None:
    monkeypatch.setattr(runner, "ArticraftAgent", AllInTotalsAgent)

    with caplog.at_level(logging.INFO):
        exit_code = asyncio.run(
            runner.run_from_input(
                "make a model with compaction",
                prompt_text="make a model with compaction",
                display_prompt="make a model with compaction",
                repo_root=tmp_path,
                image_path=None,
                provider="openai",
                thinking_level="high",
                max_turns=30,
                system_prompt_path=DESIGNER_PROMPT_NAME,
                sdk_package="sdk",
                label=None,
                tags=[],
            )
        )

    assert exit_code == 0
    assert (
        "Total tokens: {'prompt_tokens': 140, 'cached_tokens': 10, 'uncached_prompt_tokens': 130, 'candidates_tokens': 25, 'total_tokens': 165}"
        in caplog.text
    )
    assert "Total cost: $0.910000" in caplog.text


def test_successful_run_preserves_visual_meshes_as_obj_by_default(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    monkeypatch.setattr(runner, "ArticraftAgent", MeshVisualAgent)

    exit_code = asyncio.run(
        runner.run_from_input(
            "make a mesh part",
            prompt_text="make a mesh part",
            display_prompt="make a mesh part",
            repo_root=tmp_path,
            image_path=None,
            provider="openai",
            thinking_level="high",
            max_turns=30,
            system_prompt_path=DESIGNER_PROMPT_NAME,
            sdk_package="sdk",
            label=None,
            tags=[],
        )
    )

    assert exit_code == 0

    records_root = tmp_path / "data" / "records"
    record_dirs = [path for path in records_root.iterdir() if path.is_dir()]
    assert len(record_dirs) == 1
    record_dir = record_dirs[0]
    materialization_dir = tmp_path / "data" / "cache" / "record_materialization" / record_dir.name

    assert "assets/meshes/part.obj" in (materialization_dir / "model.urdf").read_text(
        encoding="utf-8"
    )
    assert not (materialization_dir / "assets" / "meshes" / "part.glb").exists()
    assert (materialization_dir / "assets" / "meshes" / "part.obj").exists()


def test_successful_run_copies_only_referenced_mesh_assets(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    monkeypatch.setattr(runner, "ArticraftAgent", MeshVisualWithUnusedAssetsAgent)

    exit_code = asyncio.run(
        runner.run_from_input(
            "make a mesh part with extras",
            prompt_text="make a mesh part with extras",
            display_prompt="make a mesh part with extras",
            repo_root=tmp_path,
            image_path=None,
            provider="openai",
            thinking_level="high",
            max_turns=30,
            system_prompt_path=DESIGNER_PROMPT_NAME,
            sdk_package="sdk",
            label=None,
            tags=[],
        )
    )

    assert exit_code == 0

    records_root = tmp_path / "data" / "records"
    record_dirs = [path for path in records_root.iterdir() if path.is_dir()]
    assert len(record_dirs) == 1
    record_dir = record_dirs[0]
    materialization_dir = tmp_path / "data" / "cache" / "record_materialization" / record_dir.name

    assert "assets/meshes/part.obj" in (materialization_dir / "model.urdf").read_text(
        encoding="utf-8"
    )
    assert not (materialization_dir / "assets" / "meshes" / "part.glb").exists()
    assert (materialization_dir / "assets" / "meshes" / "part.obj").exists()
    assert not (materialization_dir / "assets" / "meshes" / "orphan.obj").exists()
    assert not (materialization_dir / "assets" / "glb").exists()


def test_library_run_and_rerun_preserve_category_metadata(
    fake_agent: None,
    tmp_path: Path,
) -> None:
    repo_root = tmp_path
    exit_code = asyncio.run(
        runner.run_from_input(
            "make a tower crane",
            prompt_text="make a tower crane",
            display_prompt="make a tower crane",
            repo_root=repo_root,
            image_path=None,
            provider="openai",
            thinking_level="high",
            max_turns=30,
            system_prompt_path=DESIGNER_PROMPT_NAME,
            sdk_package="sdk",
            category_slug="crane_tower",
        )
    )
    assert exit_code == 0

    records_root = repo_root / "data" / "records"
    record_dirs = [path for path in records_root.iterdir() if path.is_dir()]
    assert len(record_dirs) == 1
    record_dir = record_dirs[0]

    record = json.loads((record_dir / "record.json").read_text(encoding="utf-8"))
    assert record["category_slug"] == "crane_tower"
    assert not (record_dir / "collections").exists()
    manifest_row = manifest_by_id(StorageRepo(repo_root))[record_dir.name]
    assert manifest_row["category_slug"] == "crane_tower"
    assert manifest_row["category_title"] == "Crane Tower"

    runs_root = repo_root / "data" / "cache" / "runs"
    run_dirs = [path for path in runs_root.iterdir() if path.is_dir()]
    assert len(run_dirs) == 1
    run_metadata = json.loads((run_dirs[0] / "run.json").read_text(encoding="utf-8"))
    assert run_metadata["status"] == "success"
    assert run_metadata["run_mode"] == "library_single"
    assert not (run_dirs[0] / "staging" / record_dir.name).exists()

    original_run_id = record["source"]["run_id"]
    _artifact_path(record_dir, "model.py").write_text("# stale dataset\n", encoding="utf-8")

    rerun_exit_code = asyncio.run(
        runner.rerun_record_in_place(
            repo_root=repo_root,
            record_id=record_dir.name,
        )
    )
    assert rerun_exit_code == 0

    record = json.loads((record_dir / "record.json").read_text(encoding="utf-8"))
    assert record["record_id"] == record_dir.name
    assert record["source"]["run_id"] != original_run_id
    assert record["category_slug"] == "crane_tower"
    assert (
        _artifact_path(record_dir, "model.py")
        .read_text(encoding="utf-8")
        .startswith("from __future__ import annotations")
    )

    manifest_row = manifest_by_id(StorageRepo(repo_root))[record_dir.name]
    assert manifest_row["category_slug"] == "crane_tower"
    assert manifest_row["category_title"] == "Crane Tower"

    run_dirs = sorted(path for path in runs_root.iterdir() if path.is_dir())
    assert len(run_dirs) == 2
    latest_run_metadata = json.loads((run_dirs[-1] / "run.json").read_text(encoding="utf-8"))
    assert latest_run_metadata["status"] == "success"
    assert latest_run_metadata["run_mode"] == "library_single"
    assert not (run_dirs[-1] / "staging" / record_dir.name).exists()


def test_library_run_succeeds_when_persisted_input_image_disappears(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    monkeypatch.setattr(runner, "ArticraftAgent", DeletingImageAgent)

    repo_root = tmp_path
    image_path = repo_root / "reference.png"
    image_path.write_bytes(b"\x89PNG\r\n\x1a\n")

    exit_code = asyncio.run(
        runner.run_from_input(
            [
                {"type": "input_text", "text": "make a lamp"},
                {"type": "input_image", "image_path": str(image_path), "detail": "high"},
            ],
            prompt_text="make a lamp",
            display_prompt="make a lamp",
            repo_root=repo_root,
            image_path=image_path,
            provider="openai",
            thinking_level="high",
            max_turns=30,
            system_prompt_path=DESIGNER_PROMPT_NAME,
            sdk_package="sdk",
            label="lamp try",
            tags=["lamp"],
        )
    )
    assert exit_code == 0

    record_dir = next((repo_root / "data" / "records").iterdir())
    materialization_dir = repo_root / "data" / "cache" / "record_materialization" / record_dir.name
    assert (record_dir / "record.json").exists()
    assert _artifact_path(record_dir, "provenance.json").exists()
    assert (materialization_dir / "compile_report.json").exists()
    compile_report = json.loads(
        (materialization_dir / "compile_report.json").read_text(encoding="utf-8")
    )
    assert compile_report["metrics"]["compile_level"] == "full"
    assert compile_report["metrics"]["fingerprint_inputs"]["model_py_sha256"]
    assert compile_report["metrics"]["materialization_fingerprint"]
    assert list((_active_revision_dir(record_dir) / "inputs").iterdir()) == []

    run_dir = next((repo_root / "data" / "cache" / "runs").iterdir())
    run_metadata = json.loads((run_dir / "run.json").read_text(encoding="utf-8"))
    assert run_metadata["status"] == "success"
    results = (run_dir / "results.jsonl").read_text(encoding="utf-8").splitlines()
    assert len(results) == 1
    assert json.loads(results[0])["status"] == "success"
    assert not (run_dir / "staging" / record_dir.name).exists()