diff --git a/core/agent_runtime/runner.py b/core/agent_runtime/runner.py
index 3c13850d..91a25ecc 100644
--- a/core/agent_runtime/runner.py
+++ b/core/agent_runtime/runner.py
@@ -95,6 +95,42 @@
 )
 _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:
@@ -1529,10 +1565,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 +1581,16 @@ async def _maybe_compact(
             return messages  # summarization failed → leave it to _snip_history
 
         compacted = self._build_compacted_history(messages, summary)
+        # 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}]
         if spec.post_compact_hook is not None:
             await self._call_tool_hook(spec.post_compact_hook, "auto")
         logger.info(
diff --git a/core/events/session.py b/core/events/session.py
index 1991635b..58f938d5 100644
--- a/core/events/session.py
+++ b/core/events/session.py
@@ -556,6 +556,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}"))
@@ -622,6 +626,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:
@@ -727,6 +747,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,
diff --git a/core/harness/hooks/discovery.py b/core/harness/hooks/discovery.py
index 45bf4a20..c9a0a605 100644
--- a/core/harness/hooks/discovery.py
+++ b/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(
diff --git a/core/harness/hooks/engine.py b/core/harness/hooks/engine.py
index 26f66a11..2791a80c 100644
--- a/core/harness/hooks/engine.py
+++ b/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.)."""
diff --git a/core/harness/hooks/events.py b/core/harness/hooks/events.py
index ed393156..ca5ee64a 100644
--- a/core/harness/hooks/events.py
+++ b/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"})
 
 
diff --git a/core/harness/hooks/execution.py b/core/harness/hooks/execution.py
index b11977ef..ab3a95ec 100644
--- a/core/harness/hooks/execution.py
+++ b/core/harness/hooks/execution.py
@@ -19,6 +19,7 @@
 import asyncio
 import json
 import os
+import shutil
 import time
 from dataclasses import dataclass
 from typing import Any
@@ -56,6 +57,13 @@ class HandlerDecision:
 
 def _default_shell() -> list[str]:
     if os.name == "nt":  # pragma: no cover - posix CI
+        # Hook commands follow the Claude-Code POSIX shell contract (`;`
+        # separators, single-quoted JSON, `cat` redirection). Prefer a POSIX
+        # shell (e.g. Git Bash) on Windows so those commands actually run;
+        # fall back to cmd.exe only when no POSIX shell is available.
+        sh = shutil.which("sh")
+        if sh:
+            return [sh, "-lc"]
         comspec = os.environ.get("COMSPEC", "cmd.exe")
         return [comspec, "/C"]
     shell = os.environ.get("SHELL", "/bin/sh")
@@ -250,7 +258,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,
 }
 
diff --git a/core/harness/sandbox.py b/core/harness/sandbox.py
index 5fdf38bd..67dc9a93 100644
--- a/core/harness/sandbox.py
+++ b/core/harness/sandbox.py
@@ -44,6 +44,7 @@
 import shutil
 import tempfile
 from dataclasses import dataclass, field
+from pathlib import Path
 
 # Absolute path — never resolved via PATH (PATH-injection defense).
 _MACOS_SANDBOX_EXEC = "/usr/bin/sandbox-exec"
@@ -322,8 +323,14 @@ def wrap_argv_command(
         # ``python -m core.harness.windows_sandbox -- <inner argv...>`` creates
         # a KILL_ON_JOB_CLOSE job, spawns the inner command into it suspended,
         # and resumes it — the whole process tree dies with the wrapper.
+        #
+        # The wrapper runs as ``python -m core.harness.windows_sandbox``, so the
+        # child interpreter must be able to import ``core``. The BashTool cwd is
+        # the workspace (often a tmp dir) — not on sys.path — so inject the repo
+        # root through PYTHONPATH to keep the module importable.
         import sys as _sys
 
+        repo_root = str(Path(__file__).resolve().parents[2])
         argv = [
             _sys.executable,
             "-m",
@@ -331,7 +338,11 @@ def wrap_argv_command(
             "--",
             *inner_argv,
         ]
-        return WrappedCommand(argv=argv, backend=backend)
+        return WrappedCommand(
+            argv=argv,
+            backend=backend,
+            extra_env={"PYTHONPATH": repo_root},
+        )
 
     return WrappedCommand(argv=list(inner_argv), backend="none")
 
@@ -407,6 +418,21 @@ def build_exec_command(
         bare = [shell, "-c", command] if command is not None else list(argv or [])
         return WrappedCommand(argv=bare, backend="disabled")
 
+    # On Windows the wrapped command is launched by the Job Object sandbox
+    # (``CreateProcessW``), which cannot resolve POSIX-style shell paths like
+    # ``/bin/bash``. Resolve a real executable path (e.g. Git Bash ``sh``) so
+    # the inner command starts; callers may still override ``shell`` with any
+    # Windows-resolvable value. The disabled path above keeps the bare argv
+    # untouched (upstream-locked contract).
+    if command is not None and os.name == "nt":
+        import shutil
+
+        resolved = shutil.which(shell)
+        if resolved is None and shell in ("/bin/bash", "/bin/sh", "bash", "sh"):
+            resolved = shutil.which("sh")
+        if resolved:
+            shell = resolved
+
     policy = SandboxPolicy.for_workspace(workspace, allow_network=allow_network)
     if command is not None:
         return wrap_shell_command(command, policy, shell=shell)
diff --git a/core/harness/tools/shell.py b/core/harness/tools/shell.py
index 0f1ee1fd..b20ba345 100644
--- a/core/harness/tools/shell.py
+++ b/core/harness/tools/shell.py
@@ -105,6 +105,7 @@ async def execute(self, **kwargs: Any) -> Any:
             proc = await asyncio.create_subprocess_exec(
                 *wrapped.argv,
                 cwd=self._workspace,
+                env={**os.environ, **wrapped.extra_env},
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.STDOUT,
                 **subprocess_group_kwargs(),
diff --git a/tests/test_hooks.py b/tests/test_hooks.py
index e207d76b..778923aa 100644
--- a/tests/test_hooks.py
+++ b/tests/test_hooks.py
@@ -300,7 +300,7 @@ def test_stop_block_means_keep_going():
 
 def test_payload_delivered_on_stdin(tmp_path):
     capture = tmp_path / "payload.json"
-    eng = _engine([_handler("PreToolUse", f"cat > {capture}", matcher="*")])
+    eng = _engine([_handler("PreToolUse", f"cat > {capture.as_posix()}", matcher="*")])
     asyncio.run(eng.run_pre_tool_use("Bash", {"command": "ls"}, tool_use_id="tu-9"))
     payload = json.loads(capture.read_text())
     assert payload["session_id"] == "sess-1"
@@ -567,7 +567,9 @@ def test_session_start_and_prompt_context_injected():
 
 def test_subagent_start_payload_and_plaintext_context(tmp_path):
     capture = tmp_path / "p.json"
-    eng = _engine([_handler("SubagentStart", f"cat > {capture}; echo sub-context")])
+    eng = _engine(
+        [_handler("SubagentStart", f"cat > {capture.as_posix()}; echo sub-context")]
+    )
     res = asyncio.run(eng.run_subagent_start("worker-7", "subagent"))
     assert res.additional_contexts == ["sub-context"]  # plain-text context works
     p = json.loads(capture.read_text())
@@ -802,7 +804,7 @@ async def ask(name, args):
 def test_pre_compact_hook_block_skips_and_payload(tmp_path):
     capture = tmp_path / "p.json"
     out = json.dumps({"continue": False})
-    eng = _engine([_handler("PreCompact", f"cat > {capture}; echo '{out}'")])
+    eng = _engine([_handler("PreCompact", f"cat > {capture.as_posix()}; echo '{out}'")])
     res = asyncio.run(eng.run_pre_compact("auto"))
     assert res.block is True  # continue:false → skip compaction
     p = json.loads(capture.read_text())
@@ -818,7 +820,7 @@ def test_pre_compact_matcher_matches_trigger():
 
 def test_post_compact_hook_fires_with_trigger(tmp_path):
     capture = tmp_path / "p.json"
-    eng = _engine([_handler("PostCompact", f"cat > {capture}")])
+    eng = _engine([_handler("PostCompact", f"cat > {capture.as_posix()}")])
     asyncio.run(eng.run_post_compact("auto"))
     p = json.loads(capture.read_text())
     assert p["hook_event_name"] == "PostCompact" and p["trigger"] == "auto"
@@ -829,7 +831,7 @@ def test_post_compact_hook_fires_with_trigger(tmp_path):
 
 def test_stop_payload_carries_stop_hook_active(tmp_path):
     capture = tmp_path / "p.json"
-    eng = _engine([_handler("Stop", f"cat > {capture}")])
+    eng = _engine([_handler("Stop", f"cat > {capture.as_posix()}")])
     asyncio.run(eng.run_stop(stop_hook_active=True))
     p = json.loads(capture.read_text())
     assert p["hook_event_name"] == "Stop" and p["stop_hook_active"] is True
diff --git a/tests/test_session_end_lifecycle.py b/tests/test_session_end_lifecycle.py
new file mode 100644
index 00000000..adbe1959
--- /dev/null
+++ b/tests/test_session_end_lifecycle.py
@@ -0,0 +1,418 @@
+"""SessionEnd lifecycle + PreCompact checkpoint + MCP list-discovery e2e tests.
+
+Covers the lifecycle contracts introduced by the SessionEnd / PreCompact work:
+
+- ``SessionEnd`` fires exactly once at real session termination
+  (``AgentSession.submit(Shutdown)``), never per turn; the session-exit reason
+  doubles as the matcher input (``shutdown`` / ``interrupted`` / ``error``);
+  a failing hook is non-fatal and never blocks ``ShutdownComplete``.
+- The ``PreCompact`` hook's ``additional_contexts`` survive a successful
+  compaction as a single bounded, provider-agnostic user message, and are
+  absent when the hook blocks or summarization fails.
+- The deepcode-hooks MCP ``hooks_config.json`` list format is only accepted
+  from the ``user-mcp`` source, supports ``priority`` ordering, skips disabled
+  entries, warns on invalid entries, and honours timeouts and event aliases.
+
+Hooks are exercised as REAL subprocesses (``sh -lc`` commands that echo JSON or
+exit with a code), matching ``test_hooks.py`` so we test the true execution
+path, not a mock of it.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import shutil
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from core.harness.hooks.discovery import Handler, discover_hooks  # noqa: E402
+from core.harness.hooks.engine import HooksEngine  # noqa: E402
+from core.events.protocol import Shutdown, ShutdownComplete, UserInput  # noqa: E402
+
+pytestmark = pytest.mark.skipif(
+    shutil.which("sh") is None, reason="POSIX shell required"
+)
+
+
+def _handler(event, command, *, matcher=None, order=0, timeout=30):
+    return Handler(
+        event_name=event,
+        matcher=matcher,
+        command=command,
+        timeout_sec=timeout,
+        source="project",
+        source_path="/tmp/hooks.json",
+        display_order=order,
+    )
+
+
+def _engine(handlers, cwd="/tmp"):
+    return HooksEngine(handlers, cwd, session_id="sess-1")
+
+
+def _session(hooks_engine):
+    from core.events.session import AgentSession
+
+    return AgentSession(
+        provider=None,
+        tools=_FakeTools(),
+        model="m",
+        hooks_engine=hooks_engine,
+        context_window_tokens=8000,
+    )
+
+
+class _FakeTools:
+    """Minimal tool registry: records the params each tool ran with."""
+
+    def __init__(self):
+        self.calls = []
+
+    def get_definitions(self):
+        return []
+
+    async def execute(self, name, params):
+        self.calls.append((name, params))
+        return f"ran {name} with {params}"
+
+
+# ---------------------------------------------------------------------------
+# SessionEnd lifecycle — fires exactly once at real session termination
+# ---------------------------------------------------------------------------
+
+
+def test_session_end_fires_exactly_once_on_shutdown(tmp_path):
+    count = tmp_path / "count.txt"
+    eng = _engine([_handler("SessionEnd", f"echo x >> {count}")])
+    session = _session(eng)
+
+    asyncio.run(session.submit(Shutdown()))
+
+    assert count.read_text().count("x") == 1
+    event = asyncio.run(session.next_event())
+    assert isinstance(event.msg, ShutdownComplete)
+
+
+def test_session_end_reason_matcher(tmp_path):
+    shutdown_hits = tmp_path / "shutdown.txt"
+    other_hits = tmp_path / "other.txt"
+    eng = _engine(
+        [
+            _handler("SessionEnd", f"echo x >> {shutdown_hits}", matcher="shutdown"),
+            _handler("SessionEnd", f"echo x >> {other_hits}", matcher="complete"),
+        ]
+    )
+    session = _session(eng)
+
+    asyncio.run(session.submit(Shutdown()))
+
+    assert shutdown_hits.read_text().count("x") == 1
+    assert not other_hits.exists()
+
+
+def test_session_end_hook_failure_non_fatal():
+    eng = _engine([_handler("SessionEnd", "exit 3")])
+    session = _session(eng)
+
+    # A failing SessionEnd hook must never crash the session close.
+    asyncio.run(session.submit(Shutdown()))
+    event = asyncio.run(session.next_event())
+    assert isinstance(event.msg, ShutdownComplete)
+
+
+def test_normal_turn_does_not_trigger_session_end(tmp_path):
+    count = tmp_path / "count.txt"
+    eng = _engine([_handler("SessionEnd", f"echo x >> {count}")])
+    session = _session(eng)
+
+    async def _noop_user_input(op):
+        return None
+
+    session._run_user_input = _noop_user_input  # type: ignore[assignment]
+    asyncio.run(session.submit(UserInput("hello")))
+    assert not count.exists()
+
+    asyncio.run(session.submit(Shutdown()))
+    assert count.read_text().count("x") == 1
+
+
+# ---------------------------------------------------------------------------
+# PreCompact checkpoint — bounded, provider-safe re-injection
+# ---------------------------------------------------------------------------
+
+
+def test_build_precompact_checkpoint_empty():
+    from core.agent_runtime.runner import _build_precompact_checkpoint
+
+    assert _build_precompact_checkpoint([]) is None
+    assert _build_precompact_checkpoint(["   "]) is None
+
+
+def test_build_precompact_checkpoint_limits():
+    from core.agent_runtime.runner import (
+        _PRECOMPACT_CHECKPOINT_PREFIX,
+        _PRECOMPACT_CONTEXT_LIMIT,
+        _PRECOMPACT_TOTAL_LIMIT,
+        _build_precompact_checkpoint,
+    )
+
+    long_ctx = "y" * (_PRECOMPACT_CONTEXT_LIMIT * 2)
+    checkpoint = _build_precompact_checkpoint([long_ctx])
+    assert checkpoint is not None
+    body = checkpoint[len(_PRECOMPACT_CHECKPOINT_PREFIX) + 1 :]
+    assert len(body) == _PRECOMPACT_CONTEXT_LIMIT
+
+    many = ["z" * 3000] * 10
+    checkpoint = _build_precompact_checkpoint(many)
+    assert checkpoint is not None
+    body = checkpoint[len(_PRECOMPACT_CHECKPOINT_PREFIX) + 1 :]
+    assert len(body) <= _PRECOMPACT_TOTAL_LIMIT
+
+
+def test_maybe_compact_checkpoint_injected_after_success(monkeypatch):
+    from types import SimpleNamespace
+
+    from core.agent_runtime.runner import AgentRunSpec, AgentRunner
+
+    runner = AgentRunner(provider=object())
+    monkeypatch.setattr(
+        "core.agent_runtime.runner.estimate_prompt_tokens_chain",
+        lambda *args, **kwargs: (999_999, None),
+    )
+
+    async def fake_summarize(spec, messages, *, response_observer=None):
+        return "handoff summary"
+
+    monkeypatch.setattr(runner, "_summarize", fake_summarize)
+
+    async def pre_compact_hook(trigger):
+        return SimpleNamespace(block=False, additional_contexts=["checkpoint ctx"])
+
+    spec = AgentRunSpec(
+        initial_messages=[],
+        tools=_FakeTools(),
+        model="m",
+        max_iterations=1,
+        max_tool_result_chars=100000,
+        context_window_tokens=8000,
+        pre_compact_hook=pre_compact_hook,
+    )
+    messages = [
+        {"role": "user", "content": "turn 1"},
+        {"role": "assistant", "content": "a1"},
+        {"role": "user", "content": "turn 2"},
+        {"role": "assistant", "content": "a2"},
+        {"role": "user", "content": "turn 3"},
+    ]
+    compacted = asyncio.run(runner._maybe_compact(spec, messages))
+    checkpoint_msgs = [
+        m
+        for m in compacted
+        if m.get("role") == "user" and "PreCompact checkpoint" in str(m.get("content"))
+    ]
+    assert checkpoint_msgs, "checkpoint must survive a successful compaction"
+    assert "checkpoint ctx" in checkpoint_msgs[0]["content"]
+
+
+def test_maybe_compact_block_skips_checkpoint(monkeypatch):
+    from types import SimpleNamespace
+
+    from core.agent_runtime.runner import AgentRunSpec, AgentRunner
+
+    runner = AgentRunner(provider=object())
+    monkeypatch.setattr(
+        "core.agent_runtime.runner.estimate_prompt_tokens_chain",
+        lambda *args, **kwargs: (999_999, None),
+    )
+
+    async def pre_compact_hook(trigger):
+        return SimpleNamespace(block=True, additional_contexts=["should not appear"])
+
+    spec = AgentRunSpec(
+        initial_messages=[],
+        tools=_FakeTools(),
+        model="m",
+        max_iterations=1,
+        max_tool_result_chars=100000,
+        context_window_tokens=8000,
+        pre_compact_hook=pre_compact_hook,
+    )
+    messages = [
+        {"role": "user", "content": "turn 1"},
+        {"role": "assistant", "content": "a1"},
+        {"role": "user", "content": "turn 2"},
+        {"role": "assistant", "content": "a2"},
+        {"role": "user", "content": "turn 3"},
+    ]
+    compacted = asyncio.run(runner._maybe_compact(spec, messages))
+    assert compacted is messages  # compaction aborted this turn
+    assert "PreCompact checkpoint" not in json.dumps(compacted)
+
+
+def test_maybe_compact_summarize_failure_no_checkpoint(monkeypatch):
+    from types import SimpleNamespace
+
+    from core.agent_runtime.runner import AgentRunSpec, AgentRunner
+
+    runner = AgentRunner(provider=object())
+    monkeypatch.setattr(
+        "core.agent_runtime.runner.estimate_prompt_tokens_chain",
+        lambda *args, **kwargs: (999_999, None),
+    )
+
+    async def fake_summarize_fails(spec, messages, *, response_observer=None):
+        return None  # summarization failed
+
+    monkeypatch.setattr(runner, "_summarize", fake_summarize_fails)
+
+    async def pre_compact_hook(trigger):
+        return SimpleNamespace(block=False, additional_contexts=["should not appear"])
+
+    spec = AgentRunSpec(
+        initial_messages=[],
+        tools=_FakeTools(),
+        model="m",
+        max_iterations=1,
+        max_tool_result_chars=100000,
+        context_window_tokens=8000,
+        pre_compact_hook=pre_compact_hook,
+    )
+    messages = [
+        {"role": "user", "content": "turn 1"},
+        {"role": "assistant", "content": "a1"},
+        {"role": "user", "content": "turn 2"},
+        {"role": "assistant", "content": "a2"},
+        {"role": "user", "content": "turn 3"},
+    ]
+    compacted = asyncio.run(runner._maybe_compact(spec, messages))
+    assert compacted is messages
+    assert "PreCompact checkpoint" not in json.dumps(compacted)
+
+
+# ---------------------------------------------------------------------------
+# deepcode-hooks MCP list-format discovery (hooks_config.json)
+# ---------------------------------------------------------------------------
+
+
+def _write_config(home, ws, payload):
+    (home / ".deepcode").mkdir(parents=True, exist_ok=True)
+    (ws / ".deepcode").mkdir(parents=True, exist_ok=True)
+    (home / ".deepcode" / "hooks_config.json").write_text(
+        json.dumps(payload), encoding="utf-8"
+    )
+    return str(ws), str(home)
+
+
+def test_mcp_list_format_accepted_from_user_mcp(tmp_path):
+    ws, home = _write_config(
+        tmp_path / "home",
+        tmp_path / "ws",
+        {"hooks": [{"name": "h1", "event": "PreToolUse", "handler": "echo hi"}]},
+    )
+    result = discover_hooks(ws, home)
+    assert result.warnings == []
+    assert any(
+        h.event_name == "PreToolUse" and h.command == "echo hi" for h in result.handlers
+    )
+
+
+def test_mcp_list_format_rejected_from_other_sources(tmp_path):
+    home = tmp_path / "home"
+    ws = tmp_path / "ws"
+    (home / ".deepcode").mkdir(parents=True, exist_ok=True)
+    (ws / ".deepcode").mkdir(parents=True, exist_ok=True)
+    # list shape in a project hooks.json (non user-mcp source) must be rejected
+    (ws / ".deepcode" / "hooks.json").write_text(
+        json.dumps(
+            {"hooks": [{"name": "h1", "event": "PreToolUse", "handler": "echo hi"}]}
+        ),
+        encoding="utf-8",
+    )
+    result = discover_hooks(str(ws), str(home))
+    assert any("list-shaped hooks" in w for w in result.warnings)
+    assert result.handlers == []
+
+
+def test_mcp_priority_ordering(tmp_path):
+    ws, home = _write_config(
+        tmp_path / "home",
+        tmp_path / "ws",
+        {
+            "hooks": [
+                {
+                    "name": "low",
+                    "event": "PreToolUse",
+                    "handler": "echo low",
+                    "priority": 1,
+                },
+                {
+                    "name": "high",
+                    "event": "PreToolUse",
+                    "handler": "echo high",
+                    "priority": 10,
+                },
+            ]
+        },
+    )
+    result = discover_hooks(ws, home)
+    pre_tool = [h for h in result.handlers if h.event_name == "PreToolUse"]
+    assert [h.command for h in pre_tool] == ["echo high", "echo low"]
+    assert pre_tool[0].display_order < pre_tool[1].display_order
+
+
+def test_mcp_disabled_and_invalid_entries(tmp_path):
+    ws, home = _write_config(
+        tmp_path / "home",
+        tmp_path / "ws",
+        {
+            "hooks": [
+                {
+                    "name": "disabled",
+                    "event": "PreToolUse",
+                    "handler": "echo x",
+                    "enabled": False,
+                },
+                {"name": "no-event", "handler": "echo x"},
+                {
+                    "name": "bad-type",
+                    "event": "PreToolUse",
+                    "handler": "pass",
+                    "type": "python",
+                },
+                {"name": "ok", "event": "PreToolUse", "handler": "echo ok"},
+            ]
+        },
+    )
+    result = discover_hooks(ws, home)
+    assert [h.command for h in result.handlers] == ["echo ok"]
+    assert any("without an event" in w for w in result.warnings)
+    assert any("only shell/node" in w for w in result.warnings)
+
+
+def test_mcp_timeout_and_event_alias(tmp_path):
+    ws, home = _write_config(
+        tmp_path / "home",
+        tmp_path / "ws",
+        {
+            "hooks": [
+                {
+                    "name": "aliased",
+                    "event": "sessionStart",
+                    "handler": "echo aliased",
+                    "timeout": 7,
+                }
+            ]
+        },
+    )
+    result = discover_hooks(ws, home)
+    assert result.warnings == []
+    hook = result.handlers[0]
+    assert hook.event_name == "SessionStart"
+    assert hook.timeout_sec == 7