from __future__ import annotations

import asyncio
import inspect
import json
import sqlite3
import struct
import tempfile
import unittest
from pathlib import Path

from klbr.hooks import AttentionDecision, DeliveryDecision, describe
from klbr_runtime.release import fingerprint
from klbr_runtime.wire import MAX_FRAME, ProtocolError, encode, read_frame
from peer import ROOT, behavior_copy


class FrameTests(unittest.IsolatedAsyncioTestCase):
    async def read(self,data):
        reader=asyncio.StreamReader();reader.feed_data(data);reader.feed_eof()
        return await read_frame(reader)

    async def test_length_prefix_roundtrip_unicode(self):
        value={"test":"hello 🌙"}
        frame=encode(value)
        self.assertEqual(struct.unpack('>I',frame[:4])[0],len(frame)-4)
        self.assertEqual(await self.read(frame),value)

    async def test_truncated_frames_fail(self):
        for frame in [b'\x00\x00',struct.pack('>I',20)+b'{}']:
            with self.assertRaises(asyncio.IncompleteReadError): await self.read(frame)

    async def test_oversize_rejected_before_payload_read(self):
        with self.assertRaises(ProtocolError): await self.read(struct.pack('>I',MAX_FRAME+1))
        with self.assertRaises(ProtocolError): encode({"large":"x"*MAX_FRAME})

    async def test_duplicate_keys_and_nonfinite_values_rejected(self):
        for raw in [b'{"a":1,"a":2}',b'{"x":NaN}',b'{"x":Infinity}',b'\xff',b'{']:
            with self.subTest(raw=raw),self.assertRaises(ProtocolError):
                await self.read(struct.pack('>I',len(raw))+raw)

    async def test_shared_golden_frames(self):
        fixtures=json.loads((ROOT/"tests/runtime-fixtures/envelopes.json").read_text())
        for fixture in fixtures:
            with self.subTest(kind=fixture["message"]["kind"]):
                self.assertEqual(await self.read(encode(fixture)),fixture)


class ValueTests(unittest.TestCase):
    def test_decision_shapes_are_explicit(self):
        # A mistyped public SDK slot must fail locally, not kill a workbench by
        # sending an invalid typed wire variant to the host.
        with self.assertRaises(ValueError):
            asyncio.run(describe("misspelled.slot"))
        self.assertEqual(DeliveryDecision.allow().to_wire(),{"action":"allow"})
        self.assertEqual(AttentionDecision.defer(8).to_wire(),{"action":"defer","seconds":8})
        for decision in [DeliveryDecision("allow","invalid"),DeliveryDecision.hold(""),
                         AttentionDecision.defer(float("nan")),AttentionDecision.defer(-1)]:
            with self.subTest(decision=decision),self.assertRaises(ValueError): decision.to_wire()

    def test_behavior_identity_changes_with_source(self):
        with tempfile.TemporaryDirectory() as directory:
            root=behavior_copy(Path(directory)/"candidate")
            original=fingerprint(root)
            self.assertEqual(original,fingerprint(ROOT/"behavior"))
            with (root/"klbr_hooks/defaults.py").open("a") as file: file.write("\n# a change\n")
            self.assertNotEqual(original,fingerprint(root))

    def test_behavior_symlinks_rejected(self):
        with tempfile.TemporaryDirectory() as directory:
            root=behavior_copy(Path(directory)/"candidate")
            (root/"outside.py").symlink_to('/etc/passwd')
            with self.assertRaises(ValueError): fingerprint(root)


    def test_behavior_root_symlinks_rejected(self):
        with tempfile.TemporaryDirectory() as directory:
            root=Path(directory)/"alias"
            root.symlink_to(ROOT/"behavior",target_is_directory=True)
            with self.assertRaises(ValueError): fingerprint(root)

    def test_oversized_behavior_file_rejected(self):
        with tempfile.TemporaryDirectory() as directory:
            root=behavior_copy(Path(directory)/"candidate")
            with (root/"oversized.py").open("wb") as file:
                file.truncate(2_097_153)
            with self.assertRaises(ValueError): fingerprint(root)

    def test_shared_behavior_digest(self):
        fixture=json.loads((ROOT/"tests/runtime-fixtures/behavior.json").read_text())
        self.assertEqual(fingerprint(ROOT/"behavior"),fixture["sha256"])


class SchemaTests(unittest.TestCase):
    """Executes the exact shipped migration, not Rust Store methods.

    Python's linked SQLite is only used for these serial schema tests. The Rust
    runtime separately enforces the documented WAL-reset fixed-version floor.
    """
    def setUp(self):
        self.db=sqlite3.connect(':memory:')
        self.db.execute('PRAGMA foreign_keys=ON')
        self.db.executescript((ROOT/'klbr-runtime/migrations/001_runtime.sql').read_text())
        self.db.execute("INSERT INTO sessions(id) VALUES('root')")
        self.db.execute("INSERT INTO sessions(id) VALUES('child')")
        self.db.execute("INSERT INTO releases(id,path) VALUES('revision','/test')")
        self.db.commit()

    def tearDown(self): self.db.close()

    def event(self,key=None,session='root',kind='input'):
        return self.db.execute("INSERT INTO events(session_id,kind,source_key,payload,created_ms) VALUES(?,?,?,?,0)",
                               (session,kind,key,'{}')).lastrowid

    def execution(self,identity='cell',session='root'):
        event=self.event(session=session,kind='execution.started')
        self.db.execute("INSERT INTO executions(id,session_id,generation,hook_revision,state,started_event) VALUES(?,?,'g','revision','running',?)",
                        (identity,session,event))

    def test_schema_identity(self):
        self.assertEqual(self.db.execute('PRAGMA user_version').fetchone()[0],1)
        self.assertEqual(self.db.execute('PRAGMA application_id').fetchone()[0],1263288882)

    def test_originals_cannot_be_updated_or_deleted(self):
        identity=self.event()
        for sql in ['UPDATE events SET created_ms=1 WHERE seq=?','DELETE FROM events WHERE seq=?']:
            with self.assertRaises(sqlite3.IntegrityError): self.db.execute(sql,(identity,))

    def test_source_identity_is_unique_per_session(self):
        self.event('source-1')
        with self.assertRaises(sqlite3.IntegrityError): self.event('source-1')
        self.event('source-1',session='child')

    def test_burst_has_one_inbox_row_per_input(self):
        for i in range(500):
            event=self.event(f'input-{i}')
            self.db.execute('INSERT INTO inbox(event_id) VALUES(?)',(event,))
        self.assertEqual(self.db.execute("SELECT count(*) FROM inbox WHERE status='pending'").fetchone()[0],500)

    def test_foreign_keys_prevent_orphaned_inbox(self):
        with self.assertRaises(sqlite3.IntegrityError): self.db.execute('INSERT INTO inbox(event_id) VALUES(999)')

    def test_unfinished_execution_unique_with_independent_sessions(self):
        self.execution()
        with self.assertRaises(sqlite3.IntegrityError): self.execution('second')
        self.execution('child-cell',session='child')

    def test_yield_does_not_open_a_second_foreground_slot(self):
        self.execution()
        self.db.execute("UPDATE executions SET state='yielded' WHERE id='cell'")
        with self.assertRaises(sqlite3.IntegrityError): self.execution('second')

    def test_confirmed_effect_requires_receipt(self):
        self.execution()
        with self.assertRaises(sqlite3.IntegrityError):
            self.db.execute("INSERT INTO effects(id,execution_id,request_id,text,status) VALUES('e','cell','r','hello','confirmed')")

    def test_publication_transaction_rolls_back_receipt_and_event_together(self):
        self.execution();self.db.commit()
        before=self.db.execute('SELECT count(*) FROM events').fetchone()[0]
        self.db.execute('BEGIN IMMEDIATE')
        event=self.event(kind='local.sent')
        self.db.execute("INSERT INTO effects(id,execution_id,request_id,text,status,event_id) VALUES('e','cell','r','hello','confirmed',?)",(event,))
        self.db.rollback()
        self.assertEqual(self.db.execute('SELECT count(*) FROM effects').fetchone()[0],0)
        self.assertEqual(self.db.execute('SELECT count(*) FROM events').fetchone()[0],before)

    def test_dedup_is_by_request_not_message_text(self):
        self.execution()
        for identity in ['a','b']:
            self.db.execute("INSERT INTO effects(id,execution_id,request_id,text,status) VALUES(?,'cell',?,'same','proposed')",(identity,identity))
        with self.assertRaises(sqlite3.IntegrityError):
            self.db.execute("INSERT INTO effects(id,execution_id,request_id,text,status) VALUES('c','cell','a','other','proposed')")
        self.assertEqual(self.db.execute('SELECT count(*) FROM effects').fetchone()[0],2)

    def test_impossible_ready_with_deadline_is_rejected(self):
        with self.assertRaises(sqlite3.IntegrityError):
            self.db.execute("UPDATE sessions SET state='ready',wake_at_ms=100 WHERE id='root'")


class PeerSurfaceTests(unittest.TestCase):
    """The peer surface names its subject once; the duplicated functional layer stays gone."""

    def test_no_callable_repeats_peer_id(self):
        import klbr.peer as peer
        for name in ["send", "get", "stop"]:
            signature = inspect.signature(getattr(peer.PeerSession, name))
            self.assertNotIn("peer_id", signature.parameters, name)
        for name in ["result", "cancel"]:
            signature = inspect.signature(getattr(peer.PeerRequestHandle, name))
            self.assertNotIn("peer_id", signature.parameters, name)

    def test_subjectless_verbs_are_the_whole_module_surface(self):
        import klbr.peer as peer
        self.assertEqual(sorted(peer.__all__), ["PeerRequestHandle", "PeerSession", "answer", "open", "peers"])
        for gone in ["send", "result", "cancel", "get", "stop", "list_peers", "submit_result"]:
            self.assertFalse(hasattr(peer, gone), f"peer.{gone} repeats a subject the handle already carries")

    def test_the_agents_alias_vocabulary_is_gone(self):
        with self.assertRaises(ModuleNotFoundError):
            import klbr.agents  # noqa: F401