feat(compact): align lifecycle and carry-over behavior

frontend/terminal/src/App.tsx

@@ -437,7 +437,7 @@ function AppInner({config}: {config: FrontendConfig}): React.JSX.Element {
 					setInput={setInput}
 					onSubmit={onSubmit}
 					toolName={session.busy ? currentToolName : undefined}
-					statusLabel={session.busy ? (currentToolName ? `Running ${currentToolName}...` : 'Running agent loop...') : undefined}
+					statusLabel={session.busy ? (session.busyLabel ?? (currentToolName ? `Running ${currentToolName}...` : 'Running agent loop...')) : undefined}
 					suppressSubmit={showPicker}
 				/>
 			)}

frontend/terminal/src/hooks/useBackendSession.ts

@@ -31,6 +31,7 @@ export function useBackendSession(config: FrontendConfig, onExit: (code?: number
 	const [modal, setModal] = useState<Record<string, unknown> | null>(null);
 	const [selectRequest, setSelectRequest] = useState<{title: string; command: string; options: SelectOptionPayload[]} | null>(null);
 	const [busy, setBusy] = useState(false);
+	const [busyLabel, setBusyLabel] = useState<string | undefined>(undefined);
 	const [ready, setReady] = useState(false);
 	const [todoMarkdown, setTodoMarkdown] = useState('');
 	const [swarmTeammates, setSwarmTeammates] = useState<SwarmTeammateSnapshot[]>([]);
@@ -197,6 +198,43 @@ export function useBackendSession(config: FrontendConfig, onExit: (code?: number
 				return;
 			}
 			setTranscript((items) => [...items, {role: 'status', text: message}]);
+			if (busy) {
+				setBusyLabel(message);
+			}
+			return;
+		}
+		if (event.type === 'compact_progress') {
+			const phase = String(event.compact_phase ?? '');
+			const trigger = String(event.compact_trigger ?? '');
+			const attempt = event.attempt != null ? Number(event.attempt) : undefined;
+			if (phase === 'hooks_start') {
+				setBusyLabel(
+					trigger === 'reactive'
+						? 'Preparing retry compaction…'
+						: 'Preparing conversation compaction…',
+				);
+			} else if (phase === 'context_collapse_start') {
+				setBusyLabel('Collapsing oversized context…');
+			} else if (phase === 'context_collapse_end') {
+				setBusyLabel('Context collapse complete…');
+			} else if (phase === 'session_memory_start') {
+				setBusyLabel('Condensing earlier conversation…');
+			} else if (phase === 'compact_start') {
+				setBusyLabel(
+					trigger === 'reactive'
+						? 'Context is too large. Compacting and retrying…'
+						: 'Compacting conversation memory…',
+				);
+			} else if (phase === 'compact_retry') {
+				setBusyLabel(attempt ? `Retrying compaction (${attempt})…` : 'Retrying compaction…');
+			} else if (phase === 'compact_end') {
+				setBusyLabel('Compaction complete. Continuing…');
+			} else if (phase === 'compact_failed') {
+				setBusyLabel('Compaction failed. Continuing without it…');
+			}
+			if (event.message) {
+				setTranscript((items) => [...items, {role: 'status', text: event.message!}]);
+			}
 			return;
 		}
 		if (event.type === 'assistant_delta') {
@@ -227,6 +265,7 @@ export function useBackendSession(config: FrontendConfig, onExit: (code?: number
 			setTranscript((items) => [...items, {role: 'assistant', text}]);
 			clearAssistantDelta();
 			setBusy(false);
+			setBusyLabel(undefined);
 			return;
 		}
 		if (event.type === 'line_complete') {
@@ -234,9 +273,13 @@ export function useBackendSession(config: FrontendConfig, onExit: (code?: number
 			// don't leave stale streaming text on screen.
 			clearAssistantDelta();
 			setBusy(false);
+			setBusyLabel(undefined);
 			return;
 		}
 		if ((event.type === 'tool_started' || event.type === 'tool_completed') && event.item) {
+			if (event.type === 'tool_started') {
+				setBusyLabel(event.tool_name ? `Running ${event.tool_name}...` : 'Running...');
+			}
 			const enrichedItem: TranscriptItem = {
 				...event.item,
 				tool_name: event.item.tool_name ?? event.tool_name ?? undefined,
@@ -249,6 +292,7 @@ export function useBackendSession(config: FrontendConfig, onExit: (code?: number
 		if (event.type === 'clear_transcript') {
 			setTranscript([]);
 			clearAssistantDelta();
+			setBusyLabel(undefined);
 			return;
 		}
 		if (event.type === 'select_request') {
@@ -268,6 +312,7 @@ export function useBackendSession(config: FrontendConfig, onExit: (code?: number
 			setTranscript((items) => [...items, {role: 'system', text: `error: ${event.message ?? 'unknown error'}`}]);
 			clearAssistantDelta();
 			setBusy(false);
+			setBusyLabel(undefined);
 			return;
 		}
 		if (event.type === 'todo_update') {
@@ -308,6 +353,7 @@ export function useBackendSession(config: FrontendConfig, onExit: (code?: number
 			modal,
 			selectRequest,
 			busy,
+			busyLabel,
 			ready,
 			todoMarkdown,
 			swarmTeammates,
@@ -317,6 +363,6 @@ export function useBackendSession(config: FrontendConfig, onExit: (code?: number
 			setBusy,
 			sendRequest,
 		}),
-		[assistantBuffer, bridgeSessions, busy, commands, mcpServers, modal, ready, selectRequest, status, swarmNotifications, swarmTeammates, tasks, todoMarkdown, transcript]
+		[assistantBuffer, bridgeSessions, busy, busyLabel, commands, mcpServers, modal, ready, selectRequest, status, swarmNotifications, swarmTeammates, tasks, todoMarkdown, transcript]
 	);
 }

frontend/terminal/src/types.ts

@@ -78,6 +78,11 @@ export type BackendEvent = {
 	tool_name?: string | null;
 	output?: string | null;
 	is_error?: boolean | null;
+	compact_phase?: string | null;
+	compact_trigger?: string | null;
+	attempt?: number | null;
+	compact_checkpoint?: string | null;
+	compact_metadata?: Record<string, unknown> | null;
 	// New event payloads
 	todo_items?: TodoItemSnapshot[] | null;
 	todo_markdown?: string | null;

ohmo/gateway/runtime.py

@@ -18,6 +18,7 @@ from openharness.engine.query import MaxTurnsExceeded
 from openharness.engine.stream_events import (
     AssistantTextDelta,
     AssistantTurnComplete,
+    CompactProgressEvent,
     ErrorEvent,
     StatusEvent,
     ToolExecutionCompleted,
@@ -326,6 +327,32 @@ class OhmoSessionRuntimePool:
         if isinstance(event, AssistantTextDelta):
             reply_parts.append(event.text)
             return
+        if isinstance(event, CompactProgressEvent):
+            logger.info(
+                "ohmo runtime compact progress session_key=%s session_id=%s phase=%s trigger=%s attempt=%s",
+                session_key,
+                bundle.session_id,
+                event.phase,
+                event.trigger,
+                event.attempt,
+            )
+            rendered = _format_channel_progress(
+                channel=message.channel,
+                kind="compact_progress",
+                text=event.message or "",
+                session_key=session_key,
+                content=content,
+                compact_phase=event.phase,
+                compact_trigger=event.trigger,
+                attempt=event.attempt,
+            )
+            if rendered:
+                yield GatewayStreamUpdate(
+                    kind="progress",
+                    text=rendered,
+                    metadata={"_progress": True, "_session_key": session_key, "_compact": True},
+                )
+            return
         if isinstance(event, StatusEvent):
             logger.info(
                 "ohmo runtime status session_key=%s session_id=%s message=%r",
@@ -490,6 +517,9 @@ def _format_channel_progress(
     text: str,
     session_key: str,
     content: str,
+    compact_phase: str | None = None,
+    compact_trigger: str | None = None,
+    attempt: int | None = None,
 ) -> str:
     if channel not in {
         "feishu",
@@ -525,13 +555,55 @@ def _format_channel_progress(
         if text.startswith(("🤔", "🧠", "✨", "🔎", "🪄", "🛠️", "🫧")):
             return text
         return f"🫧 {text}"
+    if kind == "compact_progress":
+        if compact_phase == "hooks_start":
+            if prefers_chinese:
+                if compact_trigger == "reactive":
+                    return "🫧 上下文有点超长，我先准备压缩一下记忆，然后立刻继续重试～"
+                return "🫧 我先把上下文和记忆准备一下，马上开始压缩重点～"
+            if compact_trigger == "reactive":
+                return "🫧 The context got too large. I’m preparing a quick memory compaction before retrying."
+            return "🫧 Let me get the context ready before I compact the conversation."
+        if compact_phase == "context_collapse_start":
+            if prefers_chinese:
+                return "🫧 我先把太长的上下文折叠一下，让后面的压缩更快一点～"
+            return "🫧 I’m collapsing the oversized context first so compaction can move faster."
+        if compact_phase == "context_collapse_end":
+            if prefers_chinese:
+                return "🫧 上下文已经先收紧了一层，继续压缩重点～"
+            return "🫧 The context is trimmed down now. Continuing with the main compaction."
+        if compact_phase in {"session_memory_start", "compact_start"}:
+            if prefers_chinese:
+                if compact_phase == "session_memory_start":
+                    return "🧠 我先把前面的聊天重点悄悄捋顺一下，马上继续～"
+                if compact_trigger == "reactive":
+                    return "🧠 这轮上下文太长了，我先压缩一下记忆，然后马上继续重试～"
+                return "🧠 聊天有点长啦，我先帮你悄悄压缩一下记忆，马上继续～"
+            if compact_phase == "session_memory_start":
+                return "🧠 Let me quickly condense the earlier parts of this chat, then I’ll keep going."
+            if compact_trigger == "reactive":
+                return "🧠 The context is too large for this turn. I’ll compact the memory and retry."
+            return "🧠 This chat is getting long. I’ll compact the memory and keep going."
+        if compact_phase == "compact_retry":
+            suffix = f" (attempt {attempt})" if attempt is not None else ""
+            if prefers_chinese:
+                return f"🔁 压缩记忆这一步有点卡，我换个方式再试一次{suffix}。"
+            return f"🔁 Compaction got stuck, trying a lighter retry{suffix}."
+        if compact_phase == "compact_failed":
+            if prefers_chinese:
+                return "⚠️ 这次记忆压缩没成功，我先跳过它继续处理你的消息。"
+            return "⚠️ Memory compaction did not complete. I’m skipping it and continuing."
+        return ""
     return text
 
 
 def _build_inbound_user_message(message: InboundMessage) -> ConversationMessage:
     """Convert an inbound channel message into user content blocks."""
     content: list[TextBlock | ImageBlock] = []
+    speaker_context = _build_speaker_context(message)
     base = (message.content or "").strip()
+    if speaker_context:
+        content.append(TextBlock(text=speaker_context))
     if base:
         content.append(TextBlock(text=base))
 
@@ -551,6 +623,26 @@ def _build_inbound_user_message(message: InboundMessage) -> ConversationMessage:
     return ConversationMessage.from_user_content(content)
 
 
+def _build_speaker_context(message: InboundMessage) -> str:
+    """Return a lightweight speaker header for group-chat messages."""
+    metadata = message.metadata or {}
+    chat_type = str(metadata.get("chat_type") or "").strip().lower()
+    sender_label = (
+        str(metadata.get("sender_display_name") or "").strip()
+        or str(metadata.get("sender_label") or "").strip()
+        or str(message.sender_id).strip()
+    )
+    if chat_type != "group":
+        return ""
+    if not sender_label:
+        sender_label = "unknown"
+    return (
+        "[Channel speaker]\n"
+        f"This message was sent in a group chat by: {sender_label}\n"
+        f"Sender id: {message.sender_id}"
+    )
+
+
 def _build_attachment_notes(media_paths: list[str]) -> str:
     """Build textual attachment notes for non-image context and persistence."""
     if not media_paths:

ohmo/runtime.py

@@ -9,7 +9,7 @@ import sys
 from pathlib import Path
 
 from openharness.api.client import SupportsStreamingMessages
-from openharness.engine.stream_events import AssistantTextDelta, AssistantTurnComplete, ErrorEvent, StatusEvent
+from openharness.engine.stream_events import AssistantTextDelta, AssistantTurnComplete, CompactProgressEvent, ErrorEvent, StatusEvent
 from openharness.ui.backend_host import run_backend_host
 from openharness.ui.runtime import build_runtime, close_runtime, handle_line, start_runtime
 from openharness.ui.react_launcher import _resolve_npm, _resolve_tsx, get_frontend_dir
@@ -173,6 +173,9 @@ async def run_ohmo_print_mode(
                 sys.stdout.flush()
             elif isinstance(event, ErrorEvent):
                 print(event.message, file=sys.stderr)
+            elif isinstance(event, CompactProgressEvent):
+                if event.message:
+                    print(event.message, file=sys.stderr)
             elif isinstance(event, StatusEvent):
                 print(event.message, file=sys.stderr)
 

src/openharness/channels/impl/feishu.py

@@ -6,6 +6,7 @@ import os
 import re
 import threading
 from collections import OrderedDict
+from dataclasses import dataclass
 from typing import Any
 
 
@@ -31,6 +32,12 @@ MSG_TYPE_MAP = {
 }
 
 
+@dataclass(frozen=True)
+class _FeishuSenderInfo:
+    open_id: str
+    display_name: str
+
+
 def _extract_share_card_content(content_json: dict, msg_type: str) -> str:
     """Extract text representation from share cards and interactive messages."""
     parts = []
@@ -253,6 +260,7 @@ class FeishuChannel(BaseChannel):
         self._ws_client: Any = None
         self._ws_thread: threading.Thread | None = None
         self._processed_message_ids: OrderedDict[str, None] = OrderedDict()  # Ordered dedup cache
+        self._sender_cache: OrderedDict[str, _FeishuSenderInfo] = OrderedDict()
         self._loop: asyncio.AbstractEventLoop | None = None
 
     async def start(self) -> None:
@@ -842,6 +850,44 @@ class FeishuChannel(BaseChannel):
         except Exception as e:
             logger.error("Error sending Feishu message: {}", e)
 
+    def _resolve_sender_display_name_sync(self, open_id: str) -> str:
+        """Resolve a human-friendly sender name from Feishu contact APIs."""
+        cached = self._sender_cache.get(open_id)
+        if cached is not None:
+            self._sender_cache.move_to_end(open_id)
+            return cached.display_name
+        if not self._client or not open_id:
+            return open_id or "unknown"
+
+        try:
+            import lark_oapi as lark
+
+            request = (
+                lark.api.contact.v3.GetUserRequest.builder()
+                .user_id(open_id)
+                .user_id_type("open_id")
+                .build()
+            )
+            response = self._client.contact.v3.user.get(request)
+            if getattr(response, "success", lambda: False)():
+                user = getattr(getattr(response, "data", None), "user", None)
+                display_name = (
+                    getattr(user, "name", None)
+                    or getattr(user, "en_name", None)
+                    or getattr(user, "nickname", None)
+                    or open_id
+                )
+            else:
+                display_name = open_id
+        except Exception:
+            logger.exception("Failed to resolve Feishu sender name open_id=%s", open_id)
+            display_name = open_id
+
+        self._sender_cache[open_id] = _FeishuSenderInfo(open_id=open_id, display_name=display_name)
+        while len(self._sender_cache) > 512:
+            self._sender_cache.popitem(last=False)
+        return display_name
+
     def _on_message_sync(self, data: "P2ImMessageReceiveV1") -> None:  # noqa: F821
         """
         Sync handler for incoming messages (called from WebSocket thread).
@@ -872,6 +918,12 @@ class FeishuChannel(BaseChannel):
                 return
 
             sender_id = sender.sender_id.open_id if sender.sender_id else "unknown"
+            loop = asyncio.get_running_loop()
+            sender_display_name = await loop.run_in_executor(
+                None,
+                self._resolve_sender_display_name_sync,
+                sender_id,
+            )
             chat_id = message.chat_id
             chat_type = message.chat_type
             msg_type = message.message_type
@@ -937,6 +989,8 @@ class FeishuChannel(BaseChannel):
                     "message_id": message_id,
                     "chat_type": chat_type,
                     "msg_type": msg_type,
+                    "sender_display_name": sender_display_name,
+                    "sender_label": sender_display_name or sender_id,
                 }
             )
 

src/openharness/commands/registry.py

@@ -41,7 +41,12 @@ from openharness.permissions import PermissionChecker, PermissionMode
 from openharness.plugins import load_plugins
 from openharness.prompts import build_runtime_system_prompt
 from openharness.plugins.installer import install_plugin_from_path, uninstall_plugin
-from openharness.services import compact_messages, estimate_conversation_tokens, summarize_messages
+from openharness.services import (
+    compact_conversation,
+    compact_messages,
+    estimate_conversation_tokens,
+    summarize_messages,
+)
 from openharness.services.session_backend import DEFAULT_SESSION_BACKEND, SessionBackend
 from openharness.skills import load_skill_registry
 from openharness.tasks import get_task_manager
@@ -280,7 +285,17 @@ def create_default_command_registry(
             except ValueError:
                 return CommandResult(message="Usage: /compact [PRESERVE_RECENT]")
         before = len(context.engine.messages)
-        compacted = compact_messages(context.engine.messages, preserve_recent=preserve_recent)
+        try:
+            compacted = await compact_conversation(
+                context.engine.messages,
+                api_client=context.engine.api_client,
+                model=context.engine.model,
+                system_prompt=context.engine.system_prompt,
+                preserve_recent=preserve_recent,
+                trigger="manual",
+            )
+        except Exception:
+            compacted = compact_messages(context.engine.messages, preserve_recent=preserve_recent)
         context.engine.load_messages(compacted)
         return CommandResult(
             message=f"Compacted conversation from {before} messages to {len(compacted)}."

src/openharness/engine/query.py

@@ -7,7 +7,7 @@ import logging
 import time
 from dataclasses import dataclass
 from pathlib import Path
-from typing import AsyncIterator, Awaitable, Callable
+from typing import Any, AsyncIterator, Awaitable, Callable
 
 from openharness.api.client import (
     ApiMessageCompleteEvent,
@@ -21,6 +21,7 @@ from openharness.engine.messages import ConversationMessage, ToolResultBlock
 from openharness.engine.stream_events import (
     AssistantTextDelta,
     AssistantTurnComplete,
+    CompactProgressEvent,
     ErrorEvent,
     StatusEvent,
     StreamEvent,
@@ -33,6 +34,7 @@ from openharness.tools.base import ToolExecutionContext
 from openharness.tools.base import ToolRegistry
 
 AUTO_COMPACT_STATUS_MESSAGE = "Auto-compacting conversation memory to keep things fast and focused."
+REACTIVE_COMPACT_STATUS_MESSAGE = "Prompt too long; compacting conversation memory and retrying."
 
 log = logging.getLogger(__name__)
 
@@ -40,6 +42,26 @@ log = logging.getLogger(__name__)
 PermissionPrompt = Callable[[str, str], Awaitable[bool]]
 AskUserPrompt = Callable[[str], Awaitable[str]]
 
+MAX_TRACKED_READ_FILES = 6
+MAX_TRACKED_SKILLS = 8
+MAX_TRACKED_ASYNC_AGENT_EVENTS = 8
+
+
+def _is_prompt_too_long_error(exc: Exception) -> bool:
+    text = str(exc).lower()
+    return any(
+        needle in text
+        for needle in (
+            "prompt too long",
+            "context length",
+            "maximum context",
+            "context window",
+            "too many tokens",
+            "too large for the model",
+            "maximum context length",
+        )
+    )
+
 
 class MaxTurnsExceeded(RuntimeError):
     """Raised when the agent exceeds the configured max_turns for one user prompt."""
@@ -67,6 +89,125 @@ class QueryContext:
     tool_metadata: dict[str, object] | None = None
 
 
+def _tool_metadata_bucket(
+    tool_metadata: dict[str, object] | None,
+    key: str,
+) -> list[Any]:
+    if tool_metadata is None:
+        return []
+    value = tool_metadata.setdefault(key, [])
+    if isinstance(value, list):
+        return value
+    replacement: list[Any] = []
+    tool_metadata[key] = replacement
+    return replacement
+
+
+def _remember_read_file(
+    tool_metadata: dict[str, object] | None,
+    *,
+    path: str,
+    offset: int,
+    limit: int,
+    output: str,
+) -> None:
+    bucket = _tool_metadata_bucket(tool_metadata, "read_file_state")
+    preview_lines = [line.strip() for line in output.splitlines()[:6] if line.strip()]
+    bucket.append(
+        {
+            "path": path,
+            "span": f"lines {offset + 1}-{offset + limit}",
+            "preview": " | ".join(preview_lines)[:320],
+        }
+    )
+    if len(bucket) > MAX_TRACKED_READ_FILES:
+        del bucket[:-MAX_TRACKED_READ_FILES]
+
+
+def _remember_skill_invocation(
+    tool_metadata: dict[str, object] | None,
+    *,
+    skill_name: str,
+) -> None:
+    bucket = _tool_metadata_bucket(tool_metadata, "invoked_skills")
+    normalized = skill_name.strip()
+    if not normalized:
+        return
+    if normalized in bucket:
+        bucket.remove(normalized)
+    bucket.append(normalized)
+    if len(bucket) > MAX_TRACKED_SKILLS:
+        del bucket[:-MAX_TRACKED_SKILLS]
+
+
+def _remember_async_agent_activity(
+    tool_metadata: dict[str, object] | None,
+    *,
+    tool_name: str,
+    tool_input: dict[str, object],
+    output: str,
+) -> None:
+    bucket = _tool_metadata_bucket(tool_metadata, "async_agent_state")
+    if tool_name == "agent":
+        description = str(tool_input.get("description") or tool_input.get("prompt") or "").strip()
+        summary = f"Spawned async agent. {description}".strip()
+        if output.strip():
+            summary = f"{summary} [{output.strip()[:180]}]".strip()
+    elif tool_name == "send_message":
+        target = str(tool_input.get("task_id") or "").strip()
+        summary = f"Sent follow-up message to async agent {target}".strip()
+    else:
+        summary = output.strip()[:220] or f"Async agent activity via {tool_name}"
+    bucket.append(summary)
+    if len(bucket) > MAX_TRACKED_ASYNC_AGENT_EVENTS:
+        del bucket[:-MAX_TRACKED_ASYNC_AGENT_EVENTS]
+
+
+def _update_plan_mode(tool_metadata: dict[str, object] | None, mode: str) -> None:
+    if tool_metadata is None:
+        return
+    tool_metadata["permission_mode"] = mode
+
+
+def _record_tool_carryover(
+    context: QueryContext,
+    *,
+    tool_name: str,
+    tool_input: dict[str, object],
+    tool_output: str,
+    is_error: bool,
+    resolved_file_path: str | None,
+) -> None:
+    if is_error:
+        return
+    if tool_name == "read_file" and resolved_file_path is not None:
+        offset = int(tool_input.get("offset") or 0)
+        limit = int(tool_input.get("limit") or 200)
+        _remember_read_file(
+            context.tool_metadata,
+            path=resolved_file_path,
+            offset=offset,
+            limit=limit,
+            output=tool_output,
+        )
+    elif tool_name == "skill":
+        _remember_skill_invocation(
+            context.tool_metadata,
+            skill_name=str(tool_input.get("name") or ""),
+        )
+    elif tool_name in {"agent", "send_message"}:
+        _remember_async_agent_activity(
+            context.tool_metadata,
+            tool_name=tool_name,
+            tool_input=tool_input,
+            output=tool_output,
+        )
+    elif tool_name == "enter_plan_mode":
+        _update_plan_mode(context.tool_metadata, "plan")
+    elif tool_name == "exit_plan_mode":
+        _update_plan_mode(context.tool_metadata, "default")
+
+
 async def run_query(
     context: QueryContext,
     messages: list[ConversationMessage],
@@ -85,20 +226,54 @@ async def run_query(
     )
 
     compact_state = AutoCompactState()
+    reactive_compact_attempted = False
+    last_compaction_result: tuple[list[ConversationMessage], bool] = (messages, False)
+
+    async def _stream_compaction(
+        *,
+        trigger: str,
+        force: bool = False,
+    ) -> AsyncIterator[tuple[StreamEvent, UsageSnapshot | None]]:
+        nonlocal last_compaction_result
+        progress_queue: asyncio.Queue[CompactProgressEvent] = asyncio.Queue()
+
+        async def _progress(event: CompactProgressEvent) -> None:
+            await progress_queue.put(event)
+
+        task = asyncio.create_task(
+            auto_compact_if_needed(
+                messages,
+                api_client=context.api_client,
+                model=context.model,
+                system_prompt=context.system_prompt,
+                state=compact_state,
+                progress_callback=_progress,
+                force=force,
+                trigger=trigger,
+                hook_executor=context.hook_executor,
+                carryover_metadata=context.tool_metadata,
+            )
+        )
+        while True:
+            try:
+                event = await asyncio.wait_for(progress_queue.get(), timeout=0.05)
+                yield event, None
+            except asyncio.TimeoutError:
+                if task.done():
+                    break
+                continue
+        while not progress_queue.empty():
+            yield progress_queue.get_nowait(), None
+        last_compaction_result = await task
+        return
 
     turn_count = 0
     while context.max_turns is None or turn_count < context.max_turns:
         turn_count += 1
         # --- auto-compact check before calling the model ---------------
-        messages, was_compacted = await auto_compact_if_needed(
-            messages,
-            api_client=context.api_client,
-            model=context.model,
-            system_prompt=context.system_prompt,
-            state=compact_state,
-        )
-        if was_compacted:
-            yield StatusEvent(message=AUTO_COMPACT_STATUS_MESSAGE), None
+        async for event, usage in _stream_compaction(trigger="auto"):
+            yield event, usage
+        messages, was_compacted = last_compaction_result
         # ---------------------------------------------------------------
 
         final_message: ConversationMessage | None = None
@@ -131,6 +306,14 @@ async def run_query(
                     usage = event.usage
         except Exception as exc:
             error_msg = str(exc)
+            if not reactive_compact_attempted and _is_prompt_too_long_error(exc):
+                reactive_compact_attempted = True
+                yield StatusEvent(message=REACTIVE_COMPACT_STATUS_MESSAGE), None
+                async for event, usage in _stream_compaction(trigger="reactive", force=True):
+                    yield event, usage
+                messages, was_compacted = last_compaction_result
+                if was_compacted:
+                    continue
             if "connect" in error_msg.lower() or "timeout" in error_msg.lower() or "network" in error_msg.lower():
                 yield ErrorEvent(message=f"Network error: {error_msg}. Check your internet connection and try again."), None
             else:
@@ -275,6 +458,14 @@ async def _execute_tool_call(
         content=result.output,
         is_error=result.is_error,
     )
+    _record_tool_carryover(
+        context,
+        tool_name=tool_name,
+        tool_input=tool_input,
+        tool_output=tool_result.content,
+        is_error=tool_result.is_error,
+        resolved_file_path=_file_path,
+    )
     if context.hook_executor is not None:
         await context.hook_executor.execute(
             HookEvent.POST_TOOL_USE,

src/openharness/engine/query_engine.py

@@ -59,6 +59,21 @@ class QueryEngine:
         """Return the maximum number of agentic turns per user input, if capped."""
         return self._max_turns
 
+    @property
+    def api_client(self) -> SupportsStreamingMessages:
+        """Return the active API client."""
+        return self._api_client
+
+    @property
+    def model(self) -> str:
+        """Return the active model identifier."""
+        return self._model
+
+    @property
+    def system_prompt(self) -> str:
+        """Return the active system prompt."""
+        return self._system_prompt
+
     @property
     def total_usage(self):
         """Return the total usage across all turns."""

src/openharness/engine/stream_events.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, Literal
 
 from openharness.api.usage import UsageSnapshot
 from openharness.engine.messages import ConversationMessage
@@ -56,6 +56,28 @@ class StatusEvent:
     message: str
 
 
+@dataclass(frozen=True)
+class CompactProgressEvent:
+    """Structured progress event for conversation compaction."""
+
+    phase: Literal[
+        "hooks_start",
+        "context_collapse_start",
+        "context_collapse_end",
+        "session_memory_start",
+        "session_memory_end",
+        "compact_start",
+        "compact_retry",
+        "compact_end",
+        "compact_failed",
+    ]
+    trigger: Literal["auto", "manual", "reactive"]
+    message: str | None = None
+    attempt: int | None = None
+    checkpoint: str | None = None
+    metadata: dict[str, Any] | None = None
+
+
 StreamEvent = (
     AssistantTextDelta
     | AssistantTurnComplete
@@ -63,4 +85,5 @@ StreamEvent = (
     | ToolExecutionCompleted
     | ErrorEvent
     | StatusEvent
+    | CompactProgressEvent
 )

src/openharness/hooks/events.py

@@ -10,5 +10,7 @@ class HookEvent(str, Enum):
 
     SESSION_START = "session_start"
     SESSION_END = "session_end"
+    PRE_COMPACT = "pre_compact"
+    POST_COMPACT = "post_compact"
     PRE_TOOL_USE = "pre_tool_use"
     POST_TOOL_USE = "post_tool_use"

src/openharness/services/__init__.py

@@ -1,6 +1,7 @@
 """Service exports."""
 
 from openharness.services.compact import (
+    compact_conversation,
     compact_messages,
     estimate_conversation_tokens,
     summarize_messages,
@@ -15,6 +16,7 @@ from openharness.services.token_estimation import estimate_message_tokens, estim
 
 __all__ = [
     "compact_messages",
+    "compact_conversation",
     "estimate_conversation_tokens",
     "estimate_message_tokens",
     "estimate_tokens",

src/openharness/services/compact/__init__.py

@@ -8,18 +8,25 @@ Faithfully translated from Claude Code's compaction system:
 
 from __future__ import annotations
 
+import asyncio
+import inspect
 import logging
 import re
 from dataclasses import dataclass
-from typing import Any
+from pathlib import Path
+from typing import Any, Awaitable, Callable, Literal
+from uuid import uuid4
 
 from openharness.engine.messages import (
     ConversationMessage,
     ContentBlock,
+    ImageBlock,
     TextBlock,
     ToolResultBlock,
     ToolUseBlock,
 )
+from openharness.engine.stream_events import CompactProgressEvent
+from openharness.hooks import HookEvent, HookExecutor
 from openharness.services.token_estimation import estimate_tokens
 
 log = logging.getLogger(__name__)
@@ -45,6 +52,17 @@ TIME_BASED_MC_CLEARED_MESSAGE = "[Old tool result content cleared]"
 AUTOCOMPACT_BUFFER_TOKENS = 13_000
 MAX_OUTPUT_TOKENS_FOR_SUMMARY = 20_000
 MAX_CONSECUTIVE_AUTOCOMPACT_FAILURES = 3
+COMPACT_TIMEOUT_SECONDS = 25
+MAX_COMPACT_STREAMING_RETRIES = 2
+MAX_PTL_RETRIES = 3
+SESSION_MEMORY_KEEP_RECENT = 12
+SESSION_MEMORY_MAX_LINES = 48
+SESSION_MEMORY_MAX_CHARS = 4_000
+CONTEXT_COLLAPSE_TEXT_CHAR_LIMIT = 2_400
+CONTEXT_COLLAPSE_HEAD_CHARS = 900
+CONTEXT_COLLAPSE_TAIL_CHARS = 500
+MAX_COMPACT_ATTACHMENTS = 6
+MAX_DISCOVERED_TOOLS = 12
 
 # Microcompact defaults
 DEFAULT_KEEP_RECENT = 5
@@ -55,6 +73,11 @@ TOKEN_ESTIMATION_PADDING = 4 / 3
 
 # Default context windows per model family
 _DEFAULT_CONTEXT_WINDOW = 200_000
+PTL_RETRY_MARKER = "[earlier conversation truncated for compaction retry]"
+ERROR_MESSAGE_INCOMPLETE_RESPONSE = "Compaction interrupted before a complete summary was returned."
+
+CompactTrigger = Literal["auto", "manual", "reactive"]
+CompactProgressCallback = Callable[[CompactProgressEvent], Awaitable[None]]
 
 
 # ---------------------------------------------------------------------------
@@ -81,6 +104,290 @@ def estimate_conversation_tokens(messages: list[ConversationMessage]) -> int:
     return estimate_message_tokens(messages)
 
 
+def _sanitize_metadata(value: Any) -> Any:
+    if isinstance(value, (str, int, float, bool)) or value is None:
+        return value
+    if isinstance(value, Path):
+        return str(value)
+    if isinstance(value, dict):
+        return {str(key): _sanitize_metadata(item) for key, item in value.items()}
+    if isinstance(value, (list, tuple, set)):
+        return [_sanitize_metadata(item) for item in value]
+    return str(value)
+
+
+def _record_compact_checkpoint(
+    carryover_metadata: dict[str, Any] | None,
+    *,
+    checkpoint: str,
+    trigger: CompactTrigger,
+    message_count: int,
+    token_count: int,
+    attempt: int | None = None,
+    details: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    payload: dict[str, Any] = {
+        "checkpoint": checkpoint,
+        "trigger": trigger,
+        "message_count": message_count,
+        "token_count": token_count,
+    }
+    if attempt is not None:
+        payload["attempt"] = attempt
+    if details:
+        payload.update(_sanitize_metadata(details))
+    if carryover_metadata is not None:
+        checkpoints = carryover_metadata.setdefault("compact_checkpoints", [])
+        if isinstance(checkpoints, list):
+            checkpoints.append(payload)
+        carryover_metadata["compact_last"] = payload
+    return payload
+
+
+async def _emit_progress(
+    callback: CompactProgressCallback | None,
+    *,
+    phase: Literal[
+        "hooks_start",
+        "context_collapse_start",
+        "context_collapse_end",
+        "session_memory_start",
+        "session_memory_end",
+        "compact_start",
+        "compact_retry",
+        "compact_end",
+        "compact_failed",
+    ],
+    trigger: CompactTrigger,
+    message: str | None = None,
+    attempt: int | None = None,
+    checkpoint: str | None = None,
+    metadata: dict[str, Any] | None = None,
+) -> None:
+    if callback is None:
+        return
+    await callback(
+        CompactProgressEvent(
+            phase=phase,
+            trigger=trigger,
+            message=message,
+            attempt=attempt,
+            checkpoint=checkpoint,
+            metadata=_sanitize_metadata(metadata) if metadata else None,
+        )
+    )
+
+
+def _is_prompt_too_long_error(exc: Exception) -> bool:
+    text = str(exc).lower()
+    return any(
+        needle in text
+        for needle in (
+            "prompt too long",
+            "context length",
+            "maximum context",
+            "context window",
+            "too many tokens",
+            "too large for the model",
+        )
+    )
+
+
+def _group_messages_by_prompt_round(
+    messages: list[ConversationMessage],
+) -> list[list[ConversationMessage]]:
+    groups: list[list[ConversationMessage]] = []
+    current: list[ConversationMessage] = []
+    for message in messages:
+        starts_new_round = (
+            message.role == "user"
+            and not any(isinstance(block, ToolResultBlock) for block in message.content)
+            and bool(message.text.strip())
+        )
+        if starts_new_round and current:
+            groups.append(current)
+            current = []
+        current.append(message)
+    if current:
+        groups.append(current)
+    return groups
+
+
+def _collapse_text(text: str) -> str:
+    if len(text) <= CONTEXT_COLLAPSE_TEXT_CHAR_LIMIT:
+        return text
+    omitted = len(text) - CONTEXT_COLLAPSE_HEAD_CHARS - CONTEXT_COLLAPSE_TAIL_CHARS
+    head = text[:CONTEXT_COLLAPSE_HEAD_CHARS].rstrip()
+    tail = text[-CONTEXT_COLLAPSE_TAIL_CHARS:].lstrip()
+    return f"{head}\n...[collapsed {omitted} chars]...\n{tail}"
+
+
+def try_context_collapse(
+    messages: list[ConversationMessage],
+    *,
+    preserve_recent: int,
+) -> list[ConversationMessage] | None:
+    """Deterministically shrink oversized text blocks before full compact."""
+    if len(messages) <= preserve_recent + 2:
+        return None
+
+    older = messages[:-preserve_recent]
+    newer = messages[-preserve_recent:]
+    changed = False
+    collapsed_older: list[ConversationMessage] = []
+    for message in older:
+        new_blocks: list[ContentBlock] = []
+        for block in message.content:
+            if isinstance(block, TextBlock):
+                collapsed = _collapse_text(block.text)
+                if collapsed != block.text:
+                    changed = True
+                new_blocks.append(TextBlock(text=collapsed))
+            else:
+                new_blocks.append(block)
+        collapsed_older.append(ConversationMessage(role=message.role, content=new_blocks))
+
+    if not changed:
+        return None
+
+    result = [*collapsed_older, *newer]
+    if estimate_message_tokens(result) >= estimate_message_tokens(messages):
+        return None
+    return result
+
+
+def truncate_head_for_ptl_retry(
+    messages: list[ConversationMessage],
+) -> list[ConversationMessage] | None:
+    """Drop the oldest prompt rounds when the compact request itself is too large."""
+    groups = _group_messages_by_prompt_round(messages)
+    if len(groups) < 2:
+        return None
+
+    drop_count = max(1, len(groups) // 5)
+    drop_count = min(drop_count, len(groups) - 1)
+    retained = [message for group in groups[drop_count:] for message in group]
+    if not retained:
+        return None
+    if retained[0].role == "assistant":
+        return [ConversationMessage.from_user_text(PTL_RETRY_MARKER), *retained]
+    return retained
+
+
+def _extract_attachment_paths(messages: list[ConversationMessage]) -> list[str]:
+    found: list[str] = []
+    seen: set[str] = set()
+    path_pattern = re.compile(r"path:\s*([^)\\n]+)")
+    attachment_pattern = re.compile(r"\[attachment:\s*([^\]]+)\]")
+    for message in messages:
+        for block in message.content:
+            if isinstance(block, ImageBlock) and block.source_path:
+                path = str(Path(block.source_path).expanduser())
+                if path not in seen:
+                    seen.add(path)
+                    found.append(path)
+            elif isinstance(block, TextBlock):
+                for match in path_pattern.findall(block.text):
+                    path = match.strip()
+                    if path and path not in seen:
+                        seen.add(path)
+                        found.append(path)
+                for match in attachment_pattern.findall(block.text):
+                    path = match.strip()
+                    if path and "download failed" not in path and path not in seen:
+                        seen.add(path)
+                        found.append(path)
+            if len(found) >= MAX_COMPACT_ATTACHMENTS:
+                return found
+    return found
+
+
+def _extract_discovered_tools(messages: list[ConversationMessage]) -> list[str]:
+    discovered: list[str] = []
+    seen: set[str] = set()
+    for message in messages:
+        for tool_use in message.tool_uses:
+            if tool_use.name and tool_use.name not in seen:
+                seen.add(tool_use.name)
+                discovered.append(tool_use.name)
+            if len(discovered) >= MAX_DISCOVERED_TOOLS:
+                return discovered
+    return discovered
+
+
+def build_compact_carryover_message(
+    messages: list[ConversationMessage],
+    *,
+    metadata: dict[str, Any] | None = None,
+    hook_note: str | None = None,
+) -> ConversationMessage | None:
+    """Preserve lightweight runtime context that should survive compaction."""
+    metadata = metadata or {}
+    attachment_paths = _extract_attachment_paths(messages)
+    discovered_tools = _extract_discovered_tools(messages)
+    permission_mode = str(metadata.get("permission_mode") or "").strip().lower()
+    read_file_state = metadata.get("read_file_state")
+    invoked_skills = metadata.get("invoked_skills")
+    async_agent_state = metadata.get("async_agent_state")
+    compact_last = metadata.get("compact_last")
+
+    lines: list[str] = []
+    if permission_mode == "plan":
+        lines.extend(
+            [
+                "Plan mode is still active for this session.",
+                "Do not execute mutating tools until the user exits plan mode.",
+            ]
+        )
+    if attachment_paths:
+        lines.append("Recent local attachments to keep in mind:")
+        lines.extend(f"- {path}" for path in attachment_paths)
+    if discovered_tools:
+        lines.append("Tools already discovered or used in this session:")
+        lines.append("- " + ", ".join(discovered_tools))
+    if isinstance(read_file_state, list) and read_file_state:
+        lines.append("Recently read files to keep in working memory:")
+        for entry in read_file_state[-4:]:
+            if not isinstance(entry, dict):
+                continue
+            path = str(entry.get("path") or "").strip()
+            span = str(entry.get("span") or "").strip()
+            preview = str(entry.get("preview") or "").strip()
+            if not path:
+                continue
+            bullet = f"- {path}"
+            if span:
+                bullet += f" ({span})"
+            lines.append(bullet)
+            if preview:
+                lines.append(f"  Preview: {preview}")
+    if isinstance(invoked_skills, list) and invoked_skills:
+        lines.append("Skills invoked earlier in the session:")
+        lines.append("- " + ", ".join(str(skill) for skill in invoked_skills[-8:]))
+    if isinstance(async_agent_state, list) and async_agent_state:
+        lines.append("Async agent / background task state:")
+        lines.extend(f"- {entry}" for entry in async_agent_state[-6:])
+    if isinstance(compact_last, dict) and compact_last:
+        checkpoint = str(compact_last.get("checkpoint") or "").strip()
+        token_count = compact_last.get("token_count")
+        if checkpoint:
+            if token_count is not None:
+                lines.append(
+                    f"Last compact checkpoint: {checkpoint} (token_count={token_count})"
+                )
+            else:
+                lines.append(f"Last compact checkpoint: {checkpoint}")
+    if hook_note:
+        lines.append("Compact hook note:")
+        lines.append(hook_note)
+
+    if not lines:
+        return None
+    return ConversationMessage.from_user_text(
+        "Carry-over context preserved after compaction:\n" + "\n".join(lines)
+    )
+
+
 # ---------------------------------------------------------------------------
 # Microcompact — clear old tool results to reduce tokens cheaply
 # ---------------------------------------------------------------------------
@@ -148,6 +455,62 @@ def microcompact_messages(
     return messages, tokens_saved
 
 
+def _summarize_message_for_memory(message: ConversationMessage) -> str:
+    text = " ".join(message.text.split())
+    if text:
+        text = text[:160]
+        return f"{message.role}: {text}"
+    tool_uses = [block.name for block in message.tool_uses]
+    if tool_uses:
+        return f"{message.role}: tool calls -> {', '.join(tool_uses[:4])}"
+    if any(isinstance(block, ToolResultBlock) for block in message.content):
+        return f"{message.role}: tool results returned"
+    return f"{message.role}: [non-text content]"
+
+
+def _build_session_memory_message(messages: list[ConversationMessage]) -> ConversationMessage | None:
+    lines: list[str] = []
+    total_chars = 0
+    for message in messages:
+        line = _summarize_message_for_memory(message)
+        if not line:
+            continue
+        projected = total_chars + len(line) + 1
+        if lines and (len(lines) >= SESSION_MEMORY_MAX_LINES or projected >= SESSION_MEMORY_MAX_CHARS):
+            lines.append("... earlier context condensed ...")
+            break
+        lines.append(line)
+        total_chars = projected
+    if not lines:
+        return None
+    body = "\n".join(lines)
+    return ConversationMessage.from_user_text(
+        "Session memory summary from earlier in this conversation:\n" + body
+    )
+
+
+def try_session_memory_compaction(
+    messages: list[ConversationMessage],
+    *,
+    preserve_recent: int = SESSION_MEMORY_KEEP_RECENT,
+) -> list[ConversationMessage] | None:
+    """Cheap deterministic compaction for long chats before full LLM compaction."""
+    if len(messages) <= preserve_recent + 4:
+        return None
+    older = messages[:-preserve_recent]
+    newer = messages[-preserve_recent:]
+    summary_message = _build_session_memory_message(older)
+    if summary_message is None:
+        return None
+    result = [summary_message, *newer]
+    if (
+        estimate_message_tokens(result) >= estimate_message_tokens(messages)
+        and len(result) >= len(messages)
+    ):
+        return None
+    return result
+
+
 # ---------------------------------------------------------------------------
 # Full compact — LLM-based summarization
 # ---------------------------------------------------------------------------
@@ -245,6 +608,7 @@ class AutoCompactState:
 
     compacted: bool = False
     turn_counter: int = 0
+    turn_id: str = ""
     consecutive_failures: int = 0
 
 
@@ -299,6 +663,11 @@ async def compact_conversation(
     preserve_recent: int = 6,
     custom_instructions: str | None = None,
     suppress_follow_up: bool = True,
+    trigger: CompactTrigger = "manual",
+    progress_callback: CompactProgressCallback | None = None,
+    emit_hooks_start: bool = True,
+    hook_executor: HookExecutor | None = None,
+    carryover_metadata: dict[str, Any] | None = None,
 ) -> list[ConversationMessage]:
     """Compact messages by calling the LLM to produce a summary.
 
@@ -337,21 +706,190 @@ async def compact_conversation(
     # Step 3: build compact request — send older messages + compact prompt
     compact_prompt = get_compact_prompt(custom_instructions)
     compact_messages = list(older) + [ConversationMessage.from_user_text(compact_prompt)]
+    attachment_paths = _extract_attachment_paths(older)
+    discovered_tools = _extract_discovered_tools(older)
+    hook_payload = {
+        "event": HookEvent.PRE_COMPACT.value,
+        "trigger": trigger,
+        "model": model,
+        "message_count": len(messages),
+        "token_count": pre_compact_tokens,
+        "preserve_recent": preserve_recent,
+        "attachments": attachment_paths,
+        "discovered_tools": discovered_tools,
+        **(carryover_metadata or {}),
+    }
+    start_checkpoint = _record_compact_checkpoint(
+        carryover_metadata,
+        checkpoint="compact_prepare",
+        trigger=trigger,
+        message_count=len(messages),
+        token_count=pre_compact_tokens,
+        details={
+            "preserve_recent": preserve_recent,
+            "attachments": attachment_paths,
+            "discovered_tools": discovered_tools,
+        },
+    )
+
+    if emit_hooks_start:
+        await _emit_progress(
+            progress_callback,
+            phase="hooks_start",
+            trigger=trigger,
+            message="Preparing conversation compaction.",
+            checkpoint="compact_hooks_start",
+            metadata=start_checkpoint,
+        )
+    if hook_executor is not None:
+        hook_result = await hook_executor.execute(HookEvent.PRE_COMPACT, hook_payload)
+        if hook_result.blocked:
+            reason = hook_result.reason or "pre-compact hook blocked compaction"
+            failed_checkpoint = _record_compact_checkpoint(
+                carryover_metadata,
+                checkpoint="compact_failed",
+                trigger=trigger,
+                message_count=len(messages),
+                token_count=pre_compact_tokens,
+                details={"reason": reason},
+            )
+            await _emit_progress(
+                progress_callback,
+                phase="compact_failed",
+                trigger=trigger,
+                message=reason,
+                checkpoint="compact_failed",
+                metadata=failed_checkpoint,
+            )
+            return messages
+    compact_start_checkpoint = _record_compact_checkpoint(
+        carryover_metadata,
+        checkpoint="compact_start",
+        trigger=trigger,
+        message_count=len(messages),
+        token_count=pre_compact_tokens,
+        details={"preserve_recent": preserve_recent},
+    )
+    await _emit_progress(
+        progress_callback,
+        phase="compact_start",
+        trigger=trigger,
+        message="Compacting conversation memory.",
+        checkpoint="compact_start",
+        metadata=compact_start_checkpoint,
+    )
 
     summary_text = ""
-    async for event in api_client.stream_message(
-        ApiMessageRequest(
-            model=model,
-            messages=compact_messages,
-            system_prompt=system_prompt or "You are a conversation summarizer.",
-            max_tokens=MAX_OUTPUT_TOKENS_FOR_SUMMARY,
-            tools=[],  # no tools for compact call
+    messages_to_summarize = compact_messages
+    retry_messages = messages_to_summarize
+    ptl_retries = 0
+
+    async def _collect_summary(summary_request_messages: list[ConversationMessage]) -> str:
+        collected = ""
+        stream = api_client.stream_message(
+            ApiMessageRequest(
+                model=model,
+                messages=summary_request_messages,
+                system_prompt=system_prompt or "You are a conversation summarizer.",
+                max_tokens=MAX_OUTPUT_TOKENS_FOR_SUMMARY,
+                tools=[],  # no tools for compact call
+            )
         )
-    ):
-        if isinstance(event, ApiMessageCompleteEvent):
-            summary_text = event.message.text
+        if inspect.isawaitable(stream):
+            stream = await stream
+        if not hasattr(stream, "__aiter__"):
+            raise RuntimeError("Compaction client did not provide a streaming response.")
+        async for event in stream:
+            if isinstance(event, ApiMessageCompleteEvent):
+                collected = event.message.text
+        if collected.strip():
+            return collected
+        raise RuntimeError(ERROR_MESSAGE_INCOMPLETE_RESPONSE)
+
+    for attempt in range(1, MAX_COMPACT_STREAMING_RETRIES + 2):
+        try:
+            summary_text = await asyncio.wait_for(
+                _collect_summary(retry_messages),
+                timeout=COMPACT_TIMEOUT_SECONDS,
+            )
+            break
+        except Exception as exc:
+            if _is_prompt_too_long_error(exc) and ptl_retries < MAX_PTL_RETRIES:
+                truncated = truncate_head_for_ptl_retry(retry_messages[:-1])
+                if truncated:
+                    ptl_retries += 1
+                    retry_messages = [*truncated, retry_messages[-1]]
+                    await _emit_progress(
+                        progress_callback,
+                        phase="compact_retry",
+                        trigger=trigger,
+                        message="Compaction prompt was too large; retrying with older context trimmed.",
+                        attempt=ptl_retries,
+                        checkpoint="compact_retry_prompt_too_long",
+                        metadata=_record_compact_checkpoint(
+                            carryover_metadata,
+                            checkpoint="compact_retry_prompt_too_long",
+                            trigger=trigger,
+                            message_count=len(retry_messages),
+                            token_count=estimate_message_tokens(retry_messages),
+                            attempt=ptl_retries,
+                            details={"ptl_retries": ptl_retries},
+                        ),
+                    )
+                    continue
+            if attempt > MAX_COMPACT_STREAMING_RETRIES:
+                await _emit_progress(
+                    progress_callback,
+                    phase="compact_failed",
+                    trigger=trigger,
+                    message=str(exc),
+                    attempt=attempt,
+                    checkpoint="compact_failed",
+                    metadata=_record_compact_checkpoint(
+                        carryover_metadata,
+                        checkpoint="compact_failed",
+                        trigger=trigger,
+                        message_count=len(retry_messages),
+                        token_count=estimate_message_tokens(retry_messages),
+                        attempt=attempt,
+                        details={"reason": str(exc)},
+                    ),
+                )
+                raise
+            await _emit_progress(
+                progress_callback,
+                phase="compact_retry",
+                trigger=trigger,
+                message=str(exc),
+                attempt=attempt,
+                checkpoint="compact_retry",
+                metadata=_record_compact_checkpoint(
+                    carryover_metadata,
+                    checkpoint="compact_retry",
+                    trigger=trigger,
+                    message_count=len(retry_messages),
+                    token_count=estimate_message_tokens(retry_messages),
+                    attempt=attempt,
+                    details={"reason": str(exc)},
+                ),
+            )
 
     if not summary_text:
+        await _emit_progress(
+            progress_callback,
+            phase="compact_failed",
+            trigger=trigger,
+            message=ERROR_MESSAGE_INCOMPLETE_RESPONSE,
+            checkpoint="compact_failed",
+            metadata=_record_compact_checkpoint(
+                carryover_metadata,
+                checkpoint="compact_failed",
+                trigger=trigger,
+                message_count=len(messages),
+                token_count=pre_compact_tokens,
+                details={"reason": ERROR_MESSAGE_INCOMPLETE_RESPONSE},
+            ),
+        )
         log.warning("Compact summary was empty — returning original messages")
         return messages
 
@@ -362,15 +900,78 @@ async def compact_conversation(
         recent_preserved=len(newer) > 0,
     )
     summary_msg = ConversationMessage.from_user_text(summary_content)
+    carryover_msg = build_compact_carryover_message(
+        older,
+        metadata=carryover_metadata,
+    )
 
-    result = [summary_msg, *newer]
+    result = [summary_msg]
+    if carryover_msg is not None:
+        result.append(carryover_msg)
+    result.extend(newer)
     post_compact_tokens = estimate_message_tokens(result)
+    post_hook_result = None
+    if hook_executor is not None:
+        post_hook_result = await hook_executor.execute(
+            HookEvent.POST_COMPACT,
+            {
+                "event": HookEvent.POST_COMPACT.value,
+                "trigger": trigger,
+                "model": model,
+                "pre_compact_message_count": len(messages),
+                "post_compact_message_count": len(result),
+                "pre_compact_tokens": pre_compact_tokens,
+                "post_compact_tokens": post_compact_tokens,
+                "attachments": attachment_paths,
+                "discovered_tools": discovered_tools,
+                **(carryover_metadata or {}),
+            },
+        )
+        hook_note = post_hook_result.reason or "\n".join(
+            result.output.strip()
+            for result in post_hook_result.results
+            if result.output.strip()
+        )
+        if hook_note:
+            carryover_msg = build_compact_carryover_message(
+                older,
+                metadata=carryover_metadata,
+                hook_note=hook_note,
+            )
+            result = [summary_msg]
+            if carryover_msg is not None:
+                result.append(carryover_msg)
+            result.extend(newer)
+            post_compact_tokens = estimate_message_tokens(result)
     log.info(
         "Compaction done: %d -> %d messages, ~%d -> ~%d tokens (saved ~%d)",
         len(messages), len(result),
         pre_compact_tokens, post_compact_tokens,
         pre_compact_tokens - post_compact_tokens,
     )
+    await _emit_progress(
+        progress_callback,
+        phase="compact_end",
+        trigger=trigger,
+        message="Conversation compaction complete.",
+        checkpoint="compact_end",
+        metadata=_record_compact_checkpoint(
+            carryover_metadata,
+            checkpoint="compact_end",
+            trigger=trigger,
+            message_count=len(result),
+            token_count=post_compact_tokens,
+            details={
+                "pre_compact_message_count": len(messages),
+                "post_compact_message_count": len(result),
+                "pre_compact_tokens": pre_compact_tokens,
+                "post_compact_tokens": post_compact_tokens,
+                "tokens_saved": pre_compact_tokens - post_compact_tokens,
+                "attachments": attachment_paths,
+                "discovered_tools": discovered_tools,
+            },
+        ),
+    )
     return result
 
 
@@ -386,6 +987,11 @@ async def auto_compact_if_needed(
     system_prompt: str = "",
     state: AutoCompactState,
     preserve_recent: int = 6,
+    progress_callback: CompactProgressCallback | None = None,
+    force: bool = False,
+    trigger: CompactTrigger = "auto",
+    hook_executor: HookExecutor | None = None,
+    carryover_metadata: dict[str, Any] | None = None,
 ) -> tuple[list[ConversationMessage], bool]:
     """Check if auto-compact should fire, and if so, compact.
 
@@ -394,17 +1000,103 @@ async def auto_compact_if_needed(
     Returns:
         (messages, was_compacted) — if compacted, messages is the new list.
     """
-    if not should_autocompact(messages, model, state):
+    if not force and not should_autocompact(messages, model, state):
         return messages, False
 
     log.info("Auto-compact triggered (failures=%d)", state.consecutive_failures)
+    _record_compact_checkpoint(
+        carryover_metadata,
+        checkpoint=f"query_{trigger}_triggered",
+        trigger=trigger,
+        message_count=len(messages),
+        token_count=estimate_message_tokens(messages),
+        details={"consecutive_failures": state.consecutive_failures},
+    )
 
     # Try microcompact first — may be enough
     messages, tokens_freed = microcompact_messages(messages)
+    _record_compact_checkpoint(
+        carryover_metadata,
+        checkpoint="query_microcompact_end",
+        trigger=trigger,
+        message_count=len(messages),
+        token_count=estimate_message_tokens(messages),
+        details={"tokens_freed": tokens_freed},
+    )
     if tokens_freed > 0 and not should_autocompact(messages, model, state):
         log.info("Microcompact freed ~%d tokens, auto-compact no longer needed", tokens_freed)
         return messages, True
 
+    context_collapsed = try_context_collapse(messages, preserve_recent=preserve_recent)
+    if context_collapsed is not None:
+        await _emit_progress(
+            progress_callback,
+            phase="context_collapse_start",
+            trigger=trigger,
+            message="Collapsing oversized context before full compaction.",
+            checkpoint="query_context_collapse_start",
+            metadata=_record_compact_checkpoint(
+                carryover_metadata,
+                checkpoint="query_context_collapse_start",
+                trigger=trigger,
+                message_count=len(messages),
+                token_count=estimate_message_tokens(messages),
+            ),
+        )
+        messages = context_collapsed
+        await _emit_progress(
+            progress_callback,
+            phase="context_collapse_end",
+            trigger=trigger,
+            message="Context collapse complete.",
+            checkpoint="query_context_collapse_end",
+            metadata=_record_compact_checkpoint(
+                carryover_metadata,
+                checkpoint="query_context_collapse_end",
+                trigger=trigger,
+                message_count=len(messages),
+                token_count=estimate_message_tokens(messages),
+            ),
+        )
+        if not force and not should_autocompact(messages, model, state):
+            return messages, True
+
+    session_memory = try_session_memory_compaction(messages, preserve_recent=max(preserve_recent, SESSION_MEMORY_KEEP_RECENT))
+    if session_memory is not None:
+        await _emit_progress(
+            progress_callback,
+            phase="session_memory_start",
+            trigger=trigger,
+            message="Condensing earlier conversation into session memory.",
+            checkpoint="query_session_memory_start",
+            metadata=_record_compact_checkpoint(
+                carryover_metadata,
+                checkpoint="query_session_memory_start",
+                trigger=trigger,
+                message_count=len(messages),
+                token_count=estimate_message_tokens(messages),
+            ),
+        )
+        await _emit_progress(
+            progress_callback,
+            phase="session_memory_end",
+            trigger=trigger,
+            message="Session memory condensation complete.",
+            checkpoint="query_session_memory_end",
+            metadata=_record_compact_checkpoint(
+                carryover_metadata,
+                checkpoint="query_session_memory_end",
+                trigger=trigger,
+                message_count=len(session_memory),
+                token_count=estimate_message_tokens(session_memory),
+            ),
+        )
+        state.compacted = True
+        state.turn_counter += 1
+        state.turn_id = uuid4().hex
+        state.consecutive_failures = 0
+        return session_memory, True
+
     # Full compact needed
     try:
         result = await compact_conversation(
@@ -414,13 +1106,26 @@ async def auto_compact_if_needed(
             system_prompt=system_prompt,
             preserve_recent=preserve_recent,
             suppress_follow_up=True,
+            trigger=trigger,
+            progress_callback=progress_callback,
+            hook_executor=hook_executor,
+            carryover_metadata=carryover_metadata,
         )
         state.compacted = True
         state.turn_counter += 1
+        state.turn_id = uuid4().hex
         state.consecutive_failures = 0
         return result, True
     except Exception as exc:
         state.consecutive_failures += 1
+        _record_compact_checkpoint(
+            carryover_metadata,
+            checkpoint=f"query_{trigger}_failed",
+            trigger=trigger,
+            message_count=len(messages),
+            token_count=estimate_message_tokens(messages),
+            details={"reason": str(exc), "consecutive_failures": state.consecutive_failures},
+        )
         log.error(
             "Auto-compact failed (attempt %d/%d): %s",
             state.consecutive_failures,

src/openharness/ui/app.py

@@ -78,6 +78,7 @@ async def run_print_mode(
     from openharness.engine.stream_events import (
         AssistantTextDelta,
         AssistantTurnComplete,
+        CompactProgressEvent,
         ErrorEvent,
         StatusEvent,
         ToolExecutionCompleted,
@@ -155,6 +156,19 @@ async def run_print_mode(
                     obj = {"type": "error", "message": event.message, "recoverable": event.recoverable}
                     print(json.dumps(obj), flush=True)
                     events_list.append(obj)
+            elif isinstance(event, CompactProgressEvent):
+                if output_format == "text" and event.message:
+                    print(event.message, file=sys.stderr)
+                elif output_format == "stream-json":
+                    obj = {
+                        "type": "compact_progress",
+                        "phase": event.phase,
+                        "trigger": event.trigger,
+                        "attempt": event.attempt,
+                        "message": event.message,
+                    }
+                    print(json.dumps(obj), flush=True)
+                    events_list.append(obj)
             elif isinstance(event, StatusEvent):
                 if output_format == "text":
                     print(event.message, file=sys.stderr)

src/openharness/ui/backend_host.py

@@ -20,6 +20,7 @@ from openharness.themes import list_themes
 from openharness.engine.stream_events import (
     AssistantTextDelta,
     AssistantTurnComplete,
+    CompactProgressEvent,
     ErrorEvent,
     StatusEvent,
     StreamEvent,
@@ -203,6 +204,19 @@ class ReactBackendHost:
             if isinstance(event, AssistantTextDelta):
                 await self._emit(BackendEvent(type="assistant_delta", message=event.text))
                 return
+            if isinstance(event, CompactProgressEvent):
+                await self._emit(
+                    BackendEvent(
+                        type="compact_progress",
+                        compact_phase=event.phase,
+                        compact_trigger=event.trigger,
+                        attempt=event.attempt,
+                        compact_checkpoint=event.checkpoint,
+                        compact_metadata=event.metadata,
+                        message=event.message,
+                    )
+                )
+                return
             if isinstance(event, AssistantTurnComplete):
                 await self._emit(
                     BackendEvent(

src/openharness/ui/output.py

@@ -10,6 +10,7 @@ from rich.syntax import Syntax
 from openharness.engine.stream_events import (
     AssistantTextDelta,
     AssistantTurnComplete,
+    CompactProgressEvent,
     StreamEvent,
     ToolExecutionCompleted,
     ToolExecutionStarted,
@@ -70,6 +71,41 @@ class OutputRenderer:
                 self._assistant_buffer = ""
             return
 
+        if isinstance(event, CompactProgressEvent):
+            self._stop_spinner()
+            if event.message:
+                label = event.message
+            elif event.phase == "hooks_start":
+                label = (
+                    "Preparing retry compaction..."
+                    if event.trigger == "reactive"
+                    else "Preparing conversation compaction..."
+                )
+            elif event.phase == "session_memory_start":
+                label = "Condensing earlier conversation..."
+            elif event.phase == "session_memory_end":
+                label = "Conversation condensed."
+            elif event.phase == "context_collapse_start":
+                label = "Collapsing oversized context..."
+            elif event.phase == "context_collapse_end":
+                label = "Context collapse complete."
+            elif event.phase == "compact_start":
+                label = (
+                    "Context is too large. Compacting and retrying..."
+                    if event.trigger == "reactive"
+                    else "Compacting conversation memory..."
+                )
+            elif event.phase == "compact_retry":
+                label = "Retrying compaction..."
+            elif event.phase == "compact_end":
+                label = "Compaction complete."
+            elif event.phase == "compact_failed":
+                label = "Compaction failed."
+            else:
+                label = "Compacting..."
+            self.console.print(f"[yellow]\u2139 {label}[/yellow]")
+            return
+
         if isinstance(event, ToolExecutionStarted):
             self._stop_spinner()
             if self._assistant_line_open:

src/openharness/ui/protocol.py

@@ -70,6 +70,7 @@ class BackendEvent(BaseModel):
         "state_snapshot",
         "tasks_snapshot",
         "transcript_item",
+        "compact_progress",
         "assistant_delta",
         "assistant_complete",
         "line_complete",
@@ -97,6 +98,11 @@ class BackendEvent(BaseModel):
     tool_input: dict[str, Any] | None = None
     output: str | None = None
     is_error: bool | None = None
+    compact_phase: str | None = None
+    compact_trigger: str | None = None
+    attempt: int | None = None
+    compact_checkpoint: str | None = None
+    compact_metadata: dict[str, Any] | None = None
     # New fields for enhanced events
     todo_markdown: str | None = None
     plan_mode: str | None = None

src/openharness/ui/runtime.py

@@ -247,6 +247,10 @@ async def build_runtime(
         extra_skill_dirs=normalized_skill_dirs,
         extra_plugin_roots=normalized_plugin_roots,
     )
+    from uuid import uuid4
+
+    session_id = uuid4().hex[:12]
+
     engine = QueryEngine(
         api_client=resolved_api_client,
         tool_registry=tool_registry,
@@ -264,6 +268,12 @@ async def build_runtime(
             "bridge_manager": bridge_manager,
             "extra_skill_dirs": normalized_skill_dirs,
             "extra_plugin_roots": normalized_plugin_roots,
+            "permission_mode": settings.permission.mode.value,
+            "session_id": session_id,
+            "read_file_state": [],
+            "invoked_skills": [],
+            "async_agent_state": [],
+            "compact_checkpoints": [],
         },
     )
     # Restore messages from a saved session if provided
@@ -273,8 +283,6 @@ async def build_runtime(
         ]
         engine.load_messages(restored)
 
-    from uuid import uuid4
-
     return RuntimeBundle(
         api_client=resolved_api_client,
         cwd=cwd,
@@ -293,7 +301,7 @@ async def build_runtime(
         ),
         external_api_client=api_client is not None,
         enforce_max_turns=enforce_max_turns or max_turns is not None,
-        session_id=uuid4().hex[:12],
+        session_id=session_id,
         settings_overrides=settings_overrides,
         session_backend=session_backend or DEFAULT_SESSION_BACKEND,
         extra_skill_dirs=normalized_skill_dirs,

src/openharness/ui/textual_app.py

@@ -19,6 +19,7 @@ from openharness.config.settings import load_settings, save_settings
 from openharness.engine.stream_events import (
     AssistantTextDelta,
     AssistantTurnComplete,
+    CompactProgressEvent,
     ErrorEvent,
     StatusEvent,
     StreamEvent,
@@ -317,6 +318,35 @@ class OpenHarnessTerminalApp(App[None]):
             self._set_current_response(f"[bold]assistant>[/bold] {self._assistant_buffer}")
             return
 
+        if isinstance(event, CompactProgressEvent):
+            if event.phase == "hooks_start":
+                if event.trigger == "reactive":
+                    self._set_current_response("[dim]Preparing retry compaction...[/dim]")
+                else:
+                    self._set_current_response("[dim]Preparing conversation compaction...[/dim]")
+            elif event.phase == "compact_start":
+                if event.trigger == "reactive":
+                    self._set_current_response("[dim]Context too large. Compacting and retrying...[/dim]")
+                else:
+                    self._set_current_response("[dim]Compacting conversation memory...[/dim]")
+            elif event.phase == "compact_retry":
+                attempt = f" (attempt {event.attempt})" if event.attempt is not None else ""
+                self._set_current_response(f"[dim]Retrying compaction{attempt}...[/dim]")
+            elif event.phase == "compact_failed":
+                self._append_line(f"system> Compaction failed: {event.message or 'unknown error'}")
+                self._set_current_response("Ready.")
+            elif event.phase == "compact_end":
+                self._set_current_response("[dim]Compaction complete.[/dim]")
+            elif event.phase == "session_memory_start":
+                self._set_current_response("[dim]Condensing earlier conversation...[/dim]")
+            elif event.phase == "session_memory_end":
+                self._set_current_response("[dim]Condensed earlier conversation.[/dim]")
+            elif event.phase == "context_collapse_start":
+                self._set_current_response("[dim]Collapsing oversized context...[/dim]")
+            elif event.phase == "context_collapse_end":
+                self._set_current_response("[dim]Context collapse complete.[/dim]")
+            return
+
         if isinstance(event, AssistantTurnComplete):
             text = self._assistant_buffer or event.message.text or "(empty response)"
             self._append_line(f"assistant> {text}")

tests/test_engine/test_query_engine.py

@@ -8,6 +8,7 @@ from pathlib import Path
 import pytest
 
 from openharness.api.client import ApiMessageCompleteEvent, ApiRetryEvent, ApiTextDeltaEvent
+from openharness.api.errors import RequestFailure
 from openharness.api.usage import UsageSnapshot
 from openharness.config.settings import PermissionSettings
 from openharness.engine.messages import ConversationMessage, TextBlock, ToolUseBlock
@@ -15,12 +16,14 @@ from openharness.engine.query_engine import QueryEngine
 from openharness.engine.stream_events import (
     AssistantTextDelta,
     AssistantTurnComplete,
+    CompactProgressEvent,
     StatusEvent,
     ToolExecutionCompleted,
     ToolExecutionStarted,
 )
 from openharness.permissions import PermissionChecker, PermissionMode
 from openharness.tools import create_default_tool_registry
+from openharness.tools.base import ToolResult
 from openharness.hooks import HookExecutionContext, HookExecutor, HookEvent
 from openharness.hooks.loader import HookRegistry
 from openharness.hooks.schemas import PromptHookDefinition
@@ -77,6 +80,28 @@ class RetryThenSuccessApiClient:
         )
 
 
+class PromptTooLongThenSuccessApiClient:
+    def __init__(self) -> None:
+        self._calls = 0
+
+    async def stream_message(self, request):
+        self._calls += 1
+        if self._calls == 1:
+            raise RequestFailure("prompt too long")
+        if self._calls == 2:
+            yield ApiMessageCompleteEvent(
+                message=ConversationMessage(role="assistant", content=[TextBlock(text="<summary>compressed</summary>")]),
+                usage=UsageSnapshot(input_tokens=1, output_tokens=1),
+                stop_reason=None,
+            )
+            return
+        yield ApiMessageCompleteEvent(
+            message=ConversationMessage(role="assistant", content=[TextBlock(text="after reactive compact")]),
+            usage=UsageSnapshot(input_tokens=1, output_tokens=1),
+            stop_reason=None,
+        )
+
+
 @pytest.mark.asyncio
 async def test_query_engine_plain_text_reply(tmp_path: Path):
     engine = QueryEngine(
@@ -220,6 +245,195 @@ async def test_query_engine_surfaces_retry_status_events(tmp_path: Path):
     assert isinstance(events[-1], AssistantTurnComplete)
 
 
+@pytest.mark.asyncio
+async def test_query_engine_emits_compact_progress_before_reply(tmp_path: Path, monkeypatch):
+    long_text = "alpha " * 50000
+    monkeypatch.setattr("openharness.services.compact.try_session_memory_compaction", lambda *args, **kwargs: None)
+    monkeypatch.setattr("openharness.services.compact.should_autocompact", lambda *args, **kwargs: True)
+    engine = QueryEngine(
+        api_client=FakeApiClient(
+            [
+                _FakeResponse(
+                    message=ConversationMessage(role="assistant", content=[TextBlock(text="<summary>trimmed</summary>")]),
+                    usage=UsageSnapshot(input_tokens=1, output_tokens=1),
+                ),
+                _FakeResponse(
+                    message=ConversationMessage(role="assistant", content=[TextBlock(text="after compact")]),
+                    usage=UsageSnapshot(input_tokens=1, output_tokens=1),
+                ),
+            ]
+        ),
+        tool_registry=create_default_tool_registry(),
+        permission_checker=PermissionChecker(PermissionSettings()),
+        cwd=tmp_path,
+        model="claude-sonnet-4-6",
+        system_prompt="system",
+    )
+    engine.load_messages(
+        [
+            ConversationMessage(role="user", content=[TextBlock(text=long_text)]),
+            ConversationMessage(role="assistant", content=[TextBlock(text=long_text)]),
+            ConversationMessage(role="user", content=[TextBlock(text=long_text)]),
+            ConversationMessage(role="assistant", content=[TextBlock(text=long_text)]),
+            ConversationMessage(role="user", content=[TextBlock(text=long_text)]),
+            ConversationMessage(role="assistant", content=[TextBlock(text=long_text)]),
+            ConversationMessage(role="user", content=[TextBlock(text=long_text)]),
+            ConversationMessage(role="assistant", content=[TextBlock(text=long_text)]),
+        ]
+    )
+
+    events = [event async for event in engine.submit_message("hello")]
+
+    hooks_start_index = next(i for i, event in enumerate(events) if isinstance(event, CompactProgressEvent) and event.phase == "hooks_start")
+    compact_start_index = next(i for i, event in enumerate(events) if isinstance(event, CompactProgressEvent) and event.phase == "compact_start")
+    final_index = next(i for i, event in enumerate(events) if isinstance(event, AssistantTurnComplete))
+    assert hooks_start_index < compact_start_index
+    assert compact_start_index < final_index
+    assert any(isinstance(event, CompactProgressEvent) and event.phase == "compact_end" for event in events)
+
+
+@pytest.mark.asyncio
+async def test_query_engine_reactive_compacts_after_prompt_too_long(tmp_path: Path, monkeypatch):
+    monkeypatch.setattr("openharness.services.compact.try_session_memory_compaction", lambda *args, **kwargs: None)
+    monkeypatch.setattr("openharness.services.compact.should_autocompact", lambda *args, **kwargs: False)
+    engine = QueryEngine(
+        api_client=PromptTooLongThenSuccessApiClient(),
+        tool_registry=create_default_tool_registry(),
+        permission_checker=PermissionChecker(PermissionSettings()),
+        cwd=tmp_path,
+        model="claude-test",
+        system_prompt="system",
+    )
+    engine.load_messages(
+        [
+            ConversationMessage(role="user", content=[TextBlock(text="one")]),
+            ConversationMessage(role="assistant", content=[TextBlock(text="two")]),
+            ConversationMessage(role="user", content=[TextBlock(text="three")]),
+            ConversationMessage(role="assistant", content=[TextBlock(text="four")]),
+            ConversationMessage(role="user", content=[TextBlock(text="five")]),
+            ConversationMessage(role="assistant", content=[TextBlock(text="six")]),
+            ConversationMessage(role="user", content=[TextBlock(text="seven")]),
+            ConversationMessage(role="assistant", content=[TextBlock(text="eight")]),
+        ]
+    )
+
+    events = [event async for event in engine.submit_message("nine")]
+
+    assert any(
+        isinstance(event, CompactProgressEvent)
+        and event.trigger == "reactive"
+        and event.phase == "compact_start"
+        for event in events
+    )
+    assert isinstance(events[-1], AssistantTurnComplete)
+    assert events[-1].message.text == "after reactive compact"
+
+
+@pytest.mark.asyncio
+async def test_query_engine_tracks_recent_read_files_and_skills(tmp_path: Path):
+    sample = tmp_path / "hello.txt"
+    sample.write_text("alpha\nbeta\n", encoding="utf-8")
+    registry = create_default_tool_registry()
+    skill_tool = registry.get("skill")
+    assert skill_tool is not None
+
+    async def _fake_skill_execute(arguments, context):
+        del context
+        return ToolResult(output=f"Loaded skill: {arguments.name}")
+
+    monkeypatch = pytest.MonkeyPatch()
+    monkeypatch.setattr(skill_tool, "execute", _fake_skill_execute)
+
+    engine = QueryEngine(
+        api_client=FakeApiClient(
+            [
+                _FakeResponse(
+                    message=ConversationMessage(
+                        role="assistant",
+                        content=[
+                            ToolUseBlock(name="read_file", input={"path": str(sample)}),
+                            ToolUseBlock(name="skill", input={"name": "demo-skill"}),
+                        ],
+                    ),
+                    usage=UsageSnapshot(input_tokens=1, output_tokens=1),
+                ),
+                _FakeResponse(
+                    message=ConversationMessage(role="assistant", content=[TextBlock(text="done")]),
+                    usage=UsageSnapshot(input_tokens=1, output_tokens=1),
+                ),
+            ]
+        ),
+        tool_registry=registry,
+        permission_checker=PermissionChecker(PermissionSettings()),
+        cwd=tmp_path,
+        model="claude-test",
+        system_prompt="system",
+        tool_metadata={},
+    )
+
+    try:
+        events = [event async for event in engine.submit_message("track context")]
+    finally:
+        monkeypatch.undo()
+
+    assert isinstance(events[-1], AssistantTurnComplete)
+    read_state = engine._tool_metadata.get("read_file_state")
+    assert isinstance(read_state, list) and read_state
+    assert read_state[-1]["path"] == str(sample.resolve())
+    assert "alpha" in read_state[-1]["preview"]
+    invoked_skills = engine._tool_metadata.get("invoked_skills")
+    assert isinstance(invoked_skills, list)
+    assert invoked_skills[-1] == "demo-skill"
+
+
+@pytest.mark.asyncio
+async def test_query_engine_tracks_async_agent_activity(tmp_path: Path, monkeypatch):
+    registry = create_default_tool_registry()
+    agent_tool = registry.get("agent")
+    assert agent_tool is not None
+
+    async def _fake_execute(arguments, context):
+        del arguments, context
+        return ToolResult(output="Spawned agent worker@team (task_id=task_123, backend=subprocess)")
+
+    monkeypatch.setattr(agent_tool, "execute", _fake_execute)
+    engine = QueryEngine(
+        api_client=FakeApiClient(
+            [
+                _FakeResponse(
+                    message=ConversationMessage(
+                        role="assistant",
+                        content=[
+                            ToolUseBlock(
+                                name="agent",
+                                input={"description": "Inspect CI", "prompt": "Inspect CI"},
+                            )
+                        ],
+                    ),
+                    usage=UsageSnapshot(input_tokens=1, output_tokens=1),
+                ),
+                _FakeResponse(
+                    message=ConversationMessage(role="assistant", content=[TextBlock(text="spawned")]),
+                    usage=UsageSnapshot(input_tokens=1, output_tokens=1),
+                ),
+            ]
+        ),
+        tool_registry=registry,
+        permission_checker=PermissionChecker(PermissionSettings(mode=PermissionMode.FULL_AUTO)),
+        cwd=tmp_path,
+        model="claude-test",
+        system_prompt="system",
+        tool_metadata={},
+    )
+
+    events = [event async for event in engine.submit_message("spawn helper")]
+
+    assert isinstance(events[-1], AssistantTurnComplete)
+    async_state = engine._tool_metadata.get("async_agent_state")
+    assert isinstance(async_state, list)
+    assert async_state[-1].startswith("Spawned async agent")
+
+
 @pytest.mark.asyncio
 async def test_query_engine_respects_pre_tool_hook_blocks(tmp_path: Path):
     sample = tmp_path / "hello.txt"

tests/test_ohmo/test_gateway.py

@@ -13,11 +13,11 @@ from openharness.channels.bus.events import InboundMessage
 from openharness.channels.bus.queue import MessageBus
 from openharness.commands import CommandResult
 from openharness.engine.messages import ConversationMessage, ImageBlock, TextBlock
-from openharness.engine.stream_events import AssistantTextDelta, StatusEvent, ToolExecutionStarted
+from openharness.engine.stream_events import AssistantTextDelta, CompactProgressEvent, ToolExecutionStarted
 
 from ohmo.gateway.bridge import OhmoGatewayBridge, _format_gateway_error
 from ohmo.gateway.models import GatewayState
-from ohmo.gateway.runtime import OhmoSessionRuntimePool
+from ohmo.gateway.runtime import OhmoSessionRuntimePool, _build_inbound_user_message, _format_channel_progress
 from ohmo.gateway.service import OhmoGatewayService, gateway_status, stop_gateway_process
 from ohmo.gateway.router import session_key_for_message
 from ohmo.session_storage import save_session_snapshot
@@ -58,6 +58,20 @@ def test_gateway_error_formats_generic_auth_failure():
     assert "Authentication failed" in _format_gateway_error(exc)
 
 
+def test_compact_progress_formats_reactive_channel_hint_in_chinese():
+    text = _format_channel_progress(
+        channel="feishu",
+        kind="compact_progress",
+        text="",
+        session_key="feishu:c1",
+        content="帮我继续处理",
+        compact_phase="compact_start",
+        compact_trigger="reactive",
+        attempt=None,
+    )
+    assert "重试" in text
+
+
 def test_gateway_status_prefers_live_config_over_stale_state(tmp_path):
     workspace = tmp_path / ".ohmo-home"
     workspace.mkdir()
@@ -199,7 +213,7 @@ async def test_runtime_pool_stream_message_formats_auto_compact_status_for_feish
                 return None
 
             async def submit_message(self, content):
-                yield StatusEvent(message="Auto-compacting conversation memory to keep things fast and focused.")
+                yield CompactProgressEvent(phase="compact_start", trigger="auto")
                 yield AssistantTextDelta(text="done")
 
         return SimpleNamespace(
@@ -220,11 +234,87 @@ async def test_runtime_pool_stream_message_formats_auto_compact_status_for_feish
     updates = [u async for u in pool.stream_message(message, "feishu:c1")]
 
     assert updates[1].kind == "progress"
-    assert updates[1].text == "🧠 聊天有点长啦，我先帮你蹦蹦跳跳压缩一下记忆，马上带着重点回来～"
+    assert updates[1].text == "🧠 聊天有点长啦，我先帮你悄悄压缩一下记忆，马上继续～"
     assert updates[-1].kind == "final"
     assert updates[-1].text == "done"
 
 
+@pytest.mark.asyncio
+async def test_runtime_pool_stream_message_formats_compact_retry_for_feishu(tmp_path, monkeypatch):
+    workspace = tmp_path / ".ohmo-home"
+    initialize_workspace(workspace)
+
+    async def fake_build_runtime(**kwargs):
+        class FakeEngine:
+            messages = []
+            total_usage = UsageSnapshot()
+
+            def set_system_prompt(self, prompt):
+                return None
+
+            async def submit_message(self, content):
+                yield CompactProgressEvent(phase="compact_retry", trigger="auto", attempt=2, message="retrying")
+                yield AssistantTextDelta(text="done")
+
+        return SimpleNamespace(
+            engine=FakeEngine(),
+            session_id="sess123",
+            current_settings=lambda: SimpleNamespace(model="gpt-5.4"),
+            commands=SimpleNamespace(lookup=lambda raw: None),
+        )
+
+    async def fake_start_runtime(bundle):
+        return None
+
+    monkeypatch.setattr("ohmo.gateway.runtime.build_runtime", fake_build_runtime)
+    monkeypatch.setattr("ohmo.gateway.runtime.start_runtime", fake_start_runtime)
+
+    pool = OhmoSessionRuntimePool(cwd=tmp_path, workspace=workspace, provider_profile="codex")
+    message = InboundMessage(channel="feishu", sender_id="u1", chat_id="c1", content="继续")
+    updates = [u async for u in pool.stream_message(message, "feishu:c1")]
+
+    assert updates[1].kind == "progress"
+    assert "再试一次" in updates[1].text
+
+
+@pytest.mark.asyncio
+async def test_runtime_pool_stream_message_formats_compact_hooks_start_for_feishu(tmp_path, monkeypatch):
+    workspace = tmp_path / ".ohmo-home"
+    initialize_workspace(workspace)
+
+    async def fake_build_runtime(**kwargs):
+        class FakeEngine:
+            messages = []
+            total_usage = UsageSnapshot()
+
+            def set_system_prompt(self, prompt):
+                return None
+
+            async def submit_message(self, content):
+                yield CompactProgressEvent(phase="hooks_start", trigger="auto")
+                yield AssistantTextDelta(text="done")
+
+        return SimpleNamespace(
+            engine=FakeEngine(),
+            session_id="sess123",
+            current_settings=lambda: SimpleNamespace(model="gpt-5.4"),
+            commands=SimpleNamespace(lookup=lambda raw: None),
+        )
+
+    async def fake_start_runtime(bundle):
+        return None
+
+    monkeypatch.setattr("ohmo.gateway.runtime.build_runtime", fake_build_runtime)
+    monkeypatch.setattr("ohmo.gateway.runtime.start_runtime", fake_start_runtime)
+
+    pool = OhmoSessionRuntimePool(cwd=tmp_path, workspace=workspace, provider_profile="codex")
+    message = InboundMessage(channel="feishu", sender_id="u1", chat_id="c1", content="继续")
+    updates = [u async for u in pool.stream_message(message, "feishu:c1")]
+
+    assert updates[1].kind == "progress"
+    assert "准备" in updates[1].text
+
+
 @pytest.mark.asyncio
 async def test_runtime_pool_stream_message_uses_english_progress_for_english_input(tmp_path, monkeypatch):
     workspace = tmp_path / ".ohmo-home"
@@ -328,6 +418,23 @@ async def test_runtime_pool_includes_media_paths_in_prompt(tmp_path, monkeypatch
     assert "text preview: Quarterly summary Revenue up 12%" in text
 
 
+def test_runtime_pool_includes_group_speaker_context():
+    built = _build_inbound_user_message(
+        InboundMessage(
+            channel="feishu",
+            sender_id="ou_123",
+            chat_id="oc_group",
+            content="请帮我看一下",
+            metadata={"chat_type": "group", "sender_display_name": "Tang Jiabin"},
+        )
+    )
+    text = "".join(block.text for block in built.content if isinstance(block, TextBlock))
+    assert "[Channel speaker]" in text
+    assert "Tang Jiabin" in text
+    assert "Sender id: ou_123" in text
+    assert "请帮我看一下" in text
+
+
 @pytest.mark.asyncio
 async def test_gateway_bridge_publishes_progress_updates():
     bus = MessageBus()

tests/test_services/test_compact.py

@@ -2,14 +2,28 @@
 
 from __future__ import annotations
 
-from openharness.engine.messages import ConversationMessage, TextBlock
+import asyncio
+
+import pytest
+
+from openharness.api.client import ApiMessageCompleteEvent
+from openharness.api.usage import UsageSnapshot
+from openharness.engine.messages import ConversationMessage, ImageBlock, TextBlock, ToolUseBlock
+from openharness.hooks import HookEvent
 from openharness.services import (
+    compact_conversation,
     compact_messages,
     estimate_conversation_tokens,
     estimate_message_tokens,
     estimate_tokens,
     summarize_messages,
 )
+from openharness.services.compact import (
+    AutoCompactState,
+    auto_compact_if_needed,
+    try_context_collapse,
+    try_session_memory_compaction,
+)
 
 
 def test_token_estimation_helpers():
@@ -34,3 +48,210 @@ def test_compact_and_summarize_messages():
     assert len(compacted) == 3
     assert "[conversation summary]" in compacted[0].text
     assert estimate_conversation_tokens(compacted) >= 1
+
+
+class _CompactApiClient:
+    def __init__(self, responses):
+        self._responses = list(responses)
+
+    async def stream_message(self, request):
+        del request
+        response = self._responses.pop(0)
+        if isinstance(response, Exception):
+            raise response
+        if asyncio.iscoroutinefunction(response):
+            await response()
+            return
+        yield ApiMessageCompleteEvent(
+            message=ConversationMessage(role="assistant", content=[TextBlock(text=response)]),
+            usage=UsageSnapshot(input_tokens=1, output_tokens=1),
+            stop_reason=None,
+        )
+
+
+class _HookExecutorStub:
+    def __init__(self) -> None:
+        self.events: list[tuple[HookEvent, dict[str, object]]] = []
+
+    async def execute(self, event: HookEvent, payload: dict[str, object]):
+        self.events.append((event, payload))
+        from openharness.hooks.types import AggregatedHookResult
+
+        return AggregatedHookResult()
+
+
+def test_try_session_memory_compaction_reduces_long_history():
+    messages = [
+        ConversationMessage(role="user", content=[TextBlock(text=(f"user {index} " * 200).strip())])
+        if index % 2 == 0
+        else ConversationMessage(role="assistant", content=[TextBlock(text=(f"assistant {index} " * 200).strip())])
+        for index in range(20)
+    ]
+
+    result = try_session_memory_compaction(messages)
+
+    assert result is not None
+    assert len(result) < len(messages)
+    assert "Session memory summary" in result[0].text
+
+
+def test_try_context_collapse_trims_oversized_messages():
+    giant = ("alpha " * 1200).strip()
+    messages = [
+        ConversationMessage(role="user", content=[TextBlock(text=giant)]),
+        ConversationMessage(role="assistant", content=[TextBlock(text=giant)]),
+        ConversationMessage(role="user", content=[TextBlock(text=giant)]),
+        ConversationMessage(role="assistant", content=[TextBlock(text=giant)]),
+        ConversationMessage(role="user", content=[TextBlock(text=giant)]),
+        ConversationMessage(role="assistant", content=[TextBlock(text="keep recent")]),
+        ConversationMessage(role="user", content=[TextBlock(text="latest")]),
+    ]
+
+    result = try_context_collapse(messages, preserve_recent=2)
+
+    assert result is not None
+    assert "[collapsed" in result[0].text
+
+
+@pytest.mark.asyncio
+async def test_compact_conversation_retries_after_incomplete_response():
+    messages = [
+        ConversationMessage(role="user", content=[TextBlock(text="alpha")]),
+        ConversationMessage(role="assistant", content=[TextBlock(text="beta")]),
+        ConversationMessage(role="user", content=[TextBlock(text="gamma")]),
+        ConversationMessage(role="assistant", content=[TextBlock(text="delta")]),
+        ConversationMessage(role="user", content=[TextBlock(text="epsilon")]),
+        ConversationMessage(role="assistant", content=[TextBlock(text="zeta")]),
+        ConversationMessage(role="user", content=[TextBlock(text="eta")]),
+    ]
+
+    compacted = await compact_conversation(
+        messages,
+        api_client=_CompactApiClient(["", "<summary>condensed</summary>"]),
+        model="claude-test",
+    )
+
+    assert compacted[0].text.startswith("This session is being continued")
+
+
+@pytest.mark.asyncio
+async def test_compact_conversation_runs_hooks_and_preserves_carryover_state(tmp_path):
+    image_path = tmp_path / "sample.png"
+    image_path.write_bytes(
+        b"\x89PNG\r\n\x1a\n"
+        b"\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x02\x00\x00\x00\x90wS\xde"
+        b"\x00\x00\x00\x0cIDAT\x08\x99c``\x00\x00\x00\x04\x00\x01\xf6\x178U"
+        b"\x00\x00\x00\x00IEND\xaeB`\x82"
+    )
+    hook_executor = _HookExecutorStub()
+    messages = [
+        ConversationMessage(role="user", content=[ImageBlock.from_path(image_path)]),
+        ConversationMessage(role="assistant", content=[TextBlock(text="Looking at the attachment")]),
+        ConversationMessage(
+            role="assistant",
+            content=[ToolUseBlock(name="read_file", input={"path": str(image_path)})],
+        ),
+        ConversationMessage(role="user", content=[TextBlock(text="Please keep going")]),
+        ConversationMessage(role="assistant", content=[TextBlock(text="Working through it")]),
+        ConversationMessage(role="user", content=[TextBlock(text="And preserve context")]),
+        ConversationMessage(role="assistant", content=[TextBlock(text="Sure")]),
+    ]
+
+    compacted = await compact_conversation(
+        messages,
+        api_client=_CompactApiClient(["<summary>condensed</summary>"]),
+        model="claude-test",
+        preserve_recent=2,
+        hook_executor=hook_executor,
+        carryover_metadata={
+            "permission_mode": "plan",
+            "session_id": "sess123",
+            "read_file_state": [
+                {
+                    "path": str(image_path),
+                    "span": "lines 1-20",
+                    "preview": "1\tPNG header",
+                }
+            ],
+            "invoked_skills": ["pikastream-video-meeting"],
+            "async_agent_state": ["Spawned async agent [task_id=task_123]"],
+            "compact_last": {"checkpoint": "query_auto_triggered", "token_count": 12345},
+        },
+    )
+
+    assert [event for event, _payload in hook_executor.events] == [HookEvent.PRE_COMPACT, HookEvent.POST_COMPACT]
+    assert compacted[0].text.startswith("This session is being continued")
+    assert "Carry-over context preserved after compaction" in compacted[1].text
+    assert "Plan mode is still active" in compacted[1].text
+    assert str(image_path) in compacted[1].text
+    assert "read_file" in compacted[1].text
+    assert "Recently read files" in compacted[1].text
+    assert "Skills invoked earlier" in compacted[1].text
+    assert "Async agent / background task state" in compacted[1].text
+    assert "Last compact checkpoint" in compacted[1].text
+
+
+@pytest.mark.asyncio
+async def test_auto_compact_records_richer_checkpoint_metadata(monkeypatch):
+    monkeypatch.setattr("openharness.services.compact.try_session_memory_compaction", lambda *args, **kwargs: None)
+    monkeypatch.setattr("openharness.services.compact.should_autocompact", lambda *args, **kwargs: True)
+    long_text = "alpha " * 50000
+    messages = [
+        ConversationMessage(role="user", content=[TextBlock(text=long_text)]),
+        ConversationMessage(role="assistant", content=[TextBlock(text=long_text)]),
+        ConversationMessage(role="user", content=[TextBlock(text=long_text)]),
+        ConversationMessage(role="assistant", content=[TextBlock(text=long_text)]),
+        ConversationMessage(role="user", content=[TextBlock(text=long_text)]),
+        ConversationMessage(role="assistant", content=[TextBlock(text=long_text)]),
+        ConversationMessage(role="user", content=[TextBlock(text=long_text)]),
+    ]
+    metadata: dict[str, object] = {}
+
+    result, was_compacted = await auto_compact_if_needed(
+        messages,
+        api_client=_CompactApiClient(["<summary>condensed</summary>"]),
+        model="claude-sonnet-4-6",
+        state=AutoCompactState(),
+        carryover_metadata=metadata,
+    )
+
+    assert was_compacted is True
+    assert result[0].text.startswith("This session is being continued")
+    checkpoints = metadata.get("compact_checkpoints")
+    assert isinstance(checkpoints, list)
+    checkpoint_names = [entry["checkpoint"] for entry in checkpoints]
+    assert "query_auto_triggered" in checkpoint_names
+    assert "query_microcompact_end" in checkpoint_names
+    assert "compact_end" in checkpoint_names
+    assert isinstance(metadata.get("compact_last"), dict)
+    assert metadata["compact_last"]["checkpoint"] == "compact_end"
+
+
+@pytest.mark.asyncio
+async def test_auto_compact_if_needed_returns_original_messages_after_timeout(monkeypatch):
+    async def _stall():
+        await asyncio.sleep(0.05)
+
+    monkeypatch.setattr("openharness.services.compact.COMPACT_TIMEOUT_SECONDS", 0.01)
+    monkeypatch.setattr("openharness.services.compact.try_session_memory_compaction", lambda *args, **kwargs: None)
+    monkeypatch.setattr("openharness.services.compact.should_autocompact", lambda *args, **kwargs: True)
+    long_text = "alpha " * 50000
+    messages = [
+        ConversationMessage(role="user", content=[TextBlock(text=long_text)]),
+        ConversationMessage(role="assistant", content=[TextBlock(text=long_text)]),
+        ConversationMessage(role="user", content=[TextBlock(text=long_text)]),
+        ConversationMessage(role="assistant", content=[TextBlock(text=long_text)]),
+        ConversationMessage(role="user", content=[TextBlock(text=long_text)]),
+        ConversationMessage(role="assistant", content=[TextBlock(text=long_text)]),
+        ConversationMessage(role="user", content=[TextBlock(text=long_text)]),
+    ]
+
+    result, was_compacted = await auto_compact_if_needed(
+        messages,
+        api_client=_CompactApiClient([_stall]),
+        model="claude-sonnet-4-6",
+        state=AutoCompactState(),
+    )
+
+    assert was_compacted is False
+    assert result == messages

tests/test_ui/test_react_backend.py

@@ -10,6 +10,7 @@ import pytest
 
 from openharness.api.client import ApiMessageCompleteEvent
 from openharness.api.usage import UsageSnapshot
+from openharness.engine.stream_events import CompactProgressEvent
 from openharness.engine.messages import ConversationMessage, TextBlock
 from openharness.ui.backend_host import BackendHostConfig, ReactBackendHost, run_backend_host
 from openharness.ui.protocol import BackendEvent
@@ -171,6 +172,50 @@ async def test_backend_host_processes_model_turn(tmp_path, monkeypatch):
     )
 
 
+@pytest.mark.asyncio
+async def test_backend_host_emits_compact_progress_event(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+
+    host = ReactBackendHost(BackendHostConfig(api_client=StaticApiClient("unused")))
+    host._bundle = await build_runtime(api_client=StaticApiClient("unused"))
+    events = []
+
+    async def _emit(event):
+        events.append(event)
+
+    async def _fake_handle_line(bundle, line, print_system, render_event, clear_output):
+        del bundle, line, print_system, clear_output
+        await render_event(
+            CompactProgressEvent(
+                phase="compact_start",
+                trigger="auto",
+                message="Compacting conversation memory.",
+                checkpoint="compact_start",
+                metadata={"token_count": 12345},
+            )
+        )
+        return True
+
+    monkeypatch.setattr("openharness.ui.backend_host.handle_line", _fake_handle_line)
+    host._emit = _emit  # type: ignore[method-assign]
+    await start_runtime(host._bundle)
+    try:
+        should_continue = await host._process_line("hi")
+    finally:
+        await close_runtime(host._bundle)
+
+    assert should_continue is True
+    assert any(
+        event.type == "compact_progress"
+        and event.compact_phase == "compact_start"
+        and event.compact_checkpoint == "compact_start"
+        and event.compact_metadata == {"token_count": 12345}
+        for event in events
+    )
+
+
 @pytest.mark.asyncio
 async def test_backend_host_surfaces_query_errors(tmp_path, monkeypatch):
     monkeypatch.chdir(tmp_path)