"""Real Git/SQLite tasks with a fake Herdr transport; never starts paid agents.""" import json import os from pathlib import Path import subprocess import sys import unittest import test_agentctl as fixtures SCRIPT = Path(__file__).resolve().parents[1] / 'herdr_agents.py' sys.path.insert(0, str(SCRIPT.parent)) from herdr_agents import PROFILES, ROUTES class HerdrTests(unittest.TestCase): git = fixtures.AgentctlTests.git cli = fixtures.AgentctlTests.cli initialized_task = fixtures.AgentctlTests.initialized_task def setUp(self): fixtures.AgentctlTests.setUp(self) self.task = self.initialized_task() self.bin = Path(self.temp.name) / 'bin'; self.bin.mkdir() self.log = Path(self.temp.name) / 'calls.jsonl' # Resolve so macOS /var → /private/var matches agentctl Repo.current strings. task_cwd = str(self.task.resolve()) self.env = dict(os.environ, HERDR_ENV='1', PATH=str(self.bin)+os.pathsep+os.environ['PATH'], MOCK_LOG=str(self.log), MOCK_CWD=task_cwd, MOCK_MODE='ok') self.env['MOCK_AGENTS'] = json.dumps(['qwen_max__alpha','qwen_flash__alpha','context_kimi__alpha','implementer_cursor__alpha','reviewer_glm__alpha']) fake = self.bin / 'herdr' fake.write_text("#!/usr/bin/env python3\n" + r'''import json,os,re,sys from pathlib import Path a=sys.argv[1:]; log=Path(os.environ['MOCK_LOG']) with log.open('a') as f:f.write(json.dumps(a)+'\n') mode=os.environ['MOCK_MODE']; cwd=os.environ['MOCK_CWD'] if a[:2]==['agent','list']: result={'agents':[{'name':n,'cwd':cwd,'agent_status':'idle'} for n in json.loads(os.environ['MOCK_AGENTS'])]} elif a[:2]==['agent','get']: result={'agent':{'agent_status':'blocked' if mode=='blocked' else 'done'}} elif a[:2]==['pane','list']: result={'panes':[{'pane_id':'w1:p2','cwd':cwd}]} elif a[:2]==['agent','prompt']: if mode=='timeout':print('timeout agent_blocked quota',file=sys.stderr);sys.exit(1) result={'status':'done'} elif a[:2]==['agent','read']: if mode=='retry': prompts=[json.loads(l) for l in log.read_text().splitlines() if json.loads(l)[:2]==['agent','prompt']] if len(prompts)==1:print(re.search(r'HERDR_RETRY:[a-f0-9]+',prompts[-1][3])[0]+':quota');sys.exit(0) print('Finished with evidence');sys.exit(0) else:result={'status':'started'} print(json.dumps({'result':result})) ''') fake.chmod(0o755) def run_cli(self, *args, env=None, cwd=None): return subprocess.run([sys.executable,'-B',str(SCRIPT),*args],cwd=cwd or self.task, env=env or self.env,capture_output=True,text=True,timeout=15) def calls(self, group): if not self.log.exists():return [] return [a for a in map(json.loads,self.log.read_text().splitlines()) if a[:2]==['agent',group]] def test_all_profiles_have_noninteractive_flags_and_explicit_pi_routes(self): for profile,(name,_,native) in PROFILES.items(): with self.subTest(profile=profile): p=self.run_cli('start','--dry-run',profile,name+'__a','w1:p2',env=dict(self.env,HERDR_ENV='')) self.assertEqual(p.returncode,0,p.stderr) for arg in native:self.assertIn(arg,p.stdout) self.assertFalse(self.log.exists()) self.assertIn('qwen3.8-max',PROFILES['pi-max'][2]) self.assertIn('--auto',PROFILES['kimi'][2]) self.assertIn('danger-full-access',PROFILES['codex'][2]) def test_outside_herdr_and_wrong_task_cwd_never_contact_transport(self): for env,cwd in [(dict(self.env,HERDR_ENV=''),self.task),(self.env,self.root)]: p=self.run_cli('dispatch','--task','alpha','report','draft',env=env,cwd=cwd) self.assertNotEqual(p.returncode,0) self.assertFalse(self.log.exists()) def test_main_or_wrong_target_pane_cannot_start(self): p=self.run_cli('start','--task','alpha','pi-max','qwen_max__a','w1:p2',env=dict(self.env,MOCK_CWD=str(self.root))) self.assertNotEqual(p.returncode,0);self.assertFalse(self.calls('start')) p=self.run_cli('start','--task','alpha','pi-max','qwen_max__a','w1:p2') self.assertEqual(p.returncode,0,p.stderr);self.assertEqual(len(self.calls('start')),1) def test_report_routes_to_max_not_kimi_or_minimax(self): p=self.run_cli('dispatch','--task','alpha','report','draft') self.assertEqual(p.returncode,0,p.stderr) self.assertEqual(self.calls('prompt')[0][2],'qwen_max__alpha') self.assertTrue(all('context_kimi' not in v for k,v in ROUTES.items() if k not in ('report-final-check','quota-emergency'))) def test_implementation_fallback_is_explicit(self): self.env['MOCK_AGENTS']=json.dumps(['reviewer_glm__alpha']) p=self.run_cli('dispatch','--task','alpha','implementation','change') self.assertEqual(p.returncode,69);self.assertFalse(self.calls('prompt')) p=self.run_cli('dispatch','--task','alpha','--implementation-fallback','Cursor quota exhausted, no writes','implementation','change') self.assertEqual(p.returncode,0,p.stderr) self.assertEqual(self.calls('prompt')[0][2],'reviewer_glm__alpha') def test_uncertain_transport_error_and_blocked_do_not_fallback(self): for mode in ['timeout','blocked']: self.log.unlink(missing_ok=True) p=self.run_cli('dispatch','--task','alpha','report','draft',env=dict(self.env,MOCK_MODE=mode)) self.assertEqual(p.returncode,75,p.stderr);self.assertEqual(len(self.calls('prompt')),1) def test_explicit_prewrite_retry_reaches_flash(self): p=self.run_cli('dispatch','--task','alpha','report','draft',env=dict(self.env,MOCK_MODE='retry')) self.assertEqual(p.returncode,0,p.stderr) self.assertEqual([a[2] for a in self.calls('prompt')],['qwen_max__alpha','qwen_flash__alpha']) def test_same_role_ambiguity_and_foreign_cwd_do_not_dispatch(self): for env in [dict(self.env,MOCK_AGENTS=json.dumps(['qwen_max__a','qwen_max__b'])),dict(self.env,MOCK_CWD=str(self.root))]: p=self.run_cli('dispatch','--task','alpha','report','draft',env=env) self.assertNotEqual(p.returncode,0);self.assertFalse(self.calls('prompt')) def test_kimi_reason_disabled_marker_and_cross_lane_budget(self): p=self.run_cli('dispatch','--task','alpha','report-final-check','check') self.assertNotEqual(p.returncode,0);self.assertFalse(self.calls('prompt')) marker=self.root/'.locks/herdr-kimi.disabled';marker.touch() args=['dispatch','--task','alpha','--kimi-reason','final report','report-final-check','check'] self.assertNotEqual(self.run_cli(*args).returncode,0);self.assertFalse(self.calls('prompt')) marker.unlink() p=self.run_cli(*args);self.assertEqual(p.returncode,0,p.stderr) p=self.run_cli('dispatch','--task','alpha','--kimi-reason','outage','quota-emergency','check') self.assertNotEqual(p.returncode,0);self.assertEqual(len(self.calls('prompt')),1) def test_kimi_atomic_limit_across_eight_processes(self): args=[sys.executable,'-B',str(SCRIPT),'dispatch','--task','alpha','--kimi-reason','final report','report-final-check','check'] processes=[subprocess.Popen(args,cwd=self.task,env=self.env,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True) for _ in range(8)] for p in processes:p.communicate(timeout=20) self.assertEqual(sum(p.returncode==0 for p in processes),1) self.assertEqual(len(self.calls('prompt')),1) def test_kimi_uncertain_send_consumes_budget(self): args=['dispatch','--task','alpha','--kimi-reason','final report','report-final-check','check'] self.assertEqual(self.run_cli(*args,env=dict(self.env,MOCK_MODE='timeout')).returncode,75) self.assertNotEqual(self.run_cli(*args).returncode,0) self.assertEqual(len(self.calls('prompt')),1) if __name__=='__main__':unittest.main()