"""Direct HTTP checks for ACP authorization, admission, restart, and MCP state.

Adapter-visible behavior belongs in acp_test.py.
"""

import base64
import json
from pathlib import Path
import secrets
import sqlite3
import sys
import threading
import time
import unittest
import urllib.error

from harness import (
    Albedo,
    Provider,
    Reply,
    exclusive,
    latest_user,
    operation_id,
    python,
    text,
    wait_until,
)

from inspect_support import Inspection

SERVER = Path(__file__).with_name("fake_mcp_server.py")
SECRET = "editor-literal-secret-4f1c"


def fake_server(name="editor", secret=SECRET):
    return {
        "name": name,
        "command": sys.executable,
        "args": [str(SERVER)],
        "env": [
            {"name": "FAKE_SECRET", "value": secret},
            {"name": "FAKE_REPORT_CWD", "value": "1"},
        ],
    }


def token():
    return base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b"=").decode()


class AcpClient:
    """One ACP connection's view of the daemon: its token and its sessions."""

    def __init__(self, app):
        self.app = app
        self.token = token()

    def headers(self, extra=None):
        return {"Albedo-Attachment-Token": self.token, **(extra or {})}

    def create(self, session_id=None, workspace=None, servers=()):
        session_id = session_id or operation_id()
        with self.app.api(
            f"/sessions/{session_id}",
            {
                "kind": "acp",
                "workspace": str(workspace or self.app.workspace),
                "mcp_servers": list(servers),
            },
            method="PUT",
            headers=self.headers({"If-None-Match": "*"}),
        ) as response:
            created = json.load(response)
        return session_id, created

    def attach(self, session_id, workspace=None, servers=()):
        with self.app.api(
            f"/sessions/{session_id}/acp-attachment",
            {
                "workspace": str(workspace or self.app.workspace),
                "mcp_servers": list(servers),
            },
            method="PUT",
            headers=self.headers(),
        ) as response:
            return json.load(response)

    def renew(self, session_id):
        with self.app.api(
            f"/sessions/{session_id}/acp-attachment/renew",
            {},
            method="POST",
            headers=self.headers(),
        ) as response:
            return response.status

    def release(self, session_id):
        with self.app.api(
            f"/sessions/{session_id}/acp-attachment",
            method="DELETE",
            headers=self.headers(),
        ) as response:
            return response.status

    def prompt(self, session_id, blocks, input_id=None):
        input_id = input_id or operation_id()
        with self.app.api(
            f"/sessions/{session_id}/inputs/{input_id}",
            {"kind": "acp_prompt", "blocks": blocks},
            method="PUT",
            headers=self.headers(),
        ) as response:
            return input_id, json.load(response)


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


def stored_with(home, value):
    """Files under the daemon's home that hold `value`. Unreadable files
    (credential backups are 0000) cannot leak to another reader either."""
    found = []
    for path in home.rglob("*"):
        try:
            if path.is_file() and value.encode() in path.read_bytes():
                found.append(str(path))
        except PermissionError:
            continue
    return found


def refused(test, call, status, code):
    with test.assertRaises(urllib.error.HTTPError) as failure:
        call()
    test.assertEqual(failure.exception.code, status)
    test.assertEqual(json.load(failure.exception)["code"], code)


class AcpAttachmentTest(unittest.TestCase):
    def setUp(self):
        self.app = Albedo()
        self.app.__enter__()
        self.addCleanup(self.app.__exit__, None, None, None)

    def test_creation_records_marker_and_attaches_its_token(self):
        editor = AcpClient(self.app)
        session_id, created = editor.create()
        self.assertEqual(created["acp"]["attachment"], "attached")
        self.assertEqual(
            created["creation"]["submitted"],
            {"kind": "acp", "workspace": str(self.app.workspace), "mcp_servers": []},
        )
        listed = session_read(self.app, session_id)
        self.assertEqual(listed["acp"]["attachment"], "attached")

    def test_creation_resend_recovers_and_changed_intent_conflicts(self):
        editor = AcpClient(self.app)
        session_id, first = editor.create()
        _, again = editor.create(session_id)
        self.assertEqual(again["id"], first["id"])
        other = self.app.workspace / "other"
        other.mkdir()
        refused(
            self,
            lambda: editor.create(session_id, workspace=other),
            409,
            "creation_conflict",
        )

    def test_creation_requires_a_token(self):
        refused(
            self,
            lambda: self.app.api(
                f"/sessions/{operation_id()}",
                {
                    "kind": "acp",
                    "workspace": str(self.app.workspace),
                    "mcp_servers": [],
                },
                method="PUT",
                headers={"If-None-Match": "*"},
            ),
            400,
            "attachment_token_required",
        )

    def test_competing_attachment_is_refused_until_released(self):
        first, second = AcpClient(self.app), AcpClient(self.app)
        session_id, _ = first.create()
        refused(self, lambda: second.attach(session_id), 409, "attachment_held")
        refused(self, lambda: second.renew(session_id), 409, "attachment_lost")
        first.attach(session_id)
        self.assertEqual(first.renew(session_id), 204)
        self.assertEqual(first.release(session_id), 204)
        self.assertEqual(first.release(session_id), 204)
        self.assertEqual(
            session_read(self.app, session_id)["acp"]["attachment"], "detached"
        )
        second.attach(session_id)
        self.assertEqual(
            session_read(self.app, session_id)["acp"]["attachment"], "attached"
        )

    def test_attach_checks_marker_and_exact_workspace(self):
        editor = AcpClient(self.app)
        plain = self.app.session()
        refused(self, lambda: editor.attach(plain), 409, "not_acp_session")
        session_id, _ = editor.create()
        other = self.app.workspace / "nested"
        other.mkdir()
        refused(
            self,
            lambda: editor.attach(session_id, workspace=other),
            409,
            "workspace_mismatch",
        )
        refused(self, lambda: editor.attach(operation_id()), 404, "session_not_found")

    def test_other_clients_read_but_cannot_drive_an_acp_session(self):
        editor = AcpClient(self.app)
        session_id, _ = editor.create()
        refused(
            self,
            lambda: self.app.prompt(session_id, "from the terminal"),
            409,
            "acp_owned",
        )
        refused(
            self,
            lambda: self.app.api(
                f"/sessions/{session_id}/inputs/{operation_id()}",
                {"kind": "acp_prompt", "blocks": [{"type": "text", "text": "hi"}]},
                method="PUT",
            ),
            409,
            "acp_owned",
        )
        stranger = AcpClient(self.app)
        refused(
            self,
            lambda: stranger.prompt(session_id, [{"type": "text", "text": "hi"}]),
            409,
            "attachment_lost",
        )
        refused(
            self,
            lambda: self.app.api(
                f"/sessions/{session_id}?view=configuration",
                {"model": "fixture-model"},
                method="PATCH",
            ),
            409,
            "acp_owned",
        )
        refused(
            self,
            lambda: self.app.api(f"/sessions/{session_id}/reload", {}, method="POST"),
            409,
            "acp_owned",
        )
        with self.app.api(f"/sessions/{session_id}?view=configuration") as response:
            etag = response.headers["ETag"]
        refused(
            self,
            lambda: self.app.api(
                f"/sessions/{session_id}?view=configuration",
                {
                    "model": "fixture-model",
                    "selection": {"extensions": {"files": False}},
                },
                method="PATCH",
                headers=editor.headers({"If-Match": etag}),
            ),
            409,
            "acp_owned",
        )
        refused(
            self,
            lambda: self.app.api(
                f"/sessions/{session_id}?view=configuration",
                {"model": "fixture-model"},
                method="PATCH",
                headers=stranger.headers({"If-Match": etag}),
            ),
            409,
            "attachment_lost",
        )
        with self.app.api(
            f"/sessions/{session_id}?view=configuration",
            {"name": "renamed by the operator"},
            method="PATCH",
            headers={"If-Match": etag},
        ):
            pass
        self.assertEqual(
            session_read(self.app, session_id)["name"], "renamed by the operator"
        )

    def test_acp_prompt_is_refused_outside_acp_sessions(self):
        editor = AcpClient(self.app)
        plain = self.app.session()
        refused(
            self,
            lambda: editor.prompt(plain, [{"type": "text", "text": "hi"}]),
            409,
            "not_acp_session",
        )

    def test_resource_links_reject_invalid_uris_and_unknown_path(self):
        editor = AcpClient(self.app)
        session_id, _ = editor.create()
        bad = [
            {
                "type": "resource_link",
                "uri": "file:///tmp/a",
                "name": "a",
                "path": "/tmp/b",
            },
            {
                "type": "resource_link",
                "uri": "file://remote/tmp/a",
                "name": "a",
            },
            {
                "type": "resource_link",
                "uri": "file:///tmp/%zz",
                "name": "a",
            },
            {
                "type": "resource_link",
                "uri": "file:///tmp/a",
                "name": "a",
                "size": -1,
            },
            {"type": "image", "data": "", "mime_type": "image/png"},
        ]
        for block in bad:
            with self.subTest(block=block):
                refused(
                    self,
                    lambda: editor.prompt(session_id, [block]),
                    400,
                    "invalid_request",
                )

    def test_token_never_reaches_responses_or_disk(self):
        editor = AcpClient(self.app)
        session_id, created = editor.create()
        attached = editor.attach(session_id)
        self.assertNotIn(editor.token, json.dumps(created))
        self.assertNotIn(editor.token, json.dumps(attached))
        self.assertNotIn(editor.token, json.dumps(session_read(self.app, session_id)))
        self.assertEqual(stored_with(self.app.home, editor.token), [])


def fast_leases(app):
    app.env["ALBEDO_TEST_TIME_SCALE"] = "100"


@exclusive
class AcpLeaseExpiryTest(unittest.TestCase):
    def test_expired_lease_drops_the_writer_only(self):
        with Albedo(prepare=fast_leases) as app:
            first, second = AcpClient(app), AcpClient(app)
            session_id, _ = first.create()
            lease = first.attach(session_id)["lease"]
            self.assertLessEqual(lease["lease_ms"], 1000)
            time.sleep(lease["lease_ms"] / 1000 + 0.3)
            refused(self, lambda: first.renew(session_id), 409, "attachment_lost")
            self.assertEqual(
                session_read(app, session_id)["acp"]["attachment"], "detached"
            )
            second.attach(session_id)


class Gate:
    """A provider that answers each request only once the test opens it."""

    def __init__(self, reply="done"):
        self.opened = threading.Event()
        self.requests = []
        self.reply = reply
        self.provider = Provider(self.answer)

    def answer(self, request):
        self.requests.append(request)
        self.opened.wait(30)
        return text(self.reply)


@exclusive
class AcpTurnTest(unittest.TestCase):
    def setUp(self):
        self.gate = Gate()
        self.addCleanup(self.gate.provider.close)
        self.addCleanup(self.gate.opened.set)
        self.app = Albedo(self.gate.provider, protocol="responses")
        self.app.__enter__()
        self.addCleanup(self.app.__exit__, None, None, None)
        self.editor = AcpClient(self.app)
        self.session_id, _ = self.editor.create()

    def receipt(self, input_id):
        with self.app.api(f"/sessions/{self.session_id}/inputs/{input_id}") as response:
            return json.load(response)

    def test_attach_captures_the_running_turn_and_later_prompts_wait(self):
        first, _ = self.editor.prompt(
            self.session_id, [{"type": "text", "text": "one"}]
        )
        wait_until(
            lambda: self.gate.requests, 30, "the first turn never reached the model"
        )
        second, _ = self.editor.prompt(
            self.session_id, [{"type": "text", "text": "two"}]
        )
        attached = self.editor.attach(self.session_id)
        self.assertEqual(attached["active_turn"]["acp_input_id"], first)
        self.assertEqual(attached["pending_input_ids"], [second])
        captured_history = self.app.history(self.session_id)["items"]
        self.assertEqual(
            attached["history_cutoff"],
            max((row["position"] for row in captured_history), default=0),
        )
        self.gate.opened.set()
        self.app.idle(self.session_id)
        later = self.app.stream_page(
            self.session_id,
            {
                "generation": attached["cursor"]["generation"],
                "cursor": attached["cursor"]["sequence"],
            },
        )["events"]
        later_positions = {
            event["data"]["entry"]["position"]
            for event in later
            if event["type"] == "message"
        }
        final_history = self.app.history(self.session_id)["items"]
        self.assertTrue(later_positions)
        self.assertTrue(
            all(
                any(row["position"] == position for row in final_history)
                for position in later_positions
            )
        )
        self.assertTrue(
            all(position > attached["history_cutoff"] for position in later_positions)
        )
        self.assertNotEqual(
            self.receipt(first)["turn"]["id"], self.receipt(second)["turn"]["id"]
        )

    def test_scoped_cancellation_requires_the_current_attachment(self):
        running, _ = self.editor.prompt(
            self.session_id, [{"type": "text", "text": "running"}]
        )
        wait_until(lambda: self.gate.requests, 30, "turn never reached the provider")
        queued, _ = self.editor.prompt(
            self.session_id, [{"type": "text", "text": "queued"}]
        )
        replacement = AcpClient(self.app)

        def cancel(input_id, headers=None):
            with self.app.api(
                f"/sessions/{self.session_id}/inputs/{input_id}/cancel",
                {},
                method="POST",
                headers=headers,
            ) as response:
                return json.load(response)["result"]

        for headers in (None, replacement.headers()):
            refused(self, lambda: cancel(running, headers), 409, "attachment_lost")
            refused(self, lambda: cancel(queued, headers), 409, "attachment_lost")
        self.editor.release(self.session_id)
        replacement.attach(self.session_id)
        refused(
            self, lambda: cancel(running, self.editor.headers()), 409, "attachment_lost"
        )
        self.assertEqual(cancel(queued, replacement.headers()), "cancelled")
        # Even a terminal receipt requires the current writer's token.
        refused(
            self, lambda: cancel(queued, self.editor.headers()), 409, "attachment_lost"
        )
        self.assertEqual(cancel(queued, replacement.headers()), "not_pending")
        self.assertEqual(cancel(running, replacement.headers()), "interrupt_requested")
        self.app.idle(self.session_id)
        self.assertEqual(self.receipt(queued)["delivery"], "cancelled")
        self.assertEqual(len(self.gate.requests), 1)

    def test_restart_cancels_waiting_prompts_and_does_not_continue_the_turn(self):
        first, _ = self.editor.prompt(
            self.session_id, [{"type": "text", "text": "one"}]
        )
        wait_until(
            lambda: self.gate.requests, 30, "the first turn never reached the model"
        )
        second, _ = self.editor.prompt(
            self.session_id, [{"type": "text", "text": "two"}]
        )
        asked = len(self.gate.requests)
        self.app.restart(crash=True)
        self.gate.opened.set()
        self.assertEqual(self.receipt(second)["delivery"], "cancelled")
        self.assertEqual(self.receipt(first)["turn"]["state"], "abandoned")
        listed = session_read(self.app, self.session_id)
        self.assertEqual(listed["acp"]["attachment"], "detached")
        time.sleep(1)
        self.assertEqual(len(self.gate.requests), asked)
        refused(
            self,
            lambda: self.editor.prompt(
                self.session_id, [{"type": "text", "text": "three"}]
            ),
            409,
            "attachment_lost",
        )
        self.editor.attach(self.session_id)
        self.editor.prompt(self.session_id, [{"type": "text", "text": "three"}])
        self.app.idle(self.session_id)


class AcpOverlayValidationTest(unittest.TestCase):
    def setUp(self):
        self.app = Albedo()
        self.app.__enter__()
        self.addCleanup(self.app.__exit__, None, None, None)

    def test_overlay_names_must_be_unique(self):
        editor = AcpClient(self.app)
        refused(
            self,
            lambda: editor.create(servers=[fake_server("twin"), fake_server("twin")]),
            409,
            "mcp_overlay_conflict",
        )

    def test_a_server_that_cannot_start_rejects_the_creation(self):
        editor = AcpClient(self.app)
        session_id = operation_id()
        broken = {**fake_server(), "command": "/nonexistent/editor-mcp"}
        refused(
            self,
            lambda: editor.create(session_id, servers=[broken]),
            409,
            "mcp_overlay_failed",
        )
        refused(
            self,
            lambda: session_read(self.app, session_id),
            404,
            "session_not_found",
        )

    def test_command_must_be_absolute(self):
        editor = AcpClient(self.app)
        relative = {**fake_server(), "command": "python3"}
        refused(self, lambda: editor.create(servers=[relative]), 400, "invalid_request")


class OverlayTool:
    """A model that calls the first editor MCP tool once, then answers."""

    def __init__(self):
        self.opened = threading.Event()
        self.opened.set()
        self.results = []
        self.provider = Provider(self.answer)

    def answer(self, request):
        self.opened.wait(30)
        inputs = request["input"]
        last_user = max(
            (index for index, item in enumerate(inputs) if item.get("role") == "user"),
            default=-1,
        )
        outputs = [
            item
            for item in inputs[last_user + 1 :]
            if item.get("type") == "function_call_output"
        ]
        if outputs:
            self.results.append(outputs[-1]["output"])
            return text("done")
        tools = sorted(tool["name"] for tool in request.get("tools", []))
        tool = next((name for name in tools if name.startswith("mcp_")), None)
        if tool is None:
            return text("no tools")
        return Reply("python", tool_name=tool, tool_arguments={"message": "hi"})


@exclusive
class AcpOverlayTest(unittest.TestCase):
    def setUp(self):
        self.model = OverlayTool()
        self.addCleanup(self.model.provider.close)
        self.addCleanup(self.model.opened.set)
        self.app = Albedo(
            self.model.provider,
            protocol="responses",
            prepare=lambda app: app.env.update(ALBEDO_INSPECT="1"),
        )
        self.app.__enter__()
        self.addCleanup(self.app.__exit__, None, None, None)
        self.editor = AcpClient(self.app)

    def test_a_running_turn_keeps_its_overlay_unless_resupplied_identically(self):
        session_id, _ = self.editor.create(servers=[fake_server()])
        self.model.opened.clear()
        self.editor.prompt(session_id, [{"type": "text", "text": "wait"}])
        wait_until(
            lambda: session_read(self.app, session_id)["status"]["phase"] != "idle",
            30,
            "the turn never started",
        )
        self.editor.attach(session_id, servers=[fake_server()])
        refused(
            self,
            lambda: self.editor.attach(
                session_id, servers=[fake_server(secret="changed")]
            ),
            409,
            "session_busy",
        )
        self.model.opened.set()
        self.app.idle(session_id)
        self.editor.attach(session_id, servers=[])

    def test_restart_loses_the_overlay_until_the_editor_resupplies_it(self):
        session_id, _ = self.editor.create(servers=[fake_server()])
        self.app.restart()
        self.editor.attach(session_id, servers=[fake_server()])
        self.editor.prompt(session_id, [{"type": "text", "text": "use the tool"}])
        self.app.idle(session_id)
        self.assertIn(SECRET, self.model.results[-1])

    def test_overlay_preparation_failure_keeps_the_previous_tools(self):
        session, _ = self.editor.create(servers=[fake_server()])
        broken = {**fake_server(), "command": "/nonexistent/editor-mcp"}
        refused(
            self,
            lambda: self.editor.attach(session, servers=[broken]),
            409,
            "mcp_overlay_failed",
        )
        self.editor.prompt(session, [{"type": "text", "text": "use original tool"}])
        self.app.idle(session)
        self.assertIn(SECRET, self.model.results[-1])

    def test_published_overlay_survives_a_runtime_reload_failure(self):
        session, _ = self.editor.create(servers=[fake_server()])
        self.editor.prompt(session, [{"type": "text", "text": "use original tool"}])
        self.app.idle(session)
        self.assertIn(SECRET, self.model.results[-1])
        with sqlite3.connect(self.app.home / "albedo.sqlite") as db:
            original = db.execute(
                "SELECT acp_overlay_identities FROM sessions WHERE id = ?", (session,)
            ).fetchone()[0]
        broken_prompt = self.app.workspace / "SYSTEM.md"
        broken_prompt.write_bytes(b"\xff")
        replacement = fake_server(secret="replacement-editor-secret")
        refused(
            self,
            lambda: self.editor.attach(session, servers=[replacement]),
            503,
            "overlay_installed_reload_failed",
        )
        refused(
            self,
            lambda: self.editor.attach(session, servers=[replacement]),
            503,
            "overlay_installed_reload_failed",
        )
        with sqlite3.connect(self.app.home / "albedo.sqlite") as db:
            published = db.execute(
                "SELECT acp_overlay_identities FROM sessions WHERE id = ?", (session,)
            ).fetchone()[0]
        self.assertNotEqual(published, original)
        requested = len(self.model.provider.requests)
        refused(
            self,
            lambda: self.editor.prompt(
                session, [{"type": "text", "text": "must not use stale tools"}]
            ),
            409,
            "overlay_installed_reload_failed",
        )
        # The runtime survives an abnormal session-actor exit. Its old cached
        # tools must remain fenced until an identical reattach repairs them.
        Inspection(self.app).evaluate(
            "Call(erlang, apply, [fun() -> "
            "{some, {subject, Pid, _}} = 'albedo@daemon@session':live(<<"
            + json.dumps(session)
            + ">>), Ref = monitor(process, Pid), exit(Pid, kill), "
            "receive {'DOWN', Ref, process, Pid, _} -> ok after 5000 -> error(actor_alive) end end, []])"
        )
        refused(
            self,
            lambda: self.editor.prompt(
                session, [{"type": "text", "text": "still must not use stale tools"}]
            ),
            409,
            "overlay_installed_reload_failed",
        )
        self.assertEqual(len(self.model.provider.requests), requested)
        self.assertEqual(len(self.model.results), 1)
        broken_prompt.unlink()
        # Retry the published replacement: execution must use its tool credentials.
        self.editor.attach(session, servers=[replacement])
        self.editor.prompt(session, [{"type": "text", "text": "use replacement tool"}])
        self.app.idle(session)
        self.assertIn("replacement-editor-secret", self.model.results[-1])
        self.assertNotIn(SECRET, self.model.results[-1])

    def test_empty_overlay_reloads_after_the_attachment_owner_restarts(self):
        session, _ = self.editor.create(servers=[fake_server()])
        self.editor.prompt(session, [{"type": "text", "text": "use original tool"}])
        self.app.idle(session)
        self.assertIn(SECRET, self.model.results[-1])
        broken_prompt = self.app.workspace / "SYSTEM.md"
        broken_prompt.write_bytes(b"\xff")
        refused(
            self,
            lambda: self.editor.attach(session, servers=[]),
            503,
            "overlay_installed_reload_failed",
        )
        Inspection(self.app).evaluate(
            "Call(erlang, apply, [fun() -> Pid = whereis(albedo_acp_attachments), "
            "Ref = monitor(process, Pid), exit(Pid, kill), "
            "receive {'DOWN', Ref, process, Pid, _} -> ok after 5000 -> error(actor_alive) end end, []])"
        )
        # No required server identities remain, but the runtime still has the
        # original tools. Missing attachment state must force a reload anyway.
        refused(
            self,
            lambda: self.editor.attach(session, servers=[]),
            503,
            "overlay_installed_reload_failed",
        )
        broken_prompt.unlink()
        self.editor.attach(session, servers=[])
        self.editor.prompt(
            session, [{"type": "text", "text": "no editor tools remain"}]
        )
        self.app.idle(session)
        self.assertEqual(len(self.model.results), 1)
        self.assertFalse(
            any(
                tool["name"].startswith("mcp_")
                for tool in self.model.provider.requests[-1]["request"].get("tools", [])
            )
        )

    def send_fixture_mail(self, session):
        def script(request):
            inputs = request.get("input", [])
            if any(item.get("type") == "function_call_output" for item in inputs):
                return text("sent")
            if "send fixture mail" in latest_user(request):
                return python(f"await mail.submit({session!r}, 'editor detached')")
            return text("mail delivered")

        self.model.provider.script = script
        sender = self.app.session()
        self.app.prompt(sender, "send fixture mail").close()
        self.app.idle(sender)
        return sender

    def test_missing_overlay_blocks_mail_until_the_editor_reattaches(self):
        session, _ = self.editor.create(servers=[fake_server()])
        self.app.restart()
        self.send_fixture_mail(session)
        wait_until(
            lambda: (
                "overlay_reattach_required"
                in json.dumps(session_read(self.app, session)["activity"])
            ),
            15,
            "missing overlay to require editor reattachment",
        )
        self.assertEqual(self.app.history(session)["items"], [])
        self.editor.attach(session, servers=[fake_server()])
        wait_until(
            lambda: any(
                row["kind"] == "assistant" for row in self.app.history(session)["items"]
            ),
            15,
            "pending mail to run after editor reattachment",
        )

    def test_empty_overlay_accepts_mail_after_restart_without_reattachment(self):
        session, _ = self.editor.create()
        self.app.restart()
        self.assertEqual(attachment_state(self.app, session), "detached")
        refused(
            self,
            lambda: self.editor.prompt(
                session, [{"type": "text", "text": "stale editor prompt"}]
            ),
            409,
            "attachment_lost",
        )

        sender = self.send_fixture_mail(session)
        try:
            wait_until(
                lambda: any(
                    row["kind"] == "assistant"
                    for row in self.app.history(session)["items"]
                ),
                15,
                "mail turn to finish with an empty editor overlay",
            )
        except AssertionError:
            self.fail(
                f"recipient activity: {session_read(self.app, session)['activity']}; "
                f"sender history: {self.app.history(sender)}"
            )

    def test_disabled_editor_mcp_does_not_block_mail_after_restart(self):
        session, _ = self.editor.create(servers=[fake_server()])
        path = self.app.home / "extensions.json"
        settings = json.loads(path.read_text()) if path.exists() else {}
        settings.setdefault("enabled", {})["editor-mcp"] = False
        path.write_text(json.dumps(settings))
        self.app.restart()

        self.send_fixture_mail(session)
        wait_until(
            lambda: any(
                row["kind"] == "assistant" for row in self.app.history(session)["items"]
            ),
            15,
            "mail turn to finish without the disabled overlay",
        )


def inject_failure(app, column):
    """Make the daemon's next write of `column` on sessions fail, as a full
    disk or a locked database would."""
    with sqlite3.connect(app.home / "albedo.sqlite", timeout=10) as db:
        db.execute(
            f"CREATE TRIGGER refuse_{column} BEFORE UPDATE OF {column} ON sessions "
            "BEGIN SELECT RAISE(ABORT, 'injected failure'); END"
        )


def clear_failure(app, column):
    with sqlite3.connect(app.home / "albedo.sqlite", timeout=10) as db:
        db.execute(f"DROP TRIGGER refuse_{column}")


def attachment_state(app, session_id):
    return session_read(app, session_id)["acp"]["attachment"]


@exclusive
class AcpOverlayFailureTest(unittest.TestCase):
    """An attach whose overlay cannot be recorded leaves the session as it was,
    live and after a restart: no writer and no overlay."""

    def test_an_overlay_that_cannot_be_recorded_is_not_installed(self):
        with Albedo() as app:
            editor = AcpClient(app)
            session_id, _ = editor.create()
            editor.release(session_id)
            inject_failure(app, "acp_overlay_identities")
            with self.assertRaises(urllib.error.HTTPError):
                editor.attach(session_id, servers=[fake_server()])
            unchanged = "detached"
            self.assertEqual(attachment_state(app, session_id), unchanged)
            app.restart()
            self.assertEqual(attachment_state(app, session_id), unchanged)
            clear_failure(app, "acp_overlay_identities")
            editor.attach(session_id, servers=[fake_server()])


@exclusive
class AcpTraceSnapshotsTest(unittest.TestCase):
    def test_large_file_edit_keeps_snapshots_inline(self):
        def script(request):
            if request["messages"][-1].get("role") == "user":
                return python("files.edit('source.txt', 'line 0650:', 'changed 0650:')")
            return text("done")

        provider = Provider(script)
        self.addCleanup(provider.close)
        with Albedo(provider) as app:
            original = "".join(
                f"line {index:04d}: existing source text\n" for index in range(1300)
            )
            (app.workspace / "source.txt").write_text(original)
            editor = AcpClient(app)
            session, _ = editor.create()
            before = app.stream_page(session)
            editor.prompt(session, [{"type": "text", "text": "edit source"}])
            app.idle(session)
            live = next(
                event["data"]
                for event in app.stream_page(session, before)["events"]
                if event["type"] == "tool"
            )
            self.assertTrue(live["content_complete"], live)
            change = live["trace"]["changes"][0]
            self.assertEqual(change["old_text"], original)
            self.assertEqual(
                change["new_text"], original.replace("line 0650:", "changed 0650:")
            )

    def test_history_preserves_snapshots_and_large_trace_reference(self):
        def script(request):
            if request["messages"][-1].get("role") == "user":
                return python(
                    "from pathlib import Path\nPath('existing.txt').write_text('changed text')\nPath('new.txt').write_text('new text')\nPath('binary.txt').write_bytes(b'\\x00binary')\nPath('oversized.txt').write_text('x' * 140000)\nPath('large.txt').write_text('x' * 100000)\nPath('large2.txt').write_text('y' * 100000)\nPath('large3.txt').write_text('z' * 100000)"
                )
            return text("done")

        provider = Provider(script)
        self.addCleanup(provider.close)
        with Albedo(provider) as app:
            (app.workspace / "existing.txt").write_text("original text")
            editor = AcpClient(app)
            session, _ = editor.create()
            before = app.stream_page(session)
            editor.prompt(session, [{"type": "text", "text": "write files"}])
            app.idle(session)
            live = next(
                event["data"]
                for event in app.stream_page(session, before)["events"]
                if event["type"] == "tool"
            )
            self.assertTrue(live["content_complete"])
            self.assertIsNotNone(live["trace"])
            self.assertIsNone(live["reference"])
            tool = next(
                e for e in app.history(session)["items"] if e["kind"] == "tool_result"
            )
            self.assertEqual(tool["tool"]["status"], "completed")
            self.assertEqual(tool["tool"]["progress_call_id"], live["progress_call_id"])
            reference = next(
                p["reference"]
                for p in tool["content"]
                if p["kind"] == "reference" and p["reference"]["field"] == "trace"
            )
            pieces = []
            url = reference["url"]
            while url:
                with app.api(url) as response:
                    page = json.load(response)
                pieces.extend(
                    part["text"] for part in page["parts"] if part["field"] == "trace"
                )
                url = (
                    reference["url"] + "?next=" + page["next"] if page["next"] else None
                )
            raw = "".join(pieces)
            self.assertEqual(len(raw.encode()), reference["bytes"])
            trace = json.loads(raw)
            change = next(c for c in trace["changes"] if c["path"].endswith("new.txt"))
            self.assertIsNone(change["old_text"])
            self.assertEqual(change["new_text"], "new text")
            changes = {Path(c["path"]).name: c for c in trace["changes"]}
            self.assertEqual(changes["existing.txt"]["old_text"], "original text")
            self.assertEqual(changes["existing.txt"]["new_text"], "changed text")
            for name in ("binary.txt", "oversized.txt"):
                self.assertEqual(changes[name]["kind"], "unavailable")
                self.assertTrue(changes[name]["reason"])
            # Snapshots stop once the 256 KiB budget is spent.
            total = sum(
                len((c.get("old_text") or "").encode())
                + len(c.get("new_text", "").encode())
                for c in changes.values()
            )
            self.assertLessEqual(total, 256 * 1024)

            # Retained transcript traces take precedence over the live journal.
            # Seed the pre-snapshot wire shape as it appeared in older sessions.
            result = next(
                p["value"] for p in tool["content"] if p.get("field") == "result"
            )
            cell_id = json.loads(result)["cell_id"]
            legacy = {
                "activities": [],
                "changes": [
                    {
                        "path": "old.txt",
                        "kind": "diff",
                        "diff": "-old\n+new\n",
                        "added": 1,
                        "removed": 1,
                    }
                ],
                "truncated": False,
            }
            with sqlite3.connect(app.home / "albedo.sqlite", timeout=10) as db:
                db.execute(
                    "INSERT INTO transcript_traces(session, cell_id, payload) VALUES (?, ?, ?)",
                    (session, cell_id, json.dumps(legacy)),
                )
            row = next(
                e for e in app.history(session)["items"] if e["kind"] == "tool_result"
            )
            self.assertEqual(
                next(p["trace"] for p in row["content"] if p["kind"] == "trace"), legacy
            )


if __name__ == "__main__":
    unittest.main()