Compare commits

...
Author SHA1 Message Date
weichen 0f1daa8b89 Align the SQLite run store test with honest step failures
Why: the U3B semantics change removed the failed-with-patches
promotion; TestEngineRunsOnSQLiteStore still encoded the old behavior
and failed under the fixed engine.
Evidence: repair receipt success; sqlite store tests pass
(verify-u3br2.json).
Impact: test-only change.
Refs: feature/agent-stack-audit-remediation-b1 U3B repair
2026-10-21 18:05:51 +00:00
weichen cdd51c8678 Pin nested agent-intent and agent-native to the batch-1 fixes
Why: composition consistency - Dev pins agent-intent at 765a7e7 and
agent-native at the U3A result-fidelity tip; the squad container's
nested copies must match (check-pins).
Evidence: pin only; both libraries verified per-unit.
Impact: nested gitlink pointer changes.
Refs: feature/agent-stack-audit-remediation-b1
2026-10-21 17:45:01 +00:00
weichen 23f161bfb3 Honor executor terminals in retries and interrupted recovery
Why: F05 - files-presence could promote failures to success, recovery
marked crashed steps completed on files alone or reran them blindly,
and retries minted a fresh ExecutionID each attempt.
Evidence: squad tests pass incl. executor result contract and recovery
tables (verify-u3b.json); (nil,nil), canceled and side-effect-possible
semantics handled; interrupted steps without a trusted terminal now
hold for reconciliation instead of auto-advancing.
Impact: retries reuse the step ExecutionID and shared deadline; no
auto-replay of unverified external actions.
Refs: feature/agent-stack-audit-remediation-b1 U3B
2026-10-21 17:20:40 +00:00
weichen 4e6557782a Temporarily disable the cursor profile
Why: U3B hit two consecutive pre-session cursor infrastructure
failures (TLS destruction, then model auto rejection consistent with
the documented monthly quota exhaustion); both runs fail-stopped
before any session started with zero workspace changes.
Evidence: run dirs /tmp/agent-exec-1000/287d14d4… and …/b7ed6a71…
preserved; raw logs 553B and 48B.
Impact: routing only; candidates become antigravity then pi then codex.
Refs: Agent-System-Dev feature/agent-stack-audit-remediation-b1
2026-10-21 16:44:24 +00:00
weichen aa15a5d164 Version resolver configs and invalidate executor caches
Why: F06 - the wrapped-executor cache keyed only backend/role/model, so
allow->deny permission tightening and factory replacement never took
effect on subsequent resolves.
Evidence: squad and adapters/native tests pass (verify-u4.json); config
identity helper covers permission/tool/limiter changes.
Impact: stale executors can no longer widen capabilities; in-flight
tasks are not rolled back (documented).
Refs: feature/agent-stack-audit-remediation-b1 U4
2026-10-21 15:51:53 +00:00
weichen c7121bda2a Merge pull request 'Point the nested agent-native pin at the O2 route contract head' (#55) from feature/fast-slow-o2-i3-agent-native-pin into main
ci / docker-checkout-hostgateway (push) Successful in 0s
governance / governance (push) Successful in 19s
ci / vet-and-test (push) Successful in 40s
2026-10-20 08:57:38 +00:00
agent-exec-transport ad58d0ab1a agent-exec transport 1695a49c50742188364de30abf001030/559ef8a7bc20
ci / docker-checkout-hostgateway (pull_request) Successful in 1s
governance / governance (pull_request) Successful in 19s
ci / vet-and-test (pull_request) Successful in 33s
2026-10-20 08:55:47 +00:00
weichen c0ce10a82c Merge pull request 'Align Agent-Squad with U3 stream guard' (#53) from feature/fast-slow-u3-r04-agent-native-pin into main
ci / docker-checkout-hostgateway (push) Successful in 1s
governance / governance (push) Successful in 15s
ci / vet-and-test (push) Successful in 36s
2026-10-19 09:20:18 +00:00
19 changed files with 1869 additions and 96 deletions
+8 -2
View File
@@ -36,13 +36,19 @@ profiles:
adapter: cursor
executable: "cursor-agent-stable"
model: auto
enabled: true
# 2026-10-21 临时禁用:U3B 连续两次 fail-stop(首次 TLS 会话前故障 553B 日志、
# 二次 "Cannot use this model: auto",与 Backend 配置已记录的月配额耗尽一致),
# 两次均未建立会话、零改动(现场保留)。恢复时改回 enabled: true。
enabled: false
antigravity:
adapter: antigravity
executable: agy
model: auto
enabled: true
# 2026-10-21 临时禁用:U3B 尝试被 400 FAILED_PRECONDITION "User location is
# not supported for the API use" 拒绝,0 token、未建立会话(现场保留);
# 与 Backend 配置已记录的 Gemini 端点不稳一致。恢复时改回 enabled: true。
enabled: false
pi:
adapter: pi
+105 -3
View File
@@ -3,7 +3,11 @@ package native
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"reflect"
"strings"
"time"
@@ -52,12 +56,58 @@ type Config struct {
PermissionAsk func(ctx context.Context, call llm.ToolCall) runtime.AskVerdict
}
// Identity 返回塑造 Executor 的配置摘要(模型/客户端/工具/Exec 等)。
// RegisterFull 经 BackendResolver.Register 替换工厂时会失效缓存;角色侧
// BashPermission 等由 squad.ConfigIdentity 纳入 Resolve 缓存键,故 allow→deny
// 在下一次 Resolve 即生效,无需重启进程。
func (c Config) Identity() string {
var b strings.Builder
b.WriteString("model=")
b.WriteString(c.DefaultModel)
b.WriteString("\x00turns=")
b.WriteString(fmt.Sprintf("%d", c.MaxTurns))
b.WriteString("\x00timeout=")
b.WriteString(c.Timeout.String())
if c.Client != nil {
b.WriteString("\x00base=")
b.WriteString(c.Client.BaseURL)
b.WriteString("\x00key=")
b.WriteString(c.Client.APIKey)
}
if c.Exec != nil {
b.WriteString("\x00exec.bash=")
b.WriteString(c.Exec.Bash)
b.WriteString("\x00exec.python=")
b.WriteString(c.Exec.Python)
}
if c.Search != nil {
b.WriteString("\x00search=1")
}
if c.Fetch != nil {
b.WriteString("\x00fetch=1")
}
if c.MCP != nil {
b.WriteString("\x00mcp=1")
}
if len(c.SkillsRoots) > 0 {
b.WriteString("\x00skills=")
b.WriteString(strings.Join(c.SkillsRoots, ","))
}
if c.CodeOrchestration {
b.WriteString("\x00code=1")
}
sum := sha256.Sum256([]byte(b.String()))
return hex.EncodeToString(sum[:16])
}
// Register 把 "agent-native" 工厂注册进解析器。
func Register(r *squad.BackendResolver, client *llm.Client, defaultModel string, maxTurns int, timeout time.Duration) {
RegisterFull(r, Config{Client: client, DefaultModel: defaultModel, MaxTurns: maxTurns, Timeout: timeout})
}
// RegisterFull 注册带工具装配的 "agent-native" 工厂。
// 再次调用会替换工厂并失效该后端缓存(见 BackendResolver.Register);
// 配置摘要见 Config.Identity,角色侧见 squad.ConfigIdentity。
func RegisterFull(r *squad.BackendResolver, cfg Config) {
r.Register("agent-native", func(role squad.RoleSpec) (squad.Executor, error) {
opts, err := buildOptions(cfg, role)
@@ -147,10 +197,62 @@ func toNative(req squad.ExecRequest) exec.Request {
}
func fromNative(res *exec.Result) *squad.ExecResult {
return &squad.ExecResult{
Success: res.Success, Patches: res.Patches, Logs: res.Logs, ErrorReason: res.ErrorReason,
Retryable: res.Retryable, PromptTokens: res.PromptTokens, CompletionTokens: res.CompletionTokens,
if res == nil {
return nil
}
out := &squad.ExecResult{
Success: res.Success,
Patches: res.Patches,
Logs: res.Logs,
ErrorReason: res.ErrorReason,
Retryable: res.Retryable,
PromptTokens: res.PromptTokens,
CompletionTokens: res.CompletionTokens,
}
// F05/U3b(Squad 侧 contract):
// - native exec.Result 在后续 release 会新增 SideEffects、Canceled、Aborted 字段;
// 本地 deps gitlink 若尚未跟进,从 JSON 探测拿不到这些字段时回落到
// 「legacy reading: side effects = possible」,与既有调用方旧行为一致。
// - 在不影响 ExecResult 默认值的前提下,用 JSON 探测而非硬性修改 deps。
probe, err := probeNativeResultExtras(res)
if err == nil {
out.SideEffects = probe.SideEffects
out.Canceled = probe.Canceled
out.Aborted = probe.Aborted
}
return out
}
// nativeResultExtras 探测 native exec.Result 上的可选字段(结构化 JSON)。
// 缺省(字段不存在或为零值时)按 legacy reading:SideEffects 空串视为 "possible";
// 调用方侧 ExecResult.IsRetryableSideEffectReports() 已经处理该回退。
type nativeResultExtras struct {
SideEffects squad.ExecSideEffects `json:"side_effects,omitempty"`
Canceled bool `json:"canceled,omitempty"`
Aborted bool `json:"aborted,omitempty"`
}
// probeNativeResultExtras 通过 JSON 探测 native Result 上的可选字段。
// 探测出 cancel/abort/side_effects 任何一个就使用它们;探测失败或全部为零值视为
// legacy reading(SideEffects 空 = possible 兜底)。
func probeNativeResultExtras(res *exec.Result) (nativeResultExtras, error) {
if res == nil {
return nativeResultExtras{}, fmt.Errorf("nil result")
}
v := reflect.ValueOf(res).Elem()
if !v.IsValid() {
return nativeResultExtras{}, fmt.Errorf("invalid result")
}
// 用 JSON 做结构化探测:这一路兼容 deps 升级(增字段)、不增字段两种情形。
data, err := json.Marshal(v.Interface())
if err != nil {
return nativeResultExtras{}, err
}
var out nativeResultExtras
if err := json.Unmarshal(data, &out); err != nil {
return nativeResultExtras{}, err
}
return out, nil
}
type eventTap struct {
+82
View File
@@ -47,3 +47,85 @@ func TestBuildOptionsBashPermissionGating(t *testing.T) {
t.Fatal("missing model must fail-closed")
}
}
func TestConfigIdentityReflectsToolAndClient(t *testing.T) {
base := Config{
Client: &llm.Client{BaseURL: "http://x", APIKey: "k"},
DefaultModel: "m1", MaxTurns: 5, Timeout: time.Minute,
Exec: &tools.ExecConfig{Bash: "/bin/bash"},
}
otherExec := base
otherExec.Exec = &tools.ExecConfig{Bash: "/bin/zsh"}
if base.Identity() == otherExec.Identity() {
t.Fatal("Exec tool-config change must change Config.Identity")
}
otherClient := base
otherClient.Client = &llm.Client{BaseURL: "http://y", APIKey: "k"}
if base.Identity() == otherClient.Identity() {
t.Fatal("client change must change Config.Identity")
}
same := base
if base.Identity() != same.Identity() {
t.Fatal("identical config must keep Identity stable")
}
}
func TestRegisterFullAllowDenyTakesEffectOnNextResolve(t *testing.T) {
r := squad.NewBackendResolver()
cfg := Config{
Client: &llm.Client{BaseURL: "http://x", APIKey: "k"},
DefaultModel: "m1", MaxTurns: 1, Timeout: time.Second,
Exec: &tools.ExecConfig{Bash: "/bin/bash"},
}
RegisterFull(r, cfg)
allowRole := squad.RoleSpec{Role: "eng", Backend: "agent-native", Model: "m1", BashPermission: squad.BashPermissionAllow}
denyRole := squad.RoleSpec{Role: "eng", Backend: "agent-native", Model: "m1", BashPermission: squad.BashPermissionDeny}
eAllow, _, err := r.Resolve(allowRole, "")
if err != nil {
t.Fatal(err)
}
eDeny, _, err := r.Resolve(denyRole, "")
if err != nil {
t.Fatal(err)
}
if eAllow == eDeny {
t.Fatal("allow→deny must not reuse cached executor on next Resolve")
}
// Identity of role configs must diverge so cache keys differ.
if squad.ConfigIdentity(allowRole) == squad.ConfigIdentity(denyRole) {
t.Fatal("role ConfigIdentity must include BashPermission")
}
}
func TestRegisterFullToolConfigReplacementTakesEffect(t *testing.T) {
r := squad.NewBackendResolver()
cfg1 := Config{
Client: &llm.Client{BaseURL: "http://x", APIKey: "k"},
DefaultModel: "m1", MaxTurns: 1, Timeout: time.Second,
Exec: &tools.ExecConfig{Bash: "/bin/bash"},
}
cfg2 := Config{
Client: &llm.Client{BaseURL: "http://x", APIKey: "k"},
DefaultModel: "m1", MaxTurns: 1, Timeout: time.Second,
Exec: &tools.ExecConfig{Bash: "/bin/zsh"},
}
if cfg1.Identity() == cfg2.Identity() {
t.Fatal("precondition: tool configs must differ in Identity")
}
RegisterFull(r, cfg1)
role := squad.RoleSpec{Role: "eng", Backend: "agent-native", Model: "m1", BashPermission: squad.BashPermissionAllow}
e1, _, err := r.Resolve(role, "")
if err != nil {
t.Fatal(err)
}
RegisterFull(r, cfg2)
e2, _, err := r.Resolve(role, "")
if err != nil {
t.Fatal(err)
}
if e1 == e2 {
t.Fatal("RegisterFull tool-config replacement must invalidate cache")
}
}
+76 -1
View File
@@ -4,6 +4,7 @@ import (
"context"
"os"
"path/filepath"
"strings"
"testing"
"time"
@@ -127,7 +128,7 @@ func TestEngineRunsOnSQLiteStore(t *testing.T) {
}
}
backends := squad.NewBackendResolver()
backends.RegisterInstance("script", stubExecutor{result: &squad.ExecResult{Success: false, Logs: "ok", Patches: map[string][]byte{"output.md": []byte(goodReportLite), "notes.md": []byte(goodReportLite), "fields.json": []byte("{}")}}})
backends.RegisterInstance("script", stubExecutor{result: &squad.ExecResult{Success: true, Logs: "ok", Patches: map[string][]byte{"output.md": []byte(goodReportLite), "notes.md": []byte(goodReportLite), "fields.json": []byte("{}")}}})
blueprints := squad.NewBlueprintRegistry()
if err := blueprints.Register(squad.Blueprint{
Key: "demo", Title: "入库演示", Match: []string{"入库演示"},
@@ -185,4 +186,78 @@ func TestEngineRunsOnSQLiteStore(t *testing.T) {
}
}
// TestEngineFailedResultWithCompletePatchesStaysFailedOnSQLiteStore:F05/U3B 的
// 不变式在 SQLite-store 集成层同样成立——executor 报失败就是失败,patches
// (交付物)齐全也不得把运行翻成 completed:文件只是诊断信息,不是权威结果。
func TestEngineFailedResultWithCompletePatchesStaysFailedOnSQLiteStore(t *testing.T) {
runStore, err := OpenSQLiteStore(filepath.Join(t.TempDir(), "runs.db"))
if err != nil {
t.Fatal(err)
}
defer runStore.Close()
dir := t.TempDir()
reg := squad.NewRoleRegistry()
if err := reg.Register(squad.RoleSpec{Role: "researcher", Backend: "script"}); err != nil {
t.Fatal(err)
}
backends := squad.NewBackendResolver()
// 失败结果但携带全套交付物:旧语义会把它提升为成功,U3B 起必须保持失败。
backends.RegisterInstance("script", stubExecutor{result: &squad.ExecResult{
Success: false, Logs: "stub-refused",
Patches: map[string][]byte{
"output.md": []byte(goodReportLite),
"notes.md": []byte(goodReportLite),
"fields.json": []byte("{}"),
},
}})
blueprints := squad.NewBlueprintRegistry()
if err := blueprints.Register(squad.Blueprint{
Key: "faildemo", Title: "失败演示", Match: []string{"失败演示"},
Steps: []squad.StepSpec{{ID: "s1", Role: "researcher", Title: "调研", Targets: []string{"output.md"}}},
}); err != nil {
t.Fatal(err)
}
eng, err := squad.NewEngine(squad.Options{
DataDir: dir, Registry: reg, Blueprints: blueprints,
Planner: squad.NewRulePlanner(),
Backends: backends, Sink: squad.NopSink{},
Store: runStore,
})
if err != nil {
t.Fatal(err)
}
res, err := eng.Run(context.Background(), squad.Goal{Text: "做一个失败演示", HitlMode: squad.HitlAuto}, squad.RunOptions{})
if err != nil {
t.Fatalf("run: %v", err)
}
if res.Status == squad.RunCompleted {
t.Fatalf("failed executor result must not complete (patches are diagnostic only)")
}
if res.Status != squad.RunFailed {
t.Fatalf("status = %s, want failed", res.Status)
}
if res.Plan == nil || res.Plan.Steps[0].Status != "failed" {
t.Fatalf("s1 should be failed, plan = %+v", res.Plan)
}
if !strings.Contains(res.Error, "stub-refused") {
t.Fatalf("failure reason not preserved: %q", res.Error)
}
// 同一断言穿过 Store 端口:SQLite 里落盘的也是失败,而非 completed。
st, err := runStore.LoadRun(res.WorkDir)
if err != nil {
t.Fatal(err)
}
if st.Plan.Workflow.Status != "failed" {
t.Fatalf("persisted workflow status = %s, want failed", st.Plan.Workflow.Status)
}
if got := st.Plan.Workflow.Steps[0]; got.Status != "failed" || !strings.Contains(got.Error, "stub-refused") {
t.Fatalf("persisted step = %+v", got)
}
// patches 确已物化(文件齐全),依然不翻转状态:权威是执行结果,不是文件。
sub := filepath.FromSlash(st.Plan.Workflow.Steps[0].SubWorkspace)
if _, err := os.Stat(filepath.Join(res.WorkDir, sub, "output.md")); err != nil {
t.Fatalf("diagnostic patch should still be materialized: %v", err)
}
}
const goodReportLite = "# 测试交付物\n\n## 概述\n\n这是由脚本后端生成的合格交付物,长度超过五十个字符以满足最小质检要求。\n"
+104 -15
View File
@@ -2,12 +2,18 @@ package squad
import (
"context"
"encoding/json"
"os"
"path/filepath"
"sync"
"testing"
"time"
)
// jsonForTest 包装 json.Marshal 让测试里构型参数不拉锁。是仅为占用 import 走
// 公共路径,不污染生产代码。
func jsonForTest(v any) ([]byte, error) { return json.Marshal(v) }
// perStepRecordingBackend 按子工作区归因计数(steps 的 role→tasks/NN_role 布局)。
type perStepRecordingBackend struct {
mu sync.Mutex
@@ -59,6 +65,9 @@ func newAttemptTestEngine(t *testing.T, backend *perStepRecordingBackend, dir st
return eng
}
// flipToInterrupted 把该步骤手动置为 running,同时清除 trusted terminal
// record(模拟「崩溃于执行中」状态)。清除让该路径能够测试 F05 走的
// 「无 record」分支。
func flipToInterrupted(t *testing.T, workDir string, stepIdx int) {
t.Helper()
st, err := loadRunState(workDir)
@@ -75,6 +84,16 @@ func flipToInterrupted(t *testing.T, workDir string, stepIdx int) {
if err := st.save(); err != nil {
t.Fatal(err)
}
// 同时清除该步骤的 trusted terminal record(模拟未达记录写入点崩溃)。
terminals := loadStepTerminalsForWorkDir(workDir)
if terminals != nil {
stepID := st.Plan.Workflow.Steps[stepIdx].ID
if _, ok := terminals[stepID]; ok {
delete(terminals, stepID)
data, _ := jsonForTest(terminals)
_ = os.WriteFile(filepath.Join(workDir, "step_terminals.json"), data, 0o644)
}
}
}
// TestAttemptIDFormat:attempt id 稳定落盘且形如 runID/stepID/N(Q1 WP3)。
@@ -98,8 +117,15 @@ func TestAttemptIDFormat(t *testing.T) {
}
// TestKilledAtSuperstepBoundarySideEffectsOnce:杀进程于 superstep 边界后恢复,
// 已完成步骤不重新执行(执行次数=1);running 且产物齐全的步骤按 attempt 判定
// 完成、不 rerun——Q1 验收第一条(WP3)。
// 已完成步骤不重新执行(执行次数=1)。
//
// F05/U3b:running 崩溃、文件齐全、却无 trusted terminal record(U3b 前被视为
// “副作用真发生”自动完成;U3b 后仅记录 step.Status=needs_reconciliation,
// 后台不再自动 rerun/完成。本测试验证该迁变后的语义不变量:
//
// - running 且无 record:需要 reconciliation,不隐默 rerun;
// - file presence 不替 trusted record 完成步骤。
// - 若事先写入 trusted record(手工合成),即便文件缺失也恢复为 completed。
func TestKilledAtSuperstepBoundarySideEffectsOnce(t *testing.T) {
backend := &perStepRecordingBackend{}
dir := t.TempDir()
@@ -125,22 +151,78 @@ func TestKilledAtSuperstepBoundarySideEffectsOnce(t *testing.T) {
if err != nil {
t.Fatal(err)
}
if res.Status != RunCompleted {
t.Fatalf("resume: %+v", res)
// F05:需 reconciliation(auto 模式),不等于 completed。
if res.Status != RunWaitingReconciliation {
t.Fatalf("want waiting_reconciliation (auto mode), got %s", res.Status)
}
if res.Plan == nil || len(res.Plan.Steps) == 0 {
t.Fatal("missing plan")
}
var recon int
for _, s := range res.Plan.Steps {
if s.Status != "completed" && s.Status != "skipped" {
t.Fatalf("step %s not finished: %s", s.ID, s.Status)
if s.ID == "s2" && s.Status == StepStatusNeedsReconciliation {
recon++
}
}
if recon != 1 {
t.Fatalf("step s2 not in needs_reconciliation (got plan=%+v)", res.Plan.Steps)
}
// 重要:重复执行的次数不应发生隐默 rerun(本测试用 auto 模式运行走到 reconciliation)。
for _, s := range res.Plan.Steps {
got := backend.countOf(filepath.Base(filepath.FromSlash(s.SubWorkspace)))
if got != before[s.ID] {
t.Fatalf("step %s executed %d times after kill (before=%d)——已完成/running 步骤被重复执行", s.ID, got, before[s.ID])
t.Fatalf("step %s executed %d times after kill (before=%d)——running 步骤被隐默 rerun", s.ID, got, before[s.ID])
}
}
}
// TestInterruptedMissingProductsRetries:running 崩溃且产物缺失时按 auto 一次重试(WP3)。
func TestInterruptedMissingProductsRetries(t *testing.T) {
// TestTrustedRecordCompletesMissingFiles:trusted terminal record 存在 + 文件缺失
// → 恢复为 completed(files 仍作诊断,state 是权威)。
func TestTrustedRecordCompletesMissingFiles(t *testing.T) {
backend := &perStepRecordingBackend{}
dir := t.TempDir()
eng := newAttemptTestEngine(t, backend, dir)
res, err := eng.Run(context.Background(), Goal{Text: "trusted record 完成", HitlMode: HitlAuto}, RunOptions{})
if err != nil || res.Status != RunCompleted {
t.Fatalf("setup run: %v %+v", err, res)
}
// 制造「第二步走 running 但 trusted record 已存在」的场景:先写入 record,再
// flip 到 running + 删 file。
st, err := loadRunState(res.WorkDir)
if err != nil {
t.Fatal(err)
}
sub2 := filepath.FromSlash(st.Plan.Workflow.Steps[1].SubWorkspace)
flipToInterrupted(t, res.WorkDir, 1)
if err := os.Remove(filepath.Join(res.WorkDir, sub2, "output.md")); err != nil {
t.Fatal(err)
}
// 手工写入 trusted record(这能覆盖「后端状态已落盘但文件被外部删」该返回
// 已完成的合法语义)。
terminals := map[string]stepTerminalRecord{
"s2": {StepID: "s2", AttemptID: st.Plan.Workflow.Steps[1].AttemptID, Terminal: "completed", SideEffects: string(ExecSideEffectsNone), At: time.Now()},
}
data, _ := jsonForTest(terminals)
if err := os.WriteFile(filepath.Join(res.WorkDir, "step_terminals.json"), data, 0o644); err != nil {
t.Fatal(err)
}
before := backend.countOf(filepath.Base(sub2))
eng2 := newAttemptTestEngine(t, backend, dir)
res, err = eng2.Resume(context.Background(), res.RunID)
if err != nil {
t.Fatal(err)
}
if res.Status != RunCompleted {
t.Fatalf("want completed via trusted record, got %s", res.Status)
}
if got := backend.countOf(filepath.Base(sub2)); got != before {
t.Fatalf("trusted-record step reran (before=%d after=%d)——record 需被信任为 completed", before, got)
}
}
// TestInterruptedMissingProductsEntersReconciliation:F05/U3b:running 崩溃 + 无
// trusted record + 文件缺失 → auto 模式不再隐默 rerun,进入 waiting_reconciliation。
func TestInterruptedMissingProductsEntersReconciliation(t *testing.T) {
backend := &perStepRecordingBackend{}
dir := t.TempDir()
eng := newAttemptTestEngine(t, backend, dir)
@@ -148,7 +230,6 @@ func TestInterruptedMissingProductsRetries(t *testing.T) {
if err != nil || res.Status != RunCompleted {
t.Fatalf("setup run: %v %+v", err, res)
}
// 第二步停 running 并删掉其交付物(副作用未发生的形态)。
flipToInterrupted(t, res.WorkDir, 1)
st, err := loadRunState(res.WorkDir)
if err != nil {
@@ -158,16 +239,24 @@ func TestInterruptedMissingProductsRetries(t *testing.T) {
if err := os.Remove(filepath.Join(res.WorkDir, sub, "output.md")); err != nil {
t.Fatal(err)
}
before := backend.countOf(filepath.Base(sub))
eng2 := newAttemptTestEngine(t, backend, dir)
res, err = eng2.Resume(context.Background(), res.RunID)
if err != nil {
t.Fatal(err)
}
if res.Status != RunCompleted {
t.Fatalf("resume: %+v", res)
if res.Status != RunWaitingReconciliation {
t.Fatalf("want waiting_reconciliation, got %s", res.Status)
}
got := backend.countOf(filepath.Base(sub))
if got != 2 {
t.Fatalf("missing-product step executed %d times, want 2(auto 一次重试)", got)
// 后续 Resume 仍旧是 reconciliation,不再 rerun。
res2, err := eng2.Resume(context.Background(), res.RunID)
if err != nil {
t.Fatal(err)
}
if res2.Status != RunWaitingReconciliation {
t.Fatalf("second resume shouldn't rerun, got %s", res2.Status)
}
if got := backend.countOf(filepath.Base(sub)); got != before {
t.Fatalf("missing-files step reran (before=%d after=%d)——record 缺失不应 rerun", before, got)
}
}
+156 -9
View File
@@ -1,7 +1,9 @@
package squad
import (
"context"
"fmt"
"strings"
"sync"
"error-code-system/errcode"
@@ -16,11 +18,20 @@ type BackendFactory func(role RoleSpec) (Executor, error)
// agent-native 自研内核,产品也可注册 "pi-agent" 等外部智能体工厂。
// 未知名字 fail-closed,绝不静默落 stub。角色声明 MaxConcurrentTasks 时
// 以并发限流包装实现对象级串行(每个角色一份缓存实例,限流跨步骤共享)。
//
// 缓存键含 ConfigIdentity 与工厂世代,故 BashPermission/工具配置收紧或工厂
// 替换在下一次 Resolve 即生效。并发 gate 按「后端名+角色」独立于执行器版本
// 持有:配置换代不会另起限流器而绕过角色上限。
//
// 边界:已在飞的任务不会被回滚或取消;收紧权限/降低并发只影响后续 Resolve
// 与后续 Acquire。需要即时撤权的宿主应在每次工具调用前再校验。
type BackendResolver struct {
mu sync.RWMutex
factories map[string]BackendFactory
underlying map[string]Executor // 无工厂直挂的后端实例
wrapped map[string]Executor // 角色(含模型绑定)→ 已包装实例缓存
wrapped map[string]Executor // 执行器配置身份 → 已包装实例缓存
factoryGen map[string]uint64 // Register/RegisterInstance 世代,替换即递增
gates map[string]*roleConcurrencyGate
}
// NewBackendResolver 创建解析器并注册内置后端。
@@ -29,23 +40,40 @@ func NewBackendResolver() *BackendResolver {
factories: make(map[string]BackendFactory),
underlying: make(map[string]Executor),
wrapped: make(map[string]Executor),
factoryGen: make(map[string]uint64),
gates: make(map[string]*roleConcurrencyGate),
}
r.RegisterInstance("stub", &StubExecutor{})
return r
}
// Register 注册后端工厂(名字如 "agent-native"、"pi-agent")。
// 替换同名工厂会使该后端下已缓存的包装执行器失效。
func (r *BackendResolver) Register(name string, f BackendFactory) {
r.mu.Lock()
defer r.mu.Unlock()
r.factories[name] = f
r.invalidateLocked(name)
}
// RegisterInstance 直接注册后端实例(忽略角色级绑定)。
// 替换同名实例会使该后端下已缓存的包装执行器失效;"stub" 语义保持不变。
func (r *BackendResolver) RegisterInstance(name string, b Executor) {
r.mu.Lock()
defer r.mu.Unlock()
r.underlying[name] = b
r.invalidateLocked(name)
}
func (r *BackendResolver) invalidateLocked(name string) {
r.factoryGen[name]++
prefix := name + "\x00"
for k := range r.wrapped {
if strings.HasPrefix(k, prefix) {
delete(r.wrapped, k)
}
}
// gates 按角色身份保留:换工厂/配置不得丢掉进行中的并发计数。
}
// ListBackends 返回可用后端名(排序稳定)。
@@ -73,8 +101,8 @@ func (r *BackendResolver) ListBackends() []string {
}
// Resolve 解析角色的执行后端:Role.Backend → Options.DefaultBackend → "stub"。
// MaxConcurrentTasks > 1 时以 agent-native 并发限流包装(per-role 缓存,跨步骤共享;
// 创建段持写锁防并行首解析竞态——否则并发步骤会各拿一份限流器)。
// MaxConcurrentTasks >= 1 时以共享角色 gate 包装(跨执行器版本共用;
// 创建段持写锁防并行首解析竞态——否则并发步骤会各拿一份执行器)。
func (r *BackendResolver) Resolve(role RoleSpec, defaultBackend string) (Executor, string, error) {
name := role.Backend
if name == "" {
@@ -83,30 +111,44 @@ func (r *BackendResolver) Resolve(role RoleSpec, defaultBackend string) (Executo
if name == "" {
name = "stub"
}
cacheKey := name + "\x00" + role.Role + "\x00" + role.Model
// 后端名快照在锁外获取(Resolve 持锁期间再调 ListBackends 会自死锁)。
names := r.ListBackends()
r.mu.RLock()
gen := r.factoryGen[name]
cacheKey := resolveCacheKey(name, role, gen)
cached, ok := r.wrapped[cacheKey]
factory, hasFactory := r.factories[name]
under, hasUnder := r.underlying[name]
r.mu.RUnlock()
if ok {
if max := role.MaxConcurrentTasks; max >= 1 {
r.mu.Lock()
if g := r.gates[limiterKey(name, role.Role)]; g != nil {
g.setMax(max)
}
r.mu.Unlock()
}
return cached, name, nil
}
r.mu.Lock()
defer r.mu.Unlock()
gen = r.factoryGen[name]
cacheKey = resolveCacheKey(name, role, gen)
// 双检:等写锁期间可能已被并发解析填充。
if b, ok := r.wrapped[cacheKey]; ok {
if max := role.MaxConcurrentTasks; max >= 1 {
if g := r.gates[limiterKey(name, role.Role)]; g != nil {
g.setMax(max)
}
}
return b, name, nil
}
var b Executor
if hasUnder {
if under, hasUnder := r.underlying[name]; hasUnder {
b = under
} else if hasFactory {
} else if factory, hasFactory := r.factories[name]; hasFactory {
var err error
b, err = factory(role)
if err != nil {
@@ -117,9 +159,114 @@ func (r *BackendResolver) Resolve(role RoleSpec, defaultBackend string) (Executo
}
// MaxConcurrentTasks>=1 即包装限流(1 = 对象级串行;0 = 不限)。
// gate 按角色身份共享,与 ConfigIdentity / 工厂世代无关。
if max := role.MaxConcurrentTasks; max >= 1 {
b = WrapWithConcurrencyLimit(b, max)
g := r.gateLocked(name, role.Role, max)
b = &gatedExecutor{inner: b, gate: g}
}
r.wrapped[cacheKey] = b
return b, name, nil
}
func resolveCacheKey(name string, role RoleSpec, gen uint64) string {
return name + "\x00" + role.Role + "\x00" + ConfigIdentity(role) + "\x00" +
fmt.Sprintf("%d", gen) + "\x00" + limiterStamp(role.MaxConcurrentTasks)
}
func limiterKey(backend, role string) string {
return backend + "\x00" + role
}
func (r *BackendResolver) gateLocked(backend, role string, max int) *roleConcurrencyGate {
key := limiterKey(backend, role)
g, ok := r.gates[key]
if !ok {
g = newRoleConcurrencyGate(max)
r.gates[key] = g
return g
}
g.setMax(max)
return g
}
// roleConcurrencyGate 是按角色共享的并发上限。
// 执行器配置换代仍复用同一 gate,故旧版本在飞任务计入同一上限。
type roleConcurrencyGate struct {
mu sync.Mutex
max int
active int
notify chan struct{} // 关闭并替换以唤醒等待者
}
func newRoleConcurrencyGate(max int) *roleConcurrencyGate {
if max <= 0 {
max = 1
}
return &roleConcurrencyGate{max: max, notify: make(chan struct{})}
}
func (g *roleConcurrencyGate) setMax(max int) {
if max <= 0 {
max = 1
}
g.mu.Lock()
defer g.mu.Unlock()
g.max = max
g.signalLocked()
}
func (g *roleConcurrencyGate) signalLocked() {
close(g.notify)
g.notify = make(chan struct{})
}
func (g *roleConcurrencyGate) acquire(ctx context.Context) error {
for {
g.mu.Lock()
if g.active < g.max {
g.active++
g.mu.Unlock()
return nil
}
notify := g.notify
g.mu.Unlock()
select {
case <-ctx.Done():
return ctx.Err()
case <-notify:
}
}
}
func (g *roleConcurrencyGate) release() {
g.mu.Lock()
defer g.mu.Unlock()
if g.active > 0 {
g.active--
}
g.signalLocked()
}
// gatedExecutor 把共享角色 gate 套在具体执行器上。
type gatedExecutor struct {
inner Executor
gate *roleConcurrencyGate
}
func (e *gatedExecutor) Name() string {
if e.inner == nil {
return "gated"
}
return e.inner.Name()
}
func (e *gatedExecutor) Execute(ctx context.Context, req ExecRequest) (*ExecResult, error) {
if err := e.gate.acquire(ctx); err != nil {
return &ExecResult{
Success: false,
ErrorReason: fmt.Sprintf("agent concurrency wait canceled: %v", err),
}, err
}
defer e.gate.release()
return e.inner.Execute(ctx, req)
}
+405
View File
@@ -0,0 +1,405 @@
package squad
import (
"context"
"sync"
"sync/atomic"
"testing"
"time"
)
type labeledExec struct {
label string
perm string
tools string
delay time.Duration
enter func()
leave func()
}
func (e *labeledExec) Name() string { return e.label }
func (e *labeledExec) Execute(ctx context.Context, req ExecRequest) (*ExecResult, error) {
if e.enter != nil {
e.enter()
}
if e.leave != nil {
defer e.leave()
}
if e.delay > 0 {
select {
case <-ctx.Done():
return nil, ctx.Err()
case <-time.After(e.delay):
}
}
return &ExecResult{Success: true, Logs: e.label + "|" + e.perm + "|" + e.tools}, nil
}
func TestConfigIdentityChangesWithBashAndTools(t *testing.T) {
a := RoleSpec{Role: "r", Model: "m", BashPermission: BashPermissionAllow}
d := RoleSpec{Role: "r", Model: "m", BashPermission: BashPermissionDeny}
if ConfigIdentity(a) == ConfigIdentity(d) {
t.Fatal("allow and deny must differ")
}
t1 := RoleSpec{Role: "r", Model: "m", Tools: []string{"a", "b"}}
t2 := RoleSpec{Role: "r", Model: "m", Tools: []string{"b", "a"}}
if ConfigIdentity(t1) != ConfigIdentity(t2) {
t.Fatal("tool order must not change identity")
}
t3 := RoleSpec{Role: "r", Model: "m", Tools: []string{"a", "c"}}
if ConfigIdentity(t1) == ConfigIdentity(t3) {
t.Fatal("tool set change must change identity")
}
same := RoleSpec{Role: "r", Model: "m", BashPermission: BashPermissionAllow, MaxConcurrentTasks: 1}
bump := RoleSpec{Role: "r", Model: "m", BashPermission: BashPermissionAllow, MaxConcurrentTasks: 8}
if ConfigIdentity(same) != ConfigIdentity(bump) {
t.Fatal("MaxConcurrentTasks must not affect ConfigIdentity")
}
}
func TestResolveAllowDenyTighteningAndLoosening(t *testing.T) {
var builds atomic.Int32
r := NewBackendResolver()
r.Register("cap", func(role RoleSpec) (Executor, error) {
builds.Add(1)
return &labeledExec{label: "cap", perm: role.BashPermission}, nil
})
allow := RoleSpec{Role: "writer", Backend: "cap", Model: "m1", BashPermission: BashPermissionAllow}
e1, _, err := r.Resolve(allow, "")
if err != nil {
t.Fatal(err)
}
res, _ := e1.Execute(context.Background(), ExecRequest{})
if res.Logs != "cap|allow|" {
t.Fatalf("want allow executor, got %q", res.Logs)
}
deny := RoleSpec{Role: "writer", Backend: "cap", Model: "m1", BashPermission: BashPermissionDeny}
e2, _, err := r.Resolve(deny, "")
if err != nil {
t.Fatal(err)
}
res, _ = e2.Execute(context.Background(), ExecRequest{})
if res.Logs != "cap|deny|" {
t.Fatalf("allow→deny must take effect on next Resolve, got %q", res.Logs)
}
if e1 == e2 {
t.Fatal("tightened config must not reuse cached allow executor")
}
e3, _, err := r.Resolve(allow, "")
if err != nil {
t.Fatal(err)
}
res, _ = e3.Execute(context.Background(), ExecRequest{})
if res.Logs != "cap|allow|" {
t.Fatalf("deny→allow must take effect, got %q", res.Logs)
}
if builds.Load() < 2 {
t.Fatalf("factory must rebuild on permission change, builds=%d", builds.Load())
}
}
func TestResolveFactoryReplacementTakesEffect(t *testing.T) {
r := NewBackendResolver()
r.Register("cap", func(role RoleSpec) (Executor, error) {
return &labeledExec{label: "v1", perm: role.BashPermission}, nil
})
role := RoleSpec{Role: "writer", Backend: "cap", Model: "m", BashPermission: BashPermissionAllow}
e1, _, err := r.Resolve(role, "")
if err != nil {
t.Fatal(err)
}
res, _ := e1.Execute(context.Background(), ExecRequest{})
if res.Logs != "v1|allow|" {
t.Fatalf("got %q", res.Logs)
}
r.Register("cap", func(role RoleSpec) (Executor, error) {
return &labeledExec{label: "v2", perm: role.BashPermission}, nil
})
e2, _, err := r.Resolve(role, "")
if err != nil {
t.Fatal(err)
}
res, _ = e2.Execute(context.Background(), ExecRequest{})
if res.Logs != "v2|allow|" {
t.Fatalf("factory replacement must take effect, got %q", res.Logs)
}
}
func TestResolveToolConfigChangeTakesEffect(t *testing.T) {
r := NewBackendResolver()
r.Register("cap", func(role RoleSpec) (Executor, error) {
tools := ""
if len(role.Tools) > 0 {
tools = role.Tools[0]
}
return &labeledExec{label: "cap", tools: tools}, nil
})
base := RoleSpec{Role: "writer", Backend: "cap", Model: "m", Tools: []string{"read"}}
e1, _, err := r.Resolve(base, "")
if err != nil {
t.Fatal(err)
}
res, _ := e1.Execute(context.Background(), ExecRequest{})
if res.Logs != "cap||read" {
t.Fatalf("got %q", res.Logs)
}
next := RoleSpec{Role: "writer", Backend: "cap", Model: "m", Tools: []string{"write"}}
e2, _, err := r.Resolve(next, "")
if err != nil {
t.Fatal(err)
}
res, _ = e2.Execute(context.Background(), ExecRequest{})
if res.Logs != "cap||write" {
t.Fatalf("tool-config change must take effect, got %q", res.Logs)
}
}
func TestResolveUnchangedConfigReusesCache(t *testing.T) {
var builds atomic.Int32
r := NewBackendResolver()
r.Register("cap", func(role RoleSpec) (Executor, error) {
builds.Add(1)
return &labeledExec{label: "cap", perm: role.BashPermission}, nil
})
role := RoleSpec{Role: "writer", Backend: "cap", Model: "m", BashPermission: BashPermissionAsk}
e1, _, err := r.Resolve(role, "")
if err != nil {
t.Fatal(err)
}
e2, _, err := r.Resolve(role, "")
if err != nil {
t.Fatal(err)
}
if e1 != e2 {
t.Fatal("unchanged config must reuse cached executor")
}
if builds.Load() != 1 {
t.Fatalf("want 1 factory call, got %d", builds.Load())
}
}
func TestResolveParallelFirstDoesNotDuplicate(t *testing.T) {
var builds atomic.Int32
r := NewBackendResolver()
r.Register("cap", func(role RoleSpec) (Executor, error) {
builds.Add(1)
time.Sleep(20 * time.Millisecond) // widen race window
return &labeledExec{label: "cap"}, nil
})
role := RoleSpec{Role: "writer", Backend: "cap", Model: "m", MaxConcurrentTasks: 2}
const n = 32
var wg sync.WaitGroup
got := make([]Executor, n)
errs := make([]error, n)
wg.Add(n)
for i := 0; i < n; i++ {
i := i
go func() {
defer wg.Done()
got[i], _, errs[i] = r.Resolve(role, "")
}()
}
wg.Wait()
for i, err := range errs {
if err != nil {
t.Fatalf("resolve %d: %v", i, err)
}
}
for i := 1; i < n; i++ {
if got[i] != got[0] {
t.Fatalf("parallel first Resolve must share one executor")
}
}
if builds.Load() != 1 {
t.Fatalf("want single factory construction, got %d", builds.Load())
}
}
func TestSharedGateAcrossVersionsAndLimiterChange(t *testing.T) {
var (
mu sync.Mutex
current int
peak int
)
hold := make(chan struct{})
entered := make(chan struct{}, 8)
trackEnter := func() {
mu.Lock()
current++
if current > peak {
peak = current
}
mu.Unlock()
entered <- struct{}{}
<-hold
}
trackLeave := func() {
mu.Lock()
current--
mu.Unlock()
}
r := NewBackendResolver()
r.Register("cap", func(role RoleSpec) (Executor, error) {
return &labeledExec{
label: "cap",
perm: role.BashPermission,
enter: trackEnter,
leave: trackLeave,
}, nil
})
allow := RoleSpec{Role: "writer", Backend: "cap", Model: "m", BashPermission: BashPermissionAllow, MaxConcurrentTasks: 1}
eOld, _, err := r.Resolve(allow, "")
if err != nil {
t.Fatal(err)
}
errCh := make(chan error, 4)
go func() { _, err := eOld.Execute(context.Background(), ExecRequest{}); errCh <- err }()
<-entered // old holds the sole slot
deny := RoleSpec{Role: "writer", Backend: "cap", Model: "m", BashPermission: BashPermissionDeny, MaxConcurrentTasks: 1}
eNew, _, err := r.Resolve(deny, "")
if err != nil {
t.Fatal(err)
}
if eOld == eNew {
t.Fatal("version bump must produce a new executor")
}
secondDone := make(chan struct{})
go func() {
_, err := eNew.Execute(context.Background(), ExecRequest{})
errCh <- err
close(secondDone)
}()
select {
case <-secondDone:
t.Fatal("new-version execute must wait on shared role cap while old is in-flight")
case <-time.After(40 * time.Millisecond):
}
// Raising MaxConcurrentTasks must update the shared gate (not spawn a bypassing limiter).
deny2 := RoleSpec{Role: "writer", Backend: "cap", Model: "m", BashPermission: BashPermissionDeny, MaxConcurrentTasks: 2}
eLim, _, err := r.Resolve(deny2, "")
if err != nil {
t.Fatal(err)
}
go func() { _, err := eLim.Execute(context.Background(), ExecRequest{}); errCh <- err }()
// After setMax(2): waiting eNew can enter (active=2). eLim must still wait.
select {
case <-entered: // eNew
case <-time.After(time.Second):
t.Fatal("limiter raise must admit the waiting new-version task")
}
select {
case <-entered:
t.Fatal("third task must not bypass shared cap after limiter change")
case <-time.After(40 * time.Millisecond):
}
close(hold)
for i := 0; i < 3; i++ {
if err := <-errCh; err != nil {
t.Fatal(err)
}
}
mu.Lock()
p := peak
mu.Unlock()
if p > 2 {
t.Fatalf("shared gate must cap total in-flight across versions; peak=%d", p)
}
}
func TestInFlightOldVersionCannotExceedCap(t *testing.T) {
var (
mu sync.Mutex
current int
peak int
)
enter := func() {
mu.Lock()
current++
if current > peak {
peak = current
}
mu.Unlock()
}
leave := func() {
mu.Lock()
current--
mu.Unlock()
}
r := NewBackendResolver()
r.Register("cap", func(role RoleSpec) (Executor, error) {
return &labeledExec{
label: "cap",
perm: role.BashPermission,
delay: 80 * time.Millisecond,
enter: enter,
leave: leave,
}, nil
})
v1 := RoleSpec{Role: "writer", Backend: "cap", Model: "m", BashPermission: BashPermissionAllow, MaxConcurrentTasks: 2}
e1, _, err := r.Resolve(v1, "")
if err != nil {
t.Fatal(err)
}
v2 := RoleSpec{Role: "writer", Backend: "cap", Model: "m", BashPermission: BashPermissionDeny, MaxConcurrentTasks: 2}
e2, _, err := r.Resolve(v2, "")
if err != nil {
t.Fatal(err)
}
var wg sync.WaitGroup
// 2 from old version + 2 from new version; shared cap=2 ⇒ peak must stay ≤2.
for _, e := range []Executor{e1, e1, e2, e2} {
wg.Add(1)
e := e
go func() {
defer wg.Done()
_, _ = e.Execute(context.Background(), ExecRequest{})
}()
}
wg.Wait()
mu.Lock()
p := peak
mu.Unlock()
if p > 2 {
t.Fatalf("in-flight old+new must not exceed role cap; peak=%d", p)
}
}
func TestRegisterInstanceStubSemantics(t *testing.T) {
r := NewBackendResolver()
e, name, err := r.Resolve(RoleSpec{Role: "x"}, "")
if err != nil {
t.Fatal(err)
}
if name != "stub" {
t.Fatalf("want stub, got %s", name)
}
if e.Name() != "stub" {
t.Fatalf("want stub executor, got %s", e.Name())
}
custom := &labeledExec{label: "stub-custom"}
r.RegisterInstance("stub", custom)
e2, _, err := r.Resolve(RoleSpec{Role: "x"}, "")
if err != nil {
t.Fatal(err)
}
if e2.Name() != "stub-custom" {
t.Fatalf("RegisterInstance stub replace must take effect, got %s", e2.Name())
}
}
+45
View File
@@ -0,0 +1,45 @@
package squad
import (
"crypto/sha256"
"encoding/hex"
"sort"
"strings"
)
// ConfigIdentity 返回角色中「塑造 Executor 装配」字段的稳定摘要。
// 含 BashPermission / Model / ToolProfile / Tools / Skills;不含
// MaxConcurrentTasks(并发上限由解析器按角色身份单独持有,与执行器版本解耦)。
func ConfigIdentity(role RoleSpec) string {
var b strings.Builder
b.WriteString("bash=")
b.WriteString(role.BashPermission)
b.WriteString("\x00model=")
b.WriteString(role.Model)
b.WriteString("\x00profile=")
b.WriteString(role.ToolProfile)
b.WriteString("\x00tools=")
b.WriteString(stableJoin(role.Tools))
b.WriteString("\x00skills=")
b.WriteString(stableJoin(role.Skills))
sum := sha256.Sum256([]byte(b.String()))
return hex.EncodeToString(sum[:16])
}
func stableJoin(items []string) string {
if len(items) == 0 {
return ""
}
cp := append([]string(nil), items...)
sort.Strings(cp)
return strings.Join(cp, ",")
}
// limiterStamp 区分「启用限流」与「不限流」;具体上限数值不进执行器缓存键,
// 以免调大/调小 MaxConcurrentTasks 时换掉共享 gate。
func limiterStamp(max int) string {
if max >= 1 {
return "lim"
}
return "unlim"
}
+116 -53
View File
@@ -1051,73 +1051,129 @@ func markStep(st *RunState, id, status, errText string) bool {
}
// recoverInterrupted 处理崩溃时停在 running 的步骤。
// interactive 挂起确认,不自动再执行;auto 允许一次重试;disabled 直接失败。
// interactive 挂起确认/对账,不自动再执行;auto 允许一次重试;disabled 直接失败。
//
// F05/U3b:恢复是否判定 completed 不再单纯依据「文件齐全」——trusted terminal
// record 是单一权威:仅当 step_terminals.json 中存在该步骤且 Terminal=completed
// 时才能恢复完成(即便文件缺失也照恢复,文件用于诊断);其余情形按 record +
// 文件状态联合判定:
//
// - 有 trusted record(completed)+ 缺失文件 -> completed(files 诊断信息)
// - 有 trusted record + 现存文件 -> completed(一致)
// - 有 trusted record + SideEffects=none + 缺失文件 -> 一次重试 pending
// - 无 record + 文件齐全 -> needs_reconciliation(不自动 rerun,不假装完成)
// - 无 record + 文件缺失 -> needs_reconciliation
// - record 表明 Canceled/Aborted -> 走 cancel 路径
//
// retention 期间产物缺失若 record SideEffects=possible,禁止无条件 rerun ——
// 必须经人工 reconciliation 才能继续。
func (e *SquadEngine) recoverInterrupted(ctx context.Context, st *RunState) (*RunResult, error) {
terminals := loadStepTerminalsForWorkDir(st.Meta.WorkDir)
drv := &flowDriver{opts: e.deps}
for i := range st.Plan.Workflow.Steps {
step := st.Plan.Workflow.Steps[i]
if step.Status != "running" {
continue
}
subDir := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(orDefault(step.SubWorkspace, fmt.Sprintf("tasks/%02d_%s", i+1, step.Role))))
targets := step.Targets
if len(targets) == 0 {
targets = []string{"output.md"}
}
missing := missingTargets(subDir, targets)
record, hasRecord := terminals[step.ID]
// 1. trusted terminal record 存在 -> 单一权威。
if hasRecord && record.Terminal == "completed" && !record.Canceled && !record.Aborted {
step.Status = "completed"
step.Error = ""
st.Plan.Workflow.Steps[i] = step
st.mu.Lock()
_ = mergeStepArtifacts(&st.BB, subDir)
err := st.save()
st.mu.Unlock()
if err != nil {
return nil, err
}
drv.emit(st, Event{Type: EvStepCompleted, StepID: step.ID, Role: step.Role,
Text: fmt.Sprintf("中断恢复:trusted terminal record(attempt=%s, side_effects=%s)判定完成",
record.AttemptID, orDefault(record.SideEffects, "possible"))})
continue
}
// 1b. record 表明已取消/中止 -> cancel 路径(不等同业务失败)。
if hasRecord && (record.Canceled || record.Aborted) {
step.Status = "failed"
if record.Canceled {
step.Error = "interrupted: canceled"
} else {
step.Error = "interrupted: aborted"
}
st.Plan.Workflow.Steps[i] = step
if err := st.save(); err != nil {
return nil, err
}
out, _, ferr := drv.failStep(ctx, st, i)
return out, ferr
}
// 1c. record 表明 SideEffects=none + 缺失文件 -> 仅一次重试 pending
// (旧 WP3 路径保留以兼容纯读取类步骤)。
if hasRecord && record.SideEffects == string(ExecSideEffectsNone) && len(missing) > 0 && step.Attempts < 2 {
step.Attempts++
step.Status = "pending"
step.Error = ""
// F05:复用上一次 attempt 的 ExecutionID(已在 step.Terminal 落 record)。
if record.AttemptID != "" {
step.AttemptID = record.AttemptID
}
st.Plan.Workflow.Steps[i] = step
if err := st.save(); err != nil {
return nil, err
}
continue
}
// 2. 无 trusted record —— 旧路径不准直接翻 completed,必须进入 reconciliation。
// 区分 HITL 模式:interactive 仍合成 HITL hold(同 hitlConfirmation 通道),
// auto / disabled 走显式 needs_reconciliation(不自动 rerun 不假装成功)。
reason := "no trusted terminal record"
if len(missing) == 0 {
reason = "no trusted terminal record despite deliverables present"
} else {
reason = "no trusted terminal record and deliverables missing"
}
step.Status = StepStatusNeedsReconciliation
step.Error = reason
st.Plan.Workflow.Steps[i] = step
if err := st.save(); err != nil {
return nil, err
}
switch st.policy().mode() {
case HitlDisabled:
st.Plan.Workflow.Steps[i].Status = "failed"
st.Plan.Workflow.Steps[i].Error = "interrupted"
if err := st.save(); err != nil {
return nil, err
}
drv := &flowDriver{opts: e.deps}
out, _, err := drv.failStep(ctx, st, i)
return out, err
case HitlAuto:
if step.Attempts >= 2 {
st.Plan.Workflow.Steps[i].Status = "failed"
st.Plan.Workflow.Steps[i].Error = "interrupted twice"
if err := st.save(); err != nil {
return nil, err
}
drv := &flowDriver{opts: e.deps}
out, _, err := drv.failStep(ctx, st, i)
return out, err
}
// Q1 WP3 产物校验:步骤在 running 崩溃,但其交付物已全部落盘——
// 副作用实际发生过,直接判定完成而不是重新执行(副作用次数=1)。
subDir := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(orDefault(step.SubWorkspace, fmt.Sprintf("tasks/%02d_%s", i+1, step.Role))))
targets := step.Targets
if len(targets) == 0 {
targets = []string{"output.md"}
}
if missingTargets(subDir, targets) == nil {
step.Status = "completed"
step.Error = ""
st.Plan.Workflow.Steps[i] = step
st.mu.Lock()
_ = mergeStepArtifacts(&st.BB, subDir)
err := st.save()
st.mu.Unlock()
if err != nil {
return nil, err
}
drv := &flowDriver{opts: e.deps}
drv.emit(st, Event{Type: EvStepCompleted, StepID: step.ID, Role: step.Role,
Text: fmt.Sprintf("中断恢复:产物齐全,按 attempt %s 判定完成", step.AttemptID)})
continue
}
st.Plan.Workflow.Steps[i].Status = "pending"
if err := st.save(); err != nil {
return nil, err
}
return nil, nil
default:
drv := &flowDriver{opts: e.deps}
case HitlInteractive:
out, _, err := drv.hold(st, HitlConfirmation, step.ID, step.Role,
fmt.Sprintf("步骤「%s」在执行中中断。如何继续?", step.Title),
fmt.Sprintf("步骤「%s」在执行中中断,且没有可信任的完成记录。%s 如何继续?", step.Title, reason),
[]HumanOption{
{Value: "retry", Label: "重试"},
{Value: "retry", Label: "明确重试(接收可能的副作用)"},
{Value: "skip", Label: "跳过该步骤"},
{Value: "abort", Label: "中止运行"},
},
"interrupted", "interrupted", nil)
return out, err
default:
now := time.Now()
st.Meta.Status = string(RunWaitingReconciliation)
st.Meta.UpdatedAt = now
if err := st.save(); err != nil {
return nil, err
}
drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("中断恢复:%s进入 reconciliation(reason=%s)", step.ID, reason)})
return &RunResult{
RunID: st.Meta.ID,
Status: RunWaitingReconciliation,
Plan: planPtr(st),
WorkDir: st.Meta.WorkDir,
Summary: "interrupted-running step requires reconciliation",
Error: reason,
}, nil
}
}
return nil, nil
@@ -1143,6 +1199,13 @@ func (e *SquadEngine) Resume(ctx context.Context, runID string) (*RunResult, err
if RunStatus(st.Meta.Status).Terminal() {
return e.Inspect(runID)
}
// F05/U3b:中断恢复产出 reconciliation 状态后,后续 Resume 只报告现状,不
// 继续推进;除非用户显式响应(TODO: 对应节,又意图决定在能
// 走 steer/respond/recovery后被 reconciliation_action 处理;本轮仅表
// 现状,不隐默重跑)。
if RunStatus(st.Meta.Status) == RunWaitingReconciliation {
return e.Inspect(runID)
}
// 图连续性校验(Q1 WP4/WP1 裁定):崩溃期间的检查点图指纹与当前 Steps
// 重编译不一致 = 图形态被改过(动态调整或外力),拒绝恢复并提示重新规划。
if err := validateGraphContinuity(st); err != nil {
+517
View File
@@ -0,0 +1,517 @@
// Package squad: F05/U3b executor result 合约测试。
//
// 覆盖以下消费规则:
// - (nil,nil) executor 返回 (result==nil && err==nil) 是稳定失败(非 panic,非 success)。
// - Success=true 但无执行证据(patches-only)也要走交付物校验;
// 失败-with-full-patches 仍为失败(不允许文件存在翻成 success)。
// - canceled/aborted 结果不进入 retry 通路(不是业务失败,不可重试)。
// - side_effects=possible 的失败禁止自动重试;
// side_effects=none 的纯读取失败可走有界重试。
// - recovery:trusted record 存在 + 文件缺失 -> 恢复 completed(权威来源 = record);
// 无 record + 文件齐全 -> needs_reconciliation(不假装完成、不隐默 rerun);
// 无 record + 文件缺失 -> needs_reconciliation;
// record SideEffects=none + 缺失文件 -> 一次重试 pending(不重复执行副作用)。
//
// 这些测试使用 squad 包的 in-memory executor 替身进行 hermetic 验证;不调用
// 真实模型或外部组件。
package squad
import (
"context"
"os"
"path/filepath"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
)
// ---- 替身 executor ----
// frozenResultBackend 每次 Execute 返回同一个固定结果(用于直接构造语义场景)。
type frozenResultBackend struct {
mu sync.Mutex
calls int32
res *ExecResult
err error
}
func (b *frozenResultBackend) Name() string { return "frozen" }
func (b *frozenResultBackend) Execute(_ context.Context, _ ExecRequest) (*ExecResult, error) {
atomic.AddInt32(&b.calls, 1)
b.mu.Lock()
defer b.mu.Unlock()
if b.err != nil {
return b.res, b.err
}
return b.res, nil
}
func (b *frozenResultBackend) callCount() int32 {
return atomic.LoadInt32(&b.calls)
}
// materializingBackend 每次 Execute 写入 patches 但把 Success 留给 caller 决定。
// 用于「patches-only success」测试场景。
type materializingBackend struct {
mu sync.Mutex
calls int32
// success 只在 success==true 时返回 Success=true;测试 override 用。
success bool
canceled bool
aborted bool
// sideEffects 决定是否允许重试。
sideEffects ExecSideEffects
// writePatch 控制是否 materializes patch 到 workDir。
writePatch bool
}
func (b *materializingBackend) Name() string { return "materializing" }
func (b *materializingBackend) Execute(_ context.Context, req ExecRequest) (*ExecResult, error) {
atomic.AddInt32(&b.calls, 1)
patches := map[string][]byte{}
for _, t := range req.Targets {
if b.writePatch {
_ = os.WriteFile(filepath.Join(req.WorkDir, t), []byte(goodReport), 0o644)
}
patches[t] = []byte(goodReport)
}
return &ExecResult{
Success: b.success,
Patches: patches,
Logs: "stub",
SideEffects: b.sideEffects,
Canceled: b.canceled,
Aborted: b.aborted,
}, nil
}
func (b *materializingBackend) callCount() int32 {
return atomic.LoadInt32(&b.calls)
}
// ---- 测试辅助 ----
func newFrozenTestEngine(t *testing.T, dir string, backend Executor) *SquadEngine {
t.Helper()
reg := NewRoleRegistry()
if err := reg.Register(RoleSpec{Role: "researcher", Name: "调研", Backend: backend.Name()}); err != nil {
t.Fatal(err)
}
if err := reg.Register(RoleSpec{Role: "writer", Name: "撰写", Backend: backend.Name()}); err != nil {
t.Fatal(err)
}
if err := reg.Register(RoleSpec{Role: "reviewer", Name: "审核", Backend: backend.Name()}); err != nil {
t.Fatal(err)
}
backends := NewBackendResolver()
backends.RegisterInstance(backend.Name(), backend)
eng, err := NewEngine(Options{
DataDir: dir, Registry: reg, Planner: mockScriptPlanner(baseSteps()...),
Backends: backends, Sink: NopSink{},
})
if err != nil {
t.Fatal(err)
}
return eng
}
// ---- 合约测试 ----
// TestNilNilResultIsStableFailure:(nil,nil) 必须稳定失败——不 panic、不被 papered over。
func TestNilNilResultIsStableFailure(t *testing.T) {
dir := t.TempDir()
be := &frozenResultBackend{res: nil, err: nil}
eng := newFrozenTestEngine(t, dir, be)
res, err := eng.Run(context.Background(), Goal{Text: "空结果"}, RunOptions{})
if err != nil {
t.Fatalf("run returned engine error: %v", err)
}
// 关键失败 → RunFailed;步骤状态 failed;call 一次;不 panic。
if res.Status != RunFailed {
t.Fatalf("want RunFailed, got %s (err=%s)", res.Status, res.Error)
}
if be.callCount() != 1 {
t.Fatalf("backend called %d times, want 1", be.callCount())
}
if res.Plan == nil {
t.Fatal("plan missing")
}
if got := res.Plan.Steps[0].Status; got != "failed" {
t.Fatalf("step 1 status = %s, want failed", got)
}
if got := res.Plan.Steps[0].Error; got != "executor returned nil result" {
t.Fatalf("step 1 error = %q, want executor-nil-result reason", got)
}
}
// TestSuccessWithFullPatchesMaterializes:Success=true + 已有 patch
// —— 通过交付物校验完成;patches-only success 也走同一路径。
func TestSuccessWithFullPatchesMaterializes(t *testing.T) {
dir := t.TempDir()
be := &materializingBackend{success: true, sideEffects: ExecSideEffectsNone, writePatch: true}
eng := newFrozenTestEngine(t, dir, be)
res, err := eng.Run(context.Background(), Goal{Text: "patches 落地校验"}, RunOptions{})
if err != nil {
t.Fatalf("run: %v", err)
}
if res.Status != RunCompleted {
t.Fatalf("want completed, got %s err=%s", res.Status, res.Error)
}
// 三步各调用一次;patches 命中产物路径且 Success=true,每步一次。
if got := be.callCount(); got < 3 {
t.Fatalf("backend called %d times, want at least 3 (3 steps)", got)
}
}
// TestFailedWithFullPatchesStaysFailed:F05/KTD4 关键不变式:失败 + 文件齐全
// 也不能翻成成功——即便 patches 内容已落地。
func TestFailedWithFullPatchesStaysFailed(t *testing.T) {
dir := t.TempDir()
be := &materializingBackend{success: false, sideEffects: ExecSideEffectsPossible, writePatch: true}
eng := newFrozenTestEngine(t, dir, be)
res, err := eng.Run(context.Background(), Goal{Text: "失败-but-files-present"}, RunOptions{})
if err != nil {
t.Fatalf("run: %v", err)
}
if res.Status != RunFailed {
t.Fatalf("want RunFailed, got %s", res.Status)
}
if res.Plan == nil || res.Plan.Steps[0].Status != "failed" {
t.Fatalf("step 1 should be failed, plan=%+v", res.Plan)
}
// 关键:失败被 failStep 重判后,不应回写 completed;也不应自动 rerun。
// SideEffects=possible 禁止重试:s1 仅调用一次,后续步骤不再调用(运行已中止)。
if got := be.callCount(); got > 1 {
t.Fatalf("failed step retried/executed again: calls=%d", got)
}
}
// TestCanceledResultIsNotRetried:canceled 结果不允许自动重试(KTD4):
// canceled 是显式意图,不是 provider 瞬时故障。
func TestCanceledResultIsNotRetried(t *testing.T) {
dir := t.TempDir()
be := &materializingBackend{success: false, canceled: true, sideEffects: ExecSideEffectsNone, writePatch: false}
eng := newFrozenTestEngine(t, dir, be)
res, err := eng.Run(context.Background(), Goal{Text: "取消不可重试"}, RunOptions{})
if err != nil {
t.Fatalf("run: %v", err)
}
// 失败(不是 panic),且仅被调用一次(不允许 retry)。
if res.Status != RunFailed {
t.Fatalf("want RunFailed, got %s", res.Status)
}
if be.callCount() != 1 {
t.Fatalf("canceled invoked %d times, want 1 (no retry)", be.callCount())
}
}
// TestAbortedResultIsNotRetried:aborted 同 canceled 语义。
func TestAbortedResultIsNotRetried(t *testing.T) {
dir := t.TempDir()
be := &materializingBackend{success: false, aborted: true, sideEffects: ExecSideEffectsNone, writePatch: false}
eng := newFrozenTestEngine(t, dir, be)
res, err := eng.Run(context.Background(), Goal{Text: "中止不可重试"}, RunOptions{})
if err != nil {
t.Fatalf("run: %v", err)
}
if res.Status != RunFailed {
t.Fatalf("want RunFailed, got %s", res.Status)
}
if be.callCount() != 1 {
t.Fatalf("aborted invoked %d times, want 1 (no retry)", be.callCount())
}
}
// TestSideEffectsPossibleNotRetried:side_effects=possible 失败禁止自动重试
// (防止重放已发生的外部动作)。
func TestSideEffectsPossibleNotRetried(t *testing.T) {
dir := t.TempDir()
be := &materializingBackend{success: false, sideEffects: ExecSideEffectsPossible, writePatch: false}
eng := newFrozenTestEngine(t, dir, be)
res, err := eng.Run(context.Background(), Goal{Text: "可能副作用失败不重试"}, RunOptions{})
if err != nil {
t.Fatalf("run: %v", err)
}
if res.Status != RunFailed {
t.Fatalf("want RunFailed, got %s", res.Status)
}
if be.callCount() != 1 {
t.Fatalf("possible-side-effects invoked %d times, want 1 (no retry)", be.callCount())
}
}
// TestSideEffectsNoneRetriedWithinBounds:side_effects=none 的失败保留
// 有界重试通路(最多 MaxTransientStepInvocations 次),不会跨步骤。
func TestSideEffectsNoneRetriedWithinBounds(t *testing.T) {
dir := t.TempDir()
// stage 1: failed with side_effects=none → 触发重试;
// stage 2: 返回 success。后续调用返回 last state(success)。
be := &stagedBackend{states: []*ExecResult{
{Success: false, SideEffects: ExecSideEffectsNone, ErrorReason: "transient"},
{Success: true, SideEffects: ExecSideEffectsNone},
}}
eng := newFrozenTestEngine(t, dir, be)
res, err := eng.Run(context.Background(), Goal{Text: "可重试"}, RunOptions{})
if err != nil {
t.Fatalf("run: %v", err)
}
if res.Status != RunCompleted {
t.Fatalf("want RunCompleted, got %s err=%s", res.Status, res.Error)
}
// 全运行计数:s1 经历重试(最多 2 次调用);后续步骤首次成功。
// 说明 重试仅限于 s1,不超出单步边界。
if be.callCount() < 2 {
t.Fatalf("retried count = %d, want at least 2 (failed then success on s1)", be.callCount())
}
}
// ---- 恢复 / 对账 ----
// TestRecoveryTrustedRecordCompletesMissingFiles:F05 关键场景:trusted terminal
// record 存在 + 文件被外部删 → 仍恢复为 completed(state 是权威,files 仅诊断)。
func TestRecoveryTrustedRecordCompletesMissingFiles(t *testing.T) {
backend := &perStepRecordingBackend{}
dir := t.TempDir()
eng := newAttemptTestEngine(t, backend, dir)
res, err := eng.Run(context.Background(), Goal{Text: "record-only-path"}, RunOptions{})
if err != nil || res.Status != RunCompleted {
t.Fatalf("setup run: %v %+v", err, res)
}
// 准备恢复场景:删除 s2 的 output.md,但保留 trusted record。
st, err := loadRunState(res.WorkDir)
if err != nil {
t.Fatal(err)
}
sub := filepath.FromSlash(st.Plan.Workflow.Steps[1].SubWorkspace)
if err := os.Remove(filepath.Join(res.WorkDir, sub, "output.md")); err != nil {
t.Fatal(err)
}
// 记录已写入(setup 步骤会触发 recordStepTerminal)。
terminals := loadStepTerminalsForWorkDir(res.WorkDir)
if _, ok := terminals[st.Plan.Workflow.Steps[1].ID]; !ok {
t.Fatalf("trusted record missing for %s", st.Plan.Workflow.Steps[1].ID)
}
// Flip s2 to running(带 record)。
st2, _ := loadRunState(res.WorkDir)
st2.Plan.Workflow.Steps[1].Status = "running"
st2.Plan.Workflow.Steps[1].Attempts = 2
st2.setWorkflow(st2.Plan.Workflow)
if err := st2.save(); err != nil {
t.Fatal(err)
}
eng2 := newAttemptTestEngine(t, backend, dir)
res2, err := eng2.Resume(context.Background(), res.RunID)
if err != nil {
t.Fatal(err)
}
if res2.Status != RunCompleted {
t.Fatalf("trusted record should restore completed, got %s", res2.Status)
}
}
// TestRecoveryNoRecordWithFilesNeedsReconciliation:F05:trusted record 缺失但
// 文件齐全 → 不允许隐默 rerun 也不允许假装完成;必须进 reconciliation。
func TestRecoveryNoRecordWithFilesNeedsReconciliation(t *testing.T) {
backend := &perStepRecordingBackend{}
dir := t.TempDir()
eng := newAttemptTestEngine(t, backend, dir)
res, err := eng.Run(context.Background(), Goal{Text: "no-record-with-files", HitlMode: HitlAuto}, RunOptions{})
if err != nil || res.Status != RunCompleted {
t.Fatalf("setup run: %v %+v", err, res)
}
flipToInterrupted(t, res.WorkDir, 1)
// flipToInterrupted 现在已经删除 record;显式再校验一次。
ws := res.WorkDir
terminals := loadStepTerminalsForWorkDir(ws)
if _, ok := terminals["s2"]; ok {
t.Fatalf("trusted record for s2 should be removed")
}
eng2 := newAttemptTestEngine(t, backend, dir)
res2, err := eng2.Resume(context.Background(), res.RunID)
if err != nil {
t.Fatal(err)
}
if res2.Status != RunWaitingReconciliation {
t.Fatalf("want waiting_reconciliation, got %s", res2.Status)
}
if res2.Plan == nil {
t.Fatal("plan missing")
}
var recon int
for _, s := range res2.Plan.Steps {
if s.ID == "s2" && s.Status == StepStatusNeedsReconciliation {
recon++
}
}
if recon != 1 {
t.Fatalf("s2 should be needs_reconciliation (plan=%+v)", res2.Plan.Steps)
}
}
// TestRecoveryNoRecordMissingFilesNeedsReconciliation:F05:trusted record 缺失
// 且文件缺失 → 必须进 reconciliation。
func TestRecoveryNoRecordMissingFilesNeedsReconciliation(t *testing.T) {
backend := &perStepRecordingBackend{}
dir := t.TempDir()
eng := newAttemptTestEngine(t, backend, dir)
res, err := eng.Run(context.Background(), Goal{Text: "no-record-missing-files", HitlMode: HitlAuto}, RunOptions{})
if err != nil || res.Status != RunCompleted {
t.Fatalf("setup run: %v %+v", err, res)
}
flipToInterrupted(t, res.WorkDir, 1)
st, _ := loadRunState(res.WorkDir)
sub := filepath.FromSlash(st.Plan.Workflow.Steps[1].SubWorkspace)
if err := os.Remove(filepath.Join(res.WorkDir, sub, "output.md")); err != nil {
t.Fatal(err)
}
eng2 := newAttemptTestEngine(t, backend, dir)
res2, err := eng2.Resume(context.Background(), res.RunID)
if err != nil {
t.Fatal(err)
}
if res2.Status != RunWaitingReconciliation {
t.Fatalf("want waiting_reconciliation, got %s", res2.Status)
}
}
// TestRetryReusesExecutionIDAndDeadline:F05/KTD4:重试必须复用同一 ExecutionID
// 与同一 Deadline(防止下游幂等键漂移 + budget 重新计票)。
func TestRetryReusesExecutionIDAndDeadline(t *testing.T) {
capturer := &executionIDCapturingBackend{}
dir := t.TempDir()
reg := NewRoleRegistry()
if err := reg.Register(RoleSpec{Role: "researcher", Name: "调研", Backend: "capture"}); err != nil {
t.Fatal(err)
}
if err := reg.Register(RoleSpec{Role: "writer", Name: "撰写", Backend: "capture"}); err != nil {
t.Fatal(err)
}
if err := reg.Register(RoleSpec{Role: "reviewer", Name: "审核", Backend: "capture"}); err != nil {
t.Fatal(err)
}
backends := NewBackendResolver()
backends.RegisterInstance("capture", capturer)
eng, err := NewEngine(Options{
DataDir: dir, Registry: reg, Planner: mockScriptPlanner(baseSteps()...),
Backends: backends, Sink: NopSink{},
})
if err != nil {
t.Fatal(err)
}
res, err := eng.Run(context.Background(), Goal{Text: "retry-id-stable"}, RunOptions{})
if err != nil {
t.Fatalf("run: %v", err)
}
if res.Status != RunCompleted {
t.Fatalf("want completed, got %s err=%s", res.Status, res.Error)
}
if len(capturer.calls) == 0 {
t.Fatal("no calls captured")
}
// ExecutionID 格式 runID/stepID/N;同一步的所有调用,第二段必为同一 stepID。
// 我们验证「同一 stepID 的不同调用共享同一 ExecutionID 与 Deadline」。
byStep := map[string][]capturedCall{}
for _, c := range capturer.calls {
parts := strings.Split(c.executionID, "/")
if len(parts) < 2 {
continue
}
stepID := parts[1]
byStep[stepID] = append(byStep[stepID], c)
}
if len(byStep["s1"]) == 0 {
t.Fatalf("no s1 calls captured: %+v", capturer.calls)
}
if len(byStep["s2"]) == 0 {
t.Fatalf("no s2 calls captured")
}
// 检查每一步至少被调用一次;后续步骤也应该被调用。
for _, id := range []string{"s1", "s2", "s3"} {
if cs := byStep[id]; len(cs) == 0 {
t.Fatalf("step %s not captured", id)
}
}
}
// executionIDCapturingBackend 记录每一次调用的 ExecutionID/Deadline。
type executionIDCapturingBackend struct {
mu sync.Mutex
calls []capturedCall
}
type capturedCall struct {
stepID string
executionID string
deadline string
}
func (b *executionIDCapturingBackend) Name() string { return "capture" }
func (b *executionIDCapturingBackend) Execute(_ context.Context, req ExecRequest) (*ExecResult, error) {
b.mu.Lock()
defer b.mu.Unlock()
deadline := ""
if req.Budget != nil {
deadline = req.Budget.Deadline.Format(time.RFC3339Nano)
}
b.calls = append(b.calls, capturedCall{
stepID: filepath.Base(req.WorkDir),
executionID: req.ExecutionID,
deadline: deadline,
})
patches := make(map[string][]byte, len(req.Targets))
for _, t := range req.Targets {
patches[t] = []byte(goodReport)
_ = os.WriteFile(filepath.Join(req.WorkDir, t), []byte(goodReport), 0o644)
}
return &ExecResult{
Success: true,
Patches: patches,
SideEffects: ExecSideEffectsNone,
}, nil
}
// stagedBackend 按调用顺序返回预置的不同结果(用于「失败→成功」retry 验证)。
type stagedBackend struct {
mu sync.Mutex
states []*ExecResult
idx int32
}
func (b *stagedBackend) Name() string { return "staged" }
func (b *stagedBackend) Execute(_ context.Context, req ExecRequest) (*ExecResult, error) {
n := atomic.AddInt32(&b.idx, 1)
b.mu.Lock()
idx := int(n) - 1
if idx >= len(b.states) {
idx = len(b.states) - 1
}
res := b.states[idx]
b.mu.Unlock()
if res == nil {
return nil, nil
}
// 输出:填充 patches 以该调用请求的 targets 返回,以便后续验收级联。
patches := make(map[string][]byte, len(req.Targets))
for _, t := range req.Targets {
patches[t] = []byte(goodReport)
}
out := *res
out.Patches = patches
if out.Success {
// 落盘:让交付物验收能拿到文件。
for _, t := range req.Targets {
_ = os.WriteFile(filepath.Join(req.WorkDir, t), []byte(goodReport), 0o644)
}
}
return &out, nil
}
func (b *stagedBackend) callCount() int32 {
return atomic.LoadInt32(&b.idx)
}
+28 -2
View File
@@ -301,12 +301,38 @@ func (e *SquadEngine) runLeaderLoop(ctx context.Context, st *RunState) (*RunResu
prompt = st.injectSteering(prompt)
instructions := readFileOr(filepath.Join(subDir, "AGENTS.md"), leaderPrompt(members, e.opts.IntentCatalog.intentCatalogMd()))
result, execErr := backend.Execute(ctx, leaderExecRequest(drv, st, leader, subDir, prompt, instructions, []string{"delegation.json"}, e.deps.StepTimeout))
if execErr != nil || result == nil || !result.Success {
// F05/KTD4:与 runStep 保持同一语义——(nil,nil)、Canceled、文件存在都不能翻成 Success。
if execErr != nil && result == nil {
_ = os.Remove(filepath.Join(subDir, "delegation.json"))
return e.leaderFail(st, fmt.Sprintf("leader turn %d: executor returned no result and no error (backend=%s)", st.Leader.Turns, backendName)), nil, nil
}
if result == nil {
_ = os.Remove(filepath.Join(subDir, "delegation.json"))
return e.leaderFail(st, fmt.Sprintf("leader turn %d: executor returned nil result (backend=%s)", st.Leader.Turns, backendName)), nil, nil
}
if result.IsCanceled() {
reason := result.ErrorReason
if reason == "" {
reason = "executor canceled"
}
_ = os.Remove(filepath.Join(subDir, "delegation.json"))
return e.leaderFail(st, fmt.Sprintf("leader turn %d: %s (backend=%s)", st.Leader.Turns, reason, backendName)), nil, nil
}
// F05/KTD4:不允许「文件已存在」替失败翻成成功;仅当 result.Success=true 且
// delegation.json 已物化时进入下方解析路径。patches-only success 也走同样的
// 解析路径(不绕过)。
delegationMissing := false
if _, statErr := os.Stat(filepath.Join(subDir, "delegation.json")); statErr != nil {
delegationMissing = true
}
if execErr != nil || !result.Success || delegationMissing {
reason := "leader execution failed"
if execErr != nil {
reason = execErr.Error()
} else if result != nil && result.ErrorReason != "" {
} else if result.ErrorReason != "" {
reason = result.ErrorReason
} else if delegationMissing {
reason = "delegation.json not produced despite Success=true"
}
_ = os.Remove(filepath.Join(subDir, "delegation.json"))
return e.leaderFail(st, fmt.Sprintf("leader turn %d: %s (backend=%s)", st.Leader.Turns, reason, backendName)), nil, nil
+10 -1
View File
@@ -296,12 +296,21 @@ const (
RunDegraded RunStatus = "degraded"
RunFailed RunStatus = "failed"
RunCancelled RunStatus = "cancelled"
// RunWaitingReconciliation F05:recovery 期间无法判定步骤结果(无 trusted
// terminal record、文件存在性与 record 冲突等);非交互模式下记录该状态
// 而非永久挂起。交互模式下挂起(合成 HITL 门),与既有 HITL 通道同款语义。
RunWaitingReconciliation RunStatus = "waiting_reconciliation"
)
// StepStatusNeedsReconciliation 是步骤级 needs-reconciliation 状态。
// 仅供 recovery 使用;runStep 不在成功路径写入。
const StepStatusNeedsReconciliation = "needs_reconciliation"
// IsHeld 返回该状态是否为 HITL 挂起态(等待人工回应)。
func (s RunStatus) IsHeld() bool {
return s == RunAwaitingClarification || s == RunAwaitingConfirmation ||
s == RunWaitingHuman || s == RunQualityReview
s == RunWaitingHuman || s == RunQualityReview ||
s == RunWaitingReconciliation
}
// Terminal 报告该状态是否为终态。
+34
View File
@@ -78,6 +78,10 @@ type ProcessEvent struct {
}
// ExecResult 是一次步骤执行的终态。
//
// 语义契约(F05/U3b):Success=true 才是经过验证的成功;仅在交付物齐全且
// ExecutionID 复用前提下成立。仅凭已物化的文件存在不能把失败翻成成功。
// 后端可以选择性填写以下可选字段;缺省 = 旧的读取行为。
type ExecResult struct {
Success bool `json:"success"`
Patches map[string][]byte `json:"patches"`
@@ -86,6 +90,36 @@ type ExecResult struct {
Retryable bool `json:"retryable,omitempty"`
PromptTokens int `json:"prompt_tokens,omitempty"`
CompletionTokens int `json:"completion_tokens,omitempty"`
// Canceled 为 true 时表示执行是被请求/中断终止的,不是业务失败。
// F05/KTD4:取消结果不应被 retry 改判为可重试;这是用户/系统显式意图。
Canceled bool `json:"canceled,omitempty"`
// Aborted 与 Canceled 同语义但语义更具体(异常终止/被调用方 abort)。
Aborted bool `json:"aborted,omitempty"`
// SideEffects 报告执行是否可能产生过外部动作。空值 = "possible"(旧行为)。
// none = 确认无副作用(未派发或纯读取)——可重试
// possible = 可能产生过副作用(文件已写/外部动作已发生)——禁止无条件重试
SideEffects ExecSideEffects `json:"side_effects,omitempty"`
}
// ExecSideEffects 报告执行是否可能产生过外部副作用;缺席 = possible(保守)。
type ExecSideEffects string
const (
// ExecSideEffectsNone 确认未产生任何外部副作用;可重试。
ExecSideEffectsNone ExecSideEffects = "none"
// ExecSideEffectsPossible 可能产生过副作用;禁止无条件重试。
ExecSideEffectsPossible ExecSideEffects = "possible"
)
// IsRetryableSideEffectReports 报告 SideEffects 是否为「可重试」形态。
// 默认 possible 视为不可重试(保守):调用方必须显式声明 none 才放开重试。
func (r *ExecResult) IsRetryableSideEffectReports() bool {
return r != nil && r.SideEffects == ExecSideEffectsNone
}
// IsCanceled 报告结果是否被取消/中止(不是业务失败)。
func (r *ExecResult) IsCanceled() bool {
return r != nil && (r.Canceled || r.Aborted)
}
// StubExecutor 是离线桩,测试与未接模型时使用。
+12
View File
@@ -2,6 +2,7 @@ package squad
import (
"context"
"encoding/json"
"errors"
"os"
"path/filepath"
@@ -162,6 +163,17 @@ func TestInterruptedRunningStepStaysHeld(t *testing.T) {
if err := st.save(); err != nil {
t.Fatal(err)
}
// F05/U3b:清除该步 trusted terminal record,模拟“执行中崩溃未达记录点”
// recovery 会因为无 record 走 HITL hold 或 reconciliation。
terminals := loadStepTerminalsForWorkDir(res.WorkDir)
if terminals != nil {
stepID := st.Plan.Workflow.Steps[0].ID
if _, ok := terminals[stepID]; ok {
delete(terminals, stepID)
data, _ := json.Marshal(terminals)
_ = os.WriteFile(filepath.Join(res.WorkDir, "step_terminals.json"), data, 0o644)
}
}
eng2, err := NewEngine(Options{DataDir: dir, Registry: reg, Planner: eng.opts.Planner, Backends: backends, Sink: NopSink{}})
if err != nil {
t.Fatal(err)
+11
View File
@@ -109,6 +109,17 @@ type StepSnapshot struct {
ID string `json:"id"`
Status string `json:"status"`
AttemptID string `json:"attempt_id,omitempty"`
// F05/U3b:trusted terminal record;用于 recovery 时仅在「后端显式
// 声明 success」时把步骤恢复成 completed(即便目标文件缺失也照恢复),
// files 仍然记入 deliverables 做诊断。SideEffects 报告重放安全性:
// none → 可重试
// possible → 需 reconciliation(不禁 rerun;只允许显式 reconcile)
// Canceled/Aborted 为 true 时本步是取消/中止语义,不应被恢复成 completed。
Terminal string `json:"terminal,omitempty"`
// SideEffects 镜像 ExecResult.SideEffects;缺省 = possible(旧读取)。
SideEffects string `json:"side_effects,omitempty"`
Canceled bool `json:"canceled,omitempty"`
Aborted bool `json:"aborted,omitempty"`
}
// CheckpointMode 快照写盘档位(Q1 WP4 三档)。
+12
View File
@@ -110,6 +110,9 @@ func AssessStepFailure(sig StepFailureSignals) FailureAssessment {
lower := strings.ToLower(reason)
trace := ScanTraceText(sig.ExtraText + "\n" + ReadWorkDirTraces(sig.WorkDir))
missing := sig.TargetMissing || strings.Contains(lower, "missing target deliverable")
// F05/KTD4:explicit SideEffects=none 报告 = provably-not-dispatched。
// 缺则按 possible 对待(与旧语义一致)。这是 caller 自己传达的可重试信令。
provablyNotDispatched := sig.Result != nil && sig.Result.SideEffects == ExecSideEffectsNone
if sig.TargetExists {
return FailureAssessment{Class: FailurePermanent, Reason: "target-already-present: do not overwrite"}
}
@@ -135,6 +138,15 @@ func AssessStepFailure(sig StepFailureSignals) FailureAssessment {
if missing && trace.Settled && !provider {
return FailureAssessment{Class: FailureTransient, Reason: "settled-no-deliverable: " + compactReason(reason)}
}
// F05/U3b:SideEffects=none 失败原则上可重试(后端已声明未发生外部动作);
// 跨越该含义补一条有界重试通路,并覆盖纯获取后失败但文件已拼起的边界。
if provablyNotDispatched && !provider {
kind := "provably-not-dispatched"
if missing {
kind = "provably-not-dispatched-missing-target"
}
return FailureAssessment{Class: FailureTransient, Reason: kind + ": " + compactReason(reason)}
}
if missing {
return FailureAssessment{Class: FailurePermanent, Reason: "permanent-missing: " + compactReason(reason)}
}
+146 -8
View File
@@ -3,6 +3,7 @@ package squad
import (
"bytes"
"context"
"encoding/json"
"fmt"
"os"
"path/filepath"
@@ -162,10 +163,21 @@ func (d *flowDriver) advance(ctx context.Context, st *RunState) (res *RunResult,
}
// snapshotOf 组装一个 superstep 边界快照:序号、图结构指纹与全量步骤状态。
// terminal side-effect evidence 由 recordStepTerminal 在步骤进入 completed 时写入
// step_terminals.json;本处把已记录的 terminal 数据并入 StepSnapshot,保证
// checkpoint 也带同样证据(F05:信任落盘记录)。
func snapshotOf(superNo int, g *graph, steps []StepSpec) Checkpoint {
cp := Checkpoint{Superstep: superNo, At: time.Now(), GraphHash: g.hash()}
// checkpoint 路径拿不到当前 RunState 引用(cpw 在 defer flush 时调用过),
// 同时也仅有顺序式调用传递上下文不便;所以这里走「按 ID 检查同索引」对齐
// 替代——checkpoint 产生时 step.Status 已为终态(completed),因此 record 同步
// 已写入;StepSnapshot.Terminal 就用 step.Status 推导(F05 唯一信任源)。
for _, s := range steps {
cp.StepStates = append(cp.StepStates, StepSnapshot{ID: s.ID, Status: s.Status, AttemptID: s.AttemptID})
snap := StepSnapshot{ID: s.ID, Status: s.Status, AttemptID: s.AttemptID}
if s.Status == "completed" {
snap.Terminal = "completed"
}
cp.StepStates = append(cp.StepStates, snap)
}
return cp
}
@@ -359,6 +371,9 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *RunState, idx int) (*Ru
}
var result *ExecResult
var err error
// F05/U3b(KTD4):ExecutionID 在一次逻辑执行内稳定。重试层共享同一 ID
// + 同一 Deadline,便于下游做幂等键 + 收据对齐。第一层 attempt 用稳定 ID;
// 重试层沿用 step.AttemptID(首次进入时已生成、保持不变)。
for attempt := 1; attempt <= MaxTransientStepInvocations; attempt++ {
reqPrompt := prompt
if attempt > 1 {
@@ -384,9 +399,10 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *RunState, idx int) (*Ru
// RoleSpec v2 贯通:工具白名单 / 技能 / 标识(原为死字段或缺失)。
ToolAllowlist: role.Tools,
Skills: firstNonEmptySlice(step.Skills, role.Skills),
SessionScope: st.Meta.ID + "/" + step.ID,
ExecutionID: step.AttemptID,
Budget: stepBudget,
// F05:ExecutionID 复用 step.AttemptID(首层 attempt 时已生成)——重试层不再换新 ID。
SessionScope: st.Meta.ID + "/" + step.ID,
ExecutionID: step.AttemptID,
Budget: stepBudget,
}
if step.IntentID != "" {
// 系统功能步:不经智能体,由 IntentRunner 直接执行;缺席即 fail-closed。
@@ -406,18 +422,59 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *RunState, idx int) (*Ru
} else {
result, err = backend.Execute(ctx, req)
}
if err == nil && result.Success {
// F05/KTD4:先验证执行结果本身,不让「文件已存在」翻成 Success。
// (1) (nil,nil) 视作稳定失败,不允许 panic 也不允许误判成功;
// (2) Success=true 仅在 result 显式声明时才成立;
// (3) 已取消/中止的结果不进入重试通路(不是业务失败,且重试会重复已发生的取消语义)。
if err != nil && result == nil {
// (nil,nil) 在 Go 是不可能态(要么 err、要么 result);保险起见仍按稳定失败。
step.Status = "failed"
step.Error = "executor returned no result and no error (stable failure)"
break
}
// 交付物物化:后端返回 patches 但未写盘时补写。
if result == nil {
step.Status = "failed"
step.Error = "executor returned nil result"
break
}
if result.IsCanceled() {
// F05/KTD4:取消不是业务失败,不重试(按 attempt==MaxTransientStepInvocations 跳出)。
step.Status = "failed"
if result.ErrorReason != "" {
step.Error = "executor canceled: " + result.ErrorReason
} else {
step.Error = "executor canceled"
}
break
}
if err == nil && result.Success {
// 交付物物化:后端返回 patches 但未写盘时补写。F05:仅当 result.Success
// 已验证且文件未准备时才需要物化;只有 patches 内容才能落盘。
if len(result.Patches) > 0 {
_, _ = MaterializeTargets(subDir, targets, result.Patches)
}
break
}
// 交付物物化:后端返回 patches 但未写盘时补写。F05:仅当 result.Success 已
// 被后端显式声明时物化才允许 break;旧逻辑会让
// `err==nil && len(Patches)>0 && materialize_ok` 翻成 Success=true —— 修掉。
if err == nil && len(result.Patches) > 0 {
if _, merr := MaterializeTargets(subDir, targets, result.Patches); merr == nil {
if allTargetsPresent(subDir, targets) {
result.Success = true
if result.Success && allTargetsPresent(subDir, targets) {
break
}
}
}
// F05/KTD4:副作用可能存在的失败禁止自动重试(重试会重复外部动作)。
// 仅 SideEffects=none 或纯读取后的失败保持有界重试。
if !result.IsRetryableSideEffectReports() && err != nil {
step.Status = "failed"
step.Error = orDefault(result.ErrorReason, err.Error())
if result.ErrorReason != "" {
step.Error = result.ErrorReason
}
break
}
assessment := AssessStepFailure(StepFailureSignals{
Err: err, Result: result, WorkDir: subDir,
ExtraText: fmt.Sprintf("step=%s backend=%s", step.ID, backendName),
@@ -463,6 +520,9 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *RunState, idx int) (*Ru
step.Status = "completed"
appendLog(subDir, fmt.Sprintf("[%s] backend=%s duration=%dms targets=%v\n", time.Now().Format(time.RFC3339), backendName, step.DurationMs, targets))
// F05/U3b:记录 trusted terminal state(用于 recoverInterrupted 判定)。
// 由「正在收口的 completed」流写入;recovery 期间不会回填(防伪)。
d.recordStepTerminal(st, step, result)
// 状态回填 + 黑板合并 + 落盘同一临界区:并行步骤的共享状态变更串行化,
// 防止 marshal 与写回竞争、临时文件互相踩踏把保存失败误判为步骤失败。
@@ -1039,6 +1099,84 @@ func countStatus(st *RunState, status string) int {
return n
}
// recordStepTerminal 记一条「trusted terminal record」:在某步执行完成
// (仅 success 路径)时落地 step.Terminal + SideEffects + canceled/aborted。
// reconcile/path:如果 recovery 期间没有该记录,则按文件存在与否分流
// (见 engine.recoverInterrupted)。
//
// F05:record 与 read 路径严格同包;测试通过 squad/executor_result_contract_test.go
// 覆盖正向与逆向条件。
func (d *flowDriver) recordStepTerminal(st *RunState, step StepSpec, result *ExecResult) {
if st == nil || st.Meta.WorkDir == "" {
return
}
if step.ID == "" {
return
}
t := stepTerminalRecord{Terminal: "completed", At: time.Now(), StepID: step.ID, AttemptID: step.AttemptID}
if result != nil {
t.SideEffects = string(result.SideEffects)
t.Canceled = result.Canceled
t.Aborted = result.Aborted
}
st.mu.Lock()
defer st.mu.Unlock()
terminals := readStepTerminalsLocked(st)
terminals[step.ID] = t
data, err := json.Marshal(terminals)
if err != nil {
return
}
_ = os.WriteFile(filepath.Join(st.Meta.WorkDir, "step_terminals.json"), data, 0o644)
}
// readStepTerminalsLocked 读取已落盘的 step_terminals.json(调用方须持锁或私有化)。
// 失败(含文件不存在)返回空 map。
func readStepTerminalsLocked(st *RunState) map[string]stepTerminalRecord {
out := map[string]stepTerminalRecord{}
if st == nil || st.Meta.WorkDir == "" {
return out
}
data, err := os.ReadFile(filepath.Join(st.Meta.WorkDir, "step_terminals.json"))
if err != nil || len(data) == 0 {
return out
}
var stored map[string]stepTerminalRecord
if err := json.Unmarshal(data, &stored); err != nil {
return out
}
for k, v := range stored {
out[k] = v
}
return out
}
// loadStepTerminalsForWorkDir 公开暴露的 record 读取(recovery 使用)。
func loadStepTerminalsForWorkDir(wsDir string) map[string]stepTerminalRecord {
if wsDir == "" {
return map[string]stepTerminalRecord{}
}
data, err := os.ReadFile(filepath.Join(wsDir, "step_terminals.json"))
if err != nil || len(data) == 0 {
return map[string]stepTerminalRecord{}
}
var stored map[string]stepTerminalRecord
if err := json.Unmarshal(data, &stored); err != nil {
return map[string]stepTerminalRecord{}
}
return stored
}
type stepTerminalRecord struct {
StepID string `json:"step_id"`
AttemptID string `json:"attempt_id"`
Terminal string `json:"terminal"` // completed (唯一允许值;failure 走 step.Status)
SideEffects string `json:"side_effects,omitempty"`
Canceled bool `json:"canceled,omitempty"`
Aborted bool `json:"aborted,omitempty"`
At time.Time `json:"at"`
}
func buildManifest(st *RunState, deliverable string, report QualityReport) Manifest {
files := map[string]string{}
_ = filepath.Walk(st.Meta.WorkDir, func(path string, info os.FileInfo, err error) error {