Native API: distinguish provider finish_reason from inferred EOF completion (follow-up to #3669)

"""Python 3 stdlib only. Usage: python repro.py /path/to/docker-agent"""
import hashlib
import json
import os
from pathlib import Path
import socket
import subprocess
import sys
import tempfile
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.request import Request, ProxyHandler, build_opener

CASES = ("explicit_stop", "done_only", "bare_eof", "transport_cut")
calls = {name: 0 for name in CASES}
http = build_opener(ProxyHandler({}))


def frame(delta, finish=None):
    return "data: " + json.dumps({
        "id": "fixture", "object": "chat.completion.chunk",
        "choices": [{"index": 0, "delta": delta, "finish_reason": finish}]
    }) + "\n\n"


class Provider(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def log_message(self, *args):
        pass

    def do_POST(self):
        request = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        name = [m["content"] for m in request["messages"] if m["role"] == "user"][-1]
        calls[name] += 1
        body = frame({"role": "assistant", "content": "fixture text"})
        if name == "explicit_stop":
            body += frame({}, "stop")
        if name in ("explicit_stop", "done_only"):
            body += "data: [DONE]\n\n"
        body = body.encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.send_header("Content-Length", str(len(body) + (1000 if name == "transport_cut" else 0)))
        self.send_header("Connection", "close")
        self.end_headers()
        self.wfile.write(body)
        self.wfile.flush()
        self.close_connection = True


def request(base, method, path, payload=None):
    data = None if payload is None else json.dumps(payload).encode("utf-8")
    return http.open(Request(base + path, data, {"Content-Type": "application/json"}, method=method), timeout=30)


def main():
    binary = str(Path(sys.argv[1]).resolve())
    print(json.dumps({"platform": sys.platform,
                      "version": subprocess.check_output([binary, "version"], text=True).strip(),
                      "sha256": hashlib.sha256(Path(binary).read_bytes()).hexdigest()}), flush=True)
    provider = ThreadingHTTPServer(("127.0.0.1", 0), Provider)
    threading.Thread(target=provider.serve_forever, daemon=True).start()
    try:
        with tempfile.TemporaryDirectory(prefix="docker-agent-eof-") as root:
            with socket.socket() as reserve:
                reserve.bind(("127.0.0.1", 0))
                port = reserve.getsockname()[1]
            config = Path(root) / "repro.yaml"
            config.write_text("""version: '15'
providers:
  fixture:
    api_type: openai_chatcompletions
    base_url: http://127.0.0.1:%d/v1
    token_key: EOF_FIXTURE_TOKEN
models:
  fixture:
    provider: fixture
    model: fixture-model
    max_tokens: 128
agents:
  root:
    model: fixture
    instruction: Fixture only.
    max_iterations: 1
""" % provider.server_port, encoding="utf-8")
            env = dict(os.environ, EOF_FIXTURE_TOKEN="local-fixture-only", DOCKER_AGENT_DISABLE_DESKTOP_PROXY="1")
            command = [binary, "serve", "api", str(config), "--listen", "127.0.0.1:%d" % port,
                       "--session-db", str(Path(root) / "native.db"),
                       "--config-dir", str(Path(root) / "config"), "--data-dir", str(Path(root) / "data")]
            with open(Path(root) / "engine.log", "w") as log:
                process = subprocess.Popen(command, env=env, stdout=log, stderr=log,
                                           creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0)
                try:
                    base = "http://127.0.0.1:%d" % port
                    for _ in range(100):
                        try:
                            with request(base, "GET", "/api/ping"):
                                break
                        except OSError:
                            if process.poll() is not None:
                                raise RuntimeError("fixture engine exited")
                            time.sleep(0.1)
                    else:
                        raise RuntimeError("fixture engine did not start")
                    for name in CASES:
                        with request(base, "POST", "/api/sessions", {"title": "EOF fixture", "working_dir": root}) as response:
                            session = json.load(response)["id"]
                        with request(base, "POST", "/api/sessions/%s/agent/repro/root" % session,
                                     {"messages": [{"role": "user", "content": name}]}) as response:
                            lines = response.read().decode("utf-8").splitlines()
                        events = [json.loads(line[5:].strip()) for line in lines if line.startswith("data:") and line[5:].strip() != "[DONE]"]
                        stop = next(e for e in reversed(events) if e["type"] == "stream_stopped")
                        print(json.dumps({"case": name, "providerRequests": calls[name],
                                          "reason": stop.get("reason"), "finish_reason": stop.get("finish_reason"),
                                          "errors": [e.get("code") for e in events if e["type"] == "error"],
                                          "hasLastMessageUsage": any(e.get("usage", {}).get("last_message") is not None
                                                                     for e in events if e["type"] == "token_usage")}), flush=True)
                finally:
                    process.terminate()
                    try:
                        process.wait(timeout=5)
                    except subprocess.TimeoutExpired:
                        process.kill()
                        process.wait()
    finally:
        provider.shutdown()
        provider.server_close()


if __name__ == "__main__":
    main()

{"platform":"win32","version":"docker-agent version v1.150.0\nCommit: 5277e7842058fc3fbb9943a6e88bc05a1c8e6fe2","sha256":"144191a21a25c0c1ecfa8c0b320af00c8373187a27fccd73256a7fc39de3d6c3"}
{"case":"explicit_stop","providerRequests":1,"reason":"normal","finish_reason":"stop","errors":[],"hasLastMessageUsage":false}
{"case":"done_only","providerRequests":1,"reason":"normal","finish_reason":"stop","errors":[],"hasLastMessageUsage":false}
{"case":"bare_eof","providerRequests":1,"reason":"normal","finish_reason":"stop","errors":[],"hasLastMessageUsage":false}
{"case":"transport_cut","providerRequests":3,"reason":"error","finish_reason":null,"errors":["model_error"],"hasLastMessageUsage":false}

{"platform":"linux","version":"docker-agent version v1.150.0\nCommit: 5277e7842058fc3fbb9943a6e88bc05a1c8e6fe2","sha256":"791fa6bac9f829847f00ab948f838bf54d11e0830988a5e5bce88d65ea65f2d1"}
{"case":"explicit_stop","providerRequests":1,"reason":"normal","finish_reason":"stop","errors":[],"hasLastMessageUsage":false}
{"case":"done_only","providerRequests":1,"reason":"normal","finish_reason":"stop","errors":[],"hasLastMessageUsage":false}
{"case":"bare_eof","providerRequests":1,"reason":"normal","finish_reason":"stop","errors":[],"hasLastMessageUsage":false}
{"case":"transport_cut","providerRequests":3,"reason":"error","finish_reason":null,"errors":["model_error"],"hasLastMessageUsage":false}