From a6bd18050cdfc8652b5fd5530fdf2711ceb694cb Mon Sep 17 00:00:00 2001 From: Autumn-27 <53844290+Autumn-27@users.noreply.github.com> Date: Wed, 19 Aug 2026 09:20:10 -0400 Subject: [PATCH] =?UTF-8?q?feat(agent):=20=E8=AF=8A=E6=96=AD=20run=20?= =?UTF-8?q?=E5=8F=96=E6=B6=88/=E7=BB=88=E6=AD=A2=E5=8E=9F=E5=9B=A0,?= =?UTF-8?q?=E5=8F=96=E4=BB=A3=E7=AC=BC=E7=BB=9F=E7=9A=84=E4=B8=AD=E6=96=AD?= =?UTF-8?q?=E6=8F=90=E7=A4=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 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。 --- agent/cancelcause.go | 116 ++++++++++++ agent/capture.go | 41 ++--- agent/planner.go | 2 +- agent/terminalreason.go | 160 ++++++++++++++++ agent/terminalreason_test.go | 172 ++++++++++++++++++ agent/worker.go | 2 +- cmd/artex/main.go | 11 +- server/conversations.go | 7 +- server/engine.go | 40 ++-- server/engine_cancelcause_test.go | 96 ++++++++++ server/engine_timeout.go | 2 +- server/orchestration.go | 2 +- server/server.go | 14 +- server/server_mgmt.go | 2 +- .../app/(main)/_components/main-content.tsx | 2 +- web/src/app/(main)/system/llm/page.tsx | 15 +- 16 files changed, 624 insertions(+), 60 deletions(-) create mode 100644 agent/cancelcause.go create mode 100644 agent/terminalreason.go create mode 100644 agent/terminalreason_test.go create mode 100644 server/engine_cancelcause_test.go diff --git a/agent/cancelcause.go b/agent/cancelcause.go new file mode 100644 index 0000000..cfda073 --- /dev/null +++ b/agent/cancelcause.go @@ -0,0 +1,116 @@ +package agent + +import ( + "context" + "errors" + "fmt" +) + +// AbortCause names WHY a planner/worker run's context was cancelled. +// +// Background: when a run's ctx is cancelled mid-flight the harness finishes with +// ReasonAbortedStreaming / ReasonAbortedTools and Terminal.Err = context.Canceled. +// ARTEX cancels that ctx from a dozen different places (pause, delete, kill_work, +// goal-met, settle drain backstop, chat stop, process shutdown, hard run timeout, +// …) but by the time it reaches captureRun they all collapse into one bare +// context.Canceled — so the trace could only print a guess-list, not a diagnosis. +// +// Fix: every cancel site uses context.WithCancelCause / WithTimeoutCause and +// attaches one of the causes below; captureRun recovers it via context.Cause and +// prints exactly who cancelled the run, why, and what happens to the intent. +// +// When you add a new cancel site, add its cause HERE — a site that cancels +// without one shows up in the trace as "canceled_no_cause", by design. +type AbortCause struct { + Code string // stable machine-readable id (logs / future filtering) + Short string // 一句话，进折叠摘要行 + Text string // 完整说明：谁取消的、为什么、这条意图接下来会怎样 +} + +func (c *AbortCause) Error() string { return c.Text } + +func cause(code, short, text string) *AbortCause { + return &AbortCause{Code: code, Short: short, Text: text} +} + +// Causef builds an ad-hoc cause with runtime detail (e.g. which intent, which +// deadline). Use the predeclared vars below when there is nothing dynamic to add. +func Causef(code, short, format string, args ...any) *AbortCause { + return &AbortCause{Code: code, Short: short, Text: fmt.Sprintf(format, args...)} +} + +// The full enumeration of cancel sites, grouped by which context gets cancelled. +var ( + // ---- 任务级 exec ctx（Engine.cancelExec：取消该任务在跑的 planner + 全部 worker）---- + + AbortPausedByUser = cause("paused_by_user", "用户暂停了任务", + "用户暂停了任务（前端「暂停」按钮 / POST /api/tasks/{id}/state action=pause）。本次运行被主动取消，该意图已退回 frontier(open)，恢复任务后会被重新领取、从头重跑") + + AbortPausedByOrchestrator = cause("paused_by_orchestrator", "编排 agent 调用 pause_task 暂停了任务", + "编排 agent 调用了 pause_task 工具暂停本任务。本次运行被主动取消，该意图已退回 frontier(open)，恢复后重跑") + + AbortTaskDeleted = cause("task_deleted", "任务被删除", + "任务被删除（DELETE /api/tasks/{id}）——删除前先取消在跑的 planner/worker。本次运行的结果不会再被使用") + + AbortPausedOnReload = cause("paused_on_reload", "后端启动时恢复了「已暂停」状态", + "后端启动时按库里持久化的「已暂停」状态恢复暂停。本次运行被取消（正常情况下此时并没有在跑的 run）") + + AbortGoalMet = cause("goal_met", "规划者判定目标已达成，任务落终态 done", + "规划者判定任务目标已达成、任务落终态 done，随即取消所有在跑的 worker——它们手头的意图跑完也没意义了。该意图标记为 stopped，不是失败") + + AbortSettleDrainTimeout = cause("settle_drain_timeout", "任务级超时收尾：drain 宽限用尽后强制取消", + "任务级超时收尾的硬兜底：到达任务 timeout 后等在跑 worker 优雅收尾，超过 90s drain 宽限仍未结束 → 强制取消。该意图标记为 exhausted，收尾阶段通常已把事实/资产写回") + + // ---- 单个 work 的 ctx（Engine.workCancel[intentID]）---- + + AbortKilledByPlanner = cause("killed_by_planner", "规划者用 kill_work 终止了这条意图", + "规划者调用 kill_work 主动终止了这条意图（判定方向跑偏/无意义，止损）。该意图标记为 stopped，不会被自动重领") + + AbortWorkFinished = cause("work_finished", "work 已正常结束，引擎释放其 context", + "work 已正常结束，引擎在 unregisterWork 里释放其 context 资源。这不是中断——若它出现在中断消息里，说明取消与收场事件存在竞态") + + AbortPausedRaceGuard = cause("paused_race_guard", "任务处于暂停态，引擎拒绝启动新 run", + "任务处于暂停态时请求执行上下文，引擎直接返回一个已取消的 context（守卫 claim→Execute 的竞态，避免暂停后仍启动新 run）。该意图会退回 frontier") + + // ---- 对话 / 主 Agent 的单轮 ctx（Server.chatCancel[busyKey]）---- + + AbortChatStoppedByUser = cause("chat_stopped_by_user", "用户点了「停止」中止本轮对话", + "用户点了「停止」，主动中止本轮对话运行（POST .../chat/stop 或 .../conversations/{id}/stop）。已产生的步骤都保留，可以直接发下一条消息") + + AbortChatTurnFinished = cause("chat_turn_finished", "本轮对话已正常结束，服务端释放其 context", + "本轮对话已正常结束，服务端在 defer 里释放其 context 资源。这不是中断——若它出现在中断消息里，说明取消与收场事件存在竞态") + + // ---- 进程级 ---- + + AbortShutdown = cause("shutdown", "后端进程正在关停（重启 / 更新 / Ctrl-C）", + "后端进程收到 SIGINT/SIGTERM 正在关停（重启 / 更新 / 手动 Ctrl-C）。所有在跑的 planner/worker 一并取消；重启后残留的 running 意图会被重置为 open 重跑") + + // ---- 单次 run 级（worker.Execute / planner.Plan 的硬兜底）---- + + AbortRunHardTimeout = cause("run_hard_timeout", "单次运行的硬超时兜底触发：某一步卡死了", + "单次运行的硬超时兜底被触发：正常情况下软墙钟预算(MaxDuration)会在回合边界优雅收尾，只有当某一步（模型请求或某个工具）自身卡死、久久不返回、导致回合边界检查根本跑不到时，才会走到「软预算 + 90s 宽限」的硬 cancel。请重点看中断前最后一个 tool_use——多半是它挂住了") +) + +// AbortReason resolves the cause attached to a cancelled run context. ok=false +// means the ctx carries no cause at all (nothing was cancelled). +func AbortReason(ctx context.Context) (code, short, text string, ok bool) { + c := context.Cause(ctx) + if c == nil { + return "", "", "", false + } + if ac, ok := errors.AsType[*AbortCause](c); ok { + return ac.Code, ac.Short, ac.Text, true + } + switch { + case errors.Is(c, context.DeadlineExceeded): + return "deadline_exceeded", "上游 context 到达 deadline（未挂具名原因）", + "上游 context 到达 deadline，但设置方没有用 WithTimeoutCause 挂具名原因: " + c.Error(), true + case errors.Is(c, context.Canceled): + // a plain Canceled means some cancel site still lacks a cause — say so + // plainly instead of inventing one. + return "canceled_no_cause", "取消方未挂具名原因（漏网的取消点）", + "上游 context 被取消，但取消方没有用 context.WithCancelCause 挂上具名原因。" + + "这属于漏网的取消点——请在 agent/cancelcause.go 补一条枚举并接到该取消处", true + } + return "other", firstLine(c.Error(), 80), c.Error(), true +} diff --git a/agent/capture.go b/agent/capture.go index 419e855..bbede4f 100644 --- a/agent/capture.go +++ b/agent/capture.go @@ -3,6 +3,8 @@ package agent import ( "context" "strings" + "time" + "unicode/utf8" "github.com/Autumn-27/norma/agentcore" "github.com/Autumn-27/norma/harness" @@ -64,13 +66,18 @@ func captureRunSession(ctx context.Context, s *agentcore.Session, input string, tbuf.WriteString(text) } + // breadcrumbs for the terminal message: when a run is aborted or times out, + // "which tool was still running, for how long" is the whole diagnosis. + lastTool := &runTrace{startedAt: time.Now()} + var finalText string var rerr error for ev, err := range s.Prompt(ctx, input) { if err != nil { flush() - if ctx.Err() != nil { // ctx cancelled = manual stop, not a failure - rec(db.Activity{Kind: "result", Summary: "（已手动停止本次运行）"}) + if ctx.Err() != nil { // ctx cancelled = stopped by the engine, not a failure + sum, detail := terminalText(ctx, &harness.Terminal{Reason: reason, Err: ctx.Err()}, lastTool) + rec(db.Activity{Kind: "result", Summary: firstLine(sum, 400), Detail: detail}) return finalText, reason, ctx.Err() } rec(db.Activity{Kind: "result", IsError: true, Summary: "执行出错: " + err.Error(), Detail: err.Error()}) @@ -84,6 +91,7 @@ func captureRunSession(ctx context.Context, s *agentcore.Session, input string, flush() toolNames[ev.ToolUse.ID] = ev.ToolUse.Name in := string(ev.ToolUse.Input) + lastTool.start(ev.ToolUse.ID, ev.ToolUse.Name, in) rec(db.Activity{Kind: "tool_use", Tool: ev.ToolUse.Name, ToolUseID: ev.ToolUse.ID, Summary: ev.ToolUse.Name + " " + firstLine(in, 200), Detail: in}) case harness.KindToolResult: @@ -92,6 +100,7 @@ func captureRunSession(ctx context.Context, s *agentcore.Session, input string, } flush() out := blocksText(ev.ToolResult.Content) + lastTool.done(ev.ToolResult.ToolUseID) rec(db.Activity{Kind: "tool_result", Tool: toolNames[ev.ToolResult.ToolUseID], ToolUseID: ev.ToolResult.ToolUseID, IsError: ev.ToolResult.IsError, Summary: firstLine(out, 200), Detail: out}) case harness.KindText: @@ -120,25 +129,13 @@ func captureRunSession(ctx context.Context, s *agentcore.Session, input string, tkind = "" } flush() // flush any trailing thinking / non-final text - sum := ev.Terminal.Text + sum, detail := ev.Terminal.Text, ev.Terminal.Text if sum == "" { - switch ev.Terminal.Reason { - case harness.ReasonAbortedTools, harness.ReasonAbortedStreaming: - // interruption (task paused / worker killed / backend restart), - // not a real completion — the intent is re-queued (open) or - // stopped, so the session icon reflects that, not this run. - sum = "（运行被中断：任务暂停 / 终止 / 重启所致；未完成）" - case harness.ReasonTimeout, harness.ReasonMaxTurns: - // budget hit → settlement ran (facts/assets were written back); - // only the trailing text summary is missing, not a failure. - sum = "（达运行预算上限，已收尾写回事实；本次无文字总结，终态: " + string(ev.Terminal.Reason) + "）" - default: - sum = "（无总结，终态: " + string(ev.Terminal.Reason) + "）" - } + sum, detail = terminalText(ctx, ev.Terminal, lastTool) } u := ev.Terminal.Usage // cumulative token usage for this session rec(db.Activity{Kind: "result", IsError: ev.Terminal.Err != nil, - Summary: firstLine(sum, 400), Detail: sum, + Summary: firstLine(sum, 400), Detail: detail, InputTokens: &u.InputTokens, OutputTokens: &u.OutputTokens, CacheReadTokens: &u.CacheReadTokens, CacheWriteTokens: &u.CacheWriteTokens}) if ev.Terminal.Err != nil { @@ -166,13 +163,15 @@ func blocksText(blocks []llm.ContentBlock) string { } // firstLine returns a single-line, length-capped preview for the summary column. +// max counts RUNES, not bytes: these previews are mostly 中文, and byte-slicing +// used to cut a multi-byte rune in half and render as "恢�…". func firstLine(s string, max int) string { s = strings.TrimSpace(s) - if i := strings.IndexByte(s, '\n'); i >= 0 { - s = s[:i] + if before, _, found := strings.Cut(s, "\n"); found { + s = before } - if len(s) > max { - s = s[:max] + "…" + if utf8.RuneCountInString(s) > max { + s = string([]rune(s)[:max]) + "…" } return s } diff --git a/agent/planner.go b/agent/planner.go index 9b46c30..b567d30 100644 --- a/agent/planner.go +++ b/agent/planner.go @@ -334,7 +334,7 @@ func (p *Planner) Plan(ctx context.Context, taskID int64, as *db.AssetStore, ts runCtx := ctx if maxDur > 0 { var cancel context.CancelFunc - runCtx, cancel = context.WithTimeout(ctx, maxDur+settleHardGrace) + runCtx, cancel = context.WithTimeoutCause(ctx, maxDur+settleHardGrace, AbortRunHardTimeout) defer cancel() } _, _, err = captureRun(runCtx, opts, input, diff --git a/agent/terminalreason.go b/agent/terminalreason.go new file mode 100644 index 0000000..90228b2 --- /dev/null +++ b/agent/terminalreason.go @@ -0,0 +1,160 @@ +package agent + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/Autumn-27/norma/harness" +) + +// This file turns a run's terminal state into something a human can act on. +// +// Before: a cancelled run produced one fixed sentence — "（运行被中断：任务暂停 / +// 终止 / 重启所致；未完成）" — a guess-list, not a diagnosis. The information to do +// better was already there (the cancel cause, the terminal reason, the turn +// count, the tool that never returned); it was just being thrown away. +// +// terminalText produces a self-contained ONE-LINE summary (the transcript's +// collapsed row, cut at the first newline by firstLine) plus a multi-line detail +// block rendered in full when the row is expanded. + +// runTrace records which tool call is in flight, so an aborted / timed-out run can +// name the step it died on — the single most useful fact when a run hangs. +type runTrace struct { + startedAt time.Time + id string // tool_use id of the most recent call + name string // its tool name + input string // its raw input JSON + at time.Time + pending bool // true = no tool_result arrived for it yet +} + +func (t *runTrace) start(id, name, input string) { + t.id, t.name, t.input, t.at, t.pending = id, name, input, time.Now(), true +} + +func (t *runTrace) done(id string) { + if id == t.id { + t.pending = false + } +} + +// reasonHint explains each harness terminal reason in plain 中文: what it means, +// whether results were still written back, and what the engine does next. Covers +// every TerminalReason the harness can emit. +var reasonHint = map[harness.TerminalReason]string{ + harness.ReasonCompleted: "模型主动结束了本轮（正常收场），但没有留下文字总结。事实/资产以工具调用为准，看上面的 tool_use 记录", + harness.ReasonMaxTurns: "撞到步数上限(MaxTurns)：SDK 已自动跑了一轮收尾(settlement)把事实/资产写回，只是缺一段文字总结。" + + "意图标记为 exhausted——表示「这个方向试过但没做完」，规划者会据此换角度，不是失败", + harness.ReasonTimeout: "撞到单次运行的软墙钟预算(MaxDuration)：SDK 在回合边界优雅收尾并写回事实/资产，只是缺文字总结。" + + "意图标记为 exhausted，不是失败", + harness.ReasonModelError: "模型/API 调用失败（网络、鉴权、限流、供应商 5xx 等）。引擎会退避后自动重跑几次；重试用尽仍失败则意图标记为 blocked", + harness.ReasonBlockingLimit: "上下文长度撞硬上限，请求在发出前就被拦下。通常是单次运行塞进了过多工具输出——考虑收窄意图粒度或压缩工具返回", + harness.ReasonPromptTooLong: "提示词过长且压缩(compaction)重试已用尽，无法继续。同样指向单次运行累积的上下文过大", + harness.ReasonImageError: "多模态内容不被当前模型支持（截图/图片块）。换支持视觉的模型，或让工具别回图", + harness.ReasonStopHookPrevented: "Stop 钩子阻止了本轮结束（守卫判定还不能收场），且随后没能继续下去。" + + "检查任务的 Guard 规则是否过严", + harness.ReasonHookStopped: "某个工具/钩子主动叫停了继续执行（守卫拦截，如越界目标、禁用命令）。看上面最后一条 tool_result 里的拦截说明", + harness.ReasonAbortedStreaming: "运行在【模型输出流式生成阶段】被取消", + harness.ReasonAbortedTools: "运行在【工具执行阶段】被取消", +} + +// terminalText renders a terminal event with no final text into (summary, detail). +// summary is one self-contained line; detail is the expandable breakdown. +func terminalText(ctx context.Context, term *harness.Terminal, tr *runTrace) (string, string) { + reason := term.Reason + aborted := reason == harness.ReasonAbortedStreaming || reason == harness.ReasonAbortedTools + + // ---- one-line summary: cause first (that's what the reader is after), then + // just enough run shape to judge it without expanding. ---- + var sum string + switch { + case aborted: + _, short, _, ok := AbortReason(ctx) + if !ok { + short = "取不到取消原因（run 的 context 上没有挂 cause）" + } + stage := "工具执行阶段" + if reason == harness.ReasonAbortedStreaming { + stage = "模型输出阶段" + } + sum = "（运行被中断：" + short + "；停在" + stage + progressSuffix(term, tr) + "，未完成）" + case reason == harness.ReasonMaxTurns || reason == harness.ReasonTimeout: + sum = "（达运行预算上限(" + string(reason) + ")，已收尾写回事实" + progressSuffix(term, tr) + "；本次无文字总结）" + default: + hint := reasonHint[reason] + if hint == "" { + hint = "未知终态（harness 新增了终态但这里还没登记，请补 reasonHint）" + } + sum = "（无文字总结，终态 " + string(reason) + "：" + firstLine(hint, 80) + "）" + } + + // ---- expandable detail ---- + var b strings.Builder + b.WriteString(sum) + b.WriteString("\n\n") + fmt.Fprintf(&b, "- **终态**: `%s` — %s\n", reason, reasonHint[reason]) + if aborted { + code, _, why, ok := AbortReason(ctx) + if ok { + fmt.Fprintf(&b, "- **中断原因** (`%s`): %s\n", code, why) + } else { + b.WriteString("- **中断原因**: 取不到——取消方没有用 context.WithCancelCause 挂上具名原因。" + + "这属于漏网的取消点，请补 agent/cancelcause.go 里的枚举\n") + } + } + if term.Err != nil { + fmt.Fprintf(&b, "- **底层错误**: `%v`\n", term.Err) + } + if term.Turns > 0 { + fmt.Fprintf(&b, "- **已执行**: %d 轮模型回合\n", term.Turns) + } + if !tr.startedAt.IsZero() { + fmt.Fprintf(&b, "- **本次运行耗时**: %s\n", roundDur(time.Since(tr.startedAt))) + } + if u := term.Usage; u.InputTokens+u.OutputTokens+u.CacheReadTokens+u.CacheWriteTokens > 0 { + fmt.Fprintf(&b, "- **累计 token**: 输入 %d / 输出 %d / 缓存读 %d / 缓存写 %d\n", + u.InputTokens, u.OutputTokens, u.CacheReadTokens, u.CacheWriteTokens) + } + if tr.name != "" { + if tr.pending { + fmt.Fprintf(&b, "- **中断时正在执行的工具**: `%s`(已跑 %s，**未返回结果**) — 若是硬超时/长时间无进展，问题多半就在这一步\n ```\n %s\n ```\n", + tr.name, roundDur(time.Since(tr.at)), firstLine(tr.input, 300)) + } else { + fmt.Fprintf(&b, "- **中断前最后一个工具**: `%s`(已正常返回)\n", tr.name) + } + } else { + b.WriteString("- **工具调用**: 本次运行还没发出任何工具调用就结束了\n") + } + return sum, b.String() +} + +// progressSuffix adds "已跑 N 轮 / 12m30s" to a summary line — how far the run got +// before it died, which decides whether an abort cost anything worth re-running. +func progressSuffix(term *harness.Terminal, tr *runTrace) string { + var parts []string + if term.Turns > 0 { + parts = append(parts, fmt.Sprintf("%d 轮", term.Turns)) + } + if !tr.startedAt.IsZero() { + parts = append(parts, roundDur(time.Since(tr.startedAt))) + } + if len(parts) == 0 { + return "" + } + return "，已跑 " + strings.Join(parts, " / ") +} + +// roundDur formats a duration at a readable granularity (no 3.400213s noise). +func roundDur(d time.Duration) string { + switch { + case d < time.Minute: + return d.Round(100 * time.Millisecond).String() + case d < time.Hour: + return d.Round(time.Second).String() + default: + return d.Round(time.Minute).String() + } +} diff --git a/agent/terminalreason_test.go b/agent/terminalreason_test.go new file mode 100644 index 0000000..a3ecd20 --- /dev/null +++ b/agent/terminalreason_test.go @@ -0,0 +1,172 @@ +package agent + +import ( + "context" + "errors" + "strings" + "testing" + "time" + "unicode/utf8" + + "github.com/Autumn-27/norma/harness" + "github.com/Autumn-27/norma/llm" +) + +// every TerminalReason the harness can emit must have an explanation — otherwise +// a run silently falls back to "未知终态", which is the bug this file exists to fix. +func TestReasonHintCoversEveryTerminalReason(t *testing.T) { + all := []harness.TerminalReason{ + harness.ReasonCompleted, harness.ReasonBlockingLimit, harness.ReasonImageError, + harness.ReasonModelError, harness.ReasonAbortedStreaming, harness.ReasonAbortedTools, + harness.ReasonPromptTooLong, harness.ReasonStopHookPrevented, harness.ReasonHookStopped, + harness.ReasonMaxTurns, harness.ReasonTimeout, + } + for _, r := range all { + if strings.TrimSpace(reasonHint[r]) == "" { + t.Errorf("terminal reason %q 没有中文说明", r) + } + } +} + +// the cause attached at cancel time must survive the ctx chain the engine builds: +// task exec ctx -> per-work child -> value ctx -> the run's hard-timeout backstop. +func TestAbortCausePropagatesThroughRunContextChain(t *testing.T) { + type key struct{} + exec, cancelExec := context.WithCancelCause(context.Background()) + work, cancelWork := context.WithCancelCause(exec) + defer cancelWork(nil) + valued := context.WithValue(work, key{}, "task-1") + run, cancelRun := context.WithTimeoutCause(valued, time.Hour, AbortRunHardTimeout) + defer cancelRun() + + cancelExec(AbortPausedByUser) + + code, _, text, ok := AbortReason(run) + if !ok || code != "paused_by_user" { + t.Fatalf("code=%q ok=%v，期望 paused_by_user", code, ok) + } + if !strings.Contains(text, "退回 frontier") { + t.Errorf("原因文案没传下来: %q", text) + } +} + +func TestAbortReasonFallbacks(t *testing.T) { + if _, _, _, ok := AbortReason(context.Background()); ok { + t.Error("未取消的 ctx 不应报出中断原因") + } + // a cancel site that forgot to attach a cause must be called out, not guessed at + plain, cancel := context.WithCancel(context.Background()) + cancel() + code, _, _, ok := AbortReason(plain) + if !ok || code != "canceled_no_cause" { + t.Errorf("code=%q ok=%v，期望 canceled_no_cause", code, ok) + } + + timed, cancelT := context.WithTimeout(context.Background(), time.Nanosecond) + defer cancelT() + <-timed.Done() + if code, _, _, _ := AbortReason(timed); code != "deadline_exceeded" { + t.Errorf("code=%q，期望 deadline_exceeded", code) + } +} + +// an aborted run's one-line summary must name the actual cause, and the detail +// must pin down the tool that never came back. +func TestTerminalTextAbortedNamesCauseAndHangingTool(t *testing.T) { + ctx, cancel := context.WithCancelCause(context.Background()) + cancel(AbortKilledByPlanner) + + tr := &runTrace{startedAt: time.Now().Add(-90 * time.Second)} + tr.start("tu_1", "Bash", `{"command":"nmap -p- 10.0.0.1"}`) + + term := &harness.Terminal{ + Reason: harness.ReasonAbortedTools, + Err: context.Canceled, + Turns: 7, + Usage: llm.Usage{InputTokens: 1200, OutputTokens: 340}, + } + sum, detail := terminalText(ctx, term, tr) + + if !strings.Contains(sum, "kill_work") { + t.Errorf("摘要没点出取消方: %q", sum) + } + if strings.Contains(sum, "\n") { + t.Errorf("摘要必须是单行（前端按首行截断）: %q", sum) + } + for _, want := range []string{"killed_by_planner", "aborted_tools", "7 轮", "Bash", "未返回结果", "1200"} { + if !strings.Contains(detail, want) { + t.Errorf("详情缺少 %q:\n%s", want, detail) + } + } +} + +// a tool that DID return must not be reported as the hang site. +func TestTerminalTextCompletedToolNotBlamed(t *testing.T) { + ctx, cancel := context.WithCancelCause(context.Background()) + cancel(AbortShutdown) + tr := &runTrace{startedAt: time.Now()} + tr.start("tu_1", "Read", `{"path":"/etc/hosts"}`) + tr.done("tu_1") + + _, detail := terminalText(ctx, &harness.Terminal{Reason: harness.ReasonAbortedStreaming}, tr) + if strings.Contains(detail, "未返回结果") { + t.Errorf("已返回的工具被误报为卡住:\n%s", detail) + } + if !strings.Contains(detail, "已正常返回") { + t.Errorf("详情没记录最后一个工具:\n%s", detail) + } +} + +// non-abort terminals keep their own explanation instead of the interruption text. +func TestTerminalTextNonAbortReasons(t *testing.T) { + ctx := context.Background() + tr := &runTrace{startedAt: time.Now()} + + sum, _ := terminalText(ctx, &harness.Terminal{Reason: harness.ReasonMaxTurns}, tr) + if !strings.Contains(sum, "达运行预算上限") || strings.Contains(sum, "中断") { + t.Errorf("max_turns 摘要不对: %q", sum) + } + + sum, detail := terminalText(ctx, &harness.Terminal{ + Reason: harness.ReasonModelError, Err: errors.New("429 rate limited")}, tr) + if !strings.Contains(sum, "model_error") { + t.Errorf("model_error 摘要不对: %q", sum) + } + if !strings.Contains(detail, "429 rate limited") { + t.Errorf("详情没带上底层错误:\n%s", detail) + } +} + +// every enumerated cause must carry both a compact headline (for the collapsed +// row) and a full explanation (for the expanded detail). +func TestAbortCausesAreWellFormed(t *testing.T) { + all := []*AbortCause{ + AbortPausedByUser, AbortPausedByOrchestrator, AbortTaskDeleted, AbortPausedOnReload, + AbortGoalMet, AbortSettleDrainTimeout, AbortKilledByPlanner, AbortWorkFinished, + AbortPausedRaceGuard, AbortChatStoppedByUser, AbortChatTurnFinished, + AbortShutdown, AbortRunHardTimeout, + } + seen := map[string]bool{} + for _, c := range all { + switch { + case c.Code == "" || seen[c.Code]: + t.Errorf("code 缺失或重复: %q", c.Code) + case c.Short == "" || utf8.RuneCountInString(c.Short) > 40: + t.Errorf("%s: Short 必须存在且 ≤40 字（进折叠摘要行）: %q", c.Code, c.Short) + case utf8.RuneCountInString(c.Text) <= utf8.RuneCountInString(c.Short): + t.Errorf("%s: Text 应比 Short 更完整", c.Code) + } + seen[c.Code] = true + } +} + +// summaries are mostly 中文; capping by bytes used to slice a rune in half. +func TestFirstLineCapsByRuneNotByte(t *testing.T) { + got := firstLine(strings.Repeat("恢复", 10), 5) + if got != "恢复恢复恢…" { + t.Errorf("firstLine 按字节截断了: %q", got) + } + if got := firstLine("头\n尾", 100); got != "头" { + t.Errorf("多行没截到首行: %q", got) + } +} diff --git a/agent/worker.go b/agent/worker.go index d21d246..71e1eb9 100644 --- a/agent/worker.go +++ b/agent/worker.go @@ -361,7 +361,7 @@ func (w *Worker) Execute(ctx context.Context, name string, taskID int64, as *db. runCtx := ctx if maxDur > 0 { var runCancel context.CancelFunc - runCtx, runCancel = context.WithTimeout(ctx, maxDur+settleHardGrace) + runCtx, runCancel = context.WithTimeoutCause(ctx, maxDur+settleHardGrace, AbortRunHardTimeout) defer runCancel() } _, reason, err := captureRunSession(runCtx, s, input, emitWrap) diff --git a/cmd/artex/main.go b/cmd/artex/main.go index 30440d4..b5579dc 100644 --- a/cmd/artex/main.go +++ b/cmd/artex/main.go @@ -16,6 +16,7 @@ import ( "syscall" "time" + "github.com/Autumn-27/artex/agent" "github.com/Autumn-27/artex/config" "github.com/Autumn-27/artex/server" ) @@ -70,8 +71,16 @@ func main() { log.Printf("[config] 配置文件: %s (不存在 — 将仅尝试环境变量 ARTEX_PG_DSN)", cfgPath) } - ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + sigCtx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() + // re-wrap with a named cause so a run cancelled by shutdown says "后端进程正在 + // 关停" in its trace instead of a bare context.Canceled (see agent/cancelcause.go). + ctx, shutdown := context.WithCancelCause(sigCtx) + defer shutdown(agent.AbortShutdown) + go func() { + <-sigCtx.Done() + shutdown(agent.AbortShutdown) + }() mgr, err := server.NewManager(*dataDir, *proxy) if err != nil { diff --git a/server/conversations.go b/server/conversations.go index 7b0c77e..69294b8 100644 --- a/server/conversations.go +++ b/server/conversations.go @@ -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) diff --git a/server/engine.go b/server/engine.go index 699b204..205f270 100644 --- a/server/engine.go +++ b/server/engine.go @@ -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). diff --git a/server/engine_cancelcause_test.go b/server/engine_cancelcause_test.go new file mode 100644 index 0000000..0669679 --- /dev/null +++ b/server/engine_cancelcause_test.go @@ -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") + } +} diff --git a/server/engine_timeout.go b/server/engine_timeout.go index 14fa3fe..751c61d 100644 --- a/server/engine_timeout.go +++ b/server/engine_timeout.go @@ -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 } diff --git a/server/orchestration.go b/server/orchestration.go index 7d3b280..011b0f6 100644 --- a/server/orchestration.go +++ b/server/orchestration.go @@ -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 }) } diff --git a/server/server.go b/server/server.go index 5e8418d..55c40f6 100644 --- a/server/server.go +++ b/server/server.go @@ -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"}) } diff --git a/server/server_mgmt.go b/server/server_mgmt.go index 04ece0a..f755afd 100644 --- a/server/server_mgmt.go +++ b/server/server_mgmt.go @@ -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 diff --git a/web/src/app/(main)/_components/main-content.tsx b/web/src/app/(main)/_components/main-content.tsx index 8c08bab..3c5a39c 100644 --- a/web/src/app/(main)/_components/main-content.tsx +++ b/web/src/app/(main)/_components/main-content.tsx @@ -62,7 +62,7 @@ export function MainContent({ children }: { children: ReactNode }) {
{version && ( - 版本·v{version} + 版本 · {version} )} diff --git a/web/src/app/(main)/system/llm/page.tsx b/web/src/app/(main)/system/llm/page.tsx index 8e5e491..f01ac53 100644 --- a/web/src/app/(main)/system/llm/page.tsx +++ b/web/src/app/(main)/system/llm/page.tsx @@ -183,7 +183,12 @@ function NewProfileDialog({ onCreated }: { onCreated: (id: string) => void }) { value={model} onChange={(e) => setModel(e.target.value)} /> - + {/* modal: this Popover lives inside a Dialog, and its content is + portaled to — outside the Dialog's react-remove-scroll + lock, which then cancels every wheel event over it (the list + rendered but refused to scroll). modal makes the Popover own + the topmost scroll lock, so its own list scrolls again. */} + {models.length > 0 && ( - + {models.map((m) => ( {models.length > 0 && ( - + {models.map((m) => (