diff --git a/README.md b/README.md index 0ad73d8..f33f364 100644 --- a/README.md +++ b/README.md @@ -69,3 +69,5 @@ python3 scripts/bridge_service.py --config /path/to/private/bridge.json 完成报告先进入私有持久outbox，有界重试并幂等确认；执行完成与集成验收分别记录。通知要求每目标最多一条未处理，其余保留并合并。当前支持execution_notification_targets显式映射：依据正式任务session_id，将任务相关事件和ETA优先交执行者；未映射事项及系统事件交配置的唯一副主控，决策直达总控。目标变更须保留旧回执并对账迁移，不盲重放积压。宿主原生队列总数与取消能力未知时如实说明，不能声称已清空。配置、实现范围与验收要求见[运行接入](references/operations.md#一层副主控的路由与可靠上报)。 重大事件驱动用户汇报，普通进度、短暂重连和自动恢复留在内部日志。故障钩子提供持久去重、分级升级和恢复记录；额外服务需按授权接入。 + +决策由主控先建卡，提交后进入待检查；接受、落实完成和真实会话归档分别记录。每项真实决策绑定一个独立咨询会话，用有限凭据只维护本决策，实质修订保留旧选择并要求重确认。详情可咨询和查看版本历史；真实会话接通及宿主归档须现场验证，不能用占位记录代替。参见[独立决策会话接入](references/decision-sessions.md)。 diff --git a/SKILL.md b/SKILL.md index bd93f17..3421d8b 100644 --- a/SKILL.md +++ b/SKILL.md @@ -112,7 +112,7 @@ description: 组织项目的多个 AI 会话并行工作，将进度、证据和 需要人类决定的事项集中登记在决策页，以卡片显示问题、影响、优先级和当前状态；点击查看背景、方案取舍、推荐及依据。仅在有实际依据时登记期限，不制造紧迫感。有明确候选方案时提供清晰可点击选项及补充意见；推荐不预选，选择不等于提交，只有明确提交后保存实际选择。 -沟通页只给简短提醒并链接具体卡片，不重复堆完整方案或内部载荷。提交使用持久幂等键，事务保存响应并排入总控投递队列；重试不得重复执行或串到另一决策/项目。标记“已提交/待核实/已决定”与运维授权、方案执行和业务验收分开。涉及权限的选择进入待核实，不能仅凭网页令牌或来源字段当成人类授权；由总控通过受支持会话与既定边界核实。 +沟通页只给简短提醒并链接具体卡片，不重复堆完整方案或内部载荷。提交使用持久幂等键，事务保存响应并排入总控投递队列；重试不得重复执行或串到另一决策/项目。状态区分待选择、待主控检查、已接受待落实、需重新考虑与完成；用户提交不直接结束决策，与运维授权、方案执行和业务验收分开。涉及权限的选择进入待核实，不能仅凭网页令牌或来源字段当成人类授权；由总控通过受支持会话与既定边界核实。 空状态不放模拟待决定事项。验收使用隔离临时实例，不把合成问题、选项或通知投进用户正式入口。检查详情展开、选项与备注、显式提交、失败保留草稿、已决状态、并发/重试/目标隔离及真实总控队列。 @@ -157,3 +157,15 @@ description: 组织项目的多个 AI 会话并行工作，将进度、证据和 任务分支视图以版本交付为主线、独立工作任务为支线，节点只表达实际登记的关键里程碑，不冒充Git提交或提交历史。交付任务显式使用kind=delivery，工作任务使用kind=work及delivery_id归属；没有归属显示未归属，不从version中的提交哈希推测。只有真实集成关系且源、目标均通过集成验收才绘制汇合。 默认显示所有未结束任务及递归必要前置，显示全部开关用于历史。悬停、键盘焦点及触屏选择突出递归上下游关系；无关项淡化但可读，退出交互恢复，不修改任务事实或自动开启全部。字段校验和迁移见references/operations.md，交互与验收见references/design-system.md。 + +## 决策修订、咨询与收尾 + +决策只来自真实需求，卡片包含简介及关联任务。用户可选择方案或自由填写意见，至少一项有效；提交保留当时版本与选项快照，进入pending_review。主控核对真实提交的影响和执行计划后接受为accepted或退回needs_reconsideration；接受不等于系统操作授权。内容实质变化保留旧快照并要求按新revision重新确认，旧记录不能静默覆盖。关联新决策显式登记来源，不自动编造。 + +独立咨询会话的创建、有限凭据交付、桥接与真实归档步骤见 [决策会话接入](references/decision-sessions.md)。 + +每项决策的独立咨询绑定真实已注册会话，未接通如实显示。咨询agent只获该决策的有限凭据，可读取历史、维护内容、请求主控核实事实；不能提交用户选择、接受、完成、归档、换绑或操作其他任务。凭据仅私有交付，不能交普通面板令牌。角色提示禁止代码、系统操作及执行决策，不声称OS级硬隔离。事实回复不是用户批准。 + +主控检查后记录结论、交接及证据才完成；完成后由获授权总控调用宿主真实归档工具，再保存回执，服务记录不冒充归档操作。隔离测试不能代替真实会话绑定与往返，尚未验证必须明示；没有真实待决定事项时不得为验收创建生产决策或会话。 + +已有专门人类入口时，主控、副主控和执行者不得直接向用户提问；问题、待选择、待授权及关键结论定向交唯一入口，由入口理解并与人类交互，再整理给主控，不经过底层事件队列。需选项先建真实决策卡，入口简短链接。决策页内嵌咨询只回答该决策范围问题，不另设一般项目入口；人类决定和审批仍走决策卡与唯一入口。 diff --git a/agent_control/server.py b/agent_control/server.py index 22110e7..4eafa48 100644 --- a/agent_control/server.py +++ b/agent_control/server.py @@ -198,10 +198,23 @@ class Handler(BaseHTTPRequestHandler): return True def auth(self): + self.decision_scope = None supplied = self.headers.get('Authorization', '') - if not hmac.compare_digest(supplied.encode('utf-8'), ('Bearer ' + self.server.token).encode('utf-8')): + if hmac.compare_digest(supplied.encode('utf-8'), ('Bearer ' + self.server.token).encode('utf-8')): + return True + scope = self.server.store.decision_credential_scope(supplied[7:]) if supplied.startswith('Bearer ') else None + if scope is None: self.output(401, {'error':'authentication required'}) return False + path = urlsplit(self.path).path + base = '/api/decisions/' + scope + allowed = ((self.command=='GET' and path in (base,base+'/history',base+'/consultation')) + or (self.command=='PATCH' and path==base) + or (self.command=='POST' and path==base+'/facts')) + if not allowed or (self.command!='GET' and self.server.store.decision_get(scope)['status']=='completed'): + self.output(403, {'error':'credential scope rejected'}) + return False + self.decision_scope = scope return True def do_GET(self): @@ -219,6 +232,12 @@ class Handler(BaseHTTPRequestHandler): state['metrics'] = self.server.metrics.snapshot() state['eta_clock'] = dict(self.server.eta_clock.status) self.output(200, state) + elif parts.path == '/api/decision-bridge-bindings' and not parts.query: + self.output(200,self.server.store.decision_bindings()) + elif re.fullmatch(r'/api/decisions/[A-Za-z0-9_.:-]+(?:/(?:history|consultation))?',parts.path) and not parts.query: + segments=parts.path.split('/') + method = {'history':self.server.store.decision_history,'consultation':self.server.store.decision_consultation}.get(segments[-1]) if len(segments)==5 else self.server.store.decision_get + self.output(200,method(segments[3])) elif parts.path == '/api/queue-health' and not parts.query: self.output(200,self.server.store.queue_health()) elif parts.path == '/api/task-reports': @@ -246,6 +265,8 @@ class Handler(BaseHTTPRequestHandler): self.output(200, {'messages':reader(query['session_id'][0])}) else: self.output(404, {'error':'not found'}) + except Conflict: + self.output(409, {'error':'state conflict'}) except (Invalid,ValueError): self.output(400, {'error':'invalid request'}) except Exception: @@ -295,8 +316,13 @@ class Handler(BaseHTTPRequestHandler): data = json.loads(self.rfile.read(length)) if not isinstance(data, dict): raise Invalid('object required') + if getattr(self,'decision_scope',None) and self.command=='PATCH': + allowed={'expected_revision','reason','title','summary','question','details','impact','recommendation','options','deadline','priority'} + if data.keys()-allowed: + self.output(403, {'error':'credential content scope rejected'}) + return store = self.server.store - routes = {'/api/task-reports':store.task_report, '/api/attention':store.attention, '/api/eta-notifications/ack':store.eta_ack, '/api/decisions':store.decision, '/api/decision-notifications/ack':store.decision_ack, + routes = {'/api/decision-notifications/submitting':store.decision_submitting, '/api/task-reports':store.task_report, '/api/attention':store.attention, '/api/eta-notifications/ack':store.eta_ack, '/api/decisions':store.decision, '/api/decision-notifications/ack':store.decision_ack, '/api/hook-notifications/submitting':store.hook_submitting, '/api/hook-notifications/coalesce':store.hook_coalesce, '/api/hooks':store.hook, '/api/hooks/escalate':store.hook_escalate, '/api/hooks/ack':store.hook_ack, '/api/sessions':store.session, '/api/session-requests':store.request, @@ -310,8 +336,18 @@ class Handler(BaseHTTPRequestHandler): result = store.integrate(self.path.split('/')[3],data) elif self.command == 'POST' and re.fullmatch(r'/api/eta-reviews/[A-Za-z0-9_.:-]+/complete',self.path): result = store.eta_complete(self.path.split('/')[3],data) - elif self.command == 'POST' and re.fullmatch(r'/api/decisions/[A-Za-z0-9_.:-]+/respond',self.path): - result = store.decision_respond(self.path.split('/')[3],data) + elif self.command == 'PATCH' and re.fullmatch(r'/api/decisions/[A-Za-z0-9_.:-]+',self.path): + result = store.decision_patch(self.path.split('/')[3],data) + elif self.command == 'POST' and re.fullmatch(r'/api/decisions/[A-Za-z0-9_.:-]+/facts/[A-Za-z0-9_.:-]+/answer',self.path): + segments=self.path.split('/') + result = store.decision_fact_answer(segments[3],segments[5],data) + elif self.command == 'POST' and re.fullmatch(r'/api/decisions/[A-Za-z0-9_.:-]+/(respond|review|complete|archive-receipt|bind-session|consultation|consultation-credential|facts)',self.path): + actions = {'respond':store.decision_respond,'review':store.decision_review,'complete':store.decision_complete, + 'archive-receipt':store.decision_archive,'bind-session':store.decision_bind, + 'consultation':store.decision_consultation,'consultation-credential':store.decision_credential,'facts':store.decision_fact} + if self.path.endswith('/bind-session') and data.get('session_id')==self.server.primary_session_id: + raise Conflict('primary session cannot be bound') + result = actions[self.path.split('/')[-1]](self.path.split('/')[3],data) elif self.command == 'POST' and self.path in routes: result = routes[self.path](data) else: diff --git a/agent_control/store.py b/agent_control/store.py index 6565d35..16bc5b4 100644 --- a/agent_control/store.py +++ b/agent_control/store.py @@ -128,6 +128,37 @@ class Store: 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'] @@ -184,9 +215,19 @@ class Store: 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 @@ -301,8 +342,8 @@ class Store: if not item['required']: continue if item.get('decision_id'): - decision = db.execute('SELECT response FROM decisions WHERE id=?',(item['decision_id'],)).fetchone() - if not decision or decision['response'] is not None: + 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() @@ -865,6 +906,10 @@ class Store: 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): @@ -1471,9 +1516,10 @@ class Store: 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_notifications',"json_extract(data,'$.response.created')", - 'delivered_at IS NULL') - decisions['cancelled_count'] = 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() @@ -1504,22 +1550,103 @@ class Store: return {'notification_id':notification_id,'status':'delivered','duplicate':already} - @staticmethod - def _decision(row): + 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 - result['response'] = response - result['status'] = ('pending_verification' if result['requires_authorization'] else 'decided') if response else 'open' + 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 decision(self, data): - fields = {'id','title','question','impact','priority','details','recommendation', - 'options','deadline','requires_authorization','task_id'} + 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)]: + ('details',8000),('recommendation',4000),('summary',1000)]: raw = data.get(name, '') if not isinstance(raw, str): raise Invalid('invalid ' + name) @@ -1558,83 +1685,345 @@ class Store: 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 json.loads(row['data']) != value: + if self._normalize_decision(json.loads(row['data'])) != value: raise Conflict('decision definition differs') return self._decision(row) - if task is not None: - self._get(db,'tasks',task) - db.execute('INSERT INTO decisions(id,data) VALUES(?,?)', - (value['id'],json.dumps(value,ensure_ascii=False))) + 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 dict(value,status='open',response=None) + '/decisions?decision='+quote(value['id'],safe=''),time.time(),decision_id=value['id']) + return value - def decision_respond(self, decision_id, data): - decision_id = identifier(decision_id,'decision_id') - if not isinstance(data,dict) or set(data) != {'option_id','comment','idempotency_key'}: - raise Invalid('invalid response fields') + 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') - if not isinstance(data['comment'],str): - raise Invalid('invalid comment') comment = string(data['comment'],'comment',8000,optional=True) - key = identifier(data['idempotency_key'],'idempotency_key') + 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') - row = db.execute('SELECT * FROM decisions WHERE id=?',(decision_id,)).fetchone() - if not row: - raise Invalid('unknown decision') - value = self._decision(row) - if row['response']: - old = value['response'] - if row['response_key'] == key and old['option_id'] == option and old['comment'] == comment: - return value - raise Conflict('decision already answered') - valid_options = {o['id'] for o in value['options']} - if (valid_options and option not in valid_options) or (not valid_options and option is not None): + 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') - if option is None and not comment.strip(): - raise Invalid('response required') - response = {'id':secrets.token_hex(16),'option_id':option,'comment':comment, - 'created':time.time(),'authorization':False} - value['response'] = response - value['status'] = 'pending_verification' if value['requires_authorization'] else 'decided' - db.execute('UPDATE decisions SET response=?,response_key=? WHERE id=?', - (json.dumps(response,ensure_ascii=False),key,decision_id)) - db.execute('INSERT INTO decision_notifications(id,decision_id,data) VALUES(?,?,?)', - (secrets.token_hex(16),decision_id,json.dumps(value,ensure_ascii=False))) - self._attention_revoke(db,{'decision_id':decision_id},response['created']) + 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 value + 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']) + 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))) + 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: - db.execute('BEGIN') - count = db.execute('SELECT count(*) FROM decision_notifications WHERE delivered_at IS NULL').fetchone()[0] - rows = db.execute('SELECT * FROM decision_notifications WHERE delivered_at IS NULL ORDER BY seq LIMIT 100').fetchall() - return {'notifications':[{'notification_id':row['id'],'target':'control', - 'decision':json.loads(row['data'])} for row in rows], - 'pending_count':count} + 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_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') + 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 delivered_at FROM decision_notifications WHERE id=?',(nid,)).fetchone() + row=db.execute('SELECT * FROM decision_workflow_notifications WHERE id=?',(nid,)).fetchone() if not row: raise Invalid('unknown notification') - duplicate = row['delivered_at'] is not None + 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_notifications SET delivered_at=? WHERE id=?',(time.time(),nid)) + 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): diff --git a/assets/dashboard/src/components/decision-chat.tsx b/assets/dashboard/src/components/decision-chat.tsx new file mode 100644 index 0000000..73d2fe0 --- /dev/null +++ b/assets/dashboard/src/components/decision-chat.tsx @@ -0,0 +1,43 @@ +import { useEffect, useRef, useState } from 'react' +import { Button } from '@/components/ui/button' +import { Textarea } from '@/components/ui/textarea' +import { request, errorText, messageKey } from '@/lib/api' +import type { Decision, Message } from '@/lib/types' +import type { ChatDraft } from '@/pages/decision-ui' +export default function DecisionChat({ decision, token, drafts }: { decision: Decision; token: string; drafts: Map }) { + const id = decision.id, bound = Boolean(decision.session_id && decision.session_thread_id) + const readOnly = decision.status === 'completed' || decision.archive_status === 'archived' + const [messages, setMessages] = useState([]), [error, setError] = useState(''), [loaded, setLoaded] = useState(false), [, redraw] = useState(0) + const refresh = useRef<() => Promise>(async () => {}) + const scroll = useRef(null) + if (!drafts.has(id)) drafts.set(id, { text: '', sending: false, error: '' }) + const draft = drafts.get(id)! + useEffect(() => { + if (!bound) return + const controller = new AbortController(); let busy = false + const poll = async () => { + if (busy || controller.signal.aborted) return + busy = true + try { const result = await request<{ messages: Message[] }>(token, `/api/decisions/${encodeURIComponent(id)}/consultation`, undefined, controller.signal); if (!controller.signal.aborted) { const box = scroll.current; const bottom = !box || box.scrollHeight - box.scrollTop - box.clientHeight < 120; setMessages(result.messages); setLoaded(true); setError(''); if (bottom) requestAnimationFrame(() => { if (scroll.current) scroll.current.scrollTop = scroll.current.scrollHeight }) } } + catch (cause) { if (!controller.signal.aborted) setError(errorText(cause)) } + finally { busy = false } + } + refresh.current = poll; void poll(); const timer = setInterval(() => void poll(), 3000) + return () => { clearInterval(timer); controller.abort() } + }, [id, token, bound]) + async function send() { + if (!bound || readOnly || draft.sending || !draft.text.trim()) return + const text = draft.text + if (!draft.attempt || draft.attempt.text !== text) draft.attempt = { text, key: messageKey() } + draft.sending = true; draft.error = ''; redraw(n => n + 1) + try { await request(token, `/api/decisions/${encodeURIComponent(id)}/consultation`, { text, client_key: draft.attempt.key }); if (draft.text === text) draft.text = ''; draft.attempt = undefined; await refresh.current() } + catch (cause) { draft.error = `${errorText(cause)} 草稿已保留，可重试同一消息。` } + finally { draft.sending = false; redraw(n => n + 1) } + } + return

本决策咨询

+

{!bound ? '咨询会话待接通，暂不能发送。' : readOnly ? '本决策已完成，咨询历史只读。' : '消息仅发送到本决策绑定的咨询会话。绑定不代表回复链路已验收。'}

+
{messages.map(message =>

{message.source === 'human' ? '你' : '咨询助手'}

{message.text}

{message.source === 'human' && ['queued','claimed'].includes(message.status) &&

等待回复…

}{['failed','uncertain'].includes(message.status) &&

回复尚未确认，请勿重复发送。

}
)}{!messages.length &&

{!bound ? '暂无已接通的咨询会话。' : loaded ? '暂无咨询消息。' : '正在读取咨询历史…'}

}
+ {(error || draft.error) &&

{draft.error || error}

} + {!readOnly &&
{ event.preventDefault(); void send() }}>