588 lines
35 KiB
Python
588 lines
35 KiB
Python
"""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
|