diff --git a/clients/terminal/tui/commands/builtin.py b/clients/terminal/tui/commands/builtin.py
index 765a8f8..64987c6 100644
--- a/clients/terminal/tui/commands/builtin.py
+++ b/clients/terminal/tui/commands/builtin.py
@@ -185,13 +185,16 @@
     meta = CommandMeta(
         name="compact",
         aliases=(),
-        description="Compact the current session (ask model to summarize context).",
+        description="Compact the current session (force context compression now).",
         keybind="ctrl+x c",
     )
 
     async def execute(self, ctx: TuiContext, args: str) -> None:
+        # Send a typed control message, not a chat message: the backend runs the
+        # real context compressor (bypassing the token threshold) instead of a
+        # full agent turn that merely produces summary text.
         if ctx.ws_client:
-            ctx.ws_client.enqueue("Please summarize and compact our conversation so far.")
+            ctx.ws_client.enqueue({"type": "compact"})
 
 
 class ThemesCommand(BaseCommand):
diff --git a/clients/terminal/tui/tui_app.py b/clients/terminal/tui/tui_app.py
index a27ea72..a144530 100644
--- a/clients/terminal/tui/tui_app.py
+++ b/clients/terminal/tui/tui_app.py
@@ -408,12 +408,29 @@
         elif msg_type in ("compression_started", "context_compressed"):
             # Compression carries a fresh context-token count (post-compress on
             # context_compressed, pre-compress on compression_started). Update
-            # the gauge; not a chat item.
+            # the gauge.
             self._status_bar.set_context(
                 payload.get("context_tokens"), payload.get("max_context_tokens")
             )
+            # Auto-compression happens mid-turn, while _streaming is True (a
+            # stream_start already kicked off the activity spinner) — just
+            # relabel it. A forced /compact has NO streaming turn, so _streaming
+            # is False: start the spinner on compression_started and stop it +
+            # emit a chat status line on context_compressed as user feedback.
             if msg_type == "compression_started":
-                self._status_bar.activity_label("compressing")
+                if self._streaming:
+                    self._status_bar.activity_label("compressing")
+                else:
+                    self._status_bar.activity_start("compressing")
+            else:  # context_compressed
+                if not self._streaming:
+                    before = payload.get("messages_before")
+                    after = payload.get("messages_after")
+                    if before is not None and after is not None:
+                        self._chat_panel.handle_ws_event(
+                            {"type": "status", "content": f"Context compacted: {before} → {after} messages"}
+                        )
+                    self._status_bar.activity_stop()
             return
         elif msg_type == "stream_delta" and self._streaming:
             # First real output: the agent is now responding, not just thinking.
diff --git a/clients/terminal/ws_client.py b/clients/terminal/ws_client.py
index 1e388ae..ee56003 100644
--- a/clients/terminal/ws_client.py
+++ b/clients/terminal/ws_client.py
@@ -40,14 +40,24 @@
             await self._ws.close()
             self._ws = None
 
-    async def send(self, content: str) -> None:
+    async def send(self, content: str | dict) -> None:
+        """Send a message to the server.
+
+        A ``str`` is wrapped as ``{"type": "message", "content": ..., "cwd": ...}``.
+        A ``dict`` is sent as-is (with ``cwd`` defaulted if absent) — used for
+        typed control messages like ``{"type": "compact"}``.
+        """
         if not self._ws:
             raise RuntimeError("WebSocket is not connected")
-        payload = {
-            "type": "message",
-            "content": content,
-            "cwd": str(self._cwd),
-        }
+        if isinstance(content, dict):
+            payload = dict(content)
+            payload.setdefault("cwd", str(self._cwd))
+        else:
+            payload = {
+                "type": "message",
+                "content": content,
+                "cwd": str(self._cwd),
+            }
         await self._ws.send(json.dumps(payload))
 
     async def receive_loop(self) -> None:
@@ -72,7 +82,7 @@
                 break
             await self.send(content)
 
-    def enqueue(self, content: str) -> None:
+    def enqueue(self, content: str | dict) -> None:
         self._input_queue.put_nowait(content)
 
     def stop_input(self) -> None:
diff --git a/navi/api/websocket.py b/navi/api/websocket.py
index 2f84206..143fc41 100644
--- a/navi/api/websocket.py
+++ b/navi/api/websocket.py
@@ -2,10 +2,12 @@
 
 Protocol (client -> server):
   {"type": "message", "content": "...", "cwd": "..."}
+  {"type": "compact"}                      # force context compression now
 
   `cwd` is optional but recommended for terminal clients; when present the
   server treats it as the user's working directory and resolves relative paths
-  against it.
+  against it. `compact` runs the context compressor immediately (bypassing the
+  token threshold) and streams back CompressionStarted / ContextCompressed.
 
 Protocol (server -> client):
   {"type": "stream_start"}
@@ -220,11 +222,49 @@
                 await websocket.send_json({"type": "error", "message": "Invalid JSON"})
                 continue
 
-            if data.get("type") != "message" or not data.get("content"):
+            msg_type = data.get("type")
+
+            if msg_type == "compact":
+                # Forced context compression (TUI /compact). Runs the real
+                # compressor (bypassing the token threshold) and streams
+                # CompressionStarted / ContextCompressed back. No stream_start
+                # — the client distinguishes a forced compact from an in-turn
+                # auto-compress by the absence of a streaming turn.
+                async with orchestrator.session_lock(session_id):
+                    if orchestrator.is_running(session_id):
+                        await websocket.send_json(
+                            {
+                                "type": "error",
+                                "message": "Agent is running; wait for it to finish before compacting.",
+                            }
+                        )
+                        continue
+                    run = orchestrator.create_run(session_id)
+                    queue = run.subscribe()
+                    current_run = run
+
+                try:
+                    run.task = asyncio.create_task(
+                        orchestrator.run_compact(session_id, session_store)
+                    )
+                except Exception:
+                    orchestrator.clear_run(session_id)
+                    raise
+
+                connected = await _stream_to_client(websocket, queue)
+                run.unsubscribe(queue)
+                queue = None
+                current_run = None
+
+                if not connected:
+                    break  # avoid calling receive_text() on a dead socket
+                continue
+
+            if msg_type != "message" or not data.get("content"):
                 await websocket.send_json(
                     {
                         "type": "error",
-                        "message": "Expected {type: 'message', content: '...', cwd?: '...'}",
+                        "message": "Expected {type: 'message', content: '...', cwd?: '...'} or {type: 'compact'}",
                     }
                 )
                 continue
diff --git a/navi/core/agent.py b/navi/core/agent.py
index 82489d0..d63ce21 100644
--- a/navi/core/agent.py
+++ b/navi/core/agent.py
@@ -33,6 +33,7 @@
     LLMBackendError,
     LLMConnectionError,
     MaxIterationsReached,
+    NothingToCompactError,
     SessionNotFound,
 )
 from navi.llm.base import LLMBackend, Message
@@ -348,6 +349,65 @@
             timeout_seconds=timeout_seconds,
         )
 
+    async def compact_stream(self, session_id: str) -> AsyncGenerator[AgentEvent, None]:
+        """Force context compression now, bypassing the token threshold.
+
+        Triggered by the TUI ``/compact`` command. Unlike the pre/mid-turn
+        auto-compression (gated by ``should_compress``), this always attempts a
+        summary pass as long as there is enough to summarize. Yields
+        ``CompressionStarted`` then ``ContextCompressed`` on success; raises
+        ``NothingToCompactError`` when the context is too small to compress
+        (the orchestrator surfaces it as an error event so the user gets
+        feedback).
+
+        Emits NO ``StreamStart``/``StreamEnd`` — the client distinguishes a
+        forced compact from an in-turn auto-compress by the absence of a
+        streaming turn (the TUI keys off its ``_streaming`` flag).
+        """
+        from navi.tools._internal.base import current_session_id as _sid_var
+
+        session = await self._sessions.get(session_id)
+        if session is None:
+            raise SessionNotFound(session_id)
+        profile = self._profiles.get(session.profile_id)
+        self._set_active_profile(profile)
+        llm = self._get_backend(profile.llm_backend)
+
+        _sid_token = _sid_var.set(session_id)
+        try:
+            if not settings.context_compression_enabled:
+                raise NothingToCompactError("Context compression is disabled.")
+
+            yield CompressionStarted(
+                context_tokens=self._compressor.estimate_context_tokens(session.context),
+                max_context_tokens=settings.ollama_num_ctx,
+            )
+            event = await self._compressor.compress_and_save_session(
+                session=session,
+                session_store=self._sessions,
+                llm=llm,
+                model=profile.model,
+                temperature=settings.context_summary_temperature,
+                session_id=session_id,
+                reason="forced",
+                keep_recent=settings.context_keep_recent,
+                max_tokens=settings.context_summary_max_tokens,
+                # Enable the intra-turn fallback so a single long autonomous turn
+                # (one user message + many tool iterations = one "turn", the
+                # common navi_code shape) can still be compressed — without this
+                # keep_recent_messages, partition_messages finds nothing to
+                # summarize when turns <= keep_recent and forced compact always
+                # reports "nothing to compact". Mirrors the midturn auto-compress.
+                keep_recent_messages=max(12, settings.context_keep_recent * 2),
+            )
+            if event is None:
+                raise NothingToCompactError(
+                    "Nothing to compact yet — the context is still small."
+                )
+            yield event
+        finally:
+            _sid_var.reset(_sid_token)
+
     async def run_stream(
         self,
         session_id: str,
diff --git a/navi/core/orchestrator.py b/navi/core/orchestrator.py
index 8134afe..651619c 100644
--- a/navi/core/orchestrator.py
+++ b/navi/core/orchestrator.py
@@ -197,6 +197,17 @@
         state.run = run
         return run
 
+    def clear_run(self, session_id: str) -> None:
+        """Drop a run that was created but never started (e.g. create_task failed).
+
+        Without this the session would stay ``is_running`` forever, blocking
+        all later turns. ``run_agent``/``run_compact`` clear ``state.run`` in
+        their own finally block, but only if their task actually starts.
+        """
+        state = self._sessions.get(session_id)
+        if state is not None:
+            state.run = None
+
     def mark_busy(self, session_id: str, stop_event: asyncio.Event | None = None) -> None:
         state = self._get_or_create_state(session_id)
         state.busy_event = stop_event or asyncio.Event()
@@ -284,6 +295,38 @@
                 state.run = None
             await self._cleanup(session_id)
 
+    # ── Forced compact (TUI /compact) ───────────────────────────────────────
+
+    async def run_compact(self, session_id: str, session_store) -> None:
+        """Run forced context compression and broadcast the events.
+
+        Mirrors ``run_agent``'s lifecycle: forwards ``CompressionStarted`` /
+        ``ContextCompressed`` from ``Agent.compact_stream`` to subscribers, and
+        surfaces ``NothingToCompactError`` (and any other failure) as an error
+        event so the user gets feedback instead of silence.
+        """
+        from navi.exceptions import NaviError, SessionNotFound
+
+        run = self._sessions[session_id].run
+        agent = self._build_agent(session_store)
+
+        try:
+            async for event in agent.compact_stream(session_id):
+                await run.broadcast(("event", event))
+        except SessionNotFound:
+            await run.broadcast(("error", "Session not found"))
+        except NaviError as e:
+            await run.broadcast(("error", str(e)))
+        except Exception as e:
+            log.exception("ws.compact_error", session_id=session_id)
+            await run.broadcast(("error", f"Internal error: {e}"))
+        finally:
+            await run.broadcast(("done", None))
+            state = self._sessions.get(session_id)
+            if state is not None:
+                state.run = None
+            await self._cleanup(session_id)
+
     # ── Headless recall run ───────────────────────────────────────────────────
 
     async def _finalize_recall(self, recall, scheduler, *, outcome: str) -> None:
diff --git a/navi/exceptions.py b/navi/exceptions.py
index 735d2ac..d42b9a5 100644
--- a/navi/exceptions.py
+++ b/navi/exceptions.py
@@ -46,3 +46,7 @@
 
 class ContextTooLargeError(NaviError):
     """Raised when the estimated context size would exceed the model's safe window."""
+
+
+class NothingToCompactError(NaviError):
+    """Raised when a forced /compact has nothing to summarize (context too small)."""
diff --git a/tests/clients/test_compact_command.py b/tests/clients/test_compact_command.py
new file mode 100644
index 0000000..c2dfae3
--- /dev/null
+++ b/tests/clients/test_compact_command.py
@@ -0,0 +1,36 @@
+"""Tests for the /compact command.
+
+/compact must send a typed ``{"type": "compact"}`` control message — not a chat
+message — so the backend runs the real context compressor instead of a full
+agent turn that merely produces summary text.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from clients.terminal.tui.commands.builtin import CompactCommand
+from clients.terminal.tui.context import TuiContext
+
+
+class _FakeWsClient:
+    def __init__(self) -> None:
+        self.enqueued: list = []
+
+    def enqueue(self, content) -> None:
+        self.enqueued.append(content)
+
+
+def test_compact_enqueues_typed_control_message() -> None:
+    ws = _FakeWsClient()
+    ctx = TuiContext(session_id="s", ws_client=ws)  # type: ignore[arg-type]
+
+    asyncio.run(CompactCommand().execute(ctx, ""))
+
+    assert ws.enqueued == [{"type": "compact"}]
+
+
+def test_compact_no_ws_client_is_noop() -> None:
+    ctx = TuiContext(session_id="s", ws_client=None)
+    # Must not raise when there is no ws_client (e.g. session not attached yet).
+    asyncio.run(CompactCommand().execute(ctx, ""))
\ No newline at end of file
diff --git a/tests/clients/test_ws_client_send.py b/tests/clients/test_ws_client_send.py
new file mode 100644
index 0000000..30733f9
--- /dev/null
+++ b/tests/clients/test_ws_client_send.py
@@ -0,0 +1,63 @@
+"""Unit tests for NaviWebSocketClient.send message framing.
+
+``send`` accepts either a plain string (wrapped as a ``message``) or a dict
+(sent as-is, with ``cwd`` defaulted) — the latter carries typed control
+messages such as ``{"type": "compact"}``. These tests fake the socket so no
+network is needed.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+
+import pytest
+
+from clients.terminal.ws_client import NaviWebSocketClient
+
+
+class _FakeWs:
+    def __init__(self) -> None:
+        self.sent: list[str] = []
+
+    async def send(self, raw: str) -> None:
+        self.sent.append(raw)
+
+
+def _client() -> NaviWebSocketClient:
+    client = NaviWebSocketClient("sess")
+    client._ws = _FakeWs()  # type: ignore[attr-defined]
+    return client
+
+
+def _sent_payloads(client: NaviWebSocketClient) -> list[dict]:
+    return [json.loads(raw) for raw in client._ws.sent]  # type: ignore[attr-defined]
+
+
+def test_send_string_wraps_as_message_with_cwd() -> None:
+    client = _client()
+    asyncio.run(client.send("hello"))
+    [payload] = _sent_payloads(client)
+    assert payload == {"type": "message", "content": "hello", "cwd": str(client._cwd)}
+
+
+def test_send_dict_passes_through_with_default_cwd() -> None:
+    client = _client()
+    asyncio.run(client.send({"type": "compact"}))
+    [payload] = _sent_payloads(client)
+    assert payload["type"] == "compact"
+    # cwd is defaulted in if absent.
+    assert payload["cwd"] == str(client._cwd)
+
+
+def test_send_dict_keeps_existing_cwd() -> None:
+    client = _client()
+    asyncio.run(client.send({"type": "compact", "cwd": "/custom"}))
+    [payload] = _sent_payloads(client)
+    assert payload == {"type": "compact", "cwd": "/custom"}
+
+
+def test_send_raises_when_not_connected() -> None:
+    client = NaviWebSocketClient("sess")  # no _ws set
+    with pytest.raises(RuntimeError, match="not connected"):
+        asyncio.run(client.send({"type": "compact"}))
\ No newline at end of file
diff --git a/tests/unit/api/test_websocket.py b/tests/unit/api/test_websocket.py
index 55c49a2..8f50964 100644
--- a/tests/unit/api/test_websocket.py
+++ b/tests/unit/api/test_websocket.py
@@ -2,7 +2,7 @@
 
 import asyncio
 import json
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 from fastapi import WebSocketDisconnect
@@ -244,3 +244,186 @@
         except asyncio.CancelledError:
             pass
     orchestrator._sessions.pop("s1", None)
+
+
+# ── /compact (forced compression) ───────────────────────────────────────────
+
+
+@pytest.mark.anyio
+async def test_compact_message_runs_forced_compression(mock_websocket, mock_session, mock_user, monkeypatch):
+    """A {"type": "compact"} message dispatches to run_compact (not run_agent),
+    and sends NO stream_start — the client distinguishes a forced compact from an
+    in-turn auto-compress by the absence of a streaming turn."""
+    monkeypatch.setattr(ws_mod, "get_current_user_ws", AsyncMock(return_value=mock_user))
+    mock_store = MagicMock()
+    mock_store.get = AsyncMock(return_value=mock_session)
+    monkeypatch.setattr(ws_mod, "get_session_store", lambda: mock_store)
+    monkeypatch.setattr(ws_mod, "_stream_to_client", AsyncMock(return_value=True))
+
+    fake_container = MagicMock()
+    fake_container.profile_registry = None
+    fake_container.tool_registry = None
+    fake_container.backend_registry = None
+    fake_container.cp_registry = None
+    fake_container.orchestrator = AgentSessionOrchestrator(fake_container)
+    monkeypatch.setattr("navi.api.deps._resolve_container", lambda: fake_container)
+
+    orchestrator = fake_container.orchestrator
+    compact_calls: list = []
+
+    async def fake_run_compact(sid, store):
+        compact_calls.append((sid, store))
+
+    monkeypatch.setattr(orchestrator, "run_compact", fake_run_compact)
+
+    received: list[str] = []
+
+    async def fake_receive_text():
+        if not received:
+            received.append(json.dumps({"type": "compact"}))
+            return received[-1]
+        raise WebSocketDisconnect()
+
+    mock_websocket.receive_text = fake_receive_text
+
+    await ws_mod.websocket_session("s1", mock_websocket)
+
+    # The handler schedules run_compact via create_task but the mocked
+    # _stream_to_client / receive_text never yield to it, so drive it here.
+    state = orchestrator._sessions.get("s1")
+    if state and state.run and state.run.task:
+        try:
+            await state.run.task
+        except asyncio.CancelledError:
+            pass
+
+    assert len(compact_calls) == 1
+    assert compact_calls[0][0] == "s1"
+    # No stream_start is sent for a compact.
+    calls = [c.args[0] for c in mock_websocket.send_json.call_args_list]
+    assert all(c.get("type") != "stream_start" for c in calls)
+
+    orchestrator._sessions.pop("s1", None)
+
+
+@pytest.mark.anyio
+async def test_compact_while_agent_running_is_rejected(mock_websocket, mock_session, mock_user, monkeypatch):
+    """A /compact while an agent turn is active is rejected — compressing
+    session.context mid-turn would race the running agent."""
+    monkeypatch.setattr(ws_mod, "get_current_user_ws", AsyncMock(return_value=mock_user))
+    mock_store = MagicMock()
+    mock_store.get = AsyncMock(return_value=mock_session)
+    monkeypatch.setattr(ws_mod, "get_session_store", lambda: mock_store)
+    monkeypatch.setattr(ws_mod, "_stream_to_client", AsyncMock(return_value=True))
+
+    fake_container = MagicMock()
+    fake_container.profile_registry = None
+    fake_container.tool_registry = None
+    fake_container.backend_registry = None
+    fake_container.cp_registry = None
+    fake_container.orchestrator = AgentSessionOrchestrator(fake_container)
+    monkeypatch.setattr("navi.api.deps._resolve_container", lambda: fake_container)
+
+    orchestrator = fake_container.orchestrator
+
+    async def fake_run_agent(*a, **kw):
+        await asyncio.sleep(3600)
+
+    monkeypatch.setattr(orchestrator, "run_agent", fake_run_agent)
+
+    message_count = 0
+
+    async def fake_receive_text():
+        nonlocal message_count
+        message_count += 1
+        if message_count == 1:
+            return json.dumps({"type": "message", "content": "first"})
+        if message_count == 2:
+            return json.dumps({"type": "compact"})
+        raise WebSocketDisconnect()
+
+    mock_websocket.receive_text = fake_receive_text
+
+    await ws_mod.websocket_session("s1", mock_websocket)
+
+    calls = [c.args[0] for c in mock_websocket.send_json.call_args_list]
+    error_calls = [c for c in calls if c["type"] == "error"]
+    assert len(error_calls) == 1
+    assert "wait for it to finish" in error_calls[0]["message"]
+
+    state = orchestrator._sessions.get("s1")
+    if state and state.run and state.run.task:
+        state.run.task.cancel()
+        try:
+            await state.run.task
+        except asyncio.CancelledError:
+            pass
+    orchestrator._sessions.pop("s1", None)
+
+
+# ── run_compact event/error broadcasting ────────────────────────────────────
+
+
+@pytest.mark.anyio
+async def test_run_compact_broadcasts_events_and_done(fake_orchestrator, monkeypatch):
+    """run_compact forwards agent.compact_stream events to subscribers and ends
+    with a done marker so _stream_to_client returns."""
+    from navi.core.events import CompressionStarted, ContextCompressed
+
+    async def fake_compact_stream(session_id):
+        yield CompressionStarted(context_tokens=100, max_context_tokens=4096)
+        yield ContextCompressed(messages_before=20, messages_after=5, context_tokens=80, max_context_tokens=4096)
+
+    class _FakeAgent:
+        async def compact_stream(self, session_id):
+            async for ev in fake_compact_stream(session_id):
+                yield ev
+
+    def fake_build_agent(self, _store):
+        return _FakeAgent()
+
+    monkeypatch.setattr(AgentSessionOrchestrator, "_build_agent", fake_build_agent)
+
+    run = fake_orchestrator.create_run("s1")
+    queue = run.subscribe()
+    await fake_orchestrator.run_compact("s1", session_store=MagicMock())
+
+    items = []
+    while not queue.empty():
+        items.append(queue.get_nowait())
+    kinds = [item[0] for item in items]
+    assert kinds == ["event", "event", "done"]
+    assert isinstance(items[0][1], CompressionStarted)
+    assert isinstance(items[1][1], ContextCompressed)
+    # run_compact's finally clears state.run, and with no websockets attached
+    # _cleanup then removes the session entry entirely.
+    assert "s1" not in fake_orchestrator._sessions
+
+
+@pytest.mark.anyio
+async def test_run_compact_surfaces_nothing_to_compact_as_error(fake_orchestrator, monkeypatch):
+    """NothingToCompactError is broadcast as an error event (user feedback),
+    followed by done — not raised into the WS handler."""
+    from navi.exceptions import NothingToCompactError
+
+    class _FakeAgent:
+        async def compact_stream(self, session_id):
+            raise NothingToCompactError("nothing")
+            yield  # pragma: no cover - makes this an async generator
+
+    def fake_build_agent(self, _store):
+        return _FakeAgent()
+
+    monkeypatch.setattr(AgentSessionOrchestrator, "_build_agent", fake_build_agent)
+
+    run = fake_orchestrator.create_run("s1")
+    queue = run.subscribe()
+    await fake_orchestrator.run_compact("s1", session_store=MagicMock())
+
+    items = []
+    while not queue.empty():
+        items.append(queue.get_nowait())
+    kinds = [item[0] for item in items]
+    assert kinds == ["error", "done"]
+    assert "nothing" in items[0][1]
+    assert "s1" not in fake_orchestrator._sessions
diff --git a/tests/unit/core/test_agent.py b/tests/unit/core/test_agent.py
index aa16acf..b454b4a 100644
--- a/tests/unit/core/test_agent.py
+++ b/tests/unit/core/test_agent.py
@@ -11,20 +11,17 @@
 
 from navi.core.agent import Agent, _is_casual_message
 from navi.core.events import (
+    CompressionStarted,
+    ContextCompressed,
     ModelInfo,
     StreamEnd,
-    StreamStopped,
     SubagentComplete,
-    TextDelta,
-    ToolEvent,
-    ToolStarted,
 )
-from navi.core.registry import BackendRegistry, ProfileRegistry, ToolRegistry
+from navi.core.registry import BackendRegistry, ProfileRegistry
 from navi.core.session import InMemorySessionStore
-from navi.exceptions import MaxIterationsReached, SessionNotFound
+from navi.exceptions import MaxIterationsReached, NothingToCompactError, SessionNotFound
 from navi.llm.base import LLMChunk, Message, ToolCallRequest
-from navi.tools._internal.base import ToolResult
-from tests.conftest_factory import FakeLLMBackend, FakeTool, make_profile, make_registry_with_tools
+from tests.conftest_factory import FakeLLMBackend, make_profile, make_registry_with_tools
 
 
 @pytest.fixture
@@ -321,6 +318,100 @@
         assert saved.messages[-1].token_count == 50
 
 
+# ─── compact_stream() tests ──────────────────────────────────────────────────
+
+
+class TestAgentCompactStream:
+    """Forced /compact: bypasses the token threshold and runs the real compressor."""
+
+    @pytest_asyncio.fixture
+    async def session_with_history(self, agent, session):
+        """A session with enough turns that there is something to summarize."""
+        # context_keep_recent defaults to 8 turns; add 10 so the oldest turns
+        # fall into to_summarize (>= 2 messages) and compress_context does work.
+        for i in range(10):
+            session.context.append(Message(role="user", content=f"user {i}"))
+            session.context.append(Message(role="assistant", content=f"assistant {i}"))
+            session.messages.append(Message(role="user", content=f"user {i}"))
+            session.messages.append(Message(role="assistant", content=f"assistant {i}"))
+        await agent._sessions.save(session)
+        return session
+
+    @pytest.mark.asyncio
+    async def test_compact_emits_started_then_compressed(self, agent, session_with_history):
+        # The agent fixture's FakeLLMBackend returns "hello" — a valid summary.
+        events = [ev async for ev in agent.compact_stream(session_with_history.id)]
+
+        assert isinstance(events[0], CompressionStarted)
+        assert isinstance(events[-1], ContextCompressed)
+        # Compression actually shrank the LLM context (10 turns -> summary + kept recent).
+        assert events[-1].messages_after < events[-1].messages_before
+
+    @pytest_asyncio.fixture
+    async def session_one_long_turn(self, agent, session):
+        """The navi_code shape: a single long autonomous turn — one user message
+        followed by many tool/assistant iterations. This is ONE turn, so the
+        turn-based partition (len(turns) <= keep_recent) finds nothing to
+        summarize. Forced compact must still compress via the intra-turn split
+        (keep_recent_messages), otherwise the user always sees
+        "Nothing to compact yet — the context is still small" regardless of how
+        long the turn grew. See keep_recent_messages in compact_stream."""
+        session.context.append(Message(role="user", content="do the task"))
+        session.messages.append(Message(role="user", content="do the task"))
+        for i in range(20):
+            session.context.append(Message(role="assistant", content=f"step {i}"))
+            session.messages.append(Message(role="assistant", content=f"step {i}"))
+        await agent._sessions.save(session)
+        return session
+
+    @pytest.mark.asyncio
+    async def test_compact_compresses_single_long_turn(self, agent, session_one_long_turn):
+        """Regression: a single long turn (navi_code shape) must compress, not
+        report 'nothing to compact'. The intra-turn fallback in compact_stream
+        (keep_recent_messages=max(12, context_keep_recent*2)) is what makes this
+        work — without it forced compact always returns NothingToCompactError
+        once the conversation is a single turn."""
+        events = [ev async for ev in agent.compact_stream(session_one_long_turn.id)]
+
+        assert isinstance(events[0], CompressionStarted)
+        assert isinstance(events[-1], ContextCompressed)
+        assert events[-1].messages_after < events[-1].messages_before
+
+    @pytest.mark.asyncio
+    async def test_compact_bypasses_threshold_even_when_context_small(self, agent, session):
+        """Forced compact is NOT gated by should_compress — but still needs >= 2
+        summarizable messages. A near-empty session raises NothingToCompactError
+        (not silently does nothing), which the orchestrator surfaces as feedback."""
+        session.context.append(Message(role="user", content="hi"))
+        session.context.append(Message(role="assistant", content="hello"))
+        await agent._sessions.save(session)
+
+        with pytest.raises(NothingToCompactError):
+            async for _ in agent.compact_stream(session.id):
+                pass
+
+    @pytest.mark.asyncio
+    async def test_compact_raises_when_compression_disabled(self, agent, session_with_history, monkeypatch):
+        import navi.core.agent as agent_mod
+        import navi.config as config
+
+        # Settings is frozen — swap the whole object the agent module sees.
+        disabled = config.Settings(
+            database_url=config.settings.database_url,
+            context_compression_enabled=False,
+        )
+        monkeypatch.setattr(agent_mod, "settings", disabled)
+        with pytest.raises(NothingToCompactError):
+            async for _ in agent.compact_stream(session_with_history.id):
+                pass
+
+    @pytest.mark.asyncio
+    async def test_compact_unknown_session_raises(self, agent):
+        with pytest.raises(SessionNotFound):
+            async for _ in agent.compact_stream("no-such-session"):
+                pass
+
+
 # ─── run_ephemeral() tests ───────────────────────────────────────────────────