dreampiler/kandev
1
0
Fork 0
Code Issues 46 Pull Requests Actions 35 Packages Projects Releases Wiki Activity

refactor(backend): drop legacy orchestrator WS handlers superseded by session.launch (#803)

Browse Source 
* refactor(backend): drop legacy orchestrator WS handlers superseded by session.launch

The orchestrator exposed six WS handlers (orchestrator.start, orchestrator.trigger,
orchestrator.prompt, orchestrator.complete, task.session.resume, task.session.prepare)
that all delegated to the unified LaunchSession or are no longer reachable from any
client. Drop the handlers, their DTOs, and the dead action constants
(orchestrator.pause, orchestrator.resume) that were never registered. Migrate the
remaining integration tests to the unified session.launch action.

* refactor(backend): drop orphan DTO request types and fix stale doc reference

Address PR review feedback: the DTO types TaskSessionStatusRequest, StopTaskRequest,
and PermissionRespondRequest had no consumers (each was shadowed by an inline
private ws*Request type in the handler). AGENTS.md still referenced the removed
task.start action; switch it to session.launch.

---------

Co-authored-by: Kandev Agent <agent@kandev.dev>
This commit is contained in:
João Salavisa
2026-05-05 08:29:46 +01:00
committed by GitHub
parent 83de4793fd
commit ba32a44cb4
8 changed files with 24 additions and 626 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AGENTS.md
+1 -1
View File
@@ -162,7 +162,7 @@ Client (WS) → Orchestrator → Lifecycle Manager → ExecutorBackend (containe
Client (WS) ← Orchestrator ← Lifecycle Manager ←──── stream updates (WS) ──────── agent subprocess
```
1. Orchestrator receives `task.start` via WS
1. Orchestrator receives `session.launch` via WS
2. Lifecycle Manager creates executor instance (container or process)
3. agentctl starts inside the instance, agent subprocess is configured and started
4. Agent events stream back via WS through the chain
apps/backend/internal/integration/cancel_after_crash_test.go
+2 -2
View File
@@ -31,7 +31,7 @@ func TestOrchestratorCancelAfterAgentCrash_UnsticksSession(t *testing.T) {
defer client.Close()
// Start the task so a session is created and an execution is registered in the simulator.
startResp, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
startResp, err := client.SendRequest("start-1", ws.ActionSessionLaunch, map[string]interface{}{
"task_id": taskID,
"agent_profile_id": "augment-agent",
})
@@ -110,7 +110,7 @@ func TestOrchestratorCancelWhenAgentHangs_UnsticksSession(t *testing.T) {
client := NewOrchestratorWSClient(t, ts.Server.URL)
defer client.Close()
startResp, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
startResp, err := client.SendRequest("start-1", ws.ActionSessionLaunch, map[string]interface{}{
"task_id": taskID,
"agent_profile_id": "augment-agent",
})
apps/backend/internal/integration/orchestrator_advanced_test.go
+6 -137
View File
@@ -30,21 +30,21 @@ func TestOrchestratorTaskPriority(t *testing.T) {
defer client.Close()
// Start low priority first
_, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
_, err := client.SendRequest("start-1", ws.ActionSessionLaunch, map[string]interface{}{
"task_id": lowPriorityTask,
"agent_profile_id": "augment-agent",
})
require.NoError(t, err)
// Then high priority
_, err = client.SendRequest("start-2", ws.ActionOrchestratorStart, map[string]interface{}{
_, err = client.SendRequest("start-2", ws.ActionSessionLaunch, map[string]interface{}{
"task_id": highPriorityTask,
"agent_profile_id": "augment-agent",
})
require.NoError(t, err)
// Then medium priority
_, err = client.SendRequest("start-3", ws.ActionOrchestratorStart, map[string]interface{}{
_, err = client.SendRequest("start-3", ws.ActionSessionLaunch, map[string]interface{}{
"task_id": medPriorityTask,
"agent_profile_id": "augment-agent",
})
@@ -54,137 +54,6 @@ func TestOrchestratorTaskPriority(t *testing.T) {
assert.Equal(t, 3, ts.AgentManager.GetLaunchCount())
}
func TestOrchestratorTriggerTask(t *testing.T) {
ts := NewOrchestratorTestServer(t)
defer ts.Close()
taskID := ts.CreateTestTask(t, "augment-agent", 2)
client := NewOrchestratorWSClient(t, ts.Server.URL)
defer client.Close()
resp, err := client.SendRequest("trigger-1", ws.ActionOrchestratorTrigger, map[string]interface{}{
"task_id": taskID,
})
require.NoError(t, err)
assert.Equal(t, ws.MessageTypeResponse, resp.Type)
var payload map[string]interface{}
require.NoError(t, resp.ParsePayload(&payload))
assert.True(t, payload["success"].(bool))
assert.Equal(t, taskID, payload["task_id"])
}
func TestOrchestratorEndToEndWorkflow(t *testing.T) {
ts := NewOrchestratorTestServer(t)
defer ts.Close()
client := NewOrchestratorWSClient(t, ts.Server.URL)
defer client.Close()
workspaceID := createOrchestratorWorkspace(t, client)
// 1. Create workflow with workflow template
workflowResp, err := client.SendRequest("workflow-1", ws.ActionWorkflowCreate, map[string]interface{}{
"workspace_id": workspaceID,
"name": "E2E Test Workflow",
"description": "End-to-end test workflow",
"workflow_template_id": "simple",
})
require.NoError(t, err)
var workflowPayload map[string]interface{}
require.NoError(t, workflowResp.ParsePayload(&workflowPayload))
workflowID := workflowPayload["id"].(string)
// 2. Get first workflow step from workflow
stepResp, err := client.SendRequest("step-list-1", ws.ActionWorkflowStepList, map[string]interface{}{
"workflow_id": workflowID,
})
require.NoError(t, err)
var stepListPayload map[string]interface{}
require.NoError(t, stepResp.ParsePayload(&stepListPayload))
steps := stepListPayload["steps"].([]interface{})
require.NotEmpty(t, steps, "workflow should have workflow steps")
workflowStepID := steps[0].(map[string]interface{})["id"].(string)
// 3. Create task with workflow step
repoResp, err := client.SendRequest("repo-1", ws.ActionRepositoryCreate, map[string]interface{}{
"workspace_id": workspaceID,
"name": "Test Repo",
"source_type": "local",
"local_path": createTempRepoDir(t),
})
require.NoError(t, err)
var repoPayload map[string]interface{}
require.NoError(t, repoResp.ParsePayload(&repoPayload))
repositoryID := repoPayload["id"].(string)
taskResp, err := client.SendRequest("task-1", ws.ActionTaskCreate, map[string]interface{}{
"workspace_id": workspaceID,
"workflow_id": workflowID,
"workflow_step_id": workflowStepID,
"title": "Implement feature X",
"description": "Create a new feature with tests",
"priority": 2,
"repository_id": repositoryID,
"base_branch": "main",
})
require.NoError(t, err)
var taskPayload map[string]interface{}
require.NoError(t, taskResp.ParsePayload(&taskPayload))
taskID := taskPayload["id"].(string)
// 4. Subscribe to task
_, err = client.SendRequest("sub-1", ws.ActionTaskSubscribe, map[string]string{"task_id": taskID})
require.NoError(t, err)
// 5. Start orchestrator
startResp, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
"task_id": taskID,
"agent_profile_id": "test-profile-id",
})
require.NoError(t, err)
var startPayload map[string]interface{}
require.NoError(t, startResp.ParsePayload(&startPayload))
assert.True(t, startPayload["success"].(bool))
agentExecutionID := startPayload["agent_execution_id"].(string)
assert.NotEmpty(t, agentExecutionID)
sessionID, _ := startPayload["session_id"].(string)
require.NotEmpty(t, sessionID)
_, err = client.SendRequest("session-sub-1", ws.ActionSessionSubscribe, map[string]string{"session_id": sessionID})
require.NoError(t, err)
// 6. Collect notifications during execution
time.Sleep(400 * time.Millisecond)
notifications := client.CollectNotifications(200 * time.Millisecond)
t.Logf("Received %d notifications", len(notifications))
// 7. Complete the task
completeResp, err := client.SendRequest("complete-1", ws.ActionOrchestratorComplete, map[string]interface{}{
"task_id": taskID,
})
require.NoError(t, err)
var completePayload map[string]interface{}
require.NoError(t, completeResp.ParsePayload(&completePayload))
assert.True(t, completePayload["success"].(bool))
// 8. Verify final task state
getResp, err := client.SendRequest("get-1", ws.ActionTaskGet, map[string]string{"id": taskID})
require.NoError(t, err)
var getPayload map[string]interface{}
require.NoError(t, getResp.ParsePayload(&getPayload))
assert.Equal(t, "COMPLETED", getPayload["state"].(string))
}
// TestOrchestratorAgentMessagePersistence validates that agent messages are stored
// correctly in the database without missing data or ordering issues.
func TestOrchestratorAgentMessagePersistence(t *testing.T) {
@@ -223,7 +92,7 @@ func TestOrchestratorAgentMessagePersistence(t *testing.T) {
require.NoError(t, err)
// Start task execution
startResp, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
startResp, err := client.SendRequest("start-1", ws.ActionSessionLaunch, map[string]interface{}{
"task_id": taskID,
"agent_profile_id": "test-profile-id",
})
@@ -462,7 +331,7 @@ func TestOrchestratorAgentMessagePersistenceWithToolCalls(t *testing.T) {
require.NoError(t, err)
// Start task execution
startResp, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
startResp, err := client.SendRequest("start-1", ws.ActionSessionLaunch, map[string]interface{}{
"task_id": taskID,
"agent_profile_id": "test-profile-id",
})
@@ -630,7 +499,7 @@ func TestOrchestratorAgentMessageChunkStreaming(t *testing.T) {
_, err := client.SendRequest("sub-1", ws.ActionTaskSubscribe, map[string]string{"task_id": taskID})
require.NoError(t, err)
startResp, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
startResp, err := client.SendRequest("start-1", ws.ActionSessionLaunch, map[string]interface{}{
"task_id": taskID,
"agent_profile_id": "test-profile-id",
})
apps/backend/internal/integration/orchestrator_basic_test.go
+3 -3
View File
@@ -69,7 +69,7 @@ func TestOrchestratorStartTask(t *testing.T) {
assert.Equal(t, ws.MessageTypeResponse, subResp.Type)
// Start task execution
resp, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
resp, err := client.SendRequest("start-1", ws.ActionSessionLaunch, map[string]interface{}{
"task_id": taskID,
"agent_profile_id": "test-profile-id",
})
@@ -111,7 +111,7 @@ func TestOrchestratorStartTaskWithAgentTypeOverride(t *testing.T) {
defer client.Close()
// Start with different agent profile
resp, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
resp, err := client.SendRequest("start-1", ws.ActionSessionLaunch, map[string]interface{}{
"task_id": taskID,
"agent_profile_id": "override-profile-id",
"priority": 3,
@@ -134,7 +134,7 @@ func TestOrchestratorStopTask(t *testing.T) {
defer client.Close()
// Start task
startResp, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
startResp, err := client.SendRequest("start-1", ws.ActionSessionLaunch, map[string]interface{}{
"task_id": taskID,
"agent_profile_id": "augment-agent",
})
apps/backend/internal/integration/orchestrator_interaction_test.go
+7 -150
View File
@@ -2,7 +2,6 @@
package integration
import (
"context"
"fmt"
"sync"
"testing"
@@ -11,130 +10,10 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/kandev/kandev/internal/task/models"
"github.com/kandev/kandev/pkg/acp/protocol"
v1 "github.com/kandev/kandev/pkg/api/v1"
ws "github.com/kandev/kandev/pkg/websocket"
)
func TestOrchestratorPromptTask(t *testing.T) {
ts := NewOrchestratorTestServer(t)
defer ts.Close()
taskID := ts.CreateTestTask(t, "augment-agent", 2)
client := NewOrchestratorWSClient(t, ts.Server.URL)
defer client.Close()
// Subscribe to task
_, err := client.SendRequest("sub-1", ws.ActionTaskSubscribe, map[string]string{"task_id": taskID})
require.NoError(t, err)
// Start task
startResp, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
"task_id": taskID,
"agent_profile_id": "augment-agent",
})
require.NoError(t, err)
var startPayload map[string]interface{}
require.NoError(t, startResp.ParsePayload(&startPayload))
require.True(t, startPayload["success"].(bool))
sessionID, _ := startPayload["session_id"].(string)
require.NotEmpty(t, sessionID)
_, err = client.SendRequest("session-sub-1", ws.ActionSessionSubscribe, map[string]string{"session_id": sessionID})
require.NoError(t, err)
// Test 1: Verify that prompting while agent is not yet ready is rejected.
// The simulated agent takes ~310ms (50ms launch + 60ms ACP + 200ms execution) before
// publishing AgentReady, so at 100ms the session is still in CREATED state.
time.Sleep(100 * time.Millisecond)
// Attempt to send prompt while agent is still initializing — should be rejected
promptResp1, err := client.SendRequest("prompt-rejected", ws.ActionOrchestratorPrompt, map[string]interface{}{
"task_id": taskID,
"session_id": sessionID,
"prompt": "This should be rejected",
})
require.NoError(t, err)
// Should receive an error response (type=error with ErrorPayload)
assert.Equal(t, ws.MessageTypeError, promptResp1.Type, "Expected error response when session is not ready")
var errPayload ws.ErrorPayload
require.NoError(t, promptResp1.ParsePayload(&errPayload))
t.Logf("Prompt rejection: code=%s message=%s", errPayload.Code, errPayload.Message)
assert.NotEmpty(t, errPayload.Message)
// Test 2: Wait for agent to complete, then send a valid prompt
// Wait for agent to reach COMPLETED or WAITING_FOR_INPUT state
waitForSessionReady := func(timeout time.Duration) bool {
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
session, err := ts.TaskRepo.GetTaskSession(context.Background(), sessionID)
if err == nil && (session.State == models.TaskSessionStateWaitingForInput || session.State == models.TaskSessionStateCompleted) {
return true
}
time.Sleep(100 * time.Millisecond)
}
return false
}
// Wait up to 5 seconds for session to be ready
if waitForSessionReady(5 * time.Second) {
// Now send a valid prompt
promptResp2, err := client.SendRequest("prompt-valid", ws.ActionOrchestratorPrompt, map[string]interface{}{
"task_id": taskID,
"session_id": sessionID,
"prompt": "Please continue and add error handling",
})
require.NoError(t, err)
var promptPayload map[string]interface{}
require.NoError(t, promptResp2.ParsePayload(&promptPayload))
assert.True(t, promptPayload["success"].(bool), "Prompt should succeed when agent is ready")
}
}
func TestOrchestratorCompleteTask(t *testing.T) {
ts := NewOrchestratorTestServer(t)
defer ts.Close()
taskID := ts.CreateTestTask(t, "augment-agent", 2)
client := NewOrchestratorWSClient(t, ts.Server.URL)
defer client.Close()
// Start task
startResp, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
"task_id": taskID,
"agent_profile_id": "augment-agent",
})
require.NoError(t, err)
var startPayload map[string]interface{}
require.NoError(t, startResp.ParsePayload(&startPayload))
require.True(t, startPayload["success"].(bool))
// Wait for agent to be running
time.Sleep(200 * time.Millisecond)
// Complete task
completeResp, err := client.SendRequest("complete-1", ws.ActionOrchestratorComplete, map[string]interface{}{
"task_id": taskID,
})
require.NoError(t, err)
var completePayload map[string]interface{}
require.NoError(t, completeResp.ParsePayload(&completePayload))
assert.True(t, completePayload["success"].(bool))
// Verify task state is COMPLETED
task, err := ts.TaskRepo.GetTask(context.Background(), taskID)
require.NoError(t, err)
assert.Equal(t, v1.TaskStateCompleted, task.State)
}
func TestOrchestratorConcurrentTasks(t *testing.T) {
ts := NewOrchestratorTestServer(t)
defer ts.Close()
@@ -157,7 +36,7 @@ func TestOrchestratorConcurrentTasks(t *testing.T) {
go func(idx int, tid string) {
defer wg.Done()
resp, err := client.SendRequest(fmt.Sprintf("start-%d", idx), ws.ActionOrchestratorStart, map[string]interface{}{
resp, err := client.SendRequest(fmt.Sprintf("start-%d", idx), ws.ActionSessionLaunch, map[string]interface{}{
"task_id": tid,
"agent_profile_id": "augment-agent",
})
@@ -256,7 +135,7 @@ func TestOrchestratorACPStreaming(t *testing.T) {
require.NoError(t, err)
// Start task
startResp, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
startResp, err := client.SendRequest("start-1", ws.ActionSessionLaunch, map[string]interface{}{
"task_id": taskID,
"agent_profile_id": "augment-agent",
})
@@ -302,7 +181,7 @@ func TestOrchestratorMultipleClients(t *testing.T) {
require.NoError(t, err)
// Client 1 starts the task
startResp, err := client1.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
startResp, err := client1.SendRequest("start-1", ws.ActionSessionLaunch, map[string]interface{}{
"task_id": taskID,
"agent_profile_id": "augment-agent",
})
@@ -339,7 +218,7 @@ func TestOrchestratorErrorHandling(t *testing.T) {
client := NewOrchestratorWSClient(t, ts.Server.URL)
defer client.Close()
resp, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{})
resp, err := client.SendRequest("start-1", ws.ActionSessionLaunch, map[string]interface{}{})
require.NoError(t, err)
assert.Equal(t, ws.MessageTypeError, resp.Type)
@@ -357,7 +236,7 @@ func TestOrchestratorErrorHandling(t *testing.T) {
client := NewOrchestratorWSClient(t, ts.Server.URL)
defer client.Close()
resp, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
resp, err := client.SendRequest("start-1", ws.ActionSessionLaunch, map[string]interface{}{
"task_id": "non-existent-task",
"agent_profile_id": "augment-agent",
})
@@ -366,28 +245,6 @@ func TestOrchestratorErrorHandling(t *testing.T) {
assert.Equal(t, ws.MessageTypeError, resp.Type)
})
t.Run("PromptTaskMissingPrompt", func(t *testing.T) {
ts := NewOrchestratorTestServer(t)
defer ts.Close()
taskID := ts.CreateTestTask(t, "augment-agent", 2)
client := NewOrchestratorWSClient(t, ts.Server.URL)
defer client.Close()
resp, err := client.SendRequest("prompt-1", ws.ActionOrchestratorPrompt, map[string]interface{}{
"task_id": taskID,
"session_id": "session-1",
})
require.NoError(t, err)
assert.Equal(t, ws.MessageTypeError, resp.Type)
var errPayload ws.ErrorPayload
require.NoError(t, resp.ParsePayload(&errPayload))
assert.Contains(t, errPayload.Message, "prompt")
})
t.Run("StopTaskNotRunning", func(t *testing.T) {
ts := NewOrchestratorTestServer(t)
defer ts.Close()
@@ -420,7 +277,7 @@ func TestOrchestratorAgentFailure(t *testing.T) {
client := NewOrchestratorWSClient(t, ts.Server.URL)
defer client.Close()
resp, err := client.SendRequest("start-1", ws.ActionOrchestratorStart, map[string]interface{}{
resp, err := client.SendRequest("start-1", ws.ActionSessionLaunch, map[string]interface{}{
"task_id": taskID,
"agent_profile_id": "augment-agent",
})
@@ -431,5 +288,5 @@ func TestOrchestratorAgentFailure(t *testing.T) {
var errPayload ws.ErrorPayload
require.NoError(t, resp.ParsePayload(&errPayload))
assert.Contains(t, errPayload.Message, "Failed to start task")
assert.Contains(t, errPayload.Message, "Failed to launch session")
}
apps/backend/internal/orchestrator/dto/dto.go
-100
View File
@@ -27,65 +27,6 @@ type QueueResponse struct {
Total int `json:"total"`
}
// TriggerTaskRequest is the payload for orchestrator.trigger
type TriggerTaskRequest struct {
TaskID string `json:"task_id"`
}
// TriggerTaskResponse is the response for orchestrator.trigger
type TriggerTaskResponse struct {
Success bool `json:"success"`
Message string `json:"message"`
TaskID string `json:"task_id"`
}
// StartTaskRequest is the payload for orchestrator.start
// If SessionID is provided, the existing session is resumed and prompted with workflow step config.
// If SessionID is not provided, a new session is created (requires AgentProfileID).
type StartTaskRequest struct {
TaskID string `json:"task_id"`
SessionID string `json:"session_id,omitempty"` // If provided, resume this existing session instead of creating new
AgentProfileID string `json:"agent_profile_id,omitempty"` // Required when creating new session (SessionID not provided)
ExecutorID string `json:"executor_id,omitempty"` // Executor to use (determines runtime: local, worktree, local_docker, etc.)
Priority int `json:"priority,omitempty"`
Prompt string `json:"prompt,omitempty"` // Initial prompt to send to the agent
WorkflowStepID string `json:"workflow_step_id,omitempty"` // If provided, prompt prefix/suffix and plan mode are applied from the step
}
// StartTaskResponse is the response for orchestrator.start
type StartTaskResponse struct {
Success bool `json:"success"`
TaskID string `json:"task_id"`
AgentExecutionID string `json:"agent_execution_id"`
TaskSessionID string `json:"session_id,omitempty"`
State string `json:"state"`
WorktreePath *string `json:"worktree_path,omitempty"`
WorktreeBranch *string `json:"worktree_branch,omitempty"`
}
// ResumeTaskSessionRequest is the payload for task.session.resume
type ResumeTaskSessionRequest struct {
TaskID string `json:"task_id"`
TaskSessionID string `json:"session_id"`
}
// ResumeTaskSessionResponse is the response for task.session.resume
type ResumeTaskSessionResponse struct {
Success bool `json:"success"`
TaskID string `json:"task_id"`
AgentExecutionID string `json:"agent_execution_id"`
TaskSessionID string `json:"session_id,omitempty"`
State string `json:"state"`
WorktreePath *string `json:"worktree_path,omitempty"`
WorktreeBranch *string `json:"worktree_branch,omitempty"`
}
// TaskSessionStatusRequest is the payload for task.session.status
type TaskSessionStatusRequest struct {
TaskID string `json:"task_id"`
TaskSessionID string `json:"session_id"`
}
// TaskSessionStatusResponse is the response for task.session.status
type TaskSessionStatusResponse struct {
// Session metadata
@@ -124,52 +65,11 @@ type TaskSessionStatusResponse struct {
Error string `json:"error,omitempty"`
}
// StopTaskRequest is the payload for orchestrator.stop
type StopTaskRequest struct {
TaskID string `json:"task_id"`
Reason string `json:"reason,omitempty"`
Force bool `json:"force,omitempty"`
}
// SuccessResponse is a generic success response
type SuccessResponse struct {
Success bool `json:"success"`
}
// PromptTaskRequest is the payload for orchestrator.prompt
type PromptTaskRequest struct {
TaskID string `json:"task_id"`
TaskSessionID string `json:"session_id"`
Prompt string `json:"prompt"`
Model string `json:"model,omitempty"` // Optional: switch to this model before processing prompt
PlanMode bool `json:"plan_mode,omitempty"` // Optional: enable plan mode (analyze before executing)
}
// PromptTaskResponse is the response for orchestrator.prompt
type PromptTaskResponse struct {
Success bool `json:"success"`
StopReason string `json:"stop_reason"`
}
// CompleteTaskRequest is the payload for orchestrator.complete
type CompleteTaskRequest struct {
TaskID string `json:"task_id"`
}
// CompleteTaskResponse is the response for orchestrator.complete
type CompleteTaskResponse struct {
Success bool `json:"success"`
Message string `json:"message"`
}
// PermissionRespondRequest is the payload for permission.respond
type PermissionRespondRequest struct {
SessionID string `json:"session_id"`
PendingID string `json:"pending_id"`
OptionID string `json:"option_id,omitempty"`
Cancelled bool `json:"cancelled,omitempty"`
}
// PermissionRespondResponse is the response for permission.respond
type PermissionRespondResponse struct {
Success bool `json:"success"`
apps/backend/internal/orchestrator/handlers/handlers.go
-220
View File
@@ -30,15 +30,9 @@ func NewHandlers(svc *orchestrator.Service, log *logger.Logger) *Handlers {
func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) {
d.RegisterFunc(ws.ActionOrchestratorStatus, h.wsGetStatus)
d.RegisterFunc(ws.ActionOrchestratorQueue, h.wsGetQueue)
d.RegisterFunc(ws.ActionOrchestratorTrigger, h.wsTriggerTask)
d.RegisterFunc(ws.ActionOrchestratorStart, h.wsStartTask)
d.RegisterFunc(ws.ActionOrchestratorStop, h.wsStopTask)
d.RegisterFunc(ws.ActionOrchestratorPrompt, h.wsPromptTask)
d.RegisterFunc(ws.ActionOrchestratorComplete, h.wsCompleteTask)
d.RegisterFunc(ws.ActionPermissionRespond, h.wsRespondToPermission)
d.RegisterFunc(ws.ActionTaskSessionResume, h.wsResumeTaskSession)
d.RegisterFunc(ws.ActionTaskSessionStatus, h.wsGetTaskSessionStatus)
d.RegisterFunc(ws.ActionTaskSessionPrepare, h.wsPrepareTaskSession)
d.RegisterFunc(ws.ActionAgentCancel, h.wsCancelAgent)
d.RegisterFunc(ws.ActionSessionLaunch, h.wsLaunchSession)
d.RegisterFunc(ws.ActionSessionEnsure, h.wsEnsureSession)
@@ -79,28 +73,6 @@ func (h *Handlers) wsGetQueue(ctx context.Context, msg *ws.Message) (*ws.Message
return ws.NewResponse(msg.ID, msg.Action, resp)
}
type wsTriggerTaskRequest struct {
TaskID string `json:"task_id"`
}
func (h *Handlers) wsTriggerTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
var req wsTriggerTaskRequest
if err := msg.ParsePayload(&req); err != nil {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
}
if req.TaskID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", 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)
}
func (h *Handlers) wsLaunchSession(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
var req orchestrator.LaunchSessionRequest
if err := msg.ParsePayload(&req); err != nil {
@@ -208,140 +180,6 @@ func (h *Handlers) wsRecoverSession(ctx context.Context, msg *ws.Message) (*ws.M
return ws.NewResponse(msg.ID, msg.Action, resp)
}
type wsStartTaskRequest struct {
TaskID string `json:"task_id"`
SessionID string `json:"session_id,omitempty"`
AgentProfileID string `json:"agent_profile_id,omitempty"`
ExecutorID string `json:"executor_id,omitempty"`
ExecutorProfileID string `json:"executor_profile_id,omitempty"`
Priority int `json:"priority,omitempty"`
Prompt string `json:"prompt,omitempty"`
WorkflowStepID string `json:"workflow_step_id,omitempty"`
PlanMode bool `json:"plan_mode,omitempty"`
}
// wsStartTask delegates to LaunchSession for backward compatibility.
func (h *Handlers) wsStartTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
var req wsStartTaskRequest
if err := msg.ParsePayload(&req); err != nil {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
}
if req.TaskID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil)
}
if req.SessionID == "" && req.AgentProfileID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "agent_profile_id is required when creating a new session", nil)
}
launchReq := &orchestrator.LaunchSessionRequest{
TaskID: req.TaskID,
SessionID: req.SessionID,
AgentProfileID: req.AgentProfileID,
ExecutorID: req.ExecutorID,
ExecutorProfileID: req.ExecutorProfileID,
Priority: req.Priority,
Prompt: req.Prompt,
WorkflowStepID: req.WorkflowStepID,
PlanMode: req.PlanMode,
}
resp, err := h.service.LaunchSession(ctx, launchReq)
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)
}
// Map to legacy StartTaskResponse format for backward compatibility
legacyResp := dto.StartTaskResponse{
Success: resp.Success,
TaskID: resp.TaskID,
AgentExecutionID: resp.AgentExecutionID,
TaskSessionID: resp.SessionID,
State: resp.State,
WorktreePath: resp.WorktreePath,
WorktreeBranch: resp.WorktreeBranch,
}
return ws.NewResponse(msg.ID, msg.Action, legacyResp)
}
type wsResumeTaskSessionRequest struct {
TaskID string `json:"task_id"`
TaskSessionID string `json:"session_id"`
}
// wsResumeTaskSession delegates to LaunchSession with resume intent for backward compatibility.
func (h *Handlers) wsResumeTaskSession(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
var req wsResumeTaskSessionRequest
if err := msg.ParsePayload(&req); err != nil {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
}
if req.TaskID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil)
}
if req.TaskSessionID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session_id is required", nil)
}
launchReq := &orchestrator.LaunchSessionRequest{
TaskID: req.TaskID,
SessionID: req.TaskSessionID,
Intent: orchestrator.IntentResume,
}
resp, err := h.service.LaunchSession(ctx, launchReq)
if err != nil {
h.logger.Error("failed to resume task session",
zap.String("task_id", req.TaskID),
zap.String("session_id", req.TaskSessionID),
zap.Error(err))
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to resume task session: "+err.Error(), nil)
}
// Map to legacy ResumeTaskSessionResponse format
legacyResp := dto.ResumeTaskSessionResponse{
Success: resp.Success,
TaskID: resp.TaskID,
AgentExecutionID: resp.AgentExecutionID,
TaskSessionID: resp.SessionID,
State: resp.State,
WorktreePath: resp.WorktreePath,
WorktreeBranch: resp.WorktreeBranch,
}
return ws.NewResponse(msg.ID, msg.Action, legacyResp)
}
type wsPrepareTaskSessionRequest struct {
TaskID string `json:"task_id"`
}
type wsPrepareTaskSessionResponse struct {
SessionID string `json:"session_id"`
}
// wsPrepareTaskSession delegates to LaunchSession with prepare intent for backward compatibility.
func (h *Handlers) wsPrepareTaskSession(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
var req wsPrepareTaskSessionRequest
if err := msg.ParsePayload(&req); err != nil {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
}
if req.TaskID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil)
}
launchReq := &orchestrator.LaunchSessionRequest{
TaskID: req.TaskID,
Intent: orchestrator.IntentPrepare,
LaunchWorkspace: true,
}
resp, err := h.service.LaunchSession(ctx, launchReq)
if err != nil {
h.logger.Error("failed to prepare task session",
zap.String("task_id", req.TaskID),
zap.Error(err))
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to prepare session: "+err.Error(), nil)
}
return ws.NewResponse(msg.ID, msg.Action, wsPrepareTaskSessionResponse{SessionID: resp.SessionID})
}
type wsStopTaskRequest struct {
TaskID string `json:"task_id"`
Reason string `json:"reason,omitempty"`
@@ -368,64 +206,6 @@ func (h *Handlers) wsStopTask(ctx context.Context, msg *ws.Message) (*ws.Message
return ws.NewResponse(msg.ID, msg.Action, dto.SuccessResponse{Success: true})
}
type wsPromptTaskRequest struct {
TaskID string `json:"task_id"`
TaskSessionID string `json:"session_id"`
Prompt string `json:"prompt"`
Model string `json:"model,omitempty"`
}
func (h *Handlers) wsPromptTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
var req wsPromptTaskRequest
if err := msg.ParsePayload(&req); err != nil {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
}
if req.TaskID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil)
}
if req.TaskSessionID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session_id is required", nil)
}
if req.Prompt == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "prompt is required", nil)
}
result, err := h.service.PromptTask(ctx, req.TaskID, req.TaskSessionID, req.Prompt, req.Model, false, nil, 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)
}
type wsCompleteTaskRequest struct {
TaskID string `json:"task_id"`
}
func (h *Handlers) wsCompleteTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
var req wsCompleteTaskRequest
if err := msg.ParsePayload(&req); err != nil {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
}
if req.TaskID == "" {
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", 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)
}
type wsPermissionRespondRequest struct {
SessionID string `json:"session_id"`
PendingID string `json:"pending_id"`
apps/backend/pkg/websocket/actions.go
+5 -13
View File
@@ -73,10 +73,8 @@ const (
ActionTaskPlanRevisionGet = "task.plan.revision.get"
ActionTaskPlanRevert = "task.plan.revert"
ActionTaskSessionList = "task.session.list"
ActionTaskSessionResume = "task.session.resume"
ActionTaskSessionStatus = "task.session.status"
ActionTaskSessionPrepare = "task.session.prepare"
ActionTaskSessionList = "task.session.list"
ActionTaskSessionStatus = "task.session.status"
// Unified session launch
ActionSessionLaunch = "session.launch"
@@ -104,15 +102,9 @@ const (
ActionAgentResize = "agent.resize" // Resize agent PTY (passthrough mode)
// Orchestrator actions
ActionOrchestratorStatus = "orchestrator.status"
ActionOrchestratorQueue = "orchestrator.queue"
ActionOrchestratorTrigger = "orchestrator.trigger"
ActionOrchestratorStart = "orchestrator.start"
ActionOrchestratorStop = "orchestrator.stop"
ActionOrchestratorPause = "orchestrator.pause"
ActionOrchestratorResume = "orchestrator.resume"
ActionOrchestratorPrompt = "orchestrator.prompt"
ActionOrchestratorComplete = "orchestrator.complete"
ActionOrchestratorStatus = "orchestrator.status"
ActionOrchestratorQueue = "orchestrator.queue"
ActionOrchestratorStop = "orchestrator.stop"
// Message Queue actions
ActionMessageQueueAdd = "message.queue.add"