test_hooks_extended_e2e.py

"""
Extended hook lifecycle tests that mirror dotnet/test/HookLifecycleAndOutputTests.cs.

E2E coverage for every handler exposed on ``SessionHooks``:
``on_pre_tool_use``, ``on_post_tool_use``, ``on_post_tool_use_failure``,
``on_user_prompt_submitted``, ``on_session_start``, ``on_session_end``,
``on_error_occurred``. Output-shape behavior (modifiedPrompt /
additionalContext / errorHandling / modifiedArgs / modifiedResult /
sessionSummary) is asserted alongside hook invocation.
"""

from __future__ import annotations

import asyncio

import pytest

from copilot.session import PermissionHandler
from copilot.tools import Tool, ToolInvocation, ToolResult

from .testharness import E2ETestContext

pytestmark = pytest.mark.asyncio(loop_scope="module")


class TestHooksExtended:
    async def test_should_invoke_userpromptsubmitted_hook_and_modify_prompt(
        self, ctx: E2ETestContext
    ):
        inputs: list[dict] = []

        async def on_user_prompt_submitted(input_data, invocation):
            inputs.append(input_data)
            assert invocation["session_id"]
            return {"modifiedPrompt": "Reply with exactly: HOOKED_PROMPT"}

        session = await ctx.client.create_session(
            on_permission_request=PermissionHandler.approve_all,
            hooks={"on_user_prompt_submitted": on_user_prompt_submitted},
        )
        try:
            response = await session.send_and_wait("Say something else")
            assert inputs
            assert "Say something else" in inputs[0].get("prompt", "")
            assert "HOOKED_PROMPT" in (response.data.content or "")
        finally:
            await session.disconnect()

    async def test_should_invoke_sessionstart_hook(self, ctx: E2ETestContext):
        inputs: list[dict] = []

        async def on_session_start(input_data, invocation):
            inputs.append(input_data)
            assert invocation["session_id"]
            return {"additionalContext": "Session start hook context."}

        session = await ctx.client.create_session(
            on_permission_request=PermissionHandler.approve_all,
            hooks={"on_session_start": on_session_start},
        )
        try:
            await session.send_and_wait("Say hi")
            assert inputs
            assert inputs[0].get("source") == "new"
            assert inputs[0].get("workingDirectory")
        finally:
            await session.disconnect()

    async def test_should_invoke_sessionend_hook(self, ctx: E2ETestContext):
        inputs: list[dict] = []
        hook_invoked: asyncio.Future = asyncio.get_event_loop().create_future()

        async def on_session_end(input_data, invocation):
            inputs.append(input_data)
            if not hook_invoked.done():
                hook_invoked.set_result(input_data)
            assert invocation["session_id"]
            return {"sessionSummary": "session ended"}

        session = await ctx.client.create_session(
            on_permission_request=PermissionHandler.approve_all,
            hooks={"on_session_end": on_session_end},
        )
        await session.send_and_wait("Say bye")
        await session.disconnect()
        await asyncio.wait_for(hook_invoked, 10.0)
        assert inputs

    async def test_should_register_erroroccurred_hook(self, ctx: E2ETestContext):
        inputs: list[dict] = []

        async def on_error_occurred(input_data, invocation):
            inputs.append(input_data)
            assert invocation["session_id"]
            return {"errorHandling": "skip"}

        session = await ctx.client.create_session(
            on_permission_request=PermissionHandler.approve_all,
            hooks={"on_error_occurred": on_error_occurred},
        )
        try:
            await session.send_and_wait("Say hi")
            # Registration-only test: a healthy turn shouldn't fire OnErrorOccurred.
            assert not inputs
            assert session.session_id
        finally:
            await session.disconnect()

    async def test_should_allow_pretooluse_to_return_modifiedargs_and_suppressoutput(
        self, ctx: E2ETestContext
    ):
        inputs: list[dict] = []

        def echo_value(invocation: ToolInvocation) -> ToolResult:
            args = invocation.arguments or {}
            return ToolResult(text_result_for_llm=str(args.get("value", "")))

        async def on_pre_tool_use(input_data, invocation):
            inputs.append(input_data)
            if input_data.get("toolName") != "echo_value":
                return {"permissionDecision": "allow"}
            return {
                "permissionDecision": "allow",
                "modifiedArgs": {"value": "modified by hook"},
                "suppressOutput": False,
            }

        session = await ctx.client.create_session(
            on_permission_request=PermissionHandler.approve_all,
            tools=[
                Tool(
                    name="echo_value",
                    description="Echoes the supplied value",
                    parameters={
                        "type": "object",
                        "properties": {
                            "value": {
                                "type": "string",
                                "description": "Value to echo",
                            }
                        },
                        "required": ["value"],
                    },
                    handler=echo_value,
                )
            ],
            hooks={"on_pre_tool_use": on_pre_tool_use},
        )
        try:
            response = await session.send_and_wait(
                "Call echo_value with value 'original', then reply with the result."
            )
            assert inputs
            assert any(inp.get("toolName") == "echo_value" for inp in inputs)
            assert "modified by hook" in (response.data.content or "")
        finally:
            await session.disconnect()

    async def test_should_allow_posttooluse_to_return_modifiedresult(self, ctx: E2ETestContext):
        inputs: list[dict] = []

        async def on_post_tool_use(input_data, invocation):
            inputs.append(input_data)
            if input_data.get("toolName") != "report_intent":
                return None
            return {
                "modifiedResult": {
                    "textResultForLlm": "modified by post hook",
                    "resultType": "success",
                    "toolTelemetry": {},
                },
                "suppressOutput": False,
            }

        session = await ctx.client.create_session(
            on_permission_request=PermissionHandler.approve_all,
            available_tools=["report_intent"],
            hooks={"on_post_tool_use": on_post_tool_use},
        )
        try:
            response = await session.send_and_wait(
                "Call the report_intent tool with intent 'Testing post hook', then reply done."
            )
            assert any(inp.get("toolName") == "report_intent" for inp in inputs)
            assert (response.data.content or "").strip().rstrip(".") in {"Done", "done"}
        finally:
            await session.disconnect()

    async def test_should_invoke_posttoolusefailure_hook_for_failed_tool_result(
        self, ctx: E2ETestContext
    ):
        failure_inputs: list[dict] = []
        post_tool_use_inputs: list[dict] = []

        async def on_post_tool_use(input_data, invocation):
            post_tool_use_inputs.append(input_data)
            return None

        async def on_post_tool_use_failure(input_data, invocation):
            failure_inputs.append(input_data)
            assert invocation["session_id"] == session.session_id
            return {"additionalContext": "HOOK_FAILURE_GUIDANCE_APPLIED"}

        session = await ctx.client.create_session(
            on_permission_request=PermissionHandler.approve_all,
            available_tools=["report_intent"],
            hooks={
                "on_post_tool_use": on_post_tool_use,
                "on_post_tool_use_failure": on_post_tool_use_failure,
            },
        )
        try:
            response = await session.send_and_wait(
                "Call the view tool with path 'missing.txt'. "
                "If it fails, use the hook guidance to answer."
            )
            assert not post_tool_use_inputs
            assert len(failure_inputs) == 1
            failure_input = failure_inputs[0]
            assert failure_input["toolName"] == "view"
            assert "does not exist" in failure_input["error"]
            assert "missing.txt" in failure_input["toolArgs"]["path"]
            assert failure_input["timestamp"].timestamp() > 0
            assert failure_input["workingDirectory"]
            assert "HOOK_FAILURE_GUIDANCE_APPLIED" in (response.data.content or "")
        finally:
            await session.disconnect()