Sync public tree to c125f70

crates/disasmer-cli/src/main.rs

@@ -5726,11 +5726,10 @@ fn coordinator_run_report(plan: RunPlan) -> Result<Value> {
     let artifact_path = format!("/vfs/artifacts/{task}-output.txt");
     let launch_task_response = if status == "started" {
         let manifest_path = plan.project.join("Cargo.toml");
-        session.request_allow_error(json!({
+        let mut launch_task_request = json!({
             "type": "launch_task",
             "tenant": tenant.clone(),
             "project": project.clone(),
-            "actor_user": user.clone(),
             "process": process.clone(),
             "task": task.clone(),
             "environment": null,
@@ -5749,7 +5748,9 @@ fn coordinator_run_report(plan: RunPlan) -> Result<Value> {
                 manifest_path.to_string_lossy()
             ],
             "artifact_path": artifact_path.clone(),
-        }))?
+        });
+        add_workflow_actor_fields(&mut launch_task_request, &plan.session, &user);
+        session.request_allow_error(launch_task_request)?
     } else {
         Value::Null
     };
@@ -6813,6 +6814,25 @@ mod tests {
                 )
                 .unwrap();
             stream.write_all(b"\n").unwrap();
+            let mut launch_line = String::new();
+            reader.read_line(&mut launch_line).unwrap();
+            assert!(launch_line.contains(r#""type":"launch_task""#));
+            assert!(launch_line.contains(r#""tenant":"tenant-live""#));
+            assert!(launch_line.contains(r#""project":"project-live""#));
+            assert!(launch_line.contains(r#""process":"vp-current""#));
+            assert!(launch_line.contains(r#""task":"build""#));
+            assert!(launch_line.contains(r#""required_capabilities":["Command"]"#));
+            assert!(launch_line.contains(r#""actor_agent":"agent-ci""#));
+            assert!(launch_line.contains(&format!(
+                r#""agent_public_key_fingerprint":"{expected_fingerprint}""#
+            )));
+            assert!(!launch_line.contains(r#""actor_user""#));
+            stream
+                .write_all(
+                    br#"{"type":"task_launched","process":"vp-current","task":"build","placement":{"node":"agent-worker","capabilities":["Command"]}}"#,
+                )
+                .unwrap();
+            stream.write_all(b"\n").unwrap();
         });
 
         let report = run_report(
@@ -6834,7 +6854,7 @@ mod tests {
         .unwrap();
         server.join().unwrap();
 
-        assert_eq!(report["status"], "started");
+        assert_eq!(report["status"], "task_launched");
         assert_eq!(report["workflow_actor"]["kind"], "agent");
         assert_eq!(report["workflow_actor"]["agent"], "agent-ci");
         assert_eq!(
@@ -6842,6 +6862,8 @@ mod tests {
             true
         );
         assert_eq!(report["run_start"]["actor"]["kind"], "agent");
+        assert_eq!(report["task_launch"]["type"], "task_launched");
+        assert_eq!(report["task_launch"]["placement"]["node"], "agent-worker");
     }
 
     #[test]
@@ -6875,10 +6897,13 @@ mod tests {
         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 [
+            for (index, 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"}"#,
-            ] {
+            ]
+            .into_iter()
+            .enumerate()
+            {
                 let (mut stream, _) = listener.accept().unwrap();
                 let mut reader = BufReader::new(stream.try_clone().unwrap());
                 let mut line = String::new();
@@ -6890,6 +6915,23 @@ mod tests {
                 assert!(line.contains(r#""restart":false"#));
                 stream.write_all(response.as_bytes()).unwrap();
                 stream.write_all(b"\n").unwrap();
+                if index == 0 {
+                    let mut launch_line = String::new();
+                    reader.read_line(&mut launch_line).unwrap();
+                    assert!(launch_line.contains(r#""type":"launch_task""#));
+                    assert!(launch_line.contains(r#""tenant":"tenant-live""#));
+                    assert!(launch_line.contains(r#""project":"project-live""#));
+                    assert!(launch_line.contains(r#""process":"vp-current""#));
+                    assert!(launch_line.contains(r#""task":"build""#));
+                    assert!(launch_line.contains(r#""actor_user":"user-live""#));
+                    assert!(launch_line.contains(r#""required_capabilities":["Command"]"#));
+                    stream
+                        .write_all(
+                            br#"{"type":"task_launched","process":"vp-current","task":"build","placement":{"node":"worker-live","capabilities":["Command"]}}"#,
+                        )
+                        .unwrap();
+                    stream.write_all(b"\n").unwrap();
+                }
             }
         });
 
@@ -6921,13 +6963,15 @@ mod tests {
         .unwrap();
         server.join().unwrap();
 
-        assert_eq!(started["status"], "started");
+        assert_eq!(started["status"], "task_launched");
         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!(started["task_launch"]["type"], "task_launched");
+        assert_eq!(started["task_launch"]["placement"]["node"], "worker-live");
 
         assert_eq!(blocked["status"], "blocked_active_process");
         assert_eq!(blocked["entry"], "test");