ACP: end_turn precedes background-shell completion and autonomous follow-up tool calls; no observable session-idle signal

"""Reproduce late ACP continuation using only a short sleep and fixed output."""

import argparse
import json
import os
import selectors
import signal
import subprocess
import tempfile
import time


parser = argparse.ArgumentParser()
parser.add_argument("--cli", default="copilot")
parser.add_argument("--model", default="gpt-5.6-luna")
args = parser.parse_args()
command = "sleep 12; printf ACP_SHELL_DONE"

with tempfile.TemporaryDirectory(prefix="acp-background-repro-") as cwd:
    proc = subprocess.Popen(
        [args.cli, "--acp", "--no-auto-update", "--disable-builtin-mcps"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        cwd=cwd,
        start_new_session=True,
    )
    selector = selectors.DefaultSelector()
    selector.register(proc.stdout, selectors.EVENT_READ)
    buffer = b""
    responses = {}
    next_id = 0
    started = time.monotonic()
    prompt_id = None
    prompt_returned = None
    late_tools = 0
    final_seen = None
    assistant_text = ""

    def write(message):
        proc.stdin.write((json.dumps(message) + "\n").encode())
        proc.stdin.flush()

    def send(method, params):
        global next_id
        next_id += 1
        write({"jsonrpc": "2.0", "id": next_id, "method": method, "params": params})
        return next_id

    def report(kind, **fields):
        print(json.dumps({"seconds": round(time.monotonic() - started, 3),
                          "event": kind, **fields}), flush=True)

    def pump():
        global buffer, prompt_returned, late_tools, final_seen, assistant_text
        if not selector.select(1):
            return
        chunk = os.read(proc.stdout.fileno(), 65536)
        if not chunk:
            raise RuntimeError("ACP process exited")
        buffer += chunk
        while b"\n" in buffer:
            line, buffer = buffer.split(b"\n", 1)
            if not line.strip():
                continue
            message = json.loads(line)
            if "id" in message and "method" not in message:
                responses[message["id"]] = message
                if message["id"] == prompt_id:
                    prompt_returned = time.monotonic()
                    report("prompt_response", result=message.get("result"),
                           error=message.get("error"))
                continue
            params = message.get("params", {})
            if message.get("method") == "session/request_permission":
                tool = params.get("toolCall", {})
                requested = tool.get("rawInput", {}).get("command")
                allow = requested == command
                options = params.get("options", [])
                option = next(
                    (o for o in options if o.get("kind") ==
                     ("allow_once" if allow else "reject_once")), None
                )
                outcome = (
                    {"outcome": "selected", "optionId": option["optionId"]}
                    if option else {"outcome": "cancelled"}
                )
                report("permission", allowed=allow)
                write({"jsonrpc": "2.0", "id": message["id"],
                       "result": {"outcome": outcome}})
            elif message.get("method") == "session/update":
                update = params.get("update", {})
                kind = update.get("sessionUpdate")
                if kind == "tool_call":
                    if prompt_returned is not None:
                        late_tools += 1
                    report("tool_call", title=update.get("title"),
                           after_prompt_response=prompt_returned is not None)
                elif kind == "agent_message_chunk":
                    text = update.get("content", {}).get("text", "")
                    assistant_text += text
                    if text:
                        report("assistant_text", text=text,
                               after_prompt_response=prompt_returned is not None)
                    if "FINAL_DONE" in assistant_text:
                        final_seen = time.monotonic()

    def request(method, params):
        request_id = send(method, params)
        deadline = time.monotonic() + 30
        while request_id not in responses:
            if time.monotonic() > deadline:
                raise TimeoutError(method)
            pump()
        response = responses.pop(request_id)
        if "error" in response:
            raise RuntimeError(response["error"])
        return response["result"]

    try:
        initialization = request("initialize", {
            "protocolVersion": 1,
            "clientCapabilities": {},
        })
        report("initialize", agentInfo=initialization.get("agentInfo"),
               agentCapabilities=initialization.get("agentCapabilities"))
        session = request("session/new", {"cwd": cwd, "mcpServers": []})
        session_id = session["sessionId"]
        request("session/set_model", {"sessionId": session_id, "modelId": args.model})
        started = time.monotonic()
        prompt_id = send("session/prompt", {
            "sessionId": session_id,
            "prompt": [{
                "type": "text",
                "text": (
                    "Lifecycle experiment in an empty temporary directory. "
                    "Do not read or write files or access the network. "
                    f"Use bash to run exactly `{command}` with mode async "
                    "(not detached). Immediately reply WAITING without calling "
                    "read_bash yet. When the shell completion notification arrives, "
                    "read that shell result with read_bash and reply FINAL_DONE. "
                    "Do not run any other command or use other tools."
                ),
            }],
        })
        deadline = time.monotonic() + 90
        while time.monotonic() < deadline:
            pump()
            if final_seen is not None and time.monotonic() - final_seen > 2:
                break
        report("summary", prompt_returned=prompt_returned is not None,
               late_tool_calls=late_tools, final_seen=final_seen is not None)
    finally:
        selector.close()
        if proc.poll() is None:
            os.killpg(proc.pid, signal.SIGTERM)
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                os.killpg(proc.pid, signal.SIGKILL)
                proc.wait()