daemon.py

"""SYNAPSE Agent Daemon — Sprint 3 Spike 2 Phase 1 scaffolding.

The daemon is the in-process host of the Claude Agent SDK. It lives in
a background thread inside a graphical Houdini session and marshals
tool dispatches onto the main thread via
``synapse.host.main_thread_exec``. Phase 1 ships the lifecycle and the
full bootstrap chain (boot gate, auth, policy, suppression) with an
**empty agent loop** — the thread starts, signals ready, and waits on
the cancel event. Phase 2 populates the loop with actual Agent SDK
turns and runs the Crucible protocol against it.

Boot chain (in order)
---------------------
1. **Boot gate** — ``hou.isUIAvailable()`` must return True. In
   headless / PDG / render-farm-worker contexts it returns False and
   the daemon refuses to boot. This is the Render Farm Fork Bomb
   guard: TOPS scheduling N hython subprocesses each booting an agent
   would N× the API cost and blow file locks.
2. **Auth** — ``synapse.host.auth.get_anthropic_api_key()``
   (hou.secure → env var). If neither source has a key, boot halts
   with an explicit error.
3. **Event loop policy (Windows only)** —
   ``asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy())``.
   Confirmed necessary in Spike 0: without it, httpx/anyio collides
   with the default ProactorEventLoop and raises
   ``APIConnectionError`` + ``NameError('closing_agens')``.
4. **User-site advisory** — ``sys.flags.no_user_site`` checked; warns
   if interpreter was started without ``-s`` / ``PYTHONNOUSERSITE=1``.
   The Spike 0 CP314/CP311 mismatch is unrecoverable from inside a
   running interpreter, so this is advisory — callers must set the
   env var at launch.
5. **Dispatcher** — built with the ``synapse_inspect_stage`` tool
   registered and a ``main_thread_executor`` that composes dialog
   suppression around ``main_thread_exec``. Per-tool-call suppression
   scope preserves the artist's own UI outside tool dispatches.
6. **Thread start** — daemon thread begins, signals ``started_event``,
   waits on ``cancel_event``. Phase 1 ends here; Phase 2 inserts the
   agent loop between signal-ready and cancel-wait.

Cancellation
------------
``cancel_event: threading.Event`` is the physical Stop button. The
agent loop (Phase 2) checks it before every API yield and before every
tool dispatch. Spike 2 Phase 1 exposes it; Phase 2 wires it into the
agent's cooperation points.
"""

from __future__ import annotations

import asyncio
import logging
import queue
import sys
import threading
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, Optional

from synapse.cognitive.agent_loop import (
    AgentTurnConfig,
    AgentTurnResult,
    STATUS_CANCELLED,
    STATUS_API_ERROR,
    run_turn,
)
from synapse.cognitive.dispatcher import Dispatcher
from synapse.cognitive.tools.inspect_stage import (
    INSPECT_STAGE_SCHEMA,
    inspect_stage,
)
from synapse.host.auth import get_anthropic_api_key
from synapse.host.dialog_suppression import suppress_modal_dialogs
from synapse.host.main_thread_executor import main_thread_exec
from synapse.host.turn_handle import TurnHandle, TurnNotComplete

logger = logging.getLogger(__name__)


DEFAULT_START_TIMEOUT_SECONDS: float = 5.0
"""Max time to wait for the daemon thread to signal started_event."""

DEFAULT_STOP_TIMEOUT_SECONDS: float = 10.0
"""Max time to wait for the daemon thread to exit after cancel."""

_REQUEST_POLL_INTERVAL_SECONDS: float = 0.25
"""How often the daemon thread checks its request queue AND the
cancel event. Short enough that ``cancel()`` → exit latency is
bounded; long enough that idle loops don't peg a core."""


@dataclass
class _AgentRequest:
    """Internal envelope for a submit_turn request queued to the daemon.

    Kept private — callers use ``SynapseDaemon.submit_turn``.

    Spike 2.4: ``handle`` replaces the old ``result_queue`` field. The
    daemon thread posts the result via ``handle._set_result(...)``;
    callers wait via ``handle.result(...)`` / ``handle.event.wait()``
    instead of blocking on a per-request ``queue.Queue``.
    """

    user_prompt: str
    config: AgentTurnConfig
    handle: TurnHandle
    submitted_at: float = field(default_factory=lambda: 0.0)


class DaemonBootError(RuntimeError):
    """Boot failed before the daemon thread started.

    Distinct from runtime errors inside the daemon thread so callers
    can route on type — boot failures are recoverable (fix and retry),
    runtime failures may leave mutable state in the thread that
    needs cleanup.
    """


class SynapseDaemon:
    """In-process agent daemon.

    Phase 1 contract:
      - ``.start()`` runs the full bootstrap chain synchronously on
        the caller's thread, then spins up the daemon thread which
        immediately idles on ``cancel_event.wait()``.
      - ``.cancel()`` sets ``cancel_event`` — the agent loop cooperates.
      - ``.stop(timeout)`` cancels and joins.
      - ``.dispatcher`` exposes the Dispatcher for direct invocation
        from tests and from Phase 2's agent loop.
      - ``.cancel_event`` exposes the Event for Phase 2 check-in points.

    Reusable: ``start()`` / ``stop()`` can cycle repeatedly. State is
    cleared at each start.
    """

    def __init__(
        self,
        *,
        api_key: Optional[str] = None,
        thread_name: str = "synapse.host.daemon",
        boot_gate: bool = True,
        main_thread_executor: Optional[
            Callable[[Callable, Dict[str, Any]], Dict[str, Any]]
        ] = None,
        anthropic_client_factory: Optional[Callable[[str], Any]] = None,
    ) -> None:
        """Construct the daemon (boot is deferred to ``start()``).

        Args:
            api_key: Optional override for the resolved API key. When
                ``None``, boot resolves via hou.secure → env var.
                Passing a value bypasses resolution — useful for tests
                and for explicit-credential launches.
            thread_name: Name assigned to the daemon ``threading.Thread``.
                Appears in debugger / thread inspection.
            boot_gate: When True (default), boot refuses unless
                ``hou.isUIAvailable()``. Tests and direct-hython
                experiments can pass ``False`` to skip.
            main_thread_executor: Optional override for the executor
                the Dispatcher uses. Default composes
                ``suppress_modal_dialogs()`` around
                ``main_thread_exec``. Tests can inject a pure-Python
                stub to exercise the daemon outside Houdini.
            anthropic_client_factory: Optional callable that accepts
                the resolved API key and returns a client object with
                a ``messages.create(...)`` method. Default imports
                ``anthropic.Anthropic`` at boot time. Tests inject a
                mock so no real API calls are made.
        """
        self._api_key_override = api_key
        self._thread_name = thread_name
        self._boot_gate_enabled = boot_gate
        self._custom_executor = main_thread_executor
        self._client_factory = anthropic_client_factory

        self._thread: Optional[threading.Thread] = None
        self._dispatcher: Optional[Dispatcher] = None
        self._anthropic_client: Optional[Any] = None
        self._request_queue: "queue.Queue[_AgentRequest]" = queue.Queue()
        self._cancel_event = threading.Event()
        self._started_event = threading.Event()
        self._exit_event = threading.Event()
        self._thread_error: Optional[BaseException] = None
        self._resolved_api_key: Optional[str] = None

    # -- Public surface --------------------------------------------------

    @property
    def cancel_event(self) -> threading.Event:
        """The cooperative cancel flag. Phase 2 checks it at yield points."""
        return self._cancel_event

    @property
    def started_event(self) -> threading.Event:
        """Set once the daemon thread is running. Useful for test sync."""
        return self._started_event

    @property
    def exit_event(self) -> threading.Event:
        """Set once the daemon thread has exited, success or failure."""
        return self._exit_event

    @property
    def dispatcher(self) -> Dispatcher:
        """The Dispatcher this daemon uses. Available after ``start()``.

        Raises:
            DaemonBootError: daemon has not been started yet.
        """
        if self._dispatcher is None:
            raise DaemonBootError(
                "Dispatcher not available — daemon has not been started"
            )
        return self._dispatcher

    @property
    def is_running(self) -> bool:
        return self._thread is not None and self._thread.is_alive()

    @property
    def thread_error(self) -> Optional[BaseException]:
        """Exception the daemon thread raised (if any). None on clean exit."""
        return self._thread_error

    def start(self) -> None:
        """Run the boot chain and start the daemon thread.

        Raises:
            DaemonBootError: any boot-chain step failed. The daemon
                thread is not started; no cleanup required.
        """
        if self.is_running:
            raise DaemonBootError("Daemon already running; stop() it first")

        self._check_boot_gate()
        self._resolved_api_key = self._resolve_api_key()
        self._apply_event_loop_policy()
        self._warn_if_user_site_active()
        self._dispatcher = self._build_dispatcher()
        self._anthropic_client = self._build_anthropic_client()

        # Reset signals for a fresh run. The request queue is drained
        # so any stale submit_turn()s from a prior cycle don't fire.
        self._drain_request_queue()
        self._cancel_event.clear()
        self._started_event.clear()
        self._exit_event.clear()
        self._thread_error = None

        self._thread = threading.Thread(
            target=self._thread_main,
            name=self._thread_name,
            daemon=True,
        )
        self._thread.start()

        if not self._started_event.wait(timeout=DEFAULT_START_TIMEOUT_SECONDS):
            raise DaemonBootError(
                f"Daemon thread did not signal started within "
                f"{DEFAULT_START_TIMEOUT_SECONDS:g}s"
            )
        logger.info("SynapseDaemon started (thread: %r)", self._thread_name)

    def cancel(self) -> None:
        """Signal cooperative cancel. Idempotent.

        Safe to call from any thread. The daemon thread's cooperation
        points (Phase 2) pick it up within one yield-point of being set.
        """
        self._cancel_event.set()

    def stop(self, timeout: float = DEFAULT_STOP_TIMEOUT_SECONDS) -> None:
        """Cancel and join the daemon thread, then drain pending requests.

        Args:
            timeout: Join timeout in seconds. If the thread does not
                exit within this budget, a warning is logged and the
                method returns — Python cannot force-kill a thread.

        The drain runs after the join so we don't race the daemon
        thread's own consumption of the queue. Any request that was
        queued but never popped has its handle cancelled, so callers
        blocked in ``handle.result(...)`` see ``TurnCancelled`` rather
        than hanging forever (design § 4.6).
        """
        self.cancel()
        if self._thread is not None and self._thread.is_alive():
            self._thread.join(timeout=timeout)
            if self._thread.is_alive():
                logger.warning(
                    "Daemon thread did not exit within %.1fs of stop()",
                    timeout,
                )
        self._thread = None
        self._drain_request_queue()

    # -- Boot chain steps ------------------------------------------------

    def _check_boot_gate(self) -> None:
        """Refuse boot unless ``hou.isUIAvailable()`` returns True.

        Render Farm Fork Bomb guard. PDG / TOPS workers spawn N hython
        subprocesses; each booting an agent multiplies API cost and
        produces file-lock contention. Headless hython returns False
        from ``hou.isUIAvailable()``; graphical Houdini returns True.
        """
        if not self._boot_gate_enabled:
            logger.debug("Boot gate bypassed (boot_gate=False)")
            return

        try:
            import hou  # type: ignore[import-not-found]
        except ImportError as exc:
            raise DaemonBootError(
                "SynapseDaemon requires a running Houdini process. "
                "Pass boot_gate=False only if you know why."
            ) from exc

        is_ui_available = getattr(hou, "isUIAvailable", None)
        if is_ui_available is None:
            raise DaemonBootError(
                "hou.isUIAvailable is missing from this Houdini build — "
                "cannot verify the boot gate. Refusing to boot."
            )

        if not is_ui_available():
            raise DaemonBootError(
                "hou.isUIAvailable() returned False. SynapseDaemon will "
                "not boot in headless / PDG contexts (Render Farm Fork "
                "Bomb prevention). Pass boot_gate=False to override."
            )

    def _resolve_api_key(self) -> str:
        """Explicit override > hou.secure > env var.

        Raises:
            DaemonBootError: no source produced a usable key.
        """
        if self._api_key_override is not None:
            key = self._api_key_override.strip()
            if not key:
                raise DaemonBootError(
                    "Explicit api_key override was empty / whitespace"
                )
            return key

        resolved = get_anthropic_api_key()
        if resolved is None:
            raise DaemonBootError(
                "No Anthropic API key available. Set one via "
                "hou.secure.setPassword('synapse_anthropic', 'sk-ant-...') "
                "or export ANTHROPIC_API_KEY=sk-ant-... before launching."
            )
        return resolved

    def _apply_event_loop_policy(self) -> None:
        """Spike 0 bootstrap lock: selector policy on Windows.

        Without it, httpx/anyio raises APIConnectionError + NameError
        ('closing_agens') during async shutdown on Python 3.11 Windows
        because the default ProactorEventLoop mishandles stream cleanup.
        """
        if sys.platform == "win32":
            policy = asyncio.WindowsSelectorEventLoopPolicy()
            asyncio.set_event_loop_policy(policy)
            logger.info(
                "Event loop policy set to WindowsSelectorEventLoopPolicy"
            )

    def _warn_if_user_site_active(self) -> None:
        """Spike 0 bootstrap lock (advisory): user site should be disabled.

        Cannot be enforced from inside a running interpreter — the flag
        is read at startup. If unset, warn; downstream wrong-ABI imports
        will fail loudly at that point.
        """
        if not getattr(sys.flags, "no_user_site", False):
            logger.warning(
                "User site-packages enabled (PYTHONNOUSERSITE not set at "
                "interpreter launch). Spike 0 found CP314 pydantic_core "
                "binaries leaking from user site into CP311 hython. Start "
                "hython with -s flag or set PYTHONNOUSERSITE=1 before "
                "launching to isolate."
            )

    def _build_dispatcher(self) -> Dispatcher:
        """Construct the Dispatcher with host-layer executor + suppression.

        The executor composes ``suppress_modal_dialogs()`` around
        ``main_thread_exec`` so every tool dispatch gets a clean,
        narrowly-scoped dialog-suppression window.
        """
        executor = self._custom_executor or _default_executor
        return Dispatcher(
            tools={"synapse_inspect_stage": inspect_stage},
            schemas={"synapse_inspect_stage": INSPECT_STAGE_SCHEMA},
            main_thread_executor=executor,
        )

    def _build_anthropic_client(self) -> Any:
        """Construct the Anthropic client for the agent loop.

        Default: imports ``anthropic.Anthropic`` lazily so stock-Python
        test paths don't pay the import cost. Tests inject
        ``anthropic_client_factory`` to bypass the real SDK entirely.

        Raises:
            DaemonBootError: anthropic SDK is not installed and no
                factory was provided. Actionable because Spike 0's
                install recipe for hython's site-packages is
                ``hython -m pip install anthropic`` (with the
                sys.path-stripping launcher documented there).
        """
        if self._client_factory is not None:
            return self._client_factory(self._resolved_api_key or "")

        try:
            from anthropic import Anthropic  # type: ignore[import-not-found]
        except ImportError as exc:
            raise DaemonBootError(
                "anthropic SDK is not installed in this Python. "
                "See Spike 0's install recipe for hython; for stock-"
                "Python tests pass anthropic_client_factory explicitly."
            ) from exc
        return Anthropic(api_key=self._resolved_api_key)

    def _drain_request_queue(self) -> None:
        """Empty the request queue and cancel any drained handles.

        Called on ``start()`` before the thread takes over, so stale
        requests from a prior cycle don't fire into the new agent
        loop, and on ``stop()`` so callers blocked in
        ``handle.result(...)`` see ``TurnCancelled`` instead of
        hanging forever (the Spike 2.4 revert path retained the
        ``queue.Empty`` surface, which left handles dangling).
        """
        try:
            while True:
                stale = self._request_queue.get_nowait()
                stale.handle.cancel()
        except queue.Empty:
            pass

    # -- submit_turn -----------------------------------------------------

    def submit_turn(
        self,
        user_prompt: str,
        *,
        system: str = "",
        model: Optional[str] = None,
        max_tokens: Optional[int] = None,
        max_iterations: Optional[int] = None,
    ) -> TurnHandle:
        """Submit a prompt to the agent loop and return a ``TurnHandle``
        immediately. Does NOT block.

        Spike 2.4: this is the non-blocking shape that closes the
        daemon ↔ main-thread deadlock. The caller polls
        ``handle.done()``, waits on ``handle.event``, or calls
        ``handle.result(timeout=...)`` from a thread that is **not**
        the Houdini main thread. Tool dispatches inside the turn
        continue to flow through the Dispatcher (and thus through the
        host-layer main-thread executor → ``hdefereval`` in GUI mode,
        direct exec in headless).

        The pre-2.4 ``wait_timeout`` parameter is removed — it
        conflated request submission with result waiting and was the
        direct vehicle for the deadlock. Callers that want a
        synchronous shape use ``submit_turn_blocking`` from a
        non-main thread, or layer their own poll on top of
        ``handle.done()``.

        Args:
            user_prompt: The user message kicking off the turn.
            system: Optional system prompt.
            model: Override the default model for this turn.
            max_tokens: Override per-response token ceiling.
            max_iterations: Override the cap on API round-trips.

        Returns:
            A ``TurnHandle`` that completes when the daemon finishes
            the turn. The handle's ``result()`` returns the
            ``AgentTurnResult``.

        Raises:
            DaemonBootError: daemon is not running.
        """
        if not self.is_running:
            raise DaemonBootError(
                "submit_turn requires a running daemon — call start() first."
            )

        config = AgentTurnConfig(
            model=model or AgentTurnConfig.model,
            max_tokens=max_tokens if max_tokens is not None
                       else AgentTurnConfig.max_tokens,
            max_iterations=(max_iterations if max_iterations is not None
                            else AgentTurnConfig.max_iterations),
            system=system,
        )
        handle = TurnHandle()
        request = _AgentRequest(
            user_prompt=user_prompt,
            config=config,
            handle=handle,
        )
        self._request_queue.put(request)
        return handle

    def submit_turn_blocking(
        self,
        user_prompt: str,
        *,
        system: str = "",
        model: Optional[str] = None,
        max_tokens: Optional[int] = None,
        max_iterations: Optional[int] = None,
        wait_timeout: Optional[float] = None,
    ) -> AgentTurnResult:
        """Synchronous shape, retained for tests and headless callers.

        Equivalent to::

            handle = self.submit_turn(...)
            return handle.result(timeout=wait_timeout)

        **DO NOT call from the Houdini main thread in GUI mode** —
        that re-introduces the Spike 2.4 deadlock. When
        ``hou.isUIAvailable()`` returns True AND the caller is on the
        main thread, this method raises ``RuntimeError`` rather than
        deadlock silently. The detection is best-effort (Python can't
        always tell), but the explicit raise steers callers off the
        cliff.

        Args:
            user_prompt: The user message kicking off the turn.
            system: Optional system prompt.
            model: Override the default model for this turn.
            max_tokens: Override per-response token ceiling.
            max_iterations: Override the cap on API round-trips.
            wait_timeout: How long to wait for the daemon thread to
                return a result. ``None`` = wait indefinitely. Tests
                should always pass a small finite timeout so stuck
                tests surface loudly.

        Returns:
            ``AgentTurnResult`` on normal completion.

        Raises:
            DaemonBootError: daemon is not running.
            RuntimeError: called from the Houdini main thread in GUI
                mode (would deadlock — see Spike 2.4 design).
            TurnNotComplete: ``wait_timeout`` elapsed without a
                result. Subclass of stdlib ``TimeoutError`` so
                ``except TimeoutError`` paths still catch it.
            TurnCancelled: handle was cancelled before completion
                (e.g. ``daemon.stop()`` drained the request queue).
        """
        self._guard_against_main_thread_blocking_call()

        handle = self.submit_turn(
            user_prompt,
            system=system,
            model=model,
            max_tokens=max_tokens,
            max_iterations=max_iterations,
        )
        try:
            return handle.result(timeout=wait_timeout)
        except TurnNotComplete as exc:
            # Re-raise with a message that keeps the prior
            # submit_turn timeout vocabulary so log scrapers don't
            # need updating. TurnNotComplete IS a TimeoutError, so
            # callers catching either still work.
            raise TurnNotComplete(
                f"submit_turn_blocking did not complete within {wait_timeout}s"
            ) from exc

    @staticmethod
    def _guard_against_main_thread_blocking_call() -> None:
        """Raise ``RuntimeError`` if invoked from the Houdini main
        thread in GUI mode.

        Best-effort detection — both checks are required:

        - ``hou.isUIAvailable()`` is True (we're in graphical Houdini,
          not headless / stock-Python).
        - ``threading.current_thread() is threading.main_thread()``.

        Either being False short-circuits the guard (headless mode
        has only one thread by design; non-main threads can block
        safely because ``hdefereval`` can drain its queue).
        """
        try:
            import hou  # type: ignore[import-not-found]
        except ImportError:
            return  # Stock Python — no main-thread coupling to Qt.

        is_ui_available = getattr(hou, "isUIAvailable", None)
        if is_ui_available is None or not is_ui_available():
            return  # Headless mode — single-threaded, safe to block.

        if threading.current_thread() is threading.main_thread():
            raise RuntimeError(
                "submit_turn_blocking was called from the Houdini main "
                "thread in GUI mode. This would deadlock — the main "
                "thread cannot block on a queue while the daemon "
                "needs it to drain hdefereval's main-thread queue "
                "(Spike 2.4). Use submit_turn(...) and poll the "
                "returned TurnHandle from a callback, or call "
                "submit_turn_blocking from a non-main thread."
            )

    # -- Thread entry point ---------------------------------------------

    def _thread_main(self) -> None:
        """Daemon loop body.

        Phase 2 (Spike 2.2) shape:
          1. Signal ``started_event`` so ``start()`` unblocks.
          2. Loop: pull requests from ``_request_queue`` with a short
             timeout so cancel latency is bounded. Each request is
             a ``_AgentRequest`` carrying a user prompt, config, and
             a reply queue.
          3. For each request: run ``synapse.cognitive.agent_loop.run_turn``
             against the dispatcher + cancel event. Post the result
             back on the request's reply queue.
          4. Exit loop when ``cancel_event`` fires.

        Cancel semantics:
          - Between requests: cancel cuts the loop immediately.
          - During a request: the agent_loop itself checks
            ``cancel_event`` at every API yield and every tool
            dispatch and returns STATUS_CANCELLED. The daemon still
            posts that result back so callers aren't left waiting.
        """
        try:
            self._started_event.set()
            logger.info("Daemon thread entering request loop")
            while not self._cancel_event.is_set():
                try:
                    request = self._request_queue.get(
                        timeout=_REQUEST_POLL_INTERVAL_SECONDS,
                    )
                except queue.Empty:
                    continue
                self._process_request(request)
            logger.info("Daemon thread received cancel — exiting loop")
        except BaseException as exc:  # noqa: BLE001 — capture everything
            self._thread_error = exc
            logger.exception("Daemon thread raised")
        finally:
            self._exit_event.set()

    def _process_request(self, request: _AgentRequest) -> None:
        """Run one agent turn against the configured client + dispatcher.

        Never lets a turn failure take down the daemon thread. API
        errors and unexpected exceptions come back via
        ``request.handle._set_result`` as ``AgentTurnResult`` values
        (with status ``STATUS_API_ERROR`` when the loop caught it, or
        a synthesized result when the failure escaped the loop
        itself).

        Idempotency: if the handle is already cancelled (caller
        invoked ``handle.cancel()`` while we were running), the
        ``_set_result`` post is logged at debug and dropped. The
        daemon does not raise.
        """
        assert self._dispatcher is not None
        assert self._anthropic_client is not None
        try:
            result = run_turn(
                self._anthropic_client,
                self._dispatcher,
                request.user_prompt,
                cancel_event=self._cancel_event,
                config=request.config,
            )
        except BaseException as exc:  # noqa: BLE001 — protect daemon thread
            logger.exception("run_turn raised outside its own guard")
            result = AgentTurnResult(
                status=STATUS_API_ERROR,
                error=f"{type(exc).__name__}: {exc}",
            )
        # Set the handle result with the (possibly synthesized) failure
        # so callers that polled .result() see the AgentTurnResult
        # shape they expect. _set_exception is reserved for daemon-
        # internal failures that don't fit AgentTurnResult.
        request.handle._set_result(result)


def _default_executor(
    fn: Callable[..., Dict[str, Any]],
    kwargs: Dict[str, Any],
) -> Dict[str, Any]:
    """Production executor: dialog suppression + main-thread marshal.

    Every tool dispatch through this executor gets its own
    suppression window that opens on enter and closes before return —
    the artist's own UI work outside this narrow interval is untouched.
    """
    with suppress_modal_dialogs():
        return main_thread_exec(fn, kwargs)