fix: sync worker directions across agents

agent/mainagent.go

@@ -2,6 +2,7 @@ package agent
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 
 	"github.com/Autumn-27/artex/db"
@@ -85,6 +86,27 @@ 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
@@ -153,7 +175,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, message, func(r db.Activity) {
+	text, _, err := captureRunSession(ctx, s, mainAgentInput(ts, message), func(r db.Activity) {
 		if emit != nil {
 			r.Worker = "mainagent"
 			emit(r)

agent/worker_direction_sync_test.go

@@ -0,0 +1,155 @@
+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)
+	}
+}

db/exploration.go

@@ -360,13 +360,13 @@ type IntentIntervention struct {
 // 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
+	ActivityID    int64
+	IntentID      int64
+	RequestID     string
+	Message       string
+	IntentState   string
 	IntentSummary string
-	CreatedAt    time.Time
+	CreatedAt     time.Time
 }
 
 // lockIntentIntervention serializes request-id reservation/finalization across

db/intent_intervention_lifecycle_test.go

@@ -202,3 +202,72 @@ 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)
+	}
+}

server/intent_intervention_protocol_test.go

@@ -181,6 +181,11 @@ 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)
@@ -248,12 +253,39 @@ 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())