Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions backend/AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 `<followups-while-busy>`-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.
Expand Down
5 changes: 3 additions & 2 deletions backend/app/gateway/github/registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
9 changes: 7 additions & 2 deletions backend/docs/GITHUB_AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -25,14 +25,19 @@ graph LR

OperatorYaml["config.yaml<br/>channels.github:<br/> enabled: true<br/> default_mention_login"]:::operator
AgentYaml["agents/{name}/config.yaml<br/>github:<br/> installation_id<br/> bot_login<br/> bindings: [{repo, triggers}]"]:::agent
Registry["build_github_agent_registry()<br/>(mtime-cached, asyncio.to_thread)"]:::route
Registry["build_github_agent_registry()<br/>(store-signature cached, asyncio.to_thread)"]:::route
Webhook["POST /api/webhooks/github<br/>(HMAC verify)"]:::route

OperatorYaml --> Registry
AgentYaml --> Registry
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
Expand Down Expand Up @@ -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)
- `app/channels/github.py` — `GitHubChannel` (log-only outbound)
4 changes: 2 additions & 2 deletions backend/packages/harness/deerflow/persistence/agents/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -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``).
"""
44 changes: 34 additions & 10 deletions backend/packages/harness/deerflow/persistence/agents/sql.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,14 +12,16 @@

from __future__ import annotations

import hashlib
import json
import logging
import shutil
import threading
import uuid
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

Expand All @@ -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__)

Expand Down Expand Up @@ -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()
32 changes: 32 additions & 0 deletions backend/tests/test_agent_store_sql.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down