Add revisioned decisions and scoped consultation lifecycle

tests/test_attention.py

@@ -95,7 +95,7 @@ class AttentionTests(unittest.TestCase):
         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'))
+        self.store.decision_respond('d',dict(option_id=None,comment='Chosen',idempotency_key='reply',expected_revision=1))
         state=self.store.state()['attention']
         self.assertEqual(state['items'],[])
         self.assertFalse(any(state['pages'].values()))

tests/test_decision_bridge.py

@@ -0,0 +1,100 @@
+"""Decision bridge boundaries without spawning a model or modifying host sessions."""
+import json
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
+import decision
+import decision_fleet
+from worker import atomic_json
+
+
+class Process:
+    pid=42
+    code=None
+    def poll(self):return self.code
+    def terminate(self):self.code=0
+    def wait(self,timeout=None):return self.code
+    def kill(self):self.code=-1
+
+
+class DecisionBridgeTests(unittest.TestCase):
+    def fixture(self,root):
+        cfg={'control':{},'codex':'codex','state_file':str(Path(root)/'worker.json'),
+             'decision_fleet':{'runtime_root':str(Path(root)/'decisions'),'enabled':True,'max_active':1}}
+        binding={'decision_id':'d','session_id':'s','thread_id':'11111111-1111-4111-8111-111111111111',
+                 'worktree':root,'status':'open','archive_status':'none'}
+        return cfg,binding
+
+    def test_only_bound_threads_and_bounded_startup_are_bridged(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,binding=self.fixture(root);fleet=decision_fleet.Fleet(cfg)
+            second={**binding,'decision_id':'other','session_id':'other','thread_id':'22222222-2222-4222-8222-222222222222'}
+            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()) as spawn:
+                status=fleet.tick([binding,second]);fleet.tick([binding,second])
+                self.assertEqual(spawn.call_count,2)
+                self.assertEqual(status['waiting'],1)
+                self.assertEqual(status['active'],1)
+                self.assertTrue(all('app-server' not in call.args[0] for call in spawn.call_args_list))
+                self.assertEqual(fleet.children['d']['cfg']['thread_id'],binding['thread_id'])
+                self.assertNotIn('notification_targets',fleet.children['d']['cfg'])
+                fleet.close()
+
+    def test_binding_change_and_duplicate_target_never_redirect_running_work(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,b=self.fixture(root);fleet=decision_fleet.Fleet(cfg)
+            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()) as spawn:
+                fleet.tick([b]);old=fleet.children['d']['cfg']['thread_id']
+                result=fleet.tick([{**b,'thread_id':'22222222-2222-4222-8222-222222222222'}])
+                self.assertEqual(result['status'],'failed')
+                self.assertEqual(spawn.call_count,2)
+                self.assertEqual(fleet.children['d']['cfg']['thread_id'],old)
+                with self.assertRaises(ValueError):fleet.tick([b,{**b,'decision_id':'other'}])
+                fleet.close()
+
+    def test_archive_waits_for_pending_completion_and_does_not_kill_model(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,b=self.fixture(root);fleet=decision_fleet.Fleet(cfg)
+            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()):
+                fleet.tick([b]);child=fleet.children['d'];state=Path(child['cfg']['state_file'])
+                atomic_json(state,{'inflight':{'message_id':'message'}})
+                result=fleet.tick([{**b,'status':'completed','archive_status':'archived'}])
+                self.assertEqual(result['status'],'failed')
+                self.assertIn('d',fleet.children)
+                self.assertTrue(all(p.poll() is None for p in child['processes'].values()))
+                atomic_json(state,{})
+                fleet.tick([{**b,'status':'completed','archive_status':'archived'}])
+                self.assertEqual(fleet.children,{})
+                self.assertTrue(state.exists())
+
+    def test_unknown_inflight_is_never_restarted(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,b=self.fixture(root);fleet=decision_fleet.Fleet(cfg)
+            with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()) as spawn:
+                fleet.tick([b]);child=fleet.children['d']
+                atomic_json(Path(child['cfg']['state_file']),{'needs_reconciliation':True})
+                child['processes']['worker'].code=1
+                self.assertEqual(fleet.tick([b])['status'],'failed')
+                self.assertEqual(spawn.call_count,2)
+                fleet.close()
+
+    def test_consultation_helper_cannot_choose_other_decision_or_review(self):
+        calls=[]
+        class Client:
+            def __init__(self,*args):pass
+            def call(self,*args):calls.append(args);return {}
+        cfg={'control':{},'decision_id':'decision-one'}
+        with patch.object(decision,'Client',Client):
+            decision.execute(cfg,'show')
+            decision.execute(cfg,'ask',{'question':'What is the dependency?'})
+            decision.execute(cfg,'ask',{'question':'What is the dependency?'})
+            self.assertEqual(calls[1][1]['idempotency_key'],calls[2][1]['idempotency_key'])
+            self.assertEqual(calls[0][0],'/api/decisions/decision-one')
+            with self.assertRaises(ValueError):decision.execute(cfg,'review',{})
+            with self.assertRaises(ValueError):decision.execute({'control':{},'decision_id':'../other'},'show')
+            with self.assertRaises(ValueError):decision.execute(cfg,'edit',{'title':'No revision'})
+
+
+if __name__=='__main__':unittest.main()

tests/test_decision_workflow.py

@@ -0,0 +1,282 @@
+"""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_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()

tests/test_decisions.py

@@ -18,7 +18,7 @@ def definition(**changes):
 
 
 def answer(**changes):
-    value = dict(option_id='a',comment='Keep this exact comment',idempotency_key='key')
+    value = dict(expected_revision=1,option_id='a',comment='Keep this exact comment',idempotency_key='key')
     value.update(changes)
     return value
 
@@ -294,16 +294,17 @@ class DecisionTests(unittest.TestCase):
         self.assertEqual(first,self.store.decision(explicit))
         with self.assertRaises(Conflict):
             self.store.decision(definition(question='Changed'))
-        self.assertEqual(self.store.decision_notifications()['pending_count'],0)
+        self.assertEqual(self.store.decision_notifications()['pending_count'],1)
 
     def test_persistence_response_dedup_and_ack(self):
         self.store.decision(definition())
+        self.store.decision_ack({'notification_id':self.store.decision_notifications()['notifications'][0]['notification_id']})
         with patch('subprocess.run',side_effect=AssertionError('must not execute')):
             first = self.store.decision_respond('choice',answer())
         reopened = Store(self.path)
         self.assertEqual(first,reopened.decision_respond('choice',answer()))
         self.assertEqual(first,reopened.decision(definition()))
-        self.assertEqual(first['status'],'decided')
+        self.assertEqual(first['status'],'pending_review')
         self.assertFalse(first['response']['authorization'])
         self.assertEqual(first['response']['comment'],answer()['comment'])
         for changes in [dict(comment='changed'),dict(idempotency_key='other'),dict(option_id=None)]:
@@ -312,7 +313,7 @@ class DecisionTests(unittest.TestCase):
         queue = reopened.decision_notifications()
         self.assertEqual(queue['pending_count'],1)
         notice = queue['notifications'][0]
-        self.assertEqual(set(notice),{'notification_id','target','decision'})
+        self.assertEqual(set(notice),{'notification_id','target','decision','event'})
         self.assertEqual(notice['target'],'control')
         self.assertEqual(notice['decision'],first)
         ack = dict(notification_id=notice['notification_id'])
@@ -325,7 +326,7 @@ class DecisionTests(unittest.TestCase):
     def test_authorization_is_only_pending_verification(self):
         self.store.decision(definition(requires_authorization=True,options=[]))
         value = self.store.decision_respond('choice',answer(option_id=None,comment='Approve sudo'))
-        self.assertEqual(value['status'],'pending_verification')
+        self.assertEqual(value['status'],'pending_review')
         self.assertIs(value['response']['authorization'],False)
 
     def test_invalid_definitions(self):
@@ -342,7 +343,7 @@ class DecisionTests(unittest.TestCase):
 
     def test_invalid_answers_and_ack(self):
         self.store.decision(definition())
-        for changes in [dict(option_id=None),dict(option_id='bad'),dict(comment=None),
+        for changes in [dict(option_id=None,comment=''),dict(option_id='bad'),dict(comment=None),
                         dict(comment='x'*8001),dict(idempotency_key=''),dict(authorization=True)]:
             with self.subTest(changes=changes):
                 with self.assertRaises(Invalid): self.store.decision_respond('choice',answer(**changes))
@@ -350,7 +351,7 @@ class DecisionTests(unittest.TestCase):
         with self.assertRaises(Invalid): self.store.decision_ack(dict(notification_id='missing'))
         with self.assertRaises(Invalid): self.store.decision_ack(dict(notification_id='missing',target='human'))
         self.assertEqual(self.store.state()['decisions'][0]['status'],'open')
-        self.assertEqual(self.store.decision_notifications()['pending_count'],0)
+        self.assertEqual(self.store.decision_notifications()['pending_count'],1)
         self.store.decision(definition(id='free',options=[]))
         for option,comment in [(None,' '),('a','text')]:
             with self.assertRaises(Invalid):
@@ -358,6 +359,7 @@ class DecisionTests(unittest.TestCase):
 
     def test_concurrent_same_key_has_one_response_and_notification(self):
         self.store.decision(definition())
+        self.store.decision_ack({'notification_id':self.store.decision_notifications()['notifications'][0]['notification_id']})
         with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
             values = list(pool.map(lambda _:self.store.decision_respond('choice',answer()),range(16)))
         self.assertTrue(all(v == values[0] for v in values))
@@ -365,6 +367,7 @@ class DecisionTests(unittest.TestCase):
 
     def test_concurrent_distinct_answers_have_one_winner(self):
         self.store.decision(definition())
+        self.store.decision_ack({'notification_id':self.store.decision_notifications()['notifications'][0]['notification_id']})
         def respond(index):
             try:
                 return self.store.decision_respond('choice',answer(idempotency_key=str(index)))
@@ -377,8 +380,9 @@ class DecisionTests(unittest.TestCase):
 
     def test_atomic_rollback_on_outbox_failure(self):
         self.store.decision(definition())
+        self.store.decision_ack({'notification_id':self.store.decision_notifications()['notifications'][0]['notification_id']})
         with self.store.db() as db:
-            db.execute("CREATE TRIGGER fail_notice BEFORE INSERT ON decision_notifications BEGIN SELECT RAISE(ABORT,'synthetic'); END")
+            db.execute("CREATE TRIGGER fail_notice BEFORE INSERT ON decision_workflow_notifications BEGIN SELECT RAISE(ABORT,'synthetic'); END")
         with self.assertRaises(Exception): self.store.decision_respond('choice',answer())
         self.assertEqual(self.store.state()['decisions'][0]['status'],'open')
         self.assertEqual(self.store.decision_notifications()['pending_count'],0)
@@ -387,6 +391,9 @@ class DecisionTests(unittest.TestCase):
         for index in range(103):
             did = 'choice-'+str(index)
             self.store.decision(definition(id=did))
+            with self.store.db() as db:
+                nid=db.execute('SELECT id FROM decision_workflow_notifications ORDER BY seq DESC LIMIT 1').fetchone()[0]
+            self.store.decision_ack({'notification_id':nid})
             self.store.decision_respond(did,answer())
         queue = self.store.decision_notifications()
         self.assertEqual(queue['pending_count'],103)
@@ -441,6 +448,8 @@ class DecisionHTTPTests(unittest.TestCase):
         self.assertEqual(self.request(notifications,headers={'Authorization':''})[0],401)
         self.assertEqual(self.request('/api/state')[1]['decisions'],[])
         self.assertEqual(self.request(create,'POST',definition())[0],200)
+        created_notice=self.request(notifications)[1]['notifications'][0]
+        self.request(ack,'POST',dict(notification_id=created_notice['notification_id']))
         self.assertEqual(self.request(create,'POST',definition(question='different'))[0],409)
         self.assertEqual(self.request(respond,'POST',answer(option_id='bad'))[0],400)
         code,value=self.request(respond,'POST',answer())
@@ -452,7 +461,7 @@ class DecisionHTTPTests(unittest.TestCase):
         self.assertEqual(self.request(notifications)[1]['pending_count'],0)
         self.assertEqual(self.request(notifications+'?target=human')[0],404)
         self.assertEqual(self.request(respond+'?x=1','POST',answer())[0],404)
-        self.assertEqual(self.request('/api/decisions/choice','PATCH',definition())[0],404)
+        self.assertEqual(self.request('/api/decisions/choice','PATCH',definition())[0],400)
         self.assertEqual(self.request('/api/state')[1]['decisions'][0],value)
 
     def test_cross_site_only_public_top_level_navigation(self):

tests/test_notifier.py

@@ -157,6 +157,63 @@ class NotifierTests(unittest.TestCase):
                 self.assertIn(cfg['execution_notification_targets']['worker'],run.call_args.args[0])
                 self.assertNotIn(cfg['internal_event_thread_id'],run.call_args.args[0])
 
+    def test_decision_workflow_uses_bound_consultation_and_never_replays(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,_,path=self.fixture(root);cfg['decision_notifications']=True
+            thread='44444444-4444-4444-8444-444444444444'
+            n={'notification_id':'fact-answer','target':'decision','decision':{'id':'one','title':'Plan','session_thread_id':thread},
+               'event':{'kind':'fact_answered','request_id':'fact-one','answer':'The prerequisite is verified.'}}
+            class Client:
+                def call(self,p,data=None):return {'notifications':[n]} if p=='/api/decision-notifications' else {'notifications':[]}
+            result=SimpleNamespace(returncode=0,stdout='Queued message 22222222-2222-4222-8222-222222222222 for thread '+thread+'.')
+            state={}
+            with patch.object(notifier.subprocess,'run',return_value=result) as run:
+                notifier.deliver(Client(),cfg,state,path);notifier.deliver(Client(),cfg,state,path)
+                self.assertEqual(run.call_count,1)
+                self.assertIn(thread,run.call_args.args[0])
+                self.assertNotIn(cfg['notification_targets']['control'],run.call_args.args[0])
+
+    def test_review_notice_is_concise_and_submission_is_not_acceptance(self):
+        n={'notification_id':'notice','decision':{'id':'one','title':'Plan','options':[{'id':'a','label':'Private option detail'}]},
+           'event':{'kind':'reviewed','reason':'Suitable in scope','impact':'No migration needed','execution_plan':'Verify before rollout'}}
+        text=notifier.notification_text(n)
+        self.assertNotIn('Private option detail',text)
+        self.assertIn('#/decisions?decision=one',text)
+        n['event']={'kind':'submitted','revision':2}
+        self.assertIn('不要把提交直接当接受',notifier.notification_text(n))
+        n['event']={'kind':'archive_required','revision':3}
+        self.assertIn('不能只写数据库冒充宿主归档',notifier.notification_text(n))
+
+    def test_failed_decision_submission_marker_does_not_queue_or_leave_unknown_model_input(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,_,path=self.fixture(root);cfg['decision_notifications']=True
+            n={'notification_id':'submit','target':'control','decision':{'id':'one','title':'Plan'},'event':{'kind':'submitted','revision':1}}
+            class Client:
+                failed=False
+                def call(self,p,data=None):
+                    if p=='/api/decision-notifications':return {'notifications':[n]}
+                    if p.endswith('/submitting') and not self.failed:
+                        self.failed=True;raise OSError('temporary marker interruption')
+                    return {'notifications':[]}
+            c=Client();state={}
+            result=SimpleNamespace(returncode=0,stdout='Queued message 22222222-2222-4222-8222-222222222222 for thread '+cfg['notification_targets']['control']+'.')
+            with patch.object(notifier.subprocess,'run',return_value=result) as run:
+                with self.assertRaises(RuntimeError):notifier.deliver(c,cfg,state,path)
+                self.assertEqual(state['deliveries'],{})
+                run.assert_not_called()
+                notifier.deliver(c,cfg,state,path)
+                self.assertEqual(run.call_count,1)
+
+    def test_consultation_target_cannot_point_to_primary_or_unbound_thread(self):
+        with tempfile.TemporaryDirectory() as root:
+            cfg,_,path=self.fixture(root);cfg['decision_notifications']=True
+            n={'notification_id':'bad','target':'decision','decision':{'id':'one','session_thread_id':cfg['notification_targets']['control']},'event':{'kind':'fact_answered'}}
+            class Client:
+                def call(self,p,data=None):return {'notifications':[n]} if p=='/api/decision-notifications' else {'notifications':[]}
+            with patch.object(notifier.subprocess,'run') as run:
+                with self.assertRaises(RuntimeError):notifier.deliver(Client(),cfg,{},path)
+                run.assert_not_called()
+
     def test_no_events_means_no_conversation_input(self):
         with tempfile.TemporaryDirectory() as root:
             cfg,n,path=self.fixture(root)

tests/test_queue_health.py

@@ -42,11 +42,13 @@ class QueueTests(unittest.TestCase):
             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?'))
+            for notice in self.store.decision_notifications()['notifications']:
+                self.store.decision_ack(dict(notification_id=notice['notification_id']))
         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.decision_respond('decision',dict(option_id=None,comment='Selected',idempotency_key='choice',expected_revision=1))
             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)
@@ -101,7 +103,7 @@ class QueueTests(unittest.TestCase):
     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'))
+        self.store.decision_respond('d',dict(option_id=None,comment='User choice',idempotency_key='choice',expected_revision=1))
         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')