Method _handle_hooks_invoke

python/copilot/session.py:2650–2721

Handle a hooks invocation from the Copilot CLI. Note: This method is internal and should not be called directly. Args: hook_type: The type of hook being invoked. input_data: The input data for the hook. Returns: The

(self, hook_type: str, input_data: Any)

        return {"sections": result}

    async def _handle_hooks_invoke(self, hook_type: str, input_data: Any) -> Any:
        """
        Handle a hooks invocation from the Copilot CLI.

        Note:
            This method is internal and should not be called directly.

        Args:
            hook_type: The type of hook being invoked.
            input_data: The input data for the hook.

        Returns:
            The hook output, or None if no handler is registered.
        """
        with self._hooks_lock:
            hooks = self._hooks

        if not hooks:
            return None

        handler_map = {
            "preToolUse": hooks.get("on_pre_tool_use"),
            "preMcpToolCall": hooks.get("on_pre_mcp_tool_call"),
            "postToolUse": hooks.get("on_post_tool_use"),
            "postToolUseFailure": hooks.get("on_post_tool_use_failure"),
            "userPromptSubmitted": hooks.get("on_user_prompt_submitted"),
            "sessionStart": hooks.get("on_session_start"),
            "sessionEnd": hooks.get("on_session_end"),
            "errorOccurred": hooks.get("on_error_occurred"),
        }

        handler = handler_map.get(hook_type)
        if not handler:
            return None

        try:
            handler_start = time.perf_counter()
            # Normalize input from the wire format:
            # - Remap wire key "cwd" to public API key "workingDirectory".
            # - Convert "timestamp" from epoch milliseconds to ``datetime`` so
            #   hook handlers see a timezone-aware ``datetime`` rather than a
            #   raw integer (matches TS PR #1357 Phase E).
            transformed: dict[str, Any] = dict(input_data)
            if "cwd" in transformed:
                transformed["workingDirectory"] = transformed.pop("cwd")
            timestamp = transformed.get("timestamp")
            if isinstance(timestamp, (int, float)):
                transformed["timestamp"] = datetime.fromtimestamp(timestamp / 1000, tz=UTC)
            # Each per-hook-type TypedDict is structurally compatible with the
            # normalized dict; cast to ``Any`` so ty doesn't try to narrow the
            # specific TypedDict variant from the runtime ``dict``.
            input_data = cast(Any, transformed)
            result = handler(input_data, {"session_id": self.session_id})
            if inspect.isawaitable(result):
                result = await result
            log_timing(
                logger,
                logging.DEBUG,