--- a/mrjunejune/inference/copilot_sidecar_test.py	Fri Aug 07 07:34:12 2026 -0700
+++ b/mrjunejune/inference/copilot_sidecar_test.py	Fri Aug 07 10:50:30 2026 -0700
@@ -1,9 +1,20 @@
 import asyncio
+import hashlib
 import types
 import unittest
+import uuid
 from dataclasses import replace
 
-from mrjunejune.inference.copilot_sidecar import Sidecar, SidecarConfig
+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):
@@ -13,6 +24,43 @@
     )
 
 
+# ---------------------------------------------------------------------------
+# 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
@@ -142,9 +190,9 @@
             base_directory="/not-used-by-fake",
         )
 
-    async def make_sidecar(self, behavior="complete"):
+    async def make_sidecar(self, behavior="complete", compile_fn=_fake_compile_fn):
         client = FakeClient(behavior)
-        sidecar = Sidecar(client, self.config, self.capture)
+        sidecar = Sidecar(client, self.config, self.capture, compile_fn=compile_fn)
         await sidecar.start()
         return sidecar, client
 
@@ -171,13 +219,17 @@
                     "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)
-        session = client.sessions["conversation-a"]
+        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")
@@ -202,6 +254,9 @@
                     "request_id": "left-request",
                     "conversation_id": "left",
                     "prompt": "left",
+                    "prompt_profile": "public_visitor",
+                    "prompt_version": 1,
+                    "knowledge_version": 1,
                 }
             ),
             sidecar.dispatch(
@@ -210,6 +265,9 @@
                     "request_id": "right-request",
                     "conversation_id": "right",
                     "prompt": "right",
+                    "prompt_profile": "public_visitor",
+                    "prompt_version": 1,
+                    "knowledge_version": 1,
                 }
             ),
         )
@@ -233,6 +291,9 @@
                 "request_id": "turn-request",
                 "conversation_id": "abort-me",
                 "prompt": "wait",
+                "prompt_profile": "public_visitor",
+                "prompt_version": 1,
+                "knowledge_version": 1,
             }
         )
         await sidecar.dispatch(
@@ -244,7 +305,7 @@
         )
         await sidecar.drain_events()
 
-        self.assertEqual(client.sessions["abort-me"].abort_calls, 1)
+        self.assertEqual(client.sessions[_fake_sdk_id("abort-me", "public_visitor")].abort_calls, 1)
         done = [
             item
             for item in self.output
@@ -268,6 +329,9 @@
                 "request_id": "error-request",
                 "conversation_id": "error-conversation",
                 "prompt": "fail",
+                "prompt_profile": "public_visitor",
+                "prompt_version": 1,
+                "knowledge_version": 1,
             }
         )
         await sidecar.drain_events()
@@ -287,6 +351,9 @@
                 "request_id": "active-request",
                 "conversation_id": "abort-error",
                 "prompt": "wait",
+                "prompt_profile": "public_visitor",
+                "prompt_version": 1,
+                "knowledge_version": 1,
             }
         )
         await sidecar.dispatch(
@@ -314,9 +381,13 @@
                 "request_id": "active",
                 "conversation_id": "delete-me",
                 "prompt": "wait",
+                "prompt_profile": "public_visitor",
+                "prompt_version": 1,
+                "knowledge_version": 1,
             }
         )
-        session = client.sessions["delete-me"]
+        _delete_sdk_id = _fake_sdk_id("delete-me", "public_visitor")
+        session = client.sessions[_delete_sdk_id]
         await sidecar.dispatch(
             {
                 "command": "conversation.delete",
@@ -333,7 +404,15 @@
         )
 
         self.assertEqual(session.disconnect_calls, 1)
-        self.assertEqual(client.delete_calls, ["delete-me"])
+        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 = [
@@ -357,12 +436,15 @@
                     "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["old"]
+        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)
@@ -371,7 +453,7 @@
 
         sidecar._conversations["new"].last_used -= 4000
         await sidecar.evict_idle_sessions()
-        self.assertEqual(client.sessions["new"].disconnect_calls, 1)
+        self.assertEqual(client.sessions[_fake_sdk_id("new", "public_visitor")].disconnect_calls, 1)
         self.assertEqual(sidecar._conversations, {})
         await sidecar.dispatch(
             {
@@ -385,7 +467,7 @@
         client = FakeClient("pending")
         client.resume_started = asyncio.Event()
         client.resume_release = asyncio.Event()
-        sidecar = Sidecar(client, self.config, self.capture)
+        sidecar = Sidecar(client, self.config, self.capture, compile_fn=_fake_compile_fn)
         await sidecar.start()
 
         start_task = asyncio.create_task(
@@ -395,6 +477,9 @@
                     "request_id": "starting",
                     "conversation_id": "race",
                     "prompt": "wait",
+                    "prompt_profile": "public_visitor",
+                    "prompt_version": 1,
+                    "knowledge_version": 1,
                 }
             )
         )
@@ -414,13 +499,16 @@
         await asyncio.gather(start_task, shutdown_task)
 
         self.assertTrue(client.stopped)
-        self.assertEqual(client.sessions["race"].disconnect_calls, 1)
+        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 = [
@@ -442,6 +530,823 @@
             )
         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()