| 1 | //! Bounded supervision over the existing manager and persisted continuation map. |
| 2 | use super::*; |
| 3 | |
| 4 | pub(super) const COMPACT_STATUS_BYTES: usize = 8192; |
| 5 | pub(super) const COMPACT_SPAWN_BYTES: usize = 4096; |
| 6 | const DETAIL_STATUS_BYTES: usize = 32 * 1024; |
| 7 | |
| 8 | pub(super) fn text_preview(value: &str, bytes: usize) -> String { |
| 9 | if value.len() <= bytes { |
| 10 | return value.to_string(); |
| 11 | } |
| 12 | let mut end = bytes.saturating_sub(3).min(value.len()); |
| 13 | while !value.is_char_boundary(end) { |
| 14 | end -= 1; |
| 15 | } |
| 16 | format!("{}...", &value[..end]) |
| 17 | } |
| 18 | |
| 19 | pub(super) fn page(input: &Value) -> Result<(usize, usize), ToolError> { |
| 20 | let number = |key: &str, default| -> Result<usize, ToolError> { |
| 21 | match input.get(key) { |
| 22 | None => Ok(default), |
| 23 | Some(value) => value |
| 24 | .as_u64() |
| 25 | .and_then(|value| usize::try_from(value).ok()) |
| 26 | .ok_or_else(|| { |
| 27 | ToolError::invalid_input(format!("{key} must be a nonnegative integer")) |
| 28 | }), |
| 29 | } |
| 30 | }; |
| 31 | let offset = number("offset", 0)?; |
| 32 | let limit = number("limit", 20)?; |
| 33 | if !(1..=20).contains(&limit) { |
| 34 | return Err(ToolError::invalid_input("limit must be 1..20")); |
| 35 | } |
| 36 | Ok((offset, limit)) |
| 37 | } |
| 38 | |
| 39 | // Keep the same typed route receipt on every surface. Exceptionally long |
| 40 | // profile/model labels are previews; detail inspection retains their source. |
| 41 | pub(super) fn compact_child_route(mut route: Value) -> Value { |
| 42 | if serde_json::to_vec(&route).is_ok_and(|bytes| bytes.len() <= 1024) { |
| 43 | return route; |
| 44 | } |
| 45 | let Some(object) = route.as_object_mut() else { |
| 46 | return route; |
| 47 | }; |
| 48 | for cap in [128, 64, 32, 16, 8] { |
| 49 | let mut truncated = false; |
| 50 | for value in object.values_mut() { |
| 51 | if let Value::String(text) = value |
| 52 | && text.len() > cap |
| 53 | { |
| 54 | *text = text_preview(text, cap); |
| 55 | truncated = true; |
| 56 | } |
| 57 | } |
| 58 | if truncated { |
| 59 | object.insert("truncated".into(), json!(true)); |
| 60 | } |
| 61 | if serde_json::to_vec(&object).is_ok_and(|bytes| bytes.len() <= 1024) { |
| 62 | break; |
| 63 | } |
| 64 | } |
| 65 | route |
| 66 | } |
| 67 | |
| 68 | pub(super) fn status_result(mut payload: Value, peek: bool) -> Result<ToolResult, ToolError> { |
| 69 | let action = if peek { "peek" } else { "status" }; |
| 70 | payload["action"] = json!(action); |
| 71 | let mut metadata = if payload.get("agent_id").is_some() { |
| 72 | json!({ |
| 73 | "action": action, "agent_id": payload["agent_id"], |
| 74 | "status": payload["status"], "terminal": payload["terminal"], |
| 75 | "child_route": payload["child_route"], |
| 76 | }) |
| 77 | } else { |
| 78 | json!({"action": action, "count": payload["count"]}) |
| 79 | }; |
| 80 | if let Some(unchanged) = payload.get("unchanged") { |
| 81 | metadata["unchanged"] = unchanged.clone(); |
| 82 | } |
| 83 | let mut result = ToolResult::json(&payload) |
| 84 | .map_err(|error| ToolError::execution_failed(error.to_string()))?; |
| 85 | result.metadata = Some(metadata); |
| 86 | Ok(result) |
| 87 | } |
| 88 | |
| 89 | pub(super) fn compact_row(manager: &SubAgentManager, agent: &SubAgent) -> Value { |
| 90 | let record = manager.worker_records.get(&agent.id); |
| 91 | let current = manager.continuation_target(&agent.id); |
| 92 | let continuable = matches!(agent.status, SubAgentStatus::Interrupted(_)) |
| 93 | && agent |
| 94 | .checkpoint |
| 95 | .as_ref() |
| 96 | .is_some_and(|cp| cp.continuable && !cp.messages.is_empty()); |
| 97 | let status = record |
| 98 | .map(|record| agent_worker_status_name(record.status)) |
| 99 | .unwrap_or_else(|| { |
| 100 | if continuable { |
| 101 | "waiting_for_user" |
| 102 | } else { |
| 103 | subagent_status_name(&agent.status) |
| 104 | } |
| 105 | }); |
| 106 | let mut row = json!({ |
| 107 | "agent_id": agent.id, "name": text_preview(&agent.session_name, 64), |
| 108 | "status": status, "terminal": agent.status != SubAgentStatus::Running, |
| 109 | "compact": true, "steps_taken": agent.steps_taken, |
| 110 | "duration_ms": u64::try_from(agent.started_at.elapsed().as_millis()).unwrap_or(u64::MAX), |
| 111 | "needs_continuation": continuable, "usage": {}, |
| 112 | }); |
| 113 | let object = row.as_object_mut().expect("row is an object"); |
| 114 | if let Some(record) = record { |
| 115 | if let Some(parent) = record |
| 116 | .parent_run_id |
| 117 | .as_ref() |
| 118 | .or(record.spec.parent_run_id.as_ref()) |
| 119 | { |
| 120 | object.insert("parent_agent_id".into(), json!(parent)); |
| 121 | } |
| 122 | object.insert("spawn_depth".into(), json!(record.spec.spawn_depth)); |
| 123 | object.insert("max_spawn_depth".into(), json!(record.spec.max_spawn_depth)); |
| 124 | if let Ok(profile) = serde_json::to_value(&record.spec.runtime_profile) { |
| 125 | let mut limits = serde_json::Map::new(); |
| 126 | for key in ["max_steps", "wall_time_secs", "wall_deadline_ms"] { |
| 127 | if let Some(value) = profile.get(key) { |
| 128 | limits.insert(key.to_string(), value.clone()); |
| 129 | } |
| 130 | } |
| 131 | object.insert("effective_limits".into(), Value::Object(limits)); |
| 132 | } |
| 133 | let usage = &record.usage; |
| 134 | // These are this worker's receipts. Shared scope expenditure must not |
| 135 | // be added once per descendant when presenting a subtree total. |
| 136 | object.insert( |
| 137 | "usage".into(), |
| 138 | json!({ |
| 139 | "input_tokens": usage.input_tokens, "output_tokens": usage.output_tokens, |
| 140 | "total_tokens": usage.total_tokens, |
| 141 | }), |
| 142 | ); |
| 143 | object.insert("last_activity_ms".into(), json!(record.updated_at_ms)); |
| 144 | if let Some(message) = &record.latest_message { |
| 145 | object.insert("activity".into(), json!(text_preview(message, 96))); |
| 146 | } |
| 147 | let mut verification = serde_json::to_value(&record.verification).unwrap_or(Value::Null); |
| 148 | if let Some(verdicts) = verification |
| 149 | .get_mut("deliverables") |
| 150 | .and_then(Value::as_array_mut) |
| 151 | { |
| 152 | let mut counts = std::collections::BTreeMap::<String, usize>::new(); |
| 153 | for verdict in verdicts.iter() { |
| 154 | *counts |
| 155 | .entry( |
| 156 | verdict |
| 157 | .get("status") |
| 158 | .and_then(Value::as_str) |
| 159 | .unwrap_or("unknown") |
| 160 | .to_string(), |
| 161 | ) |
| 162 | .or_default() += 1; |
| 163 | } |
| 164 | verdicts.sort_by_key(|verdict| { |
| 165 | matches!( |
| 166 | verdict.get("status").and_then(Value::as_str), |
| 167 | Some("present" | "pending") |
| 168 | ) |
| 169 | }); |
| 170 | let total = verdicts.len(); |
| 171 | verification["deliverables_total"] = json!(total); |
| 172 | verification["deliverables_omitted"] = json!(total.saturating_sub(4)); |
| 173 | verification["deliverable_counts"] = json!(counts); |
| 174 | } |
| 175 | bound_detail_value(&mut verification, 0, 4, &mut 1200); |
| 176 | object.insert("verification".into(), verification); |
| 177 | if let Some(route) = &record.spec.child_route { |
| 178 | object.insert( |
| 179 | "child_route".into(), |
| 180 | compact_child_route(serde_json::to_value(route).unwrap_or(Value::Null)), |
| 181 | ); |
| 182 | } |
| 183 | // #6194 item 5: live declared-vs-observed write surfacing. The child |
| 184 | // declares deliverables at spawn and the registry records every |
| 185 | // successful scoped write; the parent sees the diff while the child |
| 186 | // is still alive instead of only in the post-mortem receipt. |
| 187 | if record.spec.runtime_profile.permissions.write { |
| 188 | const MAX_LISTED_WRITES: usize = 4; |
| 189 | let declared: Vec<String> = record |
| 190 | .spec |
| 191 | .launch_manifest |
| 192 | .as_ref() |
| 193 | .map(|manifest| { |
| 194 | manifest |
| 195 | .deliverables |
| 196 | .iter() |
| 197 | .take(MAX_LISTED_WRITES) |
| 198 | .cloned() |
| 199 | .collect() |
| 200 | }) |
| 201 | .unwrap_or_default(); |
| 202 | let declared_total = record |
| 203 | .spec |
| 204 | .launch_manifest |
| 205 | .as_ref() |
| 206 | .map(|manifest| manifest.deliverables.len()) |
| 207 | .unwrap_or(0); |
| 208 | let observed: Vec<String> = record |
| 209 | .delivery_evidence |
| 210 | .observed_writes |
| 211 | .iter() |
| 212 | .take(MAX_LISTED_WRITES) |
| 213 | .cloned() |
| 214 | .collect(); |
| 215 | let observed_total = record.delivery_evidence.observed_writes.len(); |
| 216 | if declared_total > 0 || observed_total > 0 { |
| 217 | object.insert( |
| 218 | "write_progress".into(), |
| 219 | json!({ |
| 220 | "declared": declared, |
| 221 | "declared_total": declared_total, |
| 222 | "observed": observed, |
| 223 | "observed_total": observed_total, |
| 224 | }), |
| 225 | ); |
| 226 | } |
| 227 | } |
| 228 | } |
| 229 | if let Some(source) = manager.continuation_source(&agent.id) { |
| 230 | object.insert("resumed_from".into(), json!(source)); |
| 231 | } |
| 232 | match current { |
| 233 | Ok(target) if target != agent.id => { |
| 234 | object.insert("resumed_as".into(), json!(target)); |
| 235 | } |
| 236 | Err(error) => { |
| 237 | object.insert( |
| 238 | "lineage_error".into(), |
| 239 | json!(text_preview(&error.to_string(), 120)), |
| 240 | ); |
| 241 | } |
| 242 | _ => {} |
| 243 | } |
| 244 | if let Some(input) = &agent.needs_input { |
| 245 | object.insert( |
| 246 | "needs_input".into(), |
| 247 | json!(text_preview(&input.question, 160)), |
| 248 | ); |
| 249 | } |
| 250 | let stop_reason = match &agent.status { |
| 251 | SubAgentStatus::Failed(reason) | SubAgentStatus::Interrupted(reason) => Some(reason), |
| 252 | _ => agent |
| 253 | .result |
| 254 | .as_ref() |
| 255 | .filter(|_| agent.status != SubAgentStatus::Running), |
| 256 | }; |
| 257 | if let Some(reason) = stop_reason { |
| 258 | object.insert("summary".into(), json!(text_preview(reason, 160))); |
| 259 | } |
| 260 | if serde_json::to_vec(&row).is_ok_and(|bytes| bytes.len() > 3072) { |
| 261 | // A verdict containing highly escaped prose can exceed its raw-byte |
| 262 | // allowance. Keep the authoritative verdict status and name the |
| 263 | // omitted detail, rather than returning a page that cannot advance. |
| 264 | let status = row |
| 265 | .pointer("/verification/status") |
| 266 | .cloned() |
| 267 | .unwrap_or(Value::Null); |
| 268 | let counts = row.pointer("/verification/deliverable_counts").cloned(); |
| 269 | let total = row.pointer("/verification/deliverables_total").cloned(); |
| 270 | row["verification"] = json!({"status": status, "deliverable_counts": counts, "deliverables_total": total, "detail_required": true}); |
| 271 | row.as_object_mut().expect("row object").remove("activity"); |
| 272 | } |
| 273 | row |
| 274 | } |
| 275 | |
| 276 | pub(super) fn compact_roster( |
| 277 | manager: &SubAgentManager, |
| 278 | input: &Value, |
| 279 | session: &str, |
| 280 | archived: bool, |
| 281 | peek: bool, |
| 282 | ) -> Result<Value, ToolError> { |
| 283 | let (offset, limit) = page(input)?; |
| 284 | let mut agents = manager |
| 285 | .agents |
| 286 | .values() |
| 287 | .filter(|agent| manager.agent_is_owned_by_session(agent, session)) |
| 288 | .filter(|agent| archived || !manager.is_from_prior_session(agent)) |
| 289 | .collect::<Vec<_>>(); |
| 290 | // Live and waiting work comes first; tie-break with immutable ids, so a |
| 291 | // page is reproducible for unchanged state even across process restarts. |
| 292 | agents.sort_by(|a, b| { |
| 293 | (a.status != SubAgentStatus::Running, &a.id) |
| 294 | .cmp(&(b.status != SubAgentStatus::Running, &b.id)) |
| 295 | }); |
| 296 | let mut counts = std::collections::BTreeMap::<&str, usize>::new(); |
| 297 | let mut total_tokens = 0_u64; |
| 298 | let mut reported_workers = 0_usize; |
| 299 | for agent in &agents { |
| 300 | *counts |
| 301 | .entry(subagent_status_name(&agent.status)) |
| 302 | .or_default() += 1; |
| 303 | if let Some(tokens) = manager |
| 304 | .worker_records |
| 305 | .get(&agent.id) |
| 306 | .and_then(|record| record.usage.total_tokens) |
| 307 | { |
| 308 | total_tokens = total_tokens.saturating_add(tokens); |
| 309 | reported_workers += 1; |
| 310 | } |
| 311 | } |
| 312 | let total = agents.len(); |
| 313 | // A stable header carries field names once. Null cells mean absent or |
| 314 | // unreported values; a reported zero remains a numeric zero. |
| 315 | let columns = [ |
| 316 | "agent_id", |
| 317 | "parent_agent_id", |
| 318 | "resumed_from", |
| 319 | "resumed_as", |
| 320 | "spawn_depth", |
| 321 | "status", |
| 322 | "duration_ms", |
| 323 | "total_tokens", |
| 324 | "last_activity_ms", |
| 325 | "activity", |
| 326 | "needs_input", |
| 327 | "needs_continuation", |
| 328 | "verification", |
| 329 | "summary", |
| 330 | "lineage_error", |
| 331 | ]; |
| 332 | let mut rows = agents |
| 333 | .into_iter() |
| 334 | .skip(offset) |
| 335 | .take(limit) |
| 336 | .map(|agent| { |
| 337 | let mut row = compact_row(manager, agent); |
| 338 | let total_tokens = row |
| 339 | .pointer("/usage/total_tokens") |
| 340 | .cloned() |
| 341 | .unwrap_or(Value::Null); |
| 342 | row["total_tokens"] = total_tokens; |
| 343 | if let Some(record) = manager.worker_records.get(&agent.id) { |
| 344 | let mut verification = |
| 345 | json!({"status": text_preview(&record.verification.status, 64)}); |
| 346 | if !matches!( |
| 347 | record.verification.status.as_str(), |
| 348 | "self_report_only" | "deliverables_present" |
| 349 | ) && !record.verification.summary.is_empty() |
| 350 | { |
| 351 | verification["summary"] = |
| 352 | json!(text_preview(&record.verification.summary, 160)); |
| 353 | } |
| 354 | if let Some(counts) = row |
| 355 | .pointer("/verification/deliverable_counts") |
| 356 | .filter(|value| value.as_object().is_some_and(|counts| !counts.is_empty())) |
| 357 | { |
| 358 | verification["deliverable_counts"] = counts.clone(); |
| 359 | } |
| 360 | row["verification"] = verification; |
| 361 | } |
| 362 | Value::Array( |
| 363 | columns |
| 364 | .iter() |
| 365 | .map(|column| row.get(*column).cloned().unwrap_or(Value::Null)) |
| 366 | .collect(), |
| 367 | ) |
| 368 | }) |
| 369 | .collect::<Vec<_>>(); |
| 370 | loop { |
| 371 | let shown = rows.len(); |
| 372 | let next = offset.saturating_add(shown); |
| 373 | let payload = json!({ |
| 374 | "action": if peek { "peek" } else { "status" }, "compact": true, |
| 375 | "count": shown, "total_count": total, "status_counts": counts, |
| 376 | "usage": {"total_tokens": total_tokens, "reported_workers": reported_workers, "workers": total}, |
| 377 | "columns": columns, "agents": rows, "offset": offset, |
| 378 | "next_offset": (next < total).then_some(next), |
| 379 | "omitted": total.saturating_sub(shown), |
| 380 | "detail_hint": "Rows follow columns. Inspect with agent_id and detail=true.", |
| 381 | }); |
| 382 | if serde_json::to_vec(&payload) |
| 383 | .map_err(|error| ToolError::execution_failed(error.to_string()))? |
| 384 | .len() |
| 385 | <= COMPACT_STATUS_BYTES |
| 386 | { |
| 387 | return Ok(payload); |
| 388 | } |
| 389 | if rows.pop().is_none() { |
| 390 | return Err(ToolError::execution_failed( |
| 391 | "Status envelope exceeds its byte limit", |
| 392 | )); |
| 393 | } |
| 394 | } |
| 395 | } |
| 396 | |
| 397 | // Bound arbitrary message/tool-input JSON as it is copied into a diagnostic |
| 398 | // page. The transcript handle remains the authoritative unabridged source. |
| 399 | fn bound_detail_value(value: &mut Value, offset: usize, limit: usize, budget: &mut usize) { |
| 400 | if *budget == 0 { |
| 401 | *value = json!("[omitted]"); |
| 402 | return; |
| 403 | } |
| 404 | match value { |
| 405 | Value::String(text) => { |
| 406 | *text = text_preview(text, (*budget).min(1024)); |
| 407 | *budget = budget.saturating_sub(text.len()); |
| 408 | } |
| 409 | Value::Array(items) => { |
| 410 | *items = std::mem::take(items) |
| 411 | .into_iter() |
| 412 | .skip(offset) |
| 413 | .take(limit) |
| 414 | .collect(); |
| 415 | for item in items { |
| 416 | bound_detail_value(item, 0, limit, budget); |
| 417 | } |
| 418 | } |
| 419 | Value::Object(object) => { |
| 420 | object.retain(|key, _| key.len() <= 128); |
| 421 | for child in object.values_mut() { |
| 422 | bound_detail_value(child, 0, limit, budget); |
| 423 | } |
| 424 | } |
| 425 | _ => { |
| 426 | *budget = budget.saturating_sub(16); |
| 427 | } |
| 428 | } |
| 429 | } |
| 430 | |
| 431 | pub(super) fn bounded_detail( |
| 432 | mut value: Value, |
| 433 | compact: Value, |
| 434 | offset: usize, |
| 435 | limit: usize, |
| 436 | ) -> Value { |
| 437 | // The typed handle is the retrieval authority, not diagnostic prose. |
| 438 | // Large earlier fields must never consume its lookup coordinates. |
| 439 | let transcript_handle = value.get("transcript_handle").cloned(); |
| 440 | let mut verification = value.get("verification").cloned().unwrap_or(Value::Null); |
| 441 | if let Some(verdicts) = verification |
| 442 | .get_mut("deliverables") |
| 443 | .and_then(Value::as_array_mut) |
| 444 | { |
| 445 | let total = verdicts.len(); |
| 446 | *verdicts = std::mem::take(verdicts) |
| 447 | .into_iter() |
| 448 | .skip(offset) |
| 449 | .take(limit) |
| 450 | .collect(); |
| 451 | verification["deliverables_total"] = json!(total); |
| 452 | verification["deliverables_next_offset"] = (offset.saturating_add(limit) < total) |
| 453 | .then_some(offset.saturating_add(limit)) |
| 454 | .map_or(Value::Null, |next| json!(next)); |
| 455 | } |
| 456 | bound_detail_value(&mut verification, 0, limit, &mut 4000); |
| 457 | // Page the two archives at their actual boundaries, not every content |
| 458 | // array inside a message, before enforcing the byte budget. |
| 459 | for pointer in [ |
| 460 | "/checkpoint/messages", |
| 461 | "/snapshot/checkpoint/messages", |
| 462 | "/worker_record/events", |
| 463 | ] { |
| 464 | if let Some(items) = value.pointer_mut(pointer).and_then(Value::as_array_mut) { |
| 465 | let count = items.len(); |
| 466 | *items = std::mem::take(items) |
| 467 | .into_iter() |
| 468 | .skip(offset) |
| 469 | .take(limit) |
| 470 | .collect(); |
| 471 | value[pointer.replace('/', "_") + "_total"] = json!(count); |
| 472 | } |
| 473 | } |
| 474 | bound_detail_value(&mut value, 0, limit, &mut (12 * 1024)); |
| 475 | let object = value.as_object_mut().expect("projection is an object"); |
| 476 | // Keep the canonical compact facts (especially causes and delivery |
| 477 | // verdicts) visible if a very large diagnostic archive must be omitted. |
| 478 | if let Some(compact) = compact.as_object() { |
| 479 | object.extend( |
| 480 | compact |
| 481 | .iter() |
| 482 | .map(|(key, value)| (key.clone(), value.clone())), |
| 483 | ); |
| 484 | } |
| 485 | object.insert("verification".into(), verification); |
| 486 | if let Some(mut handle) = transcript_handle { |
| 487 | if let Some(preview) = handle.get("repr_preview").and_then(Value::as_str) { |
| 488 | handle["repr_preview"] = json!(text_preview(preview, 160)); |
| 489 | } |
| 490 | object.insert("transcript_handle".into(), handle); |
| 491 | } |
| 492 | object.insert("compact".into(), json!(false)); |
| 493 | object.insert("detail_bounded".into(), json!(true)); |
| 494 | object.insert("detail_offset".into(), json!(offset)); |
| 495 | object.insert("detail_limit".into(), json!(limit)); |
| 496 | object.insert("detail_hint".into(), json!("Fields are bounded; use offset/limit for messages and events, or transcript_handle for the complete retained transcript.")); |
| 497 | if serde_json::to_vec(&value).map_or(true, |bytes| bytes.len() > DETAIL_STATUS_BYTES) { |
| 498 | let object = value.as_object_mut().expect("projection is an object"); |
| 499 | for key in ["snapshot", "worker_record", "checkpoint"] { |
| 500 | object.remove(key); |
| 501 | } |
| 502 | object.insert( |
| 503 | "omitted_detail".into(), |
| 504 | json!(["snapshot", "worker_record", "checkpoint"]), |
| 505 | ); |
| 506 | } |
| 507 | if serde_json::to_vec(&value).map_or(true, |bytes| bytes.len() > DETAIL_STATUS_BYTES) { |
| 508 | let handle = value.get("transcript_handle").cloned(); |
| 509 | value = json!({ |
| 510 | "agent_id": value["agent_id"], "status": value["status"], |
| 511 | "needs_input": value["needs_input"], "summary": value["summary"], |
| 512 | "usage": value["usage"], "verification": value["verification"], |
| 513 | "transcript_handle": handle, "detail_bounded": true, |
| 514 | "omitted_detail": "Diagnostic page exceeded 32 KiB; use transcript_handle for retained detail.", |
| 515 | }); |
| 516 | } |
| 517 | value |
| 518 | } |
| 519 |