michel/clusterflux-public
1
0
Fork 0
Code Issues Pull requests Projects Releases 21 Packages Wiki Activity Actions

Public dry run dryrun-199c53541aa2

Browse source
This commit is contained in:
Michel Paulissen 2026-07-03 20:31:07 +02:00
parent c8526c39ef
commit 84ef57d84b
5 changed files with 304 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

4
DISASMER_PUBLIC_TREE.json
View file

@ -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/**",

215
crates/disasmer-cli/src/main.rs
View file

@ -1121,6 +1121,7 @@ fn process_restart_report(args: ProcessRestartArgs) -> Result<Value> {
"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<RunPlan>
})
}
fn run_report(args: RunArgs, cwd: PathBuf, session: CliSession) -> Result<Value> {
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<Value> {
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<String> {
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<Value> {
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<Value> {
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 {

10
crates/disasmer-coordinator/src/lib.rs
View file

@ -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()
}

81
crates/disasmer-coordinator/src/service.rs
View file

@ -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();

6
scripts/cli-first-contract-smoke.js
View file

@ -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/],