Files
agent-workbench/tests/test_decision_workflow.py
T

305 lines
21 KiB
Python

"""Decision lifecycle contract tests use isolated stores and HTTP servers only."""
import concurrent.futures
import hashlib
import json
import sqlite3
from pathlib import Path
import tempfile
import unittest
from agent_control.store import Store, Invalid, Conflict
import test_decisions as fixtures
from test_decisions import definition, answer
THREAD='00000000-0000-4000-8000-000000000001'
def review(**changes):
data=dict(expected_revision=1,idempotency_key='review',outcome='accept',reason='Checked facts',impact='Limited scope',execution_plan='Owner verifies execution separately')
data.update(changes)
return data
def completion(**changes):
data=dict(expected_revision=1,idempotency_key='complete',conclusion='Reviewed outcome',handoff='Owner retains implementation task',evidence=['Synthetic verification'])
data.update(changes)
return data
def bind(store,did='choice',sid='consult',thread=THREAD):
store.session(dict(id=sid,name='Consultation',role='decision',wp=did,thread_id=thread,worktree='/synthetic'))
return store.decision_bind(did,dict(expected_revision=1,idempotency_key='bind',session_id=sid,thread_id=thread))
class WorkflowTests(unittest.TestCase):
def setUp(self):
self.tmp=tempfile.TemporaryDirectory()
self.path=Path(self.tmp.name)/'state.sqlite3'
self.store=Store(self.path)
self.store.decision(definition())
def tearDown(self):
self.tmp.cleanup()
def test_accept_complete_archive_requires_review_and_real_receipt(self):
bind(self.store)
with self.assertRaises(Conflict): self.store.decision_complete('choice',completion())
with self.assertRaises(Conflict): self.store.decision_review('choice',review())
self.store.decision_respond('choice',answer())
accepted=self.store.decision_review('choice',review())
self.assertEqual(accepted['status'],'accepted')
self.assertFalse(accepted['response']['authorization'])
self.assertEqual(accepted,self.store.decision_review('choice',review()))
completed=self.store.decision_complete('choice',completion())
self.assertEqual((completed['status'],completed['archive_status']),('completed','pending'))
receipt=dict(thread_id=THREAD,idempotency_key='archive',receipt='Host tool receipt synthetic')
with self.assertRaises(Conflict): self.store.decision_archive('choice',dict(receipt,thread_id='other'))
archived=self.store.decision_archive('choice',receipt)
self.assertEqual(archived['archive_status'],'archived')
self.assertEqual(archived,Store(self.path).decision_archive('choice',receipt))
with self.assertRaises(Conflict): self.store.decision_archive('choice',dict(receipt,receipt='changed'))
kinds=[x['kind'] for x in self.store.decision_history('choice')['history']]
self.assertEqual(kinds,['version','binding','submission','review','completion','archive'])
def test_reconsider_and_free_text_with_options(self):
bind(self.store)
submitted=self.store.decision_respond('choice',answer(option_id=None,comment='Alternative approach'))
self.assertEqual(submitted['status'],'pending_review')
reconsidered=self.store.decision_review('choice',review(outcome='reconsider'))
self.assertEqual(reconsidered['status'],'needs_reconsideration')
self.assertEqual(self.store.decision_complete('choice',completion())['status'],'completed')
with self.assertRaises(Conflict): self.store.decision_respond('choice',answer(idempotency_key='again'))
def test_complete_without_bound_session_cannot_request_archive(self):
self.store.decision_respond('choice',answer())
self.store.decision_review('choice',review())
with self.assertRaises(Conflict): self.store.decision_complete('choice',completion())
self.assertEqual(self.store.decision_get('choice')['status'],'accepted')
self.assertFalse(any(n['event']['kind']=='archive_required' for n in self.store.decision_notifications()['notifications']))
def test_versions_preserve_submission_and_original_create_definition(self):
submitted=self.store.decision_respond('choice',answer())
value=self.store.decision_patch('choice',dict(expected_revision=1,reason='New facts',options=[dict(id='b',label='B',description='Different')]))
self.assertEqual((value['revision'],value['status']),(2,'needs_reconsideration'))
self.assertEqual(value['response'],submitted['response'])
self.assertEqual(value['response']['snapshot']['options'][0]['id'],'a')
self.assertEqual(self.store.decision(definition()),value)
with self.assertRaises(Conflict): self.store.decision_review('choice',review(expected_revision=2))
with self.assertRaises(Conflict): self.store.decision_respond('choice',answer(idempotency_key='new'))
with self.assertRaises(Invalid): self.store.decision_respond('choice',answer(expected_revision=2,idempotency_key='new'))
fresh=self.store.decision_respond('choice',answer(expected_revision=2,option_id='b',idempotency_key='new'))
self.assertEqual(fresh['response']['revision'],2)
history=self.store.decision_history('choice')['history']
self.assertEqual(len([e for e in history if e['kind']=='submission']),2)
self.assertEqual(history[1]['data']['snapshot']['options'][0]['id'],'a')
def test_reconsideration_attention_tracks_status_without_losing_response(self):
self.assertTrue(self.store.state()['attention']['pages']['decisions'])
first=self.store.decision_respond('choice',answer())
self.assertFalse(self.store.state()['attention']['pages']['decisions'])
value=self.store.decision_review('choice',review(outcome='reconsider'))
self.assertEqual(value['response'],first['response'])
self.assertTrue(self.store.state()['attention']['pages']['decisions'])
self.store.decision_respond('choice',answer(idempotency_key='again'))
self.assertFalse(self.store.state()['attention']['pages']['decisions'])
value=self.store.decision_patch('choice',dict(expected_revision=1,reason='Material new facts',question='New question?'))
self.assertIsNotNone(value['response'])
self.assertTrue(self.store.state()['attention']['pages']['decisions'])
self.store.decision_respond('choice',answer(expected_revision=2,idempotency_key='new-revision'))
self.assertFalse(self.store.state()['attention']['pages']['decisions'])
def test_concurrent_patch_has_single_winner(self):
def patch(index):
try: return self.store.decision_patch('choice',dict(expected_revision=1,reason='Evidence',question='Q'+str(index)))
except Conflict: return None
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
results=list(pool.map(patch,range(12)))
self.assertEqual(sum(r is not None for r in results),1)
self.assertEqual(self.store.decision_get('choice')['revision'],2)
for bad in [True,0,'2',None]:
with self.assertRaises(Invalid): self.store.decision_patch('choice',dict(expected_revision=bad,reason='why',title='new'))
with self.assertRaises(Invalid): self.store.decision_patch('choice',dict(expected_revision=2,reason='why',response={}))
def test_related_decisions_are_explicit_and_tasks_validated(self):
with self.assertRaises(Invalid): self.store.decision(definition(id='bad',supersedes_id='bad'))
with self.assertRaises(Invalid): self.store.decision(definition(id='bad',supersedes_id='missing'))
child=self.store.decision(definition(id='next',supersedes_id='choice',summary='Follow up'))
self.assertEqual(child['related_decision_ids'],['choice'])
self.assertEqual(self.store.decision_get('choice')['related_decision_ids'],['next'])
with self.assertRaises(Invalid): self.store.decision_patch('choice',dict(expected_revision=1,reason='why',task_id='missing'))
self.assertEqual(len(self.store.state()['decisions']),2)
def test_bind_isolation_and_identity_is_immutable(self):
with self.assertRaises(Conflict): self.store.decision_consultation('choice')
self.store.session(dict(id='worker',name='Executor',thread_id=THREAD))
self.store.session(dict(id='consult',name='Consult',role='decision',wp='choice',thread_id=THREAD))
data=dict(expected_revision=1,idempotency_key='bind',session_id='consult',thread_id=THREAD)
with self.assertRaises(Conflict): self.store.decision_bind('choice',data)
self.store.session(dict(id='worker',name='Executor',thread_id='00000000-0000-4000-8000-000000000002'))
bound=self.store.decision_bind('choice',data)
self.assertEqual(self.store.decision_bind('choice',data),bound)
with self.assertRaises(Conflict): self.store.decision_bind('choice',dict(data,idempotency_key='other'))
with self.assertRaises(Conflict): self.store.session(dict(id='consult',name='Mutated',role='worker',wp='choice',thread_id=THREAD))
with self.assertRaises(Conflict): self.store.session(dict(id='other',name='Reuse',thread_id=THREAD))
self.assertEqual(self.store.decision_bindings()['bindings'][0]['thread_id'],THREAD)
def test_completion_blocks_pending_inputs_and_consultation_is_readonly_after(self):
bind(self.store)
msg=self.store.decision_consultation('choice',dict(text='Question',client_key='message'))
self.assertEqual(msg,self.store.decision_consultation('choice',dict(text='Question',client_key='message')))
with self.assertRaises(Invalid): self.store.decision_consultation('choice',dict(text='x',client_key='x',session_id='other'))
self.store.decision_respond('choice',answer())
self.store.decision_review('choice',review())
for status in ['queued','claimed','uncertain']:
with self.store.db() as db: db.execute('UPDATE messages SET status=?',(status,))
with self.assertRaises(Conflict): self.store.decision_complete('choice',completion())
with self.store.db() as db: db.execute("UPDATE messages SET status='completed'")
self.store.decision_complete('choice',completion())
with self.assertRaises(Conflict): self.store.decision_consultation('choice',dict(text='Late',client_key='late'))
with self.assertRaises(Conflict): self.store.send(dict(session_id='consult',text='Late',idempotency_key='late'))
self.assertTrue(self.store.decision_consultation('choice')['messages'])
with self.store.db() as db: db.execute("UPDATE messages SET status='uncertain'")
with self.assertRaises(Conflict): self.store.decision_archive('choice',dict(thread_id=THREAD,idempotency_key='archive',receipt='receipt'))
def test_consultation_send_and_completion_are_atomic(self):
bind(self.store)
self.store.decision_respond('choice',answer())
self.store.decision_review('choice',review())
def run(action):
try:
if action=='send':
return action,self.store.decision_consultation('choice',dict(text='Concurrent input',client_key='race'))
return action,self.store.decision_complete('choice',completion())
except Conflict:
return action,None
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
outcomes=dict(pool.map(run,['send','complete']))
self.assertEqual(sum(v is not None for v in outcomes.values()),1)
if outcomes['send'] is not None:
self.assertEqual(self.store.decision_get('choice')['status'],'accepted')
else:
self.assertEqual(self.store.decision_get('choice')['status'],'completed')
self.assertEqual(self.store.messages('consult'),[])
def test_notification_snapshots_and_cancel_only_unsent(self):
notice=self.store.decision_notifications()['notifications'][0]
self.store.decision_submitting(dict(notification_id=notice['notification_id']))
bind(self.store)
self.assertEqual(self.store.decision_notifications()['notifications'][0],notice)
self.store.decision_respond('choice',answer())
submitted=next(n for n in self.store.decision_notifications()['notifications'] if n['event']['kind']=='submitted')
self.store.decision_submitting(dict(notification_id=submitted['notification_id']))
self.store.decision_patch('choice',dict(expected_revision=1,reason='Change',question='New?'))
self.assertIn(submitted,self.store.decision_notifications()['notifications'])
self.assertEqual(submitted['event']['revision'],1)
self.assertEqual(submitted['decision']['question'],'Which approach?')
def test_fact_requests_dedup_answer_isolation_and_notification(self):
bind(self.store)
request=dict(idempotency_key='facts',question='Which version?',context='Need evidence')
fact=self.store.decision_fact('choice',request)
self.assertEqual(fact,Store(self.path).decision_fact('choice',request))
self.store.decision(definition(id='other'))
with self.assertRaises(Invalid): self.store.decision_fact_answer('other',fact['id'],dict(idempotency_key='answer',answer='x'))
body=dict(idempotency_key='answer',answer='Verified version',evidence=['fixture'])
result=self.store.decision_fact_answer('choice',fact['id'],body)
self.assertEqual(result,self.store.decision_fact_answer('choice',fact['id'],body))
notices=self.store.decision_notifications()['notifications']
self.assertFalse(any(n['event']['kind']=='fact_requested' for n in notices))
notice=next(n for n in notices if n['event']['kind']=='fact_answered')
self.assertEqual((notice['target'],notice['decision']['session_thread_id']),('decision',THREAD))
self.assertEqual(self.store.decision_get('choice')['status'],'open')
self.assertFalse(any(e['kind']=='review' for e in self.store.decision_history('choice')['history']))
def test_queue_health_reads_workflow_outbox_and_submission_markers(self):
initial=self.store.queue_health()['decisions']
self.assertEqual(initial['pending_count'],1)
self.assertIsNotNone(initial['oldest_pending_at'])
nid=self.store.decision_notifications()['notifications'][0]['notification_id']
self.store.decision_submitting(dict(notification_id=nid))
self.assertEqual(self.store.queue_health()['decisions']['submission_started_unconfirmed_count'],1)
self.store.decision_ack(dict(notification_id=nid))
self.assertEqual(self.store.queue_health()['decisions']['pending_count'],0)
self.store.decision_respond('choice',answer())
self.store.decision_review('choice',review())
health=self.store.queue_health()['decisions']
self.assertEqual(health['pending_count'],1)
self.assertEqual(health['cancelled_count'],1)
def test_legacy_response_migration_never_accepts_or_loses_snapshot(self):
path=Path(self.tmp.name)/'old.sqlite3'
db=sqlite3.connect(path)
db.execute('CREATE TABLE decisions(id TEXT PRIMARY KEY,data TEXT NOT NULL,response TEXT,response_key TEXT)')
original=self.store._normalize_decision(definition())
original.pop('summary');original.pop('supersedes_id')
response=dict(id='legacy-response',option_id='a',comment='Original',created=1,authorization=False)
db.execute('INSERT INTO decisions VALUES(?,?,?,?)',('choice',json.dumps(original),json.dumps(response),'old-key'))
db.commit();db.close()
migrated=Store(path)
current=migrated.decision_get('choice')
self.assertEqual((current['status'],current['revision']),('pending_review',1))
self.assertEqual(current['response']['snapshot']['options'],original['options'])
self.assertIsNone(current['review'])
self.assertEqual(len(Store(path).decision_history('choice')['history']),2)
class WorkflowHTTPTests(unittest.TestCase):
setUp=fixtures.DecisionHTTPTests.setUp
tearDown=fixtures.DecisionHTTPTests.tearDown
request=fixtures.DecisionHTTPTests.request
def test_old_page_submission_without_revision_is_rejected_before_write(self):
self.request('/api/decisions','POST',definition())
old=answer();old.pop('expected_revision')
self.assertEqual(self.request('/api/decisions/choice/respond','POST',old)[0],400)
self.assertIsNone(self.server.store.decision_get('choice')['response'])
self.assertFalse(any(e['kind']=='submission' for e in self.server.store.decision_history('choice')['history']))
first=self.request('/api/decisions/choice/respond','POST',answer())
retry=self.request('/api/decisions/choice/respond','POST',answer())
self.assertEqual(first,retry)
self.assertEqual(first[0],200)
self.assertEqual(sum(e['kind']=='submission' for e in self.server.store.decision_history('choice')['history']),1)
def test_state_reports_current_page_asset_without_changing_decisions(self):
root=Path(self.tmp.name)/'frontend';root.mkdir()
self.server.web_root=root
(root/'index.html').write_text('<script type="module" crossorigin src="/assets/index-first.js"></script>')
self.assertEqual(self.request('/api/state')[1]['frontend_entrypoint'],'/assets/index-first.js')
(root/'index.html').write_text('<script type="module" src="/assets/index-next.js"></script>')
self.assertEqual(self.request('/api/state')[1]['frontend_entrypoint'],'/assets/index-next.js')
(root/'index.html').unlink()
self.assertIsNone(self.request('/api/state')[1]['frontend_entrypoint'])
def test_limited_credential_route_matrix_and_closed_writes(self):
store=self.server.store
store.decision(definition());bind(store)
code,issued=self.request('/api/decisions/choice/consultation-credential','POST',{})
self.assertEqual(code,200)
token=issued['token'];headers={'Authorization':'Bearer '+token}
with store.db() as db:
rows=[dict(r) for r in db.execute('SELECT * FROM decision_credentials')]
self.assertEqual(rows[0]['token_hash'],hashlib.sha256(token.encode()).hexdigest())
self.assertNotIn(token,json.dumps(rows))
for path in ['/api/decisions/choice','/api/decisions/choice/history','/api/decisions/choice/consultation']:
self.assertEqual(self.request(path,headers=headers)[0],200)
for method,path,payload in [('GET','/api/state',None),('GET','/api/decisions/other',None),('GET','/api/decision-bridge-bindings',None),
('POST','/api/decisions/choice/respond',answer()),('POST','/api/decisions/choice/review',review()),
('POST','/api/decisions/choice/complete',completion()),('POST','/api/sessions',{}),
('POST','/api/decisions/choice/consultation',dict(text='x',client_key='x')),
('POST','/api/decisions/choice/consultation-credential',{}),
('POST','/api/decisions/choice/bind-session',{}),('POST','/api/decisions/choice/archive-receipt',{})]:
with self.subTest(path=path): self.assertEqual(self.request(path,method,payload,headers)[0],403)
for field,value in [('task_id','other-task'),('supersedes_id','other'),('requires_authorization',False)]:
self.assertEqual(self.request('/api/decisions/choice','PATCH',dict(expected_revision=1,reason='Out of scope',**{field:value}),headers)[0],403)
self.assertEqual(self.request('/api/decisions/choice','PATCH',dict(expected_revision=1,reason='Evidence',summary='Updated'),headers)[0],200)
self.assertEqual(self.request('/api/decisions/choice/facts','POST',dict(idempotency_key='fact',question='Fact?'),headers)[0],200)
store.decision_respond('choice',answer(expected_revision=2));store.decision_review('choice',review(expected_revision=2));store.decision_complete('choice',completion(expected_revision=2))
self.assertEqual(self.request('/api/decisions/choice','PATCH',dict(expected_revision=2,reason='Late',summary='x'),headers)[0],403)
self.assertEqual(self.request('/api/decisions/choice/facts','POST',dict(idempotency_key='late',question='Late?'),headers)[0],403)
self.assertEqual(self.request('/api/decisions/choice/history',headers=headers)[0],200)
def test_workflow_http_endpoints_and_conflicts(self):
self.request('/api/decisions','POST',definition())
self.assertEqual(self.request('/api/decisions/choice/consultation')[0],409)
bind(self.server.store)
self.assertEqual(self.request('/api/decision-bridge-bindings')[0],200)
self.assertEqual(self.request('/api/decisions/choice/respond','POST',answer())[0],200)
self.assertEqual(self.request('/api/decisions/choice/review','POST',review())[0],200)
self.assertEqual(self.request('/api/decisions/choice/complete','POST',completion())[0],200)
self.assertEqual(self.request('/api/decisions/choice/archive-receipt','POST',dict(thread_id=THREAD,idempotency_key='archive',receipt='tool receipt'))[0],200)
self.assertEqual(self.request('/api/decisions/choice/history')[1]['history'][-1]['kind'],'archive')
if __name__=='__main__': unittest.main()