ragflow/docs/develop/agent-go-port-design.md

# Agent Canvas Go Port — Design Document

> **Last cross-checked against code:** 2026-06-17
> **Source of truth:** `internal/agent/` (canvas, component, tool, runtime, workflowx, dsl, audio, sandbox, observability/otel/) + `internal/service/` (agent.go, canvas_decode.go) + `internal/handler/` (agent.go) + `tools/` (gen-component-parity, migrate-canvas)

---

## 0. How to read this document

- **Sections 1–13** describe the current design.
- **Section 14** lists the actionable backlog for the next iteration.
- **Appendices** preserve the per-component / per-tool inventories and corner-case catalogues.

---

## 1. 概述 / Overview

### 1.1 目标

RAGFlow 的 Agent Canvas（编排 22 个 component + 21 个 tool 的 DSL 执行器）从 Python 移植到 Go。Python 端位于 `agent/canvas.py`（`Graph` / `Canvas`）+ `agent/component/base.py`（`ComponentBase` / `ComponentParamBase`）+ `agent/tools/`。Go 端独立实现于 `internal/agent/`，与 Python 端通过共享 DSL JSON schema 兼容（v1↔v2 双向转换器在 `internal/agent/dsl/`，已收敛为单一 wire 形态）。

### 1.2 核心架构决策

**State + Workflow 混血**：eino 的 `compose.Workflow` 提供声明式拓扑（节点 + exec 边）+ 并发调度；`compose.WithGenLocalState` + `WithStatePreHandler/WithStatePostHandler` 提供任意节点读任意节点输出的"状态变量"能力。State 解决 `{{cpn_id@param}}` 任意交叉引用问题，Workflow 解决执行拓扑 + cancel + checkpoint 问题。

**5-tier 移植策略**：T1（直接复用 eino 内置）→ T2（薄包装）→ T3（Lambda + State）→ T4（嵌套 Workflow 子图）→ T5（重 I/O + 第三方 lib）。判定原则：功能相当 → 优先 eino 内置，禁止复制 Python 端的黑魔法（`_feeded_deprecated_params`、partial hack、`thread_pool_exec` 异步伪装等）。

**Checkpoint 存 Redis**：eino `compose.CheckPointStore` 是纯 KV 接口，Redis String + EXPIRE 是天然 fit。业务元数据（status / canvas_id / parent_run_id）走独立 Redis Hash（**由应用层显式控制**，不依赖 eino 自动写）。

**Observability 走 OpenTelemetry**：用 OTLP HTTP exporter + eino `callbacks.Handler` 注入 span。

**AGPL-3 零容忍**：T5 DOCX 库穷举后全部 AGPL-3/维护停滞，**自实现 OOXML writer**（`archive/zip` stdlib + `text/template`）；PDF 选 `signintech/gopdf` (MIT)；Excel 选 `xuri/excelize/v2` (BSD-3)；Markdown 选 `yuin/goldmark` (MIT)。

**Wait-for-User 用 eino 原生 interrupt**：废除自实现 sentinel 链路（`__wait_for_user__` / `_user_input_provided` / synthetic Loop / `cycle_wrap.go` / `wait_for_user.go`）；改用 `compose.Interrupt` + `compose.ResumeWithData` 一等 API，节点内 `compose.GetResumeContext[T](ctx)` 读用户输入。

**Real Compile/Invoke 接入生产链**：`buildRunFunc` 驱动真正的 `canvas.Compile` → `CompiledCanvas.Invoke` 流程。

### 1.3 Reuse-First Principle

Before adding any new component, runtime abstraction, or third-party dependency, every phase must check whether the capability already exists elsewhere in the codebase or its declared dependencies.

**Decision order** (apply in sequence; first match wins):

1. **Reuse the existing RAGFlow model/service capability as-is.** If `internal/entity/models/anthropic.go`, `internal/handler/chat_session.go`, or similar already has the capability, just wire it through — don't reimplement.
2. **Wrap an existing eino / workflowx / MCP-client primitive.** If eino's `compose.NewGraphMultiBranch` or `workflowx.AddLoopNode` or `internal/utility/mcp_client.go` already provides the mechanism, build a thin adapter.
3. **Promote an already-declared-but-indirect dependency.** If the dependency is already in `go.mod` (even as `// indirect`), the work is to import it directly and use it.
4. **Add a registry alias only (no new body)** when an existing engine-level mechanism already handles the semantics.
5. **Only as a last resort** — add a new component, a new interface method, or a new third-party dependency. Each such addition must come with a written justification explaining why steps 1-4 don't apply.

**Anti-patterns** explicitly rejected: ❌ Adding `InvokeAsync` to the `Component` interface (would compete with eino `compose.Parallel`); ❌ Registering `LoopItem` / `ExitLoop` as components; ❌ Reimplementing Python's runtime path extension in Go; ❌ Building a new MCP subsystem; ❌ "Introducing" gonja (it's already a declared dep).

---

## 2. 顶层模块布局 / Module Layout

```
internal/agent/
├── canvas/              # 画布执行器（eino 编译、状态调度、checkpoint、cancel、stream、interrupt）
│   ├── canvas.go        # Canvas struct, BuildWorkflow, Run/Stream
│   ├── runner.go        # canvas.Runner; SSE event emission + interrupt catch
│   ├── scheduler.go     # State pre/post handler + 节点 lambda + legacyNoOpNames
│   ├── node_body.go     # 单节点 lambda 体 (per-class timeout via resolveTimeout)
│   ├── timeout.go       # componentDefaults map; 4-level resolver (per-class env → per-class table → uniform env → 600s fallback)
│   ├── loop_subgraph.go # Loop 宏展开 (buildSubWorkflow + translateLoopCondition)
│   ├── interrupt_resume.go # eino interrupt 封装: UserFillUpNodeBody / IsInterruptError / ExtractInterruptContexts
│   ├── multibranch.go   # Switch / Categorize 路由的 eino MultiBranch 集成
│   ├── cancel.go        # Redis cancel 协议 (watchCancel goroutine)
│   ├── stream.go        # SSE 通道
│   ├── compile.go       # eino 编译 + WithCheckPointStore + checkPointAdapter (不覆盖 InternalSerializer)
│   ├── checkpoint_store.go  # RedisCheckPointStore (Get/Set/Delete) — interface 包含 Delete
│   ├── run_tracker.go   # RunTracker (Start/MarkSucceeded/MarkFailed/MarkCancelled/AttachCheckpoint)
│   ├── state_serializer.go  # CanvasStateSerializer (encoding/json)
│   └── state_export.go  # WithState / GetStateFromContext 薄重导出
│
├── component/           # 19 components + 6 helpers (含 fixture_stubs.go + universe_a_wrappers.go)
│   ├── base.go          # Component interface + ParamError + ErrNotImplemented
│   ├── registry.go      # name → factory 映射 (auto-init)
│   ├── runtime_wire.go  # 组件与 runtime 包的桥接
│   ├── io_init.go       # T5 组件初始化
│   ├── fixture_stubs.go # IterationStub / IterationItemStub / RetrievalStub / SearchMyDataset alias / ExeSQLStub
│   ├── universe_a_wrappers.go # newRetrievalComponent / newExeSQLComponent / newTavilySearchComponent — Universe A → Universe B 委派
│   ├── production_chain_fixes_test.go   # 生产链回归 pin 测试
│   ├── agent.go         # T1 — react.NewAgent + tool artifact capture + maybeAppendCitation + Reset() interface-assert
│   ├── llm.go           # T1 — EinoChatModel 薄包装; VisualFiles / Cite / MessageHistoryWindowSize / ChatTemplateKwargs / OutputStructure / JSONOutput / TopP / MaxRetries / DelayAfterError
│   ├── llm_retry.go     # retryInvoker + Unwrap(); unwrapChatInvoker 辅助
│   ├── switch.go        # T2 — 12 of 12 operators (==/!=/contains/not contains/start with/end with/empty/not empty/>/</>=/<=)
│   ├── begin.go / message.go / categorize.go / invoke.go / browser.go
│   ├── data_operations.go / list_operations.go / string_transform.go
│   ├── variable_aggregator.go / variable_assigner.go
│   ├── fillup.go / userfillup.go
│   ├── loop.go          # T4 — no-op marker, 实际工作由 loop_subgraph 接管
│   ├── parallel.go      # T4 — workflowx.AddParallelNode 包装
│   ├── docs_generator.go / excel_processor.go   # T5
│   └── render.go        # output_format HTML/Markdown/plain renderer
│
├── tool/                # 21 tools (统一 eino tool.InvokableTool)
│   ├── registry.go      # BuildAll / BuildByName (alias: exesql=execute_sql, retrieval=search_my_dateset=search_my_dataset)
│   ├── http_helper.go   # 共用 HTTP client (context + retry + backoff)
│   ├── ssrf.go          # SSRF 防护
│   ├── mcp.go           # MCPToolAdapter — InvokableRun 调 mcpclient.CallTool over streamable-HTTP
│   ├── retrieval.go / retrieval_service.go / retrieval_nlp.go / retrieval_kg.go  # RetrievalService 双 registry: nlp + kg
│   ├── sandbox_bridge.go # CodeExec sandbox providers 桥接
│   └── akshare.go / arxiv.go / code_exec.go / code_exec_client.go / crawler.go / deepl.go
│       / duckduckgo.go / email.go / exesql.go / github.go / google.go
│       / google_scholar.go / jin10.go / pubmed.go / qweather.go
│       / searxng.go / tavily.go / tushare.go / wencai.go / wikipedia.go / yahoo_finance.go
│
├── runtime/             # canvas + component 共享的运行时契约（无 cycle）
│   ├── component.go     # Component interface (从 component/base.go 提取)
│   ├── context.go       # GetStateFromContext / withState
│   ├── state.go         # CanvasState + NewCanvasState + GetVar/SetVar/ReadVars + MarshalJSON/UnmarshalJSON + compose.RegisterSerializableType
│   ├── template.go      # ResolveTemplate (regex 快速路径)
│   ├── template_jinja.go # gonja 兜底
│   ├── selector.go      # component selector 辅助
│   └── metrics.go       # runtime metrics + Prometheus counters
│
├── workflowx/           # eino 扩展（零侵入，外部 helper）
│   ├── loop.go          # AddLoopNode[T] — 通用 do-while 循环节点
│   ├── parallel.go      # AddParallelNode[I,O] — 通用 bounded-concurrency 节点
│   └── *_test.go        # 单元 + 集成测试
│
├── sandbox/             # CodeExec 沙箱 providers
│   ├── provider.go / manager.go / http.go / result_protocol.go / artifacts.go
│   ├── self_managed.go / aliyun.go / e2b.go / local.go / ssh.go
│   └── e2b_test.go / local_test.go / manager_test.go / result_protocol_test.go / self_managed_test.go / ssh_test.go
│
├── audio/               # TTS
│   ├── tts.go                # Synthesizer interface + 错误哨兵 + 默认 stub
│   ├── model_provider_synthesizer.go  # calls models.BaseModel.AudioSpeech (60+ driver impls)
│   ├── tts_dispatch.go       # TTSDispatcher interface + NewTTSDispatchFunc
│   └── *_test.go
│
├── observability/otel/  # OTel SDK + eino callbacks.Handler
│   ├── provider.go      # TracerProvider 工厂
│   └── handler.go       # eino callbacks.Handler → OTel span
│
└── dsl/                 # DSL normalize
    ├── normalize.go     # NormalizeForCanvas (enforceHandleIds / buildGraphFromComponents / foldLegacyLoopVariants)
    ├── normalize_test.go
    └── testdata/        # 7 fixtures (all / browser / dfx_picture_parser / questions_category / resume / subaget / switch)

internal/handler/
├── agent.go             # HTTP API (RunAgent SSE with RunEvent.Type dispatch)
├── agent_wait_for_user_test.go  # 4 e2e tests pinning wait-for-user orchestrator side
└── admin_runtime.go     # POST /api/v1/admin/canvas-runtime/:tenant_id

internal/service/
├── agent.go             # AgentService.RunAgent / buildRunFunc / NewAgentService[WithOptions] / option injection
├── canvas_decode.go     # decodeCanvasFromDSL
├── canvas_decode_test.go
├── agent_run_e2e_test.go  # 4 e2e tests
└── agent_sessions.go    # session CRUD

cmd/server_main.go       # Redis CheckPointStore + RunTracker + TTS service wire-up

internal/observability/otel/
├── provider.go          # TracerProvider 工厂 (读 OTEL_EXPORTER_OTLP_ENDPOINT)
├── handler.go           # eino callbacks.Handler → OTel span
└── handler_test.go      # tracetest.SpanRecorder
```

**实际文件计数**：

- Components: **19 个** — 见 §4.2
- Tools: **21 个** — 见 §4.5
- Sandbox providers: **5 个** (self_managed, aliyun, e2b, local, ssh)
- Test files: 60+ (canvas 17, component 50+, tool 30+, runtime 4, workflowx 8, sandbox 6, audio 3, service 8+, handler 10+)

---

## 3. 架构 / Architecture

### 3.1 State + Workflow 混血

eino `compose.Workflow` 本身只支持 DAG（节点间数据通过 declared predecessor 输出传递），没有"任意节点读任意节点输出"的现成 API。RAGFlow Python 端用 `self._canvas.get_variable_value("cpn_id@param")` 实现 `{{cpn_id@param}}` 任意交叉引用。

**Go 端方案**：

1. **State 承载变量**：每个 canvas run 创建 `*CanvasState`，挂在 `context.Value` 上。所有节点通过 `runtime.GetStateFromContext(ctx)` 读写。
2. **State pre-handler**：在 `wf.AddLambdaNode(...)` 时挂 `compose.WithStatePreHandler[map[string]any, *runtime.CanvasState](canvasPre)`，从 State 提取节点输入。
3. **State post-handler**：挂 `compose.WithStatePostHandler`，把节点输出回写 State。
4. **Workflow 承载拓扑**：节点按 `downstream` / `upstream` 加 exec 边，**数据流走 State 不走边**。

```go
// internal/agent/canvas/scheduler.go
node := wf.AddLambdaNode(cpnID, nodeBody,
    compose.WithStatePreHandler[map[string]any, *runtime.CanvasState](canvasPre),
    compose.WithStatePostHandler[map[string]any, *runtime.CanvasState](canvasPost),
)
for _, upID := range comp.Upstream {
    node.AddInput(upID)
}
```

**CanvasState 序列化**：

`CanvasState` 结构包含 `sync.RWMutex`，原生无法被 `encoding/json` 序列化（`Marshaler` 接口与 mutex 不兼容）。通过：

- `MarshalJSON` / `UnmarshalJSON` 方法 — 输出/读取 `canvasStateJSON` 内部结构（不暴露 mutex）
- `compose.RegisterSerializableType[CanvasState]` — 让 eino `StatePre/PostHandler` 在 interrupt path 能 marshal/unmarshal state

eino `InternalSerializer` 是另一个独立的序列化机制（eino 内部 checkpoint payload），**不**与 `WithStateSerializer`/`compose.Serializer` 共享。生产代码只 wire `WithCheckPointStore` (保留 eino `InternalSerializer` 默认值) + CanvasState 自带 `MarshalJSON`。

### 3.2 `runtime` 包：消除 `canvas <-> component` cycle

**问题**：`component/` 大量文件（Begin/Message/Switch/Browser/...）需要调 `canvas.CanvasState` / `canvas.GetStateFromContext` / `canvas.ResolveTemplate` / `canvas.SetDefaultFactory`；同时 `canvas` 通过 `ComponentFactory` 间接依赖 `component` 的具体实现。强行 `canvas -> component` 形成 Go import cycle。

**方案**：把"运行时共用契约"提取到 `internal/agent/runtime/`，**canvas 和 component 都依赖 runtime，但不互相依赖**。

| 提取到 runtime | 留在 canvas | 留在 component |
|---------------|-------------|----------------|
| `Component` interface | DSL graph types (`Canvas`, `CanvasComponent`, `CanvasComponentObj`) | component registry + factory |
| `CanvasState` + `GetVar/SetVar/ReadVars` + MarshalJSON | 拓扑构建 (`BuildWorkflow`, `buildLoopExpansion`, scheduler wiring) | 具体 component 实现 |
| `GetStateFromContext` / `withState` / `WithState` | checkpoint / workflow 编译 orchestration | `NewBeginComponent`, `NewMessageComponent`, ... |
| `ResolveTemplate` + `template_jinja` (gonja fallback) | Loop 宏展开 logic | |
| `ParamError`, `ErrNotImplemented` | | |

### 3.3 eino interrupt 路径

```
UserFillUp 节点 → compose.Interrupt(ctx, inputSpec)
                          ↓
            返回 *InterruptSignal (实现 error 接口)
                          ↓
            图引擎捕获 → 自动 checkpoint → 向上传播
                          ↓
       Runner.Run 捕获 → SSE "waiting_for_user" + 保存 interrupt id
                          ↓
        用户提交 → Runner.Run 注入 __resume_interrupt_id__ + __resume_data__
                          ↓
            buildRunFunc 消费 → compose.ResumeWithData(ctx2, id, data)
                          ↓
          节点重入 → 顶部 compose.GetResumeContext[T](ctx) → 返回用户输入
```

**核心实现**：

```go
// internal/agent/canvas/interrupt_resume.go
func UserFillUpNodeBody(cpnID string, params map[string]any) func(ctx context.Context, input map[string]any) (map[string]any, error) {
    inputSpec := buildInputSpec(params)
    return func(ctx context.Context, input map[string]any) (map[string]any, error) {
        // Resume path: 节点重入时, 顶部检查 resume context
        if isResume, hasData, data := compose.GetResumeContext[any](ctx); isResume && hasData {
            return map[string]any{
                "user_input": data,
                cpnID:        data,
            }, nil
        }
        // 首次执行: 调 Interrupt 暂停图
        if err := compose.Interrupt(ctx, inputSpec); err != nil {
            return nil, err
        }
        return nil, errors.New("UserFillUp: interrupt did not halt execution")
    }
}
```

**Runner.Run interrupt catch**（`internal/agent/canvas/runner.go`）：

```go
if info, ok := compose.ExtractInterruptInfo(runErr); ok {
    ctxs := info.InterruptContexts // []*compose.InterruptCtx
    if len(ctxs) > 0 {
        d.saveInterruptID(canvasID, sessionID, ctxs[0].ID)
        payload, _ := json.Marshal(WaitingForUserEvent{CpnID: ctxs[0].ID})
        push(out, RunEvent{Type: "waiting_for_user", Data: string(payload)})
        return
    }
}
```

**Resume 传参**（`buildRunFunc`）：

```go
if resumeID, ok := root["__resume_interrupt_id__"].(string); ok && resumeID != "" {
    resumeData := root["__resume_data__"]
    delete(root, "__resume_interrupt_id__")
    delete(root, "__resume_data__")
    ctx2 = compose.ResumeWithData(ctx2, resumeID, resumeData)
}
```

**Cycle 处理**：前端契约保证生产画布无环（`hasCanvasCycle` 阻止保存），eino 的 DAG 检查在 `Compile()` 时自动拒绝有环图，无需额外防御。

### 3.4 真实 Compile/Invoke 接入生产链

```go
// internal/service/agent.go — buildRunFunc

func (s *AgentService) buildRunFunc(canvasID string, versionRow *entity.UserCanvasVersion, dsl map[string]any) canvas.RunFunc {
    return func(ctx context.Context, root map[string]any) (*canvas.CanvasState, error) {
        if err := ctx.Err(); err != nil {
            return nil, err
        }

        taskID := ""
        if versionRow != nil {
            taskID = versionRow.ID
        }

        c, err := decodeCanvasFromDSL(dsl)
        if err != nil {
            return nil, err
        }

        runID := canvasID
        if sessionID, ok := root["session_id"].(string); ok && sessionID != "" {
            runID = runID + "-" + sessionID
        }
        state := canvas.NewCanvasState(runID, taskID)

        userInput, _ := root["user_input"].(string)
        state.Sys["query"] = userInput
        ctx2 := runtime.WithState(ctx, state)

        if resumeID, ok := root["__resume_interrupt_id__"].(string); ok && resumeID != "" {
            resumeData := root["__resume_data__"]
            delete(root, "__resume_interrupt_id__")
            delete(root, "__resume_data__")
            ctx2 = compose.ResumeWithData(ctx2, resumeID, resumeData)
        }

        if s.runTracker != nil {
            _ = s.runTracker.Start(ctx2, runID, canvasID, tenantIDFromRoot(root), userInput)
        }

        var cc *canvas.CompiledCanvas
        if s.checkpointStore != nil && s.stateSerializer != nil {
            cc, err = canvas.Compile(ctx2, c,
                canvas.WithCheckPointStore(s.checkpointStore),
                canvas.WithStateSerializer(s.stateSerializer),
            )
        } else {
            cc, err = canvas.Compile(ctx2, c)
        }
        if err != nil {
            s.markRunFailed(ctx2, runID, "compile: "+err.Error())
            return nil, fmt.Errorf("canvas compile: %w: %w", err, ErrAgentStorageError)
        }

        if s.runTracker != nil {
            _ = s.runTracker.AttachCheckpoint(ctx2, runID, runID)
        }

        _, err = cc.Workflow.Invoke(ctx2, map[string]any{"query": userInput})
        if err != nil {
            if canvas.IsInterruptError(err) || canvas.ExtractInterruptContexts(err) != nil {
                s.markRunFailed(ctx2, runID, "interrupt: "+err.Error())
                return state, err
            }
            s.markRunFailed(ctx2, runID, "invoke: "+err.Error())
            return nil, fmt.Errorf("canvas invoke: %w: %w", err, ErrAgentStorageError)
        }

        s.markRunSucceeded(ctx2, runID)
        return state, nil
    }
}
```

**AgentService option injection**（`internal/service/agent.go`）：

```go
type AgentService struct {
    // ... existing fields
    checkpointStore canvas.CheckPointStore  // nil = in-memory (test path)
    stateSerializer canvas.StateSerializer  // nil = eino default
    runTracker      *canvas.RunTracker      // nil = best-effort no-tracking
    runner          *canvas.Runner
}

func NewAgentService() *AgentService {
    return NewAgentServiceWithOptions(nil, nil, nil)
}

func NewAgentServiceWithOptions(
    cp canvas.CheckPointStore,
    ser canvas.StateSerializer,
    rt *canvas.RunTracker,
) *AgentService {
    return &AgentService{...}
}
```

**Production boot wiring**（`cmd/server_main.go`）：

```go
// SetRedisCheckPointStore + CanvasStateSerializer + RunTracker → NewAgentServiceWithOptions
// + configureTTSSynthesizer (audio.SetModelProviderSynthesizer)
// Redis 不可达时 graceful degradation: 退化为 in-memory (nil options)
```

**DSL decoder**（`internal/service/canvas_decode.go`）：

`decodeCanvasFromDSL` 接受两种形态：
1. **IMPORT shape**: `obj.component_name` / `obj.params` (Python v1 DSL 直接写入)
2. **NormalizeForCanvas output shape**: 扁平 `name` / `params` (生产路径走 NormalizeForCanvas)

不采用 JSON round-trip — 直接 map walking 更清晰，因为生产路径已通过 `NormalizeForCanvas` 扁平化。所有失败模式 wrap `ErrAgentStorageError`。

---

## 4. Component 库 / Component Library

### 4.1 5-tier 移植策略

| Tier | 含义 | 验收 |
|------|------|------|
| **T1** | 直接用 eino 已有类型/接口，零代码 | eino 单元测试覆盖 |
| **T2** | 薄包装 1 struct + factory，对齐 Python 行为参数 | 跨 eino/RAGFlow 边界 + 1 e2e |
| **T3** | `compose.Lambda` + `StatePre/PostHandler` | 1 单测 + 1 e2e |
| **T4** | 嵌套 `compose.Workflow` + `getState[CanvasState](ctx)` | 子图单测 + 完整 e2e |
| **T5** | 重 I/O + 第三方 lib | 单测 + e2e + 失败注入 |

**判定原则**：T1 > T2 > T3 > T4 > T5 时**禁止跳级**。

### 4.2 Component 现状（19 个 .go 文件）

| Component | Python 行为 | Tier | Go 实现 | 状态 |
|-----------|------------|------|---------|------|
| **LLM** | `LLMBundle` 单轮 chat + JSON output + cite + stream | T1 | `EinoChatModel` 薄包装 `internal/entity/models/<provider>.go`；实现 `model.ToolCallingChatModel`；`retryInvoker.Unwrap()` + `unwrapChatInvoker` 实现 normal-absolute-count retry 语义 | ✅ |
| **Agent** | ReAct + tool/MCP + 多轮 stream | T1 | `react.NewAgent` + `compose.ToolsNodeConfig{Tools: tools}` + 22 tool 全注册；citation 中间件 + tool artifact 收集已实现；`Reset()` 走 `interface{ Reset() }` 类型断言 | ✅ |
| **Switch** | 多条件 (and/or) → 多 downstream + ELSE | T2 | `compose.NewGraphMultiBranch` 路由；12 of 12 operators (`==`/`!=`/`contains`/`not contains`/`start with`/`end with`/`empty`/`not empty`/`>`/`<`/`>=`/`<=`) + case-insensitive string ops | ✅ |
| **Categorize** | LLM 分类 + 路由 | T3 | Lambda 调 LLM + `compose.NewGraphMultiBranch` | ✅ |
| **Begin** | DSL 入口 + 注入 inputs + 文件 inputs | T3 | Lambda + `StatePreHandler`；文件走 `internal/service/file_service.go` | ✅ |
| **UserFillUp / Fillup** | Jinja2 + file inputs + **wait-for-user interrupt** | T3 | `text/template` 替代 Jinja2 + eino interrupt via `interrupt_resume.go` | ✅ |
| **Message** | 最终输出（jinja2 + stream + downloads + filegen + TTS + memory） | T3 | Lambda + `schema.StreamReader` + `text/template` + MinIO + TTS dispatch + MemorySaver | 🟡 真实 TTS binary + MemorySaver completion deferred |
| **Invoke** | HTTP 客户端 + HTML 清洗 + JSON | T3 | `net/http` + `golang.org/x/net/html` | ✅ |
| **Browser** | LLM + HTTP + 文件下载 + MinIO | T3 | 复用 Invoke + LLM + storage | ✅ |
| **DataOperations** | dict 7 类操作 | T3 | Lambda + `encoding/json` + `go/ast` | ✅ |
| **ListOperations** | slice 6 类操作 | T3 | Lambda + `slices` (Go 1.21+ stdlib) | ✅ |
| **StringTransform** | split/merge + Jinja2 | T3 | Lambda + `strings.Split` + `text/template` | ✅ |
| **VariableAggregator** | 多 group，first-non-empty | T3 | Lambda + State 读 | ✅ |
| **VariableAssigner** | 11 个算子原地改 State | T3 | Lambda + State 写 | ✅ |
| **Loop** | 条件循环 + `loop_variables` 初始化 + 终止评估 | T4 | `compose.NewWorkflow` + `workflowx.AddLoopNode`（loop.go 自身变为 no-op marker；实际工作由 `canvas/loop_subgraph.go` 宏展开接管） | ✅ |
| **Parallel** | 数组并行处理 | T4 | `workflowx.AddParallelNode` 包装 | ✅ |
| **DocsGenerator** | pdf/docx/txt/md/html 生成 | T5 | `signintech/gopdf` (PDF) + 自实现 OOXML writer (DOCX) + `yuin/goldmark` (MD)；`render.go` 提供 HTML/Markdown/plain rendering | 🟡 txt/md/html writers 部分缺失 |
| **ExcelProcessor** | pandas 读/合并/转换 Excel | T5 | `xuri/excelize/v2` (BSD-3) | ✅ |
| **Retrieval (Universe A)** | canvas DAG node | T2 | `newRetrievalComponent` — 委派给 Universe B `RetrievalTool` | ✅ |

### 4.3 不移植的 Python 端"遗产" / Iteration LoopItem ExitLoop 重分类

| Python 端 | 不移植原因 |
|----------|-----------|
| `_feeded_deprecated_params` / `_deprecated_params` / `_user_feeded_params` 三层装饰 | DSL v2 已去除；Go `ComponentParamBase` 不引入 |
| `ComponentParamBase.validate()` + `param_validation/*.json` 96 文件 | Go struct tag + `go-playground/validator/v10` 替代 |
| `ComponentBase.thread_limiter = asyncio.Semaphore(...)` | Go `errgroup.SetLimit(MAX_CONCURRENT_CHATS)` (stdlib x/sync) |
| `partial` 流式 hack | eino `schema.StreamReader` 原生流式 |
| `thread_pool_exec(self._invoke, **kwargs)` 异步伪装 | Go 全程 goroutine |
| `set_output("_ERROR", ...)` + `set_exception_default_value()` 双轨 | Go `error` 单一返回 + eino `OnError` callback |
| `ExitLoop` no-op 节点 | DSL v1 compat 通过 `legacyNoOpNames` 在 canvas 层吸收，**不注册 component** |
| `LoopItem` 组件 | LoopItem 角色由 `workflowx.AddLoopNode` 内部 machinery 取代，**不注册 component**；`TestLoop_Registered` enforces absence |
| `Iteration` / `IterationItem` 组件 | `IterationStub` + `IterationItemStub` 注册为 compat stubs（DSL round-trip） |

### 4.4 Two Registry Universes (Universe A vs Universe B)

```
┌──────────────────────────────────────────────────────────────┐
│ Universe A — Canvas DAG Components                            │
│   Registry: internal/agent/component/registry.go (auto-init) │
│   Interface: Component { Invoke, Stream, Inputs, Outputs }   │
│   Output: map[string]any                                     │
│   Names: PascalCase — Retrieval, TavilySearch, ExeSQL,       │
│          Answer, Generate, Begin, LLM, Switch, …              │
│   Used by: Canvas DAG nodes (placed on the canvas directly)  │
├──────────────────────────────────────────────────────────────┤
│ Universe B — Agent ReAct Tools                               │
│   Registry: internal/agent/tool/registry.go                  │
│   Interface: einotool.BaseTool { Info, InvokableRun }        │
│   Output: JSON string (envelope)                             │
│   Names: snake_case — retrieval, tavily, execute_sql, …      │
│   Used by: Agent component's tools=["…"] list, called via    │
│            eino ReAct loop                                   │
└──────────────────────────────────────────────────────────────┘
```

**Mapping table**：

| Universe A (PascalCase) | Universe B (snake_case) | 当前状态 |
|---|---|---|
| Retrieval | `retrieval` / `search_my_dateset` / `search_my_dataset` | 委派到 Universe B real (nlp + kg 双 backend) |
| ExeSQL | `execute_sql` / `exesql` | 委派到 Universe B real (mysql/pg/mssql/oceanbase/trino) |
| TavilySearch | `tavily` | 委派到 Universe B real |
| Answer | — | 需要 orchestrator-side pause/resume（已通过 eino interrupt 实现） |
| Generate | — | alias to LLM component |
| SearchMyDataset | — | 注册为 Retrieval alias (4 spellings: PascalCase + snake_case + Python-typo) |

### 4.5 Tool 实现统一模式

```go
// internal/agent/tool/registry.go
type Tool interface {
    einotool.InvokableTool  // eino 协议：Info() / InvokableRun(ctx, args, opts)
}

func BuildAll(names []string, params map[string]map[string]any) ([]einotool.BaseTool, error)
func BuildByName(name string, params map[string]any) (einotool.BaseTool, error)
```

**21 tool 表** (alias 不算新 tool): akshare, arxiv, code_exec, crawler, deepl, duckduckgo, email, exesql(=execute_sql), github, google, google_scholar, jin10, pubmed, qweather, retrieval(=search_my_dateset=search_my_dataset), searxng, tavily, tushare, wencai, wikipedia, yahoo_finance。

**Retrieval 双 registry**：
- `internal/agent/tool/retrieval_nlp.go` — `NLPRetrievalAdapter` 桥接 `nlp.RetrievalService`
- `internal/agent/tool/retrieval_kg.go` — `KGRetrievalAdapter` 桥接 `kg.Retrieval(...)` (GraphRAG, `use_kg=true`)
- `internal/agent/tool/retrieval_service.go` — 两个独立 `SetRetrievalService` / `SetKGRetrievalService` registry; un-wired 返回 `ErrRetrievalServiceMissing` / `ErrKGRetrievalServiceMissing`

**MCP tools**：`internal/agent/tool/mcp.go` — `MCPToolAdapter.InvokableRun` 通过 `mcpclient.CallTool` over streamable-HTTP dispatch。

**Tool 通用模式**：HTTP 类 tool 走 `http_helper.go` (context + retry + 指数 backoff)；ExeSQL 走 stdlib `database/sql` + 各 driver (mysql / pg / mssql / oceanbase / trino)；CodeExec 走 `internal/agent/sandbox/` 5 providers (`self_managed` / `aliyun` / `e2b` / `local` / `ssh`) + `tool/sandbox_bridge.go` 桥接；Retrieval 走进程内 `internal/service/nlp/retrieval.go` (Dealer 后端已 Go 化)。

### 4.6 Component & Tool Inventory

Parity legend: ✅ implemented & tested · 🟡 scaffolded (loud-fail sentinel) · ⚠️ implemented with a known gap vs Python.

#### Universe A — Canvas DAG components (24)

| Name | Source | Status |
|------|--------|--------|
| Agent | `internal/agent/component/agent.go` | ✅ |
| Begin | `internal/agent/component/begin.go` | ✅ |
| Browser | `internal/agent/component/browser.go` | ✅ |
| Categorize | `internal/agent/component/categorize.go` | ✅ |
| DataOperations | `internal/agent/component/data_operations.go` | ✅ |
| DocsGenerator | `internal/agent/component/docs_generator.go` | ✅ |
| ExcelProcessor | `internal/agent/component/excel_processor.go` | ✅ |
| ExeSQL | `internal/agent/component/universe_a_wrappers.go` | ⚠️ Wrapper exists; registry primary still stub |
| Fillup | `internal/agent/component/fillup.go` | ✅ |
| Generate | `internal/agent/component/fixture_stubs.go` | ✅ Legacy alias for DSL round-trip |
| Invoke | `internal/agent/component/invoke.go` | ✅ |
| Iteration | `internal/agent/component/fixture_stubs.go` | ✅ Legacy alias; compat stub |
| IterationItem | `internal/agent/component/fixture_stubs.go` | ✅ Legacy alias; compat stub |
| ListOperations | `internal/agent/component/list_operations.go` | ✅ |
| LLM | `internal/agent/component/llm.go` | ✅ |
| Loop | `internal/agent/component/loop.go` | ✅ Engine-level macro (`LoopItem`/`ExitLoop` deliberately not registered) |
| Message | `internal/agent/component/message.go` | 🟡 TTS real engine + MemorySaver completion still deferred |
| Parallel | `internal/agent/component/parallel.go` | ✅ |
| Retrieval | `internal/agent/component/universe_a_wrappers.go` | ⚠️ Wrapper exists; registry primary still stub (also covers `SearchMyDataset` alias) |
| StringTransform | `internal/agent/component/string_transform.go` | ✅ |
| Switch | `internal/agent/component/switch.go` | ✅ All 12 operators with case-folded string ops |
| TavilySearch | `internal/agent/component/universe_a_wrappers.go` | ⚠️ Wrapper exists; registry primary still stub |
| UserFillUp | `internal/agent/component/userfillup.go` | ✅ |
| VariableAggregator | `internal/agent/component/variable_aggregator.go` | ✅ |
| VariableAssigner | `internal/agent/component/variable_assigner.go` | ✅ |
| Answer | `internal/agent/component/fixture_stubs.go` | 🟡 Compat stub; canvas pause/resume is real but the Answer node is still a placeholder |

> **Stub vs wrapper**: `Retrieval` / `TavilySearch` / `ExeSQL` have real delegation wrappers in `universe_a_wrappers.go`; the registry still maps them to stubs in `fixture_stubs.go`. Tracked in §14.

#### Universe B — eino ReAct tools (25 = 23 standalone + 2 aliases)

| Name | Source | Status |
|------|--------|--------|
| akshare | `internal/agent/tool/akshare.go` | ✅ |
| arxiv | `internal/agent/tool/arxiv.go` | ✅ |
| code_exec | `internal/agent/tool/code_exec.go` + `code_exec_client.go` | ✅ All 5 sandbox providers |
| crawler | `internal/agent/tool/crawler.go` | ✅ |
| deepl | `internal/agent/tool/deepl.go` | ✅ |
| duckduckgo | `internal/agent/tool/duckduckgo.go` | ✅ |
| email | `internal/agent/tool/email.go` | ✅ |
| execute_sql | `internal/agent/tool/exesql.go` | ⚠️ SELECT-only; rejects Trino/DB2 (`ErrExeSQLUnsupportedDB`) |
| exesql | `internal/agent/tool/exesql.go` | ⚠️ Alias of `execute_sql` |
| github | `internal/agent/tool/github.go` | ✅ |
| google | `internal/agent/tool/google.go` | ✅ |
| google_scholar | `internal/agent/tool/google_scholar.go` | ✅ |
| jin10 | `internal/agent/tool/jin10.go` | ✅ |
| mcp | `internal/agent/tool/mcp.go` | 🟡 `MCPToolAdapter` wraps `mcpclient.Tool`; `InvokableRun` returns "not yet implemented" until `mcpclient.CallTools` lands |
| pubmed | `internal/agent/tool/pubmed.go` | ✅ |
| qweather | `internal/agent/tool/qweather.go` | ✅ |
| retrieval | `internal/agent/tool/retrieval.go` | ✅ Adapter + boot wiring (`cmd/server_main.go`) |
| search_my_dataset | `internal/agent/tool/registry.go` | ✅ Alias of `retrieval` |
| search_my_dateset | `internal/agent/tool/registry.go` | ✅ Python-typo alias of `retrieval` |
| searxng | `internal/agent/tool/searxng.go` | ✅ |
| tavily | `internal/agent/tool/tavily.go` | ✅ |
| tushare | `internal/agent/tool/tushare.go` | ✅ |
| wencai | `internal/agent/tool/wencai.go` | ✅ |
| wikipedia | `internal/agent/tool/wikipedia.go` | ✅ |
| yahoo_finance | `internal/agent/tool/yahoo_finance.go` | ✅ |

**Total**: 49 named entities (24 components + 25 tools).

---

## 5. DSL 单一形态

RAGFlow agent DSL 现在只有**一种** wire 形态（之前 v1/v2 双轨已删）：

```jsonc
{
  "globals":    {...},                                                  // sys.query / sys.user_id / ...
  "graph":      { "nodes": [...], "edges": [...] },                     // React-Flow 布局
  "variables":  {...},                                                  // 用户级变量
  "components": { "<Name>:<UUID>": {                                    // 执行拓扑
    "downstream": [...], "upstream": [...],
    "obj": { "component_name": "Name", "params": {...} }
  }},
  "path": [...], "retrieval": {...}, "history": [...]                    // 运行时状态
}
```

**单一 wire 的硬性保证**：

1. **后端 GET/PUT 收到的 DSL 必定同时含 `graph` + `components`**。前端 `use-build-dsl.ts` 在 PUT 时一并填充两个块，back-end 不依赖 `graph`。
2. **Go 端的唯一入口是 `dsl.NormalizeForCanvas`**（`internal/handler/agent.go:226`、`internal/service/agent.go:217,273`）。所有 Python ↔ Go 路径的 dsl 都在解码边界过一次。
3. **`internal/agent/dsl/` 包当前仅 `normalize.go` + `normalize_test.go` + `testdata/`**（v1↔v2 转换器与 `v2.go`/`loader.go`/`converter_v1_to_v2.go`/`converter_v2_to_v1.go` 已 `git rm`）。

### 5.1 NormalizeForCanvas：解码边界的三步流水线

`internal/agent/dsl/normalize.go` 的 `NormalizeForCanvas(dsl map[string]any) map[string]any`：

1. **`enforceHandleIds(dsl)`** — 把 `graph.edges[*].sourceHandle` / `targetHandle` 规约为 React-Flow 约定。
2. **`buildGraphFromComponents(components)`** — 若 `graph.nodes` 缺失，从 `components` 派生默认布局。
3. **`foldLegacyLoopVariants(dsl)`** — 把 `Loop+LoopItem` / `Iteration+IterationItem` 折叠成单个 `Loop` / `Parallel` 节点。

### 5.2 Loop / Iteration 折叠语义

- **Python 端保留** `Loop+LoopItem` / `Iteration+IterationItem` 旧类名（stable server，本次不动）。
- **Go 端** `Loop` 已经是单节点（`internal/agent/component/loop.go`），`Parallel` 已经是单节点。`Iteration` / `IterationItem` 仅作为 alias 留在 `internal/agent/component/fixture_stubs.go`，stub 体内**委托给 Parallel factory**。
- **前端** `Operator` 枚举里 `Iteration` / `IterationStart` / `LoopStart` 保留。

### 5.3 Compile 入口的兼容兜底

`canvas.Compile(ctx, c *Canvas, opts...)` 接收的 `*Canvas` **预期已经过 `NormalizeForCanvas`**。如果某条路径直接 unmarshal dsl 后丢给 Compile 而没走 decoder，`Compile` 入口会 `log.Printf` 一行 stderr warning。

### 5.4 7 个 testdata 顶层结构

`internal/agent/dsl/testdata/{all, browser, dfx_picture_parser, questions_category, resume, subaget, switch}.json` 顶层都是 `{globals, graph, variables}`（`graph.nodes` / `graph.edges` 完整）。**没有 `components` 顶层 key**。这是 import / export 文件的形态。

### 5.5 前端 dsl-bridge：单一 import 路径

`web/src/pages/agent/utils/dsl-bridge.ts` 重写为单一模式：

- 删除 `DSL_MODE` / `DslMode` / `if (DSL_MODE === 'v1')` / `if (DSL_MODE === 'v2')` 编译期分支
- `importDsl(rawParsed, isAgent)` 单一优先级：`raw.graph.nodes` 非空 → 用之；否则 fallback 到 empty seed
- `dslToGraph(dsl)` 同样只读 `dsl.graph.nodes`

---

## 6. workflowx 扩展 / workflowx Extensions

`internal/agent/workflowx/` 提供**零侵入 eino 扩展**——不修改 eino 源码，只提供外部 helper。

### 6.1 AddLoopNode[T] — 通用循环节点

**API**：
```go
func AddLoopNode[T any](
    ctx context.Context,
    wf *compose.Workflow[T, T],
    key string,
    sub *compose.Workflow[T, T],
    shouldQuit LoopCondition[T],
    opts ...LoopOption,
) (*compose.WorkflowNode, error)
```

**执行模型**（do-while 语义）：

1. 接收 `current`
2. 跑一次 sub-workflow 拿 `next`
3. `shouldQuit(ctx, iteration, current, next)` — `iteration` 从 1 开始
4. 满足 quit → 返回 `next`；否则 `current = next` 继续
5. 必须至少执行一次

**实现要点**：

- `compose.AnyLambda[T, T, struct{}](...)` 包裹 invoke + stream 双路径
- `WithLoopMaxIterations(n)` 强建议（防意外死循环）
- `WithLoopStream(mode)` — `LoopStreamFinalOnly` (默认) / `LoopStreamEveryIteration`
- 错误处理：`ErrLoopMaxIterationsExceeded` / `ErrLoopSubGraphInterrupted` / `ErrLoopResumeStateInvalid` / `ErrLoopQuitConditionFailed`
- 嵌套子 workflow 走 `compose.Runnable[T,T]` + sub-checkpoint 通过 loop-owned bridge store

**Checkpoint/Resume 合约**：

- Invoke path 嵌套 interrupt → 通过 `compose.CompositeInterrupt` 向上传播；resume 从中断的 iteration 继续（不重头）
- Stream path 走 **iteration-granular** 恢复合约：已完整发到下游的 iteration 不重放
- 稳定 child checkpoint ID 通过 `WithLoopCheckpointIDBuilder(nodeKey, iteration)`；默认 `workflowx-loop:<nodeKey>:<iteration>` 命名空间

**Loop 在 canvas 中的应用**：

- `Loop` 在 Go 端是**单节点**：registry 注册 + 工厂，但 `LoopComponent.Invoke` 是 no-op
- `BuildWorkflow` 看到名为 `Loop` 的 cpn 时：调用 `expandLoopSubgraph` 收集下游、构建 sub-`compose.Workflow[map[string]any, map[string]any]`、调 `workflowx.AddLoopNode` 把结果作为单节点插入外图
- `LoopItem` / `ExitLoop` **已删除**（v1 compat 通过 `legacyNoOpNames` 在 canvas 层吸收）

### 6.2 AddParallelNode[I, O] — 通用并发节点

**API**：
```go
func AddParallelNode[I, O any](
    ctx context.Context,
    wf *compose.Workflow[[]I, []O],
    key string,
    sub Compilable[I, O],
    opts ...ParallelOption,
) (*compose.WorkflowNode, error)
```

**实现要点**：

- 外层 invoke-only；内层 sub workflow 可 stream-capable
- `WithParallelMaxConcurrency(n int)`：0 / 1 = 顺序执行；> 1 = 信号量并发
- **顺序保持不变量**：`outputs[i]` 永远对应 `inputs[i]`
- 错误处理：`ErrParallelCompileFailed` / `ErrParallelResumeStateInvalid`；per-item 错误用 `fmt.Errorf("item %d: %w", idx, err)` 包装
- 嵌套 interrupt：累积到 `compose.CompositeInterrupt(ctx, nil, state, interruptErrs...)`
- 恢复不变量：`CompletedResults ∪ InterruptedIndices = 0..TotalCount-1`（partition 完整）

**Parallel 在 canvas 中的应用**：

- `Parallel` component 走 T4 薄包装：注册时传 `agenttool.BuildByName("parallel", params)`（实际是 `internal/agent/component/parallel.go` 的 `ParallelComponent`），内部用 `workflowx.AddParallelNode` 把 sub-workflow 插入外图

### 6.3 Canvas parallel batch (eino intrinsic, NOT workflowx parallel)

**关键发现**：Phase 4.1 "Canvas parallel batch execution" 不需要额外实现 — **eino `compose.Workflow.Run` 本身就在每个 topological wave 内 spawn 一个 `go t.execute()` per ready node**。

- `canvas/parallel_batch_test.go::TestBuildWorkflow_ParallelBatchStructure` pin 4-node sibling compile
- `canvas/parallel_timing_test.go::TestCanvas_ParallelExecution_StaticAnalysis` pin 5-node DAG compile 静态分析

`workflowx/parallel.go` 仍存在，但仅用于 `Parallel` component (Loop/Iteration 风格的 array parallel)，**不是** canvas 层的 ready-node 调度。

---

## 7. Checkpoint + Run Tracker / Persistence

### 7.1 双 key 设计

**Key 1：`agent:cp:{check_point_id}`** — eino payload 存储

- 类型：String (直接存 `[]byte`，**不走 JSON** — eino Serializer 已负责序列化)
- TTL：30 天，Set 时 `EXPIRE 30*24*3600` 一次设置
- eino `CheckPointStore` 是**纯 KV 接口** — `Get(ctx, id) ([]byte, bool, error)` / `Set(ctx, id, []byte) error`
- eino **不会**自动写入 status / canvas_id / tenant_id / run_id / parent_id / expires_at 等业务字段

**Key 2：`agent:run:{run_id}`** — 业务元数据存储 (Redis Hash)

| 字段 | 类型 | 含义 |
|------|------|------|
| `canvas_id` | string | `user_canvas.id` |
| `tenant_id` | string | 从 user-tenant lookup |
| `checkpoint_id` | string | 当前 run 的最新 checkpoint (指向 key 1) |
| `parent_run_id` | string | resume_from 源 run (续跑链)，可空 |
| `status` | int (0/1/2/3) | 0=running 1=succeeded 2=failed 3=cancelled |
| `failure_reason` | string | 失败原因 (err.Error()) |
| `cancel_requested` | int (0/1) | 1=用户/admin 已请求 cancel |
| `started_at` | int (epoch ms) | |
| `finished_at` | int (epoch ms) | 退出时填写 |

- TTL：30 天 (与 key 1 同步)
- `RunTracker.Start/MarkSucceeded/MarkFailed/MarkCancelled/AttachCheckpoint` 显式调用
- **不依赖 eino 自动写** — cancel/fail 后的 `status=failed` 由应用层自己写

### 7.2 4 个 eino payload 写入触发 (写 `agent:cp:*`)

| # | 触发点 | eino 源码 | 用途 |
|---|--------|-----------|------|
| **W1** | 节点显式 `compose.Interrupt(ctx, info)` / `StatefulInterrupt(ctx, info, state)` | `compose/interrupt.go` | human-in-the-loop、外部 API 回调、限流暂停 |
| **W2** | `compose.WithInterruptBeforeNodes([]string)` / `WithInterruptAfterNodes([]string)` 编译期拦截点 | `compose/interrupt.go` | 命中后**写盘 + 终止 run** (与 W1 共用 `handleInterrupt` 路径)；**默认开 0 个** |
| **W3** | 子 graph interrupt 向上传播 | `subGraphInterruptError` | 嵌套 subgraph / ToolsNode / agentic 抛 interrupt 时，父 graph 同步落盘 |
| **W4** | 运行退出 | `WithCheckPointID` + `WithWriteToCheckPointID` | run 退出时最后一次落盘 |

### 7.3 4 个业务元数据写入 + 1 个恢复触发

| # | 触发点 | 写入函数 |
|---|--------|---------|
| **B1** | Canvas run 启动 | `RunTracker.Start(runID, canvasID, tenantID, parentRunID)` |
| **B2** | Run 正常完成 | `RunTracker.MarkSucceeded(runID)` |
| **B3** | Run 失败 | `RunTracker.MarkFailed(runID, err.Error())` |
| **B4** | Run 被 cancel | `RunTracker.MarkCancelled(runID)` |
| **B5** | Compile 成功后 | `RunTracker.AttachCheckpoint(runID, cpID)` |
| **R1** | HTTP `POST /run?resume_from=run_xxx` | handler: `HGetAll("agent:run:run_xxx")` → `checkpoint_id` → `WithCheckPointID(cpID)` + `WithWriteToCheckPointID(newCP)` + `RunTracker.Start(newRunID, canvas, tenant, "run_xxx")` |

### 7.4 CheckPointStore / StateSerializer 接口设计

**`internal/agent/canvas/checkpoint_store.go`**：

```go
type CheckPointStore interface {
    Get(ctx context.Context, id string) ([]byte, bool, error)
    Set(ctx context.Context, id string, data []byte) error
    Delete(ctx context.Context, id string) error  // 自定义扩展, eino compose.CheckPointStore 无此方法
}
```

**`internal/agent/canvas/state_serializer.go`**：

```go
type StateSerializer interface {
    Marshal(v any) ([]byte, error)
    Unmarshal(data []byte, v any) error
}

// CanvasStateSerializer — encoding/json
type CanvasStateSerializer struct{}
func (CanvasStateSerializer) Marshal(v any) ([]byte, error)   { return json.Marshal(v) }
func (CanvasStateSerializer) Unmarshal(b []byte, v any) error { return json.Unmarshal(b, v) }
```

**`internal/agent/canvas/compile.go`** — 关键修正：

```go
// 注意: 不能用 compose.WithSerializer 覆盖 eino 的 InternalSerializer!
// eino 的 compose.Serializer 同时控制 (a) 用户提供的 state 序列化 AND (b) eino 内部
// graph state 序列化。覆盖会破坏 eino graph 内部 marshal/unmarshal 逻辑。
//
// 正确做法: 仅 wire WithCheckPointStore (custom KV 接口), 让 eino 内部
// InternalSerializer 保留默认值。同时 CanvasState 自带 MarshalJSON 让
// eino StatePre/PostHandler 能序列化 state。
func Compile(ctx context.Context, c *Canvas, opts ...CompileOption) (*CompiledCanvas, error) {
    cfg := CompileOptions{}
    for _, o := range opts { o(&cfg) }
    
    compileOpts := []compose.GraphCompileOption{
        compose.WithCheckPointStore(checkPointAdapter{cfg.Store}),  // 适配 Delete
    }
    // 显式 NOT 调用 compose.WithSerializer
    return wf.Compile(ctx, compileOpts...)
}

// checkPointAdapter drops the Delete method that compose.CheckPointStore does not declare.
type checkPointAdapter struct{ inner CheckPointStore }
func (a checkPointAdapter) Get(ctx context.Context, id string) ([]byte, bool, error) {
    return a.inner.Get(ctx, id)
}
func (a checkPointAdapter) Set(ctx context.Context, id string, data []byte) error {
    return a.inner.Set(ctx, id, data)
}
```

**CompiledCanvas struct**：

```go
type CompiledCanvas struct {
    Workflow     compose.Runnable
    CheckPointID string  // 暂时空字符串; V2.1 从 eino Runnable 表面化
}
```

### 7.5 Cancel 协议 (两段式)

```go
// internal/agent/canvas/cancel.go
func Run(ctx context.Context, taskID string, compiled compose.Runnable[...]) error {
    einoCtx, interrupt := compose.WithGraphInterrupt(ctx)
    defer close(stopCh)

    go watchCancel(taskID, func() {
        interrupt(compose.WithGraphInterruptTimeout(30 * time.Second))
    })

    return compiled.Invoke(einoCtx, input,
        compose.WithCheckPointID(genID(taskID)),
        compose.WithWriteToCheckPointID(genID(taskID)),
    )
}

func watchCancel(taskID string, onCancel func()) {
    ticker := time.NewTicker(500 * time.Millisecond)  // 500ms 轮询
    defer ticker.Stop()
    for {
        select {
        case <-stopCh: return
        case <-ticker.C:
            v, _ := redis.Get(context.Background(), fmt.Sprintf("%s-cancel", taskID))
            if v != "" { onCancel(); return }
        }
    }
}
```

**Python 兼容**：`{task_id}-cancel` Redis key 命名与 Python 端 task_service.py 协议**完全一致**。

---

## 8. OpenTelemetry 可观测性 / Observability

### 8.1 总体设计

```
Canvas run goroutine (Go)
   ↓
eino Graph Engine
   ↓ (OnStart / OnEnd / OnError auto-injected)
callbacks.Handler (业务实现)
   ├─ OTelHandler
   │   └─ 开始 span → 注入 attributes → 结束 span
   │       └─ otlphttpexporter → OTel Collector (外部)
   │           ├─ Jaeger / Tempo (trace UI)
   │           ├─ Langfuse (LLM 专门)
   │           └─ Prometheus / Grafana
   └─ SSEHandler (业务事件流) → admin UI
```

### 8.2 双通道分离

| 通道 | 用途 | 协议 | 消费者 |
|------|------|------|--------|
| **SSE** | 业务事件（"node 开始/结束/消息"） | `text/event-stream` HTTP | admin UI |
| **OTel span** | 系统可观测性（节点耗时/错误/token） | OTLP HTTP | 运维/APM |

### 8.3 eino callback → OTel 映射

| eino 时机 | OTel 行为 | Span attribute |
|-----------|-----------|----------------|
| `OnStart(ctx, info, input)` | `tracer.Start(ctx, info.Name)` → 写入 `ctx` | `eino.component.name`, `eino.component.type`, `eino.input.size` |
| `OnEnd(ctx, info, output)` | `span.End()` | `eino.output.size` |
| `OnError(ctx, info, err)` | `span.RecordError(err)` + `span.SetStatus(codes.Error, ...)` | `eino.error.message` |

### 8.4 启动配置

```bash
export OTEL_EXPORTER_OTLP_ENDPOINT="http://otel-collector:4318"
export OTEL_SERVICE_NAME="ragflow-agent"
export OTEL_RESOURCE_ATTRIBUTES="service.namespace=ragflow,deployment.environment=production"
export OTEL_TRACES_SAMPLER="parentbased_traceidratio"
export OTEL_TRACES_SAMPLER_ARG="0.1"  # 10% 采样
```

**降级**：未配置 `OTEL_EXPORTER_OTLP_ENDPOINT` → handler 退化为 noop，不影响业务。

---

## 9. 多版本 Agent 管理 / Multi-version Agents

**Go 端支持多版本并存**（**永不覆盖**）：

| 场景 | 行为 |
|------|------|
| 编辑器保存草稿 | `UPDATE user_canvas SET dsl=? WHERE id=?` (**不创建 version**) |
| 点击"发布" | `INSERT user_canvas_version(...)` 新行；`UPDATE user_canvas SET release=true, dsl=?, update_at=NOW()` |
| Run 不带 version | 拉取**最新** `user_canvas_version` (`create_time DESC LIMIT 1`) |
| Run `?version=v_xxx` | 拉取**指定** `user_canvas_version` |
| Run `?version=draft` | 拉取 `user_canvas.dsl` (编辑器未发布状态) |

**API 端**：

- `GET /api/v1/agents/{id}/versions` — 列表
- `POST /api/v1/agents/{id}/versions` — 显式发布
- `DELETE /api/v1/agents/{id}/versions/{version_id}` — 删除
- `POST /api/v1/agents/{id}/run?version=xxx` — 指定版本运行

---

## 10. 第三方库选型 / Third-party Libraries (License Gate)

### 10.1 决策结论

| 用途 | 选 | License | 备注 |
|------|-----|---------|------|
| **PDF 生成** | `signintech/gopdf` | MIT | 主选；TTF 字体注册 + CJK + header/footer 内置 |
| **DOCX 生成** | **自实现** OOXML writer | — | Go `archive/zip` stdlib + `text/template` + `//go:embed` |
| **Excel 读写** | `xuri/excelize/v2` | BSD-3 | 无 license 风险 |
| **Markdown 解析** | `yuin/goldmark` | MIT | CommonMark 标准 |
| **HTML 解析** | `golang.org/x/net/html` | BSD-3 | stdlib 旁路 |
| **OpenTelemetry SDK** | `go.opentelemetry.io/otel` v1.44.0 | Apache-2.0 | |
| **MySQL driver** | `go-sql-driver/mysql` | MPL-2.0 | ExeSQL 走 stdlib `database/sql` |
| **PG driver** | `lib/pq` | MIT | |
| **MSSQL driver** | `denisenkom/go-mssqldb` | BSD-3 | |
| **Trino driver** | `trinodb/trino-go-client v0.333.0` | Apache-2.0 | ExeSQL Trino dialect |
| **Jinja2 模板** | `nikolalohinski/gonja v1.5.3` | MIT | Phase 8a — 直接 import (from indirect) |
| **Test SQL mock** | `DATA-DOG/go-sqlmock` | MIT | ExeSQL 注入测试 |

### 10.2 AGPL-3 零容忍

RAGFlow 是 Apache-2.0；AGPL-3 强传染会让整个 RAGFlow Go 二进制被迫 AGPL-3 化。所有候选 AGPL-3 库 (unipdf / unioffice / fumiama-go-docx / baliance-gooxml) **全部排除**。

**AGPL-3 预筛规则**：
- README header 含 "AGPL" 或 "Affero" → 直接拒绝
- LICENSE 文件首行含 "Affero General Public License" → 拒绝
- GitHub license badge 显示 AGPL-3.0 / SSPL-1.0 → 拒绝
- CI 中 `go-licenses check` 命中 AGPL → 构建失败

---

## 11. HTTP 接口 / HTTP API

| Method | Path | Handler | 说明 |
|--------|------|---------|------|
| `GET`  | `/api/v1/agents` | `ListAgents` | 已存在 |
| `POST` | `/api/v1/agents` | `CreateAgent` | |
| `GET`  | `/api/v1/agents/{id}` | `GetAgent` | |
| `PATCH`| `/api/v1/agents/{id}` | `UpdateAgent` | |
| `DELETE`| `/api/v1/agents/{id}` | `DeleteAgent` | 级联删除所有 version |
| `POST` | `/api/v1/agents/{id}/run` | `RunAgent` | 同步; `?version=v_xxx` 缺省=最新 |
| `POST` | `/api/v1/agents/{id}/stream` | `StreamAgent` | SSE; emits `message` / `waiting_for_user` / `error` / `done` events |
| `POST` | `/api/v1/agents/{id}/cancel` | `CancelAgent` | 写 Redis cancel key |
| `GET`  | `/api/v1/agents/{id}/versions` | `ListVersions` | |
| `POST` | `/api/v1/agents/{id}/versions` | `PublishVersion` | |
| `GET`  | `/api/v1/agents/{id}/versions/{vid}` | `GetVersion` | |
| `DELETE`| `/api/v1/agents/{id}/versions/{vid}` | `DeleteVersion` | |
| `POST` | `/api/v1/admin/canvas-runtime/:tenant_id` | `AdminRuntime` | 翻转租户 override |

**SSE 事件 payload**：

```text
event: message
data: {"answer": "...", "reference": [...]}

event: waiting_for_user
data: {"cpn_id": "node:userfillup_1"}

event: error
data: {"error": "..."}

event: done
data: [DONE]
```

---

## 12. 验收标准 / Acceptance Criteria

| 类别 | 标准 |
|------|------|
| **功能** | 19 component × ≥3 单测 = ≥57 个 component 单测 |
| **功能** | 21 tool × ≥2 单测 = ≥42 个 tool 单测 |
| **eino 复用** | T1 组件 (LLM/Agent) 回归：跑 eino 自带 react_test.go / chatmodel_test.go / compose_test.go 不退化 |
| **功能** | `{{cpn_id@param}}` 任意节点读任意节点, 单测覆盖 |
| **功能** | SSE 事件序列与 Python `agent_api.py` 一致: `message` / `waiting_for_user` / `error` / `done` |
| **wait-for-user** | Canvas 含 UserFillUp 节点 → 首次运行到 UserFillUp 暂停 → SSE `waiting_for_user` → 用户提交后恢复运行 → 最终输出 `message` + `done` 事件 |
| **RunAgent e2e** | 4 e2e sub-tests: `TestRunAgent_RealCanvas_BeginMessage` / `_CompileFails` / `_InvokeFails` / `_WaitForUserResume` |
| **RunTracker** | miniredis-backed e2e pinning Start → AttachCheckpoint → MarkSucceeded sequence |
| **TTS dispatch** | model-provider integration wired (`audio.NewTTSDispatchFunc`) |
| **per-class timeout** | ExeSQL→3s, TavilySearch→12s, uniform fallback, env override |
| **LLM retry** | MaxRetries=5 → exactly 6 invoker calls (absolute count) |
| **可靠** | Redis 取消协议：cancel → 5s 内节点 stop (500ms 轮询下 p99 ≤ 500ms) |
| **可观测性** | OTel handler P99 overhead < 2% (100 节点) |
| **checkpoint** | Redis `RedisCheckPointStore` Get/Set/Delete 通过 eino 集成测试 |
| **代码质量** | 公共 API 100% godoc 注释；`>=80% test coverage on internal/agent/canvas` |

---

## 13. 风险 & 缓解 / Risks

| 风险 | 严重度 | 缓解 |
|------|--------|------|
| **eino State 在高并发下 mutex 竞争** | 中 | Phase 1 末 benchmark；若 > 5% 调度开销，引入分片 mutex |
| **v1 DSL 100% 兼容不可能** | 中 | 不兼容的旧 DSL 走"自动转换 + 提示"路径 |
| **Tool 外部 HTTP 失败** | 中 | 复用 `http_helper.go` 的 retry |
| **前端 DSL 编辑器只懂 v1** | 中 | Phase 5 维持 v1 写出能力 |
| **测试环境无 LLM key** | 低 | 所有 LLM 组件测试走 mock provider driver |
| **LLM retry multiplicative stacking** | 中 | `retryInvoker.Unwrap()` + `unwrapChatInvoker` 让 MaxRetries = absolute count |
| **CodeExec feature gap vs Python** | 中 | 5 sandbox providers 已 ported；`docs/develop/sandbox-python-go-diff.md` 详细记录 per-provider diff |
| **real TTS binary shape TBD** | 中 | model-provider 60+ driver 路由；real binary 由 model provider 决定 |
| **real MemorySaver 端口 partial** | 低 | partial port；user-deferred |

---

## 14. Future Work

可操作的下一轮跟进项 (按优先级)：

1. **Compile LRU cache** — LRU 按 `(canvasID, versionID, DSL-hash)` 缓存编译产物；仅在 profiling 显示 `Compile` 主导热路径时启动。1-2 周。
2. **Browser Playwright parity** — ✅ Done via `stagehand-go` local mode (see `~/.claude/plans/tingly-weaving-orbit.md`). v1 implements `Sessions.Start → Execute → End` for the natural-language `prompts` task; the v1 fixture loads and emits `content` + `downloaded_files` (always `[]` until the storage write path is ported). `upload_sources` / `persist_session` / `enable_default_extensions` / `chromium_sandbox` accepted in Update but ignored at Invoke. Multi-tenant `MODEL_API_KEY` recycling via process-singleton `StagehandRuntime`.
3. **ExcelProcessor pandas-fidelity audit** — Python 端 15.5K vs Go 当前 happy-path 覆盖。1 天 audit + 修补。
4. **Phase 8b real MemorySaver completion** — 端口 `internal/service/memory_message_service.go` 完整实现。1-2 周，user-deferred。
5. **Phase 5c DB2 support** — CGO + `github.com/ibmdb/go_ibmdb` + native client lib。仅在 e2e 需求浮现时启动。0.5-1 周。
6. **Phase 5d CodeExec 完整对等** — 5 sandbox providers + artifacts/args/timeout/per-language base image 已 ported；file output collection paths, GraphRAG adapter 仍剩余。1-2 周。
7. **Phase 6 gray + Phase 7 cleanup** — per-tenant runtime 灰度切换；`agent_api.py` 标 `@deprecated` + 兼容 proxy shim。2-4 周。
8. **DSL v3** — 类型化表达式 (编译期校验 `{{cpn_id@param}}`)。
9. **eino 生态对齐** — `AddAgenticModelNode` 替换 LLM component; `AddRetrieverNode` 替换 Retrieval component。
10. **GraphRAG component Go 化** (独立项目排期)。
11. **WebSocket 流支持** (pending demand)。
12. **Checkpoint 增强** — 跨 canvas run 复用、增量 checkpoint (仅写 diff channel)。

### Sandbox provider gaps (consolidated from the port diff)

The five Python sandbox providers are ported to Go with functional parity (self_managed, aliyun, local, ssh) and one strict superset (e2b — Go is real, Python is a stub). Admin-panel settings reader lands in `ProviderManager.LoadFromSettings` (see commit history). The remaining 7 gaps are intentional and tracked here:

- **Aliyun Go SDK gaps (v1.1.0)** — ⏸️ **blocked on upstream aliyun SDK.** Two related gaps to revisit when the SDK catches up: (1) `TemplateName` not sent on `CreateCodeInterpreter` (operators must pre-create non-default templates via Python or the aliyun console, then reference by name in metadata); (2) execute uses raw HTTP because the SDK has no execute method (the wire format was reverse-engineered from the Python SDK). Swap to the SDK calls when both APIs land. (1-2 days once the SDK releases; no in-house workaround)
- **`LocalProvider` rlimits not applied** — Go `os/exec` has no portable pre-start hook; rlimits (RLIMIT_AS/CPU/FSIZE/NOFILE) are not enforced. The Go `LocalProvider` is **not a security boundary** — for adversarial code, use `SelfManagedProvider` (executor_manager + gVisor) or `AliyunCodeInterpreterProvider` (cloud microVM). This matches the Python note that "local" is "for development / trusted environments". (no fix planned — by design)
- **`SSHProvider` uses SSH exec, not SFTP** — avoids the `github.com/pkg/sftp` dependency. For workloads with many large artifacts, switch to pkg/sftp if profiling shows exec overhead. (1 day, deferred until profiling shows it matters)
- **Windows build of `LocalProvider`** — `syscall.Setpgid` is POSIX-only. The Go side is `//go:build !windows`; the Python side runs on Windows via `process.kill()`. Tracked; not blocking because RAGFlow production is Linux. (1-2 days, deferred)
- **e2b community Go SDK is a single-maintainer port** — `github.com/eric642/e2b-go-sdk` v0.1.3 (Apache-2.0). Re-evaluate quarterly; fork to `github.com/infiniflow/e2b-go-sdk` if maintenance lags. (1 day fork if needed)
- **OTel spans on provider ops** — providers are log-free; OTel span propagation is on the HTTP client only (via `otelhttp.NewTransport`). Providers themselves do not emit OTel spans. (1 day)

---

## 15. Operations Guide

### 15.1 Boot wiring

`cmd/server_main.go` registers the runtime in three layers:

1. **ProviderManager** (`internal/agent/sandbox/manager.go`) — chooses which sandbox provider backs CodeExec. Default `self_managed`; override via `SANDBOX_PROVIDER_TYPE`. Falls back to env-driven init when the admin-panel settings table is empty/malformed.
2. **RetrievalService** (`internal/agent/tool/retrieval_service.go`) — `nlp.NewRetrievalService(docEngine, docDAO)` and `kg.NewRetrieval(...)` are wired via `tool.SetRetrievalService(...)` / `tool.SetKGRetrievalService(...)` at boot. The first backs `use_kg=false`; the second backs `use_kg=true`.
3. **AgentService** (`internal/service/agent.go`) — accepts optional Redis-backed CheckPointStore / StateSerializer / RunTracker via `NewAgentServiceWithOptions(...)`. Boot installs these when Redis is up; otherwise the fields stay nil and the service falls back to in-memory mode (transparent to callers).

Any layer that is not wired at boot produces a loud-fail sentinel (see §15.3) — stubs never silently return empty results.

### 15.2 Feature flags

| Env var | Default | Effect |
|---------|---------|--------|
| `SANDBOX_PROVIDER_TYPE` | `self_managed` | One of `self_managed` / `aliyun_codeinterpreter` / `e2b` / `local` / `ssh` |
| `SANDBOX_EXECUTOR_MANAGER_URL` | `http://sandbox-executor-manager:9385` | self-managed endpoint |
| `SANDBOX_EXECUTOR_MANAGER_TIMEOUT` | `30` (s) | self-managed per-call timeout |
| `AGENTRUN_*` (5 vars) | n/a | aliyun code interpreter |
| `E2B_API_KEY` / `E2B_ACCESS_TOKEN` | n/a | e2b (one required) |
| `E2B_TEMPLATE` | `base` | e2b sandbox template |
| `LOCAL_*` (8 vars) | n/a | local subprocess |
| `SSH_HOST` / `SSH_PORT` / `SSH_USERNAME` / `SSH_PASSWORD` / `SSH_PRIVATE_KEY` / `SSH_PRIVATE_KEY_PATH` | n/a | SSH provider |
| `COMPONENT_EXEC_TIMEOUT` | `600` (s) | canvas-level per-invocation timeout; per-class overrides via env-derived map (see `canvas/timeout.go`) |

### 15.3 Known deferred items (loud-fail sentinels)

| Sentinel | Cause | Fix |
|----------|-------|-----|
| `ErrRetrievalServiceMissing` | `tool.SetRetrievalService(...)` not called at boot | Wire `nlp.NewRetrievalService` at boot (default in `cmd/server_main.go`) |
| `ErrKGRetrievalServiceMissing` | Canvas uses `use_kg=true` and `tool.SetKGRetrievalService(...)` not called | Wire `kg.NewRetrieval(...)` at boot (default in `cmd/server_main.go`) |
| `ErrMemoryServiceMissing` | `component.SetMemorySaver(...)` not called at boot | Wire `NewMemoryMessageService(...)` (default in `cmd/server_main.go`) |
| `ErrEmbedderNotWired` | MemorySaver reached but no embedder configured | Port the embedding model — see §14 |
| `ErrSandboxNotConfigured` | `SANDBOX_PROVIDER_TYPE` set to unknown value | Set to one of the 5 supported values |
| `ErrE2BProviderNotImplemented` | `SANDBOX_PROVIDER_TYPE=e2b` and no `E2B_API_KEY`/`E2B_ACCESS_TOKEN` | Provide one of the two env vars |
| `ErrTTSEngineNotConfigured` | Message runs with `auto_play=true` and no `audio.SetSynthesizer(...)` | Wire a TTS engine at boot — see §14 |
| `ErrExeSQLUnsupportedDB` | `db_type` is `trino` or `ibm db2` | Add the driver registration — see §14 |

### 15.4 Canvas migration (Python → Go)

`tools/migrate-canvas` cross-validates Python's `normalize_chunker_dsl` against Go's `NormalizeForCanvas`. Manual equivalent until the tool ships:

1. Export canvas JSON from Python: `GET /api/v1/canvas/<id>/export`.
2. Validate Python normalizer: `uv run python -c "from agent.canvas import normalize_chunker_dsl; print(normalize_chunker_dsl(json.load(open('canvas.json'))))"`.
3. Validate Go normalizer: `go test ./internal/agent/dsl/ -run TestNormalize -v` (uses fixtures in `internal/agent/dsl/testdata/`).
4. Diff the two normalized forms. If structurally identical, the canvas is Go-portable.

### 15.5 Testing

```sh
go test -count=1 ./internal/agent/...           # all agent tests
go test -count=1 ./internal/agent/component/   # component tests
go test -count=1 ./internal/agent/tool/         # tool tests + retrieval + sandbox providers
go test -count=1 ./internal/agent/sandbox/     # 5 sandbox providers + manager
go test -count=1 ./internal/agent/canvas/      # canvas engine, parallel, interrupt/resume
go test -count=1 ./internal/agent/runtime/     # state, template, history window
```

Fixtures: `internal/agent/dsl/testdata/` (7 JSONs) drive the e2e suite and match the input corpus Python's `normalize_chunker_dsl` accepts.

---

## 附录 A · 关键文件 / Key Files

| 设计点 | 关键文件 |
|--------|---------|
| **State 模式** | `internal/agent/canvas/{state.go, scheduler.go}` + `internal/agent/runtime/{state.go, context.go, template.go, template_jinja.go}` |
| **CanvasState MarshalJSON** | `internal/agent/runtime/state.go` |
| **runtime 提取** | `internal/agent/runtime/*.go` (8 文件) + `internal/agent/canvas/state_export.go` |
| **Loop 宏展开** | `internal/agent/canvas/loop_subgraph.go` + `internal/agent/component/loop.go` (no-op marker) |
| **Parallel** | `internal/agent/component/parallel.go` + `internal/agent/workflowx/parallel.go` |
| **Loop 通用节点** | `internal/agent/workflowx/loop.go` + `loop_*_test.go` |
| **Interrupt 路径** | `internal/agent/canvas/interrupt_resume.go` + `internal/agent/canvas/runner.go` |
| **Checkpoint** | `internal/agent/canvas/{checkpoint_store.go, run_tracker.go, state_serializer.go, compile.go}` |
| **Compile 适配** | `internal/agent/canvas/compile.go` (checkPointAdapter) |
| **Per-class timeout** | `internal/agent/canvas/timeout.go` + `node_body.go` |
| **Cancel 协议** | `internal/agent/canvas/cancel.go` |
| **OTel** | `internal/observability/otel/{provider.go, handler.go, handler_test.go}` |
| **DSL normalize** | `internal/agent/dsl/{normalize.go, normalize_test.go}` + `testdata/` |
| **Tool registry** | `internal/agent/tool/{registry.go, http_helper.go, ssrf.go, mcp.go, retrieval*.go}` |
| **Component 5-tier** | `internal/agent/component/{base.go, registry.go, runtime_wire.go, fixture_stubs.go, universe_a_wrappers.go}` + 19 component .go |
| **AgentService V2** | `internal/service/agent.go` (buildRunFunc) + `internal/service/canvas_decode.go` + `internal/service/agent_run_e2e_test.go` |
| **Sandbox providers** | `internal/agent/sandbox/{self_managed.go, aliyun.go, e2b.go, local.go, ssh.go, manager.go}` + `tool/sandbox_bridge.go` |
| **TTS dispatch** | `internal/agent/audio/{tts.go, tts_dispatch.go, model_provider_synthesizer.go}` |

## 附录 B · 测试覆盖 / Test Coverage

| 包 | 测试文件数 | 覆盖点 |
|----|-----------|--------|
| `internal/agent/canvas` | 17 | `canvas_test.go, scheduler_test.go, state_test.go, variable_test.go, state_bench_test.go, state_serializer_test.go, checkpoint_store_test.go, run_tracker_test.go, cancel_test.go, stream_test.go, loop_subgraph_test.go, loop_semantics_test.go, dsl_examples_e2e_test.go, interrupt_resume_test.go, multibranch_test.go, node_body_timeout_test.go, node_body_per_class_timeout_integration_test.go, parallel_batch_test.go, parallel_timing_test.go` |
| `internal/agent/component` | 50+ | 各 component `_test.go` + `verify_p1_test.go` + `production_chain_fixes_test.go` |
| `internal/agent/tool` | 30+ | 各 tool `_test.go` + `registry_test.go` + `retrieval_nlp_test.go` + `retrieval_kg_test.go` + `exesql_trino_test.go` + `exesql_unsupported_test.go` + `http_helper_test.go` + `ssrf_test.go` + `mcp_test.go` |
| `internal/agent/runtime` | 4 | `metrics_test.go, selector_test.go, state_test.go, template_jinja_test.go` |
| `internal/agent/workflowx` | 8 | `loop_test.go, loop_options_test.go, loop_integration_test.go, loop_example_test.go, parallel_test.go, parallel_options_test.go, parallel_integration_test.go, parallel_helpers_test.go` |
| `internal/agent/dsl` | 1 | `normalize_test.go` |
| `internal/agent/audio` | 3 | `model_provider_synthesizer_test.go, tts_dispatch_test.go, tts_test.go` |
| `internal/agent/sandbox` | 6 | `e2b_test.go, local_test.go, manager_test.go, result_protocol_test.go, self_managed_test.go, ssh_test.go` |
| `internal/observability/otel` | 1 | `handler_test.go` (tracetest.SpanRecorder) |
| `internal/service` | 8+ | `canvas_decode_test.go, agent_run_e2e_test.go, agent_test.go, agent_sessions_test.go, chat_session_test.go, ...` |
| `internal/handler` | 10+ | `agent_test.go, agent_wait_for_user_test.go, admin_runtime_test.go, ...` |

## 附录 C · Deepdoc Service Endpoints (DLA/OCR/TSR)

### C.1 Endpoint summary

| Endpoint | URL | Status | Go port need |
|----------|-----|--------|--------------|
| DLA (Document Layout Analysis) | `POST {DEEPDOC_URL}/predict` | Remote HTTP (via `dla_cli.py`) | Go client with 3-retry + 18s timeout |
| OCR | **No remote endpoint** | Local ONNX only | None — `ErrNotImplemented` stub |
| TSR (Table Structure Recognition) | **No remote endpoint** | Local ONNX only | None — `ErrNotImplemented` stub |

Single toggle: `DEEPDOC_URL` (preferred) or `TENSORRT_DLA_SVR` (legacy).

### C.2 DLA HTTP contract

- **Method**: `POST {DEEPDOC_URL}/predict`
- **Body**: `multipart/form-data`, field name `request`, raw JPEG bytes
- **Response**: `{"bboxes": [[left, top, right, bottom, score, type_idx], ...]}`
- **Timeout**: 18s per request; **3 retries** per image
- **Failure sentinel**: empty list `[]`

## 附录 D · DSL v1 Corner Cases Inventory

### D.1 Top-level DSL shape

```json
{
  "components": {
    "<cpn_id>": {
      "obj": {"component_name": "Retrieval", "params": {...}},
      "downstream": ["generate_0"],
      "upstream": ["answer_0"]
    }
  },
  "path": ["begin"],
  "history": [],
  "retrieval": {"chunks": [], "doc_aggs": []},
  "globals": {"sys.query": "", "sys.user_id": "...", "sys.conversation_turns": 0,
              "sys.files": [], "sys.history": [], "sys.date": "..."},
  "variables": {},
  "memory": []
}
```

### D.2 Variable reference syntax

Two regexes:
```
variable_ref_patt    = r"\{* *\{([a-zA-Z:0-9]+@[A-Za-z0-9_.-]+|sys\.[A-Za-z0-9_.]+|env\.[A-Za-z0-9_.]+)\} *\}*"
iteration_alias_patt = r"\{* *\{(item|index|result)\} *\}*"
```

Key behaviors the Go port must mirror:
- **Brace tolerance**: `{{var}}`, `{{ var }}`, `{{{var}}}` are all valid
- **`sys.*`/`env.*`**: namespace-only (no `@`), read from `State` flat namespace
- **`cpn_id@param.nested.path`**: dot-path traversal with `json.loads` on strings, `dict.get`, `list[int]` index
- **Empty `{{...}}`**: resolves to `""`, never crashes
- **`is_reff`**: returns `True` only if `cpn_id@param` resolves to a known component

### D.3 Component-name case-insensitivity

All comparisons use `.lower()`. Stored cpn_ids may be any case. Go port must NOT key component map by case-sensitive `cpn_id`.

## 附录 E · Component & Tool Interface Inventory

### E.1 Component inventory (22 → 19 active)

| # | Component | File | `component_name` | Tier | Key behavior |
|---|-----------|------|-----------------|------|--------------|
| 1 | Begin | `begin.py` | `Begin` | T3 | Consumes `kwargs["inputs"]`, resolves file inputs via FileService |
| 2 | UserFillUp | `fillup.py` | `UserFillUp` | T3 | Renders `tips` with variable interpolation; eino interrupt |
| 3 | Fillup | (alias) | `Fillup` | T3 | Thin alias of UserFillUp (disable `enable_tips`) |
| 4 | Message | `message.py` | `Message` | T3 | jinja2 prompt + stream + TTS + filegen + memory save |
| 5 | LLM | `llm.py` | `LLM` | T1 | Sync + async paths; chatModel.Generate / Stream; structured JSON output |
| 6 | Categorize | `categorize.py` | `Categorize` | T3 | LLM one-shot classification → `_next` (routing list) + `category_name` |
| 7 | Switch | `switch.py` | `Switch` | T2 | 12 operators; `_next` = matching downstream(s) |
| 8 | Agent | `agent_with_tools.py` | `Agent` | T1 | ReAct loop with LLMBundle + tool binding + citations |
| 9 | Iteration | `iteration.py` | `Iteration` | T4 | Compat stub → Parallel (Go) |
| 10 | IterationItem | `iterationitem.py` | `IterationItem` | T4 | Compat stub |
| 11 | Loop | `loop.py` | `Loop` | T4 | workflowx.AddLoopNode (Go) |
| 12 | LoopItem | `loopitem.py` | `LoopItem` | (none) | Engine-handled, not registered |
| 13 | ExitLoop | `exit_loop.py` | `ExitLoop` | (none) | `legacyNoOpNames` (Go) |
| 14 | Invoke | `invoke.py` | `Invoke` | T3 | HTTP GET/POST/PUT/PATCH/DELETE + headers/proxy/timeout |
| 15 | Browser | `browser.py` | `Browser` | T3 | LLM-driven browsing |
| 16 | DataOperations | `data_operations.py` | `DataOperations` | T3 | 7 ops: select_keys/literal_eval/combine/filter/append_or_update/remove/rename |
| 17 | ListOperations | `list_operations.py` | `ListOperations` | T3 | 6 ops: nth/head/tail/filter/sort/drop_duplicates |
| 18 | StringTransform | `string_transform.py` | `StringTransform` | T3 | split/merge/jinja2 template ops |
| 19 | VariableAggregator | `variable_aggregator.py` | `VariableAggregator` | T3 | Returns first non-empty in each variable group |
| 20 | VariableAssigner | `variable_assigner.py` | `VariableAssigner` | T3 | 11 ops |
| 21 | DocsGenerator | `docs_generator.py` | `DocGenerator` | T5 | MD → PDF/DOCX/TXT/MD/HTML |
| 22 | ExcelProcessor | `excel_processor.py` | `ExcelProcessor` | T5 | pandas read/write/merge/convert |

### E.2 Tool inventory (21)

All tools extend `ToolBase`, expose `get_meta()` (OpenAI function-call schema), `_invoke`/`_invoke_async`.

| # | Tool | `component_name` | Behavior |
|---|------|-----------------|----------|
| 1 | AkShare | `AkShare` | Chinese financial data (HTTP) |
| 2 | ArXiv | `ArXiv` | `export.arxiv.org/api/query` search |
| 3 | CodeExec | `CodeExec` | gRPC client to Python sandbox; **5 sandbox providers** in `internal/agent/sandbox/` |
| 4 | Crawler | `Crawler` | Generic HTML scraper |
| 5 | DeepL | `DeepL` | DeepL Translate API (HTTP) |
| 6 | DuckDuckGo | `DuckDuckGo` | `html.duckduckgo.com/html` search |
| 7 | Email | `Email` | SMTP send via `smtplib` |
| 8 | ExeSQL | `ExeSQL` | MySQL/PG/MSSQL/Trino/OceanBase via stdlib `database/sql` |
| 9 | GitHub | `GitHub` | GitHub REST API search |
| 10 | Google | `Google` | SerpAPI / Google CSE search |
| 11 | GoogleScholar | `GoogleScholar` | Scholar via SerpAPI |
| 12 | Jin10 | `Jin10` | Chinese financial news feed |
| 13 | PubMed | `PubMed` | NCBI E-utilities |
| 14 | QWeather | `QWeather` | HeFeng weather API |
| 15 | Retrieval | `Retrieval` | nlp.Dealer + kg.Retrieval (Go dual-registry) |
| 16 | SearXNG | `SearXNG` | Meta-search |
| 17 | TavilySearch | `TavilySearch` | Tavily search API |
| 18 | TavilyExtract | `TavilyExtract` | Tavily extract API |
| 19 | TuShare | `TuShare` | Tushare Chinese financial data |
| 20 | WenCai | `WenCai` | 同花顺 问财 stock Q&A |
| 21 | Wikipedia | `Wikipedia` | Wikipedia REST API |
| 22 | YahooFinance | `YahooFinance` | Yahoo Finance unofficial API |
| — | MCP | (server_id) | `MCPToolAdapter` over streamable-HTTP |

## 附录 F · Open Questions (actionable)

| ID | Question | Action | Effort |
|----|----------|--------|--------|
| OQ #1 | Iteration semantic preservation | ✅ Done — engine design | — |
| OQ #2 | MCP tool priority | ✅ Done — thin wrapper | — |
| OQ #3 | DSL normalization | ✅ Done — Go-side + `tools/migrate-canvas` built | — |
| OQ #4 | History window behavior | ✅ Done — canvas-level session | — |
| OQ #5 | Citation injection scope | ✅ Done — LLM + Agent | — |
| OQ #6 | Component timeout granularity | ✅ Done — per-class table is a Go enhancement over Python's uniform 600s | — |
| OQ #7 | Universe A/B naming asymmetry | ✅ Done — keep dual-naming convention | — |
| OQ #8 | GraphRAG scope | ✅ Done — KGRetrievalAdapter wired | — |
| OQ #9 | `generate` legacy alias | ⏸️ Deferred | — |
| OQ #10 | Phase 5a vs 5b ordering | ✅ Done — single Retrieval milestone | — |
| OQ #11 | Per-component env-driven timeout | ✅ Done — canvas-level uniform 600s | — |
| OQ #12 | Embedding model port | ✅ Done — model provider architecture | — |
| OQ #13 | Switch operator coverage | ✅ Done — 12/12 | — |
| OQ #14 | Universe A `SearchMyDataset` alias | ✅ Done — 4 spellings | — |
| OQ #15 | LLM `max_retries` / `delay_after_error` | ✅ Done — `retryInvoker.Unwrap()` normal-absolute-count | — |
| OQ #16 | Phase 4.4 orchestrator side | ✅ Done — Runner.Run catches interrupt | — |
| OQ #17 | Phase 5d CodeExec full feature parity | ⏸️ Partial — 5 providers + artifacts/args/timeout/per-language base image done; GraphRAG adapter remains | 1-2 weeks |
| OQ #18 | Phase 8b real TTS engine | ✅ Done — dispatcher routes through 60+ model drivers, no shell-out needed | — |
| OQ #19 | Phase 8b real MemorySaver completion | ⏸️ Open | 1-2 weeks |
| OQ #20 | Phase 5c DB2 e2e demand | ⏸️ Open (CGO + native lib) | 0.5-1 week if needed |
| OQ #21 | Compile LRU cache | ⏸️ Open — defer until profiling | 1-2 weeks |
| OQ #22 | Phase 6 component hardening | ✅ Done — Browser on stagehand-go local mode (see `~/.claude/plans/tingly-weaving-orbit.md`); `downloaded_files` / `upload_sources` / `persist_session` deferred to follow-up PRs; ExcelProcessor audit still ⏸️ Open | — |
| OQ #23 | `tools/gen-component-parity` script | ✅ Done | — |

---

> **Last verified**: 2026-06-17