#!/usr/bin/env python3
"""Portable SIGKILL fixture for checkpoint and side-effect boundaries."""

from __future__ import annotations

import json
import os
from pathlib import Path
import sqlite3
import subprocess
import sys
import tempfile
import threading
import unittest


KEY = "synthetic-tool-call-1"


def save_checkpoint(root: Path, messages: list[dict[str, str]]) -> None:
    target = root / "session.json"
    pending = root / "session.json.tmp"
    with pending.open("w", encoding="utf-8") as handle:
        json.dump({"in_progress": True, "messages": messages}, handle)
        handle.flush()
        os.fsync(handle.fileno())
    os.replace(pending, target)
    directory_fd = os.open(root, os.O_RDONLY)
    try:
        os.fsync(directory_fd)
    finally:
        os.close(directory_fd)


def record_effect(root: Path) -> None:
    with sqlite3.connect(root / "effects.sqlite3") as database:
        database.execute(
            "CREATE TABLE IF NOT EXISTS effects (key TEXT PRIMARY KEY, outcome TEXT NOT NULL)"
        )
        database.execute(
            "INSERT INTO effects (key, outcome) VALUES (?, ?)", (KEY, "created")
        )
        database.commit()


def child(root: Path, stop_at: str) -> None:
    save_checkpoint(root, [{"role": "user", "content": "create synthetic item"}])
    print("before_effect", flush=True)
    if stop_at == "before_effect":
        threading.Event().wait()

    record_effect(root)
    print("after_effect", flush=True)
    if stop_at == "after_effect":
        threading.Event().wait()

    save_checkpoint(
        root,
        [
            {"role": "user", "content": "create synthetic item"},
            {"role": "assistant", "content": "tool call"},
            {"role": "tool", "content": "created"},
        ],
    )
    print("after_checkpoint", flush=True)
    threading.Event().wait()


def wait_for_marker(process: subprocess.Popen[str], expected: str) -> None:
    result: list[str] = []

    def read_line() -> None:
        assert process.stdout is not None
        result.append(process.stdout.readline().strip())

    while True:
        reader = threading.Thread(target=read_line, daemon=True)
        reader.start()
        reader.join(timeout=5)
        if reader.is_alive():
            raise TimeoutError(f"child did not emit {expected!r} within 5 seconds")
        if not result or result[-1] == "":
            raise RuntimeError(f"child exited before marker {expected!r}")
        if result[-1] == expected:
            return


def run_until_sigkill(root: Path, marker: str) -> None:
    process = subprocess.Popen(
        [sys.executable, __file__, "--child", str(root), marker],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    try:
        wait_for_marker(process, marker)
        process.kill()  # kills only the child PID created above
        process.wait(timeout=5)
        if process.returncode == 0:
            raise AssertionError("child was not killed")
    finally:
        if process.poll() is None:
            process.kill()
            process.wait(timeout=5)
        if process.stdout is not None:
            process.stdout.close()
        if process.stderr is not None:
            process.stderr.close()


def ledger_count(root: Path) -> int:
    database_path = root / "effects.sqlite3"
    if not database_path.exists():
        return 0
    with sqlite3.connect(database_path) as database:
        return int(database.execute("SELECT COUNT(*) FROM effects").fetchone()[0])


def message_count(root: Path) -> int:
    with (root / "session.json").open(encoding="utf-8") as handle:
        return len(json.load(handle)["messages"])


class SigkillBoundaryTest(unittest.TestCase):
    def check_boundary(self, marker: str, effects: int, messages: int) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            run_until_sigkill(root, marker)
            self.assertEqual(ledger_count(root), effects)
            self.assertEqual(message_count(root), messages)
            self.assertFalse((root / "session.json.tmp").exists())

    def test_kill_before_side_effect(self) -> None:
        self.check_boundary("before_effect", effects=0, messages=1)

    def test_kill_after_side_effect_before_result_checkpoint(self) -> None:
        self.check_boundary("after_effect", effects=1, messages=1)

    def test_kill_after_result_checkpoint(self) -> None:
        self.check_boundary("after_checkpoint", effects=1, messages=3)

    def test_duplicate_effect_is_rejected(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            record_effect(root)
            with self.assertRaises(sqlite3.IntegrityError):
                record_effect(root)
            self.assertEqual(ledger_count(root), 1)


if __name__ == "__main__":
    if len(sys.argv) == 4 and sys.argv[1] == "--child":
        child(Path(sys.argv[2]), sys.argv[3])
    else:
        unittest.main()