import asyncio
import hashlib
import types
import unittest
import uuid
from dataclasses import replace

from mrjunejune.inference.copilot_sidecar import (
    CompiledProfile,
    Sidecar,
    SidecarConfig,
    _derive_sdk_session_id,
    _SDK_SESSION_NAMESPACE,
    _validate_history,
    _HISTORY_MAX_ENTRIES,
    _HISTORY_MAX_BYTES,
)


def event(event_type, **data):
    return types.SimpleNamespace(
        type=types.SimpleNamespace(value=event_type),
        data=types.SimpleNamespace(**data),
    )


# ---------------------------------------------------------------------------
# Fake compiled-profile infrastructure
# ---------------------------------------------------------------------------

def _make_content(profile: str) -> str:
    return f"You are the {profile} assistant. Common rules apply."


_FAKE_PROFILES: dict = {
    profile: {
        "content": _make_content(profile),
        "version": 1,
        "hash": hashlib.sha256(_make_content(profile).encode()).hexdigest(),
    }
    for profile in ("public_visitor", "invited_friend", "june_admin")
}


def _fake_compile_fn(profile: str) -> dict:
    if profile not in _FAKE_PROFILES:
        raise ValueError(f"Unknown profile: {profile!r}")
    return _FAKE_PROFILES[profile]


def _fake_sdk_id(conversation_id: str, profile: str) -> str:
    """Compute the deterministic SDK session ID for use in test assertions."""
    p = _FAKE_PROFILES[profile]
    compiled = CompiledProfile(
        profile=profile,
        content=p["content"],
        prompt_version=1,
        knowledge_version=p["version"],
        hash=p["hash"],
    )
    return _derive_sdk_session_id(conversation_id, compiled)


class FakeSession:
    def __init__(self, session_id, behavior="complete"):
        self.session_id = session_id
        self.behavior = behavior
        self.handlers = []
        self.prompts = []
        self.abort_calls = 0
        self.disconnect_calls = 0
        self.disconnect_error = False

    def on(self, handler):
        self.handlers.append(handler)

        def unsubscribe():
            if handler in self.handlers:
                self.handlers.remove(handler)

        return unsubscribe

    def emit(self, value):
        for handler in list(self.handlers):
            handler(value)

    async def send(self, prompt):
        self.prompts.append(prompt)
        message_id = f"message-{len(self.prompts)}"
        if self.behavior == "complete":
            self.emit(
                event(
                    "assistant.message_delta",
                    delta_content=f"{prompt}-delta",
                    message_id=message_id,
                )
            )
            self.emit(
                event(
                    "assistant.message",
                    content=f"{prompt}-answer",
                    message_id=message_id,
                    model="test-model",
                )
            )
            self.emit(
                event(
                    "assistant.usage",
                    model="test-model",
                    input_tokens=3,
                    output_tokens=5,
                )
            )
            self.emit(event("session.idle", aborted=False))
        elif self.behavior == "error":
            self.emit(
                event(
                    "session.error",
                    error_type="provider",
                    error_code="upstream_error",
                    message="upstream failed",
                    status_code=502,
                )
            )
        return message_id

    async def abort(self):
        self.abort_calls += 1
        if self.behavior == "abort_error":
            raise RuntimeError("abort failed")
        self.emit(event("session.idle", aborted=True))

    async def disconnect(self):
        self.disconnect_calls += 1
        if self.disconnect_error:
            raise RuntimeError("disconnect failed")


class FakeClient:
    def __init__(self, behavior="complete"):
        self.behavior = behavior
        self.sessions = {}
        self.create_calls = []
        self.resume_calls = []
        self.delete_calls = []
        self.started = False
        self.stopped = False
        self.resume_started = None
        self.resume_release = None

    async def start(self):
        self.started = True

    async def stop(self):
        self.stopped = True

    async def resume_session(self, session_id, **kwargs):
        self.resume_calls.append((session_id, kwargs))
        if self.resume_started is not None:
            self.resume_started.set()
        if self.resume_release is not None:
            await self.resume_release.wait()
        if session_id not in self.sessions:
            raise LookupError(session_id)
        return self.sessions[session_id]

    async def create_session(self, session_id, **kwargs):
        self.create_calls.append((session_id, kwargs))
        session = FakeSession(session_id, self.behavior)
        self.sessions[session_id] = session
        return session

    async def delete_session(self, session_id):
        self.delete_calls.append(session_id)
        self.sessions.pop(session_id, None)


class SidecarTest(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        self.output = []

        async def capture(payload):
            self.output.append(payload)

        self.capture = capture
        self.config = SidecarConfig(
            base_url="http://litellm.invalid/v1",
            model="test-model",
            wire_api="responses",
            base_directory="/not-used-by-fake",
        )

    async def make_sidecar(self, behavior="complete", compile_fn=_fake_compile_fn):
        client = FakeClient(behavior)
        sidecar = Sidecar(client, self.config, self.capture, compile_fn=compile_fn)
        await sidecar.start()
        return sidecar, client

    async def test_health_and_invalid_protocol(self):
        sidecar, _ = await self.make_sidecar()
        await sidecar.announce_ready()
        await sidecar.dispatch({"command": "health", "request_id": "health-1"})
        await sidecar.dispatch({"command": "unknown", "request_id": "bad-1"})

        self.assertEqual(self.output[0]["type"], "ready")
        self.assertIsNone(self.output[0]["request_id"])
        self.assertEqual(self.output[1]["request_id"], "health-1")
        self.assertEqual(
            [item["type"] for item in self.output[-2:]],
            ["turn.error", "turn.done"],
        )

    async def test_multi_turn_reuses_one_session_and_configures_sdk(self):
        sidecar, client = await self.make_sidecar()
        for request_id, prompt in (("r1", "first"), ("r2", "second")):
            await sidecar.dispatch(
                {
                    "command": "turn.start",
                    "request_id": request_id,
                    "conversation_id": "conversation-a",
                    "prompt": prompt,
                    "prompt_profile": "public_visitor",
                    "prompt_version": 1,
                    "knowledge_version": 1,
                }
            )
            await sidecar.drain_events()

        self.assertEqual(len(client.create_calls), 1)
        self.assertEqual(len(client.resume_calls), 1)
        sdk_id = _fake_sdk_id("conversation-a", "public_visitor")
        session = client.sessions[sdk_id]
        self.assertEqual(session.prompts, ["first", "second"])
        options = client.create_calls[0][1]
        self.assertEqual(options["provider"]["type"], "openai")
        self.assertEqual(
            options["provider"]["base_url"], "http://litellm.invalid/v1"
        )
        self.assertEqual(options["provider"]["wire_api"], "responses")
        self.assertEqual(options["model"], "test-model")
        self.assertEqual(options["available_tools"], [])
        self.assertTrue(options["streaming"])
        self.assertEqual(
            [item["request_id"] for item in self.output if item["type"] == "turn.done"],
            ["r1", "r2"],
        )

    async def test_concurrent_conversations_keep_correlation(self):
        sidecar, _ = await self.make_sidecar()
        await asyncio.gather(
            sidecar.dispatch(
                {
                    "command": "turn.start",
                    "request_id": "left-request",
                    "conversation_id": "left",
                    "prompt": "left",
                    "prompt_profile": "public_visitor",
                    "prompt_version": 1,
                    "knowledge_version": 1,
                }
            ),
            sidecar.dispatch(
                {
                    "command": "turn.start",
                    "request_id": "right-request",
                    "conversation_id": "right",
                    "prompt": "right",
                    "prompt_profile": "public_visitor",
                    "prompt_version": 1,
                    "knowledge_version": 1,
                }
            ),
        )
        await sidecar.drain_events()

        correlated = {
            (item["request_id"], item["conversation_id"])
            for item in self.output
            if item["type"] in ("assistant.delta", "assistant.completed")
        }
        self.assertEqual(
            correlated,
            {("left-request", "left"), ("right-request", "right")},
        )

    async def test_abort_finishes_active_turn(self):
        sidecar, client = await self.make_sidecar("pending")
        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "turn-request",
                "conversation_id": "abort-me",
                "prompt": "wait",
                "prompt_profile": "public_visitor",
                "prompt_version": 1,
                "knowledge_version": 1,
            }
        )
        await sidecar.dispatch(
            {
                "command": "turn.abort",
                "request_id": "abort-request",
                "conversation_id": "abort-me",
            }
        )
        await sidecar.drain_events()

        self.assertEqual(client.sessions[_fake_sdk_id("abort-me", "public_visitor")].abort_calls, 1)
        done = [
            item
            for item in self.output
            if item["type"] == "turn.done"
            and item["request_id"] == "turn-request"
        ]
        self.assertEqual(done[0]["aborted"], True)
        abort_accept = [
            item
            for item in self.output
            if item["type"] == "turn.accepted"
            and item["request_id"] == "abort-request"
        ]
        self.assertEqual(abort_accept[0]["target_request_id"], "turn-request")

    async def test_session_error_is_terminal(self):
        sidecar, _ = await self.make_sidecar("error")
        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "error-request",
                "conversation_id": "error-conversation",
                "prompt": "fail",
                "prompt_profile": "public_visitor",
                "prompt_version": 1,
                "knowledge_version": 1,
            }
        )
        await sidecar.drain_events()

        errors = [item for item in self.output if item["type"] == "turn.error"]
        done = [item for item in self.output if item["type"] == "turn.done"]
        self.assertEqual(errors[0]["error"]["code"], "upstream_error")
        self.assertEqual(errors[0]["error"]["status_code"], 502)
        self.assertEqual(len(done), 1)
        self.assertTrue(done[0]["failed"])

    async def test_abort_error_terminates_abort_request(self):
        sidecar, _ = await self.make_sidecar("abort_error")
        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "active-request",
                "conversation_id": "abort-error",
                "prompt": "wait",
                "prompt_profile": "public_visitor",
                "prompt_version": 1,
                "knowledge_version": 1,
            }
        )
        await sidecar.dispatch(
            {
                "command": "turn.abort",
                "request_id": "failed-abort",
                "conversation_id": "abort-error",
            }
        )

        abort_events = [
            item for item in self.output if item["request_id"] == "failed-abort"
        ]
        self.assertEqual(
            [item["type"] for item in abort_events],
            ["turn.accepted", "turn.error", "turn.done"],
        )
        self.assertTrue(abort_events[-1]["failed"])

    async def test_delete_and_shutdown_release_resources(self):
        sidecar, client = await self.make_sidecar("pending")
        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "active",
                "conversation_id": "delete-me",
                "prompt": "wait",
                "prompt_profile": "public_visitor",
                "prompt_version": 1,
                "knowledge_version": 1,
            }
        )
        _delete_sdk_id = _fake_sdk_id("delete-me", "public_visitor")
        session = client.sessions[_delete_sdk_id]
        await sidecar.dispatch(
            {
                "command": "conversation.delete",
                "request_id": "delete-request",
                "conversation_id": "delete-me",
            }
        )
        await sidecar.dispatch(
            {
                "command": "shutdown",
                "request_id": "shutdown-request",
                "conversation_id": None,
            }
        )

        self.assertEqual(session.disconnect_calls, 1)
        self.assertEqual(client.delete_calls[0], _delete_sdk_id)
        self.assertEqual(
            set(client.delete_calls),
            {
                _fake_sdk_id("delete-me", "public_visitor"),
                _fake_sdk_id("delete-me", "invited_friend"),
                _fake_sdk_id("delete-me", "june_admin"),
            },
        )
        self.assertTrue(client.stopped)
        self.assertTrue(sidecar.shutting_down)
        shutdown = [
            item
            for item in self.output
            if item["request_id"] == "shutdown-request"
        ]
        self.assertEqual(shutdown[0]["type"], "turn.done")

    async def test_idle_and_overflow_sessions_are_evicted(self):
        self.config = replace(
            self.config,
            idle_timeout_seconds=3600,
            max_sessions=1,
        )
        sidecar, client = await self.make_sidecar()
        for conversation_id in ("old", "new"):
            await sidecar.dispatch(
                {
                    "command": "turn.start",
                    "request_id": f"request-{conversation_id}",
                    "conversation_id": conversation_id,
                    "prompt": conversation_id,
                    "prompt_profile": "public_visitor",
                    "prompt_version": 1,
                    "knowledge_version": 1,
                }
            )
            await sidecar.drain_events()
            await asyncio.sleep(0.01)

        old_session = client.sessions[_fake_sdk_id("old", "public_visitor")]
        old_session.disconnect_error = True
        await sidecar.evict_idle_sessions()
        self.assertEqual(old_session.disconnect_calls, 1)
        self.assertNotIn("old", sidecar._conversations)
        self.assertIn("new", sidecar._conversations)

        sidecar._conversations["new"].last_used -= 4000
        await sidecar.evict_idle_sessions()
        self.assertEqual(client.sessions[_fake_sdk_id("new", "public_visitor")].disconnect_calls, 1)
        self.assertEqual(sidecar._conversations, {})
        await sidecar.dispatch(
            {
                "command": "shutdown",
                "request_id": "shutdown-eviction",
                "conversation_id": None,
            }
        )

    async def test_shutdown_waits_for_inflight_session_creation(self):
        client = FakeClient("pending")
        client.resume_started = asyncio.Event()
        client.resume_release = asyncio.Event()
        sidecar = Sidecar(client, self.config, self.capture, compile_fn=_fake_compile_fn)
        await sidecar.start()

        start_task = asyncio.create_task(
            sidecar.dispatch(
                {
                    "command": "turn.start",
                    "request_id": "starting",
                    "conversation_id": "race",
                    "prompt": "wait",
                    "prompt_profile": "public_visitor",
                    "prompt_version": 1,
                    "knowledge_version": 1,
                }
            )
        )
        await client.resume_started.wait()
        shutdown_task = asyncio.create_task(
            sidecar.dispatch(
                {
                    "command": "shutdown",
                    "request_id": "shutdown",
                    "conversation_id": None,
                }
            )
        )
        await asyncio.sleep(0)
        self.assertTrue(sidecar.shutting_down)
        client.resume_release.set()
        await asyncio.gather(start_task, shutdown_task)

        self.assertTrue(client.stopped)
        self.assertEqual(client.sessions[_fake_sdk_id("race", "public_visitor")].disconnect_calls, 1)
        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "too-late",
                "conversation_id": "late",
                "prompt": "no",
                "prompt_profile": "public_visitor",
                "prompt_version": 1,
                "knowledge_version": 1,
            }
        )
        late_error = [
            item
            for item in self.output
            if item["request_id"] == "too-late" and item["type"] == "turn.error"
        ]
        self.assertEqual(late_error[0]["error"]["code"], "shutting_down")

    async def test_command_gates_are_released(self):
        sidecar, _ = await self.make_sidecar()
        for index in range(20):
            await sidecar.dispatch(
                {
                    "command": "unknown",
                    "request_id": f"unknown-{index}",
                    "conversation_id": f"conversation-{index}",
                }
            )
        self.assertEqual(sidecar._conversation_gates, {})

    # ------------------------------------------------------------------
    # New: profile-aware tests
    # ------------------------------------------------------------------

    async def test_three_distinct_profiles_produce_distinct_append_prompts(self):
        sidecar, client = await self.make_sidecar()
        for profile in ("public_visitor", "invited_friend", "june_admin"):
            await sidecar.dispatch(
                {
                    "command": "turn.start",
                    "request_id": f"req-{profile}",
                    "conversation_id": f"conv-{profile}",
                    "prompt": "hello",
                    "prompt_profile": profile,
                    "prompt_version": 1,
                    "knowledge_version": 1,
                }
            )
            await sidecar.drain_events()

        contents = [opts["system_message"]["content"] for _, opts in client.create_calls]
        self.assertEqual(len(contents), 3)
        self.assertEqual(len(set(contents)), 3, "all three profiles must produce distinct content")
        for _, opts in client.create_calls:
            self.assertEqual(opts["system_message"]["mode"], "append")

    async def test_session_options_memory_disabled_and_no_tools(self):
        sidecar, client = await self.make_sidecar()
        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "r-opts",
                "conversation_id": "conv-opts",
                "prompt": "test",
                "prompt_profile": "public_visitor",
                "prompt_version": 1,
                "knowledge_version": 1,
            }
        )
        await sidecar.drain_events()
        _, opts = client.create_calls[0]
        self.assertEqual(opts["memory"], {"enabled": False})
        self.assertEqual(opts["tools"], [])
        self.assertEqual(opts["available_tools"], [])
        self.assertEqual(opts["mcp_servers"], {})
        self.assertTrue(opts["enable_session_store"])

    async def test_missing_profile_rejected_before_client_calls(self):
        sidecar, client = await self.make_sidecar()
        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "r-missing",
                "conversation_id": "conv-missing",
                "prompt": "hello",
            }
        )
        errors = [e for e in self.output if e["type"] == "turn.error"]
        self.assertEqual(errors[0]["error"]["code"], "invalid_prompt_profile")
        done = [e for e in self.output if e["type"] == "turn.done"]
        self.assertTrue(done[0]["failed"])
        self.assertEqual(len(client.create_calls), 0)
        self.assertEqual(len(client.resume_calls), 0)

    async def test_unknown_profile_rejected_before_client_calls(self):
        sidecar, client = await self.make_sidecar()
        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "r-unknown",
                "conversation_id": "conv-unknown",
                "prompt": "hello",
                "prompt_profile": "hacker",
                "prompt_version": 1,
                "knowledge_version": 1,
            }
        )
        errors = [e for e in self.output if e["type"] == "turn.error"]
        self.assertEqual(errors[0]["error"]["code"], "invalid_prompt_profile")
        self.assertEqual(len(client.create_calls), 0)

    async def test_stale_prompt_version_rejected_before_client_calls(self):
        sidecar, client = await self.make_sidecar()
        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "r-stale-pv",
                "conversation_id": "conv-stale-pv",
                "prompt": "hello",
                "prompt_profile": "public_visitor",
                "prompt_version": 999,
                "knowledge_version": 1,
            }
        )
        errors = [e for e in self.output if e["type"] == "turn.error"]
        self.assertEqual(errors[0]["error"]["code"], "invalid_prompt_profile")
        self.assertEqual(len(client.create_calls), 0)

    async def test_stale_knowledge_version_rejected_before_client_calls(self):
        sidecar, client = await self.make_sidecar()
        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "r-stale-kv",
                "conversation_id": "conv-stale-kv",
                "prompt": "hello",
                "prompt_profile": "public_visitor",
                "prompt_version": 1,
                "knowledge_version": 999,
            }
        )
        errors = [e for e in self.output if e["type"] == "turn.error"]
        self.assertEqual(errors[0]["error"]["code"], "invalid_prompt_profile")
        self.assertEqual(len(client.create_calls), 0)

    async def test_bool_prompt_version_rejected_before_client_calls(self):
        sidecar, client = await self.make_sidecar()
        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "r-bool-pv",
                "conversation_id": "conv-bool-pv",
                "prompt": "hello",
                "prompt_profile": "public_visitor",
                "prompt_version": True,
                "knowledge_version": 1,
            }
        )
        errors = [e for e in self.output if e["type"] == "turn.error"]
        self.assertEqual(errors[0]["error"]["code"], "invalid_prompt_profile")
        self.assertEqual(len(client.create_calls), 0)

    async def test_bool_knowledge_version_rejected_before_client_calls(self):
        sidecar, client = await self.make_sidecar()
        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "r-bool-kv",
                "conversation_id": "conv-bool-kv",
                "prompt": "hello",
                "prompt_profile": "public_visitor",
                "prompt_version": 1,
                "knowledge_version": True,
            }
        )
        errors = [e for e in self.output if e["type"] == "turn.error"]
        self.assertEqual(errors[0]["error"]["code"], "invalid_prompt_profile")
        self.assertEqual(len(client.create_calls), 0)

    async def test_same_profile_reuses_existing_session(self):
        sidecar, client = await self.make_sidecar()
        for i in range(3):
            await sidecar.dispatch(
                {
                    "command": "turn.start",
                    "request_id": f"r-reuse-{i}",
                    "conversation_id": "conv-reuse",
                    "prompt": f"message {i}",
                    "prompt_profile": "public_visitor",
                    "prompt_version": 1,
                    "knowledge_version": 1,
                }
            )
            await sidecar.drain_events()

        self.assertEqual(len(client.create_calls), 1)
        self.assertEqual(len(client.resume_calls), 1)
        session = client.sessions[_fake_sdk_id("conv-reuse", "public_visitor")]
        self.assertEqual(session.prompts, ["message 0", "message 1", "message 2"])

    async def test_profile_switch_disconnects_and_resumes_with_new_prompt(self):
        sidecar, client = await self.make_sidecar()

        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "r-switch-1",
                "conversation_id": "conv-switch",
                "prompt": "first",
                "prompt_profile": "public_visitor",
                "prompt_version": 1,
                "knowledge_version": 1,
            }
        )
        await sidecar.drain_events()
        visitor_sdk_id = _fake_sdk_id("conv-switch", "public_visitor")
        friend_sdk_id = _fake_sdk_id("conv-switch", "invited_friend")
        first_session = client.sessions[visitor_sdk_id]
        self.assertEqual(len(client.create_calls), 1)

        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "r-switch-2",
                "conversation_id": "conv-switch",
                "prompt": "second",
                "prompt_profile": "invited_friend",
                "prompt_version": 1,
                "knowledge_version": 1,
            }
        )
        await sidecar.drain_events()

        # Old session is disconnected and permanently deleted before new one opens.
        self.assertEqual(first_session.disconnect_calls, 1)
        self.assertIn(visitor_sdk_id, client.delete_calls)

        # The new session uses the invited_friend derived ID, not the visitor one.
        resume_id, resume_opts = client.resume_calls[-1]
        self.assertEqual(resume_id, friend_sdk_id)
        self.assertNotEqual(friend_sdk_id, visitor_sdk_id)
        self.assertEqual(resume_opts["system_message"]["mode"], "append")
        expected_content = _FAKE_PROFILES["invited_friend"]["content"]
        self.assertEqual(resume_opts["system_message"]["content"], expected_content)

        # The newly opened session is a distinct object.
        second_session = client.sessions[friend_sdk_id]
        self.assertIsNot(second_session, first_session)

        done = [e for e in self.output if e["type"] == "turn.done" and e["request_id"] == "r-switch-2"]
        self.assertFalse(done[0].get("failed", False))

    async def test_concurrent_conversations_profile_isolation(self):
        sidecar, client = await self.make_sidecar()
        await asyncio.gather(
            sidecar.dispatch(
                {
                    "command": "turn.start",
                    "request_id": "req-visitor",
                    "conversation_id": "conv-visitor",
                    "prompt": "hello visitor",
                    "prompt_profile": "public_visitor",
                    "prompt_version": 1,
                    "knowledge_version": 1,
                }
            ),
            sidecar.dispatch(
                {
                    "command": "turn.start",
                    "request_id": "req-admin",
                    "conversation_id": "conv-admin",
                    "prompt": "hello admin",
                    "prompt_profile": "june_admin",
                    "prompt_version": 1,
                    "knowledge_version": 1,
                }
            ),
        )
        await sidecar.drain_events()

        visitor_sdk_id = _fake_sdk_id("conv-visitor", "public_visitor")
        admin_sdk_id = _fake_sdk_id("conv-admin", "june_admin")
        options_by_sdk_id = {sdk_id: opts for sdk_id, opts in client.create_calls}
        visitor_content = options_by_sdk_id[visitor_sdk_id]["system_message"]["content"]
        admin_content = options_by_sdk_id[admin_sdk_id]["system_message"]["content"]
        self.assertNotEqual(visitor_content, admin_content)
        self.assertEqual(visitor_content, _FAKE_PROFILES["public_visitor"]["content"])
        self.assertEqual(admin_content, _FAKE_PROFILES["june_admin"]["content"])

    async def test_profile_switch_while_active_is_rejected(self):
        sidecar, client = await self.make_sidecar("pending")

        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "active-req",
                "conversation_id": "conv-active-switch",
                "prompt": "wait",
                "prompt_profile": "public_visitor",
                "prompt_version": 1,
                "knowledge_version": 1,
            }
        )

        await sidecar.dispatch(
            {
                "command": "turn.start",
                "request_id": "switch-req",
                "conversation_id": "conv-active-switch",
                "prompt": "switch",
                "prompt_profile": "invited_friend",
                "prompt_version": 1,
                "knowledge_version": 1,
            }
        )

        errors = [e for e in self.output if e["type"] == "turn.error" and e["request_id"] == "switch-req"]
        self.assertTrue(len(errors) > 0)
        done = [e for e in self.output if e["type"] == "turn.done" and e["request_id"] == "switch-req"]
        self.assertTrue(done[0]["failed"])
        # Active-turn switch must not trigger any cleanup at all.
        sdk_id = _fake_sdk_id("conv-active-switch", "public_visitor")
        self.assertEqual(client.sessions[sdk_id].disconnect_calls, 0)
        self.assertEqual(client.delete_calls, [])

    async def test_startup_compilation_failure_prevents_readiness(self):
        call_count = {"n": 0}

        def failing_compile_fn(profile: str) -> dict:
            call_count["n"] += 1
            raise ValueError(f"corrupted assets for {profile!r}")

        client = FakeClient()
        sidecar = Sidecar(client, self.config, self.capture, compile_fn=failing_compile_fn)
        with self.assertRaises(ValueError) as ctx:
            await sidecar.start()
        self.assertIn("corrupted", str(ctx.exception))
        self.assertFalse(client.started)
        self.assertGreater(call_count["n"], 0)

    # ------------------------------------------------------------------
    # Derived session ID security tests
    # ------------------------------------------------------------------

    async def test_distinct_profiles_produce_distinct_sdk_session_ids(self):
        """Different profiles on the same conversation must never share an SDK ID."""
        ids = {
            profile: _fake_sdk_id("conv-same", profile)
            for profile in ("public_visitor", "invited_friend", "june_admin")
        }
        self.assertEqual(len(set(ids.values())), 3, "each profile needs a unique SDK ID")
        for sid in ids.values():
            uuid.UUID(sid)  # every value must be a valid UUID

    async def test_same_profile_produces_same_sdk_id_across_restarts(self):
        """The derived ID is deterministic: a sidecar restart resumes the same session."""
        client = FakeClient()
        sidecar1 = Sidecar(client, self.config, self.capture, compile_fn=_fake_compile_fn)
        await sidecar1.start()
        await sidecar1.dispatch({
            "command": "turn.start",
            "request_id": "restart-1",
            "conversation_id": "conv-restart",
            "prompt": "hello",
            "prompt_profile": "public_visitor",
            "prompt_version": 1,
            "knowledge_version": 1,
        })
        await sidecar1.drain_events()

        # Simulate sidecar restart: new Sidecar instance, same FakeClient (SDK store).
        sidecar2 = Sidecar(client, self.config, self.capture, compile_fn=_fake_compile_fn)
        await sidecar2.start()
        await sidecar2.dispatch({
            "command": "turn.start",
            "request_id": "restart-2",
            "conversation_id": "conv-restart",
            "prompt": "still here",
            "prompt_profile": "public_visitor",
            "prompt_version": 1,
            "knowledge_version": 1,
        })
        await sidecar2.drain_events()

        # Sidecar1 tries resume (fails - no session yet) then creates.
        # Sidecar2 tries resume (succeeds - session persists in SDK store).
        # Therefore exactly one create, two resume attempts, both on the same derived ID.
        self.assertEqual(len(client.create_calls), 1)
        self.assertEqual(len(client.resume_calls), 2)
        derived = _fake_sdk_id("conv-restart", "public_visitor")
        self.assertEqual(client.create_calls[0][0], derived)
        self.assertTrue(all(r == derived for r, _ in client.resume_calls))

    async def test_profile_switch_permanently_deletes_old_sdk_session(self):
        """Switching profile must delete the old derived SDK session before creating new."""
        sidecar, client = await self.make_sidecar()
        await sidecar.dispatch({
            "command": "turn.start",
            "request_id": "admin-turn",
            "conversation_id": "conv-priv",
            "prompt": "admin question",
            "prompt_profile": "june_admin",
            "prompt_version": 1,
            "knowledge_version": 1,
        })
        await sidecar.drain_events()

        admin_sdk_id = _fake_sdk_id("conv-priv", "june_admin")
        visitor_sdk_id = _fake_sdk_id("conv-priv", "public_visitor")
        self.assertNotEqual(admin_sdk_id, visitor_sdk_id)

        await sidecar.dispatch({
            "command": "turn.start",
            "request_id": "visitor-turn",
            "conversation_id": "conv-priv",
            "prompt": "public question",
            "prompt_profile": "public_visitor",
            "prompt_version": 1,
            "knowledge_version": 1,
        })
        await sidecar.drain_events()

        # Admin session must be permanently deleted before visitor session opens.
        self.assertIn(admin_sdk_id, client.delete_calls)
        # Admin session must not be in the live sessions map.
        self.assertNotIn(admin_sdk_id, client.sessions)
        # Visitor session is a distinct object.
        self.assertIn(visitor_sdk_id, client.sessions)

    async def test_delete_conversation_uses_derived_sdk_id_not_conversation_id(self):
        """conversation.delete must call delete_session with the derived UUID."""
        sidecar, client = await self.make_sidecar()
        await sidecar.dispatch({
            "command": "turn.start",
            "request_id": "setup-turn",
            "conversation_id": "conv-del-check",
            "prompt": "hi",
            "prompt_profile": "june_admin",
            "prompt_version": 1,
            "knowledge_version": 1,
        })
        await sidecar.drain_events()

        admin_sdk_id = _fake_sdk_id("conv-del-check", "june_admin")

        await sidecar.dispatch({
            "command": "conversation.delete",
            "request_id": "del-req",
            "conversation_id": "conv-del-check",
        })

        self.assertEqual(client.delete_calls[0], admin_sdk_id)
        self.assertEqual(
            set(client.delete_calls),
            {
                _fake_sdk_id("conv-del-check", "public_visitor"),
                _fake_sdk_id("conv-del-check", "invited_friend"),
                _fake_sdk_id("conv-del-check", "june_admin"),
            },
        )
        self.assertNotIn("conv-del-check", client.delete_calls)

    async def test_delete_nonexistent_conversation_cleans_known_profile_ids(self):
        """An uncached delete purges every currently known derived session ID."""
        sidecar, client = await self.make_sidecar()
        await sidecar.dispatch({
            "command": "conversation.delete",
            "request_id": "del-ghost",
            "conversation_id": "ghost-conv",
        })

        self.assertEqual(
            set(client.delete_calls),
            {
                _fake_sdk_id("ghost-conv", "public_visitor"),
                _fake_sdk_id("ghost-conv", "invited_friend"),
                _fake_sdk_id("ghost-conv", "june_admin"),
            },
        )
        self.assertNotIn("ghost-conv", client.delete_calls)
        done = [e for e in self.output if e["type"] == "turn.done" and e["request_id"] == "del-ghost"]
        self.assertEqual(done[0]["action"], "conversation.delete")

    async def test_admin_session_not_reachable_via_visitor_derived_id(self):
        """A visitor SDK ID must differ from the admin one for the same conversation."""
        admin_id = _fake_sdk_id("shared-conv", "june_admin")
        visitor_id = _fake_sdk_id("shared-conv", "public_visitor")
        self.assertNotEqual(admin_id, visitor_id)

        sidecar, client = await self.make_sidecar()
        # Establish an admin session.
        await sidecar.dispatch({
            "command": "turn.start",
            "request_id": "admin-req",
            "conversation_id": "shared-conv",
            "prompt": "secret",
            "prompt_profile": "june_admin",
            "prompt_version": 1,
            "knowledge_version": 1,
        })
        await sidecar.drain_events()

        # The admin session object must NOT be accessible under the visitor-derived ID.
        admin_session = client.sessions.get(admin_id)
        self.assertIsNotNone(admin_session)
        self.assertIsNone(client.sessions.get(visitor_id),
                          "visitor SDK ID must not map to any session object at this point")


# ---------------------------------------------------------------------------
# _validate_history unit tests
# ---------------------------------------------------------------------------

class ValidateHistoryTest(unittest.TestCase):
    def test_none_returns_empty_list(self):
        self.assertEqual(_validate_history(None), [])

    def test_empty_list_accepted(self):
        self.assertEqual(_validate_history([]), [])

    def test_valid_two_entries(self):
        hist = [
            {"role": "user", "content": "hello"},
            {"role": "assistant", "content": "hi"},
        ]
        result = _validate_history(hist)
        self.assertEqual(result, hist)

    def test_not_a_list_raises(self):
        for bad in (42, "string", True, False, {}, object()):
            with self.assertRaises(ValueError, msg=f"should reject {bad!r}"):
                _validate_history(bad)

    def test_too_many_entries_raises(self):
        entries = [{"role": "user", "content": "x"}] * (_HISTORY_MAX_ENTRIES + 1)
        with self.assertRaises(ValueError):
            _validate_history(entries)

    def test_exactly_max_entries_accepted(self):
        entries = [
            {"role": "user" if i % 2 == 0 else "assistant", "content": "x"}
            for i in range(_HISTORY_MAX_ENTRIES)
        ]
        result = _validate_history(entries)
        self.assertEqual(len(result), _HISTORY_MAX_ENTRIES)

    def test_non_object_entry_raises(self):
        for bad_entry in (42, "string", True, None, []):
            with self.assertRaises(ValueError):
                _validate_history([bad_entry])

    def test_invalid_role_raises(self):
        for bad_role in ("system", "SYSTEM", "User", "ASSISTANT", "", " user"):
            with self.assertRaises(ValueError, msg=f"role {bad_role!r} must be rejected"):
                _validate_history([{"role": bad_role, "content": "x"}])

    def test_bool_role_raises(self):
        with self.assertRaises(ValueError):
            _validate_history([{"role": True, "content": "x"}])

    def test_none_content_raises(self):
        with self.assertRaises(ValueError):
            _validate_history([{"role": "user", "content": None}])

    def test_bool_content_raises(self):
        with self.assertRaises(ValueError):
            _validate_history([{"role": "user", "content": True}])

    def test_int_content_raises(self):
        with self.assertRaises(ValueError):
            _validate_history([{"role": "user", "content": 42}])

    def test_extra_key_raises(self):
        with self.assertRaises(ValueError):
            _validate_history([{"role": "user", "content": "hi", "injected": "bad"}])

    def test_oversized_total_raises(self):
        # One entry with content just over the byte limit.
        big = "x" * (_HISTORY_MAX_BYTES + 1)
        with self.assertRaises(ValueError):
            _validate_history([{"role": "user", "content": big}])

    def test_total_at_limit_accepted(self):
        # Two entries whose combined bytes sit at or under the limit.
        chunk_size = _HISTORY_MAX_BYTES // 2 - 10  # under limit
        entries = [
            {"role": "user", "content": "a" * chunk_size},
            {"role": "assistant", "content": "b" * chunk_size},
        ]
        result = _validate_history(entries)
        self.assertEqual(len(result), 2)

    def test_special_characters_accepted(self):
        entry = {"role": "user", "content": "hello \"world\" \\ \n"}
        result = _validate_history([entry])
        self.assertEqual(result[0]["content"], entry["content"])

    def test_empty_content_string_accepted(self):
        result = _validate_history([{"role": "user", "content": ""}])
        self.assertEqual(result[0]["content"], "")


# ---------------------------------------------------------------------------
# Sidecar history integration tests
# ---------------------------------------------------------------------------

class SidecarHistoryTest(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        self.output = []

        async def capture(payload):
            self.output.append(payload)

        self.capture = capture
        self.config = SidecarConfig(
            base_url="http://litellm.invalid/v1",
            model="test-model",
            wire_api="responses",
            base_directory="/not-used-by-fake",
        )

    async def make_sidecar(self, behavior="complete"):
        client = FakeClient(behavior)
        sidecar = Sidecar(client, self.config, self.capture, compile_fn=_fake_compile_fn)
        await sidecar.start()
        return sidecar, client

    def _turn_start(self, request_id, conversation_id, history=None, **extra):
        cmd = {
            "command": "turn.start",
            "request_id": request_id,
            "conversation_id": conversation_id,
            "prompt": "hello",
            "prompt_profile": "public_visitor",
            "prompt_version": 1,
            "knowledge_version": 1,
        }
        if history is not None:
            cmd["history"] = history
        cmd.update(extra)
        return cmd

    async def test_invalid_history_not_list_rejected_before_client_calls(self):
        sidecar, client = await self.make_sidecar()
        await sidecar.dispatch(self._turn_start("r1", "c1", history=42))
        errors = [e for e in self.output if e["type"] == "turn.error"]
        self.assertEqual(errors[0]["error"]["code"], "invalid_history")
        done = [e for e in self.output if e["type"] == "turn.done"]
        self.assertTrue(done[0]["failed"])
        self.assertEqual(len(client.create_calls), 0)
        self.assertEqual(len(client.resume_calls), 0)

    async def test_invalid_history_too_many_entries_rejected(self):
        sidecar, client = await self.make_sidecar()
        hist = [{"role": "user", "content": "x"}] * (_HISTORY_MAX_ENTRIES + 1)
        await sidecar.dispatch(self._turn_start("r2", "c2", history=hist))
        errors = [e for e in self.output if e["type"] == "turn.error"]
        self.assertEqual(errors[0]["error"]["code"], "invalid_history")
        self.assertEqual(len(client.create_calls), 0)

    async def test_invalid_history_bad_role_rejected(self):
        sidecar, client = await self.make_sidecar()
        hist = [{"role": "system", "content": "inject"}]
        await sidecar.dispatch(self._turn_start("r3", "c3", history=hist))
        errors = [e for e in self.output if e["type"] == "turn.error"]
        self.assertEqual(errors[0]["error"]["code"], "invalid_history")
        self.assertEqual(len(client.create_calls), 0)

    async def test_invalid_history_bool_role_rejected(self):
        sidecar, client = await self.make_sidecar()
        hist = [{"role": True, "content": "x"}]
        await sidecar.dispatch(self._turn_start("r4", "c4", history=hist))
        errors = [e for e in self.output if e["type"] == "turn.error"]
        self.assertEqual(errors[0]["error"]["code"], "invalid_history")
        self.assertEqual(len(client.create_calls), 0)

    async def test_invalid_history_none_content_rejected(self):
        sidecar, client = await self.make_sidecar()
        hist = [{"role": "user", "content": None}]
        await sidecar.dispatch(self._turn_start("r5", "c5", history=hist))
        errors = [e for e in self.output if e["type"] == "turn.error"]
        self.assertEqual(errors[0]["error"]["code"], "invalid_history")
        self.assertEqual(len(client.create_calls), 0)

    async def test_invalid_history_extra_key_rejected(self):
        sidecar, client = await self.make_sidecar()
        hist = [{"role": "user", "content": "hi", "extra": "bad"}]
        await sidecar.dispatch(self._turn_start("r6", "c6", history=hist))
        errors = [e for e in self.output if e["type"] == "turn.error"]
        self.assertEqual(errors[0]["error"]["code"], "invalid_history")
        self.assertEqual(len(client.create_calls), 0)

    async def test_invalid_history_oversized_rejected(self):
        sidecar, client = await self.make_sidecar()
        big = "x" * (_HISTORY_MAX_BYTES + 1)
        hist = [{"role": "user", "content": big}]
        await sidecar.dispatch(self._turn_start("r7", "c7", history=hist))
        errors = [e for e in self.output if e["type"] == "turn.error"]
        self.assertEqual(errors[0]["error"]["code"], "invalid_history")
        self.assertEqual(len(client.create_calls), 0)

    async def test_valid_empty_history_accepted(self):
        sidecar, client = await self.make_sidecar()
        await sidecar.dispatch(self._turn_start("r8", "c8", history=[]))
        await sidecar.drain_events()
        done = [e for e in self.output if e["type"] == "turn.done" and e["request_id"] == "r8"]
        self.assertFalse(done[0].get("failed", False))
        self.assertEqual(len(client.create_calls), 1)

    async def test_fresh_create_receives_history_in_system_message(self):
        """When resume_session fails (no persisted session), create_session must
        include the PRIOR OWNED CONVERSATION TRANSCRIPT in system_message."""
        sidecar, client = await self.make_sidecar()
        hist = [
            {"role": "user", "content": "prior question"},
            {"role": "assistant", "content": "prior answer"},
        ]
        await sidecar.dispatch(self._turn_start("r-create", "c-create", history=hist))
        await sidecar.drain_events()

        self.assertEqual(len(client.create_calls), 1)
        _, create_opts = client.create_calls[0]
        sys_content = create_opts["system_message"]["content"]
        self.assertIn("PRIOR OWNED CONVERSATION TRANSCRIPT", sys_content)
        self.assertIn("prior question", sys_content)
        self.assertIn("prior answer", sys_content)
        self.assertIn("untrusted", sys_content.lower())
        self.assertEqual(create_opts["system_message"]["mode"], "append")

    async def test_resume_does_not_inject_history(self):
        """An existing persisted session must not receive history in system_message."""
        sidecar, client = await self.make_sidecar()
        hist = [{"role": "user", "content": "prior"}]

        # First turn: creates the session.
        await sidecar.dispatch(self._turn_start("r-resume-1", "c-resume", history=hist))
        await sidecar.drain_events()
        self.assertEqual(len(client.create_calls), 1)
        first_content = client.create_calls[0][1]["system_message"]["content"]

        # Second turn on same conversation: must resume (SDK session persists).
        await sidecar.dispatch(self._turn_start("r-resume-2", "c-resume", history=hist))
        await sidecar.drain_events()
        self.assertEqual(len(client.resume_calls), 1)
        # resume_session does not receive options from create_session call.
        resume_opts = client.resume_calls[0][1]
        resume_sys = resume_opts["system_message"]["content"]
        # The resume options (base options) must not contain the transcript block.
        self.assertNotIn("PRIOR OWNED CONVERSATION TRANSCRIPT", resume_sys)
        # The create call had the transcript; verify only one create happened.
        self.assertEqual(len(client.create_calls), 1)

    async def test_history_with_no_field_uses_base_system_message_on_create(self):
        """Absent history field: create_session uses base system_message without transcript."""
        sidecar, client = await self.make_sidecar()
        await sidecar.dispatch(self._turn_start("r-nofield", "c-nofield"))
        await sidecar.drain_events()
        self.assertEqual(len(client.create_calls), 1)
        _, opts = client.create_calls[0]
        self.assertNotIn("PRIOR OWNED CONVERSATION TRANSCRIPT", opts["system_message"]["content"])

    async def test_profile_switch_create_fallback_gets_history(self):
        """After a profile switch the new derived session is a fresh create;
        the history must be injected into the new session's system_message only."""
        sidecar, client = await self.make_sidecar()

        # First turn: public_visitor session created; no prior history yet.
        await sidecar.dispatch(self._turn_start("r-sw-1", "c-switch"))
        await sidecar.drain_events()
        visitor_sdk_id = _fake_sdk_id("c-switch", "public_visitor")
        friend_sdk_id = _fake_sdk_id("c-switch", "invited_friend")
        self.assertNotEqual(visitor_sdk_id, friend_sdk_id)
        first_create_content = client.create_calls[0][1]["system_message"]["content"]
        # No transcript on initial create (no history provided).
        self.assertNotIn("PRIOR OWNED CONVERSATION TRANSCRIPT", first_create_content)

        # Second turn: invited_friend profile — old session deleted, new fresh create.
        # Now we send history representing the prior visitor exchange.
        friend_hist = [
            {"role": "user", "content": "visitor msg"},
            {"role": "assistant", "content": "answer"},
        ]
        await sidecar.dispatch({
            "command": "turn.start",
            "request_id": "r-sw-2",
            "conversation_id": "c-switch",
            "prompt": "switch question",
            "prompt_profile": "invited_friend",
            "prompt_version": 1,
            "knowledge_version": 1,
            "history": friend_hist,
        })
        await sidecar.drain_events()

        # Old visitor session must be deleted.
        self.assertIn(visitor_sdk_id, client.delete_calls)
        # New session created under invited_friend derived ID.
        create_ids = [sid for sid, _ in client.create_calls]
        self.assertIn(friend_sdk_id, create_ids)

        # Second create must include the history transcript.
        second_create_opts = dict(client.create_calls)[friend_sdk_id]
        second_content = second_create_opts["system_message"]["content"]
        self.assertIn("PRIOR OWNED CONVERSATION TRANSCRIPT", second_content)
        self.assertIn("visitor msg", second_content)
        self.assertIn("answer", second_content)
        self.assertIn("untrusted", second_content.lower())

        done = [e for e in self.output if e["type"] == "turn.done" and e["request_id"] == "r-sw-2"]
        self.assertFalse(done[0].get("failed", False))

    async def test_history_not_exposed_in_events(self):
        """History must not appear in any turn.accepted, assistant.delta, or
        turn.done events emitted to the client."""
        sidecar, client = await self.make_sidecar()
        sensitive = "SENSITIVE_TRANSCRIPT_DATA_XYZ"
        hist = [{"role": "user", "content": sensitive}]
        await sidecar.dispatch(self._turn_start("r-safe", "c-safe", history=hist))
        await sidecar.drain_events()

        for ev in self.output:
            for field in ("delta", "content", "prompt"):
                val = ev.get(field, "")
                if isinstance(val, str):
                    self.assertNotIn(sensitive, val,
                                     f"history leaked into event[{field}]: {ev}")

    async def test_existing_conversation_history_compatibility(self):
        """Conversations without history (pre-change) work normally: no
        transcript block is injected when history is absent/empty."""
        sidecar, client = await self.make_sidecar()
        # Simulate an old-style command with no history key.
        await sidecar.dispatch({
            "command": "turn.start",
            "request_id": "r-compat",
            "conversation_id": "c-compat",
            "prompt": "legacy",
            "prompt_profile": "june_admin",
            "prompt_version": 1,
            "knowledge_version": 1,
        })
        await sidecar.drain_events()
        done = [e for e in self.output if e["type"] == "turn.done" and e["request_id"] == "r-compat"]
        self.assertFalse(done[0].get("failed", False))
        _, opts = client.create_calls[0]
        self.assertNotIn("PRIOR OWNED CONVERSATION TRANSCRIPT", opts["system_message"]["content"])


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