mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-10 05:58:36 +00:00
fix(agents): make SQL store signatures content-sensitive (#4709)
This commit is contained in:
parent
95989dfaae
commit
7910126923
@ -977,6 +977,7 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_
|
|||||||
|
|
||||||
**GitHub event-driven agents** (webhook-driven IM channel):
|
**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.
|
- 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.
|
- 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.
|
- **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.
|
- 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.
|
||||||
|
|||||||
@ -12,8 +12,9 @@ delivery. We avoid re-loading every agent on each call via a small cache keyed
|
|||||||
on the store's :meth:`~deerflow.persistence.agents.base.AgentStore.signature`
|
on the store's :meth:`~deerflow.persistence.agents.base.AgentStore.signature`
|
||||||
change token: the file backend derives it from ``config.yaml`` mtimes (any
|
change token: the file backend derives it from ``config.yaml`` mtimes (any
|
||||||
edit, addition, or deletion invalidates the cache transparently); the db backend
|
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
|
derives it from a deterministic digest of each agent's owner, name, config, and
|
||||||
``config.yaml`` see the change on the next webhook.
|
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+ /
|
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
|
APFS is ~1 µs but on some filesystems (FAT, network shares with caching) it's
|
||||||
|
|||||||
@ -25,7 +25,7 @@ graph LR
|
|||||||
|
|
||||||
OperatorYaml["config.yaml<br/>channels.github:<br/> enabled: true<br/> default_mention_login"]:::operator
|
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
|
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
|
Webhook["POST /api/webhooks/github<br/>(HMAC verify)"]:::route
|
||||||
|
|
||||||
OperatorYaml --> Registry
|
OperatorYaml --> Registry
|
||||||
@ -33,6 +33,11 @@ graph LR
|
|||||||
Registry --> Webhook
|
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.
|
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
|
## Webhook → Fan-out → Dispatch
|
||||||
|
|||||||
@ -126,6 +126,6 @@ class AgentStore(abc.ABC):
|
|||||||
|
|
||||||
Equal tokens mean "nothing changed since last read". The GitHub registry
|
Equal tokens mean "nothing changed since last read". The GitHub registry
|
||||||
keys its cache off this instead of ``stat()`` so it works for both
|
keys its cache off this instead of ``stat()`` so it works for both
|
||||||
backends (mtime triples for ``file``; ``max(updated_at)`` + row count for
|
backends (mtime triples for ``file``; a deterministic digest of stored
|
||||||
``db``).
|
agent contents for ``db``).
|
||||||
"""
|
"""
|
||||||
|
|||||||
@ -12,6 +12,8 @@ the app, so this adds no dependency.
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import hashlib
|
||||||
|
import json
|
||||||
import logging
|
import logging
|
||||||
import shutil
|
import shutil
|
||||||
import threading
|
import threading
|
||||||
@ -19,7 +21,7 @@ import uuid
|
|||||||
from collections.abc import Hashable
|
from collections.abc import Hashable
|
||||||
from datetime import UTC, datetime
|
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.exc import IntegrityError
|
||||||
from sqlalchemy.orm import Session, sessionmaker
|
from sqlalchemy.orm import Session, sessionmaker
|
||||||
|
|
||||||
@ -33,7 +35,6 @@ from deerflow.persistence.agents.base import (
|
|||||||
)
|
)
|
||||||
from deerflow.persistence.agents.model import AgentRow
|
from deerflow.persistence.agents.model import AgentRow
|
||||||
from deerflow.runtime.user_context import get_effective_user_id
|
from deerflow.runtime.user_context import get_effective_user_id
|
||||||
from deerflow.utils.time import coerce_iso
|
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@ -210,12 +211,35 @@ class SqlAgentStore(AgentStore):
|
|||||||
return "missing"
|
return "missing"
|
||||||
|
|
||||||
def signature(self) -> Hashable:
|
def signature(self) -> Hashable:
|
||||||
# MAX(updated_at) is not covered by an index (only user_id and the
|
# The GitHub registry uses this token to decide whether cached agent
|
||||||
# (user_id, name) unique constraint are), so this is a small full scan.
|
# bindings are still current. Timestamps alone are not sufficient
|
||||||
# It runs only on the webhook registry's cache-freshness check against a
|
# because two writes can reuse the same database timestamp.
|
||||||
# tiny agents table, so an index is not warranted; revisit if agents ever
|
# Computing the digest reads the small agents table only on the
|
||||||
# grow into the thousands with frequent webhook deliveries.
|
# 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:
|
with self._Session() as session:
|
||||||
max_updated, count = session.execute(select(func.max(AgentRow.updated_at), func.count(AgentRow.id))).one()
|
rows = session.execute(
|
||||||
token = coerce_iso(max_updated) if isinstance(max_updated, datetime) else str(max_updated)
|
select(
|
||||||
return (token, int(count))
|
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()
|
||||||
|
|||||||
@ -8,11 +8,13 @@ signature the GitHub registry keys its cache off.
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from datetime import datetime
|
||||||
from unittest import mock
|
from unittest import mock
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from sqlalchemy import create_engine
|
from sqlalchemy import create_engine
|
||||||
|
|
||||||
|
import deerflow.persistence.agents.model as agent_model
|
||||||
from deerflow.config.agents_config import AgentConfig
|
from deerflow.config.agents_config import AgentConfig
|
||||||
from deerflow.persistence.agents.base import AgentExistsError
|
from deerflow.persistence.agents.base import AgentExistsError
|
||||||
from deerflow.persistence.agents.model import AgentRow
|
from deerflow.persistence.agents.model import AgentRow
|
||||||
@ -160,6 +162,36 @@ def test_signature_changes_on_mutation(store):
|
|||||||
assert store.signature() != after_create
|
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):
|
def test_sync_engine_mirrors_async_pragmas(tmp_path):
|
||||||
# The db backend's sync engine must set the same per-connection SQLite PRAGMAs
|
# 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
|
# the async engine does (persistence/engine.py), not leave synchronous=FULL
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user