| 1 | //! `/v1/jobs` — the client-facing shell job surface. |
| 2 | //! |
| 3 | //! One authority: every job lives on the thread's shared `ShellManager`, the |
| 4 | //! same manager the thread's engine uses for model-launched shell work. Jobs |
| 5 | //! created here carry an `api:{thread_id}` owner scope so an engine's |
| 6 | //! per-session completion drain never claims client-launched work as model |
| 7 | //! evidence — and the model's jobs never appear "client-owned" here. |
| 8 | |
| 9 | use std::collections::HashMap; |
| 10 | |
| 11 | use axum::Json; |
| 12 | use axum::extract::{Path, Query, State}; |
| 13 | use axum::http::StatusCode; |
| 14 | use base64::Engine as _; |
| 15 | use serde::{Deserialize, Serialize}; |
| 16 | |
| 17 | use crate::tools::shell::{ |
| 18 | PtyDimensions, ShellJobSnapshot, ShellManager, ShellOutputChunk, ShellOutputStream, |
| 19 | ShellResult, ShellStatus, |
| 20 | }; |
| 21 | |
| 22 | use super::{ApiError, RuntimeApiState, map_thread_err}; |
| 23 | |
| 24 | /// `owner_session_id` prefix for jobs launched through this API. A real |
| 25 | /// engine session id can never carry it, so `*_for_session` drains stay |
| 26 | /// model-owned and API listings can tell the two apart. |
| 27 | const API_JOB_SCOPE_PREFIX: &str = "api:"; |
| 28 | |
| 29 | const COMMAND_MAX_BYTES: usize = 32 * 1024; |
| 30 | const ENV_MAX_ENTRIES: usize = 64; |
| 31 | const ENV_KEY_MAX_BYTES: usize = 128; |
| 32 | const ENV_VALUE_MAX_BYTES: usize = 8 * 1024; |
| 33 | const OUTPUT_CHUNK_DEFAULT: usize = 64 * 1024; |
| 34 | const OUTPUT_CHUNK_MAX: usize = 512 * 1024; |
| 35 | const OUTPUT_WAIT_MAX_MS: u64 = 30_000; |
| 36 | const STDIN_MAX_BYTES: usize = 64 * 1024; |
| 37 | const JOB_ID_MAX_BYTES: usize = 128; |
| 38 | |
| 39 | fn api_job_scope(thread_id: &str) -> String { |
| 40 | format!("{API_JOB_SCOPE_PREFIX}{thread_id}") |
| 41 | } |
| 42 | |
| 43 | fn job_owner(snapshot: &ShellJobSnapshot) -> &'static str { |
| 44 | if snapshot.owner_agent_id.is_some() { |
| 45 | "subagent" |
| 46 | } else if snapshot.owner_session_id.starts_with(API_JOB_SCOPE_PREFIX) { |
| 47 | "client" |
| 48 | } else { |
| 49 | "agent" |
| 50 | } |
| 51 | } |
| 52 | |
| 53 | fn map_job_err(error: anyhow::Error) -> ApiError { |
| 54 | let message = error.to_string(); |
| 55 | if message.ends_with("not found") { |
| 56 | ApiError::not_found(message) |
| 57 | } else { |
| 58 | ApiError::internal(message) |
| 59 | } |
| 60 | } |
| 61 | |
| 62 | #[derive(Debug, Serialize)] |
| 63 | pub(super) struct JobView { |
| 64 | #[serde(flatten)] |
| 65 | snapshot: ShellJobSnapshot, |
| 66 | thread_id: String, |
| 67 | /// `client` = launched through this API, `agent` = launched by the model's |
| 68 | /// shell tool, `subagent` = owned by a delegated agent. |
| 69 | owner: &'static str, |
| 70 | /// Null for historical records whose original transport is not known. |
| 71 | tty: Option<bool>, |
| 72 | terminal_size: Option<PtyDimensions>, |
| 73 | } |
| 74 | |
| 75 | impl JobView { |
| 76 | fn new(snapshot: ShellJobSnapshot, thread_id: String, manager: &ShellManager) -> Self { |
| 77 | let owner = job_owner(&snapshot); |
| 78 | let terminal_size = manager.job_terminal_size(&snapshot.job_id); |
| 79 | let tty = (!snapshot.stale).then_some(terminal_size.is_some()); |
| 80 | Self { |
| 81 | snapshot, |
| 82 | thread_id, |
| 83 | owner, |
| 84 | tty, |
| 85 | terminal_size, |
| 86 | } |
| 87 | } |
| 88 | } |
| 89 | |
| 90 | #[derive(Debug, Serialize)] |
| 91 | pub(super) struct JobListResponse { |
| 92 | jobs: Vec<JobView>, |
| 93 | } |
| 94 | |
| 95 | /// `GET /v1/jobs` — every live and known-stale job across all threads. |
| 96 | pub(super) async fn list_jobs( |
| 97 | State(state): State<RuntimeApiState>, |
| 98 | ) -> Result<Json<JobListResponse>, ApiError> { |
| 99 | let managers = state.runtime_threads.shell_managers_snapshot().await; |
| 100 | let jobs = tokio::task::spawn_blocking(move || { |
| 101 | let mut jobs = Vec::new(); |
| 102 | for (thread_id, manager) in managers { |
| 103 | let mut guard = manager.lock().unwrap_or_else(|e| e.into_inner()); |
| 104 | jobs.extend( |
| 105 | guard |
| 106 | .list_jobs() |
| 107 | .into_iter() |
| 108 | .map(|snapshot| JobView::new(snapshot, thread_id.clone(), &guard)), |
| 109 | ); |
| 110 | } |
| 111 | jobs |
| 112 | }) |
| 113 | .await |
| 114 | .map_err(|_| ApiError::internal("job listing failed"))?; |
| 115 | Ok(Json(JobListResponse { jobs })) |
| 116 | } |
| 117 | |
| 118 | async fn thread_manager( |
| 119 | state: &RuntimeApiState, |
| 120 | thread_id: &str, |
| 121 | create: bool, |
| 122 | ) -> Result<crate::tools::shell::SharedShellManager, ApiError> { |
| 123 | state |
| 124 | .runtime_threads |
| 125 | .thread_shell_manager(thread_id, create) |
| 126 | .await |
| 127 | .map_err(map_thread_err)? |
| 128 | .ok_or_else(|| ApiError::not_found(format!("thread {thread_id} has no jobs"))) |
| 129 | } |
| 130 | |
| 131 | /// `GET /v1/threads/{id}/jobs` — all jobs owned by one thread's manager: |
| 132 | /// model-launched, subagent-launched, and client-launched together. |
| 133 | pub(super) async fn list_thread_jobs( |
| 134 | State(state): State<RuntimeApiState>, |
| 135 | Path(thread_id): Path<String>, |
| 136 | ) -> Result<Json<JobListResponse>, ApiError> { |
| 137 | let Some(manager) = state |
| 138 | .runtime_threads |
| 139 | .thread_shell_manager(&thread_id, false) |
| 140 | .await |
| 141 | .map_err(map_thread_err)? |
| 142 | else { |
| 143 | return Ok(Json(JobListResponse { jobs: Vec::new() })); |
| 144 | }; |
| 145 | let jobs = tokio::task::spawn_blocking(move || { |
| 146 | let mut guard = manager.lock().unwrap_or_else(|e| e.into_inner()); |
| 147 | guard |
| 148 | .list_jobs() |
| 149 | .into_iter() |
| 150 | .map(|snapshot| JobView::new(snapshot, thread_id.clone(), &guard)) |
| 151 | .collect() |
| 152 | }) |
| 153 | .await |
| 154 | .map_err(|_| ApiError::internal("job listing failed"))?; |
| 155 | Ok(Json(JobListResponse { jobs })) |
| 156 | } |
| 157 | |
| 158 | #[derive(Deserialize)] |
| 159 | #[serde(deny_unknown_fields)] |
| 160 | pub(super) struct CreateJobRequest { |
| 161 | command: String, |
| 162 | /// Working directory. Omitted = the thread's workspace. |
| 163 | cwd: Option<String>, |
| 164 | /// Bounds the foreground-wait contract inside the manager; background |
| 165 | /// jobs are never killed at timeout. |
| 166 | timeout_ms: Option<u64>, |
| 167 | /// Run under a PTY: stderr merges into stdout and the command sees a |
| 168 | /// terminal. Required for interactive programs. |
| 169 | #[serde(default)] |
| 170 | tty: bool, |
| 171 | #[serde(default)] |
| 172 | env: HashMap<String, String>, |
| 173 | } |
| 174 | |
| 175 | #[derive(Debug, Serialize)] |
| 176 | pub(super) struct CreateJobResponse { |
| 177 | job: JobView, |
| 178 | } |
| 179 | |
| 180 | /// `POST /v1/threads/{id}/jobs` — launch a client-owned background job under |
| 181 | /// the thread's own sandbox policy. The client asking is the approval; the |
| 182 | /// thread's posture still bounds what the job may touch. |
| 183 | pub(super) async fn create_thread_job( |
| 184 | State(state): State<RuntimeApiState>, |
| 185 | Path(thread_id): Path<String>, |
| 186 | Json(request): Json<CreateJobRequest>, |
| 187 | ) -> Result<(StatusCode, Json<CreateJobResponse>), ApiError> { |
| 188 | let command = request.command.trim(); |
| 189 | if command.is_empty() { |
| 190 | return Err(ApiError::bad_request("command is required")); |
| 191 | } |
| 192 | if command.len() > COMMAND_MAX_BYTES { |
| 193 | return Err(ApiError::bad_request(format!( |
| 194 | "command must be at most {COMMAND_MAX_BYTES} bytes" |
| 195 | ))); |
| 196 | } |
| 197 | if request.env.len() > ENV_MAX_ENTRIES { |
| 198 | return Err(ApiError::bad_request(format!( |
| 199 | "env may carry at most {ENV_MAX_ENTRIES} entries" |
| 200 | ))); |
| 201 | } |
| 202 | for (key, value) in &request.env { |
| 203 | if key.len() > ENV_KEY_MAX_BYTES || key.contains(['=', '\0']) { |
| 204 | return Err(ApiError::bad_request("invalid env key")); |
| 205 | } |
| 206 | if value.len() > ENV_VALUE_MAX_BYTES || value.contains('\0') { |
| 207 | return Err(ApiError::bad_request("invalid env value")); |
| 208 | } |
| 209 | } |
| 210 | |
| 211 | let thread = state |
| 212 | .runtime_threads |
| 213 | .get_thread(&thread_id) |
| 214 | .await |
| 215 | .map_err(map_thread_err)?; |
| 216 | if !thread.allow_shell { |
| 217 | return Err(ApiError::forbidden( |
| 218 | "this thread does not allow shell commands", |
| 219 | )); |
| 220 | } |
| 221 | state |
| 222 | .runtime_threads |
| 223 | .validate_shell_access_policy( |
| 224 | &thread.workspace, |
| 225 | state.config_path.as_deref(), |
| 226 | state.config_profile.as_deref(), |
| 227 | ) |
| 228 | .await |
| 229 | .map_err(|error| ApiError::forbidden(error.to_string()))?; |
| 230 | if let Some(cwd) = request.cwd.as_deref() { |
| 231 | let resolved = std::path::Path::new(cwd); |
| 232 | let resolved = if resolved.is_absolute() { |
| 233 | resolved.to_path_buf() |
| 234 | } else { |
| 235 | thread.workspace.join(resolved) |
| 236 | }; |
| 237 | if !resolved.is_dir() { |
| 238 | return Err(ApiError::bad_request("cwd must be an existing directory")); |
| 239 | } |
| 240 | } |
| 241 | let policy = state |
| 242 | .runtime_threads |
| 243 | .thread_job_sandbox_policy(&thread) |
| 244 | .await; |
| 245 | let manager = thread_manager(&state, &thread_id, true).await?; |
| 246 | let scope = api_job_scope(&thread_id); |
| 247 | let request_timeout = request.timeout_ms; |
| 248 | let request_tty = request.tty; |
| 249 | let request_env = request.env; |
| 250 | let request_cwd = request.cwd; |
| 251 | let command = command.to_string(); |
| 252 | let job = tokio::task::spawn_blocking(move || -> Result<JobView, ApiError> { |
| 253 | let mut guard = manager.lock().unwrap_or_else(|e| e.into_inner()); |
| 254 | let result = guard |
| 255 | .execute_with_options_env_for_session( |
| 256 | &command, |
| 257 | request_cwd.as_deref(), |
| 258 | request_timeout.unwrap_or(120_000), |
| 259 | true, |
| 260 | None, |
| 261 | request_tty, |
| 262 | Some(policy), |
| 263 | request_env, |
| 264 | &scope, |
| 265 | ) |
| 266 | .map_err(|error| ApiError::internal(format!("job launch failed: {error}")))?; |
| 267 | let task_id = result.task_id.clone().unwrap_or_default(); |
| 268 | let snapshot = guard |
| 269 | .inspect_job(&task_id) |
| 270 | .map_err(|error| { |
| 271 | ApiError::internal(format!("job launched but is not tracked: {error}")) |
| 272 | })? |
| 273 | .snapshot; |
| 274 | Ok(JobView::new(snapshot, thread_id, &guard)) |
| 275 | }) |
| 276 | .await |
| 277 | .map_err(|_| ApiError::internal("job launch failed"))??; |
| 278 | Ok((StatusCode::CREATED, Json(CreateJobResponse { job }))) |
| 279 | } |
| 280 | |
| 281 | #[derive(Debug, Serialize)] |
| 282 | pub(super) struct JobDetailResponse { |
| 283 | job: JobView, |
| 284 | stdout_tail: String, |
| 285 | stderr_tail: String, |
| 286 | } |
| 287 | |
| 288 | /// `GET /v1/threads/{id}/jobs/{job_id}` — snapshot plus the retained output |
| 289 | /// tails. For the full stream, follow `output` with a cursor instead. |
| 290 | pub(super) async fn get_thread_job( |
| 291 | State(state): State<RuntimeApiState>, |
| 292 | Path((thread_id, job_id)): Path<(String, String)>, |
| 293 | ) -> Result<Json<JobDetailResponse>, ApiError> { |
| 294 | if job_id.len() > JOB_ID_MAX_BYTES { |
| 295 | return Err(ApiError::not_found("job not found")); |
| 296 | } |
| 297 | let manager = thread_manager(&state, &thread_id, false).await?; |
| 298 | tokio::task::spawn_blocking(move || { |
| 299 | let mut guard = manager.lock().unwrap_or_else(|e| e.into_inner()); |
| 300 | let detail = guard.inspect_job(&job_id).map_err(map_job_err)?; |
| 301 | Ok(Json(JobDetailResponse { |
| 302 | job: JobView::new(detail.snapshot, thread_id, &guard), |
| 303 | stdout_tail: detail.stdout, |
| 304 | stderr_tail: detail.stderr, |
| 305 | })) |
| 306 | }) |
| 307 | .await |
| 308 | .map_err(|_| ApiError::internal("job inspect failed"))? |
| 309 | } |
| 310 | |
| 311 | #[derive(Deserialize)] |
| 312 | #[serde(deny_unknown_fields)] |
| 313 | pub(super) struct JobOutputQuery { |
| 314 | /// `stdout` (default) or `stderr`; PTY jobs merge stderr into stdout. |
| 315 | #[serde(default)] |
| 316 | stream: Option<String>, |
| 317 | /// Absolute byte offset into the stream's lifetime output. |
| 318 | #[serde(default)] |
| 319 | cursor: Option<usize>, |
| 320 | /// Per-request byte ceiling, default 64 KiB, max 512 KiB. |
| 321 | #[serde(default)] |
| 322 | max_bytes: Option<usize>, |
| 323 | /// Long-poll bound for new bytes on a running job, max 30s. |
| 324 | #[serde(default)] |
| 325 | wait_ms: Option<u64>, |
| 326 | /// `base64` (default, exact bytes) or `text` (lossy UTF-8). |
| 327 | #[serde(default)] |
| 328 | format: Option<String>, |
| 329 | } |
| 330 | |
| 331 | #[derive(Debug, Serialize)] |
| 332 | pub(super) struct JobOutputResponse { |
| 333 | job_id: String, |
| 334 | stream: &'static str, |
| 335 | /// Absolute offset of `data[0]`; exceeds `cursor` when the bounded buffer |
| 336 | /// already discarded that prefix (`dropped` reports the cutoff). |
| 337 | offset: usize, |
| 338 | /// Next cursor: pass it back to continue the stream. |
| 339 | next_cursor: usize, |
| 340 | /// Total bytes the stream has produced, including discarded bytes. |
| 341 | total: usize, |
| 342 | /// Leading bytes permanently discarded by the in-flight bound. |
| 343 | dropped: usize, |
| 344 | encoding: &'static str, |
| 345 | data: String, |
| 346 | status: ShellStatus, |
| 347 | exit_code: Option<i64>, |
| 348 | /// Terminal status and no bytes remain past `next_cursor`. |
| 349 | done: bool, |
| 350 | } |
| 351 | |
| 352 | /// `GET /v1/threads/{id}/jobs/{job_id}/output` — the resumable byte stream. |
| 353 | /// Reads are non-consuming: several clients may hold independent cursors, and |
| 354 | /// polling here never steals output from the engine's own delta consumer. |
| 355 | pub(super) async fn get_thread_job_output( |
| 356 | State(state): State<RuntimeApiState>, |
| 357 | Path((thread_id, job_id)): Path<(String, String)>, |
| 358 | Query(query): Query<JobOutputQuery>, |
| 359 | ) -> Result<Json<JobOutputResponse>, ApiError> { |
| 360 | if job_id.len() > JOB_ID_MAX_BYTES { |
| 361 | return Err(ApiError::not_found("job not found")); |
| 362 | } |
| 363 | let (stream, stream_name) = match query.stream.as_deref().unwrap_or("stdout") { |
| 364 | "stdout" => (ShellOutputStream::Stdout, "stdout"), |
| 365 | "stderr" => (ShellOutputStream::Stderr, "stderr"), |
| 366 | _ => return Err(ApiError::bad_request("stream must be stdout or stderr")), |
| 367 | }; |
| 368 | let cursor = query.cursor.unwrap_or(0); |
| 369 | let max_bytes = query.max_bytes.unwrap_or(OUTPUT_CHUNK_DEFAULT); |
| 370 | if !(1..=OUTPUT_CHUNK_MAX).contains(&max_bytes) { |
| 371 | return Err(ApiError::bad_request(format!( |
| 372 | "max_bytes must be between 1 and {OUTPUT_CHUNK_MAX}" |
| 373 | ))); |
| 374 | } |
| 375 | let wait_ms = query.wait_ms.unwrap_or(0).min(OUTPUT_WAIT_MAX_MS); |
| 376 | let format = query.format.as_deref().unwrap_or("base64"); |
| 377 | if !matches!(format, "base64" | "text") { |
| 378 | return Err(ApiError::bad_request("format must be base64 or text")); |
| 379 | } |
| 380 | let manager = thread_manager(&state, &thread_id, false).await?; |
| 381 | let chunk = tokio::task::spawn_blocking({ |
| 382 | let job_id = job_id.clone(); |
| 383 | move || -> Result<ShellOutputChunk, ApiError> { |
| 384 | // Wait between non-consuming snapshots, never while owning the |
| 385 | // thread's shared ShellManager. Input, resize, and kill stay live. |
| 386 | let deadline = std::time::Instant::now() + std::time::Duration::from_millis(wait_ms); |
| 387 | loop { |
| 388 | let chunk = { |
| 389 | let mut guard = manager.lock().unwrap_or_else(|e| e.into_inner()); |
| 390 | guard |
| 391 | .read_output_chunk(&job_id, stream, cursor, max_bytes, 0) |
| 392 | .map_err(map_job_err)? |
| 393 | }; |
| 394 | let remaining = deadline.saturating_duration_since(std::time::Instant::now()); |
| 395 | if chunk.total > cursor |
| 396 | || chunk.status != ShellStatus::Running |
| 397 | || remaining.is_zero() |
| 398 | { |
| 399 | break Ok(chunk); |
| 400 | } |
| 401 | std::thread::sleep(remaining.min(std::time::Duration::from_millis(50))); |
| 402 | } |
| 403 | } |
| 404 | }) |
| 405 | .await |
| 406 | .map_err(|_| ApiError::internal("job output read failed"))??; |
| 407 | Ok(Json(encode_chunk(&job_id, stream_name, chunk, format))) |
| 408 | } |
| 409 | |
| 410 | fn encode_chunk( |
| 411 | job_id: &str, |
| 412 | stream_name: &'static str, |
| 413 | chunk: ShellOutputChunk, |
| 414 | format: &str, |
| 415 | ) -> JobOutputResponse { |
| 416 | let (encoding, data) = match format { |
| 417 | "text" => ("utf-8", String::from_utf8_lossy(&chunk.bytes).into_owned()), |
| 418 | _ => ( |
| 419 | "base64", |
| 420 | base64::engine::general_purpose::STANDARD.encode(&chunk.bytes), |
| 421 | ), |
| 422 | }; |
| 423 | let done = chunk.status != ShellStatus::Running && chunk.next_offset >= chunk.total; |
| 424 | JobOutputResponse { |
| 425 | job_id: job_id.to_string(), |
| 426 | stream: stream_name, |
| 427 | offset: chunk.offset, |
| 428 | next_cursor: chunk.next_offset, |
| 429 | total: chunk.total, |
| 430 | dropped: chunk.dropped, |
| 431 | encoding, |
| 432 | data, |
| 433 | status: chunk.status, |
| 434 | exit_code: chunk.exit_code, |
| 435 | done, |
| 436 | } |
| 437 | } |
| 438 | |
| 439 | #[derive(Deserialize)] |
| 440 | #[serde(deny_unknown_fields)] |
| 441 | pub(super) struct JobStdinRequest { |
| 442 | /// UTF-8 text (default) or base64 for arbitrary bytes. |
| 443 | data: String, |
| 444 | #[serde(default)] |
| 445 | encoding: Option<String>, |
| 446 | /// Close stdin after writing (EOF). |
| 447 | #[serde(default)] |
| 448 | close: bool, |
| 449 | } |
| 450 | |
| 451 | /// `POST /v1/threads/{id}/jobs/{job_id}/stdin` — write to a running job's |
| 452 | /// stdin. Works for PTY and piped jobs alike. |
| 453 | pub(super) async fn write_thread_job_stdin( |
| 454 | State(state): State<RuntimeApiState>, |
| 455 | Path((thread_id, job_id)): Path<(String, String)>, |
| 456 | Json(request): Json<JobStdinRequest>, |
| 457 | ) -> Result<StatusCode, ApiError> { |
| 458 | if job_id.len() > JOB_ID_MAX_BYTES { |
| 459 | return Err(ApiError::not_found("job not found")); |
| 460 | } |
| 461 | let input = match request.encoding.as_deref().unwrap_or("utf-8") { |
| 462 | "utf-8" => { |
| 463 | if request.data.len() > STDIN_MAX_BYTES { |
| 464 | return Err(ApiError::bad_request(format!( |
| 465 | "data must be at most {STDIN_MAX_BYTES} bytes" |
| 466 | ))); |
| 467 | } |
| 468 | request.data.into_bytes() |
| 469 | } |
| 470 | "base64" => { |
| 471 | if request.data.len() > STDIN_MAX_BYTES * 2 { |
| 472 | return Err(ApiError::bad_request("data exceeds the stdin limit")); |
| 473 | } |
| 474 | let bytes = base64::engine::general_purpose::STANDARD |
| 475 | .decode(&request.data) |
| 476 | .map_err(|_| ApiError::bad_request("data is not valid base64"))?; |
| 477 | if bytes.len() > STDIN_MAX_BYTES { |
| 478 | return Err(ApiError::bad_request(format!( |
| 479 | "data must be at most {STDIN_MAX_BYTES} decoded bytes" |
| 480 | ))); |
| 481 | } |
| 482 | bytes |
| 483 | } |
| 484 | _ => return Err(ApiError::bad_request("encoding must be utf-8 or base64")), |
| 485 | }; |
| 486 | let close = request.close; |
| 487 | let manager = thread_manager(&state, &thread_id, false).await?; |
| 488 | tokio::task::spawn_blocking(move || { |
| 489 | let mut guard = manager.lock().unwrap_or_else(|e| e.into_inner()); |
| 490 | guard |
| 491 | .write_stdin_bytes(&job_id, &input, close) |
| 492 | .map_err(map_job_err) |
| 493 | }) |
| 494 | .await |
| 495 | .map_err(|_| ApiError::internal("job stdin write failed"))??; |
| 496 | Ok(StatusCode::NO_CONTENT) |
| 497 | } |
| 498 | |
| 499 | #[derive(Debug, Serialize)] |
| 500 | pub(super) struct KillJobResponse { |
| 501 | job: JobView, |
| 502 | result: ShellResult, |
| 503 | } |
| 504 | |
| 505 | /// `POST /v1/threads/{id}/jobs/{job_id}/kill` — bounded SIGTERM → SIGKILL |
| 506 | /// escalation on the whole process group; the final snapshot rides along. |
| 507 | pub(super) async fn kill_thread_job( |
| 508 | State(state): State<RuntimeApiState>, |
| 509 | Path((thread_id, job_id)): Path<(String, String)>, |
| 510 | ) -> Result<Json<KillJobResponse>, ApiError> { |
| 511 | if job_id.len() > JOB_ID_MAX_BYTES { |
| 512 | return Err(ApiError::not_found("job not found")); |
| 513 | } |
| 514 | let manager = thread_manager(&state, &thread_id, false).await?; |
| 515 | tokio::task::spawn_blocking(move || { |
| 516 | let mut guard = manager.lock().unwrap_or_else(|e| e.into_inner()); |
| 517 | let result = guard.kill(&job_id).map_err(map_job_err)?; |
| 518 | let snapshot = guard.inspect_job(&job_id).map_err(map_job_err)?.snapshot; |
| 519 | Ok(Json(KillJobResponse { |
| 520 | job: JobView::new(snapshot, thread_id, &guard), |
| 521 | result, |
| 522 | })) |
| 523 | }) |
| 524 | .await |
| 525 | .map_err(|_| ApiError::internal("job kill failed"))? |
| 526 | } |
| 527 | |
| 528 | /// `POST /v1/threads/{id}/jobs/{job_id}/resize` — resize this existing PTY. |
| 529 | pub(super) async fn resize_thread_job( |
| 530 | State(state): State<RuntimeApiState>, |
| 531 | Path((thread_id, job_id)): Path<(String, String)>, |
| 532 | Json(size): Json<PtyDimensions>, |
| 533 | ) -> Result<Json<JobDetailResponse>, ApiError> { |
| 534 | let size = size |
| 535 | .validate() |
| 536 | .map_err(|error| ApiError::bad_request(error.to_string()))?; |
| 537 | if job_id.len() > JOB_ID_MAX_BYTES { |
| 538 | return Err(ApiError::not_found("job not found")); |
| 539 | } |
| 540 | let manager = thread_manager(&state, &thread_id, false).await?; |
| 541 | tokio::task::spawn_blocking(move || { |
| 542 | let mut guard = manager.lock().unwrap_or_else(|e| e.into_inner()); |
| 543 | let detail = guard.inspect_job(&job_id).map_err(map_job_err)?; |
| 544 | if detail.snapshot.stale || detail.snapshot.status != ShellStatus::Running { |
| 545 | return Err(ApiError { |
| 546 | status: StatusCode::CONFLICT, |
| 547 | message: "This job is no longer running".into(), |
| 548 | }); |
| 549 | } |
| 550 | if guard.job_terminal_size(&job_id).is_none() { |
| 551 | return Err(ApiError::bad_request("This job is not a PTY")); |
| 552 | } |
| 553 | guard.resize_pty(&job_id, size).map_err(map_job_err)?; |
| 554 | let detail = guard.inspect_job(&job_id).map_err(map_job_err)?; |
| 555 | Ok(Json(JobDetailResponse { |
| 556 | job: JobView::new(detail.snapshot, thread_id, &guard), |
| 557 | stdout_tail: detail.stdout, |
| 558 | stderr_tail: detail.stderr, |
| 559 | })) |
| 560 | }) |
| 561 | .await |
| 562 | .map_err(|_| ApiError::internal("PTY resize failed"))? |
| 563 | } |
| 564 |