native macOS codings agent orchestrator prowl.onev.cat
Something went wrong. Try again.
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362"""Opt-in real runtime / production hook bridge contracts, outside the live GUI."""from __future__ import annotations
import jsonimport osimport refrom pathlib import Pathimport secretsimport socketimport structimport sysimport subprocessimport tempfileimport threadingimport time
import agent_contracts as coreimport agent_contract_expectations as rules
def same_directory(left, right): return isinstance(left, str) and os.path.isabs(left) and os.path.realpath(left) == os.path.realpath(right)
def verify_response(output, expected): return bool(re.search(r'RESULT:\s*' + re.escape(expected.split(':')[1]) + r'(?![0-9])', output))
def terminal_signal(signal, cwd): return (isinstance(signal, dict) and signal.get('event') == 'turn-ended' and signal.get('native_event') not in ('StopFailure', None) and same_directory(signal.get('cwd'), cwd) and bool(signal.get('session_id')))
def accept_signal(payload, runtime, token, cwd): return (isinstance(payload, dict) and payload.get('runtime') == runtime and payload.get('token') == token and terminal_signal(payload.get('signal'), cwd))
def evaluate(returncode, timed_out, response, hook): if timed_out: return 'runtime_timeout' if returncode != 0: return 'runtime_failed' if not response: return 'response_missing' if not hook: return 'terminal_hook_missing' return 'verified'
class Collector: """Length-framed CLI transport. No synthetic native events are generated here.""" def __init__(self, path, runtime, token, cwd, scope="headless"): self.runtime, self.token, self.cwd = runtime, token, cwd self.scope = scope self.complete = threading.Event() self.events = [] self.terminal = False self.error = False self.stop = threading.Event() self.server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) self.server.bind(str(path)) self.server.listen(8) self.server.settimeout(0.1) self.thread = threading.Thread(target=self.run, daemon=True)
@staticmethod def read(connection, count): data = b'' while len(data) < count: part = connection.recv(count - len(data)) if not part: raise ValueError('Incomplete frame') data += part return data
def run(self): while not self.stop.is_set(): try: connection, _ = self.server.accept() except socket.timeout: continue except OSError: return with connection: connection.settimeout(0.3) try: size = struct.unpack('>I', self.read(connection, 4))[0] if not 0 < size <= core.MAX_CAPTURE_BYTES: raise ValueError('Invalid frame') document = json.loads(self.read(connection, size)) payload = document['command']['agentsHook']['_0'] if (payload.get('runtime') == self.runtime and payload.get('token') == self.token and same_directory(payload.get('signal', {}).get('cwd'), self.cwd)): signal = payload['signal'] self.events.append({key: signal[key] for key in ('event', 'native_event', 'cwd', 'session_id') if key in signal}) self.terminal |= accept_signal(payload, self.runtime, self.token, self.cwd) if rules.validate_events(self.runtime, self.events, self.cwd, self.scope) == "verified": self.complete.set() connection.sendall(struct.pack('>I', 2) + b'{}') except (OSError, ValueError, KeyError, TypeError): self.error = True
def __enter__(self): self.thread.start() return self
def __exit__(self, *_): self.stop.set() self.thread.join(timeout=1) self.server.close()
def write_text(path, text): with os.fdopen(os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), 'w') as output: output.write(text)
def runtime_configuration(runtime, route, directory): """Explicit provider routes with temporary state; values are credential references only.""" expected_wire = {'codex': 'responses', 'claude': 'anthropic-messages', 'qodercli': 'runtime-managed'} if route['wire_api'] != expected_wire.get(runtime, 'chat-completions'): raise core.PolicyError('The live recipe does not support the selected wire protocol.') if runtime != 'qodercli' and route['api_key_env'] is None: raise core.PolicyError('The live recipe requires an explicit credential reference.') model, base, key = route['model'], route['base_url'], route['api_key_env'] environment, arguments = {}, [] state = directory / 'state' state.mkdir(mode=0o700) if runtime == 'codex': environment['CODEX_HOME'] = str(state) write_text(state / 'config.toml', '\n'.join([ 'model_provider = "contract"', 'model_reasoning_effort = "low"', 'model_reasoning_summary = "none"', 'web_search = "disabled"', '[model_providers.contract]', 'name = "Contract provider"', 'base_url = ' + json.dumps(base), 'wire_api = "responses"', 'env_key = ' + json.dumps(key), 'request_max_retries = 0', 'stream_max_retries = 0', ]) + '\n') arguments = ['--skip-git-repo-check', '--ephemeral', '--sandbox', 'read-only'] elif runtime == 'claude': environment.update(CLAUDE_CONFIG_DIR=str(state), ANTHROPIC_BASE_URL=base, ANTHROPIC_MODEL=model, ANTHROPIC_DEFAULT_HAIKU_MODEL=model, ANTHROPIC_DEFAULT_SONNET_MODEL=model, ANTHROPIC_DEFAULT_OPUS_MODEL=model, CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC='1', DISABLE_AUTOUPDATER='1', CLAUDE_CODE_MAX_OUTPUT_TOKENS='512', MAX_THINKING_TOKENS='0', API_MAX_RETRIES='0') arguments = ['--tools', '', '--strict-mcp-config', '--mcp-config', '{"mcpServers":{}}', '--no-session-persistence', '--max-turns', '1', '--system-prompt', 'Answer the arithmetic question.'] elif runtime == 'copilot': environment.update(COPILOT_HOME=str(state), COPILOT_PROVIDER_TYPE='openai', COPILOT_PROVIDER_BASE_URL=base, COPILOT_PROVIDER_WIRE_API='completions', COPILOT_PROVIDER_MODEL=model, COPILOT_PROVIDER_MAX_OUTPUT_TOKENS='512', COPILOT_OFFLINE='true') arguments = ['--no-custom-instructions', '--no-auto-update', '--no-remote', '--no-ask-user', '--available-tools=', '--disable-builtin-mcps', '--stream', 'off'] elif runtime == 'droid': config = state / 'settings.json' # Droid merges the scoped settings with the user's global custom models by id. The default id is # `custom:<model>`, so a global entry for the same model (and its own credential) would win. model_id = 'custom:prowl-contract-' + model core.write_json(config, {'customModels': [{'model': model, 'id': model_id, 'displayName': 'Contract Flash', 'baseUrl': base, 'apiKey': '${' + key + '}', 'provider': 'generic-chat-completion-api', 'maxOutputTokens': 512, 'extraArgs': {'thinking': {'type': 'disabled'}}}]}) arguments = ['--settings', str(config), '-m', model_id, '--enabled-tools', '', '--disable-builtin-skills'] # Factory credentials remain in its normal home: BYOK may still require a Factory account. elif runtime in ('pi', 'omp'): environment['PI_CODING_AGENT_DIR'] = str(state) model_config = {'providers': {'contract': {'baseUrl': base, 'api': 'openai-completions', 'apiKey': ('${' + key + '}' if runtime == 'pi' else key), 'models': [{'id': model, 'name': 'Contract Flash', 'reasoning': False, 'input': ['text'], 'contextWindow': 65536, 'maxTokens': 512, 'cost': {'input': 0, 'output': 0, 'cacheRead': 0, 'cacheWrite': 0}}]}}} # YAML is a superset of JSON; this avoids a new parser dependency for OMP's models.yml. core.write_json(state / ('models.json' if runtime == 'pi' else 'models.yml'), model_config) model = 'contract/' + model arguments = ['--no-extensions', '--no-tools', '--no-skills', '--no-session', '--thinking', 'off'] if runtime == 'omp': arguments += ['--no-lsp', '--max-time', '90'] elif runtime == 'qodercli': arguments = ['--tools', '', '--max-output-tokens', '512', '--max-model-request-retries', '0', '--strict-mcp-config', '--mcp-config', '{"mcpServers":{}}', '--no-session-persistence'] elif runtime == 'opencode': environment.update(XDG_CONFIG_HOME=str(state / 'config'), XDG_DATA_HOME=str(state / 'data'), XDG_CACHE_HOME=str(state / 'cache'), XDG_STATE_HOME=str(state / 'runtime'), OPENCODE_DISABLE_AUTOUPDATE='true', OPENCODE_DISABLE_DEFAULT_PLUGINS='true') environment['OPENCODE_CONFIG_CONTENT'] = json.dumps({ 'provider': {'contract': {'npm': '@ai-sdk/openai-compatible', 'name': 'Contract', 'options': {'baseURL': base, 'apiKey': '{env:' + key + '}'}, 'models': {model: {'name': 'Contract Flash', 'limit': {'context': 65536, 'output': 512}}}}}, 'permission': {'*': 'deny'}, 'share': 'disabled', 'autoupdate': False, }) model = 'contract/' + model arguments = ['--format', 'json'] else: raise core.PolicyError('This runtime has no verified direct-provider configuration recipe.') return model, arguments, environment
def validate_export(exported, counts, nonce, requests): expected = [item['runtime'] for item in requests] try: valid = (type(exported['schema']) is int and exported['schema'] == 1 and isinstance(nonce, str) and bool(nonce) and exported['nonce'] == nonce and counts['result'] == 'Passed' and all(type(counts[k]) is int and counts[k] == v for k, v in [('totalTestCount', 1), ('passedTests', 1), ('failedTests', 0), ('skippedTests', 0)]) and [item['runtime'] for item in exported['launches']] == expected) except (KeyError, TypeError): valid = False if not valid: raise core.EvidenceError('Production launch export lacks a matching executed-test receipt.')
def export_launches(requests, directory, search_path, timeout): nonce = secrets.token_hex(16) source, receipt = directory / 'export-input.json', directory / 'export-output.json' result = directory / 'export.xcresult' core.write_json(source, requests) environment = core.probe_environment(os.environ, search_path) environment.update(PROWL_CONTRACT_EXPORT_INPUT=str(source), PROWL_CONTRACT_EXPORT_OUTPUT=str(receipt), PROWL_CONTRACT_NONCE=nonce, PROWL_CONTRACT_RESULT=str(result)) if os.environ.get('PROWL_DEVELOPMENT_TEAM'): environment['PROWL_DEVELOPMENT_TEAM'] = os.environ['PROWL_DEVELOPMENT_TEAM'] process = core.run_process(['/usr/bin/make', '_test-agent-contract-export'], search_path, timeout, environment) write_text(directory / 'export-build.log', process.stdout + process.stderr) if process.timed_out or process.returncode != 0: raise core.EvidenceError('Production launch export build/test failed; inspect export-build.log.') summary = core.run_process(['/usr/bin/xcrun', 'xcresulttool', 'get', 'test-results', 'summary', '--path', str(result), '--compact'], search_path, 30) try: counts = json.loads(summary.stdout) exported = json.loads(receipt.read_text()) validate_export(exported, counts, nonce, requests) except (OSError, ValueError, KeyError): raise core.EvidenceError('Production launch export lacks a matching executed-test receipt.') from None core.write_json(directory / 'export-test-summary.json', counts) return exported['launches'], nonce
def run_one(launch, request, row, environment, path, directory, timeout, expected): runtime = row['runtime'] scope = request.get('scenario', 'headless') log_name = runtime + '-' + scope token = secrets.token_hex(24) child_environment = core.probe_environment(environment, path) child_environment.update(request['environment']) child_environment.update(launch['environment']) key_name = row['route']['api_key_env'] key = environment.get(key_name, '') if key_name else '' if key_name: child_environment[key_name] = key if runtime == 'claude': child_environment['ANTHROPIC_AUTH_TOKEN'] = key if runtime == 'copilot': child_environment['COPILOT_PROVIDER_API_KEY'] = key started = time.monotonic() # Unix sockaddr_un on macOS is short; do not place sockets in long checkout paths. with tempfile.TemporaryDirectory(prefix='pwl-', dir='/tmp') as short: socket_path = Path(short) / 's' child_environment.update(PROWL_AGENT_HOOK_TOKEN=token, PROWL_CLI_SOCKET=str(socket_path)) with Collector(socket_path, runtime, token, launch['cwd'], scope) as collector: with tempfile.TemporaryFile() as stdout, tempfile.TemporaryFile() as stderr: process = subprocess.Popen([launch['executable'], *launch['arguments']], cwd=launch['cwd'], env=child_environment, stdin=subprocess.DEVNULL, stdout=stdout, stderr=stderr, start_new_session=True) timed_out = False try: process.wait(timeout=timeout) collector.complete.wait(timeout=2) except subprocess.TimeoutExpired: timed_out = True finally: core.stop_group(process) stdout.seek(0) stderr.seek(0) out = stdout.read(core.MAX_CAPTURE_BYTES).decode('utf-8', errors='replace') err = stderr.read(core.MAX_CAPTURE_BYTES).decode('utf-8', errors='replace') response = verify_response(out, expected) if scope == 'zero-turn': reason = 'runtime_timeout' if timed_out else rules.zero_turn_result( runtime, process.returncode, out, err, collector.events, launch['cwd']) else: reason = evaluate(process.returncode, timed_out, response, collector.terminal) if reason == 'verified': reason = rules.validate_events(runtime, collector.events, launch['cwd']) if reason == 'verified' and collector.error: reason = 'capture_protocol_error' events = list(collector.events) # Raw stdout may echo prompt/config. Persist only after replacing known launch secrets. for value in (key, token): if value: out, err = out.replace(value, '[REDACTED]'), err.replace(value, '[REDACTED]') write_text(directory / (log_name + '-stdout.log'), out) write_text(directory / (log_name + '-stderr.log'), err) return {'status': 'passed' if reason == 'verified' else ('timed_out' if timed_out else 'contract_failed'), 'reason': reason, 'response_verified': response, 'events': events, 'exit_code': process.returncode, 'elapsed_seconds': round(time.monotonic() - started, 3), 'scope': scope, 'cwd': launch['cwd'], 'response_expected': expected, 'required_events': [list(pair) for pair in (rules.ZERO_TURN if scope == 'zero-turn' else rules.HEADLESS)[runtime]], 'stdout': str(directory / (log_name + '-stdout.log')), 'stderr': str(directory / (log_name + '-stderr.log'))}
def run_suite(rows, directory, resolver, environment, timeout, build_timeout, verify=False): requests, selected = [], [] with tempfile.TemporaryDirectory(prefix='prowl-contract-') as scratch: root = Path(scratch).resolve() resources = core.ROOT / 'App/Resources' for row in rows: row['live'] = {'status': 'blocked', 'reason': 'route_not_configured'} if row['runtime'] == 'qodercli' and row['route'].get('wire_api') != 'runtime-managed': row['live']['reason'] = 'custom_provider_requires_account_model_registration' continue if row['inventory']['status'] != 'passed': row['live']['reason'] = row['inventory']['reason'] continue if row['route']['status'] != 'configured': row['live']['reason'] = row['route']['status'] continue binary = resolver(row['runtime']) if not binary or binary.path != row['executable']: row['live']['reason'] = 'binary_changed' continue scopes = (['zero-turn', 'headless'] if verify and row['runtime'] in rules.ZERO_TURN else ['headless']) row['lifecycle'] = ({'status': 'not_run'} if row['runtime'] in rules.ZERO_TURN else {'status': 'not_applicable', 'reason': rules.ZERO_TURN_UNSUPPORTED[row['runtime']]}) for scope in scopes: workspace = root / (row['runtime'] + '-' + scope) workspace.mkdir(mode=0o700) try: model, arguments, config = runtime_configuration(row['runtime'], row['route'], workspace) except core.PolicyError: row['live']['reason'] = 'unsupported_live_route' continue number = 10000 + secrets.randbelow(80000) answer = 'RESULT:' + str(number + 1739) prompt = (f'Compute {number} + 1739. Reply only RESULT: followed immediately by the decimal sum. Do not use tools.' if scope == 'headless' else '') requests.append({'runtime': row['runtime'], 'executable': binary.path, 'workspace': str(workspace), 'resources': str(resources), 'model': model, 'arguments': arguments, 'scenario': scope, 'environment': {**config, 'HOME': str(Path.home())}, 'prompt': prompt}) selected.append((row, binary.search_path, answer)) if not requests: return try: launches, nonce = export_launches(requests, directory, selected[0][1], build_timeout) except core.EvidenceError as error: for row, _, _ in selected: row['live'] = {'status': 'blocked', 'reason': 'production_export_failed'} print(str(error), file=sys.stderr, flush=True) return for launch, request, (row, path, answer) in zip(launches, requests, selected): field = "lifecycle" if request["scenario"] == "zero-turn" else "live" if launch['status'] != 'prepared': row[field] = launch continue print(request['scenario'] + ' contract: ' + row['runtime'], file=sys.stderr, flush=True) try: row[field] = run_one(launch, request, row, environment, path, directory, timeout, answer) except OSError: row[field] = {'status': 'blocked', 'reason': 'runtime_or_capture_io_failed'} print(row['runtime'] + ': ' + row[field]['reason'], file=sys.stderr, flush=True)
return nonce