"""Atomic worker receipts and execution versus acceptance state.""" 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 from agent_control.server import Server def report(**changes): data=dict(report_id='report-1',task_id='task',session_id='worker',stage='execution_completed', current_progress='Implementation tested',summary='Execution finished', evidence=['Synthetic tests passed'],version='v1') data.update(changes) return data class ReportTests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory() self.path=Path(self.tmp.name)/'reports.sqlite3' self.store=Store(self.path) for sid in ['worker','other']: self.store.session(dict(id=sid,name=sid)) self.store.task(dict(id='task',name='Work',session_id='worker',status='running')) self.store.task(dict(id='other-task',name='Other',session_id='other')) def tearDown(self): self.tmp.cleanup() def task(self): return next(t for t in self.store.state()['tasks'] if t['id']=='task') def test_execution_completion_is_waiting_not_business_completion(self): result=self.store.task_report(report()) self.assertFalse(result['duplicate']) task=result['task'] self.assertEqual(task['execution_status'],'completed') self.assertEqual(task['status'],'waiting') self.assertEqual(task['phase'],'integration') self.assertEqual(task['current_progress'],'Implementation tested') self.assertEqual(task['report_evidence'],['Synthetic tests passed']) self.assertEqual(task['last_report_id'],'report-1') self.assertEqual(self.store.task_reports('task')['reports'],[result['report']]) self.assertEqual(self.store.state()['attention']['items'],[]) def test_duplicate_persistent_receipt_returns_current_task_without_reapplying(self): first=self.store.task_report(report()) self.store.task(dict(status='completed',phase='done',current_progress='Accepted'),'task') second=Store(self.path).task_report(report(needs_attention=False)) self.assertTrue(second['duplicate']) self.assertEqual(second['report'],first['report']) self.assertEqual(second['task']['status'],'completed') self.assertEqual(second['task']['current_progress'],'Accepted') self.assertEqual(len(self.store.task_reports('task')['reports']),1) with self.assertRaises(Conflict): self.store.task_report(report(summary='Different')) with self.assertRaises(Conflict): self.store.task_report(report(task_id='other-task',session_id='other')) def test_new_late_report_does_not_roll_back_integrated_or_accepted_task(self): self.store.task_report(report()) for status in ['integrated','verified']: self.store.task(dict(integration_status=status,current_progress='Validated',version='accepted'),'task') before=self.task() result=self.store.task_report(report(report_id='late-'+status,stage='progress', current_progress='Old work',version='old')) self.assertEqual(result['task']['integration_status'],status) self.assertEqual(result['task']['current_progress'],'Validated') self.assertEqual(result['task']['execution_status'],'completed') self.assertEqual(result['task']['version'],'accepted') self.assertEqual(result['task']['progress_updated'],before['progress_updated']) self.store.task(dict(status='completed',phase='done'),'task') result=self.store.task_report(report(report_id='late-failure',stage='failed')) self.assertEqual(result['task']['status'],'completed') self.assertEqual(result['task']['phase'],'done') self.assertEqual(len(self.store.task_reports('task')['reports']),4) def test_concurrent_completion_and_late_progress_preserve_waiting_but_new_blocker_applies(self): with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: bodies=[report()]+[report(report_id='progress-'+str(i),stage='progress', current_progress='Older progress') for i in range(12)] list(pool.map(self.store.task_report,bodies)) task=self.task() self.assertEqual(task['execution_status'],'completed') self.assertEqual(task['status'],'waiting') self.assertEqual(task['phase'],'integration') self.assertEqual(task['current_progress'],'Implementation tested') progress_updated=task['progress_updated'] self.store.task_report(report(report_id='latest-arrival',stage='progress', current_progress='Still old progress')) task=self.task() self.assertEqual(task['current_progress'],'Implementation tested') self.assertEqual(task['progress_updated'],progress_updated) self.assertEqual(len(self.store.task_reports('task')['reports']),14) for stage in ['blocked','failed']: task=self.store.task_report(report(report_id='new-'+stage,stage=stage, current_progress='New issue discovered'))['task'] self.assertEqual(task['status'],stage) self.assertEqual(task['execution_status'],stage) self.assertEqual(task['current_progress'],'New issue discovered') def test_blocked_failed_progress_and_server_managed_defaults(self): self.assertEqual(self.task()['execution_status'],'pending') self.assertIsNone(self.task()['last_report_id']) for stage,expected in [('blocked','blocked'),('failed','failed'),('progress','running')]: task=self.store.task_report(report(report_id=stage,stage=stage))['task'] self.assertEqual(task['execution_status'],expected) self.assertEqual(task['status'],expected) for field,value in [('execution_status','completed'),('last_report_id','fake'), ('last_report_at',123),('report_evidence',['fake'])]: with self.assertRaises(Invalid): self.store.task({field:value},'task') self.store.task(dict(id='manual',name='Manual',status='completed',phase='done')) task=next(t for t in self.store.state()['tasks'] if t['id']=='manual') self.assertEqual(task['execution_status'],'pending') def test_identity_and_schema_reject_without_mutation(self): before=self.task() cases=[dict(session_id='other'),dict(session_id='missing'),dict(task_id='other-task'), dict(stage='completed'),dict(current_progress=None),dict(current_progress='x'*1001), dict(summary='x'*2001),dict(evidence=['x']*21),dict(evidence=['x'*1001]), dict(evidence=[{}]),dict(version=None),dict(needs_attention=1), dict(needs_attention=True,summary=''),dict(source='admin')] for change in cases: with self.subTest(change=change): with self.assertRaises(Invalid): self.store.task_report(report(**change)) self.assertEqual(self.task(),before) self.assertEqual(self.store.task_reports('task')['reports'],[]) def test_attention_only_on_explicit_true_and_same_task_item_updated(self): first=self.store.task_report(report(needs_attention=True)) items=self.store.state()['attention']['items'] self.assertEqual(len(items),1) aid=items[0]['id'] self.store.task_report(report(report_id='r2',stage='blocked',needs_attention=True,summary='Need review')) items=self.store.state()['attention']['items'] self.assertEqual(len(items),1) self.assertEqual(items[0]['id'],aid) self.assertEqual(items[0]['reason'],'Need review') self.store.attention(dict(id=aid,page='tasks',required=False,reason='')) self.store.task_report(report(needs_attention=True)) self.assertEqual(self.store.state()['attention']['items'],[]) self.assertEqual(self.store.hook_notifications()['pending_count'],0) self.assertEqual(self.store.decision_notifications()['pending_count'],0) def test_failure_rolls_back_task_and_report_and_attention(self): before=self.task() with self.store.db() as db: db.execute("CREATE TRIGGER fail_report_attention BEFORE INSERT ON attention_history BEGIN SELECT RAISE(ABORT,'synthetic'); END") with self.assertRaises(Exception): self.store.task_report(report(needs_attention=True)) self.assertEqual(self.task(),before) self.assertEqual(self.store.task_reports('task')['reports'],[]) self.assertEqual(self.store.state()['attention']['items'],[]) def test_concurrent_receipt_only_one_record(self): with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: results=list(pool.map(lambda _:self.store.task_report(report()),range(12))) self.assertEqual(sum(not r['duplicate'] for r in results),1) self.assertEqual(len(self.store.task_reports('task')['reports']),1) def test_recent_reports_are_bounded_and_task_scoped(self): for i in range(53): self.store.task_report(report(report_id='r-'+str(i),stage='progress')) self.store.task_report(report(report_id='other',task_id='other-task',session_id='other')) rows=self.store.task_reports('task')['reports'] self.assertEqual(len(rows),50) self.assertEqual(rows[0]['report_id'],'r-52') self.assertTrue(all(row['task_id']=='task' for row in rows)) self.assertEqual(len(self.store.task_reports('other-task')['reports']),1) with self.assertRaises(Invalid): self.store.task_reports('missing') class ReportHTTPTests(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_query_and_roundtrip(self): self.server.store.session(dict(id='worker',name='Worker')) self.server.store.task(dict(id='task',name='Task',session_id='worker')) path='/api/task-reports' for headers,code in [({'Authorization':''},401),({'X-CSRF-Token':''},403), ({'Origin':'https://untrusted.example'},403)]: self.assertEqual(self.request(path,'POST',report(),headers)[0],code) self.assertEqual(self.request(path+'?task_id=task',headers={'Authorization':''})[0],401) for query in ['', '?task_id=', '?task_id=task&task_id=other', '?session_id=worker']: self.assertEqual(self.request(path+query)[0],400) self.assertEqual(self.request(path,'POST',report())[0],200) self.assertTrue(self.request(path,'POST',report())[1]['duplicate']) self.assertEqual(self.request(path,'POST',report(summary='Changed'))[0],409) self.assertEqual(len(self.request(path+'?task_id=task')[1]['reports']),1)