7 Commits

Author SHA1 Message Date
RongJie G
bf2cb0e1df
fix(runtime): recover terminal Redis streams after producer loss (#7093)
Signed-off-by: CorgiBoyG <111257566+CorgiBoyG@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
2026-10-19 15:53:54 +08:00
Nan Gao
5e3c24fe94
fix(gateway): probe the Redis stream bridge and provisioner in /health/ready (#6447)
* fix(gateway): probe the Redis stream bridge and provisioner in /health/ready

GET /health/ready only probed the ORM engine and the LangGraph checkpointer.
The Redis stream bridge client connects lazily and nothing ever pinged it, so a
Gateway whose Redis was down started, answered 200 ready to the Kubernetes
readinessProbe and Compose healthcheck, and kept receiving traffic while every
run's first publish failed. Since #6328 the Redis bridge is mandatory for
multi-instance deployments, so such a replica is not serviceable at all.

- StreamBridge gains `async ping() -> bool`; the base returns True (no external
  backend), RedisStreamBridge round-trips PING and answers False on RedisError.
- /health/ready adds a `stream_bridge` probe (not_configured for the memory
  bridge; unreachable flips the endpoint to 503; a missing bridge fails closed)
  and a report-only `provisioner` probe of `{sandbox.provisioner_url}/health`
  that never changes the status code, since every replica shares one
  provisioner. The provisioner URL is snapshotted on app.state at startup.
- A probe that overruns the endpoint deadline is now the only one marked
  unreachable; finished verdicts are kept.
- The public probe gate is one lock per probe kind, so one request's probes
  still run concurrently while a thundering herd opens at most one probe
  connection per kind per process.
- Tests: bridge ping cases, stream-bridge/provisioner probe cases, verdict and
  deadline cases, and the first route-level /health/ready tests via TestClient.
- Docs: gateway AGENTS.md, CONFIGURATION.md multi-instance notes, Helm README
  and probe comment, Compose healthcheck comment.

* docs(changelog): note the readiness probe fix (#6447)

* fix(gateway): apply the sandbox proxy policy to the provisioner probe

Review follow-up for #6447.

- The provisioner readiness probe built its httpx client with the default
  trust_env=True. httpx 0.28 treats CIDR entries in NO_PROXY as host patterns
  rather than subnet bypasses, so with HTTP_PROXY set and NO_PROXY=10.0.0.0/8
  a private provisioner that the requests-based sandbox calls reach directly
  would have been probed through the proxy and reported unreachable on every
  check. The probe now reuses `sandbox_http_trust_env`: loopback, private,
  link-local and cluster-local addresses bypass the proxy, external hosts keep
  the environment's proxy settings.
- Pin the liveness/readiness separation with a route test: `GET /health`
  stays 200 while the stream bridge and database are unreachable and
  `/health/ready` is 503, so a Redis blip can never become a restart loop.
2026-10-08 06:56:33 +08:00
Wuong
e12925458a
feat(streaming): make heartbeat interval configurable (#5017)
Co-authored-by: Wuong <26929475+Wuong@users.noreply.github.com>
2026-08-30 10:46:01 +08:00
Nefelibata
c6f6a01f56
fix(runtime): prevent IndexError in MemoryStreamBridge._make_gap on empty events buffer (#5047)
* fix(runtime): prevent IndexError in MemoryStreamBridge._make_gap on empty events buffer

* fix(runtime): handle empty stream replay gap bounds across backend and frontend

- Clamp MemoryStreamBridge queue_maxsize at 1 and validate StreamBridgeConfig.queue_maxsize >= 1
- Update StreamGap docstring to clarify None retained bounds
- Allow StreamReplayGapData and parseStreamReplayGap in frontend to accept string | null bounds, safely resuming when bounds are null
- Add backend and frontend regression unit tests for queue clamping and null bounds replay gap

* docs(stream-bridge): bump config_version and document empty buffer replay gap behavior

* docs: document nullable gap bounds and sync helm config_version to 37
2026-08-29 17:21:29 +08:00
Huixin615
1cd5dea336
fix(streaming): signal replay history gaps (#4426)
* fix(streaming): signal replay history gaps

* fix(streaming): guard initial Redis replay window

* fix(frontend): align inactive gap recovery

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
2026-07-27 07:13:06 +08:00
Janlay
72f033fbbe
feat(gateway): add redis stream bridge (#3191)
* feat: add redis stream bridge

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* fix(gateway): address redis stream bridge review

Redis was imported eagerly through deerflow.runtime and declared as a hard dependency, which made memory-only installs load redis.asyncio at startup and left the lazy factory import ineffective. Move redis behind an optional extra, remove the public eager re-export, and keep make_stream_bridge as the only runtime import path with an actionable install hint when the extra is missing.

Because Docker deployments now default the stream bridge to Redis via DEER_FLOW_STREAM_BRIDGE_REDIS_URL, install the redis extra explicitly in Docker/dev container flows and teach the local uv-extra detector to infer redis from both stream_bridge.type and the Redis URL env var. This keeps Docker working while preserving slim non-Docker installs.

Harden the Redis bridge by batching XREAD replay, replacing brittle ResponseError string matching with a single fallback to 0-0 for malformed Last-Event-ID values, documenting connection/retention/fail-hard behavior, and adding fake plus opt-in real Redis coverage for XADD/XREAD, replay, invalid IDs, and MAXLEN trimming.

* fix(config): bump config version for stream bridge

* fix redis stream bridge terminal handling

* fix: repair uv.lock, format redis.py, and align Dockerfile extras test

The uv.lock file was missing a closing bracket for the redis extras
section, redis.py had a formatting issue caught by ruff, and the
Dockerfile extras test did not account for the hardcoded --extra redis
flag.

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

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
2026-07-04 09:21:19 +08:00
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>
2026-03-30 16:02:23 +08:00