Refactor orchestrator with modular service architecture

agents.md

@@ -31,7 +31,8 @@ opencode-os/
 |------|----------|-------|
 | Add API endpoint | `crates/server/src/routes/` | Add to lib.rs OpenAPI schema |
 | Task state logic | `crates/orchestrator/src/state_machine.rs` | TaskStateMachine |
-| OpenCode integration | `crates/orchestrator/src/executor.rs` | Uses opencode-client SDK |
+| OpenCode integration | `crates/orchestrator/src/executor.rs` | Thin orchestrator, delegates to services |
+| Phase services | `crates/orchestrator/src/services/` | planning, implementation, review, fix phases |
 | VCS operations | `crates/vcs/src/` | jj.rs primary, git.rs fallback |
 | Frontend component | `frontend/src/components/` | Feature dirs: kanban/, sessions/, task-detail/ |
 | Generated types | `frontend/src/types/generated/` | ts-rs from Rust |
@@ -113,8 +114,9 @@ DATABASE_URL=sqlite:./studio.db cargo run --package server
 # Frontend only
 cd frontend && pnpm dev
 
-# Tests (109 unit tests across 9 crates)
+# Tests
 cargo test --workspace
+cargo test -p orchestrator  # 55 tests
 cargo clippy --workspace --all-features -- -D warnings
 
 # Generate frontend SDK

crates/AGENTS.md

@@ -11,11 +11,11 @@
 | `core` | Domain models (NO I/O) | `Task`, `Session`, `TaskStatus`, `SessionPhase` | 10 |
 | `db` | SQLite persistence (sqlx) | `TaskRepository`, `SessionRepository`, `create_pool` | 12 |
 | `opencode-client` | OpenAPI-generated SDK | `apis::DefaultApi`, `Configuration` | 0 |
-| `orchestrator` | Task lifecycle engine | `TaskExecutor`, `TaskStateMachine`, `FileManager` | 36 |
+| `orchestrator` | Task lifecycle engine | `TaskExecutor`, `TaskStateMachine`, `services::*` | 55 |
 | `vcs` | VCS abstraction | `VersionControl`, `WorkspaceManager`, `JujutsuVcs`, `GitVcs` | 20 |
 | `events` | Event bus | `EventBus`, `TaskEvent`, `SessionEvent` | 8 |
 | `github` | GitHub API (octocrab) | `GitHubClient`, `PullRequest`, `Issue` | 11 |
-| `server` | Axum HTTP + SSE | `AppState`, `router`, `OpenApi` | 12 |
+| `server` | Axum HTTP + SSE | `AppState`, `router`, `OpenApi` | 20 |
 | `cli` | Binary: `opencode-studio` | Commands: init, serve, status, update | 0 |
 
 ## DEPENDENCY GRAPH
@@ -40,10 +40,33 @@ Foundational (no internal deps): core, events, opencode-client
 | Add API route | `server` | `src/routes/` + update `src/lib.rs` OpenAPI |
 | Task state transitions | `orchestrator` | `src/state_machine.rs` |
 | AI prompts | `orchestrator` | `src/prompts.rs` |
+| Planning phase | `orchestrator` | `src/services/planning_phase.rs` |
+| Implementation phase | `orchestrator` | `src/services/implementation_phase.rs` |
+| Review phase | `orchestrator` | `src/services/review_phase.rs` |
+| Fix phase | `orchestrator` | `src/services/fix_phase.rs` |
+| OpenCode API calls | `orchestrator` | `src/services/opencode_client.rs` |
+| Message parsing | `orchestrator` | `src/services/message_parser.rs` |
 | VCS operations | `vcs` | `src/jj.rs` (primary), `src/git.rs` (fallback) |
 | Event emission | `events` | `src/types.rs` for new event types |
 | GitHub integration | `github` | `src/client.rs` |
 
+## ORCHESTRATOR SERVICES
+
+The `orchestrator` crate uses a modular service architecture in `src/services/`:
+
+| Service | Purpose | Lines |
+|:--------|:--------|------:|
+| `executor_context.rs` | Shared context, config, transitions, persistence | 243 |
+| `planning_phase.rs` | Planning phase execution | 136 |
+| `implementation_phase.rs` | Implementation + phased execution | 646 |
+| `review_phase.rs` | AI review with JSON fallback | 353 |
+| `fix_phase.rs` | Fix iteration handling | 269 |
+| `opencode_client.rs` | OpenCode session/prompt API | 210 |
+| `message_parser.rs` | SSE parsing, ReviewResult extraction | 327 |
+| `mcp_manager.rs` | MCP server lifecycle | 108 |
+
+The main `executor.rs` (~530 lines) delegates to these services.
+
 ## CONVENTIONS
 
 - `core` exports as `opencode_core` (reserved word collision)
@@ -62,7 +85,8 @@ Foundational (no internal deps): core, events, opencode-client
 ## TEST COMMANDS
 
 ```bash
-cargo test --workspace                    # All 109 tests
-cargo test -p orchestrator               # Single crate
-cargo test -p server -- --nocapture      # With output
+cargo test --workspace                    # All tests
+cargo test -p orchestrator               # 55 tests
+cargo test -p server -- --nocapture      # 20 tests with output
+cargo clippy --workspace -- -D warnings  # Lint check
 ```

crates/cli/src/main.rs

@@ -182,10 +182,7 @@ async fn download_frontend(app_dir: &PathBuf, show_progress: bool) -> Result<()>
         .context("Failed to download frontend")?;
 
     if !response.status().is_success() {
-        anyhow::bail!(
-            "Failed to download frontend: HTTP {}",
-            response.status()
-        );
+        anyhow::bail!("Failed to download frontend: HTTP {}", response.status());
     }
 
     let total_size = response.content_length().unwrap_or(0);
@@ -436,7 +433,11 @@ fn print_init_success(project_name: &str) {
     println!("        ├── {}/", "plans".dimmed());
     println!("        └── {}/", "reviews".dimmed());
     println!();
-    println!("  {} Database stored in {}", "ℹ".blue(), "~/.opencode-studio/data/".dimmed());
+    println!(
+        "  {} Database stored in {}",
+        "ℹ".blue(),
+        "~/.opencode-studio/data/".dimmed()
+    );
     println!();
 }
 
@@ -620,10 +621,7 @@ async fn status(path: Option<PathBuf>) -> Result<()> {
     if !studio_dir.exists() {
         println!();
         println!("  {} Not an OpenCode Studio project.", "✗".red());
-        println!(
-            "     Run {} to initialize.",
-            "opencode-studio init".cyan()
-        );
+        println!("     Run {} to initialize.", "opencode-studio init".cyan());
         println!();
         return Ok(());
     }
@@ -636,11 +634,7 @@ async fn status(path: Option<PathBuf>) -> Result<()> {
         Ok(p) => p,
         Err(e) => {
             println!();
-            println!(
-                "  {} Failed to determine database path: {}",
-                "✗".red(),
-                e
-            );
+            println!("  {} Failed to determine database path: {}", "✗".red(), e);
             return Ok(());
         }
     };
@@ -664,11 +658,7 @@ async fn status(path: Option<PathBuf>) -> Result<()> {
     let tasks = task_repo.find_all().await?;
 
     println!();
-    println!(
-        "  {} {}",
-        "◆".magenta(),
-        config.project.name.white().bold()
-    );
+    println!("  {} {}", "◆".magenta(), config.project.name.white().bold());
     println!("    {}", cwd.display().to_string().dimmed());
     println!();
 
@@ -722,8 +712,7 @@ fn init_tracing() {
     tracing_subscriber::registry()
         .with(tracing_subscriber::fmt::layer().with_target(false))
         .with(
-            tracing_subscriber::EnvFilter::try_from_default_env()
-                .unwrap_or_else(|_| "warn".into()),
+            tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "warn".into()),
         )
         .init();
 }

crates/db/src/repositories/review_comment_repository.rs

@@ -59,6 +59,7 @@ impl ReviewCommentRepository {
     }
 
     /// Create a new comment
+    #[allow(clippy::too_many_arguments)]
     pub async fn create(
         &self,
         id: &str,

crates/db/src/repositories/session_activity_repository.rs

@@ -14,7 +14,8 @@ impl SessionActivityRepository {
     }
 
     pub async fn create(&self, activity: &CreateSessionActivity) -> Result<i64, DbError> {
-        let data_json = serde_json::to_string(&activity.data).unwrap_or_else(|_| "null".to_string());
+        let data_json =
+            serde_json::to_string(&activity.data).unwrap_or_else(|_| "null".to_string());
         let created_at = chrono::Utc::now().timestamp();
 
         let result = sqlx::query(
@@ -34,7 +35,10 @@ impl SessionActivityRepository {
         Ok(result.last_insert_rowid())
     }
 
-    pub async fn find_by_session_id(&self, session_id: Uuid) -> Result<Vec<SessionActivity>, DbError> {
+    pub async fn find_by_session_id(
+        &self,
+        session_id: Uuid,
+    ) -> Result<Vec<SessionActivity>, DbError> {
         let rows: Vec<SessionActivityRow> = sqlx::query_as(
             r#"
             SELECT id, session_id, activity_type, activity_id, data, created_at
@@ -72,12 +76,11 @@ impl SessionActivityRepository {
     }
 
     pub async fn count_by_session_id(&self, session_id: Uuid) -> Result<i64, DbError> {
-        let count: (i64,) = sqlx::query_as(
-            "SELECT COUNT(*) FROM session_activities WHERE session_id = ?",
-        )
-        .bind(session_id.to_string())
-        .fetch_one(&self.pool)
-        .await?;
+        let count: (i64,) =
+            sqlx::query_as("SELECT COUNT(*) FROM session_activities WHERE session_id = ?")
+                .bind(session_id.to_string())
+                .fetch_one(&self.pool)
+                .await?;
 
         Ok(count.0)
     }
@@ -177,7 +180,10 @@ mod tests {
         }
 
         // Get activities since id[2]
-        let activities = repo.find_by_session_id_since(session.id, ids[2]).await.unwrap();
+        let activities = repo
+            .find_by_session_id_since(session.id, ids[2])
+            .await
+            .unwrap();
         assert_eq!(activities.len(), 2);
         assert_eq!(activities[0].id, ids[3]);
         assert_eq!(activities[1].id, ids[4]);

crates/github/src/client.rs

@@ -71,17 +71,35 @@ impl GitHubClient {
         let pulls_handler = self.octocrab.pulls(&self.repo.owner, &self.repo.repo);
 
         let prs = match state {
-            Some(PrState::Open) => pulls_handler.list().state(octocrab::params::State::Open).send().await?,
+            Some(PrState::Open) => {
+                pulls_handler
+                    .list()
+                    .state(octocrab::params::State::Open)
+                    .send()
+                    .await?
+            }
             Some(PrState::Closed) | Some(PrState::Merged) => {
-                pulls_handler.list().state(octocrab::params::State::Closed).send().await?
+                pulls_handler
+                    .list()
+                    .state(octocrab::params::State::Closed)
+                    .send()
+                    .await?
             }
             None => pulls_handler.list().send().await?,
         };
 
-        Ok(prs.items.into_iter().map(|pr| self.convert_pr(pr)).collect())
+        Ok(prs
+            .items
+            .into_iter()
+            .map(|pr| self.convert_pr(pr))
+            .collect())
     }
 
-    pub async fn merge_pull_request(&self, number: u64, commit_message: Option<&str>) -> Result<()> {
+    pub async fn merge_pull_request(
+        &self,
+        number: u64,
+        commit_message: Option<&str>,
+    ) -> Result<()> {
         info!("Merging PR #{}", number);
 
         let pulls_handler = self.octocrab.pulls(&self.repo.owner, &self.repo.repo);
@@ -129,10 +147,7 @@ impl GitHubClient {
             state,
             head_branch: pr.head.ref_field,
             base_branch: pr.base.ref_field,
-            html_url: pr
-                .html_url
-                .map(|u| u.to_string())
-                .unwrap_or_default(),
+            html_url: pr.html_url.map(|u| u.to_string()).unwrap_or_default(),
             created_at: pr.created_at.unwrap_or_default(),
             updated_at: pr.updated_at.unwrap_or_default(),
             merged_at: pr.merged_at,
@@ -183,10 +198,7 @@ impl GitHubClient {
         poll_interval_secs: u64,
         max_wait_secs: u64,
     ) -> Result<CiStatus> {
-        info!(
-            "Waiting for CI on {} (max {}s)",
-            ref_name, max_wait_secs
-        );
+        info!("Waiting for CI on {} (max {}s)", ref_name, max_wait_secs);
 
         let start = std::time::Instant::now();
         let poll_duration = std::time::Duration::from_secs(poll_interval_secs);
@@ -225,7 +237,11 @@ impl GitHubClient {
                 None => has_pending = true,
                 Some(c) if c.contains("Success") => {}
                 Some(c) if c.contains("Skipped") || c.contains("Neutral") => {}
-                Some(c) if c.contains("Failure") || c.contains("Cancelled") || c.contains("TimedOut") => {
+                Some(c)
+                    if c.contains("Failure")
+                        || c.contains("Cancelled")
+                        || c.contains("TimedOut") =>
+                {
                     has_failure = true
                 }
                 Some(c) if c.contains("ActionRequired") => has_pending = true,
@@ -262,8 +278,20 @@ impl GitHubClient {
         let issues_handler = self.octocrab.issues(&self.repo.owner, &self.repo.repo);
 
         let issues = match state {
-            Some(IssueState::Open) => issues_handler.list().state(octocrab::params::State::Open).send().await?,
-            Some(IssueState::Closed) => issues_handler.list().state(octocrab::params::State::Closed).send().await?,
+            Some(IssueState::Open) => {
+                issues_handler
+                    .list()
+                    .state(octocrab::params::State::Open)
+                    .send()
+                    .await?
+            }
+            Some(IssueState::Closed) => {
+                issues_handler
+                    .list()
+                    .state(octocrab::params::State::Closed)
+                    .send()
+                    .await?
+            }
             None => issues_handler.list().send().await?,
         };
 
@@ -278,10 +306,8 @@ impl GitHubClient {
     pub async fn import_issue(&self, number: u64) -> Result<opencode_core::Task> {
         let issue = self.get_issue(number).await?;
 
-        let task = opencode_core::Task::new(
-            issue.title.clone(),
-            issue.body.clone().unwrap_or_default(),
-        );
+        let task =
+            opencode_core::Task::new(issue.title.clone(), issue.body.clone().unwrap_or_default());
 
         info!("Imported issue #{} as task {}", number, task.id);
         Ok(task)
@@ -443,7 +469,11 @@ mod tests {
                 None => has_pending = true,
                 Some(c) if c.contains("Success") => {}
                 Some(c) if c.contains("Skipped") || c.contains("Neutral") => {}
-                Some(c) if c.contains("Failure") || c.contains("Cancelled") || c.contains("TimedOut") => {
+                Some(c)
+                    if c.contains("Failure")
+                        || c.contains("Cancelled")
+                        || c.contains("TimedOut") =>
+                {
                     has_failure = true
                 }
                 Some(c) if c.contains("ActionRequired") => has_pending = true,