mayge/agent-workbench
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
agent-workbench/agent_control/store.py
T

2063 lines
122 KiB
Python
Raw Permalink Blame History

"""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 datetime import datetime
from pathlib import Path
from urllib.parse import urlsplit, unquote, quote
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
ATTENTION_PAGES = ('overview','tasks','cluster','communication','decisions','logs')
ATTENTION_BINDINGS = ('decision_id','task_id','hook_source','hook_event_id')
TASK_REPORT_DEFAULTS = {'execution_status':'pending','last_report_at':None,'last_report_id':None,'report_evidence':[]}
TASK_RELATION_DEFAULTS = {
'kind':'work', 'delivery_id':None,
'goal':'', 'scope':'', 'coupling_assessment':'', 'owner':'', 'integration_owner':'', 'version':'',
'parent_id':None, 'merged_into_id':None, 'depends_on':[], 'parallel_with':[], 'merge_sources':[],
'parallel_scope':'', 'acceptance_criteria':[], 'sync_points':[], 'integration_status':'not_started'}
ETA_ACTIVE = {'queued','running','blocked','waiting'}
ETA_DEFAULTS = {'eta_estimated_at':None, 'eta_revision':0, 'eta_change_reason':'',
'eta_last_reviewed_at':None, 'eta_review_status':'unestimated',
'eta_review_reason':'', 'eta_extension_count':0}
ETA_MANAGED = set(ETA_DEFAULTS) - {'eta_change_reason'}
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 decisions(
id TEXT PRIMARY KEY, data TEXT NOT NULL,
response TEXT, response_key TEXT);
CREATE TABLE IF NOT EXISTS decision_notifications(
seq INTEGER PRIMARY KEY AUTOINCREMENT,
id TEXT NOT NULL UNIQUE, decision_id TEXT NOT NULL UNIQUE,
data TEXT NOT NULL, delivered_at REAL);
CREATE INDEX IF NOT EXISTS decision_notifications_pending
ON decision_notifications(delivered_at,seq);
CREATE TABLE IF NOT EXISTS eta_reviews(
seq INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT NOT NULL UNIQUE,
task_id TEXT NOT NULL, revision INTEGER NOT NULL,
checkpoint TEXT NOT NULL, created REAL NOT NULL,
delivered_at REAL, cancelled_at REAL,
completion_body TEXT, completion_result TEXT,
UNIQUE(task_id,revision,checkpoint));
CREATE INDEX IF NOT EXISTS eta_reviews_pending
ON eta_reviews(delivered_at,cancelled_at,seq);
CREATE TABLE IF NOT EXISTS hook_observations(
source TEXT NOT NULL, event_id TEXT NOT NULL,
first_seen REAL NOT NULL, last_seen REAL NOT NULL,
repeat_count INTEGER NOT NULL, PRIMARY KEY(source,event_id));
CREATE TABLE IF NOT EXISTS hook_escalations(
source TEXT NOT NULL, event_id TEXT NOT NULL,
request TEXT NOT NULL, hook_id INTEGER NOT NULL,
PRIMARY KEY(source,event_id));
CREATE TABLE IF NOT EXISTS attention(id TEXT PRIMARY KEY, data TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS attention_history(
seq INTEGER PRIMARY KEY AUTOINCREMENT, attention_id TEXT NOT NULL,
data TEXT NOT NULL, created REAL NOT NULL);
CREATE TABLE IF NOT EXISTS task_integrations(
seq INTEGER PRIMARY KEY AUTOINCREMENT, target_id TEXT NOT NULL,
request_key TEXT NOT NULL, request TEXT NOT NULL, created REAL NOT NULL,
UNIQUE(target_id,request_key));
CREATE TABLE IF NOT EXISTS task_reports(
seq INTEGER PRIMARY KEY AUTOINCREMENT, report_id TEXT NOT NULL UNIQUE,
task_id TEXT NOT NULL, request TEXT NOT NULL, data TEXT NOT NULL);
CREATE INDEX IF NOT EXISTS task_reports_task ON task_reports(task_id,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.executescript("""
CREATE TABLE IF NOT EXISTS decision_history(
seq INTEGER PRIMARY KEY AUTOINCREMENT, decision_id TEXT NOT NULL, data TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS decision_operations(
decision_id TEXT NOT NULL, action TEXT NOT NULL, key TEXT NOT NULL,
request TEXT NOT NULL, result TEXT NOT NULL, PRIMARY KEY(decision_id,action,key));
CREATE TABLE IF NOT EXISTS decision_workflow_notifications(
seq INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT UNIQUE NOT NULL,
decision_id TEXT NOT NULL, event_key TEXT NOT NULL UNIQUE, data TEXT NOT NULL,
delivered_at REAL, cancelled_at REAL, submission_started_at REAL);
CREATE TABLE IF NOT EXISTS decision_credentials(
token_hash TEXT PRIMARY KEY, decision_id TEXT NOT NULL, created REAL NOT NULL);
CREATE TABLE IF NOT EXISTS decision_facts(
id TEXT PRIMARY KEY, decision_id TEXT NOT NULL, data TEXT NOT NULL);
""")
db.execute('BEGIN IMMEDIATE')
if 'workflow' not in {r['name'] for r in db.execute('PRAGMA table_info(decisions)')}:
db.execute('ALTER TABLE decisions ADD COLUMN workflow TEXT')
for row in db.execute('SELECT * FROM decisions WHERE workflow IS NULL').fetchall():
value = self._decision(row)
self._save_decision(db,value)
self._decision_history(db,value,'version',self._decision_content(value))
if value['response']:
self._decision_history(db,value,'submission',value['response'])
self._decision_notice(db,value,'session_required','control','session:'+value['id'])
old_notice = db.execute('SELECT * FROM decision_notifications WHERE decision_id=?',(value['id'],)).fetchone()
if old_notice:
payload = dict(notification_id=old_notice['id'],target='control',decision=value,
event=dict(kind='submitted',revision=1,created=value['response'].get('created')))
db.execute('INSERT OR IGNORE INTO decision_workflow_notifications(id,decision_id,event_key,data,delivered_at) VALUES(?,?,?,?,?)',
(old_notice['id'],value['id'],'legacy:'+value['id'],json.dumps(payload),old_notice['delivered_at']))
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']
hook_columns = {row['name'] for row in db.execute('PRAGMA table_info(hook_notifications)')}
if 'superseded_at' not in hook_columns:
db.execute('ALTER TABLE hook_notifications ADD COLUMN superseded_at REAL')
if 'submission_started_at' not in hook_columns:
db.execute('ALTER TABLE hook_notifications ADD COLUMN submission_started_at REAL')
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 {}
if old and old.get('role')!='decision' and record['role']=='decision':
raise Conflict('executor or main session cannot become consultation')
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','')
for decision_row in db.execute('SELECT * FROM decisions'):
bound=self._decision(decision_row)
if bound['session_id']==sid:
if (record['role']!='decision' or record['wp']!=bound['id']
or record['thread_id']!=bound['session_thread_id']):
raise Conflict('bound session identity immutable')
elif bound['session_thread_id'] and record['thread_id'].lower()==bound['session_thread_id'].lower():
raise Conflict('bound thread cannot be reused')
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
@staticmethod
def _attention_id(kind, *parts):
encoded = json.dumps(parts,ensure_ascii=False,separators=(',',':'))
return 'attention-'+kind+'-'+hashlib.sha256(encoded.encode('utf-8')).hexdigest()[:24]
@staticmethod
def _attention_public(item):
return {key:item[key] for key in ('id','page','required','reason','href','updated')}
@staticmethod
def _attention_href(value):
value = string(value,'href',1000)
decoded = value
for _ in range(5):
next_value = unquote(decoded)
if next_value == decoded:
break
decoded = next_value
if (unquote(decoded) != decoded or any(c in decoded for c in ('\\','//','<','>'))
or any(ord(c)<32 or ord(c)==127 for c in decoded)):
raise Invalid('invalid attention href')
parsed = urlsplit(decoded)
if (parsed.scheme or parsed.netloc or parsed.fragment
or parsed.path not in {'/'+page for page in ATTENTION_PAGES}
or urlsplit(value).path not in {'/'+page for page in ATTENTION_PAGES}
or not value.startswith('/')):
raise Invalid('invalid attention href')
return value
def _attention_write(self, db, item, now):
old_row = db.execute('SELECT data FROM attention WHERE id=?',(item['id'],)).fetchone()
old = json.loads(old_row['data']) if old_row else None
if old:
if any(old.get(key) != item.get(key) for key in ('page',)+ATTENTION_BINDINGS):
raise Conflict('attention subject differs')
if {k:v for k,v in old.items() if k!='updated'} == item:
return old, True
record = dict(item,updated=now)
self._put(db,'attention',record)
db.execute('INSERT INTO attention_history(attention_id,data,created) VALUES(?,?,?)',
(record['id'],json.dumps(record,ensure_ascii=False),now))
return record, False
def _attention_revoke(self, db, criteria, now):
for row in db.execute('SELECT data FROM attention').fetchall():
record = json.loads(row['data'])
if record['required'] and all(record.get(k)==v for k,v in criteria.items()):
record.pop('updated',None)
record['required'] = False
self._attention_write(db,record,now)
def attention(self, data):
fields = {'id','page','required','reason','href'} | set(ATTENTION_BINDINGS)
if (not isinstance(data,dict) or set(data)-fields
or not {'id','page','required','reason'} <= set(data)):
raise Invalid('invalid attention fields')
aid = identifier(data['id'])
page = data['page']
if page not in ATTENTION_PAGES or not isinstance(data['required'],bool):
raise Invalid('invalid attention page or required')
reason = data['reason']
if not isinstance(reason,str):
raise Invalid('invalid attention reason')
string(reason,'reason',240,optional=not data['required'])
if bool('hook_source' in data) != bool('hook_event_id' in data):
raise Invalid('hook identity requires both fields')
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
row = db.execute('SELECT data FROM attention WHERE id=?',(aid,)).fetchone()
old = json.loads(row['data']) if row else {}
record = dict(id=aid,page=page,required=data['required'],reason=reason,
href=self._attention_href(data.get('href',old.get('href','/'+page))))
for field in ATTENTION_BINDINGS:
record[field] = identifier(data[field],field) if field in data else old.get(field)
for field,table in (('decision_id','decisions'),('task_id','tasks')):
if record[field] is not None:
self._get(db,table,record[field])
if record['hook_source'] is not None:
found = db.execute("SELECT 1 FROM hooks WHERE source=? AND event_id=? AND status='firing'",
(record['hook_source'],record['hook_event_id'])).fetchone()
if not found:
raise Invalid('unknown hook')
# Deterministic automatic IDs cannot be pre-bound to unrelated subjects.
if aid.startswith('attention-decision-') and (
record['decision_id'] is None or aid != self._attention_id('decision',record['decision_id'])):
raise Invalid('reserved attention id')
if aid.startswith('attention-hook-') and (
record['hook_source'] is None or aid != self._attention_id('hook',record['hook_source'],record['hook_event_id'])):
raise Invalid('reserved attention id')
saved, duplicate = self._attention_write(db,record,time.time())
return dict(self._attention_public(saved),duplicate=duplicate)
def _attention_state(self, db):
items = []
for row in db.execute('SELECT data FROM attention ORDER BY id'):
item = json.loads(row['data'])
if not item['required']:
continue
if item.get('decision_id'):
decision = db.execute('SELECT * FROM decisions WHERE id=?',(item['decision_id'],)).fetchone()
if not decision or self._decision(decision)['status'] not in ('open','needs_reconsideration'):
continue
if item.get('task_id'):
task = db.execute('SELECT data FROM tasks WHERE id=?',(item['task_id'],)).fetchone()
if not task or json.loads(task['data']).get('status') == 'cancelled':
continue
if item.get('hook_source'):
resolved = db.execute("SELECT 1 FROM hooks WHERE source=? AND event_id=? AND status='resolved'",
(item['hook_source'],item['hook_event_id'])).fetchone()
if resolved:
continue
items.append(self._attention_public(item))
items.sort(key=lambda item:(item['updated'],item['id']),reverse=True)
return {'items':items,'pages':{page:any(i['page']==page for i in items) for page in ATTENTION_PAGES}}
def _attention_auto(self, db, kind, parts, page, reason, href, now, **bindings):
aid = self._attention_id(kind,*parts)
# Creating the source is the only trigger: reads and retries never recreate a withdrawn item.
item = dict(id=aid,page=page,required=True,reason=reason[:240],href=href)
item.update({key:bindings.get(key) for key in ATTENTION_BINDINGS})
return self._attention_write(db,item,now)
@staticmethod
def _task_defaults(record):
return {**{k:list(v) if isinstance(v,list) else v for k,v in TASK_RELATION_DEFAULTS.items()},
**ETA_DEFAULTS, **TASK_REPORT_DEFAULTS, 'report_evidence':[],
'current_progress':'', 'progress_updated':None, 'created_at':None, **record}
@staticmethod
def _task_ids(value, name, known, self_id=None):
if not isinstance(value,list) or len(value)>30:
raise Invalid('invalid '+name)
result = []
for value_id in value:
identifier(value_id,name)
if value_id not in known or value_id==self_id:
raise Invalid('invalid '+name+' reference')
if value_id not in result:
result.append(value_id)
return sorted(result)
def _task_relations(self, record, known):
record = self._task_defaults(record)
for field in ('goal','scope','coupling_assessment','owner','integration_owner','version','parallel_scope'):
limit = 4000 if field in ('goal','scope','coupling_assessment') else 2000 if field=='parallel_scope' else 200
if not isinstance(record[field],str):
raise Invalid('invalid '+field)
string(record[field],field,limit,True)
if record['kind'] not in ('work','delivery'):
raise Invalid('invalid task kind')
if record['kind']=='delivery' and record['delivery_id'] is not None:
raise Invalid('delivery cannot belong to another delivery')
for field in ('parent_id','merged_into_id','delivery_id'):
value = record[field]
if value is not None:
identifier(value,field)
if value==record['id'] or value not in known:
raise Invalid('invalid '+field+' reference')
for field in ('depends_on','parallel_with','merge_sources'):
record[field] = self._task_ids(record[field],field,known,record['id'])
if set(record['depends_on']) & set(record['parallel_with']) and not record['parallel_scope'].strip():
raise Invalid('parallel dependency requires scoped work and waiting point')
criteria = record['acceptance_criteria']
if not isinstance(criteria,list) or len(criteria)>20:
raise Invalid('invalid acceptance_criteria')
record['acceptance_criteria'] = [string(value,'acceptance criterion',1000) for value in criteria]
points = record['sync_points']
if not isinstance(points,list) or len(points)>20:
raise Invalid('invalid sync_points')
normalized, ids = [], set()
for point in points:
if not isinstance(point,dict) or set(point)!={'id','label','task_ids','status'}:
raise Invalid('invalid sync point fields')
point_id = identifier(point['id'],'sync point id')
if point_id in ids or point['status'] not in ('waiting','ready','completed'):
raise Invalid('invalid sync point')
ids.add(point_id)
normalized.append(dict(id=point_id,label=string(point['label'],'sync point label',200),
task_ids=self._task_ids(point['task_ids'],'sync point tasks',known),
status=point['status']))
record['sync_points'] = normalized
if record['integration_status'] not in ('not_started','waiting','integrating','integrated','verified'):
raise Invalid('invalid integration_status')
return record
@staticmethod
def _validate_task_graphs(tasks):
for task_id,task in tasks.items():
delivery = task.get('delivery_id')
if delivery is not None:
if delivery==task_id or delivery not in tasks:
raise Invalid('invalid delivery reference')
if task.get('kind','work')=='delivery' or tasks[delivery].get('kind','work')!='delivery':
raise Conflict('delivery reference requires a delivery task')
graphs = [
{key:([value['parent_id']] if value.get('parent_id') else []) for key,value in tasks.items()},
{key:value.get('depends_on',[]) for key,value in tasks.items()},
{key:([value['merged_into_id']] if value.get('merged_into_id') else []) for key,value in tasks.items()},
]
# Each relation is its own graph: containment does not impose execution order.
for graph in graphs:
indegree = {key:0 for key in graph}
for source,targets in graph.items():
for target in targets:
if target not in graph or target==source:
raise Invalid('invalid task relation reference')
indegree[target] += 1
ready = [key for key,count in indegree.items() if count==0]
visited = 0
while ready:
source = ready.pop()
visited += 1
for target in graph[source]:
indegree[target] -= 1
if indegree[target]==0:
ready.append(target)
if visited != len(graph):
raise Conflict('task relation cycle')
for key,task in tasks.items():
sources = task.get('merge_sources',[])
if len(sources)>30:
raise Invalid('too many merge sources')
for source in sources:
if source not in tasks or source==key or tasks[source].get('merged_into_id') != key:
raise Conflict('merge relationship is not reciprocal')
target = task.get('merged_into_id')
if target is not None and key not in tasks[target].get('merge_sources',[]):
raise Conflict('merge relationship is not reciprocal')
def _stage_task_relations(self, db, record, old, explicit):
tasks = {row['id']:self._task_defaults(json.loads(row['data']))
for row in db.execute('SELECT id,data FROM tasks')}
original = dict(tasks)
record = self._task_relations(record,set(tasks))
tasks[record['id']] = record
key = record['id']
def mutable(task_id):
# Copy before touching another task so change detection remains reliable.
tasks[task_id] = dict(tasks[task_id])
return tasks[task_id]
if 'merged_into_id' in explicit:
previous, target = old.get('merged_into_id'),record['merged_into_id']
if previous and target and previous!=target:
raise Conflict('source already merged elsewhere')
if previous!=target:
if previous:
parent = mutable(previous)
parent['merge_sources'] = [x for x in parent['merge_sources'] if x!=key]
if target:
parent = mutable(target)
parent['merge_sources'] = sorted(set(parent['merge_sources'])|{key})
if 'merge_sources' in explicit:
previous = set(old.get('merge_sources',[]))
wanted = set(record['merge_sources'])
for source in previous-wanted:
child = mutable(source)
if child.get('merged_into_id') != key:
raise Conflict('source relationship differs')
child['merged_into_id'] = None
for source in wanted:
child = mutable(source)
if child.get('merged_into_id') not in (None,key):
raise Conflict('source already merged elsewhere')
child['merged_into_id'] = key
self._validate_task_graphs(tasks)
changed = {task_id:task for task_id,task in tasks.items()
if task_id==key or task!=original.get(task_id)}
return tasks[key], changed
def integrate(self, target_id, data):
target_id = identifier(target_id,'target_id')
if (not isinstance(data,dict) or set(data)-{'source_ids','owner','status','version','reason'}
or not {'source_ids','owner','status','reason'} <= set(data)):
raise Invalid('invalid integration fields')
owner = string(data['owner'],'owner',200)
reason = string(data['reason'],'reason',2000)
status = data['status']
if status not in ('waiting','integrating','integrated','verified'):
raise Invalid('invalid integration status')
if 'version' in data:
if not isinstance(data['version'],str):
raise Invalid('invalid version')
string(data['version'],'version',200,True)
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
tasks = {row['id']:self._task_defaults(json.loads(row['data']))
for row in db.execute('SELECT id,data FROM tasks')}
if target_id not in tasks:
raise Invalid('unknown integration target')
sources = self._task_ids(data['source_ids'],'source_ids',set(tasks),target_id)
if not sources:
raise Invalid('integration sources required')
request = dict(source_ids=sources,owner=owner,status=status,reason=reason)
if 'version' in data:
request['version'] = data['version']
encoded = json.dumps(request,ensure_ascii=False,sort_keys=True,separators=(',',':'))
request_key = hashlib.sha256(encoded.encode('utf-8')).hexdigest()
prior = db.execute('SELECT request FROM task_integrations WHERE target_id=? AND request_key=?',
(target_id,request_key)).fetchone()
if prior:
if any(tasks[source].get('merged_into_id')!=target_id or
source not in tasks[target_id]['merge_sources'] for source in sources):
raise Conflict('previous integration relationship changed')
return {'task':tasks[target_id],'sources':[tasks[source] for source in sources],'duplicate':True}
target = dict(tasks[target_id])
target['merge_sources'] = sorted(set(target['merge_sources'])|set(sources))
target['integration_owner'] = owner
target['integration_status'] = status
if 'version' in data:
target['version'] = data['version']
target, changed = self._stage_task_relations(db,target,tasks[target_id],{'merge_sources'})
now = time.time()
for task_id,task in changed.items():
task['updated'] = now
self._put(db,'tasks',task)
tasks[task_id] = task
db.execute('INSERT INTO task_integrations(target_id,request_key,request,created) VALUES(?,?,?,?)',
(target_id,request_key,encoded,now))
self._event(db,'task.integration_updated',json.dumps(dict(target_id=target_id,**request),
ensure_ascii=False),target.get('session_id'))
return {'task':tasks[target_id],'sources':[tasks[source] for source in sources],'duplicate':False}
@staticmethod
def _eta_deadline(record):
start, upper = record.get('eta_estimated_at'), record.get('eta_max_minutes')
if any(isinstance(v,bool) or not isinstance(v,(int,float)) or not math.isfinite(v)
for v in (start,upper)) or upper <= 0:
return None
deadline = start + upper * 60
return deadline if math.isfinite(deadline) else None
def _eta_change(self, record, old, data, now):
for field, default in ETA_DEFAULTS.items():
record[field] = old.get(field,default)
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 or not math.isfinite(now + hi*60)):
raise Invalid('invalid ETA range')
if record['status'] in ETA_ACTIVE and hi <= 0:
raise Invalid('active ETA upper bound must be positive')
string(record.get('eta_basis'),'eta_basis',2000)
if 'eta_change_reason' in data:
string(data['eta_change_reason'],'eta_change_reason',4000)
changed = (lo,hi) != (old.get('eta_min_minutes'),old.get('eta_max_minutes'))
# Legacy estimates acquire a clock only when explicitly supplied.
initialize = hi is not None and not old.get('eta_revision') and (
'eta_min_minutes' in data or 'eta_max_minutes' in data)
if not changed and not initialize:
return False
first = not old.get('eta_revision')
reason = data.get('eta_change_reason') or (record.get('eta_basis') if first else None)
string(reason,'eta_change_reason',4000)
new_deadline = now + hi*60 if hi is not None else None
old_deadline = self._eta_deadline(old)
extension = (not first and new_deadline is not None and record['status'] in ETA_ACTIVE
and (old_deadline is None or new_deadline > old_deadline))
if extension and old.get('eta_extension_count',0) >= 3:
return True
record.update(eta_estimated_at=now if hi is not None else None,
eta_revision=old.get('eta_revision',0)+1, eta_change_reason=reason,
eta_review_status='scheduled' if hi is not None else 'unestimated',
eta_extension_count=old.get('eta_extension_count',0)+int(extension))
return False
def _cancel_eta(self, db, record, now):
if record.get('status') not in ETA_ACTIVE or self._eta_deadline(record) is None:
db.execute('UPDATE eta_reviews SET cancelled_at=? WHERE task_id=? AND cancelled_at IS NULL',
(now,record['id']))
else:
db.execute('UPDATE eta_reviews SET cancelled_at=? WHERE task_id=? AND revision<>? AND cancelled_at IS NULL',
(now,record['id'],record.get('eta_revision',0)))
def task(self, data, key=None):
if key is not None and 'id' in data and data['id'] != key:
raise Invalid('task id differs from target')
if 'created_at' in data or 'progress_updated' in data or (ETA_MANAGED | set(TASK_REPORT_DEFAULTS)) & set(data):
raise Invalid('server managed task field')
key = identifier(key or data.get('id') or secrets.token_hex(12))
rejected = False
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','current_progress','progress_updated','created_at'} | set(ETA_DEFAULTS) | set(TASK_RELATION_DEFAULTS) | set(TASK_REPORT_DEFAULTS)
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 ETA_ACTIVE | {'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']))
for field in ('eta_basis','decision'):
record[field] = string(record.get(field, ''), field, 4000, True)
progress = record.get('current_progress','')
if not isinstance(progress,str):
raise Invalid('invalid current_progress')
record['current_progress'] = string(progress,'current_progress',1000,True)
record, related = self._stage_task_relations(db,record,old,set(data))
now = time.time()
record['created_at'] = old.get('created_at') if row else now
rejected = self._eta_change(record,old,data,now)
if rejected:
record = self._task_defaults(old)
record['eta_review_status'] = 'needs_decision'
record['eta_review_reason'] = data['eta_change_reason']
record['eta_last_reviewed_at'] = now
if old.get('eta_review_status') != 'needs_decision':
self._event(db,'eta.needs_decision',record['name'],record.get('session_id'))
self._put(db,'tasks',record)
else:
record['progress_updated'] = now if 'current_progress' in data else old.get('progress_updated')
record['updated'] = now
self._put(db, 'tasks', record)
for related_id,related_task in related.items():
if related_id != key:
related_task['updated'] = now
self._put(db,'tasks',related_task)
self._cancel_eta(db,record,now)
if record['status'] == 'cancelled':
self._attention_revoke(db,{'task_id':record['id']},now)
self._event(db, 'task.updated', record['name'], record.get('session_id'))
if rejected:
raise Conflict('ETA extension requires decision')
return record
def task_report(self, data):
required = {'report_id','task_id','session_id','stage','current_progress','summary','evidence'}
if not isinstance(data,dict) or required-set(data) or set(data)-required-{'version','needs_attention'}:
raise Invalid('invalid task report fields')
request = {key:identifier(data[key],key) for key in ('report_id','task_id','session_id')}
if data['stage'] not in ('execution_completed','blocked','progress','failed'):
raise Invalid('invalid report stage')
request['stage'] = data['stage']
for field,limit in (('current_progress',1000),('summary',2000)):
if not isinstance(data[field],str):
raise Invalid('invalid '+field)
request[field] = string(data[field],field,limit,True)
evidence = data['evidence']
if not isinstance(evidence,list) or len(evidence)>20:
raise Invalid('invalid report evidence')
request['evidence'] = [string(value,'evidence',1000) for value in evidence]
if 'version' in data:
if not isinstance(data['version'],str):
raise Invalid('invalid version')
request['version'] = string(data['version'],'version',200,True)
attention = data.get('needs_attention',False)
if not isinstance(attention,bool) or (attention and not request['summary'].strip()):
raise Invalid('invalid report attention')
request['needs_attention'] = attention
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
old = db.execute('SELECT request,data FROM task_reports WHERE report_id=?',
(request['report_id'],)).fetchone()
if old:
if json.loads(old['request']) != request:
raise Conflict('report id already used')
return {'task':self._task_defaults(self._get(db,'tasks',request['task_id'])),
'report':json.loads(old['data']),'duplicate':True}
self._get(db,'sessions',request['session_id'])
task = self._task_defaults(self._get(db,'tasks',request['task_id']))
if task.get('session_id') != request['session_id']:
raise Conflict('report session does not own task')
now = time.time()
report = dict(request,created=now)
protected = (task.get('status') in ('completed','cancelled') or
task.get('integration_status') in ('integrated','verified') or
(request['stage']=='progress' and task.get('execution_status')=='completed'
and task.get('status')=='waiting' and task.get('phase')=='integration'))
if not protected:
stage = request['stage']
if stage == 'execution_completed':
task.update(execution_status='completed',status='waiting',phase='integration')
elif stage == 'progress':
task.update(execution_status='running',status='running')
else:
task.update(execution_status=stage,status=stage)
task['current_progress'] = request['current_progress']
task['progress_updated'] = now
if 'version' in request:
task['version'] = request['version']
task.update(last_report_at=now,last_report_id=request['report_id'],
report_evidence=list(request['evidence']),updated=now)
db.execute('INSERT INTO task_reports(report_id,task_id,request,data) VALUES(?,?,?,?)',
(request['report_id'],request['task_id'],json.dumps(request,ensure_ascii=False),
json.dumps(report,ensure_ascii=False)))
self._put(db,'tasks',task)
self._cancel_eta(db,task,now)
if attention:
aid = self._attention_id('task-report',task['id'])
item = dict(id=aid,page='tasks',required=True,reason=request['summary'][:240],
href='/tasks?task='+quote(task['id'],safe=''))
item.update({key:task['id'] if key=='task_id' else None for key in ATTENTION_BINDINGS})
self._attention_write(db,item,now)
return {'task':task,'report':report,'duplicate':False}
def task_reports(self, task_id):
task_id = identifier(task_id,'task_id')
with self.db() as db:
db.execute('BEGIN')
self._get(db,'tasks',task_id)
rows = db.execute('SELECT data FROM task_reports WHERE task_id=? ORDER BY seq DESC LIMIT 50',
(task_id,)).fetchall()
return {'reports':[json.loads(row['data']) for row in rows]}
def eta_tick(self, now=None):
now = time.time() if now is None else now
if isinstance(now,bool) or not isinstance(now,(int,float)) or not math.isfinite(now):
raise Invalid('invalid clock')
created = 0
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
for row in db.execute('SELECT data FROM tasks').fetchall():
task = self._task_defaults(json.loads(row['data']))
deadline = self._eta_deadline(task)
if task.get('status') not in ETA_ACTIVE or deadline is None or not task['eta_revision']:
continue
elapsed = now - task['eta_estimated_at']
duration = task['eta_max_minutes'] * 60
pending = False
for checkpoint,fraction in [('half',0.5),('near',0.8),('overdue',1)]:
if elapsed < duration*fraction:
continue
inserted = db.execute(
'INSERT OR IGNORE INTO eta_reviews(id,task_id,revision,checkpoint,created) VALUES(?,?,?,?,?)',
(secrets.token_hex(16),task['id'],task['eta_revision'],checkpoint,now)).rowcount
created += inserted
pending = pending or bool(inserted)
if task['eta_review_status'] != 'needs_decision':
if now >= deadline:
task['eta_review_status'] = 'overdue'
elif pending:
task['eta_review_status'] = 'pending'
self._put(db,'tasks',task)
return {'created':created}
def eta_notifications(self):
with self.db() as db:
db.execute('BEGIN')
# Current task data, not the progress snapshot at checkpoint creation.
rows = db.execute(
"SELECT r.*,t.data AS task_data FROM eta_reviews r JOIN tasks t ON t.id=r.task_id "
"WHERE r.delivered_at IS NULL AND r.cancelled_at IS NULL AND r.completion_body IS NULL "
"AND r.revision=json_extract(t.data,'$.eta_revision') "
"AND json_extract(t.data,'$.status') IN ('queued','running','blocked','waiting') "
"ORDER BY r.seq").fetchall()
return {'notifications':[{'notification_id':row['id'],'target':'control',
'task':self._task_defaults(json.loads(row['task_data'])),
'checkpoint':row['checkpoint'],'revision':row['revision']}
for row in rows[:100]], 'pending_count':len(rows)}
def eta_ack(self, data):
if not isinstance(data,dict) or set(data) != {'notification_id'}:
raise Invalid('invalid acknowledgment fields')
nid = identifier(data['notification_id'],'notification_id')
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
row = db.execute('SELECT delivered_at FROM eta_reviews WHERE id=?',(nid,)).fetchone()
if not row:
raise Invalid('unknown ETA review')
duplicate = row['delivered_at'] is not None
if not duplicate:
db.execute('UPDATE eta_reviews SET delivered_at=? WHERE id=?',(time.time(),nid))
return {'notification_id':nid,'status':'delivered','duplicate':duplicate}
def eta_complete(self, review_id, data):
review_id = identifier(review_id,'review_id')
allowed = {'result','reason','current_progress','eta_min_minutes','eta_max_minutes'}
if not isinstance(data,dict) or set(data)-allowed or not {'result','reason'} <= set(data):
raise Invalid('invalid review fields')
result = data['result']
if result not in ('keep','revise','unknown'):
raise Invalid('invalid review result')
string(data['reason'],'reason',4000)
ranges = {'eta_min_minutes','eta_max_minutes'} & set(data)
if (result == 'revise' and len(ranges) != 2) or (result != 'revise' and ranges):
raise Invalid('ETA range is only required for revise')
if 'current_progress' in data:
if not isinstance(data['current_progress'],str):
raise Invalid('invalid current_progress')
string(data['current_progress'],'current_progress',1000,True)
rejected = False
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
row = db.execute('SELECT * FROM eta_reviews WHERE id=?',(review_id,)).fetchone()
if not row:
raise Invalid('unknown ETA review')
if row['completion_body'] is not None:
if json.loads(row['completion_body']) == data:
return json.loads(row['completion_result'])
raise Conflict('review already completed')
old = self._get(db,'tasks',row['task_id'])
if (old.get('status') not in ETA_ACTIVE or old.get('eta_revision') != row['revision']
or row['cancelled_at'] is not None):
raise Conflict('obsolete ETA review')
now = time.time()
task = self._task_defaults(old)
if result == 'revise':
if data['eta_min_minutes'] is None or data['eta_max_minutes'] is None:
raise Invalid('revise requires numeric range')
update = {'eta_min_minutes':data['eta_min_minutes'],'eta_max_minutes':data['eta_max_minutes'],
'eta_change_reason':data['reason']}
task.update(update)
rejected = self._eta_change(task,old,update,now)
if rejected:
task = self._task_defaults(old)
task['eta_review_status'] = 'needs_decision'
task['eta_review_reason'] = data['reason']
task['eta_last_reviewed_at'] = now
if old.get('eta_review_status') != 'needs_decision':
self._event(db,'eta.needs_decision',task['name'],task.get('session_id'))
self._put(db,'tasks',task)
else:
if result == 'unknown' and task['eta_review_status'] != 'needs_decision':
task['eta_review_status'] = 'unknown'
elif result in ('keep','revise') and task['eta_revision'] == old['eta_revision'] and task['eta_review_status'] != 'needs_decision':
task['eta_review_status'] = 'overdue' if now >= self._eta_deadline(task) else 'reviewed'
task['eta_last_reviewed_at'] = now
task['eta_review_reason'] = data['reason']
task['updated'] = now
if 'current_progress' in data:
task['current_progress'] = data['current_progress']
task['progress_updated'] = now
self._put(db,'tasks',task)
self._cancel_eta(db,task,now)
db.execute('UPDATE eta_reviews SET completion_body=?,completion_result=? WHERE id=?',
(json.dumps(data,ensure_ascii=False),json.dumps(task,ensure_ascii=False),review_id))
self._event(db,'eta.reviewed',task['name'],task.get('session_id'))
if rejected:
raise Conflict('ETA extension requires decision')
return task
@staticmethod
def public_message(row):
from .conversation import delivery_fields
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
result.update(delivery_fields(dict(row)))
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)
for decision_row in db.execute('SELECT * FROM decisions'):
bound=self._decision(decision_row)
if bound['session_id']==sid and bound['status']=='completed':
raise Conflict('consultation closed')
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 conversation(self, sid):
from .conversation import project
from .notices import fact_answer_notice
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()
links = [json.loads(row['data']) for row in db.execute('SELECT data FROM history_links WHERE session_id=?', (sid,))]
notices = [json.loads(row['data']) for row in db.execute(
"SELECT data FROM decision_workflow_notifications WHERE json_extract(data,'$.target')='decision' "
"AND json_extract(data,'$.decision.session_id')=? AND json_extract(data,'$.event.kind')='fact_answered'", (sid,))]
return project([dict(row) for row in rows], links, {fact_answer_notice(n) for n in notices})
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','error','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 'superseded' if row['superseded_at'] is not None else 'pending',
'submission_started_at':row['submission_started_at'],
'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 id,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'])
escalation = db.execute(
"SELECT data FROM hooks WHERE source=? AND event_id=? AND status='escalated'",
(record['source'],record['event_id'])).fetchone()
record['severity'] = 'critical' if escalation else 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')
if record['status'] == 'firing':
self._observe_hook(db,saved,time.time(),repeat=True)
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
if record['status'] == 'firing':
self._observe_hook(db,record,record['created'])
else:
self._observe_hook(db,original,record['created'])
self._attention_revoke(db,{'hook_source':record['source'],'hook_event_id':record['event_id']},
record['created'])
# 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',)
notice = dict(record,log_id=str(hook_id if record['status']=='firing' else firing['id']))
if record['status']=='resolved' and escalation:
notice['escalated'] = True
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(notice,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}
@staticmethod
def _observe_hook(db, firing, now, repeat=False):
# Counts live separately so original firing/resolution audit rows stay immutable.
existing = db.execute('SELECT 1 FROM hook_observations WHERE source=? AND event_id=?',
(firing['source'],firing['event_id'])).fetchone()
if existing:
db.execute('UPDATE hook_observations SET last_seen=max(last_seen,?),repeat_count=repeat_count+? '
'WHERE source=? AND event_id=?',
(now,int(repeat),firing['source'],firing['event_id']))
else:
db.execute('INSERT INTO hook_observations(source,event_id,first_seen,last_seen,repeat_count) VALUES(?,?,?,?,?)',
(firing['source'],firing['event_id'],firing['created'],max(now,firing['created']),1+int(repeat)))
def hook_escalate(self, data):
if (not isinstance(data,dict) or set(data)-{'source','event_id','reason','impact'}
or not {'source','event_id','reason'} <= set(data)):
raise Invalid('invalid escalation fields')
request = {'source':identifier(data['source'],'source'),
'event_id':identifier(data['event_id'],'event_id'),
'reason':string(data['reason'],'reason',2000)}
if 'impact' in data:
if not isinstance(data['impact'],str):
raise Invalid('invalid impact')
request['impact'] = string(data['impact'],'impact',2000,True)
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
original_row = db.execute("SELECT * FROM hooks WHERE source=? AND event_id=? AND status='firing'",
(request['source'],request['event_id'])).fetchone()
resolved = db.execute("SELECT 1 FROM hooks WHERE source=? AND event_id=? AND status='resolved'",
(request['source'],request['event_id'])).fetchone()
if not original_row or resolved:
raise Conflict('escalation requires active firing')
original = json.loads(original_row['data'])
request.setdefault('impact',original['impact'])
prior = db.execute('SELECT * FROM hook_escalations WHERE source=? AND event_id=?',
(request['source'],request['event_id'])).fetchone()
if prior:
if json.loads(prior['request']) != request:
raise Conflict('escalation already recorded')
row = db.execute('SELECT data FROM hooks WHERE id=?',(prior['hook_id'],)).fetchone()
notifications = [self._notification(n) for n in db.execute(
'SELECT * FROM hook_notifications WHERE hook_id=? ORDER BY seq',(prior['hook_id'],))]
return {'hook':json.loads(row['data']),'duplicate':True,'notifications':notifications}
if original['severity'] == 'critical':
raise Conflict('event already critical')
record = dict(original,severity='critical',status='firing',
summary=('持续故障：'+original['summary'])[:1000],
impact=request['impact'],created=time.time())
hook_id = db.execute('INSERT INTO hooks(source,event_id,status,data) VALUES(?,?,?,?)',
(request['source'],request['event_id'],'escalated',
json.dumps(record,ensure_ascii=False))).lastrowid
db.execute('INSERT INTO hook_escalations(source,event_id,request,hook_id) VALUES(?,?,?,?)',
(request['source'],request['event_id'],json.dumps(request,ensure_ascii=False),hook_id))
self._observe_hook(db,original,record['created'])
self._event(db,'hook.escalated',json.dumps(request,ensure_ascii=False),record['session_id'])
self._attention_auto(db,'hook',(record['source'],record['event_id']),'logs',
self._safe_log_text(record['summary'],240),
'/logs?incident='+str(original_row['id']),record['created'],
hook_source=record['source'],hook_event_id=record['event_id'])
notifications = []
notice = dict(record,escalated=True,log_id=str(original_row['id']))
for target in ('control','human'):
nid = secrets.token_hex(12)
db.execute('INSERT INTO hook_notifications(id,hook_id,target,data) VALUES(?,?,?,?)',
(nid,hook_id,target,json.dumps(notice,ensure_ascii=False)))
notifications.append(self._notification(
db.execute('SELECT * FROM hook_notifications WHERE id=?',(nid,)).fetchone()))
return {'hook':record,'duplicate':False,'notifications':notifications}
@staticmethod
def _safe_log_text(value, limit=400):
# This is a human summary projection, never a raw log export.
value = re.sub(r'\x1b\[[0-?]*[ -/]*[@-~]', '', str(value))
value = re.sub(r'-----BEGIN [A-Z0-9 ]*PRIVATE KEY-----[\s\S]*?(?:-----END [A-Z0-9 ]*PRIVATE KEY-----|$)',
'[密钥已隐藏]',value,flags=re.I)
value = re.sub(r'(?i)\b(?:Bearer|Basic)\s+[A-Za-z0-9._~+/=-]+','[凭据已隐藏]',value)
value = re.sub(r"""(?ix)
\b[\w-]*(?:token|secret|password|passwd|api[_-]?key|private[_-]?key|credential|authorization|cookie|密钥|令牌|密码)[\w-]*
["']?\s*[:=]\s*(?:"[^"]*"|'[^']*'|[^\s,;}&]+)
""",'[凭据已隐藏]',value)
value = re.sub(r'(?i)(https?://)[^/\s:@]+:[^/\s@]+@',r'\1[凭据已隐藏]@',value)
value = re.sub(r'\b(?:sk-[A-Za-z0-9_-]{8,}|gh[pousr]_[A-Za-z0-9_]+|github_pat_[A-Za-z0-9_]+)\b',
'[凭据已隐藏]',value)
value = re.sub(r'\beyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\b',
'[凭据已隐藏]',value)
value = re.sub(r'\b[A-Za-z0-9_+/=-]{32,}\b','[标识已隐藏]',value)
value = re.sub(r"""(?i)file://[^\s"'<>]+""",'[路径已隐藏]',value)
value = re.sub(r"""(?i)\b[A-Z]:[\\/][^\r\n"'<>|]*""",'[路径已隐藏]',value)
value = re.sub(r"""\\\\[^\s"'<>]+""",'[路径已隐藏]',value)
value = re.sub(r"""(?<![\w:/])/(?!/)[^\s"'<>),;]+""",'[路径已隐藏]',value)
value = re.sub(r'[\x00-\x1f\x7f]+',' ',value)
return value[:limit] + ('…' if len(value)>limit else '')
def logs(self, level=None, status=None):
if level is not None and level not in ('info','warning','error','critical'):
raise Invalid('invalid log level')
if status is not None and status not in ('active','resolved'):
raise Invalid('invalid log status')
with self.db() as db:
db.execute('BEGIN')
rows = db.execute("""
SELECT f.id, f.data AS firing, r.data AS resolution, e.data AS escalation,
o.first_seen,o.last_seen,o.repeat_count
FROM hooks f
LEFT JOIN hooks r ON r.source=f.source AND r.event_id=f.event_id AND r.status='resolved'
LEFT JOIN hooks e ON e.source=f.source AND e.event_id=f.event_id AND e.status='escalated'
LEFT JOIN hook_observations o ON o.source=f.source AND o.event_id=f.event_id
WHERE f.status='firing'
""").fetchall()
logs = []
for row in rows:
firing = json.loads(row['firing'])
resolution = json.loads(row['resolution']) if row['resolution'] else None
effective = json.loads(row['escalation']) if row['escalation'] else firing
item_level = effective['severity']
if (firing['source']=='bridge-mirror' and resolution is not None and not row['escalation']
and 0 <= resolution['created']-firing['created'] <= 120):
item_level = 'warning'
item_status = 'resolved' if resolution else 'active'
if (level is not None and level != item_level) or (status is not None and status != item_status):
continue
latest = resolution or effective
first = row['first_seen'] if row['first_seen'] is not None else firing['created']
last = max(row['last_seen'] or first,latest['created'])
item = {'id':str(row['id']),'level':item_level,'status':item_status,
'first_seen':first,'last_seen':last,
'resolved_at':resolution['created'] if resolution else None,
'repeat_count':row['repeat_count'] if row['repeat_count'] is not None else 1}
for field in ('source','event_id','summary','impact'):
item[field] = self._safe_log_text(effective[field])
for field in ('action','next_step'):
item[field] = self._safe_log_text(latest[field])
item['evidence'] = [self._safe_log_text(v,240) for v in (latest['evidence'] or effective['evidence'])]
logs.append(item)
logs.sort(key=lambda item:(item['last_seen'],int(item['id'])),reverse=True)
return {'logs':logs[:100],'total_count':len(logs),'has_more':len(logs)>100}
def hook_submitting(self, data):
if not isinstance(data,dict) or set(data)!={'notification_id'}:
raise Invalid('invalid submission marker fields')
nid = identifier(data['notification_id'],'notification_id')
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
row = db.execute('SELECT * FROM hook_notifications WHERE id=?',(nid,)).fetchone()
if not row:
raise Invalid('unknown notification')
if row['superseded_at'] is not None and row['delivered_at'] is None:
raise Conflict('notification superseded')
duplicate = row['submission_started_at'] is not None
started = row['submission_started_at'] if duplicate else time.time()
if not duplicate:
db.execute('UPDATE hook_notifications SET submission_started_at=? WHERE id=?',(started,nid))
return {'notification_id':nid,'submission_started_at':started,'duplicate':duplicate,
'status':'delivered' if row['delivered_at'] is not None else 'submitting'}
def hook_coalesce(self, data):
if not isinstance(data,dict) or data:
raise Invalid('coalesce requires empty object')
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
now = time.time()
rows = db.execute("""
SELECT n.*,h.source,h.event_id,h.status AS transition
FROM hook_notifications n JOIN hooks h ON h.id=n.hook_id
WHERE n.delivered_at IS NULL AND n.superseded_at IS NULL
AND n.submission_started_at IS NULL
AND EXISTS (SELECT 1 FROM hooks r WHERE r.source=h.source AND r.event_id=h.event_id
AND r.status='resolved')
ORDER BY n.seq
""").fetchall()
count = 0
for row in rows:
if row['transition']=='resolved':
# A started/unknown submission may already be in the host: retain its recovery.
received = db.execute("""
SELECT 1 FROM hook_notifications n JOIN hooks h ON h.id=n.hook_id
WHERE h.source=? AND h.event_id=? AND n.target=?
AND h.status IN ('firing','escalated')
AND (n.delivered_at IS NOT NULL OR n.submission_started_at IS NOT NULL)
LIMIT 1
""",(row['source'],row['event_id'],row['target'])).fetchone()
if received:
continue
elif row['transition'] not in ('firing','escalated'):
continue
count += db.execute(
'UPDATE hook_notifications SET superseded_at=? WHERE id=? AND delivered_at IS NULL '
'AND superseded_at IS NULL AND submission_started_at IS NULL',(now,row['id'])).rowcount
total = db.execute('SELECT count(*) FROM hook_notifications WHERE superseded_at IS NOT NULL').fetchone()[0]
return {'superseded_count':count,'total_superseded_count':total,
'scope':'service_owned_unsubmitted_notifications'}
def queue_health(self, now=None):
now = time.time() if now is None else now
if isinstance(now,bool) or not isinstance(now,(int,float)) or not math.isfinite(now):
raise Invalid('invalid clock')
since = now-60
with self.db() as db:
db.execute('BEGIN')
def stats(table, created, pending):
row = db.execute(
'SELECT sum(CASE WHEN '+pending+' THEN 1 ELSE 0 END) AS pending_count, '
'min(CASE WHEN '+pending+' THEN '+created+' END) AS oldest_pending_at, '
'sum(CASE WHEN '+created+' BETWEEN ? AND ? THEN 1 ELSE 0 END) AS created_last_60s, '
'sum(CASE WHEN delivered_at BETWEEN ? AND ? THEN 1 ELSE 0 END) AS delivered_last_60s '
'FROM '+table,(since,now,since,now)).fetchone()
return {key:(row[key] if key=='oldest_pending_at' else row[key] or 0)
for key in row.keys()}
hooks = stats('hook_notifications',"json_extract(data,'$.created')",
'delivered_at IS NULL AND superseded_at IS NULL')
hooks['submission_started_unconfirmed_count'] = db.execute(
'SELECT count(*) FROM hook_notifications WHERE submission_started_at IS NOT NULL '
'AND delivered_at IS NULL AND superseded_at IS NULL').fetchone()[0]
hooks['superseded_count'] = db.execute(
'SELECT count(*) FROM hook_notifications WHERE superseded_at IS NOT NULL').fetchone()[0]
eta_pending = ("delivered_at IS NULL AND cancelled_at IS NULL AND completion_body IS NULL "
"AND EXISTS (SELECT 1 FROM tasks t WHERE t.id=eta_reviews.task_id "
"AND json_extract(t.data,'$.eta_revision')=eta_reviews.revision "
"AND json_extract(t.data,'$.status') IN ('queued','running','blocked','waiting'))")
eta = stats('eta_reviews','created',eta_pending)
eta['cancelled_count'] = db.execute(
'SELECT count(*) FROM eta_reviews WHERE cancelled_at IS NOT NULL').fetchone()[0]
decisions = stats('decision_workflow_notifications',"json_extract(data,'$.event.created')",
'delivered_at IS NULL AND cancelled_at IS NULL')
decisions['cancelled_count'] = db.execute('SELECT count(*) FROM decision_workflow_notifications WHERE cancelled_at IS NOT NULL').fetchone()[0]
decisions['submission_started_unconfirmed_count'] = db.execute('SELECT count(*) FROM decision_workflow_notifications WHERE submission_started_at IS NOT NULL AND delivered_at IS NULL AND cancelled_at IS NULL').fetchone()[0]
reports = db.execute(
"SELECT count(*) AS received_total, sum(CASE WHEN json_extract(data,'$.created') "
"BETWEEN ? AND ? THEN 1 ELSE 0 END) AS received_last_60s FROM task_reports",(since,now)).fetchone()
return {'scope':'service_owned_outboxes','delivery_semantics':'host_accepted_not_model_completed',
'sampled_at':now,'hooks':hooks,'eta':eta,'decisions':decisions,
'task_reports':{key:reports[key] or 0 for key in reports.keys()}}
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 AND superseded_at IS NULL').fetchone()[0]
rows = db.execute('SELECT * FROM hook_notifications WHERE delivered_at IS NULL AND superseded_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}
DECISION_FIELDS = {'id','title','question','impact','priority','details','recommendation',
'options','deadline','requires_authorization','task_id','summary','supersedes_id'}
@classmethod
def _decision_content(cls, value):
return {k:value.get(k) for k in cls.DECISION_FIELDS}
@classmethod
def _decision(cls, row):
if 'workflow' in row.keys() and row['workflow']:
return json.loads(row['workflow'])
result = json.loads(row['data'])
result.setdefault('summary','')
result.setdefault('supersedes_id',None)
response = json.loads(row['response']) if row['response'] else None
if response:
response = dict(response,revision=1,snapshot=cls._decision_content(result))
result.update(response=response,status='pending_review' if response else 'open',revision=1,
session_id=None,session_thread_id=None,archive_status='none',review=None,
completion=None,related_decision_ids=[])
return result
def _load_decision(self, db, did):
row = db.execute('SELECT * FROM decisions WHERE id=?',(identifier(did,'decision_id'),)).fetchone()
if not row:
raise Invalid('unknown decision')
return self._decision(row)
def _save_decision(self, db, value):
db.execute('UPDATE decisions SET workflow=?,response=? WHERE id=?',
(json.dumps(value,ensure_ascii=False),json.dumps(value['response']) if value['response'] else None,value['id']))
def _decision_history(self, db, value, kind, data):
event = dict(id=secrets.token_hex(16),kind=kind,revision=value['revision'],created=time.time(),data=data)
db.execute('INSERT INTO decision_history(decision_id,data) VALUES(?,?)',(value['id'],json.dumps(event,ensure_ascii=False)))
def _decision_notice(self, db, value, kind, target, event_key, **details):
nid = secrets.token_hex(16)
payload = dict(notification_id=nid,target=target,decision=value,
event=dict(kind=kind,revision=value['revision'],created=time.time(),**details))
db.execute('INSERT OR IGNORE INTO decision_workflow_notifications(id,decision_id,event_key,data) VALUES(?,?,?,?)',
(nid,value['id'],event_key,json.dumps(payload,ensure_ascii=False)))
def _cancel_decision_notices(self, db, did, kind, request_id=None):
# Once external submission has begun the immutable input must remain reconcilable.
sql = "UPDATE decision_workflow_notifications SET cancelled_at=? WHERE decision_id=? AND delivered_at IS NULL AND submission_started_at IS NULL AND json_extract(data,'$.event.kind')=?"
args = [time.time(),did,kind]
if request_id is not None:
sql += " AND json_extract(data,'$.event.request_id')=?"
args.append(request_id)
db.execute(sql,args)
@staticmethod
def _decision_revision(value, data):
rev = data.get('expected_revision')
if type(rev) is not int or rev < 1:
raise Invalid('positive expected_revision required')
if rev != value['revision']:
raise Conflict('decision revision changed')
@staticmethod
def _decision_fields(data, required, optional=()):
if not isinstance(data,dict) or not set(required) <= data.keys() or data.keys()-set(required)-set(optional):
raise Invalid('invalid workflow fields')
def _decision_replay(self, db, did, action, data):
if 'expected_revision' in data and (type(data['expected_revision']) is not int or data['expected_revision'] < 1):
raise Invalid('positive expected_revision required')
key = identifier(data.get('idempotency_key'),'idempotency_key')
row = db.execute('SELECT * FROM decision_operations WHERE decision_id=? AND action=? AND key=?',(did,action,key)).fetchone()
if row:
if json.loads(row['request']) != data:
raise Conflict('idempotency conflict')
return json.loads(row['result'])
def _decision_record(self, db, value, action, data, result=None):
result = value if result is None else result
db.execute('INSERT INTO decision_operations VALUES(?,?,?,?,?)',
(value['id'],action,data['idempotency_key'],json.dumps(data,ensure_ascii=False),json.dumps(result,ensure_ascii=False)))
return result
def decision_get(self, did):
with self.db() as db:
return self._load_decision(db,did)
def decision_history(self, did):
with self.db() as db:
self._load_decision(db,did)
return {'history':[json.loads(r['data']) for r in db.execute('SELECT data FROM decision_history WHERE decision_id=? ORDER BY seq',(did,))]}
def _normalize_decision(self, data):
fields = self.DECISION_FIELDS
if not isinstance(data, dict) or set(data) - fields or not {'id','title','question'} <= set(data):
raise Invalid('invalid decision fields')
value = {'id':identifier(data['id'])}
for name, limit in [('title',200),('question',4000),('impact',4000),
('details',8000),('recommendation',4000),('summary',1000)]:
raw = data.get(name, '')
if not isinstance(raw, str):
raise Invalid('invalid ' + name)
value[name] = string(raw,name,limit,optional=name not in ('title','question'))
priority = data.get('priority','normal')
if priority not in ('normal','high','urgent'):
raise Invalid('invalid priority')
value['priority'] = priority
options = data.get('options',[])
if not isinstance(options,list) or len(options) > 20:
raise Invalid('invalid options')
value['options'] = []
ids = set()
for option in options:
if not isinstance(option,dict) or set(option) != {'id','label','description'}:
raise Invalid('invalid option fields')
oid = identifier(option['id'],'option id')
if oid in ids or not isinstance(option['description'],str):
raise Invalid('invalid option')
ids.add(oid)
value['options'].append({'id':oid,'label':string(option['label'],'label',200),
'description':string(option['description'],'description',2000,optional=True)})
deadline = data.get('deadline')
if deadline is not None:
string(deadline,'deadline',64)
if not re.fullmatch(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?(?:Z|[+-]\d{2}:\d{2})?',deadline):
raise Invalid('invalid deadline')
try:
datetime.fromisoformat(deadline.replace('Z','+00:00'))
except ValueError:
raise Invalid('invalid deadline')
value['deadline'] = deadline
auth = data.get('requires_authorization',False)
if not isinstance(auth,bool):
raise Invalid('invalid requires_authorization')
value['requires_authorization'] = auth
task = data.get('task_id')
value['task_id'] = identifier(task,'task_id') if task is not None else None
supersedes = data.get('supersedes_id')
value['supersedes_id'] = identifier(supersedes,'supersedes_id') if supersedes is not None else None
return value
def decision(self, data):
value = self._normalize_decision(data)
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
row = db.execute('SELECT * FROM decisions WHERE id=?',(value['id'],)).fetchone()
if row:
if self._normalize_decision(json.loads(row['data'])) != value:
raise Conflict('decision definition differs')
return self._decision(row)
if value['task_id']:
self._get(db,'tasks',value['task_id'])
parent = None
if value['supersedes_id']:
if value['supersedes_id'] == value['id']:
raise Invalid('self supersedes')
parent = self._load_decision(db,value['supersedes_id'])
db.execute('INSERT INTO decisions(id,data) VALUES(?,?)',(value['id'],json.dumps(value,ensure_ascii=False)))
value.update(status='open',response=None,revision=1,session_id=None,session_thread_id=None,
archive_status='none',review=None,completion=None,
related_decision_ids=[parent['id']] if parent else [])
if parent:
parent['related_decision_ids'].append(value['id'])
self._save_decision(db,parent)
self._decision_history(db,parent,'linked',{'decision_id':value['id']})
self._save_decision(db,value)
self._decision_history(db,value,'version',self._decision_content(value))
self._decision_notice(db,value,'session_required','control','session:'+value['id'])
self._event(db,'decision.created',value['title'])
self._attention_auto(db,'decision',(value['id'],),'decisions','待决策：'+value['title'],
'/decisions?decision='+quote(value['id'],safe=''),time.time(),decision_id=value['id'])
return value
def decision_patch(self, did, data):
self._decision_fields(data,{'expected_revision','reason'},self.DECISION_FIELDS-{'id','supersedes_id'})
reason = string(data['reason'],'reason',4000)
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
value = self._load_decision(db,did)
self._decision_revision(value,data)
if value['status'] == 'completed':
raise Conflict('decision closed')
old = self._decision_content(value)
changes = {k:v for k,v in data.items() if k in self.DECISION_FIELDS}
if not changes:
raise Invalid('content changes required')
content = self._normalize_decision(dict(old,**changes))
if content['task_id']:
self._get(db,'tasks',content['task_id'])
if content == old:
return value
value.update(content)
value['revision'] += 1
if value['response']:
value['status'] = 'needs_reconsideration'
self._save_decision(db,value)
self._decision_history(db,value,'version',dict(content=content,reason=reason))
self._cancel_decision_notices(db,did,'submitted')
if value['status']=='needs_reconsideration':
self._attention_auto(db,'decision',(did,),'decisions','需要重新确认：'+value['title'],
'/decisions?decision='+quote(did,safe=''),time.time(),decision_id=did)
return value
def decision_respond(self, did, data):
self._decision_fields(data,{'expected_revision','option_id','comment','idempotency_key'})
option = data['option_id']
if option is not None:
identifier(option,'option_id')
comment = string(data['comment'],'comment',8000,optional=True)
if not isinstance(data['comment'],str) or (option is None and not comment.strip()):
raise Invalid('response required')
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
value = self._load_decision(db,did)
replay = self._decision_replay(db,did,'respond',data)
if replay is not None:
return replay
self._decision_revision(value,data)
if value['status'] not in ('open','needs_reconsideration'):
raise Conflict('decision not open')
if option is not None and option not in {o['id'] for o in value['options']}:
raise Invalid('invalid choice')
response = dict(id=secrets.token_hex(16),revision=value['revision'],option_id=option,comment=comment,
created=time.time(),authorization=False,snapshot=self._decision_content(value))
value.update(response=response,status='pending_review',review=None)
self._save_decision(db,value)
self._decision_history(db,value,'submission',response)
self._decision_notice(db,value,'submitted','control','submitted:'+response['id'])
self._attention_revoke(db,{'decision_id':did},response['created'])
self._event(db,'decision.responded',value['title'])
return self._decision_record(db,value,'respond',data)
def decision_review(self, did, data):
self._decision_fields(data,{'expected_revision','idempotency_key','outcome','reason','impact','execution_plan'}, {'authorization_evidence'})
if data['outcome'] not in ('accept','reconsider'):
raise Invalid('invalid outcome')
for k in ('reason','impact','execution_plan'):
string(data[k],k,8000)
if 'authorization_evidence' in data:
string(data['authorization_evidence'],'authorization_evidence',8000)
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
value = self._load_decision(db,did)
replay = self._decision_replay(db,did,'review',data)
if replay is not None:
return replay
self._decision_revision(value,data)
if value['status'] != 'pending_review' or not value['response'] or value['response']['revision'] != value['revision']:
raise Conflict('current submission required')
review = {k:v for k,v in data.items() if k not in ('expected_revision','idempotency_key')}
review.update(revision=value['revision'],created=time.time())
value.update(review=review,status='accepted' if data['outcome']=='accept' else 'needs_reconsideration')
self._save_decision(db,value)
self._decision_history(db,value,'review',review)
if value['status']=='needs_reconsideration':
self._attention_auto(db,'decision',(did,),'decisions','需要重新确认：'+value['title'],
'/decisions?decision='+quote(did,safe=''),time.time(),decision_id=did)
self._cancel_decision_notices(db,did,'submitted')
self._decision_notice(db,value,'reviewed','human','reviewed:'+did+':'+data['idempotency_key'])
return self._decision_record(db,value,'review',data)
@staticmethod
def _decision_evidence(data):
evidence = data.get('evidence',[])
if not isinstance(evidence,list) or len(evidence)>50:
raise Invalid('invalid evidence')
return [string(v,'evidence',4000) for v in evidence]
@staticmethod
def _decision_pending(db, value):
return db.execute("SELECT count(*) FROM messages WHERE session_id=? AND status IN ('queued','claimed','uncertain')",(value['session_id'],)).fetchone()[0]
def decision_complete(self, did, data):
self._decision_fields(data,{'expected_revision','idempotency_key','conclusion','handoff'}, {'evidence'})
for k in ('conclusion','handoff'):
string(data[k],k,8000)
evidence = self._decision_evidence(data)
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
value = self._load_decision(db,did)
replay = self._decision_replay(db,did,'complete',data)
if replay is not None:
return replay
self._decision_revision(value,data)
if value['status'] not in ('accepted','needs_reconsideration') or not value['review'] or value['review']['revision'] != value['revision']:
raise Conflict('current review required')
if not value['session_id'] or not value['session_thread_id']:
raise Conflict('bound consultation required')
if self._decision_pending(db,value):
raise Conflict('consultation messages pending')
completion = dict(revision=value['revision'],conclusion=data['conclusion'],handoff=data['handoff'],evidence=evidence,created=time.time())
value.update(completion=completion,status='completed',archive_status='pending')
self._save_decision(db,value)
self._decision_history(db,value,'completion',completion)
self._attention_revoke(db,{'decision_id':did},time.time())
self._decision_notice(db,value,'archive_required','control','archive:'+did)
return self._decision_record(db,value,'complete',data)
def decision_bind(self, did, data):
self._decision_fields(data,{'expected_revision','idempotency_key','session_id','thread_id'})
sid = identifier(data['session_id'],'session_id')
thread = data['thread_id']
if not isinstance(thread,str) or not re.fullmatch(r'[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}',thread):
raise Invalid('invalid thread UUID')
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
value = self._load_decision(db,did)
replay = self._decision_replay(db,did,'bind',data)
if replay is not None:
return replay
self._decision_revision(value,data)
if value['session_id']:
raise Conflict('binding unavailable')
session = self._get(db,'sessions',sid)
if session.get('role') != 'decision' or session.get('wp') != did or session.get('thread_id','').lower() != thread.lower():
raise Conflict('session identity mismatch')
for row in db.execute('SELECT * FROM sessions WHERE id<>?',(sid,)):
if json.loads(row['data']).get('thread_id','').lower()==thread.lower():
raise Conflict('thread reused')
for row in db.execute('SELECT * FROM decisions WHERE id<>?',(did,)):
other = self._decision(row)
if other['session_id']==sid or (other['session_thread_id'] or '').lower()==thread.lower():
raise Conflict('decision binding reused')
value.update(session_id=sid,session_thread_id=thread)
self._save_decision(db,value)
self._decision_history(db,value,'binding',dict(session_id=sid,thread_id=thread))
self._cancel_decision_notices(db,did,'session_required')
return self._decision_record(db,value,'bind',data)
def decision_archive(self, did, data):
self._decision_fields(data,{'thread_id','idempotency_key','receipt'})
string(data['receipt'],'receipt',8000)
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
value = self._load_decision(db,did)
replay = self._decision_replay(db,did,'archive',data)
if replay is not None:
return replay
if (value['status']!='completed' or not value['review'] or not value['completion']
or not value['completion']['handoff'] or not value['session_thread_id']
or value['session_thread_id'] != data['thread_id'] or value['archive_status']!='pending'):
raise Conflict('archive prerequisites missing')
if self._decision_pending(db,value):
raise Conflict('consultation messages pending')
value['archive_status']='archived'
self._save_decision(db,value)
self._decision_history(db,value,'archive',dict(thread_id=data['thread_id'],receipt=data['receipt']))
self._cancel_decision_notices(db,did,'archive_required')
return self._decision_record(db,value,'archive',data)
def decision_bindings(self):
with self.db() as db:
bindings=[]
for row in db.execute('SELECT * FROM decisions'):
value=self._decision(row)
if value['session_id']:
session=self._get(db,'sessions',value['session_id'])
rows=db.execute("SELECT * FROM messages WHERE session_id=? AND source='human' AND origin='coordination' AND status IN ('queued','claimed','uncertain')", (value['session_id'],)).fetchall()
now=time.time()
queued=[m for m in rows if m['status']=='queued' or (m['status']=='claimed' and not m['started'] and (m['lease_until'] or 0)<=now)]
uncertain=[m for m in rows if m['status']=='uncertain' or (m['status']=='claimed' and m['started'] and (m['lease_until'] or 0)<=now)]
inflight=[m for m in rows if m not in queued and m not in uncertain]
demand=dict(queued=len(queued),inflight=len(inflight),uncertain=len(uncertain),
human_queued=len(queued),oldest_queued_at=min((m['created'] for m in queued),default=None))
bindings.append(dict(decision_id=value['id'],session_id=value['session_id'],thread_id=value['session_thread_id'],
worktree=session['worktree'],status=value['status'],archive_status=value['archive_status'],
pending_messages=self._decision_pending(db,value),demand=demand))
return {'bindings':bindings}
def decision_consultation(self, did, data=None):
if data is None:
value=self.decision_get(did)
if not value['session_id']:
raise Conflict('consultation not bound')
return {'messages':self.conversation(value['session_id'])}
self._decision_fields(data,{'text','client_key'})
body=string(data['text'],'text',32000)
key=identifier(data['client_key'],'client_key')
# Status check and insertion share the same write lock as completion.
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
value=self._load_decision(db,did)
if not value['session_id'] or value['status']=='completed':
raise Conflict('consultation unavailable')
sid=value['session_id']
old=db.execute('SELECT * FROM messages WHERE session_id=? AND idempotency_key=?',(sid,key)).fetchone()
if old:
if old['text']!=body or old['source']!='human' or old['reply_to'] is not None:
raise Conflict('idempotency conflict')
return self.public_message(old)
mid=secrets.token_hex(12)
db.execute('INSERT INTO messages(id,session_id,text,source,idempotency_key,status,created) VALUES(?,?,?,?,?,?,?)',
(mid,sid,body,'human',key,'queued',time.time()))
return self.public_message(db.execute('SELECT * FROM messages WHERE id=?',(mid,)).fetchone())
def decision_credential(self, did, data):
self._decision_fields(data,set())
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
value=self._load_decision(db,did)
if not value['session_id'] or value['status']=='completed':
raise Conflict('consultation unavailable')
token=secrets.token_urlsafe(48)
db.execute('INSERT INTO decision_credentials VALUES(?,?,?)',(hashlib.sha256(token.encode()).hexdigest(),did,time.time()))
return {'token':token}
def decision_credential_scope(self, token):
with self.db() as db:
row=db.execute('SELECT decision_id FROM decision_credentials WHERE token_hash=?',(hashlib.sha256(token.encode()).hexdigest(),)).fetchone()
return row['decision_id'] if row else None
def decision_fact(self, did, data):
self._decision_fields(data,{'idempotency_key','question'},{'context'})
question=string(data['question'],'question',8000)
context=string(data.get('context',''),'context',8000,optional=True)
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
value=self._load_decision(db,did)
replay=self._decision_replay(db,did,'fact',data)
if replay is not None:
return replay
if value['status']=='completed' or not value['session_id']:
raise Conflict('consultation unavailable')
fact=dict(id=secrets.token_hex(16),revision=value['revision'],question=question,context=context,answer=None,created=time.time())
db.execute('INSERT INTO decision_facts VALUES(?,?,?)',(fact['id'],did,json.dumps(fact,ensure_ascii=False)))
self._decision_history(db,value,'fact_requested',fact)
self._decision_notice(db,value,'fact_requested','control','fact:'+fact['id'],request_id=fact['id'],question=question,context=context)
return self._decision_record(db,value,'fact',data,fact)
def decision_fact_answer(self, did, rid, data):
self._decision_fields(data,{'idempotency_key','answer'},{'evidence'})
answer=string(data['answer'],'answer',8000)
evidence=self._decision_evidence(data)
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
value=self._load_decision(db,did)
action='fact_answer:'+identifier(rid,'request_id')
replay=self._decision_replay(db,did,action,data)
if replay is not None:
return replay
row=db.execute('SELECT data FROM decision_facts WHERE id=? AND decision_id=?',(rid,did)).fetchone()
if not row:
raise Invalid('unknown fact request')
fact=json.loads(row['data'])
if fact['answer'] is not None or value['status']=='completed':
raise Conflict('fact request closed')
fact.update(answer=answer,evidence=evidence,answered_at=time.time())
db.execute('UPDATE decision_facts SET data=? WHERE id=?',(json.dumps(fact,ensure_ascii=False),rid))
self._decision_history(db,value,'fact_answered',fact)
self._cancel_decision_notices(db,did,'fact_requested',rid)
self._decision_notice(db,value,'fact_answered','decision','answer:'+rid,request_id=rid,answer=answer,evidence=evidence)
return self._decision_record(db,value,action,data,fact)
def decision_notifications(self):
with self.db() as db:
rows=db.execute('SELECT data FROM decision_workflow_notifications WHERE delivered_at IS NULL AND cancelled_at IS NULL ORDER BY seq LIMIT 100').fetchall()
count=db.execute('SELECT count(*) FROM decision_workflow_notifications WHERE delivered_at IS NULL AND cancelled_at IS NULL').fetchone()[0]
return {'notifications':[json.loads(row['data']) for row in rows],'pending_count':count}
def decision_submitting(self, data):
self._decision_fields(data,{'notification_id'})
nid=identifier(data['notification_id'],'notification_id')
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
row=db.execute('SELECT * FROM decision_workflow_notifications WHERE id=?',(nid,)).fetchone()
if not row:
raise Invalid('unknown notification')
if row['cancelled_at'] is not None:
raise Conflict('notification cancelled')
db.execute('UPDATE decision_workflow_notifications SET submission_started_at=COALESCE(submission_started_at,?) WHERE id=?',(time.time(),nid))
return {'notification_id':nid,'status':'submitting'}
def decision_ack(self, data):
self._decision_fields(data,{'notification_id'})
nid=identifier(data['notification_id'],'notification_id')
with self.db() as db:
db.execute('BEGIN IMMEDIATE')
row=db.execute('SELECT delivered_at FROM decision_workflow_notifications WHERE id=?',(nid,)).fetchone()
if not row:
raise Invalid('unknown notification')
duplicate=row['delivered_at'] is not None
if not duplicate:
db.execute('UPDATE decision_workflow_notifications SET delivered_at=? WHERE id=?',(time.time(),nid))
return {'notification_id':nid,'status':'delivered','duplicate':duplicate}
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['tasks'] = [self._task_defaults(task) for task in result['tasks']]
result['attention'] = self._attention_state(db)
result['decisions'] = [self._decision(row) for row in db.execute('SELECT * FROM decisions ORDER BY id')]
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 AND superseded_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
Reference in New Issue View Git Blame Copy Permalink