From 76210b99336a1a7d10fa1283e386c837b2568870 Mon Sep 17 00:00:00 2001 From: pphua Date: Sat, 29 Aug 2026 22:05:28 +0800 Subject: [PATCH] revert: remove cross-agent worker direction synchronization --- agent/mainagent.go | 24 +-- agent/planner.go | 20 +-- agent/tools.go | 40 ----- agent/worker_direction_sync_test.go | 155 -------------------- db/exploration.go | 61 -------- db/intent_intervention_lifecycle_test.go | 69 --------- server/engine.go | 5 - server/intent_intervention_protocol_test.go | 32 ---- server/manager.go | 30 +--- 9 files changed, 11 insertions(+), 425 deletions(-) delete mode 100644 agent/worker_direction_sync_test.go diff --git a/agent/mainagent.go b/agent/mainagent.go index 2962252..9c40eca 100644 --- a/agent/mainagent.go +++ b/agent/mainagent.go @@ -2,7 +2,6 @@ package agent import ( "context" - "encoding/json" "fmt" "github.com/Autumn-27/artex/db" @@ -86,27 +85,6 @@ func mainAgentSystem(goal, dataDir, workDir string) string { return body + artifactSpec(workDir) } -// mainAgentInput keeps Worker-side human steering visible to the Main Agent on -// every fresh Chat session. The direction records remain separate from the -// current user turn so the model does not mistake persisted context for a new -// instruction that must be sent to the Worker again. -func mainAgentInput(ts *db.ExplorationStore, message string) string { - if ts == nil { - return message - } - directions, err := ts.LatestWorkerDirections(workerDirectionOverviewLimit) - if err != nil || len(directions) == 0 { - return message - } - raw, err := json.Marshal(compactWorkerDirections(directions)) - if err != nil { - return message - } - return "【系统同步的 Worker 最新人工方向】\n" + - "以下内容来自持久化的 Worker 对话。回答、观察和调度时必须纳入，但不要把它当作本轮新消息重复下发。\n" + - string(raw) + "\n\n【用户本轮消息】\n" + message -} - // Chat handles one human message and returns the assistant reply. emit, if // non-nil, receives each execution step (thinking / tool_use / tool_result / // text / result) so the main-agent session shows its work — exactly like the @@ -175,7 +153,7 @@ func (m *MainAgent) Chat(ctx context.Context, taskID int64, as *db.AssetStore, t // C2: this session is fresh each turn; re-unlock skill-gated MCPs from prior // Skill() calls in the reloaded history so revealed tools stay callable. seedUnlockFromHistory(s.Messages(), def.UnlockSkill) - text, _, err := captureRunSession(ctx, s, mainAgentInput(ts, message), func(r db.Activity) { + text, _, err := captureRunSession(ctx, s, message, func(r db.Activity) { if emit != nil { r.Worker = "mainagent" emit(r) diff --git a/agent/planner.go b/agent/planner.go index 3f63166..95802f8 100644 --- a/agent/planner.go +++ b/agent/planner.go @@ -123,7 +123,6 @@ 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 = 被删目标文本). @@ -131,17 +130,14 @@ 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 - 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" 专用：修改后的目标文本 + Kind string + 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; @@ -166,9 +162,6 @@ 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" @@ -257,7 +250,6 @@ 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 3a34ca1..ebed6d8 100644 --- a/agent/tools.go +++ b/agent/tools.go @@ -48,38 +48,6 @@ 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 { @@ -392,14 +360,6 @@ 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_direction_sync_test.go b/agent/worker_direction_sync_test.go deleted file mode 100644 index a1cf7f8..0000000 --- a/agent/worker_direction_sync_test.go +++ /dev/null @@ -1,155 +0,0 @@ -package agent - -import ( - "context" - "iter" - "strings" - "sync" - "testing" - - "github.com/Autumn-27/artex/db" - "github.com/Autumn-27/norma/llm" -) - -type directionCaptureProvider struct { - mu sync.Mutex - requests []llm.CompletionRequest -} - -func (p *directionCaptureProvider) Stream(_ context.Context, req llm.CompletionRequest) iter.Seq2[llm.StreamEvent, error] { - p.mu.Lock() - p.requests = append(p.requests, req) - p.mu.Unlock() - return func(yield func(llm.StreamEvent, error) bool) { - if !yield(llm.StreamEvent{Type: llm.SETextDelta, Text: "同步完成"}, nil) { - return - } - if !yield(llm.StreamEvent{Type: llm.SEMessageDelta, StopReason: "end_turn"}, nil) { - return - } - yield(llm.StreamEvent{Type: llm.SEMessageStop}, nil) - } -} - -func (p *directionCaptureProvider) Complete(_ context.Context, req llm.CompletionRequest) (llm.Message, string, llm.Usage, error) { - p.mu.Lock() - p.requests = append(p.requests, req) - p.mu.Unlock() - return llm.Message{Role: llm.RoleAssistant, Content: []llm.ContentBlock{llm.TextBlock("同步完成")}}, "end_turn", llm.Usage{}, nil -} - -func (p *directionCaptureProvider) lastRequest(t *testing.T) llm.CompletionRequest { - t.Helper() - p.mu.Lock() - defer p.mu.Unlock() - if len(p.requests) == 0 { - t.Fatal("provider received no completion request") - } - return p.requests[len(p.requests)-1] -} - -func lastUserText(t *testing.T, req llm.CompletionRequest) string { - t.Helper() - for i := len(req.Messages) - 1; i >= 0; i-- { - if req.Messages[i].Role == llm.RoleUser { - return req.Messages[i].Text() - } - } - t.Fatal("completion request has no user message") - return "" -} - -type directionAgentFixture struct { - d *db.DB - task *db.Task - store *db.ExplorationStore - intentID int64 - activity db.Activity -} - -func newDirectionAgentFixture(t *testing.T) *directionAgentFixture { - t.Helper() - d := testDB(t) - task, err := d.CreateTask("agent direction sync", "synchronize worker steering", nil, 0, 0) - if err != nil { - d.Close() - t.Fatal(err) - } - t.Cleanup(func() { - _ = d.DeleteTask(task.ID) - _ = d.Close() - }) - store := d.Exploration(task.ExplorationID) - intentID, err := store.AddIntent(map[string]any{"summary": "原始接口枚举方向"}, 1, nil, "planner") - if err != nil { - t.Fatal(err) - } - if claimed, err := store.ClaimIntent(intentID, "work#direction"); err != nil || !claimed { - t.Fatalf("claim direction fixture: claimed=%v err=%v", claimed, err) - } - const requestID = "agent-direction-sync-1" - const direction = "停止接口枚举，立即聚焦登录后的水平越权" - if _, err := store.ReserveIntentIntervention(intentID, requestID, direction); err != nil { - t.Fatal(err) - } - if changed, err := store.CompareAndSetIntentState(intentID, "running", "paused"); err != nil || !changed { - t.Fatalf("pause direction fixture: changed=%v err=%v", changed, err) - } - activity, duplicate, err := store.AcceptIntentIntervention(intentID, requestID, direction) - if err != nil || duplicate { - t.Fatalf("accept direction fixture: activity=%+v duplicate=%v err=%v", activity, duplicate, err) - } - return &directionAgentFixture{d: d, task: task, store: store, intentID: intentID, activity: activity} -} - -func TestGraphOverviewIncludesLatestWorkerDirection(t *testing.T) { - f := newDirectionAgentFixture(t) - overview := NewToolSet(f.store, "planner").graphOverviewData() - directions, ok := overview["worker_directions"].([]map[string]any) - if !ok || len(directions) != 1 { - t.Fatalf("worker_directions=%T %+v", overview["worker_directions"], overview["worker_directions"]) - } - if directions[0]["activity_id"] != f.activity.ID || directions[0]["intent_id"] != f.intentID || - directions[0]["intent_summary"] != "原始接口枚举方向" || - directions[0]["message"] != "停止接口枚举，立即聚焦登录后的水平越权" { - t.Fatalf("worker direction overview=%+v", directions[0]) - } -} - -func TestPlannerRequestIncludesWorkerMessageTriggerAndPersistedDirection(t *testing.T) { - f := newDirectionAgentFixture(t) - provider := &directionCaptureProvider{} - planner := NewPlanner(provider, "test-model", t.TempDir(), nil, 0, 1) - triggerDetail := "本轮只验证登录后的对象级授权" - _, _, err := planner.Plan(context.Background(), f.task.ID, nil, f.store, f.task.Goal, []TriggerEvent{{ - Kind: TriggerWorkerMessage, ActivityID: f.activity.ID, IntentID: f.intentID, Detail: triggerDetail, - }}, nil) - if err != nil { - t.Fatal(err) - } - input := lastUserText(t, provider.lastRequest(t)) - for _, want := range []string{"【本次触发本轮的实际变动】", triggerDetail, `"worker_directions"`, "停止接口枚举，立即聚焦登录后的水平越权"} { - if !strings.Contains(input, want) { - t.Fatalf("Planner input missing %q:\n%s", want, input) - } - } -} - -func TestMainAgentNextTurnIncludesPersistedWorkerDirection(t *testing.T) { - f := newDirectionAgentFixture(t) - provider := &directionCaptureProvider{} - mainAgent := NewMainAgent(provider, "test-model", t.TempDir(), nil, 0, 1) - const userMessage = "汇报当前执行方向" - if _, err := mainAgent.Chat(context.Background(), f.task.ID, nil, f.store, f.task.Goal, userMessage, nil, nil, nil, nil); err != nil { - t.Fatal(err) - } - input := lastUserText(t, provider.lastRequest(t)) - for _, want := range []string{"【系统同步的 Worker 最新人工方向】", "停止接口枚举，立即聚焦登录后的水平越权", "【用户本轮消息】", userMessage} { - if !strings.Contains(input, want) { - t.Fatalf("Main Agent input missing %q:\n%s", want, input) - } - } - if strings.Index(input, "【系统同步的 Worker 最新人工方向】") > strings.Index(input, "【用户本轮消息】") { - t.Fatalf("Main Agent context is not separated before current user message:\n%s", input) - } -} diff --git a/db/exploration.go b/db/exploration.go index 6a05f9a..1ffb92a 100644 --- a/db/exploration.go +++ b/db/exploration.go @@ -339,7 +339,6 @@ const ( IntentInterventionPending = "pending" IntentInterventionAccepted = "accepted" IntentInterventionRejected = "rejected" - PlannerEventWorkerMessage = "worker_message" ) // IntentIntervention is the durable control record for one human interruption. @@ -356,19 +355,6 @@ type IntentIntervention struct { 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. @@ -577,7 +563,6 @@ WHERE a.exploration_id=$1 AND a.node_id=$2 "intervention_request_id": requestID, "intervention_message": message, "intervention_status": IntentInterventionAccepted, - "planner_event": PlannerEventWorkerMessage, "source": "user", }) if err != nil { @@ -624,52 +609,6 @@ WHERE a.exploration_id=$1 AND a.node_id=$2 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. diff --git a/db/intent_intervention_lifecycle_test.go b/db/intent_intervention_lifecycle_test.go index 9363e7c..c27a746 100644 --- a/db/intent_intervention_lifecycle_test.go +++ b/db/intent_intervention_lifecycle_test.go @@ -202,72 +202,3 @@ func TestIntentInterventionAcceptanceCanRetryAfterStateSettlement(t *testing.T) t.Fatalf("retry acceptance: activity=%+v duplicate=%v err=%v", activity, duplicate, err) } } - -func acceptTestWorkerDirection(t *testing.T, store *ExplorationStore, intentID int64, requestID, message string) Activity { - t.Helper() - if claimed, err := store.ClaimIntent(intentID, "direction-test-worker"); err != nil || !claimed { - t.Fatalf("claim direction intent: claimed=%v err=%v", claimed, err) - } - if _, err := store.ReserveIntentIntervention(intentID, requestID, message); err != nil { - t.Fatal(err) - } - if changed, err := store.CompareAndSetIntentState(intentID, "running", "paused"); err != nil || !changed { - t.Fatalf("pause direction intent: changed=%v err=%v", changed, err) - } - activity, duplicate, err := store.AcceptIntentIntervention(intentID, requestID, message) - if err != nil || duplicate { - t.Fatalf("accept direction: activity=%+v duplicate=%v err=%v", activity, duplicate, err) - } - return activity -} - -func TestLatestWorkerDirectionsKeepsLatestPerIntentAcrossStoreReopen(t *testing.T) { - d, err := Open(testDSN(t)) - if err != nil { - t.Skipf("postgres unavailable (%v) - skipping", err) - } - defer d.Close() - task, err := d.CreateTask("worker direction persistence", "keep latest directions", nil, 0, 0) - if err != nil { - t.Fatal(err) - } - defer d.DeleteTask(task.ID) //nolint:errcheck - store := d.Exploration(task.ExplorationID) - firstIntent, err := store.AddIntent(map[string]any{"summary": "first route"}, 1, nil, "planner") - if err != nil { - t.Fatal(err) - } - secondIntent, err := store.AddIntent(map[string]any{"summary": "second route"}, 1, nil, "planner") - if err != nil { - t.Fatal(err) - } - - acceptTestWorkerDirection(t, store, firstIntent, "direction-old", "先检查旧方向") - secondActivity := acceptTestWorkerDirection(t, store, secondIntent, "direction-second", "保留第二条意图方向") - latestActivity := acceptTestWorkerDirection(t, store, firstIntent, "direction-latest", "改为验证最新权限边界") - - d2, err := Open(testDSN(t)) - if err != nil { - t.Fatal(err) - } - defer d2.Close() - directions, err := d2.Exploration(task.ExplorationID).LatestWorkerDirections(10) - if err != nil { - t.Fatal(err) - } - if len(directions) != 2 { - t.Fatalf("directions=%+v, want one latest row per intent", directions) - } - byIntent := make(map[int64]WorkerDirection, len(directions)) - for _, direction := range directions { - byIntent[direction.IntentID] = direction - } - if got := byIntent[firstIntent]; got.ActivityID != latestActivity.ID || got.RequestID != "direction-latest" || - got.Message != "改为验证最新权限边界" || got.IntentSummary != "first route" { - t.Fatalf("latest first direction=%+v", got) - } - if got := byIntent[secondIntent]; got.ActivityID != secondActivity.ID || got.Message != "保留第二条意图方向" || - got.IntentSummary != "second route" { - t.Fatalf("second direction=%+v", got) - } -} diff --git a/server/engine.go b/server/engine.go index 2754c09..165972d 100644 --- a/server/engine.go +++ b/server/engine.go @@ -687,7 +687,6 @@ func (e *Engine) interveneWork(ctx context.Context, t *Task, intentID int64, req } if !duplicate { e.publishStoredActivity(t, activity) - t.NotifyWorkerMessage(intentID, activity.ID, message) } e.readyDirectedClaim(intentID) t.NotifyWorker() @@ -752,10 +751,6 @@ func (e *Engine) recoverWorkerMessages(ctx context.Context, t *Task, allowAdmiss 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 diff --git a/server/intent_intervention_protocol_test.go b/server/intent_intervention_protocol_test.go index 77ac2e7..8347eca 100644 --- a/server/intent_intervention_protocol_test.go +++ b/server/intent_intervention_protocol_test.go @@ -181,11 +181,6 @@ func TestWorkerMessageCancelsBeforeAcceptAndContinuesSameSessionFirst(t *testing 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) } - triggers := f.task.drainTriggers() - if len(triggers) != 1 || triggers[0].Kind != agent.TriggerWorkerMessage || - triggers[0].ActivityID != accepted.result.ActivityID || triggers[0].IntentID != f.intentID || triggers[0].Detail != message { - t.Fatalf("accepted Worker message triggers=%+v, want one matching persistent event", triggers) - } 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) @@ -253,39 +248,12 @@ func TestWorkerMessageCancelsBeforeAcceptAndContinuesSameSessionFirst(t *testing 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) } - if triggers := f.task.drainTriggers(); len(triggers) != 0 { - t.Fatalf("idempotent retries emitted Planner triggers: %+v", triggers) - } count, _ = countTranscriptWorkerMessages(t, f.tx, sessionID, requestID) if count != 1 { t.Fatalf("engine retries appended %d Worker messages, want 1", count) } } -func TestRecoverAcceptedWorkerMessageReconstructsPlannerTrigger(t *testing.T) { - f := newInterventionProtocolFixture(t, 1) - const requestID = "worker-message-recovery-1" - const message = "恢复后优先验证人工指定的权限边界" - if _, err := f.task.Store.ReserveIntentIntervention(f.intentID, requestID, message); err != nil { - t.Fatal(err) - } - if changed, err := f.task.Store.CompareAndSetIntentState(f.intentID, "running", "paused"); err != nil || !changed { - t.Fatalf("pause before recovery acceptance: changed=%v err=%v", changed, err) - } - activity, duplicate, err := f.task.Store.AcceptIntentIntervention(f.intentID, requestID, message) - if err != nil || duplicate { - t.Fatalf("accept before recovery: activity=%+v duplicate=%v err=%v", activity, duplicate, err) - } - if err := f.engine.RecoverWorkerMessages(context.Background(), f.task); err != nil { - t.Fatal(err) - } - triggers := f.task.drainTriggers() - if len(triggers) != 1 || triggers[0].Kind != agent.TriggerWorkerMessage || triggers[0].ActivityID != activity.ID || - triggers[0].IntentID != f.intentID || triggers[0].Detail != message { - t.Fatalf("recovered Planner triggers=%+v", triggers) - } -} - func TestSendWorkerMessageContinuesAfterRequestContextCancellation(t *testing.T) { f := newInterventionProtocolFixture(t, 1) workCtx, cancelWork := context.WithCancelCause(context.Background()) diff --git a/server/manager.go b/server/manager.go index d5c20e8..6fca4dc 100644 --- a/server/manager.go +++ b/server/manager.go @@ -1639,10 +1639,10 @@ 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. +// NotifyWorker wakes one idle execution slot without also scheduling a Planner +// round. The channel is separate from notify because human intervention needs +// the reopened intent reclaimed immediately, while graph-change notifications +// are Planner-facing and debounced. A nil channel is tolerated by test fixtures. func (t *Task) NotifyWorker() { if t == nil || t.workerWake == nil { return @@ -1653,28 +1653,6 @@ func (t *Task) NotifyWorker() { } } -// 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