Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323#!/usr/bin/env python3"""Hosted-only real TS/Rust HTTPS custom-root and unknown-issuer comparison.
The provider and private CA are synthetic, loopback-only and fixture-owned.This verifies provider transport, not native OS loading, catalog/OAuth clients,WebSockets, live Keychain updates, or same-process root-cache refresh."""import argparseimport http.serverimport jsonimport osfrom pathlib import Pathimport signalimport sslimport subprocessimport sysimport tempfileimport threadingimport time
from review_goal_backoff_parity import TS_ARCHIVE_SHA256, sha256from review_restore_notice_parity import textsfrom review_agent_bounce_parity import Clientfrom review_draft_disconnect_parity import stop
PROMPT = 'Complete the custom CA transport fixture.'ANSWER = 'Custom CA transport completed.'
def certificate(root, name): key, cert = root / (name + '.key'), root / (name + '.pem') subprocess.run(['openssl', 'req', '-x509', '-newkey', 'rsa:2048', '-nodes', '-keyout', str(key), '-out', str(cert), '-days', '1', '-subj', '/CN=Prime Agent fixture ' + name, '-addext', 'basicConstraints=critical,CA:TRUE', '-addext', 'keyUsage=critical,keyCertSign,cRLSign,digitalSignature', '-addext', 'subjectAltName=IP:127.0.0.1'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=15) key.chmod(0o600) return key, cert
def server_certificate(root, ca_key, ca_cert): key, csr, cert = root / 'server.key', root / 'server.csr', root / 'server.pem' extensions = root / 'server.extensions' extensions.write_text('basicConstraints=critical,CA:FALSE\n' 'keyUsage=critical,digitalSignature,keyEncipherment\n' 'extendedKeyUsage=serverAuth\nsubjectAltName=IP:127.0.0.1\n') subprocess.run(['openssl', 'req', '-new', '-newkey', 'rsa:2048', '-nodes', '-keyout', str(key), '-out', str(csr), '-subj', '/CN=Prime Agent TLS fixture'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=15) subprocess.run(['openssl', 'x509', '-req', '-in', str(csr), '-CA', str(ca_cert), '-CAkey', str(ca_key), '-CAcreateserial', '-out', str(cert), '-days', '1', '-sha256', '-extfile', str(extensions)], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=15) key.chmod(0o600) return key, cert
class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, *_args): pass
def do_POST(self): body = json.loads(self.rfile.read(int(self.headers['Content-Length']))) self.server.requests.append({'path': self.path, 'authorization': self.headers.get('Authorization'), 'body': body}) if self.path != '/v1/chat/completions': self.send_error(404) return def chunk(delta, finish=None): return {'id': 'ca-fixture', 'object': 'chat.completion.chunk', 'created': 1, 'model': 'ca-fixture', 'choices': [{'index': 0, 'delta': delta, 'finish_reason': finish}]} wire = ''.join('data: ' + json.dumps(row) + '\n\n' for row in ( chunk({'role': 'assistant', 'content': ANSWER}), chunk({}, 'stop'))) wire += 'data: [DONE]\n\n' encoded = wire.encode() self.send_response(200) self.send_header('Content-Type', 'text/event-stream') self.send_header('Content-Length', str(len(encoded))) self.end_headers() self.wfile.write(encoded)
class Server(http.server.ThreadingHTTPServer): daemon_threads = True
def __init__(self, key, cert): super().__init__(('127.0.0.1', 0), Handler) self.context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) self.context.load_cert_chain(cert, key) self.requests, self.handshake_failures = [], [] self.handshake_attempts = 0
def get_request(self): connection, address = super().get_request() connection.settimeout(5) self.handshake_attempts += 1 try: return self.context.wrap_socket(connection, server_side=True), address except ssl.SSLError as error: self.handshake_failures.append(str(error)) connection.close() raise
def owned_processes(root): expected = {('PI_CODING_AGENT_DIR=' + str(root / 'agent')).encode(), ('PRIME_AGENT_CODING_AGENT_DIR=' + str(root / 'agent')).encode()} found = {} for path in Path('/proc').iterdir(): if not path.name.isdigit(): continue try: environment = set((path / 'environ').read_bytes().split(b'\0')) if expected.isdisjoint(environment): continue start = (path / 'stat').read_text().rsplit(')', 1)[1].split()[19] found[int(path.name)] = start except (OSError, IndexError): continue return found
def confirm_cleanup(root, observed): deadline = time.monotonic() + 10 remaining = owned_processes(root) while remaining and time.monotonic() < deadline: threading.Event().wait(0.025) remaining = owned_processes(root) observed['remaining_owned_processes'] = remaining # Failure stays red, even when fixture-owned fallback cleanup succeeds. if remaining: for sig in (signal.SIGTERM, signal.SIGKILL): for pid, start in remaining.items(): if owned_processes(root).get(pid) == start: try: os.kill(pid, sig) except ProcessLookupError: pass deadline = time.monotonic() + 5 while owned_processes(root) and time.monotonic() < deadline: threading.Event().wait(0.025) observed['remaining_after_fallback'] = owned_processes(root) assert not remaining, 'owned detached processes survived normal shutdown' assert 'shutdown_error' not in observed, observed.get('shutdown_error')
def terminal_assistant(messages): assistants = [row for row in messages if row.get('role') == 'assistant'] assert len(assistants) == 1, 'expected one terminal assistant message' return assistants[0]
def request_contract(request): body = request['body'] users = ['\n'.join(texts(row.get('content'))) for row in body.get('messages', []) if row.get('role') == 'user'] count = users.count(PROMPT) assert count == 1, 'expected current fixture prompt exactly once' # Unrelated prompt layers are outside the TLS contract. Every raw message # remains in the receipt's captured provider request. return {'path': request['path'], 'authorization': request['authorization'], 'model': body.get('model'), 'stream': body.get('stream'), 'current_prompt': PROMPT, 'current_prompt_count': count}
def run_case(binary, label, root, trusted, key, cert, ca_cert, unrelated, 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} if label == 'ts': env['PI_PACKAGE_DIR'] = os.environ['PI_PACKAGE_DIR'] 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'), NO_PROXY='127.0.0.1,localhost', no_proxy='127.0.0.1,localhost', NODE_EXTRA_CA_CERTS=str(ca_cert if trusted else unrelated)) server = Server(key, cert) 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', 'baseUrl': 'https://127.0.0.1:%s/v1' % server.server_address[1], 'models': [{'id': 'ca-fixture', 'name': 'CA fixture', 'reasoning': False, 'contextWindow': 128000, 'maxTokens': 4096}]}}})) (agent / 'prompts').mkdir() (agent / 'prompts/transport-warm.md').write_text('Complete the transport fixture.') (agent / 'settings.json').write_text(json.dumps({'defaultProvider': 'fixture', 'defaultModel': 'ca-fixture', 'prewarmPythonKernel': False, 'retry': {'enabled': False}})) process, client = None, None log_path = root / 'daemon.log' log = log_path.open('wb') try: socket_path = root / 'd.sock' process = subprocess.Popen([str(binary), '--mode', 'daemon', '--daemon-socket', str(socket_path)], cwd=cwd, env=env, stdout=log, stderr=subprocess.STDOUT, start_new_session=True) client = Client(socket_path, process) created = client.call('create', name='custom-ca-fixture', config={ 'cwd': str(cwd), 'agentDir': str(agent), 'sessionDir': str(sessions), 'provider': 'fixture', 'model': 'ca-fixture', 'noExtensions': True, 'noSkills': True, 'noTools': True, 'noContextFiles': True, 'telemetryDisabled': True}) active = created.get('activeSessionId') or created.get('id') assert active, created client.call('attach', activeSessionId=active) # The ordinary UI requests its slash-command catalog before a turn. # This public read builds Rust's lazy core before user admission; a # real fixture template proves the build succeeded rather than # accepting its empty-on-build-error fallback. Cold admission is # separately blocked by the known #3560 bug, not normalized here. catalog = client.call('get_commands', activeSessionId=active) observation['startup_command_catalog'] = catalog assert any(row.get('name') == 'transport-warm' for row in catalog['commands']), catalog assert not server.requests, 'startup catalog unexpectedly invoked the provider' try: client.call('prompt_and_wait', activeSessionId=active, message=PROMPT) except AssertionError: # Certificate failure can surface as an unsuccessful wait response. # It is only valid after the actual terminal assistant error below; # any other command/transport failure remains red. response = client.wire[-1] assert not trusted and response.get('type') == 'response' and response.get('success') is False, response observation['prompt_wait_failure'] = response result = client.call('get_messages', activeSessionId=active) observation.update(messages=result, requests=server.requests, handshake_attempts=server.handshake_attempts, handshake_failures=server.handshake_failures) message = terminal_assistant(result['messages']) if trusted: assert message.get('stopReason') == 'stop', message answer = '\n'.join(texts(message.get('content'))) assert answer == ANSWER, message assert len(server.requests) == 1, 'expected one actual provider request' contract = request_contract(server.requests[0]) assert contract == {'path': '/v1/chat/completions', 'authorization': 'Bearer fixture-only', 'model': 'ca-fixture', 'stream': True, 'current_prompt': PROMPT, 'current_prompt_count': 1}, contract return {'outcome': 'completed', 'answer': answer, 'request': contract} assert server.handshake_attempts > 0, 'binary never attempted TLS' assert not server.requests, 'untrusted CA allowed an HTTP request' assert message.get('stopReason') == 'error' and message.get('errorMessage'), message return {'outcome': 'certificate_refused', 'http_requests': 0, 'assistant_stop_reason': message['stopReason']} finally: observation['owned_before_shutdown'] = owned_processes(root) if process and process.poll() is None and client: try: client.call('shutdown', force=True) observation['wire'] = client.wire client.close() process.wait(timeout=15) observation['shutdown'] = 'owned supervisor and workers stopped' except Exception as error: observation['shutdown_error'] = str(error) if process: stop(process) observation['daemon_exit_code'] = process.returncode if client: observation['wire'] = client.wire client.close() log.close() observation['daemon_log'] = log_path.read_text(errors='replace')[-10000:] server.shutdown() server.server_close() thread.join(timeout=5) confirm_cleanup(root, observation)
def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--receipt', type=Path, required=True) args = parser.parse_args() result = {'verified': False, 'runs': {}, 'scope': __doc__, 'retry_scope': 'Set documented retry.enabled=false identically in both real runtimes; compare certificate outcome, not retry policy.', 'dependency_scope': 'Warm ordinary startup get_commands before admission in both binaries. Cold lazy-first-prompt duplicate remains an unresolved dependency on #3560; no prompt deduplication or production repair in this TLS PR.', 'execution_scope': 'Explicit owned daemon, ordinary create/attach/get_commands/prompt_and_wait/get_messages, graceful shutdown acknowledgement then close socket before waiting.', 'prompt_scope': 'Compare current fixture prompt once, auth, model, stream and outcome. Unrelated layers excluded; raw messages retained.', 'cache_scope': 'Each case starts a fresh actual process. No assertion about in-process native-store refresh or TS native-root caching.'} try: assert sys.platform == 'linux', 'requires hosted Linux CI' archive_hash = sha256(os.environ['PA_TS_ARCHIVE']) assert archive_hash == TS_ARCHIVE_SHA256, 'pinned TS archive mismatch' binaries = {label: Path(os.environ['PA_' + label.upper() + '_BINARY']).resolve() for label in ('ts', 'rust')} result.update(ts_release_archive_sha256=archive_hash, binaries={label: {'path': str(path), 'sha256': sha256(path)} for label, path in binaries.items()}) with tempfile.TemporaryDirectory(prefix='custom-ca-parity-') as temporary: root = Path(temporary) ca_key, ca_cert = certificate(root, 'trusted') key, cert = server_certificate(root, ca_key, ca_cert) _, unrelated = certificate(root, 'unrelated') for scenario, trusted in [('trusted_custom_ca', True), ('untrusted_custom_ca', False)]: runs = result['runs'][scenario] = {} for label, binary in binaries.items(): observation = runs[label] = {} try: observation['contract'] = run_case(binary, label, root / scenario / label, trusted, key, cert, ca_cert, unrelated, observation) except Exception as error: observation['error'] = str(error) errors = {label: run['error'] for label, run in runs.items() if 'error' in run} assert not errors, errors assert runs['ts']['contract'] == runs['rust']['contract'], scenario + ' contract mismatch' result['verified'] = True except Exception as error: result['error'] = str(error) args.receipt.parent.mkdir(parents=True, exist_ok=True) args.receipt.write_text(json.dumps(result, indent=2) + '\n') print(json.dumps({'receipt': str(args.receipt), 'verified': result['verified'], 'error': result.get('error')})) return 0 if result['verified'] else 1
if __name__ == '__main__': sys.exit(main())