"""Python hook callables through the real daemon: a callback the model
registers in one cell gates run()/files in later cells, in the same kernel,
and goes away with it. Catches a kernel-side gate that never sees callbacks
registered by a previous cell (a registry reset per cell, or one bound to the
wrong module instance), which unit tests of albedo_hooks cannot reach."""

import json
import unittest

from harness import Albedo, Provider, python, text


class PythonHooksTests(unittest.TestCase):
    def setUp(self):
        self.replies = []
        self.provider = Provider(
            lambda request: self.replies.pop(0) if self.replies else text("done")
        )
        self.addCleanup(self.provider.close)
        self.app = Albedo(self.provider, protocol="responses")
        self.app.__enter__()
        self.addCleanup(self.app.__exit__, None, None, None)

    def session(self):
        sid = self.app.session()
        with self.app.api(
            f"/sessions/{sid}/extensions", {"name": "hooks", "enabled": True}
        ) as response:
            json.load(response)
        return sid

    def turn(self, sid):
        before = len(self.provider.requests)
        self.app.prompt(sid, "exercise hooks").close()
        self.app.idle(sid)
        return [
            json.dumps(e["request"]["input"]) for e in self.provider.requests[before:]
        ]

    def test_callback_blocks_rewrites_and_adds_context(self):
        sid = self.session()
        blocked = self.app.workspace / "blocked.txt"
        rewritten = self.app.workspace / "rewritten.txt"
        moved = self.app.workspace / "moved.txt"
        self.replies = [
            python(
                "def gate(i):\n"
                "    path = i['tool_input']['file_path']\n"
                "    if path.endswith('blocked.txt'):\n"
                "        return {'decision': 'block', 'reason': 'PY_CALLBACK_NO'}\n"
                "    if path.endswith('moved.txt'):\n"
                f"        return {{'hookSpecificOutput': {{'permissionDecision': 'allow', 'updatedInput': {{'file_path': {str(rewritten)!r}, 'content': 'x'}}}}}}\n"
                "    return {'hookSpecificOutput': {'additionalContext': 'PY_CALLBACK_CONTEXT'}}\n"
                "remove = hooks.register('PreToolUse', gate, 'Write')\n"
                "print(hooks.registered())"
            ),
            python(
                "try:\n"
                f"    files.write({str(blocked)!r}, 'no')\n"
                "except PermissionError as error:\n"
                "    print('RAISED', error)\n"
                f"files.write({str(moved)!r}, 'y')"
            ),
            python("remove(); print('REMOVED', hooks.registered())"),
            text("done"),
        ]
        requests = self.turn(sid)
        self.assertEqual(len(requests), 4)
        self.assertIn("PY_CALLBACK_NO", requests[2])
        self.assertIn("RAISED", requests[2])
        self.assertIn("PY_CALLBACK_CONTEXT", requests[2])
        self.assertIn("REMOVED", requests[3])
        self.assertFalse(blocked.exists())
        self.assertFalse(moved.exists())
        self.assertEqual(rewritten.read_text(), "x")

    def test_callbacks_on_outer_events_stop_and_outer_tool_call(self):
        sid = self.session()
        self.replies = [
            python(
                "seen = []\n"
                "def pre(i):\n"
                "    seen.append(i['tool_name'])\n"
                "    return {'hookSpecificOutput': {'additionalContext': 'PY_OUTER_' + 'PRE'}}\n"
                "def stop(i):\n"
                "    if not i['stop_hook_active']:\n"
                "        return {'decision': 'block', 'reason': 'PY_STOP_REASON'}\n"
                "hooks.register('PreToolUse', pre, '^python$')\n"
                "hooks.register('Stop', stop)\n"
                "print('registered')"
            ),
            python("print('second cell')"),
            text("first answer"),
            text("second answer"),
        ]
        requests = self.turn(sid)
        # the registering cell ran before its callbacks existed, so only the
        # second cell is gated; the Stop callback then continues the turn once
        self.assertEqual(len(requests), 4)
        self.assertNotIn("PY_OUTER_PRE", requests[1])
        self.assertIn("PY_OUTER_PRE", requests[2])
        self.assertIn("PY_STOP_REASON", requests[3])
        self.assertIn("first answer", requests[3])


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