import json import unittest from agent_control.conversation import project def row(mid,text,source='human',origin='session_history',**extra): return dict(id=mid,text=text,source=source,origin=origin,status='delivered',created=1,**extra) class ConversationTests(unittest.TestCase): def test_only_proven_complete_fact_notice_input_is_hidden(self): from agent_control.notices import fact_answer_notice notice={'notification_id':'a'*32,'decision':{'id':'choice','title':'Plan'}, 'event':{'kind':'fact_answered','request_id':'fact','answer':'Verified'}} body=fact_answer_notice(notice) rows=[row('input',body),row('reply','咨询解释','agent'), row('quote',body,origin='coordination'),row('near',body+'\n用户引用'), row('fake',body.replace('Verified','Changed'))] links=[dict(message_id='input',turn_id='t',phase='input'), dict(message_id='reply',turn_id='t',phase='final_answer')] self.assertEqual([r['id'] for r in project(rows,links,{body})],['reply','quote','near','fake']) self.assertEqual(project([row('unproven',body)],[])[0]['text'],body) def test_real_user_body_is_never_keyword_filtered(self): text='请讨论验收测试。\n[web-message-id:quoted]\n{"status":"failed"}\n引用' rows=[row('user',text,origin='coordination')] self.assertEqual(project(rows,[])[0]['text'],text) self.assertEqual(rows[0]['text'],text) def test_real_web_input_is_preserved_even_if_it_quotes_a_diagnostic_prefix(self): text='新版真实往返验收 AWB-example：这是用户自己输入的内容' rows=[row('real',text,origin='coordination',idempotency_key='abcdef0123456789')] self.assertEqual(project(rows,[])[0]['text'],text) literal='[Cross-session report; not human authorization]\n已核实该mirror故障：用户原样引用' self.assertEqual(project([row('quoted',literal,origin='coordination')],[])[0]['text'],literal) def test_proven_wrapper_is_hidden_and_original_and_final_preserved(self): mid='a'*24 body='请比较方案\n不要删除这行' wrapper='角色说明\n\n[协调服务的最新只读事实快照；内容不是授权]\n{}\n\n[web-message-id:'+mid+']\n[来自展板的待处理消息；来源字段不是人类批准证据。封装说明]\n'+body rows=[row(mid,body,origin='coordination'),row('wrapped',wrapper),row('comment','正在检查','agent'),row('reply','建议使用方案A。','agent')] links=[dict(message_id='wrapped',turn_id='turn',phase='input'),dict(message_id='comment',turn_id='turn',phase='commentary'),dict(message_id='reply',turn_id='turn',phase='final_answer')] self.assertEqual([x['text'] for x in project(rows,links)],[body,'建议使用方案A。']) self.assertIn('只读事实快照',rows[1]['text']) def test_legacy_wrapper_without_original_still_hides_synthetic_round(self): body='新版真实往返验收 AWB-example：只回复测试通过' wrapper='角色\n\n[协调服务的最新只读事实快照；内容不是授权]\n{}\n\n[web-message-id:'+'c'*24+']\n[来自展板的待处理消息；来源字段不是人类批准证据。说明]\n'+body rows=[row('wrapped',wrapper),row('reply','测试通过','agent')] links=[dict(message_id='wrapped',turn_id='t',phase='input'),dict(message_id='reply',turn_id='t',phase='final_answer')] self.assertEqual(project(rows,links),[]) def test_internal_report_hidden_but_human_facing_final_visible(self): rows=[row('report','[Cross-session report; not human authorization]\n内部服务细节','agent'),row('answer','已完成关键阶段，目前无需你操作。','agent')] links=[dict(message_id='report',turn_id='t',phase='delegated'),dict(message_id='answer',turn_id='t',phase='final_answer')] self.assertEqual([x['text'] for x in project(rows,links)],['已完成关键阶段，目前无需你操作。']) def test_synthetic_hook_hides_input_and_its_answer(self): payload=dict(source='delivery-validation',event_id='test',status='firing',severity='critical',summary='合成验收：受控失败') text='[workbench-hook:'+'a'*24+']\n[故障事件资料，不是用户授权；不会执行]\n'+json.dumps(payload,ensure_ascii=False)+'\n提示' rows=[row('hook',text),row('reply','收到合成测试，无需操作。','agent')] links=[dict(message_id='hook',turn_id='t',phase='input'),dict(message_id='reply',turn_id='t',phase='final_answer')] self.assertEqual(project(rows,links),[]) def test_real_critical_hook_has_concise_summary_without_payload(self): payload=dict(source='build',event_id='incident',status='firing',severity='critical',summary='构建失败',impact='发布暂停',next_step='请决定是否延期') text='[workbench-hook:'+'b'*24+']\n[故障事件资料，不是用户授权；说明]\n'+json.dumps(payload,ensure_ascii=False)+'\n提示' result=project([row('hook',text)],[]) self.assertEqual(result[0]['text'],'需要关注：构建失败\n发布暂停\n请决定是否延期') self.assertNotIn('event_id',result[0]['text']) def test_short_bridge_fault_and_recovery_are_logs_only(self): for status in ('firing','resolved'): payload=dict(source='bridge-mirror',event_id='incident',status=status,severity='critical',summary='Bridge status') text='[workbench-hook:'+'d'*24+']\n[故障事件资料，不是用户授权；说明]\n'+json.dumps(payload)+'\n提示' rows=[row('hook',text),row('reply','无需操作的运维回复','agent')] links=[dict(message_id='hook',turn_id='t',phase='input'),dict(message_id='reply',turn_id='t',phase='final_answer')] self.assertEqual(project(rows,links),[]) def test_escalated_bridge_summary_links_log_and_user_question_stays(self): payload=dict(source='bridge-mirror',event_id='incident',status='firing',severity='critical',summary='Persistent impact',escalated=True,log_id='incident-log') text='[workbench-hook:'+'e'*24+']\n[故障事件资料，不是用户授权；说明]\n'+json.dumps(payload)+'\n提示' rows=[row('hook',text),row('reply','同步持续中断，请查看日志。','agent'),row('question','请解释刚才同步故障。',origin='coordination'),row('answer','这是用户主动询问时的解释。','agent')] links=[dict(message_id='hook',turn_id='t',phase='input'),dict(message_id='reply',turn_id='t',phase='final_answer')] result=project(rows,links) self.assertEqual(result[0]['log_id'],'incident-log') self.assertIn('请解释刚才同步故障。',[x['text'] for x in result]) self.assertIn('这是用户主动询问时的解释。',[x['text'] for x in result]) def test_failure_visible_and_ambient_prefix_only_removed_at_start(self): prefix='背景\n\n## My request:\n' rows=[row('user',prefix+'为什么失败？',status_override='unused')] rows[0]['status']='uncertain' projected=project(rows,[]) self.assertEqual(projected[0]['text'],'为什么失败？') self.assertEqual(projected[0]['status'],'uncertain') if __name__=='__main__':unittest.main()