Omar Sobh
202e8535f0
The `decide()` helper (used by both approve and reject) wrapped `runtime.resume_run(ready).await` inside `tokio::spawn`, so the handler returned 200 to the client before the resume had even started. When a client (or test) immediately reconnected to /api/gateway with resumeFrom, the run's broadcast channel didn't exist yet — `runtime.subscribe(run_id)` returned None, the journal was still empty (no new events yet), and the SSE stream closed with zero events. Symptom: approvals_api tests flaky in CI (`unwrap on None` at line 282 of reject_over_http_executes_nothing, sometimes `missing step_finished` on the accept path). resume_run's *own* awaited portion is only the setup — claim_resume, checkpoint load, and open_channel. It internally spawns the long-running multi-step work. Awaiting it inline means we wait milliseconds for setup, then return. By the time the client reconnects, the channel exists, so subscribe() finds it and live-tail works. The sweeper still handles the crash-between-decision-and-resume case for durability.
96 lines
3.0 KiB
Rust
96 lines
3.0 KiB
Rust
use axum::extract::{Path, State};
|
|
use axum::Json;
|
|
use cm_safety::{approvals, Approval, Decision, ResumeReady, SafetyError};
|
|
use serde_json::{json, Value};
|
|
use uuid::Uuid;
|
|
|
|
use crate::{ApiError, AppState, Authed};
|
|
|
|
impl From<SafetyError> for ApiError {
|
|
fn from(err: SafetyError) -> Self {
|
|
match err {
|
|
SafetyError::NotFound => ApiError::NotFound,
|
|
SafetyError::AlreadyDecided => ApiError::Conflict,
|
|
_ => ApiError::Internal,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Loads an approval, hiding other workspaces' approvals entirely.
|
|
async fn workspace_approval(
|
|
state: &AppState,
|
|
user: &cm_auth::AuthedUser,
|
|
id: Uuid,
|
|
) -> Result<Approval, ApiError> {
|
|
let approval = approvals::get(&state.pool, id).await?;
|
|
if approval.workspace_id != user.workspace_id {
|
|
return Err(ApiError::NotFound);
|
|
}
|
|
Ok(approval)
|
|
}
|
|
|
|
/// GET /api/approvals — the pending review queue (§10), oldest first.
|
|
pub async fn list(
|
|
State(state): State<AppState>,
|
|
Authed(user): Authed,
|
|
) -> Result<Json<Vec<Approval>>, ApiError> {
|
|
let pending = approvals::list_pending(&state.pool, user.workspace_id).await?;
|
|
Ok(Json(pending))
|
|
}
|
|
|
|
/// GET /api/approvals/{id}
|
|
pub async fn get(
|
|
State(state): State<AppState>,
|
|
Authed(user): Authed,
|
|
Path(id): Path<Uuid>,
|
|
) -> Result<Json<Approval>, ApiError> {
|
|
Ok(Json(workspace_approval(&state, &user, id).await?))
|
|
}
|
|
|
|
async fn decide(
|
|
state: AppState,
|
|
user: cm_auth::AuthedUser,
|
|
id: Uuid,
|
|
decision: Decision,
|
|
) -> Result<Json<Value>, ApiError> {
|
|
workspace_approval(&state, &user, id).await?;
|
|
let approval = approvals::decide(&state.pool, id, user.user_id, decision).await?;
|
|
|
|
// Kick the resume before returning. resume_run's awaited portion is only
|
|
// the setup (claim + checkpoint load + open the broadcast channel); it
|
|
// spawns the actual multi-step work internally, so this doesn't block the
|
|
// client on the run itself — but it does guarantee the channel exists by
|
|
// the time the client reconnects via /api/gateway, closing the observable
|
|
// race where subscribe(run_id) would return None. The sweeper remains the
|
|
// durable fallback if this process dies before setup completes.
|
|
let ready = ResumeReady {
|
|
run_id: approval.run_id,
|
|
approval_id: approval.id,
|
|
approved: decision == Decision::Approve,
|
|
};
|
|
let _ = state.runtime.resume_run(ready).await;
|
|
|
|
Ok(Json(
|
|
json!({ "id": approval.id, "status": approval.status }),
|
|
))
|
|
}
|
|
|
|
/// POST /api/approvals/{id}/approve — executes the gated action (§15:
|
|
/// approve-only execution, idempotent, audited).
|
|
pub async fn approve(
|
|
State(state): State<AppState>,
|
|
Authed(user): Authed,
|
|
Path(id): Path<Uuid>,
|
|
) -> Result<Json<Value>, ApiError> {
|
|
decide(state, user, id, Decision::Approve).await
|
|
}
|
|
|
|
/// POST /api/approvals/{id}/reject — the agent learns the refusal in-band.
|
|
pub async fn reject(
|
|
State(state): State<AppState>,
|
|
Authed(user): Authed,
|
|
Path(id): Path<Uuid>,
|
|
) -> Result<Json<Value>, ApiError> {
|
|
decide(state, user, id, Decision::Reject).await
|
|
}
|