From 472c672390a4d4935a253e54b30731170591f1de Mon Sep 17 00:00:00 2001 From: Clem Date: Thu, 2 Jul 2026 15:36:05 +0200 Subject: [PATCH] feat: show OMP advisor feedback --- .../adapter/transport/acp/adapter_updates.go | 124 +++++++++++++++++- .../adapter/transport/acp/conversion_test.go | 51 +++++++ .../internal/agentctl/types/streams/agent.go | 2 + .../internal/integration/thinking_test.go | 56 ++++++++ .../orchestrator/event_handlers_streaming.go | 31 +++++ apps/backend/pkg/api/v1/task.go | 19 +-- .../task/chat/message-renderer.test.tsx | 25 +++- .../components/task/chat/message-renderer.tsx | 19 +++ .../e2e/tests/chat/advisor-feedback.spec.ts | 39 ++++++ apps/web/hooks/use-processed-messages.test.ts | 17 +++ apps/web/hooks/use-processed-messages.ts | 1 + apps/web/lib/types/http.ts | 1 + 12 files changed, 368 insertions(+), 17 deletions(-) create mode 100644 apps/web/e2e/tests/chat/advisor-feedback.spec.ts diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_updates.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_updates.go index 880484ead6..df00ce54fa 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_updates.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_updates.go @@ -10,6 +10,21 @@ import ( "go.uber.org/zap" ) +const ( + advisorMetaKeySource = "source" + advisorMetaKeyType = "type" + advisorMetaKeyKind = "kind" + advisorMetaKeyRole = "role" + advisorMetaSource = "advisor" +) + +var advisorMetaKeys = [...]string{ + advisorMetaKeySource, + advisorMetaKeyType, + advisorMetaKeyKind, + advisorMetaKeyRole, +} + // notifWork is the item type carried on notifQueue. Exactly one of the two // fields is populated: notif for a real SDK notification (the common case), // sync for a barrier posted by syncNotifQueue. The worker closes sync when @@ -180,13 +195,7 @@ func (a *Adapter) convertNotification(n acp.SessionNotification) *AgentEvent { return a.convertMessageChunk(sessionID, u.UserMessageChunk.Content, "user") case u.AgentThoughtChunk != nil: - if u.AgentThoughtChunk.Content.Text != nil { - return &AgentEvent{ - Type: streams.EventTypeReasoning, - SessionID: sessionID, - ReasoningText: u.AgentThoughtChunk.Content.Text.Text, - } - } + return a.convertThoughtChunk(sessionID, u.AgentThoughtChunk) case u.ToolCall != nil: return a.convertToolCallUpdate(sessionID, u.ToolCall) @@ -255,6 +264,88 @@ type acpUsageUpdate struct { } `json:"cost,omitempty"` } +func (a *Adapter) convertThoughtChunk(sessionID string, chunk *acp.SessionUpdateAgentThoughtChunk) *AgentEvent { + if chunk == nil || chunk.Content.Text == nil { + return nil + } + text := chunk.Content.Text.Text + if isAdvisorFeedbackMeta(chunk.Meta) { + data := copyAdvisorFeedbackMeta(chunk.Meta) + if chunk.MessageId != nil && *chunk.MessageId != "" { + data["message_id"] = *chunk.MessageId + } + return &AgentEvent{ + Type: streams.EventTypeAdvisorFeedback, + SessionID: sessionID, + Text: text, + Data: data, + } + } + return &AgentEvent{ + Type: streams.EventTypeReasoning, + SessionID: sessionID, + ReasoningText: text, + } +} + +func isAdvisorFeedbackMeta(meta map[string]any) bool { + if len(meta) == 0 { + return false + } + if advisor, ok := meta[advisorMetaSource].(bool); ok && advisor { + return true + } + for _, key := range advisorMetaKeys { + value, ok := meta[key].(string) + if !ok { + continue + } + switch strings.ToLower(value) { + case advisorMetaSource, "omp-advisor", streams.EventTypeAdvisorFeedback, "advisor-feedback": + return true + } + } + return false +} + +func copyAdvisorFeedbackMeta(meta map[string]any) map[string]any { + data := make(map[string]any, len(meta)+1) + data[advisorMetaKeySource] = advisorMetaSource + for key, value := range meta { + data[key] = value + } + return data +} + +func convertAdvisorFeedbackUpdate(sessionID string, advisor acpAdvisorFeedbackUpdate) *AgentEvent { + if advisor.SessionUpdate != streams.EventTypeAdvisorFeedback { + return nil + } + text := advisor.Text + if text == "" { + text = advisor.Message + } + if text == "" { + text = advisor.Content + } + if strings.TrimSpace(text) == "" { + return nil + } + data := copyAdvisorFeedbackMeta(advisor.Meta) + if advisor.Severity != "" { + data["severity"] = advisor.Severity + } + if advisor.AdvisorID != "" { + data["advisor_id"] = advisor.AdvisorID + } + return &AgentEvent{ + Type: streams.EventTypeAdvisorFeedback, + SessionID: sessionID, + Text: text, + Data: data, + } +} + // acpSessionInfoUpdate represents the ACP "session_info_update" notification. // TODO: Replace with the SDK type when acp-go-sdk exposes it. type acpSessionInfoUpdate struct { @@ -264,6 +355,18 @@ type acpSessionInfoUpdate struct { Meta map[string]any `json:"_meta,omitempty"` } +// acpAdvisorFeedbackUpdate represents an OMP advisor feedback session notification +// if the ACP SDK has not learned the update type yet. +type acpAdvisorFeedbackUpdate struct { + SessionUpdate string `json:"sessionUpdate"` + Text string `json:"text,omitempty"` + Message string `json:"message,omitempty"` + Content string `json:"content,omitempty"` + Severity string `json:"severity,omitempty"` + AdvisorID string `json:"advisorId,omitempty"` + Meta map[string]any `json:"_meta,omitempty"` +} + // usageTracker carries the running cumulative usage state for one ACP // session — used to infer codex-acp's per-turn deltas. Reset when the // prompt-complete handler consumes it; lastUsed sticks so subsequent @@ -363,6 +466,13 @@ func (a *Adapter) tryConvertUntypedUpdate(rawNotification []byte, sessionID stri } } + var advisor acpAdvisorFeedbackUpdate + if err := json.Unmarshal(envelope.Update, &advisor); err == nil { + if event := convertAdvisorFeedbackUpdate(sessionID, advisor); event != nil { + return event + } + } + var usage acpUsageUpdate if err := json.Unmarshal(envelope.Update, &usage); err != nil { return nil diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/conversion_test.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/conversion_test.go index 7d2c1d2ff4..8cfd1ee51a 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/conversion_test.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/conversion_test.go @@ -897,6 +897,57 @@ func TestTryConvertUntypedUpdate_UnknownUpdateType(t *testing.T) { t.Errorf("expected nil for unknown update type, got %+v", result) } } +func TestConvertNotification_AdvisorThoughtChunkMeta(t *testing.T) { + a := newTestAdapter() + messageID := "advisor-msg-1" + notification := acp.SessionNotification{ + SessionId: "s1", + Update: acp.SessionUpdate{ + AgentThoughtChunk: &acp.SessionUpdateAgentThoughtChunk{ + Meta: map[string]any{"source": "advisor", "severity": "concern"}, + Content: acp.TextBlock("Prefer the backend conversion path."), + MessageId: &messageID, + }, + }, + } + + result := a.convertNotification(notification) + + if result == nil { + t.Fatal("expected advisor feedback event, got nil") + return + } + if result.Type != streams.EventTypeAdvisorFeedback { + t.Fatalf("Type = %q, want %q", result.Type, streams.EventTypeAdvisorFeedback) + } + if result.Text != "Prefer the backend conversion path." { + t.Errorf("Text = %q", result.Text) + } + if result.Data["severity"] != "concern" { + t.Errorf("severity metadata = %v, want concern", result.Data["severity"]) + } +} + +func TestTryConvertUntypedUpdate_AdvisorFeedback(t *testing.T) { + a := newTestAdapter() + raw := []byte(`{"sessionId":"s1","update":{"sessionUpdate":"advisor_feedback","text":"Run the focused test first.","severity":"concern","advisorId":"omp"}}`) + + result := a.tryConvertUntypedUpdate(raw, "s1") + + if result == nil { + t.Fatal("expected advisor feedback event, got nil") + return + } + if result.Type != streams.EventTypeAdvisorFeedback { + t.Fatalf("Type = %q, want %q", result.Type, streams.EventTypeAdvisorFeedback) + } + if result.Text != "Run the focused test first." { + t.Errorf("Text = %q", result.Text) + } + if result.Data["advisor_id"] != "omp" { + t.Errorf("advisor_id metadata = %v, want omp", result.Data["advisor_id"]) + } +} func TestTryConvertUntypedUpdate_InvalidJSON(t *testing.T) { a := newTestAdapter() diff --git a/apps/backend/internal/agentctl/types/streams/agent.go b/apps/backend/internal/agentctl/types/streams/agent.go index 3821367856..db050dd5c1 100644 --- a/apps/backend/internal/agentctl/types/streams/agent.go +++ b/apps/backend/internal/agentctl/types/streams/agent.go @@ -8,6 +8,8 @@ const ( // EventTypeReasoning indicates chain-of-thought or thinking content. EventTypeReasoning = "reasoning" + // EventTypeAdvisorFeedback indicates per-turn feedback from an advisor side channel. + EventTypeAdvisorFeedback = "advisor_feedback" // EventTypeToolCall indicates a tool invocation has started. EventTypeToolCall = "tool_call" diff --git a/apps/backend/internal/integration/thinking_test.go b/apps/backend/internal/integration/thinking_test.go index 48a092881d..5fad9f8fa9 100644 --- a/apps/backend/internal/integration/thinking_test.go +++ b/apps/backend/internal/integration/thinking_test.go @@ -183,6 +183,62 @@ func TestThinkingEventAppend(t *testing.T) { assert.True(t, foundThinking, "should find thinking message with both chunks") } +func TestAdvisorFeedbackEventsPersistToDatabase(t *testing.T) { + ts := NewOrchestratorTestServer(t) + defer ts.Close() + + ctx := context.Background() + taskID := ts.CreateTestTask(t, "test-agent", 1) + + sessionID := uuid.New().String() + session := &models.TaskSession{ + ID: sessionID, + TaskID: taskID, + AgentProfileID: "test-agent", + State: models.TaskSessionStateRunning, + StartedAt: time.Now(), + UpdatedAt: time.Now(), + } + err := ts.TaskRepo.CreateTaskSession(ctx, session) + require.NoError(t, err) + + eventData := &lifecycle.AgentStreamEventPayload{ + Type: "agent/event", + Timestamp: time.Now().UTC().Format(time.RFC3339Nano), + TaskID: taskID, + SessionID: sessionID, + Data: &lifecycle.AgentStreamEventData{ + Type: "advisor_feedback", + Text: "Good point. Keep the patch minimal and verify the backend path.", + Data: map[string]interface{}{ + "source": "advisor", + "severity": "concern", + }, + }, + } + + subject := events.BuildAgentStreamSubject(taskID) + err = ts.EventBus.Publish(ctx, subject, bus.NewEvent(events.AgentStream, "test", eventData)) + require.NoError(t, err) + time.Sleep(100 * time.Millisecond) + + messages, err := ts.TaskSvc.ListMessages(ctx, sessionID) + require.NoError(t, err) + + var found bool + for _, msg := range messages { + if msg.Type == "advisor_feedback" { + found = true + assert.Equal(t, "Good point. Keep the patch minimal and verify the backend path.", msg.Content) + assert.Equal(t, "advisor", msg.Metadata["source"]) + assert.Equal(t, "concern", msg.Metadata["severity"]) + break + } + } + + assert.True(t, found, "should find advisor feedback message") +} + // TestCodexReasoningSummaryEvents verifies that Codex-style reasoning events // (which use ReasoningSummary instead of ReasoningText) are properly handled. // Codex sends item/reasoning/summaryTextDelta which sets ReasoningSummary. diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index fde3683cc4..a1759de95f 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming.go @@ -3,6 +3,7 @@ package orchestrator import ( "context" "fmt" + "strings" "time" "go.uber.org/zap" @@ -62,6 +63,9 @@ func (s *Service) handleAgentStreamEvent(ctx context.Context, payload *lifecycle case "thinking_streaming": s.handleThinkingStreamingEvent(ctx, payload) + case streams.EventTypeAdvisorFeedback: + s.handleAdvisorFeedbackEvent(ctx, payload) + case agentEventToolCall: s.saveAgentTextIfPresent(ctx, payload) s.handleToolCallEvent(ctx, payload) @@ -214,6 +218,33 @@ func (s *Service) handleAgentLogEvent(ctx context.Context, payload *lifecycle.Ag } } +func (s *Service) handleAdvisorFeedbackEvent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) { + taskID := payload.TaskID + sessionID := payload.SessionID + if sessionID == "" || s.messageCreator == nil { + return + } + text := payload.Data.Text + if strings.TrimSpace(text) == "" { + return + } + metadata := map[string]interface{}{"source": "advisor"} + if dataMap, ok := payload.Data.Data.(map[string]interface{}); ok { + for key, value := range dataMap { + metadata[key] = value + } + } + if err := s.messageCreator.CreateSessionMessage( + ctx, taskID, text, sessionID, + string(v1.MessageTypeAdvisorFeedback), s.getActiveTurnID(sessionID), metadata, false, + ); err != nil { + s.logger.Error("failed to create advisor feedback message", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Error(err)) + } +} + // handleToolCallEvent handles tool_call events and creates messages func (s *Service) handleToolCallEvent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) { if payload.SessionID == "" { diff --git a/apps/backend/pkg/api/v1/task.go b/apps/backend/pkg/api/v1/task.go index 1c9d8ff74e..5317b2133b 100644 --- a/apps/backend/pkg/api/v1/task.go +++ b/apps/backend/pkg/api/v1/task.go @@ -47,15 +47,16 @@ const ( type MessageType string const ( - MessageTypeMessage MessageType = "message" - MessageTypeContent MessageType = "content" - MessageTypeToolCall MessageType = "tool_call" - MessageTypeProgress MessageType = "progress" - MessageTypeLog MessageType = "log" - MessageTypeError MessageType = "error" - MessageTypeStatus MessageType = "status" - MessageTypeThinking MessageType = "thinking" - MessageTypeTodo MessageType = "todo" + MessageTypeMessage MessageType = "message" + MessageTypeContent MessageType = "content" + MessageTypeToolCall MessageType = "tool_call" + MessageTypeProgress MessageType = "progress" + MessageTypeLog MessageType = "log" + MessageTypeError MessageType = "error" + MessageTypeStatus MessageType = "status" + MessageTypeThinking MessageType = "thinking" + MessageTypeAdvisorFeedback MessageType = "advisor_feedback" + MessageTypeTodo MessageType = "todo" ) // TaskRepository represents a repository associated with a task diff --git a/apps/web/components/task/chat/message-renderer.test.tsx b/apps/web/components/task/chat/message-renderer.test.tsx index ae7be14602..0794889c24 100644 --- a/apps/web/components/task/chat/message-renderer.test.tsx +++ b/apps/web/components/task/chat/message-renderer.test.tsx @@ -10,7 +10,12 @@ vi.mock("@/hooks/use-panel-actions", () => ({ })); vi.mock("@/components/state-provider", () => ({ - useAppStore: () => null, + useAppStore: (selector: (state: unknown) => unknown) => + selector({ + turns: { bySession: {} }, + taskSessions: { items: {} }, + sessionModels: { bySessionId: {} }, + }), })); function message(overrides: Partial): Message { @@ -52,3 +57,21 @@ describe("MessageRenderer markdown file links", () => { expect(fallbackOpenFile).not.toHaveBeenCalled(); }); }); + +describe("MessageRenderer advisor feedback", () => { + it("renders advisor feedback with a distinct label", () => { + render( + , + ); + + expect(screen.getByText("OMP Advisor Feedback")).toBeTruthy(); + expect(screen.getByText("Good point. Verify the ACP conversion path.")).toBeTruthy(); + }); +}); diff --git a/apps/web/components/task/chat/message-renderer.tsx b/apps/web/components/task/chat/message-renderer.tsx index 2fcef0e5fd..b6efc79472 100644 --- a/apps/web/components/task/chat/message-renderer.tsx +++ b/apps/web/components/task/chat/message-renderer.tsx @@ -151,6 +151,25 @@ const adapters: MessageAdapter[] = [ matches: (comment) => comment.type === "thinking", render: (comment) => , }, + { + matches: (comment) => comment.type === "advisor_feedback", + render: (comment, ctx) => ( +
+
+ OMP Advisor Feedback +
+ +
+ ), + }, { matches: (comment) => comment.type === "todo", render: (comment) => , diff --git a/apps/web/e2e/tests/chat/advisor-feedback.spec.ts b/apps/web/e2e/tests/chat/advisor-feedback.spec.ts new file mode 100644 index 0000000000..a4d90d08c3 --- /dev/null +++ b/apps/web/e2e/tests/chat/advisor-feedback.spec.ts @@ -0,0 +1,39 @@ +import { test, expect } from "../../fixtures/test-base"; +import { openTaskSession } from "../../helpers/session"; + +test.describe("Advisor feedback messages", () => { + test("renders persisted advisor feedback in the session feed", async ({ + testPage, + apiClient, + seedData, + }) => { + const task = await apiClient.createTaskWithAgent( + seedData.workspaceId, + "Advisor feedback feed", + seedData.agentProfileId, + { + description: "/e2e:simple-message", + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + }, + ); + if (!task.session_id) throw new Error("createTaskWithAgent did not return a session_id"); + + const session = await openTaskSession(testPage, task.id); + await session.waitForChatIdle({ timeout: 30_000 }); + + await apiClient.seedSessionMessage(task.session_id, { + type: "advisor_feedback", + content: "Good point. Verify the ACP conversion path.", + metadata: { source: "advisor", severity: "concern" }, + }); + + await testPage.reload(); + await session.waitForLoad(); + + const chat = session.activeChat(); + await expect(chat.getByText("OMP Advisor Feedback")).toBeVisible(); + await expect(chat.getByText("Good point. Verify the ACP conversion path.")).toBeVisible(); + }); +}); diff --git a/apps/web/hooks/use-processed-messages.test.ts b/apps/web/hooks/use-processed-messages.test.ts index 824c5dcfb2..35b5531449 100644 --- a/apps/web/hooks/use-processed-messages.test.ts +++ b/apps/web/hooks/use-processed-messages.test.ts @@ -263,6 +263,23 @@ describe("buildGroupedRenderItems prepare progress placement", () => { }); }); +describe("buildGroupedRenderItems advisor feedback", () => { + it("keeps advisor feedback as a normal feed message", () => { + const advisor = makeMessage( + "advisor-1", + "advisor_feedback", + { source: "advisor", severity: "concern" }, + "Good point. Verify the ACP conversion path.", + ); + + const result = buildGroupedRenderItems([advisor], "s1", { + canAnchorPrepareProgress: false, + }); + + expect(result).toEqual([{ type: "message", message: advisor }]); + }); +}); + describe("insertLastAgentErrorItem", () => { it("inserts the notice after the nearest message before the error timestamp", () => { const before = { diff --git a/apps/web/hooks/use-processed-messages.ts b/apps/web/hooks/use-processed-messages.ts index d6dc251525..77b3bcb1fd 100644 --- a/apps/web/hooks/use-processed-messages.ts +++ b/apps/web/hooks/use-processed-messages.ts @@ -73,6 +73,7 @@ const VISIBLE_MESSAGE_TYPES: Set = new Set([ "status", "error", "thinking", + "advisor_feedback", "todo", "script_execution", "agent_plan", diff --git a/apps/web/lib/types/http.ts b/apps/web/lib/types/http.ts index ccca35c9f1..09a7ce0340 100644 --- a/apps/web/lib/types/http.ts +++ b/apps/web/lib/types/http.ts @@ -604,6 +604,7 @@ export type MessageType = | "error" | "status" | "thinking" + | "advisor_feedback" | "todo" | "permission_request" | "clarification_request"