feat(hooks): SessionEnd lifecycle + PreCompact context injection

core/agent_runtime/runner.py

@@ -88,6 +88,42 @@
 # it survives across turns and is not re-summarized every step.
 _BACKFILL_CONTENT = "[Tool result unavailable — call was interrupted or lost]"
 
+# PreCompact checkpoint re-injection (bounded, provider-safe). A PreCompact
+# hook may attach ``additional_contexts`` that must survive a successful
+# compaction so the post-compaction model can restore working context. The
+# re-injection is a plain ``role: user`` message (provider-agnostic) carrying a
+# clearly delimited prefix, with hard limits per context and in total — a
+# runaway hook can never blow the post-compaction window back open.
+_PRECOMPACT_CHECKPOINT_PREFIX = "[PreCompact checkpoint]"
+_PRECOMPACT_CONTEXT_LIMIT = 2000  # chars per additional context
+_PRECOMPACT_TOTAL_LIMIT = 8000  # chars for the whole checkpoint block
+
+
+def _build_precompact_checkpoint(contexts: list[str]) -> str | None:
+    """Bounded, delimited representation of PreCompact hook context.
+
+    Each context is stripped, truncated to ``_PRECOMPACT_CONTEXT_LIMIT`` chars
+    and the combined block capped at ``_PRECOMPACT_TOTAL_LIMIT``. Returns
+    ``None`` when nothing survives (empty input or all contexts blank).
+    """
+    if not contexts:
+        return None
+    parts: list[str] = []
+    used = 0
+    for ctx in contexts:
+        text = (ctx or "").strip()
+        if not text:
+            continue
+        text = text[:_PRECOMPACT_CONTEXT_LIMIT]
+        room = _PRECOMPACT_TOTAL_LIMIT - used
+        if room <= 0:
+            break
+        parts.append(text[:room])
+        used += min(len(text), room) + 1  # +1 for the newline separator
+    if not parts:
+        return None
+    return _PRECOMPACT_CHECKPOINT_PREFIX + "\n" + "\n".join(parts)
+
 
 @dataclass(slots=True)
 class AgentRunSpec:
@@ -1688,15 +1724,16 @@ async def _maybe_compact(
                 if estimate is None or estimate <= trigger:
                     return messages
 
+        pre_contexts: list[str] = []
         signature = history_signature(messages)
         if signature == self._refused_compaction:
             # Already tried on exactly this history and it did not shrink.
             return messages
-
         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,
@@ -1723,6 +1760,16 @@ async def _maybe_compact(
                 spec.session_key or "default",
             )
             return messages
+        # Bounded checkpoint re-injection: the PreCompact hook's
+        # ``additional_contexts`` survive a successful compaction as a single
+        # provider-agnostic user message. ``_build_precompact_checkpoint``
+        # caps each context and the total block, so a runaway hook can never
+        # blow the post-compaction window back open. When the hook blocks or
+        # summarization fails we returned above — the checkpoint only ever
+        # appears after a successful compaction.
+        checkpoint = _build_precompact_checkpoint(pre_contexts)
+        if checkpoint:
+            compacted = compacted + [{"role": "user", "content": checkpoint}]
         self._refused_compaction = None
         if spec.post_compact_hook is not None:
             await self._call_tool_hook(spec.post_compact_hook, "auto")

core/events/session.py

@@ -622,6 +622,10 @@ async def submit(self, op: Op) -> None:
             if task is not None and not task.done() and task.cancelling() == 0:
                 task.cancel()
         elif isinstance(op, Shutdown):
+            # SessionEnd fires exactly once, here — at the real session
+            # termination boundary — never per turn. Per-turn notifications
+            # are the Stop event's job (see _EVENTS_WITHOUT_MATCHER).
+            await self._run_end_hook(reason="shutdown")
             self._emit(ShutdownComplete())
         else:  # pragma: no cover - exhaustive guard
             self._emit(ErrorEvent(message=f"unknown op: {op!r}"))
@@ -690,6 +694,22 @@ async def _run_start_hook(self):
             logger.exception("start hook failed")
             return None
 
+    async def _run_end_hook(self, reason: str = "shutdown") -> None:
+        """Run SessionEnd hooks when the session itself terminates.
+
+        Notification-only: a failure is logged and never crashes the
+        shutdown. Fired exactly once per session from ``submit(Shutdown)``
+        with a documented session-exit reason (``shutdown``); per-turn
+        notifications belong to the Stop event, not SessionEnd.
+        """
+        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 shutdown
+            logger.exception("session end hook failed")
+
     async def _run_prompt_hooks(
         self, text: str, hook_contexts: list[str]
     ) -> str | None:
@@ -810,6 +830,10 @@ async def _run_user_input(self, op: UserInput | str) -> None:
             self._active_turn_task = None
             if terminal is not None:
                 self._emit(terminal)
+            # SessionEnd is NOT fired here: this finally block runs after
+            # every turn, and SessionEnd must fire exactly once at session
+            # termination (submit(Shutdown)), not per turn. Per-turn
+            # notifications are the Stop event's responsibility.
 
     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"),
     ]
@@ -83,7 +101,7 @@ def discover_hooks(workspace: str, home: str | None = None) -> DiscoveryResult:
     warnings: list[str] = []
     order = 0
     for path, source in _hook_source_files(workspace, home):
-        events = _load_hook_events(path, warnings)
+        events = _load_hook_events(path, warnings, source)
         if not events:
             continue
         for event_name, groups in events.items():
@@ -96,8 +114,17 @@ def discover_hooks(workspace: str, home: str | None = None) -> DiscoveryResult:
     return DiscoveryResult(handlers=handlers, warnings=warnings)
 
 
-def _load_hook_events(path: Path, warnings: list[str]) -> dict | None:
-    """Read one config file and return its ``hooks`` object (or ``None``)."""
+def _load_hook_events(path: Path, warnings: list[str], source: str) -> dict | None:
+    """Read one config file and return its ``hooks`` object (or ``None``).
+
+    Two shapes are accepted:
+
+    - Claude-Code dict format (``{"hooks": {"EventName": [...]}}``) — any source.
+    - deepcode-hooks MCP list format (``{"hooks": [...]}``) — **only** from the
+      ``user-mcp`` source (``~/.deepcode/hooks_config.json``). A list shape in
+      any other source is rejected with a warning so an accidental shape
+      mismatch cannot silently disable hooks.
+    """
     if not path.is_file():
         return None
     try:
@@ -106,9 +133,104 @@ 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)
+        if source != "user-mcp":
+            warnings.append(
+                f"ignoring list-shaped hooks in {path}: only "
+                "~/.deepcode/hooks_config.json supports the deepcode-hooks "
+                "list format"
+            )
+            return None
+        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,
+    matcher, ...}``.  Entries are validated explicitly — a malformed entry is
+    reported in ``warnings`` and skipped, never silently dropped.  Only
+    ``shell`` / ``node`` handlers are kept (they run as plain commands);
+    ``python``-typed snippets are skipped with a warning.  Within one event
+    the groups are ordered by ``priority`` (highest first; stable so equal
+    priorities keep declaration order).
+    """
+    events: dict[str, list] = {}
+    for hook in mcp_hooks:
+        if not isinstance(hook, dict):
+            warnings.append(f"skipping non-object hook entry in {path}")
+            continue
+        name = hook.get("name")
+        if not isinstance(name, str) or not name.strip():
+            warnings.append(f"skipping hook without a name in {path}")
+            continue
+        if hook.get("enabled") is False:
+            continue
+        event = hook.get("event")
+        if not isinstance(event, str) or not event.strip():
+            warnings.append(f"skipping hook {name!r} without an event in {path}")
+            continue
+        canonical = _MCP_EVENT_ALIASES.get(event.lower(), event)
+        if canonical not in HOOK_EVENT_NAMES:
+            warnings.append(
+                f"skipping hook {name!r} with unknown event {event!r} in {path}"
+            )
+            continue
+        handler = hook.get("handler")
+        if not isinstance(handler, str) or not handler.strip():
+            warnings.append(f"skipping hook {name!r} without a handler in {path}")
+            continue
+        htype = hook.get("type", "shell")
+        if htype not in ("shell", "node"):
+            warnings.append(
+                f"skipping {htype!r} hook {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):
+            warnings.append(
+                f"ignoring invalid timeout {timeout!r} for hook {name!r} in {path}"
+            )
+            timeout_sec = None
+        raw_matcher = hook.get("matcher")
+        matcher = (
+            raw_matcher if isinstance(raw_matcher, str) and raw_matcher.strip() else "*"
+        )
+        priority = hook.get("priority", 0)
+        try:
+            priority_int = int(priority)
+        except (TypeError, ValueError):
+            warnings.append(
+                f"ignoring invalid priority {priority!r} for hook {name!r} in {path}"
+            )
+            priority_int = 0
+        events.setdefault(canonical, []).append(
+            {
+                "matcher": matcher,
+                "priority": priority_int,
+                "hooks": [
+                    {
+                        "type": "command",
+                        "command": handler,
+                        **({"timeout": timeout_sec} if timeout_sec is not None else {}),
+                    }
+                ],
+            }
+        )
+    # Higher ``priority`` runs first (stable sort keeps equal priorities in
+    # declaration order); the transient key is dropped before _append_group.
+    for groups in events.values():
+        groups.sort(key=lambda group: group.get("priority", 0), reverse=True)
+        for group in groups:
+            group.pop("priority", None)
+    return events
 
 
 def _append_group(

core/harness/hooks/engine.py

@@ -176,6 +176,23 @@ async def run_session_start(self, source: str = "startup") -> ContextOutcome:
             additional_contexts=folded.additional_contexts,
         )
 
+    async def run_session_end(self, reason: str = "shutdown") -> ContextOutcome:
+        """Session lifecycle end — fires exactly once when the session terminates.
+
+        Called from ``AgentSession.submit(Shutdown)``, never per turn. The
+        reason doubles as the matcher input so a hook can target a specific
+        exit path (e.g. ``matcher: "shutdown"``); supported session-exit
+        reasons are ``shutdown``, ``interrupted`` and ``error``. The caller
+        logs failures so a hook can never crash the session close.
+        """
+        payload = {"hook_event_name": "SessionEnd", "reason": reason}
+        folded = await self._dispatch("SessionEnd", reason, 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 +209,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,14 +31,18 @@
     "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 whose ``matcher`` field is meaningful. ``UserPromptSubmit`` and
+# ``Stop`` fire unconditionally, so their matchers are ignored (mirrors the
+# reference). ``SessionEnd`` DOES honour its matcher: the session-exit reason
+# (``shutdown`` / ``interrupted`` / ``error``) is matched against the
+# ``matcher`` field so hooks can target a specific exit path.
 _EVENTS_WITHOUT_MATCHER: frozenset[str] = frozenset({"UserPromptSubmit", "Stop"})