feat(code): migrate legacy hooks to v2 events

libs/code/deepagents_code/config.py

@@ -112,6 +112,7 @@ class _BootstrapState:
         "BASH_ENV",
         "BASHOPTS",
         "CDPATH",
+        "COMSPEC",
         "DYLD_INSERT_LIBRARIES",
         "DYLD_LIBRARY_PATH",
         "ENV",
@@ -128,6 +129,8 @@ class _BootstrapState:
         "PYTHONSTARTUP",
         "SHELLOPTS",
         "SSH_ASKPASS",
+        "SYSTEMROOT",
+        "WINDIR",
         _INHERITED_PYTHONPATH_ENV,
     }
 )

libs/code/deepagents_code/hooks/capabilities.py

@@ -16,6 +16,8 @@
     PermissionRequestEvent,
     PostToolUseDecision,
     PostToolUseEvent,
+    PreCompactDecision,
+    PreCompactEvent,
     PreToolUseDecision,
     PreToolUseEvent,
     SessionEndDecision,
@@ -28,6 +30,8 @@
     SubagentStartEvent,
     SubagentStopDecision,
     SubagentStopEvent,
+    UserPromptSubmitDecision,
+    UserPromptSubmitEvent,
 )
 
 if TYPE_CHECKING:
@@ -52,6 +56,7 @@ class PlainOutputPolicy(StrEnum):
 class ExitCodePolicy(StrEnum):
     """How exit code 2 is interpreted for an event."""
 
+    BLOCK = "block"
     CONTEXT = "context"
     DENY = "deny"
     FEEDBACK = "feedback"
@@ -72,7 +77,7 @@ class AggregationPolicy(StrEnum):
 
 DEFAULT_COMMAND_TIMEOUT_SECONDS = 600.0
 MatcherField: TypeAlias = Literal[
-    "cause", "tool_name", "notification_type", "agent_name"
+    "cause", "tool_name", "notification_type", "agent_name", "trigger"
 ]
 
 
@@ -106,6 +111,18 @@ class HookEventSpec:
             aggregation_policy=AggregationPolicy.CONTEXT,
             supported_handler_types=frozenset({HandlerType.COMMAND}),
         ),
+        HookEvent.USER_PROMPT_SUBMIT: HookEventSpec(
+            event=HookEvent.USER_PROMPT_SUBMIT,
+            owner=HookOwner.CLIENT,
+            event_model=UserPromptSubmitEvent,
+            decision_model=UserPromptSubmitDecision,
+            matcher_field=None,
+            default_timeout_seconds=30.0,
+            exit_code_policy=ExitCodePolicy.BLOCK,
+            plain_output_policy=PlainOutputPolicy.CONTEXT,
+            aggregation_policy=AggregationPolicy.CONTEXT,
+            supported_handler_types=frozenset({HandlerType.COMMAND}),
+        ),
         HookEvent.SESSION_END: HookEventSpec(
             event=HookEvent.SESSION_END,
             owner=HookOwner.CLIENT,
@@ -166,6 +183,18 @@ class HookEventSpec:
             aggregation_policy=AggregationPolicy.FEEDBACK_AND_CONTEXT,
             supported_handler_types=frozenset({HandlerType.COMMAND}),
         ),
+        HookEvent.PRE_COMPACT: HookEventSpec(
+            event=HookEvent.PRE_COMPACT,
+            owner=HookOwner.CLIENT,
+            event_model=PreCompactEvent,
+            decision_model=PreCompactDecision,
+            matcher_field="trigger",
+            default_timeout_seconds=DEFAULT_COMMAND_TIMEOUT_SECONDS,
+            exit_code_policy=ExitCodePolicy.BLOCK,
+            plain_output_policy=PlainOutputPolicy.IGNORE,
+            aggregation_policy=AggregationPolicy.SIDE_EFFECT,
+            supported_handler_types=frozenset({HandlerType.COMMAND}),
+        ),
         HookEvent.STOP: HookEventSpec(
             event=HookEvent.STOP,
             owner=HookOwner.SERVER,
@@ -218,6 +247,8 @@ def get_event_spec(event: HookEvent) -> HookEventSpec:
     match event:
         case HookEvent.SESSION_START:
             return _HOOK_EVENT_SPECS[HookEvent.SESSION_START]
+        case HookEvent.USER_PROMPT_SUBMIT:
+            return _HOOK_EVENT_SPECS[HookEvent.USER_PROMPT_SUBMIT]
         case HookEvent.SESSION_END:
             return _HOOK_EVENT_SPECS[HookEvent.SESSION_END]
         case HookEvent.PERMISSION_REQUEST:
@@ -228,6 +259,8 @@ def get_event_spec(event: HookEvent) -> HookEventSpec:
             return _HOOK_EVENT_SPECS[HookEvent.PRE_TOOL_USE]
         case HookEvent.POST_TOOL_USE:
             return _HOOK_EVENT_SPECS[HookEvent.POST_TOOL_USE]
+        case HookEvent.PRE_COMPACT:
+            return _HOOK_EVENT_SPECS[HookEvent.PRE_COMPACT]
         case HookEvent.STOP:
             return _HOOK_EVENT_SPECS[HookEvent.STOP]
         case HookEvent.SUBAGENT_START:

libs/code/deepagents_code/hooks/env.py

@@ -10,6 +10,10 @@
 if TYPE_CHECKING:
     from collections.abc import Mapping
 
+# Shared bound for legacy hook subprocesses and the migration adapter's nested
+# `subprocess.run`. Keep the legacy dispatcher and Hooks v2 migration aligned.
+HOOK_SUBPROCESS_TIMEOUT = 5.0
+
 
 def sanitize_hook_environ(
     source: Mapping[str, str] | None = None,

libs/code/deepagents_code/hooks/legacy.py

@@ -82,6 +82,8 @@
 from concurrent.futures import ThreadPoolExecutor
 from typing import TYPE_CHECKING, Any
 
+from deepagents_code.hooks.env import HOOK_SUBPROCESS_TIMEOUT
+
 if TYPE_CHECKING:
     from collections.abc import Mapping
 
@@ -101,7 +103,6 @@
 the exact value the tool received.
 """
 
-HOOK_SUBPROCESS_TIMEOUT = 5
 """Seconds a single hook subprocess may run before it is killed.
 
 Bounds how long one misbehaving hook can block the dispatch thread. Consumed in

libs/code/deepagents_code/hooks/loading.py

@@ -172,7 +172,9 @@ def canonical_hooks_bytes(config: HooksConfig) -> bytes:
 
 
 def _canonical_group(group: MatcherGroup) -> dict[str, object]:
-    raw = group.model_dump(mode="json", by_alias=True, exclude_none=True)
+    raw = group.model_dump(
+        mode="json", by_alias=True, exclude_none=True, exclude_defaults=True
+    )
     handlers: list[dict[str, object]] = []
     hooks_raw = raw.get("hooks")
     if isinstance(hooks_raw, list):
@@ -226,8 +228,8 @@ def _read_hooks_document(
         ]
         migrated = migrate_legacy_hooks(legacy_entries)
         migration_message = (
-            f"Migrated semantically equivalent session.end hooks from {path}; "
-            "all other legacy events remain unmapped"
+            f"Migrated semantically equivalent legacy hooks from {path}; "
+            "unsupported legacy events remain unmapped"
             if migrated.hooks
             else (
                 f"Legacy hooks at {path} contained no events that are safe to "

libs/code/deepagents_code/hooks/migration.py

@@ -7,9 +7,16 @@
 
 from __future__ import annotations
 
+import base64
+import json
+import os
 import shlex
+import subprocess  # noqa: S404  # Legacy hooks are trusted user-configured commands.
+import sys
+from binascii import Error as BinasciiError
 from typing import TYPE_CHECKING
 
+from deepagents_code.hooks.env import HOOK_SUBPROCESS_TIMEOUT
 from deepagents_code.hooks.models.config import (
     CommandHandlerSpec,
     HooksConfig,
@@ -20,20 +27,32 @@
 if TYPE_CHECKING:
     from collections.abc import Mapping, Sequence
 
-# Only legacy `session.end` has the same lifecycle boundary and side-effect-only
-# behavior as its Hooks v2 counterpart. Legacy `session.start` runs once per
-# prompt execution, `context.compact` runs before the offload operation, and
-# `permission.request` is a batched observation that cannot return a decision.
 _LEGACY_EVENT_MAP: dict[str, tuple[HookEvent, str | None]] = {
+    "session.start": (HookEvent.USER_PROMPT_SUBMIT, None),
+    "user.prompt": (HookEvent.USER_PROMPT_SUBMIT, None),
+    "task.complete": (HookEvent.NOTIFICATION, "agent_completed"),
     "session.end": (HookEvent.SESSION_END, None),
+    "context.offload": (HookEvent.PRE_COMPACT, "manual"),
+    "context.compact": (HookEvent.PRE_COMPACT, "manual"),
+    "input.required": (HookEvent.NOTIFICATION, "agent_needs_input"),
 }
+# Outer runner grace so the nested adapter timeout can fire first on Windows,
+# where killing the adapter process may not reap its descendants.
+_ADAPTER_OUTER_TIMEOUT_SECONDS = HOOK_SUBPROCESS_TIMEOUT + 1.0
+_ADAPTER_MODULE = "deepagents_code.hooks.migration"
+_ADAPTER_ARGUMENT_COUNT = 2
+_THREAD_ID_EVENTS = frozenset({"session.start", "task.complete", "session.end"})
 
 
 def migrate_legacy_hooks(
     legacy_hooks: Sequence[Mapping[str, object]],
 ) -> HooksConfig:
     """Convert legacy dotted-event hook entries into Hooks v2 configuration.
 
+    Each distinct legacy event name becomes its own matcher group so a single
+    entry subscribed to both `session.start` and `user.prompt` still runs once
+    per mapped name with the matching reconstructed stdin payload.
+
     Args:
         legacy_hooks: Entries from the legacy `hooks.json` list form.
 
@@ -55,24 +74,101 @@ def migrate_legacy_hooks(
         if events is None or events == []:
             event_names = list(_LEGACY_EVENT_MAP)
         elif isinstance(events, list):
-            event_names = [name for name in events if isinstance(name, str)]
+            event_names = list(
+                dict.fromkeys(name for name in events if isinstance(name, str))
+            )
         else:
             continue
-        shell_command = shlex.join(argv)
         for event_name in event_names:
             mapped = _LEGACY_EVENT_MAP.get(event_name)
             if mapped is None:
                 continue
             event, matcher = mapped
+            adapter_argv = _adapter_argv(argv, event_name)
             grouped.setdefault(event, []).append(
                 MatcherGroup(
                     matcher=matcher,
-                    hooks=[CommandHandlerSpec(type="command", command=shell_command)],
+                    hooks=[
+                        CommandHandlerSpec(
+                            type="command",
+                            command=_shell_command(adapter_argv, os_name=os.name),
+                            argv=adapter_argv,
+                            timeout=_ADAPTER_OUTER_TIMEOUT_SECONDS,
+                        )
+                    ],
                 )
             )
     return HooksConfig(hooks=grouped)
 
 
+def _adapter_argv(argv: list[str], legacy_event: str) -> list[str]:
+    encoded_argv = base64.urlsafe_b64encode(
+        json.dumps(argv, separators=(",", ":")).encode()
+    ).decode()
+    return [sys.executable, "-m", _ADAPTER_MODULE, legacy_event, encoded_argv]
+
+
+def _shell_command(argv: Sequence[str], *, os_name: str) -> str:
+    if os_name == "nt":
+        return subprocess.list2cmdline(argv)
+    return shlex.join(argv)
+
+
+def _legacy_payload(legacy_event: str, payload: Mapping[str, object]) -> bytes:
+    legacy_payload: dict[str, object] = {"event": legacy_event}
+    if legacy_event in _THREAD_ID_EVENTS:
+        thread_id = payload.get("session_id")
+        if isinstance(thread_id, str):
+            legacy_payload["thread_id"] = thread_id
+    return json.dumps(legacy_payload, default=str).encode()
+
+
+def _decode_argv(value: str) -> list[str] | None:
+    try:
+        decoded: object = json.loads(base64.urlsafe_b64decode(value))
+    except (BinasciiError, json.JSONDecodeError, UnicodeDecodeError, ValueError):
+        return None
+    if (
+        not isinstance(decoded, list)
+        or not decoded
+        or not all(isinstance(part, str) for part in decoded)
+    ):
+        return None
+    return [part for part in decoded if isinstance(part, str)]
+
+
+def _run_adapter(args: Sequence[str]) -> int:
+    # Nested hook exit status is ignored (side-effect-only). Argument, decode,
+    # stdin, launch, and timeout failures return nonzero for runner diagnostics.
+    if len(args) != _ADAPTER_ARGUMENT_COUNT:
+        return 1
+    legacy_event, encoded_argv = args
+    argv = _decode_argv(encoded_argv)
+    if argv is None:
+        return 1
+    try:
+        payload: object = json.loads(sys.stdin.buffer.read())
+    except (json.JSONDecodeError, UnicodeDecodeError):
+        return 1
+    if not isinstance(payload, dict):
+        return 1
+    wire_payload = {str(key): value for key, value in payload.items()}
+    try:
+        subprocess.run(  # noqa: S603  # Runs the trusted legacy hook argv directly.
+            argv,
+            input=_legacy_payload(legacy_event, wire_payload),
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            check=False,
+            timeout=HOOK_SUBPROCESS_TIMEOUT,
+        )
+    except subprocess.TimeoutExpired:
+        return 1
+    except OSError:
+        return 1
+    return 0
+
+
 def is_legacy_hooks_document(data: object) -> bool:
     """Return whether `data` looks like the legacy list-shaped hooks document.
 
@@ -88,3 +184,7 @@ def is_legacy_hooks_document(data: object) -> bool:
     if not isinstance(hooks, list):
         return False
     return all(isinstance(item, dict) for item in hooks)
+
+
+if __name__ == "__main__":
+    raise SystemExit(_run_adapter(sys.argv[1:]))

libs/code/deepagents_code/hooks/models/config.py

@@ -14,7 +14,7 @@
 class _ConfigModel(BaseModel):
     # Ignore unknown keys so newer external handler fields do not fail config load.
     # Known-but-unsupported fields such as `async` are modeled explicitly and rejected.
-    model_config = ConfigDict(extra="ignore")
+    model_config = ConfigDict(extra="ignore", populate_by_name=True)
 
 
 class CommandHandlerSpec(_ConfigModel):
@@ -23,14 +23,34 @@ class CommandHandlerSpec(_ConfigModel):
     Currently only `type: "command"` is supported. Additional handler types
     remain a discriminated-union extension point and are rejected until
     implemented.
+
+    When `argv` is set, the runner launches via `create_subprocess_exec` and
+    ignores shell metacharacters in `command`.
+
+    `argv` is a temporary legacy-migration compatibility field. Remove it with
+    `hooks.legacy` and `hooks.migration` after September 1, 2026.
     """
 
     type: Literal["command"]
     command: str
+    argv: list[str] | None = None
     timeout: float | None = Field(default=None, gt=0, allow_inf_nan=False)
     status_message: str | None = Field(default=None, alias="statusMessage")
     async_: bool | None = Field(default=None, alias="async")
 
+    @field_validator("argv", mode="after")
+    @classmethod
+    def _normalize_argv(cls, value: list[str] | None) -> list[str] | None:
+        if value is None:
+            return None
+        if not value or not all(isinstance(part, str) for part in value):
+            msg = "argv must be a non-empty list of strings when provided."
+            raise ValueError(msg)
+        if not value[0].strip():
+            msg = "argv[0] must be a non-empty executable path."
+            raise ValueError(msg)
+        return value
+
     @field_validator("async_", mode="after")
     @classmethod
     def _normalize_async(cls, value: bool | None) -> None: