62 lines
4.3 KiB
Python
62 lines
4.3 KiB
Python
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_real_user_body_is_never_keyword_filtered(self):
|
|
text='请讨论验收测试。\n[web-message-id:quoted]\n{"status":"failed"}\n<in-app-browser-context>引用</in-app-browser-context>'
|
|
rows=[row('user',text,origin='coordination')]
|
|
self.assertEqual(project(rows,[])[0]['text'],text)
|
|
self.assertEqual(rows[0]['text'],text)
|
|
|
|
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_failure_visible_and_ambient_prefix_only_removed_at_start(self):
|
|
prefix='<in-app-browser-context source="ambient-ui-state">背景</in-app-browser-context>\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()
|