Combine reviewed remote observer fixture fixes

Co-Authored-By: Paperclip <noreply@paperclip.ing>

* commit 'a1145db4d8dbde17e624f8a14ef2726b33d09083':
  test(runner-e2e): require observer startup readiness
  fix(runner-e2e): bound remote observer startup requests

tests/runner-e2e/remote-native-fixtures.test.ts

@@ -220,7 +220,7 @@ describe("independent filesystem and process observations", () => {
 });
 
 describe("actual generated observer state machine", () => {
-  async function observerHarness() {
+  async function observerHarness(deferStartup = false) {
     const h = harness(); await bindRemoteNativeFixture(h.options);
     const { source, config } = h.calls[0]!.request;
     const intervals: Array<() => void> = [], timers: Array<{ fn: () => void; ms: number }> = [];
@@ -231,6 +231,9 @@ describe("actual generated observer state machine", () => {
     const missing = () => { throw Object.assign(new Error("missing"), { code: "ENOENT" }); };
     const fds = new Map<number, string>(); let nextFd = 50, runtimeInode = 4n;
     const symbolicLinks = new Set<string>();
+    const directories = new Set(["/tmp", "/workspace", "/workspace/.paperclip-runtime", "/workspace/.paperclip-runtime/paperclip-runner", "/workspace/.paperclip-runtime/paperclip-runner/sessions"]);
+    if (!deferStartup) directories.add(config.root);
+    const listeners = new Map<string, (socket: any) => void>();
     const fs = {
       constants: { O_RDONLY: 0, O_NOFOLLOW: 131072 },
       openSync(path: string, flags: number) { expect(flags).toBe(131072); if (!files.has(path)) return missing(); const fd = nextFd++; fds.set(fd, path); return fd; },
@@ -249,7 +252,7 @@ describe("actual generated observer state machine", () => {
         if (!value) return missing(); return encoding ? value.toString() : value;
       },
       lstatSync(path: string) {
-        const directory = path === config.root || path === "/workspace" || path === "/workspace/.paperclip-runtime" || path === "/workspace/.paperclip-runtime/paperclip-runner";
+        const directory = directories.has(path);
         if (!directory && !files.has(path) && !symbolicLinks.has(path)) return missing();
         return { dev: 1n, ino: path === config.root ? 2n : path === "/workspace/.paperclip-runtime/paperclip-runner" ? runtimeInode : 3n, mtimeNs: 4n, ctimeNs: 5n, isDirectory: () => directory, isFile: () => !directory, isSymbolicLink: () => symbolicLinks.has(path), size: files.get(path)?.length ?? 0 };
       },
@@ -259,13 +262,27 @@ describe("actual generated observer state machine", () => {
         const entry = { path, callback: (callback ?? options) as (_kind: string, name: string | null) => void, closed: false }; watches.push(entry);
         return Object.assign(new EventEmitter(), { close: () => { entry.closed = true; } });
       },
+      mkdirSync: vi.fn((path: string) => {
+        if (!directories.has(path.slice(0, path.lastIndexOf("/")))) return missing();
+        if (directories.has(path) || files.has(path)) throw new Error("EEXIST");
+        directories.add(path);
+      }),
+      existsSync: (path: string) => directories.has(path) || files.has(path) || listeners.has(path),
       writeFileSync(path: string, content: string, opts: { flag: string }) {
+        if (!directories.has(path.slice(0, path.lastIndexOf("/")))) return missing();
         if (opts.flag === "wx" && files.has(path)) throw new Error("EEXIST"); files.set(path, Buffer.from(content));
         for (const w of watches) if (!w.closed && path.slice(0, path.lastIndexOf("/")) === w.path) w.callback("rename", path.slice(w.path.length + 1));
       },
       rmSync: vi.fn(),
     };
-    const server = { listen: vi.fn(), close: vi.fn() };
+    const server = {
+      listen: vi.fn((path: string) => {
+        if (!directories.has(path.slice(0, path.lastIndexOf("/")))) return missing();
+        if (listeners.has(path)) throw new Error("EADDRINUSE");
+        listeners.set(path, handlers.at(-1)!);
+      }),
+      close: vi.fn(() => listeners.clear()),
+    };
     const net = { createServer(fn: (socket: any) => void) { handlers.push(fn); return server; } };
     const context = {
       require(name: string) { if (name === "node:fs") return fs; if (name === "node:net") return net; if (name === "node:child_process") return { spawn: vi.fn(() => { const child = Object.assign(new EventEmitter(), { pid: 88, exitCode: null, signalCode: null, kill: vi.fn() }); children.push(child); return child; }) }; if (name === "node:path") return { join: (...paths: string[]) => paths.join("/"), dirname: (path: string) => path.slice(0, path.lastIndexOf("/")), basename: (path: string) => path.slice(path.lastIndexOf("/") + 1) }; if (name === "node:crypto") return { createHash }; throw new Error("unexpected module"); },
@@ -274,13 +291,115 @@ describe("actual generated observer state machine", () => {
       setInterval(fn: () => void) { intervals.push(fn); return 1; }, clearInterval: vi.fn(),
       setTimeout(fn: () => void, ms: number) { timers.push({ fn, ms }); return { unref() {} }; },
     };
-    new Script(source).runInNewContext(context);
+    if (deferStartup) files.delete(`${config.root}/observer.cjs`);
+    else new Script(source).runInNewContext(context);
     function request(op: string, args: Record<string, unknown> = {}) {
       const replies: any[] = [], socket = Object.assign(new EventEmitter(), { end: (value: string) => replies.push(JSON.parse(value)), destroy: vi.fn() });
       handlers[0]!(socket); socket.emit("data", Buffer.from(JSON.stringify({ op, nonce: config.nonce, ...args }) + "\n")); return replies;
     }
-    return { request, proc, files, watches, fs, intervals, timers, config, handlers, children, symbolicLinks, replaceRuntimeRoot() { runtimeInode = 999n; } };
+    return { request, proc, files, watches, fs, intervals, timers, config, handlers, children, symbolicLinks, context, server, directories, listeners, install: h.calls[0]!, replaceRuntimeRoot() { runtimeInode = 999n; } };
   }
+  async function generatedRpc(o: Awaited<ReturnType<typeof observerHarness>>, request: Record<string, unknown>, mutateSource = (source: string) => source) {
+    const quoted = o.install.command.match(/ -e (.+) '[A-Za-z0-9+/=]+'$/su)![1]!;
+    const source = quoted.slice(1, -1).replaceAll("'\\''", "'");
+    const forwarded: string[] = [], connections: Array<{ client: any; observer: any }> = [];
+    let output = "";
+    const process = { argv: ["node", Buffer.from(JSON.stringify(request)).toString("base64")], execPath: "/node", exitCode: 0, exit: vi.fn(), stdout: { write: (value: string) => { output += value; } } };
+    const spawn = vi.fn((_node: string, argv: string[]) => {
+      expect(argv[0]).toBe(`${o.config.root}/observer.cjs`);
+      // Execute the actual installed observer, using its actual serialized config.
+      o.context.process.argv = ["node", ...argv];
+      new Script(o.files.get(argv[0]!)!.toString()).runInNewContext(o.context);
+      return { unref: vi.fn() };
+    });
+    const fs = { ...o.fs,
+      readFileSync(path: string, encoding?: string) { return path === "/node" ? Buffer.from("node") : o.fs.readFileSync(path, encoding); } };
+    const net = { connect(path: string) {
+      expect(path).toBe(`${o.config.root}/control.sock`);
+      const client: any = Object.assign(new EventEmitter(), { setTimeout: vi.fn(), destroy: vi.fn() });
+      const handler = o.listeners.get(path);
+      if (!handler) { queueMicrotask(() => client.emit("error", new Error("ECONNREFUSED"))); return client; }
+      const observer: any = Object.assign(new EventEmitter(), {
+        end(value: string) { client.emit("data", Buffer.from(value)); client.emit("end"); observer.emit("close"); },
+        destroy: vi.fn(() => { client.emit("error", new Error("observer closed socket")); observer.emit("close"); }),
+      });
+      client.write = (value: string) => { forwarded.push(value); observer.emit("data", Buffer.from(value)); };
+      connections.push({ client, observer }); handler(observer);
+      queueMicrotask(() => client.emit("connect")); return client;
+    } };
+    new Script(mutateSource(source)).runInNewContext({ require(name: string) {
+      if (name === "node:fs") return fs;
+      if (name === "node:net") return net;
+      if (name === "node:child_process") return { spawn };
+      if (name === "node:crypto") return { createHash };
+      throw new Error("unexpected RPC module");
+    }, process, Buffer, setTimeout: (fn: () => void, ms: number) => {
+      // Advance only the bounded readiness polls; retain watchdogs as inert VM timers.
+      if (ms === 10 || ms === 50) queueMicrotask(fn);
+      return { unref() {} };
+    } });
+    await new Promise<void>(resolve => setImmediate(resolve));
+    return { output, process, forwarded, spawn, connections };
+  }
+  it("executes generated install through the bounded observer socket, then closes the exact observer", async () => {
+    const o = await observerHarness(true);
+    expect(Buffer.byteLength(o.install.request.source)).toBeGreaterThan(8192);
+    const installed = await generatedRpc(o, o.install.request);
+    expect(installed.process.exitCode).toBe(0);
+    const baseline = JSON.parse(installed.output);
+    expect(baseline.ok).toBe(true); expect(baseline.result.complete).toBe(true);
+    expect(baseline.result.processes.root.pid).toBe(21);
+    expect(installed.spawn).toHaveBeenCalledTimes(1);
+    expect(installed.forwarded.map(value => JSON.parse(value))).toEqual([{ op: "snapshot", nonce: o.config.nonce }]);
+    expect(Buffer.byteLength(installed.forwarded[0]!)).toBeLessThan(8192);
+    const closed = await generatedRpc(o, { op: "close", root: o.config.root, nonce: o.config.nonce, nodeSha256: hash("node"), timeoutMs: 10_000 });
+    expect(closed.process.exitCode).toBe(0); expect(JSON.parse(closed.output)).toEqual({ ok: true, result: { closed: true } });
+    expect(closed.spawn).not.toHaveBeenCalled();
+    expect(closed.forwarded[0]).not.toContain('"source"'); expect(closed.forwarded[0]).not.toContain('"config"');
+    expect(o.watches.length).toBeGreaterThan(0); expect(o.watches.every(w => w.closed)).toBe(true);
+    expect(o.context.clearInterval).toHaveBeenCalled(); expect(o.server.close).toHaveBeenCalled();
+    o.timers.find(t => t.ms === 100)!.fn();
+    expect(closed.connections[0]!.observer.destroy).toHaveBeenCalled();
+    expect(o.fs.rmSync).toHaveBeenCalledExactlyOnceWith(o.config.root, { recursive: true, force: true });
+    expect(o.context.process.exit).toHaveBeenCalledExactlyOnceWith(0);
+  });
+  it("rejects a generated install mutant that omits its parent directory creation", async () => {
+    const o = await observerHarness(true);
+    const installed = await generatedRpc(o, o.install.request, source => {
+      const mutant = source.replace("fs.mkdirSync(c.root,{mode:0o700});", "");
+      expect(mutant).not.toBe(source); return mutant;
+    });
+    expect(installed.process.exitCode).toBe(2); expect(installed.output).toBe("");
+    expect(installed.spawn).not.toHaveBeenCalled(); expect(installed.connections).toHaveLength(0);
+    expect(o.files.has(`${o.config.root}/observer.cjs`)).toBe(false);
+    expect(o.directories.has(o.config.root)).toBe(false); expect(o.watches).toHaveLength(0);
+  });
+  it("rejects a generated observer mutant that never listens, even after handler registration", async () => {
+    const o = await observerHarness(true);
+    const source = o.install.request.source as string;
+    const mutant = source.replace("server.listen(path.join(config.root,'control.sock'));", "");
+    expect(mutant).not.toBe(source);
+    const installed = await generatedRpc(o, { ...o.install.request, source: mutant });
+    expect(installed.process.exitCode).toBe(2); expect(installed.output).toBe("");
+    expect(installed.spawn).toHaveBeenCalledTimes(1); expect(o.handlers).toHaveLength(1);
+    expect(o.server.listen).not.toHaveBeenCalled(); expect(o.listeners.size).toBe(0);
+    expect(installed.connections).toHaveLength(0); expect(installed.forwarded).toHaveLength(0);
+    // A failed connection cannot close the unlistening observer. Its owned TTL still retires it.
+    o.timers.find(t => t.ms > 1000)!.fn(); o.timers.find(t => t.ms === 100)!.fn();
+    expect(o.watches.every(w => w.closed)).toBe(true);
+    expect(o.fs.rmSync).toHaveBeenCalledExactlyOnceWith(o.config.root, { recursive: true, force: true });
+    expect(o.context.process.exit).toHaveBeenCalledExactlyOnceWith(0);
+  });
+  it("still rejects an oversized control request and can close its observer afterward", async () => {
+    const o = await observerHarness();
+    const oversized = await generatedRpc(o, { op: "snapshot", root: o.config.root, nonce: o.config.nonce, nodeSha256: hash("node"), timeoutMs: 10_000, unexpected: "x".repeat(8192) });
+    expect(oversized.process.exitCode).toBe(2); expect(oversized.output).toBe("");
+    expect(oversized.connections[0]!.observer.destroy).toHaveBeenCalled();
+    const closed = await generatedRpc(o, { op: "close", root: o.config.root, nonce: o.config.nonce, nodeSha256: hash("node"), timeoutMs: 10_000 });
+    expect(JSON.parse(closed.output)).toEqual({ ok: true, result: { closed: true } });
+    o.timers.find(t => t.ms === 100)!.fn();
+    expect(o.watches.every(w => w.closed)).toBe(true); expect(o.context.process.exit).toHaveBeenCalledExactlyOnceWith(0);
+  });
   it("acknowledges receipt-channel readiness only after the long waiter connects", async () => {
     const o = await observerHarness(); const arm = o.request("arm"); expect(arm).toHaveLength(0);
     o.request("wait"); expect(arm).toEqual([{ ok: true, result: { armed: true, sealed: false } }]);

tests/runner-e2e/remote-native-fixtures.ts

@@ -169,8 +169,8 @@ const RPC = String.raw`const fs=require('node:fs'),net=require('node:net'),cp=re
 const startedAt=Date.now();if(!Number.isInteger(r.timeoutMs)||r.timeoutMs<1000||r.timeoutMs>300000)throw Error('rpc_deadline');setTimeout(()=>process.exit(2),r.timeoutMs).unref();
 const parseStat=PARSE_STAT,runRoot=RUN_ROOT;
 async function waitRuntime(c){const root=c.binding.remoteCwd+'/'+c.runtimeRelative,boot=fs.readFileSync('/proc/sys/kernel/random/boot_id','utf8').trim(),until=Math.min(startedAt+20000,startedAt+r.timeoutMs-1000);while(Date.now()<until){try{const s=fs.lstatSync(root);if(!s.isDirectory()||s.isSymbolicLink()||fs.realpathSync(root)!==root)throw Error('runtime_root_identity');const matches=[];const entries=fs.readdirSync('/proc');if(entries.length>8192)throw Error('proc_bound');for(const name of entries){if(!/^\d+$/.test(name)||Number(name)<2)continue;try{const p=parseStat(Number(name),fs.readFileSync('/proc/'+name+'/stat','utf8'),boot),argv=fs.readFileSync('/proc/'+name+'/cmdline').toString().split('\0').filter(Boolean);if(runRoot(argv,c.binding.runId,p)){if(argv[0]!==root+'/bin/paperclip-runnerd'||hash(fs.readFileSync('/proc/'+name+'/exe'))!==c.runnerdSha256)throw Error('runtime_binary_identity');matches.push(p)}}catch(e){if(e.code!=='ENOENT'&&e.code!=='ESRCH')throw e}}if(matches.length>1)throw Error('ambiguous_run_root');if(matches.length===1)return;}catch(e){if(e.code!=='ENOENT')throw e}await new Promise(resolve=>setTimeout(resolve,50))}throw Error('runtime_not_ready')}
-(async()=>{if(hash(fs.readFileSync(process.execPath))!==r.nodeSha256)throw Error('node_identity');if(r.op==='install'){const c=r.config;await waitRuntime(c);const st=fs.lstatSync(c.sentinel.path);if(!st.isFile()||st.isSymbolicLink()||st.size>16384||fs.realpathSync(c.sentinel.path)!==c.sentinel.path)throw Error('sentinel_type');const s=JSON.parse(fs.readFileSync(c.sentinel.path,'utf8'));if(s.version!==1||s.provider!=='daytona'||s.token!==c.sentinel.token||s.companyId!==c.binding.companyId||s.environmentId!==c.binding.environmentId)throw Error('sentinel');if(fs.realpathSync(c.binding.remoteCwd)!==c.binding.remoteCwd)throw Error('cwd');fs.mkdirSync(c.root,{mode:0o700});fs.writeFileSync(c.root+'/observer.cjs',r.source,{flag:'wx',mode:0o400});c.observerTtlMs=Math.max(1,c.observerTtlMs-(Date.now()-startedAt));const child=cp.spawn(process.execPath,[c.root+'/observer.cjs',Buffer.from(JSON.stringify(c)).toString('base64')],{detached:true,stdio:'ignore',env:{PATH:'/usr/bin:/bin'}});child.unref();r.root=c.root;r.nonce=c.nonce;r.op='snapshot';}
-for(let i=0;!fs.existsSync(r.root+'/control.sock')&&i<200;i++)await new Promise(resolve=>setTimeout(resolve,10));const socket=net.connect(r.root+'/control.sock');let output='';socket.setTimeout(Math.max(1,r.timeoutMs-(Date.now()-startedAt)));socket.on('timeout',()=>{socket.destroy();process.exitCode=2});socket.on('error',()=>{process.exitCode=2});socket.on('connect',()=>socket.write(JSON.stringify(r)+'\n'));socket.on('data',b=>{output+=b;if(Buffer.byteLength(output)>262144){socket.destroy();process.exitCode=2}});socket.on('end',()=>{if(!process.exitCode)process.stdout.write(output)});
+(async()=>{let controlRequest=r;if(hash(fs.readFileSync(process.execPath))!==r.nodeSha256)throw Error('node_identity');if(r.op==='install'){const c=r.config;await waitRuntime(c);const st=fs.lstatSync(c.sentinel.path);if(!st.isFile()||st.isSymbolicLink()||st.size>16384||fs.realpathSync(c.sentinel.path)!==c.sentinel.path)throw Error('sentinel_type');const s=JSON.parse(fs.readFileSync(c.sentinel.path,'utf8'));if(s.version!==1||s.provider!=='daytona'||s.token!==c.sentinel.token||s.companyId!==c.binding.companyId||s.environmentId!==c.binding.environmentId)throw Error('sentinel');if(fs.realpathSync(c.binding.remoteCwd)!==c.binding.remoteCwd)throw Error('cwd');fs.mkdirSync(c.root,{mode:0o700});fs.writeFileSync(c.root+'/observer.cjs',r.source,{flag:'wx',mode:0o400});c.observerTtlMs=Math.max(1,c.observerTtlMs-(Date.now()-startedAt));const child=cp.spawn(process.execPath,[c.root+'/observer.cjs',Buffer.from(JSON.stringify(c)).toString('base64')],{detached:true,stdio:'ignore',env:{PATH:'/usr/bin:/bin'}});child.unref();r.root=c.root;controlRequest={op:'snapshot',nonce:c.nonce};}
+for(let i=0;!fs.existsSync(r.root+'/control.sock')&&i<200;i++)await new Promise(resolve=>setTimeout(resolve,10));const socket=net.connect(r.root+'/control.sock');let output='';socket.setTimeout(Math.max(1,r.timeoutMs-(Date.now()-startedAt)));socket.on('timeout',()=>{socket.destroy();process.exitCode=2});socket.on('error',()=>{process.exitCode=2});socket.on('connect',()=>socket.write(JSON.stringify(controlRequest)+'\n'));socket.on('data',b=>{output+=b;if(Buffer.byteLength(output)>262144){socket.destroy();process.exitCode=2}});socket.on('end',()=>{if(!process.exitCode)process.stdout.write(output)});
 })().catch(()=>{process.exitCode=2});`;
 
 function rpcSource() {