"""Run INSIDE a disposable Computer as root; every job executes as uid 1000."""
import base64
import json
import os
from pathlib import Path
import subprocess
import sqlite3
import signal
import time
import unittest
import uuid

RUNNER = '/var/lib/lazyboy-runner/runner_jobs.py'


class RunnerJobs(unittest.TestCase):
    def setUp(self):
        if os.environ.get('LAZYBOY_RUNNER_TEST_CONTAINER') != '1' or not Path('/.dockerenv').exists():
            self.fail('run only inside the disposable Runner test container')
        self.jobs = []
        self.identity = {'computer_id': 'fixture', 'bot_id': 'bot-a', 'generation': 1}

    def call(self, request, check=True, user=None):
        prefix = ['runuser', '-u', user, '--'] if user else []
        result = subprocess.run(prefix + ['python3', '-I', RUNNER], input=json.dumps(request),
                                capture_output=True, text=True, timeout=10)
        if check:
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        return json.loads(result.stdout)

    def start(self, argv, **kwargs):
        job = 'test-' + uuid.uuid4().hex
        self.jobs.append(job)
        request = dict(action='start', job_id=job, identity=self.identity, argv=argv,
                       runtime_timeout_ms=5000, **kwargs)
        return request, self.call(request)

    def follow(self, job, action='status', **kwargs):
        return self.call(dict(action=action, job_id=job, identity=self.identity, **kwargs))

    def wait(self, job):
        deadline = time.monotonic() + 8
        while time.monotonic() < deadline:
            result = self.follow(job)
            if result['status'] not in ('accepted', 'running', 'paused', 'cancelling'):
                return result
            time.sleep(.03)
        self.fail('job did not finish')

    def tearDown(self):
        for job in self.jobs:
            self.follow(job, 'cancel')
            self.wait(job)

    def test_durable_identity_replay_output_cursor_and_binary(self):
        request, first = self.start(['python3', '-c', "import sys,time;sys.stdout.buffer.write(b'\\xffA');sys.stdout.flush();time.sleep(.2);sys.stdout.write('B');sys.stderr.write('error');sys.exit(7)"])
        self.assertEqual(self.call(request)['job_id'], first['job_id'])
        changed = dict(request, argv=['true'])
        self.assertEqual(self.call(changed, check=False)['error'], 'PAYLOAD_MISMATCH')
        for identity in [dict(self.identity, bot_id='bot-b'), dict(self.identity, generation=2)]:
            denied = self.call(dict(action='status', job_id=first['job_id'], identity=identity), check=False)
            self.assertEqual(denied['error'], 'UNKNOWN_JOB')
        done = self.wait(first['job_id'])
        self.assertEqual(done['code'], 7)
        self.assertEqual(base64.b64decode(done['stdout_base64']), b'\xffAB')
        self.assertEqual(done['stderr'], 'error')
        # Every call starts a new CLI process, proving output survives client reconnects.
        rest = self.follow(first['job_id'], 'output', stdout_cursor=2, stderr_cursor=5)
        self.assertEqual(rest['stdout'], 'B')
        self.assertEqual(rest['stderr'], '')
        self.assertEqual(rest['next_stdout_cursor'], 3)
        self.assertEqual(self.follow(first['job_id'], 'cancel')['status'], 'failed')

    def test_real_pty_input_and_resize(self):
        _, started = self.start(['python3', '-u', '-c', 'import os;print(os.isatty(0));print(input().upper())'], pty=True)
        job = started['job_id']
        reply = self.follow(job, 'interact', input='hello 中文\n', rows=30, cols=100)
        self.assertTrue(reply['pty'])
        done = self.wait(job)
        self.assertEqual(done['status'], 'succeeded')
        self.assertIn('True', done['stdout'])
        self.assertIn('HELLO 中文', done['stdout'])

    def test_task_cannot_read_control_state_or_inherit_secrets(self):
        os.environ['LAZYBOY_CONTROL_TOKEN'] = 'runner-secret-canary'
        _, started = self.start(['python3', '-c', "import os; print(os.getuid()); print(os.getenv('LAZYBOY_CONTROL_TOKEN','unset')); open('/var/lib/lazyboy-runner/jobs.sqlite').read()"])
        done = self.wait(started['job_id'])
        self.assertEqual(done['status'], 'failed')
        self.assertIn('1000\nunset', done['stdout'])
        self.assertIn('PermissionError', done['stderr'])
        self.assertNotIn('runner-secret-canary', json.dumps(done))
        denied = subprocess.run(['runuser', '-u', 'lazyboy', '--', 'cat', RUNNER], capture_output=True)
        self.assertNotEqual(denied.returncode, 0)

    def test_pause_resume_cancel_are_scoped_to_one_job(self):
        _, first = self.start(['sleep', '30'])
        _, peer = self.start(['sh', '-c', 'sleep .3; printf peer'])
        job = first['job_id']
        self.assertEqual(self.follow(job, 'pause')['status'], 'paused')
        self.assertEqual(self.wait(peer['job_id'])['stdout'], 'peer')
        self.assertEqual(self.follow(job, 'resume')['status'], 'running')
        self.follow(job, 'cancel')
        self.assertEqual(self.wait(job)['status'], 'cancelled')

    def test_deadline_and_output_quota_stop_process_tree(self):
        job = 'test-' + uuid.uuid4().hex
        self.jobs.append(job)
        self.call(dict(action='start', job_id=job, identity=self.identity,
                       argv=['sh', '-c', 'sleep 30 & wait'], runtime_timeout_ms=200))
        self.assertEqual(self.wait(job)['status'], 'timed_out')
        _, loud = self.start(['python3', '-c', "import sys; sys.stdout.buffer.write(b'x'*6000000)"])
        done = self.wait(loud['job_id'])
        self.assertEqual(done['status'], 'output_limit')
        self.assertTrue(done['truncated'])
        self.assertEqual(done['next_stdout_cursor'], 65536)

    def test_local_outbox_requires_explicit_ack(self):
        _, job = self.start(['true'])
        self.wait(job['job_id'])
        before = self.call({'action': 'events'})['events']
        own = [event for event in before if event['job_id'] == job['job_id']]
        self.assertTrue(any(event['event'] == 'job.accepted' for event in own))
        self.assertTrue(any(event['event'] == 'job.succeeded' for event in own))
        self.call({'action': 'ack_events', 'ids': [event['id'] for event in own]})
        after = self.call({'action': 'events'})['events']
        self.assertFalse(any(event['job_id'] == job['job_id'] for event in after))


    def test_expired_output_keeps_replay_tombstone(self):
        request, job = self.start(['printf', 'retained'])
        self.wait(job['job_id'])
        with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
            state = json.loads(db.execute('SELECT state FROM jobs WHERE id=?', (job['job_id'],)).fetchone()[0])
            state['completed_at'] = time.time() - 90000
            db.execute('UPDATE jobs SET state=? WHERE id=?', (json.dumps(state), job['job_id']))
        # Undelivered terminal evidence prevents expiration, even after the TTL.
        events = self.call({'action': 'events'})['events']
        self.assertEqual(self.follow(job['job_id'])['stdout'], 'retained')
        own = [event['id'] for event in events if event['job_id'] == job['job_id']]
        self.call({'action': 'ack_events', 'ids': own})
        self.call({'action': 'events'})
        replay = self.call(request)
        self.assertTrue(replay['output_expired'])
        self.assertEqual(replay['status'], 'succeeded')
        self.assertFalse(Path('/var/lib/lazyboy-runner', job['job_id']).exists())
        denied = self.call(dict(action='output', job_id=job['job_id'], identity=self.identity), check=False)
        self.assertEqual(denied['error'], 'OUTPUT_EXPIRED')
        changed = self.call(dict(request, argv=['false']), check=False)
        self.assertEqual(changed['error'], 'PAYLOAD_MISMATCH')

    def test_worker_crash_recovery_checks_process_identity(self):
        _, job = self.start(['sh', '-c', 'sleep 30 & wait'])
        with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
            state = json.loads(db.execute('SELECT state FROM jobs WHERE id=?', (job['job_id'],)).fetchone()[0])
        os.kill(state['worker_pid'], signal.SIGKILL)
        time.sleep(.05)
        self.assertEqual(self.follow(job['job_id'])['status'], 'unknown')
        # A forged/reused starttime must never authorize killing the PID.
        with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
            stale = json.loads(db.execute('SELECT state FROM jobs WHERE id=?', (job['job_id'],)).fetchone()[0])
            stale['pid_start'] = 'impossible'
            db.execute('UPDATE jobs SET state=? WHERE id=?', (json.dumps(stale), job['job_id']))
        denied = self.call(dict(action='cancel', job_id=job['job_id'], identity=self.identity), check=False)
        self.assertEqual(denied['error'], 'ORPHAN_IDENTITY_UNPROVEN')
        os.kill(state['pid'], 0)
        with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
            stale['pid_start'] = state['pid_start']
            db.execute('UPDATE jobs SET state=? WHERE id=?', (json.dumps(stale), job['job_id']))
        self.assertEqual(self.follow(job['job_id'], 'cancel')['status'], 'interrupted')
        _, unattended = self.start(['sleep', '30'])
        with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
            state = json.loads(db.execute('SELECT state FROM jobs WHERE id=?', (unattended['job_id'],)).fetchone()[0])
        os.kill(state['worker_pid'], signal.SIGKILL)
        time.sleep(.05)
        self.call({'action': 'events'})
        self.assertEqual(self.follow(unattended['job_id'])['status'], 'interrupted')


    def test_retention_pressure_releases_slots_without_replaying(self):
        prefix = 'retention-' + uuid.uuid4().hex
        with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
            states = [json.loads(row[0]) for row in db.execute('SELECT state FROM jobs')]
            retained = sum(not state.get('output_expired') for state in states)
            for index in range(64 - retained):
                job = prefix + str(index)
                state = dict(job_id=job, status='succeeded', code=0, identity=self.identity,
                             completed_at=time.time() - 60, created_at=time.time() - 61,
                             stdout_size=0, stderr_size=0, pty=False)
                db.execute('INSERT INTO jobs VALUES (?,?,?,?)', (job, 'digest', '{}', json.dumps(state)))
        try:
            _, started = self.start(['true'])
            self.assertEqual(self.wait(started['job_id'])['status'], 'succeeded')
            with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
                states = [json.loads(row[0]) for row in db.execute('SELECT state FROM jobs')]
                self.assertLessEqual(sum(not state.get('output_expired') for state in states), 64)
                self.assertTrue(any(state.get('output_expired') and state['job_id'].startswith(prefix) for state in states))
        finally:
            with sqlite3.connect('/var/lib/lazyboy-runner/jobs.sqlite') as db:
                db.execute('DELETE FROM jobs WHERE id LIKE ?', (prefix + '%',))


    def test_agent_barrier_blocks_admission_and_preserves_peer(self):
        _, own = self.start(['sleep', '30'])
        _, manual = self.start(['sleep', '30'])
        self.follow(manual['job_id'], 'pause')
        peer_identity = dict(self.identity, bot_id='bot-b')
        peer = 'peer-' + uuid.uuid4().hex
        self.call(dict(action='start', job_id=peer, identity=peer_identity,
                       argv=['sh', '-c', 'sleep .2; printf peer-finished'], runtime_timeout_ms=5000))
        try:
            paused = self.call(dict(action='pause_agent', identity=self.identity))
            self.assertEqual(paused['status'], 'paused')
            self.assertEqual(self.follow(own['job_id'])['status'], 'paused')
            rejected = self.call(dict(action='start', job_id='blocked-' + uuid.uuid4().hex,
                                     identity=self.identity, argv=['true']), check=False)
            self.assertEqual(rejected['error'], 'SCOPE_PAUSED')
            bypass = self.call(dict(action='resume', job_id=own['job_id'], identity=self.identity), check=False)
            self.assertEqual(bypass['error'], 'SCOPE_PAUSED')
            time.sleep(.25)
            other = self.call(dict(action='status', job_id=peer, identity=peer_identity))
            self.assertEqual(other['status'], 'succeeded')
            self.assertEqual(other['stdout'], 'peer-finished')
        finally:
            self.call(dict(action='resume_agent', identity=self.identity))
            self.call(dict(action='cancel', job_id=peer, identity=peer_identity))
        self.assertEqual(self.follow(own['job_id'])['status'], 'running')
        self.assertEqual(self.follow(manual['job_id'])['status'], 'paused')


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