fix: prevent duplicate agent messages in database (#181)

apps/backend/cmd/kandev/adapters.go

@@ -14,7 +14,6 @@ import (
 	"github.com/kandev/kandev/internal/common/logger"
 	"github.com/kandev/kandev/internal/orchestrator"
 	"github.com/kandev/kandev/internal/orchestrator/executor"
-	taskhandlers "github.com/kandev/kandev/internal/task/handlers"
 	"github.com/kandev/kandev/internal/task/models"
 	"github.com/kandev/kandev/internal/task/repository"
 	taskservice "github.com/kandev/kandev/internal/task/service"
@@ -212,25 +211,20 @@ func (a *lifecycleAdapter) ResolveAgentProfile(ctx context.Context, profileID st
 	}, nil
 }
 
-// orchestratorAdapter adapts the orchestrator.Service to the taskhandlers.OrchestratorService interface
-type orchestratorAdapter struct {
+// orchestratorWrapper wraps orchestrator.Service to implement taskhandlers.OrchestratorService.
+// The wrapper only adapts ResumeTaskSession which returns *executor.TaskExecution that we don't need.
+type orchestratorWrapper struct {
 	svc *orchestrator.Service
 }
 
-// PromptTask forwards to the orchestrator service and converts the result type
-func (a *orchestratorAdapter) PromptTask(ctx context.Context, taskID, taskSessionID, prompt, model string, planMode bool) (*taskhandlers.PromptResult, error) {
-	result, err := a.svc.PromptTask(ctx, taskID, taskSessionID, prompt, model, planMode)
-	if err != nil {
-		return nil, err
-	}
-	return &taskhandlers.PromptResult{
-		StopReason:   result.StopReason,
-		AgentMessage: result.AgentMessage,
-	}, nil
+// PromptTask forwards directly to the orchestrator service.
+func (w *orchestratorWrapper) PromptTask(ctx context.Context, taskID, taskSessionID, prompt, model string, planMode bool) (*orchestrator.PromptResult, error) {
+	return w.svc.PromptTask(ctx, taskID, taskSessionID, prompt, model, planMode)
 }
 
-func (a *orchestratorAdapter) ResumeTaskSession(ctx context.Context, taskID, taskSessionID string) error {
-	_, err := a.svc.ResumeTaskSession(ctx, taskID, taskSessionID)
+// ResumeTaskSession forwards to the orchestrator service, discarding the TaskExecution result.
+func (w *orchestratorWrapper) ResumeTaskSession(ctx context.Context, taskID, taskSessionID string) error {
+	_, err := w.svc.ResumeTaskSession(ctx, taskID, taskSessionID)
 	return err
 }
 

apps/backend/cmd/kandev/gateway.go

@@ -17,7 +17,6 @@ import (
 	notificationservice "github.com/kandev/kandev/internal/notifications/service"
 	notificationstore "github.com/kandev/kandev/internal/notifications/store"
 	"github.com/kandev/kandev/internal/orchestrator"
-	orchestratorcontroller "github.com/kandev/kandev/internal/orchestrator/controller"
 	orchestratorhandlers "github.com/kandev/kandev/internal/orchestrator/handlers"
 	"github.com/kandev/kandev/internal/task/repository"
 	taskservice "github.com/kandev/kandev/internal/task/service"
@@ -47,8 +46,7 @@ func provideGateway(
 		gateway.SetLifecycleManager(lifecycleMgr)
 	}
 
-	orchestratorCtrl := orchestratorcontroller.NewController(orchestratorSvc)
-	orchestratorHandlers := orchestratorhandlers.NewHandlers(orchestratorCtrl, log)
+	orchestratorHandlers := orchestratorhandlers.NewHandlers(orchestratorSvc, log)
 	orchestratorHandlers.RegisterHandlers(gateway.Dispatcher)
 
 	if lifecycleMgr != nil && agentRegistry != nil {

apps/backend/cmd/kandev/main.go

@@ -525,7 +525,7 @@ func main() {
 		gateway.Dispatcher,
 		messageController,
 		taskController,
-		&orchestratorAdapter{svc: orchestratorSvc},
+		&orchestratorWrapper{svc: orchestratorSvc},
 		log,
 	)
 	taskhandlers.RegisterProcessRoutes(router, taskSvc, lifecycleMgr, log)

apps/backend/docs/add_agent.md

@@ -357,7 +357,6 @@ Configure session handling in `session_config`:
 - `can_recover`: Whether sessions can survive backend restarts
 - `session_dir_template`: Where session data is stored on host
 - `session_dir_target`: Mount path inside container (if Docker-based)
-- `reports_status_via_stream`: If true, agent reports session status via stream protocol (prevents duplicate status messages)
 
 **Session ID Tracking for Stream-Based Protocols:**
 

apps/backend/internal/agent/lifecycle/manager.go

@@ -814,10 +814,6 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecuti
 		execution.ACPSessionID = req.ACPSessionID
 	}
 
-	// Set flag indicating if agent adapter reports completion via stream
-	// When true, session manager should NOT emit a complete event after prompt returns
-	execution.ReportsStatusViaStream = agentConfig.SessionConfig.ReportsStatusViaStream
-
 	// Build agent command string for later use with StartAgentProcess
 	model := ""
 	autoApprove := false
@@ -951,18 +947,13 @@ func (m *Manager) StartAgentProcess(ctx context.Context, executionID string) err
 		return fmt.Errorf("failed to resolve MCP config: %w", err)
 	}
 
-	// Capture the original ACP session ID before initialization overwrites it.
-	// This is used to determine if we're resuming an existing session or starting a new one.
-	providedACPSessionID := execution.ACPSessionID
-
-	// Initialize ACP session
+	// Initialize ACP session.
+	// Session status events are now emitted by all adapters via the stream.
 	if err := m.initializeACPSession(ctx, execution, agentConfig, taskDescription, mcpServers); err != nil {
 		m.updateExecutionError(executionID, "failed to initialize ACP: "+err.Error())
 		return fmt.Errorf("failed to initialize ACP: %w", err)
 	}
 
-	m.emitSessionStatusEvent(execution, agentConfig, providedACPSessionID)
-
 	return nil
 }
 
@@ -1200,38 +1191,6 @@ func (m *Manager) initializeACPSession(ctx context.Context, execution *AgentExec
 	return m.sessionManager.InitializeAndPrompt(ctx, execution, agentConfig, taskDescription, mcpServers, m.MarkReady)
 }
 
-// emitSessionStatusEvent emits a session_status event indicating whether the session was resumed or new.
-// providedACPSessionID is the ACP session ID that was provided BEFORE initialization (for resumption).
-// This must be captured before initializeACPSession runs, as that overwrites execution.ACPSessionID.
-func (m *Manager) emitSessionStatusEvent(execution *AgentExecution, agentConfig *registry.AgentTypeConfig, providedACPSessionID string) {
-	// Skip emitting for agents that report their session status via the stream protocol.
-	// These agents emit session_status events from their adapter when the real session info arrives,
-	// so emitting here would create duplicate "New session started" messages.
-	if agentConfig.SessionConfig.ReportsStatusViaStream {
-		return
-	}
-
-	wasResumed := false
-	if providedACPSessionID != "" {
-		if agentConfig.SessionConfig.NativeSessionResume {
-			wasResumed = true
-		} else if agentConfig.SessionConfig.ResumeFlag != "" && agentConfig.SessionConfig.SupportsRecovery() {
-			wasResumed = true
-		}
-	}
-
-	sessionStatus := streams.SessionStatusNew
-	if wasResumed {
-		sessionStatus = streams.SessionStatusResumed
-	}
-
-	m.eventPublisher.PublishAgentStreamEvent(execution, streams.AgentEvent{
-		Type:          streams.EventTypeSessionStatus,
-		SessionID:     execution.ACPSessionID,
-		SessionStatus: sessionStatus,
-	})
-}
-
 // handleAgentEvent processes incoming agent events from the agent
 func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.AgentEvent) {
 	// Log all incoming events for debugging
@@ -1365,7 +1324,6 @@ func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.Age
 		// Check buffer content BEFORE any processing
 		execution.messageMu.Lock()
 		bufferContentBeforeFlush := execution.messageBuffer.String()
-		streamingWasUsed := execution.streamingUsedThisTurn
 		currentMsgID := execution.currentMessageID
 		execution.messageMu.Unlock()
 
@@ -1387,53 +1345,21 @@ func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.Age
 			zap.String("execution_id", execution.ID),
 			zap.String("operation_id", event.OperationID),
 			zap.String("session_id", event.SessionID),
-			zap.Bool("streaming_was_used", streamingWasUsed),
 			zap.String("current_msg_id", currentMsgID),
-			zap.String("event_text", event.Text),
 			zap.Int("buffer_length", len(bufferContentBeforeFlush)),
 			zap.String("buffer_preview", bufferPreview),
 			zap.Bool("is_error", isError))
 
-		// Get the agent message text:
-		// 1. If streaming was used, flush the buffer (which publishes final chunk) and ignore event.Text
-		// 2. Otherwise, prefer text from event (adapter-provided, e.g., Amp) or flush buffer
-		var agentText string
-		if streamingWasUsed {
-			// Streaming messages were created - flush buffer to publish final chunk
-			// and return empty so we don't create duplicate messages
-			m.flushMessageBuffer(execution)
-			agentText = ""
-		} else if event.Text != "" {
-			// No streaming used, use adapter-provided text
-			agentText = event.Text
-			// Still clear the buffer to avoid stale data
-			m.flushMessageBuffer(execution)
-		} else {
-			// No streaming used, no adapter text - flush buffer for content
-			agentText = m.flushMessageBuffer(execution)
-		}
+		// Flush the message buffer to publish any remaining content as a streaming message.
+		// All adapters now handle duplicate prevention at the adapter layer:
+		// - They send text via message_chunk events (which accumulate in the buffer)
+		// - They send complete events WITHOUT text
+		// So we just flush the buffer here - no need to track streamingUsedThisTurn.
+		m.flushMessageBuffer(execution)
 
-		m.logger.Info("complete event text resolution",
+		m.logger.Info("complete event processed",
 			zap.String("execution_id", execution.ID),
-			zap.String("operation_id", event.OperationID),
-			zap.Int("agent_text_length", len(agentText)),
-			zap.Bool("will_save_message", agentText != ""))
-
-		// Reset the streaming flag for next turn
-		execution.messageMu.Lock()
-		execution.streamingUsedThisTurn = false
-		execution.messageMu.Unlock()
-
-		// Include agent text in the event for the orchestrator to save as an agent message
-		if agentText != "" {
-			event.Text = agentText
-			// Store agent message to session history for context injection
-			if m.historyManager != nil && execution.SessionID != "" {
-				if err := m.historyManager.AppendAgentMessage(execution.SessionID, agentText); err != nil {
-					m.logger.Warn("failed to store agent message to history", zap.Error(err))
-				}
-			}
-		}
+			zap.String("operation_id", event.OperationID))
 
 		// If this was an error completion, mark the execution as failed
 		// The agent process will likely exit soon after sending an error result
@@ -1587,9 +1513,6 @@ func (m *Manager) publishStreamingMessage(execution *AgentExecution, content str
 		messageID = uuid.New().String()
 		execution.currentMessageID = messageID
 	}
-
-	// Mark that streaming was used this turn (for duplicate prevention on complete)
-	execution.streamingUsedThisTurn = true
 	execution.messageMu.Unlock()
 
 	event := AgentStreamEventData{

apps/backend/internal/agent/lifecycle/manager_test.go

@@ -4,11 +4,13 @@ import (
 	"context"
 	"io"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/kandev/kandev/internal/agent/docker"
 	"github.com/kandev/kandev/internal/agent/registry"
+	agentctl "github.com/kandev/kandev/internal/agentctl/client"
 	"github.com/kandev/kandev/internal/common/logger"
 	"github.com/kandev/kandev/internal/events/bus"
 	v1 "github.com/kandev/kandev/pkg/api/v1"
@@ -603,3 +605,391 @@ func (m *mockPassthroughProfileResolver) ResolveProfile(ctx context.Context, pro
 		CLIPassthrough: m.cliPassthrough,
 	}, nil
 }
+
+// --- Tests for duplicate message prevention in handleAgentEvent ---
+
+// MockEventBusWithTracking provides detailed tracking of published events for testing
+type MockEventBusWithTracking struct {
+	PublishedEvents []trackedEvent
+	mu              sync.Mutex
+}
+
+type trackedEvent struct {
+	Subject string
+	Event   *bus.Event
+}
+
+func (m *MockEventBusWithTracking) Publish(ctx context.Context, subject string, event *bus.Event) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.PublishedEvents = append(m.PublishedEvents, trackedEvent{Subject: subject, Event: event})
+	return nil
+}
+
+func (m *MockEventBusWithTracking) Subscribe(subject string, handler bus.EventHandler) (bus.Subscription, error) {
+	return nil, nil
+}
+
+func (m *MockEventBusWithTracking) QueueSubscribe(subject, queue string, handler bus.EventHandler) (bus.Subscription, error) {
+	return nil, nil
+}
+
+func (m *MockEventBusWithTracking) Request(ctx context.Context, subject string, event *bus.Event, timeout time.Duration) (*bus.Event, error) {
+	return nil, nil
+}
+
+func (m *MockEventBusWithTracking) Close() {}
+
+func (m *MockEventBusWithTracking) IsConnected() bool {
+	return true
+}
+
+func (m *MockEventBusWithTracking) getStreamEvents() []AgentStreamEventPayload {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	var result []AgentStreamEventPayload
+	for _, te := range m.PublishedEvents {
+		if payload, ok := te.Event.Data.(AgentStreamEventPayload); ok {
+			result = append(result, payload)
+		}
+	}
+	return result
+}
+
+// createTestManagerWithTracking creates a manager with a tracking event bus for testing
+func createTestManagerWithTracking() (*Manager, *MockEventBusWithTracking) {
+	log := newTestLogger()
+	reg := newTestRegistry()
+	eventBus := &MockEventBusWithTracking{}
+	credsMgr := &MockCredentialsManager{}
+	profileResolver := &MockProfileResolver{}
+	mgr := NewManager(reg, eventBus, nil, nil, credsMgr, profileResolver, nil, RuntimeFallbackWarn, log)
+	return mgr, eventBus
+}
+
+// createTestExecution creates a test execution with proper initialization
+func createTestExecution(id, taskID, sessionID string) *AgentExecution {
+	return &AgentExecution{
+		ID:        id,
+		TaskID:    taskID,
+		SessionID: sessionID,
+		Status:    v1.AgentStatusRunning,
+		StartedAt: time.Now(),
+	}
+}
+
+// TestHandleAgentEvent_StreamingThenComplete tests the normal flow:
+// message_chunk events followed by complete event - should NOT create duplicate
+func TestHandleAgentEvent_StreamingThenComplete(t *testing.T) {
+	mgr, eventBus := createTestManagerWithTracking()
+	execution := createTestExecution("exec-1", "task-1", "session-1")
+	mgr.executionStore.Add(execution)
+
+	// Simulate streaming chunks with newlines (which trigger publishing)
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "message_chunk",
+		Text: "Hello, world!\n",
+	})
+
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "message_chunk",
+		Text: "This is a test.\n",
+	})
+
+	// Now send complete event
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "complete",
+	})
+
+	// Verify: streaming was used, so complete should NOT have text
+	events := eventBus.getStreamEvents()
+
+	// Count message_streaming events (creates/appends)
+	var messageStreamingEvents []AgentStreamEventPayload
+	var completeEvents []AgentStreamEventPayload
+	for _, e := range events {
+		if e.Data != nil {
+			switch e.Data.Type {
+			case "message_streaming":
+				messageStreamingEvents = append(messageStreamingEvents, e)
+			case "complete":
+				completeEvents = append(completeEvents, e)
+			}
+		}
+	}
+
+	// Should have streaming messages
+	if len(messageStreamingEvents) == 0 {
+		t.Error("expected message_streaming events, got none")
+	}
+
+	// Should have exactly one complete event
+	if len(completeEvents) != 1 {
+		t.Errorf("expected 1 complete event, got %d", len(completeEvents))
+	}
+
+	// The complete event should NOT have text (streaming handled it via buffer)
+	if len(completeEvents) > 0 && completeEvents[0].Data.Text != "" {
+		t.Errorf("complete event should not have text when streaming was used, got %q", completeEvents[0].Data.Text)
+	}
+}
+
+// TestHandleAgentEvent_StreamingThenToolCallThenComplete tests the scenario that could cause duplicates:
+// message_chunk → tool_call (clears currentMessageID) → complete
+// This verifies that the buffer is properly flushed on complete after tool calls
+func TestHandleAgentEvent_StreamingThenToolCallThenComplete(t *testing.T) {
+	mgr, eventBus := createTestManagerWithTracking()
+	execution := createTestExecution("exec-1", "task-1", "session-1")
+	mgr.executionStore.Add(execution)
+
+	// Simulate streaming chunks
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "message_chunk",
+		Text: "Let me check that for you.\n",
+	})
+
+	// Verify currentMessageID is set after message_chunk
+	execution.messageMu.Lock()
+	msgIDBeforeToolCall := execution.currentMessageID
+	execution.messageMu.Unlock()
+
+	if msgIDBeforeToolCall == "" {
+		t.Error("currentMessageID should be set after message_chunk")
+	}
+
+	// Tool call - this flushes buffer and clears currentMessageID
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type:       "tool_call",
+		ToolCallID: "tool-1",
+		ToolName:   "read_file",
+	})
+
+	// After tool call, currentMessageID should be cleared
+	execution.messageMu.Lock()
+	msgIDAfterToolCall := execution.currentMessageID
+	execution.messageMu.Unlock()
+
+	if msgIDAfterToolCall != "" {
+		t.Errorf("currentMessageID should be cleared after tool_call, got %q", msgIDAfterToolCall)
+	}
+
+	// Now complete - this should flush the buffer
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "complete",
+	})
+
+	events := eventBus.getStreamEvents()
+
+	// Find the complete event
+	var completeEvents []AgentStreamEventPayload
+	for _, e := range events {
+		if e.Data != nil && e.Data.Type == "complete" {
+			completeEvents = append(completeEvents, e)
+		}
+	}
+
+	if len(completeEvents) != 1 {
+		t.Errorf("expected 1 complete event, got %d", len(completeEvents))
+	}
+
+	// The complete event should NOT have text (streaming was used)
+	if len(completeEvents) > 0 && completeEvents[0].Data.Text != "" {
+		t.Errorf("complete event should not have text when streaming was used (even after tool_call), got %q",
+			completeEvents[0].Data.Text)
+	}
+}
+
+// TestHandleAgentEvent_CompleteWithoutStreaming verifies that complete events are
+// properly handled when no streaming was used (buffer is empty).
+// All adapters now send text via message_chunk events, so this tests the empty buffer case.
+func TestHandleAgentEvent_CompleteWithoutStreaming(t *testing.T) {
+	mgr, eventBus := createTestManagerWithTracking()
+	execution := createTestExecution("exec-1", "task-1", "session-1")
+	mgr.executionStore.Add(execution)
+
+	// Complete event without any prior streaming (e.g., agent did only tool calls)
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "complete",
+	})
+
+	events := eventBus.getStreamEvents()
+
+	var completeEvents []AgentStreamEventPayload
+	for _, e := range events {
+		if e.Data != nil && e.Data.Type == "complete" {
+			completeEvents = append(completeEvents, e)
+		}
+	}
+
+	if len(completeEvents) != 1 {
+		t.Errorf("expected 1 complete event, got %d", len(completeEvents))
+	}
+
+	// Complete event should be published successfully
+	// The buffer was empty, so no message_streaming events should be generated
+	var streamingEvents []AgentStreamEventPayload
+	for _, e := range events {
+		if e.Data != nil && e.Data.Type == "message_streaming" {
+			streamingEvents = append(streamingEvents, e)
+		}
+	}
+
+	if len(streamingEvents) != 0 {
+		t.Errorf("expected 0 message_streaming events when buffer is empty, got %d", len(streamingEvents))
+	}
+}
+
+// TestHandleAgentEvent_CompleteThenMessageChunk tests the scenario where
+// message_chunk arrives after complete. This documents the behavior when
+// an adapter incorrectly sends text after the turn has completed.
+// With the new architecture, adapters should NOT send message_chunk after complete.
+func TestHandleAgentEvent_CompleteThenMessageChunk(t *testing.T) {
+	mgr, eventBus := createTestManagerWithTracking()
+	execution := createTestExecution("exec-1", "task-1", "session-1")
+	mgr.executionStore.Add(execution)
+
+	// First, simulate normal streaming during the turn
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "message_chunk",
+		Text: "Processing your request...\n",
+	})
+
+	// Complete event arrives - this flushes the buffer
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "complete",
+	})
+
+	// Now message_chunk arrives AFTER complete
+	// This shouldn't happen with properly implemented adapters,
+	// but we document the behavior: it creates a new message
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "message_chunk",
+		Text: "Done!\n",
+	})
+
+	events := eventBus.getStreamEvents()
+
+	// Count message_streaming events
+	var messageStreamingEvents []AgentStreamEventPayload
+	for _, e := range events {
+		if e.Data != nil && e.Data.Type == "message_streaming" {
+			messageStreamingEvents = append(messageStreamingEvents, e)
+		}
+	}
+
+	// Document the behavior: message_chunk after complete starts a new message
+	t.Logf("Got %d message_streaming events", len(messageStreamingEvents))
+	for i, e := range messageStreamingEvents {
+		t.Logf("  Event %d: MessageID=%s, IsAppend=%v, Text=%q",
+			i, e.Data.MessageID, e.Data.IsAppend, e.Data.Text)
+	}
+
+	// The second message_chunk (after complete) should start a NEW message
+	// since currentMessageID was cleared by the complete event
+	if len(messageStreamingEvents) >= 2 {
+		lastEvent := messageStreamingEvents[len(messageStreamingEvents)-1]
+		if !lastEvent.Data.IsAppend {
+			t.Log("Expected behavior: message_chunk after complete creates a new message")
+		}
+	}
+}
+
+// TestHandleAgentEvent_MultipleToolCalls tests streaming → tool → streaming → tool → complete
+func TestHandleAgentEvent_MultipleToolCalls(t *testing.T) {
+	mgr, eventBus := createTestManagerWithTracking()
+	execution := createTestExecution("exec-1", "task-1", "session-1")
+	mgr.executionStore.Add(execution)
+
+	// Message before first tool
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "message_chunk",
+		Text: "Let me read the file.\n",
+	})
+
+	// First tool call
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type:       "tool_call",
+		ToolCallID: "tool-1",
+		ToolName:   "read_file",
+	})
+
+	// Tool update (complete)
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type:       "tool_update",
+		ToolCallID: "tool-1",
+		ToolStatus: "complete",
+	})
+
+	// Message after first tool
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "message_chunk",
+		Text: "Now let me modify it.\n",
+	})
+
+	// Second tool call
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type:       "tool_call",
+		ToolCallID: "tool-2",
+		ToolName:   "write_file",
+	})
+
+	// Tool update (complete)
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type:       "tool_update",
+		ToolCallID: "tool-2",
+		ToolStatus: "complete",
+	})
+
+	// Final message
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "message_chunk",
+		Text: "Done with both tasks!\n",
+	})
+
+	// Complete
+	mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+		Type: "complete",
+	})
+
+	events := eventBus.getStreamEvents()
+
+	// Count different event types
+	var messageStreamingCount, toolCallCount, completeCount int
+	for _, e := range events {
+		if e.Data != nil {
+			switch e.Data.Type {
+			case "message_streaming":
+				messageStreamingCount++
+			case "tool_call":
+				toolCallCount++
+			case "complete":
+				completeCount++
+			}
+		}
+	}
+
+	t.Logf("Events: message_streaming=%d, tool_call=%d, complete=%d",
+		messageStreamingCount, toolCallCount, completeCount)
+
+	// Should have multiple streaming messages (one per "segment" before tool calls)
+	if messageStreamingCount < 3 {
+		t.Errorf("expected at least 3 message_streaming events for 3 message segments, got %d", messageStreamingCount)
+	}
+
+	if toolCallCount != 2 {
+		t.Errorf("expected 2 tool_call events, got %d", toolCallCount)
+	}
+
+	if completeCount != 1 {
+		t.Errorf("expected 1 complete event, got %d", completeCount)
+	}
+
+	// Find the complete event and verify it has no text
+	for _, e := range events {
+		if e.Data != nil && e.Data.Type == "complete" {
+			if e.Data.Text != "" {
+				t.Errorf("complete event should not have text when streaming was used, got %q", e.Data.Text)
+			}
+		}
+	}
+}

apps/backend/internal/agent/lifecycle/session.go

@@ -11,7 +11,6 @@ import (
 	"github.com/kandev/kandev/internal/agent/registry"
 	agentctl "github.com/kandev/kandev/internal/agentctl/client"
 	agentctltypes "github.com/kandev/kandev/internal/agentctl/types"
-	"github.com/kandev/kandev/internal/agentctl/types/streams"
 	"github.com/kandev/kandev/internal/common/appctx"
 	"github.com/kandev/kandev/internal/common/constants"
 	"github.com/kandev/kandev/internal/common/logger"
@@ -455,83 +454,26 @@ func (sm *SessionManager) SendPrompt(
 		return nil, fmt.Errorf("prompt failed: %w", err)
 	}
 
-	// Extract accumulated content from buffers and handle streaming message completion
-	// NOTE: When ReportsStatusViaStream is true, the adapter's complete event (via handleAgentEvent)
-	// will handle the buffer content. We should NOT clear the buffer here to avoid a race condition
-	// where the buffer is cleared before handleAgentEvent processes the complete event.
+	// Extract accumulated content from buffers for the return value.
+	// All adapters now emit complete events via the stream, so handleAgentEvent will
+	// handle the buffer content when the complete event arrives. We just peek at the
+	// buffer here to return the message content to the caller.
 	stopReason := ""
 	if resp != nil {
 		stopReason = string(resp.StopReason)
 	}
 
-	var agentMessage string
+	// Peek at the message for the return value, but don't clear the buffer.
+	// The complete event handler (handleAgentEvent) will clear the buffer.
+	execution.messageMu.Lock()
+	bufLen := execution.messageBuffer.Len()
+	agentMessage := execution.messageBuffer.String()
+	execution.messageMu.Unlock()
 
-	if execution.ReportsStatusViaStream {
-		// Agent adapter handles completion via stream - let handleAgentEvent deal with the buffer
-		// Just log for debugging. The buffer content will be handled by handleAgentEvent when
-		// the complete event arrives from the adapter.
-		execution.messageMu.Lock()
-		bufLen := execution.messageBuffer.Len()
-		// Peek at the message for the return value, but don't clear the buffer
-		agentMessage = execution.messageBuffer.String()
-		execution.messageMu.Unlock()
-
-		sm.logger.Info("ACP prompt completed (stream-based completion)",
-			zap.String("execution_id", execution.ID),
-			zap.String("stop_reason", stopReason),
-			zap.Int("buffer_length", bufLen),
-			zap.Bool("reports_status_via_stream", true))
-	} else {
-		// Agent adapter does NOT handle completion via stream - we need to handle the buffer here
-		execution.messageMu.Lock()
-		agentMessage = execution.messageBuffer.String()
-		execution.messageBuffer.Reset()
-		execution.thinkingBuffer.Reset()
-		currentMsgID := execution.currentMessageID
-		execution.currentMessageID = ""  // Clear for next turn
-		execution.currentThinkingID = "" // Clear for next turn
-		execution.messageMu.Unlock()
-
-		trimmedMessage := strings.TrimSpace(agentMessage)
-
-		sm.logger.Info("ACP prompt completed",
-			zap.String("execution_id", execution.ID),
-			zap.String("stop_reason", stopReason),
-			zap.Int("message_length", len(trimmedMessage)),
-			zap.Bool("has_streaming_msg", currentMsgID != ""))
-
-		// Handle remaining content from the buffer
-		if sm.eventPublisher != nil {
-			// If there's remaining content and an active streaming message,
-			// append it to the streaming message instead of creating a new one
-			if trimmedMessage != "" && currentMsgID != "" {
-				// Publish streaming append with the remaining content
-				sm.eventPublisher.PublishAgentStreamEventPayload(&AgentStreamEventPayload{
-					Type:      "agent/event",
-					Timestamp: time.Now().UTC().Format(time.RFC3339),
-					AgentID:   execution.AgentProfileID,
-					TaskID:    execution.TaskID,
-					SessionID: execution.SessionID,
-					Data: &AgentStreamEventData{
-						Type:      "message_streaming",
-						MessageID: currentMsgID,
-						IsAppend:  true,
-						Text:      trimmedMessage,
-					},
-				})
-				// Content was handled via streaming, publish complete with empty text
-				trimmedMessage = ""
-			}
-
-			// Publish complete event (with remaining text only if no streaming message was active)
-			completeEvent := streams.AgentEvent{
-				Type:      streams.EventTypeComplete,
-				SessionID: execution.ACPSessionID,
-				Text:      trimmedMessage,
-			}
-			sm.eventPublisher.PublishAgentStreamEvent(execution, completeEvent)
-		}
-	}
+	sm.logger.Info("prompt completed",
+		zap.String("execution_id", execution.ID),
+		zap.String("stop_reason", stopReason),
+		zap.Int("buffer_length", bufLen))
 
 	// Mark as READY for next prompt
 	if err := markReady(execution.ID); err != nil {

apps/backend/internal/agent/lifecycle/types.go

@@ -60,22 +60,12 @@ type AgentExecution struct {
 	currentMessageID  string
 	currentThinkingID string
 
-	// Track whether streaming messages were created during this turn
-	// This is set true when publishStreamingMessage is called and cleared on complete
-	// Used to avoid duplicate messages when both streaming and complete event have text
-	streamingUsedThisTurn bool
-
 	// Resume context tracking for fork_session pattern (ACP agents that don't support session/load)
 	// needsResumeContext is set to true when the session has history that should be injected
 	// resumeContextInjected is set to true after context has been injected into a prompt
 	needsResumeContext    bool
 	resumeContextInjected bool
 
-	// ReportsStatusViaStream indicates whether the agent adapter handles completion events
-	// via the stream protocol. When true, the session manager should NOT emit a complete
-	// event after prompt returns, as the adapter already sent one via the stream.
-	ReportsStatusViaStream bool
-
 	// Available commands from the agent (for slash command menu)
 	availableCommands   []streams.AvailableCommand
 	availableCommandsMu sync.RWMutex

apps/backend/internal/agent/registry/agents.json

@@ -369,8 +369,7 @@
         "resume_flag": "--resume",
         "can_recover": true,
         "session_dir_template": "{home}/.claude",
-        "session_dir_target": "",
-        "reports_status_via_stream": true
+        "session_dir_target": ""
       },
       "permission_config": {
         "permission_flag": "",
@@ -494,8 +493,7 @@
         "resume_flag": "",
         "can_recover": false,
         "session_dir_template": "{home}/.gemini",
-        "session_dir_target": "",
-        "reports_status_via_stream": false
+        "session_dir_target": ""
       },
       "permission_config": {
         "permission_flag": "",
@@ -634,8 +632,7 @@
         "resume_flag": "",
         "can_recover": true,
         "session_dir_template": "{home}/.opencode",
-        "session_dir_target": "",
-        "reports_status_via_stream": true
+        "session_dir_target": ""
       },
       "permission_config": {
         "permission_flag": "",
@@ -785,8 +782,7 @@
         "resume_flag": "--resume",
         "can_recover": true,
         "session_dir_template": "{home}/.copilot",
-        "session_dir_target": "",
-        "reports_status_via_stream": true
+        "session_dir_target": ""
       },
       "permission_config": {
         "permission_flag": "",
@@ -990,7 +986,6 @@
         "can_recover": true,
         "session_dir_template": "{home}/.config/amp",
         "session_dir_target": "",
-        "reports_status_via_stream": true,
         "fork_session_cmd": [
           "npx",
           "-y",

apps/backend/internal/agent/registry/registry.go

@@ -96,13 +96,6 @@ type SessionConfig struct {
 	// SessionDirTarget is the container path where the session directory is mounted.
 	// Example: "/root/.augment/sessions" for auggie.
 	SessionDirTarget string `json:"session_dir_target,omitempty"`
-
-	// ReportsStatusViaStream indicates that the agent reports its session status
-	// via the stream protocol (e.g., system message in stream-json).
-	// When true, the lifecycle manager will NOT emit a session_status event,
-	// since the agent's adapter will emit it when the real session info arrives.
-	// This prevents duplicate "New session started" messages.
-	ReportsStatusViaStream bool `json:"reports_status_via_stream,omitempty"`
 }
 
 // SupportsRecovery returns whether the agent supports session recovery after backend restart.

apps/backend/internal/agentctl/client/agent.go

@@ -272,7 +272,7 @@ func (c *Client) StreamUpdates(ctx context.Context, handler func(AgentEvent), mc
 	}
 
 	c.mu.Lock()
-	c.acpConn = conn
+	c.agentStreamConn = conn
 	c.mu.Unlock()
 
 	c.logger.Info("connected to updates stream", zap.String("url", wsURL))
@@ -289,7 +289,7 @@ func (c *Client) StreamUpdates(ctx context.Context, handler func(AgentEvent), mc
 	go func() {
 		defer func() {
 			c.mu.Lock()
-			c.acpConn = nil
+			c.agentStreamConn = nil
 			c.mu.Unlock()
 			if err := conn.Close(); err != nil {
 				c.logger.Debug("failed to close updates websocket", zap.Error(err))
@@ -352,11 +352,11 @@ func (c *Client) CloseUpdatesStream() {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
-	if c.acpConn != nil {
-		if err := c.acpConn.Close(); err != nil {
+	if c.agentStreamConn != nil {
+		if err := c.agentStreamConn.Close(); err != nil {
 			c.logger.Debug("failed to close agent events stream", zap.Error(err))
 		}
-		c.acpConn = nil
+		c.agentStreamConn = nil
 	}
 }
 

apps/backend/internal/agentctl/client/client.go

@@ -25,7 +25,7 @@ type Client struct {
 	logger     *logger.Logger
 
 	// WebSocket connections for streaming
-	acpConn             *websocket.Conn
+	agentStreamConn     *websocket.Conn
 	workspaceStreamConn *websocket.Conn
 	mu                  sync.RWMutex
 }

apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go

@@ -194,6 +194,22 @@ func (a *Adapter) NewSession(ctx context.Context, mcpServers []types.McpServer)
 	a.sessionID = string(resp.SessionId)
 	a.logger.Info("created new session", zap.String("session_id", a.sessionID))
 
+	// Emit session status event to normalize with other adapters.
+	// This eliminates the need for ReportsStatusViaStream flag.
+	select {
+	case a.updatesCh <- AgentEvent{
+		Type:          streams.EventTypeSessionStatus,
+		SessionID:     a.sessionID,
+		SessionStatus: streams.SessionStatusNew,
+		Data: map[string]any{
+			"session_status": streams.SessionStatusNew,
+			"init":           true,
+		},
+	}:
+	default:
+		a.logger.Warn("updates channel full, could not emit session_status event")
+	}
+
 	return a.sessionID, nil
 }
 
@@ -251,11 +267,28 @@ func (a *Adapter) LoadSession(ctx context.Context, sessionID string) error {
 	a.sessionID = sessionID
 	a.logger.Info("loaded session", zap.String("session_id", a.sessionID))
 
+	// Emit session status event to normalize with other adapters.
+	// This eliminates the need for ReportsStatusViaStream flag.
+	select {
+	case a.updatesCh <- AgentEvent{
+		Type:          streams.EventTypeSessionStatus,
+		SessionID:     a.sessionID,
+		SessionStatus: streams.SessionStatusResumed,
+		Data: map[string]any{
+			"session_status": streams.SessionStatusResumed,
+			"init":           true,
+		},
+	}:
+	default:
+		a.logger.Warn("updates channel full, could not emit session_status event")
+	}
+
 	return nil
 }
 
 // Prompt sends a prompt to the agent.
 // If pending context is set (from SetPendingContext), it will be prepended to the message.
+// When the prompt completes, a complete event is emitted via the updates channel.
 func (a *Adapter) Prompt(ctx context.Context, message string) error {
 	a.mu.Lock()
 	conn := a.acpConn
@@ -283,7 +316,24 @@ func (a *Adapter) Prompt(ctx context.Context, message string) error {
 		SessionId: acp.SessionId(sessionID),
 		Prompt:    []acp.ContentBlock{acp.TextBlock(finalMessage)},
 	})
-	return err
+	if err != nil {
+		return err
+	}
+
+	// Emit complete event via the stream.
+	// This normalizes ACP behavior to match other adapters (stream-json, amp, copilot, opencode).
+	// All adapters now emit complete events, eliminating the need for ReportsStatusViaStream flag.
+	a.logger.Debug("emitting complete event after prompt", zap.String("session_id", sessionID))
+	select {
+	case a.updatesCh <- AgentEvent{
+		Type:      streams.EventTypeComplete,
+		SessionID: sessionID,
+	}:
+	default:
+		a.logger.Warn("updates channel full, could not emit complete event")
+	}
+
+	return nil
 }
 
 // SetPendingContext sets the context to be injected into the next prompt.

apps/backend/internal/agentctl/server/adapter/transport/codex/adapter.go

@@ -458,6 +458,16 @@ func (a *Adapter) Prompt(ctx context.Context, message string) error {
 			return fmt.Errorf("turn failed: %s", completeResult.err)
 		}
 		a.logger.Info("turn completed", zap.String("turn_id", turnID), zap.Bool("success", completeResult.success))
+
+		// Emit complete event via the stream.
+		// This normalizes Codex behavior to match other adapters.
+		// All adapters now emit complete events, eliminating the need for protocol-specific flags.
+		a.sendUpdate(AgentEvent{
+			Type:        streams.EventTypeComplete,
+			SessionID:   threadID,
+			OperationID: turnID,
+		})
+
 		return nil
 	}
 }

apps/backend/internal/agentctl/server/adapter/transport/streamjson/adapter.go

@@ -95,6 +95,11 @@ type Adapter struct {
 	// Available commands from initialize response (stored until session is created)
 	pendingAvailableCommands []streams.AvailableCommand
 
+	// Track whether text was streamed this turn to prevent duplicates from result.text
+	// This is set to true when we send message_chunk events from assistant messages,
+	// and reset to false at the start of each prompt.
+	streamingTextSentThisTurn bool
+
 	// Synchronization
 	mu     sync.RWMutex
 	closed bool
@@ -698,6 +703,12 @@ func (a *Adapter) handleAssistantMessage(msg *claudecode.CLIMessage, sessionID,
 		switch block.Type {
 		case "text":
 			if block.Text != "" {
+				// Mark that we've sent streaming text this turn
+				// This prevents duplicate content from result.text
+				a.mu.Lock()
+				a.streamingTextSentThisTurn = true
+				a.mu.Unlock()
+
 				a.sendUpdate(AgentEvent{
 					Type:        streams.EventTypeMessageChunk,
 					SessionID:   sessionID,
@@ -923,7 +934,41 @@ func (a *Adapter) handleResultMessage(msg *claudecode.CLIMessage, sessionID, ope
 		})
 	}
 
-	// Send completion event
+	// Check if text was already streamed this turn via assistant messages
+	a.mu.RLock()
+	textWasStreamed := a.streamingTextSentThisTurn
+	a.mu.RUnlock()
+
+	// Only send result.text if NO text was streamed this turn.
+	// If text was already streamed via assistant messages, result.text is a duplicate.
+	// This prevents the same content from appearing twice in the conversation.
+	if !textWasStreamed {
+		var resultText string
+		if resultData := msg.GetResultData(); resultData != nil && resultData.Text != "" {
+			resultText = resultData.Text
+		} else if resultStr := msg.GetResultString(); resultStr != "" {
+			resultText = resultStr
+		}
+		if resultText != "" {
+			a.logger.Debug("sending result text as message_chunk (no streaming text this turn)",
+				zap.Int("text_length", len(resultText)))
+			a.sendUpdate(AgentEvent{
+				Type:        streams.EventTypeMessageChunk,
+				SessionID:   sessionID,
+				OperationID: operationID,
+				Text:        resultText,
+			})
+		}
+	} else {
+		a.logger.Debug("skipping result text (streaming text already sent this turn)")
+	}
+
+	// Reset the streaming flag for next turn
+	a.mu.Lock()
+	a.streamingTextSentThisTurn = false
+	a.mu.Unlock()
+
+	// Send completion event AFTER any result text chunk
 	a.sendUpdate(AgentEvent{
 		Type:        streams.EventTypeComplete,
 		SessionID:   sessionID,
@@ -938,23 +983,6 @@ func (a *Adapter) handleResultMessage(msg *claudecode.CLIMessage, sessionID, ope
 		},
 	})
 
-	// If there's result text, send it as a message chunk
-	// Result can be either a ResultData object or a plain string
-	var resultText string
-	if resultData := msg.GetResultData(); resultData != nil && resultData.Text != "" {
-		resultText = resultData.Text
-	} else if resultStr := msg.GetResultString(); resultStr != "" {
-		resultText = resultStr
-	}
-	if resultText != "" {
-		a.sendUpdate(AgentEvent{
-			Type:        streams.EventTypeMessageChunk,
-			SessionID:   sessionID,
-			OperationID: operationID,
-			Text:        resultText,
-		})
-	}
-
 	// Signal completion
 	a.mu.RLock()
 	resultCh := a.resultCh

apps/backend/internal/integration/orchestrator_test.go

@@ -29,7 +29,6 @@ import (
 	"github.com/kandev/kandev/internal/events/bus"
 	gateways "github.com/kandev/kandev/internal/gateway/websocket"
 	"github.com/kandev/kandev/internal/orchestrator"
-	orchestratorcontroller "github.com/kandev/kandev/internal/orchestrator/controller"
 	"github.com/kandev/kandev/internal/orchestrator/executor"
 	orchestratorhandlers "github.com/kandev/kandev/internal/orchestrator/handlers"
 	taskcontroller "github.com/kandev/kandev/internal/task/controller"
@@ -672,8 +671,7 @@ func NewOrchestratorTestServer(t *testing.T) *OrchestratorTestServer {
 	gateway := gateways.NewGateway(log)
 
 	// Register orchestrator handlers (Pattern A)
-	orchestratorCtrl := orchestratorcontroller.NewController(orchestratorSvc)
-	orchestratorHandlers := orchestratorhandlers.NewHandlers(orchestratorCtrl, log)
+	orchestratorHandlers := orchestratorhandlers.NewHandlers(orchestratorSvc, log)
 	orchestratorHandlers.RegisterHandlers(gateway.Dispatcher)
 
 	// Start hub

apps/backend/internal/orchestrator/controller/controller.go

@@ -1,205 +0,0 @@
-// Package controller provides the business logic coordination layer for the orchestrator.
-package controller
-
-import (
-	"context"
-
-	"github.com/kandev/kandev/internal/orchestrator"
-	"github.com/kandev/kandev/internal/orchestrator/dto"
-)
-
-// Controller coordinates orchestrator business logic
-type Controller struct {
-	service *orchestrator.Service
-}
-
-// NewController creates a new orchestrator controller
-func NewController(svc *orchestrator.Service) *Controller {
-	return &Controller{
-		service: svc,
-	}
-}
-
-// GetStatus returns the orchestrator status
-func (c *Controller) GetStatus(ctx context.Context, req dto.GetStatusRequest) (dto.StatusResponse, error) {
-	status := c.service.GetStatus()
-	return dto.StatusResponse{
-		Running:      status.Running,
-		ActiveAgents: status.ActiveAgents,
-		QueuedTasks:  status.QueuedTasks,
-	}, nil
-}
-
-// GetQueue returns the queued tasks
-func (c *Controller) GetQueue(ctx context.Context, req dto.GetQueueRequest) (dto.QueueResponse, error) {
-	queuedTasks := c.service.GetQueuedTasks()
-
-	tasks := make([]dto.QueuedTaskDTO, 0, len(queuedTasks))
-	for _, qt := range queuedTasks {
-		tasks = append(tasks, dto.QueuedTaskDTO{
-			TaskID:   qt.TaskID,
-			Priority: qt.Priority,
-			QueuedAt: qt.QueuedAt.Format("2006-01-02T15:04:05Z"),
-		})
-	}
-
-	return dto.QueueResponse{
-		Tasks: tasks,
-		Total: len(tasks),
-	}, nil
-}
-
-// TriggerTask triggers a task for execution
-func (c *Controller) TriggerTask(ctx context.Context, req dto.TriggerTaskRequest) (dto.TriggerTaskResponse, error) {
-	// For now, just return success - the service would need task details
-	return dto.TriggerTaskResponse{
-		Success: true,
-		Message: "Task triggered",
-		TaskID:  req.TaskID,
-	}, nil
-}
-
-// StartTask starts task execution.
-// If SessionID is provided, resumes the existing session and sends a prompt with workflow step config.
-// If SessionID is not provided, creates a new session (requires AgentProfileID).
-func (c *Controller) StartTask(ctx context.Context, req dto.StartTaskRequest) (dto.StartTaskResponse, error) {
-	// If session_id is provided, resume existing session with workflow step config
-	if req.SessionID != "" {
-		err := c.service.StartSessionForWorkflowStep(ctx, req.TaskID, req.SessionID, req.WorkflowStepID)
-		if err != nil {
-			return dto.StartTaskResponse{}, err
-		}
-		return dto.StartTaskResponse{
-			Success:       true,
-			TaskID:        req.TaskID,
-			TaskSessionID: req.SessionID,
-			State:         "RUNNING",
-		}, nil
-	}
-
-	// Otherwise, create a new session
-	execution, err := c.service.StartTask(ctx, req.TaskID, req.AgentProfileID, req.ExecutorID, req.Priority, req.Prompt, req.WorkflowStepID)
-	if err != nil {
-		return dto.StartTaskResponse{}, err
-	}
-
-	resp := dto.StartTaskResponse{
-		Success:          true,
-		TaskID:           execution.TaskID,
-		AgentExecutionID: execution.AgentExecutionID,
-		TaskSessionID:    execution.SessionID,
-		State:            string(execution.SessionState),
-	}
-
-	// Include worktree info if available
-	if execution.WorktreePath != "" {
-		resp.WorktreePath = &execution.WorktreePath
-	}
-	if execution.WorktreeBranch != "" {
-		resp.WorktreeBranch = &execution.WorktreeBranch
-	}
-
-	return resp, nil
-}
-
-// ResumeTaskSession resumes a specific task session execution.
-func (c *Controller) ResumeTaskSession(ctx context.Context, req dto.ResumeTaskSessionRequest) (dto.ResumeTaskSessionResponse, error) {
-	execution, err := c.service.ResumeTaskSession(ctx, req.TaskID, req.TaskSessionID)
-	if err != nil {
-		return dto.ResumeTaskSessionResponse{}, err
-	}
-
-	resp := dto.ResumeTaskSessionResponse{
-		Success:          true,
-		TaskID:           execution.TaskID,
-		AgentExecutionID: execution.AgentExecutionID,
-		TaskSessionID:    execution.SessionID,
-		State:            string(execution.SessionState),
-	}
-
-	if execution.WorktreePath != "" {
-		resp.WorktreePath = &execution.WorktreePath
-	}
-	if execution.WorktreeBranch != "" {
-		resp.WorktreeBranch = &execution.WorktreeBranch
-	}
-
-	return resp, nil
-}
-
-// GetTaskSessionStatus returns the status of a task session including whether it's resumable
-func (c *Controller) GetTaskSessionStatus(ctx context.Context, req dto.TaskSessionStatusRequest) (dto.TaskSessionStatusResponse, error) {
-	status, err := c.service.GetTaskSessionStatus(ctx, req.TaskID, req.TaskSessionID)
-	if err != nil {
-		return dto.TaskSessionStatusResponse{
-			SessionID: req.TaskSessionID,
-			TaskID:    req.TaskID,
-			Error:     err.Error(),
-		}, nil
-	}
-	return status, nil
-}
-
-// StopTask stops task execution
-func (c *Controller) StopTask(ctx context.Context, req dto.StopTaskRequest) (dto.SuccessResponse, error) {
-	reason := req.Reason
-	if reason == "" {
-		reason = "stopped via API"
-	}
-
-	if err := c.service.StopTask(ctx, req.TaskID, reason, req.Force); err != nil {
-		return dto.SuccessResponse{}, err
-	}
-
-	return dto.SuccessResponse{Success: true}, nil
-}
-
-// PromptTask sends a prompt to a running task
-func (c *Controller) PromptTask(ctx context.Context, req dto.PromptTaskRequest) (dto.PromptTaskResponse, error) {
-	result, err := c.service.PromptTask(ctx, req.TaskID, req.TaskSessionID, req.Prompt, req.Model, req.PlanMode)
-	if err != nil {
-		return dto.PromptTaskResponse{}, err
-	}
-
-	return dto.PromptTaskResponse{
-		Success:    true,
-		StopReason: result.StopReason,
-	}, nil
-}
-
-// CompleteTask marks a task as complete
-func (c *Controller) CompleteTask(ctx context.Context, req dto.CompleteTaskRequest) (dto.CompleteTaskResponse, error) {
-	if err := c.service.CompleteTask(ctx, req.TaskID); err != nil {
-		return dto.CompleteTaskResponse{}, err
-	}
-
-	return dto.CompleteTaskResponse{
-		Success: true,
-		Message: "task completed",
-	}, nil
-}
-
-// RespondToPermission responds to a permission request
-func (c *Controller) RespondToPermission(ctx context.Context, req dto.PermissionRespondRequest) (dto.PermissionRespondResponse, error) {
-	if err := c.service.RespondToPermission(ctx, req.SessionID, req.PendingID, req.OptionID, req.Cancelled); err != nil {
-		return dto.PermissionRespondResponse{}, err
-	}
-
-	return dto.PermissionRespondResponse{
-		Success:   true,
-		SessionID: req.SessionID,
-		PendingID: req.PendingID,
-	}, nil
-}
-
-// CancelAgent interrupts the current agent turn without terminating the process.
-func (c *Controller) CancelAgent(ctx context.Context, req dto.CancelAgentRequest) (dto.CancelAgentResponse, error) {
-	if err := c.service.CancelAgent(ctx, req.SessionID); err != nil {
-		return dto.CancelAgentResponse{}, err
-	}
-
-	return dto.CancelAgentResponse{
-		Success:   true,
-		SessionID: req.SessionID,
-	}, nil
-}

apps/backend/internal/orchestrator/handlers/handlers.go

@@ -5,7 +5,7 @@ import (
 	"context"
 
 	"github.com/kandev/kandev/internal/common/logger"
-	"github.com/kandev/kandev/internal/orchestrator/controller"
+	"github.com/kandev/kandev/internal/orchestrator"
 	"github.com/kandev/kandev/internal/orchestrator/dto"
 	ws "github.com/kandev/kandev/pkg/websocket"
 	"go.uber.org/zap"
@@ -13,15 +13,15 @@ import (
 
 // Handlers contains WebSocket handlers for the orchestrator API
 type Handlers struct {
-	controller *controller.Controller
-	logger     *logger.Logger
+	service *orchestrator.Service
+	logger  *logger.Logger
 }
 
 // NewHandlers creates a new WebSocket handlers instance
-func NewHandlers(ctrl *controller.Controller, log *logger.Logger) *Handlers {
+func NewHandlers(svc *orchestrator.Service, log *logger.Logger) *Handlers {
 	return &Handlers{
-		controller: ctrl,
-		logger:     log.WithFields(zap.String("component", "orchestrator-handlers")),
+		service: svc,
+		logger:  log.WithFields(zap.String("component", "orchestrator-handlers")),
 	}
 }
 
@@ -43,19 +43,28 @@ func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) {
 // WS handlers
 
 func (h *Handlers) wsGetStatus(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
-	resp, err := h.controller.GetStatus(ctx, dto.GetStatusRequest{})
-	if err != nil {
-		h.logger.Error("failed to get status", zap.Error(err))
-		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to get status", nil)
+	status := h.service.GetStatus()
+	resp := dto.StatusResponse{
+		Running:      status.Running,
+		ActiveAgents: status.ActiveAgents,
+		QueuedTasks:  status.QueuedTasks,
 	}
 	return ws.NewResponse(msg.ID, msg.Action, resp)
 }
 
 func (h *Handlers) wsGetQueue(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
-	resp, err := h.controller.GetQueue(ctx, dto.GetQueueRequest{})
-	if err != nil {
-		h.logger.Error("failed to get queue", zap.Error(err))
-		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to get queue", nil)
+	queuedTasks := h.service.GetQueuedTasks()
+	tasks := make([]dto.QueuedTaskDTO, 0, len(queuedTasks))
+	for _, qt := range queuedTasks {
+		tasks = append(tasks, dto.QueuedTaskDTO{
+			TaskID:   qt.TaskID,
+			Priority: qt.Priority,
+			QueuedAt: qt.QueuedAt.Format("2006-01-02T15:04:05Z"),
+		})
+	}
+	resp := dto.QueueResponse{
+		Tasks: tasks,
+		Total: len(tasks),
 	}
 	return ws.NewResponse(msg.ID, msg.Action, resp)
 }
@@ -73,10 +82,11 @@ func (h *Handlers) wsTriggerTask(ctx context.Context, msg *ws.Message) (*ws.Mess
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil)
 	}
 
-	resp, err := h.controller.TriggerTask(ctx, dto.TriggerTaskRequest{TaskID: req.TaskID})
-	if err != nil {
-		h.logger.Error("failed to trigger task", zap.String("task_id", req.TaskID), zap.Error(err))
-		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to trigger task: "+err.Error(), nil)
+	// TriggerTask is a placeholder that just returns success
+	resp := dto.TriggerTaskResponse{
+		Success: true,
+		Message: "Task triggered",
+		TaskID:  req.TaskID,
 	}
 	return ws.NewResponse(msg.ID, msg.Action, resp)
 }
@@ -104,19 +114,42 @@ func (h *Handlers) wsStartTask(ctx context.Context, msg *ws.Message) (*ws.Messag
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "agent_profile_id is required when creating a new session", nil)
 	}
 
-	resp, err := h.controller.StartTask(ctx, dto.StartTaskRequest{
-		TaskID:         req.TaskID,
-		SessionID:      req.SessionID,
-		AgentProfileID: req.AgentProfileID,
-		ExecutorID:     req.ExecutorID,
-		Priority:       req.Priority,
-		Prompt:         req.Prompt,
-		WorkflowStepID: req.WorkflowStepID,
-	})
+	// If session_id is provided, resume existing session with workflow step config
+	if req.SessionID != "" {
+		err := h.service.StartSessionForWorkflowStep(ctx, req.TaskID, req.SessionID, req.WorkflowStepID)
+		if err != nil {
+			h.logger.Error("failed to start task", zap.String("task_id", req.TaskID), zap.Error(err))
+			return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to start task: "+err.Error(), nil)
+		}
+		resp := dto.StartTaskResponse{
+			Success:       true,
+			TaskID:        req.TaskID,
+			TaskSessionID: req.SessionID,
+			State:         "RUNNING",
+		}
+		return ws.NewResponse(msg.ID, msg.Action, resp)
+	}
+
+	// Otherwise, create a new session
+	execution, err := h.service.StartTask(ctx, req.TaskID, req.AgentProfileID, req.ExecutorID, req.Priority, req.Prompt, req.WorkflowStepID)
 	if err != nil {
 		h.logger.Error("failed to start task", zap.String("task_id", req.TaskID), zap.Error(err))
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to start task: "+err.Error(), nil)
 	}
+
+	resp := dto.StartTaskResponse{
+		Success:          true,
+		TaskID:           execution.TaskID,
+		AgentExecutionID: execution.AgentExecutionID,
+		TaskSessionID:    execution.SessionID,
+		State:            string(execution.SessionState),
+	}
+	if execution.WorktreePath != "" {
+		resp.WorktreePath = &execution.WorktreePath
+	}
+	if execution.WorktreeBranch != "" {
+		resp.WorktreeBranch = &execution.WorktreeBranch
+	}
 	return ws.NewResponse(msg.ID, msg.Action, resp)
 }
 
@@ -137,10 +170,7 @@ func (h *Handlers) wsResumeTaskSession(ctx context.Context, msg *ws.Message) (*w
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session_id is required", nil)
 	}
 
-	resp, err := h.controller.ResumeTaskSession(ctx, dto.ResumeTaskSessionRequest{
-		TaskID:        req.TaskID,
-		TaskSessionID: req.TaskSessionID,
-	})
+	execution, err := h.service.ResumeTaskSession(ctx, req.TaskID, req.TaskSessionID)
 	if err != nil {
 		h.logger.Error("failed to resume task session",
 			zap.String("task_id", req.TaskID),
@@ -148,6 +178,20 @@ func (h *Handlers) wsResumeTaskSession(ctx context.Context, msg *ws.Message) (*w
 			zap.Error(err))
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to resume task session: "+err.Error(), nil)
 	}
+
+	resp := dto.ResumeTaskSessionResponse{
+		Success:          true,
+		TaskID:           execution.TaskID,
+		AgentExecutionID: execution.AgentExecutionID,
+		TaskSessionID:    execution.SessionID,
+		State:            string(execution.SessionState),
+	}
+	if execution.WorktreePath != "" {
+		resp.WorktreePath = &execution.WorktreePath
+	}
+	if execution.WorktreeBranch != "" {
+		resp.WorktreeBranch = &execution.WorktreeBranch
+	}
 	return ws.NewResponse(msg.ID, msg.Action, resp)
 }
 
@@ -166,16 +210,15 @@ func (h *Handlers) wsStopTask(ctx context.Context, msg *ws.Message) (*ws.Message
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil)
 	}
 
-	resp, err := h.controller.StopTask(ctx, dto.StopTaskRequest{
-		TaskID: req.TaskID,
-		Reason: req.Reason,
-		Force:  req.Force,
-	})
-	if err != nil {
+	reason := req.Reason
+	if reason == "" {
+		reason = "stopped via API"
+	}
+	if err := h.service.StopTask(ctx, req.TaskID, reason, req.Force); err != nil {
 		h.logger.Error("failed to stop task", zap.String("task_id", req.TaskID), zap.Error(err))
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to stop task: "+err.Error(), nil)
 	}
-	return ws.NewResponse(msg.ID, msg.Action, resp)
+	return ws.NewResponse(msg.ID, msg.Action, dto.SuccessResponse{Success: true})
 }
 
 type wsPromptTaskRequest struct {
@@ -200,16 +243,15 @@ func (h *Handlers) wsPromptTask(ctx context.Context, msg *ws.Message) (*ws.Messa
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "prompt is required", nil)
 	}
 
-	resp, err := h.controller.PromptTask(ctx, dto.PromptTaskRequest{
-		TaskID:        req.TaskID,
-		TaskSessionID: req.TaskSessionID,
-		Prompt:        req.Prompt,
-		Model:         req.Model,
-	})
+	result, err := h.service.PromptTask(ctx, req.TaskID, req.TaskSessionID, req.Prompt, req.Model, false)
 	if err != nil {
 		h.logger.Error("failed to send prompt", zap.String("task_id", req.TaskID), zap.Error(err))
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to send prompt: "+err.Error(), nil)
 	}
+	resp := dto.PromptTaskResponse{
+		Success:    true,
+		StopReason: result.StopReason,
+	}
 	return ws.NewResponse(msg.ID, msg.Action, resp)
 }
 
@@ -226,11 +268,14 @@ func (h *Handlers) wsCompleteTask(ctx context.Context, msg *ws.Message) (*ws.Mes
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil)
 	}
 
-	resp, err := h.controller.CompleteTask(ctx, dto.CompleteTaskRequest{TaskID: req.TaskID})
-	if err != nil {
+	if err := h.service.CompleteTask(ctx, req.TaskID); err != nil {
 		h.logger.Error("failed to complete task", zap.String("task_id", req.TaskID), zap.Error(err))
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to complete task: "+err.Error(), nil)
 	}
+	resp := dto.CompleteTaskResponse{
+		Success: true,
+		Message: "task completed",
+	}
 	return ws.NewResponse(msg.ID, msg.Action, resp)
 }
 
@@ -262,16 +307,15 @@ func (h *Handlers) wsRespondToPermission(ctx context.Context, msg *ws.Message) (
 		zap.String("option_id", req.OptionID),
 		zap.Bool("cancelled", req.Cancelled))
 
-	resp, err := h.controller.RespondToPermission(ctx, dto.PermissionRespondRequest{
-		SessionID: req.SessionID,
-		PendingID: req.PendingID,
-		OptionID:  req.OptionID,
-		Cancelled: req.Cancelled,
-	})
-	if err != nil {
+	if err := h.service.RespondToPermission(ctx, req.SessionID, req.PendingID, req.OptionID, req.Cancelled); err != nil {
 		h.logger.Error("failed to respond to permission", zap.String("session_id", req.SessionID), zap.Error(err))
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to respond to permission: "+err.Error(), nil)
 	}
+	resp := dto.PermissionRespondResponse{
+		Success:   true,
+		SessionID: req.SessionID,
+		PendingID: req.PendingID,
+	}
 	return ws.NewResponse(msg.ID, msg.Action, resp)
 }
 
@@ -293,10 +337,8 @@ func (h *Handlers) wsGetTaskSessionStatus(ctx context.Context, msg *ws.Message)
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "session_id is required", nil)
 	}
 
-	resp, err := h.controller.GetTaskSessionStatus(ctx, dto.TaskSessionStatusRequest{
-		TaskID:        req.TaskID,
-		TaskSessionID: req.TaskSessionID,
-	})
+	// Service returns dto.TaskSessionStatusResponse directly
+	resp, err := h.service.GetTaskSessionStatus(ctx, req.TaskID, req.TaskSessionID)
 	if err != nil {
 		h.logger.Error("failed to get task session status",
 			zap.String("task_id", req.TaskID),
@@ -323,10 +365,13 @@ func (h *Handlers) wsCancelAgent(ctx context.Context, msg *ws.Message) (*ws.Mess
 	h.logger.Info("cancelling agent turn",
 		zap.String("session_id", req.SessionID))
 
-	resp, err := h.controller.CancelAgent(ctx, dto.CancelAgentRequest{SessionID: req.SessionID})
-	if err != nil {
+	if err := h.service.CancelAgent(ctx, req.SessionID); err != nil {
 		h.logger.Error("failed to cancel agent", zap.String("session_id", req.SessionID), zap.Error(err))
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to cancel agent: "+err.Error(), nil)
 	}
+	resp := dto.CancelAgentResponse{
+		Success:   true,
+		SessionID: req.SessionID,
+	}
 	return ws.NewResponse(msg.ID, msg.Action, resp)
 }

apps/backend/internal/task/handlers/message_handlers.go

@@ -12,6 +12,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/kandev/kandev/internal/common/logger"
+	"github.com/kandev/kandev/internal/orchestrator"
 	"github.com/kandev/kandev/internal/orchestrator/executor"
 	"github.com/kandev/kandev/internal/task/controller"
 	"github.com/kandev/kandev/internal/task/dto"
@@ -20,15 +21,9 @@ import (
 	ws "github.com/kandev/kandev/pkg/websocket"
 )
 
-// PromptResult represents the result of prompting an agent
-type PromptResult struct {
-	StopReason   string `json:"stop_reason"`
-	AgentMessage string `json:"agent_message"`
-}
-
 // OrchestratorService defines the interface for orchestrator operations
 type OrchestratorService interface {
-	PromptTask(ctx context.Context, taskID, sessionID, prompt, model string, planMode bool) (*PromptResult, error)
+	PromptTask(ctx context.Context, taskID, sessionID, prompt, model string, planMode bool) (*orchestrator.PromptResult, error)
 	ResumeTaskSession(ctx context.Context, taskID, taskSessionID string) error
 }