"""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}