From 7e633b2056745f297856900850639609579217b1 Mon Sep 17 00:00:00 2001 From: pphua Date: Sat, 29 Aug 2026 21:44:39 +0800 Subject: [PATCH] fix: sync worker directions across agents --- agent/mainagent.go | 24 ++- agent/worker_direction_sync_test.go | 155 ++++++++++++++++++++ db/exploration.go | 12 +- db/intent_intervention_lifecycle_test.go | 69 +++++++++ server/intent_intervention_protocol_test.go | 32 ++++ 5 files changed, 285 insertions(+), 7 deletions(-) create mode 100644 agent/worker_direction_sync_test.go diff --git a/agent/mainagent.go b/agent/mainagent.go index 9c40eca..2962252 100644 --- a/agent/mainagent.go +++ b/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) diff --git a/agent/worker_direction_sync_test.go b/agent/worker_direction_sync_test.go new file mode 100644 index 0000000..a1cf7f8 --- /dev/null +++ b/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) + } +} diff --git a/db/exploration.go b/db/exploration.go index 7c40e85..6a05f9a 100644 --- a/db/exploration.go +++ b/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 diff --git a/db/intent_intervention_lifecycle_test.go b/db/intent_intervention_lifecycle_test.go index c27a746..9363e7c 100644 --- a/db/intent_intervention_lifecycle_test.go +++ b/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) + } +} diff --git a/server/intent_intervention_protocol_test.go b/server/intent_intervention_protocol_test.go index 8347eca..77ac2e7 100644 --- a/server/intent_intervention_protocol_test.go +++ b/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())