"""Terminal owner receipts recover delivery without replay or double bookkeeping."""
import asyncio
import json
from types import SimpleNamespace

from cron import jobs, scheduler
from gateway import run, session_cron
from gateway.session import SessionStore
from gateway.session_authority import SessionAuthority
from gateway.config import GatewayConfig
from hermes_state import SessionDB
from hermes_state_runtime import begin_runtime_epoch, claim_session_input, settle_session_input


def test_lost_ack_reconciles_frozen_delivery_once(tmp_path, monkeypatch):
    from cron import scheduler_authority, delivery_queue
    monkeypatch.setenv('HERMES_HOME', str(tmp_path))
    monkeypatch.setattr(run, '_load_gateway_config', lambda: {'model': {}, 'platform_toolsets': {'cli': []}})
    with jobs.use_cron_store(tmp_path / 'cron'):
        job = jobs.create_job(prompt='original', schedule='every 1h', deliver='telegram:123')
        db = SessionDB(tmp_path / 'state.db')
        runner = SimpleNamespace(_draining=False, session_store=SessionStore(tmp_path / 'sessions', GatewayConfig()), adapters={})
        authority = SessionAuthority(runner, profile_id=str(tmp_path), instance_id='test', db=db,
                                     epoch=begin_runtime_epoch(db, instance_id='test'))
        monkeypatch.setattr(authority, '_schedule', lambda ref: None)
        runner.session_authority = authority
        async def probe():
            params = {'job_id': job['id'], 'request_id': 'fire', 'extra_prompt': None}
            receipt = await session_cron.operation(authority, 'submit', params)
            row = claim_session_input(db, epoch=authority.epoch, session_id=receipt['session_id'])
            settle_session_input(db, epoch=authority.epoch, admission_id=row['admission_id'], generation=row['generation'], outcome='completed',
                                 result={'result': {'cron_result': [True, 'document', 'answer', None]}, 'usage': {}})
            root = tmp_path / 'cron' / 'admissions'; root.mkdir()
            journal = scheduler_authority.journal_path(job['id'], 'fire')
            journal.write_text(json.dumps({'params': params, 'receipt': None}))
            session_cron.bind_owner(authority)
            try:
                await asyncio.to_thread(scheduler_authority.reconcile_pending)
                assert delivery_queue.get_status('fire')['status'] == 'pending'
                assert jobs.get_job(job['id'])['last_status'] == 'delivery_queued'
                completed = jobs.get_job(job['id'])['repeat']['completed']
                journal.write_text(json.dumps({'params': params, 'receipt': None}))
                await asyncio.to_thread(scheduler_authority.reconcile_pending)
                assert jobs.get_job(job['id'])['repeat']['completed'] == completed
                assert delivery_queue.claim_next()['content'] == 'answer'
                assert delivery_queue.claim_next() is None
                unknown = dict(params, request_id='unknown')
                accepted = await session_cron.operation(authority, 'submit', unknown)
                claim_session_input(db, epoch=authority.epoch, session_id=accepted['session_id'])
                from hermes_state_runtime import recover_session_inputs
                authority.epoch = begin_runtime_epoch(db, instance_id='restart')
                recover_session_inputs(db, epoch=authority.epoch)
                unknown_journal = scheduler_authority.journal_path(job['id'], 'unknown')
                unknown_journal.write_text(json.dumps({'params': unknown, 'receipt': None}))
                await asyncio.to_thread(scheduler_authority.reconcile_pending)
                assert unknown_journal.exists()
                assert jobs.get_job(job['id'])['state'] == 'paused'
                assert delivery_queue.claim_next() is None
            finally:
                session_cron.unbind_owner(authority)
        try:
            asyncio.run(probe())
        finally:
            db.close()


def test_failed_manual_run_exits_nonzero(monkeypatch):
    from hermes_cli import cron
    monkeypatch.setattr(cron, '_cron_api', lambda **kw: {'success': True, 'job': {'executed': True, 'execution_success': False}})
    assert cron._job_action('run', 'job', 'Triggered') == 1


def _owner(tmp_path, monkeypatch):
    monkeypatch.setenv('HERMES_HOME', str(tmp_path))
    monkeypatch.setattr(run, '_load_gateway_config', lambda: {'model': {}, 'platform_toolsets': {'cli': []}})
    db = SessionDB(tmp_path / 'state.db')
    runner = SimpleNamespace(_draining=False, session_store=SessionStore(tmp_path / 'sessions', GatewayConfig()), adapters={})
    authority = SessionAuthority(runner, profile_id=str(tmp_path), instance_id='test', db=db,
                                 epoch=begin_runtime_epoch(db, instance_id='test'))
    monkeypatch.setattr(authority, '_schedule', lambda ref: None)
    runner.session_authority = authority
    return authority, db


def test_recovered_receipt_settles_the_departed_firers_execution_row(tmp_path, monkeypatch):
    """The firing process created and started the execution row, then exited after the owner
    finished. Recovery books the job from the receipt and must settle that SAME ledger row
    before the journal (its only link to the receipt) is deleted — not leave it running for
    the dead-owner sweep to rewrite as unknown."""
    import os
    import subprocess
    import sys
    from cron import scheduler_authority, executions
    with jobs.use_cron_store(tmp_path / 'cron'):
        job = jobs.create_job(prompt='original', schedule='every 1h', deliver='local')
        authority, db = _owner(tmp_path, monkeypatch)
        firer = subprocess.run([sys.executable, '-c', (
            'import sys; from cron.executions import create_execution, mark_execution_running;'
            f'r = create_execution({job["id"]!r}, source="direct"); mark_execution_running(r["id"]);'
            'print(r["id"])')], env={**os.environ, 'HERMES_HOME': str(tmp_path)},
            capture_output=True, text=True, check=True, timeout=60)
        fire = firer.stdout.strip().splitlines()[-1]
        assert executions.get_execution(fire)['status'] == 'running'

        async def probe():
            params = {'job_id': job['id'], 'request_id': fire, 'extra_prompt': None}
            receipt = await session_cron.operation(authority, 'submit', params)
            row = claim_session_input(db, epoch=authority.epoch, session_id=receipt['session_id'])
            settle_session_input(db, epoch=authority.epoch, admission_id=row['admission_id'], generation=row['generation'], outcome='completed',
                                 result={'result': {'cron_result': [True, 'document', 'answer', None]}, 'usage': {}})
            journal = scheduler_authority.journal_path(job['id'], fire)
            journal.parent.mkdir(parents=True)
            journal.write_text(json.dumps({'params': params, 'receipt': None}))
            session_cron.bind_owner(authority)
            try:
                await asyncio.to_thread(scheduler_authority.reconcile_pending)
            finally:
                session_cron.unbind_owner(authority)
            assert not journal.exists()
        try:
            asyncio.run(probe())
        finally:
            db.close()
        settled = executions.get_execution(fire)
        assert (settled['status'], settled['delivery_outcome']) == ('completed', 'suppressed')
        assert executions.recover_interrupted_executions() == 0
        assert executions.get_execution(fire)['status'] == 'completed'



def test_owner_refused_fire_is_booked_failed_and_retires_its_journal(tmp_path, monkeypatch):
    """A definite pre-admission refusal (the job's workdir was removed) is an ordinary failed
    run: it must not raise CronExecutionUnknown, keep a journal, or pause the job on every
    later tick after the operator fixes the configuration."""
    from cron import scheduler_authority
    workdir = tmp_path / 'work'
    workdir.mkdir()
    with jobs.use_cron_store(tmp_path / 'cron'):
        job = jobs.create_job(prompt='original', schedule='every 1h', deliver='local', workdir=str(workdir))
        workdir.rmdir()
        authority, db = _owner(tmp_path, monkeypatch)

        async def probe():
            session_cron.bind_owner(authority)
            try:
                result = await asyncio.to_thread(
                    scheduler_authority.run_canonical_job, jobs.get_job(job['id']), execution_id='fire')
                assert result[0] is False and 'invalid_params' in result[3]
                # Keep the definitive refusal until terminal bookkeeping, including a firing-process crash.
                assert scheduler_authority.journal_path(job['id'], 'fire').exists()
                await asyncio.to_thread(scheduler_authority.reconcile_pending)
                assert not scheduler_authority.journal_path(job['id'], 'fire').exists()
            finally:
                session_cron.unbind_owner(authority)
        try:
            asyncio.run(probe())
        finally:
            db.close()
        assert jobs.get_job(job['id'])['state'] != 'paused'


def test_admission_journal_survives_power_loss(tmp_path, monkeypatch):
    """The journal is the only evidence that a lost reply may have been admitted, so its rename
    is made durable (directory fsync), not left in the page cache until the next flush."""
    import utils
    from cron import scheduler_authority
    synced = []
    real = utils.fsync_directory
    monkeypatch.setattr(utils, 'fsync_directory', lambda path: (synced.append(str(path)), real(path)))
    workdir = tmp_path / 'work'
    workdir.mkdir()
    with jobs.use_cron_store(tmp_path / 'cron'):
        job = jobs.create_job(prompt='original', schedule='every 1h', deliver='local', workdir=str(workdir))
        workdir.rmdir()
        authority, db = _owner(tmp_path, monkeypatch)

        async def probe():
            session_cron.bind_owner(authority)
            try:
                await asyncio.to_thread(
                    scheduler_authority.run_canonical_job, jobs.get_job(job['id']), execution_id='fire')
            finally:
                session_cron.unbind_owner(authority)
        try:
            asyncio.run(probe())
        finally:
            db.close()
    journal_dir = str(scheduler_authority.journal_path(job['id'], 'fire').parent)
    assert journal_dir in synced