fix(agent): 拒绝把工作区发给数据目录不一致的 PI runner

test 启动时根 .env 的 dev runner URL 盖过了 18866,sidecar 把 test 工作区重写到 dev 空目录,模型因读不到材料报 MISSING_TARGET。端口约定优先,并在调用前核对 runner data_dir。
This commit is contained in:
weichen
2026-09-28 03:16:27 +00:00
parent 8f77053f51
commit 4f581306d7
8 changed files with 180 additions and 1 deletions
+83
View File
@@ -368,3 +368,86 @@ func TestPiAgentRemoteRunner_BodyTooLargeDoesNotWrite(t *testing.T) {
t.Fatal("must not materialize after body-too-large reject")
}
}
func TestPiAgentRemoteRunner_DataRootMismatchSkipsRun(t *testing.T) {
dir := t.TempDir()
var runHits int
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/health":
_ = json.NewEncoder(w).Encode(map[string]any{"data_dir": filepath.Join(dir, "other-root")})
case "/v1/run":
runHits++
http.Error(w, "should not be called", http.StatusInternalServerError)
default:
http.NotFound(w, r)
}
}))
defer ts.Close()
res, err := agent.NewBackend("pi-agent", &config.Config{
AgentBackend: "pi-agent",
PiAgentRunnerURL: ts.URL,
PiAgentModel: "MiniMax-M3",
PiAgentProvider: "minimax",
AgentTimeout: 5 * time.Second,
}).Execute(context.Background(), agent.Request{
WorkDir: dir, Prompt: "draft", Targets: []string{"output.md"},
})
if err != nil {
t.Fatal(err)
}
if res.Success || res.ErrorReason != "pi runner data root mismatch" {
t.Fatalf("got success=%v reason=%q", res.Success, res.ErrorReason)
}
if strings.Contains(res.ErrorReason, dir) {
t.Fatalf("reason leaked a path: %s", res.ErrorReason)
}
if runHits != 0 {
t.Fatalf("/v1/run hits=%d", runHits)
}
}
func TestPiAgentRemoteRunner_DataRootMatchStillRuns(t *testing.T) {
dir := t.TempDir()
var runHits int
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/health":
_ = json.NewEncoder(w).Encode(map[string]any{"data_dir": dir})
case "/v1/run":
runHits++
body := "ok"
runID := filepath.Base(dir)
rv := agent.BuildResultV1(runID, "success", false, "", []struct {
RelPath string
Content []byte
}{{"output.md", []byte(body)}})
_ = json.NewEncoder(w).Encode(map[string]any{
"ok": true, "run_id": runID, "runner_protocol": "typed-v1",
"patches": []map[string]string{{"rel_path": "output.md", "content": body}},
"result_v1": rv,
})
default:
http.NotFound(w, r)
}
}))
defer ts.Close()
res, err := agent.NewBackend("pi-agent", &config.Config{
AgentBackend: "pi-agent",
PiAgentRunnerURL: ts.URL,
PiAgentModel: "MiniMax-M3",
PiAgentProvider: "minimax",
AgentTimeout: 5 * time.Second,
}).Execute(context.Background(), agent.Request{
WorkDir: dir, Prompt: "draft", Targets: []string{"output.md"},
})
if err != nil {
t.Fatal(err)
}
if !res.Success {
t.Fatalf("matched data_dir must run: %s", res.ErrorReason)
}
if runHits != 1 {
t.Fatalf("/v1/run hits=%d", runHits)
}
}
+52
View File
@@ -11,6 +11,7 @@ import (
"os/exec"
"path/filepath"
"strings"
"sync"
"time"
"chat-agent-system/internal/config"
@@ -155,7 +156,58 @@ func (p *PiAgentBackend) runnerPayload(req Request, workDir string, remote bool,
return body
}
// runnerDataRoots caches a successful /health data_dir per runner URL.
// A failed probe is not cached: the next call may try again, and this call skips the check.
var runnerDataRoots sync.Map
func probeRunnerDataRoot(ctx context.Context, runnerURL string) (string, bool) {
if v, ok := runnerDataRoots.Load(runnerURL); ok {
root, _ := v.(string)
return root, root != ""
}
probeCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
defer cancel()
httpReq, err := http.NewRequestWithContext(probeCtx, http.MethodGet, runnerURL+"/health", nil)
if err != nil {
return "", false
}
client := &http.Client{Timeout: 2 * time.Second}
resp, err := client.Do(httpReq)
if err != nil {
return "", false
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return "", false
}
var body struct {
DataDir string `json:"data_dir"`
}
if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&body); err != nil {
return "", false
}
root := filepath.Clean(strings.TrimSpace(body.DataDir))
if root == "" || root == "." {
return "", false
}
runnerDataRoots.Store(runnerURL, root)
return root, true
}
func workDirWithinRunnerRoot(workDir, root string) bool {
rel, err := filepath.Rel(filepath.Clean(root), filepath.Clean(workDir))
if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
return false
}
return true
}
func (p *PiAgentBackend) executeRemote(ctx context.Context, runnerURL string, req Request, workDir string, timeout time.Duration) (*Result, error) {
if root, ok := probeRunnerDataRoot(ctx, runnerURL); ok && !workDirWithinRunnerRoot(workDir, root) {
// Do not include either path: the caller classifies this phrase, and paths leak host layout.
return &Result{Success: false, ErrorReason: "pi runner data root mismatch"}, nil
}
expectedRunID := filepath.Base(workDir)
if expectedRunID == "" || expectedRunID == "." || expectedRunID == string(filepath.Separator) {
expectedRunID = "unknown"
+1
View File
@@ -48,6 +48,7 @@ func init() {
{ID: "CHAT_TIMEOUT", Status: 504, Message: "生成回复超时,请稍后重试", Module: "stream", Class: ClassTransient, Actions: []string{"retry"}},
{ID: "NETWORK_ERROR", Status: 502, Message: "网络连接异常,请稍后重试", Module: "stream", Class: ClassTransient, Actions: []string{"retry"}},
{ID: "RUNNER_RESULT_REJECTED", Status: 502, Message: "助手执行结果校验失败,请重新开始任务", Module: "stream", Actions: []string{"restart"}},
{ID: "RUNNER_MISMATCH", Status: 500, Message: "模型执行环境与数据目录不一致,请联系管理员检查部署配置", Module: "stream"},
{ID: "TASK_INTERRUPTED", Status: 409, Message: "任务执行中断,可继续或重新开始", Module: "taskrun", Class: ClassResumable, Actions: []string{"continue", "restart"}},
{ID: "TASK_SCOPE_UNRESOLVED", Status: 400, Message: "无法确定任务所属阶段", Module: "projectinstance"},
+6
View File
@@ -71,6 +71,12 @@ func ClassifyError(err error, phaseID, requestID, sessionID, workspaceID string)
info.UserMessage = "网络连接异常,请稍后重试"
info.Actions = []string{"retry"}
case strings.Contains(msg, "runner data root mismatch"):
info.Code = "RUNNER_MISMATCH"
info.Class = ErrorClassFatal
info.UserMessage = "模型执行环境与数据目录不一致,请联系管理员检查部署配置"
info.Actions = []string{}
case isRunnerContractError(msg):
// Permanent runner wire/contract failures must not look "resumable".
info.Code = "RUNNER_RESULT_REJECTED"
+25
View File
@@ -145,6 +145,31 @@ func TestClassifyError_TypedResultRejectedIsFatalNotResumable(t *testing.T) {
}
}
func TestClassifyError_RunnerDataRootMismatch(t *testing.T) {
info := ClassifyError(
errors.New("task bd_research failed: pi runner data root mismatch"),
"dialogue_generating",
"req_mismatch",
"sess-p5-u32",
"ws_should_not_make_this_resumable",
)
if info.Code != "RUNNER_MISMATCH" {
t.Fatalf("code=%s", info.Code)
}
if info.Class != ErrorClassFatal {
t.Fatalf("class=%s", info.Class)
}
if len(info.Actions) != 0 {
t.Fatalf("actions=%v", info.Actions)
}
if strings.Contains(info.UserMessage, "/") || strings.Contains(info.Upstream, "/home/") {
t.Fatalf("leaked path: %+v", info)
}
if info.UserMessage == "" || info.Message != info.UserMessage {
t.Fatalf("message=%q user=%q", info.Message, info.UserMessage)
}
}
func TestClassifyError_BodyTooLargeIsFatalNotResumable(t *testing.T) {
info := ClassifyError(
errors.New("step step_01_organize failed: runner result body too large"),