github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-08-01 02:46:02 +00:00
Code Issues Packages Projects Releases Wiki Activity
2,889 Commits 48 Branches 8 Tags
Commit Graph

10 Commits

Author SHA1 Message Date
MiaoRuidx
8a78c264b7
fix(runtime): cancel runs across live gateway workers (#4500) 
* docs(runtime): design cross-worker cancellation

* fix(runtime): cancel runs across gateway workers

* fix(runtime): harden cross-worker cancellation races

让取消请求与 owner 终态写入通过持久化 CAS 决定先后，保证首次取消 action 在不同 worker 路由下保持一致。\n\n将 heartbeat 收敛为续租后仅发送本地中止信号，并补齐完成竞态与路由重试的回归用例。

* docs(runtime): drop implementation plan from PR

移除仅用于实现过程的跨 worker 取消设计记录，保留 README 和 backend/AGENTS.md 中面向最终行为的文档。

* fix(runtime): preserve local cancel fallback

* test(runtime): adapt worker run manager fakes

* docs(runtime): fix run cancel migration registry

---------

Co-authored-by: MiaoRuidx <12540796+MiaoRuidx@users.noreply.github.com>
 2026-07-28 21:45:20 +08:00
Zhipeng Zheng
838037188e
feat(channels): share inbound webhook dedupe across pods via Postgres (#4210) 
* feat(channels): share inbound webhook dedupe across pods via Postgres (#4120)

* ci: run cross-pod inbound dedupe integration tests in CI

Expose the job Postgres service via DEDUPE_TEST_POSTGRES_URL so the integration tests (issue #4120) actually execute instead of silently skipping. Normalize the URL for asyncpg (postgresql:// -> +asyncpg, drop libpq-only sslmode) and await the now-async _is_duplicate_inbound in test_github_dispatcher.
 2026-07-27 23:07:40 +08:00
Vanzeren
3c8b82c594
fix(runtime): serialize checkpoint writes with active runs (#4437) 
* fix(runtime): serialize checkpoint writes with active runs

* fix(runtime): address checkpoint reservation reviews

* fix(runtime): address reservation race reviews

* fix(runtime): refine reservation conflict semantics
 2026-07-25 23:18:34 +08:00
Daoyuan Li
ca3e510b7d
fix(scheduler): close duplicate dispatch race (#4105) 
Enforce one queued or running scheduled-task run per task with a partial unique index. The migration resolves legacy duplicates before creating the index, and losing inserts use the existing conflict or skip outcomes.
 2026-07-24 21:41:09 +08:00
Aari
0d4d0cb17d
feat(agents): database-backed storage for custom agent definitions (#4359) 
* feat(agents): database-backed storage for custom agent definitions

Add an agent_storage.backend switch (default file, behaviour-unchanged) with a
db backend that stores each custom agent as a row in the shared SQL persistence
layer, so a multi-instance deployment sees the same agents on every node
(#4331, #4357). Introduces an AgentStore interface routing all read/write
surfaces, an agents table + migration 0006, startup validation, and a file->db
importer. Follows the thread_meta store / run_events backend-switch /
0003_scheduled_tasks migration patterns; no new dependency.

* fix(agents): make db storage path production-ready (review round 1)

Addresses review feedback on the db/sync agent-storage path:

- sql.py: mirror the async engine's per-connection SQLite PRAGMAs on the sync
  engine (busy_timeout=30000, synchronous=NORMAL, foreign_keys=ON, WAL) so both
  engines behave identically against the shared DB; guard the engine cache with
  a lock (double-checked) so concurrent first-touch cannot build duplicate
  engines or register the connect listener twice.
- routers/agents.py + routers/assistants_compat.py: offload the sync-store reads
  that ran on the event loop (list/get/check, update's pre-read + legacy guard +
  refresh, and assistants_compat's four list routes) via asyncio.to_thread — on
  db+postgres each was a network round trip stalling the loop. Writes were
  already offloaded.
- file.py: translate the create() mkdir(exist_ok=False) race FileExistsError
  into AgentExistsError (router 409, matching SqlAgentStore's IntegrityError
  path); correct the _write docstring — per-file atomic replace, two commits
  sequential not transactional.

Tests: sync-engine PRAGMA + engine-cache reuse assertions; file create-race ->
AgentExistsError; strict Blockbuster anchor over the read endpoints so a
regression back onto the loop fails CI.

* fix(agents): address round-2 review on the db store path

- update_agent tool: align the docstring/inline comment with FileAgentStore._write.
  Cross-field write atomicity is db-only; the file backend commits config then
  soul via two sequential os.replace (a crash between them can leave a fresh
  config.yaml beside a stale SOUL.md). The dropped partial-write *reporting* is
  an intentional tradeoff — the stage-then-replace safety is preserved
  (test_update_agent_soul_failure_does_not_replace_config still holds).
- SqlAgentStore.update(): true upsert. Catch IntegrityError on the
  insert-on-missing branch, re-fetch and apply, so two concurrent first-time
  writes (e.g. two setup_agent handshakes) converge instead of surfacing a raw
  UNIQUE(user_id, name) violation as a 500. Symmetric with create().
- get_agent_store(): document the graph-subprocess config-resolution invariant
  (the except->file fallback is a genuine no-config path, not a mask for a
  misconfigured graph process) and pin it with two tests driving the real
  get_app_config() file resolution: db resolves from an on-disk config.yaml,
  file fallback when config is unresolvable.

* test(agents): cover SqlAgentStore.update() write-race upsert recovery

Mandatory-TDD test for the round-2 fix in 0680340a: two concurrent first-time
update()s where the loser's insert hits UNIQUE(user_id, name). Deterministically
forces the IntegrityError recovery path by making the first _row probe miss the
committed winner, and asserts last-writer-wins instead of a surfaced 500.
 2026-07-23 08:03:21 +08:00
Vanzeren
1769b2de0d
fix: read run stop_reason from runtime context (#4188) 
* fix: read run stop_reason from runtime context

* fix: address review feedback for #4188 stop_reason integration

   - migration 0005: use safe_add_column for consistency and drift detection
   - worker: clear runtime.context stop_reason at start of each _stream_once
     turn so a clean continuation doesn't inherit a prior cap reason
   - tests: replace circular unit test with real middleware integration
     tests that exercise LoopDetectionMiddleware._apply and
     TokenBudgetMiddleware._apply through the worker, proving the full
     middleware → runtime.context → persist pipeline

* fix(test): resume conftest

* fix: stamp stop_reason in all guard middlewares, fix clearing semantics
 2026-07-16 08:19:52 +08:00
heart-scalpel
3bc3af2530
fix(runs): close multi-worker ownership gaps in run atomicity (#3948) (#4003) 
* feat(runs): cross-process run ownership with lease + reconciliation (#3948)

Implements work items 2 and 3 of the multi-worker P0 plan
(docs/multi_worker.md). Work item 1 (Postgres startup gate, #3960)
already landed; this PR makes run creation race-safe across worker
processes and lets Postgres deployments recover orphaned inflight runs
from crashed workers without mis-marking live runs as orphans.

Work item 2 — cross-process atomic create_or_reject

- Alembic revision 0004_run_ownership adds runs.owner_worker_id,
  runs.lease_expires_at, idx_runs_lease, and a partial unique index
  uq_runs_thread_active (one pending/running run per thread). The
  index is declared on RunRow.__table_args__ with sqlite_where +
  postgresql_where (mirroring uq_channel_connection_active_identity)
  so the empty-DB bootstrap path — which runs Base.metadata.create_all
  + alembic stamp head without executing any revision's upgrade() —
  also lands it on fresh deployments. Migration 0004 additionally
  creates it idempotently for legacy/versioned upgrades.
- RunRepository.create_run_atomic is the new atomic primitive:
  - reject: INSERT directly; the partial unique index catches
    duplicate active runs; the manager surfaces the result as
    ConflictError.
  - interrupt/rollback: SELECT FOR UPDATE the conflicting rows,
    skip rows whose lease is still valid AND owned by another live
    worker (raise ConflictError — the INSERT would have failed on
    the index anyway, and a retry loop cannot make progress),
    cancel the rest in the same transaction, then INSERT the new
    row. Rows owned by this worker are interruptible regardless of
    lease state.
- RunManager.create_or_reject dispatches to the store under the
  existing local lock; same-worker in-memory cancellation runs after
  the store commit succeeds. MemoryRunStore mirrors the same
  semantics for tests and database.backend=memory.

Work item 3 — lease heartbeat + Postgres reconciliation

- RunOwnershipConfig (lease_seconds=30, grace_seconds=10,
  heartbeat_enabled=false by default), registered as startup-only in
  reload_boundary.STARTUP_ONLY_FIELDS because the heartbeat background
  task is created once in langgraph_runtime() and is not rebuilt on
  config.yaml edits.
- When heartbeat_enabled, each worker renewes leases on its own
  active runs with interval = lease_seconds / 3. The loop is bounded
  and stop-event-cancellable so shutdown is prompt.
- reconcile_orphaned_inflight_runs now runs on every backend — the
  sqlite-only gate in app/gateway/deps.py is dropped in the same
  commit so there is no window where Postgres would mis-mark live
  Worker A runs as orphans. Reconciliation errors only runs whose
  lease is NULL (legacy pre-ownership rows) or older than
  grace_seconds. In single-worker mode (heartbeat off, NULL leases)
  all inflight rows reclaim immediately, preserving the pre-ownership
  recovery latency.
- Heartbeat starts AFTER startup reconciliation and stops BEFORE the
  in-flight run drain on shutdown so the two cannot race.

GATEWAY_WORKERS=1 with heartbeat_enabled=false keeps current behavior.

Verified: 170 related tests + full backend suite (minus Docker-gated
live tests) green; ruff check + ruff format clean.

* fix(runs): tighten unique-violation handling and document clock-sync budget

Three follow-up fixes to the cross-process run ownership work in #3948,
surfacing during review.

1. _is_unique_violation: detect by driver-native signal, not message text

   The previous substring heuristic ("unique" + "violat", or "duplicate")
   missed SQLite's actual phrasing "UNIQUE constraint failed: <table>.<index>"
   — SQLite says "failed", not "violates", and never "duplicate". On SQLite
   the detector returned False, the reject path re-raised the raw
   IntegrityError, and clients saw HTTP 500 instead of ConflictError 409.
   The conversion is the load-bearing piece of the "store is source of
   truth" design but was untested — every atomic test used MemoryRunStore,
   which raises ConflictError directly and never reached this branch.

   Now prefers driver-native signals: psycopg pgcode/sqlcode "23505" and
   sqlite3 sqlite_errorcode SQLITE_CONSTRAINT_UNIQUE (reachable through
   SQLAlchemy IntegrityError.orig). Message matching stays as a fallback
   with SQLite's exact "unique constraint failed" phrase added.

2. interrupt/rollback: convert exhausted-retry IntegrityError to ConflictError

   The reject branch converts unique violations to ConflictError. The
   interrupt/rollback retry loop did not — on the 3rd attempt it re-raised
   the raw IntegrityError, leaking HTTP 500 for the same race condition
   that reject surfaces as 409. Symmetric conversion added after the loop;
   callers now see a consistent ConflictError regardless of strategy.

3. Document clock-sync requirement for multi-worker lease reconciliation

   reconcile_orphaned_inflight_runs compares another worker's UTC
   lease_expires_at against this worker's datetime.now(UTC). The only skew
   budget is grace_seconds (default 10s) — worst case, with the owning
   worker's heartbeat just about to fire, a peer whose clock is more than
   ~grace_seconds ahead can mis-reclaim a still-live run as an orphan.

   Documented in RunOwnershipConfig's docstring (with the math) and in
   config.example.yaml (with operational guidance), so operators in
   NTP-poor environments know to raise grace_seconds. Default unchanged:
   10s is reasonable for NTP-synced K8s/cloud, and bumping it would slow
   recovery of genuinely dead workers (lease_seconds + grace_seconds from
   last heartbeat to reclaim).

Tests:
- test_create_run_atomic_reject_propagates_conflict_on_unique_violation:
  end-to-end against a real SQLite-backed RunRepository, pre-inserts an
  active run, asserts reject-strategy create surfaces as ConflictError
  rather than raw IntegrityError.
- test_is_unique_violation_detects_real_sqlite_integrity_error: unit test
  for the detector against a real SQLite-raised IntegrityError; asserts
  driver-level sqlite_errorcode is SQLITE_CONSTRAINT_UNIQUE.
- test_interrupt_exhausted_retries_surface_as_conflict_error: pins the
  symmetric 409 behavior after the retry loop exhausts.

Verified: ruff check + ruff format clean; multi-worker + run_repository
+ owner_isolation + reload_boundary suites green.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): close multi-worker ownership gaps in lease heartbeat and unique-violation detection

Five code-review fixes from docs/multi_worker.md:

1. Drop unused ``claim_inflight_runs`` primitive — no caller anywhere.
   ``create_run_atomic`` does its own inline claim (SELECT FOR UPDATE +
   cancel) inside the INSERT transaction; a separate claim primitive
   would split that into two transactions and open a claim→INSERT race.
   Removes ~40 lines across base.py / memory.py / sql.py plus the
   unused ``now_iso`` parameter, freeing future RunStore implementations
   from providing it.

2. Broaden ``_renew_leases`` filter to renew pending/running runs owned
   by this worker even when ``record.task is None``. The previous
   ``task is not None`` requirement skipped the brief window between
   ``create_run_atomic`` inserting the row and the worker spawning the
   agent task; under event-loop load that window can approach
   ``lease_seconds``, after which peer reconciliation marks the run
   ``error`` (visible) or a peer's ``create_or_reject("interrupt")``
   silently kills the queued run. Filter now:
   ``task is None or not task.done()``.

3. Document the unsynchronised ``record.lease_expires_at = new_expiry``
   write. ``lease_expires_at`` is the only field on an existing record
   this path mutates; ``set_status`` / ``_persist_status`` touch other
   fields, so there is no concurrent writer to race against. Re-acquiring
   ``self._lock`` would serialise unrelated run mutations for no gain.

4. Gate ``_is_unique_violation`` message fallbacks on
   ``isinstance(current, (SAIntegrityError, sqlite3.IntegrityError))``.
   The driver-code path (pgcode/sqlite_errorcode) remains load-bearing;
   substring fallbacks are now belt-and-suspenders only for cases where
   the driver attribute isn't reachable through the cause chain. Without
   the gate, any application exception whose ``str()`` happens to contain
   "duplicate key" / "unique" + "violat" (CHECK constraint, validation
   error) would silently surface as HTTP 409 instead of 500.

5. Route ``update_lease`` through ``_call_store_with_retry`` for
   consistency with every other store call, and wrap
   ``await self._renew_leases()`` in ``_heartbeat_loop`` with
   ``except Exception: logger.warning(...)``. Previously a transient
   error from the snapshot path or an unexpected exception would kill
   the heartbeat task silently — after which no lease is ever renewed
   again and every active run eventually looks orphaned.
   ``except Exception`` lets ``CancelledError`` (BaseException since
   3.8) propagate so shutdown cancellation still works.

Regression tests:
- ``test_heartbeat_renews_pending_run_before_task_is_spawned``
- ``test_is_unique_violation_does_not_misclassify_application_exception``

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): harden multi-worker migration, memory atomicity, and tz-naive lease comparison

Three follow-up fixes to the multi-worker run ownership work:

- migration 0004 dedupe pass: cancel superseded duplicate active rows per
  thread before creating the partial UNIQUE index ``uq_runs_thread_active``
  so dirty DBs (Postgres deployments that had reconciliation skipped by the
  old sqlite-only gate, or any env that ran GATEWAY_WORKERS>1 before this PR)
  do not abort the alembic upgrade and block gateway startup. Keeps the
  newest active row per thread, marks the rest as error with an explanatory
  message.

- MemoryRunStore.create_run_atomic interrupt/rollback path: split the single-
  pass loop into two passes (collect candidates, validate, then mutate) so a
  ConflictError raised on a later candidate does not leave earlier candidates
  half-interrupted. Mirrors the SQL store's transactional rollback semantics;
  the entire test_multi_worker_run_ownership.py suite runs against memory so
  this divergence was giving false confidence.

- RunRepository.create_run_atomic interrupt path: coerce tz-naive
  ``row.lease_expires_at`` to UTC before comparing against the aware
  ``cutoff``. SQLite drops tzinfo on read despite ``DateTime(timezone=True)``
  (this file's own comment acknowledges it), so the Python-side comparison
  raised ``TypeError: can't compare offset-naive and offset-aware datetimes``
  whenever heartbeat was enabled on SQLite and a lease was non-NULL. Defaults
  (heartbeat off -> leases always NULL) masked it, but there was no guard
  against the combination. Follows the existing "naive is UTC" convention
  from ``coerce_iso``.

Each fix ships with a regression test pinning the behavior.

Co-Authored-By: heart-scalpel <heart-scalpel@users.noreply.github.com>
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): enforce heartbeat for multi-worker, fix memory-store datetime comparison, lazy-import ConflictError in store layer

Three fixes from code review:

1. Extend the startup gate (GATEWAY_WORKERS>1) to also require
   run_ownership.heartbeat_enabled=true. Without heartbeat every run has
   a NULL lease, so reconciliation treats all inflight rows as orphans
   and Worker B would kill Worker A's live runs on every rolling update
   or scale-up.

2. Fix MemoryRunStore.list_inflight_with_expired_lease to parse
   created_at as datetime instead of ISO string lexical comparison,
   and handle tz-naive lease values uniformly with the SQL store.

3. Store layer (sql.py, memory.py) now lazy-imports ConflictError
   inside create_run_atomic instead of importing from the higher
   RunManager layer at module level.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): add owner check to update_lease, document create() assumption, restore deleted comment

- update_lease (SQL + memory) now requires owner_worker_id match in WHERE
  clause so the primitive is safe by construction against misuse
- create() docstring notes it bypasses atomic create_run_atomic and
  assumes no active run exists for the thread
- restore explanatory comment in MemoryRunStore.aggregate_tokens_by_thread
  that was dropped in an earlier commit

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

* fix(runs): add psycopg3 sqlstate detection and periodic orphan reconciliation

- _is_unique_violation now checks sqlstate attribute (psycopg3 uses this
  instead of pgcode). On Postgres, the only supported multi-worker backend,
  detection was falling through to the message-substring fallback.
- _heartbeat_loop now runs reconcile_orphaned_inflight_runs every 3rd
  cycle (every lease_seconds) to catch orphans whose lease expires between
  pod restarts. Single-worker deployments are unaffected (heartbeat off).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
Co-authored-by: heart-scalpel <heart-scalpel@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
 2026-07-11 16:05:30 +08:00
Xinmin Zeng
4fc08b4f15
feat: add scheduled tasks MVP (#3898) 
* feat: add scheduled tasks MVP

* fix: harden scheduled task execution semantics

* feat(scheduled-tasks): preset-driven schedule form with timezone and live preview

Replace the raw cron input with a preset Select (hourly/daily/weekly/monthly/custom)
plus structured inputs (time picker, weekday toggles, day-of-month), datetime-local
for one-time tasks, a timezone selector defaulting to the browser timezone, and a
live human-readable preview. Reuses one ScheduledTaskScheduleInput for create and
edit; backend contract unchanged; zero new deps (pure Intl + DST-safe offset helpers).

* feat(scheduled-tasks): full-page i18n + recipe templates + E2E locale pin

Localize the rest of the scheduled-tasks page (filters, detail pane, actions,
edit form, run list, enum values) via t.scheduledTasks.* in en/zh. Add four
built-in recipe templates (GitHub Trending, news digest, issue triage, weekly
report) exposed as a chip row that pre-fills title + prompt + schedule. Pin
Playwright locale to en-US so E2E selectors stay stable against i18n. No backend
change, no new deps.

* fix(scheduled-tasks): idempotent 0003 migration, update head constants, future-date once test

Merge with main surfaced three CI failures:
- 0003_scheduled_tasks create_table collided with legacy test seeds that
  build from full metadata; guard with inspector.has_table so the revision
  no-ops when the table already exists (0004/0005 are already idempotent via
  _helpers.py).
- persistence bootstrap concurrency/regression tests pinned HEAD to main's
  0002_runs_token_usage; bump to the new head 0005_scheduled_task_thread_nullable.
- once-task router test used a fixed past run_at and tripped the
  must-be-in-the-future validation; use a future date.

* address review: ok-check, 502 for trigger failure, mock fields, migration filename, doc fences

- fetchThreadScheduledTasks now checks response.ok like the other fetchers.
- trigger endpoint returns 502 (not 409) when dispatch fails outright, so
  clients can distinguish a real conflict from a server-side failure.
- E2E mock normalizes scheduled-task objects with context_mode/last_thread_id
  and nullable thread_id, matching the backend contract the UI renders against.
- Rename 0002_scheduled_tasks.py -> 0003_scheduled_tasks.py to match its
  revision id (file was renamed in spirit already; filename now follows).
- CONFIGURATION.md: close the Tool Groups yaml fence and drop the stray fence
  after the Scheduler notes so the sections render correctly.

* fix(scheduled-tasks): harden lease, poller, config, and frontend UX after review

* fix(scheduled-tasks): harden run lifecycle, overlap skip, non_interactive gating, and DST conversion after review

- defer a once task's terminal status to the run-completion hook; the task
  stays running until the real outcome, and a startup sweep cancels once
  tasks orphaned by a crash (launch-time 'completed' could stick forever)
- record interrupted runs as a distinct 'interrupted' run status with a
  readable message; an interrupted once task ends 'cancelled', not 'failed'
- enforce overlap_policy=skip for fresh_thread_per_run via an active-run
  pre-check (same-thread ConflictError can never fire across fresh threads)
- protect terminal run statuses from the late launch-path 'running' write
- honor context.non_interactive only for internally-authenticated callers;
  arbitrary clients can no longer strip ask_clarification
- fix DST-stale timezone offset in zonedLocalToUtcIso by re-deriving the
  offset at the resolved instant (once tasks fired an hour late around
  spring-forward and the create->edit round-trip diverged)
- drop dead ScheduledTaskRunRepository.update_by_run_id; share one Gateway
  API error helper between channels and scheduled-tasks frontends

* fix(scheduled-tasks): close review round-3 gaps in guards, concurrency, and API ergonomics

- scrub internal-only context keys (non_interactive) from the assembled run
  config for non-internal callers: gating body.context alone left the same
  key smuggle-able through the free-form body.config copied verbatim by
  build_run_config
- guard update_after_launch with protect_terminal so the launch bookkeeping
  write cannot clobber a once task already finalized by a fast-failing run's
  completion hook (parent-row sibling of the run-row guard)
- reject a manual trigger while the task has an active run (409) instead of
  launching a duplicate concurrent run on fresh_thread_per_run
- re-arm a terminal once task to enabled when PATCH pushes run_at into the
  future; previously the endpoint returned 200 with a next_run_at that could
  never be claimed
- make max_concurrent_runs a real global cap: each poll claims only into the
  remaining budget of active (queued/running) scheduled runs
- paginate GET /scheduled-tasks/{id}/runs (limit<=200, offset) and push the
  thread filter of /threads/{id}/scheduled-tasks into SQL
- stamp context.user_id on scheduler-launched runs, matching IM channels, so
  user-scoped guardrail providers see the owning user

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
 2026-07-04 21:51:57 +08:00
AnoobFeng
debb0fd161
feat(persistence): wire alembic migrations, bootstrap schema on startup (#3706) 
* feat(persistence): wire alembic migrations + bootstrap schema on startup

Closes #3682. Pre-#3658 DBs lack the `runs.token_usage_by_model` column
because alembic was never wired up — startup only ran `create_all`,
which never ALTERs existing tables.

Adds a hybrid bootstrap in FastAPI lifespan (replaces bare `create_all`):
- empty DB     → create_all + stamp head
- legacy DB    → stamp 0001_baseline + upgrade head
- versioned DB → upgrade head

Concurrency: Postgres `pg_advisory_lock` (cross-process); SQLite
per-engine `asyncio.Lock` + 30s `PRAGMA busy_timeout` on both prod and
alembic engines. Column revisions use `safe_add_column` /
`safe_drop_column` idempotent helpers as fallback.

Other bits:
- 0001 baseline (chain root) + 0002 add `runs.token_usage_by_model`
- `include_object` filter so alembic ignores LangGraph checkpointer tables
- `make migrate-rev MSG="..."` for authoring new revisions
  (no migrate/stamp targets — startup is the only execution path)
- Tests: three-branch decision, concurrency, #3682 regression, env
  filter, blocking-IO gate anchor
- CLAUDE.md: new "Schema migrations" section

* fix(style): fix lint error

* perf(persistence): address review feedback on alembic bootstrap

Behavioural fixes
- _SQLITE_LOCKS now keyed via WeakKeyDictionary so id-reuse after GC
  cannot return a stale, loop-bound lock and the cache cannot leak one
  entry per disposed engine.
- safe_add_column compares nullable / server_default against the desired
  column when the name already exists and emits a warning on drift,
  surfacing manual-ALTER workarounds instead of silently no-op'ing.
- _postgres_lock issues SET LOCAL idle_in_transaction_session_timeout=0
  before pg_advisory_lock, so managed Postgres cannot kill the idle
  lock-holding session mid-upgrade and silently release the advisory
  lock.
- legacy branch now backfills missing baseline tables via a restricted
  create_all (Base.metadata.create_all scoped to _BASELINE_TABLE_NAMES).
  Restores pre-#1930 upgraders whose channel_* tables were never
  provisioned, without pre-empting future create_table revisions for
  newly-added models.

Schema parity
- runs.token_usage_by_model gains server_default=text("'{}'") in both
  the ORM model and the 0001_baseline create_table, matching what 0002
  adds via ALTER. create_all and alembic-upgrade paths now produce
  identical column definitions.
- New parity test compares Base.metadata.create_all output against a
  pure alembic upgrade base->head, asserting column-set, nullable, and
  server_default agree across all tables (normalized through the same
  helper safe_add_column's drift check uses).

Guards
- test_baseline_table_names_constant_matches_0001 pins
  _BASELINE_TABLE_NAMES to 0001_baseline.upgrade()'s actual output --
  the constant cannot drift silently when someone edits 0001.
- test_legacy_backfill_skips_non_baseline_tables verifies the restricted
  backfill does not create a phantom table on Base.metadata, modelling
  a future revision that would otherwise collide on op.create_table.

Doc residuals
- Three-branch decision table is now consistent across bootstrap.py
  top docstring, engine.py comment, test module docstring, and
  CLAUDE.md.
- Stale test anchor in blocking_io/test_persistence_engine_sqlite.py
  docstring now points at the real file.

* fix(style): fix lint error

* fix(persistence): close drift detection holes

- _check_column_drift compares column type via a family equivalence
  allowlist ({JSON, JSONB}). Catches the wrong-type workaround
  `TEXT NOT NULL DEFAULT '{}'` that previously slipped through silently,
  while keeping Postgres JSON/JSONB dialect reflection quiet. Reflected
  and desired type are also echoed in every drift warning's payload for
  operator triage.

- Extract _escape_url_for_alembic so bootstrap._alembic_safe_url and
  scripts/_autogen_revision share the ConfigParser % escape rule
  instead of duplicating it.

- backend/README.md: add `make migrate-rev MSG=...` to Commands and a
  Schema Migrations section per the repo's README/CLAUDE.md sync policy.

- test_base_to_dict.py: scope the test ORM class to an isolated MetaData
  so the create_all-vs-alembic parity test (added in the previous
  commit) is not polluted by the phantom table on the full pytest
  session.
 2026-06-24 13:57:12 +08:00
rayhpeng
d8ecaf46c9 feat(persistence): add unified persistence layer with event store, token tracking, and feedback (#1930) 
* feat(persistence): add SQLAlchemy 2.0 async ORM scaffold

Introduce a unified database configuration (DatabaseConfig) that
controls both the LangGraph checkpointer and the DeerFlow application
persistence layer from a single `database:` config section.

New modules:
- deerflow.config.database_config — Pydantic config with memory/sqlite/postgres backends
- deerflow.persistence — async engine lifecycle, DeclarativeBase with to_dict mixin, Alembic skeleton
- deerflow.runtime.runs.store — RunStore ABC + MemoryRunStore implementation

Gateway integration initializes/tears down the persistence engine in
the existing langgraph_runtime() context manager. Legacy checkpointer
config is preserved for backward compatibility.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat(persistence): add RunEventStore ABC + MemoryRunEventStore

Phase 2-A prerequisite for event storage: adds the unified run event
stream interface (RunEventStore) with an in-memory implementation,
RunEventsConfig, gateway integration, and comprehensive tests (27 cases).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat(persistence): add ORM models, repositories, DB/JSONL event stores, RunJournal, and API endpoints

Phase 2-B: run persistence + event storage + token tracking.

- ORM models: RunRow (with token fields), ThreadMetaRow, RunEventRow
- RunRepository implements RunStore ABC via SQLAlchemy ORM
- ThreadMetaRepository with owner access control
- DbRunEventStore with trace content truncation and cursor pagination
- JsonlRunEventStore with per-run files and seq recovery from disk
- RunJournal (BaseCallbackHandler) captures LLM/tool/lifecycle events,
  accumulates token usage by caller type, buffers and flushes to store
- RunManager now accepts optional RunStore for persistent backing
- Worker creates RunJournal, writes human_message, injects callbacks
- Gateway deps use factory functions (RunRepository when DB available)
- New endpoints: messages, run messages, run events, token-usage
- ThreadCreateRequest gains assistant_id field
- 92 tests pass (33 new), zero regressions

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat(persistence): add user feedback + follow-up run association

Phase 2-C: feedback and follow-up tracking.

- FeedbackRow ORM model (rating +1/-1, optional message_id, comment)
- FeedbackRepository with CRUD, list_by_run/thread, aggregate stats
- Feedback API endpoints: create, list, stats, delete
- follow_up_to_run_id in RunCreateRequest (explicit or auto-detected
  from latest successful run on the thread)
- Worker writes follow_up_to_run_id into human_message event metadata
- Gateway deps: feedback_repo factory + getter
- 17 new tests (14 FeedbackRepository + 3 follow-up association)
- 109 total tests pass, zero regressions

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* test+config: comprehensive Phase 2 test coverage + deprecate checkpointer config

- config.example.yaml: deprecate standalone checkpointer section, activate
  unified database:sqlite as default (drives both checkpointer + app data)
- New: test_thread_meta_repo.py (14 tests) — full ThreadMetaRepository coverage
  including check_access owner logic, list_by_owner pagination
- Extended test_run_repository.py (+4 tests) — completion preserves fields,
  list ordering desc, limit, owner_none returns all
- Extended test_run_journal.py (+8 tests) — on_chain_error, track_tokens=false,
  middleware no ai_message, unknown caller tokens, convenience fields,
  tool_error, non-summarization custom event
- Extended test_run_event_store.py (+7 tests) — DB batch seq continuity,
  make_run_event_store factory (memory/db/jsonl/fallback/unknown)
- Extended test_phase2b_integration.py (+4 tests) — create_or_reject persists,
  follow-up metadata, summarization in history, full DB-backed lifecycle
- Fixed DB integration test to use proper fake objects (not MagicMock)
  for JSON-serializable metadata
- 157 total Phase 2 tests pass, zero regressions

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* config: move default sqlite_dir to .deer-flow/data

Keep SQLite databases alongside other DeerFlow-managed data
(threads, memory) under the .deer-flow/ directory instead of a
top-level ./data folder.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* refactor(persistence): remove UTFJSON, use engine-level json_serializer + datetime.now()

- Replace custom UTFJSON type with standard sqlalchemy.JSON in all ORM
  models. Add json_serializer=json.dumps(ensure_ascii=False) to all
  create_async_engine calls so non-ASCII text (Chinese etc.) is stored
  as-is in both SQLite and Postgres.
- Change ORM datetime defaults from datetime.now(UTC) to datetime.now(),
  remove UTC imports.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* refactor(gateway): simplify deps.py with getter factory + inline repos

- Replace 6 identical getter functions with _require() factory.
- Inline 3 _make_*_repo() factories into langgraph_runtime(), call
  get_session_factory() once instead of 3 times.
- Add thread_meta upsert in start_run (services.py).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat(docker): add UV_EXTRAS build arg for optional dependencies

Support installing optional dependency groups (e.g. postgres) at
Docker build time via UV_EXTRAS build arg:
  UV_EXTRAS=postgres docker compose build

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* refactor(journal): fix flush, token tracking, and consolidate tests

RunJournal fixes:
- _flush_sync: retain events in buffer when no event loop instead of
  dropping them; worker's finally block flushes via async flush().
- on_llm_end: add tool_calls filter and caller=="lead_agent" guard for
  ai_message events; mark message IDs for dedup with record_llm_usage.
- worker.py: persist completion data (tokens, message count) to RunStore
  in finally block.

Model factory:
- Auto-inject stream_usage=True for BaseChatOpenAI subclasses with
  custom api_base, so usage_metadata is populated in streaming responses.

Test consolidation:
- Delete test_phase2b_integration.py (redundant with existing tests).
- Move DB-backed lifecycle test into test_run_journal.py.
- Add tests for stream_usage injection in test_model_factory.py.
- Clean up executor/task_tool dead journal references.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat(events): widen content type to str|dict in all store backends

Allow event content to be a dict (for structured OpenAI-format messages)
in addition to plain strings. Dict values are JSON-serialized for the DB
backend and deserialized on read; memory and JSONL backends handle dicts
natively. Trace truncation now serializes dicts to JSON before measuring.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(events): use metadata flag instead of heuristic for dict content detection

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat(converters): add LangChain-to-OpenAI message format converters

Pure functions langchain_to_openai_message, langchain_to_openai_completion,
langchain_messages_to_openai, and _infer_finish_reason for converting
LangChain BaseMessage objects to OpenAI Chat Completions format, used by
RunJournal for event storage. 15 unit tests added.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(converters): handle empty list content as null, clean up test

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat(events): human_message content uses OpenAI user message format

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* feat(events): ai_message uses OpenAI format, add ai_tool_call message event

- ai_message content now uses {"role": "assistant", "content": "..."} format
- New ai_tool_call message event emitted when lead_agent LLM responds with tool_calls
- ai_tool_call uses langchain_to_openai_message converter for consistent format
- Both events include finish_reason in metadata ("stop" or "tool_calls")

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat(events): add tool_result message event with OpenAI tool message format

Cache tool_call_id from on_tool_start keyed by run_id as fallback for on_tool_end,
then emit a tool_result message event (role=tool, tool_call_id, content) after each
successful tool completion.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* feat(events): summary content uses OpenAI system message format

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat(events): replace llm_start/llm_end with llm_request/llm_response in OpenAI format

Add on_chat_model_start to capture structured prompt messages as llm_request events.
Replace llm_end trace events with llm_response using OpenAI Chat Completions format.
Track llm_call_index to pair request/response events.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat(events): add record_middleware method for middleware trace events

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* test(events): add full run sequence integration test for OpenAI content format

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* feat(events): align message events with checkpoint format and add middleware tag injection

- Message events (ai_message, ai_tool_call, tool_result, human_message) now use
  BaseMessage.model_dump() format, matching LangGraph checkpoint values.messages
- on_tool_end extracts tool_call_id/name/status from ToolMessage objects
- on_tool_error now emits tool_result message events with error status
- record_middleware uses middleware:{tag} event_type and middleware category
- Summarization custom events use middleware:summarize category
- TitleMiddleware injects middleware:title tag via get_config() inheritance
- SummarizationMiddleware model bound with middleware:summarize tag
- Worker writes human_message using HumanMessage.model_dump()

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat(threads): switch search endpoint to threads_meta table and sync title

- POST /api/threads/search now queries threads_meta table directly,
  removing the two-phase Store + Checkpointer scan approach
- Add ThreadMetaRepository.search() with metadata/status filters
- Add ThreadMetaRepository.update_display_name() for title sync
- Worker syncs checkpoint title to threads_meta.display_name on run completion
- Map display_name to values.title in search response for API compatibility

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* feat(threads): history endpoint reads messages from event store

- POST /api/threads/{thread_id}/history now combines two data sources:
  checkpointer for checkpoint_id, metadata, title, thread_data;
  event store for messages (complete history, not truncated by summarization)
- Strip internal LangGraph metadata keys from response
- Remove full channel_values serialization in favor of selective fields

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: remove duplicate optional-dependencies header in pyproject.toml

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(middleware): pass tagged config to TitleMiddleware ainvoke call

Without the config, the middleware:title tag was not injected,
causing the LLM response to be recorded as a lead_agent ai_message
in run_events.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix: resolve merge conflict in .env.example

Keep both DATABASE_URL (from persistence-scaffold) and WECOM
credentials (from main) after the merge.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(persistence): address review feedback on PR #1851

- Fix naive datetime.now() → datetime.now(UTC) in all ORM models
- Fix seq race condition in DbRunEventStore.put() with FOR UPDATE
  and UNIQUE(thread_id, seq) constraint
- Encapsulate _store access in RunManager.update_run_completion()
- Deduplicate _store.put() logic in RunManager via _persist_to_store()
- Add update_run_completion to RunStore ABC + MemoryRunStore
- Wire follow_up_to_run_id through the full create path
- Add error recovery to RunJournal._flush_sync() lost-event scenario
- Add migration note for search_threads breaking change
- Fix test_checkpointer_none_fix mock to set database=None

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* chore: update uv.lock

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(persistence): address 22 review comments from CodeQL, Copilot, and Code Quality

Bug fixes:
- Sanitize log params to prevent log injection (CodeQL)
- Reset threads_meta.status to idle/error when run completes
- Attach messages only to latest checkpoint in /history response
- Write threads_meta on POST /threads so new threads appear in search

Lint fixes:
- Remove unused imports (journal.py, migrations/env.py, test_converters.py)
- Convert lambda to named function (engine.py, Ruff E731)
- Remove unused logger definitions in repos (Ruff F841)
- Add logging to JSONL decode errors and empty except blocks
- Separate assert side-effects in tests (CodeQL)
- Remove unused local variables in tests (Ruff F841)
- Fix max_trace_content truncation to use byte length, not char length

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* style: apply ruff format to persistence and runtime files

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Potential fix for pull request finding 'Statement has no effect'

Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>

* refactor(runtime): introduce RunContext to reduce run_agent parameter bloat

Extract checkpointer, store, event_store, run_events_config, thread_meta_repo,
and follow_up_to_run_id into a frozen RunContext dataclass. Add get_run_context()
in deps.py to build the base context from app.state singletons. start_run() uses
dataclasses.replace() to enrich per-run fields before passing ctx to run_agent.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* refactor(gateway): move sanitize_log_param to app/gateway/utils.py

Extract the log-injection sanitizer from routers/threads.py into a shared
utils module and rename to sanitize_log_param (public API). Eliminates the
reverse service → router import in services.py.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* perf: use SQL aggregation for feedback stats and thread token usage

Replace Python-side counting in FeedbackRepository.aggregate_by_run with
a single SELECT COUNT/SUM query. Add RunStore.aggregate_tokens_by_thread
abstract method with SQL GROUP BY implementation in RunRepository and
Python fallback in MemoryRunStore. Simplify the thread_token_usage
endpoint to delegate to the new method, eliminating the limit=10000
truncation risk.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* docs: annotate DbRunEventStore.put() as low-frequency path

Add docstring clarifying that put() opens a per-call transaction with
FOR UPDATE and should only be used for infrequent writes (currently
just the initial human_message event). High-throughput callers should
use put_batch() instead.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(threads): fall back to Store search when ThreadMetaRepository is unavailable

When database.backend=memory (default) or no SQL session factory is
configured, search_threads now queries the LangGraph Store instead of
returning 503. Returns empty list if neither Store nor repo is available.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* refactor(persistence): introduce ThreadMetaStore ABC for backend-agnostic thread metadata

Add ThreadMetaStore abstract base class with create/get/search/update/delete
interface. ThreadMetaRepository (SQL) now inherits from it. New
MemoryThreadMetaStore wraps LangGraph BaseStore for memory-mode deployments.

deps.py now always provides a non-None thread_meta_repo, eliminating all
`if thread_meta_repo is not None` guards in services.py, worker.py, and
routers/threads.py. search_threads no longer needs a Store fallback branch.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* refactor(history): read messages from checkpointer instead of RunEventStore

The /history endpoint now reads messages directly from the
checkpointer's channel_values (the authoritative source) instead of
querying RunEventStore.list_messages(). The RunEventStore API is
preserved for other consumers.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(persistence): address new Copilot review comments

- feedback.py: validate thread_id/run_id before deleting feedback
- jsonl.py: add path traversal protection with ID validation
- run_repo.py: parse `before` to datetime for PostgreSQL compat
- thread_meta_repo.py: fix pagination when metadata filter is active
- database_config.py: use resolve_path for sqlite_dir consistency

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Implement skill self-evolution and skill_manage flow (#1874)

* chore: ignore .worktrees directory

* Add skill_manage self-evolution flow

* Fix CI regressions for skill_manage

* Address PR review feedback for skill evolution

* fix(skill-evolution): preserve history on delete

* fix(skill-evolution): tighten scanner fallbacks

* docs: add skill_manage e2e evidence screenshot

* fix(skill-manage): avoid blocking fs ops in session runtime

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

* fix(config): resolve sqlite_dir relative to CWD, not Paths.base_dir

resolve_path() resolves relative to Paths.base_dir (.deer-flow),
which double-nested the path to .deer-flow/.deer-flow/data/app.db.
Use Path.resolve() (CWD-relative) instead.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Feature/feishu receive file (#1608)

* feat(feishu): add channel file materialization hook for inbound messages

- Introduce Channel.receive_file(msg, thread_id) as a base method for file materialization; default is no-op.
- Implement FeishuChannel.receive_file to download files/images from Feishu messages, save to sandbox, and inject virtual paths into msg.text.
- Update ChannelManager to call receive_file for any channel if msg.files is present, enabling downstream model access to user-uploaded files.
- No impact on Slack/Telegram or other channels (they inherit the default no-op).

* style(backend): format code with ruff for lint compliance

- Auto-formatted packages/harness/deerflow/agents/factory.py and tests/test_create_deerflow_agent.py using `ruff format`
- Ensured both files conform to project linting standards
- Fixes CI lint check failures caused by code style issues

* fix(feishu): handle file write operation asynchronously to prevent blocking

* fix(feishu): rename GetMessageResourceRequest to _GetMessageResourceRequest and remove redundant code

* test(feishu): add tests for receive_file method and placeholder replacement

* fix(manager): remove unnecessary type casting for channel retrieval

* fix(feishu): update logging messages to reflect resource handling instead of image

* fix(feishu): sanitize filename by replacing invalid characters in file uploads

* fix(feishu): improve filename sanitization and reorder image key handling in message processing

* fix(feishu): add thread lock to prevent filename conflicts during file downloads

* fix(test): correct bad merge in test_feishu_parser.py

* chore: run ruff and apply formatting cleanup
fix(feishu): preserve rich-text attachment order and improve fallback filename handling

* fix(docker): restore gateway env vars and fix langgraph empty arg issue (#1915)

Two production docker-compose.yaml bugs prevent `make up` from working:

1. Gateway missing DEER_FLOW_CONFIG_PATH and DEER_FLOW_EXTENSIONS_CONFIG_PATH
   environment overrides. Added in fb2d99f (#1836) but accidentally reverted
   by ca2fb95 (#1847). Without them, gateway reads host paths from .env via
   env_file, causing FileNotFoundError inside the container.

2. Langgraph command fails when LANGGRAPH_ALLOW_BLOCKING is unset (default).
   Empty $${allow_blocking} inserts a bare space between flags, causing
   ' --no-reload' to be parsed as unexpected extra argument. Fix by building
   args string first and conditionally appending --allow-blocking.

Co-authored-by: cooper <cooperfu@tencent.com>

* fix(frontend): resolve invalid HTML nesting and tabnabbing vulnerabilities (#1904)

* fix(frontend): resolve invalid HTML nesting and tabnabbing vulnerabilities

Fix `<button>` inside `<a>` invalid HTML in artifact components and add
missing `noopener,noreferrer` to `window.open` calls to prevent reverse
tabnabbing.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix(frontend): address Copilot review on tabnabbing and double-tab-open

Remove redundant parent onClick on web_fetch ChainOfThoughtStep to
prevent opening two tabs on link click, and explicitly null out
window.opener after window.open() for defensive tabnabbing hardening.

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

* refactor(persistence): organize entities into per-entity directories

Restructure the persistence layer from horizontal "models/ + repositories/"
split into vertical entity-aligned directories. Each entity (thread_meta,
run, feedback) now owns its ORM model, abstract interface (where applicable),
and concrete implementations under a single directory with an aggregating
__init__.py for one-line imports.

Layout:
  persistence/thread_meta/{base,model,sql,memory}.py
  persistence/run/{model,sql}.py
  persistence/feedback/{model,sql}.py

models/__init__.py is kept as a facade so Alembic autogenerate continues to
discover all ORM tables via Base.metadata. RunEventRow remains under
models/run_event.py because its storage implementation lives in
runtime/events/store/db.py and has no matching repository directory.

The repositories/ directory is removed entirely. All call sites in
gateway/deps.py and tests are updated to import from the new entity
packages, e.g.:

    from deerflow.persistence.thread_meta import ThreadMetaRepository
    from deerflow.persistence.run import RunRepository
    from deerflow.persistence.feedback import FeedbackRepository

Full test suite passes (1690 passed, 14 skipped).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(gateway): sync thread rename and delete through ThreadMetaStore

The POST /threads/{id}/state endpoint previously synced title changes
only to the LangGraph Store via _store_upsert. In sqlite mode the search
endpoint reads from the ThreadMetaRepository SQL table, so renames never
appeared in /threads/search until the next agent run completed (worker.py
syncs title from checkpoint to thread_meta in its finally block).

Likewise the DELETE /threads/{id} endpoint cleaned up the filesystem,
Store, and checkpointer but left the threads_meta row orphaned in sqlite,
so deleted threads kept appearing in /threads/search.

Fix both endpoints by routing through the ThreadMetaStore abstraction
which already has the correct sqlite/memory implementations wired up by
deps.py. The rename path now calls update_display_name() and the delete
path calls delete() — both work uniformly across backends.

Verified end-to-end with curl in gateway mode against sqlite backend.
Existing test suite (1690 passed) and focused router/repo tests pass.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* refactor(gateway): route all thread metadata access through ThreadMetaStore

Following the rename/delete bug fix in PR1, migrate the remaining direct
LangGraph Store reads/writes in the threads router and services to the
ThreadMetaStore abstraction so that the sqlite and memory backends behave
identically and the legacy dual-write paths can be removed.

Migrated endpoints (threads.py):
- create_thread: idempotency check + write now use thread_meta_repo.get/create
  instead of dual-writing the LangGraph Store and the SQL row.
- get_thread: reads from thread_meta_repo.get; the checkpoint-only fallback
  for legacy threads is preserved.
- patch_thread: replaced _store_get/_store_put with thread_meta_repo.update_metadata.
- delete_thread_data: dropped the legacy store.adelete; thread_meta_repo.delete
  already covers it.

Removed dead code (services.py):
- _upsert_thread_in_store — redundant with the immediately following
  thread_meta_repo.create() call.
- _sync_thread_title_after_run — worker.py's finally block already syncs
  the title via thread_meta_repo.update_display_name() after each run.

Removed dead code (threads.py):
- _store_get / _store_put / _store_upsert helpers (no remaining callers).
- THREADS_NS constant.
- get_store import (router no longer touches the LangGraph Store directly).

New abstract method:
- ThreadMetaStore.update_metadata(thread_id, metadata) merges metadata into
  the thread's metadata field. Implemented in both ThreadMetaRepository (SQL,
  read-modify-write inside one session) and MemoryThreadMetaStore. Three new
  unit tests cover merge / empty / nonexistent behaviour.

Net change: -134 lines. Full test suite: 1693 passed, 14 skipped.
Verified end-to-end with curl in gateway mode against sqlite backend
(create / patch / get / rename / search / delete).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: Copilot Autofix powered by AI <223894421+github-code-quality[bot]@users.noreply.github.com>
Co-authored-by: DanielWalnut <45447813+hetaoBackend@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
Co-authored-by: JilongSun <965640067@qq.com>
Co-authored-by: jie <49781832+stan-fu@users.noreply.github.com>
Co-authored-by: cooper <cooperfu@tencent.com>
Co-authored-by: yangzheli <43645580+yangzheli@users.noreply.github.com>
 2026-04-26 11:05:47 +08:00