mj_controller/server/api/
turns.rs

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

mj_controller/server/api/
turns.rs
