diff options
Diffstat (limited to 'makima/src')
| -rw-r--r-- | makima/src/bin/makima.rs | 93 | ||||
| -rw-r--r-- | makima/src/daemon/api/directive.rs | 106 | ||||
| -rw-r--r-- | makima/src/daemon/api/mod.rs | 1 | ||||
| -rw-r--r-- | makima/src/daemon/cli/directive.rs | 60 | ||||
| -rw-r--r-- | makima/src/daemon/cli/mod.rs | 40 | ||||
| -rw-r--r-- | makima/src/daemon/skills/directive.md | 88 | ||||
| -rw-r--r-- | makima/src/daemon/skills/mod.rs | 4 | ||||
| -rw-r--r-- | makima/src/db/models.rs | 333 | ||||
| -rw-r--r-- | makima/src/db/repository.rs | 888 | ||||
| -rw-r--r-- | makima/src/orchestration/directive.rs | 1685 | ||||
| -rw-r--r-- | makima/src/orchestration/mod.rs | 2 | ||||
| -rw-r--r-- | makima/src/server/handlers/contracts.rs | 18 | ||||
| -rw-r--r-- | makima/src/server/handlers/directives.rs | 785 | ||||
| -rw-r--r-- | makima/src/server/handlers/mesh_daemon.rs | 10 | ||||
| -rw-r--r-- | makima/src/server/handlers/mod.rs | 1 | ||||
| -rw-r--r-- | makima/src/server/mod.rs | 19 | ||||
| -rw-r--r-- | makima/src/server/openapi.rs | 44 |
17 files changed, 17 insertions, 4160 deletions
diff --git a/makima/src/bin/makima.rs b/makima/src/bin/makima.rs index 8115387..ee5895c 100644 --- a/makima/src/bin/makima.rs +++ b/makima/src/bin/makima.rs @@ -7,7 +7,7 @@ use std::sync::Arc; use makima::daemon::api::{ApiClient, CreateContractRequest}; use makima::daemon::cli::{ Cli, CliConfig, Commands, ConfigCommand, ContractCommand, - DirectiveCommand, SupervisorCommand, ViewArgs, + SupervisorCommand, ViewArgs, }; use makima::daemon::tui::{self, Action, App, ListItem, ViewType, TuiWsClient, WsEvent, OutputLine, OutputMessageType, WsConnectionState, RepositorySuggestion}; use makima::daemon::config::{DaemonConfig, RepoEntry}; @@ -29,7 +29,6 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> { Commands::Daemon(args) => run_daemon(args).await, Commands::Supervisor(cmd) => run_supervisor(cmd).await, Commands::Contract(cmd) => run_contract(cmd).await, - Commands::Directive(cmd) => run_directive(cmd).await, Commands::View(args) => run_view(args).await, Commands::Config(cmd) => run_config(cmd).await, } @@ -712,96 +711,6 @@ async fn run_contract( Ok(()) } -/// Run a directive subcommand. -async fn run_directive( - cmd: DirectiveCommand, -) -> Result<(), Box<dyn std::error::Error + Send + Sync>> { - match cmd { - DirectiveCommand::Status(args) => { - let client = ApiClient::new(args.api_url, args.api_key)?; - let result = client.directive_status(args.directive_id).await?; - println!("{}", serde_json::to_string(&result.0)?); - } - DirectiveCommand::Goals(args) => { - let client = ApiClient::new(args.api_url, args.api_key)?; - let result = client.directive_status(args.directive_id).await?; - // Extract goal-related fields from directive - let directive = &result.0; - let goals = serde_json::json!({ - "goal": directive.get("goal"), - "requirements": directive.get("requirements"), - "acceptanceCriteria": directive.get("acceptanceCriteria"), - "constraints": directive.get("constraints"), - "externalDependencies": directive.get("externalDependencies"), - }); - println!("{}", serde_json::to_string(&goals)?); - } - DirectiveCommand::Chains(args) => { - let client = ApiClient::new(args.api_url, args.api_key)?; - let result = client.directive_chains(args.directive_id).await?; - println!("{}", serde_json::to_string(&result.0)?); - } - DirectiveCommand::Chain(args) => { - let client = ApiClient::new(args.common.api_url, args.common.api_key)?; - let result = client - .directive_chain(args.common.directive_id, args.chain_id) - .await?; - println!("{}", serde_json::to_string(&result.0)?); - } - DirectiveCommand::Steps(args) => { - let client = ApiClient::new(args.common.api_url, args.common.api_key)?; - let result = client - .directive_chain(args.common.directive_id, args.chain_id) - .await?; - // Extract steps from chain response - let steps = result.0.get("steps").cloned().unwrap_or(serde_json::json!([])); - println!("{}", serde_json::to_string(&steps)?); - } - DirectiveCommand::UpdateStatus(args) => { - let client = ApiClient::new(args.common.api_url, args.common.api_key)?; - let req = makima::daemon::api::directive::UpdateDirectiveRequest { - status: Some(args.status), - version: None, - }; - let result = client - .directive_update(args.common.directive_id, req) - .await?; - println!("{}", serde_json::to_string(&result.0)?); - } - DirectiveCommand::Start(args) => { - let client = ApiClient::new(args.api_url, args.api_key)?; - let result = client.directive_start(args.directive_id).await?; - println!("{}", serde_json::to_string(&result.0)?); - } - DirectiveCommand::Evaluate(args) => { - let client = ApiClient::new(args.common.api_url, args.common.api_key)?; - let result = client - .directive_evaluate_step(args.common.directive_id, args.step_id) - .await?; - println!("{}", serde_json::to_string(&result.0)?); - } - DirectiveCommand::Evaluations(args) => { - let client = ApiClient::new(args.common.api_url, args.common.api_key)?; - let result = client - .directive_evaluations(args.common.directive_id, args.step_id) - .await?; - println!("{}", serde_json::to_string(&result.0)?); - } - DirectiveCommand::SubmitPlan(args) => { - let client = ApiClient::new(args.api_url, args.api_key)?; - // Read plan JSON from stdin - let mut plan_json = String::new(); - io::stdin().read_to_string(&mut plan_json)?; - let result = client - .directive_submit_plan(args.directive_id, &plan_json) - .await?; - println!("{}", serde_json::to_string(&result.0)?); - } - } - - Ok(()) -} - /// Run the TUI view command. async fn run_view(args: ViewArgs) -> Result<(), Box<dyn std::error::Error + Send + Sync>> { // Load CLI config for defaults diff --git a/makima/src/daemon/api/directive.rs b/makima/src/daemon/api/directive.rs deleted file mode 100644 index c51882b..0000000 --- a/makima/src/daemon/api/directive.rs +++ /dev/null @@ -1,106 +0,0 @@ -//! Directive API methods. - -use serde::Serialize; -use uuid::Uuid; - -use super::client::{ApiClient, ApiError}; -use super::supervisor::JsonValue; - -/// Request to update a directive. -#[derive(Serialize)] -#[serde(rename_all = "camelCase")] -pub struct UpdateDirectiveRequest { - #[serde(skip_serializing_if = "Option::is_none")] - pub status: Option<String>, - #[serde(skip_serializing_if = "Option::is_none")] - pub version: Option<i32>, -} - -impl ApiClient { - /// Get directive status and details. - pub async fn directive_status(&self, directive_id: Uuid) -> Result<JsonValue, ApiError> { - self.get(&format!("/api/v1/directives/{}", directive_id)) - .await - } - - /// List chains for a directive. - pub async fn directive_chains(&self, directive_id: Uuid) -> Result<JsonValue, ApiError> { - self.get(&format!("/api/v1/directives/{}/chains", directive_id)) - .await - } - - /// Get a chain with its steps. - pub async fn directive_chain( - &self, - directive_id: Uuid, - chain_id: Uuid, - ) -> Result<JsonValue, ApiError> { - self.get(&format!( - "/api/v1/directives/{}/chains/{}", - directive_id, chain_id - )) - .await - } - - /// Update a directive. - pub async fn directive_update( - &self, - directive_id: Uuid, - req: UpdateDirectiveRequest, - ) -> Result<JsonValue, ApiError> { - self.put(&format!("/api/v1/directives/{}", directive_id), &req) - .await - } - - /// Start a directive (transition from draft to planning). - pub async fn directive_start(&self, directive_id: Uuid) -> Result<JsonValue, ApiError> { - self.post_empty(&format!("/api/v1/directives/{}/start", directive_id)) - .await - } - - /// Trigger a manual evaluation for a step. - pub async fn directive_evaluate_step( - &self, - directive_id: Uuid, - step_id: Uuid, - ) -> Result<JsonValue, ApiError> { - self.post_empty(&format!( - "/api/v1/directives/{}/steps/{}/evaluate", - directive_id, step_id - )) - .await - } - - /// Submit a chain plan for a directive. - pub async fn directive_submit_plan( - &self, - directive_id: Uuid, - plan_json: &str, - ) -> Result<JsonValue, ApiError> { - #[derive(serde::Serialize)] - #[serde(rename_all = "camelCase")] - struct SubmitPlanBody { - plan: String, - } - self.post( - &format!("/api/v1/directives/{}/submit-plan", directive_id), - &SubmitPlanBody { - plan: plan_json.to_string(), - }, - ) - .await - } - - /// List evaluations for a step. - pub async fn directive_evaluations( - &self, - directive_id: Uuid, - step_id: Uuid, - ) -> Result<JsonValue, ApiError> { - self.get(&format!( - "/api/v1/directives/{}/steps/{}/evaluations", - directive_id, step_id - )) - .await - } -} diff --git a/makima/src/daemon/api/mod.rs b/makima/src/daemon/api/mod.rs index 2d1efbf..49d80e0 100644 --- a/makima/src/daemon/api/mod.rs +++ b/makima/src/daemon/api/mod.rs @@ -2,7 +2,6 @@ pub mod client; pub mod contract; -pub mod directive; pub mod supervisor; pub use client::ApiClient; diff --git a/makima/src/daemon/cli/directive.rs b/makima/src/daemon/cli/directive.rs deleted file mode 100644 index 4c29c14..0000000 --- a/makima/src/daemon/cli/directive.rs +++ /dev/null @@ -1,60 +0,0 @@ -//! Directive subcommand - directive orchestration commands. - -use clap::Args; -use uuid::Uuid; - -/// Common arguments for directive commands. -#[derive(Args, Debug, Clone)] -pub struct DirectiveArgs { - /// API URL - #[arg(long, env = "MAKIMA_API_URL", default_value = "https://api.makima.jp", global = true)] - pub api_url: String, - - /// API key for authentication - #[arg(long, env = "MAKIMA_API_KEY", global = true)] - pub api_key: String, - - /// Directive ID - #[arg(long, env = "MAKIMA_DIRECTIVE_ID", global = true)] - pub directive_id: Uuid, -} - -/// Arguments for chain command (get specific chain). -#[derive(Args, Debug)] -pub struct ChainArgs { - #[command(flatten)] - pub common: DirectiveArgs, - - /// Chain ID to retrieve - pub chain_id: Uuid, -} - -/// Arguments for update-status command. -#[derive(Args, Debug)] -pub struct UpdateStatusArgs { - #[command(flatten)] - pub common: DirectiveArgs, - - /// New status (draft, planning, active, paused, completed, archived, failed) - pub status: String, -} - -/// Arguments for evaluate command (trigger manual evaluation). -#[derive(Args, Debug)] -pub struct EvaluateArgs { - #[command(flatten)] - pub common: DirectiveArgs, - - /// Step ID to evaluate - pub step_id: Uuid, -} - -/// Arguments for evaluations command (list evaluation history). -#[derive(Args, Debug)] -pub struct EvaluationsArgs { - #[command(flatten)] - pub common: DirectiveArgs, - - /// Step ID to list evaluations for - pub step_id: Uuid, -} diff --git a/makima/src/daemon/cli/mod.rs b/makima/src/daemon/cli/mod.rs index 954f219..0805edd 100644 --- a/makima/src/daemon/cli/mod.rs +++ b/makima/src/daemon/cli/mod.rs @@ -3,7 +3,6 @@ pub mod config; pub mod contract; pub mod daemon; -pub mod directive; pub mod server; pub mod supervisor; pub mod view; @@ -13,7 +12,6 @@ use clap::{Parser, Subcommand}; pub use config::CliConfig; pub use contract::ContractArgs; pub use daemon::DaemonArgs; -pub use directive::DirectiveArgs; pub use server::ServerArgs; pub use supervisor::SupervisorArgs; pub use view::ViewArgs; @@ -43,10 +41,6 @@ pub enum Commands { #[command(subcommand)] Contract(ContractCommand), - /// Directive commands for autonomous goal-driven execution - #[command(subcommand)] - Directive(DirectiveCommand), - /// Interactive TUI browser for contracts and tasks /// /// Provides a drill-down interface for browsing contracts, viewing their @@ -202,40 +196,6 @@ pub enum ContractCommand { CreateFile(contract::CreateFileArgs), } -/// Directive subcommands for autonomous goal-driven execution. -#[derive(Subcommand, Debug)] -pub enum DirectiveCommand { - /// Get directive status and details - Status(DirectiveArgs), - - /// Get goal, requirements, acceptance criteria - Goals(DirectiveArgs), - - /// List chains for the directive - Chains(DirectiveArgs), - - /// Get a chain with its steps - Chain(directive::ChainArgs), - - /// List steps in a chain - Steps(directive::ChainArgs), - - /// Update directive status - UpdateStatus(directive::UpdateStatusArgs), - - /// Start a directive (create planning contract and begin orchestration) - Start(DirectiveArgs), - - /// Trigger a manual evaluation for a step - Evaluate(directive::EvaluateArgs), - - /// List evaluation history for a step - Evaluations(directive::EvaluationsArgs), - - /// Submit a chain plan for a directive (reads JSON from stdin) - SubmitPlan(DirectiveArgs), -} - impl Cli { /// Parse command-line arguments pub fn parse_args() -> Self { diff --git a/makima/src/daemon/skills/directive.md b/makima/src/daemon/skills/directive.md deleted file mode 100644 index 0d1e9d6..0000000 --- a/makima/src/daemon/skills/directive.md +++ /dev/null @@ -1,88 +0,0 @@ ---- -name: makima-directive -description: Directive orchestration tools for autonomous goal-driven execution. Use when working with directives, chains, steps, verifiers, and approvals. ---- - -# Makima Directive Commands - -These commands let orchestrators interact with directive state. Environment variables (`MAKIMA_API_URL`, `MAKIMA_API_KEY`, `MAKIMA_DIRECTIVE_ID`) are pre-configured by the daemon. - -## Status and Information - -### Get directive status -```bash -makima directive status -``` -Returns full directive details including status, autonomy level, thresholds, and tracking info. - -### Get directive goals -```bash -makima directive goals -``` -Returns the goal, requirements, acceptance criteria, constraints, and external dependencies. - -### List chains -```bash -makima directive chains -``` -Returns all chains (plan generations) for the directive, ordered by generation. - -### Get chain with steps -```bash -makima directive chain <chain_id> -``` -Returns a chain and all its steps with status, dependencies, and evaluation info. - -### List steps in a chain -```bash -makima directive steps <chain_id> -``` -Returns just the steps array from a chain. - -## Status Updates - -### Update directive status -```bash -makima directive update-status <status> -``` -Updates the directive status. Valid statuses: `draft`, `planning`, `active`, `paused`, `completed`, `archived`, `failed`. - -## Evaluation - -### Trigger manual evaluation for a step -```bash -makima directive evaluate <step_id> -``` -Triggers a monitoring evaluation for the specified step. The step must have been executed (have a contract). Sets the step to "evaluating" and dispatches a monitoring contract. - -### List evaluations for a step -```bash -makima directive evaluations <step_id> -``` -Returns the evaluation history for a step, ordered by evaluation number. - -## Output Format - -All commands output JSON to stdout. - -Example workflow: -```bash -# Check directive details and goals -makima directive status -makima directive goals - -# List execution chains -makima directive chains - -# Get details of a specific chain -makima directive chain <chain_id> - -# Trigger manual evaluation of a step -makima directive evaluate <step_id> - -# Check evaluation history -makima directive evaluations <step_id> - -# Update status to active -makima directive update-status active -``` diff --git a/makima/src/daemon/skills/mod.rs b/makima/src/daemon/skills/mod.rs index 6e5d0a8..0b05f3a 100644 --- a/makima/src/daemon/skills/mod.rs +++ b/makima/src/daemon/skills/mod.rs @@ -9,12 +9,8 @@ pub const SUPERVISOR_SKILL: &str = include_str!("supervisor.md"); /// Contract skill content - task-contract interaction commands pub const CONTRACT_SKILL: &str = include_str!("contract.md"); -/// Directive skill content - directive orchestration commands -pub const DIRECTIVE_SKILL: &str = include_str!("directive.md"); - /// All skills as (name, content) pairs for installation pub const ALL_SKILLS: &[(&str, &str)] = &[ ("makima-supervisor", SUPERVISOR_SKILL), ("makima-contract", CONTRACT_SKILL), - ("makima-directive", DIRECTIVE_SKILL), ]; diff --git a/makima/src/db/models.rs b/makima/src/db/models.rs index 6045c7d..d0a0bd6 100644 --- a/makima/src/db/models.rs +++ b/makima/src/db/models.rs @@ -1446,16 +1446,6 @@ pub struct Contract { /// Use `get_phase_config()` to get the parsed PhaseConfig. #[serde(skip_serializing_if = "Option::is_none")] pub phase_config: Option<serde_json::Value>, - /// Directive ID if this contract is part of a directive's chain - #[serde(skip_serializing_if = "Option::is_none")] - pub directive_id: Option<Uuid>, - /// Whether this contract is a directive orchestrator - #[serde(default)] - #[sqlx(default)] - pub is_directive_orchestrator: bool, - /// Reference to directive spawned by this orchestrator contract - #[serde(skip_serializing_if = "Option::is_none")] - pub spawned_directive_id: Option<Uuid>, pub version: i32, pub created_at: DateTime<Utc>, pub updated_at: DateTime<Utc>, @@ -2692,326 +2682,3 @@ mod tests { } // ============================================================================= -// Directive Types -// ============================================================================= - -/// Default autonomy level for directives -fn default_autonomy_level() -> String { - "guardrails".to_string() -} - -/// Default empty JSON array -fn default_json_array() -> serde_json::Value { - serde_json::json!([]) -} - -/// Default empty JSON object -fn default_json_object() -> serde_json::Value { - serde_json::json!({}) -} - -/// Default confidence threshold (green) -fn default_confidence_green() -> f64 { - 0.85 -} - -/// Default confidence threshold (yellow) -fn default_confidence_yellow() -> f64 { - 0.60 -} - -/// Default max rework cycles -fn default_max_rework_cycles() -> Option<i32> { - Some(3) -} - -/// Default max chain regenerations -fn default_max_chain_regenerations() -> Option<i32> { - Some(2) -} - -/// Full directive row from the database. -#[derive(Debug, Clone, FromRow, Serialize, ToSchema)] -#[serde(rename_all = "camelCase")] -pub struct Directive { - pub id: Uuid, - pub owner_id: Uuid, - pub title: String, - pub goal: String, - #[sqlx(json)] - pub requirements: serde_json::Value, - #[sqlx(json)] - pub acceptance_criteria: serde_json::Value, - #[sqlx(json)] - pub constraints: serde_json::Value, - #[sqlx(json)] - pub external_dependencies: serde_json::Value, - pub status: String, - pub autonomy_level: String, - pub confidence_threshold_green: f64, - pub confidence_threshold_yellow: f64, - pub max_total_cost_usd: Option<f64>, - pub max_wall_time_minutes: Option<i32>, - pub max_rework_cycles: Option<i32>, - pub max_chain_regenerations: Option<i32>, - pub repository_url: Option<String>, - pub local_path: Option<String>, - pub base_branch: Option<String>, - pub orchestrator_contract_id: Option<Uuid>, - pub current_chain_id: Option<Uuid>, - pub chain_generation_count: i32, - pub total_cost_usd: f64, - pub started_at: Option<DateTime<Utc>>, - pub completed_at: Option<DateTime<Utc>>, - pub version: i32, - pub created_at: DateTime<Utc>, - pub updated_at: DateTime<Utc>, -} - -/// Summary of a directive for list views. -#[derive(Debug, Clone, FromRow, Serialize, ToSchema)] -#[serde(rename_all = "camelCase")] -pub struct DirectiveSummary { - pub id: Uuid, - pub title: String, - pub goal: String, - pub status: String, - pub autonomy_level: String, - pub chain_count: i64, - pub step_count: i64, - pub total_cost_usd: f64, - pub version: i32, - pub created_at: DateTime<Utc>, - pub updated_at: DateTime<Utc>, -} - -/// Response for directive list endpoint. -#[derive(Debug, Serialize, ToSchema)] -#[serde(rename_all = "camelCase")] -pub struct DirectiveListResponse { - pub directives: Vec<DirectiveSummary>, - pub total: i64, -} - -/// Request to create a new directive. -#[derive(Debug, Clone, Deserialize, ToSchema)] -#[serde(rename_all = "camelCase")] -pub struct CreateDirectiveRequest { - pub title: String, - pub goal: String, - #[serde(default = "default_json_array")] - pub requirements: serde_json::Value, - #[serde(default = "default_json_array")] - pub acceptance_criteria: serde_json::Value, - #[serde(default = "default_json_array")] - pub constraints: serde_json::Value, - #[serde(default = "default_json_array")] - pub external_dependencies: serde_json::Value, - #[serde(default = "default_autonomy_level")] - pub autonomy_level: String, - #[serde(default = "default_confidence_green")] - pub confidence_threshold_green: f64, - #[serde(default = "default_confidence_yellow")] - pub confidence_threshold_yellow: f64, - pub max_total_cost_usd: Option<f64>, - pub max_wall_time_minutes: Option<i32>, - #[serde(default = "default_max_rework_cycles")] - pub max_rework_cycles: Option<i32>, - #[serde(default = "default_max_chain_regenerations")] - pub max_chain_regenerations: Option<i32>, - pub repository_url: Option<String>, - pub local_path: Option<String>, - pub base_branch: Option<String>, -} - -/// Request to submit a chain plan for a directive. -#[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] -#[serde(rename_all = "camelCase")] -pub struct SubmitPlanRequest { - pub plan: String, -} - -/// Request to update an existing directive. -#[derive(Debug, Clone, Deserialize, ToSchema)] -#[serde(rename_all = "camelCase")] -pub struct UpdateDirectiveRequest { - pub title: Option<String>, - pub goal: Option<String>, - pub requirements: Option<serde_json::Value>, - pub acceptance_criteria: Option<serde_json::Value>, - pub constraints: Option<serde_json::Value>, - pub external_dependencies: Option<serde_json::Value>, - pub status: Option<String>, - pub autonomy_level: Option<String>, - pub confidence_threshold_green: Option<f64>, - pub confidence_threshold_yellow: Option<f64>, - pub max_total_cost_usd: Option<f64>, - pub max_wall_time_minutes: Option<i32>, - pub max_rework_cycles: Option<i32>, - pub max_chain_regenerations: Option<i32>, - pub repository_url: Option<String>, - pub local_path: Option<String>, - pub base_branch: Option<String>, - /// Version for optimistic locking - pub version: Option<i32>, -} - -/// Lightweight contract summary attached to a chain step. -#[derive(Debug, FromRow, Serialize, ToSchema)] -#[serde(rename_all = "camelCase")] -pub struct StepContractSummary { - pub id: Uuid, - pub name: String, - pub contract_type: String, - pub phase: String, - pub status: String, - pub task_count: i64, - pub tasks_done: i64, - pub tasks_running: i64, - pub tasks_failed: i64, -} - -/// Chain step enriched with optional contract summary. -#[derive(Debug, Serialize, ToSchema)] -#[serde(rename_all = "camelCase")] -pub struct ChainStepWithContract { - #[serde(flatten)] - pub step: ChainStep, - pub contract_summary: Option<StepContractSummary>, -} - -/// Directive with its chains and steps for detail view. -#[derive(Debug, Serialize, ToSchema)] -#[serde(rename_all = "camelCase")] -pub struct DirectiveWithChains { - #[serde(flatten)] - pub directive: Directive, - pub orchestrator_contract_summary: Option<StepContractSummary>, - pub chains: Vec<ChainWithSteps>, -} - -/// Full row from directive_chains table. -#[derive(Debug, Clone, FromRow, Serialize, ToSchema)] -#[serde(rename_all = "camelCase")] -pub struct DirectiveChain { - pub id: Uuid, - pub directive_id: Uuid, - pub generation: i32, - pub name: String, - pub description: Option<String>, - pub rationale: Option<String>, - pub planning_model: Option<String>, - pub status: String, - pub total_steps: i32, - pub completed_steps: i32, - pub failed_steps: i32, - pub current_confidence: Option<f64>, - pub started_at: Option<DateTime<Utc>>, - pub completed_at: Option<DateTime<Utc>>, - pub version: i32, - pub created_at: DateTime<Utc>, - pub updated_at: DateTime<Utc>, -} - -/// Full row from chain_steps table. -#[derive(Debug, Clone, FromRow, Serialize, ToSchema)] -#[serde(rename_all = "camelCase")] -pub struct ChainStep { - pub id: Uuid, - pub chain_id: Uuid, - pub name: String, - pub description: Option<String>, - pub step_type: String, - pub contract_type: String, - pub initial_phase: Option<String>, - pub task_plan: Option<String>, - pub phases: Option<Vec<String>>, - pub depends_on: Option<Vec<Uuid>>, - pub parallel_group: Option<String>, - pub requirement_ids: Option<Vec<String>>, - pub acceptance_criteria_ids: Option<Vec<String>>, - #[sqlx(json)] - pub verifier_config: serde_json::Value, - pub status: String, - pub contract_id: Option<Uuid>, - pub supervisor_task_id: Option<Uuid>, - pub monitoring_contract_id: Option<Uuid>, - pub monitoring_task_id: Option<Uuid>, - pub confidence_score: Option<f64>, - pub confidence_level: Option<String>, - pub evaluation_count: i32, - pub rework_count: i32, - pub last_evaluation_id: Option<Uuid>, - pub editor_x: Option<f64>, - pub editor_y: Option<f64>, - pub order_index: i32, - pub started_at: Option<DateTime<Utc>>, - pub completed_at: Option<DateTime<Utc>>, - pub created_at: DateTime<Utc>, -} - -/// Chain with its steps for detail view. -#[derive(Debug, Serialize, ToSchema)] -#[serde(rename_all = "camelCase")] -pub struct ChainWithSteps { - #[serde(flatten)] - pub chain: DirectiveChain, - pub steps: Vec<ChainStepWithContract>, -} - -/// Full row from directive_evaluations table. -#[derive(Debug, Clone, FromRow, Serialize, ToSchema)] -#[serde(rename_all = "camelCase")] -pub struct DirectiveEvaluation { - pub id: Uuid, - pub directive_id: Uuid, - pub chain_id: Option<Uuid>, - pub step_id: Option<Uuid>, - pub contract_id: Option<Uuid>, - pub evaluation_type: String, - pub evaluation_number: i32, - pub evaluator: Option<String>, - pub passed: bool, - pub overall_score: Option<f64>, - pub confidence_level: Option<String>, - #[sqlx(json)] - pub programmatic_results: serde_json::Value, - #[sqlx(json)] - pub llm_results: serde_json::Value, - #[sqlx(json)] - pub criteria_results: serde_json::Value, - pub summary_feedback: String, - pub rework_instructions: Option<String>, - #[sqlx(json)] - pub directive_snapshot: Option<serde_json::Value>, - #[sqlx(json)] - pub deliverables_snapshot: Option<serde_json::Value>, - pub started_at: DateTime<Utc>, - pub completed_at: Option<DateTime<Utc>>, - pub created_at: DateTime<Utc>, -} - -/// Full row from directive_events table. -#[derive(Debug, Clone, FromRow, Serialize, ToSchema)] -#[serde(rename_all = "camelCase")] -pub struct DirectiveEvent { - pub id: Uuid, - pub directive_id: Uuid, - pub chain_id: Option<Uuid>, - pub step_id: Option<Uuid>, - pub event_type: String, - pub severity: String, - #[sqlx(json)] - pub event_data: Option<serde_json::Value>, - pub actor_type: String, - pub actor_id: Option<Uuid>, - pub created_at: DateTime<Utc>, -} - -/// Response for evaluation list endpoint. -#[derive(Debug, Serialize, ToSchema)] -#[serde(rename_all = "camelCase")] -pub struct EvaluationListResponse { - pub evaluations: Vec<DirectiveEvaluation>, - pub total: i64, -} diff --git a/makima/src/db/repository.rs b/makima/src/db/repository.rs index 4298fa5..4ed2298 100644 --- a/makima/src/db/repository.rs +++ b/makima/src/db/repository.rs @@ -6,18 +6,17 @@ use sqlx::PgPool; use uuid::Uuid; use super::models::{ - ChainStep, CheckpointPatch, CheckpointPatchInfo, Contract, ContractChatConversation, + CheckpointPatch, CheckpointPatchInfo, Contract, ContractChatConversation, ContractChatMessageRecord, ContractEvent, ContractRepository, ContractSummary, ContractTypeTemplateRecord, ConversationMessage, ConversationSnapshot, - CreateContractRequest, CreateDirectiveRequest, CreateFileRequest, CreateTaskRequest, + CreateContractRequest, CreateFileRequest, CreateTaskRequest, CreateTemplateRequest, Daemon, DaemonTaskAssignment, DaemonWithCapacity, - DeliverableDefinition, Directive, DirectiveChain, DirectiveEvaluation, DirectiveEvent, - DirectiveSummary, + DeliverableDefinition, File, FileSummary, FileVersion, HistoryEvent, HistoryQueryFilters, MeshChatConversation, MeshChatMessageRecord, PhaseChangeResult, PhaseConfig, - PhaseDefinition, StepContractSummary, SupervisorHeartbeatRecord, SupervisorState, + PhaseDefinition, SupervisorHeartbeatRecord, SupervisorState, Task, TaskCheckpoint, TaskEvent, TaskSummary, UpdateContractRequest, - UpdateDirectiveRequest, UpdateFileRequest, UpdateTaskRequest, UpdateTemplateRequest, + UpdateFileRequest, UpdateTaskRequest, UpdateTemplateRequest, }; /// Repository error types. @@ -816,10 +815,7 @@ pub async fn get_pending_tasks_for_contract( WHERE t.contract_id = $1 AND t.owner_id = $2 AND t.status = 'pending' AND t.retry_count < t.max_retries - AND (t.is_supervisor = false - OR EXISTS (SELECT 1 FROM contracts c - WHERE c.id = t.contract_id - AND (c.directive_id IS NOT NULL OR c.is_directive_orchestrator = true))) + AND t.is_supervisor = false ORDER BY t.interrupted_at DESC NULLS LAST, t.priority DESC, @@ -844,10 +840,7 @@ pub async fn get_all_pending_task_contracts( WHERE t.contract_id IS NOT NULL AND t.status = 'pending' AND t.retry_count < t.max_retries - AND (t.is_supervisor = false - OR EXISTS (SELECT 1 FROM contracts c - WHERE c.id = t.contract_id - AND (c.directive_id IS NOT NULL OR c.is_directive_orchestrator = true))) + AND t.is_supervisor = false ORDER BY t.owner_id, t.contract_id "#, ) @@ -4919,870 +4912,3 @@ fn truncate_string(s: &str, max_len: usize) -> String { } } -// ============================================================================= -// Directive CRUD -// ============================================================================= - -/// Create a new directive, scoped to owner. -pub async fn create_directive_for_owner( - pool: &PgPool, - owner_id: Uuid, - req: CreateDirectiveRequest, -) -> Result<Directive, sqlx::Error> { - sqlx::query_as::<_, Directive>( - r#" - INSERT INTO directives ( - owner_id, title, goal, - requirements, acceptance_criteria, constraints, external_dependencies, - autonomy_level, confidence_threshold_green, confidence_threshold_yellow, - max_total_cost_usd, max_wall_time_minutes, max_rework_cycles, max_chain_regenerations, - repository_url, local_path, base_branch - ) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17) - RETURNING * - "#, - ) - .bind(owner_id) - .bind(&req.title) - .bind(&req.goal) - .bind(&req.requirements) - .bind(&req.acceptance_criteria) - .bind(&req.constraints) - .bind(&req.external_dependencies) - .bind(&req.autonomy_level) - .bind(req.confidence_threshold_green) - .bind(req.confidence_threshold_yellow) - .bind(req.max_total_cost_usd) - .bind(req.max_wall_time_minutes) - .bind(req.max_rework_cycles) - .bind(req.max_chain_regenerations) - .bind(&req.repository_url) - .bind(&req.local_path) - .bind(&req.base_branch) - .fetch_one(pool) - .await -} - -/// Get a directive by ID, scoped to owner. -pub async fn get_directive_for_owner( - pool: &PgPool, - id: Uuid, - owner_id: Uuid, -) -> Result<Option<Directive>, sqlx::Error> { - sqlx::query_as::<_, Directive>( - r#" - SELECT * - FROM directives - WHERE id = $1 AND owner_id = $2 - "#, - ) - .bind(id) - .bind(owner_id) - .fetch_optional(pool) - .await -} - -/// List all directives for an owner, ordered by created_at DESC. -pub async fn list_directives_for_owner( - pool: &PgPool, - owner_id: Uuid, -) -> Result<Vec<DirectiveSummary>, sqlx::Error> { - sqlx::query_as::<_, DirectiveSummary>( - r#" - SELECT - d.id, d.title, d.goal, d.status, d.autonomy_level, - (SELECT COUNT(*) FROM directive_chains WHERE directive_id = d.id) as chain_count, - (SELECT COUNT(*) FROM chain_steps cs JOIN directive_chains dc ON cs.chain_id = dc.id WHERE dc.directive_id = d.id) as step_count, - d.total_cost_usd, d.version, d.created_at, d.updated_at - FROM directives d - WHERE d.owner_id = $1 - ORDER BY d.created_at DESC - "#, - ) - .bind(owner_id) - .fetch_all(pool) - .await -} - -/// Update a directive by ID with optimistic locking, scoped to owner. -pub async fn update_directive_for_owner( - pool: &PgPool, - id: Uuid, - owner_id: Uuid, - req: UpdateDirectiveRequest, -) -> Result<Option<Directive>, RepositoryError> { - let existing = get_directive_for_owner(pool, id, owner_id).await?; - let Some(existing) = existing else { - return Ok(None); - }; - - // Check version if provided (optimistic locking) - if let Some(expected_version) = req.version { - if existing.version != expected_version { - return Err(RepositoryError::VersionConflict { - expected: expected_version, - actual: existing.version, - }); - } - } - - // Apply updates - let title = req.title.unwrap_or(existing.title); - let goal = req.goal.unwrap_or(existing.goal); - let requirements = req.requirements.unwrap_or(existing.requirements); - let acceptance_criteria = req.acceptance_criteria.unwrap_or(existing.acceptance_criteria); - let constraints = req.constraints.unwrap_or(existing.constraints); - let external_dependencies = req.external_dependencies.unwrap_or(existing.external_dependencies); - let status = req.status.unwrap_or(existing.status); - let autonomy_level = req.autonomy_level.unwrap_or(existing.autonomy_level); - let confidence_threshold_green = req.confidence_threshold_green.unwrap_or(existing.confidence_threshold_green); - let confidence_threshold_yellow = req.confidence_threshold_yellow.unwrap_or(existing.confidence_threshold_yellow); - let max_total_cost_usd = req.max_total_cost_usd.or(existing.max_total_cost_usd); - let max_wall_time_minutes = req.max_wall_time_minutes.or(existing.max_wall_time_minutes); - let max_rework_cycles = req.max_rework_cycles.or(existing.max_rework_cycles); - let max_chain_regenerations = req.max_chain_regenerations.or(existing.max_chain_regenerations); - let repository_url = req.repository_url.or(existing.repository_url); - let local_path = req.local_path.or(existing.local_path); - let base_branch = req.base_branch.or(existing.base_branch); - - let result = if req.version.is_some() { - sqlx::query_as::<_, Directive>( - r#" - UPDATE directives - SET title = $3, goal = $4, - requirements = $5, acceptance_criteria = $6, constraints = $7, external_dependencies = $8, - status = $9, autonomy_level = $10, - confidence_threshold_green = $11, confidence_threshold_yellow = $12, - max_total_cost_usd = $13, max_wall_time_minutes = $14, - max_rework_cycles = $15, max_chain_regenerations = $16, - repository_url = $17, local_path = $18, base_branch = $19, - version = version + 1, updated_at = NOW() - WHERE id = $1 AND owner_id = $2 AND version = $20 - RETURNING * - "#, - ) - .bind(id) - .bind(owner_id) - .bind(&title) - .bind(&goal) - .bind(&requirements) - .bind(&acceptance_criteria) - .bind(&constraints) - .bind(&external_dependencies) - .bind(&status) - .bind(&autonomy_level) - .bind(confidence_threshold_green) - .bind(confidence_threshold_yellow) - .bind(max_total_cost_usd) - .bind(max_wall_time_minutes) - .bind(max_rework_cycles) - .bind(max_chain_regenerations) - .bind(&repository_url) - .bind(&local_path) - .bind(&base_branch) - .bind(req.version.unwrap()) - .fetch_optional(pool) - .await? - } else { - sqlx::query_as::<_, Directive>( - r#" - UPDATE directives - SET title = $3, goal = $4, - requirements = $5, acceptance_criteria = $6, constraints = $7, external_dependencies = $8, - status = $9, autonomy_level = $10, - confidence_threshold_green = $11, confidence_threshold_yellow = $12, - max_total_cost_usd = $13, max_wall_time_minutes = $14, - max_rework_cycles = $15, max_chain_regenerations = $16, - repository_url = $17, local_path = $18, base_branch = $19, - version = version + 1, updated_at = NOW() - WHERE id = $1 AND owner_id = $2 - RETURNING * - "#, - ) - .bind(id) - .bind(owner_id) - .bind(&title) - .bind(&goal) - .bind(&requirements) - .bind(&acceptance_criteria) - .bind(&constraints) - .bind(&external_dependencies) - .bind(&status) - .bind(&autonomy_level) - .bind(confidence_threshold_green) - .bind(confidence_threshold_yellow) - .bind(max_total_cost_usd) - .bind(max_wall_time_minutes) - .bind(max_rework_cycles) - .bind(max_chain_regenerations) - .bind(&repository_url) - .bind(&local_path) - .bind(&base_branch) - .fetch_optional(pool) - .await? - }; - - // If versioned update returned None, there was a race condition - if result.is_none() && req.version.is_some() { - if let Some(current) = get_directive_for_owner(pool, id, owner_id).await? { - return Err(RepositoryError::VersionConflict { - expected: req.version.unwrap(), - actual: current.version, - }); - } - } - - Ok(result) -} - -/// Delete a directive by ID, scoped to owner. -/// Also deletes all contracts (and their cascaded tasks/files) associated with this directive. -pub async fn delete_directive_for_owner( - pool: &PgPool, - id: Uuid, - owner_id: Uuid, -) -> Result<bool, sqlx::Error> { - // First verify the directive exists and belongs to the owner - let directive = get_directive_for_owner(pool, id, owner_id).await?; - let Some(_directive) = directive else { - return Ok(false); - }; - - // Delete all contracts linked to this directive (tasks/files cascade from contracts). - // This covers step contracts (directive_id FK) and the orchestrator contract. - sqlx::query( - r#" - DELETE FROM contracts - WHERE directive_id = $1 - OR id = (SELECT orchestrator_contract_id FROM directives WHERE id = $1) - "#, - ) - .bind(id) - .execute(pool) - .await?; - - // Now delete the directive itself (chains, steps, events, evaluations cascade via FK) - let result = sqlx::query( - r#" - DELETE FROM directives - WHERE id = $1 AND owner_id = $2 - "#, - ) - .bind(id) - .bind(owner_id) - .execute(pool) - .await?; - - Ok(result.rows_affected() > 0) -} - -/// List chains for a directive (read-only). -pub async fn list_chains_for_directive( - pool: &PgPool, - directive_id: Uuid, -) -> Result<Vec<DirectiveChain>, sqlx::Error> { - sqlx::query_as::<_, DirectiveChain>( - r#" - SELECT * - FROM directive_chains - WHERE directive_id = $1 - ORDER BY generation DESC, created_at DESC - "#, - ) - .bind(directive_id) - .fetch_all(pool) - .await -} - -/// List steps for a chain (read-only). -pub async fn list_steps_for_chain( - pool: &PgPool, - chain_id: Uuid, -) -> Result<Vec<ChainStep>, sqlx::Error> { - sqlx::query_as::<_, ChainStep>( - r#" - SELECT * - FROM chain_steps - WHERE chain_id = $1 - ORDER BY order_index ASC, created_at ASC - "#, - ) - .bind(chain_id) - .fetch_all(pool) - .await -} - -/// Batch-fetch lightweight contract summaries for a set of contract IDs. -pub async fn get_contract_summaries_batch( - pool: &PgPool, - contract_ids: &[Uuid], -) -> Result<Vec<StepContractSummary>, sqlx::Error> { - sqlx::query_as::<_, StepContractSummary>( - r#" - SELECT c.id, c.name, c.contract_type, c.phase, c.status, - COUNT(t.id) as task_count, - COUNT(t.id) FILTER (WHERE t.status IN ('done','merged')) as tasks_done, - COUNT(t.id) FILTER (WHERE t.status IN ('running','initializing','starting')) as tasks_running, - COUNT(t.id) FILTER (WHERE t.status = 'failed') as tasks_failed - FROM contracts c - LEFT JOIN tasks t ON t.contract_id = c.id - WHERE c.id = ANY($1) - GROUP BY c.id, c.name, c.contract_type, c.phase, c.status - "#, - ) - .bind(contract_ids) - .fetch_all(pool) - .await -} - -// ── Directive orchestration functions ─────────────────────────────────────── - -/// Update directive status with automatic timestamp management. -pub async fn update_directive_status( - pool: &PgPool, - id: Uuid, - new_status: &str, -) -> Result<Option<Directive>, sqlx::Error> { - sqlx::query_as::<_, Directive>( - r#" - UPDATE directives - SET status = $2, - started_at = CASE WHEN $2 = 'active' AND started_at IS NULL THEN NOW() ELSE started_at END, - completed_at = CASE WHEN $2 IN ('completed', 'failed') THEN NOW() ELSE completed_at END, - version = version + 1, - updated_at = NOW() - WHERE id = $1 - RETURNING * - "#, - ) - .bind(id) - .bind(new_status) - .fetch_optional(pool) - .await -} - -/// Set the orchestrator contract ID on a directive. -pub async fn set_directive_orchestrator_contract( - pool: &PgPool, - directive_id: Uuid, - contract_id: Uuid, -) -> Result<Option<Directive>, sqlx::Error> { - sqlx::query_as::<_, Directive>( - r#" - UPDATE directives - SET orchestrator_contract_id = $2, - version = version + 1, - updated_at = NOW() - WHERE id = $1 - RETURNING * - "#, - ) - .bind(directive_id) - .bind(contract_id) - .fetch_optional(pool) - .await -} - -/// Set the current chain ID on a directive and increment chain_generation_count. -pub async fn set_directive_current_chain( - pool: &PgPool, - directive_id: Uuid, - chain_id: Uuid, -) -> Result<Option<Directive>, sqlx::Error> { - sqlx::query_as::<_, Directive>( - r#" - UPDATE directives - SET current_chain_id = $2, - chain_generation_count = chain_generation_count + 1, - version = version + 1, - updated_at = NOW() - WHERE id = $1 - RETURNING * - "#, - ) - .bind(directive_id) - .bind(chain_id) - .fetch_optional(pool) - .await -} - -/// Increment the chain_generation_count on a directive (without setting current_chain_id). -pub async fn increment_chain_generation_count( - pool: &PgPool, - directive_id: Uuid, -) -> Result<Option<Directive>, sqlx::Error> { - sqlx::query_as::<_, Directive>( - r#" - UPDATE directives - SET chain_generation_count = chain_generation_count + 1, - version = version + 1, - updated_at = NOW() - WHERE id = $1 - RETURNING * - "#, - ) - .bind(directive_id) - .fetch_optional(pool) - .await -} - -/// Create a new directive chain. -pub async fn create_directive_chain( - pool: &PgPool, - directive_id: Uuid, - name: &str, - description: Option<&str>, - rationale: Option<&str>, - total_steps: i32, -) -> Result<DirectiveChain, sqlx::Error> { - // Get next generation number - let next_gen: (i32,) = sqlx::query_as( - "SELECT COALESCE(MAX(generation), 0) + 1 FROM directive_chains WHERE directive_id = $1", - ) - .bind(directive_id) - .fetch_one(pool) - .await?; - - sqlx::query_as::<_, DirectiveChain>( - r#" - INSERT INTO directive_chains (directive_id, generation, name, description, rationale, total_steps, status) - VALUES ($1, $2, $3, $4, $5, $6, 'running') - RETURNING * - "#, - ) - .bind(directive_id) - .bind(next_gen.0) - .bind(name) - .bind(description) - .bind(rationale) - .bind(total_steps) - .fetch_one(pool) - .await -} - -/// Create a chain step. -pub async fn create_chain_step( - pool: &PgPool, - chain_id: Uuid, - name: &str, - description: Option<&str>, - step_type: &str, - contract_type: &str, - initial_phase: Option<&str>, - task_plan: Option<&str>, - depends_on: Option<Vec<Uuid>>, - order_index: i32, -) -> Result<ChainStep, sqlx::Error> { - sqlx::query_as::<_, ChainStep>( - r#" - INSERT INTO chain_steps (chain_id, name, description, step_type, contract_type, initial_phase, task_plan, depends_on, order_index, status) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 'pending') - RETURNING * - "#, - ) - .bind(chain_id) - .bind(name) - .bind(description) - .bind(step_type) - .bind(contract_type) - .bind(initial_phase) - .bind(task_plan) - .bind(depends_on.as_deref()) - .bind(order_index) - .fetch_one(pool) - .await -} - -/// Get a single chain step by ID. -pub async fn get_chain_step( - pool: &PgPool, - step_id: Uuid, -) -> Result<Option<ChainStep>, sqlx::Error> { - sqlx::query_as::<_, ChainStep>( - "SELECT * FROM chain_steps WHERE id = $1", - ) - .bind(step_id) - .fetch_optional(pool) - .await -} - -/// Increment completed_steps counter on a directive chain. -pub async fn increment_chain_completed_steps( - pool: &PgPool, - chain_id: Uuid, -) -> Result<(), sqlx::Error> { - sqlx::query( - "UPDATE directive_chains SET completed_steps = completed_steps + 1, updated_at = NOW() WHERE id = $1", - ) - .bind(chain_id) - .execute(pool) - .await?; - Ok(()) -} - -/// Increment failed_steps counter on a directive chain. -pub async fn increment_chain_failed_steps( - pool: &PgPool, - chain_id: Uuid, -) -> Result<(), sqlx::Error> { - sqlx::query( - "UPDATE directive_chains SET failed_steps = failed_steps + 1, updated_at = NOW() WHERE id = $1", - ) - .bind(chain_id) - .execute(pool) - .await?; - Ok(()) -} - -/// Update a chain step's status with automatic timestamp management. -pub async fn update_step_status( - pool: &PgPool, - step_id: Uuid, - new_status: &str, -) -> Result<Option<ChainStep>, sqlx::Error> { - sqlx::query_as::<_, ChainStep>( - r#" - UPDATE chain_steps - SET status = $2, - started_at = CASE WHEN $2 = 'running' AND started_at IS NULL THEN NOW() ELSE started_at END, - completed_at = CASE WHEN $2 IN ('passed', 'failed') THEN NOW() ELSE completed_at END - WHERE id = $1 - RETURNING * - "#, - ) - .bind(step_id) - .bind(new_status) - .fetch_optional(pool) - .await -} - -/// Link a chain step to a contract and supervisor task. -pub async fn update_step_contract( - pool: &PgPool, - step_id: Uuid, - contract_id: Uuid, - supervisor_task_id: Uuid, -) -> Result<Option<ChainStep>, sqlx::Error> { - sqlx::query_as::<_, ChainStep>( - r#" - UPDATE chain_steps - SET contract_id = $2, - supervisor_task_id = $3 - WHERE id = $1 - RETURNING * - "#, - ) - .bind(step_id) - .bind(contract_id) - .bind(supervisor_task_id) - .fetch_optional(pool) - .await -} - -/// Find steps that are ready to execute (pending, with all dependencies passed). -pub async fn find_ready_steps( - pool: &PgPool, - chain_id: Uuid, -) -> Result<Vec<ChainStep>, sqlx::Error> { - sqlx::query_as::<_, ChainStep>( - r#" - SELECT * FROM chain_steps - WHERE chain_id = $1 - AND status = 'pending' - AND ( - depends_on IS NULL - OR array_length(depends_on, 1) IS NULL - OR NOT EXISTS ( - SELECT 1 FROM unnest(depends_on) AS dep_id - WHERE dep_id NOT IN ( - SELECT id FROM chain_steps WHERE chain_id = $1 AND status = 'passed' - ) - ) - ) - ORDER BY order_index ASC - "#, - ) - .bind(chain_id) - .fetch_all(pool) - .await -} - -/// Get a chain step by its linked contract ID. -pub async fn get_step_by_contract_id( - pool: &PgPool, - contract_id: Uuid, -) -> Result<Option<ChainStep>, sqlx::Error> { - sqlx::query_as::<_, ChainStep>( - r#"SELECT * FROM chain_steps WHERE contract_id = $1"#, - ) - .bind(contract_id) - .fetch_optional(pool) - .await -} - -/// Get a directive by its orchestrator contract ID. -pub async fn get_directive_by_orchestrator_contract( - pool: &PgPool, - contract_id: Uuid, -) -> Result<Option<Directive>, sqlx::Error> { - sqlx::query_as::<_, Directive>( - r#"SELECT * FROM directives WHERE orchestrator_contract_id = $1"#, - ) - .bind(contract_id) - .fetch_optional(pool) - .await -} - -/// Set directive-related fields on a contract (directive_id, is_directive_orchestrator). -pub async fn set_contract_directive_fields( - pool: &PgPool, - contract_id: Uuid, - directive_id: Option<Uuid>, - is_orchestrator: bool, -) -> Result<(), sqlx::Error> { - sqlx::query( - r#" - UPDATE contracts - SET directive_id = $2, - is_directive_orchestrator = $3 - WHERE id = $1 - "#, - ) - .bind(contract_id) - .bind(directive_id) - .bind(is_orchestrator) - .execute(pool) - .await?; - Ok(()) -} - -/// Get a directive by ID (no owner scoping, for internal use). -pub async fn get_directive( - pool: &PgPool, - id: Uuid, -) -> Result<Option<Directive>, sqlx::Error> { - sqlx::query_as::<_, Directive>( - r#"SELECT * FROM directives WHERE id = $1"#, - ) - .bind(id) - .fetch_optional(pool) - .await -} - -/// Update chain status. -pub async fn update_chain_status( - pool: &PgPool, - chain_id: Uuid, - new_status: &str, -) -> Result<Option<DirectiveChain>, sqlx::Error> { - sqlx::query_as::<_, DirectiveChain>( - r#" - UPDATE directive_chains - SET status = $2, - completed_at = CASE WHEN $2 IN ('completed', 'failed') THEN NOW() ELSE completed_at END, - version = version + 1, - updated_at = NOW() - WHERE id = $1 - RETURNING * - "#, - ) - .bind(chain_id) - .bind(new_status) - .fetch_optional(pool) - .await -} - -// ── Directive monitoring / evaluation functions ───────────────────────────── - -/// Create a directive evaluation record. evaluation_number is auto-incremented per step. -pub async fn create_directive_evaluation( - pool: &PgPool, - directive_id: Uuid, - chain_id: Uuid, - step_id: Uuid, - contract_id: Uuid, - evaluation_type: &str, - evaluator: Option<&str>, - passed: bool, - overall_score: Option<f64>, - confidence_level: Option<&str>, - criteria_results: &serde_json::Value, - summary_feedback: &str, - rework_instructions: Option<&str>, -) -> Result<DirectiveEvaluation, sqlx::Error> { - sqlx::query_as::<_, DirectiveEvaluation>( - r#" - INSERT INTO directive_evaluations ( - directive_id, chain_id, step_id, contract_id, - evaluation_type, evaluation_number, evaluator, - passed, overall_score, confidence_level, - criteria_results, summary_feedback, rework_instructions, - completed_at - ) - VALUES ( - $1, $2, $3, $4, - $5, COALESCE((SELECT MAX(evaluation_number) FROM directive_evaluations WHERE step_id = $3), 0) + 1, $6, - $7, $8, $9, - $10, $11, $12, - NOW() - ) - RETURNING * - "#, - ) - .bind(directive_id) - .bind(chain_id) - .bind(step_id) - .bind(contract_id) - .bind(evaluation_type) - .bind(evaluator) - .bind(passed) - .bind(overall_score) - .bind(confidence_level) - .bind(criteria_results) - .bind(summary_feedback) - .bind(rework_instructions) - .fetch_one(pool) - .await -} - -/// List evaluations for a step, ordered by evaluation_number. -pub async fn list_evaluations_for_step( - pool: &PgPool, - step_id: Uuid, -) -> Result<Vec<DirectiveEvaluation>, sqlx::Error> { - sqlx::query_as::<_, DirectiveEvaluation>( - r#" - SELECT * FROM directive_evaluations - WHERE step_id = $1 - ORDER BY evaluation_number ASC - "#, - ) - .bind(step_id) - .fetch_all(pool) - .await -} - -/// Get a single directive evaluation by ID. -pub async fn get_directive_evaluation( - pool: &PgPool, - evaluation_id: Uuid, -) -> Result<Option<DirectiveEvaluation>, sqlx::Error> { - sqlx::query_as::<_, DirectiveEvaluation>( - "SELECT * FROM directive_evaluations WHERE id = $1", - ) - .bind(evaluation_id) - .fetch_optional(pool) - .await -} - -/// Create a directive event. -pub async fn create_directive_event( - pool: &PgPool, - directive_id: Uuid, - chain_id: Option<Uuid>, - step_id: Option<Uuid>, - event_type: &str, - severity: &str, - event_data: Option<&serde_json::Value>, - actor_type: &str, - actor_id: Option<Uuid>, -) -> Result<DirectiveEvent, sqlx::Error> { - sqlx::query_as::<_, DirectiveEvent>( - r#" - INSERT INTO directive_events (directive_id, chain_id, step_id, event_type, severity, event_data, actor_type, actor_id) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8) - RETURNING * - "#, - ) - .bind(directive_id) - .bind(chain_id) - .bind(step_id) - .bind(event_type) - .bind(severity) - .bind(event_data) - .bind(actor_type) - .bind(actor_id) - .fetch_one(pool) - .await -} - -/// Update step evaluation fields after an evaluation completes. -pub async fn update_step_evaluation_fields( - pool: &PgPool, - step_id: Uuid, - confidence_score: Option<f64>, - confidence_level: Option<&str>, - last_evaluation_id: Uuid, -) -> Result<Option<ChainStep>, sqlx::Error> { - sqlx::query_as::<_, ChainStep>( - r#" - UPDATE chain_steps - SET confidence_score = $2, - confidence_level = $3, - evaluation_count = evaluation_count + 1, - last_evaluation_id = $4 - WHERE id = $1 - RETURNING * - "#, - ) - .bind(step_id) - .bind(confidence_score) - .bind(confidence_level) - .bind(last_evaluation_id) - .fetch_optional(pool) - .await -} - -/// Update step monitoring contract/task references. -pub async fn update_step_monitoring_contract( - pool: &PgPool, - step_id: Uuid, - monitoring_contract_id: Uuid, - monitoring_task_id: Uuid, -) -> Result<Option<ChainStep>, sqlx::Error> { - sqlx::query_as::<_, ChainStep>( - r#" - UPDATE chain_steps - SET monitoring_contract_id = $2, - monitoring_task_id = $3 - WHERE id = $1 - RETURNING * - "#, - ) - .bind(step_id) - .bind(monitoring_contract_id) - .bind(monitoring_task_id) - .fetch_optional(pool) - .await -} - -/// Increment step rework_count. -pub async fn increment_step_rework_count( - pool: &PgPool, - step_id: Uuid, -) -> Result<Option<ChainStep>, sqlx::Error> { - sqlx::query_as::<_, ChainStep>( - r#" - UPDATE chain_steps - SET rework_count = rework_count + 1 - WHERE id = $1 - RETURNING * - "#, - ) - .bind(step_id) - .fetch_optional(pool) - .await -} - -/// Get a chain step by its monitoring contract ID. -pub async fn get_step_by_monitoring_contract_id( - pool: &PgPool, - contract_id: Uuid, -) -> Result<Option<ChainStep>, sqlx::Error> { - sqlx::query_as::<_, ChainStep>( - r#"SELECT * FROM chain_steps WHERE monitoring_contract_id = $1"#, - ) - .bind(contract_id) - .fetch_optional(pool) - .await -} diff --git a/makima/src/orchestration/directive.rs b/makima/src/orchestration/directive.rs deleted file mode 100644 index 46d9425..0000000 --- a/makima/src/orchestration/directive.rs +++ /dev/null @@ -1,1685 +0,0 @@ -//! Directive orchestration — init, planning completion, chain advancement. - -use serde::Deserialize; -use sqlx::PgPool; -use uuid::Uuid; - -use serde::Serialize; -use crate::db::models::{ - ChainStep, CreateContractRequest, CreateTaskRequest, Directive, Task, - UpdateContractRequest, UpdateTaskRequest, -}; -use crate::db::repository; -use crate::server::state::{DaemonCommand, SharedState}; - -/// A single step in the chain plan produced by the planning supervisor. -#[derive(Debug, Deserialize)] -#[serde(rename_all = "snake_case")] -struct ChainPlanStep { - name: String, - description: String, - #[serde(alias = "taskPlan")] - task_plan: String, - #[serde(default, alias = "dependsOn")] - depends_on: Vec<String>, // names of steps this depends on -} - -/// Wrapper for the plan JSON written by the planning supervisor. -#[derive(Debug, Deserialize)] -struct ChainPlan { - steps: Vec<ChainPlanStep>, -} - -/// Result written by the monitoring supervisor after evaluating a step. -#[derive(Debug, Deserialize, Serialize)] -#[serde(rename_all = "camelCase")] -struct MonitoringResult { - passed: bool, - overall_score: Option<f64>, - confidence_level: Option<String>, - #[serde(default)] - criteria_results: serde_json::Value, - #[serde(default)] - summary_feedback: String, - rework_instructions: Option<String>, -} - -/// Dispatch a task to an available daemon. Finds a connected daemon with capacity, -/// assigns the task, and sends a SpawnTask command. -async fn dispatch_task_to_daemon( - pool: &PgPool, - state: &SharedState, - task: &Task, - contract_local_only: bool, - contract_auto_merge_local: bool, - owner_id: Uuid, -) -> Result<(), String> { - // Find available daemons - let daemons = repository::get_available_daemons_excluding(pool, owner_id, &[]) - .await - .map_err(|e| format!("Failed to get available daemons: {}", e))?; - - let available_daemon = daemons.iter().find(|d| { - d.current_task_count < d.max_concurrent_tasks - && state.daemon_connections.contains_key(&d.connection_id) - }); - - let daemon = match available_daemon { - Some(d) => d, - None => { - tracing::warn!( - task_id = %task.id, - "No daemon available to dispatch task — will be picked up by retry loop" - ); - return Ok(()); - } - }; - - // Assign task to daemon - let update_req = UpdateTaskRequest { - status: Some("starting".to_string()), - daemon_id: Some(daemon.id), - version: Some(task.version), - ..Default::default() - }; - - let updated = repository::update_task_for_owner(pool, task.id, owner_id, update_req) - .await - .map_err(|e| format!("Failed to assign task to daemon: {:?}", e))?; - - let Some(updated_task) = updated else { - return Err("Task not found when assigning to daemon".to_string()); - }; - - // Get repo URL from task or contract repositories - let repo_url = if let Some(url) = &updated_task.repository_url { - Some(url.clone()) - } else if let Some(contract_id) = updated_task.contract_id { - match repository::list_contract_repositories(pool, contract_id).await { - Ok(repos) => repos - .iter() - .find(|r| r.is_primary) - .or(repos.first()) - .and_then(|r| r.repository_url.clone().or_else(|| r.local_path.clone())), - Err(_) => None, - } - } else { - None - }; - - let cmd = DaemonCommand::SpawnTask { - task_id: updated_task.id, - task_name: updated_task.name.clone(), - plan: updated_task.plan.clone(), - repo_url, - base_branch: updated_task.base_branch.clone(), - target_branch: updated_task.target_branch.clone(), - parent_task_id: updated_task.parent_task_id, - depth: updated_task.depth, - is_orchestrator: false, - target_repo_path: updated_task.target_repo_path.clone(), - completion_action: updated_task.completion_action.clone(), - continue_from_task_id: updated_task.continue_from_task_id, - copy_files: updated_task.copy_files.as_ref().and_then(|v| serde_json::from_value(v.clone()).ok()), - contract_id: updated_task.contract_id, - is_supervisor: updated_task.is_supervisor, - autonomous_loop: updated_task.contract_id.is_some(), - resume_session: false, - conversation_history: None, - patch_data: None, - patch_base_sha: None, - local_only: contract_local_only, - auto_merge_local: contract_auto_merge_local, - supervisor_worktree_task_id: None, - }; - - if let Err(e) = state.send_daemon_command(daemon.id, cmd).await { - tracing::warn!( - task_id = %task.id, - daemon_id = %daemon.id, - error = %e, - "Failed to send spawn command — rolling back" - ); - let rollback = UpdateTaskRequest { - status: Some("pending".to_string()), - clear_daemon_id: true, - ..Default::default() - }; - let _ = repository::update_task_for_owner(pool, task.id, owner_id, rollback).await; - return Ok(()); // Non-fatal, retry loop will pick it up - } - - tracing::info!( - task_id = %task.id, - daemon_id = %daemon.id, - "Dispatched directive task to daemon" - ); - - Ok(()) -} - -/// Initialize a directive: create a planning contract and transition to "planning". -pub async fn init_directive( - pool: &PgPool, - state: &SharedState, - owner_id: Uuid, - directive_id: Uuid, -) -> Result<Directive, String> { - // 1. Get directive, verify status - let directive = repository::get_directive_for_owner(pool, directive_id, owner_id) - .await - .map_err(|e| format!("Failed to get directive: {}", e))? - .ok_or("Directive not found")?; - - if directive.status != "draft" { - return Err(format!( - "Directive must be in 'draft' status to start, current status: '{}'", - directive.status - )); - } - - // 2. Create planning contract - let contract = repository::create_contract_for_owner( - pool, - owner_id, - CreateContractRequest { - name: format!("{} - Planning", directive.title), - description: Some(format!( - "Planning contract for directive: {}", - directive.title - )), - contract_type: Some("simple".to_string()), - template_id: None, - initial_phase: Some("plan".to_string()), - autonomous_loop: Some(true), - phase_guard: None, - local_only: Some(true), - auto_merge_local: None, - }, - ) - .await - .map_err(|e| format!("Failed to create planning contract: {}", e))?; - - // 3. Mark contract as directive orchestrator - repository::set_contract_directive_fields(pool, contract.id, Some(directive_id), true) - .await - .map_err(|e| format!("Failed to set contract directive fields: {}", e))?; - - // 4. Build planning prompt - let planning_prompt = build_planning_prompt(&directive); - - // 5. Create supervisor task - let supervisor_task = repository::create_task_for_owner( - pool, - owner_id, - CreateTaskRequest { - contract_id: Some(contract.id), - name: format!("{} - Planner", directive.title), - description: Some("Decompose directive goal into executable chain steps".to_string()), - plan: planning_prompt, - parent_task_id: None, - is_supervisor: true, - priority: 10, - repository_url: directive.repository_url.clone(), - base_branch: directive.base_branch.clone(), - target_branch: None, - merge_mode: None, - target_repo_path: directive.local_path.clone(), - completion_action: None, - continue_from_task_id: None, - copy_files: None, - checkpoint_sha: None, - branched_from_task_id: None, - conversation_history: None, - supervisor_worktree_task_id: None, - }, - ) - .await - .map_err(|e| format!("Failed to create supervisor task: {}", e))?; - - // 6. Link supervisor to contract - repository::update_contract_for_owner( - pool, - contract.id, - owner_id, - UpdateContractRequest { - supervisor_task_id: Some(supervisor_task.id), - ..Default::default() - }, - ) - .await - .map_err(|e| match e { - crate::db::repository::RepositoryError::Database(e) => { - format!("Failed to link supervisor to contract: {}", e) - } - other => format!("Failed to link supervisor to contract: {:?}", other), - })?; - - // 7. Set orchestrator_contract_id on directive - repository::set_directive_orchestrator_contract(pool, directive_id, contract.id) - .await - .map_err(|e| format!("Failed to set orchestrator contract: {}", e))?; - - // 8. Transition directive to "planning" - let updated = repository::update_directive_status(pool, directive_id, "planning") - .await - .map_err(|e| format!("Failed to update directive status: {}", e))? - .ok_or("Directive not found after status update")?; - - // 9. Copy repo config to contract if repository_url is set - if let Some(ref repo_url) = directive.repository_url { - let _ = repository::add_remote_repository( - pool, - contract.id, - "directive-repo", - repo_url, - true, - ) - .await; - } else if let Some(ref local_path) = directive.local_path { - let _ = repository::add_local_repository( - pool, - contract.id, - "directive-repo", - local_path, - true, - ) - .await; - } - - tracing::info!( - directive_id = %directive_id, - contract_id = %contract.id, - task_id = %supervisor_task.id, - "Directive started: planning contract created" - ); - - // 10. Dispatch planning task to an available daemon immediately - dispatch_task_to_daemon( - pool, state, &supervisor_task, - contract.local_only, contract.auto_merge_local, - owner_id, - ).await?; - - Ok(updated) -} - -/// Submit a chain plan for a directive via the CLI/API (instead of file-based extraction). -pub async fn submit_plan( - pool: &PgPool, - state: &SharedState, - owner_id: Uuid, - directive_id: Uuid, - plan_json: &str, -) -> Result<Directive, String> { - // 1. Get directive, verify status - let directive = repository::get_directive_for_owner(pool, directive_id, owner_id) - .await - .map_err(|e| format!("Failed to get directive: {}", e))? - .ok_or("Directive not found")?; - - if directive.status != "planning" { - return Err(format!( - "Directive must be in 'planning' status to submit a plan, current status: '{}'", - directive.status - )); - } - - // 2. Idempotency: if current_chain_id already set, return existing directive - if directive.current_chain_id.is_some() { - tracing::info!( - directive_id = %directive_id, - "Plan already submitted (current_chain_id set), returning existing directive" - ); - return Ok(directive); - } - - // 3. Parse the plan JSON - let chain_plan: ChainPlan = serde_json::from_str(plan_json) - .map_err(|e| format!("Failed to parse chain plan JSON: {}", e))?; - - if chain_plan.steps.is_empty() { - return Err("Chain plan has no steps".to_string()); - } - - // 4. Create chain and steps, transition to active - create_chain_and_steps(pool, state, &directive, &chain_plan, owner_id).await?; - - // 5. Re-fetch and return the updated directive - let updated = repository::get_directive(pool, directive_id) - .await - .map_err(|e| format!("Failed to re-fetch directive: {}", e))? - .ok_or("Directive not found after plan submission")?; - - tracing::info!( - directive_id = %directive_id, - step_count = chain_plan.steps.len(), - "Plan submitted via API, directive now active" - ); - - Ok(updated) -} - -/// Called when any task completes — checks if it's directive-related and advances. -/// Called when a contract's status is updated to "completed" via the API. -/// This is the primary entry point for directive orchestration because supervisor -/// tasks do not send TaskComplete messages — they complete via contract status updates. -pub async fn on_contract_completed( - pool: &PgPool, - state: &SharedState, - contract: &crate::db::models::Contract, - owner_id: Uuid, -) -> Result<(), String> { - if contract.status != "completed" { - return Ok(()); - } - - if contract.is_directive_orchestrator { - let directive = - repository::get_directive_by_orchestrator_contract(pool, contract.id) - .await - .map_err(|e| format!("Failed to get directive by orchestrator: {}", e))?; - - if let Some(directive) = directive { - tracing::info!( - directive_id = %directive.id, - contract_id = %contract.id, - "Directive orchestrator contract completed, handling planning completion" - ); - handle_planning_completion(pool, state, &directive, owner_id).await?; - } else { - tracing::warn!( - contract_id = %contract.id, - "Directive orchestrator contract completed but no directive found" - ); - } - } else if let Some(directive_id) = contract.directive_id { - // Check if this is a monitoring contract - let monitoring_step = - repository::get_step_by_monitoring_contract_id(pool, contract.id) - .await - .map_err(|e| format!("Failed to check monitoring contract: {}", e))?; - - if let Some(step) = monitoring_step { - tracing::info!( - directive_id = %directive_id, - step_id = %step.id, - contract_id = %contract.id, - "Monitoring contract completed" - ); - process_monitoring_result(pool, state, contract, &step, owner_id).await?; - } else { - // Step contract completed - let step = repository::get_step_by_contract_id(pool, contract.id) - .await - .map_err(|e| format!("Failed to get step by contract: {}", e))?; - - if let Some(step) = step { - // Idempotency: only dispatch monitoring if step is still "running" - // (on_step_completed may also fire via the task path) - if step.status != "running" { - tracing::info!( - step_id = %step.id, - status = %step.status, - "Skipping step contract completion: step no longer running" - ); - return Ok(()); - } - - let directive = repository::get_directive(pool, directive_id) - .await - .map_err(|e| format!("Failed to get directive: {}", e))? - .ok_or("Directive not found")?; - - tracing::info!( - directive_id = %directive_id, - step_id = %step.id, - contract_id = %contract.id, - "Step contract completed, dispatching monitoring" - ); - - // Step contract completed successfully — dispatch monitoring - repository::update_step_status(pool, step.id, "evaluating") - .await - .map_err(|e| format!("Failed to update step status: {}", e))?; - - let _ = repository::create_directive_event( - pool, - directive.id, - directive.current_chain_id, - Some(step.id), - "step_evaluating", - "info", - None, - "system", - None, - ) - .await; - - dispatch_monitoring(pool, state, &directive, &step, contract, owner_id).await?; - } - } - } - - Ok(()) -} - -pub async fn on_task_completed( - pool: &PgPool, - state: &SharedState, - task: &Task, - owner_id: Uuid, -) -> Result<(), String> { - let Some(contract_id) = task.contract_id else { - return Ok(()); - }; - - let contract = repository::get_contract_for_owner(pool, contract_id, owner_id) - .await - .map_err(|e| format!("Failed to get contract: {}", e))?; - - let Some(contract) = contract else { - return Ok(()); - }; - - if contract.is_directive_orchestrator { - // This is a planning contract completion - let directive = - repository::get_directive_by_orchestrator_contract(pool, contract_id) - .await - .map_err(|e| format!("Failed to get directive by orchestrator: {}", e))?; - - if let Some(directive) = directive { - on_planning_completed(pool, state, &directive, task, owner_id).await?; - } - } else if contract.directive_id.is_some() { - // Check if this is a monitoring contract completion - let monitoring_step = - repository::get_step_by_monitoring_contract_id(pool, contract_id) - .await - .map_err(|e| format!("Failed to check monitoring contract: {}", e))?; - - if let Some(step) = monitoring_step { - on_monitoring_completed(pool, state, &contract, &step, task, owner_id).await?; - } else { - // This is a step contract completion - on_step_completed(pool, state, &contract, task, owner_id).await?; - } - } - - Ok(()) -} - -/// Handle planning task completion: parse chain plan, create steps, advance. -async fn on_planning_completed( - pool: &PgPool, - state: &SharedState, - directive: &Directive, - task: &Task, - owner_id: Uuid, -) -> Result<(), String> { - // If task failed, fail the directive - if task.status == "failed" { - tracing::warn!( - directive_id = %directive.id, - task_id = %task.id, - "Planning task failed, marking directive as failed" - ); - repository::update_directive_status(pool, directive.id, "failed") - .await - .map_err(|e| format!("Failed to update directive status: {}", e))?; - return Ok(()); - } - - // Only process when the supervisor task itself is done - if task.status != "done" || !task.is_supervisor { - return Ok(()); - } - - handle_planning_completion(pool, state, directive, owner_id).await -} - -/// Handle planning contract/task completion. -/// Checks if a plan was submitted via the CLI; if not, retries or fails. -async fn handle_planning_completion( - pool: &PgPool, - state: &SharedState, - directive: &Directive, - owner_id: Uuid, -) -> Result<(), String> { - // Re-fetch directive to check latest state - let current = repository::get_directive(pool, directive.id) - .await - .map_err(|e| format!("Failed to re-fetch directive: {}", e))? - .ok_or("Directive not found")?; - - // Idempotency: only process if still in "planning" status - if current.status != "planning" { - tracing::info!( - directive_id = %directive.id, - status = %current.status, - "Skipping handle_planning_completion: directive no longer in planning status" - ); - return Ok(()); - } - - // If plan was already submitted via CLI (current_chain_id is set), nothing to do - if current.current_chain_id.is_some() { - tracing::info!( - directive_id = %directive.id, - "Plan already submitted via CLI, skipping handle_planning_completion" - ); - return Ok(()); - } - - // No plan was submitted — check retry budget - let max_regenerations = current.max_chain_regenerations.unwrap_or(2); - if current.chain_generation_count < max_regenerations { - tracing::warn!( - directive_id = %directive.id, - attempt = current.chain_generation_count + 1, - max = max_regenerations, - "Planning completed without plan submission, retrying" - ); - - let _ = repository::create_directive_event( - pool, - directive.id, - None, - None, - "planning_retry", - "warn", - Some(&serde_json::json!({ - "attempt": current.chain_generation_count + 1, - "maxRegenerations": max_regenerations, - "reason": "Planning contract completed without submitting a plan" - })), - "system", - None, - ) - .await; - - // Increment generation count - repository::increment_chain_generation_count(pool, directive.id) - .await - .map_err(|e| format!("Failed to increment chain generation count: {}", e))?; - - // Reset to draft so init_directive can be called again - repository::update_directive_status(pool, directive.id, "draft") - .await - .map_err(|e| format!("Failed to reset directive status: {}", e))?; - - // Re-init planning - init_directive(pool, state, owner_id, directive.id).await?; - - Ok(()) - } else { - tracing::error!( - directive_id = %directive.id, - attempts = current.chain_generation_count, - max = max_regenerations, - "Planning failed: max regeneration attempts exhausted without plan submission" - ); - - let _ = repository::create_directive_event( - pool, - directive.id, - None, - None, - "planning_failed", - "error", - Some(&serde_json::json!({ - "attempts": current.chain_generation_count, - "maxRegenerations": max_regenerations, - "reason": "Max chain regeneration attempts exhausted without plan submission" - })), - "system", - None, - ) - .await; - - repository::update_directive_status(pool, directive.id, "failed") - .await - .map_err(|e| format!("Failed to update directive status: {}", e))?; - - Ok(()) - } -} - -/// Inner helper: create chain, steps, set current chain, transition to active, and advance. -/// Extracted so that `process_planning_result` can catch errors and mark the directive failed. -async fn create_chain_and_steps( - pool: &PgPool, - state: &SharedState, - directive: &Directive, - chain_plan: &ChainPlan, - owner_id: Uuid, -) -> Result<(), String> { - // Create chain - let chain = repository::create_directive_chain( - pool, - directive.id, - &format!("{} - Chain", directive.title), - Some("Auto-generated from planning"), - None, - chain_plan.steps.len() as i32, - ) - .await - .map_err(|e| format!("Failed to create directive chain: {}", e))?; - - // Create steps (two passes: first create all, then resolve dependencies) - let mut step_ids: Vec<(String, Uuid)> = Vec::new(); - - for (i, plan_step) in chain_plan.steps.iter().enumerate() { - let step = repository::create_chain_step( - pool, - chain.id, - &plan_step.name, - Some(&plan_step.description), - "task", - "simple", - Some("plan"), - Some(&plan_step.task_plan), - None, // dependencies set in second pass - i as i32, - ) - .await - .map_err(|e| format!("Failed to create chain step: {}", e))?; - - step_ids.push((plan_step.name.clone(), step.id)); - } - - // Second pass: resolve name-based dependencies to UUIDs and update - for (i, plan_step) in chain_plan.steps.iter().enumerate() { - if plan_step.depends_on.is_empty() { - continue; - } - - let dep_uuids: Vec<Uuid> = plan_step - .depends_on - .iter() - .filter_map(|dep_name| { - step_ids - .iter() - .find(|(name, _)| name == dep_name) - .map(|(_, id)| *id) - }) - .collect(); - - if !dep_uuids.is_empty() { - let step_id = step_ids[i].1; - sqlx::query( - "UPDATE chain_steps SET depends_on = $2 WHERE id = $1", - ) - .bind(step_id) - .bind(&dep_uuids) - .execute(pool) - .await - .map_err(|e| format!("Failed to update step dependencies: {}", e))?; - } - } - - // Set current chain on directive - repository::set_directive_current_chain(pool, directive.id, chain.id) - .await - .map_err(|e| format!("Failed to set current chain: {}", e))?; - - // Transition directive to active - let updated_directive = repository::update_directive_status(pool, directive.id, "active") - .await - .map_err(|e| format!("Failed to update directive status: {}", e))? - .ok_or("Directive not found after status update")?; - - tracing::info!( - directive_id = %directive.id, - chain_id = %chain.id, - step_count = chain_plan.steps.len(), - "Chain plan created, advancing chain" - ); - - // Advance chain to dispatch ready steps - advance_chain(pool, state, &updated_directive, owner_id).await -} - -/// Handle a step contract task completion. -async fn on_step_completed( - pool: &PgPool, - state: &SharedState, - contract: &crate::db::models::Contract, - task: &Task, - owner_id: Uuid, -) -> Result<(), String> { - // Only process supervisor task completions - if !task.is_supervisor { - return Ok(()); - } - - let Some(directive_id) = contract.directive_id else { - return Ok(()); - }; - - // Find the step linked to this contract - let step = repository::get_step_by_contract_id(pool, contract.id) - .await - .map_err(|e| format!("Failed to get step by contract: {}", e))?; - - let Some(step) = step else { - return Ok(()); - }; - - // Idempotency: only process if step is still "running" - // (on_contract_completed may also fire via the contract path) - if step.status != "running" { - tracing::info!( - step_id = %step.id, - status = %step.status, - "Skipping on_step_completed: step no longer running" - ); - return Ok(()); - } - - // Get the directive for threshold info - let directive = repository::get_directive(pool, directive_id) - .await - .map_err(|e| format!("Failed to get directive: {}", e))? - .ok_or("Directive not found")?; - - if task.status == "done" { - // Step task succeeded — dispatch monitoring evaluation - repository::update_step_status(pool, step.id, "evaluating") - .await - .map_err(|e| format!("Failed to update step status: {}", e))?; - - let _ = repository::create_directive_event( - pool, - directive.id, - directive.current_chain_id, - Some(step.id), - "step_evaluating", - "info", - None, - "system", - None, - ) - .await; - - tracing::info!( - directive_id = %directive_id, - step_id = %step.id, - step_name = %step.name, - "Step task done, dispatching monitoring evaluation" - ); - - dispatch_monitoring(pool, state, &directive, &step, contract, owner_id).await - } else { - // Step task failed — mark step failed and advance - repository::update_step_status(pool, step.id, "failed") - .await - .map_err(|e| format!("Failed to update step status: {}", e))?; - - let _ = repository::increment_chain_failed_steps(pool, step.chain_id).await; - - tracing::info!( - directive_id = %directive_id, - step_id = %step.id, - step_name = %step.name, - "Step failed" - ); - - advance_chain(pool, state, &directive, owner_id).await - } -} - -/// Check chain progress and dispatch ready steps or mark directive complete. -async fn advance_chain( - pool: &PgPool, - state: &SharedState, - directive: &Directive, - owner_id: Uuid, -) -> Result<(), String> { - let Some(chain_id) = directive.current_chain_id else { - return Ok(()); - }; - - let steps = repository::list_steps_for_chain(pool, chain_id) - .await - .map_err(|e| format!("Failed to list steps: {}", e))?; - - // Check if all steps passed - let all_passed = steps.iter().all(|s| s.status == "passed"); - if all_passed && !steps.is_empty() { - repository::update_chain_status(pool, chain_id, "completed") - .await - .map_err(|e| format!("Failed to update chain status: {}", e))?; - repository::update_directive_status(pool, directive.id, "completed") - .await - .map_err(|e| format!("Failed to update directive status: {}", e))?; - tracing::info!(directive_id = %directive.id, "Directive completed: all steps passed"); - return Ok(()); - } - - // Check if any step failed - let any_failed = steps.iter().any(|s| s.status == "failed"); - if any_failed { - repository::update_chain_status(pool, chain_id, "failed") - .await - .map_err(|e| format!("Failed to update chain status: {}", e))?; - repository::update_directive_status(pool, directive.id, "failed") - .await - .map_err(|e| format!("Failed to update directive status: {}", e))?; - tracing::info!(directive_id = %directive.id, "Directive failed: step failure detected"); - return Ok(()); - } - - // Find and dispatch ready steps - let ready_steps = repository::find_ready_steps(pool, chain_id) - .await - .map_err(|e| format!("Failed to find ready steps: {}", e))?; - - for step in ready_steps { - if let Err(e) = dispatch_step(pool, state, directive, &step, owner_id).await { - tracing::error!( - step_id = %step.id, - step_name = %step.name, - error = %e, - "Failed to dispatch step" - ); - } - } - - Ok(()) -} - -/// Dispatch a single chain step as a new contract with supervisor. -async fn dispatch_step( - pool: &PgPool, - state: &SharedState, - directive: &Directive, - step: &crate::db::models::ChainStep, - owner_id: Uuid, -) -> Result<(), String> { - // Mark step as running - repository::update_step_status(pool, step.id, "running") - .await - .map_err(|e| format!("Failed to update step status: {}", e))?; - - // Create contract for this step. - // Step contracts use the directive's repository config — not local_only, - // so they can branch and merge to share work across steps. - let has_repo = directive.repository_url.is_some() || directive.local_path.is_some(); - let contract = repository::create_contract_for_owner( - pool, - owner_id, - CreateContractRequest { - name: step.name.clone(), - description: step.description.clone(), - contract_type: Some(step.contract_type.clone()), - template_id: None, - initial_phase: step.initial_phase.clone(), - autonomous_loop: Some(true), - phase_guard: None, - local_only: Some(!has_repo), - auto_merge_local: if has_repo { Some(true) } else { None }, - }, - ) - .await - .map_err(|e| format!("Failed to create step contract: {}", e))?; - - // Set directive_id on contract - repository::set_contract_directive_fields(pool, contract.id, Some(directive.id), false) - .await - .map_err(|e| format!("Failed to set contract directive fields: {}", e))?; - - // Build the task plan, prepending rework instructions if this is a rework cycle - let mut task_plan = step - .task_plan - .clone() - .unwrap_or_else(|| format!("Execute step: {}", step.name)); - - if let Some(eval_id) = step.last_evaluation_id { - if let Ok(Some(evaluation)) = repository::get_directive_evaluation(pool, eval_id).await { - if let Some(ref rework) = evaluation.rework_instructions { - task_plan = format!( - "IMPORTANT — REWORK REQUIRED (attempt #{}):\n\ - The previous attempt was evaluated and did NOT pass.\n\ - Feedback: {}\n\ - Rework instructions: {}\n\n\ - ---\n\n\ - Original task plan:\n{}", - step.rework_count + 1, - evaluation.summary_feedback, - rework, - task_plan, - ); - } - } - } - - // Create supervisor task - let supervisor_task = repository::create_task_for_owner( - pool, - owner_id, - CreateTaskRequest { - contract_id: Some(contract.id), - name: format!("{} Supervisor", step.name), - description: step.description.clone(), - plan: task_plan, - parent_task_id: None, - is_supervisor: true, - priority: 5, - repository_url: directive.repository_url.clone(), - base_branch: directive.base_branch.clone(), - target_branch: None, - merge_mode: None, - target_repo_path: directive.local_path.clone(), - completion_action: None, - continue_from_task_id: None, - copy_files: None, - checkpoint_sha: None, - branched_from_task_id: None, - conversation_history: None, - supervisor_worktree_task_id: None, - }, - ) - .await - .map_err(|e| format!("Failed to create step supervisor task: {}", e))?; - - // Link supervisor to contract - repository::update_contract_for_owner( - pool, - contract.id, - owner_id, - UpdateContractRequest { - supervisor_task_id: Some(supervisor_task.id), - ..Default::default() - }, - ) - .await - .map_err(|e| match e { - crate::db::repository::RepositoryError::Database(e) => { - format!("Failed to link supervisor to step contract: {}", e) - } - other => format!("Failed to link supervisor to step contract: {:?}", other), - })?; - - // Link step to contract/task - repository::update_step_contract(pool, step.id, contract.id, supervisor_task.id) - .await - .map_err(|e| format!("Failed to update step contract link: {}", e))?; - - // Copy repo config from directive to step contract - if let Some(ref repo_url) = directive.repository_url { - let _ = repository::add_remote_repository( - pool, - contract.id, - "directive-repo", - repo_url, - true, - ) - .await; - } else if let Some(ref local_path) = directive.local_path { - let _ = repository::add_local_repository( - pool, - contract.id, - "directive-repo", - local_path, - true, - ) - .await; - } - - tracing::info!( - directive_id = %directive.id, - step_id = %step.id, - step_name = %step.name, - contract_id = %contract.id, - task_id = %supervisor_task.id, - "Step dispatched" - ); - - // Dispatch step task to an available daemon immediately - dispatch_task_to_daemon( - pool, state, &supervisor_task, - contract.local_only, contract.auto_merge_local, - owner_id, - ).await?; - - Ok(()) -} - -/// Build the planning supervisor prompt from a directive. -fn build_planning_prompt(directive: &Directive) -> String { - format!( - r#"You are planning the execution of a directive. - -DIRECTIVE: {title} -GOAL: {goal} -REQUIREMENTS: {requirements} -ACCEPTANCE CRITERIA: {acceptance_criteria} -CONSTRAINTS: {constraints} - -Your job is to decompose this goal into a sequence of executable steps. -Each step will become a separate contract with its own supervisor. - -The JSON format: -{{ - "steps": [ - {{ - "name": "Step name", - "description": "What this step accomplishes", - "task_plan": "Detailed instructions for the step's supervisor", - "depends_on": [] - }} - ] -}} - -Rules: -- Steps with no dependencies (empty depends_on array) will run in parallel. -- Steps that depend on other steps will wait until those complete. -- The depends_on array contains names of steps this step depends on. -- Each step should be a self-contained unit of work. -- Be specific in task_plan — include file paths, function names, and acceptance criteria where possible. -- Keep the number of steps reasonable (3-10 typically). - -Submit your plan by piping the JSON to stdin: - echo '<your_json_plan>' | makima directive submit-plan --directive-id {directive_id} - -After submitting the plan, mark the contract as complete: - makima supervisor complete"#, - title = directive.title, - goal = directive.goal, - requirements = serde_json::to_string_pretty(&directive.requirements).unwrap_or_default(), - acceptance_criteria = serde_json::to_string_pretty(&directive.acceptance_criteria).unwrap_or_default(), - constraints = serde_json::to_string_pretty(&directive.constraints).unwrap_or_default(), - directive_id = directive.id, - ) -} - -/// Extract JSON from file body elements. -fn extract_plan_json(body: &[crate::db::models::BodyElement]) -> Option<String> { - use crate::db::models::BodyElement; - - for element in body { - match element { - BodyElement::Code { content, .. } => { - // Try to parse as JSON - let trimmed = content.trim(); - if trimmed.starts_with('{') || trimmed.starts_with('[') { - if serde_json::from_str::<serde_json::Value>(trimmed).is_ok() { - return Some(trimmed.to_string()); - } - } - } - BodyElement::Paragraph { text } => { - let trimmed = text.trim(); - if trimmed.starts_with('{') || trimmed.starts_with('[') { - if serde_json::from_str::<serde_json::Value>(trimmed).is_ok() { - return Some(trimmed.to_string()); - } - } - } - BodyElement::Markdown { content } => { - // Try to find JSON in markdown content - let trimmed = content.trim(); - if trimmed.starts_with('{') || trimmed.starts_with('[') { - if serde_json::from_str::<serde_json::Value>(trimmed).is_ok() { - return Some(trimmed.to_string()); - } - } - // Try to find JSON in code blocks within markdown - if let Some(json_start) = trimmed.find("```json") { - let after = &trimmed[json_start + 7..]; - if let Some(json_end) = after.find("```") { - let json_str = after[..json_end].trim(); - if serde_json::from_str::<serde_json::Value>(json_str).is_ok() { - return Some(json_str.to_string()); - } - } - } - } - _ => {} - } - } - - // Fallback: concatenate all text content and try to find JSON - let all_text: String = body - .iter() - .map(|el| match el { - BodyElement::Code { content, .. } => content.clone(), - BodyElement::Paragraph { text } => text.clone(), - BodyElement::Markdown { content } => content.clone(), - _ => String::new(), - }) - .collect::<Vec<_>>() - .join("\n"); - - let trimmed = all_text.trim(); - if let Some(start) = trimmed.find('{') { - // Find matching closing brace - let substr = &trimmed[start..]; - if serde_json::from_str::<serde_json::Value>(substr).is_ok() { - return Some(substr.to_string()); - } - } - - None -} - -/// Dispatch a monitoring contract to evaluate a completed step. -async fn dispatch_monitoring( - pool: &PgPool, - state: &SharedState, - directive: &Directive, - step: &ChainStep, - step_contract: &crate::db::models::Contract, - owner_id: Uuid, -) -> Result<(), String> { - // Create monitoring contract - let contract = repository::create_contract_for_owner( - pool, - owner_id, - CreateContractRequest { - name: format!("{} - Monitor", step.name), - description: Some(format!("Monitoring evaluation for step: {}", step.name)), - contract_type: Some("monitoring".to_string()), - template_id: None, - initial_phase: Some("plan".to_string()), - autonomous_loop: Some(true), - phase_guard: None, - local_only: Some(true), - auto_merge_local: None, - }, - ) - .await - .map_err(|e| format!("Failed to create monitoring contract: {}", e))?; - - // Mark contract as directive-related (not orchestrator) - repository::set_contract_directive_fields(pool, contract.id, Some(directive.id), false) - .await - .map_err(|e| format!("Failed to set monitoring contract directive fields: {}", e))?; - - // Build evaluation prompt - let prompt = build_monitoring_prompt(directive, step, step_contract); - - // Create monitoring task (NOT a supervisor — regular task that exits when done, - // which triggers on_task_completed → on_monitoring_completed automatically) - let supervisor_task = repository::create_task_for_owner( - pool, - owner_id, - CreateTaskRequest { - contract_id: Some(contract.id), - name: format!("{} - Evaluator", step.name), - description: Some("Evaluate step output against directive criteria".to_string()), - plan: prompt, - parent_task_id: None, - is_supervisor: false, - priority: 8, - repository_url: directive.repository_url.clone(), - base_branch: directive.base_branch.clone(), - target_branch: None, - merge_mode: None, - target_repo_path: directive.local_path.clone(), - completion_action: None, - continue_from_task_id: None, - copy_files: None, - checkpoint_sha: None, - branched_from_task_id: None, - conversation_history: None, - supervisor_worktree_task_id: None, - }, - ) - .await - .map_err(|e| format!("Failed to create monitoring task: {}", e))?; - - // Link monitoring task to contract - repository::update_contract_for_owner( - pool, - contract.id, - owner_id, - UpdateContractRequest { - supervisor_task_id: Some(supervisor_task.id), - ..Default::default() - }, - ) - .await - .map_err(|e| match e { - crate::db::repository::RepositoryError::Database(e) => { - format!("Failed to link task to monitoring contract: {}", e) - } - other => format!("Failed to link task to monitoring contract: {:?}", other), - })?; - - // Link step to monitoring contract/task - repository::update_step_monitoring_contract(pool, step.id, contract.id, supervisor_task.id) - .await - .map_err(|e| format!("Failed to update step monitoring contract link: {}", e))?; - - // Copy repo config from directive to monitoring contract - if let Some(ref repo_url) = directive.repository_url { - let _ = repository::add_remote_repository( - pool, - contract.id, - "directive-repo", - repo_url, - true, - ) - .await; - } else if let Some(ref local_path) = directive.local_path { - let _ = repository::add_local_repository( - pool, - contract.id, - "directive-repo", - local_path, - true, - ) - .await; - } - - tracing::info!( - directive_id = %directive.id, - step_id = %step.id, - step_name = %step.name, - monitoring_contract_id = %contract.id, - monitoring_task_id = %supervisor_task.id, - "Monitoring evaluation dispatched" - ); - - // Dispatch monitoring task to an available daemon immediately - dispatch_task_to_daemon( - pool, state, &supervisor_task, - contract.local_only, contract.auto_merge_local, - owner_id, - ).await?; - - Ok(()) -} - -/// Build the monitoring supervisor prompt. -fn build_monitoring_prompt( - directive: &Directive, - step: &ChainStep, - step_contract: &crate::db::models::Contract, -) -> String { - format!( - r#"You are evaluating the output of a completed step in a directive chain. - -DIRECTIVE: {title} -GOAL: {goal} -REQUIREMENTS: {requirements} -ACCEPTANCE CRITERIA: {acceptance_criteria} -CONSTRAINTS: {constraints} - -STEP: {step_name} -STEP DESCRIPTION: {step_description} -STEP TASK PLAN: {task_plan} -STEP CONTRACT ID: {step_contract_id} - -CONFIDENCE THRESHOLDS: -- Green (pass): >= {threshold_green} -- Yellow (marginal): >= {threshold_yellow} -- Red (fail): < {threshold_yellow} - -INSTRUCTIONS: -1. Read the step contract's files to understand what was delivered: - makima contract files --contract-id {step_contract_id} - makima contract file <file_id> --contract-id {step_contract_id} - -2. Evaluate whether the step's output meets the directive's requirements and the step's task plan. - -3. Write your evaluation as a JSON file to this monitoring contract. Create a file called - evaluation.json with the JSON content first, then upload it: - - cat > /tmp/eval-result.json << 'EVALEOF' - {{ - "passed": true, - "overallScore": 0.85, - "confidenceLevel": "green", - "criteriaResults": [ - {{ - "criterion": "Example criterion", - "passed": true, - "score": 0.9, - "evidence": "What was found" - }} - ], - "summaryFeedback": "Summary of evaluation", - "reworkInstructions": null - }} - EVALEOF - makima contract create-file evaluation-result < /tmp/eval-result.json - - Replace the example values with your actual evaluation results. - -Scoring guidelines: -- overallScore >= {threshold_green}: confidenceLevel = "green", passed = true -- overallScore >= {threshold_yellow} and < {threshold_green}: confidenceLevel = "yellow", use judgment -- overallScore < {threshold_yellow}: confidenceLevel = "red", passed = false -- Be specific in reworkInstructions if the step fails — the step will be re-executed with these instructions. -- Set reworkInstructions to null if the step passed. - -You are done after writing the evaluation file."#, - title = directive.title, - goal = directive.goal, - requirements = serde_json::to_string_pretty(&directive.requirements).unwrap_or_default(), - acceptance_criteria = serde_json::to_string_pretty(&directive.acceptance_criteria).unwrap_or_default(), - constraints = serde_json::to_string_pretty(&directive.constraints).unwrap_or_default(), - step_name = step.name, - step_description = step.description.as_deref().unwrap_or("N/A"), - task_plan = step.task_plan.as_deref().unwrap_or("N/A"), - step_contract_id = step_contract.id, - threshold_green = directive.confidence_threshold_green, - threshold_yellow = directive.confidence_threshold_yellow, - ) -} - -/// Handle monitoring contract task completion — parse evaluation and decide step outcome. -async fn on_monitoring_completed( - pool: &PgPool, - state: &SharedState, - contract: &crate::db::models::Contract, - step: &ChainStep, - task: &Task, - owner_id: Uuid, -) -> Result<(), String> { - let Some(directive_id) = contract.directive_id else { - return Ok(()); - }; - - let directive = repository::get_directive(pool, directive_id) - .await - .map_err(|e| format!("Failed to get directive: {}", e))? - .ok_or("Directive not found")?; - - // If monitoring task itself failed, fail-open: mark step as passed - if task.status == "failed" { - tracing::warn!( - directive_id = %directive_id, - step_id = %step.id, - "Monitoring task failed, fail-open: marking step as passed" - ); - - repository::update_step_status(pool, step.id, "passed") - .await - .map_err(|e| format!("Failed to update step status: {}", e))?; - - let _ = repository::increment_chain_completed_steps(pool, step.chain_id).await; - - let _ = repository::create_directive_event( - pool, - directive_id, - directive.current_chain_id, - Some(step.id), - "monitoring_failed_open", - "warn", - None, - "system", - None, - ) - .await; - - return advance_chain(pool, state, &directive, owner_id).await; - } - - if task.status != "done" { - return Ok(()); - } - - process_monitoring_result(pool, state, contract, step, owner_id).await -} - -/// Core monitoring logic: read evaluation from files, create record, handle pass/fail/rework. -/// Called from both `on_monitoring_completed` (task path) and `on_contract_completed` (API path). -async fn process_monitoring_result( - pool: &PgPool, - state: &SharedState, - contract: &crate::db::models::Contract, - step: &ChainStep, - owner_id: Uuid, -) -> Result<(), String> { - let Some(directive_id) = contract.directive_id else { - return Ok(()); - }; - - // Idempotency guard: re-fetch step and only process if still "evaluating". - let current_step = repository::get_chain_step(pool, step.id) - .await - .map_err(|e| format!("Failed to re-fetch step: {}", e))?; - if let Some(ref s) = current_step { - if s.status != "evaluating" { - tracing::info!( - step_id = %step.id, - status = %s.status, - "Skipping process_monitoring_result: step no longer in evaluating status" - ); - return Ok(()); - } - } - - let directive = repository::get_directive(pool, directive_id) - .await - .map_err(|e| format!("Failed to get directive: {}", e))? - .ok_or("Directive not found")?; - - // Read evaluation result from monitoring contract files - let files = repository::list_files_in_contract(pool, contract.id, owner_id) - .await - .map_err(|e| format!("Failed to list monitoring contract files: {}", e))?; - - let eval_file = files.iter().find(|f| { - let name_lower = f.name.to_lowercase(); - name_lower.contains("evaluation") || name_lower.contains("eval") - }); - - let eval_file = eval_file.or_else(|| files.first()); - - let monitoring_result = if let Some(eval_file) = eval_file { - let full_file = repository::get_file(pool, eval_file.id) - .await - .map_err(|e| format!("Failed to get evaluation file: {}", e))?; - - if let Some(full_file) = full_file { - let json_str = extract_plan_json(&full_file.body); - json_str.and_then(|s| serde_json::from_str::<MonitoringResult>(&s).ok()) - } else { - None - } - } else { - None - }; - - // If we couldn't parse the result, fail-open - let Some(result) = monitoring_result else { - tracing::warn!( - directive_id = %directive_id, - step_id = %step.id, - "Could not parse monitoring result, fail-open: marking step as passed" - ); - - repository::update_step_status(pool, step.id, "passed") - .await - .map_err(|e| format!("Failed to update step status: {}", e))?; - - let _ = repository::increment_chain_completed_steps(pool, step.chain_id).await; - - let _ = repository::create_directive_event( - pool, - directive_id, - directive.current_chain_id, - Some(step.id), - "monitoring_parse_failed_open", - "warn", - None, - "system", - None, - ) - .await; - - return advance_chain(pool, state, &directive, owner_id).await; - }; - - // Create evaluation record - let chain_id = directive.current_chain_id.unwrap_or(step.chain_id); - let evaluation = repository::create_directive_evaluation( - pool, - directive_id, - chain_id, - step.id, - contract.id, - "monitoring", - Some("automated"), - result.passed, - result.overall_score, - result.confidence_level.as_deref(), - &result.criteria_results, - &result.summary_feedback, - result.rework_instructions.as_deref(), - ) - .await - .map_err(|e| format!("Failed to create directive evaluation: {}", e))?; - - // Update step evaluation fields - repository::update_step_evaluation_fields( - pool, - step.id, - result.overall_score, - result.confidence_level.as_deref(), - evaluation.id, - ) - .await - .map_err(|e| format!("Failed to update step evaluation fields: {}", e))?; - - // Create event - let event_data = serde_json::json!({ - "passed": result.passed, - "overallScore": result.overall_score, - "confidenceLevel": result.confidence_level, - "summaryFeedback": result.summary_feedback, - }); - let _ = repository::create_directive_event( - pool, - directive_id, - Some(chain_id), - Some(step.id), - if result.passed { "step_evaluation_passed" } else { "step_evaluation_failed" }, - "info", - Some(&event_data), - "system", - None, - ) - .await; - - if result.passed { - // Evaluation passed — mark step as passed - tracing::info!( - directive_id = %directive_id, - step_id = %step.id, - step_name = %step.name, - score = ?result.overall_score, - "Step evaluation passed" - ); - - repository::update_step_status(pool, step.id, "passed") - .await - .map_err(|e| format!("Failed to update step status: {}", e))?; - - let _ = repository::increment_chain_completed_steps(pool, step.chain_id).await; - - advance_chain(pool, state, &directive, owner_id).await - } else { - // Evaluation failed — check rework budget - let max_rework = directive.max_rework_cycles.unwrap_or(3); - if step.rework_count >= max_rework { - tracing::warn!( - directive_id = %directive_id, - step_id = %step.id, - step_name = %step.name, - rework_count = step.rework_count, - max_rework = max_rework, - "Step evaluation failed, max rework cycles exceeded" - ); - - repository::update_step_status(pool, step.id, "failed") - .await - .map_err(|e| format!("Failed to update step status: {}", e))?; - - let _ = repository::increment_chain_failed_steps(pool, step.chain_id).await; - - advance_chain(pool, state, &directive, owner_id).await - } else { - tracing::info!( - directive_id = %directive_id, - step_id = %step.id, - step_name = %step.name, - rework_count = step.rework_count, - "Step evaluation failed, scheduling rework" - ); - - repository::increment_step_rework_count(pool, step.id) - .await - .map_err(|e| format!("Failed to increment rework count: {}", e))?; - - // Set step back to pending so advance_chain re-dispatches it - repository::update_step_status(pool, step.id, "pending") - .await - .map_err(|e| format!("Failed to update step status: {}", e))?; - - advance_chain(pool, state, &directive, owner_id).await - } - } -} - -/// Trigger a manual evaluation for a step. Public for use by handlers. -pub async fn trigger_manual_evaluation( - pool: &PgPool, - state: &SharedState, - owner_id: Uuid, - directive_id: Uuid, - step_id: Uuid, -) -> Result<ChainStep, String> { - let directive = repository::get_directive_for_owner(pool, directive_id, owner_id) - .await - .map_err(|e| format!("Failed to get directive: {}", e))? - .ok_or("Directive not found")?; - - // Get the step — find via chain steps - let chain_id = directive.current_chain_id.ok_or("Directive has no active chain")?; - let steps = repository::list_steps_for_chain(pool, chain_id) - .await - .map_err(|e| format!("Failed to list steps: {}", e))?; - - let step = steps - .into_iter() - .find(|s| s.id == step_id) - .ok_or("Step not found in current chain")?; - - // Step must have a contract_id (must have been executed) - let contract_id = step.contract_id.ok_or("Step has no contract — it hasn't been executed yet")?; - - let contract = repository::get_contract_for_owner(pool, contract_id, owner_id) - .await - .map_err(|e| format!("Failed to get step contract: {}", e))? - .ok_or("Step contract not found")?; - - // Set step to evaluating - let updated_step = repository::update_step_status(pool, step.id, "evaluating") - .await - .map_err(|e| format!("Failed to update step status: {}", e))? - .ok_or("Step not found after status update")?; - - let _ = repository::create_directive_event( - pool, - directive.id, - directive.current_chain_id, - Some(step.id), - "manual_evaluation_triggered", - "info", - None, - "user", - None, - ) - .await; - - dispatch_monitoring(pool, state, &directive, &step, &contract, owner_id).await?; - - Ok(updated_step) -} diff --git a/makima/src/orchestration/mod.rs b/makima/src/orchestration/mod.rs index e7ffb70..8b13789 100644 --- a/makima/src/orchestration/mod.rs +++ b/makima/src/orchestration/mod.rs @@ -1 +1 @@ -pub mod directive; + diff --git a/makima/src/server/handlers/contracts.rs b/makima/src/server/handlers/contracts.rs index ad0a1ff..dc15923 100644 --- a/makima/src/server/handlers/contracts.rs +++ b/makima/src/server/handlers/contracts.rs @@ -575,24 +575,6 @@ pub async fn update_contract( }), ).await; - // Directive engine integration — process planning/step/monitoring completion - if contract.is_directive_orchestrator || contract.directive_id.is_some() { - let pool_clone = pool.clone(); - let state_clone = state.clone(); - let contract_clone = contract.clone(); - let owner = auth.owner_id; - tokio::spawn(async move { - if let Err(e) = crate::orchestration::directive::on_contract_completed( - &pool_clone, &state_clone, &contract_clone, owner, - ).await { - tracing::warn!( - contract_id = %contract_clone.id, - error = %e, - "Failed to process directive contract completion" - ); - } - }); - } } // Get summary with counts diff --git a/makima/src/server/handlers/directives.rs b/makima/src/server/handlers/directives.rs deleted file mode 100644 index 3f62a33..0000000 --- a/makima/src/server/handlers/directives.rs +++ /dev/null @@ -1,785 +0,0 @@ -//! HTTP handlers for directive CRUD operations. - -use axum::{ - extract::{Path, State}, - http::StatusCode, - response::IntoResponse, - Json, -}; -use uuid::Uuid; - -use std::collections::HashMap; - -use crate::db::models::{ - ChainStep, ChainStepWithContract, ChainWithSteps, CreateDirectiveRequest, Directive, - DirectiveChain, DirectiveListResponse, DirectiveWithChains, EvaluationListResponse, - StepContractSummary, SubmitPlanRequest, UpdateDirectiveRequest, -}; -use crate::db::repository::{self, RepositoryError}; -use crate::orchestration; -use crate::server::auth::Authenticated; -use crate::server::messages::ApiError; -use crate::server::state::SharedState; - -/// List all directives for the authenticated user's owner. -#[utoipa::path( - get, - path = "/api/v1/directives", - responses( - (status = 200, description = "List of directives", body = DirectiveListResponse), - (status = 401, description = "Unauthorized", body = ApiError), - (status = 503, description = "Database not configured", body = ApiError), - (status = 500, description = "Internal server error", body = ApiError), - ), - security( - ("bearer_auth" = []), - ("api_key" = []) - ), - tag = "Directives" -)] -pub async fn list_directives( - State(state): State<SharedState>, - Authenticated(auth): Authenticated, -) -> impl IntoResponse { - let Some(ref pool) = state.db_pool else { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), - ) - .into_response(); - }; - - match repository::list_directives_for_owner(pool, auth.owner_id).await { - Ok(directives) => { - let total = directives.len() as i64; - Json(DirectiveListResponse { directives, total }).into_response() - } - Err(e) => { - tracing::error!("Failed to list directives: {}", e); - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ApiError::new("DB_ERROR", e.to_string())), - ) - .into_response() - } - } -} - -/// Get a directive by ID with its chains. -#[utoipa::path( - get, - path = "/api/v1/directives/{id}", - params( - ("id" = Uuid, Path, description = "Directive ID") - ), - responses( - (status = 200, description = "Directive details with chains", body = DirectiveWithChains), - (status = 401, description = "Unauthorized", body = ApiError), - (status = 404, description = "Directive not found", body = ApiError), - (status = 503, description = "Database not configured", body = ApiError), - (status = 500, description = "Internal server error", body = ApiError), - ), - security( - ("bearer_auth" = []), - ("api_key" = []) - ), - tag = "Directives" -)] -pub async fn get_directive( - State(state): State<SharedState>, - Authenticated(auth): Authenticated, - Path(id): Path<Uuid>, -) -> impl IntoResponse { - let Some(ref pool) = state.db_pool else { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), - ) - .into_response(); - }; - - let directive = match repository::get_directive_for_owner(pool, id, auth.owner_id).await { - Ok(Some(d)) => d, - Ok(None) => { - return ( - StatusCode::NOT_FOUND, - Json(ApiError::new("NOT_FOUND", "Directive not found")), - ) - .into_response(); - } - Err(e) => { - tracing::error!("Failed to get directive {}: {}", id, e); - return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ApiError::new("DB_ERROR", e.to_string())), - ) - .into_response(); - } - }; - - let chains = match repository::list_chains_for_directive(pool, id).await { - Ok(c) => c, - Err(e) => { - tracing::warn!("Failed to get chains for directive {}: {}", id, e); - Vec::new() - } - }; - - // Build chains with steps - let mut all_steps_by_chain = Vec::new(); - for chain in &chains { - let steps = match repository::list_steps_for_chain(pool, chain.id).await { - Ok(s) => s, - Err(e) => { - tracing::warn!("Failed to get steps for chain {}: {}", chain.id, e); - Vec::new() - } - }; - all_steps_by_chain.push(steps); - } - - // Collect all contract IDs (from steps + orchestrator) - let mut contract_ids: Vec<Uuid> = all_steps_by_chain - .iter() - .flat_map(|steps| steps.iter().filter_map(|s| s.contract_id)) - .collect(); - if let Some(orch_id) = directive.orchestrator_contract_id { - contract_ids.push(orch_id); - } - - // Batch fetch contract summaries - let mut summary_map: HashMap<Uuid, StepContractSummary> = if contract_ids.is_empty() { - HashMap::new() - } else { - match repository::get_contract_summaries_batch(pool, &contract_ids).await { - Ok(summaries) => summaries.into_iter().map(|s| (s.id, s)).collect(), - Err(e) => { - tracing::warn!("Failed to fetch contract summaries: {}", e); - HashMap::new() - } - } - }; - - // Build enriched chains - let chains_with_steps: Vec<ChainWithSteps> = chains - .into_iter() - .zip(all_steps_by_chain.into_iter()) - .map(|(chain, steps)| { - let enriched_steps = steps - .into_iter() - .map(|step| { - let contract_summary = - step.contract_id.and_then(|id| summary_map.remove(&id)); - ChainStepWithContract { - step, - contract_summary, - } - }) - .collect(); - ChainWithSteps { - chain, - steps: enriched_steps, - } - }) - .collect(); - - let orchestrator_contract_summary = directive - .orchestrator_contract_id - .and_then(|id| summary_map.remove(&id)); - - Json(DirectiveWithChains { - directive, - orchestrator_contract_summary, - chains: chains_with_steps, - }) - .into_response() -} - -/// Create a new directive. -#[utoipa::path( - post, - path = "/api/v1/directives", - request_body = CreateDirectiveRequest, - responses( - (status = 201, description = "Directive created", body = Directive), - (status = 400, description = "Invalid request", body = ApiError), - (status = 401, description = "Unauthorized", body = ApiError), - (status = 503, description = "Database not configured", body = ApiError), - (status = 500, description = "Internal server error", body = ApiError), - ), - security( - ("bearer_auth" = []), - ("api_key" = []) - ), - tag = "Directives" -)] -pub async fn create_directive( - State(state): State<SharedState>, - Authenticated(auth): Authenticated, - Json(req): Json<CreateDirectiveRequest>, -) -> impl IntoResponse { - let Some(ref pool) = state.db_pool else { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), - ) - .into_response(); - }; - - match repository::create_directive_for_owner(pool, auth.owner_id, req).await { - Ok(directive) => (StatusCode::CREATED, Json(directive)).into_response(), - Err(e) => { - tracing::error!("Failed to create directive: {}", e); - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ApiError::new("DB_ERROR", e.to_string())), - ) - .into_response() - } - } -} - -/// Update an existing directive. -#[utoipa::path( - put, - path = "/api/v1/directives/{id}", - params( - ("id" = Uuid, Path, description = "Directive ID") - ), - request_body = UpdateDirectiveRequest, - responses( - (status = 200, description = "Directive updated", body = Directive), - (status = 401, description = "Unauthorized", body = ApiError), - (status = 404, description = "Directive not found", body = ApiError), - (status = 409, description = "Version conflict", body = ApiError), - (status = 503, description = "Database not configured", body = ApiError), - (status = 500, description = "Internal server error", body = ApiError), - ), - security( - ("bearer_auth" = []), - ("api_key" = []) - ), - tag = "Directives" -)] -pub async fn update_directive( - State(state): State<SharedState>, - Authenticated(auth): Authenticated, - Path(id): Path<Uuid>, - Json(req): Json<UpdateDirectiveRequest>, -) -> impl IntoResponse { - let Some(ref pool) = state.db_pool else { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), - ) - .into_response(); - }; - - match repository::update_directive_for_owner(pool, id, auth.owner_id, req).await { - Ok(Some(directive)) => Json(directive).into_response(), - Ok(None) => ( - StatusCode::NOT_FOUND, - Json(ApiError::new("NOT_FOUND", "Directive not found")), - ) - .into_response(), - Err(RepositoryError::VersionConflict { expected, actual }) => ( - StatusCode::CONFLICT, - Json(ApiError::new( - "VERSION_CONFLICT", - format!( - "Version conflict: expected {}, actual {}", - expected, actual - ), - )), - ) - .into_response(), - Err(RepositoryError::Database(e)) => { - tracing::error!("Failed to update directive {}: {}", id, e); - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ApiError::new("DB_ERROR", e.to_string())), - ) - .into_response() - } - } -} - -/// Delete a directive. -#[utoipa::path( - delete, - path = "/api/v1/directives/{id}", - params( - ("id" = Uuid, Path, description = "Directive ID") - ), - responses( - (status = 204, description = "Directive deleted"), - (status = 401, description = "Unauthorized", body = ApiError), - (status = 404, description = "Directive not found", body = ApiError), - (status = 503, description = "Database not configured", body = ApiError), - (status = 500, description = "Internal server error", body = ApiError), - ), - security( - ("bearer_auth" = []), - ("api_key" = []) - ), - tag = "Directives" -)] -pub async fn delete_directive( - State(state): State<SharedState>, - Authenticated(auth): Authenticated, - Path(id): Path<Uuid>, -) -> impl IntoResponse { - let Some(ref pool) = state.db_pool else { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), - ) - .into_response(); - }; - - match repository::delete_directive_for_owner(pool, id, auth.owner_id).await { - Ok(true) => StatusCode::NO_CONTENT.into_response(), - Ok(false) => ( - StatusCode::NOT_FOUND, - Json(ApiError::new("NOT_FOUND", "Directive not found")), - ) - .into_response(), - Err(e) => { - tracing::error!("Failed to delete directive {}: {}", id, e); - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ApiError::new("DB_ERROR", e.to_string())), - ) - .into_response() - } - } -} - -/// List chains for a directive. -#[utoipa::path( - get, - path = "/api/v1/directives/{id}/chains", - params( - ("id" = Uuid, Path, description = "Directive ID") - ), - responses( - (status = 200, description = "List of chains", body = Vec<DirectiveChain>), - (status = 401, description = "Unauthorized", body = ApiError), - (status = 404, description = "Directive not found", body = ApiError), - (status = 503, description = "Database not configured", body = ApiError), - (status = 500, description = "Internal server error", body = ApiError), - ), - security( - ("bearer_auth" = []), - ("api_key" = []) - ), - tag = "Directives" -)] -pub async fn list_chains( - State(state): State<SharedState>, - Authenticated(auth): Authenticated, - Path(id): Path<Uuid>, -) -> impl IntoResponse { - let Some(ref pool) = state.db_pool else { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), - ) - .into_response(); - }; - - // Verify directive exists and belongs to owner - match repository::get_directive_for_owner(pool, id, auth.owner_id).await { - Ok(Some(_)) => {} - Ok(None) => { - return ( - StatusCode::NOT_FOUND, - Json(ApiError::new("NOT_FOUND", "Directive not found")), - ) - .into_response(); - } - Err(e) => { - tracing::error!("Failed to get directive {}: {}", id, e); - return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ApiError::new("DB_ERROR", e.to_string())), - ) - .into_response(); - } - } - - match repository::list_chains_for_directive(pool, id).await { - Ok(chains) => Json(chains).into_response(), - Err(e) => { - tracing::error!("Failed to list chains for directive {}: {}", id, e); - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ApiError::new("DB_ERROR", e.to_string())), - ) - .into_response() - } - } -} - -/// Get a chain with its steps. -#[utoipa::path( - get, - path = "/api/v1/directives/{id}/chains/{chain_id}", - params( - ("id" = Uuid, Path, description = "Directive ID"), - ("chain_id" = Uuid, Path, description = "Chain ID") - ), - responses( - (status = 200, description = "Chain with steps", body = ChainWithSteps), - (status = 401, description = "Unauthorized", body = ApiError), - (status = 404, description = "Chain not found", body = ApiError), - (status = 503, description = "Database not configured", body = ApiError), - (status = 500, description = "Internal server error", body = ApiError), - ), - security( - ("bearer_auth" = []), - ("api_key" = []) - ), - tag = "Directives" -)] -pub async fn get_chain( - State(state): State<SharedState>, - Authenticated(auth): Authenticated, - Path((id, chain_id)): Path<(Uuid, Uuid)>, -) -> impl IntoResponse { - let Some(ref pool) = state.db_pool else { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), - ) - .into_response(); - }; - - // Verify directive exists and belongs to owner - match repository::get_directive_for_owner(pool, id, auth.owner_id).await { - Ok(Some(_)) => {} - Ok(None) => { - return ( - StatusCode::NOT_FOUND, - Json(ApiError::new("NOT_FOUND", "Directive not found")), - ) - .into_response(); - } - Err(e) => { - tracing::error!("Failed to get directive {}: {}", id, e); - return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ApiError::new("DB_ERROR", e.to_string())), - ) - .into_response(); - } - } - - // Get the chain and verify it belongs to this directive - let chains = match repository::list_chains_for_directive(pool, id).await { - Ok(c) => c, - Err(e) => { - tracing::error!("Failed to list chains: {}", e); - return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ApiError::new("DB_ERROR", e.to_string())), - ) - .into_response(); - } - }; - - let chain = match chains.into_iter().find(|c| c.id == chain_id) { - Some(c) => c, - None => { - return ( - StatusCode::NOT_FOUND, - Json(ApiError::new("NOT_FOUND", "Chain not found")), - ) - .into_response(); - } - }; - - let steps = match repository::list_steps_for_chain(pool, chain_id).await { - Ok(s) => s, - Err(e) => { - tracing::warn!("Failed to get steps for chain {}: {}", chain_id, e); - Vec::new() - } - }; - - // Collect contract IDs from steps - let contract_ids: Vec<Uuid> = steps.iter().filter_map(|s| s.contract_id).collect(); - - let mut summary_map: HashMap<Uuid, StepContractSummary> = if contract_ids.is_empty() { - HashMap::new() - } else { - match repository::get_contract_summaries_batch(pool, &contract_ids).await { - Ok(summaries) => summaries.into_iter().map(|s| (s.id, s)).collect(), - Err(e) => { - tracing::warn!("Failed to fetch contract summaries: {}", e); - HashMap::new() - } - } - }; - - let enriched_steps = steps - .into_iter() - .map(|step| { - let contract_summary = step.contract_id.and_then(|id| summary_map.remove(&id)); - ChainStepWithContract { - step, - contract_summary, - } - }) - .collect(); - - Json(ChainWithSteps { - chain, - steps: enriched_steps, - }) - .into_response() -} - -/// Start a directive: create a planning contract and begin orchestration. -#[utoipa::path( - post, - path = "/api/v1/directives/{id}/start", - params( - ("id" = Uuid, Path, description = "Directive ID") - ), - responses( - (status = 200, description = "Directive started", body = Directive), - (status = 400, description = "Directive not in draft status", body = ApiError), - (status = 401, description = "Unauthorized", body = ApiError), - (status = 404, description = "Directive not found", body = ApiError), - (status = 503, description = "Database not configured", body = ApiError), - (status = 500, description = "Internal server error", body = ApiError), - ), - security( - ("bearer_auth" = []), - ("api_key" = []) - ), - tag = "Directives" -)] -pub async fn start_directive( - State(state): State<SharedState>, - Authenticated(auth): Authenticated, - Path(id): Path<Uuid>, -) -> impl IntoResponse { - let Some(ref pool) = state.db_pool else { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), - ) - .into_response(); - }; - - match orchestration::directive::init_directive(pool, &state, auth.owner_id, id).await { - Ok(directive) => Json(directive).into_response(), - Err(e) if e.contains("not found") => ( - StatusCode::NOT_FOUND, - Json(ApiError::new("NOT_FOUND", e)), - ) - .into_response(), - Err(e) if e.contains("must be in 'draft'") => ( - StatusCode::BAD_REQUEST, - Json(ApiError::new("INVALID_STATUS", e)), - ) - .into_response(), - Err(e) => { - tracing::error!("Failed to start directive {}: {}", id, e); - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ApiError::new("START_FAILED", e)), - ) - .into_response() - } - } -} - -/// Trigger a manual evaluation for a step. -#[utoipa::path( - post, - path = "/api/v1/directives/{id}/steps/{step_id}/evaluate", - params( - ("id" = Uuid, Path, description = "Directive ID"), - ("step_id" = Uuid, Path, description = "Step ID") - ), - responses( - (status = 200, description = "Evaluation triggered", body = ChainStep), - (status = 400, description = "Step cannot be evaluated", body = ApiError), - (status = 401, description = "Unauthorized", body = ApiError), - (status = 404, description = "Not found", body = ApiError), - (status = 503, description = "Database not configured", body = ApiError), - (status = 500, description = "Internal server error", body = ApiError), - ), - security( - ("bearer_auth" = []), - ("api_key" = []) - ), - tag = "Directives" -)] -pub async fn evaluate_step( - State(state): State<SharedState>, - Authenticated(auth): Authenticated, - Path((id, step_id)): Path<(Uuid, Uuid)>, -) -> impl IntoResponse { - let Some(ref pool) = state.db_pool else { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), - ) - .into_response(); - }; - - match orchestration::directive::trigger_manual_evaluation(pool, &state, auth.owner_id, id, step_id).await { - Ok(step) => Json(step).into_response(), - Err(e) if e.contains("not found") || e.contains("Not found") => ( - StatusCode::NOT_FOUND, - Json(ApiError::new("NOT_FOUND", e)), - ) - .into_response(), - Err(e) if e.contains("hasn't been executed") || e.contains("no active chain") => ( - StatusCode::BAD_REQUEST, - Json(ApiError::new("INVALID_STATE", e)), - ) - .into_response(), - Err(e) => { - tracing::error!("Failed to trigger evaluation for step {}: {}", step_id, e); - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ApiError::new("EVALUATION_FAILED", e)), - ) - .into_response() - } - } -} - -/// List evaluations for a step. -#[utoipa::path( - get, - path = "/api/v1/directives/{id}/steps/{step_id}/evaluations", - params( - ("id" = Uuid, Path, description = "Directive ID"), - ("step_id" = Uuid, Path, description = "Step ID") - ), - responses( - (status = 200, description = "List of evaluations", body = EvaluationListResponse), - (status = 401, description = "Unauthorized", body = ApiError), - (status = 404, description = "Not found", body = ApiError), - (status = 503, description = "Database not configured", body = ApiError), - (status = 500, description = "Internal server error", body = ApiError), - ), - security( - ("bearer_auth" = []), - ("api_key" = []) - ), - tag = "Directives" -)] -pub async fn list_evaluations( - State(state): State<SharedState>, - Authenticated(auth): Authenticated, - Path((id, step_id)): Path<(Uuid, Uuid)>, -) -> impl IntoResponse { - let Some(ref pool) = state.db_pool else { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), - ) - .into_response(); - }; - - // Verify directive exists and belongs to owner - match repository::get_directive_for_owner(pool, id, auth.owner_id).await { - Ok(Some(_)) => {} - Ok(None) => { - return ( - StatusCode::NOT_FOUND, - Json(ApiError::new("NOT_FOUND", "Directive not found")), - ) - .into_response(); - } - Err(e) => { - tracing::error!("Failed to get directive {}: {}", id, e); - return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ApiError::new("DB_ERROR", e.to_string())), - ) - .into_response(); - } - } - - match repository::list_evaluations_for_step(pool, step_id).await { - Ok(evaluations) => { - let total = evaluations.len() as i64; - Json(EvaluationListResponse { evaluations, total }).into_response() - } - Err(e) => { - tracing::error!("Failed to list evaluations for step {}: {}", step_id, e); - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ApiError::new("DB_ERROR", e.to_string())), - ) - .into_response() - } - } -} - -/// Submit a chain plan for a directive. -#[utoipa::path( - post, - path = "/api/v1/directives/{id}/submit-plan", - params( - ("id" = Uuid, Path, description = "Directive ID") - ), - request_body = SubmitPlanRequest, - responses( - (status = 200, description = "Plan submitted, directive active", body = Directive), - (status = 400, description = "Invalid request or status", body = ApiError), - (status = 401, description = "Unauthorized", body = ApiError), - (status = 404, description = "Directive not found", body = ApiError), - (status = 503, description = "Database not configured", body = ApiError), - (status = 500, description = "Internal server error", body = ApiError), - ), - security( - ("bearer_auth" = []), - ("api_key" = []) - ), - tag = "Directives" -)] -pub async fn submit_plan( - State(state): State<SharedState>, - Authenticated(auth): Authenticated, - Path(id): Path<Uuid>, - Json(req): Json<SubmitPlanRequest>, -) -> impl IntoResponse { - let Some(ref pool) = state.db_pool else { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), - ) - .into_response(); - }; - - match orchestration::directive::submit_plan(pool, &state, auth.owner_id, id, &req.plan).await { - Ok(directive) => Json(directive).into_response(), - Err(e) if e.contains("not found") || e.contains("Not found") => ( - StatusCode::NOT_FOUND, - Json(ApiError::new("NOT_FOUND", e)), - ) - .into_response(), - Err(e) if e.contains("must be in 'planning'") || e.contains("no steps") => ( - StatusCode::BAD_REQUEST, - Json(ApiError::new("INVALID_REQUEST", e)), - ) - .into_response(), - Err(e) => { - tracing::error!("Failed to submit plan for directive {}: {}", id, e); - ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ApiError::new("SUBMIT_PLAN_FAILED", e)), - ) - .into_response() - } - } -} diff --git a/makima/src/server/handlers/mesh_daemon.rs b/makima/src/server/handlers/mesh_daemon.rs index 767d059..87b5e44 100644 --- a/makima/src/server/handlers/mesh_daemon.rs +++ b/makima/src/server/handlers/mesh_daemon.rs @@ -1303,16 +1303,6 @@ async fn handle_daemon_connection(socket: WebSocket, state: SharedState, auth_re }), ).await; - // Directive engine integration - if let Err(e) = crate::orchestration::directive::on_task_completed( - &pool, &state, &updated_task, owner_id, - ).await { - tracing::warn!( - task_id = %task_id, - error = %e, - "Failed to process directive task completion" - ); - } } Ok(None) => { tracing::warn!( diff --git a/makima/src/server/handlers/mod.rs b/makima/src/server/handlers/mod.rs index 29cd09f..ae370c9 100644 --- a/makima/src/server/handlers/mod.rs +++ b/makima/src/server/handlers/mod.rs @@ -6,7 +6,6 @@ pub mod contract_chat; pub mod contract_daemon; pub mod contract_discuss; pub mod contracts; -pub mod directives; pub mod file_ws; pub mod files; pub mod history; diff --git a/makima/src/server/mod.rs b/makima/src/server/mod.rs index 0cad050..b7a4156 100644 --- a/makima/src/server/mod.rs +++ b/makima/src/server/mod.rs @@ -18,7 +18,7 @@ use tower_http::trace::TraceLayer; use utoipa::OpenApi; use utoipa_swagger_ui::SwaggerUi; -use crate::server::handlers::{api_keys, chat, contract_chat, contract_daemon, contract_discuss, contracts, directives, file_ws, files, history, listen, mesh, mesh_chat, mesh_daemon, mesh_merge, mesh_supervisor, mesh_ws, repository_history, speak, templates, transcript_analysis, users, versions}; +use crate::server::handlers::{api_keys, chat, contract_chat, contract_daemon, contract_discuss, contracts, file_ws, files, history, listen, mesh, mesh_chat, mesh_daemon, mesh_merge, mesh_supervisor, mesh_ws, repository_history, speak, templates, transcript_analysis, users, versions}; use crate::server::openapi::ApiDoc; use crate::server::state::SharedState; @@ -170,23 +170,6 @@ pub fn make_router(state: SharedState) -> Router { "/contracts/{id}/chat/history", get(contract_chat::get_contract_chat_history).delete(contract_chat::clear_contract_chat_history), ) - // Directive endpoints - .route( - "/directives", - get(directives::list_directives).post(directives::create_directive), - ) - .route( - "/directives/{id}", - get(directives::get_directive) - .put(directives::update_directive) - .delete(directives::delete_directive), - ) - .route("/directives/{id}/start", post(directives::start_directive)) - .route("/directives/{id}/chains", get(directives::list_chains)) - .route("/directives/{id}/chains/{chain_id}", get(directives::get_chain)) - .route("/directives/{id}/submit-plan", post(directives::submit_plan)) - .route("/directives/{id}/steps/{step_id}/evaluate", post(directives::evaluate_step)) - .route("/directives/{id}/steps/{step_id}/evaluations", get(directives::list_evaluations)) // Contract supervisor resume endpoints .route("/contracts/{id}/supervisor/resume", post(mesh_supervisor::resume_supervisor)) .route("/contracts/{id}/supervisor/conversation/rewind", post(mesh_supervisor::rewind_conversation)) diff --git a/makima/src/server/openapi.rs b/makima/src/server/openapi.rs index 888269f..0b6bfba 100644 --- a/makima/src/server/openapi.rs +++ b/makima/src/server/openapi.rs @@ -4,28 +4,27 @@ use utoipa::OpenApi; use crate::db::models::{ AddLocalRepositoryRequest, AddRemoteRepositoryRequest, BranchInfo, BranchListResponse, - BranchTaskRequest, BranchTaskResponse, ChainStep, ChainStepWithContract, ChainWithSteps, + BranchTaskRequest, BranchTaskResponse, ChangePhaseRequest, Contract, ContractChatHistoryResponse, ContractChatMessageRecord, ContractEvent, ContractListResponse, ContractRepository, ContractSummary, ContractWithRelations, - CreateContractRequest, CreateDirectiveRequest, CreateFileRequest, + CreateContractRequest, CreateFileRequest, CreateManagedRepositoryRequest, CreateTaskRequest, Daemon, DaemonDirectoriesResponse, - DaemonDirectory, DaemonListResponse, Directive, DirectiveChain, DirectiveEvaluation, - DirectiveEvent, DirectiveListResponse, DirectiveSummary, DirectiveWithChains, - EvaluationListResponse, File, FileListResponse, FileSummary, + DaemonDirectory, DaemonListResponse, + File, FileListResponse, FileSummary, MergeCommitRequest, MergeCompleteCheckResponse, MergeResolveRequest, MergeResultResponse, MergeSkipRequest, MergeStartRequest, MergeStatusResponse, MeshChatConversation, MeshChatHistoryResponse, MeshChatMessageRecord, RepositoryHistoryEntry, RepositoryHistoryListResponse, RepositorySuggestionsQuery, SendMessageRequest, - StepContractSummary, SubmitPlanRequest, Task, + Task, TaskEventListResponse, TaskListResponse, TaskSummary, TaskWithSubtasks, TranscriptEntry, - UpdateContractRequest, UpdateDirectiveRequest, UpdateFileRequest, UpdateTaskRequest, + UpdateContractRequest, UpdateFileRequest, UpdateTaskRequest, }; use crate::server::auth::{ ApiKey, ApiKeyInfoResponse, CreateApiKeyRequest, CreateApiKeyResponse, RefreshApiKeyRequest, RefreshApiKeyResponse, RevokeApiKeyResponse, }; -use crate::server::handlers::{api_keys, contract_chat, contract_discuss, contracts, directives, files, listen, mesh, mesh_chat, mesh_merge, repository_history, users}; +use crate::server::handlers::{api_keys, contract_chat, contract_discuss, contracts, files, listen, mesh, mesh_chat, mesh_merge, repository_history, users}; use crate::server::messages::{ApiError, AudioEncoding, StartMessage, StopMessage, TranscriptMessage}; #[derive(OpenApi)] @@ -108,18 +107,6 @@ use crate::server::messages::{ApiError, AudioEncoding, StartMessage, StopMessage repository_history::list_repository_history, repository_history::get_repository_suggestions, repository_history::delete_repository_history, - // Directive endpoints - directives::list_directives, - directives::get_directive, - directives::create_directive, - directives::update_directive, - directives::delete_directive, - directives::start_directive, - directives::list_chains, - directives::get_chain, - directives::evaluate_step, - directives::list_evaluations, - directives::submit_plan, ), components( schemas( @@ -204,22 +191,6 @@ use crate::server::messages::{ApiError, AudioEncoding, StartMessage, StopMessage RepositoryHistoryEntry, RepositoryHistoryListResponse, RepositorySuggestionsQuery, - // Directive schemas - Directive, - DirectiveSummary, - DirectiveListResponse, - DirectiveWithChains, - DirectiveChain, - ChainStep, - ChainStepWithContract, - ChainWithSteps, - StepContractSummary, - CreateDirectiveRequest, - UpdateDirectiveRequest, - SubmitPlanRequest, - DirectiveEvaluation, - DirectiveEvent, - EvaluationListResponse, ) ), tags( @@ -230,7 +201,6 @@ use crate::server::messages::{ApiError, AudioEncoding, StartMessage, StopMessage (name = "API Keys", description = "API key management for programmatic access"), (name = "Users", description = "User account management"), (name = "Settings", description = "User settings including repository history"), - (name = "Directives", description = "Directive management for autonomous goal-driven execution"), ) )] pub struct ApiDoc; |