From 84ef57d84b648f73f4f2501c02e447284f5fca69 Mon Sep 17 00:00:00 2001 From: Michel Paulissen <862400+MichelPaulissen@users.noreply.github.com> Date: Fri, 3 Jul 2026 20:31:07 +0200 Subject: [PATCH] Public dry run dryrun-199c53541aa2 --- DISASMER_PUBLIC_TREE.json | 4 +- crates/disasmer-cli/src/main.rs | 215 ++++++++++++++++++++- crates/disasmer-coordinator/src/lib.rs | 10 + crates/disasmer-coordinator/src/service.rs | 81 ++++++++ scripts/cli-first-contract-smoke.js | 6 + 5 files changed, 304 insertions(+), 12 deletions(-) diff --git a/DISASMER_PUBLIC_TREE.json b/DISASMER_PUBLIC_TREE.json index 841a9f8..8b3ab8f 100644 --- a/DISASMER_PUBLIC_TREE.json +++ b/DISASMER_PUBLIC_TREE.json @@ -1,7 +1,7 @@ { "kind": "disasmer-filtered-public-tree", - "source_commit": "7143846e2cca94a1e37c12d8dc9b2b3344f1f692", - "release_name": "dryrun-7143846e2cca", + "source_commit": "199c53541aa226880d242f7bb5c5d122961d9d87", + "release_name": "dryrun-199c53541aa2", "filtered_out": [ "private/**", "experiments/**", diff --git a/crates/disasmer-cli/src/main.rs b/crates/disasmer-cli/src/main.rs index 4199690..7b71620 100644 --- a/crates/disasmer-cli/src/main.rs +++ b/crates/disasmer-cli/src/main.rs @@ -1121,6 +1121,7 @@ fn process_restart_report(args: ProcessRestartArgs) -> Result { "tenant": args.scope.tenant, "project": args.scope.project, "process": args.process, + "restart": true, }))?; let restart_request = process_restart_request_summary(&response, !args.yes); return Ok(json!({ @@ -1824,13 +1825,8 @@ fn main() -> Result<()> { } Commands::Run(args) => { let json_output = args.json; - let plan = run_plan(args, std::env::current_dir()?, session_from_env())?; - if should_execute_local_node(&plan) { - let report = execute_local_node_run(plan)?; - emit_report(&report, json_output)?; - } else { - emit_report(&plan, json_output)?; - } + let report = run_report(args, std::env::current_dir()?, session_from_env())?; + emit_report(&report, json_output)?; } Commands::Node { command: NodeCommands::Attach(args), @@ -3104,6 +3100,117 @@ fn run_plan(args: RunArgs, cwd: PathBuf, session: CliSession) -> Result }) } +fn run_report(args: RunArgs, cwd: PathBuf, session: CliSession) -> Result { + let plan = run_plan(args, cwd, session)?; + if should_execute_local_node(&plan) { + return Ok(serde_json::to_value(execute_local_node_run(plan)?)?); + } + coordinator_run_report(plan) +} + +fn coordinator_run_report(plan: RunPlan) -> Result { + let config = read_project_config(&plan.project)?; + let tenant = config + .as_ref() + .map(|config| config.tenant.clone()) + .unwrap_or_else(|| "tenant".to_owned()); + let project = config + .as_ref() + .map(|config| config.project.clone()) + .unwrap_or_else(|| "project".to_owned()); + let user = config + .as_ref() + .map(|config| config.user.clone()) + .unwrap_or_else(|| "user".to_owned()); + let coordinator = run_coordinator_endpoint(&plan)?; + let process = "vp-current".to_owned(); + let mut session = JsonLineSession::connect(&coordinator)?; + let response = session.request_allow_error(json!({ + "type": "start_process", + "tenant": tenant.clone(), + "project": project.clone(), + "process": process.clone(), + "restart": false, + }))?; + let run_start = run_start_summary(&response); + let status = run_start + .get("status") + .and_then(Value::as_str) + .unwrap_or("coordinator_response"); + Ok(json!({ + "command": "run", + "status": status, + "project_root": plan.project, + "entry": plan.entry, + "tenant": tenant, + "project": project, + "user": user, + "coordinator": coordinator, + "process": process, + "run_start": run_start, + "coordinator_response": response, + "coordinator_session_requests": session.requests(), + "private_website_required": false, + })) +} + +fn run_coordinator_endpoint(plan: &RunPlan) -> Result { + match &plan.coordinator { + CoordinatorSelection::Hosted => Ok(plan + .operator_endpoint + .clone() + .unwrap_or_else(default_operator_endpoint)), + CoordinatorSelection::LocalOverride(coordinator) => Ok(coordinator.clone()), + CoordinatorSelection::LocalOnly => { + anyhow::bail!("local-only run should execute through local services") + } + } +} + +fn run_start_summary(response: &Value) -> Value { + if response.get("type").and_then(Value::as_str) == Some("process_started") { + return json!({ + "status": "started", + "accepted": true, + "process": response.get("process").cloned().unwrap_or(Value::Null), + "coordinator_epoch": response.get("epoch").cloned().unwrap_or(Value::Null), + "restart": false, + "single_active_process_boundary": true, + "next_actions": [ + "disasmer process status", + "disasmer logs", + "disasmer process cancel" + ], + }); + } + + let message = response + .get("message") + .and_then(Value::as_str) + .unwrap_or("coordinator rejected run"); + let active_conflict = message.contains("already has active virtual process"); + json!({ + "status": if active_conflict { "blocked_active_process" } else { "coordinator_rejected" }, + "accepted": false, + "category": if active_conflict { "active_process_already_running" } else { "coordinator" }, + "message": message, + "restart": false, + "single_active_process_boundary": true, + "safe_failure": true, + "next_actions": if active_conflict { + json!([ + "disasmer process status", + "disasmer logs", + "disasmer process restart --yes", + "disasmer process cancel --yes", + "wait for the active process to finish" + ]) + } else { + json!(["disasmer doctor", "check coordinator status"]) + }, + }) +} + fn should_execute_local_node(plan: &RunPlan) -> bool { match &plan.coordinator { CoordinatorSelection::LocalOnly => true, @@ -3441,6 +3548,14 @@ impl JsonLineSession { } fn request(&mut self, value: Value) -> Result { + let response = self.request_allow_error(value)?; + if response.get("type").and_then(Value::as_str) == Some("error") { + anyhow::bail!("coordinator error: {response}"); + } + Ok(response) + } + + fn request_allow_error(&mut self, value: Value) -> Result { serde_json::to_writer(&mut self.writer, &value)?; self.writer.write_all(b"\n")?; self.writer.flush()?; @@ -3451,9 +3566,6 @@ impl JsonLineSession { } self.requests += 1; let response: Value = serde_json::from_str(&line)?; - if response.get("type").and_then(Value::as_str) == Some("error") { - anyhow::bail!("coordinator error: {response}"); - } Ok(response) } @@ -3621,6 +3733,89 @@ mod tests { assert_eq!(plan.operator_endpoint, None); } + #[test] + fn run_contacts_configured_coordinator_and_reports_active_process_conflicts() { + let temp = tempfile::tempdir().unwrap(); + write_project_config( + temp.path(), + &ProjectConfig { + tenant: "tenant-live".to_owned(), + project: "project-live".to_owned(), + user: "user-live".to_owned(), + coordinator: None, + }, + ) + .unwrap(); + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + let server = std::thread::spawn(move || { + for response in [ + r#"{"type":"process_started","process":"vp-current","epoch":7}"#, + r#"{"type":"error","message":"coordinator request failed: unauthorized coordinator action: project already has active virtual process vp-current; inspect it, restart it, cancel it, or wait before starting another run"}"#, + ] { + 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":"start_process""#)); + assert!(line.contains(r#""tenant":"tenant-live""#)); + assert!(line.contains(r#""project":"project-live""#)); + assert!(line.contains(r#""process":"vp-current""#)); + assert!(line.contains(r#""restart":false"#)); + stream.write_all(response.as_bytes()).unwrap(); + stream.write_all(b"\n").unwrap(); + } + }); + + let started = run_report( + RunArgs { + entry: Some("build".to_owned()), + project: Some(temp.path().to_path_buf()), + coordinator: Some(format!("http://{addr}")), + local: false, + json: false, + }, + PathBuf::from("/unused"), + CliSession::Anonymous, + ) + .unwrap(); + let blocked = run_report( + RunArgs { + entry: Some("test".to_owned()), + project: Some(temp.path().to_path_buf()), + coordinator: Some(format!("http://{addr}")), + local: false, + json: false, + }, + PathBuf::from("/unused"), + CliSession::Anonymous, + ) + .unwrap(); + server.join().unwrap(); + + assert_eq!(started["status"], "started"); + assert_eq!(started["entry"], "build"); + assert_eq!(started["tenant"], "tenant-live"); + assert_eq!(started["project"], "project-live"); + assert_eq!(started["process"], "vp-current"); + assert_eq!(started["run_start"]["restart"], false); + assert_eq!(started["run_start"]["single_active_process_boundary"], true); + + assert_eq!(blocked["status"], "blocked_active_process"); + assert_eq!(blocked["entry"], "test"); + assert_eq!( + blocked["run_start"]["category"], + "active_process_already_running" + ); + assert_eq!(blocked["run_start"]["safe_failure"], true); + assert!(blocked["run_start"]["next_actions"] + .as_array() + .unwrap() + .iter() + .any(|action| action == "disasmer process restart --yes")); + } + #[test] fn local_only_run_executes_ephemeral_local_services() { let Cli { diff --git a/crates/disasmer-coordinator/src/lib.rs b/crates/disasmer-coordinator/src/lib.rs index b11993d..4874afb 100644 --- a/crates/disasmer-coordinator/src/lib.rs +++ b/crates/disasmer-coordinator/src/lib.rs @@ -442,6 +442,16 @@ impl Coordinator { self.active_processes.get(id) } + pub fn active_process_for_project( + &self, + tenant: &TenantId, + project: &ProjectId, + ) -> Option<&ActiveProcess> { + self.active_processes + .values() + .find(|active| &active.tenant == tenant && &active.project == project) + } + pub fn active_process_count(&self) -> usize { self.active_processes.len() } diff --git a/crates/disasmer-coordinator/src/service.rs b/crates/disasmer-coordinator/src/service.rs index 9b3125a..6959ec0 100644 --- a/crates/disasmer-coordinator/src/service.rs +++ b/crates/disasmer-coordinator/src/service.rs @@ -151,6 +151,8 @@ pub enum CoordinatorRequest { tenant: String, project: String, process: String, + #[serde(default)] + restart: bool, }, ReconnectNode { node: String, @@ -1160,10 +1162,23 @@ impl CoordinatorService { tenant, project, process, + restart, } => { let tenant = TenantId::new(tenant); let project = ProjectId::new(project); let process = ProcessId::new(process); + if let Some(active) = self + .coordinator + .active_process_for_project(&tenant, &project) + { + if active.id != process || !restart { + return Err(CoordinatorError::Unauthorized(format!( + "project already has active virtual process {}; inspect it, restart it, cancel it, or wait before starting another run", + active.id + )) + .into()); + } + } self.process_cancellations .remove(&process_control_key(&tenant, &project, &process)); self.task_cancellations.retain( @@ -2299,6 +2314,7 @@ mod tests { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), + restart: false, }) .unwrap(); assert_eq!( @@ -2460,6 +2476,7 @@ mod tests { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), + restart: false, }) .unwrap(); service @@ -2596,6 +2613,7 @@ mod tests { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), + restart: false, }) .unwrap(); for node in ["node-a", "node-b"] { @@ -2703,6 +2721,63 @@ mod tests { .contains("virtual process is cancelling")); } + #[test] + fn service_rejects_second_active_process_unless_restarting_same_process() { + let mut service = CoordinatorService::new(7); + let started = service + .handle_request(CoordinatorRequest::StartProcess { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + process: "process-a".to_owned(), + restart: false, + }) + .unwrap(); + assert!(matches!( + started, + CoordinatorResponse::ProcessStarted { .. } + )); + + let same_without_restart = service + .handle_request(CoordinatorRequest::StartProcess { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + process: "process-a".to_owned(), + restart: false, + }) + .unwrap_err(); + assert!(same_without_restart + .to_string() + .contains("already has active virtual process")); + + let other_process = service + .handle_request(CoordinatorRequest::StartProcess { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + process: "process-b".to_owned(), + restart: true, + }) + .unwrap_err(); + assert!(other_process + .to_string() + .contains("already has active virtual process")); + + let restarted = service + .handle_request(CoordinatorRequest::StartProcess { + tenant: "tenant".to_owned(), + project: "project".to_owned(), + process: "process-a".to_owned(), + restart: true, + }) + .unwrap(); + assert_eq!( + restarted, + CoordinatorResponse::ProcessStarted { + process: ProcessId::from("process-a"), + epoch: 7, + } + ); + } + #[test] fn service_download_links_are_scoped_and_streaming_is_metered() { let mut service = CoordinatorService::new(7); @@ -2719,6 +2794,7 @@ mod tests { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), + restart: false, }) .unwrap(); service @@ -3067,6 +3143,7 @@ mod tests { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), + restart: false, }) .unwrap(); service @@ -3241,6 +3318,7 @@ mod tests { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "process".to_owned(), + restart: false, }) .unwrap(); @@ -3492,6 +3570,7 @@ mod tests { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "vp-control".to_owned(), + restart: false, }) .unwrap() else { @@ -3569,6 +3648,7 @@ mod tests { tenant: "tenant".to_owned(), project: "project".to_owned(), process: "vp-control".to_owned(), + restart: false, }) .unwrap(); @@ -3702,6 +3782,7 @@ mod tests { tenant: "tenant-b".to_owned(), project: "project-b".to_owned(), process: "process".to_owned(), + restart: false, }) .unwrap(); diff --git a/scripts/cli-first-contract-smoke.js b/scripts/cli-first-contract-smoke.js index 798ca94..2568eb4 100644 --- a/scripts/cli-first-contract-smoke.js +++ b/scripts/cli-first-contract-smoke.js @@ -109,6 +109,7 @@ for (const [name, pattern] of [ ["doctor ping reachability coverage", /fn doctor_pings_configured_coordinator\(\)/], ["project local config coverage", /fn project_init_select_and_status_use_local_project_config\(\)/], ["project coordinator status coverage", /fn project_status_queries_public_coordinator_state\(\)/], + ["run coordinator active-process coverage", /fn run_contacts_configured_coordinator_and_reports_active_process_conflicts\(\)/], ["quota local status coverage", /fn quota_status_uses_project_config_and_generic_public_limits\(\)/], ["quota coordinator usage coverage", /fn quota_status_queries_public_coordinator_usage\(\)/], ["task event summary coverage", /fn process_task_log_and_artifact_reports_summarize_task_events\(\)/], @@ -125,6 +126,11 @@ expect( "coordinator whole-process cancellation coverage", /fn service_cancels_whole_process_and_blocks_new_task_launches\(\)/ ); +expect( + coordinator, + "coordinator single active process coverage", + /fn service_rejects_second_active_process_unless_restarting_same_process\(\)/ +); for (const [name, pattern] of [ ["agent --json flag", /struct AgentEnrollArgs[\s\S]*#\[arg\(long\)\]\s*json: bool/],