wt
c28412ae8f
Changes: Preserve incomplete work and blocker metadata for S00-012. This checkpoint does not mark implementation accepted. Validation: Session ended with status blocked. Blocker: Required automation suite fails in existing all-prompts budget test: S21-006 exceeds even a diagnostic 90000-character prompt limit. This blocks verification of the stdin remediation; production limit remains 80000. Task: S00-012 Agent: manual-ci-cd-stdin-20261015b Role: ci-cd Action: escalate Status: blocked Task title: Fix runner prompt delivery after partial stdin timeout
386 lines
20 KiB
Python
386 lines
20 KiB
Python
#!/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
|
|
from git_commits import TaskCommits
|
|
|
|
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.commits = TaskCommits(self.root, config.get('git_commits', True))
|
|
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.
|
|
Before submit/approve/reject write state/commit_messages/{agent}.txt with a concrete
|
|
English imperative subject (15..100 chars), blank line, Changes: and Validation:
|
|
describing only actual changes/checks. Never run git add/commit/reset/push yourself.
|
|
The coordinator commits each transition before launching another agent.
|
|
'''
|
|
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)
|
|
git_head = self.commits.preflight()
|
|
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()
|
|
record = {'agent': agent, 'task': item['id'], 'action': item['action'], 'result': result}
|
|
self.checkpoint(item, agent, result, git_head, record)
|
|
return record
|
|
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
|
|
# Give the child a complete UTF-8 stream ending at EOF. Feeding a pipe
|
|
# through short communicate() retries can strand a partially sent prompt
|
|
# when the child is slow to start reading. The prompt is already persisted.
|
|
with prompt_path.open('rb') as prompt_input, (session / 'output.log').open('wb') as output:
|
|
try:
|
|
process = subprocess.Popen(argv, cwd=self.root, env=env, stdin=prompt_input,
|
|
stdout=output, stderr=subprocess.STDOUT, start_new_session=(os.name == 'posix'))
|
|
deadline = time.monotonic() + self.timeout
|
|
stopped_after_transition = False
|
|
try:
|
|
while True:
|
|
try:
|
|
process.wait(timeout=min(.2, max(.01, deadline - time.monotonic())))
|
|
break
|
|
except subprocess.TimeoutExpired:
|
|
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.wait()
|
|
stopped_after_transition = True
|
|
break
|
|
if time.monotonic() >= deadline:
|
|
stop_process_tree(process)
|
|
process.wait()
|
|
failure = 'Agent timed out; process terminated before escalation'
|
|
break
|
|
except BaseException:
|
|
stop_process_tree(process)
|
|
process.wait()
|
|
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()
|
|
if violations and self.commits.enabled:
|
|
record['commit_error'] = 'Scope violation: no automatic commit; inspect preserved changes'
|
|
else:
|
|
self.checkpoint(item, agent, result, git_head, record)
|
|
atomic_write(session / 'run.json', json.dumps(record, ensure_ascii=False, indent=2))
|
|
return record
|
|
|
|
def checkpoint(self, item, agent, result, git_head, record):
|
|
try:
|
|
record['commit'] = self.commits.commit(item, agent,
|
|
self.tracker.state()['tasks'][item['id']], result, git_head,
|
|
self.tracker.task(item['id'])['title'])
|
|
except (ValueError, OSError, subprocess.TimeoutExpired) as exc:
|
|
record['commit_error'] = str(exc)
|
|
|
|
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 records[-1].get('commit_error'):
|
|
return self.summary('git_error', records)
|
|
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': 'git_error' if stop == 'git_error' else '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,
|
|
'last_commit': records[-1].get('commit') if records else None,
|
|
'commit_error': records[-1].get('commit_error') 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 2 if result.get('stop') == 'git_error' else 3 if result.get('stop') == 'needs_input' else 0
|
|
except (ValueError, OSError, KeyError, subprocess.TimeoutExpired) 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())
|