#!/usr/bin/env python3 """Sequential fresh-process agent coordinator. Requires a configured tool-capable CLI.""" from __future__ import annotations import argparse import contextlib import fnmatch import hashlib import json import os from pathlib import Path import signal import subprocess import sys import time import uuid sys.path.insert(0, str(Path(__file__).resolve().parent)) from tasks import Tracker, atomic_write, now, read_json EXCLUDED_DIRS = {'.git', 'node_modules', '.venv', '__pycache__', '.state.lock', '.runner.lock', 'dist', 'coverage'} GENERATED = {'planning/tasks.json', 'planning/TASK_MATRIX.md', 'planning/task_matrix.csv', 'planning/ROADMAP.md', 'planning/TRACEABILITY.md'} def snapshot(root): result = {} for path in root.rglob('*'): relative = path.relative_to(root) if any(x in EXCLUDED_DIRS for x in relative.parts): continue name = relative.as_posix() if name.startswith(('state/', 'planning/tasks/', 'planning/stages/')) or name in GENERATED: continue if path.is_symlink(): result[name] = 'symlink:' + os.readlink(path) elif path.is_file(): result[name] = hashlib.sha256(path.read_bytes()).hexdigest() return result def scope_violations(before, after, scope): changed = {p for p in set(before) | set(after) if before.get(p) != after.get(p)} return sorted(p for p in changed if not any(fnmatch.fnmatchcase(p, pattern) for pattern in scope)) class Runner: def __init__(self, root, config): self.root = Path(root).resolve() self.tracker = Tracker(self.root) self.config = config self.lock_path = self.root / '.runner.lock' self.max_sessions = config.get('max_sessions', 1000) self.timeout = config.get('timeout_seconds', 1800) self.max_prompt = config.get('max_prompt_chars', 80000) if not isinstance(self.max_sessions, int) or not 1 <= self.max_sessions <= 10000: raise ValueError('max_sessions must be 1..10000') if not isinstance(self.timeout, (int, float)) or not 0 < self.timeout <= 86400: raise ValueError('timeout_seconds must be >0 and <=86400') if not isinstance(self.max_prompt, int) or not 48000 <= self.max_prompt <= 90000: raise ValueError('max_prompt_chars must be 48000..90000') @contextlib.contextmanager def lock(self): try: self.lock_path.mkdir() except FileExistsError: raise ValueError('Another runner may be active; inspect .runner.lock/owner.json, do not start a second coordinator') try: atomic_write(self.lock_path / 'owner.json', json.dumps({'pid': os.getpid(), 'at': now()})) yield finally: (self.lock_path / 'owner.json').unlink(missing_ok=True) self.lock_path.rmdir() def choose(self, stage=None): ready = self.tracker.ready() if stage: index = self.tracker.index() ready = [r for r in ready if index[r['id']]['stage'] == stage] return ready[0] if ready else None def command(self, item, agent, prompt_path): commands = self.config.get('commands_by_role', {}) argv = commands.get(item['role'], self.config.get('argv')) if not isinstance(argv, list) or not argv or not all(isinstance(x, str) and x for x in argv): raise ValueError('Configure argv as a nonempty JSON array (no shell string)') values = {'root': str(self.root), 'task': item['id'], 'role': item['role'], 'agent': agent, 'action': item['action'], 'prompt_file': str(prompt_path)} expanded = [] for argument in argv: for key, value in values.items(): argument = argument.replace('{' + key + '}', value) if 'REPLACE_' in argument: raise ValueError('Set your tool-capable CLI command in runner.local.json first') expanded.append(argument) return expanded def prompt(self, item, agent): tracker = self.tracker task = tracker.task(item['id']) assigned = f'''# Assignment from coordinator Agent id: {agent} Role: {item['role']} Task: {item['id']} Action: {item['action']} This is a fresh session. Do not resume a CLI conversation or choose another task. The coordinator has claimed executor work; reviewers only review this task. Read master_prompt and instructions below in order. Your scope is only this card. At a blocker: write one issue JSON, call tasks.py escalate, then EXIT immediately. Do not solve the new remediation, do not unblock yourself, do not launch another agent. After submit/approve/reject EXIT. The coordinator will launch the next fresh process. ''' paths = ['AGENTS.md', 'README.md', 'master_prompt.md', 'roles/common.md', f'roles/{item["role"]}.md', 'docs/WEB_PROFILE.md', 'docs/ARCHITECTURE.md', 'docs/CONTRACTS.md', 'docs/WORKFLOW.md', 'docs/AUTOMATION.md'] # Only bounded, concrete referenced upstream artifacts, never the entire repository. for raw in task['inputs']: candidate = raw.split(' (')[0] if candidate not in paths and (self.root / candidate).is_file(): paths.append(candidate) row = tracker.state()['tasks'][item['id']] if row.get('handoff'): paths.append(row['handoff']) if item['action'] == 'review' and row.get('evidence'): paths.append(row['evidence']) chunks = [assigned] for rel in dict.fromkeys(paths): path = (self.root / rel).resolve() if not path.is_relative_to(self.root): raise ValueError('Prompt input escapes project root') if not path.is_file(): raise ValueError('Required prompt input missing: ' + rel) chunks.append('\n# File: ' + rel + '\n\n' + path.read_text(encoding='utf-8')) chunks.append(tracker.packet(item['id'], agent)) result = '\n'.join(chunks) if len(result) > self.max_prompt: raise ValueError('Prompt exceeds configured character budget; split context/task, do not truncate requirements') return result def quarantine(self, task_id, previous, agent): with self.tracker.lock(): state = self.tracker.state() current = state['tasks'][task_id] # Preserve the real transition history; withdraw an untrusted approval/submission. state['tasks'][task_id] = dict(previous) state['revision'] += 1 state['events'].append({'revision': state['revision'], 'at': now(), 'action': 'runner-quarantine', 'task': task_id, 'agent': agent, 'from': current['status'], 'to': previous['status']}) state['retired_agents'] = [x for x in state.get('retired_agents', []) if x != agent] self.tracker.write_state(state) def execution_issue(self, item, agent, reason, external=False, previous=None): task = self.tracker.task(item['id']) state = self.tracker.state() row = state['tasks'][item['id']] if row['status'] == 'blocked' and agent in state.get('retired_agents', []): return {'already_escalated': True} if row['status'] not in {'in_progress', 'review'} or (item['action'] == 'claim' and row.get('owner') != agent): if previous is None: raise ValueError('Cannot safely escalate unexpected state; inspect runner record') self.quarantine(item['id'], previous, agent) issue = { 'title': 'Устранить прерывание: ' + task['title'], 'reason': reason, 'role': task['role'], 'reviewer_role': task['reviewer_role'], 'scope': task['scope'][:6], 'outcome': 'Устранить одну указанную причину прерывания и доказать готовность исходной задачи к новому запуску; не выполнять соседние задачи.', 'steps': ['Прочитать blocker report и handoff.', 'Воспроизвести конкретную причину.', 'Исправить только локальную причину в scope или уточнить внешний input.', 'Выполнить focused проверку и submit.'], 'acceptance': ['Причина воспроизводимо устранена.', 'Исходная задача не завершена обходом требований.'], 'checks': ['Повторить сценарий прерывания.', 'Проверить отсутствие изменений вне scope.'], 'requirements': [], 'requires_user_input': external, } return self.tracker.escalate(item['id'], agent, item['role'], issue) def launch(self, item): agent = 'auto-' + item['role'] + '-' + uuid.uuid4().hex[:12] session = self.root / 'state/runs' / agent prompt_path = session / 'prompt.md' # Validate command before claiming; missing local configuration is not a product task blocker. argv = self.command(item, agent, prompt_path) if item['action'] == 'claim': self.tracker.mutate('claim', item['id'], agent, role=item['role']) previous = dict(self.tracker.state()['tasks'][item['id']]) try: prompt = self.prompt(item, agent) except (ValueError, OSError) as exc: result = self.execution_issue(item, agent, 'Context preparation failed: ' + str(exc), previous=previous) self.tracker.render() return {'agent': agent, 'task': item['id'], 'action': item['action'], 'result': result} atomic_write(prompt_path, prompt) record = {'agent': agent, 'task': item['id'], 'role': item['role'], 'action': item['action'], 'started_at': now(), 'status': 'running'} atomic_write(session / 'run.json', json.dumps(record, ensure_ascii=False, indent=2)) before = snapshot(self.root) env = os.environ.copy() env.update({'LUMA_TASK_ID': item['id'], 'LUMA_ROLE': item['role'], 'LUMA_AGENT_ID': agent, 'LUMA_ACTION': item['action'], 'LUMA_PROJECT_ROOT': str(self.root), 'LUMA_PROMPT_FILE': str(prompt_path)}) failure = None stopped_after_transition = False returncode = None with (session / 'output.log').open('wb') as output: try: process = subprocess.Popen(argv, cwd=self.root, env=env, stdin=subprocess.PIPE, stdout=output, stderr=subprocess.STDOUT, start_new_session=(os.name == 'posix')) deadline = time.monotonic() + self.timeout first_communicate = True stopped_after_transition = False try: while True: try: process.communicate(prompt.encode('utf-8') if first_communicate else None, timeout=min(.2, max(.01, deadline - time.monotonic()))) break except subprocess.TimeoutExpired: first_communicate = False try: live = self.tracker.state() current = live['tasks'][item['id']] blocker_exit = current['status'] == 'blocked' and agent in live.get('retired_agents', []) executor_exit = item['action'] == 'claim' and current['status'] == 'review' and current.get('owner') == agent reviewer_exit = item['action'] == 'review' and current['status'] in {'done', 'todo'} and current.get('reviewer') == agent except ValueError: blocker_exit = executor_exit = reviewer_exit = False if blocker_exit or executor_exit or reviewer_exit: stop_process_tree(process) process.communicate() stopped_after_transition = True break if time.monotonic() >= deadline: stop_process_tree(process) process.communicate() failure = 'Agent timed out; process terminated before escalation' break except BaseException: stop_process_tree(process) process.communicate() raise finally: # Even an exited harness must not leave source-writing descendants alive. stop_process_tree(process) returncode = process.returncode except OSError as exc: failure = 'Configured agent executable failed to start: ' + type(exc).__name__ violations = scope_violations(before, snapshot(self.root), task_scope(self.tracker, item['id'])) row = self.tracker.state()['tasks'][item['id']] if violations: self.quarantine(item['id'], previous, agent) result = self.execution_issue(item, agent, 'Scope violation: ' + ', '.join(violations[:10]) + '; inspect run record before permitting further edits.', external=True, previous=previous) elif row['status'] == 'blocked' and agent in self.tracker.state().get('retired_agents', []): result = {'blocked': True, 'resolution_tasks': row.get('blocked_by', [])} elif failure or (returncode != 0 and not stopped_after_transition): result = self.execution_issue(item, agent, failure or 'Agent exited with code ' + str(returncode), external=returncode is None, previous=previous) elif item['action'] == 'claim' and row['status'] != 'review': result = self.execution_issue(item, agent, 'Agent exited without submit or escalate', previous=previous) elif item['action'] == 'review' and row['status'] not in {'done', 'todo'}: result = self.execution_issue(item, agent, 'Reviewer exited without approve/reject/escalate', previous=previous) else: result = {'status': row['status']} record.update(status='finished', finished_at=now(), returncode=returncode, stopped_after_transition=stopped_after_transition, scope_violations=violations, result=result) atomic_write(session / 'run.json', json.dumps(record, ensure_ascii=False, indent=2)) self.tracker.render() self.tracker.validate() return record def run(self, stage=None, dry_run=False): if stage and stage not in {s['id'] for s in self.tracker.definitions()['stages']}: raise ValueError('Unknown stage: ' + stage) if dry_run: item = self.choose(stage) if not item: return self.summary('idle', []) prompt = self.prompt(item, 'auto-preview') try: self.command(item, 'auto-preview', self.root / 'state/runs/preview/prompt.md') configuration_required = False except ValueError: configuration_required = True return {'dry_run': True, 'next': item, 'prompt_characters': len(prompt), 'configuration_required': configuration_required, 'state_unchanged': True} records = [] with self.lock(): self.tracker.validate() for _ in range(self.max_sessions): item = self.choose(stage) if not item: return self.summary('idle', records) records.append(self.launch(item)) if any(r.get('external_wait') for r in self.tracker.state()['tasks'].values()): return self.summary('external', records) return self.summary('session_limit', records) def summary(self, stop, records): state = self.tracker.state() done = all(r['status'] == 'done' for r in state['tasks'].values()) external = {k: r.get('reason') for k, r in state['tasks'].items() if r.get('external_wait')} return {'stop': 'complete' if done else 'needs_input' if external and stop in {'idle', 'external'} else stop, 'launched': len(records), 'revision': state['revision'], 'external_blockers': external, 'unfinished': sum(r['status'] != 'done' for r in state['tasks'].values()), 'last_session': records[-1].get('agent') if records else None} def stop_process_tree(process): if os.name == 'posix': try: os.killpg(process.pid, signal.SIGTERM) except ProcessLookupError: pass elif process.poll() is None: process.terminate() if process.poll() is None: try: process.wait(timeout=5) except subprocess.TimeoutExpired: if os.name == 'posix': try: os.killpg(process.pid, signal.SIGKILL) except ProcessLookupError: pass else: process.kill() process.wait() if os.name == 'posix': try: os.killpg(process.pid, signal.SIGKILL) except ProcessLookupError: pass def task_scope(tracker, task_id): return tracker.task(task_id)['scope'] def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--config', default='runner.local.json') parser.add_argument('--dry-run', action='store_true') parser.add_argument('--stage') parser.add_argument('--max-sessions', type=int) args = parser.parse_args() root = Path(__file__).resolve().parents[1] path = root / args.config try: if path.is_file(): config = read_json(path) elif args.dry_run: config = {} else: raise ValueError('Create runner.local.json from runner.example.json and set a tool-capable CLI argv once') if args.max_sessions is not None: config['max_sessions'] = args.max_sessions result = Runner(root, config).run(stage=args.stage, dry_run=args.dry_run) print(json.dumps(result, ensure_ascii=False, indent=2)) return 3 if result.get('stop') == 'needs_input' else 0 except (ValueError, OSError, KeyError) as exc: print(json.dumps({'stop': 'configuration_or_workflow_error', 'reason': str(exc)}, ensure_ascii=False), file=sys.stderr) return 2 if __name__ == '__main__': sys.exit(main())