From 5c98f042997b9eab35b2c3d9c0329c113de47c26 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 30 Sep 2026 04:33:09 -0500 Subject: [PATCH 1/2] fix(runner-e2e): bound remote observer startup requests Co-Authored-By: Paperclip --- .../runner-e2e/remote-native-fixtures.test.ts | 75 ++++++++++++++++++- tests/runner-e2e/remote-native-fixtures.ts | 4 +- 2 files changed, 74 insertions(+), 5 deletions(-) diff --git a/tests/runner-e2e/remote-native-fixtures.test.ts b/tests/runner-e2e/remote-native-fixtures.test.ts index efae960daee..07faa8883b2 100644 --- a/tests/runner-e2e/remote-native-fixtures.test.ts +++ b/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 }> = []; @@ -274,13 +274,82 @@ 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 = {}) { 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, install: h.calls[0]!, replaceRuntimeRoot() { runtimeInode = 999n; } }; } + async function generatedRpc(o: Awaited>, request: Record) { + 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, mkdirSync: vi.fn(), existsSync: () => o.handlers.length > 0, + 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 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 }); o.handlers[0]!(observer); + queueMicrotask(() => client.emit("connect")); return client; + } }; + new Script(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: () => ({ unref() {} }) }); + await new Promise(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("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 } }]); diff --git a/tests/runner-e2e/remote-native-fixtures.ts b/tests/runner-e2e/remote-native-fixtures.ts index bcb76e0af8c..fbdfc17cdbb 100644 --- a/tests/runner-e2e/remote-native-fixtures.ts +++ b/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()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() { From a1145db4d8dbde17e624f8a14ef2726b33d09083 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 30 Sep 2026 04:51:08 -0500 Subject: [PATCH 2/2] test(runner-e2e): require observer startup readiness Co-Authored-By: Paperclip --- .../runner-e2e/remote-native-fixtures.test.ts | 66 ++++++++++++++++--- 1 file changed, 58 insertions(+), 8 deletions(-) diff --git a/tests/runner-e2e/remote-native-fixtures.test.ts b/tests/runner-e2e/remote-native-fixtures.test.ts index 07faa8883b2..5c05950bee0 100644 --- a/tests/runner-e2e/remote-native-fixtures.test.ts +++ b/tests/runner-e2e/remote-native-fixtures.test.ts @@ -231,6 +231,9 @@ describe("actual generated observer state machine", () => { const missing = () => { throw Object.assign(new Error("missing"), { code: "ENOENT" }); }; const fds = new Map(); let nextFd = 50, runtimeInode = 4n; const symbolicLinks = new Set(); + 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 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"); }, @@ -280,9 +297,9 @@ describe("actual generated observer state machine", () => { 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, context, server, install: h.calls[0]!, 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>, request: Record) { + async function generatedRpc(o: Awaited>, request: Record, 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 }> = []; @@ -295,26 +312,32 @@ describe("actual generated observer state machine", () => { new Script(o.files.get(argv[0]!)!.toString()).runInNewContext(o.context); return { unref: vi.fn() }; }); - const fs = { ...o.fs, mkdirSync: vi.fn(), existsSync: () => o.handlers.length > 0, + 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 }); o.handlers[0]!(observer); + connections.push({ client, observer }); handler(observer); queueMicrotask(() => client.emit("connect")); return client; } }; - new Script(source).runInNewContext({ require(name: string) { + 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: () => ({ unref() {} }) }); + }, 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(resolve => setImmediate(resolve)); return { output, process, forwarded, spawn, connections }; } @@ -340,6 +363,33 @@ describe("actual generated observer state machine", () => { 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) });