| |
| """Fail unless a local OpenAI-compatible server emits a structured bash call.""" |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import urllib.request |
| from pathlib import Path |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser() |
| parser.add_argument("--base-url", default="http://127.0.0.1:8200/v1") |
| parser.add_argument("--model") |
| parser.add_argument( |
| "--system-prompt", |
| type=Path, |
| default=Path(__file__).resolve().parents[1] / "harness/system-prompt.md", |
| ) |
| args = parser.parse_args() |
|
|
| model = args.model |
| if model is None: |
| with urllib.request.urlopen(f"{args.base_url}/models", timeout=30) as response: |
| model = json.load(response)["data"][0]["id"] |
|
|
| completion_limit = 512 |
| body = { |
| "model": model, |
| "messages": [ |
| {"role": "system", "content": args.system_prompt.read_text().strip()}, |
| { |
| "role": "user", |
| "content": ( |
| "Inspect the current working directory before making any changes. " |
| "Use the bash tool now." |
| ), |
| }, |
| ], |
| "tools": [ |
| { |
| "type": "function", |
| "function": { |
| "name": "bash", |
| "description": "Execute a bash command in the current working directory.", |
| "parameters": { |
| "type": "object", |
| "properties": {"command": {"type": "string"}}, |
| "required": ["command"], |
| }, |
| }, |
| } |
| ], |
| "tool_choice": "auto", |
| "temperature": 0.2, |
| "top_p": 0.95, |
| "max_tokens": completion_limit, |
| } |
| request = urllib.request.Request( |
| f"{args.base_url}/chat/completions", |
| data=json.dumps(body).encode(), |
| headers={"Authorization": "Bearer local", "Content-Type": "application/json"}, |
| ) |
| with urllib.request.urlopen(request, timeout=180) as response: |
| result = json.load(response) |
| message = result["choices"][0]["message"] |
| calls = message.get("tool_calls") or [] |
| if not calls or calls[0].get("function", {}).get("name") != "bash": |
| raise SystemExit(f"expected a structured bash call, got: {message!r}") |
| arguments = json.loads(calls[0]["function"]["arguments"]) |
| command = arguments.get("command") |
| if not isinstance(command, str) or not command.strip(): |
| raise SystemExit(f"bash call has invalid arguments: {arguments!r}") |
| usage = result.get("usage") or {} |
| completion_tokens = usage.get("completion_tokens") |
| if not isinstance(completion_tokens, int) or completion_tokens >= completion_limit: |
| raise SystemExit( |
| "structured call did not stop before the generation ceiling: " |
| f"completion_tokens={completion_tokens!r}, ceiling={completion_limit}" |
| ) |
| print( |
| json.dumps( |
| { |
| "model": model, |
| "finish_reason": result["choices"][0].get("finish_reason"), |
| "function": "bash", |
| "command": command, |
| "completion_tokens": completion_tokens, |
| }, |
| indent=2, |
| ) |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|