"""Persistent explicit attention; opening pages never dismisses it.""" import concurrent.futures import http.client import json from pathlib import Path import tempfile import threading import unittest from unittest.mock import patch from agent_control.store import Store, Invalid, Conflict, ATTENTION_PAGES from agent_control.server import Server def item(**changes): data = dict(id='item-a',page='tasks',required=True,reason='Needs a human check') data.update(changes) return data def hook(**changes): data = dict(source='synthetic',event_id='incident',severity='warning',status='firing', summary='Connection unavailable') data.update(changes) return data class AttentionTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.path = Path(self.tmp.name)/'attention.sqlite3' self.store = Store(self.path) def tearDown(self): self.tmp.cleanup() def history_count(self): with self.store.db() as db: return db.execute('SELECT count(*) FROM attention_history').fetchone()[0] def test_empty_explicit_update_revoke_and_persist(self): self.assertEqual(self.store.state()['attention'],dict(items=[],pages={p:False for p in ATTENTION_PAGES})) with patch('agent_control.store.time.time',return_value=100): created = self.store.attention(item()) self.assertFalse(created['duplicate']) self.assertEqual(created['href'],'/tasks') self.assertEqual(created['updated'],100) self.assertTrue(Store(self.path).state()['attention']['pages']['tasks']) for _ in range(3): self.store.state() self.assertEqual(self.history_count(),1) with patch('agent_control.store.time.time',return_value=200): again = self.store.attention(item()) self.assertTrue(again['duplicate']) self.assertEqual(again['updated'],100) self.assertEqual(self.history_count(),1) self.store.attention(item(reason='Evidence changed',href='/logs?incident=5')) self.assertEqual(self.history_count(),2) self.store.attention(item(required=False,reason='')) self.assertFalse(self.store.state()['attention']['pages']['tasks']) self.assertEqual(self.history_count(),3) self.assertTrue(self.store.attention(item(required=False,reason=''))['duplicate']) self.assertEqual(self.history_count(),3) def test_many_independent_items_any_aggregate(self): self.store.attention(item()) self.store.attention(item(id='item-b')) self.store.attention(item(id='item-c',page='cluster')) self.store.attention(item(required=False,reason='Done')) state = self.store.state()['attention'] self.assertEqual(len(state['items']),2) self.assertTrue(state['pages']['tasks']) self.assertTrue(state['pages']['cluster']) self.store.attention(item(id='item-b',required=False,reason='')) self.assertFalse(self.store.state()['attention']['pages']['tasks']) self.assertTrue(self.store.state()['attention']['pages']['cluster']) self.assertEqual(set(state['items'][0]),{'id','page','required','reason','href','updated'}) def test_subject_cannot_change_and_associations_can_be_omitted_on_withdrawal(self): self.store.task(dict(id='a',name='A')) self.store.task(dict(id='b',name='B')) self.store.attention(item(task_id='a')) for data in [item(task_id='b'),item(page='cluster'),item(task_id='b',required=False,reason='')]: with self.assertRaises(Conflict): self.store.attention(data) self.assertEqual(self.history_count(),1) self.store.attention(item(required=False,reason='')) with self.store.db() as db: saved=json.loads(db.execute("SELECT data FROM attention WHERE id='item-a'").fetchone()[0]) self.assertEqual(saved['task_id'],'a') self.assertFalse(saved['required']) def test_decision_automatic_attention_response_retires_every_associated_item(self): self.store.decision(dict(id='d',title='Choose a plan',question='Which plan?')) state = self.store.state()['attention'] self.assertTrue(state['pages']['decisions']) auto=state['items'][0] self.assertEqual(auto['href'],'/decisions?decision=d') self.store.attention(item(decision_id='d')) self.store.decision_respond('d',dict(option_id=None,comment='Chosen',idempotency_key='reply')) state=self.store.state()['attention'] self.assertEqual(state['items'],[]) self.assertFalse(any(state['pages'].values())) with self.store.db() as db: self.assertTrue(all(not json.loads(r['data'])['required'] for r in db.execute('SELECT data FROM attention'))) self.store.attention(item(decision_id='d',reason='Attempt to revive already answered')) self.assertFalse(self.store.state()['attention']['pages']['tasks']) def test_manual_early_decision_withdrawal_never_recreated_by_retry_read_or_restart(self): definition=dict(id='d',title='Choose',question='What next?') self.store.decision(definition) auto=self.store.state()['attention']['items'][0] self.store.attention(dict(id=auto['id'],page='decisions',required=False,reason='')) count=self.history_count() for _ in range(3): self.store.decision(definition) self.assertFalse(Store(self.path).state()['attention']['pages']['decisions']) self.assertEqual(self.history_count(),count) def test_hook_only_upgrade_marks_attention_and_resolution_clears_related(self): self.store.hook(hook()) self.assertFalse(self.store.state()['attention']['pages']['logs']) escalation=dict(source='synthetic',event_id='incident',reason='Persistent outage') result=self.store.hook_escalate(escalation) state=self.store.state()['attention'] self.assertTrue(state['pages']['logs']) auto=state['items'][0] self.assertEqual(auto['href'],'/logs?incident='+result['notifications'][0]['hook']['log_id']) self.store.attention(item(hook_source='synthetic',hook_event_id='incident')) self.store.hook(hook(status='resolved')) self.assertEqual(self.store.state()['attention']['items'],[]) self.store.hook(hook(event_id='ordinary')) self.store.hook(hook(event_id='ordinary',status='resolved')) self.assertEqual(self.store.state()['attention']['items'],[]) def test_early_upgrade_withdrawal_is_not_recreated(self): self.store.hook(hook()) escalation=dict(source='synthetic',event_id='incident',reason='Persistent outage') self.store.hook_escalate(escalation) auto=self.store.state()['attention']['items'][0] self.store.attention(dict(id=auto['id'],page='logs',required=False,reason='Acknowledged separately')) self.store.hook_escalate(escalation) self.assertFalse(Store(self.path).state()['attention']['pages']['logs']) def test_task_completion_keeps_attention_cancellation_retires(self): self.store.task(dict(id='task',name='Task')) self.store.attention(item(task_id='task')) self.store.task(dict(status='completed',phase='done'),'task') self.assertTrue(self.store.state()['attention']['pages']['tasks']) self.store.task(dict(status='cancelled'),'task') self.assertFalse(self.store.state()['attention']['pages']['tasks']) self.store.attention(item(task_id='task',reason='Try to revive cancelled task')) self.assertFalse(self.store.state()['attention']['pages']['tasks']) def test_schema_paths_and_unknown_targets(self): for data in [item(required=1),item(reason=''),item(reason=None),item(reason='x'*241), item(page='other'),item(thread_id='arbitrary'),item(html='hi'), item(task_id='missing'),item(decision_id='missing'), item(hook_source='synthetic'),item(hook_event_id='incident'), item(hook_source='missing',hook_event_id='incident'), item(id='attention-decision-forged')]: with self.subTest(data=data): with self.assertRaises(Invalid): self.store.attention(data) for href in ['https://example.invalid','//example.invalid/logs',r'/logs\evil','/other', '/logs/../tasks','/logs#x','/logs?next=https://example.invalid', '/logs?x=%3Cscript%3E','/%6cogs','/logs?x=%255Cevil','/logs\n','javascript:alert(1)']: with self.subTest(href=href): with self.assertRaises(Invalid): self.store.attention(item(href=href)) self.assertEqual(self.history_count(),0) self.store.attention(item(href='/logs?incident=123&status=active')) self.assertEqual(self.store.state()['attention']['items'][0]['href'],'/logs?incident=123&status=active') def test_concurrent_duplicate_requests_one_history_record(self): with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: results=list(pool.map(lambda _:self.store.attention(item()),range(16))) self.assertEqual(sum(not r['duplicate'] for r in results),1) self.assertEqual(self.history_count(),1) self.assertEqual(len(self.store.state()['attention']['items']),1) def test_auto_attention_failure_rolls_back_decision_and_upgrade(self): with self.store.db() as db: db.execute("CREATE TRIGGER fail_attention BEFORE INSERT ON attention_history BEGIN SELECT RAISE(ABORT,'synthetic'); END") with self.assertRaises(Exception): self.store.decision(dict(id='d',title='Choose',question='What next?')) self.assertEqual(self.store.state()['decisions'],[]) self.store.hook(hook()) with self.assertRaises(Exception): self.store.hook_escalate(dict(source='synthetic',event_id='incident',reason='Persistent')) self.assertEqual(self.store.logs()['logs'][0]['level'],'warning') self.assertEqual(self.store.hook_notifications()['pending_count'],1) class AttentionHTTPTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() root = Path(self.tmp.name) root.chmod(0o700) token = root / 'token' self.secret = 'h'*48 token.write_text(self.secret) token.chmod(0o600) self.metrics = patch('agent_control.server.MetricsCache') mock = self.metrics.start() mock.return_value.snapshot.return_value = {'status':'synthetic'} self.server = Server(dict(bind='127.0.0.1',port=0,runtime_root=str(root),token_file=str(token))) self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() def tearDown(self): self.server.shutdown() self.server.server_close() self.thread.join() self.metrics.stop() self.tmp.cleanup() def request(self,path,method='GET',payload=None,headers=None): conn = http.client.HTTPConnection(*self.server.server_address,timeout=3) merged = {'Authorization':'Bearer '+self.secret,'X-CSRF-Token':'1','Content-Type':'application/json'} merged.update(headers or {}) conn.request(method,path,json.dumps(payload) if payload is not None else None,merged) response = conn.getresponse() status, body = response.status,json.loads(response.read()) conn.close() return status, body def test_auth_csrf_origin_and_reads_never_clear(self): for headers,status in [({'Authorization':''},401),({'X-CSRF-Token':''},403), ({'Origin':'https://untrusted.example'},403)]: self.assertEqual(self.request('/api/attention','POST',item(),headers)[0],status) self.assertEqual(self.request('/api/attention','POST',item())[0],200) self.assertEqual(self.request('/api/attention','POST',item(thread_id='no'))[0],400) for _ in range(3): self.assertTrue(self.request('/api/state')[1]['attention']['pages']['tasks']) self.assertTrue(self.request('/api/attention','POST',item())[1]['duplicate']) self.assertEqual(self.request('/api/attention','POST',item(page='logs'))[0],409) self.assertEqual(self.request('/api/attention','POST',item(required=False,reason=''))[0],200) self.assertFalse(self.request('/api/state')[1]['attention']['pages']['tasks'])