111 lines
5.7 KiB
Python
111 lines
5.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Run the shared Codex CLI and emit the grade/session files test.sh already uses."""
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
|
|
def command(args, env, output):
|
|
base = env.get('OPENAI_BASE_URL', '').strip().rstrip('/')
|
|
if not base:
|
|
anthropic = env.get('ANTHROPIC_BASE_URL', '').strip().rstrip('/')
|
|
if not anthropic.endswith('/anthropic'):
|
|
raise ValueError('Codex grading needs OPENAI_BASE_URL or the existing ANTHROPIC_BASE_URL proxy route')
|
|
base = anthropic[:-len('/anthropic')] + '/openai/v1'
|
|
key = (env.get('OPENAI_API_KEY') or env.get('ANTHROPIC_API_KEY') or '').strip()
|
|
if not key:
|
|
raise ValueError('Codex grading needs the runtime proxy API key')
|
|
metadata = '{"origin":"harbor-grading"}'
|
|
for header in env.get('ANTHROPIC_CUSTOM_HEADERS', '').splitlines():
|
|
name, _, value = header.partition(':')
|
|
if name.lower() == 'x-surge-client-metadata':
|
|
metadata = value.strip()
|
|
settings = {
|
|
'model_provider': 'grader-proxy',
|
|
'model_reasoning_effort': args.effort,
|
|
'model_providers.grader-proxy.name': 'Grader proxy',
|
|
'model_providers.grader-proxy.base_url': base,
|
|
'model_providers.grader-proxy.env_key': 'OPENAI_API_KEY',
|
|
'model_providers.grader-proxy.wire_api': 'responses',
|
|
'model_providers.grader-proxy.http_headers.X-Surge-Client-Metadata': metadata,
|
|
# The task container supplies the execution environment, as for Claude.
|
|
'approval_policy': 'never', 'sandbox_mode': 'danger-full-access',
|
|
}
|
|
if args.one_shot:
|
|
# Disabling shell alone leaves apply_patch available in Codex 0.158.0.
|
|
catalog = output.parent / 'models.json'
|
|
catalog.write_text(json.dumps({'models': [{
|
|
'slug': args.model, 'display_name': args.model, 'description': 'Tool-free grading',
|
|
'supported_reasoning_levels': [], 'shell_type': 'disabled',
|
|
'visibility': 'hide', 'supported_in_api': True, 'priority': 0,
|
|
'support_verbosity': False, 'apply_patch_tool_type': None,
|
|
'truncation_policy': {'mode': 'tokens', 'limit': 10000},
|
|
'experimental_supported_tools': [], 'tool_mode': 'direct',
|
|
'model_messages': {'instructions_template': 'Grade only the supplied evidence. You have no tools.'},
|
|
}]}))
|
|
settings.update({'model_catalog_json': str(catalog), 'features.shell_tool': False,
|
|
'features.view_image': False, 'web_search': 'disabled',
|
|
'tools.update_plan.enabled': False, 'tools.experimental_request_user_input.enabled': False,
|
|
'agents.enabled': False, 'features.goals': False})
|
|
cmd = ['codex', 'exec', '--json', '--skip-git-repo-check', '--ignore-user-config',
|
|
'--ignore-rules', '--model', args.model, '--output-last-message', str(output)]
|
|
for name, value in settings.items():
|
|
cmd += ['-c', name + '=' + json.dumps(value)]
|
|
if args.resume:
|
|
cmd += ['resume', args.resume]
|
|
return cmd + ['-'], dict(env, OPENAI_API_KEY=key)
|
|
|
|
|
|
def run(args, prompt, env):
|
|
grade = Path(args.grade)
|
|
grade.unlink(missing_ok=True)
|
|
version = subprocess.run(['codex', '--version'], env=env, capture_output=True, text=True, check=True).stdout
|
|
match = re.search(r'\b(\d+)\.(\d+)\.(\d+)([^\s]*)', version)
|
|
if not match or match[4] or tuple(map(int, match.group(1, 2, 3))) < (0, 158, 0):
|
|
raise ValueError('Codex grading requires stable CLI 0.158.0 or newer; rebuild the task image')
|
|
if not args.one_shot:
|
|
prompt += '\n\nReturn the COMPLETE grade JSON as your final message, without markdown fences, instead of writing the grade file. The verifier saves and validates it. Use shell reads to inspect the evidence and view_image for screenshots.\n'
|
|
with tempfile.TemporaryDirectory(prefix='codex-grade-') as directory:
|
|
output = Path(directory) / 'grade.txt'
|
|
cmd, child = command(args, env, output)
|
|
process = subprocess.run(cmd, input=prompt, stdout=subprocess.PIPE, text=True, env=child)
|
|
# Codex may have followed the original file-writing instruction before failing.
|
|
grade.unlink(missing_ok=True)
|
|
if process.returncode:
|
|
raise ValueError(f'Codex exited with status {process.returncode}')
|
|
events = [json.loads(line) for line in process.stdout.splitlines() if line.strip()]
|
|
session = next((e.get('thread_id') for e in events if e.get('type') == 'thread.started'), None)
|
|
completed = [e for e in events if e.get('type') == 'turn.completed']
|
|
if not session or not completed or any(e.get('type') in {'error', 'turn.failed'} for e in events):
|
|
raise ValueError('Codex did not complete a grading turn; ' + process.stdout)
|
|
if args.resume and session != args.resume:
|
|
raise ValueError('Codex resumed a different session')
|
|
grade.write_text(output.read_text())
|
|
# Preserve the result shape the existing repair loop reads.
|
|
print(json.dumps({'session_id': session, 'usage': completed[-1].get('usage', {})}))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--model', required=True)
|
|
parser.add_argument('--effort', required=True)
|
|
parser.add_argument('--grade', required=True)
|
|
parser.add_argument('--resume', default='')
|
|
parser.add_argument('--one-shot', action='store_true')
|
|
args = parser.parse_args()
|
|
try:
|
|
run(args, sys.stdin.read(), dict(os.environ))
|
|
except (OSError, ValueError, subprocess.SubprocessError) as error:
|
|
print(f'ERROR: {error}', file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|