mj_controller/server/api/
turns.rs

1use super::*;
2
3pub(super) async fn prompt(
4    State(state): State<ServerState>,
5    Path(session_id): Path<String>,
6    Json(request): Json<PromptRequest>,
7) -> Result<(StatusCode, Json<PromptResponse>), ApiFailure> {
8    let backend = backend(&state)?.clone();
9    {
10        let snapshot = state.snapshot_rx.borrow();
11        let action = ControllerAction::Prompt {
12            session_id: session_id.clone(),
13            text: request.text.clone(),
14            images: Vec::new(),
15        };
16        validate_action(&action, &snapshot)?;
17        let session = require_session_record(&snapshot, &session_id)?;
18        if !session.capabilities.prompt {
19            return Err(ApiFailure::conflict(
20                "this session cannot take a prompt right now",
21            ));
22        }
23    }
24    let turn_id = backend.prompt(session_id, request.text).await?;
25    Ok((StatusCode::ACCEPTED, Json(PromptResponse { turn_id })))
26}
27
28/// Page through a session's transcript.
29///
30/// It reads the durable projection rather than the live actor, so it answers
31/// the same way while a session runs and long after it stopped.
32#[derive(Debug, Clone, Default, Serialize, Deserialize)]
33pub struct UsageQuery {
34    pub after_seq: Option<u64>,
35    pub limit: Option<usize>,
36}
37
38pub(super) async fn usage(
39    State(state): State<ServerState>,
40    Path(session_id): Path<String>,
41    Query(query): Query<UsageQuery>,
42) -> Result<Json<crate::database::UsagePage>, ApiFailure> {
43    let page = backend(&state)?
44        .usage(
45            session_id,
46            query.after_seq.unwrap_or(0),
47            query.limit.unwrap_or(200).clamp(1, 1000),
48        )
49        .await?
50        .ok_or_else(|| ApiFailure::not_found("no usage history is recorded for that session"))?;
51    Ok(Json(page))
52}
53
54pub(super) async fn transcript(
55    State(state): State<ServerState>,
56    Path(session_id): Path<String>,
57    Query(query): Query<TranscriptQuery>,
58) -> Result<Json<TranscriptResponse>, ApiFailure> {
59    let backend = backend(&state)?.clone();
60    let limit = query
61        .limit
62        .unwrap_or(DEFAULT_TRANSCRIPT_LIMIT)
63        .clamp(1, MAX_TRANSCRIPT_LIMIT);
64    let page = backend
65        .transcript(
66            session_id.clone(),
67            query.after_seq.unwrap_or(0),
68            limit,
69            query.role,
70        )
71        .await?
72        .ok_or_else(|| ApiFailure::not_found("no transcript is recorded for that session"))?;
73    Ok(Json(TranscriptResponse {
74        next_after_seq: page.next_after_seq,
75        session_id,
76        latest_seq: page.latest_seq,
77        execution: page.execution,
78        items: page
79            .items
80            .iter()
81            .map(|item| TranscriptItemView {
82                stable_id: item.stable_id.clone(),
83                position: item.position,
84                seq: item.seq(),
85                role: mj_core::transcript::transcript_item_role(&item.body).to_owned(),
86                text: mj_transcript::transcript::transcript_item_text(item),
87                created_at_ms: item.created_at_ms,
88                last_changed_at_ms: item.last_changed_at_ms,
89                body: item.body.clone(),
90            })
91            .collect(),
92    }))
93}
94
95pub(super) async fn close(
96    State(state): State<ServerState>,
97    Path(session_id): Path<String>,
98    request: Option<Json<CloseRequest>>,
99) -> Result<StatusCode, ApiFailure> {
100    let force = request.as_ref().is_some_and(|request| request.force);
101    let active_children = if force {
102        // A force close destroys the children with the parent, so an active
103        // child is not a reason to refuse it.
104        0
105    } else {
106        let snapshot = state.snapshot_rx.borrow();
107        let session = require_session_record(&snapshot, &session_id)?;
108        session
109            .subagent_session_ids
110            .iter()
111            .filter(|child_id| {
112                snapshot.sessions.iter().any(|child| {
113                    child.id == child_id.as_str()
114                        && !matches!(
115                            child.state.as_str(),
116                            "stopped" | "lost" | "error" | "destroyed-with-data-loss"
117                        )
118                })
119            })
120            .count()
121    };
122    if active_children > 0
123        && !request
124            .as_ref()
125            .is_some_and(|request| request.acknowledge_active_subagents)
126    {
127        return Err(ApiFailure::conflict(format!(
128            "session has {} sub-agent(s); retry with acknowledge_active_subagents=true to stop children first",
129            active_children
130        )));
131    }
132    backend(&state)?.cancel_start(session_id.clone()).await?;
133    if force {
134        return send_action(&state, ControllerAction::ForceClose { session_id }).await;
135    }
136    send_action(&state, ControllerAction::Close { session_id }).await
137}
138
139#[derive(Debug, Default, serde::Deserialize)]
140#[serde(deny_unknown_fields)]
141pub(super) struct CloseRequest {
142    #[serde(default)]
143    pub(super) acknowledge_active_subagents: bool,
144    /// Destroy the session instead of checkpointing it. Irreversible.
145    #[serde(default)]
146    pub(super) force: bool,
147}
148
149pub(super) async fn cancel_turn(
150    State(state): State<ServerState>,
151    Path(session_id): Path<String>,
152) -> Result<StatusCode, ApiFailure> {
153    send_action(&state, ControllerAction::CancelTurn { session_id }).await
154}
155
156pub(super) async fn send_action(
157    state: &ServerState,
158    action: ControllerAction,
159) -> Result<StatusCode, ApiFailure> {
160    validate_action(&action, &state.snapshot_rx.borrow())?;
161    let (reply, outcome) = tokio::sync::oneshot::channel();
162    state
163        .action_tx
164        .send(ControllerRequest { action, reply })
165        .await
166        .map_err(|_| ApiFailure::unavailable("the controller is not accepting actions"))?;
167    let outcome = outcome
168        .await
169        .map_err(|_| ApiFailure::unavailable("the controller dropped this action"))?;
170    match outcome.rejection() {
171        Some(rejection) => Err(rejection.into()),
172        None => Ok(StatusCode::ACCEPTED),
173    }
174}
mj_controller/server/api/turns.rs

mj_controller/server/api/
turns.rs
