mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-01 23:43:16 +01:00
Keep background commands alive within bounded computer owner slices
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
198b04af95
commit
9a4a3c89b2
2 files changed
+135
-19
No files matched your search
@@ -5,6 +5,8 @@ import {
|
||||
readFileSync,
|
||||
realpathSync,
|
||||
rmSync,
|
||||
mkdirSync,
|
||||
writeFileSync,
|
||||
} from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
@@ -340,6 +342,85 @@ describe("computer ownership", () => {
|
||||
await f.service.reconcile();
|
||||
expect(f.backend.stop).toHaveBeenCalledOnce();
|
||||
});
|
||||
it("returns foreground output without waiting for detached command descendants", async () => {
|
||||
const f = fixture();
|
||||
const temp = realpathSync(mkdtempSync(join(tmpdir(), "computer-command-")));
|
||||
let backgroundPid: number | undefined;
|
||||
const runPath = join(temp, "systemd-run");
|
||||
// The shim starts the real supervisor. Live Boat coverage additionally
|
||||
// verifies systemd keeps the descendant in the owner slice and retires it.
|
||||
writeFileSync(
|
||||
runPath,
|
||||
`#!/usr/bin/env python3
|
||||
import sys,subprocess,json
|
||||
args=sys.argv[1:]
|
||||
assert '--property=ExitType=cgroup' in args
|
||||
assert '--property=KillMode=control-group' in args
|
||||
assert '--wait' not in args
|
||||
output=next(v.split('append:',1)[1] for v in args if v.startswith('--property=StandardOutput='))
|
||||
error=next(v.split('append:',1)[1] for v in args if v.startswith('--property=StandardError='))
|
||||
cwd=next(v.split('=',1)[1] for v in args if v.startswith('--working-directory='))
|
||||
with open(output,'wb') as out,open(error,'wb') as err:
|
||||
subprocess.Popen(args[args.index('--')+1:],stdout=out,stderr=err,cwd=cwd,start_new_session=True)
|
||||
`,
|
||||
{ mode: 0o700 },
|
||||
);
|
||||
writeFileSync(join(temp, "systemctl"), "#!/bin/sh\necho active\n", {
|
||||
mode: 0o700,
|
||||
});
|
||||
vi.mocked(f.backend.runner).mockResolvedValue({
|
||||
execute: async (input) => {
|
||||
const payload = JSON.parse(input.stdin!);
|
||||
const ownerRoot = join(temp, payload.owner);
|
||||
mkdirSync(ownerRoot, { recursive: true });
|
||||
writeFileSync(
|
||||
join(ownerRoot, "generation"),
|
||||
String(payload.generation),
|
||||
);
|
||||
const result = spawnSync(
|
||||
input.command,
|
||||
input.args!.map((arg) =>
|
||||
arg.replaceAll("/home/user/.paperclip-owners/", `${temp}/`),
|
||||
),
|
||||
{
|
||||
input: input.stdin,
|
||||
encoding: "utf8",
|
||||
timeout: 5000,
|
||||
env: { ...process.env, PATH: `${temp}:${process.env.PATH}` },
|
||||
},
|
||||
);
|
||||
if (result.error) throw result.error;
|
||||
return {
|
||||
stdout: result.stdout,
|
||||
stderr: result.stderr,
|
||||
exitCode: result.status,
|
||||
timedOut: false,
|
||||
} as any;
|
||||
},
|
||||
});
|
||||
try {
|
||||
await f.attach();
|
||||
const binding = await f.admit();
|
||||
const result = await binding.runner.execute({
|
||||
command: "sh",
|
||||
args: [
|
||||
"-c",
|
||||
"sleep 30 </dev/null >/dev/null 2>&1 & echo $!; read word; echo $word >&2; exit 7",
|
||||
],
|
||||
cwd: temp,
|
||||
stdin: "foreground-input\n",
|
||||
});
|
||||
backgroundPid = Number(result.stdout.trim());
|
||||
expect(result.exitCode).toBe(7);
|
||||
expect(result.stderr).toBe("foreground-input\n");
|
||||
expect(backgroundPid).toBeGreaterThan(1);
|
||||
expect(() => process.kill(backgroundPid!, 0)).not.toThrow();
|
||||
expect(existsSync(join(temp, binding.owner.ownerId))).toBe(true);
|
||||
} finally {
|
||||
if (backgroundPid) process.kill(backgroundPid, "SIGKILL");
|
||||
rmSync(temp, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
it("realizes probe cwd before readiness without consuming the initial personal-home seed", async () => {
|
||||
const f = fixture();
|
||||
const temp = realpathSync(mkdtempSync(join(tmpdir(), "computer-probe-")));
|
||||
|
||||
@@ -192,26 +192,61 @@ export function createComputerService(
|
||||
args: [
|
||||
"-c",
|
||||
String.raw`
|
||||
import os,sys,json,fcntl,subprocess,time
|
||||
import os,sys,json,fcntl,subprocess,time,tempfile,shutil
|
||||
p=json.load(sys.stdin);root='/home/user/.paperclip-owners/'+p['owner'];os.makedirs(root,exist_ok=True)
|
||||
with open(root+'/lock','a') as lock:
|
||||
fcntl.flock(lock,fcntl.LOCK_EX)
|
||||
if os.path.exists(root+'/retired'):raise RuntimeError('computer owner retired')
|
||||
if p.get('processClaim'):
|
||||
claim=json.load(open(root+'/claim.json'))
|
||||
if any(claim[k]!=p['processClaim'][k] for k in ['nonce','unitName','bootId','launchGeneration']) or claim['bootId']!=open('/proc/sys/kernel/random/boot_id').read().strip():raise RuntimeError('computer process superseded')
|
||||
elif not os.path.exists(root+'/generation') or int(open(root+'/generation').read())!=p['generation']:raise RuntimeError('computer owner superseded')
|
||||
i=p['input'];args=['systemd-run','--user','--pipe','--wait','--collect','--unit='+p['unit'],'--slice=paperclip-'+p['owner']+'.slice','--property=KillMode=control-group','--working-directory='+i.get('cwd',p['cwd'])]
|
||||
for k,v in i.get('env',{}).items():args.append('--setenv='+k+'='+v)
|
||||
child=subprocess.Popen(args+['--',i['command']]+i.get('args',[]),stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)
|
||||
for n in range(200):
|
||||
state=subprocess.run(['systemctl','--user','show',p['unit'],'--property=LoadState','--value'],capture_output=True,text=True).stdout.strip()
|
||||
if state=='loaded' or child.poll() is not None:break
|
||||
time.sleep(.01)
|
||||
else:
|
||||
child.kill();raise RuntimeError('command admission unconfirmed')
|
||||
output,error=child.communicate(i.get('stdin',''))
|
||||
sys.stdout.write(output);sys.stderr.write(error);sys.exit(child.returncode)
|
||||
# A receipt reports foreground completion independently from cgroup lifetime.
|
||||
# ExitType=cgroup keeps nohup/setsid descendants owned until owner retirement.
|
||||
supervisor="""
|
||||
import os,sys,json,subprocess
|
||||
payload=sys.argv[1]
|
||||
with open(payload) as f:p=json.load(f)
|
||||
os.unlink(payload)
|
||||
i=p['input']
|
||||
try:
|
||||
with open(p['stdin'],'rb') as source:
|
||||
os.unlink(p['stdin'])
|
||||
child=subprocess.Popen([i['command']]+i.get('args',[]),stdin=source)
|
||||
code=child.wait()
|
||||
except Exception as error:
|
||||
print(str(error),file=sys.stderr);code=127
|
||||
with open(p['receipt']+'.tmp','w') as f:json.dump({'exitCode':code},f)
|
||||
os.replace(p['receipt']+'.tmp',p['receipt'])
|
||||
"""
|
||||
work=None
|
||||
try:
|
||||
with open(root+'/lock','a') as lock:
|
||||
fcntl.flock(lock,fcntl.LOCK_EX)
|
||||
if os.path.exists(root+'/retired'):raise RuntimeError('computer owner retired')
|
||||
if p.get('processClaim'):
|
||||
claim=json.load(open(root+'/claim.json'))
|
||||
if any(claim[k]!=p['processClaim'][k] for k in ['nonce','unitName','bootId','launchGeneration']) or claim['bootId']!=open('/proc/sys/kernel/random/boot_id').read().strip():raise RuntimeError('computer process superseded')
|
||||
elif not os.path.exists(root+'/generation') or int(open(root+'/generation').read())!=p['generation']:raise RuntimeError('computer owner superseded')
|
||||
work=tempfile.mkdtemp(prefix='command-',dir=root)
|
||||
output=work+'/stdout';error=work+'/stderr';receipt=work+'/receipt';payload=work+'/input.json';source=work+'/stdin'
|
||||
for path in [output,error,source]:open(path,'wb').close()
|
||||
i=p['input']
|
||||
with open(source,'w') as f:f.write(i.get('stdin',''))
|
||||
with open(payload,'w') as f:json.dump({'input':i,'stdin':source,'receipt':receipt},f)
|
||||
args=['systemd-run','--user','--quiet','--collect','--unit='+p['unit'],'--slice=paperclip-'+p['owner']+'.slice','--property=Type=exec','--property=ExitType=cgroup','--property=KillMode=control-group','--property=StandardOutput=append:'+output,'--property=StandardError=append:'+error,'--working-directory='+i.get('cwd',p['cwd'])]
|
||||
for k,v in i.get('env',{}).items():args.append('--setenv='+k+'='+v)
|
||||
started=subprocess.run(args+['--','python3','-c',supervisor,payload],capture_output=True,text=True)
|
||||
if started.returncode:sys.stderr.write(started.stderr);sys.exit(started.returncode)
|
||||
with open(output,'rb') as out,open(error,'rb') as err:
|
||||
while True:
|
||||
done=os.path.exists(receipt)
|
||||
for stream,dest in [(out,sys.stdout.buffer),(err,sys.stderr.buffer)]:
|
||||
chunk=stream.read()
|
||||
if chunk:dest.write(chunk);dest.flush()
|
||||
if done:break
|
||||
state=subprocess.run(['systemctl','--user','show',p['unit'],'--property=ActiveState','--value'],capture_output=True,text=True).stdout.strip()
|
||||
if state not in ('active','activating','reloading'):
|
||||
if os.path.exists(receipt):continue
|
||||
raise RuntimeError('computer command stopped before completion')
|
||||
time.sleep(.05)
|
||||
with open(receipt) as f:code=json.load(f)['exitCode']
|
||||
sys.exit(code if code>=0 else 128-code)
|
||||
finally:
|
||||
if work:shutil.rmtree(work,ignore_errors=True)
|
||||
`,
|
||||
],
|
||||
stdin: JSON.stringify({
|
||||
|
||||
Reference in new issue
Block a user