From 4afcd62375055b1a129d3d0adb1e7e268c8c7332 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=97=B6=E4=B9=8B?= Date: Thu, 13 Aug 2026 14:39:14 +0800 Subject: [PATCH 1/2] docs: add Qoder Cloud runtime integration best practice and demo MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: 时之 --- .../integrate-qoder-cloud-runtime/index.md | 75 +++++++ demos/integrate-qoder-cloud-runtime/README.md | 64 ++++++ demos/integrate-qoder-cloud-runtime/bridge.go | 123 +++++++++++ .../bridge_test.go | 135 ++++++++++++ .../dispatcher.go | 195 ++++++++++++++++++ .../dispatcher_test.go | 79 +++++++ demos/integrate-qoder-cloud-runtime/go.mod | 3 + demos/integrate-qoder-cloud-runtime/stream.go | 147 +++++++++++++ 8 files changed, 821 insertions(+) create mode 100644 content/zh-CN/best-practices/integrate-qoder-cloud-runtime/index.md create mode 100644 demos/integrate-qoder-cloud-runtime/README.md create mode 100644 demos/integrate-qoder-cloud-runtime/bridge.go create mode 100644 demos/integrate-qoder-cloud-runtime/bridge_test.go create mode 100644 demos/integrate-qoder-cloud-runtime/dispatcher.go create mode 100644 demos/integrate-qoder-cloud-runtime/dispatcher_test.go create mode 100644 demos/integrate-qoder-cloud-runtime/go.mod create mode 100644 demos/integrate-qoder-cloud-runtime/stream.go diff --git a/content/zh-CN/best-practices/integrate-qoder-cloud-runtime/index.md b/content/zh-CN/best-practices/integrate-qoder-cloud-runtime/index.md new file mode 100644 index 0000000..eea6c59 --- /dev/null +++ b/content/zh-CN/best-practices/integrate-qoder-cloud-runtime/index.md @@ -0,0 +1,75 @@ +--- +schema_version: 1 +slug: integrate-qoder-cloud-runtime +title: 把 Qoder Cloud 接成你自己系统里的一名"员工" +summary: 让 Qoder Cloud Agent 在托管沙箱里干活，同时把你系统的业务能力做成它能调用的工具——权限、校验、边界全攥在你自己手里。 +type: best-practice +category: build-deploy +tags: + - runtime + - session + - sse + - tool-use + - idempotency +author: + name: 时之 + github: yefengzi7 +locale: zh-CN +--- + +## 适用场景与边界 + +假设你已经有一套派活系统——把任务丢给不同的编码 Agent 去做。现在你想再"招"一个：让它在云上的沙箱里跑代码，同时还能读写你系统里的工单、改状态、留评论，像个真的团队成员。 + +这就是把 Qoder Cloud 接成一种运行时的意义。两个方向的活儿它都能干：一边是内置的 Bash、文件工具在云端托管沙箱里跑（我实测时它自己写了个 Python 脚本、跑起来、把输出贴回来，整个过程在云容器里完成，不碰我本地一根汗毛）；另一边，是你把自己的业务能力做成"自定义工具"递过去——它想查工单，就发个请求，你的进程执行一次白名单内的操作，再把结果塞回同一个会话。 + +我拿它真跑过多轮：写文件、跑命令、Resume 会话接着上一轮的上下文干、中途 Cancel。云沙箱这半边确实省心——不用自己维护运行环境。 + +但有一条边界得先说清楚，不然会有错误预期：**云端那个 Bash 和文件工具，跑在托管容器里，够不着你的宿主机、代码检出、本地文件系统。** 别指望用它直接操作你本地的东西——它就是个隔离的沙箱。 + +另一条边界是你自己划的：自定义工具的白名单是硬门槛。只放你想清楚了、能校验、出事能查的那几个操作，其余一律拒。先支持定义清楚的场景（单个工单、对话），别急着开批量、定时、多 Agent 那些边界还没理清的面——那些等你有专门的 schema 和生命周期处理再说。 + +## 推荐做法 + +接这套东西，真正的难点就三件：**断了能接上、协议不乱来、令牌不串门。** + +| 决策 | 推荐方式 | 原因 | +|---|---|---| +| 事件传输 | SSE 消费，记住最新事件 ID，断线用 `Last-Event-ID` 续，重复的按 ID 去掉 | 网抖一下不至于丢事件或重复处理 | +| 会触发副作用的请求 | 带幂等键，只对 429/5xx 有限重试 | 重试不会把同一件事干两遍 | +| 一批自定义工具 | 等会话 idle 且 `stop_reason=requires_action` 把整批亮出来，先整批校验再一个个执行 | 不出现"干一半"；有问题的批次在动手前就被拦 | +| 结果回传 | 每个结果发回会话；游标只在整批成功后才前移 | 断线重放能补发没送到的结果，但不会把改数据的工具再跑一遍 | +| 兜底 | 未知动作、空批、还有活没干完就到终态、要人工授权的工具——一律当失败处理 | 拿不准的时候，宁可拒 | +| 令牌 | 云端 PAT 只在云 API 客户端里；业务任务令牌单独拿 | 任务令牌绝不进云请求、Prompt、工具输入输出、日志 | + +这里面最容易写错、也最值得画出来的，是自定义工具的执行时序。云端不是发一个工具请求你就立刻执行——它会先把这一回合要用的工具攒齐，等到 idle 且声明 `requires_action` 时，才把整批亮给你： + +```mermaid +flowchart LR + A[收到工具请求] --> B[先缓冲，不执行] + B --> C[等 idle 且 requires_action] + C --> D[整批校验] + D --> E[按顺序逐个执行] + E --> F[逐个回传结果] + F --> G[整批成功后前移游标] +``` + +为什么要这么绕？因为它让"恰好执行一次"变得可能。每个工具的结果执行完先缓存下来，万一 SSE 断了重连、或者结果发出去但没收到确认，重放时你补发的是**缓存的结果**，而不是把那个改数据的工具再执行一遍。改一次工单状态和改两次，区别可大了。 + +校验这块别手软：只认白名单里的固定工具名，多一个 JSON 字段都拒，取值、长度、任务范围挨个查。一个工单任务就只能碰它被指派的那个工单——就算云端模型脑子一热请求去改别的工单，这道墙也守在你自己进程里，它越不过去。 + +## 验证与维护 + +这类桥接靠"跑一次成功了"是不能算数的，得靠**能反复跑、覆盖到各种糟糕情况的确定性测试**来守。真连云端的冒烟测试是补充，不是主力。 + +- **用 mock 服务端把协议打一遍。** 建会话、发消息、SSE 流、整批 `requires_action`、结果回传、断线续传，全用本地假服务器覆盖。跑得起来还不够，要开竞态检测——并发问题不开这个根本看不见。 +- **专门造一个"半路失败"。** 第一个结果成功了、第二个返回 503 然后重连——这时候你要验证的是：重放只补发没送达的那个，绝不会把已经成功的改数据工具再跑一遍。这个场景不主动造，正常测试永远覆盖不到。 +- **把兜底逐个验证。** 未知动作 ID、空批、还有工具没执行完就到了 idle——这些都该判失败，而不是当成功放过去。 + +维护上有句话我想写在最显眼的地方：**别把"恰好一次"吹得比它实际能保证的大。** 进程内的恰好一次，只在这个进程活着的时候成立。进程崩了重启、或者结果发出去被云端收了但你本地没记上——这些跨进程、需要对账的情况，是接真实系统时你得自己补的功课。把它当已知边界写进文档，比假装已经解决了要诚实得多，也省得后面有人踩。 + +## 可选：Demo 源码 + +Demo 是个纯 Go 标准库、零依赖的小模块，把上面这套照着能跑的样子实现了一遍：白名单分派器、`requires_action` 整批协议、断线续传的 SSE 读取器。测试里专门有"半路 503 后重连、重放不重复执行"那一幕。跑法在 README 里，`go test` 直接看结果。 + +[查看 Demo 源码](https://github.com/QoderAI/cloud-agents-cookbook/tree/main/demos/integrate-qoder-cloud-runtime) diff --git a/demos/integrate-qoder-cloud-runtime/README.md b/demos/integrate-qoder-cloud-runtime/README.md new file mode 100644 index 0000000..98f3260 --- /dev/null +++ b/demos/integrate-qoder-cloud-runtime/README.md @@ -0,0 +1,64 @@ +# 接入 Qoder Cloud 运行时 Demo + +一个自包含的 Go 模块，演示如何把业务能力以客户端自定义工具的形式安全回接给 Qoder Cloud Agent：白名单分派器（校验 + 任务范围）、`requires_action` 整批协议（exactly-once + 重放不重复执行 + fail-closed），以及带 `Last-Event-ID` 续传和去重的 SSE 读取器。纯 Go 标准库，无第三方依赖。 + +## 对应文章 + +- 标题：把 Qoder Cloud 接成你自己系统里的一名“员工” +- Slug：`integrate-qoder-cloud-runtime` + +## 关于来源 + +这个模式提炼自 [Multica](https://github.com/multica-ai/multica) —— 一个开源的多 Agent 协作工作台，它把 Qoder Cloud 作为其中一种 Agent 运行时接入。本目录是把那套集成里最关键的协议部分（分派器 + 整批协议 + SSE 续传）提炼成一个能独立编译、能跑测试的最小模块，方便研读和复用。完整的产品与周边实现见 Multica 项目本体。 + +## 前置条件 + +- Go 1.22 或更高版本。 +- 无第三方依赖，无需网络：所有测试用进程内 fake store 与 `httptest` mock 服务端运行。 + +## 安装与配置 + +无需额外配置。模块路径为 `example.com/qoder-cloud-runtime-demo`，仅依赖 Go 标准库。 + +```bash +go mod verify +``` + +## 运行 + +运行完整测试套件： + +```bash +go test ./... +``` + +在支持 cgo 的环境上可开启竞态检测（对应文章推荐的做法）： + +```bash +CGO_ENABLED=1 go test -race ./... +``` + +## 验证结果 + +`go test ./...` 应全部通过，覆盖以下行为： + +| 测试 | 验证的行为 | +|---|---| +| `TestDispatchRejectsInvalidInput` | 未知工具、未知字段、非 UUID、非法枚举、空更新一律在触达业务前被拒 | +| `TestDispatchEnforcesIssueScope` | 工单任务只能读写被指派的那个工单 | +| `TestBatchRunsOnceThenReplaysWithoutReexecution` | 整批执行一次；重放重发缓存结果、不重复执行变更类工具 | +| `TestBatchFailsClosed` | 空批次、未知动作、带未决工具的终态 idle 均 fail-closed | +| `TestStreamResumesWithLastEventIDAndDeduplicates` | 断线用 `Last-Event-ID` 续传，重放事件被去重，事件恰好各处理一次 | + +预期输出以 `ok example.com/qoder-cloud-runtime-demo` 结尾。 + +## 清理资源 + +Demo 全程在进程内运行，不创建任何外部资源，无需清理。删除 Go 构建缓存可执行 `go clean -testcache`。 + +## 成本与安全 + +- 本 Demo 不进行任何真实网络调用，也不消耗账号额度：`httptest` 服务端在本地进程内运行。 +- 代码中不含任何真实凭证。示例里的 `placeholder-token` 仅用于占位。 +- 分派器是 fail-closed 的白名单：接入真实后端时，请把云端 PAT 与业务任务令牌分别持有，任务令牌绝不写入云请求、Prompt、工具输入/结果或日志。 +- 进程内的“恰好一次”只在单进程生命周期内成立；跨进程崩溃的调用 ID 幂等属于接入真实系统时需自行加固的部分。 diff --git a/demos/integrate-qoder-cloud-runtime/bridge.go b/demos/integrate-qoder-cloud-runtime/bridge.go new file mode 100644 index 0000000..332c317 --- /dev/null +++ b/demos/integrate-qoder-cloud-runtime/bridge.go @@ -0,0 +1,123 @@ +// SPDX-License-Identifier: Apache-2.0 + +package bridge + +import ( + "fmt" + "sync" +) + +// ToolRequest is one buffered custom_tool_use emitted by the cloud Agent. +// EventID is the source event ID, reused as the tool-use ID and as the +// idempotency key for exactly-once execution within this process. +type ToolRequest struct { + EventID string + Tool string + RawInput []byte +} + +// ToolResult is what the bridge posts back as user.custom_tool_result. +type ToolResult struct { + EventID string + Output string + IsError bool +} + +// BatchRunner buffers custom-tool requests, then executes a declared batch +// exactly once, caching results so an SSE reconnect or an ambiguous result +// POST can resend an unsent result without re-executing a mutating tool. +// +// It fails closed: an unknown declared event, an empty batch, or a terminal +// idle with unresolved custom tools all return an error before any execution. +type BatchRunner struct { + store IssueStore + scope TaskScope + + mu sync.Mutex + buffered map[string]ToolRequest // event ID -> request, filled as events arrive + results map[string]ToolResult // event ID -> cached result, for replay + executed map[string]bool // event ID -> executed, guards exactly-once +} + +// NewBatchRunner creates a runner bound to one task scope. +func NewBatchRunner(store IssueStore, scope TaskScope) *BatchRunner { + return &BatchRunner{ + store: store, + scope: scope, + buffered: map[string]ToolRequest{}, + results: map[string]ToolResult{}, + executed: map[string]bool{}, + } +} + +// Buffer records an incoming custom_tool_use without executing it. The bridge +// emits a local tool-use message here and waits for requires_action to declare +// which buffered events form the batch. +func (r *BatchRunner) Buffer(request ToolRequest) { + r.mu.Lock() + defer r.mu.Unlock() + r.buffered[request.EventID] = request +} + +// RunBatch executes the declared batch once, in the given order, and returns +// the results to post back. Calling it again with the same event IDs resends +// cached results without re-executing — this is the replay path after a +// reconnect or an ambiguous POST. It fails closed on an empty or unknown batch. +func (r *BatchRunner) RunBatch(eventIDs []string) ([]ToolResult, error) { + r.mu.Lock() + defer r.mu.Unlock() + + if len(eventIDs) == 0 { + return nil, fmt.Errorf("empty action batch: failing closed") + } + // Validate the whole declared batch before running anything. + for _, id := range eventIDs { + if _, ok := r.buffered[id]; !ok { + return nil, fmt.Errorf("unknown required action %q: failing closed", id) + } + } + + out := make([]ToolResult, 0, len(eventIDs)) + for _, id := range eventIDs { + if r.executed[id] { + // Replay: resend the cached result, never re-execute. + out = append(out, r.results[id]) + continue + } + request := r.buffered[id] + output, err := Dispatch(r.store, r.scope, request.Tool, request.RawInput) + result := ToolResult{EventID: id, Output: output} + if err != nil { + result.Output = err.Error() + result.IsError = true + } + // Mark executed and cache the result BEFORE returning, so a later + // replay of the same batch cannot run a mutating tool twice. + r.executed[id] = true + r.results[id] = result + out = append(out, result) + } + return out, nil +} + +// FinalizeIdle is called when the turn reaches a terminal idle. If any buffered +// tool never got executed, the run is incomplete and must fail closed rather +// than silently declaring success. +func (r *BatchRunner) FinalizeIdle() error { + r.mu.Lock() + defer r.mu.Unlock() + for id := range r.buffered { + if !r.executed[id] { + return fmt.Errorf("terminal idle with unresolved custom tool %q: failing closed", id) + } + } + return nil +} + +// ExecutedCount reports how many distinct tools actually ran. Tests use it to +// prove replay did not re-execute. +func (r *BatchRunner) ExecutedCount() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.executed) +} diff --git a/demos/integrate-qoder-cloud-runtime/bridge_test.go b/demos/integrate-qoder-cloud-runtime/bridge_test.go new file mode 100644 index 0000000..47c2a26 --- /dev/null +++ b/demos/integrate-qoder-cloud-runtime/bridge_test.go @@ -0,0 +1,135 @@ +// SPDX-License-Identifier: Apache-2.0 + +package bridge + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "testing" +) + +func TestBatchRunsOnceThenReplaysWithoutReexecution(t *testing.T) { + store := &fakeStore{} + runner := NewBatchRunner(store, TaskScope{Kind: TaskIssue, AssignedIssueID: assignedID}) + + // Two buffered custom_tool_use events form one declared batch. + runner.Buffer(ToolRequest{EventID: "ev-1", Tool: "multica_update_issue", RawInput: []byte(`{"issue_id":"` + assignedID + `","status":"in_progress"}`)}) + runner.Buffer(ToolRequest{EventID: "ev-2", Tool: "multica_add_issue_comment", RawInput: []byte(`{"issue_id":"` + assignedID + `","content":"working on it"}`)}) + + batch := []string{"ev-1", "ev-2"} + first, err := runner.RunBatch(batch) + if err != nil { + t.Fatalf("first run: %v", err) + } + if len(first) != 2 || first[0].IsError || first[1].IsError { + t.Fatalf("unexpected first results: %+v", first) + } + if store.updates != 1 || store.comments != 1 { + t.Fatalf("expected one update + one comment, got updates=%d comments=%d", store.updates, store.comments) + } + + // Replay the same batch (as after an SSE reconnect or ambiguous POST). + replay, err := runner.RunBatch(batch) + if err != nil { + t.Fatalf("replay: %v", err) + } + if len(replay) != 2 { + t.Fatalf("replay result count: %d", len(replay)) + } + // The mutating tools must NOT have run again. + if store.updates != 1 || store.comments != 1 { + t.Fatalf("replay re-executed mutations: updates=%d comments=%d", store.updates, store.comments) + } + if runner.ExecutedCount() != 2 { + t.Fatalf("expected 2 distinct executions, got %d", runner.ExecutedCount()) + } +} + +func TestBatchFailsClosed(t *testing.T) { + runner := NewBatchRunner(&fakeStore{}, TaskScope{Kind: TaskChat}) + runner.Buffer(ToolRequest{EventID: "ev-1", Tool: "multica_list_issues", RawInput: []byte(`{}`)}) + + if _, err := runner.RunBatch(nil); err == nil { + t.Fatal("empty batch must fail closed") + } + if _, err := runner.RunBatch([]string{"ev-unknown"}); err == nil { + t.Fatal("unknown required action must fail closed") + } + // ev-1 was buffered but never executed: a terminal idle must fail closed. + if err := runner.FinalizeIdle(); err == nil { + t.Fatal("unresolved custom tool at idle must fail closed") + } +} + +func TestBatchPartialFailureReturnsErrorResultWithoutAborting(t *testing.T) { + store := &fakeStore{} + runner := NewBatchRunner(store, TaskScope{Kind: TaskIssue, AssignedIssueID: assignedID}) + // First tool is valid; second targets another issue and must fail in scope. + runner.Buffer(ToolRequest{EventID: "ok", Tool: "multica_update_issue", RawInput: []byte(`{"issue_id":"` + assignedID + `","priority":"high"}`)}) + runner.Buffer(ToolRequest{EventID: "bad", Tool: "multica_get_issue", RawInput: []byte(`{"issue_id":"33333333-3333-4333-8333-333333333333"}`)}) + + results, err := runner.RunBatch([]string{"ok", "bad"}) + if err != nil { + t.Fatalf("batch should return per-tool results, not abort: %v", err) + } + if results[0].IsError { + t.Fatalf("first tool should succeed: %+v", results[0]) + } + if !results[1].IsError { + t.Fatal("out-of-scope tool should yield an error result") + } + if store.updates != 1 { + t.Fatalf("valid tool should have run once, got %d", store.updates) + } +} + +// TestStreamResumesWithLastEventIDAndDeduplicates drops the connection after +// the first event, then serves a reconnect that replays the last event before +// continuing. The reader must resume via Last-Event-ID, dedup the replay, and +// deliver every distinct event exactly once, in order. +func TestStreamResumesWithLastEventIDAndDeduplicates(t *testing.T) { + var hits int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits++ + flusher, _ := w.(http.Flusher) + w.Header().Set("Content-Type", "text/event-stream") + if hits == 1 { + // Deliver ev-1, then cut the stream short (no blank-line close on purpose). + fmt.Fprint(w, "id: ev-1\nevent: agent.message\ndata: first\n\n") + if flusher != nil { + flusher.Flush() + } + return // connection ends abruptly -> reader reconnects + } + // Reconnect must carry the last event ID we saw. + if got := r.Header.Get("Last-Event-ID"); got != "ev-1" { + t.Errorf("reconnect Last-Event-ID = %q, want ev-1", got) + } + // Server replays ev-1 (duplicate) then sends ev-2 and a terminal idle. + fmt.Fprint(w, "id: ev-1\nevent: agent.message\ndata: first\n\n") + fmt.Fprint(w, "id: ev-2\nevent: agent.message\ndata: second\n\n") + fmt.Fprint(w, "id: ev-3\nevent: session.status_idle\ndata: idle\n\n") + })) + defer server.Close() + + reader := NewStreamReader(server.Client(), server.URL, "placeholder-token") + var delivered []string + err := reader.Read(context.Background(), func(e Event) (bool, error) { + delivered = append(delivered, e.ID) + return e.Type == "session.status_idle", nil + }) + if err != nil { + t.Fatalf("stream read: %v", err) + } + want := []string{"ev-1", "ev-2", "ev-3"} + if len(delivered) != len(want) { + t.Fatalf("delivered %v, want %v", delivered, want) + } + for i := range want { + if delivered[i] != want[i] { + t.Fatalf("event %d = %q, want %q (delivered=%v)", i, delivered[i], want[i], delivered) + } + } +} diff --git a/demos/integrate-qoder-cloud-runtime/dispatcher.go b/demos/integrate-qoder-cloud-runtime/dispatcher.go new file mode 100644 index 0000000..50a52dc --- /dev/null +++ b/demos/integrate-qoder-cloud-runtime/dispatcher.go @@ -0,0 +1,195 @@ +// SPDX-License-Identifier: Apache-2.0 + +// Package bridge shows how a self-hosted orchestrator can expose its business +// operations to a Qoder Cloud Agent as client-side custom tools, while keeping +// the allowlist, validation, and task scope enforced in its own process. +// +// This dispatcher is a self-contained, standalone illustration of the pattern +// described in the accompanying Cookbook article. It has no third-party +// dependencies and does not reference any private system. +package bridge + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "regexp" +) + +// TaskKind is the surface a run is bound to. An issue task may touch only its +// assigned issue; a chat task may list the workspace. +type TaskKind int + +const ( + TaskChat TaskKind = iota + TaskIssue +) + +// TaskScope binds a run to a single user, workspace, and (for issue tasks) one +// assigned issue. The dispatcher enforces it on every call. +type TaskScope struct { + Kind TaskKind + AssignedIssueID string // required when Kind == TaskIssue +} + +// IssueStore is the narrow business API the dispatcher is allowed to reach. +// A real implementation would call an authenticated backend; the demo uses an +// in-memory fake in tests. +type IssueStore interface { + ListIssues(status string, limit int) (string, error) + GetIssue(id string) (string, error) + UpdateIssue(id string, fields map[string]string) (string, error) + AddComment(issueID, content, parentID string) (string, error) +} + +var ( + uuidPattern = regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$`) + allowedStatus = map[string]bool{"backlog": true, "todo": true, "in_progress": true, "in_review": true, "done": true, "blocked": true, "cancelled": true} + allowedPriority = map[string]bool{"urgent": true, "high": true, "medium": true, "low": true, "none": true} +) + +const ( + maxTitleLen = 500 + maxContentLen = 20000 + maxListLimit = 50 +) + +// ErrUnknownTool is returned for any name outside the fixed allowlist. +var ErrUnknownTool = errors.New("unknown tool") + +// Dispatch validates one custom-tool request against the allowlist and task +// scope, then performs the single allowed operation. rawInput is the tool +// input exactly as the cloud Agent sent it. It fails closed: unknown tools, +// unknown JSON fields, invalid values, and out-of-scope targets all error +// before any store call. +func Dispatch(store IssueStore, scope TaskScope, tool string, rawInput []byte) (string, error) { + switch tool { + case "multica_list_issues": + var in struct { + Status string `json:"status"` + Limit int `json:"limit"` + } + if err := strictUnmarshal(rawInput, &in); err != nil { + return "", err + } + if in.Status != "" && !allowedStatus[in.Status] { + return "", fmt.Errorf("invalid status: %q", in.Status) + } + if in.Limit != 0 && (in.Limit < 1 || in.Limit > maxListLimit) { + return "", fmt.Errorf("limit must be 1..%d", maxListLimit) + } + // An issue task is scoped to exactly its assigned issue. + if scope.Kind == TaskIssue { + return store.GetIssue(scope.AssignedIssueID) + } + return store.ListIssues(in.Status, in.Limit) + + case "multica_get_issue": + var in struct { + IssueID string `json:"issue_id"` + } + if err := strictUnmarshal(rawInput, &in); err != nil { + return "", err + } + if !uuidPattern.MatchString(in.IssueID) { + return "", errors.New("issue_id must be a UUID") + } + if err := requireScope(scope, in.IssueID); err != nil { + return "", err + } + return store.GetIssue(in.IssueID) + + case "multica_update_issue": + var in struct { + IssueID string `json:"issue_id"` + Title string `json:"title"` + Description string `json:"description"` + Status string `json:"status"` + Priority string `json:"priority"` + } + if err := strictUnmarshal(rawInput, &in); err != nil { + return "", err + } + if !uuidPattern.MatchString(in.IssueID) { + return "", errors.New("issue_id must be a UUID") + } + if err := requireScope(scope, in.IssueID); err != nil { + return "", err + } + fields := map[string]string{} + if in.Title != "" { + if len(in.Title) > maxTitleLen { + return "", fmt.Errorf("title must be 1..%d chars", maxTitleLen) + } + fields["title"] = in.Title + } + if in.Description != "" { + if len(in.Description) > maxContentLen { + return "", fmt.Errorf("description must be <=%d chars", maxContentLen) + } + fields["description"] = in.Description + } + if in.Status != "" { + if !allowedStatus[in.Status] { + return "", fmt.Errorf("invalid status: %q", in.Status) + } + fields["status"] = in.Status + } + if in.Priority != "" { + if !allowedPriority[in.Priority] { + return "", fmt.Errorf("invalid priority: %q", in.Priority) + } + fields["priority"] = in.Priority + } + if len(fields) == 0 { + return "", errors.New("update requires at least one field") + } + return store.UpdateIssue(in.IssueID, fields) + + case "multica_add_issue_comment": + var in struct { + IssueID string `json:"issue_id"` + Content string `json:"content"` + ParentID string `json:"parent_id"` + } + if err := strictUnmarshal(rawInput, &in); err != nil { + return "", err + } + if !uuidPattern.MatchString(in.IssueID) { + return "", errors.New("issue_id must be a UUID") + } + if err := requireScope(scope, in.IssueID); err != nil { + return "", err + } + if in.Content == "" || len(in.Content) > maxContentLen { + return "", fmt.Errorf("content must be 1..%d chars", maxContentLen) + } + if in.ParentID != "" && !uuidPattern.MatchString(in.ParentID) { + return "", errors.New("parent_id must be a UUID") + } + return store.AddComment(in.IssueID, in.Content, in.ParentID) + + default: + return "", fmt.Errorf("%w: %q", ErrUnknownTool, tool) + } +} + +// requireScope rejects any issue target other than the task's assigned issue. +func requireScope(scope TaskScope, issueID string) error { + if scope.Kind == TaskIssue && issueID != scope.AssignedIssueID { + return fmt.Errorf("issue task may not target another issue") + } + return nil +} + +// strictUnmarshal rejects unknown JSON fields so a cloud-supplied payload +// cannot smuggle extra keys past validation. +func strictUnmarshal(raw []byte, out any) error { + decoder := json.NewDecoder(bytes.NewReader(raw)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(out); err != nil { + return fmt.Errorf("invalid tool input: %w", err) + } + return nil +} diff --git a/demos/integrate-qoder-cloud-runtime/dispatcher_test.go b/demos/integrate-qoder-cloud-runtime/dispatcher_test.go new file mode 100644 index 0000000..3591ee7 --- /dev/null +++ b/demos/integrate-qoder-cloud-runtime/dispatcher_test.go @@ -0,0 +1,79 @@ +// SPDX-License-Identifier: Apache-2.0 + +package bridge + +import ( + "errors" + "testing" +) + +// fakeStore records calls so tests can assert exactly-once execution. +type fakeStore struct { + updates int + comments int +} + +func (f *fakeStore) ListIssues(status string, limit int) (string, error) { + return `[{"id":"issue-1"}]`, nil +} +func (f *fakeStore) GetIssue(id string) (string, error) { + return `{"id":"` + id + `"}`, nil +} +func (f *fakeStore) UpdateIssue(id string, fields map[string]string) (string, error) { + f.updates++ + return `{"id":"` + id + `","updated":true}`, nil +} +func (f *fakeStore) AddComment(issueID, content, parentID string) (string, error) { + f.comments++ + return `{"id":"comment-1"}`, nil +} + +const assignedID = "11111111-1111-4111-8111-111111111111" + +func TestDispatchRejectsInvalidInput(t *testing.T) { + store := &fakeStore{} + scope := TaskScope{Kind: TaskChat} + cases := []struct { + name string + tool string + input string + }{ + {"unknown tool", "multica_delete_everything", `{}`}, + {"unknown field", "multica_get_issue", `{"issue_id":"` + assignedID + `","extra":1}`}, + {"non-uuid", "multica_get_issue", `{"issue_id":"not-a-uuid"}`}, + {"invalid enum", "multica_update_issue", `{"issue_id":"` + assignedID + `","status":"shipped"}`}, + {"empty update", "multica_update_issue", `{"issue_id":"` + assignedID + `"}`}, + {"fractional limit", "multica_list_issues", `{"limit":1.5}`}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if _, err := Dispatch(store, scope, tc.tool, []byte(tc.input)); err == nil { + t.Fatalf("expected error for %s", tc.name) + } + }) + } + if store.updates != 0 || store.comments != 0 { + t.Fatalf("invalid inputs must not reach the store: updates=%d comments=%d", store.updates, store.comments) + } +} + +func TestDispatchEnforcesIssueScope(t *testing.T) { + store := &fakeStore{} + scope := TaskScope{Kind: TaskIssue, AssignedIssueID: assignedID} + other := "22222222-2222-4222-8222-222222222222" + if _, err := Dispatch(store, scope, "multica_get_issue", []byte(`{"issue_id":"`+other+`"}`)); err == nil { + t.Fatal("issue task must not read another issue") + } + if out, err := Dispatch(store, scope, "multica_get_issue", []byte(`{"issue_id":"`+assignedID+`"}`)); err != nil { + t.Fatalf("assigned issue read failed: %v", err) + } else if out == "" { + t.Fatal("expected issue payload") + } +} + +func TestDispatchUnknownToolIsTyped(t *testing.T) { + _, err := Dispatch(&fakeStore{}, TaskScope{Kind: TaskChat}, "nope", []byte(`{}`)) + if !errors.Is(err, ErrUnknownTool) { + t.Fatalf("expected ErrUnknownTool, got %v", err) + } +} diff --git a/demos/integrate-qoder-cloud-runtime/go.mod b/demos/integrate-qoder-cloud-runtime/go.mod new file mode 100644 index 0000000..8665b78 --- /dev/null +++ b/demos/integrate-qoder-cloud-runtime/go.mod @@ -0,0 +1,3 @@ +module example.com/qoder-cloud-runtime-demo + +go 1.22 diff --git a/demos/integrate-qoder-cloud-runtime/stream.go b/demos/integrate-qoder-cloud-runtime/stream.go new file mode 100644 index 0000000..9a2e466 --- /dev/null +++ b/demos/integrate-qoder-cloud-runtime/stream.go @@ -0,0 +1,147 @@ +// SPDX-License-Identifier: Apache-2.0 + +package bridge + +import ( + "bufio" + "context" + "fmt" + "io" + "net/http" + "strings" +) + +// Event is a decoded SSE frame from the session event stream. +type Event struct { + ID string + Type string + Data string +} + +// StreamReader consumes a session's SSE event stream. On a dropped connection +// it reconnects with Last-Event-ID set to the newest event it already saw, and +// it deduplicates any events the server replays, so no event is processed twice +// and none is lost across a reconnect. +type StreamReader struct { + client *http.Client + url string + pat string + lastEventID string + seen map[string]bool + maxReconnect int +} + +// NewStreamReader builds a reader for one session stream URL. +func NewStreamReader(client *http.Client, url, pat string) *StreamReader { + return &StreamReader{ + client: client, + url: url, + pat: pat, + seen: map[string]bool{}, + maxReconnect: 5, + } +} + +// Read streams events to handle until the stream ends or the context is done. +// A transient connection drop triggers a bounded reconnect using Last-Event-ID. +// handle returning true stops the stream (for example on a terminal idle). +func (s *StreamReader) Read(ctx context.Context, handle func(Event) (stop bool, err error)) error { + attempts := 0 + for { + done, err := s.readOnce(ctx, handle) + if done { + return err + } + if err == nil { + return nil // clean stream end + } + attempts++ + if attempts > s.maxReconnect { + return fmt.Errorf("stream failed after %d reconnects: %w", s.maxReconnect, err) + } + // Reconnect; Last-Event-ID makes the server resume after the last + // event we saw, and seen{} discards any it replays anyway. + } +} + +func (s *StreamReader) readOnce(ctx context.Context, handle func(Event) (bool, error)) (bool, error) { + request, err := http.NewRequestWithContext(ctx, http.MethodGet, s.url, nil) + if err != nil { + return true, err + } + request.Header.Set("Authorization", "Bearer "+s.pat) + request.Header.Set("Accept", "text/event-stream") + if s.lastEventID != "" { + request.Header.Set("Last-Event-ID", s.lastEventID) + } + response, err := s.client.Do(request) + if err != nil { + return false, err // transient: allow reconnect + } + defer response.Body.Close() + if response.StatusCode >= 500 || response.StatusCode == http.StatusTooManyRequests { + return false, fmt.Errorf("stream HTTP %d", response.StatusCode) + } + if response.StatusCode != http.StatusOK { + return true, fmt.Errorf("stream HTTP %d", response.StatusCode) + } + + for _, event := range parseSSE(response.Body) { + if event.ID != "" { + s.lastEventID = event.ID + if s.seen[event.ID] { + continue // dedup a replayed event + } + s.seen[event.ID] = true + } + stop, handleErr := handle(event) + if handleErr != nil { + return true, handleErr + } + if stop { + return true, nil + } + } + return false, io.ErrUnexpectedEOF // stream cut short: allow reconnect +} + +// parseSSE decodes a text/event-stream body into complete events. It is a +// minimal parser sufficient for id/event/data fields separated by blank lines. +func parseSSE(body io.Reader) []Event { + var events []Event + scanner := bufio.NewScanner(body) + scanner.Buffer(make([]byte, 0, 64*1024), 16*1024*1024) + var current Event + var data strings.Builder + flush := func() { + if current.ID == "" && current.Type == "" && data.Len() == 0 { + return + } + current.Data = data.String() + events = append(events, current) + current = Event{} + data.Reset() + } + for scanner.Scan() { + line := scanner.Text() + if line == "" { + flush() + continue + } + field, value, _ := strings.Cut(line, ":") + value = strings.TrimPrefix(value, " ") + switch field { + case "id": + current.ID = value + case "event": + current.Type = value + case "data": + if data.Len() > 0 { + data.WriteByte('\n') + } + data.WriteString(value) + } + } + flush() + return events +} From 56fa91a8b03026489fa169a8e31bb37a46560617 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=97=B6=E4=B9=8B?= Date: Fri, 14 Aug 2026 15:36:01 +0800 Subject: [PATCH 2/2] fix(demo): harden custom-tool bridge per review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Buffer: ignore duplicate event IDs so the exactly-once key can't be hijacked - strictUnmarshal: reject trailing JSON tokens (require exactly one value) - stream: parse SSE incrementally so handle/stop apply on long-lived streams Signed-off-by: 时之 --- demos/integrate-qoder-cloud-runtime/bridge.go | 6 ++ .../bridge_test.go | 22 ++++++ .../dispatcher.go | 10 +++ .../dispatcher_test.go | 1 + demos/integrate-qoder-cloud-runtime/stream.go | 70 +++++++++++-------- 5 files changed, 78 insertions(+), 31 deletions(-) diff --git a/demos/integrate-qoder-cloud-runtime/bridge.go b/demos/integrate-qoder-cloud-runtime/bridge.go index 332c317..b91ca07 100644 --- a/demos/integrate-qoder-cloud-runtime/bridge.go +++ b/demos/integrate-qoder-cloud-runtime/bridge.go @@ -56,6 +56,12 @@ func NewBatchRunner(store IssueStore, scope TaskScope) *BatchRunner { func (r *BatchRunner) Buffer(request ToolRequest) { r.mu.Lock() defer r.mu.Unlock() + // The event ID is the exactly-once idempotency key, so a duplicate or + // replayed event must not replace the original tool/input. First write + // wins; later buffers of the same ID are ignored. + if _, exists := r.buffered[request.EventID]; exists { + return + } r.buffered[request.EventID] = request } diff --git a/demos/integrate-qoder-cloud-runtime/bridge_test.go b/demos/integrate-qoder-cloud-runtime/bridge_test.go index 47c2a26..c0fb044 100644 --- a/demos/integrate-qoder-cloud-runtime/bridge_test.go +++ b/demos/integrate-qoder-cloud-runtime/bridge_test.go @@ -63,6 +63,28 @@ func TestBatchFailsClosed(t *testing.T) { } } +// A replayed event ID must not replace the original buffered tool/input, or the +// exactly-once idempotency key could be hijacked. First write wins. +func TestBufferIgnoresDuplicateEventID(t *testing.T) { + store := &fakeStore{} + runner := NewBatchRunner(store, TaskScope{Kind: TaskIssue, AssignedIssueID: assignedID}) + // Original: a read. Replay under the same ID tries to smuggle a mutation. + runner.Buffer(ToolRequest{EventID: "ev-1", Tool: "multica_get_issue", RawInput: []byte(`{"issue_id":"` + assignedID + `"}`)}) + runner.Buffer(ToolRequest{EventID: "ev-1", Tool: "multica_update_issue", RawInput: []byte(`{"issue_id":"` + assignedID + `","status":"done"}`)}) + + results, err := runner.RunBatch([]string{"ev-1"}) + if err != nil { + t.Fatalf("run: %v", err) + } + if len(results) != 1 || results[0].IsError { + t.Fatalf("unexpected results: %+v", results) + } + // The smuggled update must never have run. + if store.updates != 0 { + t.Fatalf("duplicate event ID replaced the original tool: updates=%d", store.updates) + } +} + func TestBatchPartialFailureReturnsErrorResultWithoutAborting(t *testing.T) { store := &fakeStore{} runner := NewBatchRunner(store, TaskScope{Kind: TaskIssue, AssignedIssueID: assignedID}) diff --git a/demos/integrate-qoder-cloud-runtime/dispatcher.go b/demos/integrate-qoder-cloud-runtime/dispatcher.go index 50a52dc..2775f12 100644 --- a/demos/integrate-qoder-cloud-runtime/dispatcher.go +++ b/demos/integrate-qoder-cloud-runtime/dispatcher.go @@ -14,6 +14,7 @@ import ( "encoding/json" "errors" "fmt" + "io" "regexp" ) @@ -191,5 +192,14 @@ func strictUnmarshal(raw []byte, out any) error { if err := decoder.Decode(out); err != nil { return fmt.Errorf("invalid tool input: %w", err) } + // Fail closed: require exactly one JSON value with no trailing tokens, so a + // cloud-supplied body like `{}{}` cannot smuggle a second value past + // validation. A clean end of input decodes to io.EOF. + if err := decoder.Decode(new(json.RawMessage)); !errors.Is(err, io.EOF) { + if err == nil { + return errors.New("invalid tool input: unexpected trailing data") + } + return fmt.Errorf("invalid tool input: %w", err) + } return nil } diff --git a/demos/integrate-qoder-cloud-runtime/dispatcher_test.go b/demos/integrate-qoder-cloud-runtime/dispatcher_test.go index 3591ee7..e77efab 100644 --- a/demos/integrate-qoder-cloud-runtime/dispatcher_test.go +++ b/demos/integrate-qoder-cloud-runtime/dispatcher_test.go @@ -44,6 +44,7 @@ func TestDispatchRejectsInvalidInput(t *testing.T) { {"invalid enum", "multica_update_issue", `{"issue_id":"` + assignedID + `","status":"shipped"}`}, {"empty update", "multica_update_issue", `{"issue_id":"` + assignedID + `"}`}, {"fractional limit", "multica_list_issues", `{"limit":1.5}`}, + {"trailing data", "multica_get_issue", `{"issue_id":"` + assignedID + `"}{}`}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { diff --git a/demos/integrate-qoder-cloud-runtime/stream.go b/demos/integrate-qoder-cloud-runtime/stream.go index 9a2e466..ec84211 100644 --- a/demos/integrate-qoder-cloud-runtime/stream.go +++ b/demos/integrate-qoder-cloud-runtime/stream.go @@ -86,46 +86,44 @@ func (s *StreamReader) readOnce(ctx context.Context, handle func(Event) (bool, e return true, fmt.Errorf("stream HTTP %d", response.StatusCode) } - for _, event := range parseSSE(response.Body) { - if event.ID != "" { - s.lastEventID = event.ID - if s.seen[event.ID] { - continue // dedup a replayed event - } - s.seen[event.ID] = true - } - stop, handleErr := handle(event) - if handleErr != nil { - return true, handleErr - } - if stop { - return true, nil - } - } - return false, io.ErrUnexpectedEOF // stream cut short: allow reconnect -} - -// parseSSE decodes a text/event-stream body into complete events. It is a -// minimal parser sufficient for id/event/data fields separated by blank lines. -func parseSSE(body io.Reader) []Event { - var events []Event - scanner := bufio.NewScanner(body) + // Parse the event stream incrementally: dispatch each event to handle as + // soon as its terminating blank line arrives, so a long-lived stream is + // processed in real time and handle's stop signal takes effect promptly. + scanner := bufio.NewScanner(response.Body) scanner.Buffer(make([]byte, 0, 64*1024), 16*1024*1024) var current Event var data strings.Builder - flush := func() { + + // dispatch delivers one completed event, applying Last-Event-ID tracking + // and replay deduplication. It reports whether the caller asked to stop. + dispatch := func() (stop bool, err error) { if current.ID == "" && current.Type == "" && data.Len() == 0 { - return + return false, nil } - current.Data = data.String() - events = append(events, current) + event := current + event.Data = data.String() current = Event{} data.Reset() + if event.ID != "" { + s.lastEventID = event.ID + if s.seen[event.ID] { + return false, nil // dedup a replayed event + } + s.seen[event.ID] = true + } + return handle(event) } + for scanner.Scan() { line := scanner.Text() if line == "" { - flush() + stop, handleErr := dispatch() + if handleErr != nil { + return true, handleErr + } + if stop { + return true, nil + } continue } field, value, _ := strings.Cut(line, ":") @@ -142,6 +140,16 @@ func parseSSE(body io.Reader) []Event { data.WriteString(value) } } - flush() - return events + // Flush a trailing event that ended without a final blank line. + stop, handleErr := dispatch() + if handleErr != nil { + return true, handleErr + } + if stop { + return true, nil + } + if scanErr := scanner.Err(); scanErr != nil { + return false, scanErr // transient read failure: allow reconnect + } + return false, io.ErrUnexpectedEOF // stream cut short: allow reconnect }