stack/docs/compaction-refresh/probes/p6_claude_run.py

#!/usr/bin/env python3
"""Run real Claude 2.1.x through `mosaic yolo` for P6 observation."""

from __future__ import annotations

import hashlib
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Any

HERE = Path(__file__).resolve().parent
BLOCK = "\n".join(
    [
        "GATE0_CLAUDE_ATOMIC_BEGIN",
        "segment-01=alpha-2d11",
        "segment-02=middle-8e22",
        "segment-03=omega-4f33",
        "GATE0_CLAUDE_ATOMIC_END",
    ]
)


def strings(value: Any):
    if isinstance(value, str):
        yield value
    elif isinstance(value, list):
        for item in value:
            yield from strings(item)
    elif isinstance(value, dict):
        for item in value.values():
            yield from strings(item)


def main() -> None:
    with tempfile.TemporaryDirectory(prefix="gate0-p6-claude-") as temp:
        root = Path(temp)
        workspace = root / "workspace"
        workspace.mkdir()
        settings = root / "settings.json"
        hook_log = root / "hook.jsonl"
        settings.write_text(
            json.dumps(
                {
                    "hooks": {
                        "SessionStart": [
                            {
                                "hooks": [
                                    {
                                        "type": "command",
                                        "command": f'python3 "{HERE / "p6_claude_hook.py"}"',
                                        "timeout": 20,
                                    }
                                ]
                            }
                        ]
                    }
                }
            )
        )
        env = os.environ.copy()
        env["GATE0_CLAUDE_HOOK_LOG"] = str(hook_log)
        command = [
            "mosaic",
            "yolo",
            "claude",
            "--settings",
            str(settings),
            "--model",
            "haiku",
            "--print",
            "--output-format",
            "stream-json",
            "--verbose",
            "--include-hook-events",
            "--max-budget-usd",
            "0.10",
            "Return only the exact full GATE0_CLAUDE_ATOMIC_BEGIN through GATE0_CLAUDE_ATOMIC_END block injected by SessionStart, with no code fence or commentary.",
        ]
        result = subprocess.run(
            command,
            cwd=workspace,
            env=env,
            stdin=subprocess.DEVNULL,
            text=True,
            capture_output=True,
            timeout=150,
            check=False,
        )
        events: list[dict[str, Any]] = []
        for line in result.stdout.splitlines():
            try:
                events.append(json.loads(line))
            except json.JSONDecodeError:
                continue
        hook_events = [
            event
            for event in events
            if event.get("type") == "system"
            and event.get("subtype") in {"hook_started", "hook_response"}
        ]
        full_matches = [text for event in events for text in strings(event) if BLOCK in text]
        exact_matches = [text for event in events for text in strings(event) if text == BLOCK]
        assistant_texts: list[str] = []
        for event in events:
            if event.get("type") != "assistant":
                continue
            for text in strings(event.get("message", {})):
                if "GATE0_CLAUDE_ATOMIC_BEGIN" in text:
                    assistant_texts.append(text)

        if result.returncode != 0:
            raise AssertionError(f"Claude probe exited {result.returncode}")
        if not any(event.get("subtype") == "hook_response" and event.get("outcome") == "success" for event in hook_events):
            raise AssertionError("Claude SessionStart hook did not complete successfully")
        if BLOCK not in exact_matches:
            raise AssertionError("Claude did not return an exact full-block field")

        print("$ python3 docs/compaction-refresh/probes/p6_claude_run.py")
        print("machine_assertions=PASS")
        print("command=mosaic yolo claude --settings <isolated> --model haiku --print --output-format stream-json --verbose --include-hook-events <prompt>")
        print("claude_version=" + subprocess.check_output(["claude", "--version"], text=True).strip())
        print("mosaic_version=" + subprocess.check_output(["mosaic", "--version"], text=True).strip())
        print(f"exit_code={result.returncode}")
        print("hook_process_log=" + hook_log.read_text().strip())
        for event in hook_events:
            print("hook_stream_event=" + json.dumps(event, sort_keys=True))
        print(f"block_length={len(BLOCK.encode())}")
        print(f"block_sha256={hashlib.sha256(BLOCK.encode()).hexdigest()}")
        print(f"stream_fields_containing_full_block={len(full_matches)}")
        print(f"stream_fields_exactly_equal_block={len(exact_matches)}")
        for text in assistant_texts:
            print(f"assistant_copy_length={len(text.encode())}")
            print(f"assistant_copy_sha256={hashlib.sha256(text.encode()).hexdigest()}")
            print(f"assistant_copy_exact={text == BLOCK}")
            print("assistant_copy=" + json.dumps(text))
        if result.stderr.strip():
            print("stderr_excerpt=" + json.dumps(result.stderr.splitlines()[:10]))


if __name__ == "__main__":
    main()