From 0b14889c79f60117320b67a9d631e671e99d246f Mon Sep 17 00:00:00 2001 From: Michel Paulissen <862400+MichelPaulissen@users.noreply.github.com> Date: Fri, 3 Jul 2026 21:28:56 +0200 Subject: [PATCH] Expose public task restart boundary --- DISASMER_PUBLIC_TREE.json | 4 +- crates/disasmer-cli/src/main.rs | 165 +++++++++++- crates/disasmer-coordinator/src/service.rs | 249 +++++++++++++++++++ scripts/cli-first-contract-smoke.js | 8 +- scripts/user-session-token-boundary-smoke.js | 1 + 5 files changed, 419 insertions(+), 8 deletions(-) diff --git a/DISASMER_PUBLIC_TREE.json b/DISASMER_PUBLIC_TREE.json index 5915bd8..28f49cb 100644 --- a/DISASMER_PUBLIC_TREE.json +++ b/DISASMER_PUBLIC_TREE.json @@ -1,7 +1,7 @@ { "kind": "disasmer-filtered-public-tree", - "source_commit": "c7543c680764ad3efa8d96e8c427fddcba2d3a4a", - "release_name": "dryrun-c7543c680764", + "source_commit": "e50394e956649c00db26cca4307bc244988ccecd", + "release_name": "dryrun-e50394e95664", "filtered_out": [ "private/**", "experiments/**", diff --git a/crates/disasmer-cli/src/main.rs b/crates/disasmer-cli/src/main.rs index eb802f9..5c38ff2 100644 --- a/crates/disasmer-cli/src/main.rs +++ b/crates/disasmer-cli/src/main.rs @@ -368,6 +368,7 @@ struct ProcessCancelArgs { #[derive(Clone, Debug, Subcommand)] enum TaskCommands { List(TaskListArgs), + Restart(TaskRestartArgs), } #[derive(Clone, Debug, Parser)] @@ -378,6 +379,17 @@ struct TaskListArgs { process: Option, } +#[derive(Clone, Debug, Parser)] +struct TaskRestartArgs { + #[command(flatten)] + scope: CliScopeArgs, + task: String, + #[arg(long, default_value = "vp-current")] + process: String, + #[arg(long)] + yes: bool, +} + #[derive(Clone, Debug, Parser)] struct LogsArgs { #[command(flatten)] @@ -1307,6 +1319,44 @@ fn task_list_report(args: TaskListArgs) -> Result { })) } +fn task_restart_report(args: TaskRestartArgs) -> Result { + if let Some(coordinator) = &args.scope.coordinator { + let mut session = JsonLineSession::connect(coordinator)?; + let response = session.request_allow_error(json!({ + "type": "restart_task", + "tenant": args.scope.tenant, + "project": args.scope.project, + "actor_user": args.scope.user, + "process": args.process, + "task": args.task, + }))?; + let restart_request = task_restart_request_summary(&response, !args.yes); + return Ok(json!({ + "command": "task restart", + "coordinator": coordinator, + "process": args.process, + "task": args.task, + "requires_confirmation": !args.yes, + "restart_request": restart_request, + "response": response, + "coordinator_session_requests": session.requests(), + })); + } + Ok(json!({ + "command": "task restart", + "status": "requires_coordinator", + "requires_confirmation": !args.yes, + "process": args.process, + "task": args.task, + "restart_request": { + "status": "requires_coordinator", + "operation": "restart_selected_task", + "explicit_user_action": true, + "clean_boundary_required": true, + }, + })) +} + fn logs_report(args: LogsArgs) -> Result { let events = list_task_events_if_available( args.scope.coordinator.as_deref(), @@ -2048,11 +2098,18 @@ fn main() -> Result<()> { }; emit_report(&report, json_output)?; } - Commands::Task { - command: TaskCommands::List(args), - } => { - let json_output = args.scope.json; - emit_report(&task_list_report(args)?, json_output)?; + Commands::Task { command } => { + let (report, json_output) = match command { + TaskCommands::List(args) => { + let json_output = args.scope.json; + (task_list_report(args)?, json_output) + } + TaskCommands::Restart(args) => { + let json_output = args.scope.json; + (task_restart_report(args)?, json_output) + } + }; + emit_report(&report, json_output)?; } Commands::Logs(args) => { let json_output = args.scope.json; @@ -2592,6 +2649,49 @@ fn process_cancel_request_summary(response: &Value, requires_confirmation: bool) }) } +fn task_restart_request_summary(response: &Value, requires_confirmation: bool) -> Value { + if response.get("type").and_then(Value::as_str) != Some("task_restart") { + return json!({ + "status": response.get("type").and_then(Value::as_str).unwrap_or("coordinator_response"), + "operation": "restart_selected_task", + "accepted": false, + "requires_confirmation": requires_confirmation, + "explicit_user_action": true, + "clean_boundary_required": true, + "error": response.get("message").cloned().unwrap_or(Value::Null), + }); + } + + json!({ + "status": "task_restart", + "operation": "restart_selected_task", + "accepted": response.get("accepted").cloned().unwrap_or_else(|| json!(false)), + "process": response.get("process").cloned().unwrap_or(Value::Null), + "task": response.get("task").cloned().unwrap_or(Value::Null), + "requires_confirmation": requires_confirmation, + "explicit_user_action": true, + "clean_boundary_required": true, + "clean_boundary_available": response + .get("clean_boundary_available") + .cloned() + .unwrap_or_else(|| json!(false)), + "active_task": response + .get("active_task") + .cloned() + .unwrap_or_else(|| json!(false)), + "completed_event_observed": response + .get("completed_event_observed") + .cloned() + .unwrap_or_else(|| json!(false)), + "requires_whole_process_restart": response + .get("requires_whole_process_restart") + .cloned() + .unwrap_or_else(|| json!(true)), + "message": response.get("message").cloned().unwrap_or(Value::Null), + "website_required": false, + }) +} + fn artifact_download_session_summary(response: &Value) -> Value { if response.get("type").and_then(Value::as_str) != Some("artifact_download_link") { return json!({ @@ -4368,6 +4468,7 @@ mod tests { &["disasmer", "process", "restart", "--yes"], &["disasmer", "process", "cancel", "--yes"], &["disasmer", "task", "list"], + &["disasmer", "task", "restart", "compile-linux", "--yes"], &["disasmer", "logs"], &["disasmer", "artifact", "list"], &["disasmer", "artifact", "download", "artifact"], @@ -5191,6 +5292,60 @@ mod tests { assert_eq!(cancel["cancel_request"]["new_task_launches_blocked"], true); } + #[test] + fn task_restart_reports_clean_boundary_requirements() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut reader = BufReader::new(stream.try_clone().unwrap()); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + assert!(line.contains(r#""type":"restart_task""#)); + assert!(line.contains(r#""tenant":"tenant""#)); + assert!(line.contains(r#""project":"project""#)); + assert!(line.contains(r#""actor_user":"user""#)); + assert!(line.contains(r#""process":"vp""#)); + assert!(line.contains(r#""task":"compile-linux""#)); + stream + .write_all( + br#"{"type":"task_restart","process":"vp","task":"compile-linux","actor":"user","accepted":false,"clean_boundary_available":false,"active_task":true,"completed_event_observed":false,"requires_whole_process_restart":true,"message":"selected task is still active; clean task restart requires a captured checkpoint boundary"}"#, + ) + .unwrap(); + stream.write_all(b"\n").unwrap(); + }); + + let report = task_restart_report(TaskRestartArgs { + scope: CliScopeArgs { + coordinator: Some(addr), + tenant: "tenant".to_owned(), + project: "project".to_owned(), + user: "user".to_owned(), + json: false, + }, + task: "compile-linux".to_owned(), + process: "vp".to_owned(), + yes: true, + }) + .unwrap(); + server.join().unwrap(); + + assert_eq!(report["command"], "task restart"); + assert_eq!( + report["restart_request"]["operation"], + "restart_selected_task" + ); + assert_eq!(report["restart_request"]["accepted"], false); + assert_eq!(report["restart_request"]["clean_boundary_available"], false); + assert_eq!( + report["restart_request"]["requires_whole_process_restart"], + true + ); + assert_eq!(report["restart_request"]["active_task"], true); + assert_eq!(report["restart_request"]["website_required"], false); + assert_eq!(report["coordinator_session_requests"], 1); + } + #[test] fn build_command_reuses_bundle_inspection_without_full_repo_upload() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/disasmer-coordinator/src/service.rs b/crates/disasmer-coordinator/src/service.rs index bda6934..3f0c2a8 100644 --- a/crates/disasmer-coordinator/src/service.rs +++ b/crates/disasmer-coordinator/src/service.rs @@ -221,6 +221,13 @@ pub enum CoordinatorRequest { node: String, task: String, }, + RestartTask { + tenant: String, + project: String, + actor_user: String, + process: String, + task: String, + }, DebugAttach { tenant: String, project: String, @@ -558,6 +565,17 @@ pub enum CoordinatorResponse { task: TaskId, cancel_requested: bool, }, + TaskRestart { + process: ProcessId, + task: TaskId, + actor: UserId, + accepted: bool, + clean_boundary_available: bool, + active_task: bool, + completed_event_observed: bool, + requires_whole_process_restart: bool, + message: String, + }, DebugCommand { process: ProcessId, task: TaskId, @@ -1571,6 +1589,72 @@ impl CoordinatorService { cancel_requested, }) } + CoordinatorRequest::RestartTask { + tenant, + project, + actor_user, + process, + task, + } => { + let tenant = TenantId::new(tenant); + let project = ProjectId::new(project); + let actor = UserId::new(actor_user); + let process = ProcessId::new(process); + let task = TaskId::new(task); + let context = disasmer_core::AuthContext { + tenant: tenant.clone(), + project: project.clone(), + actor: Actor::User(actor.clone()), + }; + self.coordinator.upsert_user( + tenant.clone(), + actor.clone(), + CredentialKind::BrowserSession, + ); + let authorization = self.coordinator.authorize_debug_attach(&context, &process); + if !authorization.allowed { + return Err(CoordinatorError::Unauthorized(format!( + "task restart denied: {}", + authorization.reason + )) + .into()); + } + let active_key = self + .active_tasks + .iter() + .find(|(task_tenant, task_project, task_process, _, task_id)| { + task_tenant == &tenant + && task_project == &project + && task_process == &process + && task_id == &task + }) + .cloned(); + let active_task = active_key.is_some(); + let completed_event_observed = self.task_events.iter().any(|event| { + event.tenant == tenant + && event.project == project + && event.process == process + && event.task == task + }); + let message = if active_task { + "selected task is still active; clean task restart requires a captured checkpoint boundary" + } else if completed_event_observed { + "selected task has only terminal event metadata; restart requires a captured checkpoint boundary" + } else { + "selected task is not known in the active process; restart the whole virtual process or inspect task list" + }; + Ok(CoordinatorResponse::TaskRestart { + process, + task, + actor, + accepted: false, + clean_boundary_available: false, + active_task, + completed_event_observed, + requires_whole_process_restart: true, + message: message.to_owned(), + }) + } CoordinatorRequest::DebugAttach { tenant, project, @@ -3160,6 +3244,171 @@ mod tests { assert!(authorization.reason.contains("tenant or project")); } + #[test] + fn service_reports_task_restart_boundary_through_public_api() { + let mut service = CoordinatorService::new(7); + service + .handle_request(CoordinatorRequest::CreateProject { + tenant: "tenant".to_owned(), + actor_user: "user".to_owned(), + project: "project".to_owned(), + name: "Demo".to_owned(), + }) + .unwrap(); + service + .handle_request(CoordinatorRequest::StartProcess { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + process: "process".to_owned(), + restart: false, + }) + .unwrap(); + + let denied = service + .handle_request(CoordinatorRequest::RestartTask { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + actor_user: "other-user".to_owned(), + process: "process".to_owned(), + task: "task".to_owned(), + }) + .unwrap_err(); + assert!(denied.to_string().contains("task restart denied")); + + let CoordinatorResponse::TaskRestart { + accepted, + clean_boundary_available, + active_task, + completed_event_observed, + requires_whole_process_restart, + message, + .. + } = service + .handle_request(CoordinatorRequest::RestartTask { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + actor_user: "user".to_owned(), + process: "process".to_owned(), + task: "task".to_owned(), + }) + .unwrap() + else { + panic!("expected task restart response"); + }; + assert!(!accepted); + assert!(!clean_boundary_available); + assert!(!active_task); + assert!(!completed_event_observed); + assert!(requires_whole_process_restart); + assert!(message.contains("not known")); + + service + .handle_request(CoordinatorRequest::AttachNode { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + node: "worker-linux".to_owned(), + public_key: "node-public-key".to_owned(), + }) + .unwrap(); + service + .handle_request(CoordinatorRequest::ReportNodeCapabilities { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + node: "worker-linux".to_owned(), + capabilities: linux_capabilities(), + cached_environment_digests: Vec::new(), + dependency_cache_digests: Vec::new(), + source_snapshots: Vec::new(), + artifact_locations: Vec::new(), + direct_connectivity: true, + online: true, + }) + .unwrap(); + service + .handle_request(CoordinatorRequest::LaunchTask { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + actor_user: "user".to_owned(), + process: "process".to_owned(), + task: "task".to_owned(), + environment: None, + environment_digest: None, + required_capabilities: vec![Capability::Command], + dependency_cache: None, + source_snapshot: None, + required_artifacts: Vec::new(), + quota_available: true, + policy_allowed: true, + command: "cargo".to_owned(), + command_args: vec!["test".to_owned()], + artifact_path: "/vfs/artifacts/output.txt".to_owned(), + }) + .unwrap(); + + let CoordinatorResponse::TaskRestart { + active_task, + completed_event_observed, + message, + .. + } = service + .handle_request(CoordinatorRequest::RestartTask { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + actor_user: "user".to_owned(), + process: "process".to_owned(), + task: "task".to_owned(), + }) + .unwrap() + else { + panic!("expected active task restart response"); + }; + assert!(active_task); + assert!(!completed_event_observed); + assert!(message.contains("still active")); + + service + .handle_request(CoordinatorRequest::TaskCompleted { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + process: "process".to_owned(), + node: "worker-linux".to_owned(), + task: "task".to_owned(), + terminal_state: Some(TaskTerminalState::Completed), + status_code: Some(0), + stdout_bytes: 0, + stderr_bytes: 0, + stdout_tail: String::new(), + stderr_tail: String::new(), + stdout_truncated: false, + stderr_truncated: false, + artifact_path: None, + artifact_digest: None, + artifact_size_bytes: None, + }) + .unwrap(); + + let CoordinatorResponse::TaskRestart { + active_task, + completed_event_observed, + message, + .. + } = service + .handle_request(CoordinatorRequest::RestartTask { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + actor_user: "user".to_owned(), + process: "process".to_owned(), + task: "task".to_owned(), + }) + .unwrap() + else { + panic!("expected completed task restart response"); + }; + assert!(!active_task); + assert!(completed_event_observed); + assert!(message.contains("terminal event metadata")); + } + #[test] fn service_cancels_whole_process_and_blocks_new_task_launches() { let mut service = CoordinatorService::new(7); diff --git a/scripts/cli-first-contract-smoke.js b/scripts/cli-first-contract-smoke.js index c582a9c..cb99791 100644 --- a/scripts/cli-first-contract-smoke.js +++ b/scripts/cli-first-contract-smoke.js @@ -89,7 +89,7 @@ for (const [name, pattern] of [ ["build command", /Build\(BuildArgs\)/], ["node lifecycle commands", /enum NodeCommands[\s\S]*Attach\(AttachArgs\)[\s\S]*Enroll\(NodeEnrollArgs\)[\s\S]*List\(NodeListArgs\)[\s\S]*Status\(NodeStatusArgs\)[\s\S]*Revoke\(NodeRevokeArgs\)/], ["process commands", /enum ProcessCommands[\s\S]*Status\(ProcessStatusArgs\)[\s\S]*Restart\(ProcessRestartArgs\)[\s\S]*Cancel\(ProcessCancelArgs\)/], - ["task list command", /enum TaskCommands[\s\S]*List\(TaskListArgs\)/], + ["task lifecycle commands", /enum TaskCommands[\s\S]*List\(TaskListArgs\)[\s\S]*Restart\(TaskRestartArgs\)/], ["logs command", /Logs\(LogsArgs\)/], ["artifact commands", /enum ArtifactCommands[\s\S]*List\(ArtifactListArgs\)[\s\S]*Download\(ArtifactDownloadArgs\)[\s\S]*Export\(ArtifactExportArgs\)/], ["DAP command", /Dap\(DapArgs\)/], @@ -119,6 +119,7 @@ for (const [name, pattern] of [ ["task event summary coverage", /fn process_task_log_and_artifact_reports_summarize_task_events\(\)/], ["artifact download/export report coverage", /fn artifact_download_and_export_reports_expose_safe_session_boundaries\(\)/], ["process control report coverage", /fn process_restart_and_cancel_reports_expose_control_boundaries\(\)/], + ["task restart report coverage", /fn task_restart_reports_clean_boundary_requirements\(\)/], ["build no full repo upload coverage", /fn build_command_reuses_bundle_inspection_without_full_repo_upload\(\)/], ["safe coordinator-required plans", /fn node_enroll_and_process_commands_have_safe_plan_without_coordinator\(\)/], ]) { @@ -155,6 +156,11 @@ expect( "coordinator debug attach coverage", /fn service_authorizes_debug_attach_through_public_api\(\)/ ); +expect( + coordinator, + "coordinator task restart boundary coverage", + /fn service_reports_task_restart_boundary_through_public_api\(\)/ +); for (const [name, pattern] of [ ["agent --json flag", /struct AgentEnrollArgs[\s\S]*#\[arg\(long\)\]\s*json: bool/], diff --git a/scripts/user-session-token-boundary-smoke.js b/scripts/user-session-token-boundary-smoke.js index b93f66c..1ab3307 100755 --- a/scripts/user-session-token-boundary-smoke.js +++ b/scripts/user-session-token-boundary-smoke.js @@ -63,6 +63,7 @@ for (const variant of [ "CancelTask", "CancelProcess", "PollTaskControl", + "RestartTask", "DebugAttach", "TaskCompleted", ]) {