110 lines
5.8 KiB
Python
110 lines
5.8 KiB
Python
import sys, json, tempfile, asyncio
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone, timedelta
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, 'agent-ops/skills/project/orchestrate-agent-loop/scripts')
|
|
sys.path.insert(0, 'agent-ops/skills/project/orchestrate-agent-loop/tests')
|
|
import dispatch
|
|
|
|
|
|
async def main():
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
workspace = Path(temporary)
|
|
(workspace / '.git').mkdir()
|
|
directory = workspace / 'agent-task' / 'route' / '01_blocked'
|
|
directory.mkdir(parents=True)
|
|
header = '<!-- task=route/01_blocked plan=8 tag=REVIEW_REVIEW_REVIEW_REVIEW_REVIEW_REVIEW_REVIEW_REVIEW_REVIEW_REFACTOR -->\n'
|
|
(directory / 'PLAN-local-G07.md').write_text(header, encoding='utf-8')
|
|
(directory / 'CODE_REVIEW-local-G07.md').write_text(header, encoding='utf-8')
|
|
t_blocked = dispatch.scan_tasks(workspace, None)[0]
|
|
store = dispatch.StateStore(workspace)
|
|
|
|
nighttime = datetime(2026, 7, 26, 23, 0, 0, tzinfo=timezone(timedelta(hours=9)))
|
|
selector = dispatch._selector_module()
|
|
|
|
d_blocked, spec_blocked = dispatch.persisted_execution_decision(store, t_blocked, stage='worker', evaluated_at=nighttime)
|
|
|
|
attempt_dir = workspace / 'attempt-loc'
|
|
attempt_dir.mkdir(parents=True, exist_ok=True)
|
|
loc_path = attempt_dir / 'locator.json'
|
|
stream_log = attempt_dir / 'stream.log'
|
|
stream_log.write_text('sample stream log', encoding='utf-8')
|
|
norm_log = attempt_dir / 'normalized-output.log'
|
|
norm_log.write_text('sample normalized output', encoding='utf-8')
|
|
loc_path.write_text(json.dumps({
|
|
'workspace': str(workspace.resolve()),
|
|
'task': t_blocked.name,
|
|
'plan_path': str(t_blocked.plan.resolve()),
|
|
'stream_log': str(stream_log.resolve()),
|
|
'normalized_output_log': str(norm_log.resolve()),
|
|
}), encoding='utf-8')
|
|
|
|
store.update_task(t_blocked, blocked=f'worker failure provider-quota locator={loc_path}', blocker_evidence={
|
|
'role': 'worker', 'failure_class': 'provider-quota', 'locator': str(loc_path),
|
|
'selected': d_blocked['selected'], 'work_unit_id': d_blocked['work_unit_id'],
|
|
})
|
|
|
|
store.mark_retry_quota_refresh('route/01_blocked', workspace)
|
|
|
|
invoke_calls = []
|
|
async def fake_invoke(ws, st, task, role, spec, prompt, resume_locator=None):
|
|
attempt_dir = ws / 'attempt-fake'
|
|
attempt_dir.mkdir(parents=True, exist_ok=True)
|
|
locator = attempt_dir / 'locator.json'
|
|
record = {'status': 'succeeded', 'task': task.name, 'role': role}
|
|
retry_ctx = st.task_state(task).get('retry_quota_refresh_context') if isinstance(st, dispatch.StateStore) else None
|
|
print(f' [fake_invoke] retry_ctx is None: {retry_ctx is None}')
|
|
if retry_ctx is not None:
|
|
print(f' [fake_invoke] retry_ctx keys: {list(retry_ctx.keys())}')
|
|
print(f' [fake_invoke] has locator: {bool(retry_ctx.get("locator"))}')
|
|
print(f' [fake_invoke] has handoff_id: {bool(retry_ctx.get("handoff_id"))}')
|
|
if isinstance(retry_ctx, dict) and retry_ctx.get('locator'):
|
|
record['handoff_id'] = retry_ctx.get('handoff_id') or retry_ctx.get('locator')
|
|
record['source_locator'] = retry_ctx.get('locator')
|
|
record['source_context'] = {
|
|
'role': retry_ctx.get('role'),
|
|
'failure_class': retry_ctx.get('failure_class'),
|
|
'selected': retry_ctx.get('selected'),
|
|
'work_unit_id': retry_ctx.get('work_unit_id'),
|
|
}
|
|
locator.write_text(json.dumps(record), encoding='utf-8')
|
|
invoke_calls.append((task.name, role, spec, prompt, resume_locator))
|
|
return 0, None, locator
|
|
|
|
async def fake_run_review(ws, st, task, **kwargs):
|
|
archive = ws / 'agent-task' / 'archive' / '2026' / '07' / task.name
|
|
archive.parent.mkdir(parents=True, exist_ok=True)
|
|
(task.directory / 'complete.log').write_text('simulation complete\n', encoding='utf-8')
|
|
task.directory.rename(archive)
|
|
return str(archive)
|
|
|
|
args = dispatch.argparse.Namespace(
|
|
workspace=str(workspace), task_group='route', retry_blocked=True, dry_run=False,
|
|
)
|
|
|
|
with mock.patch.object(selector, 'probe_candidate_quota', return_value={'schema_version': '1.0', 'snapshot_id': 'snap', 'source': 'fake', 'checked_at': nighttime.isoformat(), 'targets': [{'adapter': 'codex', 'target': 'gpt-5.6-sol', 'status': 'available'}], 'required_caps': [], 'reason_codes': []}), \
|
|
mock.patch.object(dispatch, 'run_review', side_effect=fake_run_review), \
|
|
mock.patch.object(dispatch, 'ensure_review_shared_state'), \
|
|
mock.patch.object(dispatch, 'invoke', side_effect=fake_invoke), \
|
|
mock.patch.object(dispatch, 'datetime') as datetime_mock, \
|
|
mock.patch.object(selector.subprocess, 'run', side_effect=AssertionError('unexpected')):
|
|
datetime_mock.now.return_value = nighttime
|
|
res = await dispatch.dispatch_with_store(args, workspace, store)
|
|
|
|
print(f'Result: {res}')
|
|
print(f'Invoke calls: {len(invoke_calls)}')
|
|
for call in invoke_calls:
|
|
print(f' task={call[0]} role={call[1]}')
|
|
|
|
attempt_locators = list(workspace.rglob('locator.json'))
|
|
attempt_locators = [p for p in attempt_locators if p != loc_path]
|
|
print(f'Attempt locators: {len(attempt_locators)}')
|
|
for p in attempt_locators:
|
|
record = json.loads(p.read_text(encoding='utf-8'))
|
|
print(f' {p}: handoff_id={record.get("handoff_id")}')
|
|
|
|
store.close()
|
|
|
|
|
|
asyncio.run(main())
|