"""Durable coordination data; no model credentials or process execution."""
import hashlib
import json
import math
import re
import secrets
import sqlite3
import time
from contextlib import contextmanager
from pathlib import Path


class Invalid(ValueError):
    pass


class Conflict(Invalid):
    pass


def string(value, name, limit=2000, optional=False):
    if optional and value is None:
        return ''
    if not isinstance(value, str) or (not optional and not value.strip()) or len(value) > limit:
        raise Invalid('invalid ' + name)
    return value


def identifier(value, name='id'):
    value = string(value, name, 128)
    if not re.fullmatch(r'[A-Za-z0-9_.:-]+', value):
        raise Invalid('invalid ' + name)
    return value


class Store:
    def __init__(self, path):
        self.path = str(path)
        with self.db() as db:
            db.executescript('''
                PRAGMA journal_mode=WAL;
                CREATE TABLE IF NOT EXISTS sessions(id TEXT PRIMARY KEY, data TEXT NOT NULL);
                CREATE TABLE IF NOT EXISTS tasks(id TEXT PRIMARY KEY, data TEXT NOT NULL);
                CREATE TABLE IF NOT EXISTS requests(id TEXT PRIMARY KEY, data TEXT NOT NULL);
                CREATE TABLE IF NOT EXISTS events(id INTEGER PRIMARY KEY AUTOINCREMENT, data TEXT NOT NULL);
                CREATE TABLE IF NOT EXISTS metadata(key TEXT PRIMARY KEY, value TEXT NOT NULL);
                CREATE TABLE IF NOT EXISTS messages(
                    id TEXT PRIMARY KEY, session_id TEXT NOT NULL, text TEXT NOT NULL,
                    source TEXT NOT NULL, reply_to TEXT, idempotency_key TEXT NOT NULL,
                    status TEXT NOT NULL, created REAL NOT NULL, worker_id TEXT,
                    lease_id TEXT, lease_until REAL, response_id TEXT,
                    UNIQUE(session_id,idempotency_key));
                CREATE INDEX IF NOT EXISTS messages_session ON messages(session_id,created);
                CREATE TABLE IF NOT EXISTS hooks(
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    source TEXT NOT NULL, event_id TEXT NOT NULL, status TEXT NOT NULL,
                    data TEXT NOT NULL, UNIQUE(source,event_id,status));
                CREATE TABLE IF NOT EXISTS hook_notifications(
                    seq INTEGER PRIMARY KEY AUTOINCREMENT,
                    id TEXT NOT NULL UNIQUE, hook_id INTEGER NOT NULL,
                    target TEXT NOT NULL, data TEXT NOT NULL,
                    delivered_at REAL, UNIQUE(hook_id,target));
                CREATE INDEX IF NOT EXISTS hook_notifications_pending
                    ON hook_notifications(delivered_at,seq);
                CREATE TABLE IF NOT EXISTS history_links(
                    session_id TEXT NOT NULL, external_id TEXT NOT NULL,
                    source TEXT NOT NULL, data TEXT NOT NULL,
                    PRIMARY KEY(session_id,external_id,source));
            ''')
            db.execute('INSERT OR IGNORE INTO metadata(key,value) VALUES(?,?)',('instance_id',secrets.token_hex(16)))
            self.instance_id=db.execute('SELECT value FROM metadata WHERE key=?',('instance_id',)).fetchone()['value']

            columns = {row['name'] for row in db.execute('PRAGMA table_info(messages)')}
            if 'started' not in columns:
                db.execute('ALTER TABLE messages ADD COLUMN started INTEGER NOT NULL DEFAULT 0')
            if 'submission' not in columns:
                db.execute('ALTER TABLE messages ADD COLUMN submission TEXT')
            if 'origin' not in columns:
                db.execute("ALTER TABLE messages ADD COLUMN origin TEXT NOT NULL DEFAULT 'coordination'")

    @contextmanager
    def db(self):
        db = sqlite3.connect(self.path, timeout=15)
        db.row_factory = sqlite3.Row
        try:
            with db:
                yield db
        finally:
            db.close()

    def _get(self, db, table, key):
        row = db.execute('SELECT data FROM ' + table + ' WHERE id=?', (key,)).fetchone()
        if not row:
            raise Invalid('unknown ' + table)
        return json.loads(row['data'])

    def _put(self, db, table, data):
        db.execute('INSERT OR REPLACE INTO ' + table + '(id,data) VALUES(?,?)',
                   (data['id'], json.dumps(data, ensure_ascii=False)))

    def _event(self, db, kind, text, session_id=None):
        data = dict(kind=kind, text=text, session_id=session_id, created=time.time())
        cursor = db.execute('INSERT INTO events(data) VALUES(?)', (json.dumps(data, ensure_ascii=False),))
        return dict(data, id=cursor.lastrowid)

    def session(self, data):
        sid = identifier(data.get('id') or secrets.token_hex(12))
        record = dict(id=sid, name=string(data.get('name'), 'name', 200),
                      wp=string(data.get('wp', ''), 'wp', 100, True),
                      role=string(data.get('role', 'worker'), 'role', 100),
                      worktree=string(data.get('worktree', ''), 'worktree', 2000, True))
        thread = string(data.get('thread_id', ''), 'thread_id', 200, True)
        thread_url=string(data.get('thread_url',''), 'thread_url', 250, True)
        if thread_url and (not thread or thread_url!='codex://threads/'+thread or not re.fullmatch(r'[0-9a-fA-F-]{36}',thread)):
            raise Invalid('invalid thread_url')
        with self.db() as db:
            db.execute('BEGIN IMMEDIATE')
            old = db.execute('SELECT data FROM sessions WHERE id=?', (sid,)).fetchone()
            old = json.loads(old['data']) if old else {}
            record.update(thread_id=thread or old.get('thread_id', ''),
                          last_seen=old.get('last_seen'), created=old.get('created', time.time()))
            record['thread_url']=thread_url or old.get('thread_url','')
            self._put(db, 'sessions', record)
            self._event(db, 'session.registered', '会话记录已登记', sid)
        return record

    def request(self, data):
        record = dict(id=secrets.token_hex(12), name=string(data.get('name'), 'name', 200),
                      wp=string(data.get('wp', ''), 'wp', 100, True),
                      prompt=string(data.get('prompt'), 'prompt', 16000),
                      status='requested', created=time.time())
        with self.db() as db:
            self._put(db, 'requests', record)
            self._event(db, 'session.requested', '新建请求等待人工或本机桥接接入')
        return record

    def task(self, data, key=None):
        key = identifier(key or data.get('id') or secrets.token_hex(12))
        with self.db() as db:
            db.execute('BEGIN IMMEDIATE')
            row = db.execute('SELECT data FROM tasks WHERE id=?', (key,)).fetchone()
            old = json.loads(row['data']) if row else {}
            record = dict(old, **data)
            record['id'] = key
            allowed = {'id','name','session_id','status','phase','eta_min_minutes','eta_max_minutes','eta_basis','decision','updated'}
            if set(record) - allowed:
                raise Invalid('unknown task field')
            record['name'] = string(record.get('name'), 'name', 200)
            record.setdefault('status', 'queued')
            record.setdefault('phase', 'planning')
            if record['status'] not in {'queued','running','blocked','waiting','completed','failed','cancelled'}:
                raise Invalid('invalid status')
            if record['phase'] not in {'planning','implementation','validation','integration','done'}:
                raise Invalid('invalid phase')
            if record['status'] == 'completed' and record['phase'] != 'done':
                raise Invalid('completed task must have done phase')
            if record.get('session_id'):
                self._get(db, 'sessions', identifier(record['session_id']))
            lo, hi = record.get('eta_min_minutes'), record.get('eta_max_minutes')
            if lo is not None or hi is not None:
                if any(isinstance(v, bool) or not isinstance(v, (int,float)) or not math.isfinite(v) or v < 0 for v in (lo,hi)) or lo > hi:
                    raise Invalid('invalid ETA range')
                string(record.get('eta_basis'), 'eta_basis', 2000)
            for field in ('eta_basis','decision'):
                record[field] = string(record.get(field, ''), field, 4000, True)
            record['updated'] = time.time()
            self._put(db, 'tasks', record)
            self._event(db, 'task.updated', record['name'], record.get('session_id'))
        return record

    @staticmethod
    def public_message(row):
        result = {k: row[k] for k in ('id','session_id','text','source','reply_to','status','created','response_id')}
        result['outcome'] = row['status'] if row['status'] in ('completed','failed','uncertain') else None
        result['origin'] = row['origin']
        result['execution'] = 'started' if row['started'] else 'not_started'
        result['submission'] = json.loads(row['submission']) if row['submission'] else None
        return result

    def send(self, data):
        sid = identifier(data.get('session_id'))
        body = string(data.get('text'), 'text', 32000)
        source = data.get('source', 'human')
        if source not in ('human','agent'):
            raise Invalid('invalid source')
        key = identifier(data.get('idempotency_key'), 'idempotency_key')
        reply = data.get('reply_to')
        with self.db() as db:
            db.execute('BEGIN IMMEDIATE')
            self._get(db, 'sessions', sid)
            old = db.execute('SELECT * FROM messages WHERE session_id=? AND idempotency_key=?', (sid,key)).fetchone()
            if old:
                if (old['text'],old['source'],old['reply_to']) != (body,source,reply):
                    raise Conflict('idempotency conflict')
                return self.public_message(old)
            if reply:
                target = db.execute('SELECT session_id FROM messages WHERE id=?', (identifier(reply),)).fetchone()
                if not target or target['session_id'] != sid:
                    raise Invalid('reply must belong to same session')
            mid = secrets.token_hex(12)
            db.execute('INSERT INTO messages(id,session_id,text,source,reply_to,idempotency_key,status,created) VALUES(?,?,?,?,?,?,?,?)',
                       (mid,sid,body,source,reply,key,'queued' if source == 'human' else 'delivered',time.time()))
            self._event(db, 'message.sent', '消息已持久保存', sid)
            return self.public_message(db.execute('SELECT * FROM messages WHERE id=?', (mid,)).fetchone())

    def submission(self, data):
        sid, worker = identifier(data.get('session_id')), identifier(data.get('worker_id'))
        mid, lease = identifier(data.get('message_id')), identifier(data.get('lease_id'))
        phase = data.get('state')
        ranks = {'submitting': 1, 'accepted': 2, 'observed': 3}
        if phase not in ranks:
            raise Invalid('invalid submission state')
        thread = string(data.get('thread_id'), 'thread_id', 200)
        digest = string(data.get('input_sha256'), 'input_sha256', 64)
        if not re.fullmatch(r'[0-9a-f]{64}', digest):
            raise Invalid('invalid input_sha256')
        update = dict(state=phase, thread_id=thread, input_sha256=digest)
        for field in ('receipt_id', 'turn_id'):
            if data.get(field) is not None:
                update[field] = string(data[field], field, 200)
        if phase == 'accepted' and not update.get('receipt_id'):
            raise Invalid('accepted requires receipt_id')
        if phase == 'observed' and not update.get('turn_id'):
            raise Invalid('observed requires turn_id')
        with self.db() as db:
            db.execute('BEGIN IMMEDIATE')
            row = db.execute('SELECT * FROM messages WHERE id=? AND session_id=?', (mid,sid)).fetchone()
            if (not row or not row['started'] or row['worker_id'] != worker or row['lease_id'] != lease
                    or row['status'] not in ('claimed','completed','failed')
                    or (row['status'] == 'claimed' and row['lease_until'] <= time.time())):
                raise Conflict('expired or invalid submission claim')
            session = self._get(db, 'sessions', sid)
            if session.get('thread_id') != thread:
                raise Conflict('submission requires matching bound thread')
            old = json.loads(row['submission']) if row['submission'] else {}
            for field in ('thread_id','input_sha256','receipt_id','turn_id'):
                if old.get(field) and update.get(field) and old[field] != update[field]:
                    raise Conflict('submission identity conflict')
            merged = dict(old, **update)
            if ranks.get(old.get('state'),0) > ranks[phase]:
                merged['state'] = old['state']
            db.execute('UPDATE messages SET submission=? WHERE id=?', (json.dumps(merged),mid))
        return {'submission': merged}

    def mirror(self, data):
        sid = identifier(data.get('session_id'))
        thread = string(data.get('thread_id'), 'thread_id', 200)
        items = data.get('items')
        if not isinstance(items, list) or len(items) > 100:
            raise Invalid('invalid mirror batch')
        prepared = []
        for item in items:
            if not isinstance(item, dict):
                raise Invalid('invalid mirror item')
            external = string(item.get('id'), 'external item id', 500)
            source = item.get('source')
            if source not in ('human','agent'):
                raise Invalid('invalid mirror source')
            body = string(item.get('text'), 'text', 262144)
            created = item.get('created', time.time())
            if isinstance(created, bool) or not isinstance(created, (int,float)) or not math.isfinite(created) or created < 0:
                raise Invalid('invalid mirror timestamp')
            key = 'mirror:' + hashlib.sha256((thread+'\0'+source+'\0'+external).encode()).hexdigest()
            related = item.get('related_message_id')
            if related is not None:
                related = identifier(related)
            phase = string(item.get('phase',''), 'phase', 100, True)
            turn_status = item.get('turn_status','')
            if turn_status not in ('','inProgress','completed','failed','interrupted'):
                raise Invalid('invalid turn status')
            turn_id = string(item.get('turn_id',''), 'turn_id', 200, True)
            input_text = string(item.get('input_text'), 'input_text', 262144) if related else None
            prepared.append((key,source,body,created,external,related,phase,turn_status,turn_id,bool(item.get('allow_update',False)),input_text))
        with self.db() as db:
            db.execute('BEGIN IMMEDIATE')
            session = self._get(db, 'sessions', sid)
            if not session.get('thread_id') or session['thread_id'] != thread:
                raise Conflict('mirror requires matching bound thread')
            imported = 0
            deferred = []
            for key,source,body,created,external,related,phase,turn_status,turn_id,allow_update,input_text in prepared:
                local_id = None
                if related:
                    original = db.execute('SELECT * FROM messages WHERE id=? AND session_id=?', (related,sid)).fetchone()
                    if not original or original['source'] != 'human':
                        raise Conflict('unknown related input')
                    submission = json.loads(original['submission']) if original['submission'] else {}
                    if (not original['started'] or original['origin'] != 'coordination'
                            or submission.get('thread_id') != thread
                            or submission.get('input_sha256') != hashlib.sha256(input_text.encode('utf-8')).hexdigest()
                            or not turn_id or (submission.get('turn_id') and submission['turn_id'] != turn_id)):
                        raise Conflict('unproven history association')
                    if not submission.get('turn_id'):
                        submission.update(turn_id=turn_id, state='observed')
                        db.execute('UPDATE messages SET submission=? WHERE id=?', (json.dumps(submission),related))
                    if source == 'human':
                        if body != original['text']:
                            raise Conflict('related input text mismatch')
                        local_id = related
                    elif phase == 'final_answer' and turn_status in ('completed','failed','interrupted'):
                        if not original['started']:
                            raise Conflict('cannot reconcile unstarted execution')
                        if original['status'] == 'claimed' and original['lease_until'] > time.time():
                            deferred.append(external)
                            continue
                        outcome = 'completed' if turn_status == 'completed' else 'failed'
                        local_id = original['response_id'] or secrets.token_hex(12)
                        if original['response_id']:
                            db.execute("UPDATE messages SET text=?,status=? WHERE id=?", (body,'delivered' if outcome=='completed' else 'failed',local_id))
                        else:
                            db.execute('INSERT INTO messages(id,session_id,text,source,reply_to,idempotency_key,status,created,origin) VALUES(?,?,?,?,?,?,?,?,?)',
                                       (local_id,sid,body,'agent',related,'reply:'+related,'delivered' if outcome=='completed' else 'failed',created,'session_history'))
                            imported += 1
                        db.execute('UPDATE messages SET status=?,response_id=? WHERE id=?', (outcome,local_id,related))
                if local_id:
                    link=dict(message_id=local_id,thread_id=thread,turn_id=turn_id,phase=phase,text_sha256=hashlib.sha256(body.encode()).hexdigest())
                    db.execute('INSERT OR REPLACE INTO history_links VALUES(?,?,?,?)',(sid,external,source,json.dumps(link)))
                    continue
                old = db.execute('SELECT * FROM messages WHERE session_id=? AND idempotency_key=?', (sid,key)).fetchone()
                if old:
                    if old['source'] != source or old['origin'] != 'session_history' or (old['text'] != body and not allow_update):
                        raise Conflict('mirror item conflict')
                    if old['text'] != body:
                        db.execute('UPDATE messages SET text=? WHERE id=?',(body,old['id']))
                    continue
                local_id = secrets.token_hex(12)
                db.execute('INSERT INTO messages(id,session_id,text,source,idempotency_key,status,created,origin) VALUES(?,?,?,?,?,?,?,?)',
                           (local_id,sid,body,source,key,'delivered',created,'session_history'))
                link=dict(message_id=local_id,thread_id=thread,turn_id=turn_id,phase=phase,text_sha256=hashlib.sha256(body.encode()).hexdigest())
                db.execute('INSERT OR REPLACE INTO history_links VALUES(?,?,?,?)',(sid,external,source,json.dumps(link)))
                imported += 1
            # History synchronization is not proof of an available execution bridge.
            session['history_synced_at'] = time.time()
            self._put(db, 'sessions', session)
        return {'imported': imported, 'received': len(items), 'deferred': deferred}

    def messages(self, sid):
        sid = identifier(sid)
        with self.db() as db:
            self._get(db, 'sessions', sid)
            rows = db.execute('SELECT * FROM (SELECT * FROM messages WHERE session_id=? ORDER BY created DESC LIMIT 500) ORDER BY created', (sid,)).fetchall()
            return [self.public_message(row) for row in rows]

    def claim(self, data):
        sid, worker = identifier(data.get('session_id')), identifier(data.get('worker_id'))
        seconds = data.get('lease_seconds', 300)
        if isinstance(seconds, bool) or not isinstance(seconds, int) or not 10 <= seconds <= 3600:
            raise Invalid('invalid lease_seconds')
        now = time.time()
        with self.db() as db:
            db.execute('BEGIN IMMEDIATE')
            session = self._get(db, 'sessions', sid)
            session['last_seen'] = now
            self._put(db, 'sessions', session)
            db.execute("UPDATE messages SET status='uncertain' WHERE session_id=? AND status='claimed' AND started=1 AND lease_until<=?", (sid,now))
            if db.execute("SELECT id FROM messages WHERE session_id=? AND status='uncertain'", (sid,)).fetchone():
                return {'message': None, 'needs_reconciliation': True}
            # Serialize each session, even when several bridge workers poll it.
            if db.execute("SELECT id FROM messages WHERE session_id=? AND status='claimed' AND lease_until>?", (sid,now)).fetchone():
                return {'message': None}
            row = db.execute("SELECT * FROM messages WHERE session_id=? AND (status='queued' OR (status='claimed' AND lease_until<=?)) ORDER BY created LIMIT 1", (sid,now)).fetchone()
            if not row:
                return {'message': None}
            lease = secrets.token_hex(24)
            db.execute("UPDATE messages SET status='claimed',worker_id=?,lease_id=?,lease_until=? WHERE id=?", (worker,lease,now+seconds,row['id']))
            result = self.public_message(db.execute('SELECT * FROM messages WHERE id=?', (row['id'],)).fetchone())
            result.update(lease_id=lease, lease_until=now+seconds)
            return {'message': result}

    def start(self, data):
        sid, worker = identifier(data.get('session_id')), identifier(data.get('worker_id'))
        mid, lease = identifier(data.get('message_id')), identifier(data.get('lease_id'))
        with self.db() as db:
            db.execute('BEGIN IMMEDIATE')
            row = db.execute('SELECT * FROM messages WHERE id=? AND session_id=?', (mid,sid)).fetchone()
            if not row or row['worker_id'] != worker or row['lease_id'] != lease or row['status'] != 'claimed' or row['lease_until'] <= time.time():
                raise Conflict('expired or invalid claim')
            db.execute('UPDATE messages SET started=1 WHERE id=?', (mid,))
        return {'started': True}

    def uncertain(self, data):
        sid, worker = identifier(data.get('session_id')), identifier(data.get('worker_id'))
        mid, lease = identifier(data.get('message_id')), identifier(data.get('lease_id'))
        with self.db() as db:
            db.execute('BEGIN IMMEDIATE')
            row = db.execute('SELECT * FROM messages WHERE id=? AND session_id=?', (mid,sid)).fetchone()
            if not row or row['worker_id'] != worker or row['lease_id'] != lease or row['status'] not in ('claimed','uncertain'):
                raise Conflict('invalid claim')
            db.execute("UPDATE messages SET status='uncertain',started=1 WHERE id=?", (mid,))
            self._event(db, 'message.uncertain', '执行结果不确定；会话已暂停自动领取，等待人工核查', sid)
        return {'needs_reconciliation': True}

    def reconcile(self, data):
        sid, mid = identifier(data.get('session_id')), identifier(data.get('message_id'))
        action = data.get('action')
        resolution = string(data.get('resolution'), 'resolution', 4000)
        if action not in ('retry','failed'):
            raise Invalid('invalid reconciliation action')
        with self.db() as db:
            db.execute('BEGIN IMMEDIATE')
            row = db.execute('SELECT * FROM messages WHERE id=? AND session_id=?', (mid,sid)).fetchone()
            if not row or row['status'] != 'uncertain':
                raise Conflict('not awaiting reconciliation')
            db.execute('UPDATE messages SET status=?,started=0,worker_id=NULL,lease_id=NULL,lease_until=NULL WHERE id=?',
                       ('queued' if action == 'retry' else 'failed',mid))
            self._event(db, 'message.reconciled', action + ': ' + resolution, sid)
        return {'status': 'queued' if action == 'retry' else 'failed'}

    def renew(self, data):
        sid, worker = identifier(data.get('session_id')), identifier(data.get('worker_id'))
        mid, lease = identifier(data.get('message_id')), identifier(data.get('lease_id'))
        seconds = data.get('lease_seconds', 300)
        if isinstance(seconds, bool) or not isinstance(seconds, int) or not 10 <= seconds <= 3600:
            raise Invalid('invalid lease_seconds')
        now = time.time()
        with self.db() as db:
            db.execute('BEGIN IMMEDIATE')
            row = db.execute('SELECT * FROM messages WHERE id=? AND session_id=?', (mid,sid)).fetchone()
            if not row or row['worker_id'] != worker or row['lease_id'] != lease or row['status'] != 'claimed' or row['lease_until'] <= now:
                raise Conflict('expired or invalid claim')
            db.execute('UPDATE messages SET lease_until=? WHERE id=?', (now+seconds,mid))
            session = self._get(db, 'sessions', sid)
            session['last_seen'] = now
            self._put(db, 'sessions', session)
        return {'lease_until': now+seconds}

    def complete(self, data):
        sid, worker = identifier(data.get('session_id')), identifier(data.get('worker_id'))
        mid, lease = identifier(data.get('message_id')), identifier(data.get('lease_id'))
        body = string(data.get('text'), 'text', 262144)
        outcome = data.get('outcome', 'completed')
        if outcome not in ('completed','failed'):
            raise Invalid('invalid outcome')
        thread = string(data.get('thread_id'), 'thread_id', 200, outcome == 'failed')
        now = time.time()
        with self.db() as db:
            db.execute('BEGIN IMMEDIATE')
            row = db.execute('SELECT * FROM messages WHERE id=? AND session_id=?', (mid,sid)).fetchone()
            if not row or row['worker_id'] != worker or row['lease_id'] != lease:
                raise Conflict('invalid claim')
            if row['status'] in ('completed','failed'):
                response = db.execute('SELECT * FROM messages WHERE id=?', (row['response_id'],)).fetchone()
                if not response or response['text'] != body or row['status'] != outcome:
                    raise Conflict('completion conflict')
                return self.public_message(response)
            if row['status'] != 'claimed' or row['lease_until'] <= now:
                raise Conflict('expired claim')
            session = self._get(db, 'sessions', sid)
            if thread and session.get('thread_id') and session['thread_id'] != thread:
                raise Conflict('thread identity conflict')
            rid = secrets.token_hex(12)
            db.execute('INSERT INTO messages(id,session_id,text,source,reply_to,idempotency_key,status,created) VALUES(?,?,?,?,?,?,?,?)',
                       (rid,sid,body,'agent',mid,'reply:'+mid,'delivered' if outcome == 'completed' else 'failed',now))
            db.execute("UPDATE messages SET status=?,response_id=? WHERE id=?", (outcome,rid,mid))
            session.update(thread_id=thread or session.get('thread_id',''),last_seen=now)
            self._put(db, 'sessions', session)
            self._event(db, 'message.' + outcome, '本机桥接已回传结果' if outcome == 'completed' else '本机桥接执行失败', sid)
            return self.public_message(db.execute('SELECT * FROM messages WHERE id=?', (rid,)).fetchone())

    def event(self, data):
        kind = string(data.get('kind'), 'kind', 100)
        body = string(data.get('text'), 'text', 8000)
        sid = data.get('session_id')
        with self.db() as db:
            if sid:
                self._get(db, 'sessions', identifier(sid))
            return self._event(db, kind, body, sid)

    @staticmethod
    def _hook_record(data):
        allowed = {'event_id','source','task_id','session_id','severity','status',
                   'summary','impact','evidence','action','next_step'}
        required = {'event_id','source','severity','status','summary'}
        if not isinstance(data, dict) or set(data) - allowed or required - set(data):
            raise Invalid('invalid hook fields')
        if data['severity'] not in ('info','warning','critical') or data['status'] not in ('firing','resolved'):
            raise Invalid('invalid hook severity or status')
        record = dict(event_id=identifier(data['event_id'], 'event_id'),
                      source=identifier(data['source'], 'source'),
                      severity=data['severity'],status=data['status'],
                      summary=string(data['summary'], 'summary', 1000))
        for field in ('task_id','session_id'):
            record[field] = identifier(data[field], field) if data.get(field) is not None else None
        for field in ('impact','action','next_step'):
            value = data.get(field, '')
            if not isinstance(value, str) or len(value) > 2000:
                raise Invalid('invalid ' + field)
            record[field] = value
        evidence = data.get('evidence', [])
        if not isinstance(evidence, list) or len(evidence) > 8:
            raise Invalid('invalid evidence')
        record['evidence'] = [string(value, 'evidence item', 1000) for value in evidence]
        return record

    @staticmethod
    def _notification(row):
        return {'notification_id': row['id'], 'target': row['target'],
                'hook': json.loads(row['data']),
                'status': 'delivered' if row['delivered_at'] is not None else 'pending',
                'authorization': False}

    def hook(self, data):
        # Hooks are untrusted observations, never executable commands or approvals.
        record = self._hook_record(data)
        with self.db() as db:
            db.execute('BEGIN IMMEDIATE')
            if record['status'] == 'resolved':
                firing = db.execute('SELECT data FROM hooks WHERE source=? AND event_id=? AND status=?',
                                    (record['source'],record['event_id'],'firing')).fetchone()
                if not firing:
                    raise Conflict('resolution requires a firing event')
                original = json.loads(firing['data'])
                record['severity'] = original['severity']
                for field in ('task_id','session_id'):
                    if record[field] is not None and record[field] != original[field]:
                        raise Conflict('resolution object conflict')
                    record[field] = original[field]
            for field, table in (('task_id','tasks'),('session_id','sessions')):
                if record[field] is not None:
                    self._get(db, table, record[field])
            old = db.execute('SELECT id,data FROM hooks WHERE source=? AND event_id=? AND status=?',
                             (record['source'],record['event_id'],record['status'])).fetchone()
            if old:
                saved = json.loads(old['data'])
                if {k:v for k,v in saved.items() if k != 'created'} != record:
                    raise Conflict('hook idempotency conflict')
                notifications = [self._notification(row) for row in db.execute(
                    'SELECT * FROM hook_notifications WHERE hook_id=? ORDER BY seq', (old['id'],))]
                return {'hook':saved,'duplicate':True,'notifications':notifications}
            record['created'] = time.time()
            cursor = db.execute('INSERT INTO hooks(source,event_id,status,data) VALUES(?,?,?,?)',
                                (record['source'],record['event_id'],record['status'],json.dumps(record,ensure_ascii=False)))
            hook_id = cursor.lastrowid
            # Persist one event per transition, with bounded text rather than repeated evidence logs.
            self._event(db, 'hook.' + record['status'],
                        json.dumps({k:record[k] for k in ('source','event_id','severity','status','summary')},
                                   ensure_ascii=False), record['session_id'])
            targets = ('control','human') if record['severity'] == 'critical' else ('control',)
            notifications = []
            for target in targets:
                notification_id = secrets.token_hex(12)
                db.execute('INSERT INTO hook_notifications(id,hook_id,target,data) VALUES(?,?,?,?)',
                           (notification_id,hook_id,target,json.dumps(record,ensure_ascii=False)))
                row = db.execute('SELECT * FROM hook_notifications WHERE id=?',(notification_id,)).fetchone()
                notifications.append(self._notification(row))
        return {'hook':record,'duplicate':False,'notifications':notifications}

    def hook_notifications(self):
        with self.db() as db:
            db.execute('BEGIN')
            count = db.execute('SELECT count(*) FROM hook_notifications WHERE delivered_at IS NULL').fetchone()[0]
            rows = db.execute('SELECT * FROM hook_notifications WHERE delivered_at IS NULL ORDER BY seq LIMIT 100').fetchall()
            return {'notifications':[self._notification(row) for row in rows],
                    'pending_count':count,'has_more':count > len(rows)}

    def hook_ack(self, data):
        if not isinstance(data, dict) or set(data) != {'notification_id'}:
            raise Invalid('invalid hook acknowledgment fields')
        notification_id = identifier(data['notification_id'], 'notification_id')
        with self.db() as db:
            db.execute('BEGIN IMMEDIATE')
            row = db.execute('SELECT * FROM hook_notifications WHERE id=?',(notification_id,)).fetchone()
            if not row:
                raise Invalid('unknown notification')
            already = row['delivered_at'] is not None
            if not already:
                db.execute('UPDATE hook_notifications SET delivered_at=? WHERE id=?',(time.time(),notification_id))
        return {'notification_id':notification_id,'status':'delivered','duplicate':already}

    def state(self):
        with self.db() as db:
            db.execute('BEGIN')
            result = {table: [json.loads(row['data']) for row in db.execute('SELECT data FROM ' + table + ' ORDER BY id')]
                      for table in ('sessions','tasks','requests')}
            result['events'] = [dict(json.loads(row['data']),id=row['id']) for row in db.execute('SELECT * FROM events ORDER BY id DESC LIMIT 100')]
            result['hooks'] = {
                'pending_notifications': db.execute('SELECT count(*) FROM hook_notifications WHERE delivered_at IS NULL').fetchone()[0],
                'active_firing': db.execute("SELECT count(*) FROM hooks f WHERE f.status='firing' AND NOT EXISTS (SELECT 1 FROM hooks r WHERE r.source=f.source AND r.event_id=f.event_id AND r.status='resolved')").fetchone()[0],
                'recent': [json.loads(row['data']) for row in db.execute('SELECT data FROM hooks ORDER BY id DESC LIMIT 20')],
            }
            result['summary'] = {row['status']: row['n'] for row in db.execute('SELECT json_extract(data,\'$.status\') status,count(*) n FROM tasks GROUP BY status')}
        now = time.time()
        for session in result['sessions']:
            session['connection'] = ('connected' if session.get('thread_id') and session.get('last_seen') and now-session['last_seen'] < 120 else 'waiting')
        result['updated'] = now
        result['instance_id'] = self.instance_id
        return result