#!/usr/bin/env python3
"""Exit-semantics tests for the receipt observer Stop-hook client."""

from __future__ import annotations

import importlib.util
import io
import json
import tempfile
import unittest
from contextlib import redirect_stderr
from pathlib import Path
from unittest import mock


TOOLS = Path(__file__).parents[2] / "framework/tools/lease-broker"
CLIENT_PATH = TOOLS / "receipt-observer-client.py"


def load_client():
    spec = importlib.util.spec_from_file_location("receipt_observer_client_test", CLIENT_PATH)
    if spec is None or spec.loader is None:
        raise RuntimeError("unable to load receipt-observer-client.py")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


CLIENT = load_client()
VALID_INPUT = json.dumps({"latest_assistant_message": "ordinary turn"}).encode()
DEEPLY_NESTED_JSON = b"[" * 2_000 + b"0" + b"]" * 2_000
ENVIRONMENT = {
    "MOSAIC_RECEIPT_OBSERVER_SOCKET": "/unused/observer.sock",
    "MOSAIC_LEASE_SESSION_ID": "a" * 64,
    "MOSAIC_RUNTIME_GENERATION": "1",
}


class FakeObserverSocket:
    def __init__(self, response: bytes) -> None:
        self.response = response

    def __enter__(self):
        return self

    def __exit__(self, *_args: object) -> None:
        return None

    def settimeout(self, _timeout: float) -> None:
        return None

    def connect(self, _path: str) -> None:
        return None

    def sendall(self, _payload: bytes) -> None:
        return None

    def shutdown(self, _how: int) -> None:
        return None

    def recv(self, _size: int) -> bytes:
        response, self.response = self.response, b""
        return response


class ReceiptObserverClientExitSemanticsTest(unittest.TestCase):
    def run_client(
        self,
        *,
        input_bytes: bytes = VALID_INPUT,
        reply: dict[str, object] | None = None,
        transport_error: OSError | None = None,
        runtime: str = "pi",
    ) -> tuple[int, str, mock.Mock]:
        request = mock.Mock(return_value=reply)
        if transport_error is not None:
            request.side_effect = transport_error
        stderr = io.StringIO()
        with (
            mock.patch.object(CLIENT.sys, "stdin", io.BytesIO(input_bytes)),
            mock.patch.object(CLIENT, "observer_request", request),
            redirect_stderr(stderr),
        ):
            arguments = ["--runtime", runtime]
            if runtime == "claude":
                arguments.append("--latest-entry")
            result = CLIENT.main(arguments, environ=ENVIRONMENT)
        return result, stderr.getvalue(), request

    def test_claude_prefers_inline_last_assistant_message(self) -> None:
        inline = "the just-finished assistant message"
        input_bytes = json.dumps({
            "last_assistant_message": inline,
            "transcript_path": "/must/not/be/opened.jsonl",
        }).encode()

        result, stderr, request = self.run_client(
            input_bytes=input_bytes,
            reply={"ok": True},
            runtime="claude",
        )

        self.assertEqual(result, 0)
        self.assertEqual(stderr, "")
        self.assertEqual(
            request.call_args.args[1]["latest_assistant_message"],
            inline,
        )

    def test_claude_falls_back_to_transcript_when_inline_field_is_absent(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            transcript = Path(directory) / "transcript.jsonl"
            transcript.write_text(
                json.dumps({
                    "message": {
                        "role": "assistant",
                        "content": [{"type": "text", "text": "fallback message"}],
                    }
                })
                + "\n",
                encoding="utf-8",
            )
            input_bytes = json.dumps({"transcript_path": str(transcript)}).encode()

            result, stderr, request = self.run_client(
                input_bytes=input_bytes,
                reply={"ok": True},
                runtime="claude",
            )

            self.assertEqual(result, 0)
            self.assertEqual(stderr, "")
            self.assertEqual(
                request.call_args.args[1]["latest_assistant_message"],
                "fallback message",
            )

    def test_claude_present_invalid_inline_field_fails_without_fallback(self) -> None:
        fallback = mock.Mock(return_value="must not be used")
        input_bytes = json.dumps({
            "last_assistant_message": None,
            "transcript_path": "/unused/transcript.jsonl",
        }).encode()

        with mock.patch.object(CLIENT, "claude_latest_entry", fallback):
            result, stderr, request = self.run_client(
                input_bytes=input_bytes,
                reply={"ok": True},
                runtime="claude",
            )

        self.assertEqual(result, 2)
        self.assertIn("Mosaic receipt observer refused", stderr)
        request.assert_not_called()
        fallback.assert_not_called()

    def test_claude_inline_message_size_guard_stays_enforced(self) -> None:
        request = mock.Mock(return_value={"ok": True})
        stderr = io.StringIO()
        with (
            mock.patch.object(CLIENT.sys, "stdin", io.BytesIO(b"{}")),
            mock.patch.object(
                CLIENT,
                "read_json",
                return_value={"last_assistant_message": "x" * (CLIENT.MAX_FRAME + 1)},
            ),
            mock.patch.object(CLIENT, "observer_request", request),
            redirect_stderr(stderr),
        ):
            result = CLIENT.main(
                ["--runtime", "claude", "--latest-entry"],
                environ=ENVIRONMENT,
            )

        self.assertEqual(result, 2)
        self.assertIn("Mosaic receipt observer refused", stderr.getvalue())
        request.assert_not_called()

    def test_pi_still_posts_only_its_runtime_message(self) -> None:
        result, stderr, request = self.run_client(reply={"ok": True})

        self.assertEqual(result, 0)
        self.assertEqual(stderr, "")
        self.assertEqual(
            request.call_args.args[1]["latest_assistant_message"],
            "ordinary turn",
        )

    def test_nothing_pending_observation_refusal_is_benign(self) -> None:
        result, stderr, request = self.run_client(
            reply={"ok": False, "code": "OBSERVATION_UNAVAILABLE"}
        )

        self.assertEqual(result, 0)
        self.assertEqual(stderr, "")
        request.assert_called_once()

    def test_success_reply_remains_successful(self) -> None:
        result, stderr, _request = self.run_client(reply={"ok": True})

        self.assertEqual(result, 0)
        self.assertEqual(stderr, "")

    def test_pending_cycle_auth_failure_stays_fail_closed(self) -> None:
        result, _stderr, _request = self.run_client(
            reply={"ok": False, "code": "ANCESTRY_MISMATCH"}
        )

        self.assertEqual(result, 2)

    def test_transport_failure_stays_fail_closed(self) -> None:
        result, stderr, _request = self.run_client(
            transport_error=ConnectionRefusedError("observer unavailable")
        )

        self.assertEqual(result, 2)
        self.assertIn("Mosaic receipt observer refused", stderr)

    def test_parse_failure_stays_fail_closed(self) -> None:
        result, stderr, request = self.run_client(input_bytes=b"{")

        self.assertEqual(result, 2)
        self.assertIn("Mosaic receipt observer refused", stderr)
        request.assert_not_called()

    def test_malformed_wire_replies_stay_fail_closed(self) -> None:
        for response in (
            b"not-json\n",
            b'{"ok":true}',
            b"{}\n{}\n",
            b'{"ok":false,"code":"OBSERVATION_UNAVAILABLE"}\n\n',
            b'{"ok":true,"ok":false,"code":"OBSERVATION_UNAVAILABLE"}\n',
            DEEPLY_NESTED_JSON + b"\n",
            b"x" * (CLIENT.MAX_FRAME + 1),
        ):
            with self.subTest(response=response):
                stderr = io.StringIO()
                with (
                    mock.patch.object(CLIENT.sys, "stdin", io.BytesIO(VALID_INPUT)),
                    mock.patch.object(
                        CLIENT.socket,
                        "socket",
                        return_value=FakeObserverSocket(response),
                    ),
                    redirect_stderr(stderr),
                ):
                    result = CLIENT.main(["--runtime", "pi"], environ=ENVIRONMENT)

                self.assertEqual(result, 2)
                self.assertIn("Mosaic receipt observer refused", stderr.getvalue())

    def test_oversized_input_stays_fail_closed(self) -> None:
        result, stderr, request = self.run_client(input_bytes=b"x" * (CLIENT.MAX_FRAME + 1))

        self.assertEqual(result, 2)
        self.assertIn("Mosaic receipt observer refused", stderr)
        request.assert_not_called()

    def test_deeply_nested_input_stays_fail_closed(self) -> None:
        result, stderr, request = self.run_client(input_bytes=DEEPLY_NESTED_JSON)

        self.assertEqual(result, 2)
        self.assertIn("Mosaic receipt observer refused", stderr)
        request.assert_not_called()

    def test_json_recursion_failure_stays_fail_closed(self) -> None:
        request = mock.Mock()
        stderr = io.StringIO()
        with (
            mock.patch.object(CLIENT.sys, "stdin", io.BytesIO(VALID_INPUT)),
            mock.patch.object(CLIENT, "observer_request", request),
            mock.patch.object(
                CLIENT.json,
                "loads",
                side_effect=RecursionError("maximum JSON nesting exceeded"),
            ),
            redirect_stderr(stderr),
        ):
            result = CLIENT.main(["--runtime", "pi"], environ=ENVIRONMENT)

        self.assertEqual(result, 2)
        self.assertIn("Mosaic receipt observer refused", stderr.getvalue())
        request.assert_not_called()

    def test_observation_unavailable_with_unexpected_fields_stays_fail_closed(self) -> None:
        result, _stderr, _request = self.run_client(
            reply={"ok": False, "code": "OBSERVATION_UNAVAILABLE", "unexpected": True}
        )

        self.assertEqual(result, 2)

    def test_non_boolean_ok_values_stay_fail_closed(self) -> None:
        for reply in (
            {"ok": 1},
            {"ok": 0, "code": "OBSERVATION_UNAVAILABLE"},
        ):
            with self.subTest(reply=reply):
                result, _stderr, _request = self.run_client(reply=reply)
                self.assertEqual(result, 2)


if __name__ == "__main__":
    unittest.main()