infiniflow / infiniflow/ragflow
Dataflow debug: handler reads Redis directly and duplicates the log key format (layering violation)
- Dominant language
- Go
- Stars
- 91k
- Forks
- 10.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 705
Description
## Background
The Go backend has a dataflow dry-run ("debug") feature for canvases. `AgentHandler.runDataflowDebug` (`internal/handler/agent.go`) runs a canvas with no knowledgebase, collects a progress log, and flushes it to Redis under the key `{canvasID}-{messageID}-logs`. The front-end then polls `GetAgentLogs` (`internal/handler/agent.go`) to render the timeline. The key format intentionally mirrors the Python agent API (`api/apps/restful_apis/agent_api.py`, `bot_api.py`) so the Go and Python sides interoperate on the same key namespace.
## Problem
`GetAgentLogs` reads Redis **directly inside the HTTP handler**:
```go
key := fmt.Sprintf("%s-%s-logs", canvasID, messageID) // inline key construction
payload, rerr := h.redisGet(key) // redis.Get().Get(key)
```
The handler reaches into the Redis client and (re)constructs the key string itself, duplicating the format already built in `task.DebugLogSink.Flush` (`internal/ingestion/task/debug_log_sink.go:226`: `s.canvasID + "-" + s.messageID + "-logs"`). The same direct-`redis.Get()` read also exists at `internal/handler/bot.go:263-264`.
## Consequences
- **Layering violation.** The repo's own design comment states the rule (`internal/ingestion/task/embedder.go:87-90`): *"the task package is the composition root for ingestion runs ... the component package must not import internal/service"* — i.e. infra implementations are injected into `task`; the transport layer must not call `redis` directly. The handler doing so breaks that contract.
- **Duplicated key format with no single source of truth.** The key is assembled in two places (writer in `task`, reader in `handler`). Changing either side silently desynchronises them: logs are written but never read back (or reads return empty) — a latent correctness bug with no compile-time guard.
- **Two parallel Redis seams for one logical store.** The handler injects `redisStore` (`task.DebugLogStore`) for the write path and a separate `redisGet` func for the read path, which complicates testing and obscures that both are the same store.
## What the correct layering should be
- **handler** = transport adapter only: parse request, authenticate (`GetUser`), authorize at the boundary (`CheckCanvasAccess` via the service), **delegate** to lower layers, serialize the response. It must not call `redis` directly and must not own orchestration.
- **task** = ingestion composition root + domain: owns the pipeline executor, the `DebugLogSink`, and **defines** the `DebugLogStore` interface that Redis satisfies. It owns the log key format as a single source of truth. Infra implementations (`redis.Get()`) are *injected into* `task` by the handler as the top-level composition root.
- The ideal home is `AgentService` (it already owns Redis run-infra such as `runTracker` / `checkpointStore`). That is **not currently possible** because `task` already imports `service` (`embedder.go:27`, flagged at `:88`); `service -> task` would be a compile-breaking import cycle. Fixing that cycle is a separate, larger refactor. Given the dependency graph, `task` is the correct owner for the debug-log read/write.
## Current state
- **Write (correct):** `runDataflowDebug` -> `task.NewDebugLogSink(canvasID, messageID, h.redisStore)` -> `Flush` writes `{canvasID}-{messageID}-logs`. The handler supplies the `redis.Get()` implementation; `task` owns the logic and the key.
- **Read (violation):** `GetAgentLogs` builds the key inline and calls `redis.Get().Get(key)` directly.
## How to fix
1. Add `func DebugLogKey(canvasID, messageID string) string` in `task` (`debug_log_sink.go`); use it inside `DebugLogSink.Flush`.
2. Extend `task.DebugLogStore` with `Get(key string) (string, error)`. `redis.Get()` already satisfies it; test fakes (`capturedStore`, `miniredisDebugStore`) add `Get`.
3. Add `func ReadDebugLog(store DebugLogStore, canvasID, messageID string) (string, error)` in `task` that returns `store.Get(DebugLogKey(canvasID, messageID))`.
4. `GetAgentLogs` becomes `payload, rerr := task.ReadDebugLog(h.redisStore, canvasID, messageID)`; remove the `redisGet` field, `WithRedisGetter`, and the direct `redis.Get()` usage from `internal/handler/agent.go`.
5. Tests (`agent_logs_test.go`): drop `WithRedisGetter`; route the read through `WithRedisStore`; fakes gain `Get`. (`agent_dataflow_test.go` already uses `WithRedisStore`.)
6. (Optional, for consistency) Apply the same `task.ReadDebugLog` pattern to `bot.go:263`.
Authorization is already enforced on the read path via `CheckCanvasAccess`, so this is a layering/robustness fix, not an access-control fix.
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.