"""Bounded research workflow: collect -> draft -> verify -> local publish.
Python 3.10+, stdlib only. This is a reliability lab, not a production service.
"""
import hashlib
import json
import sqlite3
import time
import uuid
from pathlib import Path
from memory import connect, transaction, current, revision
from provider import draft, TransientError, PermanentError

STEPS = ('collect', 'draft', 'verify', 'publish')
SCHEMA = '''
CREATE TABLE IF NOT EXISTS jobs(
 id TEXT PRIMARY KEY, spec TEXT NOT NULL, input_hash TEXT NOT NULL,
 status TEXT NOT NULL, generation INTEGER NOT NULL DEFAULT 0, owner TEXT,
 lease_until REAL NOT NULL DEFAULT 0, attempts INTEGER NOT NULL DEFAULT 0,
 available_at REAL NOT NULL DEFAULT 0, scope TEXT NOT NULL,
 memory_rev INTEGER NOT NULL, memory_expires REAL NOT NULL,
 deadline REAL NOT NULL, error TEXT);
CREATE TABLE IF NOT EXISTS checkpoints(
 job_id TEXT NOT NULL REFERENCES jobs(id), step TEXT NOT NULL, payload TEXT NOT NULL,
 PRIMARY KEY(job_id,step));
CREATE TABLE IF NOT EXISTS events(
 seq INTEGER PRIMARY KEY AUTOINCREMENT, job_id TEXT NOT NULL REFERENCES jobs(id),
 at REAL NOT NULL, generation INTEGER NOT NULL, kind TEXT NOT NULL, detail TEXT NOT NULL);
'''


def encode(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'))


def digest(value):
    return hashlib.sha256(encode(value).encode()).hexdigest()


class LostLease(Exception):
    pass


class InjectedCrash(BaseException):
    pass


class Runtime:
    def __init__(self, path, clock=time.time, lease_seconds=30):
        if lease_seconds <= 0:
            raise ValueError('lease_seconds must be positive')
        self.path = str(path)
        self.db = connect(self.path)
        self.db.executescript(SCHEMA)
        self.clock = clock
        self.lease_seconds = lease_seconds

    def event(self, job, generation, kind, detail=''):
        self.db.execute('INSERT INTO events(job_id,at,generation,kind,detail) VALUES (?,?,?,?,?)',
                        (job, self.clock(), generation, kind, detail))

    def submit(self, job, spec, scope, ttl=600):
        with transaction(self.db):
            old = self.db.execute('SELECT input_hash FROM jobs WHERE id=?', (job,)).fetchone()
            input_hash = digest({'spec': spec, 'scope': scope})
            if old:
                if old['input_hash'] != input_hash:
                    raise ValueError('same job id with different input or scope')
                return job
            now = self.clock()
            memories = current(self.db, scope, now)
            expires = min([m['expires'] for m in memories] or [now + ttl])
            self.db.execute('INSERT INTO jobs(id,spec,input_hash,status,scope,memory_rev,'
                'memory_expires,deadline) VALUES (?,?,?,\'queued\',?,?,?,?)',
                (job, encode(spec), input_hash, scope, revision(self.db, scope), expires, now + ttl))
            self.event(job, 0, 'submitted')
        return job

    def job(self, job):
        row = self.db.execute('SELECT * FROM jobs WHERE id=?', (job,)).fetchone()
        if not row:
            raise ValueError('unknown job')
        return dict(row)

    def claim(self, job, owner):
        with transaction(self.db):
            row, now = self.job(job), self.clock()
            ready = (row['status'] in ('queued', 'retry_wait') and row['available_at'] <= now
                     or row['status'] == 'running' and row['lease_until'] <= now)
            if not ready:
                return None
            if row['deadline'] <= now or row['attempts'] >= 3:
                self.db.execute("UPDATE jobs SET status='failed',error='deadline_or_attempt_budget' WHERE id=?", (job,))
                self.event(job, row['generation'], 'failed', 'deadline_or_attempt_budget')
                return None
            generation = row['generation'] + 1
            self.db.execute("UPDATE jobs SET status='running',owner=?,generation=?,lease_until=?,"
                'attempts=attempts+1 WHERE id=?', (owner, generation, now + self.lease_seconds, job))
            self.event(job, generation, 'claimed', owner)
            return generation

    def guard(self, job, owner, generation):
        row = self.job(job)
        if row['status'] != 'running' or row['owner'] != owner or row['generation'] != generation or row['lease_until'] <= self.clock():
            raise LostLease('stale worker or cancelled task')
        return row

    def validate_memory(self, row):
        if revision(self.db, row['scope']) != row['memory_rev'] or row['memory_expires'] <= self.clock():
            raise PermanentError('memory_changed_or_expired: submit a new job id')

    def before_step(self, job, owner, generation, step):
        with transaction(self.db):
            row = self.guard(job, owner, generation)
            self.validate_memory(row)
            if row['deadline'] <= self.clock():
                raise PermanentError('deadline exceeded')
            self.db.execute('UPDATE jobs SET lease_until=? WHERE id=?',
                            (self.clock() + self.lease_seconds, job))
            self.event(job, generation, 'step_started', step)

    def saved(self, job, step):
        row = self.db.execute('SELECT payload FROM checkpoints WHERE job_id=? AND step=?', (job, step)).fetchone()
        return json.loads(row['payload']) if row else None

    def commit_step(self, job, owner, generation, step, payload):
        with transaction(self.db):
            row = self.guard(job, owner, generation)
            self.validate_memory(row)
            if row['deadline'] <= self.clock():
                raise PermanentError('deadline exceeded')
            self.db.execute('INSERT INTO checkpoints VALUES (?,?,?)', (job, step, encode(payload)))
            self.event(job, generation, 'step_committed', step)
            if step == 'publish':
                self.db.execute("UPDATE jobs SET status='succeeded',owner=NULL,error=NULL WHERE id=?", (job,))

    def fail(self, job, owner, generation, exc):
        with transaction(self.db):
            row = self.guard(job, owner, generation)
            retry = isinstance(exc, TransientError) and row['attempts'] < 3 and row['deadline'] > self.clock()
            state = 'retry_wait' if retry else 'failed'
            # Deterministic backoff for repeatable tests; add jitter in a real queue.
            self.db.execute('UPDATE jobs SET status=?,owner=NULL,available_at=?,error=? WHERE id=?',
                (state, self.clock() + 2 ** row['attempts'], str(exc), job))
            self.event(job, generation, state, str(exc))

    def cancel(self, job):
        with transaction(self.db):
            row = self.job(job)
            if row['status'] in ('succeeded', 'failed', 'cancelled'):
                return False
            self.db.execute("UPDATE jobs SET status='cancelled',generation=generation+1,owner=NULL WHERE id=?", (job,))
            self.event(job, row['generation'] + 1, 'cancelled')
            return True

    def publish(self, job, payload):
        # Independent database simulates an external service's durable idempotency contract.
        db = sqlite3.connect(self.path + '.publisher.sqlite', isolation_level=None, timeout=5)
        try:
            db.execute('CREATE TABLE IF NOT EXISTS receipts(op TEXT PRIMARY KEY, payload_hash TEXT NOT NULL, receipt TEXT NOT NULL)')
            op, payload_hash = digest([job, 'publish/v1']), digest(payload)
            with transaction(db):
                old = db.execute('SELECT payload_hash,receipt FROM receipts WHERE op=?', (op,)).fetchone()
                if old:
                    if old[0] != payload_hash:
                        raise PermanentError('idempotency key reused with different payload')
                    return {'receipt': old[1], 'deduplicated': True}
                receipt = 'report-' + op[:12]
                db.execute('INSERT INTO receipts VALUES (?,?,?)', (op, payload_hash, receipt))
                return {'receipt': receipt, 'deduplicated': False}
        finally:
            db.close()

    def run(self, job, fault=None, model_fn=draft):
        owner = uuid.uuid4().hex
        generation = self.claim(job, owner)
        if generation is None:
            return self.job(job)['status']
        try:
            spec = json.loads(self.job(job)['spec'])
            for step in STEPS:
                if self.saved(job, step) is not None:
                    continue
                self.before_step(job, owner, generation, step)
                if step == 'collect':
                    row = self.job(job)
                    payload = {'docs': spec['docs'], 'memory': current(self.db, row['scope'], self.clock())}
                elif step == 'draft':
                    payload = model_fn(spec, self.saved(job, 'collect'))
                elif step == 'verify':
                    report = self.saved(job, 'draft')
                    ids = {d['id'] for d in spec['docs']}
                    citations = report.get('citations') if isinstance(report, dict) else None
                    valid = (isinstance(report, dict) and isinstance(report.get('summary'), str)
                             and bool(report['summary'].strip()) and isinstance(citations, list)
                             and bool(citations) and all(isinstance(c, str) and c in ids for c in citations))
                    if not valid:
                        raise PermanentError('invalid report or unknown source id')
                    payload = {'schema_and_source_ids': True, 'semantic_support': 'not_scored'}
                else:
                    payload = self.publish(job, self.saved(job, 'draft'))
                    if fault == 'after_effect':
                        raise InjectedCrash('publisher committed; local checkpoint absent')
                self.commit_step(job, owner, generation, step, payload)
                if fault == 'after_' + step:
                    raise InjectedCrash('checkpoint committed: ' + step)
        except (TransientError, PermanentError) as exc:
            self.fail(job, owner, generation, exc)
        return self.job(job)['status']

    def snapshot(self, job):
        row = self.job(job)
        return {'job_id': job, 'status': row['status'], 'attempts': row['attempts'],
                'generation': row['generation'], 'error': row['error'],
                'checkpoints': {s: self.saved(job, s) for s in STEPS if self.saved(job, s) is not None},
                'events': [dict(e) for e in self.db.execute(
                    'SELECT seq,at,generation,kind,detail FROM events WHERE job_id=? ORDER BY seq', (job,))]}