feat: show OMP advisor feedback

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

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()

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"
 

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.

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 == "" {

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