deer-flow/backend/tests/test_github_token_plumbing.py

"""Tests for the GitHub App installation-token plumbing.

Covers the end-to-end path that lets a GitHub-driven agent actually push
code and open PRs:

  dispatcher carries ``installation_id`` + a deterministic
  ``preferred_thread_id`` in ``InboundMessage.metadata``
    -> ChannelManager mints an installation token and injects it into
       ``run_context["github_token"]``
    -> the value flows through ``context=`` into ``runtime.context``
    -> the bash tool exposes it as ``GH_TOKEN`` / ``GITHUB_TOKEN`` via a
       per-call ``env`` overlay on ``Sandbox.execute_command``

The per-call overlay (rather than mutating ``os.environ``) is what keeps
concurrent runs on different repos from clobbering each other's token.
"""

from __future__ import annotations

import os
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

import httpx
import pytest
from langgraph_sdk.errors import ConflictError

from app.channels.manager import ChannelManager
from app.channels.message_bus import InboundMessage, InboundMessageType, MessageBus
from app.channels.store import ChannelStore
from deerflow.sandbox.local.local_sandbox import LocalSandbox
from deerflow.sandbox.tools import _github_env_from_runtime, bash_tool


def _make_conflict_error(detail: str = "thread_id already exists") -> ConflictError:
    """Mint a ConflictError that matches what langgraph_sdk would raise on a
    409 from ``POST /threads``. Constructing the SDK error directly requires
    an httpx.Response with an attached Request, so this helper hides the
    boilerplate.
    """
    req = httpx.Request("POST", "http://gateway/api/threads")
    resp = httpx.Response(409, json={"detail": detail}, request=req)
    return ConflictError(detail, response=resp, body={"detail": detail})


# ---------------------------------------------------------------------------
# Sandbox.execute_command env
# ---------------------------------------------------------------------------


def test_local_sandbox_env_overlay_reaches_subprocess(monkeypatch: pytest.MonkeyPatch) -> None:
    """``env`` is layered on top of a sanitized os.environ for the subprocess
    call — inherited benign vars survive, the injected secret wins."""
    import deerflow.sandbox.local.local_sandbox as local_sandbox

    captured: dict = {}

    def fake_run_posix(args, timeout, env=None):
        captured["env"] = env
        return ("", "", 0, False)

    runner = "_run_windows_command" if os.name == "nt" else "_run_posix_command"
    monkeypatch.setattr(LocalSandbox, runner, staticmethod(fake_run_posix))
    monkeypatch.setattr(LocalSandbox, "_get_shell", staticmethod(lambda: "/bin/bash"))
    monkeypatch.setattr(local_sandbox.os, "environ", {"PATH": "/usr/bin", "EXISTING": "kept"})

    LocalSandbox("local:t").execute_command("echo $GITHUB_TOKEN", env={"GITHUB_TOKEN": "tok-123"})

    env = captured["env"]
    assert env["GITHUB_TOKEN"] == "tok-123"
    # Inherited vars survive the overlay.
    assert env["EXISTING"] == "kept"


def test_local_sandbox_no_env_passes_sanitized_environ(monkeypatch: pytest.MonkeyPatch) -> None:
    """Without ``env`` the subprocess still gets a sanitized environ — platform
    secrets are scrubbed (#3861), only benign inherited vars survive."""
    import deerflow.sandbox.local.local_sandbox as local_sandbox

    captured: dict = {}

    def fake_run_posix(args, timeout, env=None):
        captured["env"] = env
        return ("", "", 0, False)

    runner = "_run_windows_command" if os.name == "nt" else "_run_posix_command"
    monkeypatch.setattr(LocalSandbox, runner, staticmethod(fake_run_posix))
    monkeypatch.setattr(LocalSandbox, "_get_shell", staticmethod(lambda: "/bin/bash"))
    monkeypatch.setattr(local_sandbox.os, "environ", {"PATH": "/usr/bin", "OPENAI_API_KEY": "sk-leak"})

    LocalSandbox("local:t").execute_command("echo hi")

    env = captured["env"]
    # Platform credentials are scrubbed (#3861) — never inherited by skills.
    assert "OPENAI_API_KEY" not in env
    # Benign vars survive.
    assert env["PATH"] == "/usr/bin"


def test_aio_sandbox_env_routes_through_bash_exec() -> None:
    """Per-call ``env`` is forwarded to the ``bash.exec`` API (structured env
    field on a fresh session) so secrets like ``GITHUB_TOKEN`` reach the
    command without being spliced into the command string. Replaces the old
    persistent-shell ``export … unset`` overlay, which could not keep secrets
    out of the command string.
    """
    from deerflow.community.aio_sandbox.aio_sandbox import AioSandbox

    captured: dict = {}

    class _FakeBash:
        def create_session(self, *, session_id):
            captured["created_session"] = session_id

        def exec(self, *, command, env=None, **kwargs):
            captured["command"] = command
            captured["env"] = env
            captured["exec_session"] = kwargs["session_id"]
            return SimpleNamespace(data=SimpleNamespace(stdout="ok", stderr=None))

        def close_session(self, session_id, **kwargs):
            captured["closed_session"] = session_id

    sbx = AioSandbox.__new__(AioSandbox)
    sbx._lock = __import__("threading").Lock()
    sbx._client = SimpleNamespace(bash=_FakeBash())
    sbx._DEFAULT_NO_CHANGE_TIMEOUT = 30
    sbx._DEFAULT_HARD_TIMEOUT = 30
    sbx._bash_exec_unsupported = False

    out = sbx.execute_command("gh pr create", env={"GH_TOKEN": "tok-123"})

    assert out == "ok"
    assert captured["command"] == "gh pr create"
    assert captured["env"] == {"GH_TOKEN": "tok-123"}
    assert captured["created_session"] == captured["exec_session"] == captured["closed_session"]


def test_aio_sandbox_no_env_leaves_command_unchanged() -> None:
    from deerflow.community.aio_sandbox.aio_sandbox import AioSandbox

    captured: dict = {}

    class _FakeData:
        output = "ok"

    class _FakeResult:
        data = _FakeData()

    class _FakeShell:
        def exec_command(self, *, command, no_change_timeout=None, **kwargs):
            captured["command"] = command
            return _FakeResult()

    sbx = AioSandbox.__new__(AioSandbox)
    sbx._lock = __import__("threading").Lock()
    sbx._client = SimpleNamespace(shell=_FakeShell())
    sbx._DEFAULT_NO_CHANGE_TIMEOUT = 30
    sbx._recovery_session_id = None
    sbx._default_shell_corrupted = False

    sbx.execute_command("echo hello")

    assert captured["command"] == "echo hello"


# ---------------------------------------------------------------------------
# extra_env key validation (regression pin for willem-bd #5)
# ---------------------------------------------------------------------------


@pytest.mark.parametrize(
    "bad_key",
    [
        # Shell metachar in key — the actual injection vector flagged by
        # the review (would render as `export X;rm -rf /;Y='v'; <cmd>`).
        "X;rm -rf /mnt/user-data;Y",
        "X`whoami`",
        "X$(id)",
        "X&Y",
        "X|Y",
        "X>Y",
        "X<Y",
        "X Y",  # space
        "X\tY",  # tab
        "X\nY",  # newline
        # Leading digit — not a valid POSIX env-var name even though it
        # contains no metacharacters; rejecting these too keeps the rule
        # simple and matches POSIX.
        "1FOO",
        # Empty / whitespace-only.
        "",
        "   ",
        # Non-str keys (a dict can have int keys at runtime).
        123,
    ],
)
def test_extra_env_rejects_invalid_keys(bad_key) -> None:
    """Regression pin for willem-bd's finding #5 on PR #3754.

    The abstract ``Sandbox.execute_command(env=...)`` contract validates
    keys against the POSIX env-var rule ``^[A-Za-z_][A-Za-z0-9_]*$``. Today
    no implementation splices a key into a shell string — the local sandbox
    merges them into ``subprocess.run(env=...)`` (no shell), the AIO sandbox
    forwards them via the ``bash.exec`` structured env field, and e2b
    forwards them as the SDK's ``envs``. The rule is defense-in-depth for
    the contract: future callers deriving a key from config / payload /
    user input fail fast with ``ValueError`` rather than producing a latent
    injection should a future implementation regress to splicing keys into
    a shell command string.

    The same rule applies to all implementations even though none currently
    route a key through a shell — the contract is what matters, not each
    implementation's current escaping rules.
    """
    from deerflow.sandbox.sandbox import _validate_extra_env

    with pytest.raises(ValueError, match="extra_env key"):
        _validate_extra_env({bad_key: "value"})


@pytest.mark.parametrize(
    "good_key",
    [
        "GH_TOKEN",
        "GITHUB_TOKEN",
        "_HIDDEN",
        "foo123",
        "MIXED_case_42",
        "X",
    ],
)
def test_extra_env_accepts_valid_keys(good_key: str) -> None:
    """POSIX env-var names round-trip cleanly."""
    from deerflow.sandbox.sandbox import _validate_extra_env

    # No exception => acceptance.
    _validate_extra_env({good_key: "any value with spaces and $metachars"})


def test_extra_env_none_and_empty_pass_through() -> None:
    """``None`` and empty dicts are the common case — must not raise."""
    from deerflow.sandbox.sandbox import _validate_extra_env

    _validate_extra_env(None)
    _validate_extra_env({})


def test_local_sandbox_rejects_invalid_env_key(monkeypatch: pytest.MonkeyPatch) -> None:
    """End-to-end: a bad key reaches the implementation's ``execute_command``
    and is rejected before any subprocess is spawned.
    """
    import deerflow.sandbox.local.local_sandbox as local_sandbox

    fake_popen_called = False

    def fake_popen(*args, **kwargs):
        nonlocal fake_popen_called
        fake_popen_called = True

    monkeypatch.setattr(local_sandbox.subprocess, "Popen", fake_popen)

    with pytest.raises(ValueError, match="extra_env key"):
        LocalSandbox("local:t").execute_command(
            "echo hi",
            env={"X;rm -rf /mnt/user-data;Y": "v"},
        )
    assert fake_popen_called is False, "subprocess.Popen must not run when key is invalid"


def test_aio_sandbox_rejects_invalid_env_key() -> None:
    """End-to-end on the AIO sandbox path — the injection vector flagged in
    the review never reaches the shell's ``exec_command``.
    """
    from deerflow.community.aio_sandbox.aio_sandbox import AioSandbox

    exec_called = False

    class _FakeShell:
        def exec_command(self, *, command, no_change_timeout=None, **kwargs):
            nonlocal exec_called
            exec_called = True
            return SimpleNamespace(data=SimpleNamespace(output="ok"))

    sbx = AioSandbox.__new__(AioSandbox)
    sbx._lock = __import__("threading").Lock()
    sbx._client = SimpleNamespace(shell=_FakeShell())
    sbx._DEFAULT_NO_CHANGE_TIMEOUT = 30

    with pytest.raises(ValueError, match="extra_env key"):
        sbx.execute_command(
            "echo hi",
            env={"X;rm -rf /mnt/user-data;Y": "v"},
        )
    assert exec_called is False, "shell.exec_command must not run when key is invalid"


# ---------------------------------------------------------------------------
# bash_tool token read-through
# ---------------------------------------------------------------------------


def test_github_env_from_runtime_returns_token_pair() -> None:
    runtime = SimpleNamespace(context={"github_token": "tok-abc"})
    env = _github_env_from_runtime(runtime)
    assert env == {"GH_TOKEN": "tok-abc", "GITHUB_TOKEN": "tok-abc"}


def test_github_env_from_runtime_resolves_provider_callable() -> None:
    """A callable in context["github_token"] is invoked per bash call.

    This is the refresh seam: long autonomous github runs that span past
    the 60-minute installation-token TTL need every bash invocation to
    re-ask the provider, which transparently re-mints via the app-side
    cache when the token's leeway tripped.
    """
    calls = {"n": 0}

    def _provider() -> str:
        calls["n"] += 1
        return f"tok-call-{calls['n']}"

    runtime = SimpleNamespace(context={"github_token": _provider})

    env_1 = _github_env_from_runtime(runtime)
    assert env_1 == {"GH_TOKEN": "tok-call-1", "GITHUB_TOKEN": "tok-call-1"}

    env_2 = _github_env_from_runtime(runtime)
    assert env_2 == {"GH_TOKEN": "tok-call-2", "GITHUB_TOKEN": "tok-call-2"}
    assert calls["n"] == 2  # called once per bash invocation


def test_github_env_from_runtime_returns_none_when_provider_raises() -> None:
    """A misbehaving provider must NOT crash the bash tool — it just falls
    back to the no-token path so the run can still execute read-only.
    """

    def _broken() -> str:
        raise RuntimeError("mint failed")

    runtime = SimpleNamespace(context={"github_token": _broken})
    assert _github_env_from_runtime(runtime) is None


def test_github_env_from_runtime_returns_none_when_provider_returns_empty() -> None:
    runtime = SimpleNamespace(context={"github_token": lambda: ""})
    assert _github_env_from_runtime(runtime) is None


def test_github_env_from_runtime_none_when_no_token() -> None:
    runtime = SimpleNamespace(context={"thread_id": "t1"})
    assert _github_env_from_runtime(runtime) is None


def test_github_env_from_runtime_none_when_empty() -> None:
    runtime = SimpleNamespace(context={"github_token": ""})
    assert _github_env_from_runtime(runtime) is None


def test_bash_tool_passes_token_as_env(monkeypatch: pytest.MonkeyPatch) -> None:
    """When runtime.context carries github_token, bash forwards it to the sandbox."""
    runtime = SimpleNamespace(
        state={"sandbox": {"sandbox_id": "aio:xyz"}},  # non-local -> simpler branch
        context={"thread_id": "t1", "github_token": "tok-from-manager"},
        config={},
    )

    captured: dict = {}

    class _Sandbox:
        def execute_command(self, command, env=None, timeout=None):
            captured["command"] = command
            captured["env"] = env
            return "done"

    monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: _Sandbox())
    monkeypatch.setattr("deerflow.sandbox.tools.ensure_thread_directories_exist", lambda runtime: None)

    result = bash_tool.func(runtime=runtime, description="push", command="git push")

    assert "done" in result
    assert captured["env"] == {"GH_TOKEN": "tok-from-manager", "GITHUB_TOKEN": "tok-from-manager"}


def test_bash_tool_no_env_without_token(monkeypatch: pytest.MonkeyPatch) -> None:
    runtime = SimpleNamespace(
        state={"sandbox": {"sandbox_id": "aio:xyz"}},
        context={"thread_id": "t1"},  # no github_token
        config={},
    )

    captured: dict = {}

    class _Sandbox:
        def execute_command(self, command, env=None, timeout=None):
            captured["env"] = env
            return "done"

    monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: _Sandbox())
    monkeypatch.setattr("deerflow.sandbox.tools.ensure_thread_directories_exist", lambda runtime: None)

    bash_tool.func(runtime=runtime, description="ls", command="ls")
    assert captured["env"] is None


def test_bash_tool_routes_subagent_command_to_its_shell_scope(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    runtime = SimpleNamespace(
        state={"sandbox": {"sandbox_id": "aio:xyz"}},
        context={
            "thread_id": "t1",
            "sandbox_command_scope_id": "subagent:task-1",
        },
        config={},
    )
    captured: dict = {}

    class _Sandbox:
        def execute_command(self, command, env=None, timeout=None):
            pytest.fail("subagent command must use its scoped shell session")

        def execute_command_in_scope(
            self,
            command,
            env=None,
            timeout=None,
            *,
            scope_id=None,
        ):
            captured.update(
                command=command,
                env=env,
                timeout=timeout,
                scope_id=scope_id,
            )
            return "done"

    monkeypatch.setattr(
        "deerflow.sandbox.tools.ensure_sandbox_initialized",
        lambda runtime: _Sandbox(),
    )
    monkeypatch.setattr(
        "deerflow.sandbox.tools.ensure_thread_directories_exist",
        lambda runtime: None,
    )

    result = bash_tool.func(runtime=runtime, description="ls", command="ls")

    assert result == "done"
    assert captured["scope_id"] == "subagent:task-1"
    assert captured["command"] == "cd /mnt/user-data/workspace; ls"


# ---------------------------------------------------------------------------
# ChannelManager._apply_channel_policy — the unified per-channel run-policy
# hook. The github channel registers (is_interactive=False,
# default_recursion_limit=250, credentials_provider=inject_github_credentials,
# requires_bound_identity=False) via CHANNEL_RUN_POLICY; this section
# exercises that one hook for github and the no-op path for unregistered
# channels (Slack, Telegram, …).
# ---------------------------------------------------------------------------


def _github_msg(installation_id: int | None = 140594274) -> InboundMessage:
    return InboundMessage(
        channel_name="github",
        chat_id="zhfeng/llm-gateway",
        user_id="zhfeng",
        text="a PR was opened",
        msg_type=InboundMessageType.CHAT,
        # topic_id pairs PR number with agent name to keep each agent on
        # its own deterministic thread; see app/gateway/github/dispatcher.py.
        topic_id="7:coding-llm-gateway",
        owner_user_id="default",
        metadata={
            "agent_name": "coding-llm-gateway",
            "github": {
                "repo": "zhfeng/llm-gateway",
                "number": 7,
                "installation_id": installation_id,
            },
            "preferred_thread_id": "uuid5-fixed",
        },
    )


def _new_manager() -> ChannelManager:
    bus = MessageBus()
    store = ChannelStore(path=Path("/tmp/nonexistent-store-test.json"))
    return ChannelManager(bus=bus, store=store)


@pytest.mark.asyncio
async def test_run_context_after_apply_channel_policy_is_json_serializable(monkeypatch: pytest.MonkeyPatch) -> None:
    """Regression pin: ``run_context`` survives the langgraph SDK's JSON encoder.

    The channel path calls ``client.runs.wait(thread_id, assistant_id,
    context=run_context, …)``. The SDK encodes the body with ``orjson``
    before sending it over HTTP. Anything in ``run_context`` that is not
    JSON-serializable (notably the previous closure-based token provider)
    raises ``TypeError: Type is not JSON serializable: function`` and the
    entire delivery fails. This test pins the contract so we never
    silently regress to shipping a closure again.
    """
    import json

    manager = _new_manager()
    mint = AsyncMock(return_value="ghs_installation_token")
    monkeypatch.setattr("app.gateway.github.app_auth.mint_installation_token", mint)

    run_context: dict = {"thread_id": "t1", "user_id": "u1"}
    await manager._apply_channel_policy(_github_msg(), run_context)

    # Will raise TypeError if anything in run_context is not JSON-serializable.
    encoded = json.dumps(run_context)
    assert '"github_token": "ghs_installation_token"' in encoded


@pytest.mark.asyncio
async def test_apply_channel_policy_degrades_on_mint_failure(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None:
    """A failed mint must not crash the run; the agent proceeds read-only.

    ``_apply_channel_policy`` catches credential-provider exceptions and
    logs a warning so a transient GitHub API outage or a misconfigured
    installation_id degrades to "no token injected" rather than dropping
    the delivery.
    """
    manager = _new_manager()

    async def boom(_installation_id):
        raise RuntimeError("GITHUB_APP_ID not set")

    monkeypatch.setattr("app.gateway.github.app_auth.mint_installation_token", boom)

    run_context: dict = {}
    with caplog.at_level("WARNING", logger="app.channels.manager"):
        await manager._apply_channel_policy(_github_msg(), run_context)

    # Must not crash, must not set a token, must warn.
    assert "github_token" not in run_context
    assert any("credentials_provider raised" in r.message for r in caplog.records)


@pytest.mark.asyncio
async def test_apply_channel_policy_skips_token_without_installation_id(monkeypatch: pytest.MonkeyPatch) -> None:
    """A binding with no ``installation_id`` runs without a minted token —
    no mint call, no ``github_token`` in run_context. The non-interactive
    flag still gets set because that is decided by the channel-policy
    entry, not by the credentials provider.
    """
    manager = _new_manager()
    mint = AsyncMock(return_value="should-not-be-called")
    monkeypatch.setattr("app.gateway.github.app_auth.mint_installation_token", mint)

    run_context: dict = {}
    await manager._apply_channel_policy(_github_msg(installation_id=None), run_context)

    mint.assert_not_awaited()
    assert "github_token" not in run_context
    # disable_clarification is set unconditionally for github (it is on
    # the policy entry, not the credentials provider).
    assert run_context["disable_clarification"] is True


# ---------------------------------------------------------------------------
# ChannelRunPolicy registration + _apply_channel_policy
# ---------------------------------------------------------------------------


def test_github_policy_is_registered_on_import() -> None:
    """Importing the gateway.github subpackage registers the run policy.

    The manager doesn't carry GitHub-specific branches anymore — the
    policy entry is the single source of truth. Tests, gateway
    bootstrap, and ad-hoc scripts all get the same registration via the
    same import side-effect.
    """
    # Importing app.gateway.github runs run_policy.register_policy() as
    # an import side-effect.
    import app.gateway.github  # noqa: F401
    from app.channels.run_policy import CHANNEL_RUN_POLICY

    policy = CHANNEL_RUN_POLICY.get("github")
    assert policy is not None
    assert policy.is_interactive is False
    assert policy.default_recursion_limit == 250
    assert policy.credentials_provider is not None


@pytest.mark.asyncio
async def test_apply_channel_policy_installs_token_for_github(monkeypatch: pytest.MonkeyPatch) -> None:
    """The unified policy hook installs the github credentials and the
    non-interactive flag — both in one call.

    The token in ``run_context`` is the minted **string**, not a closure,
    so the value survives the langgraph SDK's JSON encoder on its way to
    ``client.runs.wait(context=…)``.
    """
    manager = _new_manager()
    mint = AsyncMock(return_value="ghs_unified")
    monkeypatch.setattr("app.gateway.github.app_auth.mint_installation_token", mint)

    run_context: dict = {}
    await manager._apply_channel_policy(_github_msg(), run_context)

    mint.assert_awaited_once_with(140594274)
    assert run_context["github_token"] == "ghs_unified"
    # Non-interactive flag is set in the same call — one method, one place.
    assert run_context["disable_clarification"] is True


@pytest.mark.asyncio
async def test_apply_channel_policy_is_noop_for_unregistered_channels(monkeypatch: pytest.MonkeyPatch) -> None:
    """Slack/Telegram/etc. have no entry in CHANNEL_RUN_POLICY and stay untouched."""
    manager = _new_manager()
    mint = AsyncMock(return_value="should-not-be-called")
    monkeypatch.setattr("app.gateway.github.app_auth.mint_installation_token", mint)

    msg = InboundMessage(channel_name="slack", chat_id="C1", user_id="u", text="hi", metadata={})
    run_context: dict = {}
    await manager._apply_channel_policy(msg, run_context)

    mint.assert_not_awaited()
    assert "github_token" not in run_context
    assert "disable_clarification" not in run_context


# ---------------------------------------------------------------------------
# _create_thread honors preferred_thread_id
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_create_thread_uses_preferred_thread_id() -> None:
    manager = _new_manager()
    msg = _github_msg()

    created_kwargs: dict = {}

    class _FakeClient:
        class threads:
            @staticmethod
            async def create(**kwargs):
                created_kwargs.update(kwargs)
                return {"thread_id": "uuid5-fixed"}

    with patch.object(manager, "_store_thread_id", new=AsyncMock()):
        thread_id = await manager._create_thread(_FakeClient(), msg)

    assert thread_id == "uuid5-fixed"
    assert created_kwargs["thread_id"] == "uuid5-fixed"
    assert "metadata" in created_kwargs


@pytest.mark.asyncio
async def test_create_thread_without_preferred_id_omits_thread_id_kwarg() -> None:
    manager = _new_manager()
    msg = InboundMessage(
        channel_name="slack",
        chat_id="C1",
        user_id="u",
        text="hi",
        metadata={},  # no preferred_thread_id
    )

    created_kwargs: dict = {}

    class _FakeClient:
        class threads:
            @staticmethod
            async def create(**kwargs):
                created_kwargs.update(kwargs)
                return {"thread_id": "random-from-gateway"}

    with patch.object(manager, "_store_thread_id", new=AsyncMock()):
        await manager._create_thread(_FakeClient(), msg)

    assert "thread_id" not in created_kwargs


@pytest.mark.asyncio
async def test_create_thread_handles_race_on_preferred_id() -> None:
    """Two concurrent deliveries for the same (repo, number) collide on the
    deterministic thread id. The losing writer hits a 409 ConflictError
    from the underlying thread_store; we verify the thread actually exists
    and reuse the deterministic id rather than dropping the run.
    """
    manager = _new_manager()
    msg = _github_msg()  # carries preferred_thread_id="uuid5-fixed"

    class _FakeClient:
        class threads:
            @staticmethod
            async def create(**kwargs):
                raise _make_conflict_error()

            @staticmethod
            async def get(thread_id, **kwargs):
                # The winning concurrent create succeeded; threads.get
                # confirms the row is there before we cache the mapping.
                return {"thread_id": thread_id}

    stored: dict = {}

    async def _fake_store(_msg, thread_id):
        stored["thread_id"] = thread_id

    with patch.object(manager, "_store_thread_id", new=_fake_store):
        thread_id = await manager._create_thread(_FakeClient(), msg)

    # Recovered: returns the deterministic id and persisted the mapping.
    assert thread_id == "uuid5-fixed"
    assert stored["thread_id"] == "uuid5-fixed"


@pytest.mark.asyncio
async def test_create_thread_non_conflict_failure_propagates_and_does_not_poison_store() -> None:
    """Regression pin for willem-bd #1 on PR #3754.

    A transient DB/network failure on threads.create (anything other than a
    409 ConflictError) must propagate cleanly. Previously this branch
    swallowed bare Exception and wrote ``preferred_thread_id`` into the
    store, mapping every subsequent webhook for the same (repo, PR) to a
    thread that never existed — runs.create then 404'd forever with no
    retry path.

    The narrow ``except ConflictError`` lets non-conflict failures surface
    so the caller fails the delivery cleanly and the store stays clean.
    """
    manager = _new_manager()
    msg = _github_msg()  # carries preferred_thread_id="uuid5-fixed"

    class _FakeClient:
        class threads:
            @staticmethod
            async def create(**kwargs):
                # Anything that is NOT ConflictError: connection error, 500
                # from the underlying store, JSON decode error, etc.
                raise RuntimeError("HTTP 500: Failed to create thread")

            @staticmethod
            async def get(thread_id, **kwargs):
                # Should never be called on the non-conflict path.
                raise AssertionError("threads.get must not be called on non-conflict failure")

    store_calls: list = []

    async def _fake_store(_msg, thread_id):
        store_calls.append(thread_id)

    with patch.object(manager, "_store_thread_id", new=_fake_store):
        with pytest.raises(RuntimeError, match="500"):
            await manager._create_thread(_FakeClient(), msg)

    # The mapping must NOT have been written — that was the bug.
    assert store_calls == []


@pytest.mark.asyncio
async def test_create_thread_conflict_with_get_failure_propagates_and_does_not_poison_store() -> None:
    """If ConflictError fires but the follow-up threads.get also fails, the
    store underneath is in an inconsistent state. Surfacing the failure is
    better than caching a mapping to a thread that may not exist — every
    future delivery on this issue/PR would 404 forever.
    """
    manager = _new_manager()
    msg = _github_msg()

    class _FakeClient:
        class threads:
            @staticmethod
            async def create(**kwargs):
                raise _make_conflict_error()

            @staticmethod
            async def get(thread_id, **kwargs):
                # Conflict was reported but the thread is not actually there.
                raise RuntimeError("HTTP 404: thread not found")

    store_calls: list = []

    async def _fake_store(_msg, thread_id):
        store_calls.append(thread_id)

    with patch.object(manager, "_store_thread_id", new=_fake_store):
        with pytest.raises(RuntimeError, match="404"):
            await manager._create_thread(_FakeClient(), msg)

    # No mapping was cached — that's the whole point of the verify step.
    assert store_calls == []


@pytest.mark.asyncio
async def test_create_thread_without_preferred_id_propagates_error() -> None:
    """Without a deterministic id we have no recovery anchor — the original
    exception must surface so the dispatch loop can handle/report it.
    """
    manager = _new_manager()
    msg = InboundMessage(
        channel_name="slack",
        chat_id="C1",
        user_id="u",
        text="hi",
        metadata={},  # no preferred_thread_id
    )

    class _FakeClient:
        class threads:
            @staticmethod
            async def create(**kwargs):
                raise RuntimeError("HTTP 500: Failed to create thread")

    with patch.object(manager, "_store_thread_id", new=AsyncMock()):
        with pytest.raises(RuntimeError, match="500"):
            await manager._create_thread(_FakeClient(), msg)