#!/usr/bin/env node

const assert = require("assert");
const cp = require("child_process");
const crypto = require("crypto");
const fs = require("fs");
const https = require("https");
const os = require("os");
const path = require("path");
const { DapClient } = require("./dap-client");
const { agentIdentity, signedAgentWorkflowRequest } = require("./agent-signing");
const { coordinatorWireRequest } = require("./coordinator-wire");
const {
  nodeIdentity,
  nodeIdentityFromPrivateKey,
  signedNodeHeartbeat,
  signedNodeRequest,
} = require("./node-signing");
const { configurePodmanTestEnvironment } = require("./podman-test-env");

const repo = path.resolve(__dirname, "..");
configurePodmanTestEnvironment(repo);
if (
  !process.env.CLUSTERFLUX_PODMAN_NIX_SHELL &&
  cp.spawnSync("podman", ["--version"], { stdio: "ignore" }).status !== 0 &&
  cp.spawnSync("nix", ["--version"], { stdio: "ignore" }).status === 0
) {
  cp.execFileSync(
    "nix",
    ["shell", "nixpkgs#podman", "--command", "node", __filename],
    {
      cwd: repo,
      env: { ...process.env, CLUSTERFLUX_PODMAN_NIX_SHELL: "1" },
      stdio: "inherit",
    }
  );
  process.exit(0);
}

const releaseRoot = path.resolve(
  process.env.CLUSTERFLUX_PUBLIC_RELEASE_DIR ||
    path.join(repo, "target/public-release")
);
const acceptanceRoot = path.join(repo, "target/acceptance");
const manifestPath = path.join(releaseRoot, "public-release-manifest.json");
const reportPath = path.join(acceptanceRoot, "cli-happy-path-live.json");
const serviceEndpoint = "https://clusterflux.michelpaulissen.com";
const serviceAddr =
  process.env.CLUSTERFLUX_PUBLIC_RELEASE_SERVICE_ADDR ||
  "clusterflux.michelpaulissen.com:443";
const browserOpenCommand =
  process.env.CLUSTERFLUX_PUBLIC_RELEASE_BROWSER_OPEN_COMMAND;
const reuseSessionFile =
  process.env.CLUSTERFLUX_CLI_HAPPY_PATH_REUSE_SESSION_FILE;
const qualityGateEvidencePath =
  process.env.CLUSTERFLUX_QUALITY_GATE_EVIDENCE_PATH;
const enabled = process.env.CLUSTERFLUX_CLI_HAPPY_PATH_LIVE === "1";
const strictFullRelease = process.env.CLUSTERFLUX_STRICT_FULL_RELEASE === "1";
const strictVpsRestart = process.env.CLUSTERFLUX_STRICT_VPS_RESTART === "1";
const strictVpsHost = process.env.CLUSTERFLUX_STRICT_VPS_HOST;
const strictVpsIdentity = process.env.CLUSTERFLUX_STRICT_VPS_IDENTITY;
const strictServiceUnit =
  process.env.CLUSTERFLUX_STRICT_SERVICE_UNIT ||
  "clusterflux-hosted.service";
const strictSoakSeconds = Number(
  process.env.CLUSTERFLUX_STRICT_SOAK_SECONDS || "300"
);
const commands = [];

function requireEnabled() {
  if (!enabled) {
    throw new Error(
      "CLUSTERFLUX_CLI_HAPPY_PATH_LIVE=1 is required because this runs the live CLI happy path"
    );
  }
  if (!browserOpenCommand && !reuseSessionFile) {
    throw new Error(
      "CLUSTERFLUX_PUBLIC_RELEASE_BROWSER_OPEN_COMMAND or CLUSTERFLUX_CLI_HAPPY_PATH_REUSE_SESSION_FILE is required"
    );
  }
  const hasSecondTenantSession = Boolean(
    process.env.CLUSTERFLUX_SECOND_TENANT_SESSION_FILE
  );
  if (
    strictFullRelease &&
    !hasSecondTenantSession
  ) {
    throw new Error(
      "CLUSTERFLUX_STRICT_FULL_RELEASE=1 requires CLUSTERFLUX_SECOND_TENANT_SESSION_FILE so the compiled two-tenant Node and artifact collision can run"
    );
  }
  if (strictFullRelease && !process.env.CLUSTERFLUX_EXPIRED_USER_SESSION_FILE) {
    throw new Error(
      "CLUSTERFLUX_STRICT_FULL_RELEASE=1 requires CLUSTERFLUX_EXPIRED_USER_SESSION_FILE"
    );
  }
  if (strictFullRelease && !qualityGateEvidencePath) {
    throw new Error(
      "CLUSTERFLUX_STRICT_FULL_RELEASE=1 requires CLUSTERFLUX_QUALITY_GATE_EVIDENCE_PATH"
    );
  }
  if (strictFullRelease && !strictVpsRestart) {
    throw new Error(
      "CLUSTERFLUX_STRICT_FULL_RELEASE=1 requires CLUSTERFLUX_STRICT_VPS_RESTART=1"
    );
  }
  if (strictVpsRestart && (!strictVpsHost || !strictVpsIdentity)) {
    throw new Error(
      "strict VPS restart requires CLUSTERFLUX_STRICT_VPS_HOST and CLUSTERFLUX_STRICT_VPS_IDENTITY"
    );
  }
  if (
    !Number.isInteger(strictSoakSeconds) ||
    strictSoakSeconds < 120 ||
    strictSoakSeconds > 1800
  ) {
    throw new Error("CLUSTERFLUX_STRICT_SOAK_SECONDS must be an integer from 120 to 1800");
  }
}

function readJson(file) {
  return JSON.parse(fs.readFileSync(file, "utf8"));
}

function strictQualityGateEvidence(manifest) {
  if (!strictFullRelease) return null;
  const evidence = readJson(path.resolve(qualityGateEvidencePath));
  assert.strictEqual(evidence.source_commit, manifest.source_commit);
  assert.strictEqual(evidence.source_tree_digest, manifest.source_tree_digest);
  for (const gate of [evidence.private, evidence.public]) {
    assert.strictEqual(gate.status, "passed");
    assert(Number.isFinite(gate.duration_ms) && gate.duration_ms > 0);
  }
  assert.strictEqual(evidence.public.independent_filtered_tree, true);
  for (const id of [
    "formatting",
    "clippy_warnings_denied",
    "public_workspace_tests",
    "private_hosted_policy_locked_tests",
    "process_lifecycle_regressions",
    "wasm_example_builds",
    "filtered_public_tree_build_and_tests",
    "vscode_extension_candidate",
  ]) {
    const check = evidence.checks?.[id];
    assert(check, `quality evidence omitted ${id}`);
    assert.strictEqual(check.status, "passed");
    assert(Number.isFinite(check.duration_ms) && check.duration_ms > 0);
  }
  assert.strictEqual(
    evidence.checks.vscode_extension_candidate.candidate_vsix.sha256,
    manifest.release_candidate.extension_sha256
  );
  return evidence;
}

function ensureDir(dir) {
  fs.mkdirSync(dir, { recursive: true });
}

function sha256(bytes) {
  return `sha256:${crypto.createHash("sha256").update(bytes).digest("hex")}`;
}

function nonInteractiveGitEnv(extra = {}) {
  return {
    ...process.env,
    GIT_TERMINAL_PROMPT: "0",
    GIT_ASKPASS: process.env.GIT_ASKPASS || "/bin/false",
    SSH_ASKPASS: process.env.SSH_ASKPASS || "/bin/false",
    GIT_SSH_COMMAND:
      process.env.GIT_SSH_COMMAND ||
      "ssh -o BatchMode=yes -o NumberOfPasswordPrompts=0",
    ...extra,
  };
}

function commandEnv(command, env) {
  if (command !== "git") return env;
  return nonInteractiveGitEnv(env);
}

function recordCommand(command, args) {
  const redacted = [];
  let hideNext = false;
  for (const argument of args) {
    if (hideNext) {
      redacted.push("<redacted>");
      hideNext = false;
    } else {
      redacted.push(argument);
      hideNext = ["--enrollment-grant", "--admin-token"].includes(argument);
    }
  }
  commands.push([command, ...redacted].join(" "));
}

function run(command, args, options = {}) {
  recordCommand(command, args);
  return cp.execFileSync(command, args, {
    cwd: repo,
    encoding: "utf8",
    stdio: ["ignore", "pipe", "pipe"],
    timeout: 15 * 60 * 1000,
    ...options,
    env: commandEnv(command, options.env),
  });
}

function parseJsonOutput(output) {
  const trimmed = output.trim();
  if (!trimmed) throw new Error("expected JSON output, got empty stdout");
  try {
    return JSON.parse(trimmed);
  } catch (_) {
    // Commands may print progress before their final JSON report.
  }
  const lines = trimmed.split(/\r?\n/).filter(Boolean);
  for (let index = lines.length - 1; index >= 0; index -= 1) {
    try {
      return JSON.parse(lines.slice(index).join("\n"));
    } catch (_) {
      // Keep scanning for the trailing JSON value.
    }
  }
  throw new Error(`could not parse JSON output:\n${trimmed}`);
}

function runJson(command, args, options = {}) {
  return parseJsonOutput(run(command, args, options));
}

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

function sendHostedControl(payload) {
  return sendHostedControlEnvelope(
    coordinatorWireRequest(payload, "strict-live")
  );
}

function sendHostedControlEnvelope(envelope) {
  const body = Buffer.from(
    JSON.stringify(envelope)
  );
  const url = new URL("/api/v1/control", serviceEndpoint);
  return new Promise((resolve, reject) => {
    const request = https.request(
      url,
      {
        method: "POST",
        headers: {
          "content-type": "application/json",
          "content-length": body.length,
        },
        timeout: 30_000,
      },
      (response) => {
        const chunks = [];
        response.on("data", (chunk) => chunks.push(chunk));
        response.on("end", () => {
          const responseBody = Buffer.concat(chunks).toString("utf8");
          if (response.statusCode < 200 || response.statusCode >= 300) {
            reject(
              new Error(
                `hosted control HTTP ${response.statusCode}: ${responseBody}`
              )
            );
            return;
          }
          try {
            resolve(JSON.parse(responseBody));
          } catch (error) {
            reject(
              new Error(`hosted control returned invalid JSON: ${error.message}`)
            );
          }
        });
      }
    );
    request.on("timeout", () =>
      request.destroy(new Error("hosted control request timed out"))
    );
    request.on("error", reject);
    request.end(body);
  });
}

async function runMalformedIdentifierSuite({
  clusterflux,
  projectDir,
  scope,
  sessionSecret,
  tenant,
  project,
  user,
  suffix,
  securityNode,
  bundle,
  releaseArtifact,
}) {
  const scenarioStartedAt = Date.now();
  const forms = [
    { name: "empty", value: "" },
    { name: "whitespace", value: "   \t" },
    { name: "control", value: "hostile\u0000id" },
    { name: "invalid_format", value: "hostile id!" },
    { name: "oversized", value: "x".repeat(256) },
  ];
  const tokenForms = forms
    .filter((form) => form.name !== "invalid_format")
    .map((form) =>
      form.name === "oversized"
        ? { ...form, value: "x".repeat(257) }
        : form
    );
  assert(
    Buffer.byteLength(
      tokenForms.find((form) => form.name === "oversized").value
    ) > 256,
    "hostile token suite must exceed the signed-nonce and hosted-login byte limit"
  );
  const agent = `hostile-id-agent-${suffix}`;
  const agentIdentityRecord = agentIdentity("strict-hostile-id-agent", agent);
  const processId = `vp-hostile-identifiers-${suffix}`;
  const launchAttempt = `hostile-launch-attempt-${suffix}`;
  const mainTask = `ti:${processId}:main`;
  const taskSpec = {
    tenant,
    project,
    process: processId,
    task_definition: bundle.entryStableId,
    task_instance: mainTask,
    dispatch: {
      kind: "coordinator_node_wasm",
      export: bundle.entryExport,
      abi: "entrypoint_v1",
    },
    environment_id: null,
    environment: null,
    environment_digest: null,
    required_capabilities: [],
    dependency_cache: null,
    source_snapshot: null,
    required_artifacts: [],
    args: [],
    vfs_epoch: 1,
    failure_policy: "fail_fast",
    bundle_digest: bundle.digest,
  };
  const rejected = [];
  const elapsedMs = (started) =>
    Number(process.hrtime.bigint() - started) / 1_000_000;
  const rejectMalformed = async ({
    principal,
    requestVariant,
    fieldPath,
    form,
    send,
    token = false,
  }) => {
    const started = process.hrtime.bigint();
    const response = await send(form.value);
    assert.strictEqual(
      response.type,
      "error",
      `${principal} ${fieldPath} ${form.name} reached the real ${requestVariant} variant but was not rejected: ${JSON.stringify(response)}`
    );
    assert.match(
      response.message,
      token
        ? /malformed external token|token.*invalid|control characters|byte limit|empty|whitespace/i
        : /malformed external identifier|(?:Agent|Artifact|Node|Process|Project|TaskDefinition|TaskInstance|Tenant|User|LaunchAttempt)Id is invalid/i,
      `${principal} ${fieldPath} ${form.name} was not rejected for identifier validation: ${JSON.stringify(response)}`
    );
    assert.doesNotMatch(
      response.message,
      /unknown variant|unknown field|missing field/i,
      `${principal} ${fieldPath} ${form.name} only proved generic deserialization failure`
    );
    const health = await sendHostedControl({ type: "ping" });
    assert.strictEqual(
      health.type,
      "pong",
      `valid traffic failed after hostile ${principal} ${fieldPath} ${form.name}`
    );
    rejected.push({
      principal,
      request_variant: requestVariant,
      field_path: fieldPath,
      fault: form.name,
      expected: token
        ? "structured malformed external token error"
        : "structured malformed external identifier error",
      observed: response.message,
      health_after: health.type,
      duration_ms: elapsedMs(started),
    });
  };

  let agentRegistered = false;
  let processStarted = false;
  try {
    const added = runJson(
      clusterflux,
      [
        "key",
        "add",
        ...scope,
        "--agent",
        agent,
        "--public-key",
        agentIdentityRecord.publicKey,
      ],
      { cwd: projectDir }
    );
    assert.strictEqual(added.command, "key add");
    agentRegistered = true;

    const agentStartCases = [
      ["tenant", "tenant"],
      ["project", "project"],
      ["agent", "actor_agent"],
      ["process", "process"],
      ["launch_attempt", "launch_attempt"],
    ];
    for (const [principal, field] of agentStartCases) {
      for (const form of forms) {
        await rejectMalformed({
          principal: `agent_signed_${principal}`,
          requestVariant: "start_process",
          fieldPath: field,
          form,
          send: async (value) => {
            const body = {
              type: "start_process",
              tenant,
              project,
              actor_agent: agent,
              process: processId,
              launch_attempt: launchAttempt,
              restart: false,
            };
            body[field] = value;
            return sendHostedControl(
              signedAgentWorkflowRequest(agentIdentityRecord, body, {
                nonce: `hostile-agent-${principal}-${form.name}-${Date.now()}`,
              })
            );
          },
        });
      }
    }
    for (const form of tokenForms) {
      await rejectMalformed({
        principal: "agent_signed_nonce",
        requestVariant: "start_process",
        fieldPath: "agent_signature.nonce",
        form,
        token: true,
        send: async (value) => {
          const request = signedAgentWorkflowRequest(
            agentIdentityRecord,
            {
              type: "start_process",
              tenant,
              project,
              actor_agent: agent,
              process: processId,
              launch_attempt: launchAttempt,
              restart: false,
            },
            { nonce: value }
          );
          assert.strictEqual(
            request.agent_signature.nonce,
            value,
            "hostile Agent nonce was not preserved on the wire"
          );
          return sendHostedControl(request);
        },
      });
    }
    for (const form of forms) {
      await rejectMalformed({
        principal: "node_signed_heartbeat_node",
        requestVariant: "node_heartbeat",
        fieldPath: "node",
        form,
        send: async (value) =>
          sendHostedControl({
            type: "node_heartbeat",
            tenant,
            project,
            node: value,
            node_signature: signedNodeHeartbeat(
              tenant,
              project,
              value,
              securityNode.identity,
              {
                nonce: `hostile-heartbeat-node-${form.name}-${Date.now()}`,
              }
            ),
          }),
      });
    }

    const started = await sendHostedControl(
      signedAgentWorkflowRequest(
        agentIdentityRecord,
        {
          type: "start_process",
          tenant,
          project,
          actor_agent: agent,
          process: processId,
          launch_attempt: launchAttempt,
          restart: false,
        },
        { nonce: `hostile-agent-valid-start-${suffix}` }
      )
    );
    assert.strictEqual(started.type, "process_started", JSON.stringify(started));
    processStarted = true;
    taskSpec.vfs_epoch = started.epoch;

    const taskSpecCases = [
      ["tenant", "task_spec.tenant"],
      ["project", "task_spec.project"],
      ["process", "task_spec.process"],
      ["task_definition", "task_spec.task_definition"],
      ["task_instance", "task_spec.task_instance"],
    ];
    for (const [field, fieldPath] of taskSpecCases) {
      for (const form of forms) {
        await rejectMalformed({
          principal: `agent_signed_${field}`,
          requestVariant: "launch_task",
          fieldPath,
          form,
          send: async (value) => {
            const malformedTaskSpec = structuredClone(taskSpec);
            malformedTaskSpec[field] = value;
            const body = {
              type: "launch_task",
              tenant,
              project,
              actor_agent: agent,
              task_spec: malformedTaskSpec,
              wait_for_node: false,
              artifact_path: `/vfs/artifacts/${mainTask}-output.txt`,
              wasm_module_base64: bundle.moduleBase64,
            };
            return sendHostedControl(
              signedAgentWorkflowRequest(agentIdentityRecord, body, {
                nonce: `hostile-task-spec-${field}-${form.name}-${Date.now()}`,
              })
            );
          },
        });
      }
    }

    for (const form of forms) {
      await rejectMalformed({
        principal: "agent_signed_artifact_array",
        requestVariant: "launch_task",
        fieldPath: "task_spec.required_artifacts[0]",
        form,
        send: async (value) => {
          const malformedTaskSpec = structuredClone(taskSpec);
          malformedTaskSpec.required_artifacts = [releaseArtifact.artifact];
          malformedTaskSpec.args = [
            {
              Artifact: {
                id: releaseArtifact.artifact,
                digest: releaseArtifact.digest,
                size_bytes: Number(releaseArtifact.size_bytes || 1),
              },
            },
          ];
          malformedTaskSpec.required_artifacts = [value];
          const body = {
            type: "launch_task",
            tenant,
            project,
            actor_agent: agent,
            task_spec: malformedTaskSpec,
            wait_for_node: false,
            artifact_path: `/vfs/artifacts/${mainTask}-artifact-output.txt`,
            wasm_module_base64: bundle.moduleBase64,
          };
          return sendHostedControl(
            signedAgentWorkflowRequest(agentIdentityRecord, body, {
              nonce: `hostile-artifact-array-${form.name}-${Date.now()}`,
            })
          );
        },
      });
    }

    const mainLaunch = await sendHostedControl(
      signedAgentWorkflowRequest(
        agentIdentityRecord,
        {
          type: "launch_task",
          tenant,
          project,
          actor_agent: agent,
          task_spec: taskSpec,
          wait_for_node: false,
          artifact_path: `/vfs/artifacts/${mainTask}-output.txt`,
          wasm_module_base64: bundle.moduleBase64,
        },
        { nonce: `hostile-agent-valid-main-${suffix}` }
      )
    );
    assert.strictEqual(mainLaunch.type, "main_launched", JSON.stringify(mainLaunch));

    const authenticatedCases = [
      {
        principal: "authenticated_cli_project",
        requestVariant: "select_project",
        fieldPath: "request.project",
        payload: (value) => ({ type: "select_project", project: value }),
      },
      {
        principal: "authenticated_cli_process",
        requestVariant: "list_task_events",
        fieldPath: "request.process",
        payload: (value) => ({ type: "list_task_events", process: value }),
      },
      {
        principal: "authenticated_cli_task",
        requestVariant: "join_task",
        fieldPath: "request.task",
        payload: (value) => ({
          type: "join_task",
          process: processId,
          task: value,
        }),
      },
      {
        principal: "authenticated_cli_artifact",
        requestVariant: "create_artifact_download_link",
        fieldPath: "request.artifact",
        payload: (value) => ({
          type: "create_artifact_download_link",
          artifact: value,
          max_bytes: Number(releaseArtifact.size_bytes || 1),
          ttl_seconds: 60,
        }),
      },
      {
        principal: "authenticated_cli_launch_attempt",
        requestVariant: "abort_process",
        fieldPath: "request.launch_attempt",
        payload: (value) => ({
          type: "abort_process",
          process: processId,
          launch_attempt: value,
        }),
      },
    ];
    for (const testCase of authenticatedCases) {
      for (const form of forms) {
        await rejectMalformed({
          ...testCase,
          form,
          send: async (value) =>
            sendHostedControl(
              authenticatedRequest(sessionSecret, testCase.payload(value))
            ),
        });
      }
    }

    const nodeCases = [
      {
        principal: "node_signed_wrapper_node",
        requestVariant: "poll_task_assignment",
        fieldPath: "node",
        send: (value, form) =>
          sendHostedControl(
            signedNodeRequest(
              value,
              securityNode.identity,
              "poll_task_assignment",
              {
                type: "poll_task_assignment",
                tenant,
                project,
                node: securityNode.node,
              },
              { nonce: `hostile-node-wrapper-${form.name}-${Date.now()}` }
            )
          ),
      },
      {
        principal: "node_signed_poll_tenant",
        requestVariant: "poll_task_assignment",
        fieldPath: "request.tenant",
        send: (value, form) =>
          sendHostedControl(
            signedNodeRequest(
              securityNode.node,
              securityNode.identity,
              "poll_task_assignment",
              {
                type: "poll_task_assignment",
                tenant: value,
                project,
                node: securityNode.node,
              },
              { nonce: `hostile-node-tenant-${form.name}-${Date.now()}` }
            )
          ),
      },
      {
        principal: "node_signed_poll_project",
        requestVariant: "poll_task_assignment",
        fieldPath: "request.project",
        send: (value, form) =>
          sendHostedControl(
            signedNodeRequest(
              securityNode.node,
              securityNode.identity,
              "poll_task_assignment",
              {
                type: "poll_task_assignment",
                tenant,
                project: value,
                node: securityNode.node,
              },
              { nonce: `hostile-node-project-${form.name}-${Date.now()}` }
            )
          ),
      },
      {
        principal: "node_signed_poll_node",
        requestVariant: "poll_task_assignment",
        fieldPath: "request.node",
        send: (value, form) =>
          sendHostedControl(
            signedNodeRequest(
              securityNode.node,
              securityNode.identity,
              "poll_task_assignment",
              {
                type: "poll_task_assignment",
                tenant,
                project,
                node: value,
              },
              { nonce: `hostile-node-inner-${form.name}-${Date.now()}` }
            )
          ),
      },
      {
        principal: "node_signed_artifact_array",
        requestVariant: "report_node_capabilities",
        fieldPath: "request.artifact_locations[0]",
        send: (value, form) =>
          sendHostedControl(
            signedNodeRequest(
              securityNode.node,
              securityNode.identity,
              "report_node_capabilities",
              {
                ...securityNode.capability_body,
                artifact_locations: [value],
              },
              { nonce: `hostile-node-artifact-${form.name}-${Date.now()}` }
            )
          ),
      },
    ];
    for (const testCase of nodeCases) {
      for (const form of forms) {
        await rejectMalformed({
          principal: testCase.principal,
          requestVariant: testCase.requestVariant,
          fieldPath: testCase.fieldPath,
          form,
          send: (value) => testCase.send(value, form),
        });
      }
    }
    for (const form of tokenForms) {
      await rejectMalformed({
        principal: "node_signed_nonce",
        requestVariant: "node_heartbeat",
        fieldPath: "node_signature.nonce",
        form,
        token: true,
        send: async (value) => {
          const request = {
            type: "node_heartbeat",
            tenant,
            project,
            node: securityNode.node,
            node_signature: signedNodeHeartbeat(
              tenant,
              project,
              securityNode.node,
              securityNode.identity,
              { nonce: value }
            ),
          };
          assert.strictEqual(
            request.node_signature.nonce,
            value,
            "hostile Node nonce was not preserved on the wire"
          );
          return sendHostedControl(request);
        },
      });
    }

    const endpointFingerprint = sha256(
      Buffer.from(securityNode.identity.publicKey)
    );
    const rendezvous = {
      type: "request_rendezvous",
      scope: {
        tenant,
        project,
        process: processId,
        object: { Artifact: releaseArtifact.artifact },
        authorization_subject: `${securityNode.node}-self-transfer`,
      },
      source: {
        node: securityNode.node,
        advertised_addr: "127.0.0.1:4433",
        public_key_fingerprint: endpointFingerprint,
      },
      destination: {
        node: securityNode.node,
        advertised_addr: "127.0.0.1:4433",
        public_key_fingerprint: endpointFingerprint,
      },
      direct_connectivity: true,
      failure_reason: "",
    };
    const rendezvousCases = [
      ["scope.tenant", (request, value) => (request.scope.tenant = value)],
      ["scope.project", (request, value) => (request.scope.project = value)],
      ["scope.process", (request, value) => (request.scope.process = value)],
      [
        "scope.object.artifact",
        (request, value) => (request.scope.object.Artifact = value),
      ],
      ["source.node", (request, value) => (request.source.node = value)],
      [
        "destination.node",
        (request, value) => (request.destination.node = value),
      ],
    ];
    for (const [fieldPath, mutate] of rendezvousCases) {
      for (const form of forms) {
        await rejectMalformed({
          principal: "node_signed_rendezvous",
          requestVariant: "request_rendezvous",
          fieldPath: `request.${fieldPath}`,
          form,
          send: async (value) => {
            const body = structuredClone(rendezvous);
            mutate(body, value);
            return sendHostedControl(
              signedNodeRequest(
                securityNode.node,
                securityNode.identity,
                "request_rendezvous",
                body,
                {
                  nonce: `hostile-rendezvous-${fieldPath}-${form.name}-${Date.now()}`,
                }
              )
            );
          },
        });
      }
    }
    const validRendezvous = await sendHostedControl(
      signedNodeRequest(
        securityNode.node,
        securityNode.identity,
        "request_rendezvous",
        rendezvous,
        { nonce: `hostile-rendezvous-valid-${suffix}` }
      )
    );
    assert.strictEqual(
      validRendezvous.type,
      "rendezvous_plan",
      JSON.stringify(validRendezvous)
    );

    const login = await sendHostedLogin({ type: "begin_oidc_browser_login" });
    assert.strictEqual(login.type, "oidc_browser_login_started", JSON.stringify(login));
    const loginCases = [
      ["transaction_id", login.transaction_id, login.polling_secret],
      ["polling_secret", login.transaction_id, login.polling_secret],
    ];
    for (const [field, validTransaction, validSecret] of loginCases) {
      for (const form of tokenForms) {
        await rejectMalformed({
          principal: "private_hosted_login",
          requestVariant: "poll_oidc_browser_login",
          fieldPath: field,
          form,
          token: true,
          send: async (value) =>
            sendHostedLogin({
              type: "poll_oidc_browser_login",
              transaction_id:
                field === "transaction_id" ? value : validTransaction,
              polling_secret: field === "polling_secret" ? value : validSecret,
            }),
        });
        const validPoll = await sendHostedLogin({
          type: "poll_oidc_browser_login",
          transaction_id: login.transaction_id,
          polling_secret: login.polling_secret,
        });
        assert.strictEqual(
          validPoll.type,
          "oidc_browser_login_pending",
          `valid hosted login polling failed after malformed ${field}`
        );
      }
    }

    const operatorControlCases = [
      ["payload.tenant", "tenant"],
      ["payload.project", "project"],
      ["payload.process", "process"],
    ];
    for (const [fieldPath, field] of operatorControlCases) {
      for (const form of forms) {
        await rejectMalformed({
          principal: "private_hosted_operator_control",
          requestVariant: "stop_hosted_process",
          fieldPath,
          form,
          send: async (value) => {
            const payload = {
              type: "stop_hosted_process",
              tenant,
              project,
              process: processId,
            };
            payload[field] = value;
            return sendHostedControlEnvelope({
              type: "hosted_operator_request",
              protocol_version: 1,
              request_id: `hostile-operator-${field}-${form.name}-${Date.now()}`,
              operation: "stop_hosted_process",
              operator_proof: sha256(Buffer.from("invalid operator proof")),
              operator_nonce: `hostile-operator-nonce-${field}-${form.name}-${Date.now()}`,
              issued_at_epoch_seconds: Math.floor(Date.now() / 1000),
              payload,
            });
          },
        });
      }
    }

    const authenticatedHealth = await sendHostedControl(
      authenticatedRequest(sessionSecret, { type: "auth_status" })
    );
    assert.strictEqual(authenticatedHealth.type, "auth_status");
    assert.strictEqual(authenticatedHealth.tenant, tenant);
    assert.strictEqual(authenticatedHealth.project, project);
    assert.strictEqual(authenticatedHealth.actor, user);
  } finally {
    if (processStarted) {
      try {
        runJson(
          clusterflux,
          [
            "process",
            "abort",
            ...scope,
            "--process",
            processId,
            "--yes",
          ],
          { cwd: projectDir }
        );
      } catch (_) {
        // Preserve the primary validation failure while still attempting cleanup.
      }
    }
    if (agentRegistered) {
      try {
        runJson(
          clusterflux,
          ["key", "revoke", ...scope, "--agent", agent, "--yes"],
          { cwd: projectDir }
        );
      } catch (_) {
        // Preserve the primary validation failure while still attempting cleanup.
      }
    }
  }

  const coveredPrincipals = [...new Set(rejected.map((entry) => entry.principal))];
  const coveredVariants = [...new Set(rejected.map((entry) => entry.request_variant))];
  return {
    principals: coveredPrincipals,
    request_variants: coveredVariants,
    forms: forms.map((form) => form.name),
    rejected_requests: rejected,
    all_rejected_for_intended_reason: rejected.every(
      (entry) =>
        /identifier|token|Id is invalid|control characters|byte limit|empty|whitespace/i.test(
          entry.observed
        )
    ),
    valid_after_every_rejection: rejected.every(
      (entry) => entry.health_after === "pong"
    ),
    valid_authenticated_action_after_suite: true,
    valid_signed_rendezvous_after_suite: true,
    valid_private_login_poll_after_every_private_rejection: true,
    duration_ms: Date.now() - scenarioStartedAt,
  };
}

function sendHostedControlDroppingResponse(payload) {
  const body = Buffer.from(
    JSON.stringify(coordinatorWireRequest(payload, "strict-live-dropped-response"))
  );
  const url = new URL("/api/v1/control", serviceEndpoint);
  return new Promise((resolve, reject) => {
    let settled = false;
    const request = https.request(
      url,
      {
        method: "POST",
        headers: {
          "content-type": "application/json",
          "content-length": body.length,
        },
        timeout: 30_000,
      },
      (response) => {
        settled = true;
        const statusCode = response.statusCode;
        response.destroy();
        resolve({ status_code: statusCode, response_body_consumed: false });
      }
    );
    request.on("timeout", () =>
      request.destroy(new Error("hosted dropped-response request timed out"))
    );
    request.on("error", (error) => {
      if (!settled) reject(error);
    });
    request.end(body);
  });
}

function sendHostedLoginStatus(extraHeaders = {}) {
  const body = Buffer.from(
    JSON.stringify(
      coordinatorWireRequest(
        { type: "begin_oidc_browser_login" },
        `strict-live-login-${Date.now()}-${Math.random()}`
      )
    )
  );
  const url = new URL("/api/v1/login", serviceEndpoint);
  return new Promise((resolve, reject) => {
    const request = https.request(
      url,
      {
        method: "POST",
        headers: {
          "content-type": "application/json",
          "content-length": body.length,
          ...extraHeaders,
        },
        timeout: 30_000,
      },
      (response) => {
        const chunks = [];
        response.on("data", (chunk) => chunks.push(chunk));
        response.on("end", () =>
          resolve({
            status_code: response.statusCode,
            body: Buffer.concat(chunks).toString("utf8"),
          })
        );
      }
    );
    request.on("timeout", () =>
      request.destroy(new Error("hosted login request timed out"))
    );
    request.on("error", reject);
    request.end(body);
  });
}

function sendHostedLogin(payload) {
  const body = Buffer.from(
    JSON.stringify(
      coordinatorWireRequest(
        payload,
        `strict-live-login-protocol-${Date.now()}-${Math.random()}`
      )
    )
  );
  const url = new URL("/api/v1/login", serviceEndpoint);
  return new Promise((resolve, reject) => {
    const request = https.request(
      url,
      {
        method: "POST",
        headers: {
          "content-type": "application/json",
          "content-length": body.length,
        },
        timeout: 30_000,
      },
      (response) => {
        const chunks = [];
        response.on("data", (chunk) => chunks.push(chunk));
        response.on("end", () => {
          const responseBody = Buffer.concat(chunks).toString("utf8");
          if (response.statusCode < 200 || response.statusCode >= 300) {
            reject(
              new Error(
                `hosted login HTTP ${response.statusCode}: ${responseBody}`
              )
            );
            return;
          }
          try {
            resolve(JSON.parse(responseBody));
          } catch (error) {
            reject(
              new Error(`hosted login returned invalid JSON: ${error.message}`)
            );
          }
        });
      }
    );
    request.on("timeout", () =>
      request.destroy(new Error("hosted login request timed out"))
    );
    request.on("error", reject);
    request.end(body);
  });
}

async function runHostedLoginIsolationBeforeRestart() {
  const scenarioStartedAt = Date.now();
  const controlBefore = await sendHostedControl({ type: "ping" });
  assert.strictEqual(controlBefore.type, "pong");
  const statuses = [];
  let limited;
  for (let index = 0; index < 128; index += 1) {
    const response = await sendHostedLoginStatus();
    statuses.push(response.status_code);
    if (response.status_code === 429) {
      limited = response;
      break;
    }
    assert.strictEqual(response.status_code, 200, response.body);
  }
  assert(limited, `hosted login route was not rate limited: ${statuses.join(",")}`);
  const spoofed = await sendHostedLoginStatus({
    forwarded: "for=203.0.113.99",
    "x-forwarded-for": "203.0.113.99",
    "x-real-ip": "203.0.113.99",
  });
  assert.strictEqual(
    spoofed.status_code,
    429,
    "caller-supplied forwarding headers bypassed the client rate limit"
  );

  const remoteBody = JSON.stringify(
    coordinatorWireRequest(
      { type: "begin_oidc_browser_login" },
      `strict-vps-login-${Date.now()}`
    )
  );
  const remoteStatus = Number(
    run(
      "ssh",
      sshArgs(
        "curl",
        "--silent",
        "--show-error",
        "--output",
        "/dev/null",
        "--write-out=%{http_code}",
        "-HContent-Type:application/json",
        `--data-binary=${remoteBody}`,
        `${serviceEndpoint}/api/v1/login`
      )
    ).trim()
  );
  assert.strictEqual(
    remoteStatus,
    200,
    "one client exhausted the browser-login allowance for a different client"
  );
  const controlAfter = await sendHostedControl({ type: "ping" });
  assert.strictEqual(controlAfter.type, "pong");
  return {
    scenario_started_at_ms: scenarioStartedAt,
    local_statuses: statuses,
    local_limited_status: limited.status_code,
    forwarding_spoof_status: spoofed.status_code,
    independent_vps_client_status: remoteStatus,
    control_route_before: controlBefore.type,
    control_route_after: controlAfter.type,
  };
}

async function finishHostedLoginIsolationAfterRestart(evidence) {
  const control = await sendHostedControl({ type: "ping" });
  assert.strictEqual(control.type, "pong");
  const response = await sendHostedLoginStatus();
  assert(
    [200, 429].includes(response.status_code),
    `login route did not recover after service restart: ${response.status_code} ${response.body}`
  );
  evidence.after_service_restart = {
    control_route: control.type,
    login_route_status: response.status_code,
  };
  evidence.duration_ms = Date.now() - evidence.scenario_started_at_ms;
  delete evidence.scenario_started_at_ms;
  return evidence;
}

function assertDenied(response, pattern, label) {
  assert.strictEqual(response.type, "error", `${label}: ${JSON.stringify(response)}`);
  assert.match(response.message, pattern, `${label}: ${JSON.stringify(response)}`);
  return {
    denied: true,
    category: response.error?.category || null,
    message: response.message,
  };
}

function assertDeniedOrEmptyCollection(response, pattern, label) {
  const emptyCollectionField = {
    process_statuses: "processes",
    task_events: "events",
  }[response.type];
  if (!emptyCollectionField) {
    return assertDenied(response, pattern, label);
  }
  assert.deepStrictEqual(
    Object.keys(response).sort(),
    ["actor", emptyCollectionField, "type"]
      .filter((key) => key !== "actor" || Object.hasOwn(response, key))
      .sort(),
    `${label}: ${JSON.stringify(response)}`
  );
  assert.deepStrictEqual(
    response[emptyCollectionField],
    [],
    `${label}: ${JSON.stringify(response)}`
  );
  return {
    denied: false,
    scoped_empty_result: true,
    response_type: response.type,
  };
}

function readNodeCredential(projectDir, node) {
  const directory = path.join(projectDir, ".clusterflux", "nodes");
  for (const file of fs.readdirSync(directory)) {
    if (!file.endsWith(".json")) continue;
    const absolute = path.join(directory, file);
    const credential = readJson(absolute);
    if (credential.node === node) return { absolute, credential };
  }
  throw new Error(`persisted credential for ${node} was not found`);
}

function directoryStats(root) {
  if (!fs.existsSync(root)) return { files: 0, bytes: 0 };
  let files = 0;
  let bytes = 0;
  const visit = (directory) => {
    for (const entry of fs.readdirSync(directory, { withFileTypes: true })) {
      const absolute = path.join(directory, entry.name);
      if (entry.isDirectory()) visit(absolute);
      else if (entry.isFile()) {
        files += 1;
        bytes += fs.statSync(absolute).size;
      }
    }
  };
  visit(root);
  return { files, bytes };
}

function shellQuote(argument) {
  return `'${String(argument).replaceAll("'", "'\"'\"'")}'`;
}

function sshArgs(...remoteArgs) {
  assert(strictVpsHost && strictVpsIdentity);
  return [
    "-i",
    path.resolve(strictVpsIdentity.replace(/^~(?=\/)/, os.homedir())),
    "-o",
    "BatchMode=yes",
    strictVpsHost,
    remoteArgs.map(shellQuote).join(" "),
  ];
}

function strictInfrastructureSample(projectDir) {
  const memory = Object.fromEntries(
    run(
    "ssh",
    sshArgs(
      "systemctl",
      "show",
      strictServiceUnit,
      "--property=MemoryCurrent",
      "--property=MemoryPeak"
    )
  )
    .trim()
    .split(/\r?\n/)
    .map((line) => line.split("="))
  );
  const serviceDisk = Number(
    run(
      "ssh",
      sshArgs("du", "-sb", "/var/lib/clusterflux-public-release")
    )
      .trim()
      .split(/\s+/)[0]
  );
  const database = run(
    "ssh",
    sshArgs(
      "runuser",
      "-u",
      "postgres",
      "--",
      "psql",
      "-d",
      "clusterflux",
      "-AtF,"
    ),
    {
      input:
        "SELECT pg_database_size('clusterflux'), (SELECT count(*) FROM clusterflux_tenants), (SELECT count(*) FROM clusterflux_projects), (SELECT count(*) FROM clusterflux_node_identities), (SELECT count(*) FROM clusterflux_cli_sessions);\n",
      stdio: ["pipe", "pipe", "pipe"],
    }
  )
    .trim()
    .split(",")
    .map(Number);
  assert(Number.isFinite(Number(memory.MemoryCurrent)));
  assert(Number.isFinite(Number(memory.MemoryPeak)));
  assert.strictEqual(database.length, 5);
  return {
    sampled_at_epoch_ms: Date.now(),
    service_memory_current_bytes: Number(memory.MemoryCurrent),
    service_memory_peak_bytes: Number(memory.MemoryPeak),
    service_state_disk_bytes: serviceDisk,
    postgres_database_bytes: database[0],
    durable_rows: {
      tenants: database[1],
      projects: database[2],
      node_identities: database[3],
      cli_sessions: database[4],
    },
    local_node_state: directoryStats(path.join(projectDir, ".clusterflux")),
  };
}

function executable(root, name) {
  return path.join(root, "bin", process.platform === "win32" ? `${name}.exe` : name);
}

function binaryArchive(manifest) {
  const platform = `${os.platform()}-${os.arch()}`;
  const asset = manifest.assets.find(
    (candidate) =>
      candidate.name.startsWith("clusterflux-public-binaries-") &&
      candidate.name.includes(platform)
  );
  assert(asset, `missing released binary archive for ${platform}`);
  assert(fs.existsSync(asset.file), `missing released binary archive ${asset.file}`);
  return asset.file;
}

function publicRepoUrl(manifest) {
  const url =
    process.env.CLUSTERFLUX_PUBLIC_REPO_URL ||
    process.env.CLUSTERFLUX_PUBLIC_REPO_REMOTE ||
    manifest.public_repo_url ||
    manifest.public_repo_remote;
  assert(url, "public repository URL is required");
  assert(url.includes("git.michelpaulissen.com"));
  return url;
}

function stagePublicCheckout(manifest, checkout) {
  const candidateTree = process.env.CLUSTERFLUX_LIVE_PUBLIC_TREE;
  if (candidateTree) {
    const source = path.resolve(candidateTree);
    assert.strictEqual(
      source,
      path.resolve(manifest.public_tree),
      "explicit live candidate tree must be the tree recorded by the release manifest"
    );
    fs.cpSync(source, checkout, { recursive: true });
    return {
      kind: "local_filtered_release_candidate",
      path: source,
      published: false,
    };
  }
  run("git", ["clone", "--depth", "1", publicRepoUrl(manifest), checkout]);
  return {
    kind: "published_forgejo_checkout",
    url: publicRepoUrl(manifest),
    published: true,
  };
}

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

async function waitForCli(label, action, predicate, timeoutMs = 10 * 60 * 1000) {
  const deadline = Date.now() + timeoutMs;
  let lastValue;
  let lastError;
  while (Date.now() < deadline) {
    try {
      lastValue = action();
      if (predicate(lastValue)) return lastValue;
      lastError = undefined;
    } catch (error) {
      lastError = error;
    }
    await delay(250);
  }
  throw new Error(
    `${label} timed out; last value=${JSON.stringify(lastValue)}${
      lastError ? `; last error=${lastError.message}` : ""
    }`
  );
}

function spawnJsonLines(command, args, options) {
  recordCommand(command, args);
  const child = cp.spawn(command, args, {
    ...options,
    stdio: ["ignore", "pipe", "pipe"],
  });
  let stdout = "";
  let stderr = "";
  const values = [];
  const waiters = [];

  function deliver(value) {
    values.push(value);
    for (let index = waiters.length - 1; index >= 0; index -= 1) {
      const waiter = waiters[index];
      if (waiter.predicate(value)) {
        waiters.splice(index, 1);
        clearTimeout(waiter.timer);
        waiter.resolve(value);
      }
    }
  }

  child.stdout.on("data", (chunk) => {
    stdout += chunk.toString();
    while (stdout.includes("\n")) {
      const newline = stdout.indexOf("\n");
      const line = stdout.slice(0, newline).trim();
      stdout = stdout.slice(newline + 1);
      if (!line) continue;
      try {
        deliver(JSON.parse(line));
      } catch (_) {
        // Non-JSON progress stays out of the evidence report.
      }
    }
  });
  child.stderr.on("data", (chunk) => {
    stderr += chunk.toString();
  });

  function waitFor(predicate, label, timeoutMs = 120000) {
    const existing = values.find(predicate);
    if (existing) return Promise.resolve(existing);
    return new Promise((resolve, reject) => {
      const waiter = { predicate, resolve, reject, timer: undefined };
      waiter.timer = setTimeout(() => {
        const index = waiters.indexOf(waiter);
        if (index >= 0) waiters.splice(index, 1);
        reject(new Error(`${label} timed out\n${stderr}`));
      }, timeoutMs);
      waiters.push(waiter);
    });
  }

  child.once("exit", (code) => {
    for (const waiter of waiters.splice(0)) {
      clearTimeout(waiter.timer);
      waiter.reject(new Error(`${waiter.label || "child"} exited with ${code}\n${stderr}`));
    }
  });
  return { child, values, waitFor, stderr: () => stderr };
}

async function stopChild(child) {
  if (!child || child.exitCode !== null) return;
  child.kill("SIGTERM");
  const exited = new Promise((resolve) => child.once("exit", resolve));
  const forced = delay(5000).then(() => {
    if (child.exitCode === null) child.kill("SIGKILL");
  });
  await Promise.race([exited, forced]);
  if (child.exitCode === null) await exited;
}

function rawTaskEvents(report) {
  return report?.events?.response?.events || [];
}

function nodeDescriptor(status, node) {
  return status?.response?.descriptors?.find(
    (descriptor) => descriptor.id === node || descriptor.node === node
  );
}

function completedFlagship(events) {
  const completedDefinitions = new Set(
    events
      .filter((event) => event.terminal_state === "completed")
      .map((event) => event.task_definition)
  );
  return (
    completedDefinitions.has("prepare_source") &&
    completedDefinitions.has("compile_linux") &&
    completedDefinitions.has("package_release") &&
    events.some(
      (event) =>
        event.executor === "coordinator_main" && event.terminal_state === "completed"
    )
  );
}

function contentAddressedArtifactId(event, logicalName) {
  const digest = event?.artifact_digest;
  if (typeof digest !== "string" || !/^sha256:[0-9a-f]{64}$/.test(digest)) {
    return null;
  }
  const artifactId = `${logicalName}-${digest.slice("sha256:".length)}`;
  return event.artifact_path?.endsWith(`/vfs/artifacts/${artifactId}`)
    ? artifactId
    : null;
}

async function prepareLiveNodeCredentialSecurity({
  clusterflux,
  projectDir,
  scope,
  tenant,
  project,
  suffix,
}) {
  const startedAt = Date.now();
  const node = `security-node-${suffix}`;
  const grant = runJson(clusterflux, ["node", "enroll", ...scope], {
    cwd: projectDir,
  });
  const attach = runJson(
    clusterflux,
    [
      "node",
      "attach",
      "--coordinator",
      serviceEndpoint,
      "--tenant",
      tenant,
      "--project-id",
      project,
      "--node",
      node,
      "--enrollment-grant",
      grant.enrollment_grant.grant,
      "--json",
    ],
    { cwd: projectDir }
  );
  assert.strictEqual(attach.boundary.used_enrollment_exchange, true);
  const stored = readNodeCredential(projectDir, node);
  const identity = nodeIdentityFromPrivateKey(stored.credential.private_key);
  assert.strictEqual(identity.publicKey, stored.credential.public_key);

  const heartbeat = {
    type: "node_heartbeat",
    tenant,
    project,
    node,
    node_signature: signedNodeHeartbeat(tenant, project, node, identity, {
      nonce: `strict-valid-${suffix}`,
    }),
  };
  const valid = await sendHostedControl(heartbeat);
  assert.strictEqual(valid.type, "node_heartbeat");
  const replay = assertDenied(
    await sendHostedControl(heartbeat),
    /nonce.*already.*used|replay/i,
    "replayed node credential"
  );

  const expired = assertDenied(
    await sendHostedControl({
      type: "node_heartbeat",
      tenant,
      project,
      node,
      node_signature: signedNodeHeartbeat(tenant, project, node, identity, {
        nonce: `strict-expired-${suffix}`,
        issuedAtEpochSeconds: Math.floor(Date.now() / 1000) - 3600,
      }),
    }),
    /expired|clock skew/i,
    "expired node credential"
  );

  const forgedIdentity = nodeIdentity("strict-forged-node", node);
  const forged = assertDenied(
    await sendHostedControl({
      type: "node_heartbeat",
      tenant,
      project,
      node,
      node_signature: signedNodeHeartbeat(tenant, project, node, forgedIdentity, {
        nonce: `strict-forged-${suffix}`,
      }),
    }),
    /signature|public key/i,
    "forged node credential"
  );

  const originalCapabilityBody = {
    type: "report_node_capabilities",
    tenant,
    project,
    node,
    capabilities: {
      os: "Linux",
      arch: process.arch === "x64" ? "x86_64" : process.arch,
      capabilities: [],
      environment_backends: [],
      source_providers: [],
    },
    cached_environment_digests: [],
    dependency_cache_digests: [],
    source_snapshots: [],
    artifact_locations: [],
    direct_connectivity: false,
    online: true,
  };
  const modifiedEnvelope = signedNodeRequest(
    node,
    identity,
    "report_node_capabilities",
    originalCapabilityBody,
    { nonce: `strict-modified-${suffix}` }
  );
  modifiedEnvelope.request.online = false;
  const bodyModified = assertDenied(
    await sendHostedControl(modifiedEnvelope),
    /signature/i,
    "body-modified node credential"
  );

  return {
    duration_ms: Date.now() - startedAt,
    node,
    identity,
    credential_path: stored.absolute,
    credential_digest: sha256(fs.readFileSync(stored.absolute)),
    capability_body: originalCapabilityBody,
    evidence: {
      valid_request: valid.type,
      replay,
      expired,
      forged,
      body_modified: bodyModified,
    },
  };
}

async function runLiveAgentCredentialSecurity({
  clusterfluxDap,
  clusterflux,
  projectDir,
  scope,
  sessionSecret,
  tenant,
  project,
  suffix,
  bundle,
  securityNode,
  workerRuntime,
}) {
  const scenarioStartedAt = Date.now();
  const agent = `security-agent-${suffix}`;
  const identity = agentIdentity("strict-live-agent", agent);
  const added = runJson(
    clusterflux,
    ["key", "add", ...scope, "--agent", agent, "--public-key", identity.publicKey],
    { cwd: projectDir }
  );
  assert.strictEqual(added.command, "key add");

  const processId = `vp-agent-security-${suffix}`;
  const unsigned = {
    type: "start_process",
    tenant,
    project,
    actor_agent: agent,
    process: processId,
    restart: false,
  };
  const validRequest = signedAgentWorkflowRequest(identity, unsigned, {
    nonce: `strict-agent-valid-${suffix}`,
  });
  const valid = await sendHostedControl(validRequest);
  assert.strictEqual(valid.type, "process_started", JSON.stringify(valid));
  const replay = assertDenied(
    await sendHostedControl(validRequest),
    /nonce.*already.*used|replay/i,
    "replayed agent credential"
  );

  const expired = assertDenied(
    await sendHostedControl(
      signedAgentWorkflowRequest(
        identity,
        { ...unsigned, process: `${processId}-expired` },
        {
          nonce: `strict-agent-expired-${suffix}`,
          issuedAtEpochSeconds: Math.floor(Date.now() / 1000) - 3600,
        }
      )
    ),
    /expired|clock skew/i,
    "expired agent credential"
  );

  const modifiedRequest = signedAgentWorkflowRequest(
    identity,
    { ...unsigned, process: `${processId}-modified` },
    { nonce: `strict-agent-modified-${suffix}` }
  );
  modifiedRequest.restart = true;
  const bodyModified = assertDenied(
    await sendHostedControl(modifiedRequest),
    /signature/i,
    "body-modified agent credential"
  );

  const forgedIdentity = agentIdentity("strict-live-forged-agent", agent);
  const forged = assertDenied(
    await sendHostedControl(
      signedAgentWorkflowRequest(
        forgedIdentity,
        { ...unsigned, process: `${processId}-forged` },
        {
          nonce: `strict-agent-forged-${suffix}`,
          publicKeyFingerprint: identity.publicKeyFingerprint,
        }
      )
    ),
    /signature|public key/i,
    "forged agent credential"
  );

  const fakeEnvironmentDigest = sha256(
    Buffer.from(`strict-debug-missing-participant-${suffix}`)
  );
  const fakeCapabilityBody = {
    ...securityNode.capability_body,
    cached_environment_digests: [fakeEnvironmentDigest],
    online: true,
  };
  const mainTask = `ti:${processId}:main`;
  const fakeTask = `missing-debug-participant-${suffix}`;
  const mainLaunchBody = {
    type: "launch_task",
    tenant,
    project,
    actor_agent: agent,
    task_spec: {
      tenant,
      project,
      process: processId,
      task_definition: bundle.entryStableId,
      task_instance: mainTask,
      dispatch: {
        kind: "coordinator_node_wasm",
        export: bundle.entryExport,
        abi: "entrypoint_v1",
      },
      environment_id: null,
      environment: null,
      environment_digest: null,
      required_capabilities: [],
      dependency_cache: null,
      source_snapshot: null,
      required_artifacts: [],
      args: [],
      vfs_epoch: valid.epoch,
      bundle_digest: bundle.digest,
    },
    wait_for_node: true,
    artifact_path: `/vfs/artifacts/${mainTask}-output.txt`,
    wasm_module_base64: bundle.moduleBase64,
  };
  let mainLaunch;
  let directTaskV1;
  let fakeLaunch;
  let liveParentTask;
  let liveParentAssignment;
  let realDebugLaunch;
  let preconfiguredBreakpoint;
  let attachSourcePath;
  let attachBreakpointLine;
  const realDebugTask = `real-debug-participant-${suffix}`;
  let workerPaused = false;
  const baselineContainers = new Set(
    run("podman", ["ps", "-q"])
      .trim()
      .split(/\s+/)
      .filter(Boolean)
  );
  try {
    mainLaunch = await sendHostedControl(
      signedAgentWorkflowRequest(identity, mainLaunchBody, {
        nonce: `strict-agent-main-${suffix}`,
      })
    );
    assert.strictEqual(mainLaunch.type, "main_launched", JSON.stringify(mainLaunch));
    assert.strictEqual(mainLaunch.task_instance, mainTask);
    assert.strictEqual(mainLaunch.state, "running");
    assert.strictEqual(mainLaunch.actor.kind, "agent");
    assert.strictEqual(mainLaunch.actor.authenticated_without_browser, true);

    liveParentAssignment = await workerRuntime.worker.waitFor(
      (value) => {
        const taskSpec = value.task_assignment_response?.task_spec;
        return (
          value.node_status === "assignment_started" &&
          value.node === workerRuntime.node &&
          value.process === processId &&
          taskSpec?.dispatch?.abi === "task_v1" &&
          taskSpec.task_definition === "compile_linux" &&
          taskSpec.environment_id === "linux" &&
          typeof taskSpec.environment_digest === "string" &&
          typeof taskSpec.source_snapshot === "string"
        );
      },
      "agent main to dispatch an environment-bound live child to the real worker",
      120000
    );
    liveParentTask = liveParentAssignment.virtual_thread;
    assert.strictEqual(typeof liveParentTask, "string");
    assert.notStrictEqual(liveParentTask, "");
    assert.strictEqual(
      workerRuntime.child.kill("SIGSTOP"),
      true,
      "failed to pause the real worker after observing its live child"
    );
    workerPaused = true;

    const fakeCapabilities = await sendHostedControl(
      signedNodeRequest(
        securityNode.node,
        securityNode.identity,
        "report_node_capabilities",
        fakeCapabilityBody,
        { nonce: `strict-agent-fake-capabilities-${suffix}` }
      )
    );
    assert.strictEqual(
      fakeCapabilities.type,
      "node_capabilities_recorded",
      JSON.stringify(fakeCapabilities)
    );

    const fakeTaskSpec = {
    tenant,
    project,
    process: processId,
    task_definition: "task_add_one",
    task_instance: fakeTask,
    dispatch: {
      kind: "coordinator_node_wasm",
      export: null,
      abi: "task_v1",
    },
    environment_id: "strict-debug-missing-participant",
    environment: null,
    environment_digest: fakeEnvironmentDigest,
    required_capabilities: [],
    dependency_cache: null,
    source_snapshot: null,
    required_artifacts: [],
    args: [{ SmallJson: 41 }],
    vfs_epoch: valid.epoch,
    bundle_digest: bundle.digest,
    };
    directTaskV1 = assertDenied(
      await sendHostedControl(
        signedAgentWorkflowRequest(
          identity,
          {
            type: "launch_task",
            tenant,
            project,
            actor_agent: agent,
            task_spec: fakeTaskSpec,
            wait_for_node: false,
            artifact_path: `/vfs/artifacts/${fakeTask}.txt`,
            wasm_module_base64: bundle.moduleBase64,
          },
          { nonce: `strict-agent-direct-task-v1-${suffix}` }
        )
      ),
      /external callers.*EntrypointV1|TaskV1 requires an authenticated live parent/i,
      "external direct TaskV1 launch"
    );

    const liveParentSpec =
      liveParentAssignment.task_assignment_response.task_spec;
    attachSourcePath = path.join(projectDir, "src/lib.rs");
    const attachSourceLines = fs
      .readFileSync(attachSourcePath, "utf8")
      .split(/\r?\n/);
    attachBreakpointLine =
      attachSourceLines.findIndex((line) => line.includes("fn abort_probe(")) + 1;
    assert(
      attachBreakpointLine > 0,
      "agent-security fixture omitted the abort_probe debug probe"
    );
    preconfiguredBreakpoint = await sendHostedControl(
      authenticatedRequest(sessionSecret, {
        type: "set_debug_breakpoints",
        process: processId,
        revision: 1,
        probe_symbols: ["clusterflux.probe.abort_probe"],
      })
    );
    assert.strictEqual(
      preconfiguredBreakpoint.type,
      "debug_breakpoints",
      JSON.stringify(preconfiguredBreakpoint)
    );
    assert.deepStrictEqual(preconfiguredBreakpoint.probe_symbols, [
      "clusterflux.probe.abort_probe",
    ]);
    realDebugLaunch = await sendHostedControl(
      signedNodeRequest(
        workerRuntime.node,
        workerRuntime.identity,
        "launch_child_task",
        {
          type: "launch_child_task",
          tenant,
          project,
          process: processId,
          node: workerRuntime.node,
          parent_task: liveParentTask,
          task_spec: {
            tenant,
            project,
            process: processId,
            task_definition: "abort_probe",
            task_instance: realDebugTask,
            dispatch: {
              kind: "coordinator_node_wasm",
              export: null,
              abi: "task_v1",
            },
            environment_id: liveParentSpec.environment_id,
            environment: liveParentSpec.environment,
            environment_digest: liveParentSpec.environment_digest,
            required_capabilities: ["Command"],
            dependency_cache: null,
            source_snapshot: liveParentSpec.source_snapshot,
            required_artifacts: [],
            args: liveParentSpec.args,
            vfs_epoch: valid.epoch,
            bundle_digest: bundle.digest,
          },
          wait_for_node: false,
          artifact_path: `/vfs/artifacts/${realDebugTask}.txt`,
          wasm_module_base64: bundle.moduleBase64,
        },
        { nonce: `strict-real-debug-participant-${suffix}` }
      )
    );
    assert.strictEqual(
      realDebugLaunch.type,
      "task_launched",
      JSON.stringify(realDebugLaunch)
    );

    const fakeLaunchBody = {
      type: "launch_child_task",
      tenant,
      project,
      process: processId,
      node: workerRuntime.node,
      parent_task: liveParentTask,
      task_spec: fakeTaskSpec,
      wait_for_node: false,
      artifact_path: `/vfs/artifacts/${fakeTask}.txt`,
      wasm_module_base64: bundle.moduleBase64,
    };
    fakeLaunch = await sendHostedControl(
      signedNodeRequest(
        workerRuntime.node,
        workerRuntime.identity,
        "launch_child_task",
        fakeLaunchBody,
        { nonce: `strict-parent-runtime-fake-task-${suffix}` }
      )
    );
    assert.strictEqual(fakeLaunch.type, "task_launched", JSON.stringify(fakeLaunch));
    assert.strictEqual(fakeLaunch.placement.node, securityNode.node);
  } finally {
    if (workerPaused) {
      assert.strictEqual(
        workerRuntime.child.kill("SIGCONT"),
        true,
        "failed to resume the real worker after authenticated child launch"
      );
    }
  }
  await workerRuntime.worker.waitFor(
    (value) =>
      value.node_status === "assignment_started" &&
      value.virtual_thread === realDebugTask,
    "real Podman debug participant to start",
    120_000
  );
  let realDebugContainerIds = new Set();

  const debugClient = new DapClient({
    cwd: projectDir,
    command: clusterfluxDap,
    args: [],
    env: { ...process.env },
  });
  const initialize = debugClient.send("initialize", {
    adapterID: "clusterflux",
    linesStartAt1: true,
    columnsStartAt1: true,
  });
  await debugClient.response(initialize, "initialize");
  const attachRequest = debugClient.send("attach", {
    entry: "build",
    project: projectDir,
    processId,
    runtimeBackend: "live-services",
    coordinatorEndpoint: serviceEndpoint,
  });
  await debugClient.response(attachRequest, "attach");
  await debugClient.waitFor(
    (message) => message.type === "event" && message.event === "initialized"
  );
  const attachBreakpointRequest = debugClient.send("setBreakpoints", {
    source: { path: attachSourcePath },
    breakpoints: [{ line: attachBreakpointLine }],
  });
  const attachBreakpointResponse = await debugClient.response(
    attachBreakpointRequest,
    "setBreakpoints"
  );
  assert.strictEqual(
    attachBreakpointResponse.body.breakpoints[0].verified,
    false,
    "attach breakpoint was verified before runtime installation"
  );
  const configurationDone = debugClient.send("configurationDone");
  await debugClient.response(configurationDone, "configurationDone");
  const attachBreakpointInstalled = await debugClient.waitFor(
    (message) =>
      message.type === "event" &&
      message.event === "breakpoint" &&
      message.body?.breakpoint?.verified === true,
    120_000
  );
  assert.strictEqual(
    attachBreakpointInstalled.body.breakpoint.line,
    attachBreakpointLine
  );
  const attachBreakpointStop = await debugClient.waitFor(
    (message) =>
      message.seq > attachBreakpointInstalled.seq &&
      message.type === "event" &&
      message.event === "stopped" &&
      message.body.reason === "breakpoint",
    120_000
  );
  const attachStackRequest = debugClient.send("stackTrace", {
    threadId: attachBreakpointStop.body.threadId,
    startFrame: 0,
    levels: 1,
  });
  const attachStackFrames = (
    await debugClient.response(attachStackRequest, "stackTrace")
  ).body.stackFrames;
  assert.strictEqual(
    attachStackFrames[0].line,
    attachBreakpointLine,
    "attach stopped somewhere other than the installed configured breakpoint"
  );
  const attachContinue = debugClient.send("continue", {
    threadId: attachBreakpointStop.body.threadId,
  });
  const attachContinueResponse = await debugClient.response(
    attachContinue,
    "continue"
  );
  const realDebugContainerDeadline = Date.now() + 120_000;
  let lastRealDebugPodmanStates = [];
  while (Date.now() < realDebugContainerDeadline) {
    const containers = runJson("podman", ["ps", "--format", "json"]);
    lastRealDebugPodmanStates = containers;
    realDebugContainerIds = new Set(
      containers
        .map((container) => container.Id || container.ID || container.IdHex || "")
        .filter((id) => id && !baselineContainers.has(id))
    );
    if (realDebugContainerIds.size > 0) break;
    await delay(100);
  }
  assert(
    realDebugContainerIds.size > 0,
    `continued real debug participant did not start a Podman container: ${JSON.stringify(
      lastRealDebugPodmanStates
    )}`
  );
  const attachDeadline = Date.now() + 120_000;
  let attachedThreads = [];
  while (Date.now() < attachDeadline) {
    const threadsRequest = debugClient.send("threads");
    attachedThreads = (
      await debugClient.response(threadsRequest, "threads")
    ).body.threads;
    if (attachedThreads.length > 0) break;
    await delay(100);
  }
  assert(attachedThreads.length > 0, "DAP attach reported no live threads");

  const freezeStartedAt = Date.now();
  const pauseRequest = debugClient.send("pause", {
    threadId: attachedThreads[0].id,
  });
  await debugClient.response(pauseRequest, "pause");
  const dapPartialStop = await debugClient.waitFor(
    (message) =>
      message.type === "event" &&
      message.seq > attachContinueResponse.seq &&
      message.event === "stopped" &&
      message.body.reason === "pause",
    30_000
  );
  assert.strictEqual(dapPartialStop.body.allThreadsStopped, false);
  const epochRequest = debugClient.send("evaluate", {
    expression: "debug_epoch",
    context: "watch",
  });
  const freeze = {
    epoch: Number(
      (await debugClient.response(epochRequest, "evaluate")).body.result
    ),
  };
  assert(Number.isSafeInteger(freeze.epoch) && freeze.epoch > 0);
  const partialFreeze = await sendHostedControl(
    authenticatedRequest(sessionSecret, {
      type: "inspect_debug_epoch",
      process: processId,
      epoch: freeze.epoch,
    })
  );
  assert.strictEqual(partialFreeze.type, "debug_epoch_status");
  assert.strictEqual(partialFreeze.partially_frozen, true, JSON.stringify(partialFreeze));
  assert.strictEqual(partialFreeze.fully_frozen, false);
  assert.strictEqual(partialFreeze.failed, true);
  assert.match(
    partialFreeze.failure_messages.join("; "),
    /did not acknowledge frozen state within \d+ ms/
  );
  const podmanStates = runJson("podman", ["ps", "--all", "--format", "json"]);
  const pausedContainers = podmanStates.filter((container) => {
    const id = container.Id || container.ID || container.IdHex || "";
    const state = String(container.State || container.Status || "").toLowerCase();
    return realDebugContainerIds.has(id) && state.includes("paused");
  });
  assert(
    pausedContainers.length > 0,
    `partial Debug Epoch did not pause a real Podman container: ${JSON.stringify(
      podmanStates
    )}`
  );
  const pausedContainerIds = pausedContainers.map(
    (container) => container.Id || container.ID || container.IdHex
  );

  const mismatchedDigest = sha256(Buffer.from(`strict-nested-mismatch-${suffix}`));
  const mismatchedChild = await sendHostedControl(
    signedNodeRequest(
      securityNode.node,
      securityNode.identity,
      "launch_child_task",
      {
        type: "launch_child_task",
        tenant,
        project,
        process: processId,
        node: securityNode.node,
        parent_task: fakeTask,
        task_spec: {
          tenant,
          project,
          process: processId,
          task_definition: "task_add_one",
          task_instance: `${fakeTask}:child:mismatched-environment`,
          dispatch: {
            kind: "coordinator_node_wasm",
            export: null,
            abi: "task_v1",
          },
          environment_id: "strict-nested-mismatch",
          environment: null,
          environment_digest: mismatchedDigest,
          required_capabilities: [],
          dependency_cache: null,
          source_snapshot: null,
          required_artifacts: [],
          args: [{ SmallJson: 41 }],
          vfs_epoch: valid.epoch,
          bundle_digest: bundle.digest,
        },
        wait_for_node: false,
        artifact_path: `/vfs/artifacts/${fakeTask}-mismatched-child.txt`,
        wasm_module_base64: bundle.moduleBase64,
      },
      { nonce: `strict-nested-mismatch-${suffix}` }
    )
  );
  const nestedEnvironmentMismatch = assertDenied(
    mismatchedChild,
    /environment|digest|compatible|placement|no node/i,
    "nested environment mismatch"
  );

  const continueStartedAt = Date.now();
  const continueRequest = debugClient.send("continue", {
    threadId: dapPartialStop.body.threadId,
  });
  await debugClient.response(continueRequest, "continue");
  const continueResponseMs = Date.now() - continueStartedAt;
  assert(
    continueResponseMs < 2_000,
    `partial-freeze continue blocked for ${continueResponseMs} ms`
  );
  await delay(1000);
  const resumed = await sendHostedControl(
    authenticatedRequest(sessionSecret, {
      type: "inspect_debug_epoch",
      process: processId,
      epoch: freeze.epoch,
    })
  );
  assert.strictEqual(resumed.type, "debug_epoch_status");
  const resumedAcknowledgements = resumed.acknowledgements.filter(
    (acknowledgement) => acknowledgement.state === "running"
  );
  assert(resumedAcknowledgements.length > 0, JSON.stringify(resumed));
  const resumedPodmanStates = runJson("podman", ["ps", "--format", "json"]);
  assert(
    !resumedPodmanStates.some((container) => {
      const id = container.Id || container.ID || container.IdHex || "";
      const state = String(container.State || container.Status || "").toLowerCase();
      return pausedContainerIds.includes(id) && state.includes("paused");
    }),
    `DAP continue left a Clusterflux container paused: ${JSON.stringify(
      resumedPodmanStates
    )}`
  );
  await debugClient.close();

  const mainBeforeChildStartedAt = Date.now();
  const completed = await waitForCli(
    "agent-authenticated coordinator main and child workflow completion",
    () =>
      runJson(clusterflux, ["task", "list", ...scope, "--process", processId], {
        cwd: projectDir,
      }),
    (tasks) => completedFlagship(rawTaskEvents(tasks)),
    120000
  );
  const completedEvents = rawTaskEvents(completed);
  const concurrentCompileTasks = [
    ...new Set(
      completedEvents
        .filter(
          (event) =>
            event.task_definition === "compile_linux" &&
            event.terminal_state === "completed"
        )
        .map((event) => event.task)
    ),
  ];
  assert(concurrentCompileTasks.length >= 2, JSON.stringify(completedEvents));
  const mainCompletion = completedEvents.find(
    (event) =>
      event.executor === "coordinator_main" &&
      event.terminal_state === "completed"
  );
  assert(mainCompletion, "real coordinator main did not complete");
  assert(
    !completedEvents.some(
      (event) =>
        event.task === fakeTask && typeof event.terminal_state === "string"
    ),
    "the deliberately delayed final child completed before the coordinator main"
  );
  const activeAfterMain = runJson(
    clusterflux,
    ["process", "status", ...scope, "--process", processId],
    { cwd: projectDir }
  );
  assert.notStrictEqual(activeAfterMain.state, "not_active");
  assert.equal(activeAfterMain.live_process.main_state, null);
  const debugStateAfterMain = await sendHostedControl(
    authenticatedRequest(sessionSecret, {
      type: "inspect_debug_breakpoints",
      process: processId,
    })
  );
  assert.strictEqual(
    debugStateAfterMain.type,
    "debug_breakpoints",
    `main completion cleared active-child debug state: ${JSON.stringify(
      debugStateAfterMain
    )}`
  );
  const finalChildCompletion = await sendHostedControl(
    signedNodeRequest(
      securityNode.node,
      securityNode.identity,
      "task_completed",
      {
        type: "task_completed",
        tenant,
        project,
        process: processId,
        node: securityNode.node,
        task: fakeTask,
        terminal_state: "completed",
        status_code: 0,
        stdout_bytes: 2,
        stderr_bytes: 0,
        stdout_tail: "42",
        stderr_tail: "",
        stdout_truncated: false,
        stderr_truncated: false,
        artifact_path: null,
        artifact_digest: null,
        artifact_size_bytes: null,
        result: { SmallJson: 42 },
      },
      { nonce: `strict-main-before-child-complete-${suffix}` }
    )
  );
  assert.strictEqual(
    finalChildCompletion.type,
    "task_recorded",
    JSON.stringify(finalChildCompletion)
  );
  const releasedAfterFinalChild = await waitForCli(
    "final delayed child to release the active process slot",
    () =>
      runJson(
        clusterflux,
        ["process", "status", ...scope, "--process", processId],
        { cwd: projectDir }
      ),
    (status) => status.state === "not_active",
    120_000
  );
  const retainedEvents = runJson(
    clusterflux,
    ["task", "list", ...scope, "--process", processId],
    { cwd: projectDir }
  );
  assert(
    rawTaskEvents(retainedEvents).some(
      (event) =>
        event.task === fakeTask && event.terminal_state === "completed"
    ),
    "final delayed child history was not retained"
  );
  const retainedArtifacts = runJson(
    clusterflux,
    ["artifact", "list", ...scope, "--process", processId],
    { cwd: projectDir }
  );
  const retainedArtifact = retainedArtifacts.artifacts.find(
    (artifact) =>
      typeof artifact.digest === "string" &&
      /^sha256:[0-9a-f]{64}$/.test(artifact.digest) &&
      artifact.artifact.startsWith("release.tar-")
  );
  assert(retainedArtifact, "real main artifact metadata was not retained");
  const subsequentRun = runJson(
    clusterflux,
    ["run", "park-wake", "--project", ".", "--json"],
    { cwd: projectDir }
  );
  assert.strictEqual(subsequentRun.status, "main_launched");
  const subsequentAbort = runJson(
    clusterflux,
    ["process", "abort", ...scope, "--process", subsequentRun.process, "--yes"],
    { cwd: projectDir }
  );
  assert.strictEqual(subsequentAbort.abort_request.accepted, true);
  await waitForCli(
    "subsequent run cleanup after final-child slot release",
    () =>
      runJson(
        clusterflux,
        ["process", "status", ...scope, "--process", subsequentRun.process],
        { cwd: projectDir }
      ),
    (status) => status.state === "not_active",
    120_000
  );
  const mainBeforeChildLifecycle = {
    request: {
      process: processId,
      main: bundle.entryStableId,
      delayed_child: fakeTask,
      fault_injection:
        "withhold the assigned final child's signed completion until the real coordinator main completes",
    },
    expected:
      "process and debug state remain until the final child completes; final cleanup retains history and real artifact metadata; a subsequent run starts",
    observed: {
      main_terminal_state: mainCompletion.terminal_state,
      active_after_main: activeAfterMain.state !== "not_active",
      debug_state_after_main: debugStateAfterMain.type,
      child_terminal_state: "completed",
      final_process_state: releasedAfterFinalChild.state,
      retained_event_count: rawTaskEvents(retainedEvents).length,
      retained_artifact: retainedArtifact,
      subsequent_run_status: subsequentRun.status,
    },
    duration_ms: Math.max(1, Date.now() - mainBeforeChildStartedAt),
  };

  const revoked = runJson(
    clusterflux,
    ["key", "revoke", ...scope, "--agent", agent, "--yes"],
    { cwd: projectDir }
  );
  assert.strictEqual(revoked.command, "key revoke");
  const revokedUse = assertDenied(
    await sendHostedControl(
      signedAgentWorkflowRequest(
        identity,
        { ...unsigned, process: `${processId}-revoked` },
        { nonce: `strict-agent-revoked-${suffix}` }
      )
    ),
    /revoked/i,
    "revoked agent credential"
  );
  return {
    duration_ms: Date.now() - scenarioStartedAt,
    agent,
    valid_request: valid.type,
    replay,
    expired,
    forged,
    body_modified: bodyModified,
    revoked: revokedUse,
    workflow: {
      process: processId,
      main_launch: mainLaunch.type,
      external_direct_task_v1: directTaskV1,
      child_launch: fakeLaunch.type,
      child_parent: liveParentTask,
      real_debug_child_launch: realDebugLaunch.type,
      real_debug_child: realDebugTask,
      actor_kind: mainLaunch.actor.kind,
      authenticated_without_browser: mainLaunch.actor.authenticated_without_browser,
      flagship_completed: completedFlagship(completedEvents),
      concurrent_compile_tasks: concurrentCompileTasks,
    },
    main_before_child_lifecycle: mainBeforeChildLifecycle,
    attach_with_preconfigured_breakpoint: {
      request: {
        process: processId,
        source: attachSourcePath,
        line: attachBreakpointLine,
      },
      expected:
        "preconfigured before attach, unverified until adapter installation, then a real breakpoint stop at the configured line",
      observed: {
        preconfigured_revision: preconfiguredBreakpoint.revision,
        initially_verified:
          attachBreakpointResponse.body.breakpoints[0].verified,
        installation_event_verified:
          attachBreakpointInstalled.body.breakpoint.verified,
        stop_reason: attachBreakpointStop.body.reason,
        stopped_line: attachStackFrames[0].line,
      },
      duration_ms: Date.now() - scenarioStartedAt,
    },
    partial_freeze: {
      epoch: freeze.epoch,
      elapsed_ms: Date.now() - freezeStartedAt,
      partially_frozen: partialFreeze.partially_frozen,
      fully_frozen: partialFreeze.fully_frozen,
      dap_all_threads_stopped: dapPartialStop.body.allThreadsStopped,
      dap_continue_response_ms: continueResponseMs,
      podman_paused_container_ids: pausedContainerIds,
      warning: partialFreeze.failure_messages,
      resumed_participants: resumedAcknowledgements.map(
        (acknowledgement) => acknowledgement.task
      ),
    },
    nested_environment_mismatch: nestedEnvironmentMismatch,
  };
}

async function runSecondTenantIsolation({
  clusterflux,
  clusterfluxNode,
  firstScope,
  firstHelloBuild,
  firstHelloProjectDir,
  workRoot,
  firstSessionSecret,
  firstTenant,
  firstProject,
  firstProcess,
  firstNode,
  firstArtifact,
  manifest,
}) {
  const startedAt = Date.now();
  const file = process.env.CLUSTERFLUX_SECOND_TENANT_SESSION_FILE;
  const evidenceFile = process.env.CLUSTERFLUX_SECOND_TENANT_EVIDENCE_FILE;
  const targetTenant = process.env.CLUSTERFLUX_ISOLATION_TARGET_TENANT;
  const targetProject = process.env.CLUSTERFLUX_ISOLATION_TARGET_PROJECT;
  if (!file && evidenceFile) {
    const evidenceBytes = fs.readFileSync(path.resolve(evidenceFile));
    const previous = JSON.parse(evidenceBytes);
    const isolation = previous.tenant_isolation;
    assert.strictEqual(isolation?.executed, true);
    assert.notStrictEqual(isolation.second_tenant, firstTenant);
    assert.strictEqual(isolation.project?.denied, true);
    assert.strictEqual(isolation.process_hidden, true);
    assert.strictEqual(isolation.node_hidden, true);
    assert.strictEqual(isolation.tasks_and_logs?.denied, true);
    assert.strictEqual(isolation.debug?.denied, true);
    assert.strictEqual(isolation.debug?.audited, true);
    assert.strictEqual(isolation.debug?.charged_debug_read_bytes, 0);
    assert.strictEqual(isolation.artifact_and_download?.denied, true);
    assert.strictEqual(isolation.process_control?.denied, true);
    const previousDeployment = previous.release_binding?.deployment;
    const currentDeployment = deploymentProvenance();
    assert.strictEqual(
      previousDeployment?.hosted_service_sha256,
      currentDeployment.hosted_service_sha256,
      "reused isolation evidence must target the exact deployed hosted binary"
    );
    assert.deepStrictEqual(
      previous.release_binding?.binary_digests,
      manifest.binary_digests,
      "reused isolation evidence must target the exact public binaries"
    );
    return {
      evidence: {
        ...isolation,
        duration_ms: Date.now() - startedAt,
        reused_from_immutable_evidence: {
          report_sha256: sha256(evidenceBytes),
          source_commit: previous.source_commit,
          hosted_service_sha256: currentDeployment.hosted_service_sha256,
          public_binary_digests: manifest.binary_digests,
        },
      },
      runtime: null,
    };
  }
  if (!file && targetTenant && targetProject) {
    assert.notStrictEqual(
      targetTenant,
      firstTenant,
      "single-account isolation target must name a different hosted tenant"
    );
    assert.notStrictEqual(
      targetProject,
      firstProject,
      "single-account isolation target must name a different hosted project"
    );
    const call = (request) =>
      sendHostedControl(authenticatedRequest(firstSessionSecret, request));
    const project = assertDenied(
      await call({ type: "select_project", project: targetProject }),
      /outside|not visible|tenant|project|permission|unauthorized/i,
      "single-account cross-tenant project selection"
    );
    const processes = await call({ type: "list_processes" });
    assert.strictEqual(
      processes.type,
      "process_statuses",
      JSON.stringify(processes)
    );
    assert(!JSON.stringify(processes).includes(targetTenant));
    assert(!JSON.stringify(processes).includes(targetProject));
    const nodes = await call({ type: "list_node_descriptors" });
    assert.strictEqual(nodes.type, "node_descriptors", JSON.stringify(nodes));
    assert(!JSON.stringify(nodes).includes(targetTenant));
    assert(!JSON.stringify(nodes).includes(targetProject));
    const foreignProcess = `vp-foreign-isolation-${Date.now()}`;
    const foreignArtifact = `artifact-foreign-isolation-${Date.now()}`;
    const tasksAndLogsResponse = await call({
      type: "list_task_events",
      process: foreignProcess,
    });
    let tasksAndLogs;
    if (tasksAndLogsResponse.type === "error") {
      tasksAndLogs = {
        ...assertDenied(
          tasksAndLogsResponse,
          /outside|scope|tenant|project|not active|requires an active|unknown/i,
          "single-account foreign task and log listing"
        ),
        enforcement: "explicit_error",
      };
    } else {
      assert.strictEqual(
        tasksAndLogsResponse.type,
        "task_events",
        JSON.stringify(tasksAndLogsResponse)
      );
      assert.deepStrictEqual(
        tasksAndLogsResponse.events,
        [],
        "single-account foreign-looking task listing must disclose no events"
      );
      assert(!JSON.stringify(tasksAndLogsResponse).includes(targetTenant));
      assert(!JSON.stringify(tasksAndLogsResponse).includes(targetProject));
      tasksAndLogs = {
        denied: true,
        reason: "authenticated scope returned no visible task or log events",
        enforcement: "scoped_empty_result",
        response_type: tasksAndLogsResponse.type,
        event_count: tasksAndLogsResponse.events.length,
      };
    }
    const debugResponse = await call({
      type: "debug_attach",
      process: foreignProcess,
    });
    assert.strictEqual(
      debugResponse.type,
      "debug_attach",
      JSON.stringify(debugResponse)
    );
    assert.strictEqual(debugResponse.authorization?.allowed, false);
    assert.strictEqual(debugResponse.audit_event?.allowed, false);
    assert.strictEqual(debugResponse.charged_debug_read_bytes, 0);
    const artifact = assertDenied(
      await call({
        type: "create_artifact_download_link",
        artifact: foreignArtifact,
        max_bytes: 1024,
        ttl_seconds: 60,
      }),
      /outside|scope|tenant|project|not found|does not exist|unavailable/i,
      "single-account foreign artifact download"
    );
    const control = assertDenied(
      await call({ type: "abort_process", process: foreignProcess }),
      /outside|scope|tenant|project|not active|requires an active|unknown/i,
      "single-account foreign process control"
    );
    return {
      evidence: {
        executed: true,
        mode: "single_authenticated_account_foreign_project",
        second_tenant: targetTenant,
        target_project: targetProject,
        project,
        process_hidden: true,
        node_hidden: true,
        tasks_and_logs: tasksAndLogs,
        debug: {
          denied: true,
          reason: debugResponse.authorization.reason,
          audited: true,
          charged_debug_read_bytes: debugResponse.charged_debug_read_bytes,
        },
        artifact_and_download: artifact,
        process_control: control,
        duration_ms: Date.now() - startedAt,
      },
      runtime: null,
    };
  }
  if (!file) {
    return {
      evidence: {
        executed: false,
        reason: "second tenant session not supplied",
        duration_ms: Math.max(1, Date.now() - startedAt),
      },
      runtime: null,
    };
  }
  const secondSessionPath = path.resolve(file);
  const second = readJson(secondSessionPath);
  assert.strictEqual(second.coordinator, serviceEndpoint);
  const secondSessionSecret =
    second.cli_session_secret || second.session_secret;
  assert(secondSessionSecret, "second tenant session omitted its session secret");
  assert.notStrictEqual(
    second.tenant,
    firstTenant,
    "isolation proof requires a genuinely distinct hosted tenant"
  );
  assert.notStrictEqual(
    second.project,
    firstProject,
    "isolation proof requires a genuinely distinct hosted project"
  );
  const call = (request) =>
    sendHostedControl(authenticatedRequest(secondSessionSecret, request));

  const project = assertDenied(
    await call({ type: "select_project", project: firstProject }),
    /outside|not visible|tenant|project/i,
    "cross-tenant project selection"
  );
  const processes = await call({ type: "list_processes" });
  assert.strictEqual(
    processes.type,
    "process_statuses",
    JSON.stringify(processes)
  );
  assert(!JSON.stringify(processes).includes(firstProcess));
  const nodes = await call({ type: "list_node_descriptors" });
  assert.strictEqual(nodes.type, "node_descriptors", JSON.stringify(nodes));
  assert(!JSON.stringify(nodes).includes(firstNode));
  const tasksAndLogs = assertDenied(
    await call({ type: "list_task_events", process: firstProcess }),
    /outside|scope|tenant|project|not active|requires an active|unknown/i,
    "cross-tenant task and log listing"
  );
  const debugResponse = await call({
    type: "debug_attach",
    process: firstProcess,
  });
  assert.strictEqual(debugResponse.type, "debug_attach", JSON.stringify(debugResponse));
  assert.strictEqual(debugResponse.authorization?.allowed, false);
  assert.strictEqual(debugResponse.audit_event?.allowed, false);
  assert.match(
    debugResponse.authorization?.reason || "",
    /outside|scope|permission|tenant|project|not active|unknown/i
  );
  assert.strictEqual(debugResponse.charged_debug_read_bytes, 0);
  const debug = {
    denied: true,
    reason: debugResponse.authorization.reason,
    audited: true,
    charged_debug_read_bytes: debugResponse.charged_debug_read_bytes,
  };
  const artifact = assertDenied(
    await call({
      type: "create_artifact_download_link",
      artifact: firstArtifact,
      max_bytes: 1024,
      ttl_seconds: 60,
    }),
    /outside|scope|tenant|project|not found|does not exist|unavailable/i,
    "cross-tenant artifact download"
  );
  const control = assertDenied(
    await call({ type: "abort_process", process: firstProcess }),
    /outside|scope|tenant|project|not active|requires an active|unknown/i,
    "cross-tenant process control"
  );

  const secondCheckout = path.join(workRoot, "second-tenant-public-repo");
  stagePublicCheckout(manifest, secondCheckout);
  const secondProjectDir = path.join(
    secondCheckout,
    "examples",
    "hello-build"
  );
  const secondControlDir = path.join(secondProjectDir, ".clusterflux");
  ensureDir(secondControlDir);
  const secondProjectPath = path.join(
    path.dirname(secondSessionPath),
    "project.json"
  );
  assert(
    fs.existsSync(secondProjectPath),
    `second tenant session is missing adjacent project.json: ${secondProjectPath}`
  );
  fs.copyFileSync(
    secondSessionPath,
    path.join(secondControlDir, "session.json")
  );
  fs.copyFileSync(
    secondProjectPath,
    path.join(secondControlDir, "project.json")
  );
  fs.chmodSync(path.join(secondControlDir, "session.json"), 0o600);
  fs.chmodSync(path.join(secondControlDir, "project.json"), 0o644);
  const secondScope = [
    "--coordinator",
    serviceEndpoint,
    "--tenant",
    second.tenant,
    "--project-id",
    second.project,
    "--user",
    second.user,
    "--json",
  ];
  const secondProjectList = runJson(
    clusterflux,
    ["project", "list", ...secondScope],
    { cwd: secondProjectDir }
  );
  assert(secondProjectList.project_count >= 1);
  const secondProjectSelect = runJson(
    clusterflux,
    ["project", "select", ...secondScope, second.project],
    { cwd: secondProjectDir }
  );
  assert.strictEqual(secondProjectSelect.command, "project select");

  const collisionStartedAt = Date.now();
  const secondGrant = runJson(
    clusterflux,
    ["node", "enroll", ...secondScope],
    { cwd: secondProjectDir }
  );
  const secondAttach = runJson(
    clusterflux,
    [
      "node",
      "attach",
      "--coordinator",
      serviceEndpoint,
      "--tenant",
      second.tenant,
      "--project-id",
      second.project,
      "--node",
      firstNode,
      "--enrollment-grant",
      secondGrant.enrollment_grant.grant,
      "--json",
    ],
    { cwd: secondProjectDir }
  );
  assert.strictEqual(secondAttach.boundary.used_enrollment_exchange, true);
  const secondStored = readNodeCredential(secondProjectDir, firstNode);
  const secondCredentialDigest = sha256(
    fs.readFileSync(secondStored.absolute)
  );
  const secondIdentity = nodeIdentityFromPrivateKey(
    secondStored.credential.private_key
  );
  const secondWorkerArgs = [
    "--coordinator",
    serviceEndpoint,
    "--tenant",
    second.tenant,
    "--project-id",
    second.project,
    "--node",
    firstNode,
    "--worker",
    "--project-root",
    secondProjectDir,
    "--assignment-poll-ms",
    "500",
    "--emit-ready",
  ];
  const firstHelloWorker = spawnJsonLines(
    clusterfluxNode,
    [
      "--coordinator",
      serviceEndpoint,
      "--tenant",
      firstTenant,
      "--project-id",
      firstProject,
      "--node",
      firstNode,
      "--worker",
      "--project-root",
      firstHelloProjectDir,
      "--assignment-poll-ms",
      "500",
      "--emit-ready",
    ],
    {
      cwd: firstHelloProjectDir,
      env: { ...process.env },
    }
  );
  const firstHelloReady = await firstHelloWorker.waitFor(
    (value) => value.node_status === "ready",
    "first-tenant hello-build artifact worker ready for collision proof"
  );
  assert.strictEqual(firstHelloReady.node, firstNode);
  const firstCollisionHelloBuild = await runHostedHelloBuild({
    clusterflux,
    projectDir: firstHelloProjectDir,
    scope: firstScope,
    outputFile: path.join(workRoot, "hello-clusterflux-first-tenant-collision"),
  });
  assert.strictEqual(
    firstCollisionHelloBuild.artifact,
    firstHelloBuild.artifact,
    "repeated deterministic first-tenant hello-build changed its ArtifactId"
  );
  assert.strictEqual(firstCollisionHelloBuild.digest, firstHelloBuild.digest);
  assert.strictEqual(
    firstCollisionHelloBuild.executable_output,
    firstHelloBuild.executable_output
  );
  const secondWorker = spawnJsonLines(clusterfluxNode, secondWorkerArgs, {
    cwd: secondProjectDir,
    env: { ...process.env },
  });
  let secondHelloBuild;
  try {
    const secondReady = await secondWorker.waitFor(
      (value) => value.node_status === "ready",
      "same-name second-tenant worker ready"
    );
    assert.strictEqual(secondReady.node, firstNode);
    secondHelloBuild = await runHostedHelloBuild({
      clusterflux,
      projectDir: secondProjectDir,
      scope: secondScope,
      outputFile: path.join(workRoot, "hello-clusterflux-second-tenant"),
    });
  } finally {
    await stopChild(secondWorker.child);
  }
  assert.strictEqual(
    secondHelloBuild.artifact,
    firstCollisionHelloBuild.artifact,
    "deterministic two-tenant hello-build did not collide on ArtifactId"
  );
  assert.strictEqual(
    secondHelloBuild.digest,
    firstCollisionHelloBuild.digest,
    "deterministic two-tenant hello-build did not produce identical bytes"
  );
  assert.strictEqual(
    secondHelloBuild.executable_output,
    firstCollisionHelloBuild.executable_output
  );

  const firstArtifactsAfterCollision = runJson(
    clusterflux,
    [
      "artifact",
      "list",
      ...firstScope,
      "--process",
      firstCollisionHelloBuild.process,
    ],
    { cwd: firstHelloProjectDir }
  );
  assert(
    firstArtifactsAfterCollision.artifacts.some(
      (candidate) =>
        candidate.artifact === firstCollisionHelloBuild.artifact &&
        candidate.digest === firstCollisionHelloBuild.digest
    ),
    "the second tenant's same-ID artifact masked the first tenant's metadata"
  );
  const secondCollisionLink = await call({
    type: "create_artifact_download_link",
    artifact: secondHelloBuild.artifact,
    max_bytes: secondHelloBuild.downloaded_bytes,
    ttl_seconds: 60,
  });
  assert.strictEqual(
    secondCollisionLink.type,
    "artifact_download_link",
    JSON.stringify(secondCollisionLink)
  );
  const secondCollisionLinkRevoked = await call({
    type: "revoke_artifact_download_link",
    artifact: secondHelloBuild.artifact,
    token_digest: secondCollisionLink.link.scoped_token_digest,
  });
  assert.strictEqual(
    secondCollisionLinkRevoked.type,
    "artifact_download_link_revoked",
    JSON.stringify(secondCollisionLinkRevoked)
  );
  const firstDownloadAfterSecondLinkRevocation = runJson(
    clusterflux,
    [
      "artifact",
      "download",
      ...firstScope,
      firstCollisionHelloBuild.artifact,
      "--to",
      path.join(workRoot, "hello-clusterflux-first-after-second-link-revoke"),
    ],
    { cwd: firstHelloProjectDir }
  );
  assert.strictEqual(
    firstDownloadAfterSecondLinkRevocation.local_download.verified_digest,
    firstCollisionHelloBuild.digest
  );
  await stopChild(firstHelloWorker.child);

  return {
    evidence: {
      executed: true,
      second_tenant: second.tenant,
      second_project: second.project,
      project,
      process_hidden: true,
      node_hidden: true,
      tasks_and_logs: tasksAndLogs,
      debug,
      artifact_and_download: artifact,
      process_control: control,
      scoped_node_and_artifact_collision: {
        request: {
          node: firstNode,
          first_scope: {
            tenant: firstTenant,
            project: firstProject,
          },
          second_scope: {
            tenant: second.tenant,
            project: second.project,
          },
          example: "hello-build",
        },
        expected:
          "same-name Nodes coexist and deterministic hello-build artifacts share an ArtifactId without metadata, link, or download interference",
        observed: {
          first_process: firstCollisionHelloBuild.process,
          second_process: secondHelloBuild.process,
          same_artifact_id: secondHelloBuild.artifact,
          first_digest: firstCollisionHelloBuild.digest,
          second_digest: secondHelloBuild.digest,
          first_downloaded_bytes: firstCollisionHelloBuild.downloaded_bytes,
          second_downloaded_bytes: secondHelloBuild.downloaded_bytes,
          exact_executable_output: secondHelloBuild.executable_output,
          both_metadata_records_visible_to_owner: true,
          cross_tenant_unique_artifact_denied: artifact.denied,
          second_link_revoked: true,
          first_download_survived_second_link_revocation: true,
        },
        duration_ms: Math.max(1, Date.now() - collisionStartedAt),
      },
      duration_ms: Math.max(1, Date.now() - startedAt),
    },
    runtime: {
      node: firstNode,
      tenant: second.tenant,
      project: second.project,
      projectDir: secondProjectDir,
      scope: secondScope,
      workerArgs: secondWorkerArgs,
      credentialPath: secondStored.absolute,
      credentialDigest: secondCredentialDigest,
      identity: secondIdentity,
    },
  };
}

async function runLiveSignedHostileArtifactPath({
  clusterflux,
  projectDir,
  scope,
  tenant,
  project,
  suffix,
  securityNode,
}) {
  const startedAt = Date.now();
  const runReport = runJson(
    clusterflux,
    ["run", "park-wake", "--project", ".", "--json"],
    { cwd: projectDir }
  );
  assert.strictEqual(runReport.status, "main_launched");
  const process = runReport.process;
  await waitForCli(
    "hostile-path probe process to become active",
    () =>
      runJson(
        clusterflux,
        ["process", "status", ...scope, "--process", process],
        { cwd: projectDir }
      ),
    (status) => status.state !== "not_active",
    120_000
  );

  const invalidStartedAt = Date.now();
  const invalid = assertDenied(
    await sendHostedControl(
      signedNodeRequest(
        securityNode.node,
        securityNode.identity,
        "report_vfs_metadata",
        {
          type: "report_vfs_metadata",
          tenant,
          project,
          process,
          node: securityNode.node,
          task: `hostile-path-${suffix}`,
          artifact_path: "/vfs/artifacts/bad artifact!",
          artifact_digest: sha256(Buffer.from("hostile-path-invalid")),
          artifact_size_bytes: 20,
          large_bytes_uploaded: false,
        },
        { nonce: `strict-hostile-path-invalid-${suffix}-${Date.now()}` }
      )
    ),
    /invalid VFS artifact path|invalid artifact path|ArtifactId is invalid/i,
    "correctly signed hostile artifact path"
  );
  const invalidDurationMs = Math.max(1, Date.now() - invalidStartedAt);

  const validStartedAt = Date.now();
  const validArtifact = `strict-hostile-followup-${suffix}`;
  const valid = await sendHostedControl(
    signedNodeRequest(
      securityNode.node,
      securityNode.identity,
      "report_vfs_metadata",
      {
        type: "report_vfs_metadata",
        tenant,
        project,
        process,
        node: securityNode.node,
        task: `hostile-path-${suffix}`,
        artifact_path: `/vfs/artifacts/${validArtifact}`,
        artifact_digest: sha256(Buffer.from("hostile-path-valid")),
        artifact_size_bytes: 18,
        large_bytes_uploaded: false,
      },
      { nonce: `strict-hostile-path-valid-${suffix}-${Date.now()}` }
    )
  );
  assert.strictEqual(valid.type, "vfs_metadata_recorded", JSON.stringify(valid));
  const healthyHeartbeat = await sendHostedControl({
    type: "node_heartbeat",
    tenant,
    project,
    node: securityNode.node,
    node_signature: signedNodeHeartbeat(
      tenant,
      project,
      securityNode.node,
      securityNode.identity,
      { nonce: `strict-hostile-path-health-${suffix}-${Date.now()}` }
    ),
  });
  assert.strictEqual(
    healthyHeartbeat.type,
    "node_heartbeat",
    JSON.stringify(healthyHeartbeat)
  );
  const validDurationMs = Math.max(1, Date.now() - validStartedAt);
  const aborted = runJson(
    clusterflux,
    ["process", "abort", ...scope, "--process", process, "--yes"],
    { cwd: projectDir }
  );
  assert.strictEqual(aborted.abort_request.accepted, true);
  await waitForCli(
    "hostile-path probe process cleanup",
    () =>
      runJson(
        clusterflux,
        ["process", "status", ...scope, "--process", process],
        { cwd: projectDir }
      ),
    (status) => status.state === "not_active",
    120_000
  );
  return {
    request: {
      principal: "enrolled signed Node",
      variant: "report_vfs_metadata",
      process,
      malformed_path: "/vfs/artifacts/bad artifact!",
    },
    expected:
      "structured InvalidArtifactPath rejection followed immediately by valid signed metadata and heartbeat on the same service instance",
    observed: {
      rejection: invalid,
      valid_metadata_response: valid.type,
      valid_artifact: validArtifact,
      health_response: healthyHeartbeat.type,
      process_cleanup: "not_active",
    },
    invalid_duration_ms: invalidDurationMs,
    valid_followup_duration_ms: validDurationMs,
    duration_ms: Math.max(1, Date.now() - startedAt),
  };
}

async function runLiveSoak({
  clusterflux,
  projectDir,
  scope,
  securityNode,
}) {
  if (!strictVpsRestart) {
    return { executed: false, reason: "strict VPS measurement access not supplied" };
  }
  const runReport = runJson(
    clusterflux,
    ["run", "build", "--project", ".", "--json"],
    { cwd: projectDir }
  );
  assert.strictEqual(runReport.status, "main_launched");
  const processId = runReport.process;
  const parked = await waitForCli(
    "soak coordinator main to park without a capable node",
    () =>
      runJson(
        clusterflux,
        ["process", "status", ...scope, "--process", processId],
        { cwd: projectDir }
      ),
    (status) =>
      status.live_process?.main_state === "running" &&
      status.live_process?.main_wait_state === "waiting_for_task" &&
      status.current_task_count === 0 &&
      status.live_process?.connected_nodes?.length === 0,
    120000
  );

  const pollSecurityNode = (nonce) =>
    sendHostedControl(
      signedNodeRequest(
        securityNode.node,
        securityNode.identity,
        "poll_task_assignment",
        {
          type: "poll_task_assignment",
          tenant: securityNode.capability_body.tenant,
          project: securityNode.capability_body.project,
          node: securityNode.node,
        },
        { nonce }
      )
    );
  const drainedPriorAssignments = [];
  for (let drainIndex = 0; drainIndex < 16; drainIndex += 1) {
    const poll = await pollSecurityNode(
      `strict-soak-drain-${drainIndex}-${Date.now()}`
    );
    assert.strictEqual(poll.type, "task_assignment", JSON.stringify(poll));
    if (poll.assignment === null) break;
    assert.notStrictEqual(
      poll.assignment.process,
      processId,
      "soak process unexpectedly received a task assignment"
    );
    drainedPriorAssignments.push({
      process: poll.assignment.process,
      task: poll.assignment.task,
    });
    assert(drainIndex < 15, "pre-soak assignment drain did not quiesce");
  }

  const startedAt = Date.now();
  const deadline = startedAt + strictSoakSeconds * 1000;
  const samples = [];
  let sampleIndex = 0;
  while (true) {
    const capabilities = await sendHostedControl(
      signedNodeRequest(
        securityNode.node,
        securityNode.identity,
        "report_node_capabilities",
        securityNode.capability_body,
        { nonce: `strict-soak-capabilities-${sampleIndex}-${Date.now()}` }
      )
    );
    assert.strictEqual(
      capabilities.type,
      "node_capabilities_recorded",
      JSON.stringify(capabilities)
    );
    const poll = await pollSecurityNode(
      `strict-soak-poll-${sampleIndex}-${Date.now()}`
    );
    assert.strictEqual(poll.type, "task_assignment", JSON.stringify(poll));
    assert.strictEqual(poll.assignment, null);
    const status = runJson(
      clusterflux,
      ["process", "status", ...scope, "--process", processId],
      { cwd: projectDir }
    );
    assert.strictEqual(status.live_process.main_wait_state, "waiting_for_task");
    samples.push(strictInfrastructureSample(projectDir));
    sampleIndex += 1;
    if (Date.now() >= deadline) break;
    await delay(Math.min(30_000, deadline - Date.now()));
  }

  const currentMemory = samples.map(
    (sample) => sample.service_memory_current_bytes
  );
  const serviceDisk = samples.map((sample) => sample.service_state_disk_bytes);
  const databaseDisk = samples.map((sample) => sample.postgres_database_bytes);
  const localDisk = samples.map((sample) => sample.local_node_state.bytes);
  const spread = (values) => Math.max(...values) - Math.min(...values);
  assert(spread(currentMemory) <= 128 * 1024 * 1024);
  assert(spread(serviceDisk) <= 16 * 1024 * 1024);
  assert(spread(databaseDisk) <= 32 * 1024 * 1024);
  assert(spread(localDisk) <= 16 * 1024 * 1024);
  assert.deepStrictEqual(
    samples.at(-1).durable_rows,
    samples[0].durable_rows,
    "durable object counts grew during the parked-process soak"
  );
  const aborted = runJson(
    clusterflux,
    ["process", "abort", ...scope, "--process", processId, "--yes"],
    { cwd: projectDir }
  );
  assert.strictEqual(aborted.abort_request.process_slot_released, true);
  return {
    executed: true,
    process: processId,
    duration_seconds: Math.floor((Date.now() - startedAt) / 1000),
    sample_interval_seconds: 30,
    sample_count: samples.length,
    parked_main: {
      main_state: parked.live_process.main_state,
      main_wait_state: parked.live_process.main_wait_state,
    },
    drained_prior_assignments: drainedPriorAssignments,
    signed_node_poll_cycles: samples.length,
    memory_spread_bytes: spread(currentMemory),
    service_disk_spread_bytes: spread(serviceDisk),
    postgres_disk_spread_bytes: spread(databaseDisk),
    local_node_state_spread_bytes: spread(localDisk),
    samples,
  };
}

async function revokeSecurityNode({
  clusterflux,
  projectDir,
  scope,
  securityNode,
}) {
  const startedAt = Date.now();
  const revoked = runJson(
    clusterflux,
    ["node", "revoke", ...scope, "--node", securityNode.node, "--yes"],
    { cwd: projectDir }
  );
  assert.strictEqual(revoked.command, "node revoke");
  const denied = assertDenied(
    await sendHostedControl({
      type: "node_heartbeat",
      tenant: securityNode.capability_body.tenant,
      project: securityNode.capability_body.project,
      node: securityNode.node,
      node_signature: signedNodeHeartbeat(
        securityNode.capability_body.tenant,
        securityNode.capability_body.project,
        securityNode.node,
        securityNode.identity,
        { nonce: `strict-node-revoked-${Date.now()}` }
      ),
    }),
    /revoked|unknown node|not recognized|not enrolled/i,
    "revoked node credential"
  );
  return {
    node: securityNode.node,
    denied,
    duration_ms: Date.now() - startedAt,
  };
}

async function runLiveQuotaPreallocation({ sessionSecret, suffix }) {
  const startedAt = Date.now();
  const readSpawnQuota = async () => {
    const status = await sendHostedControl(
      authenticatedRequest(sessionSecret, { type: "quota_status" })
    );
    assert.strictEqual(status.type, "quota_status", JSON.stringify(status));
    const limit = status.limits?.limits?.Spawn;
    const usage = status.usage?.Spawn;
    const windowSeconds = status.window_seconds?.Spawn;
    const windowStarted = status.window_started_epoch_seconds?.Spawn;
    assert(Number.isInteger(limit) && limit > 0, JSON.stringify(status));
    assert(Number.isInteger(usage) && usage >= 0, JSON.stringify(status));
    assert(
      Number.isInteger(windowSeconds) && windowSeconds > 0,
      JSON.stringify(status)
    );
    assert(Number.isInteger(windowStarted), JSON.stringify(status));
    return { limit, usage, windowSeconds, windowStarted };
  };

  let spawnQuota = await readSpawnQuota();
  const windowElapsed = Math.max(
    0,
    Math.floor(Date.now() / 1000) - spawnQuota.windowStarted
  );
  if (spawnQuota.usage > 0 || windowElapsed > 1) {
    await delay(
      Math.max(1000, (spawnQuota.windowSeconds - windowElapsed + 1) * 1000)
    );
    spawnQuota = await readSpawnQuota();
  }
  let denial;
  let deniedProcess;
  let acceptedStarts = 0;
  const maxAttempts = spawnQuota.limit - spawnQuota.usage + 2;
  for (let index = 0; index < maxAttempts; index += 1) {
    const processId = `vp-quota-${suffix}-${index}`;
    const started = await sendHostedControl(
      authenticatedRequest(sessionSecret, {
        type: "start_process",
        process: processId,
        restart: false,
      })
    );
    if (started.type === "error") {
      assert.match(started.message, /quota|spawn|limit|community tier/i);
      denial = started;
      deniedProcess = processId;
      break;
    }
    assert.strictEqual(started.type, "process_started", JSON.stringify(started));
    acceptedStarts += 1;
    const aborted = await sendHostedControl(
      authenticatedRequest(sessionSecret, {
        type: "abort_process",
        process: processId,
      })
    );
    assert.strictEqual(aborted.type, "process_aborted", JSON.stringify(aborted));
  }
  assert(
    denial,
    `live community-tier spawn quota was not reached in ${maxAttempts} attempts`
  );
  const processes = await sendHostedControl(
    authenticatedRequest(sessionSecret, { type: "list_processes" })
  );
  assert.strictEqual(
    processes.type,
    "process_statuses",
    JSON.stringify(processes)
  );
  assert(
    !JSON.stringify(processes).includes(deniedProcess),
    "quota-denied process was allocated before denial"
  );
  const independentScope = await sendHostedControl(
    authenticatedRequest(sessionSecret, { type: "list_node_descriptors" })
  );
  assert.strictEqual(
    independentScope.type,
    "node_descriptors",
    `spawn quota exhaustion locked an independent read scope: ${JSON.stringify(independentScope)}`
  );
  return {
    duration_ms: Date.now() - startedAt,
    limit: spawnQuota.limit,
    initial_usage: spawnQuota.usage,
    window_seconds: spawnQuota.windowSeconds,
    accepted_starts_before_denial: acceptedStarts,
    denied_process: deniedProcess,
    denial: {
      type: denial.type,
      category: denial.error?.category || null,
      message: denial.message,
    },
    denied_process_allocated: false,
    independent_scope: {
      operation: "list_node_descriptors",
      response: independentScope.type,
      unaffected_by_spawn_quota: true,
    },
  };
}

async function runLongJoinProof({ clusterflux, projectDir, scope }) {
  const startedAt = Date.now();
  const runReport = runJson(
    clusterflux,
    ["run", "long-join", "--project", ".", "--json"],
    { cwd: projectDir }
  );
  assert.strictEqual(runReport.status, "main_launched");
  const terminal = await waitForCli(
    "controlled task longer than two minutes",
    () =>
      runJson(
        clusterflux,
        ["task", "list", ...scope, "--process", runReport.process],
        { cwd: projectDir }
      ),
    (tasks) =>
      rawTaskEvents(tasks).some(
        (event) =>
          event.task_definition === "long_join_probe" &&
          event.terminal_state === "completed"
      ),
    5 * 60 * 1000
  );
  const completed = rawTaskEvents(terminal).find(
    (event) =>
      event.task_definition === "long_join_probe" &&
      event.terminal_state === "completed"
  );
  const durationSeconds = Math.floor((Date.now() - startedAt) / 1000);
  assert(durationSeconds > 120);
  const released = await waitForCli(
    "long join process automatic slot release",
    () =>
      runJson(
        clusterflux,
        ["process", "status", ...scope, "--process", runReport.process],
        { cwd: projectDir }
      ),
    (status) => status.state === "not_active",
    120000
  );
  return {
    process: runReport.process,
    task: completed.task,
    terminal_state: completed.terminal_state,
    result: completed.result,
    duration_seconds: durationSeconds,
    duration_ms: Math.max(1, Date.now() - startedAt),
    process_state: released.state,
  };
}

async function runRepeatedParkWakeProof({ clusterflux, projectDir, scope }) {
  const runReport = runJson(
    clusterflux,
    ["run", "park-wake", "--project", ".", "--json"],
    { cwd: projectDir }
  );
  assert.strictEqual(runReport.status, "main_launched");
  const terminal = await waitForCli(
    "repeated coordinator-main park/wake completion",
    () =>
      runJson(
        clusterflux,
        ["task", "list", ...scope, "--process", runReport.process],
        { cwd: projectDir }
      ),
    (tasks) => {
      const completed = new Set(
        rawTaskEvents(tasks)
          .filter(
            (event) =>
              event.task_definition === "task_add_one" &&
              event.terminal_state === "completed"
          )
          .map((event) => event.task)
      );
      return completed.size >= 16;
    },
    120000
  );
  const completedTasks = [
    ...new Set(
      rawTaskEvents(terminal)
        .filter(
          (event) =>
            event.task_definition === "task_add_one" &&
            event.terminal_state === "completed"
        )
        .map((event) => event.task)
    ),
  ];
  const released = await waitForCli(
    "park/wake process automatic slot release",
    () =>
      runJson(
        clusterflux,
        ["process", "status", ...scope, "--process", runReport.process],
        { cwd: projectDir }
      ),
    (status) => status.state === "not_active",
    120000
  );
  return {
    process: runReport.process,
    completed_cycles: completedTasks.length,
    task_instances: completedTasks,
    terminal_state: released.state,
  };
}

async function runDroppedConnectionRollback({
  clusterflux,
  projectDir,
  scope,
}) {
  const scenarioStartedAt = Date.now();
  const processId = "vp-current";
  const attempted = cp.spawnSync(
    clusterflux,
    ["run", "build", "--project", ".", "--json"],
    {
      cwd: projectDir,
      env: {
        ...process.env,
        CLUSTERFLUX_TEST_DROP_RESPONSE_AFTER_OPERATION: "start_process",
      },
      encoding: "utf8",
      timeout: 5 * 60 * 1000,
    }
  );
  assert.notStrictEqual(
    attempted.status,
    0,
    `response-loss run unexpectedly succeeded: ${attempted.stdout}`
  );
  assert.match(
    `${attempted.stderr}\n${attempted.stdout}`,
    /closed.*without a response|transport|response/i
  );
  const released = await waitForCli(
    "CLI launch guard rollback after lost start response",
    () =>
      runJson(
        clusterflux,
        ["process", "status", ...scope, "--process", processId],
        { cwd: projectDir }
      ),
    (status) => status.state === "not_active",
    30000
  );
  return {
    duration_ms: Date.now() - scenarioStartedAt,
    process: processId,
    failure_injection: "control transport dropped attempt-owned start_process response",
    cli_exit_status: attempted.status,
    rollback: "automatic CLI launch guard abort_process with matching launch_attempt",
    terminal_state: released.state,
  };
}

async function runLaunchAttemptOwnershipGuards({
  clusterflux,
  projectDir,
  scope,
  sessionSecret,
  activeProcess,
}) {
  const startedAt = Date.now();
  const rejectedAttempt = `launch-guard-rejected-${Date.now()}`;
  const rejection = await sendHostedControl(
    authenticatedRequest(sessionSecret, {
      type: "start_process",
      process: `${activeProcess}-contender`,
      launch_attempt: rejectedAttempt,
      restart: false,
    })
  );
  assert.strictEqual(rejection.type, "error");
  assert.match(rejection.message, /already has active virtual process/i);

  const wrongAttempt = `launch-guard-wrong-${Date.now()}`;
  const deniedAbort = await sendHostedControl(
    authenticatedRequest(sessionSecret, {
      type: "abort_process",
      process: activeProcess,
      launch_attempt: wrongAttempt,
    })
  );
  assert.strictEqual(deniedAbort.type, "error");
  assert.match(deniedAbort.message, /does not own process/i);

  const surviving = runJson(
    clusterflux,
    ["process", "status", ...scope, "--process", activeProcess],
    { cwd: projectDir }
  );
  assert.notStrictEqual(surviving.state, "not_active");
  assert.strictEqual(surviving.process, activeProcess);
  return {
    duration_ms: Date.now() - startedAt,
    rejected_attempt: rejectedAttempt,
    rejection: rejection.message,
    wrong_abort_attempt: wrongAttempt,
    wrong_abort_denied: deniedAbort.message,
    existing_process_survived: true,
  };
}

async function runLiveBandwidthPreallocation({
  sessionSecret,
  artifact,
  maxBytes,
}) {
  const scenarioStartedAt = Date.now();
  const before = relayDurableState();
  const partialLink = await sendHostedControl(
    authenticatedRequest(sessionSecret, {
      type: "create_artifact_download_link",
      artifact,
      max_bytes: maxBytes,
      ttl_seconds: 300,
    })
  );
  assert.strictEqual(
    partialLink.type,
    "artifact_download_link",
    JSON.stringify(partialLink)
  );
  const partialToken = partialLink.link.scoped_token_digest;
  let partialChunk;
  const partialDeadline = Date.now() + 120_000;
  while (Date.now() < partialDeadline) {
    const response = await sendHostedControl(
      authenticatedRequest(sessionSecret, {
        type: "open_artifact_download_stream",
        artifact,
        max_bytes: maxBytes,
        token_digest: partialToken,
        chunk_bytes: 16,
      })
    );
    assert.strictEqual(
      response.type,
      "artifact_download_stream",
      JSON.stringify(response)
    );
    if (response.content_bytes_available === true) {
      partialChunk = response;
      break;
    }
    await delay(50);
  }
  assert(partialChunk, "artifact relay did not return the first partial chunk");
  assert.strictEqual(partialChunk.content_eof, false);
  assert.strictEqual(partialChunk.streamed_bytes, 16);
  const partialRevoked = await sendHostedControl(
    authenticatedRequest(sessionSecret, {
      type: "revoke_artifact_download_link",
      artifact,
      token_digest: partialToken,
    })
  );
  assert.strictEqual(
    partialRevoked.type,
    "artifact_download_link_revoked",
    JSON.stringify(partialRevoked)
  );
  const afterPartialAbandon = relayDurableState();
  assert(afterPartialAbandon.ingress_used > before.ingress_used);
  assert(afterPartialAbandon.egress_used > before.egress_used);
  assert(
    afterPartialAbandon.abandoned_or_failed_used >
      before.abandoned_or_failed_used
  );

  let denial;
  const acceptedReservations = [];
  for (let index = 0; index < 64; index += 1) {
    const response = await sendHostedControl(
      authenticatedRequest(sessionSecret, {
        type: "create_artifact_download_link",
        artifact,
        max_bytes: maxBytes,
        ttl_seconds: 300,
      })
    );
    if (response.type === "error") {
      assert.match(
        response.message,
        /artifact relay.*(?:concurrency|period byte budget)|bandwidth|reservation/i
      );
      denial = response;
      break;
    }
    assert.strictEqual(response.type, "artifact_download_link", JSON.stringify(response));
    acceptedReservations.push(response.link.scoped_token_digest);
  }
  assert(denial, "artifact relay preallocation was not denied in 64 reservations");
  for (const tokenDigest of acceptedReservations) {
    const revoked = await sendHostedControl(
      authenticatedRequest(sessionSecret, {
        type: "revoke_artifact_download_link",
        artifact,
        token_digest: tokenDigest,
      })
    );
    assert.strictEqual(
      revoked.type,
      "artifact_download_link_revoked",
      JSON.stringify(revoked)
    );
  }
  const afterRelease = relayDurableState();
  assert.strictEqual(
    Object.keys(afterRelease.reservations || {}).length,
    Object.keys(before.reservations || {}).length
  );
  return {
    scenario_started_at_ms: scenarioStartedAt,
    artifact,
    accepted_reservations_before_denial: acceptedReservations.length,
    bytes_served_before_denial: partialChunk.streamed_bytes,
    partial_abandon: {
      ingress_delta: afterPartialAbandon.ingress_used - before.ingress_used,
      egress_delta: afterPartialAbandon.egress_used - before.egress_used,
      abandoned_delta:
        afterPartialAbandon.abandoned_or_failed_used -
        before.abandoned_or_failed_used,
      reservation_released: !(partialToken in afterRelease.reservations),
    },
    durable_before: before,
    durable_after_release: afterRelease,
    denial: {
      type: denial.type,
      category: denial.error?.category || null,
      message: denial.message,
    },
  };
}

function relayDurableState() {
  if (!strictVpsRestart) {
    throw new Error("strict relay accounting requires VPS Postgres access");
  }
  const output = run(
    "ssh",
    sshArgs(
      "sudo",
      "-u",
      "postgres",
      "psql",
      "-d",
      "clusterflux",
      "-At",
      "-c",
      "SELECT record::text FROM clusterflux_artifact_relay_state WHERE singleton = TRUE"
    )
  ).trim();
  assert.notStrictEqual(output, "", "artifact relay durable state was not persisted");
  return JSON.parse(output.split(/\r?\n/).filter(Boolean).at(-1));
}

async function waitForHostedControlReady(timeoutMs = 60_000) {
  const deadline = Date.now() + timeoutMs;
  let lastError;
  while (Date.now() < deadline) {
    try {
      const ping = await sendHostedControl({ type: "ping" });
      if (ping.type === "pong") return ping;
      lastError = new Error(JSON.stringify(ping));
    } catch (error) {
      lastError = error;
    }
    await delay(500);
  }
  throw new Error(`hosted service did not recover: ${lastError}`);
}

async function publishRelayProbeArtifact({ clusterflux, projectDir, scope, label }) {
  const runReport = runJson(
    clusterflux,
    ["run", "build", "--project", ".", "--json"],
    { cwd: projectDir }
  );
  await waitForCli(
    `${label} flagship completion`,
    () =>
      runJson(
        clusterflux,
        ["task", "list", ...scope, "--process", runReport.process],
        { cwd: projectDir }
      ),
    (report) => completedFlagship(rawTaskEvents(report)),
    10 * 60 * 1000
  );
  const artifacts = runJson(
    clusterflux,
    ["artifact", "list", ...scope, "--process", runReport.process],
    { cwd: projectDir }
  );
  const artifact = artifacts.artifacts.find((candidate) => {
    const digest = candidate.digest;
    return (
      typeof digest === "string" &&
      /^sha256:[0-9a-f]{64}$/.test(digest) &&
      candidate.artifact === `release.tar-${digest.slice("sha256:".length)}`
    );
  });
  assert(artifact, `${label} did not publish a relay probe artifact`);
  return artifact;
}

async function runRelayEmergencyDisable({
  clusterflux,
  clusterfluxNode,
  projectDir,
  scope,
  workerArgs,
  sessionSecret,
  maxBytes,
}) {
  const startedAt = Date.now();
  if (!strictVpsRestart) {
    throw new Error("strict relay disable proof requires VPS systemd access");
  }
  const dropInDir = `/run/systemd/system/${strictServiceUnit}.d`;
  const dropInPath = `${dropInDir}/90-strict-relay-disable.conf`;
  run("ssh", sshArgs("sudo", "mkdir", "-p", dropInDir));
  cp.execFileSync("ssh", sshArgs("sudo", "tee", dropInPath), {
    cwd: repo,
    input: "[Service]\nEnvironment=CLUSTERFLUX_HOSTED_RELAY_DISABLED=true\n",
    stdio: ["pipe", "ignore", "inherit"],
  });
  let disabled;
  let disabledArtifact;
  let disabledWorker;
  try {
    run(
      "ssh",
      sshArgs(
        "sudo",
        "systemctl",
        "daemon-reload"
      )
    );
    run(
      "ssh",
      sshArgs("sudo", "systemctl", "restart", strictServiceUnit)
    );
    await waitForHostedControlReady();
    disabledWorker = spawnJsonLines(clusterfluxNode, workerArgs, {
      cwd: projectDir,
      env: { ...process.env },
    });
    await disabledWorker.waitFor(
      (value) => value.node_status === "ready",
      "relay-disabled worker ready"
    );
    disabledArtifact = await publishRelayProbeArtifact({
      clusterflux,
      projectDir,
      scope,
      label: "relay-disabled",
    });
    disabled = assertDenied(
      await sendHostedControl(
        authenticatedRequest(sessionSecret, {
          type: "create_artifact_download_link",
          artifact: disabledArtifact.artifact,
          max_bytes: maxBytes,
          ttl_seconds: 60,
        })
      ),
      /artifact relay is disabled/i,
      "emergency-disabled artifact relay"
    );
  } finally {
    if (disabledWorker) await stopChild(disabledWorker.child);
    run("ssh", sshArgs("sudo", "rm", "-f", dropInPath));
    run("ssh", sshArgs("sudo", "systemctl", "daemon-reload"));
    run(
      "ssh",
      sshArgs("sudo", "systemctl", "restart", strictServiceUnit)
    );
    await waitForHostedControlReady();
  }
  const restoredWorker = spawnJsonLines(clusterfluxNode, workerArgs, {
    cwd: projectDir,
    env: { ...process.env },
  });
  try {
    await restoredWorker.waitFor(
      (value) => value.node_status === "ready",
      "relay-restored worker ready"
    );
    const restoredArtifact = await publishRelayProbeArtifact({
      clusterflux,
      projectDir,
      scope,
      label: "relay-restored",
    });
    const restored = await sendHostedControl(
      authenticatedRequest(sessionSecret, {
        type: "create_artifact_download_link",
        artifact: restoredArtifact.artifact,
        max_bytes: maxBytes,
        ttl_seconds: 60,
      })
    );
    assert.strictEqual(restored.type, "artifact_download_link", JSON.stringify(restored));
    const revoked = await sendHostedControl(
      authenticatedRequest(sessionSecret, {
        type: "revoke_artifact_download_link",
        artifact: restoredArtifact.artifact,
        token_digest: restored.link.scoped_token_digest,
      })
    );
    assert.strictEqual(revoked.type, "artifact_download_link_revoked");
    return {
      evidence: {
        duration_ms: Date.now() - startedAt,
        disabled,
        disabled_probe_artifact: disabledArtifact.artifact,
        restored: restored.type,
        restored_probe_artifact: restoredArtifact.artifact,
        restored_reservation_released: revoked.type,
        runtime_drop_in_removed: true,
      },
      worker: restoredWorker,
    };
  } catch (error) {
    await stopChild(restoredWorker.child);
    throw error;
  }
}

function deploymentProvenance() {
  if (!strictVpsRestart) return null;
  const systemGeneration = run(
    "ssh",
    sshArgs("readlink", "-f", "/run/current-system")
  ).trim();
  const mainPid = run(
    "ssh",
    sshArgs(
      "systemctl",
      "show",
      strictServiceUnit,
      "--property=MainPID",
      "--value"
    )
  ).trim();
  assert.match(mainPid, /^[1-9][0-9]*$/, "hosted service has no live MainPID");
  const executable = run(
    "ssh",
    sshArgs("readlink", "-f", `/proc/${mainPid}/exe`)
  ).trim();
  const binarySha256 = run(
    "ssh",
    sshArgs("sha256sum", `/proc/${mainPid}/exe`)
  )
    .trim()
    .split(/\s+/)[0];
  const fragment = run(
    "ssh",
    sshArgs("systemctl", "show", strictServiceUnit, "--property=FragmentPath", "--value")
  ).trim();
  const serviceUnitSha256 = run(
    "ssh",
    sshArgs("sha256sum", fragment)
  )
    .trim()
    .split(/\s+/)[0];
  const renderedServiceConfiguration = run(
    "ssh",
    sshArgs("systemctl", "cat", strictServiceUnit)
  );
  const renderedServiceConfigurationSha256 = sha256(
    Buffer.from(renderedServiceConfiguration)
  );
  return {
    system_generation: systemGeneration,
    hosted_service_executable: executable,
    hosted_service_sha256: `sha256:${binarySha256}`,
    service_unit: fragment,
    service_unit_sha256: `sha256:${serviceUnitSha256}`,
    service_configuration_sha256: renderedServiceConfigurationSha256,
  };
}

function configurationProvenance() {
  const configRepo = path.resolve(
    process.env.CLUSTERFLUX_DEPLOYMENT_CONFIG_REPO ||
      path.join(repo, "..", "michelpaulissen.com")
  );
  const status = run("git", ["status", "--short"], { cwd: configRepo }).trim();
  const revision = run("git", ["rev-parse", "HEAD"], { cwd: configRepo }).trim();
  return {
    repository: configRepo,
    revision,
    evidence_identity: sha256(Buffer.from(revision)),
    clean: status === "",
    status: status || null,
  };
}

function proxyConfigurationProvenance() {
  if (!strictVpsRestart) return null;
  const proxyUnit = process.env.CLUSTERFLUX_STRICT_PROXY_UNIT || "nginx.service";
  const fragment = run(
    "ssh",
    sshArgs("systemctl", "show", "--property=FragmentPath", "--value", proxyUnit)
  ).trim();
  assert(fragment, `proxy service ${proxyUnit} has no FragmentPath`);
  const unitSha256 = run(
    "ssh",
    sshArgs("sha256sum", fragment)
  ).trim().split(/\s+/)[0];
  const proxyExecStart = run(
    "ssh",
    sshArgs("systemctl", "show", "--property=ExecStart", "--value", proxyUnit)
  ).trim();
  const nginxExecutable = proxyExecStart.match(/path=([^ ;]+)/)?.[1];
  const nginxConfiguration = proxyExecStart.match(/ argv\[\]=.* -c ([^ ;]+)/)?.[1];
  assert(nginxExecutable, `proxy service ${proxyUnit} has no executable identity`);
  assert(nginxConfiguration, `proxy service ${proxyUnit} has no configuration identity`);
  const renderedConfiguration = run(
    "ssh",
    sshArgs(nginxExecutable, "-T", "-c", nginxConfiguration)
  );
  const renderedConfigurationIdentity = sha256(Buffer.from(renderedConfiguration));
  const activeState = run(
    "ssh",
    sshArgs("systemctl", "is-active", proxyUnit)
  ).trim();
  assert.strictEqual(activeState, "active", `proxy service ${proxyUnit} is not active`);
  return {
    proxy_unit: proxyUnit,
    proxy_unit_fragment: fragment,
    proxy_unit_sha256: `sha256:${unitSha256}`,
    proxy_executable: nginxExecutable,
    proxy_configuration: nginxConfiguration,
    rendered_configuration_sha256: renderedConfigurationIdentity,
    evidence_identity: renderedConfigurationIdentity,
    active_state: activeState,
  };
}

async function restartHostedServiceAndResume({
  clusterflux,
  clusterfluxNode,
  projectDir,
  scope,
  workerArgs,
  workerNode,
  credentialPath,
  credentialDigest,
  scopedCollisionRuntime,
}) {
  if (!strictVpsRestart) {
    return {
      executed: false,
      reason: "strict VPS restart access not supplied",
      worker: null,
    };
  }
  const before = deploymentProvenance();
  run("ssh", sshArgs("systemctl", "restart", strictServiceUnit));
  const deadline = Date.now() + 120000;
  let ping;
  let lastError;
  while (Date.now() < deadline) {
    try {
      ping = await sendHostedControl({ type: "ping" });
      if (ping.type === "pong") break;
    } catch (error) {
      lastError = error;
    }
    await delay(500);
  }
  assert.strictEqual(
    ping?.type,
    "pong",
    `hosted service did not recover after restart: ${lastError?.message || "no pong"}`
  );
  const afterFirstRestart = deploymentProvenance();
  assert.strictEqual(afterFirstRestart.system_generation, before.system_generation);
  assert.strictEqual(
    afterFirstRestart.hosted_service_sha256,
    before.hosted_service_sha256
  );

  const projects = runJson(clusterflux, ["project", "list", ...scope], {
    cwd: projectDir,
  });
  assert(projects.project_count >= 1, "CLI session/project did not survive restart");
  const ephemeralRun = runJson(
    clusterflux,
    ["run", "build", "--project", ".", "--json"],
    { cwd: projectDir }
  );
  assert.strictEqual(ephemeralRun.status, "main_launched");
  const ephemeralProcess = ephemeralRun.process;
  await waitForCli(
    "pre-restart ephemeral main to park",
    () =>
      runJson(
        clusterflux,
        ["process", "status", ...scope, "--process", ephemeralProcess],
        { cwd: projectDir }
      ),
    (status) => status.live_process?.main_wait_state === "waiting_for_node",
    120000
  );

  run("ssh", sshArgs("systemctl", "restart", strictServiceUnit));
  const secondDeadline = Date.now() + 120000;
  ping = undefined;
  lastError = undefined;
  while (Date.now() < secondDeadline) {
    try {
      ping = await sendHostedControl({ type: "ping" });
      if (ping.type === "pong") break;
    } catch (error) {
      lastError = error;
    }
    await delay(500);
  }
  assert.strictEqual(
    ping?.type,
    "pong",
    `hosted service did not recover after ephemeral-state restart: ${
      lastError?.message || "no pong"
    }`
  );
  const after = deploymentProvenance();
  assert.strictEqual(after.system_generation, before.system_generation);
  assert.strictEqual(after.hosted_service_sha256, before.hosted_service_sha256);
  const oldStatus = runJson(
    clusterflux,
    ["process", "status", ...scope, "--process", ephemeralProcess],
    { cwd: projectDir }
  );
  assert.strictEqual(oldStatus.state, "not_active");
  assert.strictEqual(sha256(fs.readFileSync(credentialPath)), credentialDigest);

  const worker = spawnJsonLines(clusterfluxNode, workerArgs, {
    cwd: projectDir,
    env: { ...process.env },
  });
  const ready = await worker.waitFor(
    (value) => value.node_status === "ready",
    "persisted worker ready after hosted service restart"
  );
  assert.strictEqual(ready.node, workerNode);
  assert.strictEqual(sha256(fs.readFileSync(credentialPath)), credentialDigest);

  let scopedNodeCollisionAfterRestart = null;
  if (scopedCollisionRuntime) {
    assert.strictEqual(scopedCollisionRuntime.node, workerNode);
    assert.notStrictEqual(
      scopedCollisionRuntime.tenant,
      readJson(path.join(projectDir, ".clusterflux", "session.json")).tenant
    );
    assert.strictEqual(
      sha256(fs.readFileSync(scopedCollisionRuntime.credentialPath)),
      scopedCollisionRuntime.credentialDigest
    );
    const secondWorker = spawnJsonLines(
      clusterfluxNode,
      scopedCollisionRuntime.workerArgs,
      {
        cwd: scopedCollisionRuntime.projectDir,
        env: { ...process.env },
      }
    );
    try {
      const secondReady = await secondWorker.waitFor(
        (value) => value.node_status === "ready",
        "same-name second-tenant worker ready after hosted service restart"
      );
      assert.strictEqual(secondReady.node, workerNode);
      const firstStatus = await waitForCli(
        "first scoped same-name Node online after restart",
        () =>
          runJson(
            clusterflux,
            ["node", "status", ...scope, "--node", workerNode],
            { cwd: projectDir }
          ),
        (status) => nodeDescriptor(status, workerNode)?.online === true,
        30_000
      );
      const secondStatus = await waitForCli(
        "second scoped same-name Node online after restart",
        () =>
          runJson(
            clusterflux,
            [
              "node",
              "status",
              ...scopedCollisionRuntime.scope,
              "--node",
              workerNode,
            ],
            { cwd: scopedCollisionRuntime.projectDir }
          ),
        (status) => nodeDescriptor(status, workerNode)?.online === true,
        30_000
      );
      scopedNodeCollisionAfterRestart = {
        node: workerNode,
        first_scope_online:
          nodeDescriptor(firstStatus, workerNode)?.online === true,
        second_scope_online:
          nodeDescriptor(secondStatus, workerNode)?.online === true,
        both_credentials_reused_without_grants: true,
        distinct_credential_digests:
          credentialDigest !== scopedCollisionRuntime.credentialDigest,
      };
      assert.strictEqual(
        scopedNodeCollisionAfterRestart.distinct_credential_digests,
        true,
        "same-name scoped Nodes unexpectedly reused one credential"
      );
    } finally {
      await stopChild(secondWorker.child);
    }
  }

  const restartedRun = runJson(
    clusterflux,
    ["run", "build", "--project", ".", "--json"],
    { cwd: projectDir }
  );
  assert.strictEqual(restartedRun.status, "main_launched");
  const restartedProcess = restartedRun.process;
  const completed = await waitForCli(
    "new process completion after hosted service restart",
    () =>
      runJson(
        clusterflux,
        ["task", "list", ...scope, "--process", restartedProcess],
        { cwd: projectDir }
      ),
    (tasks) => completedFlagship(rawTaskEvents(tasks)),
    5 * 60 * 1000
  );
  const completedEvent = rawTaskEvents(completed).find(
    (event) =>
      event.executor === "coordinator_main" &&
      event.terminal_state === "completed"
  );
  assert(completedEvent, "post-restart flagship omitted its completed main event");
  return {
    executed: true,
    account_project_session_preserved: true,
    node_identity_preserved: true,
    old_process_terminated_honestly: true,
    terminated_ephemeral_process: ephemeralProcess,
    new_process: restartedProcess,
    new_process_result: completedEvent.result,
    scoped_node_collision_after_restart: scopedNodeCollisionAfterRestart,
    before,
    after_first_restart: afterFirstRestart,
    after,
    worker,
  };
}

async function finishScopedNodeCollisionAfterRestart({
  clusterflux,
  projectDir,
  scope,
  workerNode,
  scopedCollisionRuntime,
}) {
  const startedAt = Date.now();
  assert(scopedCollisionRuntime, "scoped collision runtime evidence is required");
  const revoked = runJson(
    clusterflux,
    [
      "node",
      "revoke",
      ...scopedCollisionRuntime.scope,
      "--node",
      scopedCollisionRuntime.node,
      "--yes",
    ],
    { cwd: scopedCollisionRuntime.projectDir }
  );
  assert.strictEqual(revoked.command, "node revoke");
  const revokedHeartbeat = assertDenied(
    await sendHostedControl({
      type: "node_heartbeat",
      tenant: scopedCollisionRuntime.tenant,
      project: scopedCollisionRuntime.project,
      node: scopedCollisionRuntime.node,
      node_signature: signedNodeHeartbeat(
        scopedCollisionRuntime.tenant,
        scopedCollisionRuntime.project,
        scopedCollisionRuntime.node,
        scopedCollisionRuntime.identity,
        {
          nonce: `strict-scoped-node-revoked-${Date.now()}`,
        }
      ),
    }),
    /not enrolled|unknown node|revoked|credential/i,
    "revoked second scoped Node"
  );
  const firstStatus = await waitForCli(
    "first same-name Node remains live after scoped revocation",
    () =>
      runJson(
        clusterflux,
        ["node", "status", ...scope, "--node", workerNode],
        { cwd: projectDir }
      ),
    (status) => nodeDescriptor(status, workerNode)?.online === true,
    30_000
  );
  return {
    request: {
      revoked_scope: {
        tenant: scopedCollisionRuntime.tenant,
        project: scopedCollisionRuntime.project,
        node: scopedCollisionRuntime.node,
      },
      retained_node: workerNode,
    },
    expected:
      "revocation affects only the selected scoped Node after both same-name credentials survive restart",
    observed: {
      revoked_command: revoked.command,
      revoked_credential_denied: revokedHeartbeat.denied,
      first_scope_node_online:
        nodeDescriptor(firstStatus, workerNode)?.online === true,
    },
    duration_ms: Math.max(1, Date.now() - startedAt),
  };
}

async function finishUserCredentialSecurity({
  clusterflux,
  projectDir,
  scope,
  sessionSecret,
}) {
  const startedAt = Date.now();
  const forged = assertDenied(
    await sendHostedControl(
      authenticatedRequest(
        `forged-session-${crypto.randomBytes(24).toString("hex")}`,
        { type: "list_projects" }
      )
    ),
    /not recognized|invalid|authentication|session/i,
    "forged user session"
  );

  let expired = null;
  const expiredFile = process.env.CLUSTERFLUX_EXPIRED_USER_SESSION_FILE;
  if (expiredFile) {
    const oldSession = readJson(path.resolve(expiredFile));
    assert(oldSession.session_secret, "expired session evidence omitted its secret");
    expired = assertDenied(
      await sendHostedControl(
        authenticatedRequest(oldSession.session_secret, { type: "list_projects" })
      ),
      /expired|not recognized/i,
      "expired user session"
    );
  } else if (strictFullRelease) {
    throw new Error(
      "strict full release requires CLUSTERFLUX_EXPIRED_USER_SESSION_FILE for a genuinely expired hosted session"
    );
  }

  const logout = runJson(clusterflux, ["logout", ...scope, "--yes"], {
    cwd: projectDir,
  });
  assert.strictEqual(logout.server_session_revocation.revoked, true);
  const revoked = assertDenied(
    await sendHostedControl(
      authenticatedRequest(sessionSecret, { type: "list_projects" })
    ),
    /revoked|not recognized/i,
    "revoked user session"
  );
  return {
    forged,
    expired,
    revoked,
    duration_ms: Date.now() - startedAt,
  };
}

async function dapVariables(client, variablesReference) {
  const request = client.send("variables", { variablesReference });
  return (await client.response(request, "variables")).body.variables;
}

async function runSameDefinitionDapIdentity({
  clusterfluxDap,
  clusterflux,
  projectDir,
  scope,
}) {
  const scenarioStartedAt = Date.now();
  const baselineContainers = new Set(
    run("podman", ["ps", "-q"])
      .trim()
      .split(/\s+/)
      .filter(Boolean)
  );
  const client = new DapClient({
    cwd: projectDir,
    command: clusterfluxDap,
    args: [],
    env: {
      ...process.env,
      CLUSTERFLUX_TEST_DAP_OBSERVER_CONNECTION_LOSS: "1",
      CLUSTERFLUX_TEST_DAP_POST_COMMIT_OBSERVATION_FAILURE: "1",
    },
  });
  let disconnected = false;
  try {
    const initialize = client.send("initialize", {
      adapterID: "clusterflux",
      linesStartAt1: true,
      columnsStartAt1: true,
    });
    await client.response(initialize, "initialize");
    const launch = client.send("launch", {
      entry: "identity",
      project: projectDir,
      runtimeBackend: "live-services",
      coordinatorEndpoint: serviceEndpoint,
    });
    await client.response(launch, "launch");
    await client.waitFor(
      (message) => message.type === "event" && message.event === "initialized"
    );
    const configurationDone = client.send("configurationDone");
    await client.response(configurationDone, "configurationDone");

    const mainThreadDeadline = Date.now() + 5 * 60 * 1000;
    let mainThread;
    while (Date.now() < mainThreadDeadline) {
      const threadsRequest = client.send("threads");
      const threads = (await client.response(threadsRequest, "threads")).body
        .threads;
      mainThread = threads.find((thread) => /coordinator main/i.test(thread.name));
      if (mainThread) break;
      await delay(100);
    }
    assert(mainThread, "identity entry did not report its coordinator main");
    const launchWithoutBreakpointMs = Date.now() - scenarioStartedAt;
    const postCommitDiagnostics = [
      "initial task observation failed after main_launched",
      "initial process observation failed after main_launched",
    ].map((expected) => {
      const message = client.messages.find(
        (candidate) =>
          candidate.type === "event" &&
          candidate.event === "output" &&
          String(candidate.body?.output || "").includes(expected)
      );
      assert(message, `post-commit launch evidence omitted: ${expected}`);
      return String(message.body.output).trim();
    });
    const prematurePostCommitEvents = client.messages.filter(
      (message) =>
        message.type === "event" &&
        ["stopped", "terminated"].includes(message.event)
    );
    assert.deepStrictEqual(prematurePostCommitEvents, []);
    const processRequest = client.send("evaluate", {
      expression: "virtual_process_id",
      context: "watch",
    });
    const processId = (await client.response(processRequest, "evaluate")).body
      .result;
    const identityContainerDeadline = Date.now() + 2 * 60 * 1000;
    let identityContainerIds = new Set();
    let maximumNewContainerCount = 0;
    let lastPodmanStates = [];
    while (Date.now() < identityContainerDeadline) {
      const containers = runJson("podman", ["ps", "--format", "json"]);
      lastPodmanStates = containers;
      identityContainerIds = new Set(
        containers
          .map((container) => container.Id || container.ID || container.IdHex || "")
          .filter((id) => id && !baselineContainers.has(id))
      );
      maximumNewContainerCount = Math.max(
        maximumNewContainerCount,
        identityContainerIds.size
      );
      if (identityContainerIds.size >= 2) break;
      await delay(100);
    }
    const identityTaskSnapshot =
      identityContainerIds.size >= 2
        ? null
        : runJson(
            clusterflux,
            ["task", "list", ...scope, "--process", processId],
            { cwd: projectDir }
          );
    assert(
      identityContainerIds.size >= 2,
      `identity entry did not start two same-definition Podman containers; max=${maximumNewContainerCount} tasks=${JSON.stringify(
        identityTaskSnapshot
      )} podman=${JSON.stringify(lastPodmanStates)}`
    );

    const pause = client.send("pause", { threadId: mainThread.id });
    await client.response(pause, "pause");
    const paused = await client.waitFor(
      (message) =>
        message.type === "event" &&
        message.event === "stopped" &&
        message.body.reason === "pause",
      30_000
    );
    assert.strictEqual(typeof paused.body.allThreadsStopped, "boolean");

    const podmanStates = runJson("podman", ["ps", "--all", "--format", "json"]);
    const clusterfluxPausedContainers = podmanStates.filter((container) => {
      const id = container.Id || container.ID || container.IdHex || "";
      const state = String(container.State || container.Status || "").toLowerCase();
      return identityContainerIds.has(id) && state.includes("paused");
    });
    assert(
      clusterfluxPausedContainers.length >= 2,
      `expected two newly paused Clusterflux containers: ${JSON.stringify(
        podmanStates
      )}`
    );

    const threadsRequest = client.send("threads");
    const threads = (await client.response(threadsRequest, "threads")).body
      .threads;
    const identityThreads = threads.filter((thread) =>
      /identity probe/i.test(thread.name)
    );
    assert.strictEqual(
      identityThreads.length,
      2,
      `expected two same-definition DAP threads: ${JSON.stringify(threads)}`
    );
    assert.notStrictEqual(identityThreads[0].id, identityThreads[1].id);

    const threadEvidence = [];
    for (const threadRecord of identityThreads) {
      const stackRequest = client.send("stackTrace", {
        threadId: threadRecord.id,
        startFrame: 0,
        levels: 1,
      });
      const frames = (await client.response(stackRequest, "stackTrace")).body
        .stackFrames;
      assert.strictEqual(frames.length, 1);
      const scopesRequest = client.send("scopes", { frameId: frames[0].id });
      const scopes = (await client.response(scopesRequest, "scopes")).body.scopes;
      const argsScope = scopes.find(
        (scopeRecord) => scopeRecord.name === "Task Args and Handles"
      );
      const runtimeScope = scopes.find(
        (scopeRecord) => scopeRecord.name === "Clusterflux Runtime"
      );
      assert(argsScope && runtimeScope);
      const args = await dapVariables(client, argsScope.variablesReference);
      const runtime = await dapVariables(client, runtimeScope.variablesReference);
      const runtimeValue = (name) =>
        runtime.find((variable) => variable.name === name)?.value;
      assert.strictEqual(runtimeValue("state"), "Frozen");
      threadEvidence.push({
        thread_id: threadRecord.id,
        task_instance: runtimeValue("virtual_thread"),
        attempt_id: runtimeValue("task_attempt_id"),
        arguments: args.map((variable) => ({
          name: variable.name,
          value: variable.value,
        })),
      });
    }
    assert.notStrictEqual(
      threadEvidence[0].task_instance,
      threadEvidence[1].task_instance
    );
    assert.notStrictEqual(threadEvidence[0].attempt_id, threadEvidence[1].attempt_id);
    const argumentText = threadEvidence
      .flatMap((record) => record.arguments.map((argument) => argument.value))
      .join(" ");
    assert.match(argumentText, /slow-first/);
    assert.match(argumentText, /fast-second/);

    const continued = client.send("continue", {
      threadId: paused.body.threadId,
    });
    await client.response(continued, "continue");
    await client.waitFor(
      (message) => message.type === "event" && message.event === "terminated",
      5 * 60 * 1000
    );

    const taskList = runJson(
      clusterflux,
      ["task", "list", ...scope, "--process", processId],
      { cwd: projectDir }
    );
    const identityEvents = rawTaskEvents(taskList).filter(
      (event) =>
        event.task_definition === "identity_probe" &&
        event.terminal_state === "completed"
    );
    assert.strictEqual(identityEvents.length, 2, JSON.stringify(identityEvents));
    assert.notStrictEqual(identityEvents[0].task, identityEvents[1].task);
    assert.notStrictEqual(identityEvents[0].attempt_id, identityEvents[1].attempt_id);
    assert.match(JSON.stringify(identityEvents[0].result), /fast-second/);
    assert.match(JSON.stringify(identityEvents[1].result), /slow-first/);

    const released = await waitForCli(
      "same-definition identity process release",
      () =>
        runJson(
          clusterflux,
          ["process", "status", ...scope, "--process", processId],
          { cwd: projectDir }
        ),
      (status) => status.state === "not_active",
      120_000
    );
    await client.close();
    disconnected = true;
    return {
      duration_ms: Date.now() - scenarioStartedAt,
      process: processId,
      launch_without_breakpoint: {
        request: {
          entry: "identity",
          runtime_backend: "live-services",
          configured_breakpoints: [],
        },
        expected: "launch commits and reports a real coordinator-main thread",
        observed: {
          thread_id: mainThread.id,
          thread_name: mainThread.name,
        },
        duration_ms: launchWithoutBreakpointMs,
      },
      post_main_launched_observation_recovery: {
        request: {
          entry: "identity",
          fault_injections: [
            "initial task snapshot observation failure",
            "initial process-status observation failure",
          ],
        },
        expected:
          "main_launched is committed, no rollback or false stop occurs, and the observer reconnects until threads appear",
        observed: {
          diagnostics: postCommitDiagnostics,
          main_thread: mainThread,
          premature_stop_or_termination_events:
            prematurePostCommitEvents.length,
        },
        duration_ms: launchWithoutBreakpointMs,
      },
      same_definition_identity: {
        expected:
          "two concurrent instances retain distinct thread, argument, attempt, output, freeze, and terminal identities",
        observed: {
          thread_evidence: threadEvidence,
          completion_order: identityEvents.map((event) => ({
            task_instance: event.task,
            attempt_id: event.attempt_id,
            result: event.result,
          })),
          paused_container_count: clusterfluxPausedContainers.length,
          terminal_state: released.state,
        },
        duration_ms: Date.now() - scenarioStartedAt,
      },
      thread_evidence: threadEvidence,
      completion_order: identityEvents.map((event) => ({
        task_instance: event.task,
        attempt_id: event.attempt_id,
        result: event.result,
      })),
      podman_paused_container_ids: clusterfluxPausedContainers.map(
        (container) => container.Id || container.ID || container.IdHex
      ),
      dap_all_threads_stopped: paused.body.allThreadsStopped,
      terminal_state: released.state,
    };
  } finally {
    if (!disconnected) {
      await client.close().catch(() => {
        if (client.child.exitCode === null) client.child.kill("SIGKILL");
      });
    }
  }
}

async function runLiveDapEditRestart({
  clusterfluxDap,
  clusterflux,
  projectDir,
  scope,
  worker,
}) {
  const scenarioStartedAt = Date.now();
  const sourcePath = fs.realpathSync(path.join(projectDir, "src/lib.rs"));
  const originalSource = fs.readFileSync(sourcePath, "utf8");
  const sourceLines = originalSource.split(/\r?\n/);
  const lineFor = (needle) => {
    const line = sourceLines.findIndex((candidate) => candidate.includes(needle)) + 1;
    assert(line > 0, `flagship source omitted ${needle}`);
    return line;
  };
  const mainLine = lineFor("pub async fn restart_main()");
  const taskLine = lineFor("fn task_trap(");
  const client = new DapClient({
    cwd: projectDir,
    command: clusterfluxDap,
    args: [],
    env: {
      ...process.env,
      CLUSTERFLUX_TEST_DAP_OBSERVER_CONNECTION_LOSS: "1",
      CLUSTERFLUX_TEST_DAP_OBSERVER_SNAPSHOT_FAILURE: "1",
      CLUSTERFLUX_TEST_DAP_OBSERVER_PROCESS_STATUS_FAILURE: "1",
      CLUSTERFLUX_TEST_DAP_OBSERVER_FALLBACK_FAILURE: "1",
      CLUSTERFLUX_TEST_DAP_OBSERVER_DEBUG_EPOCH_WAIT_FAILURE: "1",
      CLUSTERFLUX_TEST_DAP_BREAKPOINT_DELAY_REVISION: "1",
      CLUSTERFLUX_TEST_DAP_BREAKPOINT_DELAY_MS: "750",
      CLUSTERFLUX_TEST_DAP_BREAKPOINT_INSTALLATION_FAILURE_REVISION: "3",
    },
  });
  let sourceRestored = false;
  let disconnected = false;
  try {
    const initialize = client.send("initialize", {
      adapterID: "clusterflux",
      linesStartAt1: true,
      columnsStartAt1: true,
    });
    await client.response(initialize, "initialize");

    const launch = client.send("launch", {
      entry: "restart",
      project: projectDir,
      runtimeBackend: "live-services",
      coordinatorEndpoint: serviceEndpoint,
    });
    await client.response(launch, "launch");
    await client.waitFor(
      (message) => message.type === "event" && message.event === "initialized"
    );

    const configurationStartedAt = Date.now();
    const configurationDone = client.send("configurationDone");
    const configurationResponse = await client.response(
      configurationDone,
      "configurationDone"
    );
    const configurationResponseMs = Date.now() - configurationStartedAt;
    assert(
      configurationResponseMs < 2_000,
      `configurationDone blocked for ${configurationResponseMs} ms`
    );

    const threadDeadline = Date.now() + 5 * 60 * 1000;
    let runningThreads = [];
    let threadObservationResponse;
    while (Date.now() < threadDeadline) {
      const threadsRequest = client.send("threads");
      threadObservationResponse = await client.response(threadsRequest, "threads");
      runningThreads = threadObservationResponse.body.threads;
      if (runningThreads.length > 0) break;
      await delay(100);
    }
    assert(runningThreads.length > 0, "asynchronous live DAP launch reported no threads");
    const initialObserverFaults = [];
    for (const expected of [
      {
        fault: "task snapshot read transport failure",
        diagnostic: "runtime snapshot observation failed",
      },
      {
        fault: "process-status read transport failure",
        diagnostic: "runtime process observation failed",
      },
      {
        fault: "breakpoint inspection and fallback event transport failures",
        diagnostic: "runtime breakpoint and fallback event observation failed",
      },
    ]) {
      const message = await client.waitFor(
        (candidate) =>
          candidate.seq > configurationResponse.seq &&
          candidate.type === "event" &&
          candidate.event === "output" &&
          String(candidate.body?.output || "").includes(expected.diagnostic),
        120_000
      );
      initialObserverFaults.push({
        ...expected,
        expected: "reconnect and eventually report live threads without a stop",
        observed: String(message.body.output).trim(),
      });
    }
    const recoveredThreadsRequest = client.send("threads");
    const recoveredThreadsResponse = await client.response(
      recoveredThreadsRequest,
      "threads"
    );
    assert(
      recoveredThreadsResponse.body.threads.length > 0,
      "observer faults removed threads after recovery"
    );
    const prematureLaunchEvents = client.messages.filter(
      (message) =>
        message.seq > configurationResponse.seq &&
        message.seq < recoveredThreadsResponse.seq &&
        message.type === "event" &&
        ["stopped", "terminated"].includes(message.event)
    );
    assert.deepStrictEqual(
      prematureLaunchEvents,
      [],
      "post-launch observer recovery fabricated a stop or terminated the launch"
    );
    const postLaunchObservationRecoveryMs =
      Date.now() - configurationStartedAt;
    const pauseStartedAt = Date.now();
    const pause = client.send("pause", { threadId: runningThreads[0].id });
    await client.response(pause, "pause");
    const pauseResponseMs = Date.now() - pauseStartedAt;
    assert(pauseResponseMs < 2_000, `pause blocked for ${pauseResponseMs} ms`);
    const mainStop = await client.waitFor(
      (message) =>
        message.type === "event" &&
        message.event === "stopped" &&
        message.body.reason === "pause"
    );
    assert.strictEqual(mainStop.body.allThreadsStopped, true);

    const inspectStartedAt = Date.now();
    const commandStatus = client.send("evaluate", {
      expression: "command_status",
      context: "watch",
    });
    const inspected = await client.response(commandStatus, "evaluate");
    const inspectionResponseMs = Date.now() - inspectStartedAt;
    assert(
      inspectionResponseMs < 2_000,
      `inspection blocked for ${inspectionResponseMs} ms`
    );
    assert.match(inspected.body.result, /debug epoch|frozen/i);

    const breakpointStartedAt = Date.now();
    const staleRevisionRequest = client.send("setBreakpoints", {
      source: { path: sourcePath },
      breakpoints: [{ line: mainLine }],
    });
    const newestRevisionRequest = client.send("setBreakpoints", {
      source: { path: sourcePath },
      breakpoints: [{ line: taskLine }],
    });
    const staleRevisionResponse = await client.response(
      staleRevisionRequest,
      "setBreakpoints"
    );
    const newestRevisionResponse = await client.response(
      newestRevisionRequest,
      "setBreakpoints"
    );
    const breakpointResponseMs = Date.now() - breakpointStartedAt;
    assert(
      breakpointResponseMs < 2_000,
      `setBreakpoints blocked for ${breakpointResponseMs} ms`
    );
    assert.deepStrictEqual(
      staleRevisionResponse.body.breakpoints.map(
        (breakpoint) => breakpoint.verified
      ),
      [false]
    );
    assert.deepStrictEqual(
      newestRevisionResponse.body.breakpoints.map(
        (breakpoint) => breakpoint.verified
      ),
      [false]
    );
    assert.deepStrictEqual(
      newestRevisionResponse.body.breakpoints.map(
        (breakpoint) => breakpoint.message
      ),
      ["Pending coordinator breakpoint installation"]
    );
    const newestRevisionInstalled = await client.waitFor(
      (message) =>
        message.seq > newestRevisionResponse.seq &&
        message.type === "event" &&
        message.event === "breakpoint" &&
        message.body.reason === "changed" &&
        message.body.breakpoint?.verified === true &&
        message.body.breakpoint?.line === taskLine,
      30_000
    );
    await delay(1_000);
    const staleRevisionEvents = client.messages.filter(
      (message) =>
        message.seq > newestRevisionResponse.seq &&
        message.type === "event" &&
        message.event === "breakpoint" &&
        message.body.reason === "changed" &&
        message.body.breakpoint?.line === mainLine
    );
    assert.deepStrictEqual(
      staleRevisionEvents,
      [],
      "stale breakpoint revision completion overwrote the newest set"
    );

    const rejectedRevisionRequest = client.send("setBreakpoints", {
      source: { path: sourcePath },
      breakpoints: [{ line: taskLine }],
    });
    const rejectedRevisionResponse = await client.response(
      rejectedRevisionRequest,
      "setBreakpoints"
    );
    const rejectedBreakpoint = await client.waitFor(
      (message) =>
        message.seq > rejectedRevisionResponse.seq &&
        message.type === "event" &&
        message.event === "breakpoint" &&
        message.body.reason === "changed" &&
        message.body.breakpoint?.verified === false &&
        message.body.breakpoint?.line === taskLine &&
        /coordinator breakpoint installation failed/i.test(
          message.body.breakpoint?.message || ""
        ),
      30_000
    );
    assert.strictEqual(
      rejectedBreakpoint.body.breakpoint.id,
      rejectedRevisionResponse.body.breakpoints[0].id
    );
    const retryBreakpoints = client.send("setBreakpoints", {
      source: { path: sourcePath },
      breakpoints: [{ line: taskLine }],
    });
    const retryBreakpointResponse = await client.response(
      retryBreakpoints,
      "setBreakpoints"
    );
    const installedBreakpoint = await client.waitFor(
      (message) =>
        message.seq > retryBreakpointResponse.seq &&
        message.type === "event" &&
        message.event === "breakpoint" &&
        message.body.reason === "changed" &&
        message.body.breakpoint?.verified === true &&
        message.body.breakpoint?.line === taskLine,
      30_000
    );
    assert.strictEqual(
      installedBreakpoint.body.breakpoint.id,
      retryBreakpointResponse.body.breakpoints[0].id
    );
    const breakpointInstallationMs = Date.now() - breakpointStartedAt;

    const continueStartedAt = Date.now();
    const mainContinue = client.send("continue", {
      threadId: mainStop.body.threadId,
    });
    const mainContinueResponse = await client.response(mainContinue, "continue");
    const continueResponseMs = Date.now() - continueStartedAt;
    assert(
      continueResponseMs < 2_000,
      `continue blocked for ${continueResponseMs} ms`
    );
    const taskStop = await client.waitFor(
      (message) =>
        message.seq > mainStop.seq &&
        message.type === "event" &&
        message.event === "stopped" &&
        message.body.reason === "breakpoint"
    );
    const realBreakpointMs = Date.now() - breakpointStartedAt;
    assert.strictEqual(taskStop.body.allThreadsStopped, true);
    const observerReconnect = await client.waitFor(
      (message) =>
        message.seq > mainContinueResponse.seq &&
        message.type === "event" &&
        message.event === "output" &&
        String(message.body?.output || "").includes(
          "injected one transient connection loss"
        ),
      30_000
    );
    assert(
      observerReconnect.seq < taskStop.seq,
      "observer reconnection must complete before the real breakpoint stop"
    );
    assert.strictEqual(
      client.messages.filter(
        (message) =>
          message.seq > mainStop.seq &&
          message.seq < taskStop.seq &&
          message.type === "event" &&
          message.event === "stopped"
      ).length,
      0,
      "observer reconnection fabricated a stopped target before the real breakpoint"
    );
    assert.notStrictEqual(taskStop.body.threadId, mainStop.body.threadId);
    const reconnectDiagnostics = [
      "runtime observer injected one transient connection loss",
      "runtime snapshot observation failed",
      "runtime process observation failed",
      "runtime breakpoint and fallback event observation failed",
      "runtime Debug Epoch observation failed",
    ].map((diagnostic) => {
      const message = client.messages.find(
        (candidate) =>
          candidate.seq > mainContinueResponse.seq &&
          candidate.seq < taskStop.seq &&
          candidate.type === "event" &&
          candidate.event === "output" &&
          String(candidate.body?.output || "").includes(diagnostic)
      );
      assert(
        message,
        `observer reconnection evidence omitted diagnostic: ${diagnostic}`
      );
      return String(message.body.output).trim();
    });

    const stackTrace = client.send("stackTrace", {
      threadId: taskStop.body.threadId,
      startFrame: 0,
      levels: 1,
    });
    const taskFrames = (await client.response(stackTrace, "stackTrace")).body
      .stackFrames;
    assert.strictEqual(taskFrames[0].line, taskLine);

    const scopesRequest = client.send("scopes", { frameId: taskFrames[0].id });
    const scopes = (await client.response(scopesRequest, "scopes")).body.scopes;
    const argsScope = scopes.find(
      (candidate) => candidate.name === "Task Args and Handles"
    );
    const runtimeScope = scopes.find(
      (candidate) => candidate.name === "Clusterflux Runtime"
    );
    assert(argsScope && runtimeScope);
    const taskArgs = await dapVariables(client, argsScope.variablesReference);
    assert(
      taskArgs.some(
        (variable) =>
          variable.name === "arg_0" && String(variable.value).includes("0")
      )
    );
    const runtime = await dapVariables(client, runtimeScope.variablesReference);
    assert(
      runtime.some(
        (variable) =>
          variable.name === "runtime_backend" && variable.value === "LiveServices"
      )
    );
    assert(
      runtime.some(
        (variable) => variable.name === "state" && variable.value === "Frozen"
      )
    );
    const dapProcess = runtime.find(
      (variable) => variable.name === "virtual_process_id"
    )?.value;
    assert.match(dapProcess || "", /^vp-[0-9a-f]{12}$/);

    const taskContinue = client.send("continue", { threadId: taskStop.body.threadId });
    await client.response(taskContinue, "continue");
    const failedStop = await client.waitFor(
      (message) =>
        message.seq > taskStop.seq &&
        message.type === "event" &&
        message.event === "stopped" &&
        message.body.reason === "exception",
      5 * 60 * 1000
    );
    assert.strictEqual(failedStop.body.allThreadsStopped, false);

    const beforeEdit = runJson(
      clusterflux,
      ["task", "list", ...scope, "--process", dapProcess],
      { cwd: projectDir }
    );
    const originalTaskEvent = rawTaskEvents(beforeEdit).find(
      (event) =>
        event.task_definition === "task_trap" &&
        event.terminal_state === "failed"
    );
    assert(originalTaskEvent, "DAP restart entry omitted its original child event");
    assert(originalTaskEvent.attempt_id, "failed attempt omitted its identity");

    const originalTaskBody = `pub extern "C" fn task_trap(_input: i32) -> i32 {
    #[cfg(target_arch = "wasm32")]
    core::arch::wasm32::unreachable();
    #[cfg(not(target_arch = "wasm32"))]
    panic!("intentional task trap")
}`;
    const replacementTaskBody = `pub extern "C" fn task_trap(_input: i32) -> i32 {
    _input + 42 // strict hosted compatible restart probe
}`;
    const editedSource = originalSource.replace(originalTaskBody, replacementTaskBody);
    assert.notStrictEqual(editedSource, originalSource);
    fs.writeFileSync(sourcePath, editedSource);

    const restartFrame = client.send("restartFrame", {
      frameId: taskFrames[0].id,
    });
    const restartResponse = await client.response(restartFrame, "restartFrame");
    const restartedStop = await client.waitFor(
      (message) =>
        message.seq > restartResponse.seq &&
        message.type === "event" &&
        message.event === "stopped" &&
        message.body.reason === "breakpoint"
    );
    assert.strictEqual(restartedStop.body.allThreadsStopped, true);
    const restartedContinue = client.send("continue", {
      threadId: restartedStop.body.threadId,
    });
    await client.response(restartedContinue, "continue");
    const restartedNodeReport = await worker.waitFor(
      (value) =>
        value.node_status === "completed" &&
        value.virtual_thread === originalTaskEvent.task &&
        value.task_assignment_response?.task_spec?.task_definition ===
          "task_trap",
      "edited task to execute on the live restarted node",
      5 * 60 * 1000
    );
    assert.strictEqual(
      restartedNodeReport.node_status,
      "completed",
      `edited task restart failed on the live node: ${JSON.stringify(restartedNodeReport)}`
    );
    assert.match(restartedNodeReport.stdout_tail, /"SmallJson":42/);
    const restartedTask = restartedNodeReport.virtual_thread;

    const afterEdit = await waitForCli(
      "edited live task event",
      () =>
        runJson(clusterflux, ["task", "list", ...scope, "--process", dapProcess], {
          cwd: projectDir,
        }),
      (tasks) =>
        rawTaskEvents(tasks).some(
          (event) =>
            event.task === restartedTask &&
            event.terminal_state === "completed" &&
            JSON.stringify(event.result) === JSON.stringify({ SmallJson: 42 })
        ),
      5 * 60 * 1000
    );
    const restartedEvent = rawTaskEvents(afterEdit).find(
      (event) =>
        event.task === restartedTask &&
        event.terminal_state === "completed" &&
        JSON.stringify(event.result) === JSON.stringify({ SmallJson: 42 })
    );
    assert(restartedEvent, "replacement attempt omitted its completed event");
    assert(restartedEvent.attempt_id, "replacement attempt omitted its identity");
    assert.notStrictEqual(restartedEvent.attempt_id, originalTaskEvent.attempt_id);
    await client.waitFor(
      (message) => message.type === "event" && message.event === "terminated",
      5 * 60 * 1000
    );
    const disconnectStartedAt = Date.now();
    await client.close();
    disconnected = true;
    const disconnectResponseMs = Date.now() - disconnectStartedAt;
    assert(
      disconnectResponseMs < 2_000,
      `disconnect blocked for ${disconnectResponseMs} ms`
    );
    fs.writeFileSync(sourcePath, originalSource);
    sourceRestored = true;

    return {
      duration_ms: Date.now() - scenarioStartedAt,
      process: dapProcess,
      main_source_line: mainLine,
      task_breakpoint_line: taskLine,
      launch_started_without_breakpoint: true,
      configuration_response_ms: configurationResponseMs,
      pause_response_ms: pauseResponseMs,
      inspection_response_ms: inspectionResponseMs,
      breakpoint_response_ms: breakpointResponseMs,
      breakpoint_installation_ms: breakpointInstallationMs,
      continue_response_ms: continueResponseMs,
      disconnect_response_ms: disconnectResponseMs,
      pause_all_threads_stopped: mainStop.body.allThreadsStopped,
      task_all_threads_stopped: taskStop.body.allThreadsStopped,
      original_task_instance: originalTaskEvent.task,
      original_attempt: originalTaskEvent.attempt_id,
      restarted_task_instance: restartedTask,
      restarted_attempt: restartedEvent.attempt_id,
      restarted_result: restartedEvent.result,
      compatible_source_edit: "task_trap: trap -> input + 42",
      original_arguments_preserved: true,
      clean_vfs_boundary_used: true,
      breakpoint_install_failure_reported: true,
      real_breakpoint: {
        request: {
          source: sourcePath,
          line: taskLine,
          runtime_backend: "live-services",
        },
        expected:
          "coordinator installation is verified before a real all-stop breakpoint event",
        observed: {
          installed_line: installedBreakpoint.body.breakpoint.line,
          stop_reason: taskStop.body.reason,
          stopped_line: taskFrames[0].line,
          all_threads_stopped: taskStop.body.allThreadsStopped,
        },
        duration_ms: Math.max(1, realBreakpointMs),
      },
      debug_interaction: {
        request: "Pause, inspect command_status, Continue, and Disconnect",
        expected: "each control remains responsive during observer recovery",
        observed: {
          pause_response_ms: pauseResponseMs,
          inspect_response_ms: inspectionResponseMs,
          continue_response_ms: continueResponseMs,
          disconnect_response_ms: disconnectResponseMs,
          pause_stop_reason: mainStop.body.reason,
        },
        duration_ms: Math.max(
          1,
          pauseResponseMs +
            inspectionResponseMs +
            continueResponseMs +
            disconnectResponseMs
        ),
      },
      post_launch_observation_recovery: {
        fault_injections: initialObserverFaults,
        expected:
          "main_launched remains committed; reconnect; threads appear; no rollback or fabricated stop",
        observed: {
          threads: runningThreads,
          premature_stop_or_termination_events: prematureLaunchEvents.length,
          terminal_state: "completed after replacement attempt",
        },
        duration_ms: postLaunchObservationRecoveryMs,
      },
      observer_reconnection: {
        fault_injections: reconnectDiagnostics,
        expected:
          "all recoverable reads reconnect before the real breakpoint stop",
        observed: {
          real_stop_reason: taskStop.body.reason,
          false_stops_before_real_stop: 0,
        },
        duration_ms: Date.now() - continueStartedAt,
      },
      breakpoint_revisions: {
        expected: "revision 2 remains authoritative after delayed revision 1",
        delayed_revision: 1,
        authoritative_revision: 2,
        authoritative_line: newestRevisionInstalled.body.breakpoint.line,
        stale_completion_events: staleRevisionEvents.length,
        failure_revision: 3,
        recovery_revision: 4,
        observed: "newest revision verified; stale completion ignored",
        duration_ms: breakpointInstallationMs,
      },
      observer_idle_request_rate_bound_per_second: 0.8,
      observer_reconnected_without_false_stop: true,
      observer_fallback_reconnected: true,
      observer_debug_epoch_wait_reconnected: true,
    };
  } finally {
    if (!sourceRestored) fs.writeFileSync(sourcePath, originalSource);
    if (!disconnected) {
      await client.close().catch(() => {
        if (client.child.exitCode === null) client.child.kill("SIGKILL");
      });
    }
  }
}

function copyProjectControlState(sourceRoot, targetRoot) {
  const source = path.join(sourceRoot, ".clusterflux");
  const target = path.join(targetRoot, ".clusterflux");
  ensureDir(target);
  for (const file of ["session.json", "project.json"]) {
    const from = path.join(source, file);
    assert(fs.existsSync(from), `missing shared project control state ${from}`);
    fs.copyFileSync(from, path.join(target, file));
  }
  fs.chmodSync(path.join(target, "session.json"), 0o600);
  const sourceNodes = path.join(source, "nodes");
  assert(fs.existsSync(sourceNodes), "persisted node credential directory is missing");
  fs.cpSync(sourceNodes, path.join(target, "nodes"), {
    recursive: true,
    force: true,
  });
}

async function runHostedHelloBuild({
  clusterflux,
  projectDir,
  scope,
  outputFile,
}) {
  const startedAt = Date.now();
  const runReport = runJson(
    clusterflux,
    ["run", "build", "--project", ".", "--json"],
    { cwd: projectDir }
  );
  assert.strictEqual(runReport.status, "main_launched");
  const tasks = await waitForCli(
    "hosted hello-build completion",
    () =>
      runJson(
        clusterflux,
        ["task", "list", ...scope, "--process", runReport.process],
        { cwd: projectDir }
      ),
    (report) => {
      const events = rawTaskEvents(report);
      return events.some(
        (event) =>
          event.task_definition === "compile" &&
          event.terminal_state === "completed"
      );
    },
    10 * 60 * 1000
  );
  const events = rawTaskEvents(tasks);
  assert(
    events.some(
      (event) =>
        event.task_definition === "snapshot_current_project" &&
        event.terminal_state === "completed"
    )
  );
  const released = await waitForCli(
    "hosted hello-build process release",
    () =>
      runJson(
        clusterflux,
        ["process", "status", ...scope, "--process", runReport.process],
        { cwd: projectDir }
      ),
    (status) => status.state === "not_active",
    120_000
  );
  const artifacts = runJson(
    clusterflux,
    ["artifact", "list", ...scope, "--process", runReport.process],
    { cwd: projectDir }
  );
  const artifact = artifacts.artifacts.find(
    (candidate) =>
      typeof candidate.digest === "string" &&
      /^sha256:[0-9a-f]{64}$/.test(candidate.digest) &&
      candidate.artifact.startsWith("hello-clusterflux-")
  );
  assert(artifact, "hello-build did not retain its executable artifact");
  await waitForCli(
    "hosted hello-build retaining node online",
    () => runJson(clusterflux, ["node", "list", ...scope], { cwd: projectDir }),
    (report) =>
      report.response?.descriptors?.some(
        (descriptor) =>
          descriptor.online === true &&
          descriptor.artifact_locations?.includes(artifact.artifact)
      ),
    120_000
  );
  const download = runJson(
    clusterflux,
    [
      "artifact",
      "download",
      ...scope,
      artifact.artifact,
      "--to",
      outputFile,
    ],
    { cwd: projectDir }
  );
  assert.strictEqual(download.local_download.verified_digest, artifact.digest);
  assert.strictEqual(sha256(fs.readFileSync(outputFile)), artifact.digest);
  fs.chmodSync(outputFile, 0o755);
  const output = run(outputFile, []).trim();
  assert.strictEqual(output, "hello from a real Clusterflux build");
  return {
    duration_ms: Date.now() - startedAt,
    process: runReport.process,
    terminal_state: released.state,
    artifact: artifact.artifact,
    digest: artifact.digest,
    downloaded_bytes: download.local_download.bytes_written,
    downloaded_sha256: sha256(fs.readFileSync(outputFile)),
    executable_output: output,
  };
}

async function runHostedRecoveryBuild({
  clusterfluxDap,
  clusterflux,
  projectDir,
  scope,
}) {
  const startedAt = Date.now();
  const sourcePath = fs.realpathSync(path.join(projectDir, "src/lib.rs"));
  const original = fs.readFileSync(sourcePath, "utf8");
  const failing = '"exit 23".to_owned()';
  const replacement =
    '"printf \'recovered\\n\' > /clusterflux/output/recovering.txt".to_owned()';
  assert(original.includes(failing));
  const client = new DapClient({
    cwd: projectDir,
    command: clusterfluxDap,
    env: { ...process.env, CLUSTERFLUX_DAP_TIMEOUT_MS: "120000" },
  });
  let restored = false;
  let closed = false;
  try {
    const initialize = client.send("initialize", {
      adapterID: "clusterflux",
      linesStartAt1: true,
      columnsStartAt1: true,
    });
    await client.response(initialize, "initialize");
    const launch = client.send("launch", {
      entry: "build",
      project: projectDir,
      runtimeBackend: "live-services",
      coordinatorEndpoint: serviceEndpoint,
    });
    await client.response(launch, "launch");
    await client.waitFor(
      (message) => message.type === "event" && message.event === "initialized"
    );
    const configured = client.send("configurationDone");
    await client.response(configured, "configurationDone");
    const failedStop = await client.waitFor(
      (message) =>
        message.type === "event" &&
        message.event === "stopped" &&
        message.body.reason === "exception",
      10 * 60 * 1000
    );
    assert.strictEqual(failedStop.body.allThreadsStopped, false);
    const threadRequest = client.send("threads");
    const threads = (await client.response(threadRequest, "threads")).body.threads;
    const recoveringThread = threads.find((thread) => /build lane/.test(thread.name));
    assert(recoveringThread, "recovery-build failed lane is not visible in DAP");
    const processId = threads
      .map((thread) => /vp-[0-9a-f]{12}/.exec(thread.name)?.[0])
      .find(Boolean);
    assert(processId, "recovery-build DAP threads omitted the process identity");
    const before = runJson(
      clusterflux,
      ["task", "list", ...scope, "--process", processId],
      { cwd: projectDir }
    );
    const beforeEvents = rawTaskEvents(before);
    const originalFailure = beforeEvents.find(
      (event) =>
        event.task_definition === "build_lane" &&
        event.terminal_state === "failed"
    );
    assert(originalFailure?.attempt_id);
    assert(
      beforeEvents.some(
        (event) =>
          event.task_definition === "build_lane" &&
          event.task !== originalFailure.task &&
          event.terminal_state === "completed"
      )
    );
    const waiting = runJson(
      clusterflux,
      ["process", "status", ...scope, "--process", processId],
      { cwd: projectDir }
    );
    assert.strictEqual(waiting.live_process.main_state, "running");
    fs.writeFileSync(sourcePath, original.replace(failing, replacement));
    const restart = client.send("restartFrame", {
      threadId: recoveringThread.id,
    });
    await client.response(restart, "restartFrame");
    await client.waitFor(
      (message) => message.type === "event" && message.event === "terminated",
      10 * 60 * 1000
    );
    assert(
      client.messages.some(
        (message) =>
          message.type === "event" &&
          message.event === "thread" &&
          message.body.reason === "exited" &&
          message.body.threadId === recoveringThread.id
      )
    );
    const after = await waitForCli(
      "hosted recovery replacement event",
      () =>
        runJson(
          clusterflux,
          ["task", "list", ...scope, "--process", processId],
          { cwd: projectDir }
        ),
      (report) =>
        rawTaskEvents(report).some(
          (event) =>
            event.task === originalFailure.task &&
            event.terminal_state === "completed"
        ),
      120_000
    );
    const replacementEvent = rawTaskEvents(after).find(
      (event) =>
        event.task === originalFailure.task &&
        event.terminal_state === "completed"
    );
    assert(replacementEvent?.attempt_id);
    assert.notStrictEqual(replacementEvent.attempt_id, originalFailure.attempt_id);
    assert(
      rawTaskEvents(after).some(
        (event) =>
          event.executor === "coordinator_main" &&
          event.terminal_state === "completed"
      ),
      "replacement result did not satisfy the original coordinator-main join"
    );
    const artifacts = runJson(
      clusterflux,
      ["artifact", "list", ...scope, "--process", processId],
      { cwd: projectDir }
    ).artifacts;
    assert(artifacts.some((artifact) => artifact.artifact.startsWith("stable.txt-")));
    assert(
      artifacts.some((artifact) => artifact.artifact.startsWith("recovering.txt-"))
    );
    const released = await waitForCli(
      "hosted recovery process release",
      () =>
        runJson(
          clusterflux,
          ["process", "status", ...scope, "--process", processId],
          { cwd: projectDir }
        ),
      (status) => status.state === "not_active",
      120_000
    );
    fs.writeFileSync(sourcePath, original);
    restored = true;
    await client.close();
    closed = true;
    return {
      duration_ms: Date.now() - startedAt,
      process: processId,
      logical_task: originalFailure.task,
      original_attempt: originalFailure.attempt_id,
      replacement_attempt: replacementEvent.attempt_id,
      original_join_completed: true,
      terminal_state: released.state,
    };
  } finally {
    if (!restored) fs.writeFileSync(sourcePath, original);
    if (!closed) {
      await client.close().catch(() => {
        if (client.child.exitCode === null) client.child.kill("SIGKILL");
      });
    }
  }
}

async function main() {
  requireEnabled();
  const manifest = readJson(manifestPath);
  for (const asset of manifest.assets ?? []) {
    asset.file = path.isAbsolute(asset.file)
      ? asset.file
      : path.resolve(path.dirname(manifestPath), asset.file);
  }
  assert.strictEqual(manifest.kind, "clusterflux-public-release");
  assert.strictEqual(manifest.default_hosted_coordinator_endpoint, serviceEndpoint);
  assert.strictEqual(serviceAddr, "clusterflux.michelpaulissen.com:443");
  const qualityGates = strictQualityGateEvidence(manifest);

  const workRoot = fs.mkdtempSync(path.join(os.tmpdir(), "clusterflux-cli-happy-"));
  const installDir = path.join(workRoot, "install");
  const checkout = path.join(workRoot, "public-repo");
  const downloadPath = path.join(workRoot, "release.tar");
  const extractedArtifact = path.join(workRoot, "artifact");
  ensureDir(installDir);
  run("tar", ["-xzf", binaryArchive(manifest), "-C", installDir]);
  const publicSource = stagePublicCheckout(manifest, checkout);

  const clusterflux = executable(installDir, "clusterflux");
  const clusterfluxNode = executable(installDir, "clusterflux-node");
  const clusterfluxDap = executable(installDir, "clusterflux-debug-dap");
  const projectDir = path.join(checkout, "tests/fixtures/runtime-conformance");
  const helloProjectDir = path.join(checkout, "examples/hello-build");
  const recoveryProjectDir = path.join(checkout, "examples/recovery-build");
  const suffix = String(Date.now());
  const workerNode = `worker-${suffix}`;

  const loginStartedAt = Date.now();
  let loginSession;
  let receivedDefaultProject;
  if (reuseSessionFile) {
    const sourceSessionPath = path.resolve(reuseSessionFile);
    const sourceProjectPath = path.join(path.dirname(sourceSessionPath), "project.json");
    const sourceSession = readJson(sourceSessionPath);
    const sourceProject = readJson(sourceProjectPath);
    assert.strictEqual(sourceSession.kind, "human");
    assert.strictEqual(sourceSession.coordinator, serviceEndpoint);
    assert.strictEqual(sourceProject.coordinator, serviceEndpoint);
    assert.strictEqual(sourceProject.tenant, sourceSession.tenant);
    assert.strictEqual(sourceProject.project, sourceSession.project);
    assert.strictEqual(sourceProject.user, sourceSession.user);
    assert(
      Number(sourceSession.expires_at) > Math.floor(Date.now() / 1000) + 300,
      "reused CLI session expires too soon for the live journey"
    );
    const destinationConfig = path.join(projectDir, ".clusterflux");
    ensureDir(destinationConfig);
    fs.copyFileSync(sourceSessionPath, path.join(destinationConfig, "session.json"));
    fs.copyFileSync(sourceProjectPath, path.join(destinationConfig, "project.json"));
    fs.chmodSync(path.join(destinationConfig, "session.json"), 0o600);
    fs.chmodSync(path.join(destinationConfig, "project.json"), 0o644);
    loginSession = sourceSession;
    receivedDefaultProject = true;
  } else {
    const login = runJson(
      clusterflux,
      ["login", "--browser", "--json"],
      {
        cwd: projectDir,
        env: { ...process.env, CLUSTERFLUX_BROWSER_OPEN_COMMAND: browserOpenCommand },
      }
    );
    assert.strictEqual(login.plan.coordinator, serviceEndpoint);
    assert.strictEqual(login.boundary.scoped_cli_session_received, true);
    assert.strictEqual(login.boundary.provider_tokens_exposed_to_cli, false);
    assert.strictEqual(login.boundary.provider_tokens_sent_to_nodes, false);
    loginSession = login.coordinator_response.session;
    assert(loginSession, "hosted browser login omitted its scoped session");
    receivedDefaultProject =
      typeof loginSession.project === "string" && loginSession.project.length > 0;
    assert(receivedDefaultProject, "login did not return its server-owned project");
  }
  const sessionSecret =
    loginSession.cli_session_secret || loginSession.session_secret;
  const loginDurationMs = Math.max(1, Date.now() - loginStartedAt);
  assert(sessionSecret, "hosted browser login omitted the CLI session credential");
  const tenant = loginSession.tenant;
  const project = loginSession.project;
  const user = loginSession.user;
  for (const [name, value] of Object.entries({ tenant, project, user })) {
    assert.strictEqual(typeof value, "string", `hosted session omitted ${name}`);
    assert.notStrictEqual(value, "", `hosted session returned an empty ${name}`);
  }
  const scope = [
    "--coordinator",
    serviceEndpoint,
    "--tenant",
    tenant,
    "--project-id",
    project,
    "--user",
    user,
    "--json",
  ];

  if (!reuseSessionFile) {
    const projectInit = runJson(
      clusterflux,
      [
        "project",
        "init",
        ...scope,
        "--new-project",
        project,
        "--name",
        "CLI Happy Path",
        "--yes",
      ],
      { cwd: projectDir }
    );
    assert.strictEqual(projectInit.command, "project init");
    assert.strictEqual(projectInit.project_config_written, true);
    assert.strictEqual(projectInit.coordinator_response.type, "project_created");
  }

  const projectList = runJson(clusterflux, ["project", "list", ...scope], {
    cwd: projectDir,
  });
  assert(projectList.project_count >= 1);
  const projectSelect = runJson(
    clusterflux,
    ["project", "select", ...scope, project],
    { cwd: projectDir }
  );
  assert.strictEqual(projectSelect.command, "project select");

  const inspection = runJson(clusterflux, ["inspect", "--project", ".", "--json"], {
    cwd: projectDir,
  });
  assert(
    inspection.metadata.environments.some((environment) => environment.name === "linux")
  );
  assert(Array.isArray(inspection.source_provider_statuses));
  assert(inspection.source_provider_manifest);

  const runReport = runJson(
    clusterflux,
    ["run", "build", "--project", ".", "--json"],
    { cwd: projectDir }
  );
  assert.strictEqual(runReport.command, "run");
  assert.strictEqual(runReport.status, "main_launched");
  assert.strictEqual(runReport.task_launch.type, "main_launched");
  const processId = runReport.process;

  const parkedStatus = await waitForCli(
    "hosted coordinator main to park before node enrollment",
    () =>
      runJson(
        clusterflux,
        ["process", "status", ...scope, "--process", processId],
        { cwd: projectDir }
      ),
    (status) =>
      status.live_process?.main_state === "running" &&
      status.live_process?.main_wait_state === "waiting_for_node" &&
      status.live_process?.connected_nodes?.length === 0,
    120000
  );
  const launchAttemptOwnership = await runLaunchAttemptOwnershipGuards({
    clusterflux,
    projectDir,
    scope,
    sessionSecret,
    activeProcess: processId,
  });

  const nodeLifecycleStartedAt = Date.now();
  const grant = runJson(clusterflux, ["node", "enroll", ...scope], {
    cwd: projectDir,
  });
  assert.strictEqual(grant.command, "node enroll");
  const attach = runJson(
    clusterflux,
    [
      "node",
      "attach",
      "--coordinator",
      serviceEndpoint,
      "--tenant",
      tenant,
      "--project-id",
      project,
      "--node",
      workerNode,
      "--enrollment-grant",
      grant.enrollment_grant.grant,
      "--json",
    ],
    { cwd: projectDir }
  );
  assert.strictEqual(attach.command, "node attach");
  assert.strictEqual(attach.boundary.used_enrollment_exchange, true);

  const credentialDir = path.join(projectDir, ".clusterflux", "nodes");
  const credentialFiles = fs
    .readdirSync(credentialDir)
    .filter((file) => file.endsWith(".json"));
  assert.strictEqual(credentialFiles.length, 1, "expected one persisted node credential");
  const credentialPath = path.join(credentialDir, credentialFiles[0]);
  const credentialBefore = fs.readFileSync(credentialPath);
  const credentialDigest = sha256(credentialBefore);
  if (process.platform !== "win32") {
    assert.strictEqual(fs.statSync(credentialPath).mode & 0o777, 0o600);
  }

  const workerArgsFor = (projectRoot) => [
    "--coordinator",
    serviceEndpoint,
    "--tenant",
    tenant,
    "--project-id",
    project,
    "--node",
    workerNode,
    "--worker",
    "--project-root",
    projectRoot,
    "--assignment-poll-ms",
    "500",
    "--emit-ready",
  ];
  const workerArgs = workerArgsFor(projectDir);
  const spawnWorker = (projectRoot = projectDir) =>
    spawnJsonLines(clusterfluxNode, workerArgsFor(projectRoot), {
      cwd: projectRoot,
      env: { ...process.env },
    });

  let worker = spawnWorker();
  try {
    const firstReady = await worker.waitFor(
      (value) => value.node_status === "ready",
      "first persisted worker ready"
    );
    assert.strictEqual(firstReady.node, workerNode);
    const initiallyOnline = runJson(
      clusterflux,
      ["node", "status", ...scope, "--node", workerNode],
      { cwd: projectDir }
    );
    assert.strictEqual(nodeDescriptor(initiallyOnline, workerNode)?.online, true);

    const completedTasks = await waitForCli(
      "real hosted flagship completion",
      () =>
        runJson(clusterflux, ["task", "list", ...scope, "--process", processId], {
          cwd: projectDir,
        }),
      (tasks) => completedFlagship(rawTaskEvents(tasks))
    );
    const firstEvents = rawTaskEvents(completedTasks);
    assert(
      firstEvents
        .filter((event) => event.executor === "node")
        .every((event) => event.node === workerNode)
    );
    assert(
      firstEvents.some(
        (event) =>
          event.task_definition === "package_release" &&
          event.terminal_state === "completed" &&
          contentAddressedArtifactId(event, "release.tar")
      )
    );

    const releasedFlagshipStatus = await waitForCli(
      "completed flagship process to release its active slot",
      () =>
        runJson(
          clusterflux,
          ["process", "status", ...scope, "--process", processId],
          { cwd: projectDir }
        ),
      (status) => status.state === "not_active",
      120000
    );

    const offlineStartedAt = Date.now();
    await stopChild(worker.child);
    const observedOffline = await waitForCli(
      "server-derived worker stale/offline transition",
      () =>
        runJson(clusterflux, ["node", "status", ...scope, "--node", workerNode], {
          cwd: projectDir,
        }),
      (status) => nodeDescriptor(status, workerNode)?.online === false,
      120000
    );
    worker = spawnWorker();
    const secondReady = await worker.waitFor(
      (value) => value.node_status === "ready",
      "restarted persisted worker ready"
    );
    assert.strictEqual(secondReady.node, workerNode);
    assert.strictEqual(sha256(fs.readFileSync(credentialPath)), credentialDigest);

    const nodeStatus = await waitForCli(
      "server-derived worker online recovery",
      () =>
        runJson(clusterflux, ["node", "status", ...scope, "--node", workerNode], {
          cwd: projectDir,
        }),
      (status) => nodeDescriptor(status, workerNode)?.online === true,
      30000
    );
    assert(JSON.stringify(nodeStatus.response).includes(workerNode));
    assert.strictEqual(sha256(fs.readFileSync(credentialPath)), credentialDigest);
    const nodeLivenessTransition = {
      initial_online: nodeDescriptor(initiallyOnline, workerNode).online,
      stale_offline: nodeDescriptor(observedOffline, workerNode).online,
      recovered_online: nodeDescriptor(nodeStatus, workerNode).online,
      offline_detection_ms: Date.now() - offlineStartedAt,
      persisted_identity_reused: true,
      duration_ms: Math.max(1, Date.now() - nodeLifecycleStartedAt),
    };

    const processStatus = runJson(
      clusterflux,
      ["process", "status", ...scope, "--process", processId],
      { cwd: projectDir }
    );
    assert.strictEqual(processStatus.command, "process status");
    assert(processStatus.current_task_count >= firstEvents.length);

    const logs = runJson(
      clusterflux,
      ["logs", ...scope, "--process", processId],
      { cwd: projectDir }
    );
    assert(logs.log_entries.length >= 4);

    const artifacts = runJson(
      clusterflux,
      ["artifact", "list", ...scope, "--process", processId],
      { cwd: projectDir }
    );
    const releaseArtifact = artifacts.artifacts.find((artifact) => {
      const digest = artifact.digest;
      return (
        typeof digest === "string" &&
        /^sha256:[0-9a-f]{64}$/.test(digest) &&
        artifact.artifact === `release.tar-${digest.slice("sha256:".length)}`
      );
    });
    assert(releaseArtifact, "hosted flagship did not publish release.tar");
    assert.strictEqual(releaseArtifact.state, "metadata_flushed");

    const download = runJson(
      clusterflux,
      [
        "artifact",
        "download",
        ...scope,
        releaseArtifact.artifact,
        "--to",
        downloadPath,
      ],
      { cwd: projectDir }
    );
    assert.strictEqual(download.download_session.link_issued, true);
    assert.strictEqual(download.local_download.local_bytes_written_by_cli, true);
    assert.strictEqual(download.local_download.verified_digest, releaseArtifact.digest);
    assert.strictEqual(sha256(fs.readFileSync(downloadPath)), releaseArtifact.digest);

    ensureDir(extractedArtifact);
    const archiveEntries = run("tar", ["-tf", downloadPath]);
    assert.match(archiveEntries, /(?:^|\n)hello-clusterflux(?:\n|$)/);
    run("tar", ["-xf", downloadPath, "-C", extractedArtifact]);
    const builtExecutable = path.join(extractedArtifact, "hello-clusterflux");
    assert(fs.existsSync(builtExecutable));
    const builtOutput = run(builtExecutable, []).trim();
    assert.strictEqual(builtOutput, "hello from a real Clusterflux build");

    copyProjectControlState(projectDir, helloProjectDir);
    copyProjectControlState(projectDir, recoveryProjectDir);
    await stopChild(worker.child);
    worker = spawnWorker(helloProjectDir);
    await worker.waitFor(
      (value) => value.node_status === "ready",
      "hello-build worker ready"
    );
    const helloBuild = await runHostedHelloBuild({
      clusterflux,
      projectDir: helloProjectDir,
      scope,
      outputFile: path.join(workRoot, "hello-clusterflux"),
    });
    await stopChild(worker.child);
    worker = spawnWorker(recoveryProjectDir);
    await worker.waitFor(
      (value) => value.node_status === "ready",
      "recovery-build worker ready"
    );
    const recoveryBuild = await runHostedRecoveryBuild({
      clusterfluxDap,
      clusterflux,
      projectDir: recoveryProjectDir,
      scope,
    });
    await stopChild(worker.child);
    worker = spawnWorker();
    await worker.waitFor(
      (value) => value.node_status === "ready",
      "runtime-conformance worker restored"
    );

    const dapEditRestart = await runLiveDapEditRestart({
      clusterfluxDap,
      clusterflux,
      projectDir,
      scope,
      worker,
    });

    const processCancellationStartedAt = Date.now();
    const restart = runJson(
      clusterflux,
      ["process", "restart", ...scope, "--process", dapEditRestart.process, "--yes"],
      { cwd: projectDir }
    );
    assert.strictEqual(restart.restart_request.accepted, true);
    const cancel = runJson(
      clusterflux,
      ["process", "cancel", ...scope, "--process", dapEditRestart.process, "--yes"],
      { cwd: projectDir }
    );
    assert.strictEqual(cancel.cancel_request.accepted, true);
    const releasedDap = await waitForCli(
      "cancelled DAP process slot release",
      () =>
        runJson(
          clusterflux,
          ["process", "status", ...scope, "--process", dapEditRestart.process],
          { cwd: projectDir }
        ),
      (status) => status.state === "not_active",
      120000
    );
    assert.strictEqual(releasedDap.state, "not_active");
    const processCancellationLifecycle = {
      request: {
        restart_process: dapEditRestart.process,
        cancel_process: dapEditRestart.process,
      },
      expected:
        "explicit process cancellation terminates active participants and releases the slot",
      observed: {
        restart_accepted: restart.restart_request.accepted,
        cancel_accepted: cancel.cancel_request.accepted,
        terminal_state: releasedDap.state,
      },
      duration_ms: Math.max(1, Date.now() - processCancellationStartedAt),
    };

    const identityWorkerNode = `identity-worker-${suffix}`;
    const identityWorkerGrant = runJson(
      clusterflux,
      ["node", "enroll", ...scope],
      { cwd: projectDir }
    );
    runJson(
      clusterflux,
      [
        "node",
        "attach",
        "--coordinator",
        serviceEndpoint,
        "--tenant",
        tenant,
        "--project-id",
        project,
        "--node",
        identityWorkerNode,
        "--enrollment-grant",
        identityWorkerGrant.enrollment_grant.grant,
        "--json",
      ],
      { cwd: projectDir }
    );
    const identityWorker = spawnJsonLines(
      clusterfluxNode,
      [
        "--coordinator",
        serviceEndpoint,
        "--tenant",
        tenant,
        "--project-id",
        project,
        "--node",
        identityWorkerNode,
        "--worker",
        "--project-root",
        projectDir,
        "--assignment-poll-ms",
        "500",
        "--emit-ready",
      ],
      { cwd: projectDir, env: { ...process.env } }
    );
    let sameDefinitionIdentity;
    try {
      const identityWorkerReady = await identityWorker.waitFor(
        (value) => value.node_status === "ready",
        "same-definition identity worker ready"
      );
      assert.strictEqual(identityWorkerReady.node, identityWorkerNode);
      await waitForCli(
        "same-definition identity worker online",
        () =>
          runJson(
            clusterflux,
            ["node", "status", ...scope, "--node", identityWorkerNode],
            { cwd: projectDir }
          ),
        (status) => nodeDescriptor(status, identityWorkerNode)?.online === true,
        30_000
      );
      await stopChild(worker.child);
      worker = spawnWorker();
      const refreshedPrimaryReady = await worker.waitFor(
        (value) => value.node_status === "ready",
        "same-definition primary worker capability refresh"
      );
      assert.strictEqual(refreshedPrimaryReady.node, workerNode);
      const primaryStatus = await waitForCli(
        "same-definition primary worker online after refresh",
        () =>
          runJson(
            clusterflux,
            ["node", "status", ...scope, "--node", workerNode],
            { cwd: projectDir }
          ),
        (status) => nodeDescriptor(status, workerNode)?.online === true,
        30_000
      );
      const identityStatus = runJson(
        clusterflux,
        ["node", "status", ...scope, "--node", identityWorkerNode],
        { cwd: projectDir }
      );
      const primarySnapshots =
        nodeDescriptor(primaryStatus, workerNode)?.source_snapshots || [];
      const identitySnapshots =
        nodeDescriptor(identityStatus, identityWorkerNode)?.source_snapshots || [];
      assert(
        primarySnapshots.some((snapshot) => identitySnapshots.includes(snapshot)),
        `same-definition workers do not share a current source snapshot: ${JSON.stringify({
          primarySnapshots,
          identitySnapshots,
        })}`
      );
      sameDefinitionIdentity = await runSameDefinitionDapIdentity({
        clusterfluxDap,
        clusterflux,
        projectDir,
        scope,
      });
    } finally {
      await stopChild(identityWorker.child);
      runJson(
        clusterflux,
        ["node", "revoke", ...scope, "--node", identityWorkerNode, "--yes"],
        { cwd: projectDir }
      );
    }
    const repeatedParkWake = await runRepeatedParkWakeProof({
      clusterflux,
      projectDir,
      scope,
    });
    const longJoin = await runLongJoinProof({
      clusterflux,
      projectDir,
      scope,
    });

    await stopChild(worker.child);
    const tenantIsolationResult = await runSecondTenantIsolation({
      clusterflux,
      clusterfluxNode,
      firstScope: scope,
      firstHelloBuild: helloBuild,
      firstHelloProjectDir: helloProjectDir,
      workRoot,
      firstSessionSecret: sessionSecret,
      firstTenant: tenant,
      firstProject: project,
      firstProcess: dapEditRestart.process,
      firstNode: workerNode,
      firstArtifact: releaseArtifact.artifact,
      manifest,
    });
    const tenantIsolation = tenantIsolationResult.evidence;
    const scopedCollisionRuntime = tenantIsolationResult.runtime;
    worker = spawnWorker();
    const collisionRecoveryReady = await worker.waitFor(
      (value) => value.node_status === "ready",
      "primary worker restored after two-tenant collision proof"
    );
    assert.strictEqual(collisionRecoveryReady.node, workerNode);

    const securityNode = await prepareLiveNodeCredentialSecurity({
      clusterflux,
      projectDir,
      scope,
      tenant,
      project,
      suffix,
      bundle: {
        digest: runReport.bundle_digest,
        moduleBase64: fs
          .readFileSync(
            path.resolve(projectDir, runReport.bundle_build.bundle_artifact.module)
          )
          .toString("base64"),
        entryExport: runReport.entry_export,
        entryStableId: runReport.entry_stable_id,
      },
    });
    const droppedConnectionRollback = await runDroppedConnectionRollback({
      clusterflux,
      projectDir,
      scope,
    });
    const bandwidthPreallocation = await runLiveBandwidthPreallocation({
      sessionSecret,
      artifact: releaseArtifact.artifact,
      maxBytes: download.local_download.bytes_written,
    });
    await stopChild(worker.child);
    const relayEmergencyDisableResult = await runRelayEmergencyDisable({
      clusterflux,
      clusterfluxNode,
      projectDir,
      scope,
      workerArgs,
      sessionSecret,
      maxBytes: download.local_download.bytes_written,
    });
    worker = relayEmergencyDisableResult.worker;
    const relayEmergencyDisable = relayEmergencyDisableResult.evidence;
    const agentCredentialSecurity = await runLiveAgentCredentialSecurity({
      clusterfluxDap,
      clusterflux,
      projectDir,
      scope,
      sessionSecret,
      tenant,
      project,
      suffix,
      bundle: {
        digest: runReport.bundle_digest,
        moduleBase64: fs
          .readFileSync(
            path.resolve(projectDir, runReport.bundle_build.bundle_artifact.module)
          )
          .toString("base64"),
        entryExport: runReport.entry_export,
        entryStableId: runReport.entry_stable_id,
      },
      securityNode,
      workerRuntime: {
        node: workerNode,
        child: worker.child,
        worker,
        identity: nodeIdentityFromPrivateKey(
          readNodeCredential(projectDir, workerNode).credential.private_key
        ),
      },
    });
    const mainBeforeChildLifecycle =
      agentCredentialSecurity.main_before_child_lifecycle;

    await stopChild(worker.child);
    const liveSoak = await runLiveSoak({
      clusterflux,
      projectDir,
      scope,
      securityNode,
    });
    const malformedIdentifiers = await runMalformedIdentifierSuite({
      clusterflux,
      projectDir,
      scope,
      sessionSecret,
      tenant,
      project,
      user,
      suffix,
      securityNode,
      bundle: {
        digest: runReport.bundle_digest,
        moduleBase64: fs
          .readFileSync(
            path.resolve(projectDir, runReport.bundle_build.bundle_artifact.module)
          )
          .toString("base64"),
        entryExport: runReport.entry_export,
        entryStableId: runReport.entry_stable_id,
      },
      releaseArtifact,
    });
    const signedHostileArtifactPath = await runLiveSignedHostileArtifactPath({
      clusterflux,
      projectDir,
      scope,
      tenant,
      project,
      suffix,
      securityNode,
    });
    const revokedNodeCredential = await revokeSecurityNode({
      clusterflux,
      projectDir,
      scope,
      securityNode,
    });
    const quotaPreallocation = await runLiveQuotaPreallocation({
      sessionSecret,
      suffix,
    });
    const hostedLoginIsolation = await runHostedLoginIsolationBeforeRestart();
    const serviceRestart = await restartHostedServiceAndResume({
      clusterflux,
      clusterfluxNode,
      projectDir,
      scope,
      workerArgs,
      workerNode,
      credentialPath,
      credentialDigest,
      scopedCollisionRuntime,
    });
    if (serviceRestart.worker) worker = serviceRestart.worker;
    const scopedNodeRevocationAfterRestart = scopedCollisionRuntime
      ? await finishScopedNodeCollisionAfterRestart({
          clusterflux,
          projectDir,
          scope,
          workerNode,
          scopedCollisionRuntime,
        })
      : {
          executed: false,
          reason: "second tenant runtime session was not supplied",
          duration_ms: 1,
        };
    const relayAfterRestart = relayDurableState();
    assert(
      relayAfterRestart.ingress_used >=
        bandwidthPreallocation.durable_after_release.ingress_used
    );
    assert(
      relayAfterRestart.egress_used >=
        bandwidthPreallocation.durable_after_release.egress_used
    );
    assert(
      relayAfterRestart.abandoned_or_failed_used >=
        bandwidthPreallocation.durable_after_release.abandoned_or_failed_used
    );
    bandwidthPreallocation.durable_after_restart = relayAfterRestart;
    bandwidthPreallocation.restart_persistence = true;
    bandwidthPreallocation.duration_ms =
      Date.now() - bandwidthPreallocation.scenario_started_at_ms;
    delete bandwidthPreallocation.scenario_started_at_ms;
    await finishHostedLoginIsolationAfterRestart(hostedLoginIsolation);
    const userCredentialSecurity = await finishUserCredentialSecurity({
      clusterflux,
      projectDir,
      scope,
      sessionSecret,
      });
      const configProvenance = configurationProvenance();
      const proxyConfigProvenance = proxyConfigurationProvenance();
      const concurrentCompileTasks = [
        ...new Set(
          firstEvents
            .filter(
              (event) =>
                event.task_definition === "compile_linux" &&
                event.terminal_state === "completed"
            )
            .map((event) => event.task)
        ),
      ];
      assert(concurrentCompileTasks.length >= 2);
      const extensionAsset = manifest.assets.find((asset) =>
        asset.name.endsWith(".vsix")
      );
      const requirement = ({ id, passed, evidence, durationMs }) => {
        const result = {
          id,
          passed: passed === true,
          evidence,
        };
        if (Number.isFinite(durationMs) && durationMs > 0) {
          result.duration_ms = durationMs;
        } else if (strictFullRelease) {
          throw new Error(
            `strict release scenario ${id} omitted a measured positive duration`
          );
        }
        return result;
      };
      const qualityCheck = (id) => qualityGates?.checks?.[id];
      const nodeEnrollmentEvidence = {
        request: "one-time enrollment followed by a grant-free worker restart",
        expected: "the persisted node credential is reused unchanged",
        observed: {
          used_enrollment_exchange: attach.boundary.used_enrollment_exchange,
          restarted_node: secondReady.node,
          credential_digest: credentialDigest,
          liveness: nodeLivenessTransition,
          service_restart_executed: serviceRestart.executed,
        },
      };
      const loginEvidence = {
        request: reuseSessionFile
          ? "reuse a still-valid scoped browser-login session"
          : "perform browser OIDC login",
        expected: "server-owned default project persists and can be selected",
        observed: {
          fresh_login: !reuseSessionFile,
          tenant,
          project,
          user,
          received_default_project: receivedDefaultProject,
          project_select_command: projectSelect.command,
        },
      };
      const credentialSecurityEvidence = {
        expected:
          "forged, replayed, expired, body-modified, and revoked user/Node/Agent credentials are denied",
        observed: {
          user: userCredentialSecurity,
          node: {
            initial: securityNode.evidence,
            revoked: revokedNodeCredential.denied,
          },
          agent: {
            replay: agentCredentialSecurity.replay,
            expired: agentCredentialSecurity.expired,
            forged: agentCredentialSecurity.forged,
            body_modified: agentCredentialSecurity.body_modified,
            revoked: agentCredentialSecurity.revoked,
          },
        },
      };
      const credentialSecurityDurationMs =
        userCredentialSecurity.duration_ms +
        securityNode.duration_ms +
        revokedNodeCredential.duration_ms +
        agentCredentialSecurity.duration_ms;
      const relaySecurityEvidence = {
        expected:
          "limits are scoped independently and abandoned transfer bytes remain charged without a shared soft lockout",
        observed: {
          spawn_quota: quotaPreallocation,
          relay_accounting: bandwidthPreallocation,
          emergency_disable: relayEmergencyDisable,
        },
      };
      const relaySecurityDurationMs =
        quotaPreallocation.duration_ms +
        bandwidthPreallocation.duration_ms +
        relayEmergencyDisable.duration_ms;
      const strictRequirementLedger = [
        requirement({
          id: "01_formatting",
          passed: qualityCheck("formatting")?.status === "passed",
          evidence: qualityCheck("formatting"),
          durationMs: qualityCheck("formatting")?.duration_ms,
        }),
        requirement({
          id: "02_clippy_warnings_denied",
          passed: qualityCheck("clippy_warnings_denied")?.status === "passed",
          evidence: qualityCheck("clippy_warnings_denied"),
          durationMs: qualityCheck("clippy_warnings_denied")?.duration_ms,
        }),
        requirement({
          id: "03_public_workspace_tests",
          passed: qualityCheck("public_workspace_tests")?.status === "passed",
          evidence: qualityCheck("public_workspace_tests"),
          durationMs: qualityCheck("public_workspace_tests")?.duration_ms,
        }),
        requirement({
          id: "04_private_hosted_policy_locked_tests",
          passed:
            qualityCheck("private_hosted_policy_locked_tests")?.status ===
            "passed",
          evidence: qualityCheck("private_hosted_policy_locked_tests"),
          durationMs: qualityCheck("private_hosted_policy_locked_tests")
            ?.duration_ms,
        }),
        requirement({
          id: "05_wasm_example_builds",
          passed: qualityCheck("wasm_example_builds")?.status === "passed",
          evidence: qualityCheck("wasm_example_builds"),
          durationMs: qualityCheck("wasm_example_builds")?.duration_ms,
        }),
        requirement({
          id: "06_filtered_public_tree_build_and_tests",
          passed:
            qualityCheck("filtered_public_tree_build_and_tests")?.status ===
            "passed",
          evidence: qualityCheck("filtered_public_tree_build_and_tests"),
          durationMs: qualityCheck("filtered_public_tree_build_and_tests")
            ?.duration_ms,
        }),
        requirement({
          id: "07_final_vsix_checks",
          passed: Boolean(
            extensionAsset &&
              qualityCheck("vscode_extension_candidate")?.status === "passed" &&
              extensionAsset.sha256 ===
                manifest.release_candidate.extension_sha256 &&
              qualityCheck("vscode_extension_candidate")?.candidate_vsix
                ?.sha256 === extensionAsset.sha256
          ),
          evidence: qualityCheck("vscode_extension_candidate"),
          durationMs: qualityCheck("vscode_extension_candidate")?.duration_ms,
        }),
        requirement({
          id: "08_browser_login_and_persistent_default_project",
          passed: Boolean(
            sessionSecret &&
              tenant &&
              project &&
              receivedDefaultProject &&
              projectSelect.command === "project select"
          ),
          evidence: loginEvidence,
          durationMs: loginDurationMs,
        }),
        requirement({
          id: "09_one_time_node_enrollment_and_restart",
          passed:
            attach.boundary.used_enrollment_exchange === true &&
            secondReady.node === workerNode &&
            nodeLivenessTransition.persisted_identity_reused === true &&
            serviceRestart.executed === true,
          evidence: nodeEnrollmentEvidence,
          durationMs: nodeLivenessTransition.duration_ms,
        }),
        requirement({
          id: "10_real_hello_build_and_artifact_download",
          passed:
            helloBuild.executable_output ===
              "hello from a real Clusterflux build" &&
            helloBuild.downloaded_bytes > 0 &&
            helloBuild.downloaded_sha256 === helloBuild.digest,
          evidence: helloBuild,
          durationMs: helloBuild.duration_ms,
        }),
        requirement({
          id: "11_recovery_retry_and_original_join",
          passed:
            recoveryBuild.original_join_completed === true &&
            recoveryBuild.original_attempt !==
              recoveryBuild.replacement_attempt &&
            recoveryBuild.terminal_state === "not_active",
          evidence: recoveryBuild,
          durationMs: recoveryBuild.duration_ms,
        }),
        requirement({
          id: "12_long_lived_coordinator_main",
          passed:
            longJoin.duration_seconds > 120 &&
            longJoin.terminal_state === "completed" &&
            longJoin.process_state === "not_active",
          evidence: longJoin,
          durationMs: longJoin.duration_ms,
        }),
        requirement({
          id: "13_main_completes_before_child_lifecycle",
          passed:
            mainBeforeChildLifecycle.observed.active_after_main === true &&
            mainBeforeChildLifecycle.observed.debug_state_after_main ===
              "debug_breakpoints" &&
            mainBeforeChildLifecycle.observed.child_terminal_state ===
              "completed" &&
            mainBeforeChildLifecycle.observed.final_process_state ===
              "not_active" &&
            mainBeforeChildLifecycle.observed.subsequent_run_status ===
              "main_launched" &&
            qualityCheck("process_lifecycle_regressions")?.status === "passed",
          evidence: {
            live: mainBeforeChildLifecycle,
            compiled_failure_and_cancellation:
              qualityCheck("process_lifecycle_regressions"),
          },
          durationMs:
            mainBeforeChildLifecycle.duration_ms +
            (qualityCheck("process_lifecycle_regressions")?.duration_ms || 0),
        }),
        requirement({
          id: "14_dap_launch_without_breakpoint",
          passed:
            sameDefinitionIdentity.launch_without_breakpoint.observed.thread_id >
              0 &&
            sameDefinitionIdentity.terminal_state === "not_active",
          evidence: sameDefinitionIdentity.launch_without_breakpoint,
          durationMs:
            sameDefinitionIdentity.launch_without_breakpoint.duration_ms,
        }),
        requirement({
          id: "15_real_breakpoint_installation_and_hit",
          passed:
            dapEditRestart.real_breakpoint.observed.stop_reason ===
              "breakpoint" &&
            dapEditRestart.real_breakpoint.observed.stopped_line ===
              dapEditRestart.real_breakpoint.request.line &&
            dapEditRestart.real_breakpoint.observed.all_threads_stopped === true,
          evidence: dapEditRestart.real_breakpoint,
          durationMs: dapEditRestart.real_breakpoint.duration_ms,
        }),
        requirement({
          id: "16_attach_with_preconfigured_breakpoint",
          passed:
            agentCredentialSecurity.attach_with_preconfigured_breakpoint
              .observed.initially_verified === false &&
            agentCredentialSecurity.attach_with_preconfigured_breakpoint
              .observed.installation_event_verified === true &&
            agentCredentialSecurity.attach_with_preconfigured_breakpoint
              .observed.stop_reason === "breakpoint" &&
            agentCredentialSecurity.attach_with_preconfigured_breakpoint
              .observed.stopped_line ===
              agentCredentialSecurity.attach_with_preconfigured_breakpoint
                .request.line,
          evidence:
            agentCredentialSecurity.attach_with_preconfigured_breakpoint,
          durationMs:
            agentCredentialSecurity.attach_with_preconfigured_breakpoint
              .duration_ms,
        }),
        requirement({
          id: "17_breakpoint_revision_ordering",
          passed:
            dapEditRestart.breakpoint_revisions.authoritative_revision >
              dapEditRestart.breakpoint_revisions.delayed_revision &&
            dapEditRestart.breakpoint_revisions.stale_completion_events === 0,
          evidence: dapEditRestart.breakpoint_revisions,
          durationMs: dapEditRestart.breakpoint_revisions.duration_ms,
        }),
        requirement({
          id: "18_continue_pause_inspect_disconnect",
          passed:
            dapEditRestart.debug_interaction.observed.pause_response_ms <
              2_000 &&
            dapEditRestart.debug_interaction.observed.inspect_response_ms <
              2_000 &&
            dapEditRestart.debug_interaction.observed.continue_response_ms <
              2_000 &&
            dapEditRestart.debug_interaction.observed.disconnect_response_ms <
              2_000,
          evidence: dapEditRestart.debug_interaction,
          durationMs: dapEditRestart.debug_interaction.duration_ms,
        }),
        requirement({
          id: "19_distinct_same_definition_instances",
          passed:
            sameDefinitionIdentity.thread_evidence.length === 2 &&
            sameDefinitionIdentity.completion_order.length === 2 &&
            sameDefinitionIdentity.thread_evidence[0].task_instance !==
              sameDefinitionIdentity.thread_evidence[1].task_instance &&
            sameDefinitionIdentity.thread_evidence[0].attempt_id !==
              sameDefinitionIdentity.thread_evidence[1].attempt_id &&
            concurrentCompileTasks.length >= 2,
          evidence: sameDefinitionIdentity.same_definition_identity,
          durationMs:
            sameDefinitionIdentity.same_definition_identity.duration_ms,
        }),
        requirement({
          id: "20_real_podman_partial_debug_epoch",
          passed:
            agentCredentialSecurity.partial_freeze.partially_frozen === true &&
            agentCredentialSecurity.partial_freeze.dap_all_threads_stopped ===
              false &&
            agentCredentialSecurity.partial_freeze.podman_paused_container_ids
              .length > 0 &&
            agentCredentialSecurity.partial_freeze.resumed_participants.length >
              0,
          evidence: agentCredentialSecurity.partial_freeze,
          durationMs: agentCredentialSecurity.partial_freeze.elapsed_ms,
        }),
        requirement({
          id: "21_post_main_launched_observation_recovery",
          passed:
            sameDefinitionIdentity.post_main_launched_observation_recovery
              .observed.diagnostics.length === 2 &&
            sameDefinitionIdentity.post_main_launched_observation_recovery
              .observed
              .premature_stop_or_termination_events === 0 &&
            sameDefinitionIdentity.post_main_launched_observation_recovery
              .observed.main_thread.id > 0,
          evidence:
            sameDefinitionIdentity.post_main_launched_observation_recovery,
          durationMs:
            sameDefinitionIdentity.post_main_launched_observation_recovery
              .duration_ms,
        }),
        requirement({
          id: "22_observer_reconnection",
          passed:
            dapEditRestart.observer_reconnection.fault_injections.length === 5 &&
            dapEditRestart.observer_reconnection.observed
              .false_stops_before_real_stop === 0 &&
            dapEditRestart.observer_reconnection.observed.real_stop_reason ===
              "breakpoint",
          evidence: dapEditRestart.observer_reconnection,
          durationMs: dapEditRestart.observer_reconnection.duration_ms,
        }),
        requirement({
          id: "23_malformed_identifiers_preserve_service",
          passed:
            malformedIdentifiers.valid_after_every_rejection === true &&
            malformedIdentifiers.all_rejected_for_intended_reason === true &&
            malformedIdentifiers.valid_authenticated_action_after_suite ===
              true &&
            malformedIdentifiers.valid_signed_rendezvous_after_suite === true &&
            malformedIdentifiers.rejected_requests.every(
              (entry) =>
                Number.isFinite(entry.duration_ms) && entry.duration_ms > 0
            ),
          evidence: malformedIdentifiers,
          durationMs: malformedIdentifiers.duration_ms,
        }),
        requirement({
          id: "24_cross_tenant_access_denied",
          passed:
            tenantIsolation.executed === true &&
            tenantIsolation.process_hidden === true &&
            tenantIsolation.node_hidden === true &&
            tenantIsolation.tasks_and_logs.denied === true &&
            tenantIsolation.debug.denied === true &&
            tenantIsolation.artifact_and_download.denied === true,
          evidence: tenantIsolation,
          durationMs: tenantIsolation.duration_ms,
        }),
        requirement({
          id: "25_forged_replayed_expired_revoked_credentials_denied",
          passed:
            Boolean(
              userCredentialSecurity.expired &&
                userCredentialSecurity.forged &&
                userCredentialSecurity.revoked
            ) &&
            securityNode.evidence.replay.denied === true &&
            securityNode.evidence.expired.denied === true &&
            securityNode.evidence.forged.denied === true &&
            revokedNodeCredential.denied.denied === true &&
            agentCredentialSecurity.revoked.denied === true,
          evidence: credentialSecurityEvidence,
          durationMs: credentialSecurityDurationMs,
        }),
        requirement({
          id: "26_relay_limits_and_abandoned_transfer_charging",
          passed:
            quotaPreallocation.denied_process_allocated === false &&
            quotaPreallocation.independent_scope
              .unaffected_by_spawn_quota === true &&
            bandwidthPreallocation.bytes_served_before_denial > 0 &&
            bandwidthPreallocation.partial_abandon.ingress_delta > 0 &&
            bandwidthPreallocation.partial_abandon.egress_delta > 0 &&
            bandwidthPreallocation.partial_abandon.abandoned_delta > 0 &&
            bandwidthPreallocation.partial_abandon.reservation_released ===
              true &&
            bandwidthPreallocation.restart_persistence === true &&
            relayEmergencyDisable.disabled.denied === true &&
            relayEmergencyDisable.runtime_drop_in_removed === true,
          evidence: relaySecurityEvidence,
          durationMs: relaySecurityDurationMs,
        }),
        requirement({
          id: "27_existing_process_rejection_preserves_process",
          passed: launchAttemptOwnership.existing_process_survived === true,
          evidence: launchAttemptOwnership,
          durationMs: launchAttemptOwnership.duration_ms,
        }),
        requirement({
          id: "28_precommit_launch_failure_scoped_cleanup",
          passed:
            droppedConnectionRollback.cli_exit_status !== 0 &&
            droppedConnectionRollback.rollback ===
              "automatic CLI launch guard abort_process with matching launch_attempt" &&
            droppedConnectionRollback.terminal_state === "not_active",
          evidence: droppedConnectionRollback,
          durationMs: droppedConnectionRollback.duration_ms,
        }),
        requirement({
          id: "29_explicit_process_cancellation_cleanup",
          passed:
            processCancellationLifecycle.observed.restart_accepted === true &&
            processCancellationLifecycle.observed.cancel_accepted === true &&
            processCancellationLifecycle.observed.terminal_state ===
              "not_active",
          evidence: processCancellationLifecycle,
          durationMs: processCancellationLifecycle.duration_ms,
        }),
        requirement({
          id: "30_scoped_node_and_artifact_collision",
          passed:
            tenantIsolation.scoped_node_and_artifact_collision?.observed
              ?.both_metadata_records_visible_to_owner === true &&
            tenantIsolation.scoped_node_and_artifact_collision?.observed
              ?.first_download_survived_second_link_revocation === true &&
            tenantIsolation.scoped_node_and_artifact_collision?.observed
              ?.same_artifact_id === helloBuild.artifact &&
            serviceRestart.scoped_node_collision_after_restart
              ?.first_scope_online === true &&
            serviceRestart.scoped_node_collision_after_restart
              ?.second_scope_online === true &&
            serviceRestart.scoped_node_collision_after_restart
              ?.both_credentials_reused_without_grants === true &&
            scopedNodeRevocationAfterRestart.observed
              ?.revoked_credential_denied === true &&
            scopedNodeRevocationAfterRestart.observed
              ?.first_scope_node_online === true,
          evidence: {
            live_collision:
              tenantIsolation.scoped_node_and_artifact_collision,
            restart:
              serviceRestart.scoped_node_collision_after_restart,
            scoped_revocation: scopedNodeRevocationAfterRestart,
          },
          durationMs:
            tenantIsolation.scoped_node_and_artifact_collision?.duration_ms +
            scopedNodeRevocationAfterRestart.duration_ms,
        }),
        requirement({
          id: "31_signed_hostile_artifact_path_preserves_service",
          passed:
            signedHostileArtifactPath.observed.rejection.denied === true &&
            signedHostileArtifactPath.observed.valid_metadata_response ===
              "vfs_metadata_recorded" &&
            signedHostileArtifactPath.observed.health_response ===
              "node_heartbeat" &&
            signedHostileArtifactPath.observed.process_cleanup ===
              "not_active",
          evidence: signedHostileArtifactPath,
          durationMs: signedHostileArtifactPath.duration_ms,
        }),
      ];
      const fullReleasePassed =
        strictFullRelease &&
        proxyConfigProvenance?.active_state === "active" &&
        strictRequirementLedger.every((requirement) => requirement.passed === true);
      const namedScenarios = strictRequirementLedger.map((requirement) => ({
        id: requirement.id,
        name: requirement.id
          .replace(/^\d+_/, "")
          .replaceAll("_", " "),
        status: requirement.passed ? "passed" : "failed",
        ...(requirement.duration_ms
          ? { duration_ms: requirement.duration_ms }
          : {}),
      }));

    const report = {
      kind: "clusterflux-cli-happy-path-live",
      release_name: manifest.release_name,
      source_commit: manifest.source_commit,
      source_tree_digest: manifest.source_tree_digest,
      public_tree_identity: manifest.public_tree_identity,
      binary_digests: manifest.binary_digests,
      service_addr: serviceAddr,
      default_hosted_coordinator_endpoint: serviceEndpoint,
      public_repository_url: publicSource.published ? publicSource.url : null,
      public_source: publicSource,
      tenant,
      project,
      worker_node: workerNode,
      process: processId,
      signed_in: true,
      fresh_hosted_login: !reuseSessionFile,
      reused_scoped_cli_session: Boolean(reuseSessionFile),
      received_default_project: receivedDefaultProject,
      process_started_before_node: true,
      parked_main_observed: {
        main_state: parkedStatus.live_process.main_state,
        main_wait_state: parkedStatus.live_process.main_wait_state,
        connected_nodes: parkedStatus.live_process.connected_nodes,
      },
      node_enrollment_grants_used: 2,
      persisted_node_credential_mode:
        process.platform === "win32" ? null : fs.statSync(credentialPath).mode & 0o777,
      persisted_node_credential_digest: credentialDigest,
        node_restarted_without_grant: true,
        restarted_node_ready: secondReady.node,
        node_liveness_transition: nodeLivenessTransition,
      flagship_terminal_state: releasedFlagshipStatus.state,
      flagship_task_definitions: [
        "prepare_source",
        "compile_linux",
        "package_release",
      ],
      rootless_podman_flagship_completed: true,
      downstream_release_artifact: releaseArtifact,
      artifact_download: {
        bytes_written: download.local_download.bytes_written,
        verified_digest: download.local_download.verified_digest,
        executable_output: builtOutput,
      },
        flagship_process_released_automatically: true,
        concurrent_same_definition_tasks: sameDefinitionIdentity,
        flagship_same_definition_task_instances: concurrentCompileTasks,
        repeated_park_wake: repeatedParkWake,
        long_join: longJoin,
        main_before_child_lifecycle: mainBeforeChildLifecycle,
      live_dap_edit_restart: dapEditRestart,
      process_restart_requested: true,
      process_cancel_requested: true,
      dap_process_aborted: true,
      process_cancellation_lifecycle: processCancellationLifecycle,
      tenant_isolation: tenantIsolation,
      scoped_node_revocation_after_restart:
        scopedNodeRevocationAfterRestart,
      credential_security: {
        user: userCredentialSecurity,
        node: {
          node: securityNode.node,
          credential_digest: securityNode.credential_digest,
          ...securityNode.evidence,
          revoked: revokedNodeCredential.denied,
        },
        agent: agentCredentialSecurity,
      },
        quota_preallocation: quotaPreallocation,
        bandwidth_preallocation: bandwidthPreallocation,
        relay_emergency_disable: relayEmergencyDisable,
        hosted_login_isolation: hostedLoginIsolation,
        dropped_connection_rollback: droppedConnectionRollback,
        launch_attempt_ownership: launchAttemptOwnership,
      live_soak: liveSoak,
      hosted_service_restart: {
        ...serviceRestart,
        worker: undefined,
      },
      release_binding: {
        manifest: manifestPath,
        source_commit: manifest.source_commit,
        source_tree_digest: manifest.source_tree_digest,
        source_tree_clean: manifest.source_tree_clean,
        public_tree_identity: manifest.public_tree_identity,
        binary_digests: manifest.binary_digests,
        release_candidate: manifest.release_candidate,
        deployment: serviceRestart.after || deploymentProvenance(),
        configuration: configProvenance,
        proxy_configuration: proxyConfigProvenance,
      },
        commands,
        quality_gates: qualityGates,
        hello_build: helloBuild,
        recovery_build: recoveryBuild,
        malformed_identifiers: malformedIdentifiers,
        signed_hostile_artifact_path: signedHostileArtifactPath,
        named_scenarios: namedScenarios,
        scenario_skips: [],
        strict_requirement_ledger: strictRequirementLedger,
        acceptance_result: fullReleasePassed ? "passed" : "partial",
    };
    ensureDir(path.dirname(reportPath));
    fs.writeFileSync(reportPath, `${JSON.stringify(report, null, 2)}\n`);
    if (strictFullRelease && !fullReleasePassed) {
      const failed = strictRequirementLedger
        .filter((requirement) => requirement.passed !== true)
        .map((requirement) => requirement.id);
      throw new Error(`strict full release failed: ${failed.join(", ")}`);
    }
    console.log(`CLI happy-path live smoke passed: ${reportPath}`);
  } finally {
    await stopChild(worker?.child);
  }
}

main().catch((error) => {
  console.error(error.stack || error.message);
  process.exit(1);
});