test_codex_service_chain.py

# Copyright (c) 2026 OceanBase.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

import asyncio
import json
import os
import shutil
import socket
import subprocess
import sys
import threading
import time
from pathlib import Path

import httpx
import pytest
import uvicorn
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport
from pydantic import SecretStr

from powercontext.builtin.artifacts.memory import EmbeddingProfile
from powercontext.builtin.inference import EmbeddingResult, InferenceUnavailableError
from powercontext.builtin.persistence.sqlite import SQLiteConfig
from powercontext.builtin.runtime import InferenceConfig, RuntimeConfig
from powercontext.client import PowerContextClient
from powercontext.http import (
    AtomicMemoryLifecycleRequest,
    ListMemoryEntriesRequest,
    PrepareContextRequest,
    SearchMemoryRequest,
)
from powercontext.server.factory import create_server_app
from powercontext.server.settings import AccessControlConfig, BearerAuthConfig, McpConfig, ServerSettings
from tests.e2e.atomic_memory_models import independent_atomic_memory_model

PROJECT_ROOT = Path(__file__).resolve().parents[2]
CODEX_PLUGIN = PROJECT_ROOT / "integrations" / "codex" / "plugins" / "powercontext"
AUTH_TOKEN = "codex-e2e-token"  # noqa: S105 - non-secret test credential.
AUTHORIZATION = f"Bearer {AUTH_TOKEN}"


@pytest.mark.parametrize("recall_gate_enabled", [False, True], ids=["default", "gate-enabled"])
@pytest.mark.parametrize("answer_format", [False, True], ids=["tool-constraints", "answer-format"])
def test_execution_constraints_preserve_fts_facts_through_codex_hook(tmp_path, recall_gate_enabled, answer_format):
    app = create_server_app(
        settings=ServerSettings(
            auth=BearerAuthConfig(token=SecretStr(AUTH_TOKEN)),
            access=AccessControlConfig(mode="enforced"),
            database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'recall.db'}"),
            runtime=RuntimeConfig(artifact_processing_families=(), recall_gate_enabled=recall_gate_enabled),
            mcp=McpConfig(enabled=False),
        ),
        scheduler_path=tmp_path / "scheduler.db",
    )
    listener = socket.socket()
    listener.bind(("127.0.0.1", 0))
    host, port = listener.getsockname()
    base_url = f"http://{host}:{port}"
    server = uvicorn.Server(uvicorn.Config(app, log_level="critical"))
    thread = threading.Thread(target=server.run, kwargs={"sockets": [listener]}, daemon=True)
    thread.start()
    facts = {
        "The synthetic Quartz application has deployment codename QUARTZ-8413.",
        "The validation command for the synthetic Quartz application is `python -m pytest -q`.",
    }
    unrelated = "PostgreSQL advisory locks coordinate leader election."
    instruction_only = (
        "Use only supplied context. Do not call tools, read files, inspect old sessions, or delegate.",
        "Use only the context already supplied to you. If the facts are absent, say unknown.",
        *(
            f"Execution guideline {index}. Do not call tools, read files, inspect old sessions, or delegate."
            for index in range(36)
        ),
    )
    question = "For the synthetic Quartz application, what are the deployment codename and validation command?"
    suffix = (
        " Use only the context already supplied to you. Do not call tools, read files, inspect old sessions, or delegate."
        " If the facts are absent, say unknown."
    )
    if answer_format:
        suffix = (
            " Answer briefly and cite the artifact reference from the provided context."
            " If no reliable context is present, say so."
        )
    try:
        _wait_until_started(server, thread)
        scope_id = _create_scope(base_url, authorization=AUTHORIZATION)
        plugin = tmp_path / "plugin"
        shutil.copytree(CODEX_PLUGIN, plugin, ignore=shutil.ignore_patterns("__pycache__", ".venv"))
        config = json.loads((plugin / ".mcp.json").read_text())
        config["mcpServers"]["powercontext"]["url"] = f"{base_url}/mcp"
        (plugin / ".mcp.json").write_text(json.dumps(config))
        with httpx.Client(base_url=base_url, headers={"Authorization": AUTHORIZATION}, timeout=10) as client:
            for text in (*sorted(facts), unrelated, *instruction_only):
                remembered = client.post(
                    "/v1/memory/remember", json={"scope_id": scope_id, "kind": "fact", "text": text}
                )
                remembered.raise_for_status()
            for repeat in range(2):
                for index, query in enumerate((question, question + suffix)):
                    found = client.post("/v1/memory/search", json={"scope_id": scope_id, "query": query, "mode": "fts"})
                    found.raise_for_status()
                    assert {hit["memory"]["text"] for hit in found.json()["hits"]} == facts
                    prepared = client.post(
                        "/v1/context/prepare", json={"scope_id": scope_id, "query": query, "max_bytes": 8000}
                    )
                    prepared.raise_for_status()
                    assert prepared.json()["status"] == "ready"
                    recalled = _run_hook(
                        plugin,
                        prompt=query,
                        turn_id=f"recall-{repeat}-{index}",
                        authorization=AUTHORIZATION,
                        scope_id=scope_id,
                        flush_on_capture=False,
                    )
                    context = json.loads(recalled.stdout)["hookSpecificOutput"]["additionalContext"]
                    for content in (prepared.json()["content"], context):
                        assert all(fact in content for fact in facts)
                        assert unrelated not in content
                        assert all(text not in content for text in instruction_only)
    finally:
        server.should_exit = True
        thread.join(timeout=10)
        listener.close()
    assert not thread.is_alive()


@pytest.mark.parametrize("with_topic", [False, True], ids=["empty-topics", "existing-topic"])
@pytest.mark.parametrize("recall_gate_enabled", [False, True], ids=["default", "gate-enabled"])
def test_codex_hook_injects_fts_memory_while_optional_embedding_is_stalled(
    tmp_path: Path, with_topic: bool, recall_gate_enabled: bool
) -> None:
    class StalledEmbedding:
        profile = EmbeddingProfile(profile_id="stalled", model="stalled", dimension=2)
        stalled = False
        available = False

        async def embed(self, texts: tuple[str, ...], /) -> EmbeddingResult:
            if self.stalled:
                await asyncio.sleep(20)
            if self.available:
                return EmbeddingResult(vectors=tuple((1.0, 0.0) for _ in texts))
            raise InferenceUnavailableError("embed")

    embedding = StalledEmbedding()
    app = create_server_app(
        settings=ServerSettings(
            database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'stalled.db'}"),
            auth=BearerAuthConfig(token=SecretStr(AUTH_TOKEN)),
            access=AccessControlConfig(mode="enforced"),
            runtime=RuntimeConfig(recall_gate_enabled=recall_gate_enabled),
        ),
        embedding_model=embedding,
    )
    listener = socket.socket()
    listener.bind(("127.0.0.1", 0))
    base_url = f"http://127.0.0.1:{listener.getsockname()[1]}"
    server = uvicorn.Server(uvicorn.Config(app, log_level="critical", timeout_graceful_shutdown=1))
    thread = threading.Thread(target=server.run, kwargs={"sockets": [listener]}, daemon=True)
    thread.start()
    try:
        _wait_until_started(server, thread)
        plugin = _copy_plugin(tmp_path, base_url)
        scope_id = _create_scope(base_url, authorization=AUTHORIZATION)
        text = "For ORCHID the release codename is ORCHID-728 and the required validation command is pytest -q."
        # Configured vector writes require a matching embedding. Seed while
        # available, then stall only the optional retrieval provider.
        embedding.available = True
        with httpx.Client(base_url=base_url, headers={"Authorization": AUTHORIZATION}) as http:
            http.post(
                "/v1/memory/remember", json={"scope_id": scope_id, "kind": "fact", "text": text}
            ).raise_for_status()
            if with_topic:
                http.post(
                    f"/v1/scopes/{scope_id}/artifacts",
                    json={
                        "family": "topic-memory",
                        "content": {"title": "ORCHID release TOPIC-1665", "summary": text, "detail": text},
                    },
                ).raise_for_status()
        embedding.available = False
        embedding.stalled = True
        environment = {key: value for key, value in os.environ.items() if not key.startswith("POWERCONTEXT_")}
        environment.update(
            POWERCONTEXT_CODEX_SCOPE_ID=scope_id,
            POWERCONTEXT_CODEX_AUTHORIZATION=AUTHORIZATION,
            POWERCONTEXT_CLIENT_CONFIG_FILE=str(tmp_path / "client.json"),
            POWERCONTEXT_DIAGNOSTIC_STATE_FILE=str(tmp_path / "diagnostics.json"),
            POWERCONTEXT_HOME=str(tmp_path / "home"),
        )
        recalled = subprocess.run(
            [sys.executable, str(plugin / "hooks" / "recall.py")],
            env=environment,
            input=json.dumps({
                "hook_event_name": "UserPromptSubmit",
                "cwd": str(tmp_path),
                "prompt": "What are the ORCHID release codename and required validation command?",
                "session_id": "stalled-embedding",
            }),
            text=True,
            capture_output=True,
            check=True,
            timeout=10,
        )
        output = json.loads(recalled.stdout)
        assert "hookSpecificOutput" in output, (recalled.stdout, recalled.stderr)
        assert "ORCHID-728" in output["hookSpecificOutput"]["additionalContext"]
        assert "pytest -q" in output["hookSpecificOutput"]["additionalContext"]
        if with_topic:
            assert "TOPIC-1665" in output["hookSpecificOutput"]["additionalContext"]
    finally:
        server.should_exit = True
        thread.join(timeout=10)
        listener.close()
        assert not thread.is_alive()


@pytest.mark.parametrize("authentication_enabled", [False, True], ids=["public", "authenticated"])
def test_codex_hook_http_sdk_and_mcp_share_one_composed_context(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    authentication_enabled: bool,
) -> None:
    monkeypatch.setattr(
        "pydantic_ai.models.infer_model",
        lambda _: independent_atomic_memory_model("Use PowerContext as the composition root."),
    )
    app = create_server_app(
        settings=ServerSettings(
            auth=BearerAuthConfig(
                token=SecretStr(AUTH_TOKEN) if authentication_enabled else None,
            ),
            access=AccessControlConfig(mode="enforced" if authentication_enabled else "disabled"),
            database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'runtime.db'}"),
            inference=InferenceConfig(generation_model="test"),
            mcp=McpConfig(enabled=True),
        )
    )

    listener = socket.socket()
    listener.bind(("127.0.0.1", 0))
    host, port = listener.getsockname()
    base_url = f"http://{host}:{port}"
    server = uvicorn.Server(
        uvicorn.Config(
            app,
            log_level="critical",
            lifespan="on",
        )
    )
    thread = threading.Thread(
        target=server.run,
        kwargs={"sockets": [listener]},
        daemon=True,
    )
    thread.start()
    try:
        _wait_until_started(server, thread)
        plugin = tmp_path / "plugin"
        shutil.copytree(
            CODEX_PLUGIN,
            plugin,
            ignore=shutil.ignore_patterns("__pycache__", ".venv"),
        )
        mcp_configuration = json.loads((plugin / ".mcp.json").read_text())
        mcp_configuration["mcpServers"]["powercontext"]["url"] = f"{base_url}/mcp"
        (plugin / ".mcp.json").write_text(json.dumps(mcp_configuration))
        scope_id = _create_scope(
            base_url,
            authorization=AUTHORIZATION if authentication_enabled else None,
        )

        first = _run_hook(
            plugin,
            prompt="Remember which object is the composition root.",
            turn_id="turn-1",
            authorization=AUTHORIZATION if authentication_enabled else None,
            scope_id=scope_id,
        )
        assert first.stdout == ""
        assert AUTH_TOKEN not in first.stderr

        recalled = _run_hook(
            plugin,
            prompt="Which composition root should this project use?",
            turn_id="turn-2",
            authorization=AUTHORIZATION if authentication_enabled else None,
            scope_id=scope_id,
        )
        context = json.loads(recalled.stdout)["hookSpecificOutput"]["additionalContext"]
        envelope = json.loads(context.splitlines()[-2])
        assert envelope["items"][0]["content"] == "Use PowerContext as the composition root."
        assert envelope["items"][0]["citation"]["artifact_ref"]["family"] == "atomic-memory"
        assert AUTH_TOKEN not in recalled.stderr

        async def verify_transport_surfaces() -> None:
            async with PowerContextClient(base_url, token=AUTH_TOKEN if authentication_enabled else None) as sdk:
                found = await sdk.search_memory(
                    SearchMemoryRequest(
                        scope_id=scope_id,
                        query="PowerContext composition root",
                    )
                )
                prepared = await sdk.prepare_context(
                    PrepareContextRequest(
                        scope_id=scope_id,
                        query="PowerContext composition root",
                    )
                )
                entries = await sdk.list_memory_entries(
                    ListMemoryEntriesRequest(scope_id=scope_id),
                )
                assert found.hits
                assert {hit.memory.text for hit in found.hits} == {"Use PowerContext as the composition root."}
                assert prepared.content is not None
                prepared_envelope = json.loads(prepared.content.splitlines()[-2])
                assert "Use PowerContext as the composition root." in {
                    item["content"] for item in prepared_envelope["items"]
                }
                assert entries.entries
                selected = entries.entries[0].artifact
                exact = await sdk.get_artifact_revision(
                    scope_id, selected.family, selected.artifact_id, selected.revision
                )
                assert exact.sources[0].source_type == "content"

                transport = StreamableHttpTransport(
                    f"{base_url}/mcp",
                    headers={"Authorization": AUTHORIZATION} if authentication_enabled else None,
                )
                async with Client(transport) as mcp:
                    result = await mcp.call_tool(
                        "search_memory",
                        {
                            "scope_id": scope_id,
                            "query": "PowerContext composition root",
                        },
                    )
                structured = result.structured_content or {}
                hits = structured.get("hits")
                assert isinstance(hits, list)
                assert hits[0]["memory"]["text"] == "Use PowerContext as the composition root."

                retired_entry_ids: set[str] = set()
                current = entries
                while current.entries:
                    record = current.entries[0]
                    retired = await sdk.change_atomic_memory_lifecycle(
                        AtomicMemoryLifecycleRequest.model_validate({
                            "scope_id": scope_id,
                            "target": {
                                "artifact": record.artifact.model_dump(mode="json"),
                                "state_version": record.state_version,
                            },
                            "state": "forgotten",
                        })
                    )
                    assert len(retired.records) == 1
                    retired_entry_ids.add(retired.records[0].artifact.artifact_id)
                    current = await sdk.list_memory_entries(
                        ListMemoryEntriesRequest(scope_id=scope_id),
                    )
                audited = await sdk.list_memory_entries(
                    ListMemoryEntriesRequest(scope_id=scope_id, include_inactive=True),
                )
                assert current.entries == []
                assert {entry.artifact.artifact_id for entry in audited.entries} == retired_entry_ids
                assert all(entry.state == "forgotten" for entry in audited.entries)

        asyncio.run(verify_transport_surfaces())

        excluded = _run_hook(
            plugin,
            prompt="Which composition root should this project use?",
            turn_id="turn-3",
            authorization=AUTHORIZATION if authentication_enabled else None,
            scope_id=scope_id,
        )
        assert excluded.stdout == ""
        assert AUTH_TOKEN not in excluded.stderr
    finally:
        server.should_exit = True
        thread.join(timeout=10)
        listener.close()
        assert not thread.is_alive()


def test_codex_session_binding_switch_resume_and_child_scope_flow(tmp_path: Path) -> None:
    app = create_server_app(
        settings=ServerSettings(
            database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'scope-flow.db'}"),
            mcp=McpConfig(enabled=True),
        )
    )
    listener = socket.socket()
    listener.bind(("127.0.0.1", 0))
    host, port = listener.getsockname()
    base_url = f"http://{host}:{port}"
    server = uvicorn.Server(uvicorn.Config(app, log_level="critical", lifespan="on"))
    thread = threading.Thread(target=server.run, kwargs={"sockets": [listener]}, daemon=True)
    thread.start()
    try:
        _wait_until_started(server, thread)
        plugin = _copy_plugin(tmp_path, base_url)
        root_scope_id = _create_named_scope(base_url, title="Feature", key="codex-feature")
        first_child_id = _create_named_scope(
            base_url,
            title="Research result",
            key="codex-research",
            parent_scope_id=root_scope_id,
        )
        second_child_id = _create_named_scope(
            base_url,
            title="Validation result",
            key="codex-validation",
            parent_scope_id=root_scope_id,
        )

        _run_codex_plugin_hook(plugin, "session_binding.py", session_id="session-a")
        default_resolution = _run_codex_plugin_hook(
            plugin,
            "bind_tools.py",
            session_id="session-a",
            tool_name="mcp__powercontext__resolve_scope_binding",
            tool_input={"binding_keys": []},
        )
        default_input = json.loads(default_resolution.stdout)["hookSpecificOutput"]["updatedInput"]
        default_scope_id = httpx.post(
            f"{base_url}/v1/scope-bindings/resolve",
            json=default_input,
            timeout=5,
        ).json()["scope_id"]

        _set_session_binding(base_url, "session-a", first_child_id)
        _run_codex_plugin_hook(plugin, "session_binding.py", session_id="session-a")
        first_tool = _run_codex_plugin_hook(
            plugin,
            "bind_tools.py",
            session_id="session-a",
            tool_name="mcp__powercontext__search_memory",
            tool_input={"scope_id": second_child_id, "query": "current state"},
        )
        first_input = json.loads(first_tool.stdout)["hookSpecificOutput"]["updatedInput"]
        assert first_input["scope_id"] == first_child_id

        _run_codex_plugin_hook(plugin, "session_binding.py", session_id="session-b")
        second_tool = _run_codex_plugin_hook(
            plugin,
            "bind_tools.py",
            session_id="session-b",
            tool_name="mcp__powercontext__search_memory",
            tool_input={"scope_id": first_child_id, "query": "current state"},
        )
        second_input = json.loads(second_tool.stdout)["hookSpecificOutput"]["updatedInput"]
        assert second_input["scope_id"] == default_scope_id

        _set_session_binding(base_url, "session-b", second_child_id)
        switched_tool = _run_codex_plugin_hook(
            plugin,
            "bind_tools.py",
            session_id="session-b",
            tool_name="mcp__powercontext__search_memory",
            tool_input={"query": "current state"},
        )
        switched_input = json.loads(switched_tool.stdout)["hookSpecificOutput"]["updatedInput"]
        assert switched_input["scope_id"] == second_child_id

        _set_session_binding(base_url, "session-a", second_child_id)
        _run_codex_plugin_hook(plugin, "session_binding.py", session_id="session-a")
        resumed_after_switch = _run_codex_plugin_hook(
            plugin,
            "bind_tools.py",
            session_id="session-a",
            tool_name="mcp__powercontext__search_memory",
            tool_input={"scope_id": first_child_id, "query": "current state"},
        )
        resumed_after_switch_input = json.loads(resumed_after_switch.stdout)["hookSpecificOutput"]["updatedInput"]
        assert resumed_after_switch_input["scope_id"] == second_child_id

        peer_agent = _run_codex_plugin_hook(
            plugin,
            "bind_tools.py",
            session_id="session-b",
            tool_name="mcp__powercontext__remember_memory",
            tool_input={"scope_id": first_child_id, "kind": "decision", "text": "Shared result"},
        )
        peer_agent_input = json.loads(peer_agent.stdout)["hookSpecificOutput"]["updatedInput"]
        assert peer_agent_input["scope_id"] == second_child_id

        scopes = {scope["scope_id"]: scope for scope in httpx.get(f"{base_url}/v1/scopes", timeout=5).json()["items"]}
        assert scopes[first_child_id]["parent_scope_id"] == root_scope_id
        assert scopes[second_child_id]["parent_scope_id"] == root_scope_id
    finally:
        server.should_exit = True
        thread.join(timeout=10)
        listener.close()
        assert not thread.is_alive()


def test_codex_workspace_cli_preserves_session_and_repository_boundaries(tmp_path: Path) -> None:
    app = create_server_app(
        settings=ServerSettings(
            database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'workspace-flow.db'}"),
            mcp=McpConfig(enabled=False),
        ),
        scheduler_path=tmp_path / "workspace-scheduler.db",
    )
    listener = socket.socket()
    listener.bind(("127.0.0.1", 0))
    host, port = listener.getsockname()
    base_url = f"http://{host}:{port}"
    server = uvicorn.Server(uvicorn.Config(app, log_level="critical", lifespan="on"))
    thread = threading.Thread(target=server.run, kwargs={"sockets": [listener]}, daemon=True)
    thread.start()
    try:
        _wait_until_started(server, thread)
        plugin = _copy_plugin(tmp_path, base_url)
        checkout = tmp_path / "checkout"
        other = tmp_path / "other"
        git = shutil.which("git")
        assert git is not None
        for directory in (checkout, other):
            directory.mkdir()
            subprocess.run([git, "init", str(directory)], check=True, capture_output=True, timeout=10)
        nested = checkout / "nested"
        nested.mkdir()
        environment: dict[str, str] = {
            **os.environ,
            "CODEX_HOME": str(tmp_path / "codex-home"),
            "NO_PROXY": "127.0.0.1,localhost,::1",
            "POWERCONTEXT_CODEX_HTTP_BUDGET_SECONDS": "10",
            "POWERCONTEXT_CODEX_REQUEST_TIMEOUT_SECONDS": "5",
        }
        environment.pop("POWERCONTEXT_CODEX_SCOPE_ID", None)
        environment.pop("POWERCONTEXT_CODEX_AUTHORIZATION", None)

        def cli(directory: Path, *arguments: str, check: bool = True) -> subprocess.CompletedProcess[str]:
            return subprocess.run(
                [sys.executable, str(plugin / "scripts" / "scope_binding.py"), "--cwd", str(directory), *arguments],
                env=environment,
                capture_output=True,
                text=True,
                check=check,
                timeout=15,
            )

        def session_scope(session_id: str, directory: Path = nested) -> str:
            payload: dict[str, object] = {"session_id": session_id, "cwd": str(directory)}
            for script in ("session_binding.py", "bind_tools.py"):
                if script == "bind_tools.py":
                    payload.update({
                        "tool_name": "mcp__powercontext__search_memory",
                        "tool_input": {"query": "current state"},
                    })
                result = subprocess.run(
                    [sys.executable, str(plugin / "hooks" / script)],
                    input=json.dumps(payload),
                    env=environment,
                    capture_output=True,
                    text=True,
                    check=True,
                    timeout=15,
                )
            return json.loads(result.stdout)["hookSpecificOutput"]["updatedInput"]["scope_id"]

        default_id = cli(checkout).stdout.strip()
        assert session_scope("old") == default_id
        project_id = _create_named_scope(base_url, title="Project", key="workspace-project")
        override_id = _create_named_scope(base_url, title="Override", key="workspace-override")
        scopes_before = httpx.get(f"{base_url}/v1/scopes", timeout=5).json()
        assert cli(nested, "--bind-scope", project_id).stdout.strip() == project_id
        assert cli(checkout).stdout.strip() == project_id
        assert cli(nested).stdout.strip() == project_id
        assert cli(other).stdout.strip() == default_id
        assert session_scope("old") == default_id
        assert session_scope("new") == project_id
        assert session_scope("other", other) == default_id

        environment["POWERCONTEXT_CODEX_SCOPE_ID"] = override_id
        assert cli(nested).stdout.strip() == override_id
        assert session_scope("new") == override_id
        environment.pop("POWERCONTEXT_CODEX_SCOPE_ID")
        assert session_scope("new") == project_id

        rejected = cli(nested, "--bind-scope", "scp_00000000000000000000000000", check=False)
        assert rejected.returncode != 0
        assert cli(nested).stdout.strip() == project_id
        conflicting = cli(nested, "--bind-scope", project_id, "--clear-scope", check=False)
        assert conflicting.returncode == 2
        assert cli(nested).stdout.strip() == project_id
        assert cli(nested, "--clear-scope").stdout.strip() == default_id
        assert cli(checkout, "--clear-scope").stdout.strip() == default_id
        assert session_scope("new") == project_id
        assert session_scope("after-clear") == default_id
        assert httpx.get(f"{base_url}/v1/scopes", timeout=5).json() == scopes_before
    finally:
        server.should_exit = True
        thread.join(timeout=10)
        listener.close()
        assert not thread.is_alive()


def _run_hook(
    plugin: Path,
    *,
    prompt: str,
    turn_id: str,
    authorization: str | None,
    scope_id: str,
    flush_on_capture: bool = True,
) -> subprocess.CompletedProcess[str]:
    environment: dict[str, str] = {
        **os.environ,
        "POWERCONTEXT_CODEX_FLUSH_ON_CAPTURE": "true" if flush_on_capture else "false",
        "POWERCONTEXT_CODEX_HTTP_BUDGET_SECONDS": "10",
        "POWERCONTEXT_CODEX_REQUEST_TIMEOUT_SECONDS": "5",
        "POWERCONTEXT_CODEX_SCOPE_ID": scope_id,
    }
    environment.pop("POWERCONTEXT_CODEX_AUTHORIZATION", None)
    if authorization is not None:
        environment["POWERCONTEXT_CODEX_AUTHORIZATION"] = authorization
    return subprocess.run(
        [sys.executable, str(plugin / "hooks" / "recall.py")],
        cwd=PROJECT_ROOT,
        env=environment,
        input=json.dumps({
            "hook_event_name": "UserPromptSubmit",
            "cwd": str(PROJECT_ROOT),
            "prompt": prompt,
            "session_id": "session-e2e",
            "turn_id": turn_id,
        }),
        text=True,
        capture_output=True,
        check=True,
        timeout=15,
    )


def _create_scope(base_url: str, *, authorization: str | None) -> str:
    headers = {"Authorization": authorization} if authorization is not None else None
    response = httpx.post(
        f"{base_url}/v1/scopes",
        headers=headers,
        json={
            "title": "Codex end-to-end work",
            "summary": "Shared context for the Codex integration test.",
            "idempotency_key": "codex-e2e-work",
        },
        timeout=5,
    )
    response.raise_for_status()
    return response.json()["scope_id"]


def _copy_plugin(tmp_path: Path, base_url: str) -> Path:
    plugin = tmp_path / "scope-plugin"
    shutil.copytree(CODEX_PLUGIN, plugin, ignore=shutil.ignore_patterns("__pycache__", ".venv"))
    configuration = json.loads((plugin / ".mcp.json").read_text())
    configuration["mcpServers"]["powercontext"]["url"] = f"{base_url}/mcp"
    (plugin / ".mcp.json").write_text(json.dumps(configuration))
    return plugin


def _create_named_scope(
    base_url: str,
    *,
    title: str,
    key: str,
    parent_scope_id: str | None = None,
) -> str:
    response = httpx.post(
        f"{base_url}/v1/scopes",
        json={
            "title": title,
            "summary": f"Codex integration test Scope for {title}.",
            "parent_scope_id": parent_scope_id,
            "idempotency_key": key,
        },
        timeout=5,
    )
    response.raise_for_status()
    return response.json()["scope_id"]


def _set_session_binding(base_url: str, session_id: str, scope_id: str) -> None:
    response = httpx.put(
        f"{base_url}/v1/scope-bindings",
        json={
            "key": {"integration": "codex", "kind": "session", "external_id": session_id},
            "scope_id": scope_id,
        },
        timeout=5,
    )
    response.raise_for_status()


def _run_codex_plugin_hook(
    plugin: Path,
    script: str,
    *,
    session_id: str,
    tool_name: str | None = None,
    tool_input: dict[str, object] | None = None,
) -> subprocess.CompletedProcess[str]:
    payload: dict[str, object] = {
        "session_id": session_id,
        "cwd": str(PROJECT_ROOT),
    }
    if tool_name is not None:
        payload.update({"tool_name": tool_name, "tool_input": tool_input or {}})
    environment: dict[str, str] = {
        **os.environ,
        "POWERCONTEXT_CODEX_HTTP_BUDGET_SECONDS": "10",
        "POWERCONTEXT_CODEX_REQUEST_TIMEOUT_SECONDS": "5",
    }
    environment.pop("POWERCONTEXT_CODEX_SCOPE_ID", None)
    environment.pop("POWERCONTEXT_CODEX_AUTHORIZATION", None)
    return subprocess.run(
        [sys.executable, str(plugin / "hooks" / script)],
        cwd=PROJECT_ROOT,
        env=environment,
        input=json.dumps(payload),
        text=True,
        capture_output=True,
        check=True,
        timeout=15,
    )


def _wait_until_started(server: uvicorn.Server, thread: threading.Thread) -> None:
    deadline = time.monotonic() + 10
    while thread.is_alive() and not server.started and time.monotonic() < deadline:
        time.sleep(0.01)
    assert server.started