返回 CodeWhale
task_projection.rs
根目录 / crates / tui / src / tui / ui / task_projection.rs
1 //! Task-panel and shell projection: task-panel refresh, shell live-output
2 //! reconciliation, detached-job projection, and RLM task entries
3 //! (TUI_MODULARIZATION.md slice 5). Pure projection — no dispatch here.
4
5 use super::*;
6 use crate::tui::automation_panel::AutomationScan;
7
8 pub(super) async fn refresh_active_task_panel(
9 app: &mut App,
10 task_manager: &SharedTaskManager,
11 ) -> bool {
12 let namespace_changed = app.task_panel_session_id != app.current_session_id;
13 if namespace_changed {
14 app.task_panel.clear();
15 app.task_panel_session_id = app.current_session_id.clone();
16 app.task_panel_unavailable = false;
17 }
18 let tasks = match app.current_session_id.as_deref() {
19 Some(session_id) => match task_manager
20 .list_tasks_for_owner(None, None, session_id)
21 .await
22 {
23 Ok(tasks) => tasks,
24 Err(error) => {
25 let changed = namespace_changed || !app.task_panel_unavailable;
26 if !app.task_panel_unavailable {
27 app.push_status_toast(
28 codewhale_localization::tr(
29 app.ui_locale,
30 codewhale_localization::MessageId::TaskInventoryUnavailable,
31 )
32 .to_string(),
33 crate::tui::app::StatusToastLevel::Warning,
34 Some(8_000),
35 );
36 tracing::warn!(%error, "Task inventory unavailable; preserving scoped snapshot");
37 }
38 app.task_panel_unavailable = true;
39 return changed;
40 }
41 },
42 None => Vec::new(),
43 };
44 let was_unavailable = std::mem::replace(&mut app.task_panel_unavailable, false);
45 let previously_active_durable_ids = app
46 .task_panel
47 .iter()
48 .filter(|entry| matches!(entry.status.as_str(), "queued" | "running"))
49 .map(|entry| entry.id.as_str())
50 .collect::<HashSet<_>>();
51 let durable_background_completed = newly_completed_id(
52 previously_active_durable_ids,
53 tasks
54 .iter()
55 .filter(|task| task.status == TaskStatus::Completed)
56 .map(|task| task.id.as_str()),
57 );
58 let mut lifecycle_changed = false;
59 if let (Some(work), Some(session_id)) = (
60 app.runtime_services.work.as_ref(),
61 app.current_session_id.as_deref(),
62 ) {
63 for task in &tasks {
64 if !task.execution_binding_known {
65 continue;
66 }
67 let external = format!("task:{}", task.id);
68 if !work.has_operation_binding(Some(session_id), &external) {
69 continue;
70 }
71 match work.reconcile_operation(
72 session_id,
73 task_owner_snapshot(
74 &task.id,
75 task.status,
76 task.lifecycle_seq,
77 task.created_at,
78 task.started_at,
79 task.ended_at,
80 ),
81 ) {
82 Ok(changed) => lifecycle_changed |= changed,
83 Err(err) => {
84 tracing::warn!(task_id = %task.id, error = %err, "failed to reconcile durable task lifecycle");
85 }
86 }
87 }
88 }
89 if lifecycle_changed && let Err(err) = persist_pending_work_checkpoint(app).await {
90 tracing::warn!(error = %err, "durable task lifecycle checkpoint remains pending");
91 }
92 let session_started_at = app.session_started_at;
93 let mut entries: Vec<TaskPanelEntry> =
94 select_work_sidebar_tasks(tasks, session_started_at, app.current_session_id.as_deref())
95 .into_iter()
96 .map(|summary| {
97 let unverified = !summary.execution_binding_known
98 && matches!(summary.status, TaskStatus::Queued | TaskStatus::Running);
99 let mut entry = task_summary_to_panel_entry(summary);
100 if unverified {
101 entry.stale = true;
102 entry.role = Some(
103 codewhale_localization::tr(
104 app.ui_locale,
105 codewhale_localization::MessageId::TaskOwnershipUnverified,
106 )
107 .to_string(),
108 );
109 }
110 entry
111 })
112 .collect();
113
114 entries.extend(active_rlm_task_entries(app));
115
116 // #3804: this is a render-only read of shell jobs and must not block the
117 // async UI loop on the shell manager's std::sync Mutex. Use try_lock; on
118 // contention, retain the previous frame's background shell entries so
119 // running shells don't flicker out of the Work panel. Shell ownership,
120 // cancellation, approval state, and output capture never depend on this
121 // refresh succeeding.
122 let prev_shell_entries: Vec<TaskPanelEntry> = app
123 .task_panel
124 .iter()
125 .filter(|entry| matches!(entry.kind, TaskPanelEntryKind::Background))
126 .cloned()
127 .collect();
128 let prev_shell_ids = prev_shell_entries
129 .iter()
130 .map(|entry| entry.id.clone())
131 .collect::<HashSet<_>>();
132 let (shell_entries, shell_background_completed): (Vec<TaskPanelEntry>, bool) = match app
133 .runtime_services
134 .shell_manager
135 .as_ref()
136 {
137 Some(shell_mgr) => match shell_mgr.try_lock() {
138 Ok(mut mgr) => {
139 let jobs = mgr
140 .list_jobs_for_session(app.current_session_id.as_deref().unwrap_or_default());
141 let completed = newly_completed_id(
142 prev_shell_ids.iter().map(String::as_str).collect(),
143 jobs.iter()
144 .filter(|job| {
145 matches!(job.status, crate::tools::shell::ShellStatus::Completed)
146 })
147 .map(|job| job.id.as_str()),
148 );
149 let entries = jobs
150 .into_iter()
151 .filter(|job| matches!(job.status, crate::tools::shell::ShellStatus::Running))
152 .map(|job| TaskPanelEntry {
153 id: job.id,
154 status: "running".to_string(),
155 prompt_summary: format!("shell: {}", job.command),
156 duration_ms: Some(job.elapsed_ms),
157 kind: TaskPanelEntryKind::Background,
158 stale: job.stale,
159 elapsed_since_output_ms: job.elapsed_since_output_ms,
160 owner_agent_id: job.owner_agent_id,
161 owner_agent_name: job.owner_agent_name,
162 current_tool: None,
163 role: None,
164 files_touched: 0,
165 })
166 .collect();
167 (entries, completed)
168 }
169 // Contended: keep the last known snapshot rather than blocking.
170 // A retained frame could belong to the session that was just
171 // replaced. Fail closed on contention instead of showing it
172 // in the new conversation.
173 Err(_) => (Vec::new(), false),
174 },
175 None => (Vec::new(), false),
176 };
177 entries.extend(shell_entries);
178
179 // Report whether anything visible changed so the idle tick can skip the
180 // redraw: an unconditional 2.5 s repaint kept the app from ever going
181 // quiescent (#3757).
182 let changed =
183 namespace_changed || was_unavailable || lifecycle_changed || app.task_panel != entries;
184 app.task_panel = entries;
185 let tip_shown = (durable_background_completed || shell_background_completed)
186 && app.maybe_show_behavioral_tip(
187 crate::tui::behavioral_tips::BehavioralTip::BackgroundJobReceipt,
188 );
189 changed || tip_shown
190 }
191
192 pub(super) fn newly_completed_id<'a>(
193 previously_active_ids: HashSet<&'a str>,
194 completed_ids: impl IntoIterator<Item = &'a str>,
195 ) -> bool {
196 completed_ids
197 .into_iter()
198 .any(|id| previously_active_ids.contains(id))
199 }
200
201 /// Newest runs scanned per automation when refreshing the automation
202 /// projection. Live runs sit at the head of the newest-first listing, and a
203 /// failure once seen is held unacknowledged by the projection until the
204 /// operator engages the automation surface, so the band never needs a full
205 /// run-history scan on the render cadence.
206 const AUTOMATION_PANEL_RUN_SCAN: usize = 25;
207
208 /// Refresh the activity band's scheduled-work projection
209 /// (AUTOMATION-VISIBILITY-SPEC §2.1) from the durable automation store.
210 ///
211 /// The store is files on disk — every definition plus up to
212 /// `AUTOMATION_PANEL_RUN_SCAN` run files per definition — so the scan never
213 /// runs on the async UI loop: it is taken on a blocking thread, and this
214 /// tick folds whatever scan has finished, then starts the next one. At most
215 /// one scan is in flight; a slow disk costs the band latency, never the
216 /// frame. Returns whether the visible state changed, so the idle tick can
217 /// skip the redraw (#3757).
218 pub(super) async fn refresh_automation_panel(app: &mut App) -> bool {
219 let mut changed = false;
220 if let Some(scan) = app.automation_scan.take() {
221 if scan.is_finished() {
222 match scan.await {
223 Ok(scan) => changed = fold_automation_scan(app, &scan),
224 Err(err) => {
225 tracing::warn!(error = %err, "automation panel scan task failed");
226 }
227 }
228 } else {
229 app.automation_scan = Some(scan);
230 return false;
231 }
232 }
233 app.automation_scan = start_automation_scan(app, false);
234 changed
235 }
236
237 /// Startup variant: take one scan and wait for it, so the first frame
238 /// already carries the band count (the task panel gets the same courtesy).
239 /// The startup pass is FULL — every run file — so a long-running task that
240 /// already sits behind more than a window of newer runs is visible from the
241 /// first frame. The manager lock is retried briefly: the scheduler tick
242 /// holds it only while persisting, and giving up here would blank the band
243 /// on a contended startup.
244 pub(super) async fn refresh_automation_panel_blocking(app: &mut App) -> bool {
245 let scan = 'retry: {
246 for _ in 0..STARTUP_SCAN_LOCK_RETRIES {
247 if let Some(scan) = start_automation_scan(app, true) {
248 break 'retry Some(scan);
249 }
250 tokio::time::sleep(STARTUP_SCAN_LOCK_RETRY_DELAY).await;
251 }
252 None
253 };
254 let Some(scan) = scan else {
255 return false;
256 };
257 match scan.await {
258 Ok(scan) => fold_automation_scan(app, &scan),
259 Err(err) => {
260 tracing::warn!(error = %err, "automation panel scan task failed");
261 false
262 }
263 }
264 }
265
266 /// Startup lock-retry budget: the scheduler's persist phase is short; ten
267 /// 50 ms attempts covers it without parking startup on a stuck lock.
268 const STARTUP_SCAN_LOCK_RETRIES: usize = 10;
269 const STARTUP_SCAN_LOCK_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(50);
270
271 /// Start one store scan on a blocking thread. The manager is cloned out
272 /// from under its tokio Mutex (`try_lock`, same rule as the shell snapshot:
273 /// the scheduler tick holds that lock while persisting, and the UI loop
274 /// must not park behind it; on contention the next tick retries) so the
275 /// scan holds no lock while it reads — the store's writes are atomic
276 /// renames, so a concurrent read sees a whole file either way.
277 ///
278 /// `full` reads every run file (startup only). The cadence scan reads the
279 /// newest `AUTOMATION_PANEL_RUN_SCAN` runs per automation PLUS the runs
280 /// this session already watched go live, wherever they sit in history — a
281 /// frequent automation can stack newer runs behind a long-running task,
282 /// and neither the live count nor the settle receipt may depend on the
283 /// task staying inside the newest window.
284 fn start_automation_scan(app: &App, full: bool) -> Option<tokio::task::JoinHandle<AutomationScan>> {
285 let automations = app.runtime_services.automations.as_ref()?;
286 let manager = automations.try_lock().ok()?.clone();
287 let live_owners = app.automation_panel.live_run_owners();
288 Some(tokio::task::spawn_blocking(move || {
289 let records = match manager.list_automations() {
290 Ok(records) => records,
291 Err(err) => {
292 tracing::warn!(error = %err, "automation panel refresh could not list automations");
293 return AutomationScan::default();
294 }
295 };
296 let mut runs = Vec::new();
297 for record in &records {
298 let limit = if full {
299 None
300 } else {
301 Some(AUTOMATION_PANEL_RUN_SCAN)
302 };
303 match manager.list_runs(&record.id, limit) {
304 Ok(recent) => runs.extend(recent),
305 Err(err) => {
306 tracing::warn!(automation_id = %record.id, error = %err, "automation panel refresh could not list runs");
307 }
308 }
309 if !full {
310 let wanted: std::collections::BTreeSet<String> = live_owners
311 .iter()
312 .filter(|(_, owner)| owner.as_str() == record.id.as_str())
313 .map(|(run_id, _)| run_id.clone())
314 .collect();
315 if !wanted.is_empty() {
316 match manager.get_runs_by_ids(&record.id, &wanted) {
317 Ok(found) => runs.extend(found),
318 Err(err) => {
319 tracing::warn!(automation_id = %record.id, error = %err, "automation panel refresh could not re-read live runs");
320 }
321 }
322 }
323 }
324 }
325 // A re-read live run may also sit inside the newest window; the
326 // fold counts by id, so dedupe before handing the scan over.
327 let mut seen = std::collections::BTreeSet::new();
328 runs.retain(|run| seen.insert(run.id.clone()));
329 AutomationScan { records, runs }
330 }))
331 }
332
333 /// Fold a finished scan into the projection and post the typed receipt
334 /// (spec §2.2) for every run this session watched go live and settle — the
335 /// transcript learns about background work from the same scan that
336 /// repaints the band.
337 fn fold_automation_scan(app: &mut App, scan: &AutomationScan) -> bool {
338 let session_started_at = app.session_started_at;
339 let delta = app
340 .automation_panel
341 .fold_scan(&scan.records, &scan.runs, session_started_at);
342 let locale = app.ui_locale;
343 for run in &delta.settled {
344 app.add_message(crate::tui::automation_routing::settled_run_receipt(
345 locale, run,
346 ));
347 }
348 delta.changed || !delta.settled.is_empty()
349 }
350
351 pub(super) fn refresh_shell_exec_live_output(app: &mut App) -> bool {
352 let Some(shell_mgr) = app.runtime_services.shell_manager.as_ref().cloned() else {
353 return false;
354 };
355 // #3804: render-only read — try_lock so a contended shell Mutex can never
356 // block the async UI loop; skip this frame's live-output update on
357 // contention (the next refresh picks it up).
358 let jobs = {
359 let Ok(mut mgr) = shell_mgr.try_lock() else {
360 return false;
361 };
362 mgr.list_jobs_for_session(app.current_session_id.as_deref().unwrap_or_default())
363 .into_iter()
364 .map(|job| (job.id.clone(), job))
365 .collect::<std::collections::HashMap<_, _>>()
366 };
367 let mut changed = false;
368 for index in 0..app.virtual_cell_count() {
369 let Some(ShellExecLiveUpdate {
370 task_id,
371 status: next_status,
372 output: next_live,
373 duration_ms: next_duration,
374 finalized,
375 stale_elapsed_since_output_ms,
376 }) = shell_exec_live_update(app, index, &jobs)
377 else {
378 continue;
379 };
380 let Some(HistoryCell::Tool(ToolCell::Exec(exec))) = app.cell_at_virtual_index_mut(index)
381 else {
382 continue;
383 };
384 if exec.output.is_some() || exec.shell_task_id.as_deref() != Some(task_id.as_str()) {
385 continue;
386 }
387 exec.status = next_status;
388 exec.duration_ms = Some(next_duration);
389 exec.stale_elapsed_since_output_ms = stale_elapsed_since_output_ms;
390 if finalized {
391 exec.output = next_live;
392 exec.output_summary = exec
393 .output
394 .as_deref()
395 .map(crate::tui::history::summarize_tool_output);
396 exec.live_output = None;
397 exec.stale_elapsed_since_output_ms = None;
398 } else {
399 exec.live_output = next_live;
400 }
401 changed = true;
402 }
403 changed
404 }
405
406 pub(super) struct ShellExecLiveUpdate {
407 pub(super) task_id: String,
408 pub(super) status: ToolStatus,
409 pub(super) output: Option<String>,
410 pub(super) duration_ms: u64,
411 pub(super) finalized: bool,
412 pub(super) stale_elapsed_since_output_ms: Option<u64>,
413 }
414
415 pub(super) fn shell_exec_live_update(
416 app: &App,
417 index: usize,
418 jobs: &std::collections::HashMap<String, ShellJobSnapshot>,
419 ) -> Option<ShellExecLiveUpdate> {
420 let HistoryCell::Tool(ToolCell::Exec(exec)) = app.cell_at_virtual_index(index)? else {
421 return None;
422 };
423 if exec.output.is_some() {
424 return None;
425 }
426 let task_id = exec.shell_task_id.as_deref()?;
427 let Some(job) = jobs.get(task_id) else {
428 return Some(ShellExecLiveUpdate {
429 task_id: task_id.to_string(),
430 status: ToolStatus::Failed,
431 output: detached_shell_job_output(task_id, exec),
432 duration_ms: exec.duration_ms.unwrap_or_default(),
433 finalized: true,
434 stale_elapsed_since_output_ms: None,
435 });
436 };
437 let next_status = shell_job_tool_status(&job.status);
438 let next_live = shell_job_live_output(job).or_else(|| exec.live_output.clone());
439 let finalized = !matches!(job.status, ShellStatus::Running);
440 let stale_elapsed_since_output_ms = if matches!(job.status, ShellStatus::Running) && job.stale {
441 Some(job.elapsed_since_output_ms.unwrap_or(0))
442 } else {
443 None
444 };
445 if exec.status == next_status
446 && exec.live_output == next_live
447 && exec.duration_ms == Some(job.elapsed_ms)
448 && exec.stale_elapsed_since_output_ms == stale_elapsed_since_output_ms
449 {
450 return None;
451 }
452 Some(ShellExecLiveUpdate {
453 task_id: task_id.to_string(),
454 status: next_status,
455 output: next_live,
456 duration_ms: job.elapsed_ms,
457 finalized,
458 stale_elapsed_since_output_ms,
459 })
460 }
461
462 pub(super) fn detached_shell_job_output(task_id: &str, exec: &ExecCell) -> Option<String> {
463 let mut output = exec.live_output.clone().unwrap_or_default();
464 if !output.trim().is_empty() {
465 output.push_str("\n\n");
466 }
467 output.push_str(&format!(
468 "Shell job `{task_id}` is no longer attached to this TUI session."
469 ));
470 Some(output)
471 }
472
473 pub(super) fn shell_job_tool_status(status: &ShellStatus) -> ToolStatus {
474 match status {
475 ShellStatus::Running => ToolStatus::Running,
476 ShellStatus::Completed => ToolStatus::Success,
477 ShellStatus::Failed | ShellStatus::Killed | ShellStatus::TimedOut => ToolStatus::Failed,
478 }
479 }
480
481 pub(super) fn shell_job_live_output(job: &ShellJobSnapshot) -> Option<String> {
482 match (job.stdout_tail.is_empty(), job.stderr_tail.is_empty()) {
483 (true, true) => None,
484 (false, true) => Some(job.stdout_tail.clone()),
485 (true, false) => Some(format!("STDERR:\n{}", job.stderr_tail)),
486 (false, false) => Some(format!(
487 "{}\n\nSTDERR:\n{}",
488 job.stdout_tail, job.stderr_tail
489 )),
490 }
491 }
492
493 pub(super) fn active_rlm_task_entries(app: &App) -> Vec<TaskPanelEntry> {
494 let Some(active) = app.active_cell.as_ref() else {
495 return Vec::new();
496 };
497 let duration_ms = app
498 .turn_started_at
499 .map(|started| u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX));
500 active
501 .entries()
502 .iter()
503 .enumerate()
504 .filter_map(|(idx, entry)| {
505 let HistoryCell::Tool(ToolCell::Generic(generic)) = entry else {
506 return None;
507 };
508 if !matches!(
509 generic.name.as_str(),
510 "rlm_open" | "rlm_eval" | "rlm_configure" | "rlm_close" | "rlm"
511 ) || generic.status != ToolStatus::Running
512 {
513 return None;
514 }
515 let summary = generic
516 .input_summary
517 .as_deref()
518 .filter(|summary| !summary.trim().is_empty())
519 .unwrap_or("running chunked analysis");
520 Some(TaskPanelEntry {
521 id: format!("rlm-{}", idx + 1),
522 status: "running".to_string(),
523 prompt_summary: format!("RLM: {summary}"),
524 duration_ms,
525 kind: TaskPanelEntryKind::Background,
526 stale: false,
527 elapsed_since_output_ms: None,
528 owner_agent_id: None,
529 owner_agent_name: None,
530 current_tool: None,
531 role: None,
532 files_touched: 0,
533 })
534 })
535 .collect()
536 }
537
537 lines RUST