diff options
Diffstat (limited to 'makima/src')
| -rw-r--r-- | makima/src/db/repository.rs | 185 | ||||
| -rw-r--r-- | makima/src/orchestration/directive.rs | 10 | ||||
| -rw-r--r-- | makima/src/server/handlers/directive_documents.rs | 85 | ||||
| -rw-r--r-- | makima/src/server/handlers/directives.rs | 54 | ||||
| -rw-r--r-- | makima/src/server/mod.rs | 4 | ||||
| -rw-r--r-- | makima/src/server/openapi.rs | 1 |
6 files changed, 291 insertions, 48 deletions
diff --git a/makima/src/db/repository.rs b/makima/src/db/repository.rs index 20f3268..ee4b561 100644 --- a/makima/src/db/repository.rs +++ b/makima/src/db/repository.rs @@ -6132,6 +6132,11 @@ pub async fn archive_directive_document( /// or queue it. Returns the updated row, or `Ok(None)` if the contract /// doesn't exist. Errors with `RepositoryError::Validation` if the /// contract is in any state other than `draft`. +/// +/// Side effect: if the contract enters `active`, the parent directive +/// is flipped to `active` (from `draft|paused|idle|inactive`). This is +/// what makes the orchestrator reconciler pick the directive up — its +/// gate is `directive.status = 'active' AND orchestrator_task_id IS NULL`. pub async fn start_contract( pool: &PgPool, contract_id: Uuid, @@ -6184,6 +6189,13 @@ pub async fn start_contract( .fetch_optional(&mut *tx) .await?; + // Flip the parent directive to active so the reconciler picks it up. + // Only when this contract is actually entering the active slot — a + // queued contract doesn't drive planning by itself. + if new_status == "active" { + activate_parent_directive(&mut tx, current.directive_id).await?; + } + tx.commit().await?; Ok(updated) } @@ -6234,6 +6246,16 @@ pub async fn pause_contract( // position, excluding the one we just paused). promote_next_queued_contract(&mut tx, current.directive_id).await?; + // If no contract is active after the pause+promote, pause the + // directive too — stops the reconciler from spawning new planners + // on what is now an idle directive. + deactivate_parent_directive_if_no_active( + &mut tx, + current.directive_id, + "paused", + ) + .await?; + tx.commit().await?; Ok(updated) } @@ -6289,6 +6311,17 @@ pub async fn complete_contract( promote_next_queued_contract(&mut tx, current.directive_id).await?; + // If the ship freed the active slot AND no queued contract was + // available to promote, the directive itself goes inactive — its + // iteration is shipped; the next cycle starts via reopen or a new + // contract. + deactivate_parent_directive_if_no_active( + &mut tx, + current.directive_id, + "inactive", + ) + .await?; + tx.commit().await?; Ok(updated) } @@ -6339,12 +6372,164 @@ pub async fn unlock_contract( if was_active { promote_next_queued_contract(&mut tx, current.directive_id).await?; + // If unlocking the active contract leaves no other active under + // the directive, pause the directive too. + deactivate_parent_directive_if_no_active( + &mut tx, + current.directive_id, + "paused", + ) + .await?; } tx.commit().await?; Ok(updated) } +/// Reopen a shipped contract for amendment. Flips the contract back to +/// `active`, re-activates the parent directive, and clears the +/// directive's PR linkage + orchestrator task so the reconciler spawns a +/// fresh planner. The planner uses `get_latest_merged_revision` to +/// detect the previously-shipped PR and frame the new plan as a delta. +pub async fn reopen_contract( + pool: &PgPool, + contract_id: Uuid, +) -> Result<Option<DirectiveDocument>, RepositoryError> { + let mut tx = pool.begin().await?; + + let current = sqlx::query_as::<_, DirectiveDocument>( + r#"SELECT * FROM directive_documents WHERE id = $1"#, + ) + .bind(contract_id) + .fetch_optional(&mut *tx) + .await?; + + let current = match current { + Some(c) => c, + None => return Ok(None), + }; + + if current.status != "shipped" { + return Err(RepositoryError::Validation(format!( + "contract is in status '{}'; only 'shipped' contracts can be reopened", + current.status + ))); + } + + let updated = sqlx::query_as::<_, DirectiveDocument>( + r#" + UPDATE directive_documents + SET status = 'active', + version = version + 1, + updated_at = NOW() + WHERE id = $1 + RETURNING * + "#, + ) + .bind(contract_id) + .fetch_optional(&mut *tx) + .await?; + + // Re-activate the directive and clear the prior PR + orchestrator + // linkage. Status is forced to `active` regardless of prior value + // (except archived — guard against re-opening under an archived + // directive). + sqlx::query( + r#" + UPDATE directives + SET status = 'active', + orchestrator_task_id = NULL, + pr_url = NULL, + pr_branch = NULL, + updated_at = NOW(), + version = version + 1 + WHERE id = $1 AND status <> 'archived' + "#, + ) + .bind(current.directive_id) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + Ok(updated) +} + +/// Resolve the directive's currently-active contract id. Returns +/// `Ok(None)` when no active contract exists. Used by the +/// auto-complete-on-PR path so the contract row can be shipped at the +/// same moment the directive registers its PR url. +pub async fn get_active_contract_id_for_directive( + pool: &PgPool, + directive_id: Uuid, +) -> Result<Option<Uuid>, sqlx::Error> { + let row: Option<(Uuid,)> = sqlx::query_as( + r#" + SELECT id FROM directive_documents + WHERE directive_id = $1 AND status = 'active' + ORDER BY position ASC, created_at ASC + LIMIT 1 + "#, + ) + .bind(directive_id) + .fetch_optional(pool) + .await?; + Ok(row.map(|r| r.0)) +} + +/// Flip the parent directive to `active` when a child contract just +/// became active. Only promotes from `draft|paused|idle|inactive` — +/// leaves `archived` directives untouched. +async fn activate_parent_directive( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + directive_id: Uuid, +) -> Result<(), sqlx::Error> { + sqlx::query( + r#" + UPDATE directives + SET status = 'active', + updated_at = NOW(), + version = version + 1 + WHERE id = $1 + AND status IN ('draft', 'paused', 'idle', 'inactive') + "#, + ) + .bind(directive_id) + .execute(&mut **tx) + .await?; + Ok(()) +} + +/// After a contract lifecycle change that may have left no active +/// contract under the directive, transition the directive to the +/// supplied `new_status` (typically `'paused'` for unlock/pause flows, +/// `'inactive'` for ship). No-op if the directive still has an active +/// contract or is already past the destination state. +async fn deactivate_parent_directive_if_no_active( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + directive_id: Uuid, + new_status: &str, +) -> Result<(), sqlx::Error> { + sqlx::query( + r#" + UPDATE directives + SET status = $2, + updated_at = NOW(), + version = version + 1 + WHERE id = $1 + AND status = 'active' + AND NOT EXISTS ( + SELECT 1 FROM directive_documents + WHERE directive_id = $1 AND status = 'active' + ) + "#, + ) + .bind(directive_id) + .bind(new_status) + .execute(&mut **tx) + .await?; + Ok(()) +} + /// Find the lowest-position `queued` contract under a directive and /// flip it to `active`. No-op when no queued contract exists. /// diff --git a/makima/src/orchestration/directive.rs b/makima/src/orchestration/directive.rs index 384fa23..7f90bcd 100644 --- a/makima/src/orchestration/directive.rs +++ b/makima/src/orchestration/directive.rs @@ -97,7 +97,7 @@ impl DirectiveOrchestrator { .spawn_orchestrator_task( directive.id, directive.owner_id, - format!("Plan: {}", directive.title), + "orchestrator".to_string(), plan, directive.repository_url.as_deref(), directive.base_branch.as_deref(), @@ -517,7 +517,7 @@ impl DirectiveOrchestrator { .spawn_orchestrator_task( directive.id, directive.owner_id, - format!("Re-plan: {}", directive.title), + "orchestrator (re-plan)".to_string(), plan, directive.repository_url.as_deref(), directive.base_branch.as_deref(), @@ -844,7 +844,7 @@ impl DirectiveOrchestrator { .spawn_completion_task( directive.id, directive.owner_id, - format!("PR: {}", directive.title), + "completion".to_string(), prompt, directive.repository_url.as_deref(), directive.base_branch.as_deref(), @@ -1367,9 +1367,9 @@ pub async fn trigger_completion_task( let prompt = build_completion_prompt(&directive, &contract_body, &step_tasks, &step_branches, &directive_branch, base_branch); let task_name = if directive.pr_url.is_some() { - format!("Update PR: {}", directive.title) + "completion (update)".to_string() } else { - format!("PR: {}", directive.title) + "completion".to_string() }; // Create the completion task FIRST so we have a real task ID for the FK diff --git a/makima/src/server/handlers/directive_documents.rs b/makima/src/server/handlers/directive_documents.rs index 23081b5..ee98a61 100644 --- a/makima/src/server/handlers/directive_documents.rs +++ b/makima/src/server/handlers/directive_documents.rs @@ -680,7 +680,7 @@ pub async fn reorder_contract( /// closure. Cuts the boilerplate from start/pause/complete/unlock down /// to a couple of lines each. async fn run_contract_transition<F, Fut>( - pool: sqlx::PgPool, + state: SharedState, owner_id: Uuid, contract_id: Uuid, f: F, @@ -691,6 +691,14 @@ where Output = Result<Option<crate::db::models::DirectiveDocument>, RepositoryError>, >, { + let Some(pool) = state.db_pool.clone() else { + return ( + StatusCode::SERVICE_UNAVAILABLE, + Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), + ) + .into_response(); + }; + match load_owned_document(&pool, owner_id, contract_id).await { Ok(Some(_)) => {} Ok(None) => { @@ -710,7 +718,14 @@ where } match f(pool, contract_id).await { - Ok(Some(doc)) => Json(doc).into_response(), + Ok(Some(doc)) => { + // Any successful lifecycle transition may have flipped the + // parent directive's status (see repository helpers). Wake + // the reconciler so the user doesn't wait up to 15s before + // the orchestrator daemon spawns / stops. + state.kick_directive_reconciler(); + Json(doc).into_response() + } Ok(None) => ( StatusCode::NOT_FOUND, Json(ApiError::new("NOT_FOUND", "Contract not found")), @@ -759,14 +774,7 @@ pub async fn start_contract( Authenticated(auth): Authenticated, Path(document_id): Path<Uuid>, ) -> impl IntoResponse { - let Some(pool) = state.db_pool.clone() else { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), - ) - .into_response(); - }; - run_contract_transition(pool, auth.owner_id, document_id, |pool, id| async move { + run_contract_transition(state, auth.owner_id, document_id, |pool, id| async move { repository::start_contract(&pool, id).await }) .await @@ -793,14 +801,7 @@ pub async fn pause_contract( Authenticated(auth): Authenticated, Path(document_id): Path<Uuid>, ) -> impl IntoResponse { - let Some(pool) = state.db_pool.clone() else { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), - ) - .into_response(); - }; - run_contract_transition(pool, auth.owner_id, document_id, |pool, id| async move { + run_contract_transition(state, auth.owner_id, document_id, |pool, id| async move { repository::pause_contract(&pool, id).await }) .await @@ -829,14 +830,7 @@ pub async fn complete_contract( Path(document_id): Path<Uuid>, Json(req): Json<CompleteContractRequest>, ) -> impl IntoResponse { - let Some(pool) = state.db_pool.clone() else { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), - ) - .into_response(); - }; - run_contract_transition(pool, auth.owner_id, document_id, move |pool, id| async move { + run_contract_transition(state, auth.owner_id, document_id, move |pool, id| async move { repository::complete_contract(&pool, id, req.pr_url.as_deref(), req.pr_branch.as_deref()).await }) .await @@ -863,16 +857,39 @@ pub async fn unlock_contract( Authenticated(auth): Authenticated, Path(document_id): Path<Uuid>, ) -> impl IntoResponse { - let Some(pool) = state.db_pool.clone() else { - return ( - StatusCode::SERVICE_UNAVAILABLE, - Json(ApiError::new("DB_UNAVAILABLE", "Database not configured")), - ) - .into_response(); - }; - run_contract_transition(pool, auth.owner_id, document_id, |pool, id| async move { + run_contract_transition(state, auth.owner_id, document_id, |pool, id| async move { repository::unlock_contract(&pool, id).await }) .await .into_response() } + +/// Reopen a shipped contract for amendment. The contract goes back to +/// `active`, the parent directive flips to `active`, and the directive's +/// PR linkage + orchestrator task id are cleared so the reconciler +/// spawns a fresh planner. The planner reads +/// `get_latest_merged_revision` and frames the new plan as a delta on +/// top of the previously-merged PR. +#[utoipa::path( + post, + path = "/api/v1/contracts/{document_id}/reopen", + params(("document_id" = Uuid, Path, description = "Contract ID")), + responses( + (status = 200, description = "Contract reopened", body = crate::db::models::DirectiveDocument), + (status = 400, description = "Invalid state transition", body = ApiError), + (status = 404, description = "Not found", body = ApiError), + ), + security(("bearer_auth" = []), ("api_key" = [])), + tag = "Directive Documents" +)] +pub async fn reopen_contract( + State(state): State<SharedState>, + Authenticated(auth): Authenticated, + Path(document_id): Path<Uuid>, +) -> impl IntoResponse { + run_contract_transition(state, auth.owner_id, document_id, |pool, id| async move { + repository::reopen_contract(&pool, id).await + }) + .await + .into_response() +} diff --git a/makima/src/server/handlers/directives.rs b/makima/src/server/handlers/directives.rs index 6d99179..35a46a0 100644 --- a/makima/src/server/handlers/directives.rs +++ b/makima/src/server/handlers/directives.rs @@ -232,15 +232,51 @@ pub async fn update_directive( ); } - // Transition the contract to 'inactive' now that its - // iteration is "shipped" — editing the goal again starts - // an amendment cycle, surfaced via the New draft action. - if let Err(e) = repository::set_directive_inactive(pool, directive.id).await { - tracing::warn!( - directive_id = %directive.id, - error = %e, - "Failed to mark directive inactive after PR creation" - ); + // Auto-complete the active contract — flips its status + // to `shipped`, records pr_url/pr_branch, and (via the + // contract↔directive sync in repository) transitions + // the directive itself to `inactive`. This removes the + // need for a manual "Mark complete" click; the PR + // raise IS the completion signal. + match repository::get_active_contract_id_for_directive(pool, directive.id).await { + Ok(Some(contract_id)) => { + if let Err(e) = repository::complete_contract( + pool, + contract_id, + Some(new_pr_url.as_str()), + directive.pr_branch.as_deref(), + ) + .await + { + tracing::warn!( + directive_id = %directive.id, + contract_id = %contract_id, + error = %e, + "Failed to auto-complete contract after PR creation — \ + directive status not synced; user may need to manually ship" + ); + } + } + Ok(None) => { + // No active contract — fall back to the old + // behaviour (mark directive inactive). This is + // the legacy path for directives without + // contracts attached yet. + if let Err(e) = repository::set_directive_inactive(pool, directive.id).await { + tracing::warn!( + directive_id = %directive.id, + error = %e, + "Failed to mark directive inactive after PR creation" + ); + } + } + Err(e) => { + tracing::warn!( + directive_id = %directive.id, + error = %e, + "Failed to resolve active contract for auto-complete" + ); + } } } } diff --git a/makima/src/server/mod.rs b/makima/src/server/mod.rs index 604caea..a6c7787 100644 --- a/makima/src/server/mod.rs +++ b/makima/src/server/mod.rs @@ -257,6 +257,10 @@ pub fn make_router(state: SharedState) -> Router { post(directive_documents::unlock_contract), ) .route( + "/contracts/{document_id}/reopen", + post(directive_documents::reopen_contract), + ) + .route( "/contracts/{document_id}/tasks", get(directive_documents::list_document_tasks), ) diff --git a/makima/src/server/openapi.rs b/makima/src/server/openapi.rs index 437285f..5bbd0fe 100644 --- a/makima/src/server/openapi.rs +++ b/makima/src/server/openapi.rs @@ -127,6 +127,7 @@ use crate::server::messages::{ApiError, AudioEncoding, StartMessage, StopMessage directive_documents::pause_contract, directive_documents::complete_contract, directive_documents::unlock_contract, + directive_documents::reopen_contract, directive_documents::list_document_tasks, // Order endpoints orders::list_orders, |