diff --git a/apps/decodex/src/orchestrator/lane_control.rs b/apps/decodex/src/orchestrator/lane_control.rs index 562088df6..c3fd920e8 100644 --- a/apps/decodex/src/orchestrator/lane_control.rs +++ b/apps/decodex/src/orchestrator/lane_control.rs @@ -147,7 +147,7 @@ impl LaneRunInspect { event_count: run.event_count, worktree_path: run.worktree_path.clone(), soft_interrupt_available: soft_interrupt_available_for_run(run), - hard_interrupt_available: run.process_id.is_some() && run.process_alive != Some(false), + hard_interrupt_available: hard_interrupt_available_for_run(run), hard_interrupt_requires_force: true, } } @@ -397,6 +397,10 @@ fn soft_interrupt_allows_hard_fallback( soft: &LaneSoftInterruptReport, run: &OperatorRunStatus, ) -> bool { + if run.phase == "terminal_pending" { + return false; + } + match soft.status.as_str() { "pending" | "failed" | "unavailable" => soft.error_class.as_deref() != Some("lane_not_active") @@ -519,6 +523,10 @@ fn soft_interrupt_available_for_run(run: &OperatorRunStatus) -> bool { && run.control_capability.as_ref().is_some_and(|capability| capability.status == "active") } +fn hard_interrupt_available_for_run(run: &OperatorRunStatus) -> bool { + run.phase != "terminal_pending" && run.process_id.is_some() && run.process_alive != Some(false) +} + fn lane_control_operator_context(run: &OperatorRunStatus) -> Value { let control_capability = run.control_capability.as_ref().map(|capability| { serde_json::json!({ diff --git a/apps/decodex/src/orchestrator/operator_http.rs b/apps/decodex/src/orchestrator/operator_http.rs index 4ac24df49..851fb1f01 100644 --- a/apps/decodex/src/orchestrator/operator_http.rs +++ b/apps/decodex/src/orchestrator/operator_http.rs @@ -12,6 +12,7 @@ const OPERATOR_DASHBOARD_LOGO_ICO: &[u8] = const OPERATOR_DASHBOARD_LOGO_TOUCH_PNG: &[u8] = include_bytes!(concat!(env!("CARGO_MANIFEST_DIR"), "/src/orchestrator/assets/logo-touch.png")); const OPERATOR_HTTP_READ_TIMEOUT: Duration = Duration::from_millis(250); +const DASHBOARD_MAX_WEBSOCKET_CLIENTS: usize = 64; const DASHBOARD_RUN_ACTIVITY_FINGERPRINT_VOLATILE_FIELDS: &[&str] = &[ "idle_for_seconds", "protocol_idle_for_seconds", @@ -50,22 +51,50 @@ enum DashboardClientFrame { #[derive(Clone, Default)] struct DashboardEventHub { - clients: Arc>>>, + clients: Arc>>, + last_run_activity: Arc>>, + next_client_id: Arc>, } impl DashboardEventHub { - fn subscribe(&self) -> Result> { + fn subscribe(&self) -> Result { let (event_tx, event_rx) = mpsc::channel(); let mut clients = self .clients .lock() .map_err(|error| eyre::eyre!("Dashboard event client lock poisoned: {error}"))?; - clients.push(event_tx); + if clients.len() >= DASHBOARD_MAX_WEBSOCKET_CLIENTS { + eyre::bail!( + "Dashboard websocket client limit reached ({DASHBOARD_MAX_WEBSOCKET_CLIENTS})." + ); + } + + let mut next_client_id = self + .next_client_id + .lock() + .map_err(|error| eyre::eyre!("Dashboard event client id lock poisoned: {error}"))?; + let id = *next_client_id; + + *next_client_id = next_client_id.saturating_add(1); - Ok(event_rx) + clients.push(DashboardClientHandle { id, sender: event_tx }); + + Ok(DashboardClientRegistration { + id, + receiver: event_rx, + clients: Arc::clone(&self.clients), + }) } fn broadcast(&self, event_type: &'static str, payload: Value) { + let event = DashboardBroadcastEvent { event_type, payload }; + + if event_type == "runActivity" + && let Ok(mut last_run_activity) = self.last_run_activity.lock() + { + *last_run_activity = Some(event.clone()); + } + let Ok(mut clients) = self.clients.lock() else { tracing::warn!( "Skipped dashboard event broadcast because the client list lock is poisoned." @@ -73,23 +102,63 @@ impl DashboardEventHub { return; }; - let event = DashboardBroadcastEvent { event_type, payload }; - clients.retain(|client| client.send(event.clone()).is_ok()); + clients.retain(|client| client.sender.send(event.clone()).is_ok()); } fn has_clients(&self) -> bool { self.clients.lock().is_ok_and(|clients| !clients.is_empty()) } + fn cached_run_activity_event( + &self, + subscription: &DashboardClientSubscription, + ) -> Option { + self.last_run_activity + .lock() + .ok() + .and_then(|event| event.as_ref().and_then(|event| dashboard_event_for_subscription(event, subscription))) + } + #[cfg(test)] fn close_clients_for_test(&self) { if let Ok(mut clients) = self.clients.lock() { clients.clear(); } } + + #[cfg(test)] + fn client_count_for_test(&self) -> usize { + self.clients.lock().map(|clients| clients.len()).unwrap_or_default() + } } +#[derive(Debug)] +struct DashboardClientHandle { + id: u64, + sender: Sender, +} + +struct DashboardClientRegistration { + id: u64, + receiver: Receiver, + clients: Arc>>, +} +impl DashboardClientRegistration { + fn recv_timeout( + &self, + timeout: Duration, + ) -> std::result::Result { + self.receiver.recv_timeout(timeout) + } +} +impl Drop for DashboardClientRegistration { + fn drop(&mut self) { + if let Ok(mut clients) = self.clients.lock() { + clients.retain(|client| client.id != self.id); + } + } +} #[derive(Clone, Debug)] struct DashboardBroadcastEvent { event_type: &'static str, @@ -403,7 +472,7 @@ fn handle_operator_dashboard_websocket_connection( write_dashboard_websocket_event(&mut stream, "snapshot", &payload)?; } - write_current_dashboard_run_activity_event(&mut stream, state_store, &session.subscription); + write_cached_dashboard_run_activity_event(&mut stream, dashboard_events, &session.subscription); loop { for frame in read_dashboard_websocket_client_frames(&mut stream, &mut client_frame_buffer)? @@ -421,9 +490,9 @@ fn handle_operator_dashboard_websocket_connection( write_dashboard_websocket_event(&mut stream, "snapshot", &payload)?; } if dashboard_control_ack_should_push_run_activity(&response) { - write_current_dashboard_run_activity_event( + write_cached_dashboard_run_activity_event( &mut stream, - state_store, + dashboard_events, &session.subscription, ); } @@ -638,29 +707,22 @@ fn dashboard_control_ack_should_push_run_activity(ack: &Value) -> bool { ) } -fn write_current_dashboard_run_activity_event( +fn write_cached_dashboard_run_activity_event( stream: &mut TcpStream, - state_store: &StateStore, + dashboard_events: &DashboardEventHub, subscription: &DashboardClientSubscription, ) { - match build_operator_run_activity_event(state_store).and_then(|event| { - if let Some(event) = dashboard_event_for_subscription(&event.event, subscription) { - if !dashboard_run_activity_event_has_active_runs(&event) { - return Ok(()); + match dashboard_events.cached_run_activity_event(subscription) { + Some(event) if dashboard_run_activity_event_has_active_runs(&event) => { + if let Err(error) = write_dashboard_websocket_event(stream, event.event_type, &event.payload) + { + tracing::warn!( + ?error, + "Skipped cached dashboard run activity snapshot for a WebSocket client." + ); } - - write_dashboard_websocket_event(stream, event.event_type, &event.payload)?; - } - - Ok(()) - }) { - Ok(()) => {}, - Err(error) => { - tracing::warn!( - ?error, - "Skipped immediate dashboard run activity snapshot for a WebSocket client." - ); }, + Some(_) | None => {}, } } diff --git a/apps/decodex/src/orchestrator/status.rs b/apps/decodex/src/orchestrator/status.rs index ab97e0bc7..5b6349e6f 100644 --- a/apps/decodex/src/orchestrator/status.rs +++ b/apps/decodex/src/orchestrator/status.rs @@ -210,6 +210,26 @@ struct OperatorRunProtocolSummary { event_count: i64, } +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +struct OperatorTerminalFinalizeProjection { + status: &'static str, + phase: &'static str, + wait_reason: &'static str, + current_operation: &'static str, +} + +struct OperatorRunLifecycleProjection { + status: String, + phase: String, + wait_reason: Option, + current_operation: String, + suspected_stall: bool, + execution_liveness: String, + active_lease: bool, + retry_kind: Option, + retry_ready_at_unix_epoch: Option, +} + struct LiveOperatorStatusObserverContext<'a, T> { tracker: &'a T, project: &'a ServiceConfig, @@ -4897,45 +4917,30 @@ fn operator_run_status( let timing = operator_run_timing(&run, marker.as_ref(), now_unix_epoch); let app_server_state = operator_run_app_server_state(&run, marker.as_ref()); let protocol_summary = operator_run_protocol_summary(&run, marker.as_ref()); - let marker_current_operation = marker.as_ref().and_then(RunActivityMarker::current_operation); - let status = operator_run_visible_status( - run.status(), + let terminal_finalize_projection = + operator_run_terminal_finalize_projection(project, state_store, &run)?; + let lifecycle = operator_run_lifecycle_projection( + &run, + marker.as_ref(), + terminal_finalize_projection, + &timing, &app_server_state, &protocol_summary, - &timing, - marker_current_operation, - ); - let (retry_kind, retry_ready_at_unix_epoch) = visible_operator_run_retry_schedule( - &status, - marker.as_ref().and_then(RunActivityMarker::retry_kind), - marker.as_ref().and_then(RunActivityMarker::retry_ready_at_unix_epoch), - now_unix_epoch, - ); - let (phase, wait_reason) = classify_operator_run_phase( - &status, - retry_kind.as_deref(), - retry_ready_at_unix_epoch, now_unix_epoch, ); - let current_operation = classify_operator_run_operation( - &phase, - marker_current_operation, - ); - let suspected_stall = operator_run_is_suspected_stall( - &phase, - timing.last_progress_unix_epoch, - now_unix_epoch, - run_activity_idle_timeout(marker.as_ref()), - ); let child_agent_activity = operator_run_child_agent_activity(marker.as_ref(), now_unix_epoch); let protocol_activity = operator_run_protocol_activity( marker.as_ref(), &app_server_state, child_agent_activity.as_ref(), timing.protocol_idle_for_seconds, - matches!(status.as_str(), "starting" | "running"), + matches!(lifecycle.status.as_str(), "starting" | "running"), + ); + let wait_reason = operator_run_wait_reason( + &lifecycle.phase, + lifecycle.wait_reason.clone(), + protocol_activity.as_ref(), ); - let wait_reason = operator_run_wait_reason(&phase, wait_reason, protocol_activity.as_ref()); let (account, accounts) = operator_run_accounts(marker.as_ref()); let branch_name = run.branch_name().map(str::to_owned); let worktree_path = operator_run_relative_worktree_path(project, &run); @@ -4946,10 +4951,15 @@ fn operator_run_status( ); let private_evidence = operator_run_private_evidence(project, &run, issue_identifier.as_deref()); let loop_status = - operator_run_loop_status(project, state_store, &run, &status, &phase, ¤t_operation)?; + operator_run_loop_status( + project, + state_store, + &run, + &lifecycle.status, + &lifecycle.phase, + &lifecycle.current_operation, + )?; let control_capability = operator_run_control_capability(&run, &app_server_state); - let execution_liveness = - operator_run_execution_liveness(&status, &timing, &app_server_state, &protocol_summary); Ok(OperatorRunStatus { project_id: project.service_id().to_owned(), @@ -4960,20 +4970,20 @@ fn operator_run_status( title: None, author: None, attempt_number: run.attempt_number(), - status, + status: lifecycle.status, attempt_status: run.status().to_owned(), - phase, + phase: lifecycle.phase, wait_reason, - current_operation, + current_operation: lifecycle.current_operation, thread_id: app_server_state.thread_id, turn_id: app_server_state.turn_id, thread_status: app_server_state.thread_status, thread_active_flags: app_server_state.thread_active_flags, interactive_requested: app_server_state.interactive_requested, continuation_pending: app_server_state.continuation_pending, - active_lease: run.active_lease(), - queue_lease_state: operator_run_queue_lease_state(run.active_lease()), - execution_liveness, + active_lease: lifecycle.active_lease, + queue_lease_state: operator_run_queue_lease_state(lifecycle.active_lease), + execution_liveness: lifecycle.execution_liveness, updated_at: run.updated_at().to_owned(), last_run_activity_at: format_optional_unix_timestamp(timing.last_run_activity_unix_epoch), last_protocol_activity_at: format_optional_unix_timestamp( @@ -4982,7 +4992,7 @@ fn operator_run_status( last_progress_at: format_optional_unix_timestamp(timing.last_progress_unix_epoch), idle_for_seconds: timing.idle_for_seconds, protocol_idle_for_seconds: timing.protocol_idle_for_seconds, - suspected_stall, + suspected_stall: lifecycle.suspected_stall, last_event_type: protocol_summary.last_event_type, last_event_at: protocol_summary.last_event_at, event_count: protocol_summary.event_count, @@ -4992,8 +5002,8 @@ fn operator_run_status( process_id: timing.process_id, process_alive: timing.process_alive, process_liveness_reason: timing.process_liveness_reason, - retry_kind, - next_retry_at: format_optional_unix_timestamp(retry_ready_at_unix_epoch), + retry_kind: lifecycle.retry_kind, + next_retry_at: format_optional_unix_timestamp(lifecycle.retry_ready_at_unix_epoch), effective_model: app_server_state.effective_model, effective_model_provider: app_server_state.effective_model_provider, effective_cwd: app_server_state.effective_cwd, @@ -5009,6 +5019,74 @@ fn operator_run_status( }) } +fn operator_run_lifecycle_projection( + run: &ProjectRunStatus, + marker: Option<&RunActivityMarker>, + terminal_finalize_projection: Option, + timing: &OperatorRunTiming, + app_server_state: &OperatorRunAppServerState, + protocol_summary: &OperatorRunProtocolSummary, + now_unix_epoch: i64, +) -> OperatorRunLifecycleProjection { + let marker_current_operation = marker.and_then(RunActivityMarker::current_operation); + let status = terminal_finalize_projection + .map(|projection| projection.status.to_owned()) + .unwrap_or_else(|| { + operator_run_visible_status( + run.status(), + app_server_state, + protocol_summary, + timing, + marker_current_operation, + ) + }); + let (retry_kind, retry_ready_at_unix_epoch) = visible_operator_run_retry_schedule( + &status, + marker.and_then(RunActivityMarker::retry_kind), + marker.and_then(RunActivityMarker::retry_ready_at_unix_epoch), + now_unix_epoch, + ); + let (phase, wait_reason) = + if let Some(projection) = terminal_finalize_projection { + (String::from(projection.phase), Some(String::from(projection.wait_reason))) + } else { + classify_operator_run_phase( + &status, + retry_kind.as_deref(), + retry_ready_at_unix_epoch, + now_unix_epoch, + ) + }; + let current_operation = terminal_finalize_projection + .map(|projection| projection.current_operation.to_owned()) + .unwrap_or_else(|| classify_operator_run_operation(&phase, marker_current_operation)); + let suspected_stall = terminal_finalize_projection.is_none() + && operator_run_is_suspected_stall( + &phase, + timing.last_progress_unix_epoch, + now_unix_epoch, + run_activity_idle_timeout(marker), + ); + let execution_liveness = if terminal_finalize_projection.is_some() { + String::from("not_running") + } else { + operator_run_execution_liveness(&status, timing, app_server_state, protocol_summary) + }; + let active_lease = terminal_finalize_projection.is_none() && run.active_lease(); + + OperatorRunLifecycleProjection { + status, + phase, + wait_reason, + current_operation, + suspected_stall, + execution_liveness, + active_lease, + retry_kind, + retry_ready_at_unix_epoch, + } +} + fn operator_run_wait_reason( phase: &str, wait_reason: Option, @@ -5102,12 +5180,17 @@ fn operator_run_has_terminal_lifecycle( current_operation: &str, ) -> bool { phase == "completed" + || phase == "terminal_pending" || current_operation == "ledger_outcome" || matches!( status, "succeeded" | "failed" | "interrupted" + | "review_handoff_pending" + | "review_repair_pending" + | "closeout_pending" + | "manual_attention_pending" | "cleanup_complete" | "closeout" | "landed" @@ -5661,6 +5744,56 @@ fn operator_run_protocol_summary( } } +fn operator_run_terminal_finalize_projection( + project: &ServiceConfig, + state_store: &StateStore, + run: &ProjectRunStatus, +) -> crate::prelude::Result> { + let events = state_store.list_private_execution_events( + project.service_id(), + run.issue_id(), + run.run_id(), + run.attempt_number(), + )?; + let Some(path) = events + .iter() + .rev() + .find(|event| event.event_type() == "terminal_finalize") + .and_then(|event| event.payload().get("path")) + .and_then(Value::as_str) + else { + return Ok(None); + }; + + Ok(match path { + "review_handoff" => Some(OperatorTerminalFinalizeProjection { + status: "review_handoff_pending", + phase: "terminal_pending", + wait_reason: "review_handoff_writeback", + current_operation: RUN_OPERATION_REVIEW_WRITEBACK, + }), + "review_repair" => Some(OperatorTerminalFinalizeProjection { + status: "review_repair_pending", + phase: "terminal_pending", + wait_reason: "review_repair_writeback", + current_operation: RUN_OPERATION_REVIEW_WRITEBACK, + }), + "closeout" => Some(OperatorTerminalFinalizeProjection { + status: "closeout_pending", + phase: "terminal_pending", + wait_reason: "closeout_writeback", + current_operation: RUN_OPERATION_REVIEW_WRITEBACK, + }), + "manual_attention" => Some(OperatorTerminalFinalizeProjection { + status: "manual_attention_pending", + phase: "terminal_pending", + wait_reason: "manual_attention_writeback", + current_operation: RUN_OPERATION_REVIEW_WRITEBACK, + }), + _ => None, + }) +} + fn operator_run_visible_status( attempt_status: &str, app_server_state: &OperatorRunAppServerState, diff --git a/apps/decodex/src/orchestrator/tests/operator/status/http.rs b/apps/decodex/src/orchestrator/tests/operator/status/http.rs index 36d1e67f5..3688078c4 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/http.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/http.rs @@ -5,6 +5,8 @@ use orchestrator::OperatorControlRequests; use state::RUN_CONTROL_CHANNEL_DIR; use state::RUN_CONTROL_CHANNEL_TRANSPORT_LOCAL_FILE; use process::Child; +use orchestrator::DASHBOARD_MAX_WEBSOCKET_CLIENTS; +use orchestrator::DashboardClientSubscription; use crate::runtime; @@ -357,7 +359,7 @@ fn operator_dashboard_websocket_sends_current_snapshot_on_connect() { } #[test] -fn operator_dashboard_websocket_sends_current_run_activity_on_connect() { +fn operator_dashboard_websocket_sends_cached_run_activity_on_connect() { let (_temp_dir, config, _workflow) = temp_project_layout(); let listener = TcpListener::bind("127.0.0.1:0").expect("listener should bind"); let address = listener.local_addr().expect("listener address should resolve"); @@ -415,6 +417,11 @@ fn operator_dashboard_websocket_sends_current_run_activity_on_connect() { ) .expect("account marker should write"); + let run_activity = + orchestrator::build_operator_run_activity_event(&state_store).expect("run activity should build"); + + dashboard_events.broadcast(run_activity.event.event_type, run_activity.event.payload); + let server = thread::spawn(move || { let (stream, _) = listener.accept().expect("listener should accept a connection"); @@ -1251,6 +1258,92 @@ fn operator_dashboard_run_activity_fingerprint_ignores_volatile_timing_fields() ); } +#[test] +fn dashboard_event_hub_unregisters_websocket_clients_and_caps_fanout() { + let hub = DashboardEventHub::default(); + let mut registrations = Vec::new(); + + for _ in 0..DASHBOARD_MAX_WEBSOCKET_CLIENTS { + registrations.push(hub.subscribe().expect("client should subscribe below cap")); + } + + assert_eq!( + hub.client_count_for_test(), + orchestrator::DASHBOARD_MAX_WEBSOCKET_CLIENTS + ); + assert!( + hub.subscribe().is_err(), + "client fanout should be capped instead of growing unbounded" + ); + + drop(registrations.pop()); + + assert_eq!( + hub.client_count_for_test(), + orchestrator::DASHBOARD_MAX_WEBSOCKET_CLIENTS - 1 + ); + + let replacement = hub.subscribe().expect("slot should reopen after client drop"); + + assert_eq!( + hub.client_count_for_test(), + orchestrator::DASHBOARD_MAX_WEBSOCKET_CLIENTS + ); + + drop(replacement); + drop(registrations); + + assert_eq!(hub.client_count_for_test(), 0); +} + +#[test] +fn dashboard_event_hub_caches_and_filters_last_run_activity_event() { + let hub = DashboardEventHub::default(); + let payload = serde_json::json!({ + "emittedAtUnixEpoch": 1_774_000_000, + "accountControl": { + "mode": "balanced", + "account_selector": null, + }, + "accounts": [], + "activeRuns": [ + { + "project_id": "decodex", + "issue_id": "issue-1", + "run_id": "run-1", + }, + { + "project_id": "decodex", + "issue_id": "issue-2", + "run_id": "run-2", + }, + ], + "activeRunsComplete": true, + "activeRunScope": "complete", + }); + let subscription = DashboardClientSubscription { + project_id: Some(String::from("decodex")), + issue_id: Some(String::from("issue-1")), + run_id: None, + }; + + hub.broadcast("runActivity", payload); + hub.broadcast("snapshot", serde_json::json!({"ignored": true})); + + let event = hub + .cached_run_activity_event(&subscription) + .expect("cached run activity should remain available after other event types"); + let active_runs = event.payload["activeRuns"] + .as_array() + .expect("filtered active runs should be an array"); + + assert_eq!(event.event_type, "runActivity"); + assert_eq!(active_runs.len(), 1); + assert_eq!(active_runs[0]["issue_id"], "issue-1"); + assert_eq!(event.payload["activeRunsComplete"], false); + assert_eq!(event.payload["activeRunScope"], "filtered"); +} + #[test] fn operator_dashboard_run_activity_event_includes_disabled_project_active_runs() { let temp_dir = TempDir::new().expect("temp dir should exist"); diff --git a/apps/decodex/src/orchestrator/tests/operator/status/running_lanes.rs b/apps/decodex/src/orchestrator/tests/operator/status/running_lanes.rs index c24f7f09a..0e05e643c 100644 --- a/apps/decodex/src/orchestrator/tests/operator/status/running_lanes.rs +++ b/apps/decodex/src/orchestrator/tests/operator/status/running_lanes.rs @@ -1413,6 +1413,162 @@ fn operator_status_snapshot_keeps_terminal_status_live_process_in_running_lanes( assert_eq!(project.retained_worktree_count, 0); } +#[test] +fn operator_status_projects_terminal_finalized_run_as_pending_not_active() { + let (_temp_dir, config, _workflow) = temp_project_layout(); + let state_store = StateStore::open_in_memory().expect("state store should open"); + let registration = ProjectRegistration::from_config( + config.service_id(), + &service_config_path(config.repo_root()), + &config, + true, + "test-fingerprint", + ); + let issue = sample_issue("Todo", &[]); + let worktree_path = config.worktree_root().join("PUB-101"); + + state_store.upsert_project(®istration).expect("project should register"); + state_store + .record_run_attempt("run-1", &issue.id, 1, "running") + .expect("run attempt should record"); + state_store.update_run_thread("run-1", "thread-1").expect("thread should record"); + state_store.update_run_turn("run-1", "turn-1").expect("turn should record"); + state_store + .upsert_lease("pubfi", &issue.id, "run-1", "In Progress") + .expect("lease should record"); + state_store + .upsert_worktree( + "pubfi", + &issue.id, + "x/pubfi-pub-101", + &worktree_path.display().to_string(), + ) + .expect("worktree should record"); + state_store + .append_event("run-1", 1, "skills/changed", "{}") + .expect("protocol event should record"); + state_store + .append_private_execution_event( + "pubfi", + &issue.id, + "run-1", + 1, + "terminal_finalize", + serde_json::json!({ + "path": "review_handoff", + "mode": "handoff", + "branch": "x/pubfi-pub-101", + "worktree_path": worktree_path.display().to_string(), + }), + ) + .expect("terminal finalize event should record"); + + fs::create_dir_all(&worktree_path).expect("worktree path should exist"); + + let mut child = Command::new("sleep").arg("30").spawn().expect("sleep child should start"); + let child_pid = child.id(); + + state::write_run_activity_marker_for_process(&worktree_path, "run-1", 1, child_pid) + .expect("live process marker should write"); + + let snapshot = orchestrator::build_operator_status_snapshot(&config, &state_store, 10) + .expect("snapshot should build"); + + assert_terminal_pending_status_projection(&snapshot); + assert_terminal_pending_lane_inspect(&state_store); + assert_terminal_pending_interrupt_rejects_force(&state_store); + + if matches!(child.try_wait(), Ok(None)) { + child.kill().expect("sleep child should be killable"); + } + + child.wait().expect("sleep child should reap"); +} + +fn assert_terminal_pending_status_projection(snapshot: &OperatorStatusSnapshot) { + let project = snapshot.projects.first().expect("project summary should exist"); + let run = snapshot + .recent_runs + .iter() + .find(|run| run.run_id == "run-1") + .expect("terminal-pending run should remain inspectable in recent runs"); + + assert!( + snapshot.active_runs.is_empty(), + "terminal-finalized runs must not keep presenting as active execution" + ); + assert_eq!(project.active_run_count, 0); + assert_eq!(project.running_lane_count, 0); + assert_eq!(run.status, "review_handoff_pending"); + assert_eq!(run.attempt_status, "running"); + assert_eq!(run.phase, "terminal_pending"); + assert_eq!(run.wait_reason.as_deref(), Some("review_handoff_writeback")); + assert_eq!(run.current_operation, state::RUN_OPERATION_REVIEW_WRITEBACK); + assert!(!run.active_lease); + assert_eq!(run.queue_lease_state, "not_held"); + assert_eq!(run.execution_liveness, "not_running"); + assert!(!run.suspected_stall); + assert_eq!(run.last_event_type.as_deref(), Some("skills/changed")); + assert_eq!( + run.loop_status.as_ref().map(|status| status.summary.as_str()), + Some("terminal lifecycle: review_handoff_pending") + ); +} + +fn assert_terminal_pending_lane_inspect(state_store: &StateStore) { + let response = String::from_utf8(orchestrator::build_operator_lane_inspect_http_response( + state_store, + format!( + "GET {}?projectId=pubfi&issue=PUB-101&runId=run-1 HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n", + orchestrator::OPERATOR_LANE_INSPECT_ENDPOINT_PATH + ) + .as_bytes(), + )) + .expect("lane inspect response should be utf-8"); + let data = operator_status_response_json_body(&response, "lane inspect"); + + assert!(response.starts_with("HTTP/1.1 200 OK\r\n")); + assert_eq!(data["matchedRunCount"], 1); + assert_eq!(data["runs"][0]["status"], "review_handoff_pending"); + assert_eq!(data["runs"][0]["phase"], "terminal_pending"); + assert_eq!(data["runs"][0]["waitReason"], "review_handoff_writeback"); + assert_eq!(data["runs"][0]["currentOperation"], state::RUN_OPERATION_REVIEW_WRITEBACK); + assert_eq!(data["runs"][0]["activeLease"], false); + assert_eq!(data["runs"][0]["executionLiveness"], "not_running"); + assert_eq!(data["runs"][0]["softInterruptAvailable"], false); + assert_eq!(data["runs"][0]["hardInterruptAvailable"], false); +} + +fn assert_terminal_pending_interrupt_rejects_force(state_store: &StateStore) { + let body = br#"{"projectId":"pubfi","issue":"PUB-101","runId":"run-1","force":true}"#; + let response = String::from_utf8(orchestrator::build_operator_lane_interrupt_http_response( + state_store, + format!( + "POST {} HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + orchestrator::OPERATOR_LANE_INTERRUPT_ENDPOINT_PATH, + body.len(), + String::from_utf8_lossy(body) + ) + .as_bytes(), + )) + .expect("lane interrupt response should be utf-8"); + let data = operator_status_response_json_body(&response, "lane interrupt"); + + assert!(response.starts_with("HTTP/1.1 200 OK\r\n")); + assert_eq!(data["classification"], "soft_interrupt_unavailable"); + assert_eq!(data["softInterrupt"]["errorClass"], "lane_not_active"); + assert_eq!(data["hardInterrupt"], Value::Null); +} + +fn operator_status_response_json_body(response: &str, context: &str) -> Value { + let body = response + .split_once("\r\n\r\n") + .map(|(_, body)| body) + .unwrap_or_else(|| panic!("{context} response should include body")); + + serde_json::from_str(body).unwrap_or_else(|_| panic!("{context} response should be json")) +} + #[test] fn operator_status_snapshot_promotes_starting_after_app_server_activity() { let (_temp_dir, config, _workflow) = temp_project_layout(); diff --git a/docs/reference/operator-control-plane.md b/docs/reference/operator-control-plane.md index a7631d68d..76c51475c 100644 --- a/docs/reference/operator-control-plane.md +++ b/docs/reference/operator-control-plane.md @@ -498,6 +498,13 @@ map to an operator decision. - `status` is the operator-facing run status. If the raw attempt is still `starting` after app-server thread, model, or protocol evidence exists, `status` is shown as `running` and `attempt_status` preserves the raw persisted attempt value. +- When a worker has recorded `issue_terminal_finalize` but the surrounding retained + lifecycle has not finished writing back, the operator projection uses + `phase = terminal_pending`. Statuses such as `review_handoff_pending`, + `review_repair_pending`, `closeout_pending`, and `manual_attention_pending` mean + Decodex is finishing the terminal path. They are not active execution, do not hold a + queue lease, do not count as suspected stalls, and should not expose hard-interrupt + fallback as an available control. - Child-agent activity comes from `.decodex-run-activity` when the app-server recorder captured model/tool/tracker/browser/image buckets. - The child-agent breakdown is diagnostic. It explains where observed wall time went; diff --git a/plugins/decodex/.codex-plugin/plugin.json b/plugins/decodex/.codex-plugin/plugin.json index ae7814c8c..9dfd31a71 100644 --- a/plugins/decodex/.codex-plugin/plugin.json +++ b/plugins/decodex/.codex-plugin/plugin.json @@ -21,7 +21,7 @@ "interface": { "displayName": "Decodex", "shortDescription": "Use Decodex for research/design intake, planning, manual CLI work, or retained-lane automation.", - "longDescription": "Routes Codex agents through Decodex's natural-language-first surfaces: ordinary requests like \"research X\" enter Decodex-native research/design and produce latent Decision Contracts rather than automatic implementation; later follow-ups such as \"arrange this\", \"push this forward\", \"推进\", or \"做\" can promote accepted contracts into planning and issue shaping; planning splits accepted work into normal Linear issues and queues only ready nodes through registered project policy; manual CLI workflows such as commit, land, status, project registration, account selection, and dry runs stay explicit; and runtime-owned automation workflows such as serve, registered project contracts, Linear labels, tracker tools, review handoff, landing, closeout, and operator status keep their narrow authority. The plugin keeps graph, DAG, goal, and queue mechanics backstage for ordinary users while preserving Decodex-specific instructions with the runtime instead of duplicating them in generic playbooks.", + "longDescription": "Routes Codex agents through Decodex natural-language-first research/design intake, such as \"research X\", which produces latent Decision Contracts before execution. Follow-ups such as \"arrange this\" or \"推进\" can promote accepted work into planning that queues only ready nodes, while graph, DAG, goal, and queue mechanics backstage remain hidden from ordinary users.", "developerName": "hack-ink", "category": "Development", "capabilities": [ @@ -34,10 +34,7 @@ "defaultPrompt": [ "Research how Decodex should handle this.", "Arrange the accepted Decodex research contract into executable issues.", - "Use Decodex to inspect this lane.", - "Split this feature into Decodex-friendly independent issues.", - "Run the Decodex automation preflight.", - "Land this PR through Decodex." + "Use Decodex to inspect this lane." ], "brandColor": "#0F766E" } diff --git a/plugins/decodex/skills/automation/SKILL.md b/plugins/decodex/skills/automation/SKILL.md index 9cd835e57..9a2f27fcf 100644 --- a/plugins/decodex/skills/automation/SKILL.md +++ b/plugins/decodex/skills/automation/SKILL.md @@ -1,6 +1,6 @@ --- name: automation -description: "Use for Decodex runtime-owned automation: registered projects, `serve`, automatic issue intake, retained lanes, tracker tools, review handoff, repair, landing, closeout, cleanup, and operator recovery. Does not own manual commit or manual PR landing details." +description: "Use when Decodex owns runtime automation: registered projects, serve, issue intake, retained lanes, tracker tools, review handoff, repair, landing, closeout, cleanup, or operator recovery." --- # Automation @@ -131,6 +131,12 @@ terminal automation signal. that the lane is currently blocked, when the Run Ledger shows terminal closeout, cleanup, or landed completion and the active/backlog/recovery/post-review sections are empty. +- After an owned agent records `issue_terminal_finalize`, status or lane inspect may + briefly project `phase = terminal_pending` with statuses such as + `review_handoff_pending`, `review_repair_pending`, `closeout_pending`, or + `manual_attention_pending`. Treat this as terminal lifecycle writeback, not active + execution or a stall. Do not hard-interrupt, requeue, or clear labels from that row; + wait for the retained review, closeout, cleanup, or manual-attention path to settle. - When app-server preflight mentions `skills/list`, distinguish non-blocking scan diagnostics from real blockers. If the run cwd is present and at least one skill is enabled, preserve `error_count`, `first_error_path`, and `first_error` as evidence diff --git a/plugins/decodex/skills/decodex/SKILL.md b/plugins/decodex/skills/decodex/SKILL.md index b56089924..57d965938 100644 --- a/plugins/decodex/skills/decodex/SKILL.md +++ b/plugins/decodex/skills/decodex/SKILL.md @@ -1,6 +1,6 @@ --- name: decodex -description: Use as the conductor for Decodex work whenever the user asks to use, configure, operate, debug, or author Decodex. Routes between manual CLI workflows and runtime-owned automation workflows, and keeps Decodex-specific authority in the Decodex repo rather than generic Playbook guidance. +description: Use when the user asks to use, configure, operate, debug, or author Decodex. Routes research/design, planning, manual CLI, and runtime-owned automation workflows through Decodex-specific authority. --- # Decodex diff --git a/plugins/decodex/skills/manual-cli/SKILL.md b/plugins/decodex/skills/manual-cli/SKILL.md index a7c166902..4e66ef178 100644 --- a/plugins/decodex/skills/manual-cli/SKILL.md +++ b/plugins/decodex/skills/manual-cli/SKILL.md @@ -1,6 +1,6 @@ --- name: manual-cli -description: "Use for human-driven Decodex CLI workflows: project registration, probe, status, dry-run probes, live-run routing, commit, land, archive hygiene, and local operator inspection. Does not own retained-lane automation policy beyond routing to the automation skill." +description: "Use when a human is driving Decodex CLI workflows: project registration, probe, status, dry-run checks, live-run routing, commit, land, archive hygiene, or local operator inspection." --- # Manual CLI @@ -144,6 +144,11 @@ Manual commit and landing are separate narrow workflows: attempts under `history_lanes[].attempts` are retained for diagnosis and should not be read as current lane failure when the primary sections show no active, queued, recovery, or post-review work for that issue. +- If `status` or `lane inspect` shows `phase = terminal_pending`, the agent already + called `issue_terminal_finalize` and Decodex is finishing terminal lifecycle + writeback. Statuses such as `review_handoff_pending`, `review_repair_pending`, + `closeout_pending`, and `manual_attention_pending` are not active lanes and should + not be interrupted, requeued, or manually cleaned up from the side. - Use `status --live` when the operator needs fresh Linear/GitHub observer readback before acting. Use `/api/accounts?refresh=1` for fresh ChatGPT account usage probes. - Use `run --dry-run` before live automation to validate project loading, issue