fix(openshell): use confirmed engine cleanup inventories (#2603)

Podman registers its native /libpod routes only under a version prefix, so the observer's unversioned /libpod container list and inspect calls got the router's plain-text Not Found, which the reader then failed to parse as JSON; an owned delete stopped unconfirmed at the cleanup inventory. The observer now uses the compatible /containers/json?all=true and /containers/<id>/json routes, which Podman serves with or without a prefix and whose Id, Names, Labels and Mounts fields the existing selection already reads. Volume absence no longer comes from a per-volume 404: one /volumes listing must be a 200 with an array of named Volumes and no Warnings, and presence is name membership in it; anything else is environment_cleanup_unconfirmed with the HTTP status. The reader parses JSON only for a 200, and malformed 200 inventory is a typed unconfirmed error. Socket identity, workload and supervisor selection, ownership and lifecycle are unchanged, and nothing is sent to the engine but GETs. Limits: container rows are still checked only for status and array shape, so a 200 of [{}] with an empty volume list would read as absent (unchanged here; Podman emits full records or an error); a thrown cleanup error still omits its stage and dispatch status; this is a snapshot, not an atomic inventory; maintained native owned-delete isn't requalified by this change.

packages/daemon/src/adapters/openshell-podman-observer.ts

@@ -12,7 +12,11 @@ const read: EngineRead = (socket, path) => new Promise((resolve, reject) => {
   const req = request({ socketPath: socket, path, method: "GET", timeout: 5000 }, res => {
     const chunks: Buffer[] = []; let bytes = 0;
     res.on("data", chunk => { bytes += chunk.length; if (bytes > 2 * 1024 * 1024) { req.destroy(new Error("Engine inventory exceeds its bound")); return; } chunks.push(chunk); });
-    res.on("error", reject); res.on("end", () => { try { resolve({ status: res.statusCode ?? 0, body: chunks.length ? JSON.parse(Buffer.concat(chunks).toString("utf8")) : null }); } catch (error) { reject(error); } });
+    res.on("error", reject); res.on("end", () => {
+      const status = res.statusCode ?? 0, text = Buffer.concat(chunks).toString("utf8");
+      try { resolve({ status, body: status === 200 && text ? JSON.parse(text) : text }); }
+      catch { reject(new EnvironmentError("environment_cleanup_unconfirmed", "The local engine returned malformed inventory JSON")); }
+    });
   });
   req.on("timeout", () => req.destroy(new Error("Engine inventory timed out"))); req.on("error", reject); req.end();
 });
@@ -27,8 +31,8 @@ export class OpenShellPodmanObserver {
     return { socket: path, socketDevice: stat.dev, socketInode: stat.ino, uid, host: hostname() };
   }
   private async rows(socket: string): Promise<Array<Record<string, any>>> {
-    const result = await this.get(socket, "/libpod/containers/json?all=true");
-    if (result.status !== 200 || !Array.isArray(result.body)) throw new EnvironmentError("environment_cleanup_unconfirmed", "Complete local engine inventory is unavailable");
+    const result = await this.get(socket, "/containers/json?all=true");
+    if (result.status !== 200 || !Array.isArray(result.body)) throw new EnvironmentError("environment_cleanup_unconfirmed", `Complete local engine inventory is unavailable (HTTP ${result.status})`);
     return result.body;
   }
   async capture(binding: EnvironmentBinding, path: string): Promise<PodmanCleanupSelection> {
@@ -44,7 +48,7 @@ export class OpenShellPodmanObserver {
     const mounted = new Set<string>();
     for (const container of containers) {
       if (typeof container.Id !== "string" || !/^[0-9a-f]{64}$/.test(container.Id)) throw new EnvironmentError("environment_cleanup_unconfirmed", "Engine container identity is malformed");
-      const inspected = await this.get(path, `/libpod/containers/${container.Id}/json`);
+      const inspected = await this.get(path, `/containers/${container.Id}/json`);
       const actual = inspected.body as Record<string, any>;
       if (inspected.status !== 200 || actual?.Id !== container.Id || !Array.isArray(actual.Mounts)) throw new EnvironmentError("environment_cleanup_unconfirmed", "The captured container cannot be inspected");
       for (const mount of actual.Mounts) if (mount.Type === "volume" && typeof mount.Name === "string") mounted.add(mount.Name);
@@ -61,12 +65,14 @@ export class OpenShellPodmanObserver {
     const rows = await this.rows(selected.socket);
     const workload = rows.some(row => row.Id === selected.workloadId || row.Labels?.["openshell.ai/sandbox-id"] === selected.resourceId && row.Labels?.["openshell.ai/isolation-role"] === "sandbox");
     const supervisor = rows.some(row => row.Id === selected.supervisorId || Array.isArray(row.Names) && row.Names.some((name: string) => name.replace(/^\//, "") === `openshell-supervisor-${selected.resourceId}`));
-    let volumes = false;
-    for (const name of selected.volumes) {
-      const observed = await this.get(selected.socket, `/libpod/volumes/${encodeURIComponent(name)}/json`);
-      if (observed.status === 200) volumes = true;
-      else if (observed.status !== 404) throw new EnvironmentError("environment_cleanup_unconfirmed", "Volume lookup failed; it is not absence");
-    }
+    // A route-level 404 is not volume absence. Require a complete positive inventory.
+    const observed = await this.get(selected.socket, "/volumes");
+    const listing = observed.body as { Volumes?: unknown; Warnings?: unknown } | null;
+    const inventory = listing?.Volumes, warnings = listing?.Warnings;
+    if (observed.status !== 200 || !Array.isArray(inventory) || inventory.some(volume => typeof volume?.Name !== "string")
+      || warnings != null && (!Array.isArray(warnings) || warnings.length !== 0))
+      throw new EnvironmentError("environment_cleanup_unconfirmed", `Complete local volume inventory is unavailable (HTTP ${observed.status})`);
+    const volumes = selected.volumes.some(name => inventory.some(volume => volume.Name === name));
     return { workload, supervisor, volumes };
   }
 }

packages/daemon/test/openshell-podman-observer.test.ts

@@ -1,6 +1,7 @@
-import { afterEach, expect, it } from "vitest";
+import { afterEach, expect, it, vi } from "vitest";
 import { createServer, type Server } from "node:net";
-import { mkdtempSync, realpathSync, rmSync } from "node:fs";
+import { createServer as createHttpServer } from "node:http";
+import { lstatSync, mkdtempSync, realpathSync, rmSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { OpenShellPodmanObserver } from "../src/adapters/openshell-podman-observer.js";
@@ -15,14 +16,14 @@ async function fixture() {
   const id = "compute-one", workloadId = "a".repeat(64), supervisorId = "b".repeat(64);
   let rows: Record<string, any>[] = [{ Id: workloadId, Names: ["workload"], Labels: { "openshell.managed": "true", "openshell.ai/sandbox-id": id, "openshell.ai/isolation-role": "sandbox", "openshell.ai/sandbox-workspace": "workspace" } },
     { Id: supervisorId, Names: [`openshell-supervisor-${id}`], Labels: {} }];
-  let volumeStatus = 404;
+  let volumeStatus = 200;
   const paths: string[] = [];
   const observer = new OpenShellPodmanObserver(async (_socket, path) => {
     paths.push(path);
-    if (path.startsWith("/libpod/containers/json")) return { status: 200, body: rows };
-    if (path.startsWith("/libpod/containers/")) return { status: 200, body: { Id: path.includes(workloadId) ? workloadId : supervisorId,
+    if (path.startsWith("/containers/json")) return { status: 200, body: rows };
+    if (path.startsWith("/containers/")) return { status: 200, body: { Id: path.includes(workloadId) ? workloadId : supervisorId,
       Mounts: [`openshell-sandbox-${id}-workspace`, `openshell-control-${id}`, `openshell-channel-${id}`].map(Name => ({ Type: "volume", Name })) } };
-    return { status: volumeStatus, body: {} };
+    return { status: volumeStatus, body: { Volumes: [], Warnings: [] } };
   });
   const binding = { resource: { id }, placement: { workspace: "workspace" } } as EnvironmentBinding;
   return { observer, binding, socket, paths, disappear: () => { rows = [{ Id: "c".repeat(64), Names: ["unrelated"], Labels: {} }]; }, failVolume: () => { volumeStatus = 500; } };
@@ -31,10 +32,71 @@ it.skipIf(!supported)("captures exact local workload/supervisor/volumes then ign
   const f = await fixture(), selected = await f.observer.capture(f.binding, f.socket);
   expect(selected).toMatchObject({ resourceId: "compute-one", workloadId: "a".repeat(64), supervisorId: "b".repeat(64) });
   f.disappear(); expect(await f.observer.observe(f.binding, selected)).toEqual({ workload: false, supervisor: false, volumes: false });
-  expect(f.paths.every(path => path.startsWith("/libpod/containers/") || path.startsWith("/libpod/volumes/"))).toBe(true);
+  expect(f.paths.every(path => path.startsWith("/containers/") || path === "/volumes")).toBe(true);
 });
 it.skipIf(!supported)("does not treat an engine error or missing original inventory as cleanup", async () => {
   const f = await fixture(), selected = await f.observer.capture(f.binding, f.socket); f.disappear(); f.failVolume();
   await expect(f.observer.observe(f.binding, selected)).rejects.toMatchObject({ code: "environment_cleanup_unconfirmed" });
   await expect(f.observer.observe(f.binding)).rejects.toMatchObject({ code: "environment_cleanup_unconfirmed" });
 });
+
+async function httpFixture(failure?: "404" | "malformed") {
+  const root = realpathSync(mkdtempSync(join(tmpdir(), "os-")));
+  closers.push(() => rmSync(root, { recursive: true, force: true }));
+  const socket = join(root, "s"), id = "compute-one", workloadId = "a".repeat(64), supervisorId = "b".repeat(64);
+  const ownedVolumes = [`openshell-sandbox-${id}-workspace`, `openshell-control-${id}`, `openshell-channel-${id}`];
+  const unrelated = { Id: "c".repeat(64), Names: ["unrelated"], Labels: {} };
+  let containers = [unrelated,
+    { Id: workloadId, Names: ["workload"], Labels: { "openshell.managed": "true", "openshell.ai/sandbox-id": id, "openshell.ai/isolation-role": "sandbox", "openshell.ai/sandbox-workspace": "workspace" } },
+    { Id: supervisorId, Names: [`openshell-supervisor-${id}`], Labels: {} }];
+  let volumes = [...ownedVolumes, "unrelated-volume"], volumeMissingRoute = false;
+  const requests: Array<{ method: string | undefined; path: string | undefined }> = [];
+  const server = createHttpServer((req, res) => {
+    requests.push({ method: req.method, path: req.url });
+    if (failure === "404" || req.url?.startsWith("/libpod/") || volumeMissingRoute && req.url === "/volumes") {
+      res.writeHead(404, { "Content-Type": "text/plain" }); res.end("Not Found\n"); return;
+    }
+    if (failure === "malformed") { res.writeHead(200); res.end("not JSON\n"); return; }
+    let body: unknown;
+    if (req.url === "/containers/json?all=true") body = containers;
+    else if (req.url === `/containers/${workloadId}/json` || req.url === `/containers/${supervisorId}/json`)
+      body = { Id: req.url.split("/")[2], Mounts: ownedVolumes.map(Name => ({ Type: "volume", Name })) };
+    else if (req.url === "/volumes") body = { Volumes: volumes.map(Name => ({ Name })), Warnings: [] };
+    else { res.writeHead(404); res.end("Not Found\n"); return; }
+    res.writeHead(200, { "Content-Type": "application/json" }); res.end(JSON.stringify(body));
+  });
+  await new Promise<void>(resolve => server.listen(socket, resolve));
+  closers.push(() => new Promise<void>(resolve => server.close(() => resolve())));
+  const observer = new OpenShellPodmanObserver(), st = lstatSync(socket);
+  // Exercise the real HTTP reader on macOS too; rootless socket selection has its Linux cells above.
+  const selection = vi.spyOn(observer as unknown as { socket(path: string): unknown }, "socket")
+    .mockImplementation(path => ({ socket: path, socketDevice: st.dev, socketInode: st.ino, uid: st.uid, host: "engine.test" }));
+  closers.push(() => selection.mockRestore());
+  return {
+    observer, socket, requests, binding: { resource: { id }, placement: { workspace: "workspace" } } as EnvironmentBinding,
+    retireContainers: () => { containers = [unrelated]; volumes = [ownedVolumes[0]!, "unrelated-volume"]; },
+    retireVolumes: () => { volumes = ["unrelated-volume"]; },
+    loseVolumeRoute: () => { volumeMissingRoute = true; },
+  };
+}
+
+it("uses the engine's HTTP routes and positive inventories to verify exact dependent cleanup", async () => {
+  const f = await httpFixture(), selected = await f.observer.capture(f.binding, f.socket);
+  expect(selected).toMatchObject({ resourceId: "compute-one", workloadId: "a".repeat(64), supervisorId: "b".repeat(64) });
+  expect(await f.observer.observe(f.binding, selected)).toEqual({ workload: true, supervisor: true, volumes: true });
+  f.retireContainers();
+  expect(await f.observer.observe(f.binding, selected)).toEqual({ workload: false, supervisor: false, volumes: true });
+  f.retireVolumes();
+  expect(await f.observer.observe(f.binding, selected)).toEqual({ workload: false, supervisor: false, volumes: false });
+  f.loseVolumeRoute();
+  await expect(f.observer.observe(f.binding, selected)).rejects.toMatchObject({ code: "environment_cleanup_unconfirmed" });
+  expect(f.requests.every(req => req.method === "GET")).toBe(true);
+  expect(f.requests.some(req => req.path?.startsWith("/libpod/"))).toBe(false);
+});
+
+it.each(["404", "malformed"] as const)("keeps an HTTP %s inventory response unconfirmed", async failure => {
+  const f = await httpFixture(failure);
+  await expect(f.observer.capture(f.binding, f.socket)).rejects.toMatchObject({ code: "environment_cleanup_unconfirmed" });
+  expect(f.requests).toHaveLength(1);
+  expect(f.requests[0]?.method).toBe("GET");
+});