"""Service-owned queue diagnostics; no claims about a host private queue.""" import http.client import concurrent.futures 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 from agent_control.server import Server def firing(**changes): data=dict(source='synthetic',event_id='event',severity='warning',status='firing',summary='Synthetic failure') data.update(changes) return data class QueueTests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory() self.store=Store(Path(self.tmp.name)/'queue.sqlite3') def tearDown(self): self.tmp.cleanup() def test_empty_real_counts_and_explicit_scope(self): health=self.store.queue_health(100) self.assertEqual(health['scope'],'service_owned_outboxes') self.assertEqual(health['delivery_semantics'],'host_accepted_not_model_completed') for queue in ['hooks','eta','decisions']: self.assertEqual(health[queue]['pending_count'],0) self.assertIsNone(health[queue]['oldest_pending_at']) self.assertEqual(health[queue]['created_last_60s'],0) self.assertEqual(health[queue]['delivered_last_60s'],0) def test_sql_counts_timestamps_and_receipts_are_real(self): with patch('agent_control.store.time.time',return_value=10): first=self.store.hook(firing()) self.store.session(dict(id='worker',name='Worker')) self.store.task(dict(id='task',name='Task',session_id='worker', eta_min_minutes=1,eta_max_minutes=2,eta_basis='Measured')) self.store.decision(dict(id='decision',title='Choose',question='Which approach?')) self.store.eta_tick(70) with patch('agent_control.store.time.time',return_value=80): self.store.hook(firing(event_id='recent',severity='critical')) self.store.hook_ack(dict(notification_id=first['notifications'][0]['notification_id'])) self.store.decision_respond('decision',dict(option_id=None,comment='Selected',idempotency_key='choice')) self.store.task_report(dict(report_id='receipt',task_id='task',session_id='worker', stage='progress',current_progress='Verified',summary='Progress',evidence=[])) health=self.store.queue_health(100) self.assertEqual(health['hooks']['pending_count'],2) self.assertEqual(health['hooks']['oldest_pending_at'],80) self.assertEqual(health['hooks']['created_last_60s'],2) self.assertEqual(health['hooks']['delivered_last_60s'],1) self.assertEqual(health['eta']['pending_count'],1) self.assertEqual(health['eta']['oldest_pending_at'],70) self.assertEqual(health['decisions']['pending_count'],1) self.assertEqual(health['decisions']['oldest_pending_at'],80) self.assertEqual(health['task_reports'],dict(received_total=1,received_last_60s=1)) with patch('agent_control.store.time.time',return_value=101): self.store.task(dict(status='cancelled'),'task') self.assertEqual(self.store.queue_health(102)['eta']['cancelled_count'],1) self.assertEqual(self.store.queue_health(102)['eta']['pending_count'],0) def test_coalesce_resolved_backlog_idempotent_without_delivery_claim(self): with patch('agent_control.store.time.time',return_value=10): first=self.store.hook(firing()) for _ in range(10): self.store.hook(firing()) with patch('agent_control.store.time.time',return_value=20): self.store.hook(firing(status='resolved')) before=self.store.state()['hooks']['recent'] with patch('agent_control.store.time.time',return_value=30): result=self.store.hook_coalesce({}) self.assertEqual(result['superseded_count'],2) self.assertEqual(self.store.hook_notifications()['pending_count'],0) self.assertEqual(self.store.hook_coalesce({})['superseded_count'],0) self.assertEqual(self.store.state()['hooks']['recent'],before) self.assertEqual(self.store.queue_health(40)['hooks']['delivered_last_60s'],0) self.assertEqual(self.store.queue_health(40)['hooks']['superseded_count'],2) repeated=self.store.hook(firing()) self.assertEqual(repeated['notifications'][0]['status'],'superseded') self.assertEqual(len(self.store.state()['events']),2) def test_coalesce_keeps_active_major_and_recovery_for_delivered_target(self): warning=self.store.hook(firing()) self.store.hook_ack(dict(notification_id=warning['notifications'][0]['notification_id'])) self.store.hook_escalate(dict(source='synthetic',event_id='event',reason='Persistent')) self.store.hook(firing(status='resolved')) active=self.store.hook(firing(event_id='active-major',severity='critical')) self.store.hook_coalesce({}) notices=self.store.hook_notifications()['notifications'] recoveries=[n for n in notices if n['hook']['event_id']=='event'] self.assertEqual(len(recoveries),1) self.assertEqual(recoveries[0]['target'],'control') self.assertEqual(recoveries[0]['hook']['status'],'resolved') self.assertEqual(sum(n['hook']['event_id']=='active-major' for n in notices),2) def test_coalesce_does_not_touch_user_decision_or_message(self): self.store.session(dict(id='s',name='Synthetic')) self.store.decision(dict(id='d',title='Choose',question='Which approach?')) self.store.decision_respond('d',dict(option_id=None,comment='User choice',idempotency_key='choice')) before=self.store.decision_notifications() self.store.send(dict(session_id='s',text='Real user request',idempotency_key='user-message')) messages=self.store.messages('s') self.store.hook(firing()) self.store.hook(firing(status='resolved')) self.store.hook_coalesce({}) self.assertEqual(self.store.decision_notifications(),before) self.assertEqual(self.store.messages('s'),messages) with self.assertRaises(Invalid): self.store.hook_coalesce({'clear_all':True}) def test_unknown_submission_marker_protects_firing_and_its_recovery(self): first=self.store.hook(firing()) nid=first['notifications'][0]['notification_id'] with patch('agent_control.store.time.time',return_value=10): marker=self.store.hook_submitting(dict(notification_id=nid)) with patch('agent_control.store.time.time',return_value=20): again=self.store.hook_submitting(dict(notification_id=nid)) self.assertFalse(marker['duplicate']) self.assertTrue(again['duplicate']) self.assertEqual(again['submission_started_at'],10) self.assertEqual(again['status'],'submitting') self.store.hook(firing(status='resolved')) self.assertEqual(self.store.hook_coalesce({})['superseded_count'],0) self.assertEqual(self.store.hook_notifications()['pending_count'],2) health=self.store.queue_health(30)['hooks'] self.assertEqual(health['submission_started_unconfirmed_count'],1) self.assertEqual(health['delivered_last_60s'],0) self.assertEqual(self.store.hook_notifications()['notifications'][0]['submission_started_at'],10) def test_superseded_cannot_begin_submission_but_real_late_ack_is_recorded(self): first=self.store.hook(firing()) nid=first['notifications'][0]['notification_id'] self.store.hook(firing(status='resolved')) self.store.hook_coalesce({}) with self.assertRaises(Conflict): self.store.hook_submitting(dict(notification_id=nid)) # An actual accepted receipt can arrive late; never erase that fact. self.store.hook_ack(dict(notification_id=nid)) self.assertEqual(self.store.hook(firing())['notifications'][0]['status'],'delivered') with self.assertRaises(Invalid): self.store.hook_submitting(dict(notification_id='missing')) with self.assertRaises(Invalid): self.store.hook_submitting(dict(notification_id=nid,accepted=True)) def test_submission_and_coalesce_are_serialized(self): notice=self.store.hook(firing())['notifications'][0] self.store.hook(firing(status='resolved')) def submit(): try: return self.store.hook_submitting(dict(notification_id=notice['notification_id'])) except Conflict: return None with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: attempt=pool.submit(submit) cleanup=pool.submit(self.store.hook_coalesce,{}) result=attempt.result() cleanup.result() with self.store.db() as db: row=db.execute('SELECT * FROM hook_notifications WHERE id=?',(notice['notification_id'],)).fetchone() if result is not None: self.assertIsNotNone(row['submission_started_at']) self.assertIsNone(row['superseded_at']) else: self.assertIsNone(row['submission_started_at']) self.assertIsNotNone(row['superseded_at']) class QueueHTTPTests(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_security_and_coalesce_routes(self): first=self.server.store.hook(firing()) self.server.store.hook(firing(status='resolved')) self.assertEqual(self.request('/api/queue-health',headers={'Authorization':''})[0],401) self.assertEqual(self.request('/api/queue-health?x=1')[0],404) for path,body in [('/api/hook-notifications/coalesce',{}), ('/api/hook-notifications/submitting',dict(notification_id=first['notifications'][0]['notification_id']))]: for headers,status in [({'Authorization':''},401),({'X-CSRF-Token':''},403), ({'Origin':'https://untrusted.example'},403)]: self.assertEqual(self.request(path,'POST',body,headers)[0],status) self.assertEqual(self.request('/api/hook-notifications/coalesce','POST',{'clear':True})[0],400) self.assertEqual(self.request('/api/hook-notifications/coalesce','POST',{})[1]['superseded_count'],2) self.assertEqual(self.request('/api/queue-health')[1]['hooks']['pending_count'],0) self.assertEqual(self.request('/api/queue-health')[1]['hooks']['superseded_count'],2)