feat(hooks): SessionEnd lifecycle + PreCompact context injection

core/agent_runtime/runner.py

@@ -1529,10 +1529,12 @@ async def _maybe_compact(
         if estimate <= int(budget * _COMPACT_TRIGGER_FRACTION):
             return messages
 
+        pre_contexts: list[str] = []
         if spec.pre_compact_hook is not None:
             pre = await self._call_tool_hook(spec.pre_compact_hook, "auto")
             if pre is not None and getattr(pre, "block", False):
                 return messages  # a PreCompact hook aborted compaction this turn
+            pre_contexts = list(getattr(pre, "additional_contexts", None) or [])
 
         summary = await self._summarize(
             spec,
@@ -1543,6 +1545,15 @@ async def _maybe_compact(
             return messages  # summarization failed → leave it to _snip_history
 
         compacted = self._build_compacted_history(messages, summary)
+        # memento 式 checkpoint 回注: PreCompact hook 的 additionalContext
+        # 作为独立 user 消息追加, 随压缩历史一起幸存 (供压缩后模型恢复上下文)。
+        if pre_contexts:
+            compacted = compacted + [
+                {
+                    "role": "user",
+                    "content": "[PreCompact checkpoint]\n" + "\n".join(pre_contexts),
+                }
+            ]
         if spec.post_compact_hook is not None:
             await self._call_tool_hook(spec.post_compact_hook, "auto")
         logger.info(

core/events/session.py

@@ -622,6 +622,21 @@ async def _run_start_hook(self):
             logger.exception("start hook failed")
             return None
 
+    async def _run_end_hook(self, reason: str = "complete") -> None:
+        """Run SessionEnd hooks at the close of a turn.
+
+        Notification-only: a failure is logged and never crashes the turn.
+        Fired on every terminal path (complete / interrupted / error) so
+        summaries can be persisted even when compaction never ran.
+        """
+        engine = self._hooks_engine
+        if engine is None or not engine.has_event("SessionEnd"):
+            return
+        try:
+            await engine.run_session_end(reason=reason)
+        except Exception:  # noqa: BLE001 - hooks never crash a turn
+            logger.exception("session end hook failed")
+
     async def _run_prompt_hooks(
         self, text: str, hook_contexts: list[str]
     ) -> str | None:
@@ -726,6 +741,13 @@ async def _run_user_input(self, op: UserInput | str) -> None:
             self._active_turn_task = None
             if terminal is not None:
                 self._emit(terminal)
+            reason = (
+                terminal.stop_reason
+                if terminal is not None
+                and terminal.stop_reason in ("interrupted", "error")
+                else "complete"
+            )
+            await self._run_end_hook(reason)
 
     async def _execute_turn(
         self,

core/harness/hooks/discovery.py

@@ -18,8 +18,9 @@
 fold order when several hooks fire for one event — is stable and deterministic:
 
     1. user     ``~/.deepcode/hooks.json``
-    2. project  ``<workspace>/.deepcode/hooks.json``
-    3. project  ``<workspace>/.claude/settings.json``  (Claude-Code-compatible)
+    2. user-mcp ``~/.deepcode/hooks_config.json``  (deepcode-hooks MCP list format)
+    3. project  ``<workspace>/.deepcode/hooks.json``
+    4. project  ``<workspace>/.claude/settings.json``  (Claude-Code-compatible)
 
 Only ``type: command`` handlers are supported; ``prompt`` / ``agent`` handlers
 and ``async: true`` are skipped with a warning (the reference does the same).
@@ -39,6 +40,22 @@
 
 _DEFAULT_TIMEOUT_SEC = 600
 
+# deepcode-hooks MCP stores camelCase event names; core uses the reference
+# agent's PascalCase names. Keys are matched case-insensitively via .lower().
+_MCP_EVENT_ALIASES: dict[str, str] = {
+    "sessionstart": "SessionStart",
+    "sessionend": "SessionEnd",
+    "pretooluse": "PreToolUse",
+    "posttooluse": "PostToolUse",
+    "userpromptsubmit": "UserPromptSubmit",
+    "permissionrequest": "PermissionRequest",
+    "precompact": "PreCompact",
+    "postcompact": "PostCompact",
+    "subagentstart": "SubagentStart",
+    "subagentstop": "SubagentStop",
+    "stop": "Stop",
+}
+
 
 @dataclass(slots=True)
 class Handler:
@@ -48,7 +65,7 @@ class Handler:
     matcher: str | None
     command: str
     timeout_sec: int
-    source: str  # "user" | "project" — for reporting only
+    source: str  # "user" | "user-mcp" | "project" — for reporting only
     source_path: str
     display_order: int
     status_message: str | None = None
@@ -67,6 +84,7 @@ def _hook_source_files(workspace: str, home: str | None) -> list[tuple[Path, str
     ws = Path(workspace)
     return [
         (home_dir / ".deepcode" / "hooks.json", "user"),
+        (home_dir / ".deepcode" / "hooks_config.json", "user-mcp"),
         (ws / ".deepcode" / "hooks.json", "project"),
         (ws / ".claude" / "settings.json", "project"),
     ]
@@ -97,7 +115,12 @@ def discover_hooks(workspace: str, home: str | None = None) -> DiscoveryResult:
 
 
 def _load_hook_events(path: Path, warnings: list[str]) -> dict | None:
-    """Read one config file and return its ``hooks`` object (or ``None``)."""
+    """Read one config file and return its ``hooks`` object (or ``None``).
+
+    Accepts both shapes:
+    - Claude-Code dict format: ``{"hooks": {"EventName": [...]}}``
+    - deepcode-hooks MCP list format: ``{"hooks": [ {name, event, handler, ...} ]}``
+    """
     if not path.is_file():
         return None
     try:
@@ -106,9 +129,61 @@ def _load_hook_events(path: Path, warnings: list[str]) -> dict | None:
         warnings.append(f"failed to read hooks config {path}: {exc}")
         return None
     hooks = data.get("hooks") if isinstance(data, dict) else None
-    if not isinstance(hooks, dict):
-        return None
-    return hooks
+    if isinstance(hooks, dict):
+        return hooks  # Claude-Code format
+    if isinstance(hooks, list):
+        # deepcode-hooks MCP list format (hooks_config.json)
+        return _mcp_hooks_to_events(hooks, warnings, path)
+    return None
+
+
+def _mcp_hooks_to_events(mcp_hooks: list, warnings: list[str], path: Path) -> dict:
+    """Convert the deepcode-hooks MCP ``hooks`` list to the events-dict shape.
+
+    Each entry: ``{name, event, handler, type, priority, timeout, enabled, ...}``.
+    Only ``shell`` / ``node`` handlers are kept — they runnable as plain
+    commands; ``python``-typed snippets are skipped with a warning.
+    """
+    events: dict[str, list] = {}
+    for hook in mcp_hooks:
+        if not isinstance(hook, dict):
+            continue
+        if hook.get("enabled") is False:
+            continue
+        event = hook.get("event")
+        if not isinstance(event, str):
+            continue
+        canonical = _MCP_EVENT_ALIASES.get(event.lower(), event)
+        if canonical not in HOOK_EVENT_NAMES:
+            continue  # 与未知事件键一致：静默跳过 (forward-compat)
+        handler = hook.get("handler")
+        if not isinstance(handler, str) or not handler.strip():
+            continue
+        htype = hook.get("type", "shell")
+        if htype not in ("shell", "node"):
+            warnings.append(
+                f"skipping {htype!r} hook {hook.get('name', '')!r} in {path}: "
+                "only shell/node handlers are runnable as commands"
+            )
+            continue
+        timeout = hook.get("timeout")
+        try:
+            timeout_sec = max(1, int(timeout)) if timeout is not None else None
+        except (TypeError, ValueError):
+            timeout_sec = None
+        events.setdefault(canonical, []).append(
+            {
+                "matcher": "*",
+                "hooks": [
+                    {
+                        "type": "command",
+                        "command": handler,
+                        **({"timeout": timeout_sec} if timeout_sec is not None else {}),
+                    }
+                ],
+            }
+        )
+    return events
 
 
 def _append_group(

core/harness/hooks/engine.py

@@ -176,6 +176,22 @@ async def run_session_start(self, source: str = "startup") -> ContextOutcome:
             additional_contexts=folded.additional_contexts,
         )
 
+    async def run_session_end(self, reason: str = "complete") -> ContextOutcome:
+        """Session lifecycle end — a notification hook (summary persistence, etc.).
+
+        Fires unconditionally at the close of a turn (complete / interrupted /
+        error), unlike ``PreCompact`` which only fires when a summarization pass
+        actually runs. Matchers are ignored (see ``_EVENTS_WITHOUT_MATCHER``);
+        the caller logs failures so a hook can never crash the turn.
+        """
+        payload = {"hook_event_name": "SessionEnd", "reason": reason}
+        folded = await self._dispatch("SessionEnd", None, payload)
+        return ContextOutcome(
+            block=folded.block,
+            block_reason=folded.block_reason,
+            additional_contexts=folded.additional_contexts,
+        )
+
     async def run_user_prompt_submit(self, prompt: str) -> ContextOutcome:
         payload = {"hook_event_name": "UserPromptSubmit", "prompt": prompt}
         folded = await self._dispatch("UserPromptSubmit", None, payload)
@@ -192,10 +208,18 @@ async def run_stop(self, stop_hook_active: bool = False) -> StopOutcome:
 
     async def run_pre_compact(self, trigger: str = "auto") -> ContextOutcome:
         """Before a summarization pass. A ``block`` (continue:false) asks to skip
-        compaction this turn; the matcher runs against ``trigger`` (auto/manual)."""
+        compaction this turn; the matcher runs against ``trigger`` (auto/manual).
+
+        ``additional_contexts`` from hook ``hookSpecificOutput.additionalContext``
+        are passed through so a PreCompact hook can inject a checkpoint summary
+        (memento-style) that survives the compaction."""
         payload = {"hook_event_name": "PreCompact", "trigger": trigger}
         folded = await self._dispatch("PreCompact", trigger, payload)
-        return ContextOutcome(block=folded.block, block_reason=folded.block_reason)
+        return ContextOutcome(
+            block=folded.block,
+            block_reason=folded.block_reason,
+            additional_contexts=folded.additional_contexts,
+        )
 
     async def run_post_compact(self, trigger: str = "auto") -> ContextOutcome:
         """After a summarization pass — a notification hook (state saved, etc.)."""

core/harness/hooks/events.py

@@ -31,15 +31,19 @@
     "PreCompact",
     "PostCompact",
     "SessionStart",
+    "SessionEnd",
     "UserPromptSubmit",
     "SubagentStart",
     "SubagentStop",
     "Stop",
 )
 
-# Events whose ``matcher`` field is meaningful. ``UserPromptSubmit`` and ``Stop``
-# fire unconditionally, so their matchers are ignored (mirrors the reference).
-_EVENTS_WITHOUT_MATCHER: frozenset[str] = frozenset({"UserPromptSubmit", "Stop"})
+# Events whose ``matcher`` field is meaningful. ``UserPromptSubmit``, ``Stop``
+# and ``SessionEnd`` fire unconditionally, so their matchers are ignored
+# (mirrors the reference).
+_EVENTS_WITHOUT_MATCHER: frozenset[str] = frozenset(
+    {"UserPromptSubmit", "Stop", "SessionEnd"}
+)
 
 
 def matcher_applies_to_event(event_name: str, matcher: str | None) -> str | None:

core/harness/hooks/execution.py

@@ -250,7 +250,9 @@ def _decode_permission_request(obj: dict) -> HandlerDecision:
     "Stop": lambda o: _block_from_decision(o, "Stop"),
     "SubagentStop": lambda o: _block_from_decision(o, "SubagentStop"),
     "SessionStart": _decode_additional_context_only,
+    "SessionEnd": _decode_additional_context_only,
     "SubagentStart": _decode_additional_context_only,
+    "PreCompact": lambda o: _block_from_decision(o, "PreCompact"),
     "PermissionRequest": _decode_permission_request,
 }