diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 566baca45fb..0f0e328c827 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -975,6 +975,7 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_ **GitHub event-driven agents** (webhook-driven IM channel): - Custom agents declare a `github:` block in their `config.yaml` to bind to repos and event triggers; the webhook route is fail-closed by default (mounted only when `GITHUB_WEBHOOK_SECRET` is set) and exempt from auth/CSRF because authenticity is enforced by HMAC. +- Registry caching is keyed by the configured agent store's opaque signature: file storage uses agent config mtimes, while database storage hashes the ordered owner/name/config/soul contents so same-timestamp writes still invalidate webhook routing. - Outbound is **log-only** by design: each agent posts its own reply mid-run via the `gh` CLI from its sandbox, so the manager uses `fire_and_forget=True` and `runs.create()` returns once pending. - **Follow-up buffering while busy** (issue #4121): because outbound is log-only, the pre-existing `THREAD_BUSY_MESSAGE` reply on a `ConflictError` was invisible to the commenter — a comment posted while a run was already active looked like it had been silently ignored. When `ChannelRunPolicy.buffer_followups_on_busy=True` (GitHub's default), a `ConflictError` on `runs.create()` now also appends the triggering message to a per-thread, in-memory buffer (`ChannelManager._followup_buffers`) — deduped by GitHub delivery id, capped at `FOLLOWUP_BUFFER_MAX_PER_THREAD` (20, oldest dropped with a WARNING log on overflow). The first successful `runs.create()` on a thread now captures its `run_id` and spawns a background watcher that subscribes to that run's `StreamBridge` stream; once it observes `END_SENTINEL`, the watcher drains up to `FOLLOWUP_DRAIN_BATCH_SIZE` (10) buffered entries into one ``-wrapped input and fires a follow-up `runs.create()` — itself watched the same way, so a backlog deeper than one batch chains into further drain cycles instead of growing one unbounded input. If that follow-up `runs.create()` itself hits `ConflictError` (e.g. a manual Web UI turn or a scheduled run raced onto the same thread), the batch is requeued rather than lost, and is retried whenever this manager next successfully creates and watches a run on that thread. Reactions/acknowledgment (e.g. GitHub's `eyes`/`confused` reaction API) on buffered comments are intentionally **out of scope** for this mechanism and left to a follow-up — comments are coalesced silently. **Plumbing**: the watcher needs the Gateway's `StreamBridge` singleton, which `ChannelManager` did not previously have access to; it is threaded from `app.py`'s lifespan (where `app.state.stream_bridge` is already set by `langgraph_runtime`) through `start_channel_service(get_stream_bridge=...)` → `ChannelService.__init__` → `ChannelManager.__init__`, as a zero-arg closure mirroring the existing `launch_run=lambda **kwargs: launch_scheduled_thread_run(app=app, **kwargs)` pattern used for `ScheduledTaskService` in the same lifespan function. A `ChannelManager` constructed without it (e.g. directly in a test) still buffers safely — it just has no watcher to auto-drain. **Scope limitation**: the buffer and watcher state are per-process, in-memory. Under `GATEWAY_WORKERS>1` or multi-pod, a follow-up comment routed to a different worker process than the one running the busy thread's agent will not see that buffer. This is a known, deliberately deferred limitation with the same shape as the cross-pod gap described for issue #4120 (a shared buffer store or IM-leader election would be needed to close it) — single-process/single-pod deployments, the safe default, see no correctness issue from this, only the documented per-process scope. - See [backend/docs/GITHUB_AGENTS.md](docs/GITHUB_AGENTS.md) for the architecture diagrams: webhook → fan-out → `InboundMessage` dispatch, `preferred_thread_id = UUID5(repo, number, agent_name)` thread determinism, mention-handle precedence chain, GH token lifecycle via `GH_TOKEN`/`GITHUB_TOKEN` per-call `extra_env`, and the narrow `ConflictError` (HTTP 409) thread-create race recovery. diff --git a/backend/app/gateway/github/registry.py b/backend/app/gateway/github/registry.py index 9ae68cc6aad..1dd718924f6 100644 --- a/backend/app/gateway/github/registry.py +++ b/backend/app/gateway/github/registry.py @@ -12,8 +12,9 @@ on the store's :meth:`~deerflow.persistence.agents.base.AgentStore.signature` change token: the file backend derives it from ``config.yaml`` mtimes (any edit, addition, or deletion invalidates the cache transparently); the db backend -derives it from ``max(updated_at)`` + the row count. Operators who hand-edit a -``config.yaml`` see the change on the next webhook. +derives it from a deterministic digest of each agent's owner, name, config, and +soul. Operators who hand-edit a ``config.yaml`` see the change on the next +webhook. Cache invalidation caveat (file backend): mtime granularity on macOS HFS+ / APFS is ~1 µs but on some filesystems (FAT, network shares with caching) it's diff --git a/backend/docs/GITHUB_AGENTS.md b/backend/docs/GITHUB_AGENTS.md index d46004cec29..6d65975d6f0 100644 --- a/backend/docs/GITHUB_AGENTS.md +++ b/backend/docs/GITHUB_AGENTS.md @@ -25,7 +25,7 @@ graph LR OperatorYaml["config.yaml
channels.github:
enabled: true
default_mention_login"]:::operator AgentYaml["agents/{name}/config.yaml
github:
installation_id
bot_login
bindings: [{repo, triggers}]"]:::agent - Registry["build_github_agent_registry()
(mtime-cached, asyncio.to_thread)"]:::route + Registry["build_github_agent_registry()
(store-signature cached, asyncio.to_thread)"]:::route Webhook["POST /api/webhooks/github
(HMAC verify)"]:::route OperatorYaml --> Registry @@ -33,6 +33,11 @@ graph LR Registry --> Webhook ``` +Registry cache invalidation uses the configured agent store's opaque signature. +The file backend derives it from agent config mtimes; the database backend uses +a deterministic digest of the stored owner, name, config, and soul, so an update +still invalidates the registry when two writes share the same timestamp. + Each agent binding lists the **events it cares about** under `triggers:`. Events absent from `triggers:` are not delivered to that agent — the dispatcher never loads the agent for them. `DEFAULT_TRIGGERS` only supplies **field-level defaults** (e.g. `require_mention: true`) for events a binding did declare; it is no longer an enablement list. ## Webhook → Fan-out → Dispatch @@ -300,4 +305,4 @@ This is also why the GitHub channel registers `ChannelRunPolicy.fire_and_forget= - `app/gateway/github/triggers.py` — `event_should_fire`, `DEFAULT_TRIGGERS` - `app/gateway/github/run_policy.py` — `inject_github_credentials`, `register_policy` - `app/gateway/routers/github_webhooks.py` — HMAC verify, route mount predicate -- `app/channels/github.py` — `GitHubChannel` (log-only outbound) \ No newline at end of file +- `app/channels/github.py` — `GitHubChannel` (log-only outbound) diff --git a/backend/packages/harness/deerflow/persistence/agents/base.py b/backend/packages/harness/deerflow/persistence/agents/base.py index 725dc765ff3..d0782c8725e 100644 --- a/backend/packages/harness/deerflow/persistence/agents/base.py +++ b/backend/packages/harness/deerflow/persistence/agents/base.py @@ -126,6 +126,6 @@ def signature(self) -> Hashable: Equal tokens mean "nothing changed since last read". The GitHub registry keys its cache off this instead of ``stat()`` so it works for both - backends (mtime triples for ``file``; ``max(updated_at)`` + row count for - ``db``). + backends (mtime triples for ``file``; a deterministic digest of stored + agent contents for ``db``). """ diff --git a/backend/packages/harness/deerflow/persistence/agents/sql.py b/backend/packages/harness/deerflow/persistence/agents/sql.py index 50ad21a9726..d474a2c8694 100644 --- a/backend/packages/harness/deerflow/persistence/agents/sql.py +++ b/backend/packages/harness/deerflow/persistence/agents/sql.py @@ -12,6 +12,8 @@ from __future__ import annotations +import hashlib +import json import logging import shutil import threading @@ -19,7 +21,7 @@ from collections.abc import Hashable from datetime import UTC, datetime -from sqlalchemy import Engine, create_engine, delete, event, func, select +from sqlalchemy import Engine, create_engine, delete, event, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session, sessionmaker @@ -33,7 +35,6 @@ ) from deerflow.persistence.agents.model import AgentRow from deerflow.runtime.user_context import get_effective_user_id -from deerflow.utils.time import coerce_iso logger = logging.getLogger(__name__) @@ -210,12 +211,35 @@ def delete(self, name: str, *, user_id: str | None = None) -> AgentDeleteOutcome return "missing" def signature(self) -> Hashable: - # MAX(updated_at) is not covered by an index (only user_id and the - # (user_id, name) unique constraint are), so this is a small full scan. - # It runs only on the webhook registry's cache-freshness check against a - # tiny agents table, so an index is not warranted; revisit if agents ever - # grow into the thousands with frequent webhook deliveries. + # The GitHub registry uses this token to decide whether cached agent + # bindings are still current. Timestamps alone are not sufficient + # because two writes can reuse the same database timestamp. + # Computing the digest reads the small agents table only on the + # registry's cache-freshness check; revisit if agent counts or webhook + # delivery rates grow enough for this scan to become material. with self._Session() as session: - max_updated, count = session.execute(select(func.max(AgentRow.updated_at), func.count(AgentRow.id))).one() - token = coerce_iso(max_updated) if isinstance(max_updated, datetime) else str(max_updated) - return (token, int(count)) + rows = session.execute( + select( + AgentRow.user_id, + AgentRow.name, + AgentRow.config, + AgentRow.soul, + ).order_by(AgentRow.user_id, AgentRow.name) + ).all() + + payload = [ + { + "user_id": user_id, + "name": name, + "config": config or {}, + "soul": soul or "", + } + for user_id, name, config, soul in rows + ] + serialized = json.dumps( + payload, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(serialized.encode("utf-8")).hexdigest() diff --git a/backend/tests/test_agent_store_sql.py b/backend/tests/test_agent_store_sql.py index f416073e383..09398ad88fc 100644 --- a/backend/tests/test_agent_store_sql.py +++ b/backend/tests/test_agent_store_sql.py @@ -8,11 +8,13 @@ from __future__ import annotations +from datetime import datetime from unittest import mock import pytest from sqlalchemy import create_engine +import deerflow.persistence.agents.model as agent_model from deerflow.config.agents_config import AgentConfig from deerflow.persistence.agents.base import AgentExistsError from deerflow.persistence.agents.model import AgentRow @@ -160,6 +162,36 @@ def test_signature_changes_on_mutation(store): assert store.signature() != after_create +def test_signature_changes_when_update_reuses_timestamp(store, monkeypatch): + store.create( + "x", + {"name": "x", "description": "before"}, + "s", + user_id="u1", + ) + after_create = store.signature() + + with store._Session() as session: + frozen_timestamp = session.query(AgentRow).one().updated_at + + class FrozenDateTime(datetime): + @classmethod + def now(cls, tz=None): + return frozen_timestamp + + monkeypatch.setattr(agent_model, "datetime", FrozenDateTime) + + store.update( + "x", + {"name": "x", "description": "changed"}, + None, + user_id="u1", + ) + + assert store.get("x", user_id="u1").description == "changed" + assert store.signature() != after_create + + def test_sync_engine_mirrors_async_pragmas(tmp_path): # The db backend's sync engine must set the same per-connection SQLite PRAGMAs # the async engine does (persistence/engine.py), not leave synchronous=FULL