180 lines
11 KiB
Python
180 lines
11 KiB
Python
"""Blocker -> bounded remediation task -> fresh execution. Uses Tracker's lock/journal."""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
|
|
def check_dag(data):
|
|
index = {t['id']: t for t in data['tasks']}
|
|
visited, active = set(), set()
|
|
def visit(key):
|
|
if key in active:
|
|
raise ValueError('Remediation would create a dependency cycle: ' + key)
|
|
if key in visited:
|
|
return
|
|
if key not in index:
|
|
raise ValueError('Unknown remediation dependency: ' + key)
|
|
active.add(key)
|
|
for dep in index[key]['depends_on']:
|
|
visit(dep)
|
|
active.remove(key)
|
|
visited.add(key)
|
|
for key in index:
|
|
visit(key)
|
|
|
|
|
|
def escalate(tracker, task_id, agent, role, issue, handoff=None):
|
|
if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]{0,79}', agent or ''):
|
|
raise ValueError('A safe agent id is required')
|
|
if not isinstance(issue, dict):
|
|
raise ValueError('Issue must be a JSON object')
|
|
for field in ['title', 'reason', 'outcome', 'role']:
|
|
if not isinstance(issue.get(field), str) or not issue[field].strip():
|
|
raise ValueError('Issue requires nonempty ' + field)
|
|
for field, maximum in [('scope', 6), ('steps', 8), ('acceptance', 6), ('checks', 6)]:
|
|
value = issue.get(field)
|
|
if not isinstance(value, list) or not 1 <= len(value) <= maximum or not all(isinstance(x, str) and x.strip() for x in value):
|
|
raise ValueError('Issue requires a bounded nonempty list: ' + field)
|
|
if 'requires_user_input' in issue and not isinstance(issue['requires_user_input'], bool):
|
|
raise ValueError('requires_user_input must be a JSON boolean')
|
|
for section_key, maximum in [('spec_sections', 27), ('ui_sections', 26)]:
|
|
sections = issue.get(section_key, [])
|
|
if not isinstance(sections, list) or len(sections) > 6 or any(type(x) is not int or not 1 <= x <= maximum for x in sections):
|
|
raise ValueError('Invalid/broad remediation context sections')
|
|
if len(json.dumps(issue, ensure_ascii=False)) > 16000:
|
|
raise ValueError('Issue too large; describe one atomic cause')
|
|
with tracker.lock():
|
|
data, state = tracker.definitions(), tracker.state()
|
|
if agent in state.get('retired_agents', []):
|
|
raise ValueError('Agent has ended its session after a blocker; use a fresh agent')
|
|
index = {t['id']: t for t in data['tasks']}
|
|
if task_id not in index:
|
|
raise ValueError('Unknown original task')
|
|
parent, row = index[task_id], state['tasks'][task_id]
|
|
allowed_owner = row['status'] == 'in_progress' and row.get('owner') == agent and role == parent['role']
|
|
allowed_reviewer = row['status'] == 'review' and agent != row.get('owner') and role == parent['reviewer_role']
|
|
if not (allowed_owner or allowed_reviewer):
|
|
raise ValueError('Only current executor or assigned-role reviewer may escalate its current task')
|
|
roles = tracker.roles()
|
|
if issue['role'] not in roles or issue.get('reviewer_role', parent['reviewer_role']) not in roles:
|
|
raise ValueError('Unknown remediation operator/reviewer')
|
|
for scope in issue['scope']:
|
|
if Path(scope).is_absolute() or '..' in Path(scope).parts or scope in {'*', '**', '**/*', '.'}:
|
|
raise ValueError('Remediation must have bounded project-local scope')
|
|
if scope.startswith(('specs/', 'planning/', 'state/')):
|
|
raise ValueError('Remediation source scope cannot rewrite specs/workflow state')
|
|
extra = issue.get('depends_on', [])
|
|
if not isinstance(extra, list) or any(d not in index for d in extra):
|
|
raise ValueError('Remediation dependencies must reference existing task IDs')
|
|
if task_id in extra:
|
|
raise ValueError('Remediation cannot depend on its blocked parent')
|
|
depth = parent.get('blocker_depth', 0) + 1
|
|
root_id = parent.get('root_task', task_id)
|
|
previous = sum(t.get('kind') == 'blocker' and t.get('root_task') == root_id for t in data['tasks'])
|
|
external = bool(issue.get('requires_user_input')) or depth > 3 or previous >= 6
|
|
trigger = issue['reason']
|
|
if depth > 3 or previous >= 6:
|
|
trigger += '\nAutomatic blocker limit reached; explicit external resolution required.'
|
|
ordinal = max(int(t['id'].split('-')[1]) for t in data['tasks'] if t['stage'] == parent['stage']) + 1
|
|
if ordinal > 999:
|
|
raise ValueError('Stage task ID space exhausted')
|
|
child_id = f'{parent["stage"]}-{ordinal:03}'
|
|
requirements = issue.get('requirements', parent['requirements'])
|
|
allowed_reqs = {r['id'] for r in tracker.requirements() if r['applicability'] != 'out_of_scope'}
|
|
if not isinstance(requirements, list) or any(r not in allowed_reqs for r in requirements):
|
|
raise ValueError('Unknown/out-of-scope remediation requirement')
|
|
report = f'state/blockers/{child_id}.md'
|
|
child = {
|
|
'id': child_id, 'stage': parent['stage'], 'kind': 'blocker', 'root_task': root_id,
|
|
'blocker_depth': depth, 'unblocks': [task_id], 'role': issue['role'],
|
|
'reviewer_role': issue.get('reviewer_role', parent['reviewer_role']),
|
|
'title': issue['title'], 'outcome': issue['outcome'],
|
|
'depends_on': list(dict.fromkeys(parent['depends_on'] + extra)),
|
|
'requirements': list(dict.fromkeys(requirements)), 'scope': issue['scope'],
|
|
'inputs': list(dict.fromkeys(parent['inputs'] + [report])),
|
|
'spec_sections': issue.get('spec_sections', []), 'ui_sections': issue.get('ui_sections', []),
|
|
'steps': issue['steps'], 'acceptance': issue['acceptance'], 'checks': issue['checks'],
|
|
'context_budget_chars': 48000,
|
|
}
|
|
# Keep the original stage gate last; dynamic task IDs are never reused.
|
|
gate_id = next(s.get('exit_gate') for s in data['stages'] if s['id'] == parent['stage'])
|
|
insert_at = next((i for i, t in enumerate(data['tasks']) if t['id'] == gate_id), len(data['tasks']))
|
|
data['tasks'].insert(insert_at, child)
|
|
parent['depends_on'].append(child_id)
|
|
check_dag(data)
|
|
old_status, old_owner = row['status'], row.get('owner')
|
|
row.update(status='blocked', reason=trigger, blocked_by=[child_id], resume_status=old_status,
|
|
blocker_report=report, updated_at=tracker.timestamp())
|
|
if handoff:
|
|
row['handoff'] = tracker.safe_file(handoff, 'state/handoffs')
|
|
else:
|
|
row['handoff'] = f'state/handoffs/{task_id}.md'
|
|
state['tasks'][child_id] = {'status': 'blocked' if external else 'todo', 'owner': None,
|
|
'external_wait': external, 'reason': trigger if external else None}
|
|
state.setdefault('retired_agents', []).append(agent)
|
|
state['retired_agents'] = list(dict.fromkeys(state['retired_agents']))
|
|
state['revision'] += 1
|
|
state['events'].append({'revision': state['revision'], 'at': tracker.timestamp(), 'action': 'escalate',
|
|
'task': task_id, 'resolution_task': child_id, 'agent': agent, 'role': role,
|
|
'from': old_status, 'to': 'blocked', 'reason': trigger})
|
|
issue_text = '\n'.join([
|
|
'# Blocker ' + child_id, '', '- Original task: ' + task_id, '- Root task: ' + root_id,
|
|
'- Reporter: ' + agent, '- Role: ' + role, '- Original owner: ' + str(old_owner),
|
|
'- Waits for user/external condition: ' + str(external), '', '## Cause', '', trigger,
|
|
'', '## One remediation outcome', '', issue['outcome'], '',
|
|
'## Evidence / reproduction', '', issue.get('reproduction', 'See original handoff and task evidence.'), '',
|
|
'The reporting agent must stop now. A new agent implements only ' + child_id + '.', '',
|
|
])
|
|
artifacts = {report: issue_text, f"planning/tasks/{child_id}.md": tracker.card(child)}
|
|
if not handoff:
|
|
existing = tracker.root / row['handoff']
|
|
previous_text = existing.read_text(encoding='utf-8') if existing.is_file() else ''
|
|
artifacts[row['handoff']] = (previous_text + '\n\n# Blocked handoff\n\n' +
|
|
f'Original: {task_id}; reporter: {agent}; resolution: {child_id}.\n\n' + trigger +
|
|
'\n\nNext: complete and review the remediation; only a fresh session may continue.\n')
|
|
tracker.commit_plan_state(data, state, artifacts)
|
|
return {'task': task_id, 'status': 'blocked', 'resolution_task': child_id,
|
|
'resolution_status': state['tasks'][child_id]['status'], 'agent_must_exit': True,
|
|
'requires_user_input': external, 'revision': state['revision']}
|
|
|
|
|
|
def unblock_resolved(tracker, state, completed_id):
|
|
"""Called under lock as part of approval; no human/old-agent unblock needed."""
|
|
for parent_id, row in state['tasks'].items():
|
|
blockers = row.get('blocked_by', [])
|
|
if row['status'] != 'blocked' or completed_id not in blockers:
|
|
continue
|
|
if not all(state['tasks'][b]['status'] == 'done' for b in blockers):
|
|
continue
|
|
row['previous_owner'] = row.get('owner')
|
|
if row.get('resume_status') == 'review':
|
|
row.update(status='review', reason=None)
|
|
else:
|
|
row.update(status='todo', owner=None, reason=None)
|
|
row.update(blocked_by=[], resumed_after_blocker=True, updated_at=tracker.timestamp())
|
|
state['events'].append({'revision': state['revision'] + 1, 'at': tracker.timestamp(),
|
|
'action': 'auto-unblock', 'task': parent_id, 'resolved_by': completed_id,
|
|
'to': row['status'], 'fresh_agent_required': True})
|
|
|
|
|
|
def release_external(tracker, task_id, agent, role, reason):
|
|
if role != 'orchestrator' or not reason or not reason.strip():
|
|
raise ValueError('External release requires orchestrator and concrete resolution reason')
|
|
with tracker.lock():
|
|
state = tracker.state()
|
|
if agent in state.get('retired_agents', []):
|
|
raise ValueError('Use a fresh coordinator session')
|
|
row = state['tasks'].get(task_id)
|
|
if not row or row['status'] != 'blocked' or not row.get('external_wait'):
|
|
raise ValueError('Not an externally waiting remediation')
|
|
row.update(status='todo', owner=None, external_wait=False, reason=reason)
|
|
state['revision'] += 1
|
|
state['events'].append({'revision': state['revision'], 'at': tracker.timestamp(),
|
|
'action': 'release-external', 'task': task_id, 'agent': agent, 'reason': reason})
|
|
tracker.write_state(state)
|
|
return {'id': task_id, 'status': 'todo', 'verification_required': True}
|