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

Expose public task restart boundary

Browse source
This commit is contained in:
Michel Paulissen 2026-07-03 21:28:56 +02:00
parent f586b48893
commit 0b14889c79
5 changed files with 419 additions and 8 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": "c7543c680764ad3efa8d96e8c427fddcba2d3a4a",
"release_name": "dryrun-c7543c680764",
"source_commit": "e50394e956649c00db26cca4307bc244988ccecd",
"release_name": "dryrun-e50394e95664",
"filtered_out": [
"private/**",
"experiments/**",

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

@ -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<String>,
}
#[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<Value> {
}))
}
fn task_restart_report(args: TaskRestartArgs) -> Result<Value> {
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<Value> {
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),
} => {
Commands::Task { command } => {
let (report, json_output) = match command {
TaskCommands::List(args) => {
let json_output = args.scope.json;
emit_report(&task_list_report(args)?, json_output)?;
(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();

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

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

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

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

1
scripts/user-session-token-boundary-smoke.js
View file

@ -63,6 +63,7 @@ for (const variant of [
"CancelTask",
"CancelProcess",
"PollTaskControl",
"RestartTask",
"DebugAttach",
"TaskCompleted",
]) {