Public dry run dryrun-309831e1e021

Source commit: 309831e1e021f962c118452336776fd9a94025f9

Public tree identity: sha256:6fa95c1745579bd6256dbeb3d476db0b07c2f24aa9213b0f7783ce1adfc8aca5

crates/disasmer-node/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "disasmer-node"
+version = "0.1.0"
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[dependencies]
+disasmer-core = { path = "../disasmer-core" }
+quinn.workspace = true
+rcgen.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+thiserror.workspace = true
+tokio.workspace = true
+wasmtime.workspace = true

crates/disasmer-node/src/bin/disasmer-podman-smoke.rs

@@ -0,0 +1,84 @@
+use std::fs;
+use std::path::PathBuf;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use disasmer_core::{
+    CommandInvocation, Digest, EnvironmentKind, EnvironmentRequirements, EnvironmentResource,
+    NodeId, ProcessId, TaskId, VfsOverlay, VfsPath,
+};
+use disasmer_node::{LinuxRootlessPodmanBackend, LocalSourceCheckout, StdProcessRunner};
+use serde_json::json;
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let workspace = create_workspace()?;
+    let env_dir = workspace.join("envs/linux");
+    fs::create_dir_all(&env_dir)?;
+    fs::write(
+        env_dir.join("Containerfile"),
+        "FROM docker.io/library/alpine:3.20\nWORKDIR /workspace\n",
+    )?;
+    fs::write(workspace.join("input.txt"), "node-local source\n")?;
+
+    let env = EnvironmentResource {
+        name: "linux".to_owned(),
+        kind: EnvironmentKind::Containerfile,
+        recipe_path: env_dir.join("Containerfile"),
+        context_path: env_dir,
+        digest: Digest::sha256("phase2-podman-smoke-linux-env"),
+        requirements: EnvironmentRequirements::linux_container(),
+    };
+    let invocation = CommandInvocation {
+        program: "sh".to_owned(),
+        args: vec![
+            "-c".to_owned(),
+            "printf 'podman-ok:' && cat input.txt".to_owned(),
+        ],
+        env: Some(env),
+    };
+    let checkout = LocalSourceCheckout {
+        host_path: workspace.clone(),
+        snapshot: Digest::sha256("phase2-podman-smoke-checkout"),
+    };
+    let task = TaskId::from("podman-smoke");
+    let mut overlay = VfsOverlay::new(task.clone(), NodeId::from("node-podman-smoke"));
+    let mut runner = StdProcessRunner;
+    let output = LinuxRootlessPodmanBackend.execute_local_checkout_task(
+        ProcessId::from("vp-podman-smoke"),
+        task,
+        &invocation,
+        checkout,
+        Some(VfsPath::new("/vfs/artifacts/podman-smoke.txt")?),
+        &mut runner,
+        &mut overlay,
+    )?;
+    let manifest = overlay.flush();
+
+    println!(
+        "{}",
+        serde_json::to_string(&json!({
+            "podman_status": "completed",
+            "status_code": output.status_code,
+            "stdout": output.stdout,
+            "stderr": output.stderr,
+            "staged_artifact": output.staged_artifact,
+            "large_bytes_uploaded": manifest.large_bytes_uploaded,
+            "uses_full_repo_tarball": false,
+            "coordinator_routed_file_reads": false,
+        }))?
+    );
+    let _ = fs::remove_dir_all(workspace);
+    Ok(())
+}
+
+fn create_workspace() -> Result<PathBuf, std::io::Error> {
+    let nanos = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|duration| duration.as_nanos())
+        .unwrap_or_default();
+    let workspace = std::env::temp_dir().join(format!(
+        "disasmer-podman-smoke-{}-{nanos}",
+        std::process::id()
+    ));
+    fs::create_dir_all(&workspace)?;
+    Ok(workspace)
+}

crates/disasmer-node/src/bin/disasmer-quic-smoke.rs

@@ -0,0 +1,138 @@
+use std::{net::SocketAddr, sync::Arc};
+
+use disasmer_core::{
+    ArtifactId, DataPlaneObject, DataPlaneScope, Digest, NativeQuicTransport, NodeEndpoint, NodeId,
+    ProcessId, ProjectId, RendezvousRequest, TenantId, Transport,
+};
+use quinn::rustls::{
+    pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer},
+    RootCertStore,
+};
+use quinn::{ClientConfig, Endpoint, ServerConfig};
+use serde::{Deserialize, Serialize};
+use serde_json::json;
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+struct QuicTransferRequest {
+    scope: DataPlaneScope,
+    authorization_digest: Digest,
+    requested_bytes: u64,
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let payload = b"artifact-bytes-over-rust-native-quic".to_vec();
+    let (cert, key) = self_signed_localhost_cert()?;
+    let server_endpoint = Endpoint::server(
+        ServerConfig::with_single_cert(vec![cert.clone()], key)?,
+        "127.0.0.1:0".parse()?,
+    )?;
+    let server_addr = server_endpoint.local_addr()?;
+
+    let scope = DataPlaneScope {
+        tenant: TenantId::from("tenant"),
+        project: ProjectId::from("project"),
+        process: ProcessId::from("vp-quic"),
+        object: DataPlaneObject::Artifact(ArtifactId::from("quic-artifact")),
+        authorization_subject: "node-a-to-node-b".to_owned(),
+    };
+    let transport = NativeQuicTransport;
+    let plan = transport.plan_authenticated_direct_bulk_transfer(
+        RendezvousRequest {
+            scope: scope.clone(),
+            source: endpoint("node-a", server_addr),
+            destination: endpoint("node-b", "127.0.0.1:0".parse()?),
+        },
+        true,
+        "",
+    )?;
+
+    let expected_scope = plan.scope.clone();
+    let expected_digest = plan.authorization_digest.clone();
+    let expected_payload = payload.clone();
+    let server = tokio::spawn(async move {
+        let incoming = server_endpoint
+            .accept()
+            .await
+            .ok_or("server endpoint closed before accepting a QUIC connection")?;
+        let connection = incoming.await?;
+        let (mut send, mut recv) = connection.accept_bi().await?;
+        let request_bytes = recv.read_to_end(64 * 1024).await?;
+        let request: QuicTransferRequest = serde_json::from_slice(&request_bytes)?;
+        if request.scope != expected_scope {
+            return Err("QUIC request scope did not match the authorized data-plane scope".into());
+        }
+        if request.authorization_digest != expected_digest {
+            return Err(
+                "QUIC request authorization digest did not match the rendezvous plan".into(),
+            );
+        }
+        send.write_all(&expected_payload).await?;
+        send.finish()?;
+        server_endpoint.wait_idle().await;
+        Ok::<usize, Box<dyn std::error::Error + Send + Sync>>(request_bytes.len())
+    });
+
+    let mut roots = RootCertStore::empty();
+    roots.add(cert)?;
+    let client_config = ClientConfig::with_root_certificates(Arc::new(roots))?;
+    let mut client_endpoint = Endpoint::client("127.0.0.1:0".parse()?)?;
+    client_endpoint.set_default_client_config(client_config);
+
+    let connection = client_endpoint.connect(server_addr, "localhost")?.await?;
+    let (mut send, mut recv) = connection.open_bi().await?;
+    let request = QuicTransferRequest {
+        scope: plan.scope.clone(),
+        authorization_digest: plan.authorization_digest.clone(),
+        requested_bytes: payload.len() as u64,
+    };
+    let request_bytes = serde_json::to_vec(&request)?;
+    send.write_all(&request_bytes).await?;
+    send.finish()?;
+    let received = recv.read_to_end(64 * 1024).await?;
+    connection.close(0u32.into(), b"done");
+    client_endpoint.wait_idle().await;
+    let server_received_request_bytes = server.await??;
+
+    if received != payload {
+        return Err("QUIC artifact payload did not round trip".into());
+    }
+
+    println!(
+        "{}",
+        json!({
+            "kind": "disasmer_quic_smoke",
+            "transport": format!("{:?}", transport.kind()),
+            "rust_native_quic": true,
+            "authenticated_direct_connection": transport.authenticated_direct_connections(),
+            "coordinator_assisted_rendezvous": plan.coordinator_assisted_rendezvous,
+            "coordinator_bulk_relay_allowed": plan.coordinator_bulk_relay_allowed,
+            "source_node": plan.source.node,
+            "destination_node": plan.destination.node,
+            "scope": plan.scope,
+            "request_bytes": request_bytes.len(),
+            "server_received_request_bytes": server_received_request_bytes,
+            "payload_bytes": received.len(),
+            "authorization_digest": plan.authorization_digest,
+        })
+    );
+
+    Ok(())
+}
+
+fn self_signed_localhost_cert() -> Result<
+    (CertificateDer<'static>, PrivateKeyDer<'static>),
+    Box<dyn std::error::Error + Send + Sync>,
+> {
+    let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_owned()])?;
+    let key = PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(cert.signing_key.serialize_der()));
+    Ok((cert.cert.into(), key))
+}
+
+fn endpoint(name: &str, addr: SocketAddr) -> NodeEndpoint {
+    NodeEndpoint {
+        node: NodeId::from(name),
+        advertised_addr: addr.to_string(),
+        public_key_fingerprint: Digest::sha256(format!("{name}-public-key")),
+    }
+}

crates/disasmer-node/src/bin/disasmer-wasmtime-smoke.rs

@@ -0,0 +1,133 @@
+use std::fs;
+use std::path::PathBuf;
+
+use disasmer_core::{CommandInvocation, NodeId, TaskId, VfsPath};
+use disasmer_node::{LocalCommandExecutor, VirtualThreadCommand, WasmtimeTaskRuntime};
+use serde_json::json;
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let args: Vec<String> = std::env::args().collect();
+    if args.len() == 2 && args[1] == "--host-command" {
+        return run_host_command_smoke();
+    }
+    if args.len() == 6 && args[1] == "--debug-freeze-resume" {
+        return run_debug_freeze_resume_smoke(&args[2], &args[3], &args[4], &args[5]);
+    }
+    if args.len() != 5 {
+        return Err(
+            "usage: disasmer-wasmtime-smoke <module.wasm> <export> <i32-arg> <expected-i32> | --host-command | --debug-freeze-resume <module.wasm> <export> <i32-arg> <expected-i32>".into(),
+        );
+    }
+
+    let module = PathBuf::from(&args[1]);
+    let export = &args[2];
+    let arg: i32 = args[3].parse()?;
+    let expected: i32 = args[4].parse()?;
+
+    let wasm = fs::read(&module)?;
+    let runtime = WasmtimeTaskRuntime::new()?;
+    let result = runtime.run_i32_export(&wasm, export, arg)?;
+    if result != expected {
+        return Err(format!("expected {expected}, got {result} from export `{export}`").into());
+    }
+
+    println!(
+        "{}",
+        serde_json::to_string(&json!({
+            "type": "wasmtime_task_smoke",
+            "module": module,
+            "export": export,
+            "arg": arg,
+            "result": result,
+        }))?
+    );
+    Ok(())
+}
+
+fn run_debug_freeze_resume_smoke(
+    module: &str,
+    export: &str,
+    arg: &str,
+    expected: &str,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let module = PathBuf::from(module);
+    let arg: i32 = arg.parse()?;
+    let expected: i32 = expected.parse()?;
+    let wasm = fs::read(&module)?;
+    let runtime = WasmtimeTaskRuntime::new()?;
+    let probe = runtime.freeze_resume_i32_export_probe(&wasm, export, arg)?;
+    if probe.result != expected {
+        return Err(format!(
+            "expected {expected}, got {} from export `{export}`",
+            probe.result
+        )
+        .into());
+    }
+
+    println!(
+        "{}",
+        serde_json::to_string(&json!({
+            "type": "wasmtime_debug_freeze_resume_smoke",
+            "module": module,
+            "export": export,
+            "task": probe.task,
+            "frozen_state": probe.frozen_state,
+            "resumed_state": probe.resumed_state,
+            "stack_frames": probe.stack_frames,
+            "local_values": probe.local_values,
+            "wasm_function": probe.wasm_function,
+            "wasm_pc": probe.wasm_pc,
+            "arg": arg,
+            "result": probe.result,
+            "node_runtime_reached_wasm_task": true,
+            "node_runtime_captured_wasm_locals": true,
+        }))?
+    );
+    Ok(())
+}
+
+fn run_host_command_smoke() -> Result<(), Box<dyn std::error::Error>> {
+    let runtime = WasmtimeTaskRuntime::new()?;
+    let result = runtime.run_i32_export_with_command_import(
+        r#"
+        (module
+          (import "disasmer" "cmd_run" (func $cmd_run (result i32)))
+          (func (export "compile-linux") (result i32)
+            call $cmd_run))
+        "#,
+        "compile-linux",
+        LocalCommandExecutor {
+            node: NodeId::from("node-wasmtime"),
+            hosted_control_plane: false,
+            has_command_capability: true,
+        },
+        VirtualThreadCommand {
+            virtual_thread: TaskId::from("compile-linux"),
+            invocation: CommandInvocation {
+                program: "sh".to_owned(),
+                args: vec!["-c".to_owned(), "printf linux-build-artifact".to_owned()],
+                env: None,
+            },
+            stage_stdout_as: Some(VfsPath::new("/vfs/artifacts/linux/app.tar.zst")?),
+        },
+    )?;
+
+    println!(
+        "{}",
+        serde_json::to_string(&json!({
+            "type": "wasmtime_host_command_smoke",
+            "export": "compile-linux",
+            "export_result": result.export_result,
+            "virtual_thread": result.command_output.virtual_thread,
+            "stdout": result.command_output.stdout,
+            "staged_artifact": result.command_output.staged_artifact,
+            "large_bytes_uploaded": result.manifest.large_bytes_uploaded,
+            "manifest_objects": result.manifest.objects.len(),
+            "node_host_import": "disasmer.cmd_run",
+            "flagship_linux_build_task": true,
+            "node_executed_host_command": true,
+            "hosted_control_plane_ran_command": false,
+        }))?
+    );
+    Ok(())
+}

crates/disasmer-node/src/main.rs

@@ -0,0 +1,637 @@
+use std::io::{BufRead, BufReader, Write};
+use std::net::TcpStream;
+use std::path::PathBuf;
+use std::time::{Duration, Instant};
+
+use disasmer_core::{
+    Capability, CommandInvocation, NodeCapabilities, NodeId, TaskId, VfsOverlay, VfsPath,
+};
+use disasmer_node::{CommandOutput, LocalCommandExecutor, VirtualThreadCommand};
+use serde_json::{json, Value};
+
+#[derive(Debug)]
+struct Args {
+    coordinator: String,
+    tenant: String,
+    project: String,
+    node: String,
+    process: String,
+    task: String,
+    command: String,
+    command_args: Vec<String>,
+    artifact: String,
+    enrollment_grant: Option<String>,
+    public_key: Option<String>,
+    control_poll_ms: u64,
+    assignment_poll_ms: u64,
+    emit_ready: bool,
+    worker: bool,
+}
+
+#[derive(Clone, Debug)]
+struct RuntimeTask {
+    process: String,
+    task: String,
+    command: String,
+    command_args: Vec<String>,
+    artifact: String,
+    epoch: Option<u64>,
+    task_assignment_response: Value,
+}
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let args = parse_args()?;
+
+    let mut session = CoordinatorSession::connect(&args.coordinator)?;
+
+    let registration = register_node(&mut session, &args)?;
+    let heartbeat = session.request(json!({
+        "type": "node_heartbeat",
+        "node": &args.node,
+    }))?;
+    let capability_report = session.request(json!({
+        "type": "report_node_capabilities",
+        "tenant": &args.tenant,
+        "project": &args.project,
+        "node": &args.node,
+        "capabilities": NodeCapabilities::detect_current(),
+        "cached_environment_digests": [],
+        "dependency_cache_digests": [],
+        "source_snapshots": [],
+        "artifact_locations": [],
+        "direct_connectivity": true,
+        "online": true,
+    }))?;
+
+    if args.worker {
+        return worker_loop(
+            &args,
+            &mut session,
+            registration,
+            heartbeat,
+            capability_report,
+        );
+    }
+
+    let task_assignment = session.request(json!({
+        "type": "schedule_task",
+        "tenant": &args.tenant,
+        "project": &args.project,
+        "environment": null,
+        "environment_digest": null,
+        "required_capabilities": [Capability::Command],
+        "dependency_cache": null,
+        "source_snapshot": null,
+        "required_artifacts": [],
+        "quota_available": true,
+        "policy_allowed": true,
+        "prefer_node": &args.node,
+    }))?;
+    let runtime_task = RuntimeTask {
+        process: args.process.clone(),
+        task: args.task.clone(),
+        command: args.command.clone(),
+        command_args: args.command_args.clone(),
+        artifact: args.artifact.clone(),
+        epoch: None,
+        task_assignment_response: task_assignment,
+    };
+    let report = run_runtime_task(
+        &args,
+        &mut session,
+        runtime_task,
+        registration,
+        heartbeat,
+        capability_report,
+    )?;
+    println!("{}", serde_json::to_string(&report)?);
+    Ok(())
+}
+
+fn worker_loop(
+    args: &Args,
+    session: &mut CoordinatorSession,
+    registration: Value,
+    heartbeat: Value,
+    capability_report: Value,
+) -> Result<(), Box<dyn std::error::Error>> {
+    if args.emit_ready {
+        println!(
+            "{}",
+            serde_json::to_string(&json!({
+                "node_status": "ready",
+                "mode": "worker",
+                "node": &args.node,
+            }))?
+        );
+        std::io::stdout().flush()?;
+    }
+
+    loop {
+        let response = session.request(json!({
+            "type": "poll_task_assignment",
+            "tenant": &args.tenant,
+            "project": &args.project,
+            "node": &args.node,
+        }))?;
+        let Some(assignment) = response.get("assignment").filter(|value| !value.is_null()) else {
+            std::thread::sleep(Duration::from_millis(args.assignment_poll_ms));
+            continue;
+        };
+        let runtime_task = runtime_task_from_assignment(assignment)?;
+        let report = run_runtime_task(
+            args,
+            session,
+            runtime_task,
+            registration.clone(),
+            heartbeat.clone(),
+            capability_report.clone(),
+        )?;
+        println!("{}", serde_json::to_string(&report)?);
+        std::io::stdout().flush()?;
+    }
+}
+
+fn runtime_task_from_assignment(value: &Value) -> Result<RuntimeTask, Box<dyn std::error::Error>> {
+    Ok(RuntimeTask {
+        process: required_string(value, "process")?,
+        task: required_string(value, "task")?,
+        command: required_string(value, "command")?,
+        command_args: value
+            .get("command_args")
+            .and_then(Value::as_array)
+            .map(|items| {
+                items
+                    .iter()
+                    .filter_map(Value::as_str)
+                    .map(str::to_owned)
+                    .collect::<Vec<_>>()
+            })
+            .unwrap_or_default(),
+        artifact: required_string(value, "artifact_path")?,
+        epoch: value.get("epoch").and_then(Value::as_u64),
+        task_assignment_response: value.clone(),
+    })
+}
+
+fn required_string(value: &Value, field: &str) -> Result<String, Box<dyn std::error::Error>> {
+    value
+        .get(field)
+        .and_then(Value::as_str)
+        .map(str::to_owned)
+        .ok_or_else(|| format!("task assignment missing string field `{field}`").into())
+}
+
+fn run_runtime_task(
+    args: &Args,
+    session: &mut CoordinatorSession,
+    task: RuntimeTask,
+    registration: Value,
+    heartbeat: Value,
+    capability_report: Value,
+) -> Result<Value, Box<dyn std::error::Error>> {
+    let epoch = match task.epoch {
+        Some(epoch) => epoch,
+        None => {
+            let started = session.request(json!({
+                "type": "start_process",
+                "tenant": &args.tenant,
+                "project": &args.project,
+                "process": &task.process,
+            }))?;
+            started
+                .get("epoch")
+                .and_then(Value::as_u64)
+                .ok_or("coordinator start_process response missing epoch")?
+        }
+    };
+    session.request(json!({
+        "type": "reconnect_node",
+        "node": &args.node,
+        "process": &task.process,
+        "epoch": epoch,
+    }))?;
+    let debug_command = session.request(json!({
+        "type": "poll_debug_command",
+        "tenant": &args.tenant,
+        "project": &args.project,
+        "process": &task.process,
+        "node": &args.node,
+        "task": &task.task,
+    }))?;
+    if args.emit_ready && !args.worker {
+        println!(
+            "{}",
+            serde_json::to_string(&json!({
+                "node_status": "ready",
+                "node": &args.node,
+                "process": &task.process,
+                "task": &task.task,
+            }))?
+        );
+        std::io::stdout().flush()?;
+    }
+    if args.control_poll_ms > 0 && wait_for_cancellation(session, args, &task)? {
+        let recorded = session.request(json!({
+            "type": "task_completed",
+            "tenant": &args.tenant,
+            "project": &args.project,
+            "process": &task.process,
+            "node": &args.node,
+            "task": &task.task,
+            "terminal_state": "cancelled",
+            "status_code": null,
+            "stdout_bytes": 0,
+            "stderr_bytes": 0,
+            "artifact_path": null,
+            "artifact_digest": null,
+            "artifact_size_bytes": null,
+        }))?;
+        return Ok(cancelled_node_report(
+            args,
+            &task,
+            registration,
+            heartbeat,
+            capability_report,
+            task.task_assignment_response.clone(),
+            debug_command,
+            recorded,
+            session.requests(),
+        ));
+    }
+
+    let executor = LocalCommandExecutor {
+        node: NodeId::new(args.node.clone()),
+        hosted_control_plane: false,
+        has_command_capability: true,
+    };
+    let task_id = TaskId::new(task.task.clone());
+    let mut overlay = VfsOverlay::new(task_id.clone(), NodeId::new(args.node.clone()));
+    let output = executor.run(
+        VirtualThreadCommand {
+            virtual_thread: task_id,
+            invocation: CommandInvocation {
+                program: task.command.clone(),
+                args: task.command_args.clone(),
+                env: None,
+            },
+            stage_stdout_as: Some(VfsPath::new(task.artifact.clone())?),
+        },
+        &mut overlay,
+    )?;
+    let manifest = overlay.flush();
+    let staged = output.staged_artifact.as_ref();
+    let artifact_digest = staged.map(|artifact| artifact.digest.clone());
+    let artifact_path = staged.map(|artifact| artifact.path.as_str().to_owned());
+    let artifact_size_bytes = staged.map(|artifact| artifact.size);
+
+    let log_event = session.request(json!({
+        "type": "report_task_log",
+        "tenant": &args.tenant,
+        "project": &args.project,
+        "process": &task.process,
+        "node": &args.node,
+        "task": &task.task,
+        "stdout_bytes": output.stdout.len(),
+        "stderr_bytes": output.stderr.len(),
+        "stdout_tail": &output.stdout,
+        "stderr_tail": &output.stderr,
+        "stdout_truncated": output.stdout_truncated,
+        "stderr_truncated": output.stderr_truncated,
+        "backpressured": output.log_backpressured,
+    }))?;
+    let vfs_metadata = session.request(json!({
+        "type": "report_vfs_metadata",
+        "tenant": &args.tenant,
+        "project": &args.project,
+        "process": &task.process,
+        "node": &args.node,
+        "task": &task.task,
+        "artifact_path": artifact_path,
+        "artifact_digest": artifact_digest,
+        "artifact_size_bytes": artifact_size_bytes,
+        "large_bytes_uploaded": manifest.large_bytes_uploaded,
+    }))?;
+    let recorded = session.request(json!({
+        "type": "task_completed",
+        "tenant": &args.tenant,
+        "project": &args.project,
+        "process": &task.process,
+        "node": &args.node,
+        "task": &task.task,
+        "status_code": output.status_code,
+        "stdout_bytes": output.stdout.len(),
+        "stderr_bytes": output.stderr.len(),
+        "stdout_tail": &output.stdout,
+        "stderr_tail": &output.stderr,
+        "stdout_truncated": output.stdout_truncated,
+        "stderr_truncated": output.stderr_truncated,
+        "artifact_path": staged.map(|artifact| artifact.path.as_str().to_owned()),
+        "artifact_digest": staged.map(|artifact| artifact.digest.clone()),
+        "artifact_size_bytes": artifact_size_bytes,
+    }))?;
+
+    Ok(node_report(
+        output,
+        manifest.large_bytes_uploaded,
+        registration,
+        heartbeat,
+        capability_report,
+        task.task_assignment_response,
+        debug_command,
+        log_event,
+        vfs_metadata,
+        recorded,
+        session.requests(),
+    ))
+}
+
+fn node_report(
+    output: CommandOutput,
+    large_bytes_uploaded: bool,
+    registration_response: Value,
+    heartbeat_response: Value,
+    capability_response: Value,
+    task_assignment_response: Value,
+    debug_command_response: Value,
+    log_event_response: Value,
+    vfs_metadata_response: Value,
+    coordinator_response: Value,
+    session_requests: usize,
+) -> Value {
+    json!({
+        "node_status": "completed",
+        "virtual_thread": output.virtual_thread,
+        "terminal_state": if output.status_code == Some(0) { "completed" } else { "failed" },
+        "status_code": output.status_code,
+        "stdout_bytes": output.stdout.len(),
+        "stderr_bytes": output.stderr.len(),
+        "stdout_tail": &output.stdout,
+        "stderr_tail": &output.stderr,
+        "stdout_truncated": output.stdout_truncated,
+        "stderr_truncated": output.stderr_truncated,
+        "log_backpressured": output.log_backpressured,
+        "staged_artifact": output.staged_artifact,
+        "large_bytes_uploaded": large_bytes_uploaded,
+        "registration_response": registration_response,
+        "heartbeat_response": heartbeat_response,
+        "capability_response": capability_response,
+        "task_assignment_response": task_assignment_response,
+        "debug_command_response": debug_command_response,
+        "log_event_response": log_event_response,
+        "vfs_metadata_response": vfs_metadata_response,
+        "session_requests": session_requests,
+        "coordinator_response": coordinator_response,
+    })
+}
+
+fn cancelled_node_report(
+    _args: &Args,
+    task: &RuntimeTask,
+    registration_response: Value,
+    heartbeat_response: Value,
+    capability_response: Value,
+    task_assignment_response: Value,
+    debug_command_response: Value,
+    coordinator_response: Value,
+    session_requests: usize,
+) -> Value {
+    json!({
+        "node_status": "cancelled",
+        "virtual_thread": &task.task,
+        "terminal_state": "cancelled",
+        "status_code": null,
+        "stdout_bytes": 0,
+        "stderr_bytes": 0,
+        "stdout_tail": "",
+        "stderr_tail": "",
+        "stdout_truncated": false,
+        "stderr_truncated": false,
+        "log_backpressured": false,
+        "staged_artifact": null,
+        "large_bytes_uploaded": false,
+        "registration_response": registration_response,
+        "heartbeat_response": heartbeat_response,
+        "capability_response": capability_response,
+        "task_assignment_response": task_assignment_response,
+        "debug_command_response": debug_command_response,
+        "log_event_response": null,
+        "vfs_metadata_response": null,
+        "session_requests": session_requests,
+        "coordinator_response": coordinator_response,
+    })
+}
+
+fn register_node(
+    session: &mut CoordinatorSession,
+    args: &Args,
+) -> Result<Value, Box<dyn std::error::Error>> {
+    let public_key = args
+        .public_key
+        .clone()
+        .unwrap_or_else(|| format!("{}-public-key", args.node));
+    if let Some(grant) = &args.enrollment_grant {
+        session.request(json!({
+            "type": "exchange_node_enrollment_grant",
+            "tenant": &args.tenant,
+            "project": &args.project,
+            "node": &args.node,
+            "public_key": public_key,
+            "enrollment_grant": grant,
+            "now_epoch_seconds": 0,
+        }))
+    } else {
+        session.request(json!({
+            "type": "attach_node",
+            "tenant": &args.tenant,
+            "project": &args.project,
+            "node": &args.node,
+            "public_key": public_key,
+        }))
+    }
+}
+
+fn wait_for_cancellation(
+    session: &mut CoordinatorSession,
+    args: &Args,
+    task: &RuntimeTask,
+) -> Result<bool, Box<dyn std::error::Error>> {
+    let deadline = Instant::now() + Duration::from_millis(args.control_poll_ms);
+    loop {
+        let control = session.request(json!({
+            "type": "poll_task_control",
+            "tenant": &args.tenant,
+            "project": &args.project,
+            "process": &task.process,
+            "node": &args.node,
+            "task": &task.task,
+        }))?;
+        if control
+            .get("cancel_requested")
+            .and_then(Value::as_bool)
+            .unwrap_or(false)
+        {
+            return Ok(true);
+        }
+        let now = Instant::now();
+        if now >= deadline {
+            return Ok(false);
+        }
+        std::thread::sleep((deadline - now).min(Duration::from_millis(100)));
+    }
+}
+
+struct CoordinatorSession {
+    writer: TcpStream,
+    reader: BufReader<TcpStream>,
+    requests: usize,
+}
+
+impl CoordinatorSession {
+    fn connect(addr: &str) -> Result<Self, Box<dyn std::error::Error>> {
+        let transport_addr = json_line_transport_addr(addr);
+        let writer = TcpStream::connect(&transport_addr)?;
+        let reader = BufReader::new(writer.try_clone()?);
+        Ok(Self {
+            writer,
+            reader,
+            requests: 0,
+        })
+    }
+
+    fn request(&mut self, value: Value) -> Result<Value, Box<dyn std::error::Error>> {
+        serde_json::to_writer(&mut self.writer, &value)?;
+        self.writer.write_all(b"\n")?;
+        self.writer.flush()?;
+
+        let mut line = String::new();
+        if self.reader.read_line(&mut line)? == 0 {
+            return Err("coordinator closed session without a response".into());
+        }
+        let response: Value = serde_json::from_str(&line)?;
+        self.requests += 1;
+        if response.get("type").and_then(Value::as_str) == Some("error") {
+            return Err(format!("coordinator error: {response}").into());
+        }
+        Ok(response)
+    }
+
+    fn requests(&self) -> usize {
+        self.requests
+    }
+}
+
+fn json_line_transport_addr(endpoint: &str) -> String {
+    let endpoint = endpoint.trim();
+    for (scheme, default_port) in [("https://", 443), ("http://", 80)] {
+        if let Some(rest) = endpoint.strip_prefix(scheme) {
+            let authority = rest.split('/').next().unwrap_or(rest);
+            if authority.contains(':') {
+                return authority.to_owned();
+            }
+            return format!("{authority}:{default_port}");
+        }
+    }
+    endpoint.to_owned()
+}
+
+fn parse_args() -> Result<Args, Box<dyn std::error::Error>> {
+    let mut coordinator = None;
+    let mut tenant = "tenant".to_owned();
+    let mut project = "project".to_owned();
+    let mut node = "node".to_owned();
+    let mut process = "process".to_owned();
+    let mut task = "compile-linux".to_owned();
+    let mut command = None;
+    let mut command_args = Vec::new();
+    let mut artifact = "/vfs/artifacts/node-output.txt".to_owned();
+    let mut enrollment_grant = None;
+    let mut public_key = None;
+    let mut control_poll_ms = 0;
+    let mut assignment_poll_ms = 500;
+    let mut emit_ready = false;
+    let mut worker = false;
+
+    let mut args = std::env::args().skip(1);
+    while let Some(arg) = args.next() {
+        match arg.as_str() {
+            "--coordinator" => coordinator = args.next(),
+            "--tenant" => tenant = args.next().ok_or("--tenant requires a value")?,
+            "--project-id" => project = args.next().ok_or("--project-id requires a value")?,
+            "--node" => node = args.next().ok_or("--node requires a value")?,
+            "--process" => process = args.next().ok_or("--process requires a value")?,
+            "--task" => task = args.next().ok_or("--task requires a value")?,
+            "--command" => command = args.next(),
+            "--arg" => command_args.push(args.next().ok_or("--arg requires a value")?),
+            "--artifact" => artifact = args.next().ok_or("--artifact requires a value")?,
+            "--enrollment-grant" => enrollment_grant = args.next(),
+            "--public-key" => public_key = args.next(),
+            "--control-poll-ms" => {
+                control_poll_ms = args
+                    .next()
+                    .ok_or("--control-poll-ms requires a value")?
+                    .parse()?
+            }
+            "--assignment-poll-ms" => {
+                assignment_poll_ms = args
+                    .next()
+                    .ok_or("--assignment-poll-ms requires a value")?
+                    .parse()?
+            }
+            "--emit-ready" => emit_ready = true,
+            "--worker" => worker = true,
+            "--project" => {
+                let project_path = PathBuf::from(args.next().ok_or("--project requires a path")?);
+                command_args.extend([
+                    "test".to_owned(),
+                    "--manifest-path".to_owned(),
+                    project_path
+                        .join("Cargo.toml")
+                        .to_string_lossy()
+                        .into_owned(),
+                ]);
+            }
+            other => return Err(format!("unknown argument: {other}").into()),
+        }
+    }
+
+    Ok(Args {
+        coordinator: coordinator.ok_or("--coordinator is required")?,
+        tenant,
+        project,
+        node,
+        process,
+        task,
+        command: command.unwrap_or_else(|| "cargo".to_owned()),
+        command_args,
+        artifact,
+        enrollment_grant,
+        public_key,
+        control_poll_ms,
+        assignment_poll_ms,
+        emit_ready,
+        worker,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::json_line_transport_addr;
+
+    #[test]
+    fn hosted_operator_url_maps_to_json_line_transport_address() {
+        assert_eq!(
+            json_line_transport_addr("https://disasmer.michelpaulissen.com:9443"),
+            "disasmer.michelpaulissen.com:9443"
+        );
+        assert_eq!(
+            json_line_transport_addr("https://disasmer.michelpaulissen.com:9443/auth/device"),
+            "disasmer.michelpaulissen.com:9443"
+        );
+        assert_eq!(
+            json_line_transport_addr("http://operator.example.test"),
+            "operator.example.test:80"
+        );
+        assert_eq!(json_line_transport_addr("127.0.0.1:7999"), "127.0.0.1:7999");
+    }
+}