#!/usr/bin/env node

const assert = require("assert");
const cp = require("child_process");
const crypto = require("crypto");
const fs = require("fs");
const net = require("net");
const path = require("path");
const { coordinatorWireRequest } = require("./coordinator-wire");
const { nodeIdentity } = require("./node-signing");
const { configurePodmanTestEnvironment } = require("./podman-test-env");

const repo = path.resolve(__dirname, "..");
const project = path.join(repo, "examples/launch-build-demo");
const bundleDirectory = path.join(repo, "target/acceptance/sdk-runtime-bundle");
const editedBundleDirectory = path.join(
  repo,
  "target/acceptance/sdk-runtime-edited-bundle"
);
const incompatibleBundleDirectory = path.join(
  repo,
  "target/acceptance/sdk-runtime-incompatible-bundle"
);
const wasmPath = path.join(bundleDirectory, "module.wasm");
const projectSourcePath = path.join(project, "src/build.rs");
const node = "wasmtime-assignment-node";
const identity = nodeIdentity("wasmtime-assignment-smoke", node);
const trapNode = "wasmtime-trap-node";
const trapIdentity = nodeIdentity("wasmtime-assignment-smoke", trapNode);
const clientSessionSecret = "wasmtime-assignment-client-session-secret";

class JsonLineReader {
  constructor(stream) {
    this.buffer = "";
    this.messages = [];
    this.waiters = [];
    stream.on("data", (chunk) => {
      this.buffer += chunk.toString();
      while (this.buffer.includes("\n")) {
        const newline = this.buffer.indexOf("\n");
        const line = this.buffer.slice(0, newline).trim();
        this.buffer = this.buffer.slice(newline + 1);
        if (line) this.messages.push(JSON.parse(line));
      }
      this.flush();
    });
  }

  next(timeoutMs = 240000, label = "JSON line") {
    if (this.messages.length > 0) return Promise.resolve(this.messages.shift());
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        this.waiters = this.waiters.filter((waiter) => waiter.resolve !== resolve);
        reject(new Error(`timed out waiting for ${label}`));
      }, timeoutMs);
      this.waiters.push({ resolve, timer });
    });
  }

  flush() {
    while (this.messages.length > 0 && this.waiters.length > 0) {
      const waiter = this.waiters.shift();
      clearTimeout(waiter.timer);
      waiter.resolve(this.messages.shift());
    }
  }
}

function send(addr, message) {
  return new Promise((resolve, reject) => {
    const socket = net.connect(addr.port, addr.host, () => {
      socket.write(`${JSON.stringify(coordinatorWireRequest(message))}\n`);
    });
    let buffer = "";
    const timer = setTimeout(() => {
      socket.destroy();
      reject(new Error(`timed out waiting for ${message.type}`));
    }, 30000);
    socket.on("data", (chunk) => {
      buffer += chunk.toString();
      const newline = buffer.indexOf("\n");
      if (newline < 0) return;
      clearTimeout(timer);
      socket.destroy();
      try {
        resolve(JSON.parse(buffer.slice(0, newline)));
      } catch (error) {
        reject(error);
      }
    });
    socket.on("error", (error) => {
      clearTimeout(timer);
      reject(error);
    });
  });
}

function authenticated(request) {
  return {
    type: "authenticated",
    session_secret: clientSessionSecret,
    request,
  };
}

function stopChild(child) {
  if (!child || child.exitCode !== null) return;
  child.kill("SIGTERM");
}

function delay(milliseconds) {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}

function commandWithPodman(program, args) {
  if (cp.spawnSync("podman", ["--version"], { stdio: "ignore" }).status === 0) {
    return { program, args };
  }
  if (cp.spawnSync("nix", ["--version"], { stdio: "ignore" }).status === 0) {
    return {
      program: "nix",
      args: ["shell", "nixpkgs#podman", "--command", program, ...args],
    };
  }
  throw new Error(
    "Wasmtime assignment smoke requires rootless Podman (or Nix to provide it)"
  );
}

function procChildren(pid) {
  try {
    return fs
      .readFileSync(`/proc/${pid}/task/${pid}/children`, "utf8")
      .trim()
      .split(/\s+/)
      .filter(Boolean)
      .map(Number);
  } catch (_) {
    return [];
  }
}

function procDescendants(rootPid) {
  const pending = [rootPid];
  const descendants = [];
  while (pending.length > 0) {
    const parent = pending.shift();
    for (const child of procChildren(parent)) {
      descendants.push(child);
      pending.push(child);
    }
  }
  return descendants;
}

function procCommand(pid) {
  try {
    return fs.readFileSync(`/proc/${pid}/cmdline`, "utf8").replaceAll("\0", " ");
  } catch (_) {
    return "";
  }
}

function procState(pid) {
  const status = fs.readFileSync(`/proc/${pid}/status`, "utf8");
  return status.match(/^State:\s+([^\n]+)$/m)?.[1] || "unknown";
}

async function waitForNativeSleep(rootPid) {
  for (let attempt = 0; attempt < 2400; attempt += 1) {
    const sleep = procDescendants(rootPid).find((pid) =>
      /(?:^|\s)sleep\s+30(?:\s|$)/.test(procCommand(pid))
    );
    if (sleep) return sleep;
    await delay(25);
  }
  throw new Error(`worker ${rootPid} did not start the abort probe's native sleep process`);
}

async function waitForDebugEpochState(addr, process, epoch, field) {
  for (let attempt = 0; attempt < 1200; attempt += 1) {
    const status = await send(addr, authenticated({
      type: "inspect_debug_epoch",
      process,
      epoch,
    }));
    assert.strictEqual(status.type, "debug_epoch_status", JSON.stringify(status));
    if (status.failed) {
      throw new Error(
        `debug epoch ${epoch} failed: ${status.failure_messages.join("; ")}`
      );
    }
    if (status[field]) return status;
    await delay(25);
  }
  throw new Error(`timed out waiting for debug epoch ${epoch} ${field}`);
}

async function waitForBreakpointHit(addr, process) {
  for (let attempt = 0; attempt < 1200; attempt += 1) {
    const status = await send(addr, authenticated({
      type: "inspect_debug_breakpoints",
      process,
    }));
    assert.strictEqual(status.type, "debug_breakpoints", JSON.stringify(status));
    if (status.hit_epoch != null) return status;
    await delay(25);
  }
  throw new Error(`timed out waiting for executing Wasm breakpoint in ${process}`);
}

async function waitForRunningTaskInstances(addr, process, taskInstances) {
  for (let attempt = 0; attempt < 1200; attempt += 1) {
    const response = await send(addr, authenticated({
      type: "list_task_snapshots",
      process,
    }));
    assert.strictEqual(response.type, "task_snapshots", JSON.stringify(response));
    const running = new Set(
      response.snapshots
        .filter((snapshot) => snapshot.current && snapshot.state === "running")
        .map((snapshot) => snapshot.task)
    );
    if (taskInstances.every((task) => running.has(task))) return;
    await delay(50);
  }
  throw new Error(
    `timed out waiting for running task instances ${taskInstances.join(", ")} in ${process}`
  );
}

async function waitForTaskJoin(addr, process, task, timeoutMs = 60000) {
  const deadline = Date.now() + timeoutMs;
  while (Date.now() < deadline) {
    const response = await send(addr, authenticated({
      type: "join_task",
      process,
      task,
    }));
    assert.strictEqual(response.type, "task_joined", JSON.stringify(response));
    if (response.join.state !== "pending") return response;
    await delay(25);
  }
  throw new Error(`timed out waiting to join task ${task} in ${process}`);
}

(async () => {
  configurePodmanTestEnvironment(repo);
  const build = JSON.parse(
    cp.execFileSync(
      "cargo",
      [
        "run",
        "-q",
        "-p",
        "clusterflux-cli",
        "--bin",
        "clusterflux",
        "--",
        "build",
        "--project",
        project,
        "--output",
        bundleDirectory,
        "--json",
      ],
      { cwd: repo, encoding: "utf8" }
    )
  );
  assert.strictEqual(build.status, "built");
  assert.strictEqual(build.bundle_artifact.module, wasmPath);
  assert(build.bundle_artifact.task_descriptor_count >= 2);
  const wasm = fs.readFileSync(wasmPath);
  const wasmModuleBase64 = wasm.toString("base64");
  const bundleDigest = `sha256:${crypto.createHash("sha256").update(wasm).digest("hex")}`;
  assert.strictEqual(build.bundle_artifact.bundle_digest, bundleDigest);
  const linuxEnvironment = build.bundle.metadata.environments.find(
    (environment) => environment.name === "linux"
  );
  assert(linuxEnvironment, "build bundle must declare the Linux environment");

  const coordinator = cp.spawn(
    "cargo",
    [
      "run",
      "-q",
      "-p",
      "clusterflux-coordinator",
      "--bin",
      "clusterflux-coordinator",
      "--",
      "--listen",
      "127.0.0.1:0",
    ],
    {
      cwd: repo,
      env: {
        ...process.env,
        CLUSTERFLUX_SELF_HOSTED_SESSION_SECRET: clientSessionSecret,
        CLUSTERFLUX_SELF_HOSTED_TENANT: "tenant",
        CLUSTERFLUX_SELF_HOSTED_PROJECT: "project",
        CLUSTERFLUX_SELF_HOSTED_USER: "user",
      },
    }
  );
  const coordinatorLines = new JsonLineReader(coordinator.stdout);
  let worker;
  let projectSourceBeforeEdit;
  let coordinatorStderr = "";
  let workerStderr = "";
  coordinator.stderr.on("data", (chunk) => {
    coordinatorStderr += chunk.toString();
  });

  try {
    const ready = await coordinatorLines.next();
    const [host, portText] = ready.listen.split(":");
    const addr = { host, port: Number(portText) };
    assert.strictEqual(ready.client_authority, "strict");
    assert.strictEqual(ready.self_hosted_session_bootstrapped, true);
    assert.strictEqual((await send(addr, { type: "ping" })).type, "pong");

    const grant = await send(addr, authenticated({
      type: "create_node_enrollment_grant",
      ttl_seconds: 900,
    }));
    assert.strictEqual(
      grant.type,
      "node_enrollment_grant_created",
      JSON.stringify(grant)
    );

    const initialWorkerInvocation = commandWithPodman("cargo", [
        "run",
        "-q",
        "-p",
        "clusterflux-node",
        "--bin",
        "clusterflux-node",
        "--",
        "--coordinator",
        ready.listen,
        "--tenant",
        "tenant",
        "--project-id",
        "project",
        "--node",
        node,
        "--public-key",
        identity.publicKey,
        "--enrollment-grant",
        grant.grant,
        "--worker",
        "--assignment-poll-ms",
        "25",
        "--emit-ready",
        "--project-root",
        project,
      ]);
    worker = cp.spawn(
      initialWorkerInvocation.program,
      initialWorkerInvocation.args,
      {
        cwd: repo,
        env: { ...process.env, CLUSTERFLUX_NODE_PRIVATE_KEY: identity.privateKey },
      }
    );
    worker.stderr.on("data", (chunk) => {
      workerStderr += chunk.toString();
    });
    const workerLines = new JsonLineReader(worker.stdout);
    const workerReady = await workerLines.next();
    assert.strictEqual(workerReady.node_status, "ready");
    assert.strictEqual(workerReady.node, node);

    const started = await send(addr, authenticated({
      type: "start_process",
      process: "vp-wasmtime-assignment",
      restart: false,
    }));
    assert.strictEqual(started.type, "process_started");

    const sdkRun = JSON.parse(
      cp.execFileSync(
        "cargo",
        [
          "run",
          "-q",
          "-p",
          "launch-build-demo",
          "--bin",
          "sdk-product-runtime",
        ],
        {
          cwd: repo,
          encoding: "utf8",
          env: {
            ...process.env,
            CLUSTERFLUX_SDK_COORDINATOR: ready.listen,
            CLUSTERFLUX_SDK_TENANT: "tenant",
            CLUSTERFLUX_SDK_PROJECT: "project",
            CLUSTERFLUX_SDK_PROCESS: "vp-wasmtime-assignment",
            CLUSTERFLUX_SDK_USER: "user",
            CLUSTERFLUX_SDK_SESSION_SECRET: clientSessionSecret,
            CLUSTERFLUX_SDK_BUNDLE_DIGEST: bundleDigest,
            CLUSTERFLUX_SDK_WASM_MODULE_PATH: wasmPath,
            CLUSTERFLUX_SDK_VFS_EPOCH: String(started.epoch),
          },
        }
      )
    );
    assert.strictEqual(sdkRun.kind, "clusterflux-sdk-product-runtime");
    assert.strictEqual(sdkRun.remote_result, 42);
    assert.strictEqual(sdkRun.local_function_invoked_by_spawn, false);
    assert.strictEqual(sdkRun.task_spec.task_definition, "task_add_one");
    assert.match(sdkRun.task_spec.task_instance, /^ti:vp-wasmtime-assignment:\d+$/);
    assert.strictEqual(sdkRun.task_spec.environment_id, "linux");
    assert.strictEqual(sdkRun.task_spec.vfs_epoch, started.epoch);
    assert.strictEqual(sdkRun.task_spec.bundle_digest, bundleDigest);
    assert.strictEqual(
      sdkRun.task_spec.dispatch.kind,
      "coordinator_node_wasm"
    );
    assert.strictEqual(sdkRun.task_spec.dispatch.export, null);
    assert.strictEqual(sdkRun.task_spec.dispatch.abi, "task_v1");
    assert.deepStrictEqual(sdkRun.task_spec.args, [{ SmallJson: 41 }]);

    const nodeRun = await workerLines.next();
    assert.strictEqual(nodeRun.node_status, "completed");
    assert.strictEqual(nodeRun.status_code, 0);
    assert.strictEqual(nodeRun.stdout_tail, '{"SmallJson":42}\n');
    assert.strictEqual(nodeRun.staged_artifact, null);
    assert.strictEqual(nodeRun.large_bytes_uploaded, false);
    assert.strictEqual(
      nodeRun.task_assignment_response.task,
      sdkRun.task_spec.task_instance
    );
    assert.strictEqual(
      nodeRun.task_assignment_response.task_spec.dispatch.abi,
      "task_v1"
    );
    assert.strictEqual(nodeRun.vfs_metadata_response.type, "vfs_metadata_recorded");
    assert.strictEqual(nodeRun.coordinator_response.type, "task_recorded");

    const events = await send(addr, authenticated({
      type: "list_task_events",
      process: "vp-wasmtime-assignment",
    }));
    assert.strictEqual(events.type, "task_events");
    assert.strictEqual(events.events.length, 1);
    assert.strictEqual(events.events[0].task, sdkRun.task_spec.task_instance);
    assert.strictEqual(events.events[0].stdout_tail, '{"SmallJson":42}\n');
    assert.deepStrictEqual(events.events[0].result, { SmallJson: 42 });

    // Prove the edited-task restart boundary with a real Rust source edit and
    // real Wasm rebuild. The first execution uses the original bundle, then the
    // coordinator accepts a compatible replacement bundle, preserves the
    // original argument envelope, assigns a fresh task-instance ID, and the
    // same node executes the changed code from the clean task-entry checkpoint.
    const editableLaunch = await send(addr, authenticated({
      type: "launch_task",
      task_spec: {
        tenant: "tenant",
        project: "project",
        process: "vp-wasmtime-assignment",
        task_definition: "task_add_one",
        task_instance: "task_add_one-editable-1",
        dispatch: {
          kind: "coordinator_node_wasm",
          export: null,
          abi: "task_v1",
        },
        environment_id: null,
        environment: null,
        environment_digest: null,
        required_capabilities: [],
        dependency_cache: null,
        source_snapshot: null,
        required_artifacts: [],
        args: [{ SmallJson: 41 }],
        vfs_epoch: started.epoch,
        bundle_digest: bundleDigest,
      },
      wait_for_node: false,
      artifact_path: "/vfs/artifacts/task-add-one-editable.json",
      wasm_module_base64: wasmModuleBase64,
    }));
    assert.strictEqual(editableLaunch.type, "task_launched", JSON.stringify(editableLaunch));
    const initialEditableRun = await workerLines.next(
      60000,
      "initial editable task completion"
    );
    assert.strictEqual(initialEditableRun.node_status, "completed");
    assert.strictEqual(initialEditableRun.task_assignment_response.task, "task_add_one-editable-1");
    assert.strictEqual(initialEditableRun.stdout_tail, '{"SmallJson":42}\n');

    const initialEditableJoin = await send(addr, authenticated({
      type: "join_task",
      process: "vp-wasmtime-assignment",
      task: "task_add_one-editable-1",
    }));
    assert.strictEqual(initialEditableJoin.type, "task_joined");
    assert.strictEqual(initialEditableJoin.join.state, "completed");
    assert.deepStrictEqual(initialEditableJoin.join.result, { SmallJson: 42 });

    projectSourceBeforeEdit = fs.readFileSync(projectSourcePath, "utf8");
    const originalBody = "pub extern \"C\" fn task_add_one(input: i32) -> i32 {\n    input + 1\n}";
    const editedBody =
      "pub extern \"C\" fn task_add_one(input: i32) -> i32 {\n" +
      "    input + 2 // edited-task restart integration probe\n" +
      "}";
    assert(
      projectSourceBeforeEdit.includes(originalBody),
      "flagship task_add_one source must retain the expected editable body"
    );
    fs.writeFileSync(
      projectSourcePath,
      projectSourceBeforeEdit.replace(originalBody, editedBody)
    );

    let editedBuild;
    try {
      editedBuild = JSON.parse(
        cp.execFileSync(
          "cargo",
          [
            "run",
            "-q",
            "-p",
            "clusterflux-cli",
            "--bin",
            "clusterflux",
            "--",
            "build",
            "--project",
            project,
            "--output",
            editedBundleDirectory,
            "--json",
          ],
          { cwd: repo, encoding: "utf8" }
        )
      );
    } finally {
      fs.writeFileSync(projectSourcePath, projectSourceBeforeEdit);
      projectSourceBeforeEdit = undefined;
    }
    assert.strictEqual(editedBuild.status, "built");
    const editedWasmPath = path.join(editedBundleDirectory, "module.wasm");
    assert.strictEqual(editedBuild.bundle_artifact.module, editedWasmPath);
    const editedWasm = fs.readFileSync(editedWasmPath);
    const editedBundleDigest = `sha256:${crypto
      .createHash("sha256")
      .update(editedWasm)
      .digest("hex")}`;
    assert.strictEqual(editedBuild.bundle_artifact.bundle_digest, editedBundleDigest);
    assert.notStrictEqual(
      editedBundleDigest,
      bundleDigest,
      "the real Rust source edit must produce changed Wasm bytes"
    );

    const originalTaskDescriptor = JSON.parse(
      fs.readFileSync(path.join(bundleDirectory, "task-descriptors.json"), "utf8")
    ).find((descriptor) => descriptor.name === "task_add_one");
    const editedTaskDescriptor = JSON.parse(
      fs.readFileSync(path.join(editedBundleDirectory, "task-descriptors.json"), "utf8")
    ).find((descriptor) => descriptor.name === "task_add_one");
    assert(originalTaskDescriptor, "original bundle must describe task_add_one");
    assert(editedTaskDescriptor, "edited bundle must describe task_add_one");
    assert.strictEqual(editedTaskDescriptor.stable_id, originalTaskDescriptor.stable_id);
    assert.strictEqual(
      editedTaskDescriptor.restart_compatibility_hash,
      originalTaskDescriptor.restart_compatibility_hash,
      "body-only edit must remain restart-compatible"
    );

    const editedRestart = await send(addr, authenticated({
      type: "restart_task",
      process: "vp-wasmtime-assignment",
      task: "task_add_one-editable-1",
      replacement_bundle: {
        bundle_digest: editedBundleDigest,
        wasm_module_base64: editedWasm.toString("base64"),
      },
    }));
    assert.strictEqual(editedRestart.type, "task_restart", JSON.stringify(editedRestart));
    assert.strictEqual(editedRestart.accepted, true);
    assert.strictEqual(editedRestart.clean_boundary_available, true);
    assert.strictEqual(editedRestart.active_task, false);
    assert.strictEqual(editedRestart.completed_event_observed, true);
    assert.strictEqual(editedRestart.requires_whole_process_restart, false);
    assert.strictEqual(
      editedRestart.restarted_task_instance,
      "task_add_one-editable-1",
      "a restart attempt must retain the logical task instance identity"
    );

    const editedNodeRun = await workerLines.next(
      60000,
      "edited replacement task completion"
    );
    assert.strictEqual(editedNodeRun.node_status, "completed");
    assert.strictEqual(
      editedNodeRun.task_assignment_response.task,
      editedRestart.restarted_task_instance
    );
    assert.strictEqual(
      editedNodeRun.task_assignment_response.task_spec.bundle_digest,
      editedBundleDigest
    );
    assert.deepStrictEqual(
      editedNodeRun.task_assignment_response.task_spec.args,
      [{ SmallJson: 41 }],
      "edited restart must preserve the original serialized argument envelope"
    );
    assert.strictEqual(
      editedNodeRun.task_assignment_response.task_spec.vfs_epoch,
      started.epoch,
      "edited restart must use the original clean VFS entry boundary"
    );
    assert.strictEqual(
      editedNodeRun.stdout_tail,
      '{"SmallJson":43}\n',
      "the restarted assignment must execute the edited Rust code"
    );

    const editedJoin = await send(addr, authenticated({
      type: "join_task",
      process: "vp-wasmtime-assignment",
      task: editedRestart.restarted_task_instance,
    }));
    assert.strictEqual(editedJoin.type, "task_joined");
    assert.strictEqual(editedJoin.join.state, "completed");
    assert.deepStrictEqual(editedJoin.join.result, { SmallJson: 43 });

    // A real signature edit must be rejected before a replacement assignment
    // is created, while the compatible restarted result remains intact.
    projectSourceBeforeEdit = fs.readFileSync(projectSourcePath, "utf8");
    const incompatibleBody =
      "pub extern \"C\" fn task_add_one(input: i32) -> i64 {\n" +
      "    i64::from(input) + 1\n" +
      "}";
    assert(projectSourceBeforeEdit.includes(originalBody));
    const restartJoinOriginal =
      '            .expect("restart probe child should complete");';
    const restartJoinWithConversion =
      '            .expect("restart probe child should complete")\n' +
      '            .try_into()\n' +
      '            .expect("incompatible fixture result should fit i32");';
    const restartHostOriginal = "    task_add_one(41)\n}";
    const restartHostWithConversion =
      '    i32::try_from(task_add_one(41))\n' +
      '        .expect("incompatible fixture result should fit i32")\n' +
      "}";
    assert(projectSourceBeforeEdit.includes(restartJoinOriginal));
    assert(projectSourceBeforeEdit.includes(restartHostOriginal));
    const incompatibleSource = projectSourceBeforeEdit
      .replace(originalBody, incompatibleBody)
      .replace(restartJoinOriginal, restartJoinWithConversion)
      .replace(restartHostOriginal, restartHostWithConversion);
    fs.writeFileSync(
      projectSourcePath,
      incompatibleSource
    );
    let incompatibleBuild;
    try {
      incompatibleBuild = JSON.parse(
        cp.execFileSync(
          "cargo",
          [
            "run",
            "-q",
            "-p",
            "clusterflux-cli",
            "--bin",
            "clusterflux",
            "--",
            "build",
            "--project",
            project,
            "--output",
            incompatibleBundleDirectory,
            "--json",
          ],
          { cwd: repo, encoding: "utf8" }
        )
      );
    } finally {
      fs.writeFileSync(projectSourcePath, projectSourceBeforeEdit);
      projectSourceBeforeEdit = undefined;
    }
    assert.strictEqual(incompatibleBuild.status, "built");
    const incompatibleWasmPath = path.join(incompatibleBundleDirectory, "module.wasm");
    const incompatibleWasm = fs.readFileSync(incompatibleWasmPath);
    const incompatibleBundleDigest = `sha256:${crypto
      .createHash("sha256")
      .update(incompatibleWasm)
      .digest("hex")}`;
    assert.strictEqual(
      incompatibleBuild.bundle_artifact.bundle_digest,
      incompatibleBundleDigest
    );
    const incompatibleTaskDescriptor = JSON.parse(
      fs.readFileSync(
        path.join(incompatibleBundleDirectory, "task-descriptors.json"),
        "utf8"
      )
    ).find((descriptor) => descriptor.name === "task_add_one");
    assert(incompatibleTaskDescriptor, "incompatible bundle must still name task_add_one");
    assert.notStrictEqual(
      incompatibleTaskDescriptor.restart_compatibility_hash,
      originalTaskDescriptor.restart_compatibility_hash,
      "signature edit must change restart compatibility metadata"
    );

    const incompatibleRestart = await send(addr, authenticated({
      type: "restart_task",
      process: "vp-wasmtime-assignment",
      task: "task_add_one-editable-1",
      replacement_bundle: {
        bundle_digest: incompatibleBundleDigest,
        wasm_module_base64: incompatibleWasm.toString("base64"),
      },
    }));
    assert.strictEqual(
      incompatibleRestart.type,
      "task_restart",
      JSON.stringify(incompatibleRestart)
    );
    assert.strictEqual(incompatibleRestart.accepted, false);
    assert.strictEqual(incompatibleRestart.requires_whole_process_restart, true);
    assert.strictEqual(incompatibleRestart.restarted_task_instance, null);
    assert.match(
      incompatibleRestart.message,
      /task (?:entrypoint|ABI) changed/,
      "signature edit must explain which restart-compatibility boundary changed"
    );

    const preservedEditedJoin = await send(addr, authenticated({
      type: "join_task",
      process: "vp-wasmtime-assignment",
      task: editedRestart.restarted_task_instance,
    }));
    assert.strictEqual(preservedEditedJoin.type, "task_joined");
    assert.strictEqual(preservedEditedJoin.join.state, "completed");
    assert.deepStrictEqual(
      preservedEditedJoin.join.result,
      { SmallJson: 43 },
      "incompatible edit denial must not corrupt the compatible restarted result"
    );

    const retiredFirstWorker = await send(addr, authenticated({
      type: "revoke_node_credential",
      node,
    }));
    assert.strictEqual(retiredFirstWorker.type, "node_credential_revoked");
    stopChild(worker);
    await delay(100);

    const trapGrant = await send(addr, authenticated({
      type: "create_node_enrollment_grant",
      ttl_seconds: 900,
    }));
    assert.strictEqual(trapGrant.type, "node_enrollment_grant_created");
    const trapWorkerInvocation = commandWithPodman("cargo", [
        "run",
        "-q",
        "-p",
        "clusterflux-node",
        "--bin",
        "clusterflux-node",
        "--",
        "--coordinator",
        ready.listen,
        "--tenant",
        "tenant",
        "--project-id",
        "project",
        "--node",
        trapNode,
        "--public-key",
        trapIdentity.publicKey,
        "--enrollment-grant",
        trapGrant.grant,
        "--worker",
        "--assignment-poll-ms",
        "25",
        "--emit-ready",
        "--project-root",
        project,
      ]);
    worker = cp.spawn(
      trapWorkerInvocation.program,
      trapWorkerInvocation.args,
      {
        cwd: repo,
        env: {
          ...process.env,
          CLUSTERFLUX_NODE_PRIVATE_KEY: trapIdentity.privateKey,
        },
      }
    );
    worker.stderr.on("data", (chunk) => {
      workerStderr += chunk.toString();
    });
    const trapWorkerLines = new JsonLineReader(worker.stdout);
    const trapWorkerReady = await trapWorkerLines.next();
    assert.strictEqual(trapWorkerReady.node_status, "ready");
    assert.strictEqual(trapWorkerReady.node, trapNode);
    const nodeDescriptors = await send(addr, authenticated({
      type: "list_node_descriptors",
    }));
    assert.strictEqual(
      nodeDescriptors.type,
      "node_descriptors",
      JSON.stringify(nodeDescriptors)
    );
    const trapNodeDescriptor = nodeDescriptors.descriptors.find(
      (descriptor) => descriptor.id === trapNode
    );
    assert(trapNodeDescriptor, "trap worker descriptor must be visible");
    assert.deepStrictEqual(
      trapNodeDescriptor.source_snapshots.length,
      1,
      "trap worker must report its authoritative checkout snapshot"
    );
    const sourceSnapshot = trapNodeDescriptor.source_snapshots[0];

    const trapLaunch = await send(addr, authenticated({
      type: "launch_task",
      task_spec: {
        tenant: "tenant",
        project: "project",
        process: "vp-wasmtime-assignment",
        task_definition: "task_trap",
        task_instance: "task_trap-1",
        dispatch: {
          kind: "coordinator_node_wasm",
          export: "task_trap",
          abi: "task_v1",
        },
        environment_id: null,
        environment: null,
        environment_digest: null,
        required_capabilities: [],
        dependency_cache: null,
        source_snapshot: null,
        required_artifacts: [],
        args: [{ SmallJson: 0 }],
        vfs_epoch: started.epoch,
        bundle_digest: bundleDigest,
      },
      wait_for_node: false,
      artifact_path: "/vfs/artifacts/wasmtime-trap.txt",
      wasm_module_base64: wasmModuleBase64,
    }));
    assert.strictEqual(trapLaunch.type, "task_launched", JSON.stringify(trapLaunch));
    const trapNodeRun = await trapWorkerLines.next();
    assert.strictEqual(trapNodeRun.node_status, "failed");
    assert.strictEqual(trapNodeRun.terminal_state, "failed");
    assert.match(trapNodeRun.stderr_tail, /wasmtime task failed|unreachable|trap/i);
    assert.strictEqual(trapNodeRun.log_event_response.type, "task_log_recorded");
    assert.strictEqual(trapNodeRun.coordinator_response.type, "task_recorded");

    const trapJoin = await send(addr, authenticated({
      type: "join_task",
      process: "vp-wasmtime-assignment",
      task: "task_trap-1",
    }));
    assert.strictEqual(trapJoin.type, "task_joined");
    assert.strictEqual(trapJoin.join.state, "failed");
    assert.strictEqual(trapJoin.join.remote_completion_observed, true);
    assert.match(trapJoin.join.message, /wasmtime task failed|unreachable|trap/i);

    const trapRestart = await send(addr, authenticated({
      type: "restart_task",
      process: "vp-wasmtime-assignment",
      task: "task_trap-1",
    }));
    assert.strictEqual(trapRestart.type, "task_restart", JSON.stringify(trapRestart));
    assert.strictEqual(trapRestart.accepted, true);
    assert.strictEqual(trapRestart.clean_boundary_available, true);
    assert.strictEqual(trapRestart.active_task, false);
    assert.strictEqual(trapRestart.completed_event_observed, true);
    assert.strictEqual(trapRestart.requires_whole_process_restart, false);
    assert.strictEqual(
      trapRestart.restarted_task_instance,
      "task_trap-1",
      "a failed-task retry must retain the logical task instance identity"
    );
    assert.match(trapRestart.message, /clean VFS entry boundary epoch/);
    const restartedTrapNodeRun = await trapWorkerLines.next(
      60000,
      "restarted trap task completion"
    );
    assert.strictEqual(restartedTrapNodeRun.node_status, "failed");
    assert.strictEqual(restartedTrapNodeRun.terminal_state, "failed");
    assert.strictEqual(
      restartedTrapNodeRun.task_assignment_response.task,
      trapRestart.restarted_task_instance
    );

    const configuredBreakpoints = await send(addr, authenticated({
      type: "set_debug_breakpoints",
      process: "vp-wasmtime-assignment",
      probe_symbols: ["clusterflux.probe.cooperative_cancellation_probe"],
    }));
    assert.strictEqual(configuredBreakpoints.type, "debug_breakpoints");
    assert.deepStrictEqual(configuredBreakpoints.probe_symbols, [
      "clusterflux.probe.cooperative_cancellation_probe",
    ]);
    const cancellationLaunch = await send(addr, authenticated({
      type: "launch_task",
      task_spec: {
        tenant: "tenant",
        project: "project",
        process: "vp-wasmtime-assignment",
        task_definition: "cooperative_cancellation_probe",
        task_instance: "cooperative_cancellation_probe-1",
        dispatch: {
          kind: "coordinator_node_wasm",
          export: null,
          abi: "task_v1",
        },
        environment_id: null,
        environment: null,
        environment_digest: null,
        required_capabilities: [],
        dependency_cache: null,
        source_snapshot: null,
        required_artifacts: [],
        args: [],
        vfs_epoch: started.epoch,
        bundle_digest: bundleDigest,
      },
      wait_for_node: false,
      artifact_path: "/vfs/artifacts/cooperative-cancellation-probe.txt",
      wasm_module_base64: wasmModuleBase64,
    }));
    assert.strictEqual(cancellationLaunch.type, "task_launched");
    const breakpointHit = await waitForBreakpointHit(
      addr,
      "vp-wasmtime-assignment"
    );
    assert.strictEqual(
      breakpointHit.hit_probe_symbol,
      "clusterflux.probe.cooperative_cancellation_probe"
    );
    assert.strictEqual(
      breakpointHit.hit_task,
      "cooperative_cancellation_probe-1"
    );
    const debugEpoch = { epoch: breakpointHit.hit_epoch };
    const frozenDebugEpoch = await waitForDebugEpochState(
      addr,
      "vp-wasmtime-assignment",
      debugEpoch.epoch,
      "fully_frozen"
    );
    assert.strictEqual(frozenDebugEpoch.acknowledgements.length, 1);
    assert.strictEqual(frozenDebugEpoch.acknowledgements[0].state, "frozen");
    assert.strictEqual(
      frozenDebugEpoch.acknowledgements[0].task,
      "cooperative_cancellation_probe-1"
    );
    const debugResume = await send(addr, authenticated({
      type: "resume_debug_epoch",
      process: "vp-wasmtime-assignment",
      epoch: debugEpoch.epoch,
    }));
    assert.strictEqual(debugResume.type, "debug_epoch");
    assert.strictEqual(debugResume.command, "resume");
    const resumedDebugEpoch = await waitForDebugEpochState(
      addr,
      "vp-wasmtime-assignment",
      debugEpoch.epoch,
      "fully_resumed"
    );
    assert.strictEqual(resumedDebugEpoch.acknowledgements.length, 1);
    assert.strictEqual(resumedDebugEpoch.acknowledgements[0].state, "running");
    const cancellationRequested = await send(addr, authenticated({
      type: "cancel_process",
      process: "vp-wasmtime-assignment",
    }));
    assert.strictEqual(
      cancellationRequested.type,
      "process_cancellation_requested"
    );
    assert.strictEqual(cancellationRequested.cancelled_tasks.length, 1);
    assert.strictEqual(
      cancellationRequested.cancelled_tasks[0].task,
      "cooperative_cancellation_probe-1"
    );
    const cancellationNodeRun = await trapWorkerLines.next(
      10000,
      "cooperative cancellation task completion"
    );
    assert.strictEqual(cancellationNodeRun.node_status, "completed");
    assert.strictEqual(cancellationNodeRun.terminal_state, "completed");
    const cancellationJoin = await send(addr, authenticated({
      type: "join_task",
      process: "vp-wasmtime-assignment",
      task: "cooperative_cancellation_probe-1",
    }));
    assert.strictEqual(cancellationJoin.type, "task_joined");
    assert.strictEqual(cancellationJoin.join.state, "completed");
    assert.deepStrictEqual(cancellationJoin.join.result, { SmallJson: 17 });

    // Completing the cooperatively cancelled task releases the process slot.
    // A redundant forced abort must not resurrect or mutate that process.
    const cancelledProcessCleanup = await send(addr, authenticated({
      type: "abort_process",
      process: "vp-wasmtime-assignment",
    }));
    assert.strictEqual(cancelledProcessCleanup.type, "error");
    assert.match(
      cancelledProcessCleanup.message,
      /process abort requires an active virtual process/
    );

    const multiParticipantProcess = await send(addr, authenticated({
      type: "start_process",
      process: "vp-wasmtime-multi-debug",
      restart: false,
    }));
    assert.strictEqual(multiParticipantProcess.type, "process_started");
    const multiParticipantLaunch = await send(addr, authenticated({
      type: "launch_task",
      task_spec: {
        tenant: "tenant",
        project: "project",
        process: multiParticipantProcess.process,
        task_definition: "debug_parent_probe",
        task_instance: "debug_parent_probe-1",
        dispatch: {
          kind: "coordinator_node_wasm",
          export: null,
          abi: "task_v1",
        },
        environment_id: null,
        environment: null,
        environment_digest: null,
        required_capabilities: [],
        dependency_cache: null,
        source_snapshot: null,
        required_artifacts: [],
        args: [],
        vfs_epoch: multiParticipantProcess.epoch,
        bundle_digest: bundleDigest,
      },
      wait_for_node: false,
      artifact_path: "/vfs/artifacts/debug-parent-probe.txt",
      wasm_module_base64: wasmModuleBase64,
    }));
    assert.strictEqual(multiParticipantLaunch.type, "task_launched");
    const debugChildInstance = "debug_parent_probe-1:child:1";
    await waitForRunningTaskInstances(addr, multiParticipantProcess.process, [
      "debug_parent_probe-1",
      debugChildInstance,
    ]);
    const multiParticipantFreeze = await send(addr, authenticated({
      type: "create_debug_epoch",
      process: multiParticipantProcess.process,
      stopped_task: debugChildInstance,
      reason: "prove all-stop across active parent and child Wasm participants",
    }));
    assert.strictEqual(
      multiParticipantFreeze.type,
      "debug_epoch",
      JSON.stringify(multiParticipantFreeze)
    );
    const multiParticipantFrozen = await waitForDebugEpochState(
      addr,
      multiParticipantProcess.process,
      multiParticipantFreeze.epoch,
      "fully_frozen"
    );
    assert.strictEqual(multiParticipantFreeze.affected_tasks.length, 2);
    assert.strictEqual(multiParticipantFrozen.acknowledgements.length, 2);
    assert.deepStrictEqual(
      multiParticipantFrozen.acknowledgements
        .map((acknowledgement) => acknowledgement.task)
        .sort(),
      [debugChildInstance, "debug_parent_probe-1"].sort()
    );
    const multiParticipantResume = await send(addr, authenticated({
      type: "resume_debug_epoch",
      process: multiParticipantProcess.process,
      epoch: multiParticipantFreeze.epoch,
    }));
    assert.strictEqual(multiParticipantResume.type, "debug_epoch");
    const multiParticipantResumed = await waitForDebugEpochState(
      addr,
      multiParticipantProcess.process,
      multiParticipantFreeze.epoch,
      "fully_resumed"
    );
    assert.strictEqual(multiParticipantResumed.acknowledgements.length, 2);
    const multiParticipantCancellation = await send(addr, authenticated({
      type: "cancel_process",
      process: multiParticipantProcess.process,
    }));
    assert.strictEqual(
      multiParticipantCancellation.type,
      "process_cancellation_requested"
    );
    assert.strictEqual(multiParticipantCancellation.cancelled_tasks.length, 2);
    const multiParticipantJoin = await waitForTaskJoin(
      addr,
      multiParticipantProcess.process,
      "debug_parent_probe-1"
    );
    const multiParticipantNodeRun = await trapWorkerLines.next(
      10000,
      "multi-participant parent completion"
    );
    assert.strictEqual(multiParticipantNodeRun.node_status, "completed");
    assert.strictEqual(multiParticipantJoin.join.state, "completed");
    assert.deepStrictEqual(multiParticipantJoin.join.result, { SmallJson: 23 });
    const multiParticipantCleanup = await send(addr, authenticated({
      type: "abort_process",
      process: multiParticipantProcess.process,
    }));
    assert.strictEqual(multiParticipantCleanup.type, "error");
    assert.match(
      multiParticipantCleanup.message,
      /process abort requires an active virtual process/
    );

    const abortProcess = await send(addr, authenticated({
      type: "start_process",
      process: "vp-wasmtime-abort",
      restart: false,
    }));
    assert.strictEqual(abortProcess.type, "process_started");

    const abortLaunch = await send(addr, authenticated({
      type: "launch_task",
      task_spec: {
        tenant: "tenant",
        project: "project",
        process: abortProcess.process,
        task_definition: "abort_probe",
        task_instance: "abort_probe-1",
        dispatch: {
          kind: "coordinator_node_wasm",
          export: null,
          abi: "task_v1",
        },
        environment_id: "linux",
        environment: linuxEnvironment.requirements,
        environment_digest: linuxEnvironment.digest,
        required_capabilities: ["Command"],
        dependency_cache: null,
        source_snapshot: sourceSnapshot,
        required_artifacts: [],
        args: [],
        vfs_epoch: abortProcess.epoch,
        bundle_digest: bundleDigest,
      },
      wait_for_node: false,
      artifact_path: "/vfs/artifacts/abort-probe.txt",
      wasm_module_base64: wasmModuleBase64,
    }));
    assert.strictEqual(abortLaunch.type, "task_launched");
    assert.strictEqual(process.platform, "linux", "native all-stop proof is the Linux MVP path");
    const nativeSleepPid = await waitForNativeSleep(worker.pid);
    const nativeFreeze = await send(addr, authenticated({
      type: "create_debug_epoch",
      process: abortProcess.process,
      stopped_task: "abort_probe-1",
      reason: "freeze while the abort probe native command is running",
    }));
    assert.strictEqual(nativeFreeze.type, "debug_epoch", JSON.stringify(nativeFreeze));
    assert.strictEqual(nativeFreeze.command, "freeze");
    const nativeFrozen = await waitForDebugEpochState(
      addr,
      abortProcess.process,
      nativeFreeze.epoch,
      "fully_frozen"
    );
    assert.strictEqual(nativeFrozen.acknowledgements.length, 1);
    assert.strictEqual(nativeFrozen.acknowledgements[0].task, "abort_probe-1");
    assert.match(procState(nativeSleepPid), /^T \(stopped\)/);
    const nativeResume = await send(addr, authenticated({
      type: "resume_debug_epoch",
      process: abortProcess.process,
      epoch: nativeFreeze.epoch,
    }));
    assert.strictEqual(nativeResume.type, "debug_epoch");
    assert.strictEqual(nativeResume.command, "resume");
    const nativeResumed = await waitForDebugEpochState(
      addr,
      abortProcess.process,
      nativeFreeze.epoch,
      "fully_resumed"
    );
    assert.strictEqual(nativeResumed.acknowledgements.length, 1);
    assert.doesNotMatch(procState(nativeSleepPid), /^T \(stopped\)/);
    const abortStartedAt = Date.now();
    const aborted = await send(addr, authenticated({
      type: "abort_process",
      process: abortProcess.process,
    }));
    assert.strictEqual(aborted.type, "process_aborted");
    assert.strictEqual(aborted.aborted_tasks.length, 1);
    assert.strictEqual(aborted.aborted_tasks[0].task, "abort_probe-1");
    const abortedNodeRun = await trapWorkerLines.next(
      10000,
      "forced-abort task completion"
    );
    assert.strictEqual(
      abortedNodeRun.node_status,
      "cancelled",
      JSON.stringify(abortedNodeRun)
    );
    assert.strictEqual(abortedNodeRun.terminal_state, "cancelled");
    const abortElapsedMs = Date.now() - abortStartedAt;
    assert(
      abortElapsedMs < 10000,
      `abort should stop the running command promptly; observed ${abortElapsedMs}ms`
    );
    const afterAbort = await send(addr, authenticated({ type: "list_processes" }));
    assert.strictEqual(afterAbort.type, "process_statuses");
    assert.deepStrictEqual(afterAbort.processes, []);

    const report = {
      kind: "clusterflux-wasmtime-assignment-smoke",
      coordinator: ready.listen,
      client_authority: ready.client_authority,
      process: started.process,
      epoch: started.epoch,
      task: sdkRun.task_spec.task,
      node,
      bundle_digest: bundleDigest,
      sdk_product_runtime: sdkRun,
      assignment_response: nodeRun.task_assignment_response.type,
      assignment_wasm_abi: nodeRun.task_assignment_response.task_spec.dispatch.abi,
      node_runtime_status: nodeRun.node_status,
      node_runtime_stdout: nodeRun.stdout_tail,
      vfs_metadata_response: nodeRun.vfs_metadata_response.type,
      task_completion_response: nodeRun.coordinator_response.type,
      task_events: events.events.length,
      large_bytes_uploaded: nodeRun.large_bytes_uploaded,
      edited_task_restart: {
        source_edit: "task_add_one: input + 1 -> input + 2",
        original_bundle_digest: bundleDigest,
        replacement_bundle_digest: editedBundleDigest,
        original_task_instance: "task_add_one-editable-1",
        restarted_task_instance: editedRestart.restarted_task_instance,
        original_args: [{ SmallJson: 41 }],
        clean_vfs_epoch: started.epoch,
        initial_result: initialEditableJoin.join.result,
        edited_result: editedJoin.join.result,
        incompatible_source_edit: "task_add_one result: i32 -> i64",
        incompatible_bundle_digest: incompatibleBundleDigest,
        incompatible_requires_whole_process_restart:
          incompatibleRestart.requires_whole_process_restart,
        preserved_result_after_incompatible_denial:
          preservedEditedJoin.join.result,
      },
      trapped_task: {
        task: trapLaunch.task,
        node_status: trapNodeRun.node_status,
        terminal_state: trapNodeRun.terminal_state,
        log_response: trapNodeRun.log_event_response.type,
        completion_response: trapNodeRun.coordinator_response.type,
        join_state: trapJoin.join.state,
        remote_completion_observed:
          trapJoin.join.remote_completion_observed,
        checkpoint_restart: {
          accepted: trapRestart.accepted,
          clean_boundary_available: trapRestart.clean_boundary_available,
          requires_whole_process_restart:
            trapRestart.requires_whole_process_restart,
          rerun_terminal_state: restartedTrapNodeRun.terminal_state,
        },
      },
      cooperative_cancellation: {
        task: cancellationLaunch.task,
        request: cancellationRequested.type,
        task_observed_request: true,
        terminal_state: cancellationNodeRun.terminal_state,
        result: cancellationJoin.join.result,
        process_cleanup_was_explicit: false,
        process_slot_released: true,
      },
      debug_epoch: {
        epoch: debugEpoch.epoch,
        wasm_probe_hit: breakpointHit.hit_probe_symbol,
        freeze_acknowledgements: frozenDebugEpoch.acknowledgements.length,
        fully_frozen: frozenDebugEpoch.fully_frozen,
        resume_acknowledgements: resumedDebugEpoch.acknowledgements.length,
        fully_resumed: resumedDebugEpoch.fully_resumed,
      },
      native_command_debug_epoch: {
        process: abortProcess.process,
        task: abortLaunch.task,
        native_sleep_pid: nativeSleepPid,
        epoch: nativeFreeze.epoch,
        freeze_acknowledgements: nativeFrozen.acknowledgements.length,
        sleep_process_observed_stopped: true,
        resume_acknowledgements: nativeResumed.acknowledgements.length,
        sleep_process_observed_resumed: true,
      },
      multi_participant_debug_epoch: {
        process: multiParticipantProcess.process,
        parent_task: multiParticipantLaunch.task,
        epoch: multiParticipantFreeze.epoch,
        expected_participants: multiParticipantFreeze.affected_tasks.length,
        freeze_acknowledgements:
          multiParticipantFrozen.acknowledgements.length,
        fully_frozen: multiParticipantFrozen.fully_frozen,
        resume_acknowledgements:
          multiParticipantResumed.acknowledgements.length,
        fully_resumed: multiParticipantResumed.fully_resumed,
        parent_result: multiParticipantJoin.join.result,
      },
      aborted_running_task: {
        process: abortProcess.process,
        task: abortLaunch.task,
        terminal_state: abortedNodeRun.terminal_state,
        process_slot_released: afterAbort.processes.length === 0,
        elapsed_ms: abortElapsedMs,
      },
    };
    const reportPath = path.join(repo, "target", "acceptance", "wasmtime-assignment.json");
    fs.mkdirSync(path.dirname(reportPath), { recursive: true });
    fs.writeFileSync(reportPath, `${JSON.stringify(report, null, 2)}\n`);
    console.log(`Wasmtime assignment smoke passed: ${reportPath}`);
  } catch (error) {
    if (coordinatorStderr.trim()) console.error(coordinatorStderr);
    if (workerStderr.trim()) console.error(workerStderr);
    throw error;
  } finally {
    if (projectSourceBeforeEdit !== undefined) {
      fs.writeFileSync(projectSourcePath, projectSourceBeforeEdit);
    }
    stopChild(worker);
    stopChild(coordinator);
  }
})().catch((error) => {
  console.error(error.stack || error.message);
  process.exit(1);
});