diff --git a/plugins/memory/openviking/__init__.py b/plugins/memory/openviking/__init__.py
index 937def6b9ad4..513e0d1ec23d 100644
--- a/plugins/memory/openviking/__init__.py
+++ b/plugins/memory/openviking/__init__.py
@@ -82,6 +82,7 @@ _SYNC_TRACE_ENV = "HERMES_OPENVIKING_SYNC_TRACE"
 _DEFAULT_RECALL_LIMIT = 6
 _DEFAULT_RECALL_SCORE_THRESHOLD = 0.15
 _DEFAULT_RECALL_MAX_INJECTED_CHARS = 4000
+_DEFAULT_PROFILE_MAX_CHARS = 4000
 _DEFAULT_RECALL_TIMEOUT_SECONDS = 4.0
 _DEFAULT_RECALL_REQUEST_TIMEOUT_SECONDS = 3.0
 _DEFAULT_RECALL_FULL_READ_LIMIT = 2
@@ -89,6 +90,12 @@ _RECALL_QUERY_MIN_CHARS = 5
 _RECALL_MIN_TIMEOUT_SECONDS = 0.05
 _READ_BATCH_LIMIT = 3
 _READ_BATCH_FULL_LIMIT = 2500
+_PROFILE_READS = (
+    ("viking://user/memories/profile.md", "full"),
+)
+_PREFERENCES_OVERVIEW_URI = "viking://user/memories/preferences/"
+_ENTITIES_OVERVIEW_URI = "viking://user/memories/entities/"
+_SESSION_START_TRUNCATION_SUFFIX = "[...] truncated"
 
 # Maps the viking_remember `category` enum to a viking:// subdirectory.
 # Keep in sync with REMEMBER_SCHEMA.parameters.properties.category.enum.
@@ -1842,6 +1849,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
         self._runtime_start_thread: Optional[threading.Thread] = None
         self._memory_write_lock = threading.Lock()
         self._memory_write_threads: Set[threading.Thread] = set()
+        self._profile_prefetched_sessions: Set[str] = set()
         # Set on shutdown so deferred-commit / writer finalizers stop issuing
         # network writes against a torn-down provider.
         self._shutting_down = False
@@ -1915,6 +1923,12 @@ class OpenVikingMemoryProvider(MemoryProvider):
                 "default": _DEFAULT_RECALL_MAX_INJECTED_CHARS,
                 "env_var": "OPENVIKING_RECALL_MAX_INJECTED_CHARS",
             },
+            {
+                "key": "profile_max_chars",
+                "description": "Maximum session-start memory characters injected",
+                "default": _DEFAULT_PROFILE_MAX_CHARS,
+                "env_var": "OPENVIKING_PROFILE_MAX_CHARS",
+            },
             {
                 "key": "recall_timeout_seconds",
                 "description": "Total timeout for recall (seconds)",
@@ -2170,6 +2184,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
                 hermes_home = str(Path.home() / ".hermes")
         self._hermes_home = hermes_home
         self._acquire_run_lock()
+        self._profile_prefetched_sessions.clear()
         warning_callback = (
             kwargs.get("warning_callback")
             if kwargs.get("platform") == "cli"
@@ -2250,7 +2265,8 @@ class OpenVikingMemoryProvider(MemoryProvider):
                 "# OpenViking Knowledge Base\n"
                 f"Active. Endpoint: {self._endpoint}\n"
                 "Use viking_search, viking_read, viking_browse, "
-                "viking_remember, viking_forget, viking_add_resource. "
+                "viking_remember, viking_forget, "
+                "viking_add_resource. "
                 "If repeated searches "
                 "return the same evidence or no stronger evidence, answer "
                 "from available evidence and state uncertainty if needed."
@@ -2259,17 +2275,24 @@ class OpenVikingMemoryProvider(MemoryProvider):
     def prefetch(self, query: str, *, session_id: str = "") -> str:
         """Return recall context for this query/session."""
         query_text = _derive_openviking_user_text(query).strip()
-        if not self._client or len(query_text) < _RECALL_QUERY_MIN_CHARS:
+        if not self._client:
             return ""
 
         effective_session_id = str(session_id or self._session_id or "").strip()
-        result = self._search_prefetch_context(
-            query_text,
-            session_id=effective_session_id,
-        )
-        if not result:
+        parts: List[str] = []
+        session_memory = self._session_start_memory_context(effective_session_id)
+        if session_memory:
+            parts.append(session_memory)
+        if len(query_text) >= _RECALL_QUERY_MIN_CHARS:
+            result = self._search_prefetch_context(
+                query_text,
+                session_id=effective_session_id,
+            )
+            if result:
+                parts.append(result)
+        if not parts:
             return ""
-        return f"## OpenViking Context\n{result}"
+        return "## OpenViking Context\n" + "\n\n".join(parts)
 
     @staticmethod
     def _remaining_recall_timeout(deadline: float, per_request_timeout: float) -> float:
@@ -2946,6 +2969,262 @@ class OpenVikingMemoryProvider(MemoryProvider):
             "resources": self._env_bool("OPENVIKING_RECALL_RESOURCES", False),
         }
 
+    def _profile_max_chars(self) -> int:
+        return self._env_int(
+            "OPENVIKING_PROFILE_MAX_CHARS",
+            _DEFAULT_PROFILE_MAX_CHARS,
+            minimum=200,
+            maximum=50000,
+        )
+
+    @staticmethod
+    def _extract_text_content(resp: Any) -> str:
+        result = OpenVikingMemoryProvider._unwrap_result(resp)
+        if isinstance(result, str):
+            return result.strip()
+        if isinstance(result, dict):
+            return str(result.get("content") or result.get("text") or "").strip()
+        return ""
+
+    @staticmethod
+    def _is_placeholder_overview(content: str) -> bool:
+        normalized = " ".join((content or "").strip().lower().split())
+        return normalized in {
+            "[directory overview is not ready]",
+            "[directory overview is not generated]",
+            "[directory abstract is not ready]",
+            "[directory abstract is not generated]",
+        } or normalized.endswith((
+            "[directory overview is not ready]",
+            "[directory overview is not generated]",
+            "[directory abstract is not ready]",
+            "[directory abstract is not generated]",
+        ))
+
+    @staticmethod
+    def _weighted_context_len(content: str) -> int:
+        total = 0
+        for ch in content:
+            total += 2 if ord(ch) >= 0x3000 else 1
+        return total
+
+    @classmethod
+    def _take_weighted_prefix(cls, content: str, max_chars: int) -> str:
+        if max_chars <= 0:
+            return ""
+        used = 0
+        end = 0
+        for end, ch in enumerate(content):
+            used += 2 if ord(ch) >= 0x3000 else 1
+            if used > max_chars:
+                return content[:end]
+        return content
+
+    @classmethod
+    def _take_weighted_suffix(cls, content: str, max_chars: int) -> str:
+        if max_chars <= 0:
+            return ""
+        used = 0
+        start = len(content)
+        for idx in range(len(content) - 1, -1, -1):
+            ch = content[idx]
+            used += 2 if ord(ch) >= 0x3000 else 1
+            if used > max_chars:
+                return content[start:]
+            start = idx
+        return content
+
+    @classmethod
+    def _truncate_text_content(cls, content: str, max_chars: int) -> str:
+        content = content.strip()
+        if cls._weighted_context_len(content) <= max_chars:
+            return content
+        suffix = f"\n\n{_SESSION_START_TRUNCATION_SUFFIX}"
+        suffix_len = cls._weighted_context_len(suffix)
+        if max_chars <= suffix_len:
+            return cls._take_weighted_prefix(content, max_chars)
+        return cls._take_weighted_prefix(content, max_chars - suffix_len).rstrip() + suffix
+
+    @classmethod
+    def _truncate_profile_content(cls, content: str, max_chars: int) -> str:
+        content = content.strip()
+        if cls._weighted_context_len(content) <= max_chars:
+            return content
+        marker = f"\n\n{_SESSION_START_TRUNCATION_SUFFIX}\n\n"
+        marker_len = cls._weighted_context_len(marker)
+        if max_chars <= marker_len + 40:
+            return cls._truncate_text_content(content, max_chars)
+        remaining = max_chars - marker_len
+        head_budget = max(1, remaining // 2)
+        tail_budget = max(1, remaining - head_budget)
+        head = cls._take_weighted_prefix(content, head_budget).rstrip()
+        tail = cls._take_weighted_suffix(content, tail_budget).lstrip()
+        if not head or not tail:
+            return cls._truncate_text_content(content, max_chars)
+        return f"{head}{marker}{tail}"
+
+    def _read_session_start_text(
+        self,
+        client: _VikingClient,
+        endpoint: str,
+        uri: str,
+        *,
+        skip_placeholder: bool = False,
+        timeout: Optional[float] = None,
+    ) -> Optional[str]:
+        try:
+            kwargs = {"timeout": timeout} if timeout is not None else {}
+            resp = client.get(endpoint, params={"uri": uri}, **kwargs)
+        except Exception as e:
+            if _status_code_from_error(e) in {404, 410}:
+                return ""
+            return None
+        content = self._extract_text_content(resp)
+        if skip_placeholder and self._is_placeholder_overview(content):
+            return ""
+        return content
+
+    def _read_session_start_memory_parts(
+        self,
+        *,
+        client: Optional[_VikingClient] = None,
+        request_timeout: Optional[float] = None,
+    ) -> Dict[str, Optional[str]]:
+        active_client = client or self._client
+        if not active_client:
+            return {}
+
+        profile_uri, _level = _PROFILE_READS[0]
+        return {
+            "profile": self._read_session_start_text(
+                active_client,
+                "/api/v1/content/read",
+                profile_uri,
+                timeout=request_timeout,
+            ),
+            "preferences": self._read_session_start_text(
+                active_client,
+                "/api/v1/content/overview",
+                _PREFERENCES_OVERVIEW_URI,
+                skip_placeholder=True,
+                timeout=request_timeout,
+            ),
+            "entities": self._read_session_start_text(
+                active_client,
+                "/api/v1/content/overview",
+                _ENTITIES_OVERVIEW_URI,
+                skip_placeholder=True,
+                timeout=request_timeout,
+            ),
+        }
+
+    @staticmethod
+    def _format_session_start_memory_block(
+        *,
+        profile: str = "",
+        preferences: str = "",
+        entities: str = "",
+    ) -> str:
+        lines: List[str] = ["## Session Memory"]
+        if profile:
+            lines.extend([
+                '  <user-profile uri="viking://user/memories/profile.md">',
+                profile,
+                "  </user-profile>",
+            ])
+        if preferences or entities:
+            lines.append("  <available-memories>")
+            if preferences:
+                lines.extend([
+                    f'    <preferences uri="{_PREFERENCES_OVERVIEW_URI}">',
+                    preferences,
+                    "    </preferences>",
+                ])
+            if entities:
+                lines.extend([
+                    f'    <entities uri="{_ENTITIES_OVERVIEW_URI}">',
+                    entities,
+                    "    </entities>",
+                ])
+            lines.append("  </available-memories>")
+        if len(lines) == 1:
+            return ""
+        return "\n".join(lines)
+
+    def _budget_session_start_memory_parts(self, parts: Dict[str, str], max_chars: int) -> Dict[str, str]:
+        profile = parts.get("profile", "").strip()
+        preferences = parts.get("preferences", "").strip()
+        entities = parts.get("entities", "").strip()
+        full = self._format_session_start_memory_block(
+            profile=profile,
+            preferences=preferences,
+            entities=entities,
+        )
+        if not full or self._weighted_context_len(full) <= max_chars:
+            return {"profile": profile, "preferences": preferences, "entities": entities}
+
+        placeholder = "\0"
+        placeholder_block = self._format_session_start_memory_block(
+            profile=placeholder if profile else "",
+            preferences=placeholder if preferences else "",
+            entities=placeholder if entities else "",
+        )
+        present_count = sum(1 for value in (profile, preferences, entities) if value)
+        overhead = self._weighted_context_len(placeholder_block) - present_count
+        content_budget = max_chars - overhead
+        if content_budget <= 0:
+            return {"profile": "", "preferences": "", "entities": ""}
+
+        has_listings = bool(preferences or entities)
+        result = {"profile": "", "preferences": "", "entities": ""}
+        remaining = content_budget
+        if profile:
+            profile_budget = remaining if not has_listings else min(
+                self._weighted_context_len(profile),
+                max(0, int(content_budget * 0.6)),
+            )
+            result["profile"] = self._truncate_profile_content(profile, profile_budget)
+            remaining -= self._weighted_context_len(result["profile"])
+
+        listing_values = {"preferences": preferences, "entities": entities}
+        listing_keys = [key for key, value in listing_values.items() if value]
+        for idx, key in enumerate(listing_keys):
+            if remaining <= 0:
+                break
+            value = listing_values[key]
+            budget = remaining if idx == len(listing_keys) - 1 else max(0, remaining // 2)
+            result[key] = self._truncate_text_content(value, budget)
+            remaining -= self._weighted_context_len(result[key])
+        return result
+
+    def _session_start_memory_context(self, session_id: str) -> str:
+        session_key = session_id or self._session_id or "__openviking_default_session__"
+        if session_key in self._profile_prefetched_sessions:
+            return ""
+        try:
+            cfg = self._recall_config()
+            raw_parts = self._read_session_start_memory_parts(
+                request_timeout=cfg["request_timeout_seconds"],
+            )
+        except Exception as e:
+            logger.debug("OpenViking session-start memory prefetch failed: %s", e)
+            return ""
+        profile_failed = raw_parts.get("profile") is None
+        parts = {key: value or "" for key, value in raw_parts.items()}
+        if not any(value.strip() for value in parts.values()):
+            if not profile_failed:
+                self._profile_prefetched_sessions.add(session_key)
+            return ""
+        budgeted = self._budget_session_start_memory_parts(parts, self._profile_max_chars())
+        block = self._format_session_start_memory_block(**budgeted)
+        if not block:
+            if not profile_failed:
+                self._profile_prefetched_sessions.add(session_key)
+            return ""
+        if not profile_failed:
+            self._profile_prefetched_sessions.add(session_key)
+        return block
+
     @staticmethod
     def _clamp_score(value: Any) -> float:
         try:
@@ -3580,6 +3859,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
             return
 
         rewound = bool(kwargs.get("rewound"))
+        compression = kwargs.get("reason") == "compression"
 
         # Rotate cached session state synchronously (cheap, in-memory) and
         # snapshot the old session under the lock so a concurrent sync_turn
@@ -3596,6 +3876,9 @@ class OpenVikingMemoryProvider(MemoryProvider):
                 self._session_id = new_id
                 self._turn_count = 0
 
+        if compression:
+            self._profile_prefetched_sessions.discard(old_session_id or new_id)
+
         if not rotate:
             # Same-session rewind (/undo) or no-op rotation: no commit and no
             # counter reset.
diff --git a/tests/openviking_plugin/test_openviking.py b/tests/openviking_plugin/test_openviking.py
index 70f65fa62cf8..f5a4f84ed7fe 100644
--- a/tests/openviking_plugin/test_openviking.py
+++ b/tests/openviking_plugin/test_openviking.py
@@ -106,6 +106,7 @@ def make_prefetch_provider(monkeypatch, responses, **env):
         "OPENVIKING_RECALL_FULL_READ_LIMIT",
         "OPENVIKING_RECALL_PREFER_ABSTRACT",
         "OPENVIKING_RECALL_RESOURCES",
+        "OPENVIKING_PROFILE_MAX_CHARS",
     ):
         monkeypatch.delenv(key, raising=False)
     for key, value in env.items():
@@ -987,6 +988,9 @@ class TestOpenVikingAutoRecallPrefetch:
                 if parsed.path == "/api/v1/content/read":
                     query = parse_qs(parsed.query)
                     uri = query.get("uri", [""])[0]
+                    if uri.startswith("viking://user/memories/"):
+                        self.send_error(404)
+                        return
                     records["reads"].append(uri)
                     self._send_json({"result": {"content": "E2E full L2 memory content."}})
                     return
@@ -1029,6 +1033,7 @@ class TestOpenVikingAutoRecallPrefetch:
             "OPENVIKING_RECALL_MAX_INJECTED_CHARS",
             "OPENVIKING_RECALL_PREFER_ABSTRACT",
             "OPENVIKING_RECALL_RESOURCES",
+            "OPENVIKING_PROFILE_MAX_CHARS",
             "OPENVIKING_API_KEY",
         ):
             monkeypatch.delenv(key, raising=False)
diff --git a/tests/plugins/memory/test_openviking_provider.py b/tests/plugins/memory/test_openviking_provider.py
index 9d8e39924835..69a5e8619aa7 100644
--- a/tests/plugins/memory/test_openviking_provider.py
+++ b/tests/plugins/memory/test_openviking_provider.py
@@ -35,6 +35,7 @@ def _clear_openviking_env(monkeypatch):
         "OPENVIKING_USER",
         "OPENVIKING_AGENT",
         "OPENVIKING_CLI_CONFIG_FILE",
+        "OPENVIKING_PROFILE_MAX_CHARS",
     ):
         monkeypatch.delenv(key, raising=False)
 
@@ -1521,14 +1522,36 @@ def test_tool_add_resource_sends_git_remote_sources_as_path(url):
     })
 
 
-def test_get_tool_schemas_includes_narrow_forget_tool():
+def test_get_tool_schemas_omits_profile_and_keeps_narrow_forget_tools():
     provider = OpenVikingMemoryProvider()
 
     names = [schema["name"] for schema in provider.get_tool_schemas()]
 
+    assert "viking_profile" not in names
     assert "viking_forget" in names
 
 
+def test_handle_tool_call_profile_returns_unknown_tool():
+    provider = OpenVikingMemoryProvider()
+    provider._client = MagicMock()
+
+    result = json.loads(provider.handle_tool_call("viking_profile", {}))
+
+    assert result["error"] == "Unknown tool: viking_profile"
+    provider._client.get.assert_not_called()
+
+
+def test_system_prompt_block_omits_removed_profile_tool_guidance():
+    provider = OpenVikingMemoryProvider()
+    provider._client = MagicMock()
+    provider._client.get.return_value = {"result": [{"name": "memories"}]}
+
+    prompt = provider.system_prompt_block()
+
+    assert "viking_profile" not in prompt
+    assert "viking_search" in prompt
+
+
 def test_handle_tool_call_forget_deletes_exact_memory_file_uri():
     uri = "viking://user/peers/hermes/memories/preferences/mem_abc123.md"
     provider = OpenVikingMemoryProvider()
@@ -3557,6 +3580,337 @@ def _make_prefetch_provider() -> OpenVikingMemoryProvider:
     return provider
 
 
+def _mock_session_start_reads(provider: OpenVikingMemoryProvider, responses: dict[tuple[str, str], object]):
+    calls = []
+
+    def fake_get(path, params=None, **kwargs):
+        uri = (params or {}).get("uri", "")
+        calls.append((path, uri))
+        response = responses.get((path, uri), "")
+        if isinstance(response, Exception):
+            raise response
+        return {"result": response}
+
+    provider._client.get.side_effect = fake_get
+    return calls
+
+
+def test_prefetch_prepends_session_start_memory_context_once_per_session():
+    provider = _make_prefetch_provider()
+    calls = _mock_session_start_reads(
+        provider,
+        {
+            ("/api/v1/content/read", "viking://user/memories/profile.md"): (
+                "User prefers concise answers."
+            ),
+            ("/api/v1/content/overview", "viking://user/memories/preferences/"): (
+                "# Preferences\n- Keep replies compact."
+            ),
+            ("/api/v1/content/overview", "viking://user/memories/entities/"): (
+                "# Entities\n- Ada Lovelace: collaborator."
+            ),
+        },
+    )
+    provider._search_prefetch_context = MagicMock(return_value="- [events]\n  recalled context")
+
+    first = provider.prefetch("What should we recall?", session_id="sid-123")
+    second = provider.prefetch("What should we recall?", session_id="sid-123")
+
+    assert "## Session Memory" in first
+    assert '<user-profile uri="viking://user/memories/profile.md">' in first
+    assert "User prefers concise answers." in first
+    assert '<preferences uri="viking://user/memories/preferences/">' in first
+    assert "Keep replies compact." in first
+    assert '<entities uri="viking://user/memories/entities/">' in first
+    assert "Ada Lovelace: collaborator." in first
+    assert "recalled context" in first
+    assert "## Session Memory" not in second
+    assert "recalled context" in second
+    assert calls == [
+        ("/api/v1/content/read", "viking://user/memories/profile.md"),
+        ("/api/v1/content/overview", "viking://user/memories/preferences/"),
+        ("/api/v1/content/overview", "viking://user/memories/entities/"),
+    ]
+    assert provider._search_prefetch_context.call_count == 2
+
+
+def test_prefetch_can_return_session_start_memory_for_short_query():
+    provider = _make_prefetch_provider()
+    _mock_session_start_reads(
+        provider,
+        {
+            ("/api/v1/content/read", "viking://user/memories/profile.md"): (
+                "User profile is Ada."
+            ),
+            ("/api/v1/content/overview", "viking://user/memories/preferences/"): "",
+            ("/api/v1/content/overview", "viking://user/memories/entities/"): "",
+        },
+    )
+    provider._search_prefetch_context = MagicMock(return_value="should not run")
+
+    context = provider.prefetch("hi", session_id="sid-123")
+
+    assert "## OpenViking Context" in context
+    assert "## Session Memory" in context
+    assert "User profile is Ada." in context
+    provider._search_prefetch_context.assert_not_called()
+
+
+def test_prefetch_session_start_memory_reads_use_bounded_timeout(monkeypatch):
+    monkeypatch.setenv("OPENVIKING_RECALL_REQUEST_TIMEOUT_SECONDS", "0.25")
+    provider = _make_prefetch_provider()
+    calls = []
+
+    def fake_get(path, params=None, **kwargs):
+        calls.append((path, (params or {}).get("uri", ""), kwargs.get("timeout")))
+        if (params or {}).get("uri") == "viking://user/memories/profile.md":
+            return {"result": "User profile is Ada."}
+        return {"result": ""}
+
+    provider._client.get.side_effect = fake_get
+    provider._search_prefetch_context = MagicMock(return_value="should not run")
+
+    context = provider.prefetch("hi", session_id="sid-123")
+
+    assert "User profile is Ada." in context
+    assert calls == [
+        ("/api/v1/content/read", "viking://user/memories/profile.md", 0.25),
+        ("/api/v1/content/overview", "viking://user/memories/preferences/", 0.25),
+        ("/api/v1/content/overview", "viking://user/memories/entities/", 0.25),
+    ]
+
+
+def test_prefetch_retries_session_start_memory_after_empty_failed_attempt():
+    provider = _make_prefetch_provider()
+    profile_attempts = 0
+
+    def fake_get(path, params=None, **kwargs):
+        nonlocal profile_attempts
+        uri = (params or {}).get("uri", "")
+        if uri == "viking://user/memories/profile.md":
+            profile_attempts += 1
+            if profile_attempts == 1:
+                raise RuntimeError("transient profile read failure")
+            return {"result": "Recovered user profile."}
+        return {"result": ""}
+
+    provider._client.get.side_effect = fake_get
+    provider._search_prefetch_context = MagicMock(return_value="should not run")
+
+    first = provider.prefetch("hi", session_id="sid-123")
+    provider._turn_count = 1
+    second = provider.prefetch("hi", session_id="sid-123")
+
+    assert first == ""
+    assert "Recovered user profile." in second
+    assert profile_attempts == 2
+
+
+def test_prefetch_marks_successful_empty_session_start_memory_as_checked():
+    provider = _make_prefetch_provider()
+    calls = _mock_session_start_reads(
+        provider,
+        {
+            ("/api/v1/content/read", "viking://user/memories/profile.md"): "",
+            ("/api/v1/content/overview", "viking://user/memories/preferences/"): "",
+            ("/api/v1/content/overview", "viking://user/memories/entities/"): "",
+        },
+    )
+    provider._search_prefetch_context = MagicMock(return_value="should not run")
+
+    first = provider.prefetch("hi", session_id="sid-123")
+    second = provider.prefetch("hi", session_id="sid-123")
+
+    assert first == ""
+    assert second == ""
+    assert calls == [
+        ("/api/v1/content/read", "viking://user/memories/profile.md"),
+        ("/api/v1/content/overview", "viking://user/memories/preferences/"),
+        ("/api/v1/content/overview", "viking://user/memories/entities/"),
+    ]
+    provider._search_prefetch_context.assert_not_called()
+
+
+def test_prefetch_marks_checked_when_secondary_session_memory_read_fails():
+    provider = _make_prefetch_provider()
+    calls = []
+
+    def fake_get(path, params=None, **kwargs):
+        uri = (params or {}).get("uri", "")
+        calls.append((path, uri))
+        if uri == "viking://user/memories/profile.md":
+            return {"result": "User profile is Ada."}
+        if uri == "viking://user/memories/entities/":
+            raise RuntimeError("transient entities overview failure")
+        return {"result": ""}
+
+    provider._client.get.side_effect = fake_get
+    provider._search_prefetch_context = MagicMock(return_value="should not run")
+
+    first = provider.prefetch("hi", session_id="sid-123")
+    provider._turn_count = 1
+    second = provider.prefetch("hi", session_id="sid-123")
+
+    assert "User profile is Ada." in first
+    assert second == ""
+    assert calls == [
+        ("/api/v1/content/read", "viking://user/memories/profile.md"),
+        ("/api/v1/content/overview", "viking://user/memories/preferences/"),
+        ("/api/v1/content/overview", "viking://user/memories/entities/"),
+    ]
+    provider._search_prefetch_context.assert_not_called()
+
+
+def test_prefetch_reinjects_after_in_place_compression_same_session():
+    provider = _make_prefetch_provider()
+    provider._session_id = "sid-123"
+    profiles = iter(["Profile before compression.", "Profile after compression."])
+
+    def fake_get(path, params=None, **kwargs):
+        uri = (params or {}).get("uri", "")
+        if uri == "viking://user/memories/profile.md":
+            return {"result": next(profiles)}
+        return {"result": ""}
+
+    provider._client.get.side_effect = fake_get
+    provider._search_prefetch_context = MagicMock(return_value="should not run")
+
+    first = provider.prefetch("hi", session_id="sid-123")
+    provider._turn_count = 3
+    provider.on_session_switch("sid-123", reason="compression")
+    second = provider.prefetch("hi", session_id="sid-123")
+
+    assert "Profile before compression." in first
+    assert "Profile after compression." in second
+
+
+def test_prefetch_reinjects_for_new_session_id():
+    provider = _make_prefetch_provider()
+    profiles = iter(["Session A profile.", "Session B profile."])
+
+    def fake_get(path, params=None, **kwargs):
+        uri = (params or {}).get("uri", "")
+        if uri == "viking://user/memories/profile.md":
+            return {"result": next(profiles)}
+        return {"result": ""}
+
+    provider._client.get.side_effect = fake_get
+    provider._search_prefetch_context = MagicMock(return_value="should not run")
+
+    first = provider.prefetch("hi", session_id="sid-a")
+    second = provider.prefetch("hi", session_id="sid-b")
+
+    assert "Session A profile." in first
+    assert "Session B profile." in second
+
+
+def test_prefetch_degrades_cleanly_when_some_session_memory_parts_are_missing():
+    provider = _make_prefetch_provider()
+    _mock_session_start_reads(
+        provider,
+        {
+            ("/api/v1/content/read", "viking://user/memories/profile.md"): RuntimeError("missing"),
+            ("/api/v1/content/overview", "viking://user/memories/preferences/"): (
+                "# Preferences\n- Likes source-backed answers."
+            ),
+            ("/api/v1/content/overview", "viking://user/memories/entities/"): "",
+        },
+    )
+    provider._search_prefetch_context = MagicMock(return_value="should not run")
+
+    context = provider.prefetch("hi", session_id="sid-123")
+
+    assert "Likes source-backed answers." in context
+    assert "<user-profile" not in context
+    assert "<entities" not in context
+    assert '<preferences uri="viking://user/memories/preferences/">' in context
+
+
+def test_prefetch_omits_placeholder_directory_overviews_from_session_memory():
+    provider = _make_prefetch_provider()
+    _mock_session_start_reads(
+        provider,
+        {
+            ("/api/v1/content/read", "viking://user/memories/profile.md"): (
+                "User profile is Ada."
+            ),
+            ("/api/v1/content/overview", "viking://user/memories/preferences/"): (
+                "# viking://user/memories/preferences/\n\n[Directory abstract is not ready]"
+            ),
+            ("/api/v1/content/overview", "viking://user/memories/entities/"): (
+                "# viking://user/memories/entities/\n\n[Directory overview is not generated]"
+            ),
+        },
+    )
+    provider._search_prefetch_context = MagicMock(return_value="should not run")
+
+    context = provider.prefetch("hi", session_id="sid-123")
+
+    assert "User profile is Ada." in context
+    assert "<available-memories>" not in context
+    assert "Directory abstract is not ready" not in context
+    assert "Directory overview is not generated" not in context
+
+
+def test_session_start_memory_context_respects_total_budget_and_preserves_profile_tail(monkeypatch):
+    monkeypatch.setenv("OPENVIKING_PROFILE_MAX_CHARS", "700")
+    provider = _make_prefetch_provider()
+    long_profile = "\n".join(
+        ["Profile head: user is Ada."]
+        + [f"profile middle {i}: {'x' * 30}" for i in range(40)]
+        + ["Profile tail: recent work is OpenViking."]
+    )
+    _mock_session_start_reads(
+        provider,
+        {
+            ("/api/v1/content/read", "viking://user/memories/profile.md"): long_profile,
+            ("/api/v1/content/overview", "viking://user/memories/preferences/"): (
+                "Preferences overview " + ("p" * 500)
+            ),
+            ("/api/v1/content/overview", "viking://user/memories/entities/"): (
+                "Entities overview " + ("e" * 500)
+            ),
+        },
+    )
+    provider._search_prefetch_context = MagicMock(return_value="should not run")
+
+    context = provider.prefetch("hi", session_id="sid-budget")
+    block = context.removeprefix("## OpenViking Context\n")
+
+    assert len(block) <= 700
+    assert "Profile head: user is Ada." in block
+    assert "Profile tail: recent work is OpenViking." in block
+    assert "[...] truncated" in block
+    assert "viking_profile" not in block
+
+
+def test_prefetch_does_not_auto_inject_memory_overview_when_profile_missing():
+    provider = _make_prefetch_provider()
+    calls = _mock_session_start_reads(
+        provider,
+        {
+            ("/api/v1/content/read", "viking://user/memories/profile.md"): RuntimeError("missing"),
+            ("/api/v1/content/overview", "viking://user/memories/preferences/"): "",
+            ("/api/v1/content/overview", "viking://user/memories/entities/"): "",
+        },
+    )
+    provider._search_prefetch_context = MagicMock(return_value="- [events]\n  recalled context")
+
+    context = provider.prefetch("What should we recall?", session_id="sid-123")
+
+    assert "## Session Memory" not in context
+    assert "recalled context" in context
+    assert calls == [
+        ("/api/v1/content/read", "viking://user/memories/profile.md"),
+        ("/api/v1/content/overview", "viking://user/memories/preferences/"),
+        ("/api/v1/content/overview", "viking://user/memories/entities/"),
+    ]
+    provider._search_prefetch_context.assert_called_once_with(
+        "What should we recall?",
+        session_id="sid-123",
+    )
+
+
 def test_queue_prefetch_is_noop_for_openviking_recall(monkeypatch):
     provider = _make_prefetch_provider()
     constructed_clients = []