"""Daemon integration: provider binding, history, tools, workspaces, and recovery."""

import json
from pathlib import Path
import sqlite3
import time
import unittest
import urllib.error

from harness import Albedo, Provider, exclusive, operation_id, python, text


def contents(item):
    content = item.get("content", "")
    if isinstance(content, str):
        return content
    return " ".join(part.get("text", "") for part in content if isinstance(part, dict))


def script(request):
    inputs = request.get("messages", request.get("input", []))
    user = next(
        (contents(item) for item in reversed(inputs) if item.get("role") == "user"), ""
    )
    tools = sum(
        item.get("role") == "tool" or item.get("type") == "function_call_output"
        for item in inputs
    )
    if "continue from branch" in user:
        return text("finished")
    if "no first token" in user:
        return text("never streamed", hang=True)
    if "over 100 turns" in user:
        return text("finished") if tools >= 105 else python("1")
    if "workspace probe" in user:
        code = (
            "import os\nfrom pathlib import Path\n"
            "assert 'workspace_marker' not in globals()\n"
            "workspace_marker = os.getcwd()\nPath('cwd-probe').write_text(os.getcwd())"
        )
        done = (
            inputs[-1].get("role") == "tool"
            or inputs[-1].get("type") == "function_call_output"
        )
        return text("finished") if done else python(code)
    code = (
        "import asyncio, os\nfrom pathlib import Path\n"
        "Path('example.txt').write_text('hello\\n')\n"
        "saved = Path('example.txt').read_text()\n"
        "assert 'ALBEDO_API_KEY' not in os.environ\n"
        "assert 'ALBEDO_TOKEN' not in os.environ\n"
        + (
            "Path('recovery-started').write_text('ready')\n"
            if "hang then recover" in user
            else ""
        )
        + f"await asyncio.sleep({30 if 'hang' in user else 2 if 'first task' in user else 0.4})\nlen(saved)"
    )
    return (
        text("finished")
        if tools
        else python(code, reasoning="reasoning about the task")
    )


class IntegrationTest(unittest.TestCase):
    def setUp(self):
        self.provider = Provider(script)
        self.addCleanup(self.provider.close)

    def app_for(self, protocol, *, prepare=None, providers=None):
        app = Albedo(
            self.provider, protocol=protocol, prepare=prepare, providers=providers
        )
        app.__enter__()
        self.addCleanup(app.__exit__, None, None, None)
        return app

    def read(self, app, path):
        with app.api(path) as response:
            return json.load(response)

    def send(self, app, session, message):
        app.prompt(session, message).close()
        app.idle(session, timeout=90)

    @exclusive
    def test_session_start_refuses_an_unsaved_default_effort(self):
        app = self.app_for("chat_completions")
        session = app.session()

        def prepare(app):
            # A newly available catalog gives the persisted session a default.
            (app.home / "models.json").write_text("{}")
            with sqlite3.connect(app.home / "albedo.sqlite") as db:
                db.execute(
                    "UPDATE sessions SET model='o3-unlisted', effort=NULL WHERE id=?",
                    (session,),
                )
                db.execute(
                    "CREATE TRIGGER reject_effort BEFORE UPDATE OF effort ON sessions "
                    "BEGIN SELECT RAISE(FAIL, 'fixture effort write failed'); END"
                )

        app.restart(prepare=prepare)
        # Malformed submissions must be rejected before attempting the worker
        # activation whose effort write this fixture deliberately blocks.
        with self.assertRaises(urllib.error.HTTPError) as caught:
            app.api(
                f"/sessions/{session}/events",
                {"type": "continu", "content": "must not submit"},
            ).close()
        self.assertEqual(caught.exception.code, 400)
        self.assertEqual(json.load(caught.exception)["code"], "invalid_submission")
        with self.assertRaises(urllib.error.HTTPError) as caught:
            self.read(app, f"/sessions/{session}/status")
        self.assertIn("session could not start", json.load(caught.exception)["error"])
        with sqlite3.connect(app.home / "albedo.sqlite") as db:
            self.assertIsNone(
                db.execute(
                    "SELECT effort FROM sessions WHERE id=?", (session,)
                ).fetchone()[0]
            )
            db.execute("DROP TRIGGER reject_effort")
        self.assertFalse(self.provider.requests)
        self.assertIn(
            "fixture effort write failed", (app.home / "daemon.log").read_text()
        )
        self.read(app, f"/sessions/{session}/status")
        with sqlite3.connect(app.home / "albedo.sqlite") as db:
            self.assertEqual(
                db.execute(
                    "SELECT effort FROM sessions WHERE id=?", (session,)
                ).fetchone()[0],
                "medium",
            )

    @exclusive
    def test_scheduler_reports_a_failed_occurrence_write_and_retries(self):
        def prepare(app):
            app.env["ALBEDO_SCHEDULE_TICK_MS"] = "50"

        app = self.app_for("chat_completions", prepare=prepare)
        session = app.session()
        with sqlite3.connect(app.home / "albedo.sqlite") as db:
            db.execute(
                "CREATE TRIGGER reject_advance BEFORE DELETE ON schedules "
                "BEGIN SELECT RAISE(FAIL, 'fixture advance failed'); END"
            )
            db.execute(
                "INSERT INTO schedules(session,kind,prompt,next_at) VALUES(?,'once',?,0)",
                (session, "scheduled fixture"),
            )
        deadline = time.monotonic() + 15
        while "fixture advance failed" not in (app.home / "daemon.log").read_text():
            self.assertLess(time.monotonic(), deadline)
            time.sleep(0.05)
        self.assertIn(
            "schedule occurrence could not be advanced",
            (app.home / "daemon.log").read_text(),
        )
        with sqlite3.connect(app.home / "albedo.sqlite") as db:
            self.assertEqual(
                db.execute("SELECT COUNT(*) FROM schedules").fetchone()[0], 1
            )
            db.execute("DROP TRIGGER reject_advance")
        while True:
            with sqlite3.connect(app.home / "albedo.sqlite") as db:
                if db.execute("SELECT COUNT(*) FROM schedules").fetchone()[0] == 0:
                    break
            self.assertLess(time.monotonic(), deadline)
            time.sleep(0.05)
        app.idle(session)
        self.assertTrue(self.provider.requests)

    @exclusive
    def test_deletion_refuses_a_failed_child_lookup(self):
        app = self.app_for("chat_completions")
        session = app.session()
        with sqlite3.connect(app.home / "albedo.sqlite") as db:
            db.execute("ALTER TABLE session_family RENAME TO unavailable_family")
        for suffix in ("", "?tree=1"):
            with self.subTest(tree=bool(suffix)):
                with self.assertRaises(urllib.error.HTTPError) as caught:
                    app.api(f"/sessions/{session}{suffix}", method="DELETE")
                self.assertEqual(caught.exception.code, 409)
                self.assertIn("session_family", json.load(caught.exception)["error"])
                with sqlite3.connect(app.home / "albedo.sqlite") as db:
                    self.assertIsNotNone(
                        db.execute(
                            "SELECT id FROM sessions WHERE id=?", (session,)
                        ).fetchone()
                    )
        with sqlite3.connect(app.home / "albedo.sqlite") as db:
            db.execute("ALTER TABLE unavailable_family RENAME TO session_family")
        with app.api(f"/sessions/{session}?tree=1", method="DELETE") as response:
            self.assertEqual(json.load(response)["deleted"], 1)

    def settings(self, app, protocol, active="alpha", **models):
        other = "responses" if protocol == "chat_completions" else "chat_completions"
        configured = {
            "alpha": ("alpha-2", models.get("alpha", "fixture-alpha"), protocol),
            "beta": ("beta-1", models.get("beta", "fixture-beta"), protocol),
            "gamma": ("gamma-1", models.get("gamma", "fixture-gamma"), other),
        }
        providers = {
            name: {
                "baseUrl": self.provider.url + f"/{name}/v1",
                "apiKey": key,
                "model": model,
                "protocol": binding,
            }
            for name, (key, model, binding) in configured.items()
        }
        (app.home / "config.json").write_text(
            json.dumps({"active": active, "providers": providers})
        )

    def records(self, start):
        return self.provider.requests[start:]

    def wait_for_request(self, start):
        deadline = time.monotonic() + 15
        while len(self.provider.requests) <= start and time.monotonic() < deadline:
            time.sleep(0.05)
        self.assertGreater(len(self.provider.requests), start)

    def assert_projected(self, record, protocol, users):
        items = record["request"][
            "messages" if protocol == "chat_completions" else "input"
        ]
        serialized = json.dumps(items)
        for user in users:
            self.assertIn(user, serialized)
        self.assertIn("finished", serialized)
        if protocol == "chat_completions":
            calls = [
                call["id"]
                for item in items
                if item.get("role") == "assistant"
                for call in item.get("tool_calls", [])
            ]
            outputs = [
                item["tool_call_id"] for item in items if item.get("role") == "tool"
            ]
        else:
            calls = [
                item["call_id"] for item in items if item.get("type") == "function_call"
            ]
            outputs = [
                item["call_id"]
                for item in items
                if item.get("type") == "function_call_output"
            ]
        self.assertTrue(any(call.startswith("fixture-call-") for call in calls))
        self.assertTrue(set(calls) & set(outputs))

    @exclusive
    def test_unconfigured_startup_and_legacy_provider_migration(self):
        def prepare(app):
            workspace = app.root / "legacy-workspace"
            workspace.mkdir(exist_ok=True)
            with sqlite3.connect(app.home / "albedo.sqlite") as db:
                db.execute(
                    "CREATE TABLE sessions(id TEXT PRIMARY KEY,cwd TEXT NOT NULL,model TEXT NOT NULL,protocol TEXT NOT NULL,stage TEXT NOT NULL DEFAULT 'idle')"
                )
                for protocol in ("responses", "chat_completions"):
                    db.execute(
                        "INSERT INTO sessions(id,cwd,model,protocol) VALUES(?,?,?,?)",
                        (
                            "legacy-" + protocol,
                            str(workspace),
                            "legacy-model",
                            protocol,
                        ),
                    )

        app = self.app_for("responses", prepare=prepare, providers={})
        health = self.read(app, "/health")
        self.assertTrue(health["ok"])
        self.assertEqual(health["version"], 2)
        self.assertTrue(
            {"session_provider", "session_workspace"} <= set(health["capabilities"])
        )
        with self.assertRaises(urllib.error.HTTPError) as caught:
            app.api("/sessions", {"workspace": str(app.workspace)}).close()
        self.assertEqual(caught.exception.code, 400)
        self.assertIn("/login", json.load(caught.exception)["error"])
        for protocol in ("responses", "chat_completions"):
            with self.subTest(protocol=protocol):
                legacy = "legacy-" + protocol
                self.settings(app, protocol)
                config = json.loads((app.home / "config.json").read_text())
                config["providers"]["default"] = {
                    "baseUrl": self.provider.url + "/legacy/v1",
                    "apiKey": "legacy-key",
                    "model": "legacy-model",
                    "protocol": protocol,
                }
                (app.home / "config.json").write_text(json.dumps(config))
                start = len(self.provider.requests)
                self.send(app, legacy, "inspect the legacy workspace")
                requests = self.records(start)
                self.assertTrue(requests)
                self.assertTrue(all("/legacy/v1/" in r["path"] for r in requests))
                self.assertTrue(
                    all(r["authorization"] == "Bearer legacy-key" for r in requests)
                )
                session = app.session()
                info = next(
                    s for s in self.read(app, "/sessions") if s["id"] == session
                )
                self.assertEqual(
                    (info["provider"], info["title"]), ("alpha", "new session")
                )
                pid = app.connection["pid"]
                app.cli("sessions")
                self.assertEqual(
                    json.loads((app.home / "daemon.json").read_text())["pid"], pid
                )
                self.restart(app)
                info = next(s for s in self.read(app, "/sessions") if s["id"] == legacy)
                self.assertEqual(
                    (info["provider"], info["title"]),
                    ("default", "inspect the legacy workspace"),
                )

    def restart(self, app):
        app.restart(crash=True)

    def test_image_metadata_is_verified_and_only_metadata_is_exposed(self):
        for protocol in ("responses", "chat_completions"):
            with self.subTest(protocol=protocol):
                app = self.app_for(protocol)
                session = app.session()
                png = "iVBORw0KGgoAAAANSUhEUgAAAAIAAAAD"
                attachment = {
                    "mimeType": "image/png",
                    "data": png,
                    "width": 2,
                    "height": 3,
                    "bytes": 24,
                }
                with self.assertRaises(urllib.error.HTTPError) as caught:
                    app.api(
                        f"/sessions/{session}/events",
                        {
                            "content": "reject this",
                            "image": dict(attachment, width=4),
                        },
                    ).close()
                self.assertEqual(caught.exception.code, 400)
                self.assertEqual(
                    json.load(caught.exception)["code"], "invalid_submission"
                )
                start = len(self.provider.requests)
                app.api(
                    f"/sessions/{session}/events",
                    {
                        "content": "describe this image",
                        "image": attachment,
                    },
                ).close()
                app.idle(session)
                self.send(app, session, "confirm image history")
                requests = self.records(start)
                self.assertEqual(len(requests), 3)
                self.assertTrue(
                    all(
                        "data:image/png;base64," + png in json.dumps(r["request"])
                        for r in requests
                    )
                )
                image = next(
                    e["image"]
                    for e in app.events(session)
                    if e.get("type") == "user" and e.get("image")
                )
                self.assertEqual(
                    image,
                    {"mimeType": "image/png", "width": 2, "height": 3, "bytes": 24},
                )
                self.assertNotIn(png, json.dumps(image))

    def test_invalid_submissions_do_not_activate_or_admit_work(self):
        app = self.app_for("responses")
        session = app.session()
        before = app.events(session)
        invalid = [
            {},
            [],
            {"type": "usr", "content": "must not submit"},
            {"type": "continu", "content": "must not submit"},
            {"type": None, "content": "must not submit"},
            {"content": None},
            {"content": 4},
            {"content": "must not submit", "clientId": False},
            {"type": "continue", "content": False},
            {"type": "continue", "image": {}},
            {
                "content": "must not submit",
                "image": {
                    "mimeType": "image/png",
                    "data": "not base64",
                    "width": 1,
                    "height": 1,
                    "bytes": 1,
                },
            },
        ]
        for payload in invalid:
            for identity in (session, "nonexistent-submission-session"):
                with self.subTest(payload=payload, identity=identity):
                    with self.assertRaises(urllib.error.HTTPError) as caught:
                        app.api(f"/sessions/{identity}/events", payload).close()
                    self.assertEqual(caught.exception.code, 400)
                    self.assertEqual(
                        json.load(caught.exception)["code"], "invalid_submission"
                    )
        self.assertEqual(app.events(session), before)
        self.assertFalse(self.provider.requests)

    def test_submission_kinds_preserve_user_and_continuation_behavior(self):
        app = self.app_for("responses")
        session = app.session()
        fixture = json.loads(
            (Path(__file__).parents[1] / "fixtures/mutation_responses.json").read_text()
        )["submission"]
        for kind in (None, "user", "continue"):
            payload = {
                "operationId": operation_id(),
                "content": "valid submission " + str(kind),
            }
            if kind is not None:
                payload["type"] = kind
            with app.api(f"/sessions/{session}/events", payload) as response:
                self.assertEqual(response.status, fixture["status"])
                self.assertEqual(
                    json.load(response),
                    {**fixture["body"], "operationId": payload["operationId"]},
                )
            app.idle(session)
        users = [
            event["text"] for event in app.events(session) if event["type"] == "user"
        ]
        self.assertEqual(users, ["valid submission None", "valid submission user"])
        self.assertGreaterEqual(len(self.provider.requests), 3)

    def test_busy_messages_join_next_model_request_in_order(self):
        app = self.app_for("responses")
        session = app.session()
        start = len(self.provider.requests)
        with app.api(
            f"/sessions/{session}/events", {"content": "first task"}
        ) as response:
            self.assertFalse(json.load(response)["queued"])
        self.wait_for_request(start)
        with app.api(
            f"/sessions/{session}/events", {"content": "queued direction"}
        ) as response:
            self.assertTrue(json.load(response)["queued"])
        app.prompt(session, "another direction").close()
        app.idle(session)
        requests = self.records(start)
        self.assertEqual(len(requests), 2)
        first, second = (json.dumps(r["request"]) for r in requests)
        self.assertNotIn("queued direction", first)
        self.assertIn("queued direction", second)
        self.assertIn("another direction", second)
        self.assertLess(
            second.index("queued direction"), second.index("another direction")
        )
        self.assertTrue(
            any(
                e.get("type") == "user" and e.get("text") == "queued direction"
                for e in app.events(session)
            )
        )

    @exclusive
    def test_workspace_repair_validates_path_and_preserves_history(self):
        app = self.app_for("responses")
        old = app.root / "original-workspace"
        old.mkdir()
        moved = app.root / "moved-workspace"
        session = app.session(old)
        self.send(app, session, "workspace probe before move")
        self.assertEqual(Path((old / "cwd-probe").read_text()).resolve(), old.resolve())
        old.rename(moved)
        pid = app.connection["pid"]
        before = app.events(session)
        operation = operation_id()
        request = {"operationId": operation, "content": "workspace probe after move"}
        with app.api(f"/sessions/{session}/events", request) as response:
            self.assertEqual(response.status, 202)
        receipt = self.read(app, f"/operations/{operation}")
        self.assertEqual(receipt["deliveryStatus"], "pending")
        for invalid in ("relative", str(old), str(moved / "cwd-probe")):
            with self.assertRaises(urllib.error.HTTPError) as caught:
                app.api(
                    f"/sessions/{session}/workspace", {"workspace": invalid}
                ).close()
            self.assertEqual(caught.exception.code, 409)
        with app.api(
            f"/sessions/{session}/workspace", {"workspace": str(moved)}
        ) as response:
            self.assertEqual(json.load(response)["workspace"], str(moved))
        self.assertEqual(
            next(s for s in self.read(app, "/sessions") if s["id"] == session)[
                "workspace"
            ],
            str(moved),
        )
        self.assertEqual(json.loads((app.home / "daemon.json").read_text())["pid"], pid)
        deadline = time.monotonic() + 35
        while time.monotonic() < deadline:
            receipt = self.read(app, f"/operations/{operation}")
            if receipt["deliveryStatus"] == "committed":
                break
            time.sleep(0.05)
        self.assertEqual(receipt["deliveryStatus"], "committed")
        app.idle(session)
        self.assertEqual(
            Path((moved / "cwd-probe").read_text()).resolve(), moved.resolve()
        )
        self.assertFalse(old.exists())
        after = app.events(session)
        self.assertEqual(
            [e for e in before if e.get("type") == "user"],
            [e for e in after if e.get("type") == "user"][:-1],
        )
        with sqlite3.connect(app.home / "albedo.sqlite") as db:
            self.assertEqual(
                db.execute(
                    "SELECT cwd FROM sessions WHERE id=?", (session,)
                ).fetchone()[0],
                str(moved),
            )
        self.restart(app)
        info = next(s for s in self.read(app, "/sessions") if s["id"] == session)
        self.assertEqual(info["workspace"], str(moved))

    @exclusive
    def test_tools_trace_detach_and_fork_checkpoint(self):
        for protocol in ("responses", "chat_completions"):
            with self.subTest(protocol=protocol):
                app = self.app_for(protocol)
                session = app.session()
                app.prompt(session, "write and inspect a file").close()
                app.events(session)
                app.idle(session)
                events = app.events(session)
                self.assertEqual((app.workspace / "example.txt").read_text(), "hello\n")
                self.assertTrue(
                    any(
                        e.get("type") == "message" and e.get("text") == "finished"
                        for e in events
                    )
                )
                self.assertTrue(
                    any(
                        e.get("type") == "thinking"
                        and e.get("text") == "reasoning about the task"
                        for e in events
                    )
                )
                tool = next(e for e in events if e.get("type") == "tool")
                self.assertEqual(json.loads(tool["result"])["status"], "ok")
                self.assertTrue(
                    any(
                        a["kind"] == "read" and a["target"].endswith("example.txt")
                        for a in tool["trace"]["activities"]
                    )
                )
                self.assertTrue(
                    any(
                        c["path"].endswith("example.txt")
                        for c in tool["trace"]["changes"]
                    )
                )
                source = self.read(app, f"/sessions/{session}/tree?after=0&limit=100")
                items = source["items"]
                self.assertFalse(source["hasMore"])
                self.assertEqual(
                    {i["type"] for i in items}, {"user", "assistant", "tool"}
                )
                if protocol == "responses":
                    self.assertTrue(
                        any(i["preview"].startswith("[reasoning]") for i in items)
                    )
                checkpoint = next(i for i in items if i["preview"] == "call python")
                with app.api(
                    f"/sessions/{session}/fork", {"checkpoint": checkpoint["id"]}
                ) as response:
                    branch = json.load(response)
                self.assertEqual(branch["title"], "write and inspect a file")
                self.assertEqual(
                    (branch["provider"], branch["model"], branch["protocol"]),
                    ("fixture", "fixture-model", protocol),
                )
                branch_items = self.read(
                    app, f"/sessions/{branch['id']}/tree?after=0&limit=100"
                )["items"]
                self.assertEqual(
                    branch_items[-1]["preview"], "not executed after branch checkpoint"
                )
                self.assertEqual(
                    [i["preview"] for i in branch_items[:-1]],
                    [i["preview"] for i in items if i["id"] <= checkpoint["id"]],
                )
                start = len(self.provider.requests)
                self.send(app, branch["id"], "continue from branch")
                requests = self.records(start)
                self.assertEqual(len(requests), 1)
                self.assertTrue(
                    all(
                        "/chat/completions" in r["path"]
                        if protocol == "chat_completions"
                        else "/responses" in r["path"]
                        for r in requests
                    )
                )
                inputs = requests[0]["request"][
                    "messages" if protocol == "chat_completions" else "input"
                ]
                outputs = [
                    i
                    for i in inputs
                    if i.get("role") == "tool"
                    or i.get("type") == "function_call_output"
                ]
                self.assertEqual(
                    [i.get("content", i.get("output")) for i in outputs],
                    ["not executed after branch checkpoint"],
                )
                self.assertTrue(
                    any(
                        e.get("type") == "message" and e.get("text") == "finished"
                        for e in app.events(branch["id"])
                    )
                )
                self.assertEqual(
                    self.read(app, f"/sessions/{session}/tree?after=0&limit=100")[
                        "items"
                    ],
                    items,
                )
                self.assertEqual(
                    next(s for s in self.read(app, "/sessions") if s["id"] == session)[
                        "title"
                    ],
                    "write and inspect a file",
                )
                self.send(app, session, "review the result")
                self.assertEqual(
                    next(s for s in self.read(app, "/sessions") if s["id"] == session)[
                        "title"
                    ],
                    "review the result",
                )
                timestamps = [
                    (e["type"], e["text"], e.get("timestamp"))
                    for e in app.events(session)
                    if e.get("type") in ("user", "message")
                ]
                self.assertTrue(timestamps)
                self.assertTrue(
                    all(isinstance(stamp, int) for _, _, stamp in timestamps)
                )
                self.restart(app)
                self.assertEqual(
                    [
                        (e["type"], e["text"], e.get("timestamp"))
                        for e in app.events(session)
                        if e.get("type") in ("user", "message")
                    ],
                    timestamps,
                )

    @exclusive
    def test_provider_switch_projects_history_across_protocols_and_restart(self):
        for protocol in ("responses", "chat_completions"):
            with self.subTest(protocol=protocol):
                app = self.app_for(protocol)
                self.settings(app, protocol)
                session = app.session()
                route = f"/sessions/{session}/commands"
                app.api(
                    route,
                    {
                        "name": "/model",
                        "args": {
                            "provider": "alpha",
                            "model": "initial-alpha",
                        },
                    },
                ).close()
                start = len(self.provider.requests)
                self.send(app, session, "build switch history")
                self.assertGreaterEqual(len(self.records(start)), 2)
                self.assertTrue(
                    all("/alpha/v1/" in r["path"] for r in self.records(start))
                )
                other = (
                    "responses"
                    if protocol == "chat_completions"
                    else "chat_completions"
                )
                for provider, model, expected in (
                    ("gamma", "chosen-gamma", other),
                    (None, "renamed-gamma", other),
                ):
                    args = {"model": model}
                    if provider:
                        args["provider"] = provider
                    with app.api(route, {"name": "/model", "args": args}) as response:
                        result = json.load(response)
                    self.assertEqual(
                        result,
                        {
                            "result": {
                                "provider": "gamma",
                                "model": model,
                                "protocol": expected,
                                "effort": None,
                            }
                        },
                    )
                    defaults = json.loads((app.home / "config.json").read_text())
                    self.assertEqual(defaults["active"], "gamma")
                    self.assertEqual(defaults["providers"]["gamma"]["model"], model)
                with app.api(
                    "/sessions", {"workspace": str(app.workspace)}
                ) as response:
                    default = json.load(response)
                self.assertEqual(
                    (default["provider"], default["model"]), ("gamma", "renamed-gamma")
                )
                with self.assertRaises(urllib.error.HTTPError) as caught:
                    app.api(
                        route,
                        {
                            "name": "/model",
                            "args": {
                                "provider": "unknown",
                                "model": "wrong",
                            },
                        },
                    ).close()
                self.assertEqual(caught.exception.code, 409)
                start = len(self.provider.requests)
                self.send(app, session, "after first switch")
                gamma = self.records(start)
                self.assertTrue(gamma)
                self.assertTrue(
                    all(
                        "/gamma/v1/" in r["path"]
                        and r["model"] == "renamed-gamma"
                        and r["authorization"] == "Bearer gamma-1"
                        for r in gamma
                    )
                )
                self.assert_projected(
                    gamma[0], other, ["build switch history", "after first switch"]
                )
                with app.api(
                    route,
                    {
                        "name": "/model",
                        "args": {
                            "provider": "alpha",
                            "model": "returned-alpha",
                        },
                    },
                ) as response:
                    result = json.load(response)
                self.assertEqual(
                    result,
                    {
                        "result": {
                            "provider": "alpha",
                            "model": "returned-alpha",
                            "protocol": protocol,
                            "effort": None,
                        }
                    },
                )
                defaults = json.loads((app.home / "config.json").read_text())
                self.assertEqual(defaults["active"], "alpha")
                self.assertEqual(
                    defaults["providers"]["alpha"]["model"], "returned-alpha"
                )
                start = len(self.provider.requests)
                self.send(app, session, "after second switch")
                returned = self.records(start)
                self.assertTrue(returned)
                self.assertTrue(
                    all(
                        "/alpha/v1/" in r["path"] and r["model"] == "returned-alpha"
                        for r in returned
                    )
                )
                users = [
                    "build switch history",
                    "after first switch",
                    "after second switch",
                ]
                self.assert_projected(returned[0], protocol, users)
                info = next(
                    s for s in self.read(app, "/sessions") if s["id"] == session
                )
                self.assertEqual(
                    (info["provider"], info["model"], info["protocol"]),
                    ("alpha", "returned-alpha", protocol),
                )
                self.restart(app)
                start = len(self.provider.requests)
                self.send(app, session, "after persisted restart")
                restarted = self.records(start)
                self.assertTrue(restarted)
                self.assertTrue(
                    all(
                        "/alpha/v1/" in r["path"] and r["model"] == "returned-alpha"
                        for r in restarted
                    )
                )
                self.assert_projected(
                    restarted[0], protocol, users + ["after persisted restart"]
                )

    @exclusive
    def test_provider_config_reload_and_interrupt_recovery(self):
        for protocol in ("responses", "chat_completions"):
            with self.subTest(protocol=protocol):
                app = self.app_for(protocol)
                self.settings(app, protocol)
                initial = json.loads((app.home / "config.json").read_text())
                initial["providers"]["alpha"]["apiKey"] = "alpha-1"
                (app.home / "config.json").write_text(json.dumps(initial))
                session = app.session()
                self.settings(app, protocol, active="beta")
                start = len(self.provider.requests)
                self.send(app, session, "write and inspect a file")
                alpha = self.records(start)
                self.assertTrue(alpha)
                self.assertTrue(
                    all(
                        "/alpha/v1/" in r["path"]
                        and r["authorization"] == "Bearer alpha-2"
                        and r["model"] == "fixture-alpha"
                        for r in alpha
                    )
                )
                second = app.session()
                info = next(s for s in self.read(app, "/sessions") if s["id"] == second)
                self.assertEqual(
                    (info["provider"], info["model"]), ("beta", "fixture-beta")
                )
                start = len(self.provider.requests)
                app.prompt(second, "hang").close()
                self.wait_for_request(start)
                for operation, body in (
                    ("workspace", {"workspace": str(app.root)}),
                    (
                        "commands",
                        {
                            "name": "/model",
                            "args": {
                                "provider": "gamma",
                                "model": "busy-rejected",
                            },
                        },
                    ),
                ):
                    with self.assertRaises(urllib.error.HTTPError) as caught:
                        app.api(f"/sessions/{second}/{operation}", body).close()
                    self.assertEqual(caught.exception.code, 409)
                info = next(s for s in self.read(app, "/sessions") if s["id"] == second)
                self.assertEqual(
                    (info["workspace"], info["provider"]), (str(app.workspace), "beta")
                )
                app.api(f"/sessions/{second}/interrupt", {}).close()
                app.idle(second)
                beta = self.records(start)
                self.assertTrue(beta)
                self.assertTrue(
                    all(
                        "/beta/v1/" in r["path"]
                        and r["authorization"] == "Bearer beta-1"
                        for r in beta
                    )
                )
                self.assertTrue(
                    any(e.get("type") == "user" for e in app.events(second))
                )
                third = app.session()
                app.prompt(third, "hang then recover").close()
                ready = app.workspace / "recovery-started"
                deadline = time.monotonic() + 15
                while not ready.exists() and time.monotonic() < deadline:
                    time.sleep(0.05)
                self.assertTrue(ready.exists(), "recovery tool did not start")
                stamp = (app.workspace / "example.txt").stat().st_mtime_ns
                self.settings(
                    app, protocol, active="alpha", beta="changed-beta-default"
                )
                config = json.loads((app.home / "config.json").read_text())
                config["providers"]["beta"]["apiKey"] = "beta-2"
                (app.home / "config.json").write_text(json.dumps(config))
                start = len(self.provider.requests)
                self.restart(app)
                app.idle(third)
                restored = app.events(third)
                self.assertTrue(
                    any(
                        e.get("type") == "message" and e.get("text") == "finished"
                        for e in restored
                    )
                )
                resumed = self.records(start)
                self.assertTrue(resumed)
                self.assertTrue(
                    all(
                        "/beta/v1/" in r["path"]
                        and r["authorization"] == "Bearer beta-2"
                        and r["model"] == "fixture-beta"
                        for r in resumed
                    )
                )
                inputs = resumed[0]["request"][
                    "messages" if protocol == "chat_completions" else "input"
                ]
                note = [contents(i) for i in inputs if i.get("role") == "user"][-1]
                self.assertTrue(
                    note.startswith(
                        '<system-note origin="daemon restart">albedo restarted'
                    )
                )
                # The detached kernel outlives the crash: no reset notice.
                self.assertNotIn("<system-note>The python kernel", note)
                markers = [
                    e
                    for e in restored
                    if e.get("type") == "user" and e.get("source") == "daemon restart"
                ]
                self.assertEqual(len(markers), 1)
                self.assertTrue(markers[0]["text"].startswith("albedo restarted"))
                sessions = {s["id"]: s for s in self.read(app, "/sessions")}
                self.assertEqual(
                    (sessions[session]["provider"], sessions[session]["title"]),
                    ("alpha", "write and inspect a file"),
                )
                self.assertEqual(
                    (sessions[third]["provider"], sessions[third]["title"]),
                    ("beta", "hang then recover"),
                )
                self.assertEqual(
                    (app.workspace / "example.txt").stat().st_mtime_ns, stamp
                )
                self.assertTrue(
                    any(e.get("type") == "message" for e in app.events(session))
                )

    def test_interrupt_before_first_token_settles_quickly(self):
        for protocol in ("responses", "chat_completions"):
            with self.subTest(protocol=protocol):
                app = self.app_for(protocol)
                session = app.session()
                start = len(self.provider.requests)
                app.prompt(session, "no first token").close()
                self.wait_for_request(start)
                before = time.monotonic()
                app.api(f"/sessions/{session}/interrupt", {}).close()
                app.idle(session)
                # The provider holds the stream open without ever sending a
                # token, so only the worker noticing the stop latch while it
                # waits can settle the interrupt: the Abort backstop 2.5s in
                # is the old path, and the provider answers after 60s.
                self.assertLess(time.monotonic() - before, 1.0)
                # The interrupted session takes a new turn and commits it.
                self.send(app, session, "continue from branch")
                self.assertTrue(
                    any(
                        e.get("type") == "message" and e.get("text") == "finished"
                        for e in app.events(session)
                    )
                )

    def test_more_than_hundred_tool_turns_complete(self):
        for protocol in ("responses", "chat_completions"):
            with self.subTest(protocol=protocol):
                app = self.app_for(protocol)
                session = app.session()
                start = len(self.provider.requests)
                self.send(app, session, "continue for over 100 turns")
                self.assertEqual(len(self.records(start)), 106)
                events = app.events(session)
                self.assertEqual(sum(e.get("type") == "tool" for e in events), 105)
                self.assertTrue(
                    any(
                        e.get("type") == "message" and e.get("text") == "finished"
                        for e in events
                    )
                )