shipped_tool.py

#!/usr/bin/env python3
"""Standalone agentic coding tool for OpenAI-compatible endpoints."""
from __future__ import annotations

import argparse
import os
import subprocess
import sys

from openai import OpenAI

from agent_core import coding_task_message, openai_generate_fn, run_episode


def run_bash(cmd: str, timeout_s: int = 30) -> str:
    cp = subprocess.run(
        cmd,
        shell=True,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        timeout=timeout_s,
    )
    out = cp.stdout[-12000:]
    return f"Exit code: {cp.returncode}\n{out}"


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--base-url", default=os.environ.get("OPENAI_BASE_URL", "http://127.0.0.1:8000/v1"))
    ap.add_argument("--model", default=os.environ.get("MODEL", "bk1dr/qwen3-8b-code-pkpo"))
    ap.add_argument("--api-key", default=os.environ.get("OPENAI_API_KEY", "EMPTY"))
    ap.add_argument("--max-turns", type=int, default=1,
                    help="One-turn deployment default used for training and evaluation.")
    ap.add_argument("--cp", action="store_true", help="Wrap stdin as a competitive-programming task.")
    ap.add_argument("--starter-code", default="")
    args = ap.parse_args()

    prompt = sys.stdin.read()
    if args.cp:
        first = coding_task_message(prompt, starter_code=args.starter_code, max_turns=args.max_turns)
    else:
        first = prompt

    client = OpenAI(base_url=args.base_url, api_key=args.api_key)
    generate = openai_generate_fn(client, args.model)
    ep = run_episode(generate, run_bash, first, max_turns=args.max_turns)
    if ep.final_answer is not None:
        print(ep.final_answer)
    else:
        print(f"Episode ended without final answer: {ep.end_reason}", file=sys.stderr)
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())