[SYMPP-2DFF] Certify rotating ownership with Windows Job clients

plugins/symphony-plus-plus-mcp/tests/launcher/cold-start-singleton-smoke.js

@@ -214,6 +214,157 @@ function startClient(barrier, launcher, environment, clients, latencies, readyTa
   return client;
 }
 
+function startJobClient(root, barrier, launcher, environment, clients, latencies, readyTarget, index) {
+  const wrapper = path.join(root, `job-client-${index}.cmd`);
+  const state = path.join(root, `job-client-${index}.state`);
+  fs.writeFileSync(wrapper, `@echo off\r\n:wait\r\nif not exist "${barrier}" goto wait\r\ncall "${launcher}"\r\nexit /b %ERRORLEVEL%\r\n`);
+  const child = spawn("pwsh.exe", ["-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-File", process.env.SYMPP_JOB_CLIENT, state, wrapper], { env: environment, stdio: ["pipe", "pipe", "pipe"], windowsHide: true });
+  const client = { child, state, stderr: "", stdout: "", ready: false, jobReady: false, rootPid: 0, result: null, pending: new Map() };
+  clients.push(client);
+  child.stderr.on("data", (chunk) => {
+    client.stderr += chunk;
+    const match = client.stderr.match(/JOB_READY:(\d+)/);
+    if (match) { client.jobReady = true; client.rootPid = Number(match[1]); }
+  });
+  child.stdin.on("error", (error) => { client.stderr += `${error.code || error.message}\n`; });
+  child.stdout.on("data", (chunk) => {
+    client.stdout += chunk;
+    const lines = client.stdout.split(/\r?\n/); client.stdout = lines.pop();
+    for (const line of lines) {
+      if (!line) continue;
+      const response = JSON.parse(line);
+      if (response.id === 1) {
+        latencies.push(Date.now() - readyTarget.startedAt);
+        child.stdin.write(`${JSON.stringify({ jsonrpc: "2.0", id: 2, method: "tools/list", params: {} })}\n`);
+      } else if (response.id === 2) {
+        assert.deepEqual(response.result?.tools?.map((tool) => tool.name), expectedTools);
+        client.ready = true;
+        readyTarget.count++;
+        if (readyTarget.count === readyTarget.target) readyTarget.resolve();
+      } else if (client.pending.has(response.id)) {
+        client.pending.get(response.id)(response);
+        client.pending.delete(response.id);
+      }
+    }
+  });
+  client.result = new Promise((resolve) => child.on("exit", (code) => resolve({ code, stderr: client.stderr })));
+  return client;
+}
+
+function jobState(client) {
+  try {
+    const [active = "", seen = ""] = fs.readFileSync(client.state, "utf8").split(/\r?\n/);
+    const parse = (value) => value.split(",").filter(Boolean).map(Number);
+    return { active: parse(active), seen: parse(seen) };
+  } catch (_) { return { active: [], seen: [] }; }
+}
+function activeLeasePids(symppHome) {
+  const directory = path.join(symppHome, "runtime", "codex-plugin-leases");
+  try { return fs.readdirSync(directory).map((name) => readJson(path.join(directory, name))?.pid).filter((pid) => pid && processAlive(pid)); }
+  catch (_) { return []; }
+}
+function runtimeEpoch(runtimeFile) {
+  const state = readJson(runtimeFile);
+  return `${Number(state?.backend?.pid || 0)}:${String(state?.publication?.backend?.process_start_time_utc_ticks || "")}`;
+}
+function listenerPids(shell, port) {
+  const result = spawnSync(shell, ["-NoProfile", "-NonInteractive", "-Command", "@(Get-NetTCPConnection -LocalPort $env:FIXTURE_PORT -State Listen -ErrorAction SilentlyContinue | Select-Object -ExpandProperty OwningProcess -Unique) | ConvertTo-Json -Compress"], { env: { ...process.env, FIXTURE_PORT: String(port) }, encoding: "utf8", windowsHide: true });
+  assert.equal(result.status, 0, result.stderr);
+  return result.stdout.trim() ? [].concat(JSON.parse(result.stdout.trim())) : [];
+}
+async function closeJob(client, graceful = false) {
+  if (client.child.exitCode === null) {
+    if (graceful) client.child.stdin.end();
+    else process.kill(client.child.pid);
+  }
+  await client.result;
+  await waitFor(() => jobState(client).seen.every((pid) => !processAlive(pid)), `Closed Job retained a process. ${JSON.stringify(jobState(client))}`);
+}
+async function jobOwner(clients, pid) {
+  return waitFor(() => clients.find((client) => client.child.exitCode === null && jobState(client).active.includes(pid)), `No active Job owns PID ${pid}.`);
+}
+
+async function certifyJobs({ clients, shell, runtimeFile, backendState, backendPort, traceDir, symppHome }) {
+  const initial = readJson(backendState);
+  assert.equal(clients.length, 32);
+  assert.equal(new Set(clients.map((client) => client.rootPid)).size, 32);
+  assert.equal(initial.starts, 1);
+  assert.equal(traceCount(traceDir, "runtime_ready_published"), 1);
+  assert.deepEqual(listenerPids(shell, backendPort), [initial.pid]);
+  await waitFor(() => activeLeasePids(symppHome).length === 32, "Initial adapters did not publish 32 active leases.");
+  const initialOwner = await jobOwner(clients, initial.pid);
+  assert.ok(jobState(initialOwner).active.includes(initial.pid), "Initial backend was not owned by its adapter Job.");
+  const sentinel = clients.find((client) => client !== initialOwner);
+  const sentinelAdapter = await waitFor(() => activeLeasePids(symppHome).find((pid) => jobState(sentinel).active.includes(pid)), "Original follower adapter was not observed in its Job.");
+  let requestId = 4000;
+
+  const follower = clients.find((client) => client !== initialOwner && client !== sentinel);
+  const stableEpoch = runtimeEpoch(runtimeFile);
+  await closeJob(follower);
+  assert.equal(runtimeEpoch(runtimeFile), stableEpoch);
+  assert.deepEqual(listenerPids(shell, backendPort), [initial.pid]);
+  assert.equal((await requestClient(sentinel, requestId++, "tools/list")).result?.tools?.length, expectedTools.length);
+
+  async function rotate(trigger) {
+    const before = readJson(backendState);
+    const owner = await jobOwner(clients, before.pid);
+    assert.ok(jobState(owner).active.includes(before.pid), "Published owner Job did not contain the backend.");
+    await closeJob(owner);
+    await waitFor(() => !processAlive(before.pid) && portAvailable(backendPort), "Owner Job close did not kill its backend.");
+    assert.ok(processAlive(sentinelAdapter), "Original follower adapter did not survive owner Job close.");
+    const response = await requestClient(trigger, requestId++, "tools/list");
+    assert.equal(response.result?.tools?.length, expectedTools.length);
+    const after = await waitFor(() => { const value = readJson(backendState); return value?.starts === before.starts + 1 && value; }, "Owner rotation did not create one replacement epoch.");
+    assert.notEqual(after.pid, before.pid);
+    assert.deepEqual(listenerPids(shell, backendPort), [after.pid]);
+    await delay(250);
+    assert.equal(readJson(backendState).starts, before.starts + 1);
+    assert.equal((await requestClient(sentinel, requestId++, "tools/list")).result?.tools?.length, expectedTools.length);
+    assert.ok(processAlive(sentinelAdapter));
+    return after;
+  }
+
+  const firstTrigger = clients.find((client) => client.child.exitCode === null && client !== sentinel && client !== initialOwner);
+  await rotate(firstTrigger);
+  const secondTrigger = clients.find((client) => client.child.exitCode === null && client !== sentinel && client !== firstTrigger);
+  await rotate(secondTrigger);
+
+  const currentOwner = await jobOwner(clients, readJson(backendState).pid);
+  const beforePrune = runtimeEpoch(runtimeFile);
+  for (const client of clients) if (client.child.exitCode === null && client !== sentinel && client !== currentOwner) await closeJob(client);
+  assert.equal(runtimeEpoch(runtimeFile), beforePrune);
+  assert.deepEqual(listenerPids(shell, backendPort), [readJson(backendState).pid]);
+  assert.deepEqual(clients.filter((client) => client.child.exitCode === null), [sentinel, currentOwner]);
+  await rotate(sentinel);
+
+  const beforeBackendCrash = readJson(backendState);
+  const sentinelRoot = sentinel.rootPid;
+  process.kill(beforeBackendCrash.pid);
+  await waitFor(() => !processAlive(beforeBackendCrash.pid), "Backend-only crash did not stop the backend.");
+  assert.ok(processAlive(sentinelAdapter) && processAlive(sentinelRoot), "Backend-only crash killed the original STDIO client.");
+  assert.equal((await requestClient(sentinel, requestId++, "tools/list")).result?.tools?.length, expectedTools.length);
+  const afterBackendCrash = await waitFor(() => { const value = readJson(backendState); return value?.starts === beforeBackendCrash.starts + 1 && value; }, "Backend-only crash did not recover once.");
+  assert.deepEqual(listenerPids(shell, backendPort), [afterBackendCrash.pid]);
+
+  const indeterminate = await requestClient(sentinel, requestId++, "tools/call", { name: "fixture.mutate", arguments: {} });
+  assert.equal(indeterminate.error?.code, -32001);
+  assert.equal(indeterminate.error?.data?.replayed, false);
+  const afterMutation = await waitFor(() => { const value = readJson(backendState); return value?.starts === afterBackendCrash.starts + 1 && value; }, "Ambiguous mutation did not recover one backend.");
+  assert.equal(afterMutation.mutations, 1);
+  assert.equal((await requestClient(sentinel, requestId++, "tools/list")).result?.tools?.length, expectedTools.length);
+  assert.equal(readJson(backendState).mutations, 1);
+  assert.equal(traceCount(traceDir, "runtime_ready_published"), 6);
+
+  await closeJob(sentinel, true);
+  await waitFor(() => portAvailable(backendPort), "Final Job close retained the backend listener.");
+  await waitFor(() => fs.readdirSync(path.join(symppHome, "runtime", "codex-plugin-leases"), { withFileTypes: true }).filter((entry) => entry.isFile()).length === 0, "Final Job close retained an adapter lease file.");
+  const seen = [...new Set(clients.flatMap((client) => jobState(client).seen))];
+  assert.ok(seen.length >= clients.length * 2, "Job snapshots did not observe launcher descendants.");
+  assert.ok(seen.every((pid) => !processAlive(pid)), `Final Job close retained test-owned processes: ${seen.filter(processAlive)}`);
+  assert.ok(clients.every((client) => client.child.exitCode !== null));
+  return { mode: "job_certification", clients: 32, initial_epochs: 1, owner_rotations: 3, backend_recoveries: 2, mutations: 1, original_stdio: true, processes_after: 0, listeners_after: 0, active_leases_after: 0 };
+}
+
 function requestClientLine(client, id, line) {
   return new Promise((resolve, reject) => {
     const timeout = setTimeout(() => { client.pending.delete(id); reject(new Error(`Client request ${id} timed out. ${client.stderr}`)); }, 90000);
@@ -292,8 +443,12 @@ async function runCase(clientCount, shell, mode = "normal") {
     const readyTarget = { count: 0, target: clientCount, startedAt: 0, resolve: () => readyResolve() };
     const allReady = new Promise((resolve) => { readyResolve = resolve; });
     const latencies = [];
-    for (let index = 0; index < clientCount; index++) startClient(barrier, path.join(installedRoot, "scripts", "start-sympp-mcp.cmd"), environment, clients, latencies, readyTarget);
-    await waitFor(() => clients.every((client) => client.stderr.includes("BARRIER_READY")), "Clients did not reach the start barrier.");
+    const jobCertification = mode === "job_certification";
+    for (let index = 0; index < clientCount; index++) {
+      if (jobCertification) startJobClient(root, barrier, path.join(installedRoot, "scripts", "start-sympp-mcp.cmd"), environment, clients, latencies, readyTarget, index);
+      else startClient(barrier, path.join(installedRoot, "scripts", "start-sympp-mcp.cmd"), environment, clients, latencies, readyTarget);
+    }
+    await waitFor(() => jobCertification ? clients.every((client) => client.jobReady) : clients.every((client) => client.stderr.includes("BARRIER_READY")), "Clients did not reach the start barrier.");
     readyTarget.startedAt = Date.now();
     for (const client of clients) client.child.stdin.write(`${JSON.stringify({ jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "cold-herd", version: "1" } } })}\n`);
     fs.writeFileSync(barrier, "go");
@@ -329,6 +484,11 @@ async function runCase(clientCount, shell, mode = "normal") {
     const firstBackend = readJson(backendState);
     const firstOwnerPid = Number(readJson(runtimeFile)?.publication?.owner_adapter_pid || 0);
     const backendOnlyReadRecovery = mode.endsWith("backend_only_read_recovery");
+    if (jobCertification) {
+      const result = await certifyJobs({ clients, shell, runtimeFile, backendState, backendPort, traceDir, symppHome });
+      backendPid = 0;
+      return result;
+    }
     let recoveryClients = clients;
     if (mode === "powershell_fallback_initialize_retry") {
       fs.writeFileSync(failAfterProbeFile, "ready");
@@ -538,6 +698,11 @@ async function main() {
     process.stdout.write(`${JSON.stringify(result)}\n`);
     return;
   }
+  if (process.env.SYMPP_JOB_CERTIFICATION) {
+    const result = await runCase(32, pwsh, "job_certification");
+    process.stdout.write(`${JSON.stringify(result)}\n`);
+    return;
+  }
   const results = [];
   results.push(await runCase(30, windowsPowerShell));
   results.push(await runCase(100, pwsh));

plugins/symphony-plus-plus-mcp/tests/launcher/job-client.cs

@@ -0,0 +1,63 @@
+using System;
+using System.ComponentModel;
+using System.Diagnostics;
+using System.IO;
+using System.Runtime.InteropServices;
+using System.Threading.Tasks;
+
+public sealed class JobHandle : IDisposable
+{
+    const int ExtendedLimitInformation = 9, BasicProcessIdList = 3;
+    const uint KillOnClose = 0x2000, BreakawayOk = 0x800, SilentBreakawayOk = 0x1000;
+    IntPtr handle;
+    [StructLayout(LayoutKind.Sequential)] struct IoCounters { public ulong ReadOps, WriteOps, OtherOps, ReadBytes, WriteBytes, OtherBytes; }
+    [StructLayout(LayoutKind.Sequential)] struct BasicLimits
+    {
+        public long PerProcessTime, PerJobTime;
+        public uint Flags;
+        public UIntPtr MinimumWorkingSet, MaximumWorkingSet;
+        public uint ActiveProcessLimit;
+        public UIntPtr Affinity;
+        public uint PriorityClass, SchedulingClass;
+    }
+    [StructLayout(LayoutKind.Sequential)] struct ExtendedLimits { public BasicLimits Basic; public IoCounters Io; public UIntPtr ProcessMemory, JobMemory, PeakProcessMemory, PeakJobMemory; }
+    [DllImport("kernel32.dll", SetLastError = true)] static extern IntPtr CreateJobObject(IntPtr attributes, string name);
+    [DllImport("kernel32.dll", SetLastError = true)] static extern bool SetInformationJobObject(IntPtr job, int type, IntPtr value, uint length);
+    [DllImport("kernel32.dll", SetLastError = true)] static extern bool QueryInformationJobObject(IntPtr job, int type, IntPtr value, uint length, out uint returned);
+    [DllImport("kernel32.dll", SetLastError = true)] static extern bool AssignProcessToJobObject(IntPtr job, IntPtr process);
+    [DllImport("kernel32.dll")] static extern bool CloseHandle(IntPtr handle);
+    static void Require(bool value, string action) { if (!value) throw new Win32Exception(Marshal.GetLastWin32Error(), action); }
+
+    public JobHandle()
+    {
+        handle = CreateJobObject(IntPtr.Zero, null); Require(handle != IntPtr.Zero, "CreateJobObject");
+        var limits = new ExtendedLimits { Basic = new BasicLimits { Flags = KillOnClose } };
+        var buffer = Marshal.AllocHGlobal(Marshal.SizeOf<ExtendedLimits>());
+        try
+        {
+            Marshal.StructureToPtr(limits, buffer, false);
+            Require(SetInformationJobObject(handle, ExtendedLimitInformation, buffer, (uint)Marshal.SizeOf<ExtendedLimits>()), "SetInformationJobObject");
+            uint returned;
+            Require(QueryInformationJobObject(handle, ExtendedLimitInformation, buffer, (uint)Marshal.SizeOf<ExtendedLimits>(), out returned), "QueryInformationJobObject");
+            var flags = Marshal.PtrToStructure<ExtendedLimits>(buffer).Basic.Flags;
+            if ((flags & KillOnClose) == 0 || (flags & (BreakawayOk | SilentBreakawayOk)) != 0) throw new InvalidOperationException("Job limits are not non-breakaway kill-on-close.");
+        }
+        finally { Marshal.FreeHGlobal(buffer); }
+    }
+    public void Assign(Process process) { Require(AssignProcessToJobObject(handle, process.Handle), "AssignProcessToJobObject"); }
+    public int[] Pids()
+    {
+        var buffer = Marshal.AllocHGlobal(4096);
+        try
+        {
+            uint returned;
+            Require(QueryInformationJobObject(handle, BasicProcessIdList, buffer, 4096, out returned), "QueryInformationJobObject");
+            var result = new int[Marshal.ReadInt32(buffer, 4)];
+            for (var i = 0; i < result.Length; i++) result[i] = (int)Marshal.ReadIntPtr(buffer, 8 + i * IntPtr.Size);
+            return result;
+        }
+        finally { Marshal.FreeHGlobal(buffer); }
+    }
+    public static async Task ProxyInput(Stream input, Stream output) { await input.CopyToAsync(output); output.Close(); }
+    public void Dispose() { if (handle != IntPtr.Zero) { CloseHandle(handle); handle = IntPtr.Zero; } }
+}

plugins/symphony-plus-plus-mcp/tests/launcher/job-client.ps1

@@ -0,0 +1,29 @@
+param([Parameter(Mandatory)][string]$StateFile, [Parameter(Mandatory)][string]$ClientScript)
+$ErrorActionPreference = "Stop"
+Add-Type -Path $env:SYMPP_JOB_HELPER_ASSEMBLY
+$job = [JobHandle]::new()
+try {
+  $start = [System.Diagnostics.ProcessStartInfo]::new("cmd.exe", "/d /s /c call `"$ClientScript`"")
+  $start.UseShellExecute = $false
+  $start.RedirectStandardInput = $true
+  $start.RedirectStandardOutput = $true
+  $start.RedirectStandardError = $true
+  $start.CreateNoWindow = $true
+  $process = [System.Diagnostics.Process]::Start($start)
+  $job.Assign($process)
+  [Console]::Error.WriteLine("JOB_READY:$($process.Id)")
+  $inputTask = [JobHandle]::ProxyInput([Console]::OpenStandardInput(), $process.StandardInput.BaseStream)
+  $outputTask = $process.StandardOutput.BaseStream.CopyToAsync([Console]::OpenStandardOutput())
+  $errorTask = $process.StandardError.BaseStream.CopyToAsync([Console]::OpenStandardError())
+  $seen = [System.Collections.Generic.HashSet[int]]::new()
+  do {
+    $active = @($job.Pids())
+    foreach ($processId in $active) { [void]$seen.Add($processId) }
+    [System.IO.File]::WriteAllText("$StateFile.tmp", (($active -join ",") + "`n" + (@($seen) -join ",")))
+    [System.IO.File]::Move("$StateFile.tmp", $StateFile, $true)
+  } while (-not $process.WaitForExit(25))
+  [System.Threading.Tasks.Task]::WaitAll(@($outputTask, $errorTask))
+  exit $process.ExitCode
+} finally {
+  $job.Dispose()
+}

plugins/symphony-plus-plus-mcp/tests/launcher/run-job-object-certification.ps1

@@ -0,0 +1,25 @@
+[System.Diagnostics.Process]::GetCurrentProcess().PriorityClass = [System.Diagnostics.ProcessPriorityClass]::BelowNormal
+$ErrorActionPreference = "Stop"
+$inheritedPriority = & (Get-Command pwsh -ErrorAction Stop).Source -NoProfile -NonInteractive -Command '[System.Diagnostics.Process]::GetCurrentProcess().PriorityClass'
+if ($inheritedPriority -ne "BelowNormal") { throw "Job certification descendants must inherit BelowNormal priority." }
+$root = Join-Path ([System.IO.Path]::GetTempPath()) ("sympp-job-client-" + [guid]::NewGuid().ToString("N"))
+$previousRunner = $env:SYMPP_JOB_CLIENT
+$previousAssembly = $env:SYMPP_JOB_HELPER_ASSEMBLY
+$previousCertification = $env:SYMPP_JOB_CERTIFICATION
+try {
+  New-Item -ItemType Directory -Path $root | Out-Null
+  $assembly = Join-Path $root "job-client.dll"
+  Add-Type -TypeDefinition (Get-Content -LiteralPath (Join-Path $PSScriptRoot "job-client.cs") -Raw) -Language CSharp -OutputAssembly $assembly
+  $env:SYMPP_JOB_CLIENT = Join-Path $PSScriptRoot "job-client.ps1"
+  $env:SYMPP_JOB_HELPER_ASSEMBLY = $assembly
+  $env:SYMPP_JOB_CERTIFICATION = "1"
+  & (Get-Command node.exe -ErrorAction Stop).Source (Join-Path $PSScriptRoot "cold-start-singleton-smoke.js")
+  if ($LASTEXITCODE -ne 0) { throw "Windows Job Object certification failed with exit code $LASTEXITCODE." }
+} finally {
+  $env:SYMPP_JOB_CLIENT = $previousRunner
+  $env:SYMPP_JOB_HELPER_ASSEMBLY = $previousAssembly
+  $env:SYMPP_JOB_CERTIFICATION = $previousCertification
+  $tempRoot = [System.IO.Path]::GetFullPath([System.IO.Path]::GetTempPath())
+  $resolvedRoot = [System.IO.Path]::GetFullPath($root)
+  if ($resolvedRoot.StartsWith($tempRoot, [System.StringComparison]::OrdinalIgnoreCase)) { Remove-Item -LiteralPath $resolvedRoot -Recurse -Force -ErrorAction SilentlyContinue }
+}