From cb146d27a1aa2c457aa0c0bd31aece3f889b79a4 Mon Sep 17 00:00:00 2001 From: ddcat666 <26564566+ddcat666@users.noreply.github.com> Date: Tue, 29 Sep 2026 21:41:36 +0800 Subject: [PATCH] =?UTF-8?q?feat(deploy):=20=E7=94=BB=E5=B8=83=20Agent=20-?= =?UTF-8?q?=20=E7=8B=AC=E7=AB=8B=20yingce-agent=20=E5=AE=B9=E5=99=A8?= =?UTF-8?q?=E5=B9=B6=E4=BF=AE=E5=A4=8D=E8=BF=90=E8=A1=8C=E4=B8=AD=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 生产会话不再因 stdout 管道提前关闭被误判失败。设置 YINGCE_AGENT_URL 后只走独立容器，失败不回退到内嵌进程。 --- .env.example | 3 + .github/workflows/publish-images.yml | 7 +- AGENTS.md | 26 +++ backend/agent-runtime/pi/agent-runtime.mjs | 38 +++- backend/agent-runtime/pi/package-lock.json | 12 +- backend/agent-runtime/pi/package.json | 3 +- backend/internal/agent/runtime/runtime.go | 170 +++++++++++++----- .../internal/agent/runtime/runtime_test.go | 83 +++++++++ backend/internal/agent/yingceagent/client.go | 69 +++++++ .../internal/agent/yingceagent/client_test.go | 119 ++++++++++++ .../internal/app/cloud_agent_media_test.go | 38 ++-- .../app/cloud_agent_pi_coordinator.go | 46 +++-- .../internal/app/cloud_agent_pi_runtime.go | 11 +- backend/internal/app/cloud_agent_runtime.go | 3 +- .../cloud_agent_single_active_task_test.go | 14 ++ backend/internal/app/cloud_agent_tools.go | 4 +- backend/internal/hostupdate/manager_ops.go | 34 +++- .../internal/hostupdate/manager_ops_test.go | 4 +- .../internal/prompts/agent-media-policy.md | 8 +- backend/internal/prompts/agent_policy_test.go | 6 +- docker-compose.deploy.yml | 19 ++ scripts/install-server-image.sh | 12 +- yingce-agent/Dockerfile | 19 ++ yingce-agent/README.md | 7 + yingce-agent/server.mjs | 167 +++++++++++++++++ 25 files changed, 821 insertions(+), 101 deletions(-) create mode 100644 backend/internal/agent/runtime/runtime_test.go create mode 100644 backend/internal/agent/yingceagent/client.go create mode 100644 backend/internal/agent/yingceagent/client_test.go create mode 100644 yingce-agent/Dockerfile create mode 100644 yingce-agent/README.md create mode 100644 yingce-agent/server.mjs diff --git a/.env.example b/.env.example index 7c774204a..d050c64b1 100644 --- a/.env.example +++ b/.env.example @@ -24,6 +24,9 @@ CANVAS_SHUTDOWN_TIMEOUT=10m CANVAS_AUTO_MIGRATE=true # 生产在线更新由宿主机 Host Updater 提供；安装脚本生成随机 Token，不要手工提交真实值。 CANVAS_UPDATER_TOKEN= +# 生产 Agent 运行在独立的 yingce-agent 容器。安装脚本生成随机 Token，不要手工提交真实值。 +# YINGCE_AGENT_TOKEN= +# CANVAS_YINGCE_AGENT_IMAGE=ghcr.io//open-ai-canvas-yingce-agent:@sha256:... CANVAS_UPDATER_SOCKET_DIR=/run/open-ai-canvas-updater # 新服务器请直接运行 README 的一键部署命令；脚本会在 .env 中生成以下 PostgreSQL 配置。 diff --git a/.github/workflows/publish-images.yml b/.github/workflows/publish-images.yml index 24aba594c..fae278173 100644 --- a/.github/workflows/publish-images.yml +++ b/.github/workflows/publish-images.yml @@ -75,6 +75,11 @@ jobs: arch: arm64 platforms: linux/arm64 runner: ubuntu-24.04-arm + - image: open-ai-canvas-yingce-agent + dockerfile: yingce-agent/Dockerfile + arch: multi + platforms: linux/amd64,linux/arm64 + runner: ubuntu-24.04 env: IMAGE: ghcr.io/${{ needs.prepare.outputs.owner }}/${{ matrix.image }} steps: @@ -138,7 +143,7 @@ jobs: strategy: fail-fast: false matrix: - image: [open-ai-canvas-web, open-ai-canvas-backend] + image: [open-ai-canvas-web, open-ai-canvas-backend, open-ai-canvas-yingce-agent] # 不同提交可并行编译，但同一镜像的可变标签更新必须串行。 concurrency: group: image-tags-${{ matrix.image }}-${{ github.ref }} diff --git a/AGENTS.md b/AGENTS.md index 16b9f09ce..88576265f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,6 +18,32 @@ ## 2. 开始工作前 +### Agent 本地开发 + +详细说明请参考 [docs/AGENT_LOCAL_DEVELOPMENT.md](docs/AGENT_LOCAL_DEVELOPMENT.md)。 + +**快速启动**： + +```bash +# 1. 安装 Agent runtime 依赖 +cd backend/agent-runtime/pi && npm install && cd ../../.. + +# 2. 启动后端（自动使用嵌入模式） +cd backend +CANVAS_BACKEND_DATA_DIR=../.local/project-workbench-debug go run ./cmd/server + +# 3. 启动前端 +cd web && bun run dev +``` + +本地开发不设置 `YINGCE_AGENT_URL` 时，后端在本进程内启动 Node，无需单独启动 yingce-agent。 + +生产 Compose 设置 `YINGCE_AGENT_URL` 后只调用 yingce-agent 容器。远程服务失败时不要回退到内嵌进程，否则同一步可能执行两次。不要删除 `yingce-agent/` 或把它从发布流程里撤掉。 + +--- + +### 其他本地开发说明 + 1. 先读取任务涉及的入口、调用方、配置、锁文件和相邻测试；先理解现状，再决定是否抽象或重构。 2. 使用 `rg` / `rg --files` 搜索，优先并行读取相关文件。不要为了“统一风格”改动无关模块、依赖、格式或用户已有修改。 3. 先形成目标边界：页面负责什么、service 负责什么、handler/service/repository 如何分层、数据和错误如何流动。新增 helper 必须消除真实重复或隔离明确协议，不能只透传参数。 diff --git a/backend/agent-runtime/pi/agent-runtime.mjs b/backend/agent-runtime/pi/agent-runtime.mjs index 45c1e85f5..fa13ee101 100644 --- a/backend/agent-runtime/pi/agent-runtime.mjs +++ b/backend/agent-runtime/pi/agent-runtime.mjs @@ -1,9 +1,20 @@ -import { mkdir, readFile, writeFile } from "node:fs/promises"; -import { dirname } from "node:path"; -import { createAgentSession, DefaultResourceLoader, ModelRuntime, SessionManager } from "@earendil-works/pi-coding-agent"; +import { rmSync } from "node:fs"; +import { mkdir, mkdtemp, readFile, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { dirname, join } from "node:path"; +import { Agent, setGlobalDispatcher } from "undici"; +import { createAgentSession, DefaultResourceLoader, ModelRuntime, SessionManager, SettingsManager } from "@earendil-works/pi-coding-agent"; import { createAssistantMessageEventStream } from "@earendil-works/pi-ai"; import { Type } from "typebox"; +// Node 自带 undici 的 header/body 超时默认 300 秒，短于文本任务 8 分钟和视频任务 60 分钟。 +// 设为 0 后只由 Go 侧任务期限取消 bridge；这里不再提前切断仍在排队或生成中的步骤。 +setGlobalDispatcher(new Agent({ + connectTimeout: 30_000, + headersTimeout: 0, + bodyTimeout: 0, +})); + const providerID = "infinite-canvas"; const api = "openai-completions"; @@ -69,7 +80,19 @@ async function run() { throw new Error("Agent runtime is missing its bridge, session, or model configuration"); } - const modelRuntime = await ModelRuntime.create({ agentDir: request.agentDir }); + // 每次运行使用独立的空配置目录：SDK 的 auth.json / models.json / 锁文件都落在这里， + // 不读取宿主 ~/.pi，也不会执行预置 auth.json 里的 !command。运行结束即删除。 + const isolatedDir = await mkdtemp(join(tmpdir(), "agent-runtime-")); + process.env.HOME = isolatedDir; + process.env.PI_CODING_AGENT_DIR = isolatedDir; + process.on("exit", () => { + try { rmSync(isolatedDir, { recursive: true, force: true }); } catch {} + }); + request.agentDir = isolatedDir; + const modelRuntime = await ModelRuntime.create({ + authPath: join(isolatedDir, "auth.json"), + modelsPath: join(isolatedDir, "models.json"), + }); modelRuntime.registerProvider(providerID, { name: "影策模型任务", baseUrl: "http://agent-runtime.invalid/v1", @@ -192,15 +215,18 @@ async function run() { } else { sessionManager = SessionManager.create(request.cwd, request.sessionDir); } + // 严格隔离：不探索文件系统的 packages/skills/extensions const resourceLoader = new DefaultResourceLoader({ cwd: request.cwd, agentDir: request.agentDir, noExtensions: true, + noSkills: true, noPromptTemplates: true, noThemes: true, noContextFiles: true, additionalSkillPaths: request.skillPaths ?? [], systemPrompt: request.systemPrompt, + settingsManager: SettingsManager.inMemory({}), }); await resourceLoader.reload(); @@ -249,6 +275,10 @@ async function run() { sessionConfig.permissions = request.permissions; } + // 只启用平台声明的工具，不使用 SDK 默认工具 + const platformToolNames = tools.map(t => t.name); + sessionConfig.tools = platformToolNames; + ({ session } = await createAgentSession(sessionConfig)); const compactionConfig = request.compaction || { diff --git a/backend/agent-runtime/pi/package-lock.json b/backend/agent-runtime/pi/package-lock.json index 1364b967a..741bec47c 100644 --- a/backend/agent-runtime/pi/package-lock.json +++ b/backend/agent-runtime/pi/package-lock.json @@ -8,7 +8,8 @@ "dependencies": { "@earendil-works/pi-ai": "0.87.1", "@earendil-works/pi-coding-agent": "0.87.1", - "typebox": "1.3.27" + "typebox": "1.3.27", + "undici": "7.28.0" }, "engines": { "node": ">=22.19.0" @@ -2965,6 +2966,15 @@ "integrity": "sha512-zu+jc1pcy4UiNThxikUr36f0Rybk9PEeCg/NE6adeWr/SKsdNO4EzZHYRDlv2YCVAfj3Odq3dESSo/jNyoBXzA==", "license": "MIT" }, + "node_modules/undici": { + "version": "7.28.0", + "resolved": "https://registry.npmjs.org/undici/-/undici-7.28.0.tgz", + "integrity": "sha512-cRZYrTDwWznlnRiPjggAGxZXanty6M8RV1ff8Wm4LWXBp7/IG8v5DnOm74DtUBp9OONpK75YlPnIjQqX0dBDtA==", + "license": "MIT", + "engines": { + "node": ">=20.18.1" + } + }, "node_modules/undici-types": { "version": "8.9.0", "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-8.9.0.tgz", diff --git a/backend/agent-runtime/pi/package.json b/backend/agent-runtime/pi/package.json index f8aa31d1e..9f3e4db0e 100644 --- a/backend/agent-runtime/pi/package.json +++ b/backend/agent-runtime/pi/package.json @@ -8,6 +8,7 @@ "dependencies": { "@earendil-works/pi-ai": "0.87.1", "@earendil-works/pi-coding-agent": "0.87.1", - "typebox": "1.3.27" + "typebox": "1.3.27", + "undici": "7.28.0" } } diff --git a/backend/internal/agent/runtime/runtime.go b/backend/internal/agent/runtime/runtime.go index 4ad757ced..b38f84795 100644 --- a/backend/internal/agent/runtime/runtime.go +++ b/backend/internal/agent/runtime/runtime.go @@ -15,7 +15,9 @@ import ( "os/exec" "path/filepath" "runtime" + "strconv" "strings" + "sync" "time" ) @@ -54,7 +56,21 @@ type ProcessRequest struct { Model map[string]any `json:"model"` } +// BridgeServer is the per-run callback endpoint. The Node runtime calls it for +// model, tool, and event steps. Authorization is the bearer token; binding can +// be loopback for the embedded process or the container network for yingce-agent. +type BridgeServer struct { + URL string + Token string + Close func() +} + func Run(ctx context.Context, request ProcessRequest, bridge Bridge) error { + release, err := acquireAgentProcess(ctx) + if err != nil { + return err + } + defer release() if bridge.Model == nil || bridge.Tool == nil || bridge.Event == nil { return errors.New("Agent bridge handlers are incomplete") } @@ -62,34 +78,27 @@ func Run(ctx context.Context, request ProcessRequest, bridge Bridge) error { if err != nil { return err } - tokenBytes := make([]byte, 32) - if _, err := rand.Read(tokenBytes); err != nil { - return fmt.Errorf("create Agent bridge credential: %w", err) - } - token := hex.EncodeToString(tokenBytes) - listener, err := net.Listen("tcp", "127.0.0.1:0") + server, err := StartBridge("127.0.0.1:0", "127.0.0.1", bridge) if err != nil { - return fmt.Errorf("listen for Agent bridge: %w", err) + return err } - defer listener.Close() - mux := http.NewServeMux() - mux.HandleFunc("POST /model", func(w http.ResponseWriter, r *http.Request) { BridgeCall(w, r, token, bridge.Model) }) - mux.HandleFunc("POST /tool", func(w http.ResponseWriter, r *http.Request) { BridgeCall(w, r, token, bridge.Tool) }) - mux.HandleFunc("POST /event", func(w http.ResponseWriter, r *http.Request) { BridgeCall(w, r, token, bridge.Event) }) - server := &http.Server{Handler: mux, ReadHeaderTimeout: 3 * time.Second, ReadTimeout: 2 * time.Minute} - serverDone := make(chan error, 1) - go func() { serverDone <- server.Serve(listener) }() - defer func() { _ = server.Shutdown(context.Background()) }() - - request.BridgeURL = "http://" + listener.Addr().String() - request.BridgeToken = token + defer server.Close() + request.BridgeURL = server.URL + request.BridgeToken = server.Token payload, err := json.Marshal(request) if err != nil { return fmt.Errorf("encode Agent runtime request: %w", err) } - cmd := exec.CommandContext(ctx, "node", filepath.Join(runtimeDir, "agent-runtime.mjs")) + cmd := exec.CommandContext(ctx, "node", "--max-old-space-size="+nodeMemoryMB(), filepath.Join(runtimeDir, "agent-runtime.mjs")) cmd.Dir = runtimeDir cmd.Stdin = strings.NewReader(string(payload)) + // 环境变量白名单：只传 Node 运行所需的最少变量。 + cmd.Env = []string{ + "PATH=" + os.Getenv("PATH"), + "HOME=" + request.AgentDir, + "NODE_ENV=production", + "PI_OFFLINE=1", + } stdout, err := cmd.StdoutPipe() if err != nil { return fmt.Errorf("open Agent runtime output: %w", err) @@ -98,32 +107,10 @@ func Run(ctx context.Context, request ProcessRequest, bridge Bridge) error { if err := cmd.Start(); err != nil { return fmt.Errorf("start Agent runtime: %w", err) } - stdoutErr := make(chan error, 1) - go func() { - scanner := bufio.NewScanner(stdout) - scanner.Buffer(make([]byte, 64*1024), lineLimit) - for scanner.Scan() { - var event struct { - Event string `json:"event"` - Message string `json:"message"` - } - if err := json.Unmarshal(scanner.Bytes(), &event); err != nil { - stdoutErr <- fmt.Errorf("decode Agent runtime event: %w", err) - return - } - if event.Event == "runtime_error" || event.Event == "bridge_error" { - stdoutErr <- errors.New(firstNonEmpty(event.Message, "Agent runtime failed")) - return - } - } - if err := scanner.Err(); err != nil { - stdoutErr <- fmt.Errorf("read Agent runtime output: %w", err) - return - } - stdoutErr <- nil - }() + // StdoutPipe 的读取必须在 Wait 之前结束。Wait 会关闭管道，提前调用会把 + // 正常退出误报为 "file already closed"。 + lineErr := ReadOutput(stdout) waitErr := cmd.Wait() - lineErr := <-stdoutErr if lineErr != nil { return lineErr } @@ -136,11 +123,65 @@ func Run(ctx context.Context, request ProcessRequest, bridge Bridge) error { return nil } -func BridgeCall(w http.ResponseWriter, r *http.Request, expectedToken string, handler func(context.Context, map[string]json.RawMessage) (any, error)) { - if r.RemoteAddr == "" || !strings.HasPrefix(r.RemoteAddr, "127.0.0.1:") { - http.Error(w, `{"error":"loopback bridge only"}`, http.StatusForbidden) - return +func StartBridge(listenAddr, advertiseHost string, bridge Bridge) (*BridgeServer, error) { + if bridge.Model == nil || bridge.Tool == nil || bridge.Event == nil { + return nil, errors.New("Agent bridge handlers are incomplete") } + advertiseHost = strings.TrimSpace(advertiseHost) + if advertiseHost == "" || strings.ContainsAny(advertiseHost, `:/\\ `) { + return nil, errors.New("Agent bridge host must be a hostname or IP without a scheme or port") + } + tokenBytes := make([]byte, 32) + if _, err := rand.Read(tokenBytes); err != nil { + return nil, fmt.Errorf("create Agent bridge credential: %w", err) + } + token := hex.EncodeToString(tokenBytes) + listener, err := net.Listen("tcp", listenAddr) + if err != nil { + return nil, fmt.Errorf("listen for Agent bridge: %w", err) + } + tcpAddr, ok := listener.Addr().(*net.TCPAddr) + if !ok || tcpAddr.Port == 0 { + _ = listener.Close() + return nil, errors.New("Agent bridge did not receive a TCP port") + } + mux := http.NewServeMux() + mux.HandleFunc("POST /model", func(w http.ResponseWriter, r *http.Request) { BridgeCall(w, r, token, bridge.Model) }) + mux.HandleFunc("POST /tool", func(w http.ResponseWriter, r *http.Request) { BridgeCall(w, r, token, bridge.Tool) }) + mux.HandleFunc("POST /event", func(w http.ResponseWriter, r *http.Request) { BridgeCall(w, r, token, bridge.Event) }) + server := &http.Server{Handler: mux, ReadHeaderTimeout: 3 * time.Second, ReadTimeout: 2 * time.Minute} + go func() { _ = server.Serve(listener) }() + return &BridgeServer{ + URL: fmt.Sprintf("http://%s:%d", advertiseHost, tcpAddr.Port), + Token: token, + Close: func() { _ = server.Shutdown(context.Background()) }, + }, nil +} + +func ReadOutput(stdout io.Reader) error { + scanner := bufio.NewScanner(stdout) + scanner.Buffer(make([]byte, 64*1024), lineLimit) + var lastError error + for scanner.Scan() { + var event struct { + Event string `json:"event"` + Message string `json:"message"` + } + if err := json.Unmarshal(scanner.Bytes(), &event); err != nil { + lastError = fmt.Errorf("decode Agent runtime event: %w", err) + continue + } + if event.Event == "runtime_error" || event.Event == "bridge_error" { + lastError = errors.New(firstNonEmpty(event.Message, "Agent runtime failed")) + } + } + if err := scanner.Err(); err != nil { + return fmt.Errorf("read Agent runtime output: %w", err) + } + return lastError +} + +func BridgeCall(w http.ResponseWriter, r *http.Request, expectedToken string, handler func(context.Context, map[string]json.RawMessage) (any, error)) { if r.Header.Get("Authorization") != "Bearer "+expectedToken { http.Error(w, `{"error":"unauthorized bridge"}`, http.StatusUnauthorized) return @@ -181,6 +222,37 @@ func RuntimeDir() (string, error) { return "", errors.New("Agent runtime files are missing; set CANVAS_PI_RUNTIME_DIR") } +var ( + agentSlotsOnce sync.Once + agentSlots chan struct{} +) + +func acquireAgentProcess(ctx context.Context) (func(), error) { + agentSlotsOnce.Do(func() { + limit := 8 + if raw := strings.TrimSpace(os.Getenv("CANVAS_AGENT_MAX_PROCESSES")); raw != "" { + if parsed, err := strconv.Atoi(raw); err == nil && parsed > 0 && parsed <= 1000 { + limit = parsed + } + } + agentSlots = make(chan struct{}, limit) + }) + select { + case agentSlots <- struct{}{}: + return func() { <-agentSlots }, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +func nodeMemoryMB() string { + raw := strings.TrimSpace(os.Getenv("YINGCE_AGENT_NODE_MEMORY_MB")) + if parsed, err := strconv.Atoi(raw); err == nil && parsed >= 128 && parsed <= 8192 { + return strconv.Itoa(parsed) + } + return "512" +} + func firstNonEmpty(values ...string) string { for _, value := range values { if strings.TrimSpace(value) != "" { diff --git a/backend/internal/agent/runtime/runtime_test.go b/backend/internal/agent/runtime/runtime_test.go new file mode 100644 index 000000000..147fef974 --- /dev/null +++ b/backend/internal/agent/runtime/runtime_test.go @@ -0,0 +1,83 @@ +package runtime + +import ( + "context" + "encoding/json" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" +) + +// fakeRuntime 用一个假的 agent-runtime.mjs 替换真实运行时，只验证 Go 侧的进程与输出处理。 +func fakeRuntime(t *testing.T, script string) { + t.Helper() + if _, err := exec.LookPath("node"); err != nil { + t.Skip("node is not installed") + } + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, "agent-runtime.mjs"), []byte(script), 0o600); err != nil { + t.Fatal(err) + } + t.Setenv("CANVAS_PI_RUNTIME_DIR", dir) +} + +func noopBridge() Bridge { + handler := func(context.Context, map[string]json.RawMessage) (any, error) { return map[string]any{"ok": true}, nil } + return Bridge{Model: handler, Tool: handler, Event: handler} +} + +// 回归：Wait 早于读完 stdout 时，大量输出后正常退出会被判成 +// "read |0: file already closed"。 +func TestRunDrainsStdoutBeforeWait(t *testing.T) { + fakeRuntime(t, ` +const line = JSON.stringify({ event: "progress", pad: "x".repeat(2048) }) + "\n"; +for (let i = 0; i < 2000; i++) process.stdout.write(line); +process.stdout.write(JSON.stringify({ event: "settled" }) + "\n"); +`) + for i := 0; i < 5; i++ { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + err := Run(ctx, ProcessRequest{Model: map[string]any{"id": "m"}}, noopBridge()) + cancel() + if err != nil { + t.Fatalf("attempt %d: normal exit reported as failure: %v", i, err) + } + } +} + +// runtime_error 之后仍要读到 EOF，并把该错误作为运行结果返回。 +func TestRunReportsRuntimeErrorAfterDrainingOutput(t *testing.T) { + fakeRuntime(t, ` +process.stdout.write(JSON.stringify({ event: "runtime_error", message: "boom" }) + "\n"); +const line = JSON.stringify({ event: "progress", pad: "x".repeat(2048) }) + "\n"; +for (let i = 0; i < 500; i++) process.stdout.write(line); +process.exitCode = 1; +`) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + err := Run(ctx, ProcessRequest{Model: map[string]any{"id": "m"}}, noopBridge()) + if err == nil || err.Error() != "boom" { + t.Fatalf("want runtime_error message, got %v", err) + } +} + +// 子进程只能拿到白名单环境变量，拿不到数据库连接等密钥。 +func TestRunDoesNotLeakBackendEnvironment(t *testing.T) { + t.Setenv("DATABASE_URL", "postgres://secret") + fakeRuntime(t, ` +if (process.env.DATABASE_URL) { + process.stdout.write(JSON.stringify({ event: "runtime_error", message: "leaked" }) + "\n"); + process.exitCode = 1; +} +`) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := Run(ctx, ProcessRequest{Model: map[string]any{"id": "m"}}, noopBridge()); err != nil { + if strings.Contains(err.Error(), "leaked") { + t.Fatal("backend environment leaked into the Agent runtime") + } + t.Fatal(err) + } +} diff --git a/backend/internal/agent/yingceagent/client.go b/backend/internal/agent/yingceagent/client.go new file mode 100644 index 000000000..27f93df8b --- /dev/null +++ b/backend/internal/agent/yingceagent/client.go @@ -0,0 +1,69 @@ +package yingceagent + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "strings" + "time" + + agentruntime "infinite-canvas/backend/internal/agent/runtime" +) + +// Run executes one Agent session on the independent yingce-agent service. +// The Go process keeps the bridge, billing, and tool execution. A configured +// remote endpoint never falls back to an embedded Node process: doing so can +// run the same step twice. +func Run(ctx context.Context, endpoint, serviceToken, bridgeHost string, request agentruntime.ProcessRequest, bridge agentruntime.Bridge) error { + endpoint = strings.TrimRight(strings.TrimSpace(endpoint), "/") + serviceToken = strings.TrimSpace(serviceToken) + if endpoint == "" { + return fmt.Errorf("YINGCE_AGENT_URL is empty") + } + if len(serviceToken) < 32 { + return fmt.Errorf("YINGCE_AGENT_TOKEN must contain at least 32 characters") + } + server, err := agentruntime.StartBridge("0.0.0.0:0", bridgeHost, bridge) + if err != nil { + return err + } + defer server.Close() + request.BridgeURL = server.URL + request.BridgeToken = server.Token + payload, err := json.Marshal(request) + if err != nil { + return fmt.Errorf("encode remote Agent request: %w", err) + } + httpRequest, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint+"/v1/runs", bytes.NewReader(payload)) + if err != nil { + return fmt.Errorf("create remote Agent request: %w", err) + } + httpRequest.Header.Set("Authorization", "Bearer "+serviceToken) + httpRequest.Header.Set("Content-Type", "application/json") + response, err := remoteClient.Do(httpRequest) + if err != nil { + return fmt.Errorf("call yingce-agent: %w", err) + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + detail, _ := io.ReadAll(io.LimitReader(response.Body, 4096)) + return fmt.Errorf("yingce-agent returned HTTP %d: %s", response.StatusCode, strings.TrimSpace(string(detail))) + } + return agentruntime.ReadOutput(response.Body) +} + +var remoteClient = &http.Client{ + Transport: &http.Transport{ + Proxy: http.ProxyFromEnvironment, + DialContext: (&net.Dialer{ + Timeout: 10 * time.Second, + KeepAlive: 30 * time.Second, + }).DialContext, + ForceAttemptHTTP2: true, + IdleConnTimeout: 90 * time.Second, + }, +} diff --git a/backend/internal/agent/yingceagent/client_test.go b/backend/internal/agent/yingceagent/client_test.go new file mode 100644 index 000000000..d76302f67 --- /dev/null +++ b/backend/internal/agent/yingceagent/client_test.go @@ -0,0 +1,119 @@ +package yingceagent + +import ( + "bufio" + "context" + "encoding/json" + "net/http" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "testing" + "time" + + agentruntime "infinite-canvas/backend/internal/agent/runtime" +) + +func TestRemoteRunCallsBridgeAndRejectsBadToken(t *testing.T) { + if _, err := exec.LookPath("node"); err != nil { + t.Skip("node is not installed") + } + _, source, _, ok := runtime.Caller(0) + if !ok { + t.Fatal("locate test file") + } + serverPath := filepath.Clean(filepath.Join(filepath.Dir(source), "../../../../yingce-agent/server.mjs")) + runtimePath := filepath.Join(t.TempDir(), "agent-runtime.mjs") + script := ` +let raw = ""; +for await (const chunk of process.stdin) raw += chunk; +const request = JSON.parse(raw); +const response = await fetch(new URL("/tool", request.bridgeURL), { + method: "POST", + headers: { "content-type": "application/json", authorization: "Bearer " + request.bridgeToken }, + body: JSON.stringify({ name: "lookup" }), +}); +if (!response.ok) { + process.stdout.write(JSON.stringify({ event: "runtime_error", message: "bridge " + response.status }) + "\n"); + process.exitCode = 1; +} else { + process.stdout.write(JSON.stringify({ event: "settled" }) + "\n"); +} +` + if err := os.WriteFile(runtimePath, []byte(script), 0o600); err != nil { + t.Fatal(err) + } + token := strings.Repeat("ab", 16) + cmd := exec.Command("node", serverPath) + cmd.Env = append(os.Environ(), + "YINGCE_AGENT_TOKEN="+token, + "YINGCE_AGENT_RUNTIME="+runtimePath, + "PORT=0", + "MAX_CONCURRENT_SESSIONS=2", + ) + stdout, err := cmd.StdoutPipe() + if err != nil { + t.Fatal(err) + } + cmd.Stderr = os.Stderr + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = cmd.Process.Kill() + _, _ = cmd.Process.Wait() + }) + reader := bufio.NewReader(stdout) + line, err := reader.ReadString('\n') + if err != nil { + t.Fatal(err) + } + if !strings.HasPrefix(line, "YINGCE_AGENT_READY ") { + t.Fatalf("server did not report its port: %q", line) + } + endpoint := "http://127.0.0.1:" + strings.TrimSpace(strings.TrimPrefix(line, "YINGCE_AGENT_READY ")) + + health, err := http.Get(endpoint + "/healthz") + if err != nil { + t.Fatal(err) + } + defer health.Body.Close() + if health.StatusCode != http.StatusOK { + t.Fatalf("health status %d", health.StatusCode) + } + + unauthorized, err := http.Post(endpoint+"/v1/runs", "application/json", strings.NewReader(`{}`)) + if err != nil { + t.Fatal(err) + } + defer unauthorized.Body.Close() + if unauthorized.StatusCode != http.StatusUnauthorized { + t.Fatalf("missing token status %d", unauthorized.StatusCode) + } + + called := false + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + err = Run(ctx, endpoint, token, "127.0.0.1", agentruntime.ProcessRequest{ + Model: map[string]any{"id": "m"}, + }, agentruntime.Bridge{ + Model: func(context.Context, map[string]json.RawMessage) (any, error) { + return map[string]any{"ok": true}, nil + }, + Tool: func(context.Context, map[string]json.RawMessage) (any, error) { + called = true + return map[string]any{"content": "ok"}, nil + }, + Event: func(context.Context, map[string]json.RawMessage) (any, error) { + return map[string]any{"ok": true}, nil + }, + }) + if err != nil { + t.Fatal(err) + } + if !called { + t.Fatal("remote runtime did not call the Go tool bridge") + } +} diff --git a/backend/internal/app/cloud_agent_media_test.go b/backend/internal/app/cloud_agent_media_test.go index 243111829..5a6ea5265 100644 --- a/backend/internal/app/cloud_agent_media_test.go +++ b/backend/internal/app/cloud_agent_media_test.go @@ -619,7 +619,9 @@ func TestCloudAgentMediaPreviousDraftRequiresNewApproval(t *testing.T) { } } -func TestCloudAgentAutoMediaSubmitsWithoutApproval(t *testing.T) { +// 界面承诺“图片、视频始终先创建草稿，再经独立审批才提交”：auto 也必须先审批， +// 审批前不建任务、不扣费；批准后恰好提交一次。 +func TestCloudAgentAutoMediaRequiresApprovalBeforeSubmit(t *testing.T) { s, db, a := agentMediaFixture(t) // Omitted size and duration should be filled by the selected model's catalog // defaults during admission instead of producing a repair turn. @@ -638,16 +640,28 @@ func TestCloudAgentAutoMediaSubmitsWithoutApproval(t *testing.T) { if err != nil { t.Fatal(err) } - if run.Status == "waiting_approval" || state.Approval != nil { - t.Fatalf("auto mode unexpectedly requested approval: status=%s approval=%+v", run.Status, state.Approval) + if run.Status != "waiting_approval" || state.Approval == nil { + t.Fatalf("auto mode submitted media without approval: status=%s", run.Status) + } + var pendingTasks, pendingOrders int64 + db.Model(&model.Task{}).Where("type = ?", "canvas_video").Count(&pendingTasks) + db.Model(&model.BillingOrder{}).Count(&pendingOrders) + if pendingTasks != 0 || pendingOrders != ordersBefore { + t.Fatalf("charged/submitted before approval: tasks=%d orders=%d before=%d", pendingTasks, pendingOrders, ordersBefore) + } + if err := s.DecideCloudAgentApproval("user", run.ID, state.Approval.ID, "approve", ""); err != nil { + t.Fatal(err) + } + if err := s.advanceCloudAgentByID("user", run.ID); err != nil { + t.Fatal(err) + } + run, _ = s.repo.CloudAgent("user", run.ID) + state, err = cloudAgentDecode(run) + if err != nil { + t.Fatal(err) } if state.MediaTaskID == "" { - t.Fatalf("auto mode did not submit a media task: %+v", state.Events) - } - for _, event := range state.Events { - if event.Type == "approval_requested" { - t.Fatal("auto mode emitted an approval event") - } + t.Fatalf("approved auto media was not submitted: status=%s", run.Status) } task, err := s.repo.TaskForUser("user", state.MediaTaskID) if err != nil { @@ -665,7 +679,7 @@ func TestCloudAgentAutoMediaSubmitsWithoutApproval(t *testing.T) { nodes, _ := creationObjects(doc["nodes"]) meta, _ := nodes[a.NodeID]["metadata"].(map[string]any) if meta["status"] != "loading" || meta["taskId"] != task.ID || meta["agentDraftRunId"] != nil { - t.Fatalf("auto submission did not finalize the draft: %+v", meta) + t.Fatalf("approved submission did not finalize the draft: %+v", meta) } var tasks, orders int64 if err := db.Model(&model.Task{}).Where("type = ?", "canvas_video").Count(&tasks).Error; err != nil { @@ -675,7 +689,7 @@ func TestCloudAgentAutoMediaSubmitsWithoutApproval(t *testing.T) { t.Fatal(err) } if tasks != 1 || orders != ordersBefore+1 { - t.Fatalf("auto submission created unexpected records: tasks=%d orders=%d before=%d", tasks, orders, ordersBefore) + t.Fatalf("approved submission created unexpected records: tasks=%d orders=%d before=%d", tasks, orders, ordersBefore) } // Re-entering a submitted call is idempotent and must not reserve/submit a // second generation. @@ -686,7 +700,7 @@ func TestCloudAgentAutoMediaSubmitsWithoutApproval(t *testing.T) { t.Fatal(err) } if tasks != 1 { - t.Fatalf("re-entering auto media duplicated task: %d", tasks) + t.Fatalf("re-entering approved media duplicated task: %d", tasks) } // Admission failures are still non-billed tool failures and do not create an diff --git a/backend/internal/app/cloud_agent_pi_coordinator.go b/backend/internal/app/cloud_agent_pi_coordinator.go index 5cdef7833..e92ef8202 100644 --- a/backend/internal/app/cloud_agent_pi_coordinator.go +++ b/backend/internal/app/cloud_agent_pi_coordinator.go @@ -231,9 +231,11 @@ func (s *Service) recoverCloudAgentPiRunners() { if run.Status == "waiting_approval" || run.Status == "waiting_user" { continue } - // 批准后的媒体任务仍在生成：由 finishApprovedCloudAgentMedia 等结果回写后再恢复运行。 - // 这时提前启动运行时，会在媒体任务未结束时再入队模型步骤，形成两个活动任务。 - if s.cloudAgentAwaitingMedia(run.UserID, run.ID) { + // 批准后的媒体任务仍在生成。进程重启后内存 waiter 已经消失，这里按 + // 持久化的任务 ID 重建；不能同时再启动一轮模型步骤。 + state, decodeErr := cloudAgentDecode(&run) + if decodeErr == nil && state.MediaTaskID != "" { + s.startApprovedCloudAgentMediaWaiter(run.UserID, run.ID, state.MediaTaskID) continue } s.startCloudAgentPi(run.UserID, run.ID) @@ -245,18 +247,6 @@ func (s *Service) recoverCloudAgentPiRunners() { } } -// cloudAgentAwaitingMedia reports whether the run is waiting on an approved -// media task. The runtime must not start (and schedule a model step) until that -// task has been written back; one run owns at most one active task. -func (s *Service) cloudAgentAwaitingMedia(userID, runID string) bool { - run, err := s.repo.CloudAgent(userID, runID) - if err != nil { - return false - } - state, err := cloudAgentDecode(run) - return err == nil && state.MediaTaskID != "" -} - // runCloudAgentPiSession 核心会话管理 func (s *Service) runCloudAgentPiSession(ctx context.Context, userID, runID string) error { // 1. 加载运行状态 @@ -1085,6 +1075,32 @@ func (s *Service) cloudAgentPiTool(ctx context.Context, userID, runID string, pa if call.Function.Name == "" { return nil, fmt.Errorf("tool call is missing a name") } + + // 工具白名单校验：拒绝未声明的工具调用 + run, err := s.repo.CloudAgent(userID, runID) + if err != nil { + return nil, err + } + state, err := cloudAgentDecode(run) + if err != nil { + return nil, err + } + + allowedTools := make(map[string]bool) + for _, tool := range state.Canonical.Tools { + if fn, ok := tool["function"].(map[string]interface{}); ok { + name := stringValue(fn["name"]) + if name != "" { + allowedTools[name] = true + } + } + } + + if !allowedTools[call.Function.Name] { + log.Printf("[Agent] rejected undeclared tool call: %s in run %s", call.Function.Name, runID) + return nil, fmt.Errorf("未声明的工具: %s", call.Function.Name) + } + return s.executeCloudAgentRuntimeTool(ctx, userID, runID, call) } diff --git a/backend/internal/app/cloud_agent_pi_runtime.go b/backend/internal/app/cloud_agent_pi_runtime.go index 822341e99..850d1fcd8 100644 --- a/backend/internal/app/cloud_agent_pi_runtime.go +++ b/backend/internal/app/cloud_agent_pi_runtime.go @@ -2,8 +2,11 @@ package app import ( "context" + "os" + "strings" agentruntime "infinite-canvas/backend/internal/agent/runtime" + "infinite-canvas/backend/internal/agent/yingceagent" ) // These aliases keep the app coordinator's request shape stable while the @@ -37,7 +40,7 @@ type cloudAgentPiProcessRequest struct { } func runCloudAgentPi(ctx context.Context, request cloudAgentPiProcessRequest, bridge cloudAgentPiBridge) error { - return agentruntime.Run(ctx, agentruntime.ProcessRequest{ + runtimeRequest := agentruntime.ProcessRequest{ BridgeURL: request.BridgeURL, BridgeToken: request.BridgeToken, SessionFile: request.SessionFile, SessionJSONL: request.SessionJSONL, SessionDir: request.SessionDir, SessionID: request.SessionId, UserID: request.UserId, CanvasID: request.CanvasId, RunID: request.RunId, @@ -45,5 +48,9 @@ func runCloudAgentPi(ctx context.Context, request cloudAgentPiProcessRequest, br SkillPaths: request.SkillPaths, EnabledSkills: request.EnabledSkills, Profile: request.Profile, Memory: request.Memory, Canvas: request.Canvas, Features: request.Features, Tools: request.Tools, Compaction: request.Compaction, Permissions: request.Permissions, Model: request.Model, - }, agentruntime.Bridge(bridge)) + } + if endpoint := strings.TrimSpace(os.Getenv("YINGCE_AGENT_URL")); endpoint != "" { + return yingceagent.Run(ctx, endpoint, os.Getenv("YINGCE_AGENT_TOKEN"), os.Getenv("YINGCE_AGENT_BRIDGE_HOST"), runtimeRequest, agentruntime.Bridge(bridge)) + } + return agentruntime.Run(ctx, runtimeRequest, agentruntime.Bridge(bridge)) } diff --git a/backend/internal/app/cloud_agent_runtime.go b/backend/internal/app/cloud_agent_runtime.go index 44162769f..4466eea69 100644 --- a/backend/internal/app/cloud_agent_runtime.go +++ b/backend/internal/app/cloud_agent_runtime.go @@ -1712,6 +1712,7 @@ func (s *Service) advanceCloudAgentTool(run *model.CloudAgentExecution, state *c } allowed := cloudAgentToolAllowed(state.Request, call.Function.Name) mediaTool := call.Function.Name == "generate_media" || call.Function.Name == "image_layer_split" + // 媒体工具在所有模式下都进入审批；其他写入只在 request_approval 下审批。 if allowed && cloudAgentWrite(call.Function.Name) && (state.Request.PermissionMode == "request_approval" || mediaTool) && state.Approval == nil { var plan *cloudAgentMediaPlan var modelName string @@ -1753,7 +1754,7 @@ func (s *Service) advanceCloudAgentTool(run *model.CloudAgentExecution, state *c return s.cloudAgentMediaError(run, state, "admission", false, false, err) } } - if plan != nil && state.Request.PermissionMode == "auto" { + if plan != nil && state.Request.PermissionMode == "auto" && !mediaTool { if plan.Prepared != nil { // The draft and quote were already checkpointed. Reuse them after a // worker restart instead of dry-admitting and mutating the canvas again. diff --git a/backend/internal/app/cloud_agent_single_active_task_test.go b/backend/internal/app/cloud_agent_single_active_task_test.go index e29469ec5..bfeefea42 100644 --- a/backend/internal/app/cloud_agent_single_active_task_test.go +++ b/backend/internal/app/cloud_agent_single_active_task_test.go @@ -26,6 +26,20 @@ func TestCloudAgentModelStepWaitsForPendingMediaTask(t *testing.T) { if err != nil { t.Fatal(err) } + // 媒体生成在所有模式下都要先审批；批准后才提交媒体任务。 + if state.Approval == nil { + t.Fatalf("media did not request approval (status=%s)", run.Status) + } + s.disablePiRuntime = true + if err := s.DecideCloudAgentApproval("user", run.ID, state.Approval.ID, "approve", ""); err != nil { + t.Fatal(err) + } + if run, err = s.repo.CloudAgent("user", run.ID); err != nil { + t.Fatal(err) + } + if state, err = cloudAgentDecode(run); err != nil { + t.Fatal(err) + } if state.MediaTaskID == "" { t.Fatalf("fixture did not submit a media task (status=%s failure=%q)", run.Status, run.FailureMessage) } diff --git a/backend/internal/app/cloud_agent_tools.go b/backend/internal/app/cloud_agent_tools.go index e13974e6f..ba977abb7 100644 --- a/backend/internal/app/cloud_agent_tools.go +++ b/backend/internal/app/cloud_agent_tools.go @@ -536,7 +536,7 @@ func compileCloudAgentTools(req CloudAgentRequest, includeProfileTool bool) []ma "topic", "category", "situation") } if req.PermissionMode != "read_only" && len(req.ContextScope) > 0 { - add("image_layer_split", "将图片按用户指定对象拆分为独立透明图层。参数与 generate_media 的图片生成参数一致，但 mode 固定为 image；request_approval 进入界面独立审批，auto 由服务端准入成功后直接提交，不会再次弹出模型生成审批。", map[string]any{ + add("image_layer_split", "将图片按用户指定对象拆分为独立透明图层。参数与 generate_media 的图片生成参数一致，但 mode 固定为 image；所有权限模式都会先创建草稿并进入界面独立审批，用户批准后才提交生成任务。", map[string]any{ "prompt": str("需要拆分的对象与透明背景要求"), "logicalModelId": str("selection.logicalModelId"), "channelId": str("selection.channelId"), "channelModelKey": str("selection.channelModelKey"), "quality": str("模型支持的质量档位"), "snapshotHash": str("最近画布读取返回的 mediaSnapshotHash，可省略"), "nodeId": str("新的结果节点ID"), "title": str("结果节点名称"), "referenceNodeIds": map[string]any{"type": "array", "maxItems": 16, "items": str("源图片节点ID")}, }, "prompt", "nodeId", "title", "referenceNodeIds") @@ -610,7 +610,7 @@ func compileCloudAgentTools(req CloudAgentRequest, includeProfileTool bool) []ma }, "snapshotHash") } if req.PermissionMode != "read_only" && len(req.ContextScope) > 0 { - add("generate_media", "提交媒体生成：先准备草稿和引用；request_approval 需要用户确认，auto 在服务端完成模型、能力、价格、预算和资源校验后直接提交，不会再次弹出模型生成审批。auto 只在准入成功后提交。创建节点/改提示词/连线用 canvas_apply_ops。先读画布和模型目录并遵守其能力。已有任务或产物的节点不可覆盖；状态用 generation/task_get。sourceNodeId 为文本输入，referenceNodeIds 为媒体输入，referenceTransientIds 仅接受标注工具返回值，不接受URL。已提交失败要告知用户，重试须用户明确要求并重新审批。", map[string]any{ + add("generate_media", "提交媒体生成：先准备草稿和引用；所有权限模式都会先经服务端完成模型、能力、价格、预算和资源校验，再进入界面独立审批，用户批准后才提交生成任务。创建节点/改提示词/连线用 canvas_apply_ops。先读画布和模型目录并遵守其能力。已有任务或产物的节点不可覆盖；状态用 generation/task_get。sourceNodeId 为文本输入，referenceNodeIds 为媒体输入，referenceTransientIds 仅接受标注工具返回值，不接受URL。已提交失败要告知用户，重试须用户明确要求并重新审批。", map[string]any{ "mode": map[string]any{"type": "string", "enum": cloudAgentGenerationModeNames()}, "prompt": str("完整生成提示词；引用素材时在对应描述中使用 @图片1、@视频1、@音频1，各类型按 referenceNodeIds 中出现顺序独立编号，文本来源不占媒体编号。服务端会为遗漏的已选素材补齐引用标签，不推断素材用途"), "logicalModelId": str("selection.logicalModelId；与channelId/channelModelKey互斥"), "channelId": str("selection.channelId"), "channelModelKey": str("selection.channelModelKey"), "durationSeconds": map[string]any{"type": "integer", "minimum": 0}, "size": str("模型支持的画幅，例如9:16"), "quality": str("目录支持的分辨率或质量"), "videoGenerateAudio": map[string]any{"type": "boolean", "description": "是否生成音频，仅视频可用"}, diff --git a/backend/internal/hostupdate/manager_ops.go b/backend/internal/hostupdate/manager_ops.go index 9f3ef34ef..72dfc3550 100644 --- a/backend/internal/hostupdate/manager_ops.go +++ b/backend/internal/hostupdate/manager_ops.go @@ -4,6 +4,7 @@ import ( "archive/zip" "bytes" "context" + "crypto/rand" "crypto/sha256" "encoding/hex" "encoding/json" @@ -246,6 +247,7 @@ func (m *Manager) restartSelf() { type deploymentImages struct { backend string web string + agent string } func immutableImageRefs(repository, version string) deploymentImages { @@ -254,6 +256,7 @@ func immutableImageRefs(repository, version string) deploymentImages { return deploymentImages{ backend: "ghcr.io/" + owner + "/open-ai-canvas-backend:" + tag, web: "ghcr.io/" + owner + "/open-ai-canvas-web:" + tag, + agent: "ghcr.io/" + owner + "/open-ai-canvas-yingce-agent:" + tag, } } @@ -329,12 +332,16 @@ func (m *Manager) writeComposeEnvOverride(images deploymentImages) (string, erro cleanup() return "", err } + if err := setEnvValue(path, "CANVAS_YINGCE_AGENT_IMAGE", images.agent); err != nil { + cleanup() + return "", err + } return path, nil } func (m *Manager) verifyImages(targetVersion string) (deploymentImages, error) { images := immutableImageRefs(m.config.Repository, targetVersion) - refs := []string{images.backend, images.web} + refs := []string{images.backend, images.web, images.agent} digests := make([]string, 0, len(refs)) for _, image := range refs { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) @@ -362,7 +369,7 @@ func (m *Manager) verifyImages(targetVersion string) (deploymentImages, error) { } digests = append(digests, digest) } - return deploymentImages{backend: digests[0], web: digests[1]}, nil + return deploymentImages{backend: digests[0], web: digests[1], agent: digests[2]}, nil } func setDeploymentImages(path, version string, images deploymentImages) error { @@ -372,7 +379,28 @@ func setDeploymentImages(path, version string, images deploymentImages) error { if err := setEnvValue(path, "CANVAS_BACKEND_IMAGE", images.backend); err != nil { return err } - return setEnvValue(path, "CANVAS_WEB_IMAGE", images.web) + if err := setEnvValue(path, "CANVAS_WEB_IMAGE", images.web); err != nil { + return err + } + if err := setEnvValue(path, "CANVAS_YINGCE_AGENT_IMAGE", images.agent); err != nil { + return err + } + return ensureAgentToken(path) +} + +func ensureAgentToken(path string) error { + values, err := readEnvFile(path) + if err != nil { + return err + } + if len(strings.TrimSpace(values["YINGCE_AGENT_TOKEN"])) >= 32 { + return nil + } + token := make([]byte, 32) + if _, err := rand.Read(token); err != nil { + return fmt.Errorf("生成 Agent 服务凭证：%w", err) + } + return setEnvValue(path, "YINGCE_AGENT_TOKEN", hex.EncodeToString(token)) } func (m *Manager) createBackup(version string) (Backup, error) { diff --git a/backend/internal/hostupdate/manager_ops_test.go b/backend/internal/hostupdate/manager_ops_test.go index 6b2f7ee9c..b9acc705e 100644 --- a/backend/internal/hostupdate/manager_ops_test.go +++ b/backend/internal/hostupdate/manager_ops_test.go @@ -172,7 +172,7 @@ func TestWriteComposeEnvOverrideReplacesImageRefs(t *testing.T) { t.Fatal(err) } manager := &Manager{config: Config{InstallDir: installDir, EnvFile: ".env", StateDir: stateDir}} - path, err := manager.writeComposeEnvOverride(deploymentImages{backend: "new-backend", web: "new-web"}) + path, err := manager.writeComposeEnvOverride(deploymentImages{backend: "new-backend", web: "new-web", agent: "new-agent"}) if err != nil { t.Fatal(err) } @@ -182,7 +182,7 @@ func TestWriteComposeEnvOverrideReplacesImageRefs(t *testing.T) { t.Fatal(err) } value := string(data) - if !strings.Contains(value, "CANVAS_BACKEND_IMAGE=new-backend\n") || !strings.Contains(value, "CANVAS_WEB_IMAGE=new-web\n") || !strings.Contains(value, "POSTGRES_DB=canvas\n") { + if !strings.Contains(value, "CANVAS_BACKEND_IMAGE=new-backend\n") || !strings.Contains(value, "CANVAS_WEB_IMAGE=new-web\n") || !strings.Contains(value, "CANVAS_YINGCE_AGENT_IMAGE=new-agent\n") || !strings.Contains(value, "POSTGRES_DB=canvas\n") { t.Fatalf("unexpected override env: %q", value) } } diff --git a/backend/internal/prompts/agent-media-policy.md b/backend/internal/prompts/agent-media-policy.md index 2a449463d..ad9cc7e21 100644 --- a/backend/internal/prompts/agent-media-policy.md +++ b/backend/internal/prompts/agent-media-policy.md @@ -1,16 +1,16 @@ --- id: cloud-agent-media -version: 4 +version: 5 --- # 影策 Cloud Agent 媒体策略 生成前读取节点能力和当前画布，向 model_list 传实际 mode 与 referenceNodeIds，按真实参考素材筛选模型。多图生视频不能选择仅文生视频模型，也不能丢弃参考图来迁就模型。返回的 options / profiles 是配置能力，不凭模型名称猜测；同一配置必须同时满足参考数量、时长、画幅和音频要求，不能拼接不同 profiles 的能力。空候选时说明能力缺口，不编造可用模型；素材或模式改变后旧的筛选结果不再适用。 -用户要求生成且参数齐备时，本轮直接调用 generate_media。request_approval 会进入界面独立审批；auto 会由服务端完成模型、能力、价格、预算和资源准入后直接提交，不会再产生一张模型生成审批卡。不要把仅创建节点、连线或填写提示词说成已提交。用户仅要求改词或准备草稿时不要擅自生成。已有未提交草稿应优先续用原节点：无任务的空白媒体占位节点、当前运行草稿、已结束且清理完成运行的遗留草稿均需最新快照；request_approval 需要新的审批，auto 由服务端重新准入。另一运行仍在处理的草稿先完成或取消原运行；已有任务或成品不可覆盖。不要通过不断换新 ID 逃避错误，不把所有旧节点都误认为不可操作。 +用户要求生成且参数齐备时，本轮直接调用 generate_media。图片、视频在所有权限模式下都先创建草稿，由服务端完成模型、能力、价格、预算和资源准入后进入界面独立审批，用户批准后才提交生成任务；auto 只豁免其他画布修改，不豁免媒体生成。不要把仅创建节点、连线或填写提示词说成已提交。用户仅要求改词或准备草稿时不要擅自生成。已有未提交草稿应优先续用原节点：无任务的空白媒体占位节点、当前运行草稿、已结束且清理完成运行的遗留草稿均需最新快照；续用草稿同样需要新的审批。另一运行仍在处理的草稿先完成或取消原运行；已有任务或成品不可覆盖。不要通过不断换新 ID 逃避错误，不把所有旧节点都误认为不可操作。 媒体生成先区分参数选择与收费授权。用户明确指定的模型、画幅、时长、音频必须遵守；用户说“随便”“你决定”“按默认”时允许安全默认，不要求四项逐个确认：从当前目录选择支持参考方式、参数且预算可覆盖的模型，优先用户最近明确使用且仍可用的模型，否则选择价格清楚的低成本可用项。画幅优先匹配参考图原始宽高比且必须在模型支持范围内，否则用目录默认；时长用目录默认，不编造；模型不支持音频且用户未要求声音时直接用默片。 -用户对上一条明确推荐方案说“可以”“开始吧”，即接受该方案的参数，不要求回复特定确认口令。只有引用对象不明确、用户硬性要求与能力冲突、无有效默认值或预算不足时，一次性询问真正缺少的事项。选择后简短说明实际参数和默认来源；request_approval 创建草稿交给审批卡，auto 直接提交并用 task_get/generation 核实结果，不添加额外文字确认轮次。 +用户对上一条明确推荐方案说“可以”“开始吧”，即接受该方案的参数，不要求回复特定确认口令。只有引用对象不明确、用户硬性要求与能力冲突、无有效默认值或预算不足时，一次性询问真正缺少的事项。选择后简短说明实际参数和默认来源；创建草稿交给审批卡，批准提交后用 task_get/generation 核实结果，不添加额外文字确认轮次。 -创建媒体草稿及连线不等于收费授权；request_approval 必须等待界面独立审批，auto 只有在服务端准入成功后才提交，不能再等待或伪造审批事件。“随便”“开始”和默认参数都不能绕过服务端准入。已提交的收费任务失败不得自动重提、换模型或盲重试；把失败原因告诉用户并结束本步，不要把整轮对话停掉。查询计费时，authorizedChargeMicrocredits / chargeLimitMicrocredits 是预授权或上限，不是实际消费；只有 billing.status=settled 时才使用 actualCredits（或 actualAmountMicrocredits）描述已扣费，reservedCredits 不能当作已扣费。 +创建媒体草稿及连线不等于收费授权；所有权限模式都必须等待界面独立审批，不能伪造审批事件。“随便”“开始”和默认参数都不能绕过服务端准入。已提交的收费任务失败不得自动重提、换模型或盲重试；把失败原因告诉用户并结束本步，不要把整轮对话停掉。查询计费时，authorizedChargeMicrocredits / chargeLimitMicrocredits 是预授权或上限，不是实际消费；只有 billing.status=settled 时才使用 actualCredits（或 actualAmountMicrocredits）描述已扣费，reservedCredits 不能当作已扣费。 diff --git a/backend/internal/prompts/agent_policy_test.go b/backend/internal/prompts/agent_policy_test.go index ff0fe8634..3374e1f91 100644 --- a/backend/internal/prompts/agent_policy_test.go +++ b/backend/internal/prompts/agent_policy_test.go @@ -10,15 +10,15 @@ func TestLoadAgentPoliciesUsesDocumentMetadata(t *testing.T) { if err != nil { t.Fatal(err) } - if system.ID != "cloud-agent-system" || system.Version != 12 || media.ID != "cloud-agent-media" || media.Version != 4 { + if system.ID != "cloud-agent-system" || system.Version != 12 || media.ID != "cloud-agent-media" || media.Version != 5 { t.Fatalf("unexpected policy metadata: system=%+v media=%+v", system, media) } if strings.Contains(system.Text, "id: cloud-agent-system") || !strings.HasPrefix(system.Text, "# 影策 Cloud Agent") { t.Fatalf("metadata leaked into compiled policy body: %q", system.Text) } for _, phrase := range []string{ - "auto 会由服务端完成模型、能力、价格、预算和资源准入后直接提交", - "request_approval 必须等待界面独立审批", + "auto 只豁免其他画布修改，不豁免媒体生成", + "所有权限模式都必须等待界面独立审批", "authorizedChargeMicrocredits / chargeLimitMicrocredits 是预授权或上限，不是实际消费", } { if !strings.Contains(media.Text, phrase) { diff --git a/docker-compose.deploy.yml b/docker-compose.deploy.yml index ad53cdfef..5e76324a7 100644 --- a/docker-compose.deploy.yml +++ b/docker-compose.deploy.yml @@ -38,6 +38,20 @@ services: postgres: condition: service_healthy + yingce-agent: + image: ${CANVAS_YINGCE_AGENT_IMAGE:?请先配置 CANVAS_YINGCE_AGENT_IMAGE（必须包含镜像 digest）} + environment: + YINGCE_AGENT_TOKEN: ${YINGCE_AGENT_TOKEN:?请先配置 YINGCE_AGENT_TOKEN} + MAX_CONCURRENT_SESSIONS: ${YINGCE_AGENT_MAX_SESSIONS:-8} + NODE_MAX_OLD_SPACE_SIZE: ${YINGCE_AGENT_NODE_MEMORY_MB:-512} + healthcheck: + test: ["CMD", "wget", "-qO-", "http://127.0.0.1:8081/healthz"] + interval: 10s + timeout: 3s + retries: 10 + start_period: 10s + restart: unless-stopped + backend: image: ${CANVAS_BACKEND_IMAGE:?请先配置 CANVAS_BACKEND_IMAGE（必须包含镜像 digest）} environment: @@ -57,6 +71,9 @@ services: CANVAS_PUBLIC_BASE_URL: ${CANVAS_PUBLIC_BASE_URL:-} CANVAS_UPDATER_SOCKET: /run/open-ai-canvas-updater/updater.sock CANVAS_UPDATER_TOKEN: ${CANVAS_UPDATER_TOKEN:-} + YINGCE_AGENT_URL: http://yingce-agent:8081 + YINGCE_AGENT_TOKEN: ${YINGCE_AGENT_TOKEN:?请先配置 YINGCE_AGENT_TOKEN} + YINGCE_AGENT_BRIDGE_HOST: backend GIN_MODE: release volumes: - ${CANVAS_DATA_PATH:-backend-data}:/data @@ -68,6 +85,8 @@ services: condition: service_healthy migrate: condition: service_completed_successfully + yingce-agent: + condition: service_healthy restart: unless-stopped stop_grace_period: ${CANVAS_SHUTDOWN_TIMEOUT:-10m} diff --git a/scripts/install-server-image.sh b/scripts/install-server-image.sh index 5a56a1ac8..5acb61b4e 100755 --- a/scripts/install-server-image.sh +++ b/scripts/install-server-image.sh @@ -118,9 +118,14 @@ prepare_environment() { fi CANVAS_BACKEND_IMAGE="${IMAGE_REPOSITORY}-backend:${CANVAS_IMAGE_TAG}" CANVAS_WEB_IMAGE="${IMAGE_REPOSITORY}-web:${CANVAS_IMAGE_TAG}" + CANVAS_YINGCE_AGENT_IMAGE="${IMAGE_REPOSITORY}-yingce-agent:${CANVAS_IMAGE_TAG}" set_env_value .env CANVAS_IMAGE_TAG "$CANVAS_IMAGE_TAG" set_env_value .env CANVAS_BACKEND_IMAGE "$CANVAS_BACKEND_IMAGE" set_env_value .env CANVAS_WEB_IMAGE "$CANVAS_WEB_IMAGE" + set_env_value .env CANVAS_YINGCE_AGENT_IMAGE "$CANVAS_YINGCE_AGENT_IMAGE" + if ! grep -Eq '^YINGCE_AGENT_TOKEN=.{32,}$' .env; then + set_env_value .env YINGCE_AGENT_TOKEN "$(openssl rand -hex 32)" + fi return fi @@ -140,6 +145,8 @@ CANVAS_HTTP_PORT=${CANVAS_HTTP_PORT} CANVAS_IMAGE_TAG=${CANVAS_IMAGE_TAG} CANVAS_BACKEND_IMAGE=${IMAGE_REPOSITORY}-backend:${CANVAS_IMAGE_TAG} CANVAS_WEB_IMAGE=${IMAGE_REPOSITORY}-web:${CANVAS_IMAGE_TAG} +CANVAS_YINGCE_AGENT_IMAGE=${IMAGE_REPOSITORY}-yingce-agent:${CANVAS_IMAGE_TAG} +YINGCE_AGENT_TOKEN=$(openssl rand -hex 32) CANVAS_REGISTRATION_ENABLED=false CANVAS_ALLOW_PRIVATE_UPSTREAMS=false CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS= @@ -185,13 +192,16 @@ start_services() { if ! docker compose --env-file .env -f "$COMPOSE_FILE" pull; then fail "GHCR 镜像拉取失败；如果容器包尚未公开，请通过 GHCR_USERNAME 和 GHCR_TOKEN 登录后重试" fi - local backend_digest web_digest + local backend_digest web_digest agent_digest backend_digest="$(docker image inspect "$CANVAS_BACKEND_IMAGE" --format '{{range .RepoDigests}}{{println .}}{{end}}' | awk -v repository="${IMAGE_REPOSITORY}-backend" '$0 ~ "^" repository "@sha256:" { print; exit }')" web_digest="$(docker image inspect "$CANVAS_WEB_IMAGE" --format '{{range .RepoDigests}}{{println .}}{{end}}' | awk -v repository="${IMAGE_REPOSITORY}-web" '$0 ~ "^" repository "@sha256:" { print; exit }')" + agent_digest="$(docker image inspect "$CANVAS_YINGCE_AGENT_IMAGE" --format '{{range .RepoDigests}}{{println .}}{{end}}' | awk -v repository="${IMAGE_REPOSITORY}-yingce-agent" '$0 ~ "^" repository "@sha256:" { print; exit }')" [[ "$backend_digest" =~ ^${IMAGE_REPOSITORY//\//\/}-backend@sha256:[a-f0-9]{64}$ ]] || fail "后端镜像未返回可验证的仓库 digest" [[ "$web_digest" =~ ^${IMAGE_REPOSITORY//\//\/}-web@sha256:[a-f0-9]{64}$ ]] || fail "Web 镜像未返回可验证的仓库 digest" + [[ "$agent_digest" =~ ^${IMAGE_REPOSITORY//\//\/}-yingce-agent@sha256:[a-f0-9]{64}$ ]] || fail "Agent 镜像未返回可验证的仓库 digest" set_env_value .env CANVAS_BACKEND_IMAGE "$backend_digest" set_env_value .env CANVAS_WEB_IMAGE "$web_digest" + set_env_value .env CANVAS_YINGCE_AGENT_IMAGE "$agent_digest" docker compose --env-file .env -f "$COMPOSE_FILE" up -d --remove-orphans --wait --wait-timeout 600 } diff --git a/yingce-agent/Dockerfile b/yingce-agent/Dockerfile new file mode 100644 index 000000000..1f9584100 --- /dev/null +++ b/yingce-agent/Dockerfile @@ -0,0 +1,19 @@ +# syntax=docker/dockerfile:1.7 + +FROM node:22.19-alpine + +RUN apk add --no-cache ca-certificates wget \ + && addgroup -S app \ + && adduser -S app -G app + +WORKDIR /app +COPY backend/agent-runtime/pi/package.json backend/agent-runtime/pi/package-lock.json /app/agent-runtime/ +RUN npm ci --omit=dev --ignore-scripts --prefix /app/agent-runtime +COPY backend/agent-runtime/pi/agent-runtime.mjs /app/agent-runtime/agent-runtime.mjs +COPY yingce-agent/server.mjs /app/server.mjs + +ENV NODE_ENV=production +USER app +EXPOSE 8081 +HEALTHCHECK --interval=30s --timeout=3s --start-period=10s --retries=3 CMD wget -qO- http://127.0.0.1:8081/healthz >/dev/null || exit 1 +CMD ["node", "server.mjs"] diff --git a/yingce-agent/README.md b/yingce-agent/README.md new file mode 100644 index 000000000..dce36de5a --- /dev/null +++ b/yingce-agent/README.md @@ -0,0 +1,7 @@ +# yingce-agent + +独立的 Agent Node 运行时。它不连接数据库，也不执行计费、审批或画布写入。 + +后端在 `YINGCE_AGENT_URL` 非空时把一轮会话交给本服务，并在自己的网络地址上提供一次性 bridge token。Node 只通过该 bridge 把模型、工具和事件交回 Go。远程调用失败时不会回退到后端内嵌进程，避免同一步执行两次。 + +本地开发不设置 `YINGCE_AGENT_URL`，后端仍在本进程启动 Node。生产 Compose 必须同时提供镜像、`YINGCE_AGENT_TOKEN` 和 `YINGCE_AGENT_BRIDGE_HOST`。 diff --git a/yingce-agent/server.mjs b/yingce-agent/server.mjs new file mode 100644 index 000000000..c64794e34 --- /dev/null +++ b/yingce-agent/server.mjs @@ -0,0 +1,167 @@ +import { spawn } from "node:child_process"; +import { timingSafeEqual } from "node:crypto"; +import { createServer } from "node:http"; +import { dirname } from "node:path"; + +const token = process.env.YINGCE_AGENT_TOKEN || ""; +const maxSessions = positiveInt(process.env.MAX_CONCURRENT_SESSIONS, 8, 1000); +const memoryMB = positiveInt(process.env.NODE_MAX_OLD_SPACE_SIZE, 512, 8192); +const runtimePath = process.env.YINGCE_AGENT_RUNTIME || "/app/agent-runtime/agent-runtime.mjs"; +const port = Number(process.env.PORT || 8081); +const bodyLimit = 32 * 1024 * 1024; + +if (token.length < 32) { + console.error("YINGCE_AGENT_TOKEN must contain at least 32 characters"); + process.exit(1); +} + +let active = 0; +const waiters = []; + +function positiveInt(raw, fallback, max) { + const parsed = Number(raw); + if (!Number.isInteger(parsed) || parsed < 1) return fallback; + return Math.min(parsed, max); +} + +function authorized(request) { + const actual = Buffer.from(request.headers.authorization || ""); + const expected = Buffer.from(`Bearer ${token}`); + return actual.length === expected.length && timingSafeEqual(actual, expected); +} + +function acquire(request) { + if (active < maxSessions) { + active += 1; + return Promise.resolve(); + } + return new Promise((resolve, reject) => { + const waiter = { resolve, reject }; + waiters.push(waiter); + const abort = () => { + const index = waiters.indexOf(waiter); + if (index >= 0) waiters.splice(index, 1); + reject(new Error("Agent request cancelled while queued")); + }; + if (request.aborted || request.destroyed) abort(); + else request.once("aborted", abort); + }); +} + +function release() { + const waiter = waiters.shift(); + if (waiter) { + waiter.resolve(); + return; + } + active = Math.max(0, active - 1); +} + +function readBody(request) { + return new Promise((resolve, reject) => { + const chunks = []; + let size = 0; + request.on("data", (chunk) => { + size += chunk.length; + if (size > bodyLimit) { + reject(new Error("request body too large")); + request.destroy(); + return; + } + chunks.push(chunk); + }); + request.on("end", () => resolve(Buffer.concat(chunks))); + request.on("error", reject); + }); +} + +function writeRuntimeError(response, message) { + if (!response.headersSent) { + response.writeHead(200, { "content-type": "application/x-ndjson" }); + } + response.end(`${JSON.stringify({ event: "runtime_error", message })}\n`); +} + +const server = createServer(async (request, response) => { + if (request.method === "GET" && request.url === "/healthz") { + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ ok: true, active, limit: maxSessions })); + return; + } + if (request.method !== "POST" || request.url !== "/v1/runs") { + response.writeHead(404, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: "not found" })); + return; + } + if (!authorized(request)) { + response.writeHead(401, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: "unauthorized" })); + return; + } + let body; + try { + body = await readBody(request); + JSON.parse(body.toString("utf8")); + } catch (error) { + response.writeHead(400, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: error instanceof Error ? error.message : "invalid request" })); + return; + } + try { + await acquire(request); + } catch (error) { + response.writeHead(499, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: error instanceof Error ? error.message : "cancelled" })); + return; + } + + const child = spawn(process.execPath, [`--max-old-space-size=${memoryMB}`, runtimePath], { + cwd: dirname(runtimePath), + env: { + PATH: process.env.PATH || "", + NODE_ENV: "production", + PI_OFFLINE: "1", + }, + stdio: ["pipe", "pipe", "pipe"], + }); + let released = false; + const finish = () => { + if (released) return; + released = true; + release(); + }; + let sawRuntimeError = false; + const stop = () => { + if (child.exitCode === null && child.signalCode === null) child.kill("SIGTERM"); + }; + request.on("aborted", stop); + response.writeHead(200, { "content-type": "application/x-ndjson" }); + child.stdout.on("data", (chunk) => { + const text = chunk.toString("utf8"); + if (text.includes('"runtime_error"') || text.includes('"bridge_error"')) sawRuntimeError = true; + response.write(chunk); + }); + child.stderr.on("data", (chunk) => process.stderr.write(chunk)); + child.on("error", (error) => { + writeRuntimeError(response, error.message); + finish(); + }); + child.on("close", (code) => { + if (!response.writableEnded) { + if (code && !sawRuntimeError) { + response.write(`${JSON.stringify({ event: "runtime_error", message: `Agent runtime exited with code ${code}` })}\n`); + } + response.end(); + } + finish(); + }); + child.stdin.end(body); +}); + +server.requestTimeout = 0; +server.headersTimeout = 0; +server.timeout = 0; +server.listen(port, "0.0.0.0", () => { + const address = server.address(); + console.log(`YINGCE_AGENT_READY ${address.port}`); +});