"""Safety contracts tested with a fake native world, never a second live owner."""
import copy
import json
from pathlib import Path
import socket
import tempfile
import threading
import time
import unittest
from unittest.mock import patch

from witcher_devkit import debugger, native
from witcher_devkit.debug_journal import Journal, durable_json, verify, recover_torn_tail
from witcher_devkit.debug_runtime import OperationUncertain, Runtime, validate


class World:
    def __init__(self):
        self.ticks=0; self.calls=[]; self.hp=100; self.gen=1; self.paused=True; self.events=[]
    def snapshot(self):
        return dict(ok=True,generation=self.gen,data=dict(ready=True,ticks=self.ticks,input_ticks=0,remaining_ticks=0,
            world=dict(devkit_paused=self.paused,player=dict(hp=self.hp,alive=self.hp>0))))
    def call(self,command,args='',timeout=5):
        self.calls.append((command,args))
        if command=='snapshot': return self.snapshot()
        if command in ('advance','input'):
            self.ticks+=int(args.split('|')[-1]); self.hp-=1
        if command=='event_page':
            return dict(ok=True,generation=self.gen,data=dict(latest=len(self.events),oldest=1,events=self.events[int(args.split('|')[1]):]))
        return dict(ok=True,generation=self.gen,data={})


class DebuggerTests(unittest.TestCase):
    def setUp(self):
        self.tmp=tempfile.TemporaryDirectory(); self.addCleanup(self.tmp.cleanup)
        self.root=Path(self.tmp.name); self.journal=Journal(self.root/'journal')
        self.world=World(); self.cancel=threading.Event(); self.runtime=Runtime(self.journal,self.cancel)
        self.patches=[patch.object(native,'call',side_effect=self.world.call),
                      patch.object(native,'state',return_value={'uncertain':None}),patch.object(native,'ROOT',self.root),
                      patch.object(native,'STATE_ROOT',self.root)]
        for p in self.patches: p.start(); self.addCleanup(p.stop)
        self.receipt=dict(token='secret',session_id='s',broker=debugger.identity())
        self.broker=debugger.Broker(self.receipt,self.journal)

    def request(self,method,**args): return self.broker.request(dict(method=method,token='secret',**args))
    def submit(self,opid='a',**extra):
        return self.request('submit',kind='step',args={'ticks':3},operation_id=opid,wall_seconds=5,expected_generation=None,**extra)
    def done(self):
        end=time.monotonic()+3
        while self.broker.active and time.monotonic()<end: time.sleep(.005)
        self.assertIsNone(self.broker.active)
        return self.broker.operations['a']

    def test_idempotent_operation_and_conflicting_arguments(self):
        self.submit(); self.assertEqual(self.done()['status'],'completed')
        self.submit(); self.assertEqual(self.world.calls.count(('advance','3')),1)
        with self.assertRaisesRegex(ValueError,'operation_id_conflict'):
            self.request('submit',kind='step',args={'ticks':4},operation_id='a',wall_seconds=5,expected_generation=None)

    def test_busy_cancel_single_writer(self):
        entered=threading.Event(); release=threading.Event()
        def work(*args): entered.set(); release.wait(2); return {'cancel_seen':self.broker.cancelled.is_set()}
        with patch.object(self.broker.runtime,'execute',side_effect=work):
            self.submit(); self.assertTrue(entered.wait(1))
            with self.assertRaisesRegex(ValueError,'busy'): self.submit('b')
            self.assertTrue(self.request('cancel',operation_id='a')['cancel_requested'])
            self.assertEqual(self.request('status')['active_operation'],'a')
            release.set(); self.assertTrue(self.done()['result']['cancel_seen'])

    def test_fsync_intent_exists_before_native_dispatch(self):
        def work(*args):
            rows=self.journal.rows(); self.assertEqual(rows[-1]['kind'],'intent'); return {}
        with patch.object(self.broker.runtime,'execute',side_effect=work): self.submit(); self.done()
        self.assertEqual(verify(self.root/'journal'/'manifest.json')['pending_operations'],[])

    def test_timeout_retains_uncertainty_and_never_replays(self):
        def failure(*args):
            self.broker.runtime.mutation_dispatched=True
            raise TimeoutError('lost ack')
        with patch.object(self.broker.runtime,'execute',side_effect=failure):
            self.submit(); self.assertEqual(self.done()['status'],'uncertain')
        with self.assertRaisesRegex(ValueError,'uncertain'): self.submit('b')
        self.assertEqual(self.submit()['operation']['status'],'uncertain')

    def test_restart_preserves_pending_operation(self):
        self.journal.append('intent',dict(operation_id='old',status='accepted',fingerprint='f'))
        recovered=debugger.Broker(self.receipt,Journal(self.journal.folder))
        self.assertEqual(recovered.operations['old']['outcome'],'unknown')
        self.assertTrue(recovered.uncertain)

    def test_close_allowed_with_host_uncertainty_and_certain_native_receipt(self):
        def failure(*args):
            self.broker.runtime.mutation_dispatched=True
            raise TimeoutError('lost completion observation')
        with patch.object(self.broker.runtime,'execute',side_effect=failure):
            self.submit()
            self.assertEqual(self.done()['status'],'uncertain')
        with patch.object(native,'stop',return_value=dict(ok=True,state='stopped')):
            self.request('submit',kind='close',args={},operation_id='b',wall_seconds=5,expected_generation=None)
            end=time.monotonic()+3
            while self.broker.active and time.monotonic()<end: time.sleep(.005)
        self.assertEqual(self.broker.operations['b']['status'],'completed')
        self.assertEqual(self.broker.operations['a']['status'],'uncertain')
        self.assertTrue(self.broker.uncertain)
        self.assertTrue(self.broker.closed)

    def test_close_stays_blocked_for_unresolved_native_delivery(self):
        self.broker.uncertain=True
        with patch.object(native,'state',return_value={'uncertain':{'command':'input','request':7}}), \
             patch.object(native,'stop',side_effect=AssertionError('must not exit')):
            with self.assertRaisesRegex(ValueError,'unresolved native delivery blocks close'):
                self.request('submit',kind='close',args={},operation_id='b',wall_seconds=5,expected_generation=None)
        self.assertNotIn('b',self.broker.operations)
        self.assertTrue(self.broker.uncertain)

    def test_close_stays_blocked_when_native_receipt_cannot_be_verified(self):
        self.broker.uncertain=True
        with patch.object(native,'state',side_effect=RuntimeError('bad receipt')):
            with self.assertRaisesRegex(ValueError,'native receipt unavailable'):
                self.request('submit',kind='close',args={},operation_id='b',wall_seconds=5,expected_generation=None)

    def test_close_does_not_assume_missing_native_uncertainty_field_is_certain(self):
        self.broker.uncertain=True
        with patch.object(native,'state',return_value={}):
            with self.assertRaisesRegex(ValueError,'native receipt unavailable'):
                self.request('submit',kind='close',args={},operation_id='b',wall_seconds=5,expected_generation=None)

    def test_explicit_uncertainty_does_not_depend_on_error_text_or_mutation_flag(self):
        with patch.object(self.broker.runtime,'execute',side_effect=OperationUncertain('incomplete compound evidence')):
            self.submit()
            self.assertEqual(self.done()['status'],'uncertain')
        self.assertTrue(self.broker.uncertain)

    def test_cancel_remains_a_request_not_a_native_uncertainty_bypass(self):
        self.broker.uncertain=True
        self.broker.active='active-op'
        result=self.request('cancel',operation_id='active-op')
        self.assertTrue(result['cancel_requested'])
        self.assertTrue(self.broker.cancelled.is_set())
        self.assertTrue(self.broker.uncertain)
        self.assertEqual(self.world.calls,[])
        self.broker.active=None

    def test_stale_generation_rejects_without_action(self):
        self.request('submit',kind='step',args={'ticks':3},operation_id='a',wall_seconds=5,expected_generation=2)
        self.assertEqual(self.done()['status'],'rejected')
        self.assertFalse(any(c=='advance' for c,a in self.world.calls))

    def test_token_and_legacy_ownership(self):
        with self.assertRaisesRegex(ValueError,'token'): self.broker.request({'method':'status','token':'wrong'})
        with patch.object(debugger,'descriptor',return_value={'broker':{'pid':-1},'session_id':'owned'}), patch.object(debugger,'identity',return_value={'pid':-1}):
            with self.assertRaisesRegex(RuntimeError,'session_owned'): debugger.reject_legacy()

    def test_input_is_single_dispatch_and_delivery_not_effect(self):
        result=self.runtime.execute('step',dict(ticks=3,input='0|0|0|0|AttackLight|3'),5)
        self.assertEqual(self.world.calls.count(('input','0|0|0|0|AttackLight|3')),1)
        self.assertTrue(result['input_delivered']); self.assertFalse(result['effect_verified'])
        self.assertEqual(result['ticks_observed'],3)

    def test_cancellation_before_dispatch(self):
        self.cancel.set(); r=self.runtime.execute('step',{'ticks':3},5)
        self.assertEqual(r['stop_reason'],'cancelled')
        self.assertFalse(any(c=='advance' for c,a in self.world.calls))

    def test_terminal_precedence_prevents_any_new_action(self):
        self.world.hp=0
        for kind,args in [('step',{'ticks':3}),('step',{'ticks':3,'input':'0|0|0|0|Dodge|3'}),
                          ('until',{'ticks':3,'condition':{'kind':'player_dead'}})]:
            r=self.runtime.execute(kind,args,5)
            self.assertEqual(r['stop_reason'],'player_death'); self.assertEqual(r['ticks_observed'],0)
        self.assertFalse(any(c in ('advance','input') for c,a in self.world.calls))

    def test_completed_step_retains_collection_failure_separately(self):
        calls=0; original=self.world.call
        def call(command,args='',timeout=5):
            nonlocal calls
            if command=='event_page':
                calls+=1
                if calls>1: raise TimeoutError('event page unavailable after completed step')
            return original(command,args,timeout)
        with patch.object(native,'call',side_effect=call): result=self.runtime.execute('step',{'ticks':3},5)
        self.assertEqual(result['ticks_observed'],3)
        self.assertEqual(result['stop_reason'],'observation_gap')
        self.assertFalse(result['events']['complete']); self.assertTrue(result['events']['errors'])
        self.assertTrue(result['after']['data']['world']['devkit_paused'])

    def test_actor_death_precedes_condition_and_budget(self):
        original=self.world.call
        def call(command,args='',timeout=5):
            if command=='actor': return {'ok':True,'generation':1,'data':{'id':2,'alive':False}}
            return original(command,args,timeout)
        with patch.object(native,'call',side_effect=call):
            r=self.runtime.execute('until',{'ticks':3,'condition':{'kind':'actor_dead','generation':1,'actor_id':2}},5)
        self.assertEqual(r['stop_reason'],'actor_death'); self.assertEqual(r['ticks_observed'],0)
        self.assertFalse(any(c=='advance' for c,a in self.world.calls))

    def test_until_initial_true_does_not_advance(self):
        r=self.runtime.execute('until',dict(ticks=10,condition={'kind':'player_health_below','value':101}),5)
        self.assertEqual(r['ticks_observed'],0); self.assertEqual(r['stop_reason'],'condition')

    def test_until_uses_step_bound_and_observed_state(self):
        r=self.runtime.execute('until',dict(ticks=10,step=2,condition={'kind':'player_health_below','value':98}),5)
        self.assertEqual(r['ticks_observed'],6); self.assertEqual(r['stop_reason'],'condition')
        self.assertEqual(r['pause_tick'],6)

    def test_event_baseline_excludes_historical_contact(self):
        self.world.events=[dict(sequence=1,kind='contact',time=9)]
        r=self.runtime.execute('until',dict(ticks=2,condition={'kind':'event','event_kind':'contact'}),5)
        self.assertEqual(r['stop_reason'],'tick_budget')

    def test_generation_change_stops_without_further_actions(self):
        old=self.world.call
        def changing(command,args='',timeout=5):
            r=old(command,args,timeout)
            if command=='advance': self.world.gen+=1
            return r
        with patch.object(native,'call',side_effect=changing):
            with self.assertRaisesRegex(RuntimeError,'generation_changed'): self.runtime.execute('step',{'ticks':3},5)
        self.assertEqual(self.world.calls.count(('advance','3')),1)

    def test_torn_journal_and_tamper_are_not_valid_evidence(self):
        self.journal.append('note',{}); self.journal.manifest()
        with self.journal.path.open('ab') as f: f.write(b'{')
        with self.assertRaisesRegex(ValueError,'torn_tail'): Journal(self.journal.folder)
        with self.assertRaisesRegex(ValueError,'mismatch'): verify(self.journal.folder/'manifest.json')

    def test_concurrent_journal_writers_allocate_sequences_atomically(self):
        from witcher_devkit import debug_journal
        original=debug_journal.encode
        errors=[]
        def slow_encode(value):
            time.sleep(.001)  # Release the GIL after sequence allocation.
            return original(value)
        def writer(index):
            try:
                for n in range(20):
                    self.journal.append('note',dict(writer=index,index=n))
            except Exception as error:
                errors.append(error)
        with patch.object(debug_journal,'encode',side_effect=slow_encode):
            threads=[threading.Thread(target=writer,args=(n,)) for n in range(6)]
            for thread in threads: thread.start()
            for thread in threads: thread.join(5)
        self.assertFalse(errors)
        self.assertTrue(all(not thread.is_alive() for thread in threads))
        rows=Journal(self.journal.folder).rows()
        self.assertEqual([row['sequence'] for row in rows],list(range(1,121)))
        self.journal.manifest()
        self.assertTrue(verify(self.journal.folder/'manifest.json')['ok'])

    def test_corrupt_journal_cannot_block_post_exit_settings_cleanup(self):
        self.journal.path.write_bytes(b'{"sequence":1,"kind":"note","payload":{}}\n'
                                      b'{"sequence":1,"kind":"note","payload":{}}\n')
        raw=self.journal.path.read_bytes()
        receipt=dict(self.receipt,folder=str(self.journal.folder))
        with patch('witcher_devkit.debug_recovery.closed_cleanup',return_value=dict(ok=True,process_confirmed_closed=True)) as cleanup:
            result=debugger.recover_closed(receipt)
        cleanup.assert_called_once_with(receipt)
        self.assertTrue(result['cleanup_ok'])
        self.assertFalse(result['journal_integrity'])
        self.assertIn('journal_sequence_gap',result['journal_error'])
        self.assertEqual(self.journal.path.read_bytes(),raw)
        self.assertFalse(self.world.calls)
        with patch.object(debugger,'identity',return_value=None):
            status=debugger.Client(receipt).request('status')
        self.assertEqual(status['state'],'closed')
        self.assertFalse(status['recovery']['journal_integrity'])

    def test_recovery_preserves_torn_tail_and_marks_pending_unknown(self):
        self.journal.append('intent',{'operation_id':'lost','fingerprint':'f','kind':'step','status':'accepted'})
        with self.journal.path.open('ab') as f: f.write(b'{"partial":')
        original=self.journal.path.read_bytes()
        journal=recover_torn_tail(self.journal.folder)
        backup=next(self.journal.folder.glob('journal-torn-*.jsonl'))
        self.assertEqual(backup.read_bytes(),original)
        broker=debugger.Broker(self.receipt,journal)
        self.assertTrue(broker.uncertain)
        self.assertEqual(Journal(journal.folder).operations['lost']['status'],'interrupted')
        journal.manifest(); self.assertEqual(verify(journal.folder/'manifest.json')['pending_operations'],[])

    def test_quota_refuses_intent_before_native_work(self):
        self.journal.quota=self.journal.reserve
        with self.assertRaisesRegex(OSError,'journal_quota'): self.submit()
        self.assertEqual(self.world.calls,[])

    def test_validation_rejects_unbounded_and_unknown_conditions(self):
        for args in ({'ticks':0},{'ticks':601},{'ticks':True},{'ticks':1,'input':'0|0|0|0|Guard|1'}):
            with self.assertRaises(ValueError): validate('step',args)
        with self.assertRaises(ValueError): validate('until',{'ticks':1,'condition':{'kind':'eval','code':'arbitrary'}})

    def test_load_waits_for_save_list_and_dispatches_once(self):
        import hashlib
        docs=self.root/'docs'; (docs/'gamesaves').mkdir(parents=True)
        (docs/'gamesaves'/'Fixture.SAV').write_bytes(b'save')
        calls=[]; lists=iter([[],['fixture']]); gen=iter([1,None,2,2])
        def call(command,args='',timeout=5):
            calls.append((command,args))
            if command=='saves': return {'ok':True,'generation':1,'data':next(lists)}
            if command=='snapshot':
                generation=next(gen)
                if generation is None: raise TimeoutError('read during loading')
                self.world.gen=generation; return self.world.snapshot()
            return {'ok':True,'generation':2,'data':{}}
        with patch.object(native.legacy.manage,'DOCS',docs), patch.object(native,'call',side_effect=call):
            r=self.runtime.execute('load',{'name':'fixture','sha256':hashlib.sha256(b'save').hexdigest()},3)
        self.assertEqual(calls.count(('load','fixture')),1)
        self.assertEqual(r['snapshot']['generation'],2)

    def test_read_timeout_does_not_poison_certain_session(self):
        with patch.object(self.broker.runtime,'execute',side_effect=TimeoutError('observation timed out')):
            self.request('submit',kind='inspect',args={'view':'world'},operation_id='a',wall_seconds=2,expected_generation=None)
            self.assertEqual(self.done()['status'],'rejected')
        self.assertFalse(self.broker.uncertain)

    def test_handoff_requires_pause_and_invalidates_old_token(self):
        self.request('submit',kind='handoff',args={},operation_id='a',wall_seconds=5,expected_generation=None)
        self.assertTrue(self.done()['result']['token_rotated'])
        with self.assertRaisesRegex(ValueError,'token'): self.request('status')
        result=self.broker.request(dict(method='status',token=self.broker.receipt['token']))
        self.assertEqual(result['state'],'paused')

    def test_wall_budget_stops_certain_running_work(self):
        running=False
        def call(command,args='',timeout=5):
            nonlocal running
            self.world.calls.append((command,args))
            if command=='event_page': return {'ok':True,'generation':1,'data':{'latest':0,'oldest':1,'events':[]}}
            if command=='advance': running=True
            if command=='release': running=False
            if command=='snapshot':
                s=self.world.snapshot()
                if running:
                    self.world.ticks+=1
                    s['data']['remaining_ticks']=600
                    s['data']['world']['devkit_paused']=False
                return s
            return {'ok':True,'generation':1,'data':{}}
        with patch.object(native,'call',side_effect=call): r=self.runtime.execute('run',{'ticks':600},.25)
        self.assertEqual(r['stop_reason'],'wall_budget')
        self.assertTrue(r['input_release_verified'])
        self.assertEqual(self.world.calls.count(('advance','600')),1)
        self.assertIn(('release',''),self.world.calls)

    def test_record_window_survives_owner_restart(self):
        self.request('submit',kind='record-start',args={},operation_id='a',wall_seconds=5,expected_generation=None)
        window=self.done()['result']['window_id']
        recovered=debugger.Broker(self.receipt,Journal(self.journal.folder))
        self.assertEqual(recovered.runtime.window_id,window)
        result=recovered.runtime.execute('record-stop',{},5)
        self.assertEqual(result['window_id'],window)
        self.assertFalse(result['recording'])

    def test_dead_owner_still_blocks_legacy_game_access(self):
        receipt={'broker':{'pid':999,'start':'x'},'session_id':'old','process':{'pid':123}}
        with patch.object(debugger,'descriptor',return_value=receipt), patch.object(debugger,'identity',return_value=None), patch.object(native.legacy,'process_identity',return_value={'pid':123}):
            with self.assertRaisesRegex(RuntimeError,'session_owned'): debugger.reject_legacy()

    def test_protocol_and_request_fields_fail_before_native_work(self):
        with self.assertRaisesRegex(ValueError,'unsupported_protocol_version'):
            self.broker.request({'protocol_version':2,'method':'status','token':'secret'})
        with self.assertRaisesRegex(ValueError,'unknown operation request fields'):
            self.request('submit',kind='step',args={'ticks':1},operation_id='a',wall_seconds=1,extra=True)
        self.assertEqual(self.world.calls,[])
        with self.assertRaises(ValueError): validate('inspect',{'view':'profile'})
        with self.assertRaises(ValueError): validate('step',{'ticks':2,'input':''})

    def test_condition_precedes_same_boundary_cancellation(self):
        original=self.world.call
        def call(command,args='',timeout=5):
            r=original(command,args,timeout)
            if command=='advance': self.cancel.set()
            return r
        with patch.object(native,'call',side_effect=call):
            r=self.runtime.execute('until',{'ticks':10,'condition':{'kind':'player_health_below','value':100}},5)
        self.assertEqual(r['stop_reason'],'condition'); self.assertEqual(r['ticks_observed'],1)

    def test_real_socket_client_result_lookup(self):
        path=str(self.root/'debug.sock')
        with debugger.Server(path,debugger.Handler) as server:
            server.broker=self.broker
            thread=threading.Thread(target=server.serve_forever,daemon=True); thread.start()
            try:
                c=debugger.Client(dict(socket=path,token='secret',broker=debugger.identity()))
                result=c.submit('step',{'ticks':3},'a',5)
                self.assertEqual(result['operation']['operation_id'],'a')
                self.assertEqual(c.wait('a',3)['operation']['status'],'completed')
            finally: server.shutdown(); thread.join()

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