Public dry run dryrun-22f391eb394a

This commit is contained in:
Michel Paulissen 2026-07-03 19:37:59 +02:00
parent 17e52b5a2f
commit 16cb76e6e2
3 changed files with 300 additions and 4 deletions

View file

@ -1,3 +1,4 @@
use std::collections::BTreeSet;
use std::io::{BufRead, BufReader, Write};
use std::net::{TcpListener, TcpStream, ToSocketAddrs};
use std::path::{Path, PathBuf};
@ -880,6 +881,7 @@ fn project_init_report(args: ProjectInitArgs, cwd: PathBuf) -> Result<Value> {
fn project_status_report(args: ProjectStatusArgs, cwd: PathBuf) -> Result<Value> {
let config = read_project_config(&cwd)?;
let effective_scope = effective_project_scope(&args.scope, config.as_ref());
let inspection = bundle_inspection(
BundleInspectArgs {
project: Some(cwd.clone()),
@ -888,14 +890,40 @@ fn project_status_report(args: ProjectStatusArgs, cwd: PathBuf) -> Result<Value>
cwd.clone(),
)
.ok();
let coordinator = effective_scope.coordinator.clone();
let attached_nodes =
list_attached_nodes_if_available(coordinator.as_deref(), &effective_scope)?;
let coordinator_response =
list_task_events_if_available(args.scope.coordinator.as_deref(), &args.scope, None)?;
list_task_events_if_available(coordinator.as_deref(), &effective_scope, None)?;
let discovered_environments = discovered_environment_names(inspection.as_ref());
let active_process = active_process_from_task_events(coordinator_response.as_ref())
.unwrap_or_else(|| {
if coordinator.is_some() {
"none_observed".to_owned()
} else {
"unknown_without_coordinator".to_owned()
}
});
let quota_posture = project_quota_posture(&attached_nodes, coordinator_response.as_ref());
Ok(json!({
"command": "project status",
"cwd": cwd,
"tenant": effective_scope.tenant,
"project": effective_scope.project,
"user": effective_scope.user,
"coordinator": coordinator,
"project_identity": {
"tenant": effective_scope.tenant,
"project": effective_scope.project,
"user": effective_scope.user,
"source": if config.is_some() { "project_config_with_cli_overrides" } else { "cli_scope" }
},
"project_config": config,
"bundle": inspection,
"active_process": "vp-current",
"discovered_environments": discovered_environments,
"active_process": active_process,
"attached_nodes": attached_nodes,
"quota_posture": quota_posture,
"coordinator_response": coordinator_response,
}))
}
@ -1378,6 +1406,7 @@ fn human_report(value: &Value) -> String {
push_string_field(&mut lines, value, "tenant", "tenant");
push_string_field(&mut lines, value, "project", "project");
push_string_field(&mut lines, value, "process", "process");
push_string_field(&mut lines, value, "active_process", "active process");
push_string_field(&mut lines, value, "node", "node");
push_string_field(&mut lines, value, "artifact", "artifact");
push_string_field(&mut lines, value, "entry", "entry");
@ -1416,6 +1445,30 @@ fn human_report(value: &Value) -> String {
push_nested_string_field(&mut lines, metadata, "identity", "bundle");
}
}
if let Some(environments) = value
.get("discovered_environments")
.and_then(Value::as_array)
{
let names = environments
.iter()
.filter_map(Value::as_str)
.collect::<Vec<_>>();
if !names.is_empty() {
lines.push(format!("environments: {}", names.join(", ")));
}
}
if let Some(attached_nodes) = value.get("attached_nodes") {
if let Some(count) = attached_nodes.get("count").and_then(Value::as_u64) {
let online = attached_nodes
.get("online")
.and_then(Value::as_u64)
.unwrap_or(0);
lines.push(format!("attached nodes: {count} ({online} online)"));
}
}
if let Some(current_usage) = value.pointer("/quota_posture/current_usage") {
lines.push(format!("quota usage: {}", compact_json(current_usage)));
}
if let Some(flow) = value.get("human_flow") {
if let Some(device) = flow.get("Device") {
lines.push("flow: device".to_owned());
@ -1850,6 +1903,43 @@ fn write_project_config(project: &Path, config: &ProjectConfig) -> Result<()> {
.with_context(|| format!("failed to write {}", file.display()))
}
fn effective_project_scope(scope: &CliScopeArgs, config: Option<&ProjectConfig>) -> CliScopeArgs {
CliScopeArgs {
coordinator: scope
.coordinator
.clone()
.or_else(|| config.and_then(|config| config.coordinator.clone())),
tenant: effective_scope_value(
&scope.tenant,
config.map(|config| config.tenant.as_str()),
"tenant",
),
project: effective_scope_value(
&scope.project,
config.map(|config| config.project.as_str()),
"project",
),
user: effective_scope_value(
&scope.user,
config.map(|config| config.user.as_str()),
"user",
),
json: scope.json,
}
}
fn effective_scope_value(
cli_value: &str,
config_value: Option<&str>,
default_value: &str,
) -> String {
if cli_value == default_value {
config_value.unwrap_or(cli_value).to_owned()
} else {
cli_value.to_owned()
}
}
fn auth_state_value() -> Value {
match session_from_env() {
CliSession::Anonymous => json!({
@ -1953,6 +2043,120 @@ fn list_task_events_if_available(
})))
}
fn list_attached_nodes_if_available(
coordinator: Option<&str>,
scope: &CliScopeArgs,
) -> Result<Value> {
let Some(coordinator) = coordinator else {
return Ok(json!({
"checked": false,
"source": "no_coordinator",
"count": 0,
"online": 0,
"response": null,
}));
};
let mut session = JsonLineSession::connect(coordinator)?;
let response = session.request(json!({
"type": "list_node_descriptors",
"tenant": scope.tenant,
"project": scope.project,
"actor_user": scope.user,
}))?;
let descriptors = response
.get("descriptors")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
let online = descriptors
.iter()
.filter(|descriptor| {
descriptor
.get("online")
.and_then(Value::as_bool)
.unwrap_or(false)
})
.count();
Ok(json!({
"checked": true,
"source": "coordinator",
"coordinator": coordinator,
"count": descriptors.len(),
"online": online,
"response": response,
"coordinator_session_requests": session.requests(),
}))
}
fn discovered_environment_names(inspection: Option<&BundleInspection>) -> Vec<String> {
inspection
.map(|inspection| {
inspection
.metadata
.environments
.iter()
.map(|environment| environment.name.clone())
.collect()
})
.unwrap_or_default()
}
fn active_process_from_task_events(task_events: Option<&Value>) -> Option<String> {
let events = task_events?
.pointer("/response/events")
.and_then(Value::as_array)?;
let processes = events
.iter()
.filter_map(|event| event.get("process").and_then(Value::as_str))
.map(str::to_owned)
.collect::<BTreeSet<_>>();
match processes.len() {
0 => None,
1 => processes.into_iter().next(),
_ => Some(format!(
"multiple_observed:{}",
processes.into_iter().collect::<Vec<_>>().join(",")
)),
}
}
fn task_event_count(task_events: Option<&Value>) -> usize {
task_events
.and_then(|task_events| task_events.pointer("/response/events"))
.and_then(Value::as_array)
.map(Vec::len)
.unwrap_or(0)
}
fn project_quota_posture(attached_nodes: &Value, task_events: Option<&Value>) -> Value {
let attached_node_count = attached_nodes
.get("count")
.and_then(Value::as_u64)
.unwrap_or(0);
let online_node_count = attached_nodes
.get("online")
.and_then(Value::as_u64)
.unwrap_or(0);
json!({
"source": "cli_project_status_summary",
"current_usage": {
"attached_nodes": attached_node_count,
"online_nodes": online_node_count,
"observed_task_events": task_event_count(task_events),
},
"limits": {
"artifact_download_bytes": "checked before download link creation",
"rendezvous_attempts": "checked before rendezvous plan creation",
"hosted_wasm": "checked before hosted coordinator-side work starts",
},
"next_blocked_action": null,
"billing_required": false,
"private_abuse_heuristics_exposed": false,
})
}
fn login_plan(args: LoginArgs) -> LoginPlan {
login_plan_with_nonce(args, login_nonce())
}
@ -3525,6 +3729,97 @@ mod tests {
read_project_config(temp.path()).unwrap().unwrap().project,
"project-b"
);
let status = project_status_report(
ProjectStatusArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
)
.unwrap();
assert_eq!(status["command"], "project status");
assert_eq!(status["project_identity"]["project"], "project-b");
assert_eq!(status["project_identity"]["tenant"], "tenant");
assert_eq!(status["active_process"], "unknown_without_coordinator");
assert_eq!(status["attached_nodes"]["checked"], false);
assert_eq!(status["quota_posture"]["billing_required"], false);
}
#[test]
fn project_status_queries_public_coordinator_state() {
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 _ in 0..2 {
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();
if line.contains("\"type\":\"list_node_descriptors\"") {
assert!(line.contains("\"project\":\"project-live\""));
stream
.write_all(
br#"{"type":"node_descriptors","descriptors":[{"id":"node-a","online":true}],"actor":"user"}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
} else if line.contains("\"type\":\"list_task_events\"") {
assert!(line.contains("\"project\":\"project-live\""));
stream
.write_all(
br#"{"type":"task_events","events":[{"process":"vp-live","task":"task-a"}]}"#,
)
.unwrap();
stream.write_all(b"\n").unwrap();
} else {
panic!("unexpected coordinator request: {line}");
}
}
});
let temp = tempfile::tempdir().unwrap();
write_project_config(
temp.path(),
&ProjectConfig {
tenant: "tenant-live".to_owned(),
project: "project-live".to_owned(),
user: "user".to_owned(),
coordinator: Some(addr.clone()),
},
)
.unwrap();
let status = project_status_report(
ProjectStatusArgs {
scope: CliScopeArgs {
coordinator: None,
tenant: "tenant".to_owned(),
project: "project".to_owned(),
user: "user".to_owned(),
json: false,
},
},
temp.path().to_path_buf(),
)
.unwrap();
server.join().unwrap();
assert_eq!(status["coordinator"], addr);
assert_eq!(status["project_identity"]["project"], "project-live");
assert_eq!(status["attached_nodes"]["checked"], true);
assert_eq!(status["attached_nodes"]["count"], 1);
assert_eq!(status["attached_nodes"]["online"], 1);
assert_eq!(status["active_process"], "vp-live");
assert_eq!(
status["quota_posture"]["current_usage"]["observed_task_events"],
1
);
}
#[test]