"""Isolated blocker orchestration tests. No real model/provider/product task is run.""" import importlib.util import json from pathlib import Path import subprocess import sys import unittest import uuid from unittest.mock import Mock, patch from test_tasks import WorkflowTests, ROOT, MODULE RUN_SPEC = importlib.util.spec_from_file_location('luma_runner', ROOT / 'scripts/runner.py') RUN_MODULE = importlib.util.module_from_spec(RUN_SPEC) RUN_SPEC.loader.exec_module(RUN_MODULE) class BlockerTests(WorkflowTests): # Inherit temporary fixture helpers, but do not re-run the inherited test suite. def issue(self, **changes): result = {'title': 'Fix concrete cause', 'reason': 'A required local condition is unavailable', 'role': 'front', 'reviewer_role': 'qa', 'scope': ['src/fix.*'], 'outcome': 'Make the single missing condition verifiably available', 'steps': ['Reproduce', 'Fix local cause', 'Check'], 'acceptance': ['Original failure resolved'], 'checks': ['Positive and negative fixture'], 'requirements': ['REQ-001']} result.update(changes) return result def finish(self, ident, owner, reviewer): self.tracker.mutate('claim', ident, owner, role='front') self.tracker.mutate('submit', ident, owner, evidence=self.evidence(ident + '-' + owner + '.md'), handoff=self.evidence(ident + '.md', 'state/handoffs')) self.tracker.mutate('approve', ident, reviewer, role='qa', evidence=self.evidence(ident + '-review.md')) def test_escalation_blocks_parent_creates_ready_child_and_retires(self): self.claim() result = self.tracker.escalate('S00-001', 'owner', 'front', self.issue()) child = result['resolution_task'] self.assertTrue(result['agent_must_exit']) self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'blocked') self.assertIn(child, self.tracker.task('S00-001')['depends_on']) self.assertNotIn('S00-001', self.tracker.task(child)['depends_on']) self.assertEqual(self.tracker.ready()[0]['id'], child) with self.assertRaisesRegex(ValueError, 'fresh session'): self.claim('S00-003', 'owner') self.tracker.validate() # Child card is in the same journal, no render race. self.tracker.render() def test_reviewed_resolution_requeues_parent_for_new_agent(self): self.claim() child = self.tracker.escalate('S00-001', 'owner', 'front', self.issue())['resolution_task'] self.finish(child, 'resolver', 'resolver-reviewer') row = self.tracker.state()['tasks']['S00-001'] self.assertEqual((row['status'], row['owner']), ('todo', None)) self.assertEqual(self.tracker.ready()[0]['id'], 'S00-001') with self.assertRaises(ValueError): self.claim('S00-001', 'owner') self.claim('S00-001', 'fresh-owner') self.assertIn('Previous handoff', self.tracker.packet('S00-001', 'fresh-owner')) def test_submit_only_does_not_unblock(self): self.claim() child = self.tracker.escalate('S00-001', 'owner', 'front', self.issue())['resolution_task'] self.tracker.mutate('claim', child, 'resolver', role='front') self.tracker.mutate('submit', child, 'resolver', evidence=self.evidence('resolve.md'), handoff=self.evidence('resolve.md', 'state/handoffs')) self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'blocked') def test_reviewer_can_escalate_then_new_reviewer_continues(self): self.claim() self.submit() child = self.tracker.escalate('S00-001', 'reviewer', 'qa', self.issue())['resolution_task'] self.finish(child, 'resolver', 'resolve-review') self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'review') with self.assertRaises(ValueError): self.tracker.mutate('approve', 'S00-001', 'reviewer', role='qa', evidence=self.evidence('review.md')) self.tracker.mutate('approve', 'S00-001', 'fresh-reviewer', role='qa', evidence=self.evidence('fresh-review.md')) def test_cycle_or_unbounded_scope_is_rejected_without_partial_state(self): self.claim() before = self.tracker.state() for changes in [{'depends_on': ['S00-001']}, {'depends_on': ['S00-002']}, {'scope': ['**']}, {'scope': ['../outside/**']}, {'scope': ['specs/tz_clear.md']}]: with self.assertRaises(ValueError): self.tracker.escalate('S00-001', 'owner', 'front', self.issue(**changes)) self.assertEqual(self.tracker.state(), before) self.assertEqual(len(self.tracker.definitions()['tasks']), 3) def test_external_blocker_waits_until_release_and_verification(self): self.claim() child = self.tracker.escalate('S00-001', 'owner', 'front', self.issue(requires_user_input=True))['resolution_task'] self.assertEqual(self.tracker.state()['tasks'][child]['status'], 'blocked') self.assertNotIn(child, [r['id'] for r in self.tracker.ready()]) self.tracker.render() self.tracker.validate() MODULE.task_extensions.release_external(self.tracker, child, 'coordinator', 'orchestrator', 'Required test service restored') self.assertEqual(self.tracker.state()['tasks'][child]['status'], 'todo') self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'blocked') def test_duplicate_escalation_cannot_spawn_second_child(self): self.claim() self.tracker.escalate('S00-001', 'owner', 'front', self.issue()) before = self.tracker.state() with self.assertRaises(ValueError): self.tracker.escalate('S00-001', 'owner', 'front', self.issue()) self.assertEqual(before, self.tracker.state()) self.assertEqual(len(self.tracker.definitions()['tasks']), 4) def test_invalid_context_or_string_boolean_rejected(self): self.claim() for change in [{'requires_user_input': 'false'}, {'ui_sections': [999]}, {'spec_sections': 'all'}]: with self.assertRaises(ValueError): self.tracker.escalate('S00-001', 'owner', 'front', self.issue(**change)) self.assertEqual(len(self.tracker.definitions()['tasks']), 3) def test_nested_blocker_limit_waits_for_external_resolution(self): self.claim() current, agent = 'S00-001', 'owner' for depth in range(1, 5): result = self.tracker.escalate(current, agent, 'front', self.issue()) current = result['resolution_task'] if depth < 4: agent = 'resolver-' + str(depth) self.claim(current, agent) self.assertTrue(result['requires_user_input']) self.assertEqual(self.tracker.task(current)['blocker_depth'], 4) def test_plan_state_journal_recovers_after_interrupted_artifact_write(self): self.claim() real = MODULE.atomic_write calls = [] def fail_after_journal(path, text): calls.append(str(path)) if len(calls) == 2: raise OSError('Simulated interruption') return real(path, text) with patch.object(MODULE, 'atomic_write', side_effect=fail_after_journal): with self.assertRaises(OSError): self.tracker.escalate('S00-001', 'owner', 'front', self.issue()) self.assertTrue(self.tracker.journal_path.exists()) recovered = self.tracker.state() self.assertFalse(self.tracker.journal_path.exists()) self.assertEqual(recovered['tasks']['S00-001']['status'], 'blocked') child = recovered['tasks']['S00-001']['blocked_by'][0] self.assertIn(child, self.tracker.index()) self.tracker.render() self.tracker.validate() class RunnerTests(WorkflowTests): def setUp(self): super().setUp() for rel in ['AGENTS.md', 'README.md', 'master_prompt.md', 'roles/common.md', 'docs/WEB_PROFILE.md', 'docs/ARCHITECTURE.md', 'docs/CONTRACTS.md', 'docs/WORKFLOW.md', 'docs/AUTOMATION.md', 'docs/GIT_WORKFLOW.md']: path = self.root / rel path.parent.mkdir(parents=True, exist_ok=True) path.write_text('# Fresh one-task instructions\n', encoding='utf-8') fake = self.root / 'fake_agent.py' fake.write_text('''import importlib.util,json,os,pathlib,sys spec=importlib.util.spec_from_file_location('tool',sys.argv[1]);m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m) r=pathlib.Path(os.environ['LUMA_PROJECT_ROOT']);t=m.Tracker(r) id=os.environ['LUMA_TASK_ID'];agent=os.environ['LUMA_AGENT_ID'];role=os.environ['LUMA_ROLE'];action=os.environ['LUMA_ACTION'] prompt=sys.stdin.read();assert 'fresh session' in prompt log=r/'state/test-launches.json';entries=json.loads(log.read_text()) if log.exists() else [];entries.append({'task':id,'agent':agent,'action':action});log.write_text(json.dumps(entries)) if id=='S00-001' and action=='claim' and not (r/'state/first-blocker').exists(): (r/'state/first-blocker').write_text('recorded') issue={'title':'Fix local missing input','reason':'fixture blocker','role':'front','reviewer_role':'qa','scope':['src/fix.*'],'outcome':'Restore required local input','steps':['Restore local input'],'acceptance':['Input available'],'checks':['Original check succeeds'],'requirements':['REQ-001']} t.escalate(id,agent,role,issue);sys.exit(0) e=r/'state/evidence';h=r/'state/handoffs';e.mkdir(exist_ok=True);h.mkdir(exist_ok=True) evidence=e/(id+'-'+agent+'.md');evidence.write_text('# Evidence\\n\\nActual isolated fixture check succeeded; expected negative outcome checked.\\n') if action=='review':t.mutate('approve',id,agent,role=role,evidence=str(evidence)) else: handoff=h/(id+'.md');handoff.write_text('# Handoff\\n\\nFixture implementation completed; next step independent review of this task.\\n') t.mutate('submit',id,agent,evidence=str(evidence),handoff=str(handoff)) ''', encoding='utf-8') self.config = {'git_commits': False, 'argv': [sys.executable, str(fake), str(ROOT / 'scripts/tasks.py')], 'max_sessions': 20, 'timeout_seconds': 5} def test_fresh_process_blocker_resolver_review_resume_chain(self): runner = RUN_MODULE.Runner(self.root, self.config) result = runner.run() self.assertEqual(result['stop'], 'complete') launches = json.loads((self.root / 'state/test-launches.json').read_text()) self.assertEqual(len(launches), 9) self.assertEqual(len({x['agent'] for x in launches}), 9) original = [x for x in launches if x['task'] == 'S00-001' and x['action'] == 'claim'] self.assertEqual(len(original), 2) self.assertNotEqual(original[0]['agent'], original[1]['agent']) self.assertIn(original[0]['agent'], self.tracker.state()['retired_agents']) self.assertFalse(runner.lock_path.exists()) def test_dry_run_does_not_claim_or_write(self): runner = RUN_MODULE.Runner(self.root, {}) before = self.tracker.state() result = runner.run(dry_run=True) self.assertTrue(result['dry_run']) self.assertEqual(self.tracker.state(), before) self.assertFalse((self.root / 'state/runs').exists()) def test_runner_lock_prevents_second_coordinator(self): runner = RUN_MODULE.Runner(self.root, self.config) with runner.lock(): with self.assertRaises(ValueError): runner.run() def test_missing_executable_creates_external_resolution_and_stops(self): runner = RUN_MODULE.Runner(self.root, {'git_commits': False, 'argv': ['missing-tool-8b5e6d']}) result = runner.run() self.assertEqual(result['stop'], 'needs_input') self.assertEqual(result['launched'], 1) self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'blocked') self.assertTrue(result['external_blockers']) def test_incomplete_agent_exit_creates_remediation(self): runner = RUN_MODULE.Runner(self.root, {'git_commits': False, 'argv': [sys.executable, '-c', 'pass'], 'max_sessions': 1}) result = runner.run() self.assertEqual(result['stop'], 'session_limit') state = self.tracker.state() self.assertEqual(state['tasks']['S00-001']['status'], 'blocked') self.assertTrue(state['retired_agents']) def test_delayed_reader_receives_complete_large_utf8_prompt_and_eof(self): fake = self.root / 'fake_agent.py' fake.write_text(fake.read_text().replace( "prompt=sys.stdin.read();assert 'fresh session' in prompt", "import time;time.sleep(.5)\n" "payload=sys.stdin.buffer.read()\n" "assert payload == pathlib.Path(os.environ['LUMA_PROMPT_FILE']).read_bytes()\n" "prompt=payload.decode('utf-8');assert 'fresh session' in prompt\n" "print('Complete UTF-8 prompt and EOF:',len(payload),flush=True)")) (self.root / 'state/first-blocker').write_text('skip fixture blocker') prompt = 'This is a fresh session.\n' + '\U0001f642\u0416\n' * 20000 runner = RUN_MODULE.Runner(self.root, dict(self.config, max_sessions=1, timeout_seconds=3)) with patch.object(runner, 'prompt', return_value=prompt): result = runner.run() self.assertEqual(result['launched'], 1) self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'review') record_path = next((self.root / 'state/runs').glob('*/run.json')) record = json.loads(record_path.read_text()) self.assertEqual(record['result'], {'status': 'review'}) self.assertEqual(record['scope_violations'], []) self.assertIn('Complete UTF-8 prompt and EOF: ' + str(len(prompt.encode('utf-8'))), record_path.with_name('output.log').read_text()) def test_scope_violation_quarantines_and_requires_input(self): runner = RUN_MODULE.Runner(self.root, {'git_commits': False, 'argv': [sys.executable, '-c', "from pathlib import Path;Path('bad-edit.txt').write_text('wrong scope')"]}) result = runner.run() self.assertEqual(result['stop'], 'needs_input') self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'blocked') record = json.loads(next((self.root / 'state/runs').glob('*/run.json')).read_text()) self.assertEqual(record['scope_violations'], ['bad-edit.txt']) def test_reporter_process_is_stopped_after_escalate_without_waiting_timeout(self): import time fake = self.root / 'fake_agent.py' fake.write_text(fake.read_text().replace('t.escalate(id,agent,role,issue);sys.exit(0)', 't.escalate(id,agent,role,issue);import time;time.sleep(10);sys.exit(0)')) config = dict(self.config, max_sessions=1, timeout_seconds=5) started = time.monotonic() result = RUN_MODULE.Runner(self.root, config).run() self.assertLess(time.monotonic() - started, 3) self.assertEqual(result['launched'], 1) record = json.loads(next((self.root / 'state/runs').glob('*/run.json')).read_text()) self.assertTrue(record['stopped_after_transition']) self.assertEqual(len(self.tracker.definitions()['tasks']), 4) self.assertEqual(record['result']['blocked'], True) def test_invalid_stage_is_not_silent_idle(self): with self.assertRaisesRegex(ValueError, 'Unknown stage'): RUN_MODULE.Runner(self.root, {}).run(stage='S99', dry_run=True) def test_timeout_terminates_before_queuing_resolution(self): runner = RUN_MODULE.Runner(self.root, {'git_commits': False, 'argv': [sys.executable, '-c', 'import time;time.sleep(2)'], 'timeout_seconds': .05, 'max_sessions': 1}) result = runner.run() self.assertEqual(result['stop'], 'session_limit') self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'blocked') def _check_descendant_cleanup(self, transition): import time ready = self.root / 'state/descendant-ready' late_write = self.root / 'src/late-descendant-write.txt' child_code = ( "import pathlib,signal,time;signal.signal(signal.SIGTERM,signal.SIG_IGN);" f"pathlib.Path({str(ready)!r}).write_text('ready');" "time.sleep(1.5);" f"pathlib.Path({str(late_write)!r}).write_text('unexpected')" ) startup = ( "import subprocess,time\n" f"subprocess.Popen([sys.executable,'-c',{child_code!r}])\n" f"while not pathlib.Path({str(ready)!r}).exists():time.sleep(.01)\n" ) fake = self.root / 'fake_agent.py' source = fake.read_text().replace( "prompt=sys.stdin.read();assert 'fresh session' in prompt", startup + ("prompt=sys.stdin.read();assert 'fresh session' in prompt" if transition else "time.sleep(10)")) fake.write_text(source + '\nimport time;time.sleep(10)\n') (self.root / 'state/first-blocker').write_text('skip fixture blocker') runner = RUN_MODULE.Runner(self.root, dict(self.config, max_sessions=1, timeout_seconds=2 if transition else .5)) result = runner.run() self.assertEqual(result['launched'], 1) self.assertTrue(ready.is_file(), 'Descendant must start before cleanup') record = json.loads(next((self.root / 'state/runs').glob('*/run.json')).read_text()) self.assertEqual(record['stopped_after_transition'], transition) self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'review' if transition else 'blocked') time.sleep(1.6) self.assertFalse(late_write.exists(), 'Descendant survived process-group cleanup') self.assertEqual(record['scope_violations'], []) @unittest.skipUnless(sys.platform != 'win32', 'POSIX process-group cleanup') def test_timeout_stops_source_writing_descendant(self): self._check_descendant_cleanup(transition=False) @unittest.skipUnless(sys.platform != 'win32', 'POSIX process-group cleanup') def test_submit_transition_stops_source_writing_descendant(self): self._check_descendant_cleanup(transition=True) @unittest.skipUnless(sys.platform != 'win32', 'POSIX process-group cleanup') def test_completed_cleanup_does_not_signal_reaped_group_again(self): process = Mock(pid=12345, returncode=-15) process.poll.return_value = -15 with patch.object(RUN_MODULE.os, 'killpg') as killpg: RUN_MODULE.stop_process_tree(process) self.assertEqual(killpg.call_args_list, [unittest.mock.call(12345, RUN_MODULE.signal.SIGTERM), unittest.mock.call(12345, RUN_MODULE.signal.SIGKILL)]) killpg.side_effect = PermissionError('Reaped process group inaccessible') RUN_MODULE.stop_process_tree(process) self.assertEqual(killpg.call_count, 2) @unittest.skipUnless(sys.platform != 'win32', 'POSIX process-group cleanup') def test_failed_descendant_kill_is_not_marked_complete(self): process = Mock(pid=12345, returncode=-15) process.poll.return_value = -15 with patch.object(RUN_MODULE.os, 'killpg', side_effect=[None, PermissionError('Live descendant inaccessible')]) as killpg: with self.assertRaises(PermissionError): RUN_MODULE.stop_process_tree(process) killpg.side_effect = None RUN_MODULE.stop_process_tree(process) self.assertEqual(killpg.call_count, 4) def test_scope_matcher_checks_modified_created_deleted_files(self): self.assertEqual(RUN_MODULE.scope_violations({'a': 'old', 'gone': 'old'}, {'a': 'new', 'created': 'new'}, ['a']), ['created', 'gone']) class FoundationRunnerTests(unittest.TestCase): def test_ui_acceptance_prompt_keeps_exact_context_and_explicit_upstream_reads(self): runner = RUN_MODULE.Runner(ROOT, {}) task = runner.tracker.task('S21-006') for action, role in [('claim', task['role']), ('review', task['reviewer_role'])]: with self.subTest(action=action): prompt = runner.prompt({'id': task['id'], 'role': role, 'action': action}, 'auto-preview') self.assertLessEqual(len(prompt), 80000) self.assertIn(runner.tracker.packet(task['id'], 'auto-preview'), prompt) self.assertIn('Required task input reads (not inlined)', prompt) self.assertIn('Before implementation or review, read these task inputs', prompt) self.assertIn('a missing required artifact is a blocker', prompt) for raw in task['inputs']: self.assertIn(raw, prompt) if ' (upstream artifact' in raw: self.assertIn('- ' + raw, prompt) self.assertNotIn('# File: ' + raw.split(' (')[0] + '\n', prompt) # Every UI Kit section stays verbatim, including tables and # unnumbered details beyond the short requirement registry. spec = (ROOT / 'specs/uikit_clear.md').read_text(encoding='utf-8') for section in task['ui_sections']: self.assertIn(runner.tracker.section(spec, section), prompt) self.assertIn('This is a fresh session.', prompt) self.assertIn('After submit/approve/reject EXIT.', prompt) def test_all_current_fresh_agent_prompts_fit_configured_budget(self): runner = RUN_MODULE.Runner(ROOT, {}) lengths = [] for task in runner.tracker.definitions()['tasks']: for action, role in [('claim', task['role']), ('review', task['reviewer_role'])]: with self.subTest(task=task['id'], action=action): lengths.append(len(runner.prompt({'id': task['id'], 'role': role, 'action': action}, 'auto-preview'))) self.assertLessEqual(max(lengths), 80000) def test_unconfigured_runner_does_not_claim_product_work(self): before = (ROOT / 'state/tasks.json').read_bytes() result = subprocess.run([sys.executable, str(ROOT / 'scripts/runner.py'), '--config', 'missing-' + uuid.uuid4().hex + '.json'], capture_output=True, text=True) self.assertEqual(result.returncode, 2) self.assertIn('configuration', result.stderr) self.assertEqual(before, (ROOT / 'state/tasks.json').read_bytes()) # Imported helper class is not a second copy of its original tests. # Remove inherited test methods from these subclasses' discovery via explicit suite. def load_tests(loader, tests, pattern): import unittest suite = unittest.TestSuite() for cls in [BlockerTests, RunnerTests, FoundationRunnerTests]: suite.addTests(cls(name) for name in cls.__dict__ if name.startswith('test_')) return suite