feat(agent): 诊断 run 取消/终止原因,取代笼统的中断提示
- 新增 AbortCause(cancelcause.go):各取消点用 WithCancelCause/WithTimeoutCause 附上原因(暂停/删除/kill_work/目标达成/drain 兜底/聊天停止/进程关闭/硬超时…), captureRun 用 context.Cause 还原出到底是谁、为何取消。 - 新增 terminalreason.go:把终端状态转成可读的一行摘要 + 多行详情(取消原因、 终止 reason、轮数、卡住的工具),替换原来固定那句猜测式提示。 - engine/worker/planner/capture 及 conversations/orchestration/server 接入; cmd/artex/main.go、前端 llm/main-content 相应调整。 - 配套测试 terminalreason_test.go / engine_cancelcause_test.go。
This commit is contained in:
1 parent
8e9b126972
commit
a6bd18050c
16 files changed
+624
-60
No files matched your search
@@ -11,6 +11,7 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Autumn-27/artex/agent"
|
||||
"github.com/Autumn-27/artex/db"
|
||||
"github.com/Autumn-27/artex/intercept"
|
||||
)
|
||||
@@ -243,7 +244,7 @@ func (s *Server) pgStopConversation(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, 200, map[string]any{"status": "idle"}) // nothing running
|
||||
return
|
||||
}
|
||||
cancel()
|
||||
cancel(agent.AbortChatStoppedByUser)
|
||||
writeJSON(w, 200, map[string]any{"status": "stopping"})
|
||||
}
|
||||
|
||||
@@ -331,12 +332,12 @@ func (s *Server) runConversation(c *db.Conversation, msg, busyKey string) {
|
||||
func (s *Server) runConversationSync(c *db.Conversation, msg, busyKey string) {
|
||||
// Per-run cancellable context so a manual stop (pgStopConversation) can abort
|
||||
// just this session. Registered under chatMu so the stop handler can find it.
|
||||
ctx, cancel := context.WithCancel(intercept.WithConvID(s.ctx, c.ID))
|
||||
ctx, cancel := context.WithCancelCause(intercept.WithConvID(s.ctx, c.ID))
|
||||
s.chatMu.Lock()
|
||||
s.chatCancel[busyKey] = cancel
|
||||
s.chatMu.Unlock()
|
||||
defer func() {
|
||||
cancel()
|
||||
cancel(agent.AbortChatTurnFinished)
|
||||
s.chatMu.Lock()
|
||||
delete(s.chatBusy, busyKey)
|
||||
delete(s.chatCancel, busyKey)
|
||||
|
||||
+23
-17
@@ -97,14 +97,17 @@ type Engine struct {
|
||||
// per-task execution context: each planner.Plan / worker.Execute runs under it,
|
||||
// so pausing can CANCEL an in-flight run (not just skip the next one). Recreated
|
||||
// on resume since cancelling is one-shot.
|
||||
// CancelCauseFunc (not CancelFunc): every cancel site attaches a named
|
||||
// *agent.AbortCause so the run's trace can say WHICH of the ~10 cancel
|
||||
// paths fired instead of one catch-all "运行被中断" line.
|
||||
execMu sync.Mutex
|
||||
execCancel map[string]context.CancelFunc
|
||||
execCancel map[string]context.CancelCauseFunc
|
||||
execCtx map[string]context.Context
|
||||
|
||||
// per-work (per running intent) cancel, so the planner's kill_work can stop a
|
||||
// single worker without pausing the whole task. Keyed by globally-unique intent id.
|
||||
workMu sync.Mutex
|
||||
workCancel map[int64]context.CancelFunc
|
||||
workCancel map[int64]context.CancelCauseFunc
|
||||
|
||||
// steerBox queues planner course-corrections for a running work (keyed by intent
|
||||
// id). The worker's PreToolUse hook drains it before its next tool call and hands
|
||||
@@ -136,18 +139,21 @@ func (e *Engine) nextPlannerRound(taskID string) int {
|
||||
|
||||
// Pause stops a task: marks it paused AND cancels any in-flight planner/worker run
|
||||
// for it (a long worker.Execute would otherwise keep going until it finishes).
|
||||
func (e *Engine) Pause(taskID string) {
|
||||
// cause says WHY (user button / orchestrator tool / delete / reload) — it surfaces
|
||||
// verbatim in the cancelled run's trace, so pass the matching agent.Abort* value.
|
||||
func (e *Engine) Pause(taskID string, cause error) {
|
||||
e.paused.Store(taskID, true)
|
||||
e.cancelExec(taskID)
|
||||
e.cancelExec(taskID, cause)
|
||||
}
|
||||
|
||||
// cancelExec cancels a task's current per-task exec context (any in-flight
|
||||
// planner.Plan / worker.Execute), if present. Shared by Pause and the settle
|
||||
// sequence's hard-drain backstop.
|
||||
func (e *Engine) cancelExec(taskID string) {
|
||||
// sequence's hard-drain backstop. cause is recovered by the aborted run via
|
||||
// context.Cause and printed in its result record.
|
||||
func (e *Engine) cancelExec(taskID string, cause error) {
|
||||
e.execMu.Lock()
|
||||
if cancel := e.execCancel[taskID]; cancel != nil {
|
||||
cancel()
|
||||
cancel(cause)
|
||||
}
|
||||
e.execMu.Unlock()
|
||||
}
|
||||
@@ -166,14 +172,14 @@ func (e *Engine) execContextFor(parent context.Context, taskID string) context.C
|
||||
defer e.execMu.Unlock()
|
||||
if e.IsPaused(taskID) {
|
||||
// never hand out a live context while paused (guards the claim→Execute race)
|
||||
c, cancel := context.WithCancel(parent)
|
||||
cancel()
|
||||
c, cancel := context.WithCancelCause(parent)
|
||||
cancel(agent.AbortPausedRaceGuard)
|
||||
return c
|
||||
}
|
||||
if c := e.execCtx[taskID]; c != nil && c.Err() == nil {
|
||||
return c
|
||||
}
|
||||
c, cancel := context.WithCancel(parent)
|
||||
c, cancel := context.WithCancelCause(parent)
|
||||
e.execCtx[taskID] = c
|
||||
e.execCancel[taskID] = cancel
|
||||
return c
|
||||
@@ -204,12 +210,12 @@ func (e *Engine) touch(taskID string) { e.lastAct.Store(taskID, time.Now().Unix(
|
||||
|
||||
func NewEngine(m *Manager) *Engine {
|
||||
return &Engine{m: m, debounce: 800 * time.Millisecond, bc: NewBroadcaster(),
|
||||
execCancel: map[string]context.CancelFunc{}, execCtx: map[string]context.Context{},
|
||||
workCancel: map[int64]context.CancelFunc{}, steerBox: map[int64][]string{}}
|
||||
execCancel: map[string]context.CancelCauseFunc{}, execCtx: map[string]context.Context{},
|
||||
workCancel: map[int64]context.CancelCauseFunc{}, steerBox: map[int64][]string{}}
|
||||
}
|
||||
|
||||
// registerWork records the cancel for the work currently running intentID.
|
||||
func (e *Engine) registerWork(intentID int64, cancel context.CancelFunc) {
|
||||
func (e *Engine) registerWork(intentID int64, cancel context.CancelCauseFunc) {
|
||||
e.workMu.Lock()
|
||||
e.workCancel[intentID] = cancel
|
||||
e.workMu.Unlock()
|
||||
@@ -220,7 +226,7 @@ func (e *Engine) unregisterWork(intentID int64) {
|
||||
e.workMu.Lock()
|
||||
if c := e.workCancel[intentID]; c != nil {
|
||||
delete(e.workCancel, intentID)
|
||||
c() // release context resources (no-op if already cancelled)
|
||||
c(agent.AbortWorkFinished) // release context resources (no-op if already cancelled)
|
||||
}
|
||||
e.workMu.Unlock()
|
||||
e.steerMu.Lock()
|
||||
@@ -306,7 +312,7 @@ func (e *Engine) KillWork(intentID int64) error {
|
||||
if c == nil {
|
||||
return fmt.Errorf("意图 %d 当前没有运行中的 work(可能已结束或未被领取)", intentID)
|
||||
}
|
||||
c()
|
||||
c(agent.AbortKilledByPlanner)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -530,7 +536,7 @@ func (e *Engine) plannerLoop(ctx context.Context, t *Task) {
|
||||
// 任务已判完成 → 立刻取消在跑的 worker:它们手头的意图跑出来也没意义了。
|
||||
// 下一轮 worker 循环撞终态门就不再领新意图;被取消的这批走下方"任务已完成"分支
|
||||
// 归为 stopped(而非 blocked)。
|
||||
e.cancelExec(t.ID)
|
||||
e.cancelExec(t.ID, agent.AbortGoalMet)
|
||||
default:
|
||||
log.Printf("[planner] task %s 规划完成", t.ID)
|
||||
}
|
||||
@@ -597,7 +603,7 @@ func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) {
|
||||
emit := func(r db.Activity) { e.emitActivity(t, r) }
|
||||
ectx := e.clockCtx(e.execContextFor(ctx, t.ID), t, false) // cancellable by Pause; 带任务 deadline
|
||||
// per-work child context so the planner's kill_work can stop just this work.
|
||||
workCtx, workCancel := context.WithCancel(ectx)
|
||||
workCtx, workCancel := context.WithCancelCause(ectx)
|
||||
e.registerWork(intent.ID, workCancel)
|
||||
// wrap the guard hooks so steer_work can inject a mid-run course-correction
|
||||
// for THIS intent (drained before the worker's next tool call).
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/Autumn-27/artex/agent"
|
||||
)
|
||||
|
||||
// The engine's cancel paths must hand a NAMED cause to the run being cancelled —
|
||||
// that cause is the whole content of the "运行被中断" line in the trace. These
|
||||
// exercise the context machinery only (no Manager / DB needed).
|
||||
|
||||
func TestPauseCancelsWithNamedCause(t *testing.T) {
|
||||
e := NewEngine(nil)
|
||||
ectx := e.execContextFor(context.Background(), "t1")
|
||||
// the per-work child the worker loop actually runs under
|
||||
workCtx, workCancel := context.WithCancelCause(ectx)
|
||||
defer workCancel(nil)
|
||||
|
||||
e.Pause("t1", agent.AbortPausedByUser)
|
||||
|
||||
if code, _, _, ok := agent.AbortReason(workCtx); !ok || code != "paused_by_user" {
|
||||
t.Fatalf("work ctx 拿到的原因 code=%q ok=%v,期望 paused_by_user", code, ok)
|
||||
}
|
||||
}
|
||||
|
||||
func TestKillWorkCancelsOnlyThatWorkWithNamedCause(t *testing.T) {
|
||||
e := NewEngine(nil)
|
||||
ectx := e.execContextFor(context.Background(), "t1")
|
||||
workCtx, workCancel := context.WithCancelCause(ectx)
|
||||
e.registerWork(42, workCancel)
|
||||
|
||||
if err := e.KillWork(42); err != nil {
|
||||
t.Fatalf("KillWork: %v", err)
|
||||
}
|
||||
if code, _, _, ok := agent.AbortReason(workCtx); !ok || code != "killed_by_planner" {
|
||||
t.Fatalf("code=%q ok=%v,期望 killed_by_planner", code, ok)
|
||||
}
|
||||
// the task ctx must survive — kill_work stops one work, not the whole task
|
||||
if ectx.Err() != nil {
|
||||
t.Error("kill_work 不该取消整个任务的 exec ctx")
|
||||
}
|
||||
if err := e.KillWork(999); err == nil {
|
||||
t.Error("对没有在跑的意图 KillWork 应该报错")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSettleDrainAndGoalMetCausesAreDistinct(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
cause *agent.AbortCause
|
||||
want string
|
||||
}{
|
||||
{"settle", agent.AbortSettleDrainTimeout, "settle_drain_timeout"},
|
||||
{"goalMet", agent.AbortGoalMet, "goal_met"},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
e := NewEngine(nil)
|
||||
ectx := e.execContextFor(context.Background(), "t1")
|
||||
e.cancelExec("t1", tc.cause)
|
||||
if code, _, _, _ := agent.AbortReason(ectx); code != tc.want {
|
||||
t.Fatalf("code=%q,期望 %q", code, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// asking for an exec ctx while paused must return an already-cancelled ctx that
|
||||
// still SAYS why — this is the claim→Execute race guard, easily mistaken for a
|
||||
// mystery abort.
|
||||
func TestExecContextWhilePausedCarriesRaceGuardCause(t *testing.T) {
|
||||
e := NewEngine(nil)
|
||||
e.Pause("t1", agent.AbortPausedByUser)
|
||||
ctx := e.execContextFor(context.Background(), "t1")
|
||||
if ctx.Err() == nil {
|
||||
t.Fatal("暂停期间不该发出可用的 exec ctx")
|
||||
}
|
||||
if code, _, _, _ := agent.AbortReason(ctx); code != "paused_race_guard" {
|
||||
t.Fatalf("code=%q,期望 paused_race_guard", code)
|
||||
}
|
||||
}
|
||||
|
||||
// Resume must hand out a live context again (cancelling is one-shot).
|
||||
func TestResumeYieldsFreshContext(t *testing.T) {
|
||||
e := NewEngine(nil)
|
||||
old := e.execContextFor(context.Background(), "t1")
|
||||
e.Pause("t1", agent.AbortPausedByUser)
|
||||
if old.Err() == nil {
|
||||
t.Fatal("Pause 没取消旧 ctx")
|
||||
}
|
||||
e.paused.Delete("t1") // Resume() without the Task/Notify plumbing
|
||||
if fresh := e.execContextFor(context.Background(), "t1"); fresh.Err() != nil {
|
||||
t.Fatal("恢复后应拿到全新可用的 exec ctx")
|
||||
}
|
||||
}
|
||||
@@ -159,7 +159,7 @@ func (e *Engine) settleTask(ctx context.Context, t *Task) {
|
||||
for e.inflightCount(t.ID) > 0 {
|
||||
if time.Now().After(hardStop) {
|
||||
log.Printf("[deadline] task %s drain 超时(%s),硬取消在跑 run", t.ID, settleDrainGrace)
|
||||
e.cancelExec(t.ID)
|
||||
e.cancelExec(t.ID, agent.AbortSettleDrainTimeout)
|
||||
_ = sleepCtx(ctx, 3*time.Second) // 给 worker 分支一点时间落库/归类
|
||||
break
|
||||
}
|
||||
|
||||
@@ -304,7 +304,7 @@ func (s *Server) toolPauseTask() actool.CoreTool {
|
||||
if _, ok := s.m.Task(a.TaskID); !ok {
|
||||
return actool.Errorf("task 不存在: " + a.TaskID), nil
|
||||
}
|
||||
s.engine.Pause(a.TaskID)
|
||||
s.engine.Pause(a.TaskID, agent.AbortPausedByOrchestrator)
|
||||
return actool.Text("task paused: " + a.TaskID), nil
|
||||
})
|
||||
}
|
||||
|
||||
+7
-7
@@ -60,7 +60,7 @@ type Server struct {
|
||||
// convBusyKey), so a manual stop can abort JUST that session's agent run. Set/
|
||||
// cleared alongside chatBusy under chatMu. Aborting a run does NOT touch the P3
|
||||
// trigger queue — the drain goroutine simply proceeds to the next queued fire.
|
||||
chatCancel map[string]context.CancelFunc
|
||||
chatCancel map[string]context.CancelCauseFunc
|
||||
|
||||
// triggerQ buffers P3 trigger fires PER AGENT. A per-agent "pump" launches runs up
|
||||
// to a concurrency limit derived from the agent's策略: serial → limit 1 (+ optional
|
||||
@@ -123,7 +123,7 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
|
||||
log.Fatalf("[auth] JWT key: %v", err)
|
||||
}
|
||||
s := &Server{m: m, engine: NewEngine(m), ctx: ctx, skillDir: skillDir, jwtKey: key, chatBusy: map[string]bool{},
|
||||
chatCancel: map[string]context.CancelFunc{}, triggerQ: map[string][]triggeredRun{},
|
||||
chatCancel: map[string]context.CancelCauseFunc{}, triggerQ: map[string][]triggeredRun{},
|
||||
triggerActive: map[string]int{}, triggerCfg: map[string]triggerBehavior{},
|
||||
profAgents: map[int64]*profBundle{}, profChatAgents: map[int64]*agent.ChatAgent{},
|
||||
provByProfile: map[int64]*provEntry{}}
|
||||
@@ -217,7 +217,7 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
|
||||
log.Printf("[engine] task %s 重置 %d 个残留 running 意图为 open", t.ID, n)
|
||||
}
|
||||
if t.Paused {
|
||||
s.engine.Pause(t.ID)
|
||||
s.engine.Pause(t.ID, agent.AbortPausedOnReload)
|
||||
}
|
||||
// 任务级超时:为每个未终态、带 timeout 的任务起 deadline 协调器,独立于 planner/worker
|
||||
// loop——非活跃任务重启后也能在到点后被收尾(deadline 已过则立即走收尾时序)。
|
||||
@@ -902,7 +902,7 @@ func (s *Server) control(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
switch req.Action {
|
||||
case "pause":
|
||||
s.engine.Pause(t.ID)
|
||||
s.engine.Pause(t.ID, agent.AbortPausedByUser)
|
||||
case "resume":
|
||||
s.engine.Run(s.ctx, t) // ensure loops are alive, then un-pause + nudge
|
||||
s.engine.Resume(t)
|
||||
@@ -2330,7 +2330,7 @@ func (s *Server) chat(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
// Cancellable context so stopChat can abort this turn without affecting
|
||||
// the server's root context (mirrors the conversation stop pattern).
|
||||
ctx, cancel := context.WithCancel(s.ctx)
|
||||
ctx, cancel := context.WithCancelCause(s.ctx)
|
||||
s.chatBusy[t.ID] = true
|
||||
s.chatCancel[t.ID] = cancel
|
||||
s.chatMu.Unlock()
|
||||
@@ -2342,7 +2342,7 @@ func (s *Server) chat(w http.ResponseWriter, r *http.Request) {
|
||||
// handler returns immediately and the browser never needs to hold the request.
|
||||
go func() {
|
||||
defer func() {
|
||||
cancel()
|
||||
cancel(agent.AbortChatTurnFinished)
|
||||
s.chatMu.Lock()
|
||||
delete(s.chatBusy, t.ID)
|
||||
delete(s.chatCancel, t.ID)
|
||||
@@ -2386,7 +2386,7 @@ func (s *Server) stopChat(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, 200, map[string]any{"status": "idle"})
|
||||
return
|
||||
}
|
||||
cancel()
|
||||
cancel(agent.AbortChatStoppedByUser)
|
||||
writeJSON(w, 200, map[string]any{"status": "stopping"})
|
||||
}
|
||||
|
||||
|
||||
@@ -68,7 +68,7 @@ func decode(r *http.Request, v any) error { return json.NewDecoder(r.Body).Decod
|
||||
|
||||
func (s *Server) pgDeleteTask(w http.ResponseWriter, r *http.Request) {
|
||||
id := r.PathValue("id")
|
||||
s.engine.Pause(id) // cancel running loops before removing
|
||||
s.engine.Pause(id, agent.AbortTaskDeleted) // cancel running loops before removing
|
||||
if err := s.m.DeleteTask(id); err != nil {
|
||||
writeErr(w, 500, err.Error())
|
||||
return
|
||||
|
||||
Reference in new issue
Block a user