"""ACP over real stdio pipes and daemon admission, execution, and history.

Session tests use private daemons because ACP selects the global default provider.
"""

from __future__ import annotations

import http.server
import json
import os
import shutil
import signal
import sys
import tempfile
import threading
import time
import unittest
import urllib.error
from pathlib import Path
from urllib.parse import quote

from acp_fault_support import DaemonProxy
from acp_support import AcpClient
from provider_events import responses_events
from harness import (
    Albedo,
    Provider,
    Reply,
    current_daemon,
    exclusive,
    latest_user,
    operation_id,
    python,
    release_fifo,
    text,
    wait_until,
)

PROTOCOL = 1
# The literal env value an editor supplies for the overlay server. It must reach
# the server and never reach a file under ALBEDO_HOME or the daemon log.
SECRET = "acp-secret-value"
SERVER = Path(__file__).with_name("fake_mcp_server.py")


def answered(request):
    """Whether a request carries a tool result, so the scripted turn is past its tool call."""
    return any(
        item.get("role") == "tool" for item in request.get("messages", [])
    ) or any(
        item.get("type") == "function_call_output" for item in request.get("input", [])
    )


def tool_names(request):
    return [
        tool.get("name") or tool.get("function", {}).get("name")
        for tool in request.get("tools", [])
    ]


def overlay_tool(request):
    return next(name for name in tool_names(request) if name.startswith("mcp_overlay"))


def user_messages(request):
    """The user turns of a request as JSON, so a check can find each block's fields."""
    items = request.get("messages", request.get("input", []))
    return json.dumps([item for item in items if item.get("role") == "user"])


def chunks(updates, kind):
    return [update for update in updates if update["sessionUpdate"] == kind]


def chunk_text(updates, kind):
    return "".join(update["content"]["text"] for update in chunks(updates, kind))


def text_block(value):
    return {"type": "text", "text": value}


def link_block(uri, name):
    return {"type": "resource_link", "uri": uri, "name": name}


def position(client, message):
    """Where a message object sits in the adapter's stdout, by identity."""
    return next(
        index for index, item in enumerate(client.messages()) if item is message
    )


def gate_code(started, gate):
    """A python cell that announces it started, then blocks until the gate is released."""
    return (
        "import pathlib\n"
        f"pathlib.Path({str(started)!r}).touch()\n"
        f"open({str(gate)!r}, 'rb').read(1)\n"
    )


def background_code(gate):
    """A python cell that starts a background job which waits on the gate, then wakes the session."""
    program = f"open({str(gate)!r}, 'rb').read(1); print('wake-done')"
    return f"import sys\njob = run(sys.executable, '-c', {program!r})\njob.id"


def chat_event(delta, finish=None):
    return {
        "id": "fixture",
        "choices": [{"index": 0, "delta": delta, "finish_reason": finish}],
    }


def tool_call(index, call_id, code):
    return {
        "index": index,
        "id": call_id,
        "type": "function",
        "function": {
            "name": "python",
            "arguments": json.dumps({"code": code, "timeout_ms": 60000}),
        },
    }


def two_tool_calls(first, second):
    """One assistant message that makes two python tool calls at once."""
    events = [
        chat_event({"tool_calls": [tool_call(0, "call-a", first)]}),
        chat_event({"tool_calls": [tool_call(1, "call-b", second)]}),
        chat_event({}, "tool_calls"),
    ]
    return Reply("text", "", events=events)


def text_then_tool(value, code):
    """One assistant message that says ``value`` and then makes a python tool call."""
    events = [
        chat_event({"content": value}),
        chat_event({"tool_calls": [tool_call(0, "call-a", code)]}),
        chat_event({}, "tool_calls"),
    ]
    return Reply("text", "", events=events)


def frames(*messages):
    """JSON-RPC messages as one stdin write, so the adapter reads them in one batch."""
    return b"".join(json.dumps(message).encode() + b"\n" for message in messages)


def prompt_message(request_id, session, value):
    return {
        "jsonrpc": "2.0",
        "id": request_id,
        "method": "session/prompt",
        "params": {"sessionId": session, "prompt": [text_block(value)]},
    }


def cancel_message(session):
    return {
        "jsonrpc": "2.0",
        "method": "session/cancel",
        "params": {"sessionId": session},
    }


def wait_until_file(path, timeout):
    """Read a file once it exists, for a server that writes it as it starts."""
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if path.exists() and path.stat().st_size:
            return path.read_text()
        time.sleep(0.05)
    raise AssertionError(f"{path} was not written")


def holds(path, needle):
    """Whether a readable file contains `needle`. Credential backups are
    0000, so nothing else can read them either."""
    try:
        return path.is_file() and needle in path.read_bytes()
    except PermissionError:
        return False


def overlay_server(root):
    """An absolute executable that records its cwd and env, then runs the fake MCP server."""
    record = root / "overlay-record.json"
    script = root / "overlay-server"
    script.write_text(
        f"#!{sys.executable}\n"
        "import json, os, sys\n"
        "with open(%r, 'w') as file:\n"
        "    json.dump({'cwd': os.getcwd(), 'secret': os.environ.get('FAKE_SECRET'),\n"
        "               'ambient': os.environ.get('ALBEDO_MCP_AMBIENT')}, file)\n"
        "os.execv(sys.executable, [sys.executable, %r])\n" % (str(record), str(SERVER))
    )
    script.chmod(0o755)
    return script


class AcpCase(unittest.TestCase):
    """Shared helpers: an adapter per test, and the JSON-RPC results it returns."""

    def setUp(self):
        self.daemon = current_daemon()

    def connect(self, env=None):
        client = AcpClient(self.daemon.env if env is None else env)
        self.addCleanup(client.kill)
        return client

    def initialize(self, client):
        return self.result(
            client.request(
                "initialize", {"protocolVersion": PROTOCOL, "clientCapabilities": {}}
            )
        )

    def result(self, exchange):
        self.assertNotIn("error", exchange.response, exchange.response)
        return exchange.response["result"]

    def assertError(self, exchange, code, reason=None, *, codes=None):
        response = exchange.response
        self.assertNotIn("result", response, response)
        self.assertIn(response["error"]["code"], codes or (code,), response)
        if reason is not None:
            self.assertEqual(
                response["error"].get("data", {}).get("reason"), reason, response
            )
        return response["error"]


class AcpSessionCase(AcpCase):
    """A daemon of its own, a scripted model, and a workspace for one scenario."""

    albedo_options = {}

    def setUp(self):
        super().setUp()
        self.script = lambda request: text("ok")
        self.provider = Provider(lambda request: self.script(request))
        self.addCleanup(self.provider.close)
        self.app = Albedo(self.provider, **self.albedo_options)
        self.app.__enter__()
        self.addCleanup(self.app.__exit__, None, None, None)
        self.workspace = self.app.workspace

    def new_session(self, client, *, servers=(), cwd=None):
        result = self.result(
            client.request(
                "session/new",
                {"cwd": str(cwd or self.workspace), "mcpServers": list(servers)},
            )
        )
        return result["sessionId"]

    def send_prompt(self, client, session, *blocks):
        return client.send(
            "session/prompt", {"sessionId": session, "prompt": list(blocks)}
        )

    def prompt(self, client, session, *blocks):
        return client.response(self.send_prompt(client, session, *blocks), timeout=60)

    def stop_reason(self, exchange):
        return self.result(exchange)["stopReason"]

    def snapshot(self, session):
        with self.app.api(f"/sessions/{session}?tail=0") as response:
            return json.load(response)

    def acp_state(self, session):
        return self.snapshot(session)["acp"]

    def session_ids(self):
        with self.app.api("/sessions?scope=all") as response:
            return {item["id"] for item in json.load(response)["items"]}

    def history_json(self, session):
        return json.dumps(self.app.history(session))

    def wait_requests(self, count):
        wait_until(
            lambda: len(self.provider.requests) >= count,
            30,
            f"model request {count}",
        )

    def wait_detached(self, session, timeout=15):
        return self.wait_for_state(session, "detached", timeout)

    def wait_for_state(self, session, attachment, timeout):
        deadline = time.monotonic() + timeout
        state = None
        while time.monotonic() < deadline:
            state = self.acp_state(session)
            if state["attachment"] == attachment:
                return state
            time.sleep(0.1)
        self.fail(f"attachment is {state}, expected {attachment}")

    def hang_then(self, final):
        """The first model request never answers; later ones answer ``final``."""
        self.script = lambda request: (
            text("stalled", hang=True)
            if len(self.provider.requests) == 1
            else text(final)
        )

    def gate_turn(self, final):
        """A turn whose python tool blocks until the test releases the gate."""
        gate = self.workspace / "gate"
        started = self.workspace / "started"
        os.mkfifo(gate)
        code = gate_code(started, gate)
        self.script = lambda request: (
            python(code) if not answered(request) else text(final)
        )
        return gate, started

    def release(self, gate):
        wait_until(lambda: release_fifo(gate), 30, "the gated tool to start")

    def interrupt(self, session):
        """Stop the session's running turn from outside the adapter, as another client would."""
        captured = self.snapshot(session)
        self.app.api(
            f"/sessions/{session}/interrupt",
            {
                "run_id": captured["status"]["run_id"],
                "through_input_order": captured["input_order"],
            },
        ).close()

    def start_gated_prompt(self, client, session, final):
        gate, started = self.gate_turn(final)
        prompt_id = self.send_prompt(client, session, text_block("work"))
        wait_until(started.exists, 30, "the tool to start")
        client.wait_for(
            lambda: any(
                update["sessionUpdate"] == "tool_call"
                for update in client.updates(session)
            ),
            30,
            "the tool call update",
        )
        return gate, prompt_id


class AcpWireTests(AcpCase):
    """Framing and dispatch on the wire, before any session exists.

    These run on the shared daemon: none of them creates a session or contacts
    the daemon, so they cannot interfere with the other suites' sessions.
    """

    def test_initialize_answers_from_static_data_without_starting_a_daemon(self):
        home = Path(tempfile.mkdtemp(prefix="acp-home-"))
        self.addCleanup(shutil.rmtree, home, True)
        client = self.connect(env=dict(self.daemon.env, ALBEDO_HOME=str(home)))
        initialized = self.result(
            client.request("initialize", {"protocolVersion": 999})
        )
        self.assertEqual(initialized["protocolVersion"], PROTOCOL)
        capabilities = initialized["agentCapabilities"]
        self.assertTrue(capabilities["loadSession"])
        self.assertTrue(
            {"list", "resume"} <= capabilities["sessionCapabilities"].keys()
        )
        self.assertFalse(any(capabilities["promptCapabilities"].values()))
        self.assertFalse(any(capabilities["mcpCapabilities"].values()))
        self.assertEqual(initialized["authMethods"], [])
        # The adapter exits before close returns, and it answers initialize
        # without contacting the daemon, so nothing can start after this.
        client.close()
        # No daemon was discovered, started, or attached in the empty home.
        self.assertFalse((home / "daemon.json").exists())
        self.assertFalse((home / "daemon.log").exists())

    def test_fragmented_and_combined_lines_are_framed_by_newline_only(self):
        client = self.connect()
        client.send("initialize", {"protocolVersion": PROTOCOL}, id=7, fragment=5)
        self.assertEqual(client.response(7).response["id"], 7)
        lines = b"".join(
            json.dumps(
                {
                    "jsonrpc": "2.0",
                    "id": request_id,
                    "method": "initialize",
                    "params": {"protocolVersion": 1},
                }
            ).encode()
            + b"\n"
            for request_id in (900719925474099312345, "request-9")
        )
        client.expect(900719925474099312345)
        client.expect("request-9")
        client.write(lines)
        self.assertEqual(
            client.response(900719925474099312345).response["id"], 900719925474099312345
        )
        self.assertEqual(client.response("request-9").response["id"], "request-9")

    def test_malformed_json_returns_parse_error_with_null_id_and_keeps_connection(self):
        client = self.connect()
        client.write(b'{"jsonrpc": "2.0", "id": 1,\n')
        error = client.wait_for(
            lambda: next(
                (
                    message
                    for message in client.messages()
                    if message.get("error", {}).get("code") == -32700
                ),
                None,
            ),
            10,
            "a parse error",
        )
        self.assertIsNone(error["id"])
        self.initialize(client)

    def test_invalid_requests_return_errors_and_keep_the_connection(self):
        client = self.connect()
        for message, code in (
            ({"jsonrpc": "2.0", "id": 5}, -32600),
            ({"jsonrpc": "2.0", "id": 5, "method": "no/such-method"}, -32601),
            (
                {
                    "jsonrpc": "2.0",
                    "id": 5,
                    "method": "authenticate",
                    "params": {"methodId": "terminal"},
                },
                -32602,
            ),
        ):
            with self.subTest(message=message):
                response_id = None if code == -32600 else 5
                client.expect(response_id)
                client.write(frames(message))
                self.assertError(client.response(response_id), code)
        self.initialize(client)

    def test_unknown_notifications_get_no_response(self):
        client = self.connect()
        before = len(client.messages())
        client.notify("$/cancel_request", {"requestId": 1})
        client.notify("x-vendor/unknown", {"value": 1})
        exchange = client.request("authenticate", {"methodId": "x"})
        self.assertEqual(exchange.notifications, [])
        # The only new stdout message is the answer to the request.
        self.assertEqual(len(client.messages()), before + 1)


@exclusive
class AcpCreationTests(AcpSessionCase):
    """Validation of session/new: nothing is created when the request is refused."""

    def test_relative_cwd_is_rejected_before_any_session_exists(self):
        client = self.connect()
        self.initialize(client)
        before = self.session_ids()
        self.assertError(
            client.request(
                "session/new", {"cwd": "relative/project", "mcpServers": []}
            ),
            -32602,
        )
        self.assertEqual(self.session_ids(), before)

    def test_non_stdio_mcp_servers_are_rejected_before_any_session_exists(self):
        client = self.connect()
        self.initialize(client)
        before = self.session_ids()
        for kind in ("http", "sse"):
            with self.subTest(kind=kind):
                server = {
                    "type": kind,
                    "name": "remote",
                    "url": "https://example.invalid/mcp",
                    "headers": [],
                }
                self.assertError(
                    client.request(
                        "session/new",
                        {"cwd": str(self.workspace), "mcpServers": [server]},
                    ),
                    -32602,
                )
        self.assertEqual(self.session_ids(), before)


@exclusive
class AcpModelTests(AcpSessionCase):
    """The editor's choice reaches the provider without changing global defaults."""

    def setUp(self):
        def prepare(app):
            app.provider.catalog = {
                "data": [{"id": "fixture-model"}, {"id": "other-model"}]
            }
            app.providers = {
                name: {
                    "extension": "openai",
                    "baseUrl": app.provider.url,
                    "apiKey": "fixture-key",
                    "model": "fixture-model",
                    "protocol": "chat_completions",
                }
                for name in ("first", "second")
            }

        self.albedo_options = {"prepare": prepare}
        super().setUp()

    def test_model_choices_switch_only_the_session_and_survive_reattachment(self):
        client = self.connect()
        self.initialize(client)
        created = self.result(
            client.request(
                "session/new", {"cwd": str(self.workspace), "mcpServers": []}
            )
        )
        session = created["sessionId"]
        selector = created["configOptions"][0]
        self.assertEqual(selector["currentValue"], "first/fixture-model")
        self.assertIn("second/other-model", {m["value"] for m in selector["options"]})
        selection = {
            "sessionId": session,
            "configId": "model",
            "value": "second/other-model",
        }
        self.assertError(
            client.request(
                "session/set_config_option", {**selection, "value": "missing/model"}
            ),
            -32602,
        )
        changed = self.result(client.request("session/set_config_option", selection))
        self.assertEqual(
            changed["configOptions"][0]["currentValue"], selection["value"]
        )
        self.assertEqual(
            self.stop_reason(
                self.prompt(client, session, text_block("use the selected model"))
            ),
            "end_turn",
        )
        self.assertEqual(self.provider.requests[-1]["model"], "other-model")
        self.assertEqual(self.snapshot(session)["provider_profile"], "second")
        untouched = self.new_session(client)
        self.assertEqual(self.snapshot(untouched)["model"], "fixture-model")
        for method in ("session/resume", "session/load"):
            loaded = self.result(
                client.request(
                    method,
                    {
                        "sessionId": session,
                        "cwd": str(self.workspace),
                        "mcpServers": [],
                    },
                )
            )
            self.assertEqual(
                loaded["configOptions"][0]["currentValue"], selection["value"]
            )

    def test_only_unknown_model_write_outcomes_require_resume(self):
        proxy = DaemonProxy(self.daemon)
        self.addCleanup(proxy.close)
        client = self.connect(env=proxy.env)
        self.initialize(client)
        for phase in ("providers", "configuration", "refresh", "write"):
            with self.subTest(phase=phase):
                session = self.new_session(client)
                selection = {
                    "sessionId": session,
                    "configId": "model",
                    "value": "second/other-model",
                }
                path = (
                    "/settings"
                    if phase in ("providers", "refresh")
                    else f"/sessions/{session}?view=configuration"
                )
                failed = proxy.failed
                writes = sum(method == "PATCH" for method, _ in proxy.requests)
                proxy.fail(
                    "PATCH" if phase == "write" else "GET",
                    path,
                    skip=1 if phase == "refresh" else 0,
                    after_write=phase == "write",
                )
                response = client.request("session/set_config_option", selection)
                self.assertEqual(proxy.failed, failed + 1)
                if phase == "refresh":
                    selector = self.result(response)["configOptions"][0]
                    self.assertEqual(selector["currentValue"], selection["value"])
                    self.assertIn(
                        "first/fixture-model", {m["value"] for m in selector["options"]}
                    )
                else:
                    self.assertError(response, -32603)
                expected = (
                    "other-model" if phase in ("refresh", "write") else "fixture-model"
                )
                self.assertEqual(self.snapshot(session)["model"], expected)
                if phase == "write":
                    self.assertError(
                        self.prompt(client, session, text_block("must resume")),
                        -32603,
                        "observation_uncertain",
                    )
                    continue
                if phase != "refresh":
                    self.assertEqual(
                        sum(method == "PATCH" for method, _ in proxy.requests), writes
                    )
                self.assertEqual(
                    self.stop_reason(
                        self.prompt(client, session, text_block("still attached"))
                    ),
                    "end_turn",
                )
                self.assertEqual(self.provider.requests[-1]["model"], expected)
                switched = self.result(
                    client.request("session/set_config_option", selection)
                )
                self.assertEqual(
                    switched["configOptions"][0]["currentValue"], selection["value"]
                )

    def test_model_switch_is_refused_while_a_turn_runs(self):
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        gate, prompt = self.start_gated_prompt(client, session, "finished")
        self.assertError(
            client.request(
                "session/set_config_option",
                {
                    "sessionId": session,
                    "configId": "model",
                    "value": "second/other-model",
                },
            ),
            -32602,
            "busy",
        )
        self.release(gate)
        self.assertEqual(
            self.stop_reason(client.response(prompt, timeout=60)), "end_turn"
        )
        self.assertEqual(self.snapshot(session)["model"], "fixture-model")


@exclusive
class AcpNoProviderTests(AcpSessionCase):
    """A home with no default provider answers auth_required and stays usable."""

    albedo_options = {"providers": {}}

    def test_session_new_without_a_provider_returns_auth_required_and_stays_usable(
        self,
    ):
        client = self.connect()
        self.initialize(client)
        error = self.assertError(
            client.request(
                "session/new", {"cwd": str(self.workspace), "mcpServers": []}
            ),
            -32000,
            "provider_missing",
        )
        self.assertIn("albedo login", error["data"]["instruction"])
        self.initialize(client)
        self.assertError(
            client.request(
                "session/new", {"cwd": str(self.workspace), "mcpServers": []}
            ),
            -32000,
            "provider_missing",
        )


@exclusive
class AcpPromptTests(AcpSessionCase):
    """Prompt admission, updates, and stop reasons for turns that finish."""

    def test_text_prompt_streams_message_chunks_and_ends_turn(self):
        self.script = lambda request: text("hello from the model")
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        result = self.prompt(client, session, text_block("say hello"))
        self.assertEqual(self.stop_reason(result), "end_turn")
        updates = client.updates(session)
        self.assertEqual(
            chunk_text(updates, "agent_message_chunk"), "hello from the model"
        )
        message_ids = {
            update.get("messageId") for update in chunks(updates, "agent_message_chunk")
        }
        self.assertEqual(len(message_ids), 1)
        self.assertIsNotNone(next(iter(message_ids)))

    def test_no_update_for_the_turn_arrives_after_its_response(self):
        self.script = lambda request: text("done talking")
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        response = self.prompt(client, session, text_block("talk"))
        after = position(client, response.response) + 1
        # The adapter writes in order, so this answer comes after any late update.
        self.result(client.request("session/list", {"cwd": str(self.workspace)}))
        late = [
            message
            for message in client.messages()[after:]
            if message.get("method") == "session/update"
        ]
        self.assertEqual(late, [])

    def test_thinking_is_sent_as_agent_thought_chunks(self):
        self.script = lambda request: text("answer", reasoning="weighing options")
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        self.assertEqual(
            self.stop_reason(self.prompt(client, session, text_block("think"))),
            "end_turn",
        )
        updates = client.updates(session)
        self.assertEqual(chunk_text(updates, "agent_thought_chunk"), "weighing options")
        self.assertEqual(chunk_text(updates, "agent_message_chunk"), "answer")

    def test_resource_link_file_uri_reaches_the_model_as_decoded_path_and_original_uri(
        self,
    ):
        target = self.workspace / "with space" / "a+b 100%20.txt"
        target.parent.mkdir()
        target.write_text("notes")
        uri = "file://" + quote(str(target), safe="/+")
        self.script = lambda request: text("read it")
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        result = self.prompt(
            client,
            session,
            text_block("summarize"),
            {
                **link_block(uri, "a+b 100%20.txt"),
                "description": "editor supplied notes",
            },
        )
        self.assertEqual(self.stop_reason(result), "end_turn")
        sent = user_messages(self.provider.requests[0]["request"])
        self.assertIn(str(target), sent)
        self.assertIn(uri, sent)
        self.assertIn("editor supplied notes", sent)

    def test_invalid_file_uris_are_rejected_before_admission(self):
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        before = self.app.history(session)
        for uri in (
            "file://remote/tmp/a",
            "file:///tmp/%zz",
            "file:relative",
            "relative",
            "file://user@localhost/tmp/a",
            "file://localhost:1/tmp/a",
            "file:///tmp/a?",
            "file:///tmp/a#",
            "file:///tmp/%00",
            "file://localhost",
        ):
            with self.subTest(uri=uri):
                self.assertError(
                    client.request(
                        "session/prompt",
                        {
                            "sessionId": session,
                            "prompt": [text_block("hi"), link_block(uri, "x")],
                        },
                    ),
                    -32602,
                )
        self.assertEqual(self.app.history(session), before)
        self.assertEqual(len(self.provider.requests), 0)

    def test_non_file_uri_is_passed_as_a_reference_and_never_fetched(self):
        hits = []

        class Handler(http.server.BaseHTTPRequestHandler):
            def do_GET(self):
                hits.append(self.path)
                self.send_response(200)
                self.end_headers()
                self.wfile.write(b"fetched")

            def log_message(self, format, *args):
                pass

        server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        threading.Thread(target=server.serve_forever, daemon=True).start()
        self.addCleanup(server.server_close)
        self.addCleanup(server.shutdown)
        uri = f"http://127.0.0.1:{server.server_port}/fetch-me"
        self.script = lambda request: text("noted")
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        result = self.prompt(
            client, session, text_block("look"), link_block(uri, "fetch-me")
        )
        self.assertEqual(self.stop_reason(result), "end_turn")
        self.assertEqual(hits, [])
        self.assertIn(uri, user_messages(self.provider.requests[0]["request"]))

    def test_unsupported_blocks_are_rejected_with_nothing_admitted(self):
        blocks = {
            "negative resource size": {**link_block("file:///tmp/x", "x"), "size": -1},
            "image": {"type": "image", "mimeType": "image/png", "data": "iVBORw0KGgo="},
            "audio": {"type": "audio", "mimeType": "audio/wav", "data": "UklGRg=="},
            "embedded resource": {
                "type": "resource",
                "resource": {"uri": "file:///tmp/x.txt", "text": "inline"},
            },
        }
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        before = self.app.history(session)
        for name, block in blocks.items():
            with self.subTest(block=name):
                self.assertError(
                    client.request(
                        "session/prompt",
                        {"sessionId": session, "prompt": [text_block("hi"), block]},
                    ),
                    -32602,
                )
        self.assertEqual(self.app.history(session), before)
        self.assertEqual(len(self.provider.requests), 0)

    def test_tool_call_reports_one_stable_id_then_a_completed_update(self):
        self.script = lambda request: (
            text("done") if answered(request) else python("print('tool ran')")
        )
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        self.assertEqual(
            self.stop_reason(self.prompt(client, session, text_block("run"))),
            "end_turn",
        )
        calls = [
            update
            for update in client.updates(session)
            if update["sessionUpdate"] in ("tool_call", "tool_call_update")
        ]
        self.assertEqual(calls[0]["sessionUpdate"], "tool_call")
        self.assertEqual(len({update["toolCallId"] for update in calls}), 1)
        self.assertEqual(
            sum(update["sessionUpdate"] == "tool_call" for update in calls), 1
        )
        self.assertEqual(calls[-1]["sessionUpdate"], "tool_call_update")
        self.assertEqual(calls[-1]["status"], "completed")

    def test_concurrent_tool_calls_keep_separate_ids(self):
        self.script = lambda request: (
            text("both")
            if answered(request)
            else two_tool_calls("print('one')", "print('two')")
        )
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        self.assertEqual(
            self.stop_reason(self.prompt(client, session, text_block("both"))),
            "end_turn",
        )
        calls = [
            update
            for update in client.updates(session)
            if update["sessionUpdate"] in ("tool_call", "tool_call_update")
        ]
        announced = {
            update["toolCallId"]
            for update in calls
            if update["sessionUpdate"] == "tool_call"
        }
        final = {
            update["toolCallId"]: update["status"]
            for update in calls
            if update["sessionUpdate"] == "tool_call_update"
        }
        self.assertEqual(len(announced), 2)
        self.assertEqual(set(final), announced)
        self.assertEqual(set(final.values()), {"completed"})

    def test_empty_and_tool_only_turns_complete_without_a_text_reply(self):
        for program in (None, "x = 1"):
            with self.subTest(program=program):
                self.script = lambda request, program=program: (
                    text("")
                    if program is None or answered(request)
                    else python(program)
                )
                client = self.connect()
                self.initialize(client)
                session = self.new_session(client)
                exchange = self.prompt(client, session, text_block("compute"))
                self.assertEqual(self.stop_reason(exchange), "end_turn")

    def test_repeated_identical_text_and_separate_messages_are_all_delivered(self):
        self.script = lambda request: (
            text("same") if answered(request) else text_then_tool("same", "x = 1")
        )
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        self.assertEqual(
            self.stop_reason(self.prompt(client, session, text_block("twice"))),
            "end_turn",
        )
        updates = client.updates(session)
        self.assertEqual(chunk_text(updates, "agent_message_chunk"), "samesame")
        # Text before a tool call and text after it are two daemon messages.
        message_ids = {
            update["messageId"] for update in chunks(updates, "agent_message_chunk")
        }
        self.assertGreaterEqual(len(message_ids), 2)


@exclusive
class AcpCancelTests(AcpSessionCase):
    """Busy prompts, cancellation against queued and running turns, and the successor rule."""

    def test_second_prompt_while_one_is_active_returns_busy(self):
        self.hang_then("after")
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        first = self.send_prompt(client, session, text_block("first"))
        self.wait_requests(1)
        self.assertError(
            client.request(
                "session/prompt",
                {"sessionId": session, "prompt": [text_block("second")]},
            ),
            -32602,
            "busy",
        )
        client.notify("session/cancel", {"sessionId": session})
        self.assertEqual(self.stop_reason(client.response(first)), "cancelled")

    def test_cancel_still_returns_cancelled_when_the_daemon_crashes(self):
        gate = self.workspace / "cancel-gate"
        started = self.workspace / "cancel-started"
        os.mkfifo(gate)
        code = (
            "import signal, pathlib\n"
            "signal.signal(signal.SIGINT, signal.SIG_IGN)\n"
            f"pathlib.Path({str(started)!r}).touch()\n"
            f"open({str(gate)!r}, 'rb').read(1)\n"
        )
        self.script = lambda request: python(code)
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        pending = self.send_prompt(client, session, text_block("hold through cancel"))
        wait_until(started.exists, 30, "the cancellation-resistant tool to start")
        client.notify("session/cancel", {"sessionId": session})
        self.result(client.request("session/list", {"cwd": str(self.workspace)}))
        self.app.restart(crash=True)
        self.assertEqual(self.stop_reason(client.response(pending)), "cancelled")
        self.assertError(
            self.prompt(client, session, text_block("before resume")),
            -32603,
            "observation_uncertain",
        )
        self.result(
            client.request(
                "session/resume",
                {
                    "sessionId": session,
                    "cwd": str(self.workspace),
                    "mcpServers": [],
                },
            )
        )
        self.script = lambda request: text("recovered")
        self.assertEqual(
            self.stop_reason(self.prompt(client, session, text_block("after resume"))),
            "end_turn",
        )

    def test_cancel_while_queued_behind_a_system_turn_returns_cancelled(self):
        gate = self.workspace / "job-gate"
        os.mkfifo(gate)

        def script(request):
            count = len(self.provider.requests)
            if count == 1:
                return python(background_code(gate))
            if count == 2:
                return text("started")
            # The job's wake turn, which is non-ACP work, never finishes.
            return text("waking", hang=True)

        self.script = script
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        self.assertEqual(
            self.stop_reason(self.prompt(client, session, text_block("start job"))),
            "end_turn",
        )
        self.release(gate)
        self.wait_requests(3)
        queued = self.send_prompt(client, session, text_block("queued"))
        client.wait_for(
            lambda: self.snapshot(session)["pending_inputs"], 20, "the queued input"
        )
        self.assertEqual(len(self.provider.requests), 3)
        client.notify("session/cancel", {"sessionId": session})
        self.assertEqual(self.stop_reason(client.response(queued)), "cancelled")
        # Cancelling a queued input leaves the system turn running.
        self.assertEqual(len(self.provider.requests), 3)
        self.assertNotEqual(self.snapshot(session)["status"]["phase"], "idle")
        self.interrupt(session)
        self.app.idle(session)

    def test_cancel_after_completion_keeps_the_end_turn_outcome(self):
        self.script = lambda request: text("quick")
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        first = self.send_prompt(client, session, text_block("quick"))
        # The response is written only after the turn has ended, so the
        # completion is established before the cancel is sent.
        self.assertEqual(self.stop_reason(client.response(first)), "end_turn")
        self.app.idle(session)
        answered = len(client.messages())
        client.notify("session/cancel", {"sessionId": session})
        # The adapter reads messages in order, so this answer means the cancel was processed.
        self.result(client.request("session/list", {"cwd": str(self.workspace)}))
        late = [
            message
            for message in client.messages()[answered:]
            if message.get("method") == "session/update"
        ]
        self.assertEqual(late, [])
        self.assertEqual(len(self.provider.requests), 1)
        self.assertIn("quick", self.history_json(session))
        self.assertEqual(
            chunk_text(client.updates(session), "agent_message_chunk"), "quick"
        )

    def test_successor_prompt_after_cancel_is_admitted_only_after_the_predecessor_settles(
        self,
    ):
        self.hang_then("successor answer")
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        first = self.send_prompt(client, session, text_block("predecessor"))
        self.wait_requests(1)
        for _ in range(3):
            client.notify("session/cancel", {"sessionId": session})
        second = self.send_prompt(client, session, text_block("successor"))
        predecessor = client.response(first)
        successor = client.response(second)
        self.assertEqual(self.stop_reason(predecessor), "cancelled")
        self.assertEqual(self.stop_reason(successor), "end_turn")
        settled = position(client, predecessor.response)
        first_successor_chunk = min(
            index
            for index, message in enumerate(client.messages())
            if message.get("method") == "session/update"
            and message["params"]["update"]["sessionUpdate"] == "agent_message_chunk"
        )
        self.assertGreater(first_successor_chunk, settled)

        answers = [
            message
            for message in client.messages()
            if message.get("id") == first
            and ("result" in message or "error" in message)
        ]
        self.assertEqual(len(answers), 1)

    def test_third_prompt_while_a_successor_is_held_is_busy(self):
        # The predecessor's model request never answers, so only its cancel
        # ends the turn. The cancel, the successor, and the third prompt go out
        # in one write, so the adapter reads all three before the daemon can
        # settle the predecessor: the successor is held when the third arrives.
        self.hang_then("successor answer")
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        first = self.send_prompt(client, session, text_block("predecessor"))
        self.wait_requests(1)
        client.expect("successor")
        client.expect("third")
        client.write(
            frames(
                cancel_message(session),
                prompt_message("successor", session, "successor"),
                prompt_message("third", session, "third"),
            )
        )
        self.assertError(client.response("third"), -32602, "busy")
        self.assertEqual(self.stop_reason(client.response(first)), "cancelled")
        self.assertEqual(self.stop_reason(client.response("successor")), "end_turn")

    def test_cancel_before_the_successor_is_admitted_cancels_the_successor_too(self):
        # As above: the second cancel is read in the same batch as the successor,
        # so it arrives before the daemon can admit the successor.
        self.hang_then("never sent")
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        first = self.send_prompt(client, session, text_block("predecessor"))
        self.wait_requests(1)
        client.expect("successor")
        client.write(
            frames(
                cancel_message(session),
                prompt_message("successor", session, "successor"),
                cancel_message(session),
            )
        )
        self.assertEqual(self.stop_reason(client.response(first)), "cancelled")
        self.assertEqual(self.stop_reason(client.response("successor")), "cancelled")
        self.assertEqual(len(self.provider.requests), 1)
        self.app.idle(session)


@exclusive
class AcpAttachmentTests(AcpSessionCase):
    """Disconnects, the single attachment, resume from a second adapter, and other clients."""

    def test_closing_stdin_mid_turn_leaves_the_turn_running_and_detaches(self):
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        gate, _ = self.start_gated_prompt(client, session, "finished after disconnect")
        client.close()
        self.wait_detached(session)
        self.assertNotEqual(self.snapshot(session)["status"]["phase"], "idle")
        self.assertIn(session, self.session_ids())
        self.release(gate)
        self.app.idle(session)
        self.wait_requests(2)
        self.assertIn("finished after disconnect", self.history_json(session))

    def test_sigterm_releases_the_attachment_and_the_turn_continues(self):
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        gate, _ = self.start_gated_prompt(client, session, "finished after sigterm")
        client.signal(signal.SIGTERM)
        client.wait_exit()
        self.wait_detached(session)
        self.release(gate)
        self.app.idle(session)
        self.assertIn("finished after sigterm", self.history_json(session))

    def test_killed_adapter_keeps_the_attachment_until_its_lease_expires(self):
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        gate, _ = self.start_gated_prompt(client, session, "finished after kill")
        client.kill()
        # A killed adapter cannot release; the lease still holds the attachment.
        self.assertEqual(self.acp_state(session)["attachment"], "attached")
        self.wait_detached(session, timeout=60)
        self.release(gate)
        self.app.idle(session)
        self.assertIn("finished after kill", self.history_json(session))

    def test_a_second_adapter_is_refused_while_the_attachment_is_live(self):
        first = self.connect()
        self.initialize(first)
        session = self.new_session(first)
        second = self.connect()
        self.initialize(second)
        self.assertError(
            second.request(
                "session/resume",
                {"sessionId": session, "cwd": str(self.workspace), "mcpServers": []},
            ),
            -32602,
            "attachment_held",
            codes=(-32602, -32603),
        )
        self.assertEqual(
            self.stop_reason(self.prompt(first, session, text_block("still mine"))),
            "end_turn",
        )

    def test_resume_refuses_a_cwd_that_is_not_the_session_workspace(self):
        first = self.connect()
        self.initialize(first)
        session = self.new_session(first)
        self.assertError(
            first.request(
                "session/resume",
                {"sessionId": session, "cwd": str(self.app.root), "mcpServers": []},
            ),
            -32602,
            "workspace_mismatch",
            codes=(-32602, -32603),
        )

        self.assertEqual(
            self.stop_reason(self.prompt(first, session, text_block("still attached"))),
            "end_turn",
        )

    def test_resume_of_an_unknown_session_returns_resource_not_found(self):
        client = self.connect()
        self.initialize(client)
        self.assertError(
            client.request(
                "session/resume",
                {
                    "sessionId": operation_id(),
                    "cwd": str(self.workspace),
                    "mcpServers": [],
                },
            ),
            -32002,
        )

    def test_resume_accepts_an_omitted_or_null_mcp_server_list(self):
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        for optional in ({}, {"mcpServers": None}):
            with self.subTest(optional=optional):
                self.result(
                    client.request(
                        "session/resume",
                        {
                            "sessionId": session,
                            "cwd": str(self.workspace),
                            **optional,
                        },
                    )
                )
                self.assertEqual(
                    self.stop_reason(
                        self.prompt(client, session, text_block("resumed"))
                    ),
                    "end_turn",
                )

    def test_resume_of_an_idle_session_accepts_a_new_prompt(self):
        self.script = lambda request: text("first answer")
        first = self.connect()
        self.initialize(first)
        session = self.new_session(first)
        self.assertEqual(
            self.stop_reason(self.prompt(first, session, text_block("one"))), "end_turn"
        )
        first.close()
        second = self.connect()
        self.initialize(second)
        resumed = second.request(
            "session/resume",
            {"sessionId": session, "cwd": str(self.workspace), "mcpServers": []},
        )
        self.result(resumed)
        # Nothing from history is replayed on resume.
        self.assertEqual(resumed.notifications, [])
        self.assertEqual(second.updates(session), [])
        self.assertEqual(
            self.stop_reason(self.prompt(second, session, text_block("two"))),
            "end_turn",
        )
        self.assertEqual(
            chunk_text(second.updates(session), "agent_message_chunk"), "first answer"
        )

    def test_resume_during_a_running_turn_forwards_only_later_updates(self):
        first = self.connect()
        self.initialize(first)
        session = self.new_session(first)
        gate, _ = self.start_gated_prompt(first, session, "after resume")
        first.close()
        second = self.connect()
        self.initialize(second)
        resumed = second.request(
            "session/resume",
            {"sessionId": session, "cwd": str(self.workspace), "mcpServers": []},
        )
        self.result(resumed)
        self.assertEqual(resumed.notifications, [])
        time.sleep(0.5)
        # The tool started before the attach, so nothing of it is replayed while it runs.
        self.assertEqual(second.updates(session), [])
        self.release(gate)
        second.wait_for(
            lambda: any(
                update["sessionUpdate"] == "tool_call_update"
                and update["status"] == "completed"
                for update in second.updates(session)
            ),
            30,
            "the inherited tool to complete",
        )
        second.wait_for(
            lambda: (
                chunk_text(second.updates(session), "agent_message_chunk")
                == "after resume"
            ),
            30,
            "the inherited turn's text",
        )
        self.app.idle(session)
        # The second adapter never received a prompt response: none was synthesized.
        answered_ids = {
            message["id"]
            for message in second.messages()
            if "id" in message and "method" not in message
        }
        self.assertEqual(answered_ids, {1, 2})

    def test_resume_queues_one_prompt_until_the_inherited_turn_ends(self):
        first = self.connect()
        self.initialize(first)
        session = self.new_session(first)
        gate, _ = self.start_gated_prompt(first, session, "after busy")
        first.close()
        second = self.connect()
        self.initialize(second)
        self.result(
            second.request(
                "session/resume",
                {"sessionId": session, "cwd": str(self.workspace), "mcpServers": []},
            )
        )
        pending = self.send_prompt(second, session, text_block("next"))
        # This answer proves stdin has processed the successor while the
        # inherited turn is still held at the FIFO, not merely slow to finish.
        self.result(second.request("session/list", {"cwd": str(self.workspace)}))
        self.assertFalse(
            any(message.get("id") == pending for message in second.messages())
        )
        self.assertError(
            self.prompt(second, session, text_block("one too many")), -32602, "busy"
        )
        self.release(gate)
        self.assertEqual(self.stop_reason(second.response(pending)), "end_turn")
        self.assertIn("next", user_messages(self.provider.requests[-1]["request"]))
        self.assertNotIn("one too many", self.history_json(session))

    def test_resume_cancel_interrupts_the_inherited_turn(self):
        self.hang_then("never-finished")
        first = self.connect()
        self.initialize(first)
        session = self.new_session(first)
        self.send_prompt(first, session, text_block("stall"))
        self.wait_requests(1)
        first.close()
        second = self.connect()
        self.initialize(second)
        self.result(
            second.request(
                "session/resume",
                {"sessionId": session, "cwd": str(self.workspace), "mcpServers": []},
            )
        )
        queued = self.send_prompt(second, session, text_block("waiting"))
        second.notify("session/cancel", {"sessionId": session})
        self.assertEqual(self.stop_reason(second.response(queued)), "cancelled")
        self.app.idle(session, timeout=30)
        # The interrupted turn never reaches the model again.
        self.assertEqual(len(self.provider.requests), 1)
        self.assertNotIn("never-finished", self.history_json(session))

    def test_other_clients_cannot_submit_to_an_acp_session_but_can_read_and_stop_it(
        self,
    ):
        self.hang_then("after")
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        first = self.send_prompt(client, session, text_block("stall"))
        self.wait_requests(1)
        with self.assertRaises(urllib.error.HTTPError) as refused:
            self.app.prompt(session, "from another client")
        self.assertEqual(refused.exception.code, 409)
        self.assertIn("acp_owned", refused.exception.read().decode())
        self.assertIsNotNone(self.app.history(session))
        self.interrupt(session)
        # An interrupt from another client ends the turn as interrupted, which the adapter reports as cancelled.
        self.assertEqual(self.stop_reason(client.response(first)), "cancelled")
        self.app.idle(session)

    def test_slow_reader_overflow_closes_the_adapter_and_the_turn_completes(self):
        # Large enough to fill the adapter's 4 MiB outbound bound several times over.
        self.provider.chunk_size = 256 * 1024
        self.script = lambda request: text("x" * (6 * 1024 * 1024) + "END-MARKER")
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        client.pause_reading()
        self.send_prompt(client, session, text_block("long answer"))
        self.assertNotEqual(client.wait_exit(timeout=90), 0)
        self.app.idle(session, timeout=120)
        # The whole answer was committed; history hands back one that large
        # as a reference to its bytes.
        self.assertIn(
            f'"bytes": {6 * 1024 * 1024 + len("END-MARKER")}',
            self.history_json(session),
        )
        self.wait_detached(session)


@exclusive
class AcpMcpTests(AcpSessionCase):
    """The per-session MCP overlay: what the server sees, and what is never stored."""

    def overlay(self, client):
        closed = self.app.root / "overlay-closed"
        script = overlay_server(self.app.root)
        servers = [
            {
                "name": "overlay",
                "command": str(script),
                "args": [],
                "env": [
                    {"name": "FAKE_SECRET", "value": SECRET},
                    {"name": "FAKE_MCP_CLOSED", "value": str(closed)},
                ],
            }
        ]
        session = self.new_session(client, servers=servers)
        return session, self.app.root / "overlay-record.json"

    def call_overlay_tool(self):
        self.script = lambda request: (
            Reply(
                "python",
                tool_name=overlay_tool(request),
                tool_arguments={"message": "fail"},
            )
            if not answered(request)
            else text("done")
        )

    def test_stdio_server_accepts_null_args_and_env(self):
        client = self.connect()
        self.initialize(client)
        session = self.new_session(
            client,
            servers=[
                {
                    "name": "overlay",
                    "command": str(overlay_server(self.app.root)),
                    "args": None,
                    "env": None,
                }
            ],
        )
        self.call_overlay_tool()
        self.assertEqual(
            self.stop_reason(self.prompt(client, session, text_block("call tool"))),
            "end_turn",
        )
        self.assertTrue(
            any(
                name.startswith("mcp_overlay")
                for name in tool_names(self.provider.requests[0]["request"])
            )
        )

    def test_prompt_after_failed_reattach_returns_the_resume_remedy(self):
        client = self.connect()
        self.initialize(client)
        session, _ = self.overlay(client)
        self.assertEqual(
            self.stop_reason(self.prompt(client, session, text_block("warm"))),
            "end_turn",
        )
        broken_prompt = self.workspace / "SYSTEM.md"
        broken_prompt.write_bytes(b"\xff")
        attachment = {
            "sessionId": session,
            "cwd": str(self.workspace),
            "mcpServers": [],
        }
        self.assertError(
            client.request("session/load", attachment),
            -32603,
            "overlay_installed_reload_failed",
        )
        requested = len(self.provider.requests)
        failure = self.assertError(
            client.request(
                "session/prompt",
                {"sessionId": session, "prompt": [text_block("must be refused")]},
                timeout=5,
            ),
            -32603,
            "observation_uncertain",
        )
        self.assertIn("resume from the editor", failure["message"])
        self.assertEqual(len(self.provider.requests), requested)
        broken_prompt.unlink()
        self.result(client.request("session/resume", attachment))
        self.assertEqual(
            self.stop_reason(self.prompt(client, session, text_block("recovered"))),
            "end_turn",
        )

    def test_overlay_tool_is_visible_and_its_server_runs_in_the_workspace_with_the_literal_env(
        self,
    ):
        self.call_overlay_tool()
        client = self.connect()
        self.initialize(client)
        session, record_path = self.overlay(client)
        record = json.loads(wait_until_file(record_path, 30))
        self.assertEqual(record["cwd"], str(self.workspace))
        self.assertEqual(record["secret"], SECRET)
        self.assertIsNone(record["ambient"])
        self.assertEqual(
            self.stop_reason(
                self.prompt(client, session, text_block("use the overlay"))
            ),
            "end_turn",
        )
        self.assertTrue(
            any(
                name.startswith("mcp_overlay")
                for name in tool_names(self.provider.requests[0]["request"])
            )
        )
        # The call reached the server, which answered with its error.
        # The daemon reports an MCP error to the model without the server's text.
        self.assertIn(
            "MCP request failed", json.dumps(self.provider.requests[1]["request"])
        )
        self.app.idle(session)
        self.assertEqual(
            [
                str(path)
                for path in self.app.home.rglob("*")
                if holds(path, SECRET.encode())
            ],
            [],
        )


@exclusive
class AcpRestartTests(AcpSessionCase):
    """Restart: an interrupted prompt is never replayed, and resume attaches again."""

    def test_restart_never_replays_an_interrupted_prompt(self):
        self.hang_then("after restart")
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        self.send_prompt(client, session, text_block("INTERRUPTED-PROMPT"))
        self.wait_requests(1)
        client.close()
        requests_before = len(self.provider.requests)
        # A crash, so the stalled turn cannot hold up a graceful stop.
        self.app.restart(crash=True)
        second = self.connect()
        self.initialize(second)
        resumed = second.request(
            "session/resume",
            {"sessionId": session, "cwd": str(self.workspace), "mcpServers": []},
        )
        self.result(resumed)
        self.assertEqual(resumed.notifications, [])
        time.sleep(1)
        # Nothing ran after the restart: the prompt is not submitted again.
        self.assertEqual(len(self.provider.requests), requests_before)
        self.assertEqual(second.updates(session), [])

    def test_resume_after_restart_attaches_and_accepts_a_new_prompt(self):
        self.script = lambda request: text("fresh answer")
        first = self.connect()
        self.initialize(first)
        session = self.new_session(first)
        self.assertEqual(
            self.stop_reason(self.prompt(first, session, text_block("before restart"))),
            "end_turn",
        )
        first.close()
        self.app.restart()
        second = self.connect()
        self.initialize(second)
        self.result(
            second.request(
                "session/resume",
                {"sessionId": session, "cwd": str(self.workspace), "mcpServers": []},
            )
        )
        self.assertEqual(self.acp_state(session)["attachment"], "attached")
        self.assertEqual(
            self.stop_reason(self.prompt(second, session, text_block("fresh prompt"))),
            "end_turn",
        )
        self.assertIn(
            "fresh prompt", latest_user(self.provider.requests[-1]["request"])
        )
        self.assertEqual(
            chunk_text(second.updates(session), "agent_message_chunk"), "fresh answer"
        )


@exclusive
class AcpMultipleOutputTests(AcpSessionCase):
    """Live replacement identities must not collapse different output items."""

    albedo_options = {"protocol": "responses"}

    def test_hidden_provider_record_does_not_claim_the_streamed_reply(self):
        reply = "The real reply follows hidden provider state."
        events = list(responses_events(text(reply), "", "", 8))
        for event in events:
            if event["type"] == "response.output_text.delta":
                event["output_index"] = 1
        events[-1]["response"]["output"].insert(
            0,
            {
                "id": "hidden-reasoning",
                "type": "reasoning",
                "summary": [],
                "encrypted_content": "opaque-provider-state",
            },
        )
        self.script = lambda request: Reply("text", "", events=events)
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        self.assertEqual(
            self.stop_reason(self.prompt(client, session, text_block("reply"))),
            "end_turn",
        )
        self.assertEqual(
            chunk_text(client.updates(session), "agent_message_chunk"), reply
        )

    def test_multiple_messages_in_one_response_are_not_duplicated(self):
        output = []
        events = []
        for index, value in enumerate(["same", "same"]):
            item = {
                "id": f"message-{index}",
                "type": "message",
                "role": "assistant",
                "status": "completed",
                "content": [{"type": "output_text", "text": value, "annotations": []}],
            }
            events.append(
                {
                    "type": "response.output_text.delta",
                    "output_index": index,
                    "content_index": 0,
                    "delta": value,
                }
            )
            output.append(item)
        events.append(
            {
                "type": "response.completed",
                "response": {"id": "fixture", "status": "completed", "output": output},
            }
        )
        self.script = lambda request: Reply("text", "", events=events)
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        self.assertEqual(
            self.stop_reason(self.prompt(client, session, text_block("two messages"))),
            "end_turn",
        )
        self.assertEqual(
            chunk_text(client.updates(session), "agent_message_chunk"), "samesame"
        )


@exclusive
class AcpToolPresentationTests(AcpSessionCase):
    """Recorded tool evidence has the same presentation live and after load."""

    def start(self):
        client = self.connect()
        self.initialize(client)
        return client, self.new_session(client)

    def run_program(self, client, session, code):
        """Run ``code`` as the only tool call of one turn.

        Returns the call's final update and every update the turn sent.
        """
        before_requests = len(self.provider.requests)
        before_updates = len(client.updates(session))
        self.script = lambda request: (
            python(code)
            if len(self.provider.requests) == before_requests + 1
            else text("done")
        )
        self.assertEqual(
            self.stop_reason(self.prompt(client, session, text_block(code))),
            "end_turn",
        )
        new = client.updates(session)[before_updates:]
        finals = [
            update
            for update in new
            if update["sessionUpdate"] == "tool_call_update"
            and update["status"] in ("completed", "failed")
        ]
        self.assertEqual(len(finals), 1, new)
        return finals[0], new

    def test_fs_capabilities_are_ignored_and_file_tools_use_the_disk(self):
        # An editor that advertises fs callbacks gets no fs/* requests: the
        # file tools read and write the disk like any other session.
        (self.workspace / "notes.txt").write_text("disk text\n")
        client = self.connect()
        self.result(
            client.request(
                "initialize",
                {
                    "protocolVersion": PROTOCOL,
                    "clientCapabilities": {
                        "fs": {"readTextFile": True, "writeTextFile": True}
                    },
                },
            )
        )
        session = self.new_session(client)
        update, _ = self.run_program(
            client,
            session,
            "print(files.read('notes.txt'))\nfiles.write('notes.txt', 'written text\\n')",
        )
        self.assertEqual(update["status"], "completed")
        self.assertIn("disk text", json.dumps(update))
        self.assertEqual((self.workspace / "notes.txt").read_text(), "written text\n")
        self.assertEqual(client.requests_from_adapter(), [])

    def test_replayed_tool_calls_equal_the_live_updates(self):
        (self.workspace / "binary").write_bytes(bytes([0, 255]))
        client, session = self.start()
        programs = [
            "files.write('created.txt', 'one\\n')",
            "files.edit('created.txt', 'one', 'two')",
            "print(" + repr("x" * (300 * 1024)) + ")",
            "files.write('binary', 'text')",
            "raise ValueError('tool-failure' + 'x' * (300 * 1024))",
        ]
        turns = [self.run_program(client, session, code) for code in programs]
        live = [final for final, _ in turns]
        for update, old_text, new_text in (
            (live[0], None, "one\n"),
            (live[1], "one\n", "two\n"),
        ):
            diffs = [part for part in update["content"] if part["type"] == "diff"]
            self.assertEqual(len(diffs), 1, update)
            self.assertTrue(diffs[0]["path"].endswith("created.txt"))
            self.assertEqual(diffs[0]["oldText"], old_text)
            self.assertEqual(diffs[0]["newText"], new_text)
        self.assertTrue(all(update["status"] == "completed" for update in live[:-1]))
        self.assertIn(
            "[albedo: tool output truncated to fit 256 KiB]", json.dumps(live[2])
        )
        for update in [update for _, updates in turns for update in updates]:
            self.assertLessEqual(
                len(json.dumps(update, separators=(",", ":")).encode()), 256 * 1024
            )
        self.assertFalse(any(part["type"] == "diff" for part in live[3]["content"]))
        self.assertTrue(
            any(
                Path(location["path"]).name == "binary"
                for location in live[3]["locations"]
            )
        )
        self.assertEqual(live[-1]["status"], "failed")
        self.assertIn("tool-failure", json.dumps(live[-1]))
        failure = next(
            row
            for row in self.app.history(session)["items"]
            if row["kind"] == "tool_result" and row["tool"]["status"] == "failed"
        )
        self.assertTrue(
            any(
                part["kind"] == "reference" and part["reference"]["field"] == "result"
                for part in failure["content"]
            ),
            failure,
        )
        client.close()
        self.wait_detached(session)
        replay = self.connect()
        self.initialize(replay)
        self.result(
            replay.request(
                "session/load",
                {"sessionId": session, "cwd": str(self.workspace), "mcpServers": []},
            )
        )
        loaded = [
            update
            for update in replay.updates(session)
            if update["sessionUpdate"] == "tool_call_update"
            and update["status"] in ("completed", "failed")
        ]
        self.assertEqual(loaded, live)


@exclusive
class AcpLoadTests(AcpSessionCase):
    """History replay is bounded by the attachment capture and never owns listing."""

    def test_list_filters_pages_and_rejects_bad_cursor_without_attaching(self):
        client = self.connect()
        self.initialize(client)
        sessions = {self.new_session(client) for _ in range(51)}
        other = self.new_session(client, cwd=self.app.root)
        self.app.session()
        observer = self.connect()
        self.initialize(observer)
        first = self.result(
            observer.request("session/list", {"cwd": str(self.workspace)})
        )
        self.assertEqual(len(first["sessions"]), 50)
        second = self.result(
            observer.request(
                "session/list",
                {"cwd": str(self.workspace), "cursor": first["nextCursor"]},
            )
        )
        listed = first["sessions"] + second["sessions"]
        self.assertEqual({item["sessionId"] for item in listed}, sessions)
        self.assertTrue(all(item["cwd"] == str(self.workspace) for item in listed))
        self.assertError(
            observer.request("session/list", {"cursor": "invalid"}), -32602
        )
        self.assertError(
            observer.request(
                "session/load",
                {"sessionId": other, "cwd": str(self.app.root), "mcpServers": []},
            ),
            -32602,
            "attachment_held",
        )
        self.assertEqual(self.acp_state(other)["attachment"], "attached")

    def test_load_keeps_answer_when_page_ends_inside_provider_row(self):
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        for _ in range(49):
            self.assertEqual(
                self.stop_reason(self.prompt(client, session, text_block("same"))),
                "end_turn",
            )
        self.script = lambda request: Reply(
            "text",
            "",
            events=[
                chat_event({"reasoning_content": "boundary-thought"}),
                chat_event({"content": "boundary-answer"}),
                chat_event({}, "stop"),
            ],
        )
        self.assertEqual(
            self.stop_reason(self.prompt(client, session, text_block("last"))),
            "end_turn",
        )
        with self.app.api(f"/sessions/{session}/history?after=0&limit=100") as response:
            first = json.load(response)
        with self.app.api(
            f"/sessions/{session}/history?limit=100&next={quote(first['newer'])}"
        ) as response:
            second = json.load(response)
        self.assertEqual(len(first["items"]), 100)
        self.assertEqual(first["items"][-1]["kind"], "thinking")
        self.assertEqual(second["items"][0]["kind"], "assistant")
        self.assertEqual(first["items"][-1]["position"], second["items"][0]["position"])
        client.close()
        self.wait_detached(session)
        loaded = self.connect()
        self.initialize(loaded)
        self.result(
            loaded.request(
                "session/load",
                {"sessionId": session, "cwd": str(self.workspace), "mcpServers": []},
            )
        )
        replay = loaded.updates(session)
        self.assertEqual(
            chunk_text(replay, "agent_message_chunk"), "ok" * 49 + "boundary-answer"
        )
        self.assertEqual(chunk_text(replay, "agent_thought_chunk"), "boundary-thought")
        self.assertEqual(chunk_text(replay, "user_message_chunk"), "same" * 49 + "last")

    def test_load_replays_repeated_text_then_observes_a_running_turn(self):
        client = self.connect()
        self.initialize(client)
        session = self.new_session(client)
        # Each turn contributes user and assistant rows, crossing the 100-row page.
        for _ in range(51):
            self.assertEqual(
                self.stop_reason(self.prompt(client, session, text_block("same"))),
                "end_turn",
            )
        gate, _ = self.start_gated_prompt(client, session, "after-load")
        client.close()
        self.wait_detached(session)
        loaded = self.connect()
        self.initialize(loaded)
        exchange = loaded.request(
            "session/load",
            {"sessionId": session, "cwd": str(self.workspace), "mcpServers": []},
        )
        self.result(exchange)
        replay = loaded.updates(session)
        self.assertEqual(chunk_text(replay, "agent_message_chunk"), "ok" * 51)
        self.assertEqual(chunk_text(replay, "user_message_chunk"), "same" * 51 + "work")
        successor = self.send_prompt(loaded, session, text_block("after load"))
        self.release(gate)
        loaded.wait_for(
            lambda: chunk_text(loaded.updates(session), "agent_message_chunk").endswith(
                "after-load"
            ),
            30,
            "live text after load",
        )
        self.assertEqual(self.stop_reason(loaded.response(successor)), "end_turn")
        self.assertEqual(
            chunk_text(loaded.updates(session), "agent_message_chunk"),
            "ok" * 51 + "after-load" * 2,
        )

    def test_commit_racing_load_is_delivered_exactly_once(self):
        first = self.connect()
        self.initialize(first)
        session = self.new_session(first)
        gate, _ = self.start_gated_prompt(first, session, "concurrent-commit")
        first.close()
        self.wait_detached(session)
        loaded = self.connect()
        self.initialize(loaded)
        request = loaded.send(
            "session/load",
            {"sessionId": session, "cwd": str(self.workspace), "mcpServers": []},
        )
        self.release(gate)
        self.result(loaded.response(request))
        loaded.wait_for(
            lambda: (
                "concurrent-commit"
                in chunk_text(loaded.updates(session), "agent_message_chunk")
            ),
            30,
            "the concurrent commit",
        )
        self.app.idle(session)
        self.result(loaded.request("session/list", {}))
        self.assertEqual(
            chunk_text(loaded.updates(session), "agent_message_chunk"),
            "concurrent-commit",
        )
        tools = [
            update
            for update in loaded.updates(session)
            if update["sessionUpdate"] == "tool_call_update"
            and update["status"] == "completed"
        ]
        self.assertEqual(len(tools), 1)