deer-flow/backend/app/gateway/github/registry.py

"""Build the GitHub webhook → agent registry.

Walks every user's custom-agent directory under ``{base_dir}/users/`` plus
the legacy shared layout at ``{base_dir}/agents/`` and indexes every agent
that declares a ``github:`` block by the ``(repo, event)`` pairs it
declares an interest in.

The dispatcher calls :func:`build_github_agent_registry` once per webhook
delivery. We avoid re-parsing every ``config.yaml`` on each call via a
small mtime-keyed cache: the directory listing + ``stat()`` per config
file is cheap (~µs), while ``yaml.safe_load`` is the dominant cost
(~hundreds of µs per file). The cache key is the sorted tuple of
``(user_id, agent_name, config.yaml mtime)`` triples; any mtime change,
addition, or deletion invalidates the cache transparently. Operators
who hand-edit ``config.yaml`` see the change on the next webhook.

Cache invalidation caveat: mtime granularity on macOS HFS+ / APFS is
~1 µs but on some filesystems (FAT, network shares with caching) it's
1 s. Two edits inside the same coarse-tick would look identical. For
the dispatch path that's fine — webhooks are rare relative to operator
edits, and the next non-coincident write reconciles. If we ever land
operator tooling that batches sub-second edits, we can extend the
signature with file size or a content hash.
"""

from __future__ import annotations

import logging
import threading
from dataclasses import dataclass
from pathlib import Path

from app.gateway.github.triggers import _resolved_trigger
from deerflow.config.agents_config import (
    AgentConfig,
    GitHubTriggerConfig,
    load_agent_config,
)
from deerflow.config.paths import get_paths
from deerflow.runtime.user_context import DEFAULT_USER_ID

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class GitHubAgentMatch:
    """One ``(user, agent, _resolved_trigger)`` row in the ``(repo, event)`` index.

    The trigger is the binding override merged with per-event field defaults
    (see :func:`app.gateway.github.triggers._resolved_trigger`), so the
    dispatcher does not have to re-resolve it at fan-out time. Pre-resolving
    here also folds the per-binding lookup out of the hot path: the registry
    already chose the right binding for this ``(repo, event)``, so the
    dispatcher's old "find the binding whose ``.repo`` matches" loop —
    which silently dropped events when an agent had multiple bindings on
    one repo (PR feedback R3) — disappears entirely. Single-binding-per-repo
    is enforced upstream by :class:`GitHubAgentConfig`'s validator, so
    each ``(repo, event)`` resolves to exactly one trigger per agent.

    The ``github:`` block is read off ``agent.github`` (always non-None
    here — the rebuild filters agents without one before constructing a
    match), so we don't carry a separate ``github`` field.
    """

    user_id: str
    agent: AgentConfig
    trigger: GitHubTriggerConfig


# Cache: (signature, registry). ``signature`` is a tuple of
# ``(user_id, agent_name, mtime)`` triples. Identical signature → registry
# is still valid, skip the YAML parses.
_Signature = tuple[tuple[str, str, float], ...]
_Registry = dict[tuple[str, str], list[GitHubAgentMatch]]
_cache: tuple[_Signature, _Registry] | None = None
# Threading lock (not asyncio): build_github_agent_registry is invoked
# from asyncio.to_thread in the dispatcher, so the lock is acquired on
# the worker thread. A plain Lock is the right primitive here.
_cache_lock = threading.Lock()


def _discover_user_ids() -> list[str]:
    """Return all user-id directories under ``base_dir/users/``.

    Falls back to ``[DEFAULT_USER_ID]`` so the no-auth dev setup (which
    keeps everything in ``users/default/``) is always covered even before
    the directory has been created on disk.
    """
    paths = get_paths()
    users_dir: Path = paths.base_dir / "users"
    if not users_dir.exists():
        return [DEFAULT_USER_ID]

    found: list[str] = []
    for entry in sorted(users_dir.iterdir()):
        if entry.is_dir() and (entry / "agents").exists():
            found.append(entry.name)
    if DEFAULT_USER_ID not in found:
        found.append(DEFAULT_USER_ID)
    return found


def _gather_agent_signature() -> tuple[_Signature, list[tuple[str, str]]]:
    """Return (signature, [(user_id, agent_name)]) for every agent on disk.

    The signature lets us skip the YAML parse on warm hits; the
    discovered list lets the rebuilder process exactly the agents that
    the signature covers. Doing iterdir + stat is cheap (~µs each); the
    full cost we avoid is the ``yaml.safe_load`` per config.

    Includes the legacy shared layout at ``{base_dir}/agents/`` under the
    :data:`DEFAULT_USER_ID` bucket so unmigrated installations still
    receive webhook fan-out. Per-user entries shadow legacy entries with
    the same name (matching :func:`list_custom_agents`' precedence), so
    once an install runs ``migrate_user_isolation.py`` the legacy entry
    is silently superseded rather than producing duplicate rows.
    """
    paths = get_paths()
    sig: list[tuple[str, str, float]] = []
    discovered: list[tuple[str, str]] = []
    seen: set[tuple[str, str]] = set()
    for user_id in _discover_user_ids():
        agent_root = paths.user_agents_dir(user_id)
        if not agent_root.exists():
            continue
        for entry in sorted(agent_root.iterdir()):
            config = entry / "config.yaml"
            if not entry.is_dir() or not config.exists():
                continue
            try:
                mtime = config.stat().st_mtime
            except OSError:
                # Vanished between iterdir and stat — racing operator
                # edit. Drop from this round; next call picks it up.
                continue
            sig.append((user_id, entry.name, mtime))
            discovered.append((user_id, entry.name))
            seen.add((user_id, entry.name))

    # Legacy shared layout: {base_dir}/agents/{name}/. CLAUDE.md commits
    # to this as a read-only fallback for unmigrated installs, and
    # load_agent_config() / list_custom_agents() honour it — the webhook
    # path must too, or an unmigrated install with a ``github:`` block on
    # a shared agent silently fans out to nothing. Legacy entries map
    # onto the DEFAULT_USER_ID bucket because that is the user-id
    # ``load_agent_config(name)`` resolves them under at run-time.
    legacy_root = paths.agents_dir
    if legacy_root.exists():
        for entry in sorted(legacy_root.iterdir()):
            config = entry / "config.yaml"
            if not entry.is_dir() or not config.exists():
                continue
            # Per-user shadow: if users/default/agents/{name} already
            # exists, skip the legacy entry so we don't index the same
            # agent twice with conflicting trigger sets.
            if (DEFAULT_USER_ID, entry.name) in seen:
                continue
            try:
                mtime = config.stat().st_mtime
            except OSError:
                continue
            sig.append((DEFAULT_USER_ID, entry.name, mtime))
            discovered.append((DEFAULT_USER_ID, entry.name))
            seen.add((DEFAULT_USER_ID, entry.name))
    return tuple(sig), discovered


def _rebuild(discovered: list[tuple[str, str]]) -> _Registry:
    """Parse every agent's config.yaml and build the (repo, event) index.

    Each ``(repo, event)`` slot stores :class:`GitHubAgentMatch` rows — the
    user_id + AgentConfig + the trigger already resolved (binding override
    merged with per-event field defaults). The dispatcher then only needs
    to apply the trigger; it never re-walks ``bindings`` to find the right
    one. Single-binding-per-repo is enforced by
    :class:`GitHubAgentConfig`'s validator, so a duplicate-repo config
    fails to load here (logged as a skip) instead of producing duplicate
    rows in this index.
    """
    index: _Registry = {}
    for user_id, agent_name in discovered:
        try:
            cfg = load_agent_config(agent_name, user_id=user_id)
        except Exception as exc:  # noqa: BLE001 — one bad agent must not kill the scan
            logger.warning("github_registry: skipping agent %s/%s: %s", user_id, agent_name, exc)
            continue
        if cfg is None or cfg.github is None:
            continue
        for binding in cfg.github.bindings:
            for event, override in binding.triggers.items():
                resolved = _resolved_trigger(event, {event: override})
                if resolved is None:
                    # ``_resolved_trigger`` only returns None when the
                    # event is not in the dict we passed — by construction
                    # it is here, so this branch is unreachable. Keep the
                    # guard for type-checker happiness.
                    continue
                index.setdefault((binding.repo, event), []).append(GitHubAgentMatch(user_id=user_id, agent=cfg, trigger=resolved))
    return index


def build_github_agent_registry() -> _Registry:
    """Return ``{(repo, event): [GitHubAgentMatch, ...]}`` across all users.

    Each agent appears in the index once per ``(repo, declared_event)`` pair,
    with the per-event trigger pre-resolved by merging the binding override
    with :data:`app.gateway.github.triggers.DEFAULT_TRIGGERS`. Events are
    opt-in per binding: an agent only registers for the events it explicitly
    lists under ``github.bindings[].triggers``. An agent that declares an
    empty ``triggers:`` map (or omits it) registers for nothing and the
    dispatcher will never fan a webhook out to it.

    Warm path (no agents added/removed/edited since the last call) costs
    only the iterdir + stat pass — no YAML parsing. Cold path parses
    every config.yaml and refreshes the cache. The result is shared
    across callers (returned by reference) since :class:`GitHubAgentMatch`
    is frozen and the registry is intended as read-only.
    """
    global _cache
    with _cache_lock:
        signature, discovered = _gather_agent_signature()
        if _cache is not None and _cache[0] == signature:
            return _cache[1]
        registry = _rebuild(discovered)
        _cache = (signature, registry)
        return registry


def _invalidate_cache() -> None:
    """Drop the cached registry. Test-only helper."""
    global _cache
    with _cache_lock:
        _cache = None


def lookup_agents(
    registry: _Registry,
    repo: str,
    event: str,
) -> list[GitHubAgentMatch]:
    """Convenience: return the list of agent matches for ``(repo, event)``.

    Each match carries the user, AgentConfig (with ``.github`` attached),
    and the pre-resolved trigger config for this specific event, so the
    caller does not need to walk the agent's ``bindings`` again.
    """
    return registry.get((repo, event), [])