fix(squad): propagate AgentTimeout to pi-agent step execution

E2E v2 still hit 30min runner timeout because runOneStep ignored ExecuteOptions.Timeout; inject shared KB config in pi-agent E2E test.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-13 02:50:23 +00:00
committed by weichen
co-authored by Cursor
parent f4619757e2
commit 033ef70ac9
4 changed files with 74 additions and 39 deletions
@@ -41,7 +41,26 @@ Phase 4 完成后,交付验收需满足两项缺口:
## 五、 实际测试结果与判定 (Actual Results & Evaluation)
### 用例 1:pi-agent 全链路 E2E(第二次运行,含 auth 修复)
### 用例 1a:pi-agent 全链路 E2E v2(60m AgentTimeout 配置,2026-09-12 09:50 UTC)
- **Input**: `go test -count=1 -timeout 2h -v -run TestPiAgent_FullLifecycle ./internal/server/...`
- **Actual Output**(节选):
```
[kb] 无法连接到 PostgreSQL 知识库 (postgres://:@:0/...), 将降级运行在离线模式
✓ [0. 环境校验] Health检查正常: Env=dev, AgentBackend=pi-agent, SharedKBPort=0
⏱ [步骤 1 对话完成] 累计耗时 3m0s
ExecuteStage V0.1 failed: step step_01_organize failed: Execution timed out after 1800000ms
--- FAIL: TestPiAgent_FullLifecycle_MultiRole_RealWorld_E2E (1980.29s)
```
- **Verdict**: **FAIL**
- **Duration**: 1980s(约 33min)
- **根因**:
1. E2E 未注入 `KB_DB_*`,知识库离线降级;
2. `squad/workflow.go:runOneStep` 硬编码 `Timeout: 30*time.Minute`,未传递 `ExecuteOptions.Timeout`,导致 pi-agent runner 仍收到 1800000ms(30min)而非 60min。
**已修复(待 v3 复验)**:`mergePiAgentE2EKbConfig` 注入共享 KB;`runOneStep` 透传 `stepTimeout`。
### 用例 1b:pi-agent 全链路 E2E(历史 PASS,含 auth 修复)
- **Input**: `go test ./internal/server/ -run TestPiAgent_FullLifecycle -timeout 1h -v`
- **Actual Output**(节选):
@@ -106,43 +125,11 @@ ok chat-agent-system/internal/bizdev 0.003s TestEnsureTaskStageAccess_De
## 七、 测试结论与后续优化建议 (Conclusion & Next Steps)
**结论**:两项交付目标均已达成,可合并入 dev 阶段节点。
**结论**:bizdev ABAC、Blended 上下文过滤与 PG 治理链已验收通过;pi-agent E2E v2 因 30min 硬编码超时 **FAIL**,修复已落地,待 v3 复验。
| 交付项 | 状态 |
|--------|------|
| pi-agent 1h E2E 全链路 | ✅ PASS(676s) |
| bizdev Agent 上下文 ABAC | ✅ 已实现并注入 server |
| 多角色 docgov Bearer 鉴权 | ✅ E2E 验证通过 |
### 用例 3:BlendedSquadContext 阶段 ABAC 过滤
- **Input**: `go test ./internal/bizdev/... -run TestBuildBlendedSquadContextForUser_FiltersByStage -v`
- **Actual Output**:
```
--- PASS: TestBuildBlendedSquadContextForUser_FiltersByStage (0.00s)
PASS
```
- **Verdict**: **PASS**(里程碑决策按 `stage_key` 过滤,仅保留当前允许阶段)
### 用例 4:Phase 5 PG 治理全链路(activate → approve → advance)
- **Input**: `go test ./internal/server/ -run TestWorkflowGovernance_PG -v`
- **Actual Output**:
```
--- PASS: TestWorkflowGovernance_PG_ActivateApproveAdvanceChain (0.36s)
PASS
ok chat-agent-system/internal/server 0.367s
```
- **Verdict**: **PASS**(含 stage_ready 通知、访客 403 ABAC、创建者阶段推进)
- **Duration**: ~0.36s
## 七、 测试结论与后续优化建议 (Conclusion & Next Steps)
**结论**:pi-agent E2E、bizdev ABAC、Blended 上下文过滤与 PG 治理链均已验收通过。
| 交付项 | 状态 |
|--------|------|
| pi-agent 1h E2E 全链路 | ✅ PASS(676s) |
| pi-agent 1h E2E 全链路 | ⚠️ v2 FAIL(1980s);历史 PASS(676s);v3 待跑 |
| bizdev Agent 上下文 ABAC | ✅ |
| BuildBlendedSquadContext ABAC | ✅ |
| PG activate→approve→advance | ✅ |
+44
View File
@@ -16,10 +16,44 @@ import (
"chat-agent-system/internal/config"
"chat-agent-system/internal/ctxmgr"
"chat-agent-system/internal/docgov"
"chat-agent-system/internal/kb"
"chat-agent-system/internal/server"
"chat-agent-system/internal/squad"
)
// mergePiAgentE2EKbConfig 从 .env / 环境变量注入共享知识库与向量检索配置(E2E 必须走真实 KB,不可 stub)。
func mergePiAgentE2EKbConfig(cfg *config.Config) {
envCfg, err := config.Load("")
if err == nil && envCfg != nil {
cfg.KbDbHost = envCfg.KbDbHost
cfg.KbDbPort = envCfg.KbDbPort
cfg.KbDbUser = envCfg.KbDbUser
cfg.KbDbPassword = envCfg.KbDbPassword
cfg.KbDbName = envCfg.KbDbName
cfg.EmbeddingEndpointURL = envCfg.EmbeddingEndpointURL
cfg.EmbeddingAPIKey = envCfg.EmbeddingAPIKey
cfg.EmbeddingModel = envCfg.EmbeddingModel
cfg.RerankerEndpointURL = envCfg.RerankerEndpointURL
cfg.RerankerAPIKey = envCfg.RerankerAPIKey
cfg.RerankerModel = envCfg.RerankerModel
}
if cfg.KbDbHost == "" {
cfg.KbDbHost = "127.0.0.1"
}
if cfg.KbDbPort == 0 {
cfg.KbDbPort = 5432
}
if cfg.KbDbUser == "" {
cfg.KbDbUser = "regkb"
}
if cfg.KbDbPassword == "" {
cfg.KbDbPassword = "shared_kb_secret_pass"
}
if cfg.KbDbName == "" {
cfg.KbDbName = "regulations"
}
}
// TestPiAgent_FullLifecycle_MultiRole_RealWorld_E2E
// 贯穿:新建项目 -> 角色权限配置 -> 分角色材料补充与真实对话 -> 自由非标任务执行 -> 工作流推进 -> 审核驳回/再批 -> 数据库落盘与全量审计追溯
//
@@ -50,6 +84,7 @@ func TestPiAgent_FullLifecycle_MultiRole_RealWorld_E2E(t *testing.T) {
DataDir: t.TempDir(),
AllowInMemory: true,
}
mergePiAgentE2EKbConfig(cfg)
svc := bizdev.MustNewService(t, cfg)
srv := server.NewServer(cfg, svc)
@@ -76,6 +111,15 @@ func TestPiAgent_FullLifecycle_MultiRole_RealWorld_E2E(t *testing.T) {
respHealth.Body.Close()
t.Logf("✓ [0. 环境校验] Health检查正常: Env=%s, AgentBackend=%s, SharedKBPort=%d",
healthData.Env, healthData.AgentBackend, healthData.SharedKBPort)
kbClient := kb.NewClient(cfg)
defer kbClient.Close()
if stats, err := kbClient.Stats(ctx); err != nil {
t.Logf("⚠ [0. 知识库] 未连接共享 KB(%s:%d/%s): %v", cfg.KbDbHost, cfg.KbDbPort, cfg.KbDbName, err)
} else if reg, ok := stats[kb.DomainRegulatory]; ok {
t.Logf("✓ [0. 知识库] 已连接共享 KB: documents=%d chunks=%d", reg.DocumentCount, reg.ChunkCount)
} else {
t.Logf("✓ [0. 知识库] 已连接共享 KB")
}
// 3. 项目基本信息与三级角色定义
projectID := int64(time.Now().UnixNano()%900000 + 100000)
+1 -1
View File
@@ -312,7 +312,7 @@ func (q *QualityGate) EnsureQualityOrRepair(
_ = os.WriteFile(repairTaskPath, []byte(b.String()), 0o644)
// 执行修复步骤(治理入账:轨迹 + 预算)
runErr := driver.runOneStep(ctx, wsDir, repairStep, backend, opts.StepNotify, nil, &[]string{}, ledger)
runErr := driver.runOneStep(ctx, wsDir, repairStep, backend, opts.Timeout, opts.StepNotify, nil, &[]string{}, ledger)
if runErr != nil {
break
}
+7 -3
View File
@@ -263,7 +263,7 @@ func (d *WorkflowDriver) Execute(ctx context.Context, wsDir string, backend agen
wg.Add(1)
go func() {
defer wg.Done()
if err := d.runOneStep(runCtx, wsDir, sCopy, backend, opts.StepNotify, &mu, &producedFiles, ledger); err != nil {
if err := d.runOneStep(runCtx, wsDir, sCopy, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil {
mu.Lock()
if groupErr == nil {
groupErr = err
@@ -325,7 +325,7 @@ func (d *WorkflowDriver) Execute(ctx context.Context, wsDir string, backend agen
}
// 治理:受控迁移至执行态(legacy 重跑场景直接放行)
markStepExecuting(wf, step.ID)
if err := d.runOneStep(runCtx, wsDir, step, backend, opts.StepNotify, &mu, &producedFiles, ledger); err != nil {
if err := d.runOneStep(runCtx, wsDir, step, backend, timeout, opts.StepNotify, &mu, &producedFiles, ledger); err != nil {
wf.Status = "failed"
_ = d.SaveWorkflow(wsDir, wf)
_ = d.updateWorkspaceStatus(wsDir, "failed")
@@ -403,6 +403,7 @@ func (d *WorkflowDriver) runOneStep(
wsDir string,
step StepSpec,
backend agent.Backend,
stepTimeout time.Duration,
notify func(StepEvent),
mu *sync.Mutex,
producedFiles *[]string,
@@ -491,12 +492,15 @@ func (d *WorkflowDriver) runOneStep(
prevContent = string(b)
}
if stepTimeout <= 0 {
stepTimeout = 30 * time.Minute
}
req := agent.Request{
WorkDir: subDir,
Prompt: string(taskData),
Instructions: string(agentsData),
Targets: targets,
Timeout: 30 * time.Minute,
Timeout: stepTimeout,
}
res, err := backend.Execute(ctx, req)