Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267#!/usr/bin/env python3"""Capture mandatory-thinking summary requests from real TS/Rust binaries.
The pinned TS v0.9.8 release and the current Rust binary run identical Pythoncells through their offered REPL tool. Only auxiliary request thinking controlsare compared, recording TS summary thinking-off as the intended correction:prompt wording, token budgets, and whole UI/frame parity are notclaimed. Branch summary uses an owned daemon and its real navigate_tree command.All providers, homes, sessions, and subprocesses belong to this fixture."""import argparseimport http.serverimport jsonimport osfrom pathlib import Pathimport sysimport socketimport subprocessimport timeimport tempfileimport threading
from review_goal_backoff_parity import TS_ARCHIVE_SHA256, sha256from review_restore_notice_parity import invoke, probe_fromfrom review_draft_disconnect_parity import stop
MARKER = 'MINIMUM_REASONING_RESULT='
class Server(http.server.ThreadingHTTPServer): daemon_threads = True
def __init__(self, scenario): super().__init__(('127.0.0.1', 0), Handler) self.scenario = scenario self.requests = [] self.main_requests = 0 self.tool_name = None self.error = None
class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, *_args): pass
def do_POST(self): if self.path != '/v1/chat/completions': self.send_error(404) return body = json.loads(self.rfile.read(int(self.headers['Content-Length']))) auxiliary = not body.get('tools') self.server.requests.append({'auxiliary': auxiliary, 'body': body}) if len(self.server.requests) > 12: self.server.error = 'request limit exceeded' self.send_error(422, self.server.error) return if auxiliary: content = '## Goal\nVerify auxiliary thinking controls.\n## Progress\nSeed fixture complete.' delta, finish = {'content': content}, 'stop' else: self.server.main_requests += 1 number = self.server.main_requests if number == 1 or self.server.scenario == 'branch': # Enough history for a real summary after a second user turn. delta, finish = {'content': 'Seed observation. ' * 300}, 'stop' elif number == 2: offered = {tool.get('function', {}).get('name') for tool in body.get('tools', [])} name = next((name for name in ('python_repl', 'ipython') if name in offered), None) if name is None: self.server.error = 'no supported REPL tool offered' self.send_error(422, self.server.error) return self.server.tool_name = name scenario = self.server.scenario code = (f'import {scenario}, json\n' f'result = await {scenario}.run("Verify mandatory thinking")\n' f'print({MARKER!r} + json.dumps(result))') delta = {'tool_calls': [{'index': 0, 'id': 'call_schedule', 'type': 'function', 'function': {'name': name, 'arguments': json.dumps({'code': code})}}]} finish = 'tool_calls' else: delta, finish = {'content': 'Fixture complete.'}, 'stop' def chunk(value, reason=None): return {'id': 'minimum-reasoning', 'object': 'chat.completion.chunk', 'created': 1, 'model': 'mandatory-fixture', 'choices': [{'index': 0, 'delta': value, 'finish_reason': reason}]} wire = ''.join('data: ' + json.dumps(item) + '\n\n' for item in ( chunk({'role': 'assistant', **delta}), chunk({}, finish))) + 'data: [DONE]\n\n' self.send_response(200) self.send_header('Content-Type', 'text/event-stream') self.send_header('Content-Length', str(len(wire.encode()))) self.end_headers() self.wfile.write(wire.encode())
def summarize_branch(binary, root, cwd, env, session_file, observation): entries = [json.loads(line) for line in session_file.read_text().splitlines()] target = next(row['id'] for row in entries if row.get('type') == 'message' and row.get('message', {}).get('role') == 'assistant') path = root / 'daemon.sock' with (root / 'daemon.log').open('wb') as log: process = subprocess.Popen([str(binary), '--mode', 'daemon', '--daemon-socket', str(path)], cwd=cwd, env=env, stdout=log, stderr=log, start_new_session=True) connection = None try: deadline = time.monotonic() + 35 while time.monotonic() < deadline: assert process.poll() is None, 'owned daemon exited before socket readiness' try: connection = socket.socket(socket.AF_UNIX) connection.settimeout(60) connection.connect(str(path)) break except (FileNotFoundError, ConnectionRefusedError): connection.close() connection = None threading.Event().wait(0.05) assert connection is not None, 'owned daemon socket never became ready' with connection.makefile('rb') as reader: hello = json.loads(reader.readline()) assert hello.get('type') == 'daemon_hello', hello observation['daemon_hello'] = hello def command(identifier, payload): connection.sendall((json.dumps({'type': 'command', 'id': identifier, 'protocol': hello['protocol'], 'command': payload}) + '\n').encode()) deadline = time.monotonic() + 60 while time.monotonic() < deadline: line = reader.readline() assert line, 'owned daemon closed before response' row = json.loads(line) observation.setdefault('daemon_wire', []).append(row) if row.get('type') == 'response' and row.get('id') == identifier: assert row.get('success') is True, row return row raise TimeoutError('no response for ' + identifier) created = command('create', {'type': 'create', 'sessionPath': str(session_file), 'config': {'cwd': str(cwd), 'sessionDir': str(session_file.parent)}}) active = created['data'].get('id') or created['data'].get('sessionId') assert active, created moved = command('branch', {'type': 'navigate_tree', 'activeSessionId': active, 'targetId': target, 'summarize': True, 'customInstructions': 'Verify mandatory thinking'}) assert moved.get('data', {}).get('cancelled') is not True, moved finally: if connection is not None: connection.close() stop(process) observation['daemon_log'] = (root / 'daemon.log').read_text(errors='replace')
def run_case(binary, label, scenario, root, observation): for name in ('home', 'agent', 'sessions', 'work', 'tmp'): (root / name).mkdir(parents=True) agent, sessions, cwd = root / 'agent', root / 'sessions', root / 'work' env = {key: os.environ[key] for key in ('PATH', 'LANG', 'LC_ALL', 'TZ', 'PRIME_AGENT_KERNEL_PYTHON', 'PA_E2E_KERNEL_PYTHON') if key in os.environ} library = Path(os.environ['PI_PACKAGE_DIR']) / 'skills' paths = sorted(str(path) for path in library.glob('*/src') if path.is_dir()) for skill in ('compact',): assert (library / skill / 'src' / skill / '__init__.py').is_file(), skill + ' skill missing' env.update(HOME=str(root / 'home'), TMPDIR=str(root / 'tmp'), DO_NOT_TRACK='1', PI_OFFLINE='1', PI_CODING_AGENT_DIR=str(agent), PRIME_AGENT_CODING_AGENT_DIR=str(agent), PI_SESSION_DIR=str(sessions), PRIME_AGENT_SESSION_DIR=str(sessions), XDG_CONFIG_HOME=str(root / 'config'), XDG_CACHE_HOME=str(root / 'cache'), XDG_DATA_HOME=str(root / 'data'), PYTHONPATH=os.pathsep.join(paths)) if label == 'ts': env['PI_PACKAGE_DIR'] = os.environ['PI_PACKAGE_DIR'] server = Server(scenario) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() (agent / 'models.json').write_text(json.dumps({'providers': {'fixture': { 'api': 'openai-completions', 'apiKey': 'fixture-only', # Provider-level compat is supported by both loaders. TS also merges # model-level compat; Rust's existing custom loader ignores that field. # Keep this fixture on their common configuration surface and require # the exact zai wire toggle, without accepting another adapter's format. 'compat': {'thinkingFormat': 'zai'}, 'baseUrl': 'http://127.0.0.1:%s/v1' % server.server_address[1], 'models': [{'id': 'mandatory-fixture', 'name': 'Mandatory thinking fixture', 'reasoning': True, 'input': ['text'], 'contextWindow': 128000, 'maxTokens': 8192, 'thinkingLevelMap': {'off': None, 'minimal': None, 'low': 'low', 'medium': None, 'high': 'high', 'xhigh': None, 'max': 'max'}}]}}})) (agent / 'settings.json').write_text(json.dumps({ 'defaultProvider': 'fixture', 'defaultModel': 'mandatory-fixture', 'prewarmPythonKernel': False, 'compaction': {'enabled': False, 'keepRecentTokens': 100, 'reserveTokens': 1024}, 'autoRefine': {'enabled': False}})) command = [str(binary), '--mode', 'json', '--provider', 'fixture', '--model', 'mandatory-fixture', '--thinking', 'low', '--session-dir', str(sessions), 'Seed the fixture.', 'Schedule the requested auxiliary operation.'] try: observation['version'] = invoke([str(binary), '--version'], cwd, env) observation['process'] = invoke(command, cwd, env) assert server.error is None, server.error bodies = [row['body'] for row in server.requests] if scenario == 'branch': session_files = list(sessions.rglob('*.jsonl')) assert len(session_files) == 1, session_files summarize_branch(binary, root, cwd, env, session_files[0], observation) scheduled = {'scheduled': True} else: scheduled = probe_from(bodies, MARKER) observation['scheduled'] = scheduled assert scheduled.get('scheduled') is True, scheduled auxiliary = [row['body'] for row in server.requests if row['auxiliary']] assert auxiliary, 'scheduled operation made no auxiliary provider request' controls = [{key: body[key] for key in ('enable_thinking', 'thinking', 'reasoning_effort', 'reasoning', 'chat_template_kwargs') if key in body} for body in auxiliary] expected = {'enable_thinking': label == 'rust'} assert all(control == expected for control in controls), {'expected': expected, 'actual': controls} # A scheduled cell alone is insufficient: check the persisted operation result. entries = [json.loads(line) for path in sessions.rglob('*.jsonl') for line in path.read_text().splitlines() if line.strip()] outcomes = [entry for entry in entries if (scenario == 'compact' and entry.get('type') == 'compaction') or (scenario == 'branch' and entry.get('type') == 'branch_summary')] assert outcomes, 'no durable completed ' + scenario + ' outcome' observation.update(controls=controls, outcome_count=len(outcomes), tool_name=server.tool_name) return {'scheduled': scheduled['scheduled'], 'completed': bool(outcomes)} finally: observation['requests'] = server.requests server.shutdown() server.server_close() thread.join(timeout=2)
def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--receipt', type=Path, required=True) args = parser.parse_args() result = {'verified': False, 'runs': {}, 'scope': 'Compaction and branch summary auxiliary thinking controls; no frame parity claim.', 'intended_difference': 'TS v0.9.8 compaction and branch summary requests disable zai thinking; Rust enables mandatory thinking.', 'fixture_compat_scope': 'Provider-level zai compat used identically by both custom-model loaders; model-level compat is not relied on.', 'comparison': 'Both schedule and persist completed outcomes. Every captured auxiliary request must match the exact expected control; request counts and full bodies remain in runs.'} try: assert sys.platform == 'linux', 'requires Linux hosted CI and pinned Linux TS binary' archive_hash = sha256(os.environ['PA_TS_ARCHIVE']) assert archive_hash == TS_ARCHIVE_SHA256, 'TS archive hash mismatch' result['ts_release_archive_sha256'] = archive_hash binaries = {label: Path(os.environ['PA_' + label.upper() + '_BINARY']).resolve() for label in ('ts', 'rust')} result['binaries'] = {label: {'path': str(path), 'sha256': sha256(path)} for label, path in binaries.items()} with tempfile.TemporaryDirectory(prefix='minimum-reasoning-parity-') as temp: for scenario in ('compact', 'branch'): results = {} for label, binary in binaries.items(): observation = {} result['runs'][scenario + '_' + label] = observation results[label] = run_case(binary, label, scenario, Path(temp) / scenario / label, observation) assert results['ts'] == results['rust'], results result['verified'] = True except Exception as error: result['error'] = str(error) raise finally: args.receipt.parent.mkdir(parents=True, exist_ok=True) args.receipt.write_text(json.dumps(result, indent=2) + '\n')
if __name__ == '__main__': main()