From 3244d4a8ff2d6734c7561021a8805dc06263d70a Mon Sep 17 00:00:00 2001 From: kimi wang Date: Fri, 13 Nov 2026 13:55:36 +0800 Subject: [PATCH] fix(managed-subagents): make signature() change when an update lands in one clock tick (#8035) (#8076) * fix(managed-subagents): make signature() change when an update lands in one clock tick SqlManagedSubagentStore.signature() returns the tuple of (id, updated_at), so a definition change is only visible to peer processes when the row's stamp moves. The ORM onupdate callback reads the wall clock, which has ~15.6 ms resolution on Windows and can also sit behind a stamp written by a node with a skewed clock. In both cases the new value equals or precedes the stored one and the registry snapshot is never invalidated. Force updated_at strictly forward in update(): normalise SQLite's naive datetime to UTC and, when the clock has not advanced past the stored value, bump it by one microsecond. Add a regression test that freezes both clocks to the stored timestamp; it fails without the fix (signature unchanged) and passes with it. Refs #8035 * fix(managed-subagents): serialize signature timestamp updates --------- Co-authored-by: wangkangmao Co-authored-by: Willem Jiang --- .../harness/deerflow/persistence/AGENTS.md | 2 + .../persistence/managed_subagents/sql.py | 25 +++- backend/tests/test_managed_subagent_store.py | 137 +++++++++++++++++- 3 files changed, 154 insertions(+), 10 deletions(-) diff --git a/backend/packages/harness/deerflow/persistence/AGENTS.md b/backend/packages/harness/deerflow/persistence/AGENTS.md index 8c281eedb6..ea36586720 100644 --- a/backend/packages/harness/deerflow/persistence/AGENTS.md +++ b/backend/packages/harness/deerflow/persistence/AGENTS.md @@ -33,6 +33,8 @@ change to a query, a lock or a JSON predicate is verified on both dialects. Otherwise unrelated writers fail with `database is locked`. - **`FOR UPDATE SKIP LOCKED` does not serialize SQLite readers.** A claim on SQLite reserves the writer before it selects. +- **Managed subagent updates lock before reading `updated_at`.** Peer registry + caches depend on each committed update advancing that timestamp. - **A create-or-update by unique key must survive two first writers.** Both can miss the SELECT. Use the dialect's `on_conflict_do_update`, or roll back and retry the select-then-write. diff --git a/backend/packages/harness/deerflow/persistence/managed_subagents/sql.py b/backend/packages/harness/deerflow/persistence/managed_subagents/sql.py index c299fdc7fe..0e882aaf88 100644 --- a/backend/packages/harness/deerflow/persistence/managed_subagents/sql.py +++ b/backend/packages/harness/deerflow/persistence/managed_subagents/sql.py @@ -4,8 +4,9 @@ from __future__ import annotations import uuid from collections.abc import Hashable +from datetime import UTC, datetime, timedelta -from sqlalchemy import delete, select +from sqlalchemy import delete, select, text from sqlalchemy.exc import IntegrityError from deerflow.persistence.agents.sql import get_sync_sessionmaker @@ -58,11 +59,29 @@ class SqlManagedSubagentStore(ManagedSubagentStore): raise ManagedSubagentExistsError(f"Managed subagent '{definition.name}' already exists") from exc def update(self, definition: ManagedSubagentDefinition) -> None: + document = definition.model_dump(mode="json") + stmt = select(ManagedSubagentRow).where(ManagedSubagentRow.name == definition.name) with self._Session() as session: - row = session.execute(select(ManagedSubagentRow).where(ManagedSubagentRow.name == definition.name)).scalar_one_or_none() + # Serialize writers before reading updated_at, so each sees the + # preceding commit's stamp even when both clocks return the same time. + if session.get_bind().dialect.name == "sqlite": + session.execute(text("BEGIN IMMEDIATE")) + else: + stmt = stmt.with_for_update() + row = session.execute(stmt).scalar_one_or_none() if row is None: raise FileNotFoundError(f"Managed subagent not found: {definition.name}") - row.definition = definition.model_dump(mode="json") + row.definition = document + # Peer registries key their snapshots by (id, updated_at), so + # advance the stamp even when the wall clock stalls or moves back. + now = datetime.now(UTC) + current = row.updated_at + if current is not None: + if current.tzinfo is None: # SQLite returns naive datetimes + current = current.replace(tzinfo=UTC) + if now <= current: + now = current + timedelta(microseconds=1) + row.updated_at = now session.commit() def delete(self, name: str) -> bool: diff --git a/backend/tests/test_managed_subagent_store.py b/backend/tests/test_managed_subagent_store.py index b0dabc9d17..d8477d34c6 100644 --- a/backend/tests/test_managed_subagent_store.py +++ b/backend/tests/test_managed_subagent_store.py @@ -3,12 +3,18 @@ from __future__ import annotations import json +import os +import uuid +from concurrent.futures import ThreadPoolExecutor from datetime import UTC, datetime, timedelta +from threading import Event, get_ident from types import SimpleNamespace import pytest -from sqlalchemy import create_engine, select +from sqlalchemy import create_engine, event, select, text +from sqlalchemy.engine import make_url +from deerflow.config.subagents_config import SubagentsAppConfig from deerflow.persistence.base import Base from deerflow.persistence.managed_subagents import ( ManagedSubagentDefinition, @@ -19,6 +25,8 @@ from deerflow.persistence.managed_subagents.base import normalize_managed_subage from deerflow.persistence.managed_subagents.file import FileManagedSubagentStore from deerflow.persistence.managed_subagents.model import ManagedSubagentRow from deerflow.persistence.managed_subagents.sql import SqlManagedSubagentStore +from deerflow.persistence.postgres_schema import dsn_with_search_path +from deerflow.subagents import registry def _definition(name: str = "researcher", **changes) -> ManagedSubagentDefinition: @@ -38,12 +46,35 @@ def file_store(tmp_path, monkeypatch): @pytest.fixture() -def sql_store(tmp_path): - url = f"sqlite:///{tmp_path}/managed.db" - engine = create_engine(url) - Base.metadata.create_all(engine, tables=[ManagedSubagentRow.__table__]) - engine.dispose() - return SqlManagedSubagentStore(url) +def sql_store(tmp_path, request): + if getattr(request, "param", "sqlite") == "sqlite": + store = SqlManagedSubagentStore(f"sqlite:///{tmp_path}/managed.db") + engine = store._Session.kw["bind"] + try: + Base.metadata.create_all(engine, tables=[ManagedSubagentRow.__table__]) + yield store + finally: + engine.dispose() + return + + uri = os.environ.get("TEST_POSTGRES_URI") + if not uri: + pytest.skip("requires TEST_POSTGRES_URI (managed subagents on PostgreSQL)") + schema = f"managed_subagents_{uuid.uuid4().hex}" + admin = create_engine(make_url(uri).set(drivername="postgresql+psycopg")) + with admin.begin() as connection: + connection.execute(text(f'CREATE SCHEMA "{schema}"')) + url = make_url(dsn_with_search_path(uri, schema)).set(drivername="postgresql+psycopg") + store = SqlManagedSubagentStore(url.render_as_string(hide_password=False)) + engine = store._Session.kw["bind"] + try: + Base.metadata.create_all(engine, tables=[ManagedSubagentRow.__table__]) + yield store + finally: + engine.dispose() + with admin.begin() as connection: + connection.execute(text(f'DROP SCHEMA "{schema}" CASCADE')) + admin.dispose() @pytest.mark.parametrize("store_fixture", ["file_store", "sql_store"]) @@ -109,6 +140,98 @@ def test_sql_signature_detects_update_behind_existing_max_timestamp(sql_store): assert sql_store.signature() != before +def test_sql_signature_changes_when_the_clock_does_not_advance(sql_store, monkeypatch): + """An update inside one clock tick must still invalidate peer snapshots.""" + sql_store.create(_definition()) + fixed = datetime.now(UTC) + + # Pin the stored timestamp, then freeze the wall clock at exactly that + # value. The ORM onupdate callback writes the very same timestamp back, + # so without the fix the row's (id, updated_at) pair never moves. + with sql_store._Session() as session: + row = session.execute(select(ManagedSubagentRow).where(ManagedSubagentRow.name == "researcher")).scalar_one() + row.updated_at = fixed.replace(tzinfo=None) # SQLite persists naive datetimes + session.commit() + + before = sql_store.signature() + + class FrozenDatetime(datetime): + @classmethod + def now(cls, tz=None): + return fixed if tz is not None else fixed.replace(tzinfo=None) + + monkeypatch.setattr("deerflow.persistence.managed_subagents.model.datetime", FrozenDatetime) + # The store's own clock only exists once the fix is in place; raising=False + # keeps this test a behavioural check against unpatched code too. + monkeypatch.setattr("deerflow.persistence.managed_subagents.sql.datetime", FrozenDatetime, raising=False) + + sql_store.update(_definition().model_copy(update={"description": "second"})) + + assert sql_store.get("researcher").description == "second" + assert sql_store.signature() != before + + +@pytest.mark.parametrize("sql_store", ["sqlite", "postgresql"], indirect=True) +def test_concurrent_sql_updates_invalidate_the_registry_after_each_commit(sql_store, monkeypatch): + fixed = datetime(2026, 1, 1, tzinfo=UTC) + + class FrozenDatetime(datetime): + @classmethod + def now(cls, tz=None): + return fixed if tz is not None else fixed.replace(tzinfo=None) + + monkeypatch.setattr("deerflow.persistence.managed_subagents.model.datetime", FrozenDatetime) + monkeypatch.setattr("deerflow.persistence.managed_subagents.sql.datetime", FrozenDatetime) + definition = _definition() + sql_store.create(definition) + original_signature = sql_store.signature() + first = definition.model_copy(update={"description": "First commit"}) + second = definition.model_copy(update={"description": "Second commit"}) + + observer = SqlManagedSubagentStore(sql_store._url) + monkeypatch.setattr(registry, "get_managed_subagent_store", lambda *_: observer) + monkeypatch.setattr(registry, "_managed_definitions_cache", {}) + registry_clock = SimpleNamespace(monotonic=lambda: 0.0) + monkeypatch.setattr(registry, "time", registry_clock) + config = SubagentsAppConfig() + + writer_ready = Event() + resume_writer = Event() + reader_thread = get_ident() + + def pause_before_write(_conn, _cursor, statement, _params, _context, _executemany): + if get_ident() == reader_thread: + return + if statement.startswith(("BEGIN IMMEDIATE", "UPDATE ")) or " FOR UPDATE" in statement: + writer_ready.set() + if not resume_writer.wait(timeout=10): + raise TimeoutError("Second writer was not released") + + # Pause the second writer just before it takes a lock or issues its UPDATE. + # The old code has already read the timestamp when it reaches this point. + engine = sql_store._Session.kw["bind"] + event.listen(engine, "before_cursor_execute", pause_before_write) + try: + with ThreadPoolExecutor(max_workers=1) as pool: + pending = pool.submit(sql_store.update, second) + try: + assert writer_ready.wait(timeout=10) + sql_store.update(first) + first_signature = observer.signature() + assert first_signature[0][1] == original_signature[0][1] + timedelta(microseconds=1) + assert registry.get_subagent_config(definition.name, app_config=config).description == first.description + finally: + resume_writer.set() + pending.result(timeout=10) + finally: + event.remove(engine, "before_cursor_execute", pause_before_write) + + assert observer.get(definition.name).description == second.description + registry_clock.monotonic = lambda: registry._MANAGED_SIGNATURE_TTL_SECONDS + assert registry.get_subagent_config(definition.name, app_config=config).description == second.description + assert observer.signature()[0][1] == first_signature[0][1] + timedelta(microseconds=1) + + @pytest.mark.parametrize("store_fixture", ["file_store", "sql_store"]) def test_duplicate_create_is_a_conflict(request, store_fixture): store = request.getfixturevalue(store_fixture)