Public dry run dryrun-7a9befca72b4

crates/disasmer-cli/src/main.rs

@@ -1095,10 +1095,20 @@ fn process_status_report(args: ProcessStatusArgs) -> Result<Value> {
         &args.scope,
         Some(args.process.clone()),
     )?;
+    let current_tasks = task_summaries(events.as_ref());
+    let current_task_count = current_tasks.as_array().map(Vec::len).unwrap_or(0);
+    let state = process_state_from_tasks(events.as_ref());
     Ok(json!({
         "command": "process status",
         "process": args.process,
-        "state": if events.is_some() { "inspectable" } else { "unknown_without_coordinator" },
+        "state": if events.is_some() { state } else { "unknown_without_coordinator" },
+        "started_entrypoint": "unknown_from_task_events",
+        "current_task_count": current_task_count,
+        "current_tasks": current_tasks,
+        "debug_state": {
+            "frozen": null,
+            "status": "unknown_from_cli_task_events"
+        },
         "events": events,
     }))
 }
@@ -1163,9 +1173,11 @@ fn task_list_report(args: TaskListArgs) -> Result<Value> {
         &args.scope,
         args.process.clone(),
     )?;
+    let tasks = task_summaries(events.as_ref());
     Ok(json!({
         "command": "task list",
         "process": args.process,
+        "tasks": tasks,
         "events": events,
     }))
 }
@@ -1176,10 +1188,12 @@ fn logs_report(args: LogsArgs) -> Result<Value> {
         &args.scope,
         args.process.clone(),
     )?;
+    let log_entries = log_entries(events.as_ref(), args.task.as_deref());
     Ok(json!({
         "command": "logs",
         "process": args.process,
         "task": args.task,
+        "log_entries": log_entries,
         "logs_are_capped": true,
         "secret_redaction_policy": "configured-redaction-boundary",
         "events": events,
@@ -1192,10 +1206,13 @@ fn artifact_list_report(args: ArtifactListArgs) -> Result<Value> {
         &args.scope,
         args.process.clone(),
     )?;
+    let artifacts = artifact_summaries(events.as_ref());
     Ok(json!({
         "command": "artifact list",
         "process": args.process,
         "source": "task_events",
+        "artifacts": artifacts,
+        "default_durable_store_assumed": false,
         "events": events,
     }))
 }
@@ -1314,7 +1331,6 @@ fn quota_status_report(args: QuotaStatusArgs, cwd: PathBuf) -> Result<Value> {
         "task_events": task_events,
         "next_blocked_action": quota_next_blocked_action(&current_usage),
         "community_tier_language": true,
-        "billing_required": false,
         "private_abuse_heuristics_exposed": false,
     }))
 }
@@ -1482,6 +1498,18 @@ fn human_report(value: &Value) -> String {
     if let Some(current_usage) = value.pointer("/quota_posture/current_usage") {
         lines.push(format!("quota usage: {}", compact_json(current_usage)));
     }
+    if let Some(current_task_count) = value.get("current_task_count").and_then(Value::as_u64) {
+        lines.push(format!("tasks: {current_task_count}"));
+    }
+    if let Some(tasks) = value.get("tasks").and_then(Value::as_array) {
+        lines.push(format!("tasks: {}", tasks.len()));
+    }
+    if let Some(log_entries) = value.get("log_entries").and_then(Value::as_array) {
+        lines.push(format!("log entries: {}", log_entries.len()));
+    }
+    if let Some(artifacts) = value.get("artifacts").and_then(Value::as_array) {
+        lines.push(format!("artifacts: {}", artifacts.len()));
+    }
     if let Some(current_usage) = value.get("current_usage") {
         lines.push(format!("quota usage: {}", compact_json(current_usage)));
     }
@@ -2149,6 +2177,158 @@ fn active_process_from_task_events(task_events: Option<&Value>) -> Option<String
     }
 }
 
+fn task_event_values(task_events: Option<&Value>) -> Vec<&Value> {
+    task_events
+        .and_then(|task_events| task_events.pointer("/response/events"))
+        .and_then(Value::as_array)
+        .map(|events| events.iter().collect())
+        .unwrap_or_default()
+}
+
+fn event_string(event: &Value, field: &str) -> Option<String> {
+    event.get(field).and_then(Value::as_str).map(str::to_owned)
+}
+
+fn event_u64(event: &Value, field: &str) -> Option<u64> {
+    event.get(field).and_then(Value::as_u64)
+}
+
+fn task_summaries(task_events: Option<&Value>) -> Value {
+    Value::Array(
+        task_event_values(task_events)
+            .into_iter()
+            .map(|event| {
+                let task = event_string(event, "task").unwrap_or_else(|| "unknown".to_owned());
+                let terminal_state =
+                    event_string(event, "terminal_state").unwrap_or_else(|| "unknown".to_owned());
+                let node = event_string(event, "node");
+                json!({
+                    "process": event_string(event, "process"),
+                    "task": task,
+                    "state": terminal_state,
+                    "environment": event.get("environment").cloned().unwrap_or_else(|| json!("unknown_from_task_event")),
+                    "environment_digest": event.get("environment_digest").cloned().unwrap_or(Value::Null),
+                    "node_placement": {
+                        "node": node,
+                        "source": "coordinator_task_event",
+                    },
+                    "failure_reason": task_failure_reason(event),
+                    "stdout_bytes": event_u64(event, "stdout_bytes").unwrap_or(0),
+                    "stderr_bytes": event_u64(event, "stderr_bytes").unwrap_or(0),
+                })
+            })
+            .collect(),
+    )
+}
+
+fn task_failure_reason(event: &Value) -> Value {
+    match event.get("terminal_state").and_then(Value::as_str) {
+        Some("failed") => {
+            if let Some(stderr) = event.get("stderr_tail").and_then(Value::as_str) {
+                if !stderr.is_empty() {
+                    return json!(stderr);
+                }
+            }
+            if let Some(status_code) = event.get("status_code").and_then(Value::as_i64) {
+                return json!(format!("task exited with status {status_code}"));
+            }
+            json!("task failed")
+        }
+        Some("cancelled") => json!("task cancelled"),
+        _ => Value::Null,
+    }
+}
+
+fn process_state_from_tasks(task_events: Option<&Value>) -> &'static str {
+    let events = task_event_values(task_events);
+    if events.is_empty() {
+        return "no_tasks_observed";
+    }
+    if events.iter().any(|event| {
+        event
+            .get("terminal_state")
+            .and_then(Value::as_str)
+            .is_some_and(|state| state == "failed")
+    }) {
+        return "has_failed_tasks";
+    }
+    if events.iter().any(|event| {
+        event
+            .get("terminal_state")
+            .and_then(Value::as_str)
+            .is_some_and(|state| state == "cancelled")
+    }) {
+        return "has_cancelled_tasks";
+    }
+    if events.iter().all(|event| {
+        event
+            .get("terminal_state")
+            .and_then(Value::as_str)
+            .is_some_and(|state| state == "completed")
+    }) {
+        return "completed_tasks_observed";
+    }
+    "tasks_observed"
+}
+
+fn log_entries(task_events: Option<&Value>, task_filter: Option<&str>) -> Value {
+    Value::Array(
+        task_event_values(task_events)
+            .into_iter()
+            .filter(|event| {
+                task_filter.map_or(true, |task_filter| {
+                    event
+                        .get("task")
+                        .and_then(Value::as_str)
+                        .is_some_and(|task| task == task_filter)
+                })
+            })
+            .map(|event| {
+                json!({
+                    "process": event_string(event, "process"),
+                    "task": event_string(event, "task"),
+                    "node": event_string(event, "node"),
+                    "stdout_bytes": event_u64(event, "stdout_bytes").unwrap_or(0),
+                    "stderr_bytes": event_u64(event, "stderr_bytes").unwrap_or(0),
+                    "stdout_tail": event.get("stdout_tail").cloned().unwrap_or_else(|| json!("")),
+                    "stderr_tail": event.get("stderr_tail").cloned().unwrap_or_else(|| json!("")),
+                    "stdout_truncated": event.get("stdout_truncated").and_then(Value::as_bool).unwrap_or(false),
+                    "stderr_truncated": event.get("stderr_truncated").and_then(Value::as_bool).unwrap_or(false),
+                    "capped": true,
+                })
+            })
+            .collect(),
+    )
+}
+
+fn artifact_summaries(task_events: Option<&Value>) -> Value {
+    Value::Array(
+        task_event_values(task_events)
+            .into_iter()
+            .filter_map(|event| {
+                let path = event.get("artifact_path").and_then(Value::as_str)?;
+                let node = event_string(event, "node");
+                Some(json!({
+                    "artifact": artifact_name_from_path(path),
+                    "path": path,
+                    "producer_task": event_string(event, "task"),
+                    "producer_node": node,
+                    "process": event_string(event, "process"),
+                    "digest": event.get("artifact_digest").cloned().unwrap_or(Value::Null),
+                    "size_bytes": event.get("artifact_size_bytes").cloned().unwrap_or(Value::Null),
+                    "state": if event.get("artifact_digest").is_some() { "metadata_flushed" } else { "metadata_without_digest" },
+                    "known_locations": node.into_iter().collect::<Vec<_>>(),
+                    "durable_storage": false,
+                }))
+            })
+            .collect(),
+    )
+}
+
+fn artifact_name_from_path(path: &str) -> String {
+    path.rsplit('/').next().unwrap_or(path).to_owned()
+}
+
 fn task_event_count(task_events: Option<&Value>) -> usize {
     task_events
         .and_then(|task_events| task_events.pointer("/response/events"))
@@ -2165,7 +2345,6 @@ fn project_quota_posture(attached_nodes: &Value, task_events: Option<&Value>) ->
         "current_usage": current_usage,
         "limits": quota_limits_value(),
         "next_blocked_action": next_blocked_action,
-        "billing_required": false,
         "private_abuse_heuristics_exposed": false,
     })
 }
@@ -3816,7 +3995,6 @@ mod tests {
         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]
@@ -3924,7 +4102,6 @@ mod tests {
         assert_eq!(status["current_usage"]["attached_nodes"], 0);
         assert_eq!(status["limits"]["artifact_download_bytes"], 268_435_456);
         assert_eq!(status["community_tier_language"], true);
-        assert_eq!(status["billing_required"], false);
         assert_eq!(
             status["next_blocked_action"]["action"],
             "node_work_requires_online_attached_node"
@@ -3998,6 +4175,76 @@ mod tests {
         assert_eq!(status["task_events"]["response"]["type"], "task_events");
     }
 
+    #[test]
+    fn process_task_log_and_artifact_reports_summarize_task_events() {
+        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 response = concat!(
+                r#"{"type":"task_events","events":["#,
+                r#"{"tenant":"tenant","project":"project","process":"vp","node":"node-a","task":"task-a","terminal_state":"completed","status_code":0,"stdout_bytes":12,"stderr_bytes":0,"stdout_tail":"ok","stderr_tail":"","stdout_truncated":false,"stderr_truncated":false,"artifact_path":"/vfs/artifacts/app.txt","artifact_digest":"sha256:artifact","artifact_size_bytes":12},"#,
+                r#"{"tenant":"tenant","project":"project","process":"vp","node":"node-b","task":"task-b","terminal_state":"failed","status_code":1,"stdout_bytes":0,"stderr_bytes":7,"stdout_tail":"","stderr_tail":"boom","stdout_truncated":false,"stderr_truncated":false,"artifact_path":null,"artifact_digest":null,"artifact_size_bytes":null}"#,
+                r#"]}"#
+            );
+            for _ in 0..4 {
+                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("\"type\":\"list_task_events\""));
+                assert!(line.contains("\"process\":\"vp\""));
+                stream.write_all(response.as_bytes()).unwrap();
+                stream.write_all(b"\n").unwrap();
+            }
+        });
+        let scope = CliScopeArgs {
+            coordinator: Some(addr),
+            tenant: "tenant".to_owned(),
+            project: "project".to_owned(),
+            user: "user".to_owned(),
+            json: false,
+        };
+
+        let process = process_status_report(ProcessStatusArgs {
+            scope: scope.clone(),
+            process: "vp".to_owned(),
+        })
+        .unwrap();
+        let tasks = task_list_report(TaskListArgs {
+            scope: scope.clone(),
+            process: Some("vp".to_owned()),
+        })
+        .unwrap();
+        let logs = logs_report(LogsArgs {
+            scope: scope.clone(),
+            process: Some("vp".to_owned()),
+            task: Some("task-a".to_owned()),
+        })
+        .unwrap();
+        let artifacts = artifact_list_report(ArtifactListArgs {
+            scope,
+            process: Some("vp".to_owned()),
+        })
+        .unwrap();
+        server.join().unwrap();
+
+        assert_eq!(process["state"], "has_failed_tasks");
+        assert_eq!(process["current_task_count"], 2);
+        assert_eq!(
+            process["current_tasks"][0]["node_placement"]["node"],
+            "node-a"
+        );
+        assert_eq!(tasks["tasks"][1]["failure_reason"], "boom");
+        assert_eq!(logs["log_entries"].as_array().unwrap().len(), 1);
+        assert_eq!(logs["log_entries"][0]["task"], "task-a");
+        assert_eq!(logs["log_entries"][0]["stdout_tail"], "ok");
+        assert_eq!(artifacts["artifacts"].as_array().unwrap().len(), 1);
+        assert_eq!(artifacts["artifacts"][0]["artifact"], "app.txt");
+        assert_eq!(artifacts["artifacts"][0]["size_bytes"], 12);
+        assert_eq!(artifacts["artifacts"][0]["known_locations"][0], "node-a");
+        assert_eq!(artifacts["default_durable_store_assumed"], false);
+    }
+
     #[test]
     fn build_command_reuses_bundle_inspection_without_full_repo_upload() {
         let temp = tempfile::tempdir().unwrap();