mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
32 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> |
||
|
Ryker_Feng
|
fcbf0609b0
|
feat(chat): edit and rerun latest user turn (#4377)
Co-authored-by: Willem Jiang <willem.jiang@gmail.com> |
||
|
Huixin615
|
090e80c1dd
|
fix(runtime): fail-stop runs when lease ownership cannot be confirmed (#4431)
* fix(runtime): fail-stop runs after lease expiry * test(runtime): cover late successful lease renewal --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com> |
||
|
DanielWalnut
|
bb9f67aaf1
|
fix(runtime): close cancelled replacement admission (#4472) | ||
|
Vanzeren
|
1c7531242c
|
feat(runtime): record terminal artifact delivery receipts (slice 1 of #4272) (#4365)
* feat(runtime): record terminal artifact delivery receipts (#4272) * fix(runtime): persist delivery receipts across recovery * test(runtime): cover delivery receipt invariants * fix(runtime): preserve terminal status on receipt outages |
||
|
MiaoRuidx
|
735f67a5b2
|
fix: guard pending run startup cancellation (#4450)
* fix: guard pending run startup cancellation * fix(run): address startup review feedback * fix(run): narrow start_run store contract --------- Co-authored-by: MiaoRuidx <12540796+MiaoRuidx@users.noreply.github.com> Co-authored-by: Willem Jiang <willem.jiang@gmail.com> |
||
|
Huixin615
|
8af760fc30
|
fix(runtime): make orphan reconciliation lease-aware (#4427) | ||
|
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 |
||
|
H Haidong
|
c7538cfb35
|
fix(runs): terminate orphaned streams after lease recovery (#4420)
* fix(runs): terminate orphaned streams after lease recovery * fix(runs): include recovered ids in callback warnings * fix(runs): harden orphan recovery lifecycle |
||
|
MiaoRuidx
|
80c06414f8
|
fix: make orphan reconciliation lease-aware (#4434)
让启动/孤儿 run 恢复在最终写入前通过 claim_for_takeover 原子重查 lease，避免 owner 在扫描后续约成功仍被误标为 error。 补充扫描后续约的回归测试，并把 reconciliation 写失败测试迁移到 takeover claim 路径。 Co-authored-by: MiaoRuidx <12540796+MiaoRuidx@users.noreply.github.com> Co-authored-by: Willem Jiang <willem.jiang@gmail.com> |
||
|
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 |
||
|
AnoobFeng
|
446fa03801
|
fix(context): resolve context compress bug (#4065)
* fix(runtime): persist original human input outside model sanitization * refactor(history): load thread messages by global event sequence * fix(frontend): make summarization rescue a transient history bridge * fix(frontend): old message not append tail 1. add identity anchor 2. add bridgeOrder * fix(frontend): lint error fix * fix: address review feedback and harden pagination coverage - defer transient history ref writes until after render commit - cover large middleware-only history scans - verify infinite-query refetch recalculates page cursors - document AI event types and anchor-weaving differences * fix: harden message pagination and enrichment - append unmatched live tails after canonical history - warn and stop when pagination has_more lacks a cursor - deep-copy restored UI messages to isolate model-facing content - log invalid event sequence and non-advancing cursor errors - pass user_id explicitly through event-store history queries - cover middleware-only AI runs across memory, JSONL, and DB stores * fix: address pagination review feedback * fix(frontend): checkpoint has unknow redener content, optimize the anchor policy * fix(frontend): unit test issue missed previously, remove the TanStack cache trimming * fix(gateway): harden message history queries and provenance - reject externally forged original_user_content metadata - validate provenance metadata in upload and sanitization middleware - make run lookups fail closed by default - batch feedback queries by run ID - align memory message filtering with persistent stores |
||
|
heart-scalpel
|
b53c1ae0e0
|
fix(runs): cancel degrades to lease takeover for multi-worker (#4064)
* fix(runs): cancel degrades to lease takeover for multi-worker Work item 4 of the multi-worker ownership epic (https://github.com/bytedance/deer-flow/issues/3948). Problem: POST /runs/{run_id}/cancel landing on a non-owning worker returns 409 — the cancel button silently fails under GATEWAY_WORKERS>1 with no sticky routing. cancel() required the current worker to hold the in-memory task/abort_event, which any non-owner pod cannot satisfy. Changes: - RunManager.cancel() returns CancelOutcome enum (cancelled / taken_over / lease_valid_elsewhere / not_active_locally / not_cancellable / unknown) instead of bool, so the router can map each outcome to the right HTTP response. - New store primitive claim_for_takeover(): a single atomic conditional UPDATE that marks a run as error only when status IN (pending, running) AND (lease IS NULL OR lease < now - grace). Closes the stale-read / concurrent-heartbeat race — if the owner renews between our read and write, the UPDATE matches 0 rows and we surface lease_valid_elsewhere. - HTTP cancel + stream-join endpoints route on CancelOutcome: cancelled -> 202 (or 204 with wait=true); taken_over -> 202 immediately (no SSE streaming — the run is terminal on another worker, streaming would hang); lease_valid_elsewhere -> 409 + Retry-After header computed from lease_expires_at + grace_seconds. - RunManager.grace_seconds exposed as a public property; the router no longer reaches into _run_ownership_config. - _is_lease_expired extracted to a module-level function, shared by RunManager.cancel() and MemoryRunStore.claim_for_takeover(). - GATEWAY_WORKERS=1 + heartbeat_enabled=false is zero-regression: the non-local path short-circuits to not_active_locally, preserving the original 409 behaviour the existing tests pin. Tests: 12 new (5 store primitive + 4 cancel-takeover unit + 3 HTTP including a regression guard verifying POST /stream?action=interrupt on a dead-owner run returns 202 instead of hanging on SSE). 244 directly-related tests pass; 36/36 blocking-IO gate pass. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> * fix(runs): guard update_status and self-terminate on takeover Two defenses close a split-brain window where the original owner could overwrite a peer's takeover status: - update_status (SQL + memory store) now guards on status IN ('pending','running'). When takeover already set the row to 'error', the owner's final status write matches 0 rows and is dropped. - _persist_status: when update_status returns False, check whether the row exists before attempting recovery via put(). If the row exists (takeover by another worker), skip recovery instead of blindly upserting over the takeover. - Heartbeat _renew_leases: when update_lease returns False (row no longer pending/running or owner changed), cancel the local task so wasted CPU is bounded to the next heartbeat tick (~10s) instead of the full task lifetime. Also fix three reviewer feedback items: - Re-fetch the store row when cancel() returns lease_valid_elsewhere, so Retry-After uses the owner's freshly-renewed lease instead of a stale value from request start. - Fallback 'unknown' in takeover error message when owner_worker_id is NULL (pre-ownership data). - Remove dead else-10 branch from grace_seconds property (unreachable — all callers are downstream of the heartbeat_enabled guard). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> * test(runs): pin split-brain defences from update_status guard + heartbeat Three tests lock down the takeover authoritativeness so a late-running owner cannot overwrite a peer's claim: - update_status must reject writes when the store row is already terminal (taken over by another worker). - _persist_status must skip row-recovery via put() when the row exists but has been taken over. - Heartbeat _renew_leases must cancel the local task when update_lease returns False (row claimed by another worker). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> * fix(runs): precise outcome + log when local cancel loses to peer takeover Two reviewer precision nits on the split-brain defence: - _persist_status: branch the skip-reason log on existing["status"]. error → WARNING "peer takeover" (anomalous); interrupted/success → INFO "local cancel/completion race" (expected when user hits stop as the run finishes). Stops noisy false-positive takeover warnings in operator logs. - cancel() local path: when _persist_status returns False, re-check the store. If a peer's claim_for_takeover flipped the row to error between our in-memory cancel and the guarded update_status, surface taken_over instead of cancelled so the client sees a status consistent with the store. Test: test_cancel_returns_taken_over_when_peer_claims_during_local_cancel pins the race outcome. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> * fix(runs): widen update_status guard, de-duplicate lease helpers, add coverage Round 3 of reviewer feedback: - Widen update_status guard to status IN ('pending','running','interrupted'). The original guard blocked interrupted→error (the rollback finalize path), losing the "Rolled back by user" message. interrupted is now permitted while error/success stay locked — takeover protection unchanged. - claim_for_takeover False now re-reads the store row to distinguish causes: owner renewed lease → lease_valid_elsewhere; row went terminal → not_cancellable; another worker already took it over → taken_over. - Extract _raise_lease_valid_elsewhere() helper to de-duplicate the 409+Retry-After block shared across cancel_run and stream_existing_run. - Extract _lease_expired_or_null() in persistence/run/sql.py to de-duplicate the lease-expiry SQL WHERE clause shared by claim_for_takeover and list_inflight_with_expired_lease. - 11 new tests: 5 SQL-layer claim_for_takeover (expired/valid/NULL/ terminal/nonexistent), 3 _compute_retry_after unit (NULL/unparseable/ normal), 2 claim re-read precision (terminal/takeover), 1 stream endpoint 409+Retry-After. Not addressed (non-blocking, reviewer agreed): - The 2–3 store.gets in the takeover cold path: optimizing the API to accept a pre-fetched record would couple the router to the manager more tightly than justified by the perf gain. - The lease-expiry inline loop in MemoryRunStore.list_inflight_with_- expired_lease pre-computes cutoff once for all rows; switching to the shared _is_lease_expired helper would recompute datetime.now() per row with no real benefit. 260 related tests pass; 36/36 blocking-IO gate pass; ruff clean. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> * fix(runs): de-duplicate lease-expiry helper, restore defensive fallback Address final round of review feedback: - Extract is_lease_expired to deerflow.utils.time (no _ prefix, public utility). Manager and MemoryRunStore now import from the same place instead of the store reaching backward into the manager for a private function. - Restore defensive else-10 fallback in grace_seconds property (removed in an earlier round). The guard is unreachable for current callers but protects future ones from AttributeError. - Comment the transient in-memory interrupted vs store error state when a local cancel is superseded by a peer takeover. - Comment the max(1, ...) floor in _compute_retry_after — the floor is a lower bound, not a poll interval; clients should apply jitter. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com> Co-authored-by: rayhpeng <rayhpeng@gmail.com> |
||
|
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> |
||
|
xiawiie
|
2e15e3fe0d
|
fix: generate fallback title for interrupted first-turn runs (#3874)
* fix: generate title for interrupted first turn * test(title): cover partial-exchange + dict-form messages Harden the interrupted-run fallback path added in 19fc34fd: - TitleMiddleware._should_generate_title now accepts a lone first-turn user message when allow_partial_exchange=True, so the worker can still derive a title if cancellation lands before any AI chunk is checkpointed. - runtime/runs/worker._ensure_interrupted_title computes the next checkpoint step defensively (treat missing/non-int step as 0) and renames a shadowed ckpt_config local for readability. - Add four unit tests in tests/test_title_middleware_core_logic.py: partial-exchange allows user-only, partial-exchange still respects an existing title, dict-form messages are recognized, and the sync fallback path derives a title from dict-form messages — matching what channel_values stores in the checkpoint. Refs #3859. * fix: persist interrupted-title via channel_versions bump Address PR #3874 review feedback: ``_ensure_interrupted_title`` previously called ``aput(..., new_versions={})``. LangGraph's DB-backed savers (``PostgresSaver`` and the v4 ``SqliteSaver`` blob layout) strip inline ``channel_values`` from ``put`` and only persist blobs for channels named in ``new_versions`` — so the fallback ``title`` channel was dropped on read-back and ``threads_meta.display_name`` stayed ``"Untitled"`` after refresh on those backends. The original in-memory e2e passed because ``InMemorySaver`` keeps the inline snapshot verbatim. Fix mirrors ``_rollback_to_pre_run_checkpoint`` in the same file: bump ``channel_versions["title"]`` (via ``checkpointer.get_next_version`` when available, else int/string fallbacks), persist the new version on the checkpoint, and declare it in ``new_versions`` so the DB savers actually write the blob. Regression coverage in ``tests/test_run_worker_rollback.py``: - ``test_ensure_interrupted_title_bumps_channel_version_and_declares_it_in_new_versions`` — exact ``aput`` invariants: ``new_versions == {"title": 1}``, the written checkpoint's ``channel_versions["title"]`` is bumped, and the pre-existing ``messages`` version is preserved. - ``test_ensure_interrupted_title_bumps_existing_string_version`` — string-shaped prior version (some savers use UUID-style versions); bumped value must differ from the prior, no overwrite-in-place. - ``test_ensure_interrupted_title_skips_when_title_already_set`` — title short-circuit; no extra ``aput``. - ``test_ensure_interrupted_title_returns_none_when_no_checkpoint`` — no checkpoint yet; returns ``None`` without writing. - ``test_ensure_interrupted_title_round_trip_with_real_sqlite_checkpointer`` — full round-trip against a real ``AsyncSqliteSaver`` on a disk-backed DB, then closes and re-opens the saver to simulate a fresh connection. The fallback title must still be present on the second ``aget_tuple``. This is the exact scenario the review flagged. Validated locally with the full backend suite: 5195 passed, 18 skipped. Refs #3859. Addresses review on #3874. * test(worker, title): harden interrupted-title fallback for every saver Defensive coverage on top of the channel_versions fix (commit 05253957), addressing edge cases surfaced during a second-pass review of #3874. Worker: - Extract version bump into ``_bump_channel_version(checkpointer, current)`` with explicit fallbacks for int / float / numeric-string / UUID-shaped string / None / bool, AND a wrap-around defense when the saver's ``get_next_version`` raises or returns an unchanged value. The invariant is: returned version MUST differ from the prior. Without this, a saver bug (or a custom backend) could leave ``new_versions={"title": v}`` no-op on DB savers — the very class of bug the original review pointed out. Title middleware: - Coerce ``state.get("messages")`` from ``None`` to ``[]`` on both ``_should_generate_title`` and ``_build_title_prompt``. A partially-initialized checkpoint can carry ``messages=None`` on the channel_values channel (the worker reads raw channel_values, not BaseMessages), and the default kwarg only protects against a missing key. Repro: ``TypeError: 'NoneType' object is not iterable`` from the next() generator — confirmed by reverting the fix and watching ``test_*_handles_none_messages_channel`` go red. Tests (TDD-verified red→green for the new asserts): - ``test_run_worker_rollback.py``: * ``_bump_channel_version`` — 8 tests covering every version type (int, float, numeric string, UUID-style string, None, bool) and every saver-side fault mode (no ``get_next_version`` / raising / stuck on identity). * ``test_ensure_interrupted_title_*`` — 5 additional helper boundary tests: title.enabled=false short-circuit; empty messages list; messages=None; aput-error propagation (helper contract: caller swallows, not the helper); idempotency on a real InMemorySaver across two invocations. * ``test_ensure_interrupted_title_preserves_non_title_channel_versions`` — pins that ``new_versions`` only contains ``"title"`` and that other channels' versions are untouched (regression anchor for a sloppier draft that bumped every channel). * ``test_worker_finally_block_swallows_helper_exceptions`` — pins the integration contract: even if the helper raises, the worker's threads_meta status sync still runs and ``publish_end`` is still awaited so the SSE stream closes cleanly. - ``test_title_middleware_core_logic.py``: * 4 additional tests: ``messages=None`` on both ``_should_generate_title`` and ``_build_title_prompt``; the ``role: user`` / ``role: assistant`` (OpenAI-style) dict normalization; partial-exchange path with a dict-form message. Verification: - ``PYTHONPATH=. uv run pytest tests/ -x --ignore=tests/blocking_io -q`` → 5215 passed, 18 skipped. - ``ruff check`` + ``ruff format --check`` clean on every touched file. - Red/green TDD verification: temporarily reverted the ``new_versions={}`` fix → 4 new tests went red as expected; restored and the suite is green again. Same red/green dance for the ``messages=None`` coercion. Refs #3859. Addresses second-pass review on #3874. * fix(title): ignore dict context reminders in fallback * fix(worker): link interrupted-title checkpoint to its parent The title-bump checkpoint written by ``_ensure_interrupted_title`` was landing without a ``parent_checkpoint_id`` — a real orphan in the LangGraph history graph. Reproduction (disk-backed AsyncSqliteSaver): [seed] checkpoint_id = 1f173dbc... [helper] wrote title = "Why is the sky blue?" [issue 1] new checkpoint = 1f173dbc..., parent = None [issue 1] is new checkpoint orphaned? True Root cause: ``_ensure_interrupted_title`` built ``write_config`` as ``{"thread_id": ..., "checkpoint_ns": ...}`` only. ``BaseCheckpointSaver`` implementations read ``configurable.checkpoint_id`` from that config as the *parent* id when inserting (see ``langgraph/checkpoint/sqlite/aio.py`` ``aput``: ``config["configurable"].get("checkpoint_id")`` becomes the ``parent_checkpoint_id`` column). With no value, the saver writes NULL — the new checkpoint is a tree root. Consequences: - Any future LangGraph ``runs.resume_from`` / time-travel feature has no backward edge to walk past the title-bump. - History-visualization UIs built on ``alist()`` render the title-bump as a sibling of the prior checkpoint, not its descendant. Fix: read ``checkpoint_id`` off the tuple's own config and thread it into ``write_config["configurable"]["checkpoint_id"]`` before calling ``aput``, the same pattern every middleware-driven write uses. Three new regression tests against real ``AsyncSqliteSaver`` (disk-backed, fresh connections so we exercise the on-disk read path): - ``test_ensure_interrupted_title_links_new_checkpoint_to_its_parent`` — asserts ``latest.parent_config["configurable"]["checkpoint_id"]`` equals the seeded checkpoint id. TDD red-green verified: reverting the fix flips this test red with ``AssertionError: title-bump checkpoint must have a parent_config``. - ``test_ensure_interrupted_title_appears_in_history_with_audit_marker`` — pins the audit contract: the title-bump entry in ``alist()`` carries ``metadata.source == "update"`` and ``metadata.writes`` contains ``runtime_interrupt_title``. This is a deliberate design choice — we do NOT hide the entry from history (audit trail belongs in the saver), but its source and writes marker MUST be unambiguous so UIs/tools can identify it. - ``test_ensure_interrupted_title_survives_immediate_next_turn`` — cancel → immediate user follow-up scenario. Simulates the agent's next turn appending a (user, ai) pair without touching the title channel, then opens a fresh saver and verifies the title is still present after the next-turn checkpoint write. Pins the channel-version-blob invariant established by commit 05253957 — without the ``new_versions={"title": v}`` declaration there, the title blob would vanish from the DB and this test would read back ``None``. Verification: - ``PYTHONPATH=. uv run pytest tests/ -x --ignore=tests/blocking_io -q`` → 5222 passed, 15 skipped. - ``ruff check`` + ``ruff format --check`` clean on every touched file. - Reproduction script confirms ``parent_checkpoint_id`` is now non-null and the next-turn read-back preserves the fallback title. Refs #3859. * Revert "fix(worker): link interrupted-title checkpoint to its parent" This reverts commit c763ed9334781db1acdce0f5f33d663d8d5f80ff. * test: trim over-engineered test coverage Reduce review surface area on PR #3874 by dropping defensive tests that don't pin a real invariant. After self-review: - ``_bump_channel_version``: 8 tests → 2 (happy path + saver-error fallback). Dropped float / bool / numeric-string / UUID-string / missing-get-next-version / stuck-get-next-version branches — those are speculative scaffolding for savers we don't ship. - ``_ensure_interrupted_title``: dropped ``returns_none_when_title_disabled``, ``returns_none_with_no_user_message``, ``returns_none_when_no_checkpoint`` — boundary guards already exercised by the e2e test and the ``handles_none_messages_channel`` regression anchor. Net: -107 lines of test code. Remaining coverage still pins every red-green-verified invariant (channel_versions bump, string-version bump, idempotency, sqlite round-trip, non-title channel preservation, aput-error contract, worker finally swallowing, partial-exchange). Verification: 5209 passed, 15 skipped. * fix: harden interrupted title finalization * fix: serialize interrupted title finalization * fix: preserve interrupt semantics during title finalization * fix: preserve delayed interrupted title recovery --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com> |
||
|
AnoobFeng
|
e7a03e5243
|
fix(gateway): attribute token usage to actual models (#3658)
* fix(gateway): attribute token usage to actual models
Capture per-call model names from LLM response metadata for lead, middleware, and subagent calls.
Persist a per-run token_usage_by_model breakdown and aggregate by that map in both SQL and memory stores, with legacy fallback to the run-level model_name for older rows.
Add regression coverage for by_model totals, caller consistency, active progress snapshots, store parity, and SubagentTokenCollector model propagation.
* fix(gateway): harden by-model token aggregation
Use usage.get("total_tokens", 0) when reducing per-model token usage maps so aggregation tolerates partially written or manually edited JSON blobs without changing behavior for journal-written rows.
* docs(gateway): clarify by-model run count semantics
Document that by_model[*].runs counts the number of runs in which a model appeared, so multi-model runs can increment multiple model buckets.
|
||
|
ly-wang19
|
0d3bfe0a76
|
perf(runtime): index runs by thread_id to avoid O(n) scans in RunManager (#3499)
* perf(runtime): index runs by thread_id to avoid O(n) scans in RunManager RunManager.list_by_thread, create_or_reject (inflight check), and has_inflight each filtered every in-memory run by thread_id — an O(total in-memory runs) scan that grows with overall gateway traffic rather than the queried thread's depth. Add a thread_id -> run_ids secondary index (an insertion-ordered dict used as an ordered set) maintained in lockstep with _runs under the existing lock at every add/remove site (create, create_or_reject, both rollbacks, cleanup). The three per-thread queries now run in O(runs-in-thread); insertion order is preserved so list_by_thread keeps stable tie-breaking. Behavior unchanged. Adds 6 regression tests; full RunManager suite 146 passed. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * test(runtime): cover create_or_reject rollback + clarify thread-index guard docstrings Address review on #3499 (fancyboi999): - Reword _thread_records_locked docstring: lockstep under self._lock is the correctness guarantee; self._runs.get is one-directional defense-in-depth (drops stale ids, cannot recover index-missing ids), not reconciliation. - Add test_failed_create_or_reject_unindexes_run covering the create_or_reject rollback/unindex mutation site (the last untested mutation path). - Fix _FailingPutRunStore docstring ("initial put" -> "every put"). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: ly-wang19 <ly-wang19@users.noreply.github.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> |
||
|
DanielWalnut
|
aa015462a7
|
feat(im): Add user-owned IM channel connections (#3487)
* Add user-owned IM channel connections
* Fix dev startup and channel connect popup
* Use async channel connect flow
* Harden dev service daemon startup
* Support local IM channel connections
* Align IM connections with local channels
* Fix safe user id digest algorithm
* Address Copilot IM channel feedback
* Address IM channel review comments
* Support all integrated IM channel connections
* Format additional channel connection tests
* Keep unavailable channel connect buttons clickable
* Fix IM channel provider icons
* Add runtime setup for enabled IM channels
* Guard global shortcut key handling
* Keep configured IM channels editable
* Avoid password autofill for channel secrets
* Make channel threads visible to connection owners
* Persist IM runtime config locally
* Allow disconnecting runtime IM channels
* Route no-auth channel sessions to local user
* Use default user for auth-disabled local mode
* Show IM channel source on threads
* Prefill IM channel runtime config
* Reflect IM channel runtime health
* Ignore Feishu message read events
* Ignore Feishu non-content message events
* Let setup wizard enable IM channels
* Fix frontend formatting after merge
* Stabilize backend tests without local config
* Isolate channel runtime config tests
* Address channel connection review comments
* Use sha256 user buckets with legacy migration
* Ensure runtime IM channels are ready after restart
* Persist disconnected IM channel state
* Address channel connection review comments
* Address channel connection review findings
Frontend connect flow:
- Open the runtime-config dialog only when a provider still needs
credentials; configured providers go straight to the connect flow, so
the binding-code/deep-link path is reachable from the UI again.
- After saving credentials, continue into the connect flow when a user
binding is still required (multi-user mode) instead of stopping at a
"Connected" toast.
- Extract shared provider-state helpers to core/channels/provider-state
and add unit + e2e coverage for the direct-connect and
configure-then-connect paths.
Provider status semantics:
- Report connection_status from the user's newest connection row;
with no binding it is not_connected, except in auth-disabled local
mode where a configured running channel is effectively connected.
Concurrency and event-loop correctness:
- Offload ChannelRuntimeConfigStore construction and writes, channel
service construction, and Slack connection replies to threads; add a
tests/blocking_io/ anchor for the runtime-config handlers.
- Consume binding codes with a conditional UPDATE so a code can only be
used once under concurrent workers; retry upsert_connection as an
update when a concurrent insert wins the unique constraint.
- Serialize ensure_channel_ready per channel so concurrent provider
polls cannot double-start a channel worker.
Config and migration hardening:
- Stop mutating the get_app_config()-cached Telegram provider config;
the runtime store now owns the UI-entered bot username.
- Register channel_connections in STARTUP_ONLY_FIELDS with the
standardized startup-only Field description.
- Match the legacy unsafe-id bucket by recomputing its exact SHA-1 name
so another user's same-prefix bucket can never be migrated.
- Remove the unused Telegram process_webhook_update path and document
src/core/channels in the frontend docs.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
* Address PR review comments on authz scoping and channel runtime
Security (review feedback from ShenAC-SAC):
- Scope internal-token callers to the connection owner carried in
X-DeerFlow-Owner-User-Id instead of bypassing owner checks outright,
in both require_permission(owner_check=True) and the stateless run
endpoints. Internal callers keep access to their own and
shared/legacy threads, and may claim a default-owned channel thread
for its real owner, but a leaked internal token no longer grants
cross-user thread access.
- Require admin privileges for POST/DELETE /api/channels/{provider}/
runtime-config: runtime credentials and channel workers are
instance-wide shared state (same model as the MCP config API).
Read-only provider listing stays available to all users.
Performance (review feedback from willem-bd):
- Skip the redundant thread channel-metadata PATCH after the first
successful backfill per thread.
- Reuse the per-connection Slack WebClient until its token changes
instead of constructing one per outbound message.
- Reconcile channel readiness for all providers concurrently in
GET /api/channels/providers.
Also resolve the code-quality unused-import flag in the blocking-io
anchor by pre-importing the channel service via importlib.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
* Fix prettier formatting in provider-state test
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
* Reconcile UI runtime channel config with config reload on restart
Main now reloads a channel's config.yaml entry on restart_channel()
(#3514, issue #3497). Adapt the user-owned connection flow to coexist:
- configure_channel() restarts with reload_config=False — the caller
just supplied the authoritative config (browser-entered credentials
that are never written to config.yaml), so a file reload must not
clobber it with the stale on-disk entry.
- _load_channel_config() re-applies the UI runtime-store overlay used
at startup, so an operator-triggered restart keeps browser-entered
credentials for channels without a config.yaml entry and does not
resurrect a channel disconnected from the UI.
- Offload the reload's disk IO (config.yaml + runtime store) with
asyncio.to_thread, matching the blocking-IO policy on this branch.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---------
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
|
||
|
Xinmin Zeng
|
268fdd6968
|
fix(gateway): drain in-flight runs before closing checkpointer on shutdown (#3381)
* fix(gateway): drain in-flight runs before closing checkpointer on shutdown Chat runs execute in fire-and-forget background asyncio tasks that write checkpoints through a shared checkpointer. On shutdown, langgraph_runtime's AsyncExitStack tore down the checkpointer's postgres connection pool while those run tasks were still mid-graph. langgraph's AsyncPregelLoop._checkpointer_put_after_previous then ran its `finally: await checkpointer.aput(...)` against the closed pool, raising psycopg_pool.PoolClosed. Because that put runs in a langgraph-internal task (not on run_agent's call stack), run_agent's try/except cannot catch it and it surfaces as "unhandled exception during asyncio.run() shutdown". Add RunManager.shutdown() to cancel and bounded-await all in-flight runs, and call it from langgraph_runtime BEFORE the AsyncExitStack closes the checkpointer, so the final checkpoint write lands while the pool is still open. The drain is bounded by a timeout so a stuck run cannot hang worker shutdown, and is shielded so a second shutdown signal cannot abandon it mid-drain and reopen the race. Closes #3373 * fix(gateway): address review — preserve completed-run status, bound drain persistence Addresses Copilot review on #3381: - RunManager.shutdown(): decide run status AFTER the drain. Under the lock it now only requests cancellation; after asyncio.wait it marks/persists `interrupted` only for runs still pending or ended cancelled. A run that completes (e.g. `success`) during the drain window keeps its real terminal status instead of being unconditionally overwritten. - Bound the trailing status persistence within the timeout budget (deadline = loop.time()+timeout; gather wrapped in asyncio.wait_for) so a slow store backing off under DB pressure cannot push shutdown past the deadline. - deps: use asyncio.create_task instead of asyncio.ensure_future. - tests: wait deterministically for the run to be in-flight (poll the first checkpoint) instead of a fixed sleep; init shutdown_calls explicitly in the recovery test double; add regression test asserting a run completing during the drain keeps its status (in memory and in the store). * fix(gateway): address maintainer review — surface failed drain persists, clarify timeout constant Addresses @WillemJiang review on #3381: - shutdown(): inspect the gather result of the trailing interrupted-status persistence. _persist_status is best-effort (it catches + logs its own failure with exc_info and returns False, so it never raises out of the gather), but the aggregate result was never checked — a partial failure had no shutdown-level visibility. Now any escaped Exception is logged, and any False (a persist that did not confirm) is logged with the run_id. Added regression test test_shutdown_surfaces_failed_interrupted_persist. - deps: clarify the _RUN_DRAIN_TIMEOUT_SECONDS comment — state the actual value of _SHUTDOWN_HOOK_TIMEOUT_SECONDS (5.0s) and that both count toward the lifespan shutdown window. Kept as two separate constants (independent teardown steps that may diverge) rather than one shared "must match" value. - Verified no other test fake needs the shutdown stub: _FakeRunManager in test_worker_langfuse_metadata.py is a run_agent() argument (worker path), never injected into langgraph_runtime, so it never receives shutdown(). |
||
|
rayhpeng
|
0fb05825a2
|
fix(runtime): make run creation persistence atomic (#3152)
* fix runtime run creation persistence atomicity * fix run creation cancellation rollback * fix run manager test cleanup await * clarify run creation rollback on cancellation * document new run persistence rollback boundary --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com> |
||
|
AochenShen99
|
66d6a6a4e8
|
fix: harden run finalization persistence (#3155)
* fix: harden run finalization persistence * style: format gateway recovery test * fix: align run repository return types * fix: harden completion recovery follow-up |
||
|
Lawrance_YXLiao
|
2eeb597985
|
fix(runs): expose active progress counters (#3148)
* fix(runs): expose active progress counters * fix(runs): avoid delayed progress flush on completion * fix(runs): tighten progress snapshot semantics * fix(runs): preserve omitted progress fields * chore(runs): remove duplicate journal initialization |
||
|
john lee
|
9b19cca91c
|
fix(runtime): make RunManager.cancel() idempotent for already-interrupted runs (#3055) (#3058)
A second cancel() call on an interrupted run returned False, causing the cancel and stream_existing_run router endpoints to raise 409 on double-stop. Fix: return True inside the lock when record.status == RunStatus.interrupted. This covers both the POST /cancel and POST /join endpoints without any re-fetch or extra get() call — the idempotency lives at the source. Also fixes stream_existing_run (the LangGraph SDK stop-button path), which had the identical cancel() → 409 pattern and was not covered by the original PR. Both endpoints share the fix automatically. Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com> |
||
|
He Wang
|
c810e9f809
|
fix(harness)!: hydrate runs from RunStore and persist interrupted status (#2932)
* fix(harness): hydrate run history from RunStore and persist cancellation status fix: - Make RunManager.get() async and hydrate from RunStore when in-memory record is missing - Merge store rows into list_by_thread() with in-memory precedence for active runs - Persist interrupted status to RunStore in cancel() and create_or_reject(interrupt|rollback) - Extract _persist_status() to reuse the best-effort store update pattern - Await run_mgr.get() in all gateway endpoints - Return 409 with distinct message for store-only runs not active on current worker Closes #2812, Closes #2813 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> * fix(harness): consistent sort and guarded hydration in RunManager fix: - list_by_thread() now sorts by created_at desc (newest first) even when no RunStore is configured, matching the store-backed code path - guard _record_from_store() call sites in get() and list_by_thread() with best-effort error handling so a single malformed store row cannot turn read paths into 500s test: - update test_list_by_thread assertion to expect newest-first order - seed MemoryRunStore via public put() API instead of writing to _runs * fix(harness): guard store-only runs from streaming and fix get() TOCTOU Add RunRecord.store_only flag set by _record_from_store so callers can distinguish hydrated history from live in-memory runs. join_run and stream_existing_run (action=None) now return 409 instead of hanging forever on an empty MemoryStreamBridge channel. Re-check _runs under lock after the store await in RunManager.get() so a concurrent create() that lands between the two checks returns the authoritative in-memory record rather than a stale store-hydrated copy. Co-Authored-By: Claude Sonnet 4 <noreply@anthropic.com> * fix(harness): reorder bridge fetch in join_run and make list_by_thread limit explicit Move get_stream_bridge() after the store_only guard in join_run so a missing bridge cannot produce 503 for historical runs before the 409 guard fires. Add limit parameter to RunManager.list_by_thread (default 100, matching the store's page size) and pass it explicitly to the store call. Update docstring to document the limit instead of claiming all runs are returned. Co-Authored-By: Claude Sonnet 4 <noreply@anthropic.com> * fix(harness): cap list_by_thread result to limit after merge Apply [:limit] to all return paths in list_by_thread so the method consistently returns at most limit records regardless of how many in-memory runs exist, making the limit parameter a true upper bound on the response size rather than just a store-query hint. Co-Authored-By: Claude Sonnet 4 <noreply@anthropic.com> * fix `list_by_thread` docstring Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * fix(runtime): add update_model_name to RunStore to prevent SQL integrity errors RunManager.update_model_name() was calling _persist_to_store() which uses RunStore.put(), but RunRepository.put() is insert-only. This caused integrity errors when updating model_name for existing runs in SQL-backed stores. fix: - Add abstract update_model_name method to RunStore base class - Implement update_model_name in MemoryRunStore - Implement update_model_name in RunRepository with proper normalization - Add _persist_model_name helper in RunManager - Update RunManager.update_model_name to use the new method test: - Add tests for update_model_name functionality - Add integration tests for RunManager with SQL-backed store Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> * fix(runtime): handle NULL status/on_disconnect in _record_from_store `dict.get(key, default)` only uses the default when the key is absent, so a SQL row with an explicit NULL status would pass `None` to `RunStatus(None)` and raise, breaking hydration for otherwise valid rows. Switch to `row.get(...) or fallback` so both missing and NULL values get a safe default. Add tests for get() and list_by_thread() with a NULL status row to prevent regression. Co-Authored-By: Claude Sonnet 4 <noreply@anthropic.com> * fix(runs): address PR review feedback on store consistency changes - Fix list_by_thread limit semantics: pass store_limit = max(0, limit - len(memory_records)) to store so newer store records are not crowded out by in-memory records - Remove dead code: cancelled guard after raise is always True, simplify to if wait and record.task - Document _record_from_store NULL fallback policy (status→pending, on_disconnect→cancel) in docstring Co-Authored-By: Claude Sonnet 4 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com> Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> |
||
|
Willem Jiang
|
39f901d3a5
|
fix(runs): restore historical runs from persistent store after gateway restart (#2989)
* fix(runs): restore historical runs from persistent store after gateway restart
RunManager.list_by_thread() and get() only queried the in-memory _runs
dict, returning empty results after a restart even when PostgreSQL had
the records. Add store fallback to both read paths and a new async
aget() for the API endpoint, keeping sync get() for internal callers
that need live task/abort_event state.
Fixes #2984
* Apply suggestions from code review
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
* fix(runs): scope run store fallback reads by user id
Agent-Logs-Url: https://github.com/bytedance/deer-flow/sessions/e73daada-1215-4bc1-ab7d-7117826c5013
Co-authored-by: WillemJiang <219644+WillemJiang@users.noreply.github.com>
* test(runs): clarify ordering expectation and mock store filters
Agent-Logs-Url: https://github.com/bytedance/deer-flow/sessions/e73daada-1215-4bc1-ab7d-7117826c5013
Co-authored-by: WillemJiang <219644+WillemJiang@users.noreply.github.com>
* test(runs): make user filter fallback assertions explicit
Agent-Logs-Url: https://github.com/bytedance/deer-flow/sessions/e73daada-1215-4bc1-ab7d-7117826c5013
Co-authored-by: WillemJiang <219644+WillemJiang@users.noreply.github.com>
* test(runs): verify user-isolated fallback behavior with memory store
Agent-Logs-Url: https://github.com/bytedance/deer-flow/sessions/e73daada-1215-4bc1-ab7d-7117826c5013
Co-authored-by: WillemJiang <219644+WillemJiang@users.noreply.github.com>
* update the code with feedback from issue-2984
---------
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
|
||
|
Yi Tang
|
de253e4a0a
|
feat(run): Propagates model_name from the gateway request through the runtime and persistence stack to the SQLite database. (#2775)
* feat(run): propagate model_name from gateway request context to persistence layer Pass model_name through the full run creation pipeline — from RunCreateRequest.context in the gateway, through RunManager, to the RunStore interface and SQL persistence. This enables client-specified model selection to be recorded per-run in the database. * feat(run): add model allowlist validation and effective model name capture - Validate model_name against allowlist in gateway services.py using get_app_config().get_model_config() - Truncate model_name to 128 chars to match DB column constraint - In worker.py, capture effective model name from agent.metadata after agent creation and persist if resolved differently than requested * feat(run): add defense-in-depth model_name normalization and round-trip persistence tests - Add _normalize_model_name() to RunRepository for whitespace stripping and 128-char truncation before DB writes. - Add round-trip unit tests for model_name creation and default None in test_run_manager.py. * fix(run): coerce non-string model_name values before strip/truncate in _normalize_model_name * fix(gateway): add runtime type guard for model_name coercion in gateway services Add isinstance check and str() coercion before calling .strip() to prevent AttributeError when non-string types (int, None, etc.) flow through the gateway. Paired with SQL integration test for end-to-end model_name persistence across gateway → langgraph → persistence layer. * fix(run): drop Alembic migration for model_name (no-op) and expose public update method on RunManager - Drop a1b2c3d4e5f6 migration: model_name already exists in RunRow schema and is auto-created via Base.metadata.create_all() at startup - Add update_model_name() public method to RunManager to replace the private _persist_to_store call in worker.py, preserving internal locking/persistence |
||
|
Xinmin Zeng
|
ca3332f8bf
|
fix(gateway): return ISO 8601 timestamps from threads endpoints (#2599)
* fix(gateway): return ISO 8601 timestamps from threads endpoints (#2594) ThreadResponse documents created_at / updated_at as ISO timestamps, matching the LangGraph Platform schema (langgraph_sdk.schema.Thread exposes them as datetime, JSON-encoded as ISO 8601). The gateway threads router was instead emitting str(time.time()) — unix-second floats — breaking frontend new Date() parsing and producing a mixed ISO/unix wire format that also corrupted the search sort order. Centralize timestamp generation in deerflow.utils.time: - now_iso() — datetime.now(UTC).isoformat() - coerce_iso(x) — heals legacy unix-timestamp strings on read so the store converges to ISO without a one-shot migration threads.py: replace 6 time.time() call sites with now_iso(); wrap all read paths and Phase-2 checkpoint metadata with coerce_iso(); _store_upsert opportunistically heals legacy created_at on update; drop unused time import. thread_runs.py: reuse now_iso() instead of a private duplicate _now_iso(), preventing future drift between the two timestamp call sites. Tests: 9 unit tests for the helper; 5 integration tests pinning the ISO contract for create/get/patch/search and the legacy-healing path on the internal store upsert. Full suite: 2144 passed, 15 skipped, 0 failed. Closes #2594 * fix(gateway): coerce checkpoint metadata timestamps to ISO on read After the merge with main, three additional read paths in ``threads.py`` were still emitting raw ``str(metadata.get("created_at", ""))`` — ``get_thread_state``, ``update_thread_state``, and ``get_thread_history``. Same root cause as #2594: when the checkpoint metadata's ``created_at`` is a unix-second float (legacy data, or a checkpoint written by an older Gateway version), ``str(float)`` produces ``"1777252410.411327"`` and the frontend's ``new Date(...)`` returns ``Invalid Date``. The fix on the ``/threads/{id}`` GET path was already in place; these three sibling endpoints needed the same treatment. All four call sites now flow through ``coerce_iso``, so: - legacy float metadata heals to ISO on the way out, - ISO metadata passes through unchanged, - ``datetime`` instances (which the new ``coerce_iso`` branch handles explicitly) emit with the ``T`` separator instead of falling through to the space-separated ``str(datetime)`` form. Coverage added for the two endpoints not already pinned by the merge: - ``test_get_thread_state_returns_iso_for_legacy_checkpoint_metadata`` - ``test_get_thread_history_returns_iso_for_legacy_checkpoint_metadata`` Both pre-seed a checkpoint whose metadata carries the literal float from the issue body and assert the wire format is ISO. |
||
|
JeffJiang
|
db5ad86381 |
feat: enhance chat history loading with new hooks and UI components (#2338)
* Refactor API fetch calls to use a unified fetch function; enhance chat history loading with new hooks and UI components - Replaced `fetchWithAuth` with a generic `fetch` function across various API modules for consistency. - Updated `useThreadStream` and `useThreadHistory` hooks to manage chat history loading, including loading states and pagination. - Introduced `LoadMoreHistoryIndicator` component for better user experience when loading more chat history. - Enhanced message handling in `MessageList` to accommodate new loading states and history management. - Added support for run messages in the thread context, improving the overall message handling logic. - Updated translations for loading indicators in English and Chinese. * Fix test assertions for run ordering in RunManager tests - Updated assertions in `test_list_by_thread` to reflect correct ordering of runs. - Modified `test_list_by_thread_is_stable_when_timestamps_tie` to ensure stable ordering when timestamps are tied. |
||
|
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>
|
||
|
Admire
|
aae59a8ba8
|
fix: surface configured sandbox mounts to agents (#1638)
* fix: surface configured sandbox mounts to agents * fix: address PR review feedback --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com> |
||
|
Admire
|
3ff15423d6
|
fix Windows Docker sandbox path mounting (#1634)
* fix windows docker sandbox paths * fix windows sandbox mount validation * fix backend checks for windows sandbox path PR |
||
|
rayhpeng
|
34e835bc33
|
feat(gateway): implement LangGraph Platform API in Gateway, replace langgraph-cli (#1403)
* feat(gateway): implement LangGraph Platform API in Gateway, replace langgraph-cli
Implement all core LangGraph Platform API endpoints in the Gateway,
allowing it to fully replace the langgraph-cli dev server for local
development. This eliminates a heavyweight dependency and simplifies
the development stack.
Changes:
- Add runs lifecycle endpoints (create, stream, wait, cancel, join)
- Add threads CRUD and search endpoints
- Add assistants compatibility endpoints (search, get, graph, schemas)
- Add StreamBridge (in-memory pub/sub for SSE) and async provider
- Add RunManager with atomic create_or_reject (eliminates TOCTOU race)
- Add worker with interrupt/rollback cancel actions and runtime context injection
- Route /api/langgraph/* to Gateway in nginx config
- Skip langgraph-cli startup by default (SKIP_LANGGRAPH_SERVER=0 to restore)
- Add unit tests for RunManager, SSE format, and StreamBridge
* fix: drain bridge queue on client disconnect to prevent backpressure
When on_disconnect=continue, keep consuming events from the bridge
without yielding, so the worker is not blocked by a full queue.
Only on_disconnect=cancel breaks out immediately.
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
* fix: remove pytest import
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
* fix: Fix default stream_mode to ["values", "messages-tuple"]
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
* fix: Remove unused if_exists field from ThreadCreateRequest
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
* fix: address review comments on gateway LangGraph API
- Mount runs.py router in app.py (missing include_router)
- Normalize interrupt_before/after "*" to node list before run_agent()
- Use entry.id for SSE event ID instead of counter
- Drain bridge queue on disconnect when on_disconnect=continue
- Reuse serialization helper in wait_run() for consistent wire format
- Reject unsupported multitask_strategy with 400
- Remove SKIP_LANGGRAPH_SERVER fallback, always use Gateway
* feat: extract app.state access into deps.py
Encapsulate read/write operations for singleton objects (RunManager,
StreamBridge, checkpointer) held in app.state into a shared utility,
reducing repeated access patterns across router modules.
* feat: extract deerflow.runtime.serialization module with tests
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
* refactor: replace duplicated serialization with deerflow.runtime.serialization
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
* feat: extract app/gateway/services.py with run lifecycle logic
Create a service layer that centralizes SSE formatting, input/config
normalization, and run lifecycle management. Router modules will delegate
to these functions instead of using private cross-imported helpers.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
* refactor: wire routers to use services layer, remove cross-module private imports
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
* style: apply ruff formatting to refactored files
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
* feat(runtime): support LangGraph dev server and add compat route
- Enable official LangGraph dev server for local development workflow
- Decouple runtime components from agents package for better separation
- Provide gateway-backed fallback route when dev server is skipped
- Simplify lifecycle management using context manager in gateway
* feat(runtime): add Store providers with auto-backend selection
- Add async_provider.py and provider.py under deerflow/runtime/store/
- Support memory, sqlite, postgres backends matching checkpointer config
- Integrate into FastAPI lifespan via AsyncExitStack in deps.py
- Replace hardcoded InMemoryStore with config-driven factory
* refactor(gateway): migrate thread management from checkpointer to Store and resolve multiple endpoint failures
- Add Store-backed CRUD helpers (_store_get, _store_put, _store_upsert)
- Replace checkpoint-scanning search with two-phase strategy:
phase 1 reads Store (O(threads)), phase 2 backfills from checkpointer
for legacy/LangGraph Server threads with lazy migration
- Extend Store record schema with values field for title persistence
- Sync thread title from checkpoint to Store after run completion
- Fix /threads/{id}/runs/{run_id}/stream 405 by accepting both
GET and POST methods; POST handles interrupt/rollback actions
- Fix /threads/{id}/state 500 by separating read_config and
write_config, adding checkpoint_ns to configurable, and
shallow-copying checkpoint/metadata before mutation
- Sync title to Store on state update for immediate search reflection
- Move _upsert_thread_in_store into services.py, remove duplicate logic
- Add _sync_thread_title_after_run: await run task, read final
checkpoint title, write back to Store record
- Spawn title sync as background task from start_run when Store exists
* refactor(runtime): deduplicate store and checkpointer provider logic
Extract _ensure_sqlite_parent_dir() helper into checkpointer/provider.py
and use it in all three places that previously inlined the same mkdir logic.
Consolidate duplicate error constants in store/async_provider.py by importing
from store/provider.py instead of redefining them.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* refactor(runtime): move SQLite helpers to runtime/store, checkpointer imports from store
_resolve_sqlite_conn_str and _ensure_sqlite_parent_dir now live in
runtime/store/provider.py. agents/checkpointer/provider and
agents/checkpointer/async_provider import from there, reversing the
previous dependency direction (store → checkpointer becomes
checkpointer → store).
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* refactor(runtime): extract SQLite helpers into runtime/store/_sqlite_utils.py
Move resolve_sqlite_conn_str and ensure_sqlite_parent_dir out of
checkpointer/provider.py into a dedicated _sqlite_utils module.
Functions are now public (no underscore prefix), making cross-module
imports semantically correct. All four provider files import from
the single shared location.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* fix(gateway): use adelete_thread to fully remove thread checkpoints on delete
AsyncSqliteSaver has no adelete method — the previous hasattr check
always evaluated to False, silently leaving all checkpoint rows in the
database. Switch to adelete_thread(thread_id) which deletes every
checkpoint and pending-write row for the thread across all namespaces
(including sub-graph checkpoints).
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* fix(gateway): remove dead bridge_cm/ckpt_cm code and fix StrEnum lint
app.py had unreachable code after the async-with lifespan refactor:
bridge_cm and ckpt_cm were referenced but never defined (F821), and
the channel service startup/shutdown was outside the langgraph_runtime
block so it never ran. Move channel service lifecycle inside the
async-with block where it belongs.
Replace str+Enum inheritance in RunStatus and DisconnectMode with
StrEnum as suggested by UP042.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
* style: format with ruff
---------
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: JeffJiang <for-eleven@hotmail.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
|