package orch import ( "database/sql" "os" "path/filepath" "testing" "time" ) func TestOrchRunDispatchReconcileLifecycle(t *testing.T) { t.Parallel() dbPath := filepath.Join(t.TempDir(), "coord.db") runOrchCommand( t, "--db", dbPath, "--json", "run", "init", "--run", "run_blog_001", "--goal", "Build blog MVP", "--summary", "Public blog plus admin CRUD", ) taskOut := runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_001", "--task", "T1", "--title", "Implement retry policy", "--summary", "Add retry policy to HTTP client", "--default-to", "worker-a", ) var taskResp map[string]any mustDecodeJSON(t, taskOut, &taskResp) if got := nestedString(t, taskResp, "data", "task", "status"); got != "ready" { t.Fatalf("expected new task to become ready, got %q", got) } readyOut := runOrchCommand( t, "--db", dbPath, "--json", "ready", "--run", "run_blog_001", ) var readyResp map[string]any mustDecodeJSON(t, readyOut, &readyResp) readyTasks := nestedArray(t, readyResp, "data", "tasks") if len(readyTasks) != 1 { t.Fatalf("expected one ready task, got %#v", readyTasks) } readyTask, ok := readyTasks[0].(map[string]any) if !ok { t.Fatalf("expected ready task object, got %#v", readyTasks[0]) } if taskID, _ := readyTask["task_id"].(string); taskID != "T1" { t.Fatalf("expected ready task T1, got %#v", readyTask["task_id"]) } dispatchOut := runOrchCommand( t, "--db", dbPath, "--json", "dispatch", "--run", "run_blog_001", "--task", "T1", "--body", "Implement retry handling for the HTTP client.", ) var dispatchResp map[string]any mustDecodeJSON(t, dispatchOut, &dispatchResp) if got := nestedString(t, dispatchResp, "data", "task", "status"); got != "dispatched" { t.Fatalf("expected dispatched task, got %q", got) } threadID := nestedString(t, dispatchResp, "data", "attempt", "thread_id") runInboxCommand( t, "--db", dbPath, "--json", "claim", "--agent", "worker-a", "--thread", threadID, ) runInboxCommand( t, "--db", dbPath, "--json", "update", "--agent", "worker-a", "--thread", threadID, "--status", "in_progress", "--summary", "Implementation started", ) reconcileOut := runOrchCommand( t, "--db", dbPath, "--json", "reconcile", "--run", "run_blog_001", ) var reconcileResp map[string]any mustDecodeJSON(t, reconcileOut, &reconcileResp) updatedTasks := nestedArray(t, reconcileResp, "data", "updated_tasks") if len(updatedTasks) != 1 { t.Fatalf("expected one updated task after running reconcile, got %#v", updatedTasks) } runningTask, ok := updatedTasks[0].(map[string]any) if !ok { t.Fatalf("expected updated task object, got %#v", updatedTasks[0]) } if status, _ := runningTask["status"].(string); status != "running" { t.Fatalf("expected running task after reconcile, got %#v", runningTask["status"]) } runInboxCommand( t, "--db", dbPath, "--json", "done", "--agent", "worker-a", "--thread", threadID, "--summary", "Retry policy implemented", "--body", "The HTTP client now retries transient failures.", ) reconcileDoneOut := runOrchCommand( t, "--db", dbPath, "--json", "reconcile", "--run", "run_blog_001", ) var reconcileDoneResp map[string]any mustDecodeJSON(t, reconcileDoneOut, &reconcileDoneResp) updatedTasks = nestedArray(t, reconcileDoneResp, "data", "updated_tasks") if len(updatedTasks) != 1 { t.Fatalf("expected one updated task after done reconcile, got %#v", updatedTasks) } doneTask, ok := updatedTasks[0].(map[string]any) if !ok { t.Fatalf("expected updated task object, got %#v", updatedTasks[0]) } if status, _ := doneTask["status"].(string); status != "done" { t.Fatalf("expected done task after reconcile, got %#v", doneTask["status"]) } statusOut := runOrchCommand( t, "--db", dbPath, "--json", "status", "--run", "run_blog_001", ) var statusResp map[string]any mustDecodeJSON(t, statusOut, &statusResp) if got := nestedString(t, statusResp, "data", "run", "status"); got != "done" { t.Fatalf("expected run status done, got %q", got) } tasks := nestedArray(t, statusResp, "data", "tasks") if len(tasks) != 1 { t.Fatalf("expected one task in status response, got %#v", tasks) } task, ok := tasks[0].(map[string]any) if !ok { t.Fatalf("expected task object, got %#v", tasks[0]) } if got, _ := task["status"].(string); got != "done" { t.Fatalf("expected status task done, got %#v", task["status"]) } } func TestOrchDependencyBlockedAndAnswerFlow(t *testing.T) { t.Parallel() dbPath := filepath.Join(t.TempDir(), "coord.db") runOrchCommand( t, "--db", dbPath, "--json", "run", "init", "--run", "run_blog_002", "--goal", "Build dependency-aware workflow", ) runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_002", "--task", "T1", "--title", "Build backend", "--summary", "Implement backend APIs", "--default-to", "worker-a", ) runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_002", "--task", "T2", "--title", "Build frontend", "--summary", "Implement frontend flows", "--default-to", "worker-b", ) runOrchCommand( t, "--db", dbPath, "--json", "dep", "add", "--run", "run_blog_002", "--task", "T2", "--depends-on", "T1", ) readyOut := runOrchCommand( t, "--db", dbPath, "--json", "ready", "--run", "run_blog_002", ) var readyResp map[string]any mustDecodeJSON(t, readyOut, &readyResp) readyTasks := nestedArray(t, readyResp, "data", "tasks") if len(readyTasks) != 1 { t.Fatalf("expected only dependency-free task ready, got %#v", readyTasks) } readyTask, ok := readyTasks[0].(map[string]any) if !ok { t.Fatalf("expected ready task object, got %#v", readyTasks[0]) } if taskID, _ := readyTask["task_id"].(string); taskID != "T1" { t.Fatalf("expected T1 ready before dependency clears, got %#v", readyTask["task_id"]) } dispatchBackendOut := runOrchCommand( t, "--db", dbPath, "--json", "dispatch", "--run", "run_blog_002", "--task", "T1", ) var dispatchBackendResp map[string]any mustDecodeJSON(t, dispatchBackendOut, &dispatchBackendResp) threadBackend := nestedString(t, dispatchBackendResp, "data", "attempt", "thread_id") runInboxCommand( t, "--db", dbPath, "--json", "claim", "--agent", "worker-a", "--thread", threadBackend, ) runInboxCommand( t, "--db", dbPath, "--json", "done", "--agent", "worker-a", "--thread", threadBackend, "--summary", "Backend complete", ) runOrchCommand( t, "--db", dbPath, "--json", "reconcile", "--run", "run_blog_002", ) readyAfterDoneOut := runOrchCommand( t, "--db", dbPath, "--json", "ready", "--run", "run_blog_002", ) var readyAfterDoneResp map[string]any mustDecodeJSON(t, readyAfterDoneOut, &readyAfterDoneResp) readyTasks = nestedArray(t, readyAfterDoneResp, "data", "tasks") if len(readyTasks) != 1 { t.Fatalf("expected dependent task to become ready, got %#v", readyTasks) } readyTask, ok = readyTasks[0].(map[string]any) if !ok { t.Fatalf("expected ready task object, got %#v", readyTasks[0]) } if taskID, _ := readyTask["task_id"].(string); taskID != "T2" { t.Fatalf("expected T2 ready after T1 completion, got %#v", readyTask["task_id"]) } dispatchFrontendOut := runOrchCommand( t, "--db", dbPath, "--json", "dispatch", "--run", "run_blog_002", "--task", "T2", ) var dispatchFrontendResp map[string]any mustDecodeJSON(t, dispatchFrontendOut, &dispatchFrontendResp) threadFrontend := nestedString(t, dispatchFrontendResp, "data", "attempt", "thread_id") runInboxCommand( t, "--db", dbPath, "--json", "claim", "--agent", "worker-b", "--thread", threadFrontend, ) runInboxCommand( t, "--db", dbPath, "--json", "update", "--agent", "worker-b", "--thread", threadFrontend, "--status", "blocked", "--summary", "Need logging decision", "--payload-json", `{"question":"stdout or stderr?"}`, ) runOrchCommand( t, "--db", dbPath, "--json", "reconcile", "--run", "run_blog_002", ) blockedOut := runOrchCommand( t, "--db", dbPath, "--json", "blocked", "--run", "run_blog_002", ) var blockedResp map[string]any mustDecodeJSON(t, blockedOut, &blockedResp) blockedTasks := nestedArray(t, blockedResp, "data", "blocked") if len(blockedTasks) != 1 { t.Fatalf("expected one blocked task, got %#v", blockedTasks) } blockedTask, ok := blockedTasks[0].(map[string]any) if !ok { t.Fatalf("expected blocked task object, got %#v", blockedTasks[0]) } question, ok := blockedTask["question"].(map[string]any) if !ok { t.Fatalf("expected blocked question object, got %#v", blockedTask["question"]) } if kind, _ := question["kind"].(string); kind != "question" { t.Fatalf("expected blocked question kind, got %#v", question["kind"]) } answerOut := runOrchCommand( t, "--db", dbPath, "--json", "answer", "--run", "run_blog_002", "--task", "T2", "--body", "Use stdout for MVP.", ) var answerResp map[string]any mustDecodeJSON(t, answerOut, &answerResp) if got := nestedString(t, answerResp, "data", "message", "kind"); got != "answer" { t.Fatalf("expected answer message kind, got %q", got) } showOut := runInboxCommand( t, "--db", dbPath, "--json", "show", "--thread", threadFrontend, ) var showResp map[string]any mustDecodeJSON(t, showOut, &showResp) messages := nestedArray(t, showResp, "data", "messages") if len(messages) < 4 { t.Fatalf("expected answer to append a message, got %#v", messages) } lastMessage, ok := messages[len(messages)-1].(map[string]any) if !ok { t.Fatalf("expected last message object, got %#v", messages[len(messages)-1]) } if kind, _ := lastMessage["kind"].(string); kind != "answer" { t.Fatalf("expected latest message to be answer, got %#v", lastMessage["kind"]) } runInboxCommand( t, "--db", dbPath, "--json", "update", "--agent", "worker-b", "--thread", threadFrontend, "--status", "in_progress", "--summary", "Decision applied", ) runInboxCommand( t, "--db", dbPath, "--json", "done", "--agent", "worker-b", "--thread", threadFrontend, "--summary", "Frontend complete", ) runOrchCommand( t, "--db", dbPath, "--json", "reconcile", "--run", "run_blog_002", ) statusOut := runOrchCommand( t, "--db", dbPath, "--json", "status", "--run", "run_blog_002", ) var statusResp map[string]any mustDecodeJSON(t, statusOut, &statusResp) if got := nestedString(t, statusResp, "data", "run", "status"); got != "done" { t.Fatalf("expected run status done after both tasks, got %q", got) } } func TestOrchDispatchRejectsNonReadyTask(t *testing.T) { t.Parallel() dbPath := filepath.Join(t.TempDir(), "coord.db") runOrchCommand( t, "--db", dbPath, "--json", "run", "init", "--run", "run_blog_003", "--goal", "Validate ready gating", ) runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_003", "--task", "T1", "--title", "Backend", ) runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_003", "--task", "T2", "--title", "Frontend", ) runOrchCommand( t, "--db", dbPath, "--json", "dep", "add", "--run", "run_blog_003", "--task", "T2", "--depends-on", "T1", ) stdout, _, exitCode := executeOrchCommand( "--db", dbPath, "--json", "dispatch", "--run", "run_blog_003", "--task", "T2", ) if exitCode != 30 { t.Fatalf("expected invalid_state exit code 30, got %d\nstdout:\n%s", exitCode, stdout) } assertErrorJSON(t, stdout, "invalid_state") } func TestOrchDispatchCreatesStrictWorktree(t *testing.T) { t.Parallel() dbPath := filepath.Join(t.TempDir(), "coord.db") repoPath := initGitRepo(t) runOrchCommand( t, "--db", dbPath, "--json", "run", "init", "--run", "run_blog_worktree_001", "--goal", "Validate strict worktree dispatch", ) runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_worktree_001", "--task", "T1", "--title", "Implement backend", "--default-to", "worker-a", ) dispatchOut := runOrchCommand( t, "--db", dbPath, "--json", "dispatch", "--run", "run_blog_worktree_001", "--task", "T1", "--repo-path", repoPath, "--workspace-root", ".orch/worktrees", "--strict-worktree", "--body", "Implement inside isolated worktree.", ) var dispatchResp map[string]any mustDecodeJSON(t, dispatchOut, &dispatchResp) attempt, ok := nestedValue(t, dispatchResp, "data", "attempt").(map[string]any) if !ok { t.Fatalf("expected attempt object, got %#v", nestedValue(t, dispatchResp, "data", "attempt")) } if got, _ := attempt["base_ref"].(string); got != "HEAD" { t.Fatalf("expected base_ref HEAD, got %#v", attempt["base_ref"]) } expectedCommit := gitHeadCommit(t, repoPath) if got, _ := attempt["base_commit"].(string); got != expectedCommit { t.Fatalf("expected base_commit %q, got %#v", expectedCommit, attempt["base_commit"]) } if got, _ := attempt["branch_name"].(string); got != "orch/run-blog-worktree-001/T1/attempt-1" { t.Fatalf("unexpected branch name %#v", attempt["branch_name"]) } worktreePath, _ := attempt["worktree_path"].(string) if worktreePath == "" { t.Fatalf("expected worktree_path, got %#v", attempt["worktree_path"]) } if got, _ := attempt["workspace_status"].(string); got != "created" { t.Fatalf("expected workspace_status created, got %#v", attempt["workspace_status"]) } if _, err := os.Stat(worktreePath); err != nil { t.Fatalf("stat worktree path %s: %v", worktreePath, err) } if _, err := os.Stat(filepath.Join(worktreePath, "README.md")); err != nil { t.Fatalf("expected README.md in worktree: %v", err) } message, ok := nestedValue(t, dispatchResp, "data", "message").(map[string]any) if !ok { t.Fatalf("expected message object, got %#v", nestedValue(t, dispatchResp, "data", "message")) } payload, ok := message["payload_json"].(map[string]any) if !ok { t.Fatalf("expected payload_json object, got %#v", message["payload_json"]) } if got, _ := payload["worktree_path"].(string); got != worktreePath { t.Fatalf("expected payload worktree path %q, got %#v", worktreePath, payload["worktree_path"]) } threadID := nestedString(t, dispatchResp, "data", "attempt", "thread_id") runInboxCommand( t, "--db", dbPath, "--json", "claim", "--agent", "worker-a", "--thread", threadID, ) runInboxCommand( t, "--db", dbPath, "--json", "update", "--agent", "worker-a", "--thread", threadID, "--status", "in_progress", "--summary", "Started inside worktree", ) reconcileOut := runOrchCommand( t, "--db", dbPath, "--json", "reconcile", "--run", "run_blog_worktree_001", ) var reconcileResp map[string]any mustDecodeJSON(t, reconcileOut, &reconcileResp) updatedTasks := nestedArray(t, reconcileResp, "data", "updated_tasks") if len(updatedTasks) != 1 { t.Fatalf("expected one updated task after worktree reconcile, got %#v", updatedTasks) } } func TestOrchStrictWorktreeRejectsDirtyRepoWithoutBaseRef(t *testing.T) { t.Parallel() dbPath := filepath.Join(t.TempDir(), "coord.db") repoPath := initGitRepo(t) if err := os.WriteFile(filepath.Join(repoPath, "dirty.txt"), []byte("dirty\n"), 0o644); err != nil { t.Fatalf("write dirty file: %v", err) } runOrchCommand( t, "--db", dbPath, "--json", "run", "init", "--run", "run_blog_worktree_002", "--goal", "Validate dirty repo rejection", ) runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_worktree_002", "--task", "T1", "--title", "Implement backend", "--default-to", "worker-a", ) stdout, _, exitCode := executeOrchCommand( "--db", dbPath, "--json", "dispatch", "--run", "run_blog_worktree_002", "--task", "T1", "--repo-path", repoPath, "--workspace-root", ".orch/worktrees", "--strict-worktree", ) if exitCode != 30 { t.Fatalf("expected invalid_state exit code 30, got %d\nstdout:\n%s", exitCode, stdout) } assertErrorJSON(t, stdout, "invalid_state") if _, err := os.Stat(filepath.Join(repoPath, ".orch", "worktrees", "run_blog_worktree_002", "T1", "attempt-1")); !os.IsNotExist(err) { t.Fatalf("expected no worktree directory on strict failure, got err=%v", err) } } func TestOrchStrictWorktreeAllowsExplicitBaseRefOnDirtyRepo(t *testing.T) { t.Parallel() dbPath := filepath.Join(t.TempDir(), "coord.db") repoPath := initGitRepo(t) baseCommit := gitHeadCommit(t, repoPath) if err := os.WriteFile(filepath.Join(repoPath, "dirty.txt"), []byte("dirty\n"), 0o644); err != nil { t.Fatalf("write dirty file: %v", err) } runOrchCommand( t, "--db", dbPath, "--json", "run", "init", "--run", "run_blog_worktree_003", "--goal", "Validate explicit base ref on dirty repo", ) runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_worktree_003", "--task", "T1", "--title", "Implement backend", "--default-to", "worker-a", ) dispatchOut := runOrchCommand( t, "--db", dbPath, "--json", "dispatch", "--run", "run_blog_worktree_003", "--task", "T1", "--repo-path", repoPath, "--workspace-root", ".orch/worktrees", "--strict-worktree", "--base-ref", "HEAD", ) var dispatchResp map[string]any mustDecodeJSON(t, dispatchOut, &dispatchResp) if got := nestedString(t, dispatchResp, "data", "attempt", "base_ref"); got != "HEAD" { t.Fatalf("expected explicit base_ref HEAD, got %q", got) } if got := nestedString(t, dispatchResp, "data", "attempt", "base_commit"); got != baseCommit { t.Fatalf("expected base_commit %q, got %q", baseCommit, got) } } func TestOrchDispatchAutoEnablesWorktreeForCodeLikeTask(t *testing.T) { dbPath := filepath.Join(t.TempDir(), "coord.db") repoPath := initGitRepo(t) runOrchCommand( t, "--db", dbPath, "--json", "run", "init", "--run", "run_blog_auto_worktree_001", "--goal", "Validate auto worktree detection", ) runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_auto_worktree_001", "--task", "T1", "--title", "Implement backend API", "--default-to", "backend-worker", ) dispatchOut := runOrchCommand( t, "--db", dbPath, "--json", "dispatch", "--run", "run_blog_auto_worktree_001", "--task", "T1", "--repo-path", repoPath, ) var dispatchResp map[string]any mustDecodeJSON(t, dispatchOut, &dispatchResp) attempt := nestedValue(t, dispatchResp, "data", "attempt").(map[string]any) worktreePath, _ := attempt["worktree_path"].(string) if worktreePath == "" { t.Fatalf("expected auto-detected code task to allocate a worktree, got %#v", attempt) } if got, _ := attempt["workspace_status"].(string); got != "created" { t.Fatalf("expected created workspace status, got %#v", attempt["workspace_status"]) } if _, err := os.Stat(worktreePath); err != nil { t.Fatalf("stat auto worktree path %s: %v", worktreePath, err) } } func TestOrchDispatchDoesNotAutoEnableWorktreeForNonCodeTask(t *testing.T) { t.Parallel() dbPath := filepath.Join(t.TempDir(), "coord.db") repoPath := initGitRepo(t) runOrchCommand( t, "--db", dbPath, "--json", "run", "init", "--run", "run_blog_auto_worktree_002", "--goal", "Validate non-code dispatch fallback", ) runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_auto_worktree_002", "--task", "T1", "--title", "Review QA findings", "--summary", "Summarize test failures and next steps", "--default-to", "qa-worker", ) dispatchOut := runOrchCommand( t, "--db", dbPath, "--json", "dispatch", "--run", "run_blog_auto_worktree_002", "--task", "T1", "--repo-path", repoPath, ) var dispatchResp map[string]any mustDecodeJSON(t, dispatchOut, &dispatchResp) attempt := nestedValue(t, dispatchResp, "data", "attempt").(map[string]any) if got, _ := attempt["worktree_path"].(string); got != "" { t.Fatalf("expected non-code task to stay on non-worktree path, got %#v", attempt["worktree_path"]) } if got, _ := attempt["workspace_status"].(string); got != "" { t.Fatalf("expected no workspace status for non-code task, got %#v", attempt["workspace_status"]) } } func TestOrchWaitWakesOnBlockedEvent(t *testing.T) { t.Parallel() dbPath := filepath.Join(t.TempDir(), "coord.db") runOrchCommand( t, "--db", dbPath, "--json", "run", "init", "--run", "run_blog_wait_001", "--goal", "Validate wait wake behavior", ) runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_wait_001", "--task", "T1", "--title", "Implement backend", "--default-to", "worker-a", ) dispatchOut := runOrchCommand( t, "--db", dbPath, "--json", "dispatch", "--run", "run_blog_wait_001", "--task", "T1", ) var dispatchResp map[string]any mustDecodeJSON(t, dispatchOut, &dispatchResp) threadID := nestedString(t, dispatchResp, "data", "attempt", "thread_id") type waitResult struct { stdout string stderr string exitCode int } resultCh := make(chan waitResult, 1) go func() { stdout, stderr, exitCode := executeOrchCommand( "--db", dbPath, "--json", "wait", "--run", "run_blog_wait_001", "--for", "task_blocked", "--after-event", "0", "--timeout-seconds", "2", ) resultCh <- waitResult{stdout: stdout, stderr: stderr, exitCode: exitCode} }() time.Sleep(200 * time.Millisecond) runInboxCommand( t, "--db", dbPath, "--json", "claim", "--agent", "worker-a", "--thread", threadID, ) runInboxCommand( t, "--db", dbPath, "--json", "update", "--agent", "worker-a", "--thread", threadID, "--status", "blocked", "--summary", "Need logging decision", "--payload-json", `{"question":"stdout or stderr?"}`, ) select { case result := <-resultCh: if result.exitCode != 0 { t.Fatalf("wait exited with %d\nstderr:\n%s\nstdout:\n%s", result.exitCode, result.stderr, result.stdout) } var waitResp map[string]any mustDecodeJSON(t, result.stdout, &waitResp) if woke, _ := nestedValue(t, waitResp, "data", "woke").(bool); !woke { t.Fatalf("expected wait to wake, got %#v", waitResp) } events := nestedArray(t, waitResp, "data", "events") if len(events) != 1 { t.Fatalf("expected one wait event, got %#v", events) } event, ok := events[0].(map[string]any) if !ok { t.Fatalf("expected wait event object, got %#v", events[0]) } if got, _ := event["type"].(string); got != "task_blocked" { t.Fatalf("expected task_blocked event, got %#v", event["type"]) } if got, _ := event["summary"].(string); got != "Need logging decision" { t.Fatalf("expected blocked summary to surface question summary, got %#v", event["summary"]) } payload, ok := event["payload"].(map[string]any) if !ok { t.Fatalf("expected event payload object, got %#v", event["payload"]) } if got, _ := payload["question"].(string); got != "stdout or stderr?" { t.Fatalf("expected question payload, got %#v", payload["question"]) } case <-time.After(3 * time.Second): t.Fatal("timed out waiting for orch wait result") } } func TestOrchWaitTimesOutWithoutMatchingEvent(t *testing.T) { t.Parallel() dbPath := filepath.Join(t.TempDir(), "coord.db") runOrchCommand( t, "--db", dbPath, "--json", "run", "init", "--run", "run_blog_wait_002", "--goal", "Validate wait timeout behavior", ) stdout, stderr, exitCode := executeOrchCommand( "--db", dbPath, "--json", "wait", "--run", "run_blog_wait_002", "--for", "task_done", "--after-event", "0", "--timeout-seconds", "1", ) if exitCode != 0 { t.Fatalf("wait exited with %d\nstderr:\n%s\nstdout:\n%s", exitCode, stderr, stdout) } var waitResp map[string]any mustDecodeJSON(t, stdout, &waitResp) if woke, _ := nestedValue(t, waitResp, "data", "woke").(bool); woke { t.Fatalf("expected wait timeout, got %#v", waitResp) } if nextEventID, _ := nestedValue(t, waitResp, "data", "next_event_id").(float64); nextEventID != 0 { t.Fatalf("expected next_event_id 0 on timeout, got %#v", nextEventID) } } func TestOrchRetryCreatesNewAttempt(t *testing.T) { t.Parallel() dbPath := filepath.Join(t.TempDir(), "coord.db") repoPath := initGitRepo(t) runOrchCommand( t, "--db", dbPath, "--json", "run", "init", "--run", "run_blog_retry_001", "--goal", "Validate retry behavior", ) runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_retry_001", "--task", "T1", "--title", "Implement backend", "--default-to", "worker-a", ) dispatchOut := runOrchCommand( t, "--db", dbPath, "--json", "dispatch", "--run", "run_blog_retry_001", "--task", "T1", "--repo-path", repoPath, "--workspace-root", ".orch/worktrees", "--strict-worktree", ) var dispatchResp map[string]any mustDecodeJSON(t, dispatchOut, &dispatchResp) threadID := nestedString(t, dispatchResp, "data", "attempt", "thread_id") firstWorktreePath := nestedString(t, dispatchResp, "data", "attempt", "worktree_path") runInboxCommand( t, "--db", dbPath, "--json", "claim", "--agent", "worker-a", "--thread", threadID, ) runInboxCommand( t, "--db", dbPath, "--json", "fail", "--agent", "worker-a", "--thread", threadID, "--summary", "Build failed", ) runOrchCommand( t, "--db", dbPath, "--json", "reconcile", "--run", "run_blog_retry_001", ) retryOut := runOrchCommand( t, "--db", dbPath, "--json", "retry", "--run", "run_blog_retry_001", "--task", "T1", "--body", "Retry after fixing the failure.", ) var retryResp map[string]any mustDecodeJSON(t, retryOut, &retryResp) if got := nestedString(t, retryResp, "data", "task", "status"); got != "dispatched" { t.Fatalf("expected retried task to be dispatched, got %q", got) } if got := nestedValue(t, retryResp, "data", "attempt", "attempt_no").(float64); got != 2 { t.Fatalf("expected retry attempt 2, got %#v", got) } secondThreadID := nestedString(t, retryResp, "data", "attempt", "thread_id") if secondThreadID == threadID { t.Fatalf("expected retry to create a new thread, got same thread %q", secondThreadID) } secondWorktreePath := nestedString(t, retryResp, "data", "attempt", "worktree_path") if secondWorktreePath == firstWorktreePath { t.Fatalf("expected retry to create a new worktree, got reused path %q", secondWorktreePath) } if _, err := os.Stat(secondWorktreePath); err != nil { t.Fatalf("stat retry worktree %s: %v", secondWorktreePath, err) } } func TestOrchReassignCancelsOldThreadAndDispatchesNewAttempt(t *testing.T) { t.Parallel() dbPath := filepath.Join(t.TempDir(), "coord.db") repoPath := initGitRepo(t) runOrchCommand( t, "--db", dbPath, "--json", "run", "init", "--run", "run_blog_reassign_001", "--goal", "Validate reassign behavior", ) runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_reassign_001", "--task", "T1", "--title", "Implement backend", "--default-to", "worker-a", ) dispatchOut := runOrchCommand( t, "--db", dbPath, "--json", "dispatch", "--run", "run_blog_reassign_001", "--task", "T1", "--repo-path", repoPath, "--workspace-root", ".orch/worktrees", "--strict-worktree", ) var dispatchResp map[string]any mustDecodeJSON(t, dispatchOut, &dispatchResp) originalThreadID := nestedString(t, dispatchResp, "data", "attempt", "thread_id") runInboxCommand( t, "--db", dbPath, "--json", "claim", "--agent", "worker-a", "--thread", originalThreadID, ) runInboxCommand( t, "--db", dbPath, "--json", "update", "--agent", "worker-a", "--thread", originalThreadID, "--status", "blocked", "--summary", "Need product decision", "--payload-json", `{"question":"Proceed with v1 scope?"}`, ) runOrchCommand( t, "--db", dbPath, "--json", "reconcile", "--run", "run_blog_reassign_001", ) reassignOut := runOrchCommand( t, "--db", dbPath, "--json", "reassign", "--run", "run_blog_reassign_001", "--task", "T1", "--to", "worker-b", "--reason", "Try another worker with clearer ownership.", ) var reassignResp map[string]any mustDecodeJSON(t, reassignOut, &reassignResp) if got := nestedString(t, reassignResp, "data", "attempt", "assigned_to"); got != "worker-b" { t.Fatalf("expected reassigned attempt to target worker-b, got %q", got) } if got := nestedValue(t, reassignResp, "data", "attempt", "attempt_no").(float64); got != 2 { t.Fatalf("expected reassign attempt 2, got %#v", got) } newThreadID := nestedString(t, reassignResp, "data", "attempt", "thread_id") if newThreadID == originalThreadID { t.Fatalf("expected reassignment to create a new thread, got %q", newThreadID) } showOut := runInboxCommand( t, "--db", dbPath, "--json", "show", "--thread", originalThreadID, ) var showResp map[string]any mustDecodeJSON(t, showOut, &showResp) if got := nestedString(t, showResp, "data", "thread", "status"); got != "cancelled" { t.Fatalf("expected old reassigned thread to be cancelled, got %q", got) } } func TestOrchCancelTaskAndRun(t *testing.T) { t.Parallel() dbPath := filepath.Join(t.TempDir(), "coord.db") runOrchCommand( t, "--db", dbPath, "--json", "run", "init", "--run", "run_blog_cancel_001", "--goal", "Validate cancel behavior", ) runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_cancel_001", "--task", "T1", "--title", "Implement backend", "--default-to", "worker-a", ) runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_cancel_001", "--task", "T2", "--title", "Implement frontend", "--default-to", "worker-b", ) dispatchOut := runOrchCommand( t, "--db", dbPath, "--json", "dispatch", "--run", "run_blog_cancel_001", "--task", "T1", ) var dispatchResp map[string]any mustDecodeJSON(t, dispatchOut, &dispatchResp) threadID := nestedString(t, dispatchResp, "data", "attempt", "thread_id") runOrchCommand( t, "--db", dbPath, "--json", "cancel", "--run", "run_blog_cancel_001", "--task", "T1", "--reason", "Task is no longer needed.", ) statusOut := runOrchCommand( t, "--db", dbPath, "--json", "status", "--run", "run_blog_cancel_001", ) var statusResp map[string]any mustDecodeJSON(t, statusOut, &statusResp) tasks := nestedArray(t, statusResp, "data", "tasks") taskStatuses := map[string]string{} for _, item := range tasks { task, ok := item.(map[string]any) if !ok { t.Fatalf("expected task object, got %#v", item) } taskStatuses[task["task_id"].(string)] = task["status"].(string) } if taskStatuses["T1"] != "cancelled" { t.Fatalf("expected T1 cancelled, got %q", taskStatuses["T1"]) } if taskStatuses["T2"] == "cancelled" { t.Fatalf("expected T2 to remain active before run cancel, got %q", taskStatuses["T2"]) } showOut := runInboxCommand( t, "--db", dbPath, "--json", "show", "--thread", threadID, ) var showResp map[string]any mustDecodeJSON(t, showOut, &showResp) if got := nestedString(t, showResp, "data", "thread", "status"); got != "cancelled" { t.Fatalf("expected cancelled task thread to be cancelled, got %q", got) } runOrchCommand( t, "--db", dbPath, "--json", "cancel", "--run", "run_blog_cancel_001", "--reason", "Stop the run.", ) statusOut = runOrchCommand( t, "--db", dbPath, "--json", "status", "--run", "run_blog_cancel_001", ) mustDecodeJSON(t, statusOut, &statusResp) if got := nestedString(t, statusResp, "data", "run", "status"); got != "cancelled" { t.Fatalf("expected cancelled run, got %q", got) } tasks = nestedArray(t, statusResp, "data", "tasks") for _, item := range tasks { task, ok := item.(map[string]any) if !ok { t.Fatalf("expected task object, got %#v", item) } if got, _ := task["status"].(string); got != "cancelled" { t.Fatalf("expected all tasks cancelled after run cancel, got %#v", task["status"]) } } } func TestOrchCleanupRemovesCompletedWorktree(t *testing.T) { t.Parallel() dbPath := filepath.Join(t.TempDir(), "coord.db") repoPath := initGitRepo(t) runOrchCommand( t, "--db", dbPath, "--json", "run", "init", "--run", "run_blog_cleanup_001", "--goal", "Validate cleanup behavior", ) runOrchCommand( t, "--db", dbPath, "--json", "task", "add", "--run", "run_blog_cleanup_001", "--task", "T1", "--title", "Implement backend", "--default-to", "worker-a", ) dispatchOut := runOrchCommand( t, "--db", dbPath, "--json", "dispatch", "--run", "run_blog_cleanup_001", "--task", "T1", "--repo-path", repoPath, "--workspace-root", ".orch/worktrees", "--strict-worktree", ) var dispatchResp map[string]any mustDecodeJSON(t, dispatchOut, &dispatchResp) threadID := nestedString(t, dispatchResp, "data", "attempt", "thread_id") worktreePath := nestedString(t, dispatchResp, "data", "attempt", "worktree_path") runInboxCommand( t, "--db", dbPath, "--json", "claim", "--agent", "worker-a", "--thread", threadID, ) runInboxCommand( t, "--db", dbPath, "--json", "done", "--agent", "worker-a", "--thread", threadID, "--summary", "Backend complete", ) runOrchCommand( t, "--db", dbPath, "--json", "reconcile", "--run", "run_blog_cleanup_001", ) cleanupOut := runOrchCommand( t, "--db", dbPath, "--json", "cleanup", "--run", "run_blog_cleanup_001", "--task", "T1", ) var cleanupResp map[string]any mustDecodeJSON(t, cleanupOut, &cleanupResp) cleaned := nestedArray(t, cleanupResp, "data", "cleaned") if len(cleaned) != 1 { t.Fatalf("expected one cleaned attempt, got %#v", cleaned) } if _, err := os.Stat(worktreePath); !os.IsNotExist(err) { t.Fatalf("expected cleaned worktree path to be removed, err=%v", err) } } func TestOrchCouncilStartDispatchesThreeReviewers(t *testing.T) { t.Parallel() dbPath := filepath.Join(t.TempDir(), "coord.db") startOut := runOrchCommand( t, "--db", dbPath, "--json", "council", "start", "--run", "council_blog_001", "--target", "Review the current blog architecture and propose optimizations.", "--target-type", "mixed", "--output", "both", ) var startResp map[string]any mustDecodeJSON(t, startOut, &startResp) if got := nestedString(t, startResp, "data", "run_id"); got != "council_blog_001" { t.Fatalf("expected council run id, got %q", got) } if got := nestedString(t, startResp, "data", "mode"); got != "brainstorm" { t.Fatalf("expected default council mode brainstorm, got %q", got) } reviewers := nestedArray(t, startResp, "data", "reviewers") if len(reviewers) != 3 { t.Fatalf("expected three council reviewers, got %#v", reviewers) } sqlDB, err := openOrchDB(t.Context(), dbPath) if err != nil { t.Fatalf("open orch db: %v", err) } defer sqlDB.Close() var ( mode string targetType string outputMode string onlyUnanimous int ) if err := sqlDB.QueryRowContext( t.Context(), `SELECT mode, target_type, output_mode, only_unanimous FROM council_runs WHERE run_id = ?`, "council_blog_001", ).Scan(&mode, &targetType, &outputMode, &onlyUnanimous); err != nil { t.Fatalf("query council_runs: %v", err) } if mode != "brainstorm" || targetType != "mixed" || outputMode != "both" || onlyUnanimous != 0 { t.Fatalf("unexpected council run metadata: mode=%q targetType=%q outputMode=%q onlyUnanimous=%d", mode, targetType, outputMode, onlyUnanimous) } var ( prompt string targetFile string repoPath string targetTaskID string ) if err := sqlDB.QueryRowContext( t.Context(), `SELECT prompt, target_file, repo_path, target_task_id FROM council_inputs WHERE run_id = ?`, "council_blog_001", ).Scan(&prompt, &targetFile, &repoPath, &targetTaskID); err != nil { t.Fatalf("query council_inputs: %v", err) } if prompt == "" || targetFile != "" || repoPath != "" || targetTaskID != "" { t.Fatalf("unexpected council input row: prompt=%q targetFile=%q repoPath=%q targetTaskID=%q", prompt, targetFile, repoPath, targetTaskID) } rows, err := sqlDB.QueryContext( t.Context(), `SELECT reviewer_role, task_id, status FROM council_reviewers WHERE run_id = ? ORDER BY reviewer_role ASC`, "council_blog_001", ) if err != nil { t.Fatalf("query council_reviewers: %v", err) } defer rows.Close() var reviewerRows int for rows.Next() { var ( reviewerRole string taskID string status string ) if err := rows.Scan(&reviewerRole, &taskID, &status); err != nil { t.Fatalf("scan council reviewer row: %v", err) } reviewerRows++ if status != "dispatched" { t.Fatalf("expected council reviewer status dispatched, got %q for %s", status, reviewerRole) } var worktreePath sql.NullString if err := sqlDB.QueryRowContext( t.Context(), `SELECT a.worktree_path FROM task_attempts a WHERE a.run_id = ? AND a.task_id = ? AND a.attempt_no = 1`, "council_blog_001", taskID, ).Scan(&worktreePath); err != nil { t.Fatalf("query council attempt worktree path for %s: %v", taskID, err) } if worktreePath.Valid && worktreePath.String != "" { t.Fatalf("expected council reviewer task %s to avoid worktree allocation, got %q", taskID, worktreePath.String) } } if err := rows.Err(); err != nil { t.Fatalf("iterate council reviewers: %v", err) } if reviewerRows != 3 { t.Fatalf("expected three stored council reviewers, got %d", reviewerRows) } statusOut := runOrchCommand( t, "--db", dbPath, "--json", "status", "--run", "council_blog_001", ) var statusResp map[string]any mustDecodeJSON(t, statusOut, &statusResp) if got := nestedString(t, statusResp, "data", "run", "status"); got != "running" { t.Fatalf("expected council run status running, got %q", got) } tasks := nestedArray(t, statusResp, "data", "tasks") if len(tasks) != 3 { t.Fatalf("expected three council tasks, got %#v", tasks) } }