diff --git a/agent/cancelcause.go b/agent/cancelcause.go
index f4705bc..e61cfaf 100644
--- a/agent/cancelcause.go
+++ b/agent/cancelcause.go
@@ -45,6 +45,8 @@ var (
"规划者调用 kill_work 主动终止了这条意图,通常表示方向跑偏或已无继续价值;意图会标记为 stopped,不会自动重新领取")
AbortWorkPausedByUser = cause("work_paused_by_user", "用户暂停了这条 Worker 意图",
"用户暂停了正在运行的 Worker。本次调用被取消,意图转为 paused;已经登记的意图、事实、漏洞和活动记录全部保留,恢复后从头重新执行")
+ AbortWorkIntervenedByUser = cause("work_intervened_by_user", "用户暂停 Worker 并发起了新对话",
+ "用户向正在运行的 Worker 发送了新的对话意图。本次调用被立即取消并短暂进入 paused;新消息作为同一会话的下一条用户消息,随后自动恢复并优先重新领取")
AbortWorkCancelledByUser = cause("work_cancelled_by_user", "用户取消了这条 Worker 意图",
"用户取消了正在运行的 Worker。本次调用被取消;Worker 退出写入区后,服务端会事务性删除该意图及其直接产生的事实、漏洞和执行记录")
AbortWorkFinished = cause("work_finished", "Worker 已正常结束并释放 context",
diff --git a/agent/planner.go b/agent/planner.go
index 95802f8..3f63166 100644
--- a/agent/planner.go
+++ b/agent/planner.go
@@ -123,6 +123,7 @@ func renderPlannerTodos(items []actool.Todo) string {
//
// "done" — a worker finished intent IntentID (its output conclusion is fetched).
// "finding" — a worker reported a finding on intent IntentID (Detail = 摘要).
+// "worker_message" — the human changed one Worker's direction (Detail = 新方向).
// "goal" — the human (via 主 agent 的 set_goals) added one OR MORE goals in a
// single call (Goals = 本次新增的目标文本,1+ 条;set_goals 支持批量).
// "goal_deleted" — the human deleted a goal from 总览的目标管理 (Detail = 被删目标文本).
@@ -130,14 +131,17 @@ func renderPlannerTodos(items []actool.Todo) string {
// "cancelled" — the human deleted intent IntentID (Detail = 删除原因). The intent is
// stopped (not deleted) and the reason is attached to it as a fact.
type TriggerEvent struct {
- Kind string
- IntentID int64
- Detail string
- Goals []string // Kind=="goal" 专用:本次 set_goals 新增的目标文本(1 条或多条)
- OldGoal string // Kind=="goal_edited" 专用:修改前的目标文本
- NewGoal string // Kind=="goal_edited" 专用:修改后的目标文本
+ Kind string
+ ActivityID int64 // persistent activity id; worker_message notifications use it for de-duplication
+ IntentID int64
+ Detail string
+ Goals []string // Kind=="goal" 专用:本次 set_goals 新增的目标文本(1 条或多条)
+ OldGoal string // Kind=="goal_edited" 专用:修改前的目标文本
+ NewGoal string // Kind=="goal_edited" 专用:修改后的目标文本
}
+const TriggerWorkerMessage = db.PlannerEventWorkerMessage
+
// renderTriggers spells out the change(s) that fired this round: for a finished
// worker — which intent + its output conclusion; for a finding — which intent +
// what was found. Empty for time/heartbeat wakes. Reads the store (best-effort;
@@ -162,6 +166,9 @@ func renderTriggers(ts *db.ExplorationStore, evs []TriggerEvent) string {
b.WriteString(fmt.Sprintf("\n- 人修改了目标,由「%s」变为「%s」—— 请据新目标调整探索方向(原方向若已不适用请停派)。", ev.OldGoal, ev.NewGoal))
case "finding":
b.WriteString(fmt.Sprintf("\n- 意图 #%d(%s)的 worker 报告了一个 finding:%s", ev.IntentID, intentSummary(ts, ev.IntentID), ev.Detail))
+ case TriggerWorkerMessage:
+ b.WriteString(fmt.Sprintf("\n- 用户向意图 #%d(%s)的 Worker 下发了新的人工方向:%s —— 该方向已优先交给同一 Worker;后续规划必须以它替代或补充原意图描述,避免生成冲突、回退或重复方向。",
+ ev.IntentID, intentSummary(ts, ev.IntentID), truncOutput(ev.Detail, 2000)))
case "cancelled":
b.WriteString(fmt.Sprintf("\n- 意图 #%d 由用户删除,意图内容是:%s、删除原因是:%s。该意图已停止(不再执行),其原因已作为事实挂在该意图上;请据此重新规划。", ev.IntentID, intentSummary(ts, ev.IntentID), ev.Detail))
default: // "done"
@@ -250,6 +257,7 @@ const plannerDefaultTmpl = `你是一个授权渗透测试系统的"规划者"
决策流程(每次唤醒):
1. **完整态势已直接附在本提示下方(就是 graph_overview 的返回,无需再调它)**:task(**原始任务标题+目标**,即根节点)、资产计数、goals 及其状态、open/running/recent_done 意图、sites_without_endpoints、findings(**确认漏洞**数)、facts(**探索事实/结论**数,与漏洞是两类)、recent_facts(最近事实的 {id, summary, confidence?},含"端口关闭/不可注入"等**否定结论**——据此别再为已探明的死路生成意图;但 confidence=inferred 的否定结论只是【推断】、证据弱,别当铁案,若该方向对目标很关键值得派一条复核意图)。**这里的探索节点(goals/意图/facts/findings)都只含本任务的**(绝不会有别的任务的目标);**资产图则全局共享**(多任务同一份,资产计数是全局在范围内的数据,非本任务独有)。需要更深的细节时才按需调:list_facts 分页列事实(最新在前,默认 20 条,可传 q 关键词过滤、before 翻页,返回带 total/has_more)、list_findings 列全部漏洞、**node_detail(id)** 取某条的完整证据/详情(列表/recent_facts 只给摘要)。
- open_intents / running_intents / recent_done_intents——"哪些方向已经有意图在覆盖/已尝试"。每个意图还带 **parents(上游:它派生自哪些事实/意图)和 yields(下游:它产生了哪些事实/发现)**——这就是探索图的**血缘关系**,据此理解"哪些事实来自哪个方向、能否综合成新方向"。recent_facts 里每个事实带 **from_intent**(由哪个意图产生)。
+ - worker_directions——用户在 Worker 对话里对每个意图最近一次下发的人工方向(持久化、重启后仍存在)。它比该意图最初的 summary 更新;判断在跑方向、补意图和使用 steer/kill 时必须以它为准,禁止重新派发与人工方向冲突的工作。
- sites_without_endpoints / findings——"哪些方向【可能】需要探索"。
- 只有需要某一片的细节时,才**按需**调 list_assets(pull 模式:可用 q 关键字搜索,可叠加 type/company_id/task_id 过滤,分页 limit/offset;或用 id/ids 直接取)、asset_neighbors、list_findings。资产图全局共享,别默认拉全量。资产中可能包含非本次任务涉及到的资产,所以需要主要出现非本次任务相关的资产时忽略这些资产。
2. 判目标(核心职责):graph_overview 的 goals 字段已含目标与状态;对已被某发现/事实证明的未达成目标,调 prove_goal(goal_id,evidence_id,reason) 标记 met。**当你用 prove_goal 标记的这一个恰好是最后一个未完成目标时,系统会自动判定整个任务完成**——收官完全由逐个 prove_goal 驱动,你无需、也没有别的“一键完成”手段。
diff --git a/agent/tools.go b/agent/tools.go
index ebed6d8..3a34ca1 100644
--- a/agent/tools.go
+++ b/agent/tools.go
@@ -48,6 +48,38 @@ func compactIntents(ns []*db.Node, parentsOf, yieldsOf map[int64][]int64) []map[
return out
}
+const workerDirectionOverviewLimit = 300
+const workerDirectionMessageRunes = 1200
+
+// compactWorkerDirections keeps the latest durable human direction for each
+// intent small enough for every Planner/Main Agent turn. The full message stays
+// available in the Worker activity trace when an unusually long direction is
+// truncated here.
+func compactWorkerDirections(items []db.WorkerDirection) []map[string]any {
+ out := make([]map[string]any, 0, len(items))
+ for _, item := range items {
+ message := item.Message
+ truncated := false
+ if runes := []rune(message); len(runes) > workerDirectionMessageRunes {
+ message = string(runes[:workerDirectionMessageRunes]) + "…"
+ truncated = true
+ }
+ row := map[string]any{
+ "activity_id": item.ActivityID,
+ "intent_id": item.IntentID,
+ "intent_state": item.IntentState,
+ "intent_summary": item.IntentSummary,
+ "message": message,
+ "updated_at": item.CreatedAt,
+ }
+ if truncated {
+ row["truncated"] = true
+ }
+ out = append(out, row)
+ }
+ return out
+}
+
// ToolSet exposes the PG-backed dual graph (asset + exploration) to an LLM agent.
// One ToolSet is created per planner/worker run; per-run signals live here.
type ToolSet struct {
@@ -360,6 +392,14 @@ func (t *ToolSet) graphOverview() actool.CoreTool {
// an empty context and always needs this first).
func (t *ToolSet) graphOverviewData() map[string]any {
out := map[string]any{}
+ // worker_directions is the persistent cross-agent steering channel. It has
+ // one latest accepted human message per local intent, so Planner and Main
+ // Agent do not have to infer a changed direction from the original summary.
+ if directions, err := t.ts.LatestWorkerDirections(workerDirectionOverviewLimit); err == nil {
+ out["worker_directions"] = compactWorkerDirections(directions)
+ } else {
+ out["worker_directions"] = []map[string]any{}
+ }
// goals summary folded in so the planner needn't call list_goals each round.
goals, _ := t.ts.ListByKind(db.KindGoal, 100)
gsum := make([]map[string]any, 0, len(goals))
diff --git a/agent/worker.go b/agent/worker.go
index 2c77722..84f693b 100644
--- a/agent/worker.go
+++ b/agent/worker.go
@@ -7,6 +7,7 @@ import (
"os"
"path/filepath"
"strconv"
+ "strings"
"time"
"github.com/Autumn-27/artex/db"
@@ -68,6 +69,30 @@ type Worker struct {
nonStreamingFn func() bool
}
+// WorkerSessionID returns the stable transcript key used by a worker intent.
+// Worker slots are reusable, so the intent id (rather than work#N) is the
+// session identity. Keep this helper public so the Worker message API and UI
+// can refer to exactly the conversation that will be resumed.
+func WorkerSessionID(explorationID, intentID int64) string {
+ return fmt.Sprintf("exp%d-worker-i%d", explorationID, intentID)
+}
+
+const workerChatMarkerPrefix = ""
+}
+
+func hasWorkerChatMessage(messages []llm.Message, requestID string) bool {
+ marker := workerChatMarker(requestID)
+ for _, message := range messages {
+ if message.Role == llm.RoleUser && strings.Contains(message.Text(), marker) {
+ return true
+ }
+ }
+ return false
+}
+
// SetNonStreaming wires a resolver deciding whether runs use the non-streaming
// model path (true = non-streaming). nil/unset = streaming (default). Read per
// run so a profile or task-chain toggle takes effect without rebuilding.
@@ -274,6 +299,18 @@ func renderWorkerGraphOverview(data map[string]any) string {
// explored but persisted nothing isn't mistaken for done, and the engine can log
// facts/assets/findings separately instead of lumping them under "facts").
func (w *Worker) Execute(ctx context.Context, name string, taskID int64, as *db.AssetStore, ts *db.ExplorationStore, intent *db.Node, hooks harness.HookRunner, emit func(db.Activity), enr EnrichTrigger, notifyFinding func(int64, string)) (harness.TerminalReason, WriteCounts, error) {
+ return w.execute(ctx, name, taskID, as, ts, intent, hooks, emit, enr, notifyFinding, "", "")
+}
+
+// ExecuteWithMessage runs the next turn in the same intent conversation with a
+// human-authored message. The HTTP handler does not edit the transcript;
+// agentcore records the message as a normal user turn when this Worker starts.
+// This keeps Worker continuation identical to the regular agent chat flow.
+func (w *Worker) ExecuteWithMessage(ctx context.Context, name string, taskID int64, as *db.AssetStore, ts *db.ExplorationStore, intent *db.Node, hooks harness.HookRunner, emit func(db.Activity), enr EnrichTrigger, notifyFinding func(int64, string), requestID, message string) (harness.TerminalReason, WriteCounts, error) {
+ return w.execute(ctx, name, taskID, as, ts, intent, hooks, emit, enr, notifyFinding, strings.TrimSpace(requestID), strings.TrimSpace(message))
+}
+
+func (w *Worker) execute(ctx context.Context, name string, taskID int64, as *db.AssetStore, ts *db.ExplorationStore, intent *db.Node, hooks harness.HookRunner, emit func(db.Activity), enr EnrichTrigger, notifyFinding func(int64, string), requestID, message string) (harness.TerminalReason, WriteCounts, error) {
tsx := NewToolSet(ts, name)
tsx.SetTaskID(taskID)
coverageEnabled := as == nil || as.CoverageEnabled(taskID)
@@ -358,7 +395,7 @@ func (w *Worker) Execute(ctx context.Context, name string, taskID int64, as *db.
}
if w.tx != nil { // persist raw LLM conversation; one file per worked intent
opts.Transcript = w.tx
- opts.SessionID = fmt.Sprintf("exp%d-worker-i%d", ts.ID(), intent.ID)
+ opts.SessionID = WorkerSessionID(ts.ID(), intent.ID)
}
intentID := intent.ID
emitWrap := func(r db.Activity) {
@@ -398,11 +435,26 @@ func (w *Worker) Execute(ctx context.Context, name string, taskID int64, as *db.
// being re-run. The transcript ID is deterministic per intent, so if a prior
// session exists the worker continues from where it left off instead of
// restarting from scratch.
+ alreadyRecorded := false
if w.tx != nil {
_ = s.Resume(opts.SessionID)
- if len(s.Messages()) > 0 {
+ alreadyRecorded = requestID != "" && hasWorkerChatMessage(s.Messages(), requestID)
+ if len(s.Messages()) > 0 && message == "" {
seedUnlockFromHistory(s.Messages(), def.UnlockSkill)
input = "继续执行。"
+ } else if len(s.Messages()) > 0 {
+ seedUnlockFromHistory(s.Messages(), def.UnlockSkill)
+ }
+ }
+ if message != "" {
+ if alreadyRecorded {
+ input = "继续执行上一次人工对话输入的新意图。不要重复已经完成的动作。"
+ } else if len(s.Messages()) > 0 {
+ input = workerChatMarker(requestID) + "\n【人工对话输入的新意图】\n" + message +
+ "\n\n请立即按这条人工输入执行,完成后再根据上下文决定原任务是否需要继续。"
+ } else {
+ input += "\n\n" + workerChatMarker(requestID) + "\n【人工对话输入的新意图】\n" + message +
+ "\n\n请优先执行这条人工输入。"
}
}
diff --git a/agent/worker_intervention_test.go b/agent/worker_intervention_test.go
new file mode 100644
index 0000000..ab8977e
--- /dev/null
+++ b/agent/worker_intervention_test.go
@@ -0,0 +1,28 @@
+package agent
+
+import (
+ "testing"
+
+ "github.com/Autumn-27/norma/llm"
+)
+
+func TestWorkerSessionIDIsStablePerIntent(t *testing.T) {
+ if got := WorkerSessionID(12, 34); got != "exp12-worker-i34" {
+ t.Fatalf("session id = %q, want exp12-worker-i34", got)
+ }
+}
+
+func TestWorkerChatMarkerOnlyMatchesItsNormalUserTurn(t *testing.T) {
+ requestID := "worker-message-123"
+ messages := []llm.Message{
+ {Role: llm.RoleAssistant, Content: []llm.ContentBlock{llm.TextBlock(workerChatMarker(requestID))}},
+ llm.UserText(workerChatMarker("worker-message-other") + "\nother"),
+ llm.UserText(workerChatMarker(requestID) + "\nnew intent"),
+ }
+ if !hasWorkerChatMessage(messages, requestID) {
+ t.Fatal("expected the matching user turn to be detected")
+ }
+ if hasWorkerChatMessage(messages, "worker-message-missing") {
+ t.Fatal("unrelated request id matched a Worker user turn")
+ }
+}
diff --git a/db/exploration.go b/db/exploration.go
index eb0e677..7c40e85 100644
--- a/db/exploration.go
+++ b/db/exploration.go
@@ -14,6 +14,11 @@ import (
// it to distinguish an expected control race (HTTP 409) from a storage failure.
var ErrIntentStateConflict = errors.New("intent state changed concurrently")
+// ErrIntentInterventionConflict marks a rejected or conflicting worker
+// intervention request. It covers state races, a reused request id with a
+// different payload, and a second request trying to overtake a pending one.
+var ErrIntentInterventionConflict = errors.New("intent intervention conflict")
+
// utf8Clean makes a string safe for a PostgreSQL text column. It (1) replaces
// invalid UTF-8 byte sequences with U+FFFD and (2) strips NUL (0x00) bytes.
// Tool output (nmap/curl/… stdout) can carry raw/truncated bytes that PostgreSQL's
@@ -330,6 +335,437 @@ WHERE id=$2 AND exploration_id=$3 AND kind='intent' AND state=$4`, state, id, s.
return n == 1, err
}
+const (
+ IntentInterventionPending = "pending"
+ IntentInterventionAccepted = "accepted"
+ IntentInterventionRejected = "rejected"
+ PlannerEventWorkerMessage = "worker_message"
+)
+
+// IntentIntervention is the durable control record for one human interruption.
+// It deliberately lives in activity rather than a new table: activity already
+// participates in task archive/restore, and the accepted row is also the user
+// turn rendered in the intent transcript.
+type IntentIntervention struct {
+ RequestID string
+ Message string
+ Status string
+ ActivityID int64
+ IntentID int64
+ IntentState string
+ CreatedAt time.Time
+}
+
+// WorkerDirection is the latest accepted human direction for one Worker
+// intent. The accepted activity is the durable planning event, so this view
+// survives restarts and is shared by Planner and Main Agent context builders.
+type WorkerDirection struct {
+ ActivityID int64
+ IntentID int64
+ RequestID string
+ Message string
+ IntentState string
+ IntentSummary string
+ CreatedAt time.Time
+}
+
+// lockIntentIntervention serializes request-id reservation/finalization across
+// processes. exploration_nodes.id is a global primary key, so the intent id is
+// sufficient input to the namespaced advisory key.
+func lockIntentIntervention(tx *sql.Tx, intentID int64) error {
+ _, err := tx.Exec(`SELECT pg_advisory_xact_lock(hashtextextended('artex-intent-intervention:' || $1::text, 0))`, intentID)
+ return err
+}
+
+func scanIntentIntervention(row *sql.Row) (*IntentIntervention, error) {
+ var it IntentIntervention
+ err := row.Scan(&it.ActivityID, &it.RequestID, &it.Message, &it.Status,
+ &it.IntentID, &it.IntentState, &it.CreatedAt)
+ if err == sql.ErrNoRows {
+ return nil, nil
+ }
+ if err != nil {
+ return nil, err
+ }
+ return &it, nil
+}
+
+// IntentInterventionByRequest finds a prior reservation/acceptance. Callers do
+// this before ordinary lifecycle validation so retrying an already accepted
+// request remains idempotent even if the worker has since finished naturally.
+func (s *ExplorationStore) IntentInterventionByRequest(intentID int64, requestID string) (*IntentIntervention, error) {
+ return scanIntentIntervention(s.db.QueryRow(`
+SELECT COALESCE(NULLIF(a.metadata->>'accepted_activity_id','')::bigint, a.id),
+ COALESCE(a.metadata->>'intervention_request_id',''),
+ COALESCE(a.metadata->>'intervention_message', a.detail, ''),
+ COALESCE(a.metadata->>'intervention_status',''),
+ n.id, n.state, a.created_at
+FROM activity a
+JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id
+WHERE a.exploration_id=$1 AND a.node_id=$2
+ AND a.metadata->>'intervention_request_id'=$3
+ORDER BY a.id
+LIMIT 1`, s.expID, intentID, requestID))
+}
+
+// ReserveIntentIntervention durably claims a request id while the intent is
+// still running. A hidden intervention_control activity is the write-ahead
+// record; acceptance inserts a newer visible kind='user' row so SSE cursors can
+// never skip a message whose control reservation had an older id.
+func (s *ExplorationStore) ReserveIntentIntervention(intentID int64, requestID, message string) (*IntentIntervention, error) {
+ tx, err := s.db.Begin()
+ if err != nil {
+ return nil, err
+ }
+ defer tx.Rollback()
+ if err := lockIntentIntervention(tx, intentID); err != nil {
+ return nil, err
+ }
+
+ existing, err := scanIntentIntervention(tx.QueryRow(`
+SELECT COALESCE(NULLIF(a.metadata->>'accepted_activity_id','')::bigint, a.id),
+ COALESCE(a.metadata->>'intervention_request_id',''),
+ COALESCE(a.metadata->>'intervention_message', a.detail, ''),
+ COALESCE(a.metadata->>'intervention_status',''),
+ n.id, n.state, a.created_at
+FROM activity a
+JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id
+WHERE a.exploration_id=$1 AND a.node_id=$2
+ AND a.metadata->>'intervention_request_id'=$3
+ORDER BY a.id
+LIMIT 1`, s.expID, intentID, requestID))
+ if err != nil {
+ return nil, err
+ }
+ if existing != nil {
+ if existing.Message != message {
+ return nil, fmt.Errorf("%w: request_id 已用于不同的 Worker 消息", ErrIntentInterventionConflict)
+ }
+ if err := tx.Commit(); err != nil {
+ return nil, err
+ }
+ return existing, nil
+ }
+
+ var state string
+ if err := tx.QueryRow(`SELECT state FROM exploration_nodes
+ WHERE id=$1 AND exploration_id=$2 AND kind='intent' FOR UPDATE`, intentID, s.expID).Scan(&state); err != nil {
+ if err == sql.ErrNoRows {
+ return nil, fmt.Errorf("%w: intent not found", ErrIntentInterventionConflict)
+ }
+ return nil, err
+ }
+ if state != "running" && state != "paused" {
+ return nil, fmt.Errorf("%w: intent state %s cannot receive a message", ErrIntentInterventionConflict, state)
+ }
+ var otherRequest string
+ err = tx.QueryRow(`SELECT COALESCE(metadata->>'intervention_request_id','')
+ FROM activity
+ WHERE exploration_id=$1 AND node_id=$2
+ AND metadata->>'intervention_status'=$3
+ ORDER BY id LIMIT 1`, s.expID, intentID, IntentInterventionPending).Scan(&otherRequest)
+ if err != nil && err != sql.ErrNoRows {
+ return nil, err
+ }
+ if err == nil {
+ return nil, fmt.Errorf("%w: intent 已有待处理的 Worker 消息 %s", ErrIntentInterventionConflict, otherRequest)
+ }
+
+ metadata, err := json.Marshal(map[string]any{
+ "intervention": true,
+ "intervention_handoff": "pending",
+ "intervention_request_id": requestID,
+ "intervention_message": message,
+ "intervention_status": IntentInterventionPending,
+ "source": "user",
+ })
+ if err != nil {
+ return nil, err
+ }
+ var out IntentIntervention
+ err = tx.QueryRow(`
+INSERT INTO activity(exploration_id,node_id,worker,kind,metadata)
+VALUES ($1,$2,'system','intervention_control',$3)
+RETURNING id, created_at`, s.expID, intentID, string(metadata)).Scan(&out.ActivityID, &out.CreatedAt)
+ if err != nil {
+ return nil, err
+ }
+ out.RequestID, out.Message, out.Status = requestID, message, IntentInterventionPending
+ out.IntentID, out.IntentState = intentID, state
+ if err := tx.Commit(); err != nil {
+ return nil, err
+ }
+ return &out, nil
+}
+
+// AcceptIntentIntervention is the DB commit point: the hidden reservation
+// becomes an audited user turn in the same transaction that moves paused back
+// to open. Thus no Worker can claim the intent before its pending message and
+// visible activity are both durable.
+func (s *ExplorationStore) AcceptIntentIntervention(intentID int64, requestID, message string) (Activity, bool, error) {
+ var activity Activity
+ tx, err := s.db.Begin()
+ if err != nil {
+ return activity, false, err
+ }
+ defer tx.Rollback()
+ if err := lockIntentIntervention(tx, intentID); err != nil {
+ return activity, false, err
+ }
+
+ var controlID int64
+ var storedMessage, status, state, taskStatus string
+ var taskPaused, taskQueued bool
+ var taskDeletedAt sql.NullTime
+ var createdAt time.Time
+ err = tx.QueryRow(`
+SELECT a.id,
+ COALESCE(a.metadata->>'intervention_message', a.detail, ''),
+ COALESCE(a.metadata->>'intervention_status',''),
+ n.state, a.created_at, COALESCE(t.status,''), t.paused, t.queued, t.deleted_at
+FROM activity a
+JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id
+JOIN tasks t ON t.exploration_id=a.exploration_id
+WHERE a.exploration_id=$1 AND a.node_id=$2
+ AND a.metadata->>'intervention_request_id'=$3
+ ORDER BY a.id LIMIT 1
+ FOR UPDATE OF a,n,t`, s.expID, intentID, requestID).
+ Scan(&controlID, &storedMessage, &status, &state, &createdAt,
+ &taskStatus, &taskPaused, &taskQueued, &taskDeletedAt)
+ if err == sql.ErrNoRows {
+ return activity, false, fmt.Errorf("%w: intervention reservation not found", ErrIntentInterventionConflict)
+ }
+ if err != nil {
+ return activity, false, err
+ }
+ if storedMessage != message {
+ return activity, false, fmt.Errorf("%w: request_id 已用于不同的 Worker 消息", ErrIntentInterventionConflict)
+ }
+ if status == IntentInterventionAccepted {
+ var acceptedID int64
+ if err := tx.QueryRow(`SELECT COALESCE((metadata->>'accepted_activity_id')::bigint,0),
+ COALESCE((SELECT created_at FROM activity accepted
+ WHERE accepted.id=(activity.metadata->>'accepted_activity_id')::bigint), created_at)
+ FROM activity WHERE id=$1`, controlID).Scan(&acceptedID, &createdAt); err != nil {
+ return activity, false, err
+ }
+ if acceptedID <= 0 {
+ return activity, false, fmt.Errorf("accepted intervention has no activity id")
+ }
+ if err := tx.Commit(); err != nil {
+ return activity, false, err
+ }
+ nid := intentID
+ activity.ID = acceptedID
+ activity.NodeID, activity.Worker, activity.Kind = &nid, "user", "user"
+ activity.Summary, activity.Detail, activity.CreatedAt = message, message, createdAt
+ return activity, true, nil
+ }
+ if status != IntentInterventionPending {
+ return activity, false, fmt.Errorf("%w: intervention status %s cannot be accepted", ErrIntentInterventionConflict, status)
+ }
+ if state != "paused" {
+ return activity, false, fmt.Errorf("%w: intent state %s cannot be reopened", ErrIntentInterventionConflict, state)
+ }
+ if taskPaused || taskQueued || taskDeletedAt.Valid || IsTerminal(taskStatus) {
+ return activity, false, fmt.Errorf("%w: task lifecycle changed (status=%s paused=%v queued=%v deleted=%v)",
+ ErrIntentInterventionConflict, taskStatus, taskPaused, taskQueued, taskDeletedAt.Valid)
+ }
+
+ metadata, err := json.Marshal(map[string]any{
+ "intervention": true,
+ "intervention_request_id": requestID,
+ "intervention_message": message,
+ "intervention_status": IntentInterventionAccepted,
+ "planner_event": PlannerEventWorkerMessage,
+ "source": "user",
+ })
+ if err != nil {
+ return activity, false, err
+ }
+ if err := tx.QueryRow(`INSERT INTO activity(
+ exploration_id,node_id,worker,kind,summary,detail,metadata)
+ VALUES ($1,$2,'user','user',$3,$3,$4)
+ RETURNING id,created_at`, s.expID, intentID, utf8Clean(message), string(metadata)).
+ Scan(&activity.ID, &createdAt); err != nil {
+ return activity, false, err
+ }
+ var acceptedMeta map[string]any
+ if err := json.Unmarshal(metadata, &acceptedMeta); err != nil {
+ return activity, false, err
+ }
+ acceptedMeta["accepted_activity_id"] = activity.ID
+ controlMetadata, err := json.Marshal(acceptedMeta)
+ if err != nil {
+ return activity, false, err
+ }
+ if _, err := tx.Exec(`UPDATE activity SET metadata=$1 WHERE id=$2`, string(controlMetadata), controlID); err != nil {
+ return activity, false, err
+ }
+ res, err := tx.Exec(`UPDATE exploration_nodes
+ SET state='open', owner=NULL, completed_at=NULL, blocked_reason=NULL
+ WHERE id=$1 AND exploration_id=$2 AND kind='intent' AND state='paused'`, intentID, s.expID)
+ if err != nil {
+ return activity, false, err
+ }
+ if n, err := res.RowsAffected(); err != nil || n != 1 {
+ if err != nil {
+ return activity, false, err
+ }
+ return activity, false, fmt.Errorf("%w: intent is no longer paused", ErrIntentInterventionConflict)
+ }
+ if err := tx.Commit(); err != nil {
+ return activity, false, err
+ }
+ nid := intentID
+ activity.NodeID, activity.Worker, activity.Kind = &nid, "user", "user"
+ activity.Summary, activity.Detail, activity.CreatedAt = message, message, createdAt
+ activity.Metadata = metadata
+ return activity, false, nil
+}
+
+// LatestWorkerDirections returns at most one accepted human direction per
+// local intent, newest directions first. It reads the visible user activity
+// rather than the hidden hand-off row so archive/restore and activity history
+// share one authoritative record.
+func (s *ExplorationStore) LatestWorkerDirections(limit int) ([]WorkerDirection, error) {
+ if limit <= 0 {
+ limit = 300
+ }
+ rows, err := s.db.Query(`
+SELECT activity_id,intent_id,request_id,message,intent_state,intent_summary,created_at
+FROM (
+ SELECT DISTINCT ON (a.node_id)
+ a.id AS activity_id,
+ a.node_id AS intent_id,
+ COALESCE(a.metadata->>'intervention_request_id','') AS request_id,
+ COALESCE(a.metadata->>'intervention_message',a.detail,'') AS message,
+ n.state AS intent_state,
+ COALESCE(n.payload->>'summary','') AS intent_summary,
+ a.created_at
+ FROM activity a
+ JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id
+ WHERE a.exploration_id=$1
+ AND a.kind='user'
+ AND n.kind='intent'
+ AND a.metadata->>'intervention_status'=$2
+ AND a.metadata->>'planner_event'=$3
+ ORDER BY a.node_id,a.id DESC
+) latest
+ORDER BY activity_id DESC
+LIMIT $4`, s.expID, IntentInterventionAccepted, PlannerEventWorkerMessage, limit)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ out := make([]WorkerDirection, 0)
+ for rows.Next() {
+ var item WorkerDirection
+ if err := rows.Scan(&item.ActivityID, &item.IntentID, &item.RequestID, &item.Message,
+ &item.IntentState, &item.IntentSummary, &item.CreatedAt); err != nil {
+ return nil, err
+ }
+ out = append(out, item)
+ }
+ return out, rows.Err()
+}
+
+// RecoverableIntentInterventions returns control operations that must be
+// reconstructed before ordinary frontier claiming after a process restart:
+// pending transcript commits, plus accepted hand-offs not yet claimed once.
+func (s *ExplorationStore) RecoverableIntentInterventions() ([]IntentIntervention, error) {
+ rows, err := s.db.Query(`
+SELECT COALESCE(NULLIF(a.metadata->>'accepted_activity_id','')::bigint, a.id),
+ COALESCE(a.metadata->>'intervention_request_id',''),
+ COALESCE(a.metadata->>'intervention_message',''),
+ COALESCE(a.metadata->>'intervention_status',''),
+ n.id,n.state,a.created_at
+FROM activity a
+JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id
+WHERE a.exploration_id=$1 AND a.kind='intervention_control'
+ AND (a.metadata->>'intervention_status'=$2 OR (
+ a.metadata->>'intervention_status'=$3
+ AND COALESCE(a.metadata->>'intervention_handoff','pending')='pending'))
+ORDER BY a.id`, s.expID, IntentInterventionPending, IntentInterventionAccepted)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ out := []IntentIntervention{}
+ for rows.Next() {
+ var item IntentIntervention
+ if err := rows.Scan(&item.ActivityID, &item.RequestID, &item.Message, &item.Status,
+ &item.IntentID, &item.IntentState, &item.CreatedAt); err != nil {
+ return nil, err
+ }
+ out = append(out, item)
+ }
+ return out, rows.Err()
+}
+
+// MarkIntentInterventionHandoffClaimed records that the accepted instruction
+// has been preferentially claimed once. Idempotency status remains accepted.
+func (s *ExplorationStore) MarkIntentInterventionHandoffClaimed(intentID int64) error {
+ _, err := s.db.Exec(`UPDATE activity SET metadata=jsonb_set(
+ metadata,'{intervention_handoff}',to_jsonb('claimed'::text))
+ WHERE id=(SELECT id FROM activity
+ WHERE exploration_id=$1 AND node_id=$2 AND kind='intervention_control'
+ AND metadata->>'intervention_status'=$3
+ AND COALESCE(metadata->>'intervention_handoff','pending')='pending'
+ ORDER BY id DESC LIMIT 1)`, s.expID, intentID, IntentInterventionAccepted)
+ return err
+}
+
+// PendingIntentInterventionMessage returns the accepted human chat turn that
+// should be consumed by the next Worker run. The handoff remains pending until
+// the run has actually started, so a process restart can reconstruct the
+// directed claim without losing the user's message.
+func (s *ExplorationStore) PendingIntentInterventionMessage(intentID int64) (string, string, bool, error) {
+ var requestID, message string
+ err := s.db.QueryRow(`SELECT
+ COALESCE(metadata->>'intervention_request_id',''),
+ COALESCE(metadata->>'intervention_message','')
+ FROM activity
+ WHERE exploration_id=$1 AND node_id=$2 AND kind='intervention_control'
+ AND metadata->>'intervention_status'=$3
+ AND COALESCE(metadata->>'intervention_handoff','pending')='pending'
+ ORDER BY id DESC LIMIT 1`, s.expID, intentID, IntentInterventionAccepted).
+ Scan(&requestID, &message)
+ if err == sql.ErrNoRows {
+ return "", "", false, nil
+ }
+ if err != nil {
+ return "", "", false, err
+ }
+ return requestID, message, true, nil
+}
+
+// RejectIntentIntervention closes a pending control reservation without making
+// it visible in task activity. It is used when lifecycle validation changes or
+// the transcript could not be made durable; a different future request is then
+// free to proceed after the user explicitly resumes the paused intent.
+func (s *ExplorationStore) RejectIntentIntervention(intentID int64, requestID, reason string) error {
+ tx, err := s.db.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+ if err := lockIntentIntervention(tx, intentID); err != nil {
+ return err
+ }
+ _, err = tx.Exec(`UPDATE activity
+ SET metadata=jsonb_set(jsonb_set(metadata,'{intervention_status}',to_jsonb($1::text)),
+ '{intervention_reject_reason}',to_jsonb($2::text))
+ WHERE exploration_id=$3 AND node_id=$4
+ AND metadata->>'intervention_request_id'=$5
+ AND metadata->>'intervention_status'=$6`,
+ IntentInterventionRejected, utf8Clean(reason), s.expID, intentID, requestID, IntentInterventionPending)
+ if err != nil {
+ return err
+ }
+ return tx.Commit()
+}
+
// IntentCleanup summarizes the blackboard records removed when a user cancels
// one worker intent. Global assets and traffic are deliberately not part of this
// operation; exploration anchors disappear only because their owning nodes do.
@@ -1006,7 +1442,8 @@ func (d *DB) TokenTotalsAll() (map[int64]TokenUsage, error) {
// Persisted (unlike Engine.LastActivity's in-memory map), so it survives restarts
// and gives终态任务 a stable "ran until" time for computing run duration.
func (d *DB) LastActivityAll() (map[int64]int64, error) {
- rows, err := d.Query(`SELECT exploration_id, EXTRACT(EPOCH FROM MAX(created_at))::bigint FROM activity GROUP BY exploration_id`)
+ rows, err := d.Query(`SELECT exploration_id, EXTRACT(EPOCH FROM MAX(created_at))::bigint
+ FROM activity WHERE kind IS DISTINCT FROM 'intervention_control' GROUP BY exploration_id`)
if err != nil {
return nil, err
}
@@ -1057,7 +1494,7 @@ func (d *DB) TaskListMetricsAll() (map[int64]TaskListMetrics, error) {
LEFT JOIN LATERAL (
SELECT EXTRACT(EPOCH FROM created_at)::bigint AS created_at
FROM activity
- WHERE exploration_id=task.exploration_id
+ WHERE exploration_id=task.exploration_id AND kind IS DISTINCT FROM 'intervention_control'
ORDER BY created_at DESC
LIMIT 1
) latest_activity ON true
@@ -1188,10 +1625,10 @@ func (s *ExplorationStore) ActivityList(nodeID *int64, sinceID int64, limit int)
const cols = `id, node_id, COALESCE(worker,''), COALESCE(kind,''), COALESCE(tool,''), COALESCE(tool_use_id,''), is_error, COALESCE(summary,''), metadata, created_at, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens`
if nodeID != nil {
rows, err = s.db.Query(`SELECT `+cols+`
-FROM activity WHERE exploration_id=$1 AND node_id=$2 AND id>$3 ORDER BY id LIMIT $4`, s.expID, *nodeID, sinceID, limit)
+FROM activity WHERE exploration_id=$1 AND node_id=$2 AND kind IS DISTINCT FROM 'intervention_control' AND id>$3 ORDER BY id LIMIT $4`, s.expID, *nodeID, sinceID, limit)
} else {
rows, err = s.db.Query(`SELECT `+cols+`
-FROM activity WHERE exploration_id=$1 AND id>$2 ORDER BY id LIMIT $3`, s.expID, sinceID, limit)
+FROM activity WHERE exploration_id=$1 AND kind IS DISTINCT FROM 'intervention_control' AND id>$2 ORDER BY id LIMIT $3`, s.expID, sinceID, limit)
}
if err != nil {
return nil, sinceID, err
@@ -1228,7 +1665,7 @@ func (s *ExplorationStore) ActivityListForTerminalIntent(nodeID, sinceID int64,
JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id
WHERE a.exploration_id=$1 AND a.node_id=$2 AND a.id>$3
AND n.kind='intent' AND n.state IN ('done','blocked','exhausted','stopped')
- AND a.kind NOT IN ('thinking','usage')
+ AND a.kind NOT IN ('thinking','usage','intervention_control')
ORDER BY a.id LIMIT $4`, s.expID, nodeID, sinceID, limit)
if err != nil {
return nil, sinceID, err
@@ -1291,7 +1728,7 @@ func (s *ExplorationStore) ActivityPage(f ActivitySessionFilter, before int64, l
limitArg := len(args) + 1
args = append(args, limit+1) // one extra row to detect older history
q := fmt.Sprintf(`SELECT `+cols+`
-FROM activity WHERE exploration_id=$1%s AND ($%d <= 0 OR id < $%d)
+FROM activity WHERE exploration_id=$1 AND kind IS DISTINCT FROM 'intervention_control'%s AND ($%d <= 0 OR id < $%d)
ORDER BY id DESC LIMIT $%d`, cond, beforeArg, beforeArg, limitArg)
rows, err := s.db.Query(q, args...)
if err != nil {
@@ -1336,7 +1773,7 @@ func (s *ExplorationStore) ActivityPageForTerminalIntent(nodeID, before int64, l
JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id
WHERE a.exploration_id=$1 AND a.node_id=$2
AND n.kind='intent' AND n.state IN ('done','blocked','exhausted','stopped')
- AND a.kind NOT IN ('thinking','usage')
+ AND a.kind NOT IN ('thinking','usage','intervention_control')
AND ($3 <= 0 OR a.id < $3)
ORDER BY a.id DESC LIMIT $4`, s.expID, nodeID, before, limit+1)
if err != nil {
@@ -1372,7 +1809,7 @@ func (s *ExplorationStore) ActivityPageForTerminalIntent(nodeID, before int64, l
// session, since one SSE covers the whole task.
func (s *ExplorationStore) ActivityMaxID() (int64, error) {
var max sql.NullInt64
- err := s.db.QueryRow(`SELECT MAX(id) FROM activity WHERE exploration_id=$1`, s.expID).Scan(&max)
+ err := s.db.QueryRow(`SELECT MAX(id) FROM activity WHERE exploration_id=$1 AND kind IS DISTINCT FROM 'intervention_control'`, s.expID).Scan(&max)
if err != nil {
return 0, err
}
@@ -1382,7 +1819,7 @@ func (s *ExplorationStore) ActivityMaxID() (int64, error) {
// ActivityDetail lazily returns the full detail blob for one step.
func (s *ExplorationStore) ActivityDetail(id int64) (string, error) {
var d sql.NullString
- err := s.db.QueryRow(`SELECT detail FROM activity WHERE id=$1 AND exploration_id=$2`, id, s.expID).Scan(&d)
+ err := s.db.QueryRow(`SELECT detail FROM activity WHERE id=$1 AND exploration_id=$2 AND kind IS DISTINCT FROM 'intervention_control'`, id, s.expID).Scan(&d)
if err == sql.ErrNoRows {
return "", nil
}
@@ -1415,7 +1852,7 @@ func (s *ExplorationStore) ActivityTrace(nodeID int64, limit int) ([]Activity, e
limit = 500
}
rows, err := s.db.Query(`SELECT `+traceCols+`
-FROM activity WHERE exploration_id=$1 AND node_id=$2 AND kind NOT IN ('thinking','usage')
+FROM activity WHERE exploration_id=$1 AND node_id=$2 AND kind NOT IN ('thinking','usage','intervention_control')
ORDER BY id LIMIT $3`, s.expID, nodeID, limit)
if err != nil {
return nil, err
@@ -1435,7 +1872,7 @@ func (s *ExplorationStore) ActivityTraceForTerminalIntent(nodeID int64, limit in
JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id
WHERE a.exploration_id=$1 AND a.node_id=$2 AND n.kind='intent'
AND n.state IN ('done','blocked','exhausted','stopped')
- AND a.kind NOT IN ('thinking','usage')
+ AND a.kind NOT IN ('thinking','usage','intervention_control')
ORDER BY a.id LIMIT $3`, s.expID, nodeID, limit)
if err != nil {
return nil, err
@@ -1457,11 +1894,11 @@ func (s *ExplorationStore) ActivityTraceSearch(nodeID *int64, q string, limit in
var err error
if nodeID != nil {
rows, err = s.db.Query(`SELECT `+traceCols+`
-FROM activity WHERE exploration_id=$1 AND node_id=$2 AND kind NOT IN ('thinking','usage')
+FROM activity WHERE exploration_id=$1 AND node_id=$2 AND kind NOT IN ('thinking','usage','intervention_control')
AND (summary ILIKE $3 OR detail ILIKE $3) ORDER BY id LIMIT $4`, s.expID, *nodeID, like, limit)
} else {
rows, err = s.db.Query(`SELECT `+traceCols+`
-FROM activity WHERE exploration_id=$1 AND node_id IS NOT NULL AND kind NOT IN ('thinking','usage')
+FROM activity WHERE exploration_id=$1 AND node_id IS NOT NULL AND kind NOT IN ('thinking','usage','intervention_control')
AND (summary ILIKE $2 OR detail ILIKE $2) ORDER BY id LIMIT $3`, s.expID, like, limit)
}
if err != nil {
@@ -1483,7 +1920,7 @@ func (s *ExplorationStore) ActivityTraceSearchForTerminalIntent(nodeID int64, q
JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id
WHERE a.exploration_id=$1 AND a.node_id=$2 AND n.kind='intent'
AND n.state IN ('done','blocked','exhausted','stopped')
- AND a.kind NOT IN ('thinking','usage')
+ AND a.kind NOT IN ('thinking','usage','intervention_control')
AND (a.summary ILIKE $3 OR a.detail ILIKE $3)
ORDER BY a.id LIMIT $4`, s.expID, nodeID, like, limit)
if err != nil {
@@ -1505,7 +1942,7 @@ func (s *ExplorationStore) ActivityTraceSearchTerminalIntents(q string, limit in
JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id
WHERE a.exploration_id=$1 AND n.kind='intent'
AND n.state IN ('done','blocked','exhausted','stopped')
- AND a.kind NOT IN ('thinking','usage')
+ AND a.kind NOT IN ('thinking','usage','intervention_control')
AND (a.summary ILIKE $2 OR a.detail ILIKE $2)
ORDER BY a.id LIMIT $3`, s.expID, like, limit)
if err != nil {
@@ -1576,7 +2013,7 @@ func (s *ExplorationStore) ActivityTraceSearchExcluding(excludeNodeID int64, q s
like := "%" + q + "%"
rows, err := s.db.Query(`SELECT `+traceCols+`
FROM activity WHERE exploration_id=$1 AND node_id IS NOT NULL AND node_id <> $2
-AND kind NOT IN ('thinking','usage')
+AND kind NOT IN ('thinking','usage','intervention_control')
AND (summary ILIKE $3 OR detail ILIKE $3) ORDER BY id LIMIT $4`, s.expID, excludeNodeID, like, limit)
if err != nil {
return nil, err
@@ -1600,7 +2037,7 @@ func (s *ExplorationStore) ActivityByIDs(ids []int64) ([]Activity, error) {
args[i+1] = id
}
rows, err := s.db.Query(`SELECT id, node_id, COALESCE(kind,''), COALESCE(tool,''), is_error, COALESCE(detail,'')
-FROM activity WHERE exploration_id=$1 AND kind<>'thinking' AND id IN (`+strings.Join(ph, ",")+`) ORDER BY id`, args...)
+FROM activity WHERE exploration_id=$1 AND kind NOT IN ('thinking','intervention_control') AND id IN (`+strings.Join(ph, ",")+`) ORDER BY id`, args...)
if err != nil {
return nil, err
}
@@ -1634,7 +2071,7 @@ func (s *ExplorationStore) ActivityByIDsForTerminalIntents(ids []int64) ([]Activ
rows, err := s.db.Query(`SELECT a.id, a.node_id, COALESCE(a.kind,''), COALESCE(a.tool,''), a.is_error, COALESCE(a.detail,'')
FROM activity a
JOIN exploration_nodes n ON n.id=a.node_id AND n.exploration_id=a.exploration_id
- WHERE a.exploration_id=$1 AND a.kind NOT IN ('thinking','usage') AND n.kind='intent'
+ WHERE a.exploration_id=$1 AND a.kind NOT IN ('thinking','usage','intervention_control') AND n.kind='intent'
AND n.state IN ('done','blocked','exhausted','stopped')
AND a.id IN (`+strings.Join(ph, ",")+`) ORDER BY a.id`, args...)
if err != nil {
diff --git a/db/intent_intervention_lifecycle_test.go b/db/intent_intervention_lifecycle_test.go
new file mode 100644
index 0000000..c27a746
--- /dev/null
+++ b/db/intent_intervention_lifecycle_test.go
@@ -0,0 +1,204 @@
+package db
+
+import (
+ "encoding/json"
+ "errors"
+ "sync"
+ "testing"
+)
+
+func newRunningInterventionIntent(t *testing.T, d *DB, label string) (*ExplorationStore, int64, func()) {
+ t.Helper()
+ task, err := d.CreateTask("worker intervention "+label, "verify intervention protocol", nil, 0, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ store := d.Exploration(task.ExplorationID)
+ intentID, err := store.AddIntent(map[string]any{"summary": label}, 1, nil, "planner")
+ if err != nil {
+ _ = d.DeleteTask(task.ID)
+ t.Fatal(err)
+ }
+ if claimed, err := store.ClaimIntent(intentID, "work#1"); err != nil || !claimed {
+ _ = d.DeleteTask(task.ID)
+ t.Fatalf("claim intervention intent: claimed=%v err=%v", claimed, err)
+ }
+ return store, intentID, func() { _ = d.DeleteTask(task.ID) }
+}
+
+func TestIntentInterventionReservationAndAcceptanceAreIdempotent(t *testing.T) {
+ d, err := Open(testDSN(t))
+ if err != nil {
+ t.Skipf("postgres unavailable (%v) - skipping", err)
+ }
+ defer d.Close()
+ store, intentID, cleanup := newRunningInterventionIntent(t, d, "idempotent")
+ defer cleanup()
+
+ const requestID = "intervention-idempotent-1"
+ const message = "停止枚举,优先验证登录越权"
+ reserved, err := store.ReserveIntentIntervention(intentID, requestID, message)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if reserved.Status != IntentInterventionPending || reserved.ActivityID <= 0 || reserved.IntentState != "running" {
+ t.Fatalf("unexpected reservation: %+v", reserved)
+ }
+
+ replayed, err := store.ReserveIntentIntervention(intentID, requestID, message)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if replayed.ActivityID != reserved.ActivityID || replayed.Status != IntentInterventionPending {
+ t.Fatalf("reservation retry created a different operation: first=%+v retry=%+v", reserved, replayed)
+ }
+ if _, err := store.ReserveIntentIntervention(intentID, requestID, "改成另一条 Worker 消息"); !errors.Is(err, ErrIntentInterventionConflict) {
+ t.Fatalf("same request_id with different message error=%v, want intervention conflict", err)
+ }
+ if _, err := store.ReserveIntentIntervention(intentID, "intervention-other", message); !errors.Is(err, ErrIntentInterventionConflict) {
+ t.Fatalf("second pending request error=%v, want intervention conflict", err)
+ }
+
+ if changed, err := store.CompareAndSetIntentState(intentID, "running", "paused"); err != nil || !changed {
+ t.Fatalf("settle intent before acceptance: changed=%v err=%v", changed, err)
+ }
+ activity, duplicate, err := store.AcceptIntentIntervention(intentID, requestID, message)
+ if err != nil || duplicate {
+ t.Fatalf("accept intervention: duplicate=%v err=%v", duplicate, err)
+ }
+ if activity.ID <= reserved.ActivityID || activity.NodeID == nil || *activity.NodeID != intentID ||
+ activity.Worker != "user" || activity.Kind != "user" || activity.Summary != message || activity.Detail != message {
+ t.Fatalf("accepted audit mismatch: %+v", activity)
+ }
+ var metadata map[string]any
+ if err := json.Unmarshal(activity.Metadata, &metadata); err != nil {
+ t.Fatal(err)
+ }
+ if metadata["intervention_request_id"] != requestID || metadata["intervention_status"] != IntentInterventionAccepted {
+ t.Fatalf("accepted metadata=%v", metadata)
+ }
+ node, err := store.GetNode(intentID)
+ if err != nil || node == nil || node.State != "open" || node.Owner != "" {
+ t.Fatalf("accepted intent=%+v err=%v, want reopened and unowned", node, err)
+ }
+
+ again, duplicate, err := store.AcceptIntentIntervention(intentID, requestID, message)
+ if err != nil || !duplicate || again.ID != activity.ID {
+ t.Fatalf("accept retry: activity=%+v duplicate=%v err=%v", again, duplicate, err)
+ }
+ lookup, err := store.IntentInterventionByRequest(intentID, requestID)
+ if err != nil || lookup == nil || lookup.Status != IntentInterventionAccepted || lookup.ActivityID != activity.ID {
+ t.Fatalf("accepted lookup=%+v err=%v", lookup, err)
+ }
+ var rows int
+ if err := d.QueryRow(`SELECT COUNT(*) FROM activity
+ WHERE exploration_id=$1 AND node_id=$2
+ AND metadata->>'intervention_request_id'=$3`, store.ID(), intentID, requestID).Scan(&rows); err != nil {
+ t.Fatal(err)
+ }
+ if rows != 2 {
+ t.Fatalf("Worker message rows=%d, want one hidden control row and one visible user row", rows)
+ }
+ activities, cursor, err := store.ActivityList(&intentID, 0, 10)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(activities) != 1 || activities[0].ID != activity.ID || cursor != activity.ID {
+ t.Fatalf("public activities leaked control row: activities=%+v cursor=%d", activities, cursor)
+ }
+ maxID, err := store.ActivityMaxID()
+ if err != nil || maxID != activity.ID {
+ t.Fatalf("public activity cursor=%d err=%v, want visible user row %d", maxID, err, activity.ID)
+ }
+ if detail, err := store.ActivityDetail(reserved.ActivityID); err != nil || detail != "" {
+ t.Fatalf("hidden control detail=%q err=%v", detail, err)
+ }
+}
+
+func TestIntentInterventionConcurrentReservationHasSingleDurableWinner(t *testing.T) {
+ d, err := Open(testDSN(t))
+ if err != nil {
+ t.Skipf("postgres unavailable (%v) - skipping", err)
+ }
+ defer d.Close()
+ store, intentID, cleanup := newRunningInterventionIntent(t, d, "concurrent")
+ defer cleanup()
+
+ const contenders = 12
+ const requestID = "intervention-concurrent-1"
+ const message = "只保留被动请求并验证当前假设"
+ start := make(chan struct{})
+ results := make(chan *IntentIntervention, contenders)
+ errs := make(chan error, contenders)
+ var wg sync.WaitGroup
+ for range contenders {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ <-start
+ reservation, reserveErr := store.ReserveIntentIntervention(intentID, requestID, message)
+ results <- reservation
+ errs <- reserveErr
+ }()
+ }
+ close(start)
+ wg.Wait()
+ close(results)
+ close(errs)
+ for err := range errs {
+ if err != nil {
+ t.Fatalf("concurrent reservation: %v", err)
+ }
+ }
+ var activityID int64
+ for result := range results {
+ if result == nil || result.ActivityID <= 0 {
+ t.Fatalf("invalid concurrent reservation: %+v", result)
+ }
+ if activityID == 0 {
+ activityID = result.ActivityID
+ } else if result.ActivityID != activityID {
+ t.Fatalf("concurrent requests produced activity ids %d and %d", activityID, result.ActivityID)
+ }
+ }
+ var rows int
+ if err := d.QueryRow(`SELECT COUNT(*) FROM activity
+ WHERE exploration_id=$1 AND node_id=$2
+ AND metadata->>'intervention_request_id'=$3`, store.ID(), intentID, requestID).Scan(&rows); err != nil {
+ t.Fatal(err)
+ }
+ if rows != 1 {
+ t.Fatalf("concurrent intervention rows=%d, want 1", rows)
+ }
+}
+
+func TestIntentInterventionAcceptanceCanRetryAfterStateSettlement(t *testing.T) {
+ d, err := Open(testDSN(t))
+ if err != nil {
+ t.Skipf("postgres unavailable (%v) - skipping", err)
+ }
+ defer d.Close()
+ store, intentID, cleanup := newRunningInterventionIntent(t, d, "settlement retry")
+ defer cleanup()
+
+ const requestID = "intervention-settle-1"
+ const message = "立刻停止当前动作并检查权限边界"
+ reserved, err := store.ReserveIntentIntervention(intentID, requestID, message)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := store.AcceptIntentIntervention(intentID, requestID, message); !errors.Is(err, ErrIntentInterventionConflict) {
+ t.Fatalf("accept before worker settlement error=%v, want conflict", err)
+ }
+ pending, err := store.IntentInterventionByRequest(intentID, requestID)
+ if err != nil || pending == nil || pending.Status != IntentInterventionPending || pending.ActivityID != reserved.ActivityID {
+ t.Fatalf("failed acceptance lost reservation: pending=%+v err=%v", pending, err)
+ }
+ if changed, err := store.CompareAndSetIntentState(intentID, "running", "paused"); err != nil || !changed {
+ t.Fatalf("settle intent: changed=%v err=%v", changed, err)
+ }
+ activity, duplicate, err := store.AcceptIntentIntervention(intentID, requestID, message)
+ if err != nil || duplicate || activity.ID <= reserved.ActivityID {
+ t.Fatalf("retry acceptance: activity=%+v duplicate=%v err=%v", activity, duplicate, err)
+ }
+}
diff --git a/server/engine.go b/server/engine.go
index f3c7074..2754c09 100644
--- a/server/engine.go
+++ b/server/engine.go
@@ -75,6 +75,7 @@ const (
)
var errWorkControlConflict = errors.New("work control conflict")
+var errWorkInterventionConflict = errors.New("work intervention conflict")
// retryableWorkerModelError excludes errors already handled by the task router.
// In particular, a quota error after partial streaming advances the task cursor
@@ -130,6 +131,17 @@ type Engine struct {
// runWorkerStep has stopped writing and committed its final state.
workMu sync.Mutex
work map[int64]*workExecution
+ // interventionLocks serializes reservation, cancellation and message hand-off
+ // per intent. PostgreSQL separately serializes the durable transaction steps.
+ interventionLocks sync.Map // intentID -> *sync.Mutex
+
+ // directedClaims keeps intents with a new human message ahead of the normal
+ // frontier without mutating their persisted priority. runWorkerStep installs
+ // the entry before releasing ControlWork, so a single worker slot cannot grab
+ // unrelated work in the paused->open hand-off gap.
+ directedMu sync.Mutex
+ directedClaims map[string][]int64
+ directedReady map[int64]bool
// 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
@@ -162,7 +174,14 @@ type taskRuntime struct {
type workExecution struct {
cancel context.CancelCauseFunc
done chan error
- action string // user action: pause | cancel
+ action string // user action: pause | cancel | intervene
+}
+
+type WorkInterventionResult struct {
+ RequestID string
+ ActivityID int64
+ State string
+ Duplicate bool
}
// nextPlannerRound returns the next planner round number for a task (1-based).
@@ -285,6 +304,12 @@ func (e *Engine) StopTask(taskID string) {
e.stamped.Delete(taskID)
e.inflight.Delete(taskID)
e.coordStarted.Delete(taskID)
+ e.directedMu.Lock()
+ for _, intentID := range e.directedClaims[taskID] {
+ delete(e.directedReady, intentID)
+ }
+ delete(e.directedClaims, taskID)
+ e.directedMu.Unlock()
e.deleteMu.Lock()
e.deleting.Delete(taskID)
e.deleteMu.Unlock()
@@ -390,7 +415,7 @@ func NewEngine(m *Manager) *Engine {
return &Engine{m: m, debounce: 800 * time.Millisecond, bc: NewBroadcaster(),
execCancel: map[string]context.CancelCauseFunc{}, execCtx: map[string]context.Context{},
work: map[int64]*workExecution{}, steerBox: map[int64][]string{},
- runtimes: map[string]*taskRuntime{}}
+ directedClaims: map[string][]int64{}, directedReady: map[int64]bool{}, runtimes: map[string]*taskRuntime{}}
}
// registerWork records the cancel for the work currently running intentID.
@@ -425,7 +450,7 @@ func (e *Engine) detachWork(intentID int64) (action string, complete func(error)
// worker has fully stopped writing. Cancellation cleanup is performed by the API
// handler after this returns; pause state is committed by runWorkerStep itself.
func (e *Engine) ControlWork(ctx context.Context, intentID int64, action string) error {
- if action != "pause" && action != "cancel" {
+ if action != "pause" && action != "cancel" && action != "intervene" {
return fmt.Errorf("unsupported work action %q", action)
}
if ctx == nil {
@@ -444,8 +469,11 @@ func (e *Engine) ControlWork(ctx context.Context, intentID int64, action string)
run.action = action
done := run.done
cause := error(agent.AbortWorkPausedByUser)
- if action == "cancel" {
+ switch action {
+ case "cancel":
cause = agent.AbortWorkCancelledByUser
+ case "intervene":
+ cause = agent.AbortWorkIntervenedByUser
}
run.cancel(cause)
e.workMu.Unlock()
@@ -464,6 +492,287 @@ func (e *Engine) ControlWork(ctx context.Context, intentID int64, action string)
}
}
+func (e *Engine) lockWorkIntervention(intentID int64) func() {
+ value, _ := e.interventionLocks.LoadOrStore(intentID, &sync.Mutex{})
+ mu := value.(*sync.Mutex)
+ mu.Lock()
+ return mu.Unlock
+}
+
+func (e *Engine) validateWorkInterventionTask(t *Task, allowAdmissionPause bool) error {
+ if t == nil {
+ return fmt.Errorf("%w: task not found", errWorkInterventionConflict)
+ }
+ lifecycle := t.lifecycleSnapshot()
+ switch {
+ case e.IsDeleting(t.ID):
+ return fmt.Errorf("%w: 任务正在删除,无法向 Worker 发送消息", errWorkInterventionConflict)
+ case lifecycle.Paused || (!allowAdmissionPause && e.IsPaused(t.ID)):
+ return fmt.Errorf("%w: 任务已暂停,无法向 Worker 发送消息", errWorkInterventionConflict)
+ case lifecycle.Queued:
+ return fmt.Errorf("%w: 排队中的任务无法向 Worker 发送消息", errWorkInterventionConflict)
+ case isTerminalStatus(lifecycle.Status):
+ return fmt.Errorf("%w: 终态任务无法向 Worker 发送消息", errWorkInterventionConflict)
+ case e.isSettling(t.ID):
+ return fmt.Errorf("%w: 任务正在收尾,无法向 Worker 发送消息", errWorkInterventionConflict)
+ default:
+ return nil
+ }
+}
+
+func (e *Engine) abandonWorkIntervention(t *Task, intentID int64, requestID string, reason error) {
+ e.removeDirectedClaim(t.ID, intentID)
+ target := "open"
+ lifecycle := t.lifecycleSnapshot()
+ if e.IsDeleting(t.ID) || isTerminalStatus(lifecycle.Status) {
+ target = "stopped"
+ } else if e.isSettling(t.ID) {
+ target = "exhausted"
+ }
+ if changed, err := t.Store.CompareAndSetIntentState(intentID, "paused", target); err != nil {
+ log.Printf("[worker intervention] task %s 意图 #%d 补偿 paused->%s 失败: %v", t.ID, intentID, target, err)
+ } else if changed && target == "open" {
+ t.NotifyWorker()
+ }
+ if err := t.Store.RejectIntentIntervention(intentID, requestID, reason.Error()); err != nil {
+ log.Printf("[worker intervention] task %s 意图 #%d 标记请求 %s rejected 失败: %v", t.ID, intentID, requestID, err)
+ }
+}
+
+// InterveneWork performs the complete Worker message protocol. Once the request
+// has a durable pending reservation it no longer depends on the HTTP request
+// context: the caller supplies a server-owned bounded context so disconnects do
+// not strand the intent. The accepted DB transition is running -> paused ->
+// open, with the visible user message and reopen committed atomically.
+func (e *Engine) InterveneWork(ctx context.Context, t *Task, intentID int64, requestID, message string) (WorkInterventionResult, error) {
+ return e.interveneWork(ctx, t, intentID, requestID, message, false)
+}
+
+func (e *Engine) interveneWork(ctx context.Context, t *Task, intentID int64, requestID, message string, allowAdmissionPause bool) (WorkInterventionResult, error) {
+ out := WorkInterventionResult{RequestID: requestID}
+ if t == nil || t.Store == nil {
+ return out, fmt.Errorf("%w: task not found", errWorkInterventionConflict)
+ }
+ if ctx == nil {
+ ctx = context.Background()
+ }
+ unlock := e.lockWorkIntervention(intentID)
+ defer unlock()
+
+ existing, err := t.Store.IntentInterventionByRequest(intentID, requestID)
+ if err != nil {
+ return out, err
+ }
+ if existing != nil {
+ if existing.Message != message {
+ return out, fmt.Errorf("%w: request_id 已用于不同的 Worker 消息", errWorkInterventionConflict)
+ }
+ if existing.Status == db.IntentInterventionAccepted {
+ out.ActivityID, out.State, out.Duplicate = existing.ActivityID, existing.IntentState, true
+ if out.State == "open" && e.validateWorkInterventionTask(t, allowAdmissionPause) == nil {
+ e.queueDirectedClaim(t.ID, intentID)
+ e.readyDirectedClaim(intentID)
+ t.NotifyWorker()
+ }
+ return out, nil
+ }
+ if existing.Status != db.IntentInterventionPending {
+ return out, fmt.Errorf("%w: Worker 消息请求状态为 %s", errWorkInterventionConflict, existing.Status)
+ }
+ }
+ if err := e.validateWorkInterventionTask(t, allowAdmissionPause); err != nil {
+ lifecycle := t.lifecycleSnapshot()
+ if existing != nil && (e.IsDeleting(t.ID) || e.isSettling(t.ID) || isTerminalStatus(lifecycle.Status)) {
+ e.abandonWorkIntervention(t, intentID, requestID, err)
+ }
+ return out, err
+ }
+ _, worker := e.snapshotFor(t)
+ if worker == nil {
+ return out, fmt.Errorf("%w: worker 尚未就绪", errWorkInterventionConflict)
+ }
+ if existing == nil {
+ existing, err = t.Store.ReserveIntentIntervention(intentID, requestID, message)
+ if err != nil {
+ if errors.Is(err, db.ErrIntentInterventionConflict) {
+ return out, fmt.Errorf("%w: %v", errWorkInterventionConflict, err)
+ }
+ return out, err
+ }
+ }
+ out.ActivityID = existing.ActivityID
+
+ // A pending reservation can be recovered after a process restart. An open
+ // row is fenced back to paused before publishing the user turn; a running row
+ // is stopped through the normal live-work protocol so no late writer races us.
+ for {
+ node, getErr := t.Store.GetNode(intentID)
+ if getErr != nil {
+ return out, getErr
+ }
+ if node == nil {
+ return out, fmt.Errorf("%w: intent not found", errWorkInterventionConflict)
+ }
+ switch node.State {
+ case "running":
+ controlErr := e.ControlWork(ctx, intentID, "intervene")
+ if controlErr != nil {
+ // The per-work 30s wait may expire before the server-owned operation
+ // context. Keep waiting for the named cancellation to settle; timeout
+ // release marks the run no-handoff so a late exit cannot install a
+ // permanent directed-claim barrier behind our back. A freshly claimed
+ // intent can also be running just before registerWork; retrying closes
+ // that small claim-to-registration window.
+ for {
+ settled, readErr := t.Store.GetNode(intentID)
+ if readErr != nil {
+ return out, readErr
+ }
+ if settled != nil && settled.State == "paused" {
+ break
+ }
+ if ctx.Err() != nil || settled == nil || settled.State != "running" {
+ e.removeDirectedClaim(t.ID, intentID)
+ return out, fmt.Errorf("%w: %v", errWorkInterventionConflict, controlErr)
+ }
+ if sleepCtx(ctx, 50*time.Millisecond) {
+ e.removeDirectedClaim(t.ID, intentID)
+ return out, fmt.Errorf("%w: %v", errWorkInterventionConflict, controlErr)
+ }
+ if errors.Is(controlErr, errWorkControlConflict) {
+ controlErr = e.ControlWork(ctx, intentID, "intervene")
+ }
+ }
+ }
+ case "open":
+ changed, setErr := t.Store.CompareAndSetIntentState(intentID, "open", "paused")
+ if setErr != nil {
+ return out, setErr
+ }
+ if !changed {
+ continue
+ }
+ e.queueDirectedClaim(t.ID, intentID)
+ case "paused":
+ e.queueDirectedClaim(t.ID, intentID)
+ default:
+ e.abandonWorkIntervention(t, intentID, requestID,
+ fmt.Errorf("意图已进入 %s,无法发送新消息", node.State))
+ return out, fmt.Errorf("%w: 意图已进入 %s,无法发送新消息", errWorkInterventionConflict, node.State)
+ }
+ break
+ }
+ e.queueDirectedClaim(t.ID, intentID)
+
+ // Close lifecycle races that began while cancellation was settling. A task
+ // pause or queue preserves the pending message and paused intent for a later
+ // admission; terminal, settling and deleting transitions close the request.
+ if err := e.validateWorkInterventionTask(t, allowAdmissionPause); err != nil {
+ // The user message has not entered the transcript yet. Preserve the
+ // reservation and paused intent so the same request can be retried after a
+ // transient task pause; do not reopen it behind a terminal transition.
+ e.removeDirectedClaim(t.ID, intentID)
+ lifecycle := t.lifecycleSnapshot()
+ if e.IsDeleting(t.ID) || e.isSettling(t.ID) || isTerminalStatus(lifecycle.Status) {
+ e.abandonWorkIntervention(t, intentID, requestID, err)
+ }
+ return out, err
+ }
+ activity, duplicate, err := t.Store.AcceptIntentIntervention(intentID, requestID, message)
+ if err != nil {
+ // Keep this intent paused so a retry of the same request can safely finish;
+ // remove only the task-wide claim barrier so unrelated workers keep moving.
+ e.removeDirectedClaim(t.ID, intentID)
+ return out, err
+ }
+ if !duplicate {
+ e.publishStoredActivity(t, activity)
+ t.NotifyWorkerMessage(intentID, activity.ID, message)
+ }
+ e.readyDirectedClaim(intentID)
+ t.NotifyWorker()
+ node, err := t.Store.GetNode(intentID)
+ if err != nil {
+ return out, err
+ }
+ out.Duplicate = duplicate
+ if node != nil {
+ out.State = node.State
+ }
+ return out, nil
+}
+
+// RecoverWorkerMessages reconstructs unfinished Worker chat handoffs before
+// ordinary frontier claiming. Pending messages are fenced in paused until an
+// agent runtime is available; accepted messages regain their directed priority
+// and are consumed by ExecuteWithMessage on the next claim.
+func (e *Engine) RecoverWorkerMessages(ctx context.Context, t *Task) error {
+ return e.recoverWorkerMessages(ctx, t, false)
+}
+
+// recoverWorkerMessagesForAdmission runs while the scheduler still owns the
+// Engine pause barrier but has already committed a runnable task lifecycle. It
+// restores directed hand-offs before Resume wakes any worker, so ordinary
+// frontier work cannot overtake the human message.
+func (e *Engine) recoverWorkerMessagesForAdmission(ctx context.Context, t *Task) error {
+ return e.recoverWorkerMessages(ctx, t, true)
+}
+
+func (e *Engine) recoverWorkerMessages(ctx context.Context, t *Task, allowAdmissionPause bool) error {
+ if t == nil || t.Store == nil {
+ return nil
+ }
+ items, err := t.Store.RecoverableIntentInterventions()
+ if err != nil {
+ return err
+ }
+ lifecycle := t.lifecycleSnapshot()
+ runnable := !e.IsDeleting(t.ID) && !e.isSettling(t.ID) && !isTerminalStatus(lifecycle.Status) &&
+ !lifecycle.Paused && !lifecycle.Queued && (allowAdmissionPause || !e.IsPaused(t.ID))
+ if !runnable {
+ for _, item := range items {
+ e.removeDirectedClaim(t.ID, item.IntentID)
+ }
+ return nil
+ }
+ for _, item := range items {
+ switch item.Status {
+ case db.IntentInterventionPending:
+ if item.IntentState == "open" {
+ if changed, setErr := t.Store.CompareAndSetIntentState(item.IntentID, "open", "paused"); setErr != nil {
+ return setErr
+ } else if changed {
+ item.IntentState = "paused"
+ }
+ }
+ if _, worker := e.snapshotFor(t); worker == nil {
+ continue
+ }
+ if _, recoverErr := e.interveneWork(ctx, t, item.IntentID, item.RequestID, item.Message, allowAdmissionPause); recoverErr != nil {
+ log.Printf("[worker chat] task %s 恢复意图 #%d 待处理消息失败: %v", t.ID, item.IntentID, recoverErr)
+ }
+ case db.IntentInterventionAccepted:
+ // Reconstruct the Planner-facing event while the Worker hand-off is still
+ // pending. The durable graph_overview view remains authoritative even if
+ // this notification was already consumed before a later restart.
+ t.NotifyWorkerMessage(item.IntentID, item.ActivityID, item.Message)
+ if item.IntentState == "paused" {
+ if changed, setErr := t.Store.CompareAndSetIntentState(item.IntentID, "paused", "open"); setErr != nil {
+ return setErr
+ } else if changed {
+ item.IntentState = "open"
+ }
+ }
+ if item.IntentState == "open" {
+ e.queueDirectedClaim(t.ID, item.IntentID)
+ e.readyDirectedClaim(item.IntentID)
+ t.NotifyWorker()
+ }
+ }
+ }
+ return nil
+}
+
// releaseWorkControl drops only this caller's reservation after its wait is
// cancelled. The work context stays cancelled; runWorkerStep recognizes the
// named cancellation cause and settles the intent into the recoverable paused
@@ -471,7 +780,11 @@ func (e *Engine) ControlWork(ctx context.Context, intentID int64, action string)
func (e *Engine) releaseWorkControl(intentID int64, run *workExecution, action string) {
e.workMu.Lock()
if current := e.work[intentID]; current == run && current.action == action {
- current.action = ""
+ if action == "intervene" {
+ current.action = "intervene_no_handoff"
+ } else {
+ current.action = ""
+ }
}
e.workMu.Unlock()
}
@@ -610,6 +923,19 @@ func (e *Engine) emitActivity(t *Task, r db.Activity) db.Activity {
return r
}
+// publishStoredActivity broadcasts an activity inserted by a larger DB
+// transaction, avoiding a second INSERT through emitActivity.
+func (e *Engine) publishStoredActivity(t *Task, r db.Activity) {
+ if t == nil || r.ID <= 0 {
+ return
+ }
+ if r.CreatedAt.IsZero() {
+ r.CreatedAt = time.Now()
+ }
+ e.bc.Publish(t.ID, r)
+ e.touch(t.ID)
+}
+
// appendActivity persists one activity row, retrying briefly on write failure.
// Concurrent planner + worker inserts into the same exploration's activity log
// occasionally fail; a couple of quick retries recover most. Crucially, every
@@ -876,13 +1202,13 @@ func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) {
}
_, worker := e.snapshotFor(t)
if worker == nil {
- if sleepCtx(ctx, 1500*time.Millisecond) {
+ if waitWorker(ctx, t, 1500*time.Millisecond) {
return
}
continue
}
if e.IsPaused(t.ID) {
- if sleepCtx(ctx, 1000*time.Millisecond) {
+ if waitWorker(ctx, t, 1000*time.Millisecond) {
return
}
continue // user-paused: don't claim/execute intents
@@ -891,13 +1217,13 @@ func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) {
return
}
if e.isSettling(t.ID) {
- if sleepCtx(ctx, 1000*time.Millisecond) {
+ if waitWorker(ctx, t, 1000*time.Millisecond) {
return
}
continue // 任务超时收尾中:不再领新意图(在跑的自行收尾,协调器等其 drain)
}
if isTerminalStatus(e.m.TaskStatus(t.ID)) {
- if sleepCtx(ctx, 1000*time.Millisecond) {
+ if waitWorker(ctx, t, 1000*time.Millisecond) {
return
}
continue // 任务已终态(done/failed/timeout):停止领取遗留意图,别在完成后空跑 frontier
@@ -907,7 +1233,7 @@ func (e *Engine) workerLoop(ctx context.Context, t *Task, name string) {
}
claimed := e.runWorkerStep(ctx, t, name, worker)
e.decInflight(t.ID)
- if !claimed && sleepCtx(ctx, 800*time.Millisecond) {
+ if !claimed && waitWorker(ctx, t, 800*time.Millisecond) {
return
}
}
@@ -946,9 +1272,31 @@ func (e *Engine) runWorkerStep(ctx context.Context, t *Task, name string, worker
workCtx = intercept.WithTaskContext(workCtx, t.ID, fmt.Sprintf("%s · #%d", name, iid), taskEmit)
hooks := steerHooks{inner: t.Guard.Hooks(), drain: func() (string, bool) { return e.drainSteer(iid) }}
wTaskID, _ := strconv.ParseInt(t.ID, 10, 64)
+ chatRequestID, chatMessage, hasChatMessage, chatErr := t.Store.PendingIntentInterventionMessage(intent.ID)
+ if chatErr != nil {
+ log.Printf("[worker %s] task %s 读取意图 #%d 待处理人工对话失败: %v", name, t.ID, intent.ID, chatErr)
+ if err := transitionIntentState(t.Store, intent.ID, "running", "open"); err != nil {
+ log.Printf("[worker %s] task %s 意图 #%d 人工对话读取失败后回退失败: %v", name, t.ID, intent.ID, err)
+ }
+ _, completeWork := e.detachWork(intent.ID)
+ completeWork(chatErr)
+ return true
+ }
e.BeginLLMCall(t.ID)
- reason, wrote, err := worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding)
+ var reason harness.TerminalReason
+ var wrote agent.WriteCounts
+ var err error
+ if hasChatMessage {
+ reason, wrote, err = worker.ExecuteWithMessage(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding, chatRequestID, chatMessage)
+ } else {
+ reason, wrote, err = worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding)
+ }
e.EndLLMCall(t.ID)
+ if hasChatMessage {
+ if markErr := t.Store.MarkIntentInterventionHandoffClaimed(intent.ID); markErr != nil {
+ log.Printf("[worker %s] task %s 标记意图 #%d 人工对话已消费失败: %v", name, t.ID, intent.ID, markErr)
+ }
+ }
// model_error 收场 → 额外重跑几次(退避后再试)。仅在意图仍属本 work、任务
// 未暂停/未终止/未取消【且未进入收尾】时重试;否则让位给对应分支处理(收尾期不
// 再重试,避免退避挤占其他 worker 的优雅收尾窗口)。
@@ -961,7 +1309,11 @@ func (e *Engine) runWorkerStep(ctx context.Context, t *Task, name string, worker
break // 退避期间被取消(终止/暂停)→ 交给下方分支处理
}
e.BeginLLMCall(t.ID)
- reason, wrote, err = worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding)
+ if hasChatMessage {
+ reason, wrote, err = worker.ExecuteWithMessage(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding, chatRequestID, chatMessage)
+ } else {
+ reason, wrote, err = worker.Execute(workCtx, name, wTaskID, e.m.assets, t.Store, intent, hooks, emit, e.m.enrich, t.NotifyFinding)
+ }
e.EndLLMCall(t.ID)
}
// Capture kill state before detachWork cancels workCtx. kill = this work's
@@ -979,6 +1331,8 @@ func (e *Engine) runWorkerStep(ctx context.Context, t *Task, name string, worker
switch {
case errors.Is(workCause, agent.AbortWorkPausedByUser):
action = "pause"
+ case errors.Is(workCause, agent.AbortWorkIntervenedByUser):
+ action = "intervene"
case errors.Is(workCause, agent.AbortWorkCancelledByUser):
action = "cancel"
}
@@ -995,6 +1349,22 @@ func (e *Engine) runWorkerStep(ctx context.Context, t *Task, name string, worker
e.touch(t.ID)
return true
}
+ if action == "intervene" || action == "intervene_no_handoff" {
+ controlErr = transitionIntentState(t.Store, intent.ID, "running", "paused")
+ if controlErr != nil {
+ log.Printf("[worker %s] task %s 意图 #%d 对话暂停落库失败: %v", name, t.ID, intent.ID, controlErr)
+ return true
+ }
+ // Install the directed barrier before completeWork releases ControlWork.
+ // With one worker slot this prevents its next loop from taking unrelated
+ // frontier work during the message hand-off.
+ if action == "intervene" {
+ e.queueDirectedClaim(t.ID, intent.ID)
+ }
+ log.Printf("[worker %s] task %s 意图 #%d 已暂停,等待新的对话意图", name, t.ID, intent.ID)
+ e.touch(t.ID)
+ return true
+ }
if action == "cancel" {
// Park the stopped run in paused before handing cleanup to the API. If the
// request disconnects after cancellation, the intent remains recoverable and
@@ -1099,7 +1469,114 @@ func sleepCtx(ctx context.Context, d time.Duration) (done bool) {
}
}
+func waitWorker(ctx context.Context, t *Task, d time.Duration) (done bool) {
+ timer := time.NewTimer(d)
+ defer timer.Stop()
+ var wake <-chan struct{}
+ if t != nil {
+ wake = t.workerWake
+ }
+ select {
+ case <-ctx.Done():
+ return true
+ case <-wake:
+ return false
+ case <-timer.C:
+ return false
+ }
+}
+
+func (e *Engine) queueDirectedClaim(taskID string, intentID int64) {
+ if taskID == "" || intentID <= 0 {
+ return
+ }
+ e.directedMu.Lock()
+ defer e.directedMu.Unlock()
+ for _, id := range e.directedClaims[taskID] {
+ if id == intentID {
+ return
+ }
+ }
+ e.directedClaims[taskID] = append(e.directedClaims[taskID], intentID)
+ e.directedReady[intentID] = false
+}
+
+func (e *Engine) readyDirectedClaim(intentID int64) {
+ e.directedMu.Lock()
+ if _, exists := e.directedReady[intentID]; exists {
+ e.directedReady[intentID] = true
+ }
+ e.directedMu.Unlock()
+}
+
+func (e *Engine) removeDirectedClaim(taskID string, intentID int64) {
+ e.directedMu.Lock()
+ defer e.directedMu.Unlock()
+ items := e.directedClaims[taskID]
+ for i, id := range items {
+ if id != intentID {
+ continue
+ }
+ items = append(items[:i], items[i+1:]...)
+ delete(e.directedReady, intentID)
+ if len(items) == 0 {
+ delete(e.directedClaims, taskID)
+ } else {
+ e.directedClaims[taskID] = items
+ }
+ return
+ }
+}
+
+// claimDirected tries every pending human-message hand-off before ordinary work
+// frontier. blocked=true means at least one hand-off is still paused/running;
+// callers must not let a just-released single worker slot overtake it.
+func (e *Engine) claimDirected(t *Task, name string) (intent *db.Node, blocked bool) {
+ e.directedMu.Lock()
+ ids := append([]int64(nil), e.directedClaims[t.ID]...)
+ ready := make(map[int64]bool, len(ids))
+ for _, id := range ids {
+ ready[id] = e.directedReady[id]
+ }
+ e.directedMu.Unlock()
+ for _, id := range ids {
+ node, err := t.Store.GetNode(id)
+ if err != nil {
+ return nil, true
+ }
+ if node == nil {
+ e.removeDirectedClaim(t.ID, id)
+ continue
+ }
+ switch node.State {
+ case "open":
+ if !ready[id] {
+ blocked = true
+ continue
+ }
+ ok, err := t.Store.ClaimIntent(id, name)
+ if err != nil {
+ return nil, true
+ }
+ if ok {
+ e.removeDirectedClaim(t.ID, id)
+ node.State, node.Owner = "running", name
+ return node, false
+ }
+ blocked = true
+ case "paused", "running":
+ blocked = true
+ default:
+ e.removeDirectedClaim(t.ID, id)
+ }
+ }
+ return nil, blocked
+}
+
func (e *Engine) claimNext(t *Task, name string) *db.Node {
+ if intent, blocked := e.claimDirected(t, name); intent != nil || blocked {
+ return intent
+ }
fr, _ := t.Store.Frontier(20)
for _, in := range fr {
if ok, _ := t.Store.ClaimIntent(in.ID, name); ok {
diff --git a/server/engine_cancelcause_test.go b/server/engine_cancelcause_test.go
index d1f4f7a..63c7324 100644
--- a/server/engine_cancelcause_test.go
+++ b/server/engine_cancelcause_test.go
@@ -45,6 +45,7 @@ func TestControlWorkCarriesUserActionCause(t *testing.T) {
}{
{action: "pause", code: "work_paused_by_user"},
{action: "cancel", code: "work_cancelled_by_user"},
+ {action: "intervene", code: "work_intervened_by_user"},
} {
t.Run(tc.action, func(t *testing.T) {
e := NewEngine(nil)
diff --git a/server/goals.go b/server/goals.go
index c8e6d98..05f2c7e 100644
--- a/server/goals.go
+++ b/server/goals.go
@@ -225,6 +225,13 @@ func (s *Server) hasReadyQueuedTask(excludeID string) bool {
}
func (s *Server) startAdmittedTask(t *Task, mode string) {
+ recoverMessages := s.engine.RecoverWorkerMessages
+ if s.engine.IsPaused(t.ID) {
+ recoverMessages = s.engine.recoverWorkerMessagesForAdmission
+ }
+ if err := recoverMessages(s.ctx, t); err != nil {
+ log.Printf("[worker chat] task %s 恢复待处理人工消息失败: %v", t.ID, err)
+ }
if mode == "bootstrap" {
if !s.engine.beginTaskOperation(t.ID) {
return
diff --git a/server/intent_intervention.go b/server/intent_intervention.go
new file mode 100644
index 0000000..6df99fa
--- /dev/null
+++ b/server/intent_intervention.go
@@ -0,0 +1,125 @@
+package server
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "net/http"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/Autumn-27/artex/db"
+)
+
+const maxWorkerMessageBytes = 64 << 10
+
+func validWorkerMessageRequestID(id string) bool {
+ if id == "" || len(id) > 128 {
+ return false
+ }
+ for _, r := range id {
+ if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') ||
+ (r >= '0' && r <= '9') || r == '-' || r == '_' || r == '.' || r == ':' {
+ continue
+ }
+ return false
+ }
+ return true
+}
+
+// sendWorkerMessage pauses a running Worker when necessary, records one user
+// turn for its existing conversation, and immediately schedules that same
+// intent to continue. A disconnected browser cannot strand an accepted message.
+func (s *Server) sendWorkerMessage(w http.ResponseWriter, r *http.Request) {
+ t, ok := s.m.Task(r.PathValue("id"))
+ if !ok {
+ writeErr(w, http.StatusNotFound, "task not found")
+ return
+ }
+ iid, err := strconv.ParseInt(r.PathValue("iid"), 10, 64)
+ if err != nil || iid <= 0 {
+ writeErr(w, http.StatusBadRequest, "bad intent id")
+ return
+ }
+ r.Body = http.MaxBytesReader(w, r.Body, maxWorkerMessageBytes)
+ var req struct {
+ Message string `json:"message"`
+ RequestID string `json:"request_id"`
+ }
+ if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+ var tooLarge *http.MaxBytesError
+ if errors.As(err, &tooLarge) {
+ writeErr(w, http.StatusRequestEntityTooLarge, "请求体过大")
+ return
+ }
+ writeErr(w, http.StatusBadRequest, "bad json: "+err.Error())
+ return
+ }
+ message := strings.TrimSpace(req.Message)
+ requestID := strings.TrimSpace(req.RequestID)
+ if message == "" {
+ writeErr(w, http.StatusBadRequest, "消息不能为空")
+ return
+ }
+ if len([]rune(message)) > 4000 {
+ writeErr(w, http.StatusBadRequest, "消息不能超过 4000 个字符")
+ return
+ }
+ if !validWorkerMessageRequestID(requestID) {
+ writeErr(w, http.StatusBadRequest, "request_id 必须是 1-128 位字母、数字、-、_、. 或 :")
+ return
+ }
+
+ if !s.engine.beginTaskOperation(t.ID) {
+ writeErr(w, http.StatusConflict, "任务正在删除,无法向 Worker 发送消息")
+ return
+ }
+ defer s.engine.decInflight(t.ID)
+
+ node, err := t.Store.GetNode(iid)
+ if err != nil {
+ writeErr(w, http.StatusInternalServerError, err.Error())
+ return
+ }
+ if node == nil {
+ if inherited, sourceErr := t.Store.GetNodeWithSources(iid); sourceErr == nil && inherited != nil && inherited.Inherited {
+ writeErr(w, http.StatusConflict, "继承意图为只读,不能发送 Worker 消息")
+ return
+ }
+ writeErr(w, http.StatusNotFound, "intent not found")
+ return
+ }
+ if node.Kind != db.KindIntent {
+ writeErr(w, http.StatusConflict, "node is not an intent")
+ return
+ }
+ base := s.ctx
+ if base == nil {
+ base = context.Background()
+ }
+ opCtx, cancel := context.WithTimeout(base, workControlWaitTimeout+15*time.Second)
+ defer cancel()
+ result, err := s.engine.InterveneWork(opCtx, t, iid, requestID, message)
+ if err != nil {
+ switch {
+ case errors.Is(err, context.DeadlineExceeded), errors.Is(err, context.Canceled):
+ writeErr(w, http.StatusGatewayTimeout, err.Error())
+ case errors.Is(err, errWorkInterventionConflict), errors.Is(err, errWorkControlConflict),
+ errors.Is(err, db.ErrIntentInterventionConflict), errors.Is(err, db.ErrIntentStateConflict):
+ writeErr(w, http.StatusConflict, err.Error())
+ default:
+ writeErr(w, http.StatusInternalServerError, err.Error())
+ }
+ return
+ }
+
+ writeJSON(w, http.StatusOK, map[string]any{
+ "id": iid,
+ "state": result.State,
+ "accepted": true,
+ "activity_seq": result.ActivityID,
+ "request_id": result.RequestID,
+ "duplicate": result.Duplicate,
+ })
+}
diff --git a/server/intent_intervention_protocol_test.go b/server/intent_intervention_protocol_test.go
new file mode 100644
index 0000000..8347eca
--- /dev/null
+++ b/server/intent_intervention_protocol_test.go
@@ -0,0 +1,378 @@
+package server
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "iter"
+ "net/http"
+ "net/http/httptest"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/Autumn-27/artex/agent"
+ "github.com/Autumn-27/artex/db"
+ "github.com/Autumn-27/norma/llm"
+ "github.com/Autumn-27/norma/transcript"
+)
+
+type workerMessageTestProvider struct{}
+
+func (workerMessageTestProvider) Stream(_ context.Context, _ llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error] {
+ return func(yield func(llm.StreamEvent, error) bool) {
+ yield(llm.StreamEvent{Type: llm.SETextDelta, Text: "已按新的对话意图继续执行"}, nil)
+ yield(llm.StreamEvent{Type: llm.SEMessageDelta, StopReason: "end_turn"}, nil)
+ yield(llm.StreamEvent{Type: llm.SEMessageStop}, nil)
+ }
+}
+
+func (p workerMessageTestProvider) Complete(ctx context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) {
+ acc := llm.NewAccumulator()
+ for event, err := range p.Stream(ctx, req) {
+ if err != nil {
+ return llm.Message{}, "", llm.Usage{}, err
+ }
+ acc.Add(event)
+ }
+ return acc.Message(), acc.StopReason, acc.Usage, nil
+}
+
+type interventionProtocolFixture struct {
+ m *Manager
+ task *Task
+ engine *Engine
+ worker *agent.Worker
+ tx *transcript.Store
+ intentID int64
+}
+
+func newInterventionProtocolFixture(t *testing.T, targetPriority int) *interventionProtocolFixture {
+ t.Helper()
+ m, err := NewManager(t.TempDir(), "")
+ if err != nil {
+ t.Skipf("postgres unavailable (%v) - skipping", err)
+ }
+ task, err := m.CreateTask("worker message protocol", "verify immediate conversation handoff", nil, 0, 0)
+ if err != nil {
+ m.Close()
+ t.Fatal(err)
+ }
+ intentID, err := task.Store.AddIntent(map[string]any{"summary": "target worker intent"}, targetPriority, nil, "planner")
+ if err != nil {
+ m.Close()
+ t.Fatal(err)
+ }
+ if claimed, err := task.Store.ClaimIntent(intentID, "work#1"); err != nil || !claimed {
+ m.Close()
+ t.Fatalf("claim target intent: claimed=%v err=%v", claimed, err)
+ }
+ tx := transcript.NewStore(m.dir + "/transcripts")
+ worker := agent.NewWorker(workerMessageTestProvider{}, "test-model", m.dir, tx, 0, 1)
+ engine := NewEngine(m)
+ engine.UseLLM(nil, worker)
+ fixture := &interventionProtocolFixture{m: m, task: task, engine: engine, worker: worker, tx: tx, intentID: intentID}
+ t.Cleanup(func() {
+ taskID, _ := strconv.ParseInt(task.ID, 10, 64)
+ _ = m.pg.DeleteTask(taskID)
+ _ = m.Close()
+ })
+ return fixture
+}
+
+// settleRegisteredIntervention emulates the ordering inside runWorkerStep after
+// a cancellation-aware provider returns: detach the live work handle, commit the
+// running->paused final state, install the directed hand-off, then release the
+// controller waiting on workExecution.done.
+func settleRegisteredIntervention(t *testing.T, f *interventionProtocolFixture) string {
+ t.Helper()
+ action, complete := f.engine.detachWork(f.intentID)
+ if action != "intervene" {
+ t.Fatalf("settled action=%q, want intervene", action)
+ }
+ if changed, err := f.task.Store.CompareAndSetIntentState(f.intentID, "running", "paused"); err != nil || !changed {
+ t.Fatalf("settle running->paused: changed=%v err=%v", changed, err)
+ }
+ f.engine.queueDirectedClaim(f.task.ID, f.intentID)
+ complete(nil)
+ return action
+}
+
+func countTranscriptWorkerMessages(t *testing.T, tx *transcript.Store, sessionID, requestID string) (int, []transcript.Record) {
+ t.Helper()
+ records, err := tx.Load(tx.MainPath(sessionID))
+ if err != nil {
+ t.Fatal(err)
+ }
+ marker := "ARTEX_WORKER_CHAT:" + requestID
+ count := 0
+ for _, record := range records {
+ if record.Message != nil && strings.Contains(record.Message.Text(), marker) {
+ count++
+ }
+ }
+ return count, records
+}
+
+func TestWorkerMessageCancelsBeforeAcceptAndContinuesSameSessionFirst(t *testing.T) {
+ f := newInterventionProtocolFixture(t, 1)
+ competitorID, err := f.task.Store.AddIntent(map[string]any{"summary": "higher priority competitor"}, 100, nil, "planner")
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ // Pre-seed the same stable session so the message must continue its UUID
+ // chain rather than creating a worker-slot-specific conversation.
+ sessionID := agent.WorkerSessionID(f.task.ExpID, f.intentID)
+ seed := f.tx.NewWriter(sessionID, "")
+ seed.RecordMessage(llm.UserText("original intent turn"), llm.Usage{})
+ if err := seed.Err(); err != nil {
+ t.Fatal(err)
+ }
+ before, err := f.tx.Load(f.tx.MainPath(sessionID))
+ if err != nil || len(before) != 1 {
+ t.Fatalf("seed transcript records=%d err=%v", len(before), err)
+ }
+
+ workCtx, cancelWork := context.WithCancelCause(context.Background())
+ f.engine.registerWork(f.intentID, cancelWork)
+ live, unsubscribe := f.engine.Broadcaster().Subscribe(f.task.ID)
+ defer unsubscribe()
+
+ const requestID = "worker-message-engine-1"
+ const message = "停止主动枚举,立即验证登录越权"
+ type outcome struct {
+ result WorkInterventionResult
+ err error
+ }
+ done := make(chan outcome, 1)
+ go func() {
+ result, err := f.engine.InterveneWork(context.Background(), f.task, f.intentID, requestID, message)
+ done <- outcome{result: result, err: err}
+ }()
+
+ select {
+ case <-workCtx.Done():
+ if code, _, _, _ := agent.AbortReason(workCtx); code != "work_intervened_by_user" {
+ t.Fatalf("old execution cancel code=%q, want work_intervened_by_user", code)
+ }
+ case <-time.After(time.Second):
+ t.Fatal("Worker message did not immediately cancel the old execution")
+ }
+ select {
+ case early := <-done:
+ t.Fatalf("Worker message accepted before old worker settled: %+v", early)
+ case <-time.After(25 * time.Millisecond):
+ }
+ pending, err := f.task.Store.IntentInterventionByRequest(f.intentID, requestID)
+ if err != nil || pending == nil || pending.Status != db.IntentInterventionPending {
+ t.Fatalf("durable pending reservation=%+v err=%v", pending, err)
+ }
+ settleRegisteredIntervention(t, f)
+
+ var accepted outcome
+ select {
+ case accepted = <-done:
+ case <-time.After(2 * time.Second):
+ t.Fatal("timed out waiting for accepted Worker message")
+ }
+ if accepted.err != nil || accepted.result.ActivityID <= pending.ActivityID || accepted.result.Duplicate || accepted.result.State != "open" {
+ t.Fatalf("accepted Worker message=%+v control=%+v err=%v", accepted.result, pending, accepted.err)
+ }
+ node, err := f.task.Store.GetNode(f.intentID)
+ if err != nil || node == nil || node.State != "open" {
+ t.Fatalf("reopened target=%+v err=%v", node, err)
+ }
+
+ // The API publishes the visible user activity but never edits transcript files.
+ // agentcore appends the standard user turn only after the same Worker is claimed.
+ count, records := countTranscriptWorkerMessages(t, f.tx, sessionID, requestID)
+ if count != 0 || len(records) != 1 {
+ t.Fatalf("HTTP handler edited Worker transcript: count=%d records=%+v", count, records)
+ }
+ select {
+ case activity := <-live:
+ if activity.ID != accepted.result.ActivityID || activity.NodeID == nil || *activity.NodeID != f.intentID ||
+ activity.Worker != "user" || activity.Kind != "user" || activity.Summary != message {
+ t.Fatalf("broadcast activity=%+v", activity)
+ }
+ case <-time.After(time.Second):
+ t.Fatal("accepted Worker message was not broadcast")
+ }
+ select {
+ case duplicate := <-live:
+ t.Fatalf("accepted Worker message broadcast twice: %+v", duplicate)
+ default:
+ }
+
+ // The directed hand-off must outrank the normal frontier without changing the
+ // persisted priority. This catches the one-worker race where the released slot
+ // used to take an unrelated, higher-priority intent before the target reopened.
+ claimed := f.engine.claimNext(f.task, "work#2")
+ if claimed == nil || claimed.ID != f.intentID {
+ t.Fatalf("directed claim=%+v, want target %d ahead of competitor %d", claimed, f.intentID, competitorID)
+ }
+ competitor, err := f.task.Store.GetNode(competitorID)
+ if err != nil || competitor == nil || competitor.State != "open" {
+ t.Fatalf("competitor was claimed before Worker message target: node=%+v err=%v", competitor, err)
+ }
+ request, nextMessage, hasMessage, err := f.task.Store.PendingIntentInterventionMessage(f.intentID)
+ if err != nil || !hasMessage || request != requestID || nextMessage != message {
+ t.Fatalf("pending Worker message request=%q message=%q present=%v err=%v", request, nextMessage, hasMessage, err)
+ }
+ taskID, _ := strconv.ParseInt(f.task.ID, 10, 64)
+ if _, _, err := f.worker.ExecuteWithMessage(context.Background(), "work#2", taskID, nil, f.task.Store,
+ claimed, nil, nil, nil, nil, request, nextMessage); err != nil {
+ t.Fatalf("continue Worker conversation: %v", err)
+ }
+ if err := f.task.Store.MarkIntentInterventionHandoffClaimed(f.intentID); err != nil {
+ t.Fatal(err)
+ }
+ count, records = countTranscriptWorkerMessages(t, f.tx, sessionID, requestID)
+ if count != 1 || len(records) != 3 || records[1].ParentUUID != records[0].UUID || records[2].ParentUUID != records[1].UUID {
+ t.Fatalf("Worker message did not form one normal chained turn: count=%d records=%+v", count, records)
+ }
+ if records[1].Message == nil || records[1].Message.Role != llm.RoleUser || !strings.Contains(records[1].Message.Text(), message) {
+ t.Fatalf("Worker message was not recorded as the next user turn: %+v", records[1])
+ }
+ if _, _, pendingAfterRun, err := f.task.Store.PendingIntentInterventionMessage(f.intentID); err != nil || pendingAfterRun {
+ t.Fatalf("Worker message handoff remains pending=%v err=%v", pendingAfterRun, err)
+ }
+
+ retry, err := f.engine.InterveneWork(context.Background(), f.task, f.intentID, requestID, message)
+ if err != nil || !retry.Duplicate || retry.ActivityID != accepted.result.ActivityID {
+ t.Fatalf("idempotent engine retry=%+v err=%v", retry, err)
+ }
+ if _, err := f.engine.InterveneWork(context.Background(), f.task, f.intentID, requestID, "同 request_id 的另一条消息"); !errors.Is(err, errWorkInterventionConflict) {
+ t.Fatalf("same request_id different message error=%v, want conflict", err)
+ }
+ count, _ = countTranscriptWorkerMessages(t, f.tx, sessionID, requestID)
+ if count != 1 {
+ t.Fatalf("engine retries appended %d Worker messages, want 1", count)
+ }
+}
+
+func TestSendWorkerMessageContinuesAfterRequestContextCancellation(t *testing.T) {
+ f := newInterventionProtocolFixture(t, 1)
+ workCtx, cancelWork := context.WithCancelCause(context.Background())
+ f.engine.registerWork(f.intentID, cancelWork)
+ s := &Server{m: f.m, engine: f.engine, ctx: context.Background()}
+
+ requestBody, _ := json.Marshal(map[string]string{
+ "message": "中止当前动作,只验证权限边界", "request_id": "worker-message-disconnect-1",
+ })
+ requestCtx, disconnect := context.WithCancel(context.Background())
+ disconnect()
+ req := httptest.NewRequest(http.MethodPost, "/api/tasks/1/intents/1/messages", bytes.NewReader(requestBody)).WithContext(requestCtx)
+ req.SetPathValue("id", f.task.ID)
+ req.SetPathValue("iid", strconv.FormatInt(f.intentID, 10))
+ recorder := httptest.NewRecorder()
+ done := make(chan struct{})
+ go func() {
+ s.sendWorkerMessage(recorder, req)
+ close(done)
+ }()
+
+ select {
+ case <-workCtx.Done():
+ case <-time.After(time.Second):
+ t.Fatal("disconnected request did not continue to cancel the worker")
+ }
+ settleRegisteredIntervention(t, f)
+ select {
+ case <-done:
+ case <-time.After(2 * time.Second):
+ t.Fatal("server-owned Worker message did not finish after browser disconnect")
+ }
+ if recorder.Code != http.StatusOK {
+ t.Fatalf("disconnected Worker message status=%d body=%s", recorder.Code, recorder.Body.String())
+ }
+ accepted, err := f.task.Store.IntentInterventionByRequest(f.intentID, "worker-message-disconnect-1")
+ if err != nil || accepted == nil || accepted.Status != db.IntentInterventionAccepted {
+ t.Fatalf("disconnected request acceptance=%+v err=%v", accepted, err)
+ }
+}
+
+func TestWorkerMessageRejectsUnavailableTaskStatesWithoutReservation(t *testing.T) {
+ cases := []struct {
+ name string
+ block func(*interventionProtocolFixture)
+ }{
+ {name: "paused", block: func(f *interventionProtocolFixture) {
+ f.task.updateLifecycle(func(state *taskLifecycleState) { state.Paused = true })
+ f.engine.Pause(f.task.ID, agent.AbortPausedByUser)
+ }},
+ {name: "queued", block: func(f *interventionProtocolFixture) {
+ f.task.updateLifecycle(func(state *taskLifecycleState) { state.Queued = true })
+ }},
+ {name: "terminal", block: func(f *interventionProtocolFixture) {
+ f.task.updateLifecycle(func(state *taskLifecycleState) { state.Status = "done" })
+ }},
+ {name: "settling", block: func(f *interventionProtocolFixture) { f.engine.markSettling(f.task.ID) }},
+ {name: "deleting", block: func(f *interventionProtocolFixture) { f.engine.BeginDelete(f.task.ID) }},
+ }
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ f := newInterventionProtocolFixture(t, 1)
+ tc.block(f)
+ requestID := "worker-message-unavailable-" + tc.name
+ if _, err := f.engine.InterveneWork(context.Background(), f.task, f.intentID, requestID, "must not persist"); !errors.Is(err, errWorkInterventionConflict) {
+ t.Fatalf("unavailable state error=%v, want conflict", err)
+ }
+ reservation, err := f.task.Store.IntentInterventionByRequest(f.intentID, requestID)
+ if err != nil || reservation != nil {
+ t.Fatalf("unavailable state persisted reservation=%+v err=%v", reservation, err)
+ }
+ node, err := f.task.Store.GetNode(f.intentID)
+ if err != nil || node == nil || node.State != "running" {
+ t.Fatalf("unavailable state mutated intent=%+v err=%v", node, err)
+ }
+ count, _ := countTranscriptWorkerMessages(t, f.tx, agent.WorkerSessionID(f.task.ExpID, f.intentID), requestID)
+ if count != 0 {
+ t.Fatalf("unavailable state appended %d transcript interventions", count)
+ }
+ })
+ }
+}
+
+func TestWorkerWakeInterruptsLongIdlePoll(t *testing.T) {
+ task := &Task{workerWake: make(chan struct{}, 1)}
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ started := make(chan struct{})
+ returned := make(chan bool, 1)
+ go func() {
+ close(started)
+ returned <- waitWorker(ctx, task, 5*time.Second)
+ }()
+ <-started
+ start := time.Now()
+ task.NotifyWorker()
+ select {
+ case stopped := <-returned:
+ if stopped {
+ t.Fatal("worker wake was mistaken for context shutdown")
+ }
+ if elapsed := time.Since(start); elapsed > 250*time.Millisecond {
+ t.Fatalf("worker wake took %s, want immediate wake", elapsed)
+ }
+ case <-time.After(time.Second):
+ t.Fatal("worker wake waited for the idle polling interval")
+ }
+}
+
+func TestSendWorkerMessageRejectsOversizedBodyAsEntityTooLarge(t *testing.T) {
+ f := newInterventionProtocolFixture(t, 1)
+ s := &Server{m: f.m, engine: f.engine, ctx: context.Background()}
+ body := `{"message":"` + strings.Repeat("x", maxWorkerMessageBytes) + `","request_id":"oversized"}`
+ req := httptest.NewRequest(http.MethodPost, "/api/tasks/1/intents/1/messages", strings.NewReader(body))
+ req.SetPathValue("id", f.task.ID)
+ req.SetPathValue("iid", strconv.FormatInt(f.intentID, 10))
+ recorder := httptest.NewRecorder()
+ s.sendWorkerMessage(recorder, req)
+ if recorder.Code != http.StatusRequestEntityTooLarge {
+ t.Fatalf("oversized Worker message status=%d body=%s", recorder.Code, recorder.Body.String())
+ }
+}
diff --git a/server/manager.go b/server/manager.go
index cb198c2..d5c20e8 100644
--- a/server/manager.go
+++ b/server/manager.go
@@ -63,6 +63,7 @@ type Task struct {
Store *pgdb.ExplorationStore `json:"-"`
Guard *guard.Guard `json:"-"`
notify chan struct{}
+ workerWake chan struct{}
lifecycleMu sync.RWMutex
llmMu sync.RWMutex
@@ -737,7 +738,7 @@ func taskFromPG(pt *pgdb.Task, store *pgdb.ExplorationStore, ic *intercept.Inter
TimeoutSeconds: pt.TimeoutSeconds, PlanHeartbeatSeconds: pt.PlanHeartbeatSeconds,
CoverageEnabled: pt.CoverageEnabled,
FirstRunAt: unixOrZero(pt.FirstRunAt), DeadlineAt: unixOrZero(pt.DeadlineAt),
- Store: store, Guard: guard.NewWithInterceptor(ic), notify: make(chan struct{}, 1),
+ Store: store, Guard: guard.NewWithInterceptor(ic), notify: make(chan struct{}, 1), workerWake: make(chan struct{}, 64),
}
}
@@ -1638,6 +1639,42 @@ func (t *Task) Notify() {
}
}
+// NotifyWorker wakes one idle execution slot without itself scheduling a Planner
+// round. Worker-message acceptance separately records a persistent planning event
+// and calls NotifyWorkerMessage; keeping execution wake-up independent prevents
+// Planner debounce from delaying the directed Worker claim.
+func (t *Task) NotifyWorker() {
+ if t == nil || t.workerWake == nil {
+ return
+ }
+ select {
+ case t.workerWake <- struct{}{}:
+ default:
+ }
+}
+
+// NotifyWorkerMessage immediately schedules a Planner round for an accepted,
+// persistent Worker direction. ActivityID identifies the durable kind=user row
+// and prevents retries in the same debounce window from duplicating the prompt.
+func (t *Task) NotifyWorkerMessage(intentID, activityID int64, message string) {
+ message = strings.TrimSpace(message)
+ if intentID <= 0 || activityID <= 0 || message == "" {
+ return
+ }
+ t.trigMu.Lock()
+ for _, event := range t.pendingTriggers {
+ if event.Kind == agent.TriggerWorkerMessage && event.ActivityID == activityID {
+ t.trigMu.Unlock()
+ return
+ }
+ }
+ t.pendingTriggers = append(t.pendingTriggers, agent.TriggerEvent{
+ Kind: agent.TriggerWorkerMessage, ActivityID: activityID, IntentID: intentID, Detail: message,
+ })
+ t.trigMu.Unlock()
+ t.Notify()
+}
+
// NotifyDone is Notify plus a hint: a worker just finished intentID and that is
// what triggered this wake-up. The planner reads the accumulated triggers next
// round so it can spell out which intent finished (+ its output). Events pile up
diff --git a/server/server.go b/server/server.go
index f9ed5f4..b6a4182 100644
--- a/server/server.go
+++ b/server/server.go
@@ -249,6 +249,9 @@ func New(ctx context.Context, m *Manager, skillDir string, dataDir string, keyDi
if n, _ := t.Store.ResetRunningIntents(); n > 0 {
log.Printf("[engine] task %s 重置 %d 个残留 running 意图为 open", t.ID, n)
}
+ if err := s.engine.RecoverWorkerMessages(ctx, t); err != nil {
+ log.Printf("[worker chat] task %s 恢复待处理人工消息失败: %v", t.ID, err)
+ }
if lifecycle.Paused {
s.engine.Pause(t.ID, agent.AbortPausedOnReload)
}
@@ -471,6 +474,11 @@ func (s *Server) applyLLM(cfg agent.Config) error {
s.llmOn = true
s.cfgMu.Unlock()
s.invalidateTaskAgents()
+ for _, task := range s.m.List() {
+ if err := s.engine.RecoverWorkerMessages(s.ctx, task); err != nil {
+ log.Printf("[worker chat] task %s 在 LLM 就绪后恢复待处理人工消息失败: %v", task.ID, err)
+ }
+ }
// wake the active task so a task created while idle starts exploring.
if t := s.m.ActiveTask(); t != nil {
@@ -749,6 +757,7 @@ func (s *Server) Handler() http.Handler {
mux.HandleFunc("PUT /api/tasks/{id}/llm", s.updateTaskLLMProfiles)
mux.HandleFunc("GET /api/tasks/{id}/llm/resolution", s.taskLLMResolutionHandler)
mux.HandleFunc("POST /api/tasks/{id}/intents/{iid}/control", s.controlIntent)
+ mux.HandleFunc("POST /api/tasks/{id}/intents/{iid}/messages", s.sendWorkerMessage)
mux.HandleFunc("POST /api/tasks/{id}/intents/{iid}/rerun", s.rerunIntent) // 重跑单条 blocked/exhausted/stopped 意图
mux.HandleFunc("POST /api/tasks/{id}/intents/rerun-blocked", s.rerunBlocked) // 批量重跑本任务全部 blocked 意图
mux.HandleFunc("POST /api/active", s.setActive)
diff --git a/web/src/app/(main)/function/tasks/detail/_tabs/sessions-tab.tsx b/web/src/app/(main)/function/tasks/detail/_tabs/sessions-tab.tsx
index 937b97e..46ed163 100644
--- a/web/src/app/(main)/function/tasks/detail/_tabs/sessions-tab.tsx
+++ b/web/src/app/(main)/function/tasks/detail/_tabs/sessions-tab.tsx
@@ -14,7 +14,6 @@ import {
Loader2Icon,
PaperclipIcon,
PauseIcon,
- PlayIcon,
RadioIcon,
RotateCwIcon,
ShieldAlertIcon,
@@ -43,6 +42,7 @@ import { Badge } from "@/components/ui/badge";
import { Button } from "@/components/ui/button";
import { InputGroup, InputGroupAddon, InputGroupButton, InputGroupTextarea } from "@/components/ui/input-group";
import { ScrollArea } from "@/components/ui/scroll-area";
+import { Spinner } from "@/components/ui/spinner";
import { Textarea } from "@/components/ui/textarea";
import { Tooltip, TooltipContent, TooltipProvider, TooltipTrigger } from "@/components/ui/tooltip";
import { api, sseUrl } from "@/lib/api";
@@ -93,6 +93,18 @@ const PAGE = 200; // history page size
const SYSTEM_SCAN_PAGE = 500; // generic activity pages scanned to recover sparse system audit events
const MAX_KEEP = 4000; // per-session in-memory cap; older pages re-fetched on scroll-up
const STREAM_WINDOW_MS = 5000; // "live" = activity seen within this window
+const MAX_WORKER_MESSAGE_CHARS = 4000;
+
+function newWorkerMessageRequestID(): string {
+ return (
+ globalThis.crypto?.randomUUID?.() ??
+ `worker-message-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 12)}`
+ );
+}
+
+function workerMessageCharCount(value: string): number {
+ return Array.from(value).length;
+}
// One session's lazily-loaded, reverse-paginated cache. `lastTs`/`unread` are kept
// live even for sessions that were never opened, so the list shows liveness + unread
@@ -313,8 +325,6 @@ function SessionItem({
hasPending,
unread,
onClick,
- onPause,
- onResume,
onCancel,
controlling,
deleted,
@@ -325,8 +335,6 @@ function SessionItem({
hasPending?: boolean;
unread?: number;
onClick: () => void;
- onPause?: () => void;
- onResume?: () => void;
onCancel?: () => void;
controlling?: boolean;
deleted?: boolean;
@@ -339,7 +347,8 @@ function SessionItem({