From cf9a0d4d757a07935e0ea0949ff97add8108142b Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 21 Oct 2026 06:00:51 +0000 Subject: [PATCH 1/3] chore(task): start fix/resolve-hitl-run-status-ordering --- .task/fix-resolve-hitl-run-status-ordering.md | 66 +++++++++++++++++++ 1 file changed, 66 insertions(+) create mode 100644 .task/fix-resolve-hitl-run-status-ordering.md diff --git a/.task/fix-resolve-hitl-run-status-ordering.md b/.task/fix-resolve-hitl-run-status-ordering.md new file mode 100644 index 0000000..497b358 --- /dev/null +++ b/.task/fix-resolve-hitl-run-status-ordering.md @@ -0,0 +1,66 @@ +# fix/resolve-hitl-run-status-ordering 任务卡 + +任务主记录：。 +状态：进行中；owner：weichen + +## 基线与授权 + +- 权威仓：ssh://git-ssh.u1680832.nyat.app:31411/weichen/Agent-System-Backend.git；root_target：main；immediate_target：main +- branch_kind：standalone；initiative_id： +- R0（initial_main_sha）：07228aa38139d974e88573a84ad992fd66166168；工作区：/home/weichen/projects/Agent-System-Backend-wt-fix-resolve-hitl-run-status-ordering（worktree）；分支：fix/resolve-hitl-run-status-ordering +- 父任务信息（仅子任务填写）：无 +- accepting_requirements：false +- 需求分流：verdict=NEW；selected_initiative=无；候选=无；理由=no accepting initiative；triage 收据=/tmp/be-triage-runhitl-20261021.json；receipt_sha256=39dc1067007f31d2871e03c86a389f882925d5279198b0d3bdf9d8ccd253a6f7；requirement_sha256=2756850b023061065b52aac9b8f318a9802d56bdc97882ac24b749e940ea8559 +- 授权与交付终点：<本次允许的动作与终点；未授权事项写明> +- 实施执行：agent-per-unit（配置：AGENT-EXECUTION.md；外部 Agent 一单元一新进程/会话/工作区） + +## 目标与非目标 + +<预期结果、受影响入口/模块/契约；明确不做什么。> + +## Agent 实施收据 + +按根目录 AGENT-EXECUTION.md 路由；外部 Agent 一单元一新进程/会话/隔离工作区。实施、修复与测试必须通过登记的候选执行；未配置或候选安全耗尽停在 unavailable/blocked，禁止 host 隐式代写。 + +| 单元 ID | 预期范围 | 角色 | 候选顺序 | attempt | CLI/模型 | 终态 | transport | fallback | 源码外收据路径 | +|---|---|---|---|---|---|---|---|---|---| +| 待定义 | — | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | NOT_RUN | — | + +## 复用材料（task-start 自动检索结果摘录） + +- 相关近期提交： 命中；无则写无>。 +- 相关经验条目：<检索命中；无则写无>。 + +## 同步记录 + +| 节点 | sync_base_sha | tested_head_sha | checked_at | 结果 | +|---|---|---|---|---| +| 启动 | | | <日期时间> | <一致/已同步> | +| 验收前 | <填写> | <填写> | <填写> | <填写> | + +字段没有发生的写"无"，不省略。本卡与 PR 描述同源：开 PR 时将"目标与非目标、结果与验证、影响与剩余项"复制进 PR 描述，不另写第二份。 + +## 交付调度（有多个就绪任务时） + +- 调度记录 / entry_id / 当前 attempt_id：<平台或调度服务链接及标识；未入队写无>。 +- 入队源 HEAD / owner / 依赖：<完整 SHA、身份、依赖或无>。 +- 固定候选引用：<调度尝试中的 base/head、组件与配置；未固定写无>。 +- 窗口、最终 CI、Review 与真实合入结果：<引用调度记录及原始证据，不复制全队列>。 + +定义见 [合入调度与 CI 分配](../standards/merge-queue.md)。本段在候选固定前整理；固定后先使用源码之外的记录保存运行结果，再按项目闭合流程回填。 + +## 结果与验证（完成后填写） + +- 结果：<交付了什么；未发生写无>。 +- 验证：<结论性摘要：PASS/FAIL/SKIP/NOT_RUN 逐项；逐条证据见下表，来源标注与写入规则见 [按影响选择验证与证据](../standards/verification.md)>。 +- 影响与剩余项：<影响面、遗留风险、后续任务>。 + +| 时间(UTC) | 命令 | 被测 HEAD | exit | 来源 | 摘要/位置 | +|---|---|---|---|---|---| +| <时间> | <命令；由项目 evidence 工具写入，无工具时附命令+对象 SHA+输出 digest 并标 SELF> | | | GATE/REPLAY/SELF | <摘要/位置> | + +## 回收（task-finish） + +- 合入核对：<分支已包含于目标分支的核对输出>。 +- 未跟踪/ignored 文件处置：<归档/可再生/存疑 逐项；无则写无>。 +- 回收动作：。 -- 2.54.0 From 789f67f3173d3cbaf2cb02a40590547186f9de9d Mon Sep 17 00:00:00 2001 From: agent-exec-transport Date: Wed, 21 Oct 2026 07:08:29 +0000 Subject: [PATCH 2/3] agent-exec transport d7af319ad1e36046a46be5ad16007582/d3f2d0786d5b --- internal/runmgr/runmgr.go | 40 +++- internal/runmgr/runmgr_test.go | 238 +++++++++++++++++++++++ internal/store/messages.go | 31 +++ internal/store/messages_test.go | 120 ++++++++++++ internal/store/run_account_write.go | 37 ++++ internal/store/run_account_write_test.go | 105 ++++++++++ 6 files changed, 561 insertions(+), 10 deletions(-) create mode 100644 internal/runmgr/runmgr_test.go create mode 100644 internal/store/messages_test.go create mode 100644 internal/store/run_account_write_test.go diff --git a/internal/runmgr/runmgr.go b/internal/runmgr/runmgr.go index 2141f38..1e708f6 100644 --- a/internal/runmgr/runmgr.go +++ b/internal/runmgr/runmgr.go @@ -120,6 +120,10 @@ type Manager struct { // genuiAuto：Generative UI 自动建议闸门（nil = 未接线；默认关闭行为）。 // 仅由任务阶段/材料工件语义事件调用；present/surface/action 不得接入。 genuiAuto GenUIAutoTrigger + + // hitlAnsweredHook 仅测试接缝：HITL 答案交给 runner 之后同步一次，供用例 + // 确定性构造"runner 抢先结算"的交错。生产装配不注入（nil = 直通）。 + hitlAnsweredHook func(runID core.ID) } // GenUIAutoTrigger 自动建议页面闸门（由 genui.Auto 实现）。 @@ -2062,6 +2066,15 @@ func (m *Manager) setRunStatus(r *core.Run, status core.RunStatus, phase core.Ru return m.st.UpdateRun(r) } +// resumeRunFromWaitingHITL 把等待人工输入的 run 推进回 running：条件更新， +// 由 store 报告是否命中（见 store.ResumeRunFromWaitingHITL）。 +func (m *Manager) resumeRunFromWaitingHITL(r *core.Run) (bool, error) { + if m.accountEffectiveAuthorization != nil { + return m.st.ResumeRunFromWaitingHITLAccountChecked(string(r.Owner), r, m.accountAllowImplicitLocal) + } + return m.st.ResumeRunFromWaitingHITL(r) +} + // emit 经批处理器持久化事件并广播；返回 seq（缓冲中的增量返回 0）。 func (m *Manager) emit(conversationID, runID string, typ core.StreamEventType, payload any) (int64, error) { return m.events.emit(conversationID, runID, typ, payload) @@ -2240,21 +2253,28 @@ func (m *Manager) ResolveHITL(owner, cardID string, resp core.HITLResponse) (piv if !ok { return "", false, fmt.Errorf("runmgr: run %s not active (may have ended)", run.ID) } + // 恢复运行状态必须先于把答案交给 runner（U7 双实例验收实测缺陷修复）： + // 反序时 runner 立即被唤醒并抢先结算，写下 completed/ended_at，随后这里的 + // 迟到写把行覆盖回 running/ended_at=NULL——run 永远卡在 running 且无执行者， + // 会话永不释放，之后每次提交都是 409。前置 + 条件更新后，交出答案之后本函数 + // 不再写 run 行，runner 是该行此后唯一的写者。 + if run.Status == core.RunWaitingHITL { + applied, err := m.resumeRunFromWaitingHITL(&run) + if err != nil { + return "", false, err + } + if !applied { + // 行已被 runner 或取消路径推进：不改写，只记录，交由权威写者收口。 + log.Printf("[runmgr] run %s already advanced past waiting_hitl on resolve; resume skipped", run.ID) + } + } select { case h.hitlCh <- hitlAnswer{resp: resp}: default: return "", false, fmt.Errorf("runmgr: hitl channel busy") } - // 恢复运行状态。 - if run.Status == core.RunWaitingHITL { - run.Status = core.RunRunning - if m.accountEffectiveAuthorization != nil { - if err := m.st.UpdateRunAccountChecked(string(run.Owner), &run, m.accountAllowImplicitLocal); err != nil { - return "", false, err - } - } else if err := m.st.UpdateRun(&run); err != nil { - return "", false, err - } + if m.hitlAnsweredHook != nil { + m.hitlAnsweredHook(run.ID) } return "", false, nil } diff --git a/internal/runmgr/runmgr_test.go b/internal/runmgr/runmgr_test.go new file mode 100644 index 0000000..658f2ce --- /dev/null +++ b/internal/runmgr/runmgr_test.go @@ -0,0 +1,238 @@ +package runmgr + +import ( + "path/filepath" + "testing" + "time" + + "agent-system-backend/internal/bus" + "agent-system-backend/internal/profiles" + "agent-system-backend/internal/runtime" + "agent-system-backend/internal/settings" + "agent-system-backend/internal/store" + "agent-system-backend/internal/tools" + core "agent-system-core" +) + +// hitlOrderRunner 问 rounds 轮 HITL 后正常返回，让 runmgr 走完结算。 +type hitlOrderRunner struct { + rounds int + asked chan struct{} +} + +func (*hitlOrderRunner) Name() string { return "hitl-order" } + +func (r *hitlOrderRunner) Execute(rc runtime.RunCtx) error { + for i := 0; i < r.rounds; i++ { + // AskHITL 在拿到答案前不返回：先报"已提问"，卡与 waiting_hitl 行由 runmgr 落库。 + r.asked <- struct{}{} + if _, err := rc.AskHITL(core.HITLCard{ + Question: "继续？", + Kind: core.HITLKindChoice, + Choices: []core.HITLChoice{{ID: "ok", Label: "继续"}}, + AllowFreeInput: true, + }); err != nil { + return err + } + } + return rc.SetAssistantParts([]core.Part{{Type: core.PartText, Text: "完成"}}) +} + +func newHITLOrderEnv(t *testing.T, runner runtime.Runner) (*Manager, *store.Store, core.Conversation) { + t.Helper() + st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "hitl-order.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { st.Close() }) + b := bus.New() + reg := tools.NewRegistry() + if err := reg.RegisterBuiltin(); err != nil { + t.Fatal(err) + } + proj, err := st.CreateProject(store.OwnerDefault, "p") + if err != nil { + t.Fatal(err) + } + conv, err := st.CreateConversation(store.OwnerDefault, proj.ID, "c", "") + if err != nil { + t.Fatal(err) + } + mgr := New(st, b, reg, profiles.New(st), settings.NewService(st), runner, runner, runner, runner) + t.Cleanup(mgr.Shutdown) + return mgr, st, conv +} + +func startHITLOrderRun(t *testing.T, mgr *Manager, st *store.Store, conv core.Conversation) core.Run { + t.Helper() + user := core.Message{ConversationID: conv.ID, Role: core.RoleUser, + Parts: []core.Part{{Type: core.PartText, Text: "开工"}}} + if err := st.AppendMessage(&user); err != nil { + t.Fatal(err) + } + run, err := mgr.StartConversationRun(conv, user, nil, nil) + if err != nil { + t.Fatal(err) + } + return run +} + +// pendingCard 返回该 run 当前未决的 HITL 卡（卡 ID 由 runmgr 生成，runner 侧拿不到）。 +func pendingCard(t *testing.T, st *store.Store, conv core.Conversation, round int) core.HITLCard { + t.Helper() + cards, err := st.ActiveHITLCards(conv.Owner, conv.ID) + if err != nil { + t.Fatal(err) + } + unresolved := []core.HITLCard{} + for _, c := range cards { + if !c.Resolved { + unresolved = append(unresolved, c) + } + } + if len(unresolved) == 0 { + t.Fatalf("round %d: no unresolved hitl card", round+1) + } + return unresolved[len(unresolved)-1] +} + +// runState 渲染 run 行（含失败原因），只用于用例诊断。 +func runState(t *testing.T, st *store.Store, runID core.ID) string { + t.Helper() + got, err := st.GetRunByID(runID) + if err != nil { + return "read failed: " + err.Error() + } + msg := "status=" + string(got.Status) + if got.Error != nil { + msg += " error=" + got.Error.Code + ": " + got.Error.Message + } + return msg +} + +// runTerminalWithin 轮询到 run 进入终态（bounded wait，非定时睡眠）。 +func runTerminalWithin(t *testing.T, st *store.Store, runID core.ID, bound time.Duration) core.Run { + t.Helper() + var last core.Run + waitUntilWithin(t, bound, func() bool { + got, err := st.GetRunByID(runID) + if err != nil { + t.Fatal(err) + } + last = got + return got.Status == core.RunCompleted || got.Status == core.RunFailed || got.Status == core.RunCancelled + }) + return last +} + +// ResolveHITL 的写序回归：答案交给 runner 之后 ResolveHITL 不得再落库 run 状态。 +// 命中旧实现（交出答案 → 再写 running）时，runner 抢先结算的 completed/ended_at +// 会被这次迟到写覆盖回 running/ended_at=NULL，run 永远卡住、会话永不释放 +// （Dev 双实例验收：'run not completed within 15s' / 'conversation still busy'）。 +func TestResolveHITLRunnerSettlementSurvivesResolveReturn(t *testing.T) { + runner := &hitlOrderRunner{rounds: 1, asked: make(chan struct{}, 1)} + mgr, st, conv := newHITLOrderEnv(t, runner) + run := startHITLOrderRun(t, mgr, st, conv) + + select { + case <-runner.asked: + case <-time.After(10 * time.Second): + t.Fatalf("runner did not reach the HITL question: %s", runState(t, st, run.ID)) + } + waitUntilWithin(t, 10*time.Second, func() bool { + got, err := st.GetRunByID(run.ID) + return err == nil && got.Status == core.RunWaitingHITL + }) + card := pendingCard(t, st, conv, 0) + if got, err := st.GetRunByID(run.ID); err != nil || got.Status != core.RunWaitingHITL { + t.Fatalf("run status=%s err=%v (want waiting_hitl)", got.Status, err) + } + + // 同步点：答案交付之后阻塞到 runner 结算落库，再让 ResolveHITL 收尾。 + // 旧实现在这之后写 running（覆盖终态），新实现此处已无写入。 + mgr.hitlAnsweredHook = func(id core.ID) { + runTerminalWithin(t, st, id, 10*time.Second) + } + type resolveResult struct { + pivot string + turned bool + err error + } + done := make(chan resolveResult, 1) + go func() { + pivot, turned, err := mgr.ResolveHITL(conv.Owner, string(card.ID), core.HITLResponse{ChoiceID: "ok"}) + done <- resolveResult{pivot, turned, err} + }() + var res resolveResult + select { + case res = <-done: + case <-time.After(20 * time.Second): + t.Fatal("ResolveHITL did not return") + } + if res.err != nil || res.pivot != "" || res.turned { + t.Fatalf("resolve = %+v", res) + } + + final, err := st.GetRunByID(run.ID) + if err != nil { + t.Fatal(err) + } + if final.Status != core.RunCompleted { + t.Fatalf("run status = %s, want completed（迟到的 resolve 写覆盖了 runner 的终态）", final.Status) + } + if final.EndedAt == nil { + t.Fatal("ended_at 被迟到写抹掉：run 会永远卡在 running") + } + mgr.convMu.Lock() + busy := mgr.convActive[conv.ID] + mgr.convMu.Unlock() + if busy { + t.Fatal("conversation still in flight after the run reached a terminal status") + } +} + +// 第二轮 HITL：第 2 轮的 waiting_hitl 行同样不能被任何迟到写改回 running， +// 且第 2 轮结算的终态必须留存。 +func TestResolveHITLSecondRoundKeepsTerminalStatus(t *testing.T) { + runner := &hitlOrderRunner{rounds: 2, asked: make(chan struct{}, 2)} + mgr, st, conv := newHITLOrderEnv(t, runner) + run := startHITLOrderRun(t, mgr, st, conv) + + select { + case <-runner.asked: + case <-time.After(10 * time.Second): + t.Fatalf("runner did not reach the first HITL question: %s", runState(t, st, run.ID)) + } + waitUntilWithin(t, 10*time.Second, func() bool { + got, err := st.GetRunByID(run.ID) + return err == nil && got.Status == core.RunWaitingHITL + }) + first := pendingCard(t, st, conv, 0) + if _, _, err := mgr.ResolveHITL(conv.Owner, string(first.ID), core.HITLResponse{ChoiceID: "ok"}); err != nil { + t.Fatalf("resolve round 1: %v", err) + } + + select { + case <-runner.asked: + case <-time.After(10 * time.Second): + t.Fatal("runner did not reach the second HITL question") + } + second := pendingCard(t, st, conv, 1) + if got, err := st.GetRunByID(run.ID); err != nil || got.Status != core.RunWaitingHITL { + t.Fatalf("round 2 status=%s err=%v (want waiting_hitl)", got.Status, err) + } + mgr.hitlAnsweredHook = func(id core.ID) { + runTerminalWithin(t, st, id, 10*time.Second) + } + if _, _, err := mgr.ResolveHITL(conv.Owner, string(second.ID), core.HITLResponse{ChoiceID: "ok"}); err != nil { + t.Fatalf("resolve round 2: %v", err) + } + + final, err := st.GetRunByID(run.ID) + if err != nil { + t.Fatal(err) + } + if final.Status != core.RunCompleted || final.EndedAt == nil { + t.Fatalf("run status=%s ended=%v (want completed with ended_at)", final.Status, final.EndedAt) + } +} diff --git a/internal/store/messages.go b/internal/store/messages.go index 1800203..0bea24e 100644 --- a/internal/store/messages.go +++ b/internal/store/messages.go @@ -231,6 +231,37 @@ func (s *Store) UpdateRun(r *core.Run) error { return err } +// ResumeRunFromWaitingHITL 把仍在等待人工输入的 run 推进回 running。 +// +// 条件更新（status 仍为 waiting_hitl）是排序护栏，不是装饰：命中人工输入的 +// 行在这之后只能由 runner 推进，runner 的终态（completed/ended_at/usage/error） +// 不会被这里读到的旧快照覆盖。返回值区分三种结果：applied=true 本次推进； +// applied=false 行仍在但已被别的写者推进（runner 结算或取消），调用方不得改写； +// ErrNotFound 行不存在。 +func (s *Store) ResumeRunFromWaitingHITL(r *core.Run) (bool, error) { + if r == nil || r.ID == "" { + return false, ErrNotFound + } + res, err := s.db.Exec(s.q(`UPDATE runs SET status=? WHERE id=? AND status=?`), + string(core.RunRunning), r.ID, string(core.RunWaitingHITL)) + if err != nil { + return false, err + } + if n, err := res.RowsAffected(); err != nil { + return false, err + } else if n == 1 { + return true, nil + } + // 未命中：区分"行不存在"与"行已被推进"，两者对调用方含义不同。 + var one int + if err := s.db.QueryRow(s.q(`SELECT 1 FROM runs WHERE id=?`), r.ID).Scan(&one); errors.Is(err, sql.ErrNoRows) { + return false, ErrNotFound + } else if err != nil { + return false, err + } + return false, nil +} + const runColumns = `id,parent_run_id,conversation_id,project_id,owner,kind,profile_id,status,phase,agent_name,created_at,ended_at,usage,error` func scanRun(scan func(...any) error) (core.Run, error) { diff --git a/internal/store/messages_test.go b/internal/store/messages_test.go new file mode 100644 index 0000000..fb213c0 --- /dev/null +++ b/internal/store/messages_test.go @@ -0,0 +1,120 @@ +package store + +import ( + "errors" + "testing" + "time" + + core "agent-system-core" +) + +// settleRunFixture 把 run 写成终态（completed + ended_at/usage），模拟 runner 结算。 +func settleRunFixture(t *testing.T, st *Store, run core.Run, status core.RunStatus) core.Run { + t.Helper() + got, err := st.GetRunByID(run.ID) + if err != nil { + t.Fatal(err) + } + usage := core.Usage{PromptTokens: 7, CompletionTokens: 5, TotalTokens: 12} + now := time.Now().UTC() + got.Status = status + got.Phase = core.PhaseDone + got.Usage = &usage + got.EndedAt = &now + if err := st.UpdateRun(&got); err != nil { + t.Fatal(err) + } + return got +} + +// 条件推进的三个结果必须可区分：命中、已被别的写者推进（不写）、行不存在。 +func TestResumeRunFromWaitingHITLReportsOutcome(t *testing.T) { + st := openTestStore(t) + conv := seedConversation(t, st) + run := seedRun(t, st, conv, core.RunWaitingHITL) + + applied, err := st.ResumeRunFromWaitingHITL(&run) + if err != nil || !applied { + t.Fatalf("first resume: applied=%v err=%v", applied, err) + } + got, err := st.GetRunByID(run.ID) + if err != nil || got.Status != core.RunRunning { + t.Fatalf("resumed row: status=%s err=%v", got.Status, err) + } + + // 已推进但未终态（重复/跨实例转交）：不写、不报错。 + applied, err = st.ResumeRunFromWaitingHITL(&run) + if err != nil || applied { + t.Fatalf("second resume of an advanced row: applied=%v err=%v", applied, err) + } + + // 行不存在：与"已被推进"分开报告。 + missing := core.Run{ID: NewID("run"), ConversationID: conv.ID, Owner: conv.Owner} + applied, err = st.ResumeRunFromWaitingHITL(&missing) + if !errors.Is(err, ErrNotFound) || applied { + t.Fatalf("missing row: applied=%v err=%v", applied, err) + } + applied, err = st.ResumeRunFromWaitingHITL(&core.Run{}) + if !errors.Is(err, ErrNotFound) || applied { + t.Fatalf("empty id: applied=%v err=%v", applied, err) + } +} + +// 回归：终态行不得被条件推进覆盖，ended_at/usage 等审计字段必须留存。 +// 命中缺陷实现（无条件的整行写）时，这里就是 run 被拉回 running 的那一步。 +func TestResumeRunFromWaitingHITLCannotOverwriteTerminalRow(t *testing.T) { + for _, status := range []core.RunStatus{core.RunCompleted, core.RunFailed, core.RunCancelled} { + t.Run(string(status), func(t *testing.T) { + st := openTestStore(t) + conv := seedConversation(t, st) + run := seedRun(t, st, conv, core.RunWaitingHITL) + settled := settleRunFixture(t, st, run, status) + + applied, err := st.ResumeRunFromWaitingHITL(&settled) + if err != nil || applied { + t.Fatalf("resume touched a terminal row: applied=%v err=%v", applied, err) + } + after, err := st.GetRunByID(run.ID) + if err != nil { + t.Fatal(err) + } + if after.Status != status { + t.Fatalf("status = %s, want %s", after.Status, status) + } + if after.EndedAt == nil { + t.Fatal("ended_at wiped by the resume") + } + if after.Usage == nil || after.Usage.TotalTokens != 12 { + t.Fatalf("usage = %+v, want the runner's value", after.Usage) + } + if after.Phase != core.PhaseDone { + t.Fatalf("phase = %s, want done", after.Phase) + } + }) + } +} + +// 条件推进按当前行状态判定：runner 又问一轮时（新 waiting_hitl 行）仍可推进。 +func TestResumeRunFromWaitingHITLAppliesPerWaitingRound(t *testing.T) { + st := openTestStore(t) + conv := seedConversation(t, st) + first := seedRun(t, st, conv, core.RunWaitingHITL) + if applied, err := st.ResumeRunFromWaitingHITL(&first); err != nil || !applied { + t.Fatalf("round 1: applied=%v err=%v", applied, err) + } + settled := settleRunFixture(t, st, first, core.RunCompleted) + // 第二轮：runner 再问一次，行回到 waiting_hitl。 + round2 := settled + round2.Status = core.RunWaitingHITL + if err := st.UpdateRun(&round2); err != nil { + t.Fatal(err) + } + applied, err := st.ResumeRunFromWaitingHITL(&round2) + if err != nil || !applied { + t.Fatalf("round 2: applied=%v err=%v", applied, err) + } + got, err := st.GetRunByID(first.ID) + if err != nil || got.Status != core.RunRunning { + t.Fatalf("round 2 row: status=%s err=%v", got.Status, err) + } +} diff --git a/internal/store/run_account_write.go b/internal/store/run_account_write.go index f5d57b6..5ef2a91 100644 --- a/internal/store/run_account_write.go +++ b/internal/store/run_account_write.go @@ -34,6 +34,43 @@ func (s *Store) UpdateRunAccountChecked(actor string, r *core.Run, allowImplicit }) } +// ResumeRunFromWaitingHITLAccountChecked fences the HITL resume against account +// disable and keeps the same conditional guard as ResumeRunFromWaitingHITL: +// only a row still waiting for input advances, and the caller learns which case +// it hit (applied / already advanced by the runner / row missing). +func (s *Store) ResumeRunFromWaitingHITLAccountChecked(actor string, r *core.Run, allowImplicitDefault bool) (bool, error) { + if r == nil || actor == "" || string(r.Owner) != actor { + return false, ErrAccountForbidden + } + applied := false + err := s.withEffectiveAccountWrite(actor, allowImplicitDefault, func(tx *sql.Tx, _ Account) error { + var owner string + if err := tx.QueryRow(s.q(`SELECT owner FROM runs WHERE id=?`), r.ID).Scan(&owner); errors.Is(err, sql.ErrNoRows) { + return ErrNotFound + } else if err != nil { + return err + } + if owner != actor { + return ErrNotFound + } + res, err := tx.Exec(s.q(`UPDATE runs SET status=? WHERE id=? AND status=?`), + string(core.RunRunning), r.ID, string(core.RunWaitingHITL)) + if err != nil { + return err + } + n, err := res.RowsAffected() + if err != nil { + return err + } + applied = n == 1 + return nil + }) + if err != nil { + return false, err + } + return applied, nil +} + // CreateRunAccountChecked fences run admission against account disable. The // conversation's stored owner is checked in the same final transaction. func (s *Store) CreateRunAccountChecked(r *core.Run, allowImplicitDefault bool) error { diff --git a/internal/store/run_account_write_test.go b/internal/store/run_account_write_test.go new file mode 100644 index 0000000..afc90cb --- /dev/null +++ b/internal/store/run_account_write_test.go @@ -0,0 +1,105 @@ +package store + +import ( + "errors" + "testing" + + core "agent-system-core" +) + +func runAccountFixture(t *testing.T, st *Store, owner string) (core.Project, core.Conversation, core.Run) { + t.Helper() + for _, a := range []struct { + owner string + admin bool + }{{"admin", true}, {"alice", false}, {"bob", false}} { + if err := st.CreateAccount(a.owner, "", "", a.admin); err != nil { + t.Fatal(err) + } + } + project, err := st.CreateProject(owner, "resume") + if err != nil { + t.Fatal(err) + } + conv, err := st.CreateConversation(owner, project.ID, "resume", "") + if err != nil { + t.Fatal(err) + } + run := core.Run{ConversationID: conv.ID, ProjectID: project.ID, Owner: owner, + Kind: core.RunConversation, Status: core.RunWaitingHITL, Phase: core.PhaseThinking} + if err := st.CreateRun(&run); err != nil { + t.Fatal(err) + } + return project, conv, run +} + +// 账号围栏版条件推进：命中、已被推进、越权三者各自可辨，终态不被覆盖。 +func TestResumeRunFromWaitingHITLAccountCheckedOutcome(t *testing.T) { + st := openTestStore(t) + _, _, run := runAccountFixture(t, st, "alice") + + applied, err := st.ResumeRunFromWaitingHITLAccountChecked("alice", &run, false) + if err != nil || !applied { + t.Fatalf("owner resume: applied=%v err=%v", applied, err) + } + applied, err = st.ResumeRunFromWaitingHITLAccountChecked("alice", &run, false) + if err != nil || applied { + t.Fatalf("second resume: applied=%v err=%v", applied, err) + } + + settled := settleRunFixture(t, st, run, core.RunCompleted) + applied, err = st.ResumeRunFromWaitingHITLAccountChecked("alice", &settled, false) + if err != nil || applied { + t.Fatalf("terminal row resumed: applied=%v err=%v", applied, err) + } + after, err := st.GetRunByID(run.ID) + if err != nil { + t.Fatal(err) + } + if after.Status != core.RunCompleted || after.EndedAt == nil { + t.Fatalf("terminal row changed: status=%s ended=%v", after.Status, after.EndedAt) + } + + // 越权 actor、行不存在与“已被推进”三者分开报告。 + spoof := run + spoof.Owner = "bob" // 调用方自称 owner= bob，行 owner 仍是 alice。 + applied, err = st.ResumeRunFromWaitingHITLAccountChecked("bob", &spoof, false) + if !errors.Is(err, ErrNotFound) || applied { + t.Fatalf("foreign actor: applied=%v err=%v", applied, err) + } + foreign := run + foreign.ID = NewID("run") + applied, err = st.ResumeRunFromWaitingHITLAccountChecked("alice", &foreign, false) + if !errors.Is(err, ErrNotFound) || applied { + t.Fatalf("missing row: applied=%v err=%v", applied, err) + } + forged := run + forged.Owner = "bob" + applied, err = st.ResumeRunFromWaitingHITLAccountChecked("alice", &forged, false) + if !errors.Is(err, ErrAccountForbidden) || applied { + t.Fatalf("forged owner: applied=%v err=%v", applied, err) + } + if applied, err := st.ResumeRunFromWaitingHITLAccountChecked("", &run, false); !errors.Is(err, ErrAccountForbidden) || applied { + t.Fatalf("empty actor: applied=%v err=%v", applied, err) + } +} + +// 账号停用后不得再推进 run（与其它 per-table 围栏写同一约定）。 +func TestResumeRunFromWaitingHITLAccountCheckedStopsAfterDisable(t *testing.T) { + st := openTestStore(t) + _, _, run := runAccountFixture(t, st, "alice") + account, err := st.GetAccountByOwner("alice") + if err != nil { + t.Fatal(err) + } + if _, err := st.SetAccountEnabled("admin", account.ID, account.StateVersion, false); err != nil { + t.Fatal(err) + } + if applied, err := st.ResumeRunFromWaitingHITLAccountChecked("alice", &run, false); !errors.Is(err, ErrAccountForbidden) || applied { + t.Fatalf("disabled account resumed run: applied=%v err=%v", applied, err) + } + got, err := st.GetRunByID(run.ID) + if err != nil || got.Status != core.RunWaitingHITL { + t.Fatalf("disabled account changed the row: status=%s err=%v", got.Status, err) + } +} -- 2.54.0 From f39b68318833b8eccca9759db4d28b1b9a2269a8 Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 21 Oct 2026 07:12:13 +0000 Subject: [PATCH 3/3] Record task authorization and agent receipt for the run-status fix Why: the card must record the scope, the (r1 scope-expansion -> granted expansion -> r2 success) trajectory, and the rebase onto 241e1b3 before delivery. Evidence: /tmp/be-runhitl-impl-r2-20261021.json (pi, success, transport 789f67f); no PR/merge facts yet. Impact: card metadata only; no behaviour change. Refs: fix/resolve-hitl-run-status-ordering --- .task/fix-resolve-hitl-run-status-ordering.md | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/.task/fix-resolve-hitl-run-status-ordering.md b/.task/fix-resolve-hitl-run-status-ordering.md index 497b358..b8385b5 100644 --- a/.task/fix-resolve-hitl-run-status-ordering.md +++ b/.task/fix-resolve-hitl-run-status-ordering.md @@ -11,13 +11,16 @@ - 父任务信息（仅子任务填写）：无 - accepting_requirements：false - 需求分流：verdict=NEW；selected_initiative=无；候选=无；理由=no accepting initiative；triage 收据=/tmp/be-triage-runhitl-20261021.json；receipt_sha256=39dc1067007f31d2871e03c86a389f882925d5279198b0d3bdf9d8ccd253a6f7；requirement_sha256=2756850b023061065b52aac9b8f318a9802d56bdc97882ac24b749e940ea8559 -- 授权与交付终点：<本次允许的动作与终点；未授权事项写明> +- 授权与交付终点：修复 Manager.ResolveHITL 的 run 状态写入顺序竞态并补回归测试；按 Backend 仓现行门禁完成 Review、PR、CI、合入、任务卡闭合与回收。不得改其它子系统、不得改 HTTP/server 层、不得放宽断言。 - 实施执行：agent-per-unit（配置：AGENT-EXECUTION.md；外部 Agent 一单元一新进程/会话/工作区） ## 目标与非目标 -<预期结果、受影响入口/模块/契约；明确不做什么。> +目标：Manager.ResolveHITL 先向 `hitlCh` 投递答案、再无条件 `UpdateRun(running)`，会在 runner 已把 run 写成 `completed/ended_at` 之后被覆盖回 `running/ended_at=NULL`，导致 run 永久卡在 running（无执行者）、会话永不释放、后续每次提交都是 409。修复为先以条件更新 `WHERE status='waiting_hitl'` 把 run 推进到 running（由 store 报告是否命中），再交答案给 runner；交答案后该 run 行此后唯一写者是 runner，不会被反向覆盖。 +非目标：不改 HTTP/server 层、不改其它 store 表的写入语义、不放宽任何断言、不引入新依赖、不改 schema。 + +验收条件：(1) 新回归测试在修复前红（runner 抢先结算后 DB 仍为 `completed/ended_at`）、修复后绿；(2) `go test ./internal/runmgr ./internal/store` 全绿；(3) 独立 Review 无阻塞项。 ## Agent 实施收据 按根目录 AGENT-EXECUTION.md 路由；外部 Agent 一单元一新进程/会话/隔离工作区。实施、修复与测试必须通过登记的候选执行；未配置或候选安全耗尽停在 unavailable/blocked，禁止 host 隐式代写。 -- 2.54.0