mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-01 15:35:12 +01:00
Add persistent computer ownership ledger and Boat backend
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed
1 parent
9525aa80e5
commit
5dcaa3b1a0
24 files changed
+60289
-3
No files matched your search
@@ -0,0 +1,385 @@
|
||||
import { generateKeyPairSync, randomUUID } from "node:crypto";
|
||||
import {
|
||||
createSshCommandManagedRuntimeRunner,
|
||||
shellQuote,
|
||||
} from "@paperclipai/adapter-utils/ssh";
|
||||
import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
|
||||
import type { ComputerBackend } from "../application/ports.js";
|
||||
import {
|
||||
ComputerError,
|
||||
segment,
|
||||
type ComputerRecord,
|
||||
type Owner,
|
||||
} from "../domain/ledger.js";
|
||||
import { remoteProgram } from "./remote-program.js";
|
||||
|
||||
const apiOrigin = "https://boat.dev/api/v1";
|
||||
const transportCache = new Map<string, Promise<CommandManagedRuntimeRunner>>();
|
||||
const desktopCache = new Map<
|
||||
string,
|
||||
{ viewerUrl: string; expiresAt: string }
|
||||
>();
|
||||
export function boatBackend(
|
||||
resolveKey: (record: ComputerRecord) => Promise<string>,
|
||||
fetcher: typeof fetch = fetch,
|
||||
): ComputerBackend {
|
||||
async function api(
|
||||
record: ComputerRecord,
|
||||
method: string,
|
||||
suffix = "",
|
||||
body?: unknown,
|
||||
): Promise<any> {
|
||||
const response = await fetcher(
|
||||
`${apiOrigin}/sandboxes/${segment(record.providerId)}${suffix}`,
|
||||
{
|
||||
method,
|
||||
headers: {
|
||||
authorization: `Bearer ${await resolveKey(record)}`,
|
||||
"content-type": "application/json",
|
||||
},
|
||||
...(body === undefined ? {} : { body: JSON.stringify(body) }),
|
||||
signal: AbortSignal.timeout(90_000),
|
||||
},
|
||||
);
|
||||
if (!response.ok)
|
||||
throw new ComputerError(
|
||||
"provider_error",
|
||||
`Boat request failed (${response.status})`,
|
||||
);
|
||||
const data = await response.json();
|
||||
if (!data || typeof data !== "object")
|
||||
throw new ComputerError("provider_error", "Invalid Boat response");
|
||||
return data;
|
||||
}
|
||||
async function inspect(record: ComputerRecord) {
|
||||
const data = await api(record, "GET");
|
||||
const sandbox = data.sandbox ?? data;
|
||||
if (sandbox.id !== record.providerId || typeof sandbox.state !== "string")
|
||||
throw new ComputerError(
|
||||
"provider_error",
|
||||
"Boat returned a different computer",
|
||||
);
|
||||
return {
|
||||
state: sandbox.state as string,
|
||||
snapshots: sandbox.snapshots === true,
|
||||
stop: sandbox.stop ?? null,
|
||||
};
|
||||
}
|
||||
async function runner(
|
||||
record: ComputerRecord,
|
||||
): Promise<CommandManagedRuntimeRunner> {
|
||||
const key = `${record.id}:${record.ledger.secretRef.secretId}:${record.ledger.secretRef.version ?? "latest"}`;
|
||||
let pending = transportCache.get(key);
|
||||
if (!pending) {
|
||||
pending = (async () => {
|
||||
const pair = generateKeyPairSync("ed25519");
|
||||
const jwk = pair.publicKey.export({ format: "jwk" });
|
||||
const publicBytes = Buffer.from(jwk.x!, "base64url");
|
||||
const name = Buffer.from("ssh-ed25519");
|
||||
const length = (n: number) => {
|
||||
const b = Buffer.alloc(4);
|
||||
b.writeUInt32BE(n);
|
||||
return b;
|
||||
};
|
||||
const publicKey = `ssh-ed25519 ${Buffer.concat([length(name.length), name, length(publicBytes.length), publicBytes]).toString("base64")} paperclip-computer`;
|
||||
// OpenSSH cannot read PKCS8 Ed25519 keys; emit its unencrypted private format in memory.
|
||||
const seed = Buffer.from(
|
||||
pair.privateKey.export({ format: "jwk" }).d!,
|
||||
"base64url",
|
||||
);
|
||||
const field = (b: Buffer) => Buffer.concat([length(b.length), b]);
|
||||
const check = Buffer.from("70617065", "hex");
|
||||
let privateBody = Buffer.concat([
|
||||
check,
|
||||
check,
|
||||
field(name),
|
||||
field(publicBytes),
|
||||
field(Buffer.concat([seed, publicBytes])),
|
||||
field(Buffer.from("paperclip-computer")),
|
||||
]);
|
||||
const padding = 8 - (privateBody.length % 8);
|
||||
privateBody = Buffer.concat([
|
||||
privateBody,
|
||||
Buffer.from(Array.from({ length: padding }, (_, i) => i + 1)),
|
||||
]);
|
||||
const publicBlob = Buffer.concat([field(name), field(publicBytes)]);
|
||||
const blob = Buffer.concat([
|
||||
Buffer.from("openssh-key-v1\0"),
|
||||
field(Buffer.from("none")),
|
||||
field(Buffer.from("none")),
|
||||
field(Buffer.alloc(0)),
|
||||
length(1),
|
||||
field(publicBlob),
|
||||
field(privateBody),
|
||||
]);
|
||||
const privateKey = `-----BEGIN OPENSSH PRIVATE KEY-----\n${blob
|
||||
.toString("base64")
|
||||
.match(/.{1,70}/g)!
|
||||
.join("\n")}\n-----END OPENSSH PRIVATE KEY-----\n`;
|
||||
const result = await api(record, "POST", "/sshkey", { key: publicKey });
|
||||
if (
|
||||
typeof result.hostKey !== "string" ||
|
||||
!/^ssh-ed25519 [A-Za-z0-9+/=]+$/.test(result.hostKey.trim())
|
||||
)
|
||||
throw new ComputerError(
|
||||
"provider_error",
|
||||
"Boat did not provide a valid SSH host key",
|
||||
);
|
||||
const endpoint =
|
||||
typeof result.sshEndpoint === "string" ? result.sshEndpoint : null;
|
||||
const parsed = endpoint?.match(/^([A-Za-z0-9.-]+):(\d+)$/);
|
||||
const host = parsed?.[1] ?? result.machineIp;
|
||||
const port = parsed ? Number(parsed[2]) : 22;
|
||||
if (
|
||||
typeof host !== "string" ||
|
||||
!host ||
|
||||
/[\s/]/.test(host) ||
|
||||
!Number.isInteger(port) ||
|
||||
port < 1 ||
|
||||
port > 65535
|
||||
)
|
||||
throw new ComputerError(
|
||||
"provider_error",
|
||||
"Invalid Boat SSH endpoint",
|
||||
);
|
||||
const knownHost = port === 22 ? host : `[${host}]:${port}`;
|
||||
return createSshCommandManagedRuntimeRunner({
|
||||
spec: {
|
||||
host,
|
||||
port,
|
||||
username: "user",
|
||||
remoteWorkspacePath: "/home/user",
|
||||
remoteCwd: "/home/user",
|
||||
privateKey,
|
||||
knownHosts: `${knownHost} ${result.hostKey.trim()}\n`,
|
||||
strictHostKeyChecking: true,
|
||||
},
|
||||
maxBufferBytes: 24 * 1024 * 1024,
|
||||
});
|
||||
})();
|
||||
transportCache.set(key, pending);
|
||||
pending.catch(() => transportCache.delete(key));
|
||||
}
|
||||
return pending;
|
||||
}
|
||||
async function execute(
|
||||
record: ComputerRecord,
|
||||
code: string,
|
||||
payload: unknown,
|
||||
) {
|
||||
const result = await (
|
||||
await runner(record)
|
||||
).execute({
|
||||
command: "python3",
|
||||
args: ["-c", code],
|
||||
stdin: JSON.stringify(payload),
|
||||
timeoutMs: 120_000,
|
||||
});
|
||||
if (result.exitCode !== 0)
|
||||
throw new ComputerError("provider_error", "Computer operation failed");
|
||||
let value: any;
|
||||
try {
|
||||
value = JSON.parse(result.stdout);
|
||||
} catch {
|
||||
throw new ComputerError(
|
||||
"provider_error",
|
||||
"Invalid computer operation response",
|
||||
);
|
||||
}
|
||||
if (value?.error)
|
||||
throw new ComputerError(
|
||||
["not_found", "conflict", "invalid"].includes(value.error)
|
||||
? value.error
|
||||
: "provider_error",
|
||||
`Computer operation: ${value.error}`,
|
||||
);
|
||||
return value;
|
||||
}
|
||||
const processProgram = String.raw`
|
||||
import os,sys,json,fcntl,subprocess
|
||||
p=json.load(sys.stdin);base='/home/user/.paperclip-owners';os.makedirs(base,exist_ok=True)
|
||||
owner=p['owner'];oid=owner['id'];root=os.path.join(base,oid);os.makedirs(root,exist_ok=True)
|
||||
with open(os.path.join(root,'lock'),'a') as lock:
|
||||
fcntl.flock(lock,fcntl.LOCK_EX)
|
||||
tombstone=os.path.join(root,'retired');unit='paperclip-'+oid+'.service';slice='paperclip-'+oid+'.slice'
|
||||
boot=open('/proc/sys/kernel/random/boot_id').read().strip()
|
||||
generation_path=os.path.join(root,'generation')
|
||||
generation=int(open(generation_path).read()) if os.path.exists(generation_path) else 0
|
||||
if owner['generation']<generation:print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
if p['action']=='advance':
|
||||
if os.path.exists(tombstone):print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
with open(generation_path,'w') as f:f.write(str(owner['generation']))
|
||||
print('{}')
|
||||
elif p['action']=='retire':
|
||||
open(tombstone,'a').close()
|
||||
subprocess.run(['systemctl','--user','stop',slice,unit],capture_output=True)
|
||||
state=subprocess.run(['systemctl','--user','show',slice,'--property=ActiveState','--value'],capture_output=True,text=True).stdout.strip()
|
||||
if state not in ('inactive','failed',''):raise RuntimeError('process retirement unconfirmed')
|
||||
print('{}')
|
||||
elif p['action']=='inspect':
|
||||
claim=owner.get('process');state=subprocess.run(['systemctl','--user','is-active',unit],capture_output=True,text=True).stdout.strip()
|
||||
print(json.dumps({'running':bool(claim and claim['bootId']==boot and state=='active' and not os.path.exists(tombstone)),'claim':claim}))
|
||||
else:
|
||||
if os.path.exists(tombstone):print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
claim=owner['process'];claim['bootId']=boot
|
||||
marker=os.path.join(root,'claim.json')
|
||||
if os.path.exists(marker):
|
||||
old=json.load(open(marker))
|
||||
if old['nonce']!=claim['nonce']:print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
with open(marker,'w') as f:json.dump(claim,f)
|
||||
payload=p['input'];args=['systemd-run','--user','--unit='+unit,'--slice='+slice,'--collect','--property=KillMode=control-group','--working-directory='+payload.get('cwd','/home/user')]
|
||||
for key,value in payload.get('env',{}).items():args.append('--setenv='+key+'='+value)
|
||||
state=subprocess.run(['systemctl','--user','is-active',unit],capture_output=True,text=True).stdout.strip()
|
||||
if state!='active':
|
||||
result=subprocess.run(args+['--',payload['command']]+payload.get('args',[]),capture_output=True)
|
||||
if result.returncode:raise RuntimeError('launch failed')
|
||||
print(json.dumps(claim))
|
||||
`;
|
||||
async function host(record: ComputerRecord, port: number) {
|
||||
if (!Number.isInteger(port) || port < 1024 || port > 65535)
|
||||
throw new ComputerError("invalid", "Invalid preview port");
|
||||
const result = await api(record, "POST", "/host", {
|
||||
port,
|
||||
access: "private",
|
||||
});
|
||||
const url = new URL(result.url);
|
||||
if (
|
||||
url.protocol !== "https:" ||
|
||||
!url.hostname.endsWith(".on.boat.dev") ||
|
||||
url.username ||
|
||||
url.password ||
|
||||
!url.searchParams.get("_token") ||
|
||||
result.access === "public"
|
||||
)
|
||||
throw new ComputerError(
|
||||
"provider_error",
|
||||
"Boat did not provide private hosting",
|
||||
);
|
||||
return url;
|
||||
}
|
||||
return {
|
||||
inspect,
|
||||
runner,
|
||||
async ready(record) {
|
||||
const state = await inspect(record);
|
||||
if (!state.snapshots)
|
||||
throw new ComputerError("invalid", "Boat snapshots must be enabled");
|
||||
if (state.stop && ["pending", "failing"].includes(state.stop.status))
|
||||
throw new ComputerError(
|
||||
"conflict",
|
||||
"Boat is still saving a previous stop",
|
||||
);
|
||||
if (!["ready", "idle", "running"].includes(state.state)) {
|
||||
await api(record, "POST", "/resume", { ttlSeconds: 300 });
|
||||
transportCache.delete(
|
||||
`${record.id}:${record.ledger.secretRef.secretId}:${record.ledger.secretRef.version ?? "latest"}`,
|
||||
);
|
||||
for (let i = 0; i < 90; i++) {
|
||||
const status = await inspect(record);
|
||||
if (["ready", "idle", "running"].includes(status.state)) return;
|
||||
await new Promise((r) => setTimeout(r, 1000));
|
||||
}
|
||||
throw new ComputerError("provider_error", "Boat did not become ready");
|
||||
}
|
||||
},
|
||||
async claim(record) {
|
||||
await execute(
|
||||
record,
|
||||
String.raw`
|
||||
import os,sys,json,fcntl
|
||||
p=json.load(sys.stdin);path='/home/user/.paperclip-controller.json'
|
||||
fd=os.open(path,os.O_CREAT|os.O_RDWR|os.O_NOFOLLOW,0o600)
|
||||
with os.fdopen(fd,'r+') as f:
|
||||
fcntl.flock(f,fcntl.LOCK_EX);old=f.read()
|
||||
if old and json.loads(old)!=p:print(json.dumps({'error':'conflict'}));sys.exit(0)
|
||||
if not old:f.write(json.dumps(p));f.flush();os.fsync(f.fileno())
|
||||
print('{}')
|
||||
`,
|
||||
{
|
||||
controllerId: record.ledger.controllerId,
|
||||
companyId: record.companyId,
|
||||
computerId: record.id,
|
||||
},
|
||||
);
|
||||
},
|
||||
async advance(record, owner) {
|
||||
await execute(record, processProgram, { action: "advance", owner });
|
||||
},
|
||||
async launch(record, owner, input) {
|
||||
return execute(record, processProgram, {
|
||||
action: "launch",
|
||||
owner,
|
||||
input,
|
||||
});
|
||||
},
|
||||
async inspectProcess(record, owner) {
|
||||
return execute(record, processProgram, { action: "inspect", owner });
|
||||
},
|
||||
async retire(record, owner) {
|
||||
await execute(record, processProgram, { action: "retire", owner });
|
||||
},
|
||||
async stop(record) {
|
||||
const value = await api(record, "POST", "/stop", {});
|
||||
const stop = value.stop ?? value.sandbox?.stop;
|
||||
if (!stop?.id)
|
||||
throw new ComputerError("provider_error", "Boat stop receipt missing");
|
||||
return stop;
|
||||
},
|
||||
async stopStatus(record, id) {
|
||||
const value = await api(record, "GET", `/stops/${segment(id)}`);
|
||||
return value.stop ?? value;
|
||||
},
|
||||
async renew(record) {
|
||||
await api(record, "PATCH", "", { ttlSeconds: 300 });
|
||||
},
|
||||
async desktop(record) {
|
||||
const cached = desktopCache.get(record.id);
|
||||
if (cached && Date.parse(cached.expiresAt) > Date.now() + 60_000)
|
||||
return cached;
|
||||
const value = await api(record, "POST", "/desktop", {});
|
||||
const url = new URL(value.desktopUrl);
|
||||
if (url.protocol !== "https:" || !url.hostname.endsWith(".on.boat.dev"))
|
||||
throw new ComputerError("provider_error", "Invalid Boat desktop URL");
|
||||
url.searchParams.set("overlay", "0");
|
||||
const result = {
|
||||
viewerUrl: url.toString(),
|
||||
expiresAt: new Date(Date.now() + 9 * 60_000).toISOString(),
|
||||
};
|
||||
desktopCache.set(record.id, result);
|
||||
return result;
|
||||
},
|
||||
async preview(record, port) {
|
||||
return { url: (await host(record, port)).toString() };
|
||||
},
|
||||
async ingress(record, port, path) {
|
||||
if (
|
||||
!path.startsWith("/") ||
|
||||
path.startsWith("//") ||
|
||||
/[?#\r\n]/.test(path)
|
||||
)
|
||||
throw new ComputerError("invalid", "Invalid runner ingress path");
|
||||
const url = await host(record, port);
|
||||
const response = await fetcher(url, {
|
||||
redirect: "manual",
|
||||
signal: AbortSignal.timeout(30_000),
|
||||
});
|
||||
const cookie = response.headers
|
||||
.getSetCookie()
|
||||
.find((value) => value.startsWith("_port_auth="))
|
||||
?.split(";")[0];
|
||||
if (!cookie || response.status < 200 || response.status >= 400)
|
||||
throw new ComputerError(
|
||||
"provider_error",
|
||||
"Boat private hosting authentication failed",
|
||||
);
|
||||
url.search = "";
|
||||
url.hash = "";
|
||||
url.protocol = "wss:";
|
||||
url.pathname = path;
|
||||
return { url: url.toString(), secretHeaders: { Cookie: cookie } };
|
||||
},
|
||||
async remote(record, input) {
|
||||
return execute(record, remoteProgram, input);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
import { eq } from "drizzle-orm";
|
||||
import { computers } from "@paperclipai/db";
|
||||
import type { CompanyDeletionParticipant } from "../../../lib/company-deletion.js";
|
||||
import { ComputerError, liveOwners, type Ledger } from "../domain/ledger.js";
|
||||
|
||||
export const computerCompanyDeletion: CompanyDeletionParticipant = {
|
||||
async deleteCompanyData(tx, companyId) {
|
||||
const records = await tx.select().from(computers).where(eq(computers.companyId, companyId)).for("update");
|
||||
for (const record of records) {
|
||||
const ledger = record.ledger as unknown as Ledger;
|
||||
if (ledger.status !== "detached" || ledger.action || liveOwners(ledger).length) {
|
||||
throw new ComputerError("conflict", "Detach company computers and wait for their snapshots before deleting the company");
|
||||
}
|
||||
}
|
||||
await tx.delete(computers).where(eq(computers.companyId, companyId));
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,15 @@
|
||||
import {afterEach,describe,expect,it} from 'vitest';
|
||||
import {mkdtempSync,realpathSync,rmSync,symlinkSync,writeFileSync,mkdirSync} from 'node:fs';
|
||||
import {tmpdir} from 'node:os';
|
||||
import {join} from 'node:path';
|
||||
import {spawnSync} from 'node:child_process';
|
||||
import {remoteProgram} from './remote-program.js';
|
||||
const roots:string[]=[];
|
||||
function fixture(){const temp=realpathSync(mkdtempSync(join(tmpdir(),'computer-files-')));roots.push(temp);const root=join(temp,'home');const call=(input:Record<string,unknown>)=>{const result=spawnSync('python3',['-c',remoteProgram.replaceAll('/home/user/paperclip/',`${temp}/`)],{input:JSON.stringify({...input,root}),encoding:'utf8'});if(result.status!==0)throw new Error(result.stderr);return JSON.parse(result.stdout);};return {root,temp,call};}
|
||||
afterEach(()=>{for(const root of roots.splice(0))rmSync(root,{recursive:true,force:true});});
|
||||
describe('confined computer files',()=>{
|
||||
it('seeds absent home only and preserves existing content',()=>{const f=fixture();expect(f.call({action:'seed',files:{'nested/one':Buffer.from('first').toString('base64')}})).toEqual({seeded:true});expect(f.call({action:'seed',files:{'nested/one':Buffer.from('overwrite').toString('base64')}})).toEqual({seeded:false});expect(Buffer.from(f.call({action:'read',path:'nested/one'}).base64,'base64').toString()).toBe('first');});
|
||||
it('rejects traversal, symlink escapes, and oversized reads',()=>{const f=fixture();f.call({action:'seed',files:{'ok':Buffer.from('12345').toString('base64')}});mkdirSync(join(f.temp,'outside'));writeFileSync(join(f.temp,'outside','secret'),'private');symlinkSync(join(f.temp,'outside'),join(f.root,'escape'));expect(f.call({action:'read',path:'../outside/secret'})).toEqual({error:'invalid'});expect(f.call({action:'read',path:'escape/secret'})).toEqual({error:'invalid'});expect(f.call({action:'read',path:'ok',maxBytes:4})).toEqual({error:'invalid'});});
|
||||
it('rejects stale edits and uses no-clobber move semantics',()=>{const f=fixture();f.call({action:'seed',files:{}});const first=f.call({action:'write',path:'file',expectedSha256:null,base64:Buffer.from('one').toString('base64')});const second=f.call({action:'write',path:'file',expectedSha256:first.sha256,base64:Buffer.from('two').toString('base64')});expect(f.call({action:'write',path:'file',expectedSha256:first.sha256,base64:''})).toEqual({error:'conflict'});expect(f.call({action:'move',path:'file',to:'other',expectedSha256:second.sha256})).toEqual({sha256:second.sha256});expect(f.call({action:'list'}).map((e:any)=>e.name)).toEqual(['other']);expect(f.call({action:'remove',path:'other',expectedSha256:first.sha256})).toEqual({error:'conflict'});});
|
||||
it('roundtrips binary bytes without interpreting text',()=>{const f=fixture();const bytes=Buffer.from([0,255,254,13,10,128]);f.call({action:'seed',files:{binary:bytes.toString('base64')}});expect(Buffer.from(f.call({action:'read',path:'binary'}).base64,'base64')).toEqual(bytes);});
|
||||
});
|
||||
@@ -0,0 +1,142 @@
|
||||
/** Fixed remote program. Inputs travel as JSON, never interpolated into Python or shell. */
|
||||
export const remoteProgram = String.raw`
|
||||
import os,sys,json,hashlib,base64,tempfile,subprocess,fcntl,stat,shutil
|
||||
p=json.load(sys.stdin)
|
||||
def fail(code):
|
||||
print(json.dumps({'error':code}));sys.exit(0)
|
||||
def digest(data):return hashlib.sha256(data).hexdigest()
|
||||
def run(args,cwd=None):
|
||||
r=subprocess.run(args,cwd=cwd,capture_output=True,text=True)
|
||||
if r.returncode:fail('command_failed')
|
||||
return r.stdout.strip()
|
||||
root=p['root']
|
||||
if not root.startswith('/home/user/paperclip/') or '..' in root.split('/'):fail('invalid')
|
||||
# Reject symlinks even in existing ancestors; one Unix user is not an OS boundary.
|
||||
def safe(path):
|
||||
if not isinstance(path,str) or path.startswith('/') or '\x00' in path or '..' in path.split('/'):fail('invalid')
|
||||
current=root
|
||||
for part in path.split('/'):
|
||||
if part in ('','.'):continue
|
||||
current=os.path.join(current,part)
|
||||
if os.path.islink(current):fail('invalid')
|
||||
if os.path.commonpath([root,os.path.realpath(current)])!=root:fail('invalid')
|
||||
return current
|
||||
current='/'
|
||||
for part in root.split('/'):
|
||||
if not part:continue
|
||||
current=os.path.join(current,part)
|
||||
if os.path.islink(current):fail('invalid')
|
||||
act=p['action']
|
||||
if act=='owned-port':
|
||||
port=p['port'];owner=p['ownerId'];inodes=set()
|
||||
for table in ['/proc/net/tcp','/proc/net/tcp6']:
|
||||
for line in open(table).readlines()[1:]:
|
||||
cols=line.split()
|
||||
if int(cols[1].split(':')[1],16)==port and cols[3]=='0A':inodes.add(cols[9])
|
||||
for pid in os.listdir('/proc'):
|
||||
if not pid.isdigit():continue
|
||||
try:
|
||||
if 'paperclip-'+owner+'.slice' not in open('/proc/'+pid+'/cgroup').read():continue
|
||||
for fd in os.listdir('/proc/'+pid+'/fd'):
|
||||
target=os.readlink('/proc/'+pid+'/fd/'+fd)
|
||||
if target.startswith('socket:[') and target[8:-1] in inodes:print('{}');sys.exit(0)
|
||||
except (FileNotFoundError,PermissionError,ProcessLookupError):pass
|
||||
fail('conflict')
|
||||
if act=='seed':
|
||||
if os.path.exists(root):print(json.dumps({'seeded':False}));sys.exit(0)
|
||||
os.makedirs(os.path.dirname(root),exist_ok=True)
|
||||
temp=tempfile.mkdtemp(prefix='.seed-',dir=os.path.dirname(root))
|
||||
try:
|
||||
for name,value in p['files'].items():
|
||||
dest=safe(name);relative=os.path.relpath(dest,root)
|
||||
if relative=='.':fail('invalid')
|
||||
target=os.path.join(temp,relative);os.makedirs(os.path.dirname(target),exist_ok=True)
|
||||
with open(target,'wb') as f:f.write(base64.b64decode(value,validate=True))
|
||||
try:os.rename(temp,root);result=True
|
||||
except FileExistsError:result=False
|
||||
print(json.dumps({'seeded':result}))
|
||||
finally:
|
||||
if os.path.exists(temp):shutil.rmtree(temp)
|
||||
sys.exit(0)
|
||||
if act=='workspace':
|
||||
os.makedirs(root,exist_ok=True)
|
||||
repo=p.get('repositoryUrl');branch=p.get('branch');task=p.get('taskId');base_ref=p.get('baseRef')
|
||||
if base_ref and (base_ref.startswith('-') or '\n' in base_ref):fail('invalid')
|
||||
checkout=os.path.join(root,'checkout')
|
||||
if repo:
|
||||
if not (repo.startswith('https://') or repo.startswith('ssh://') or repo.startswith('git@')) or '\n' in repo:fail('invalid')
|
||||
if not os.path.exists(checkout):run(['git','clone','--',repo,checkout])
|
||||
elif run(['git','remote','get-url','origin'],checkout)!=repo:fail('conflict')
|
||||
else:
|
||||
os.makedirs(checkout,exist_ok=True)
|
||||
if not os.path.isdir(os.path.join(checkout,'.git')):run(['git','init',checkout])
|
||||
if branch:
|
||||
if branch.startswith('-') or '\n' in branch:fail('invalid')
|
||||
run(['git','check-ref-format','--branch',branch])
|
||||
if p.get('mode')=='worktree':
|
||||
if not task:fail('invalid')
|
||||
target=os.path.join(root,'tasks',task);os.makedirs(os.path.dirname(target),exist_ok=True)
|
||||
if not os.path.exists(target):run(['git','worktree','add','-b','paperclip/'+task,target,base_ref or branch or 'HEAD'],checkout)
|
||||
run(['git','rev-parse','--show-toplevel'],target)
|
||||
else:
|
||||
target=checkout
|
||||
if branch:run(['git','checkout',branch],checkout)
|
||||
print(json.dumps({'remoteCwd':target}));sys.exit(0)
|
||||
if not os.path.isdir(root):fail('not_found')
|
||||
path=safe(p.get('path',''))
|
||||
lock=os.open(os.path.join(root,'.paperclip-editor.lock'),os.O_CREAT|os.O_RDWR|os.O_NOFOLLOW,0o600)
|
||||
fcntl.flock(lock,fcntl.LOCK_EX)
|
||||
try:
|
||||
if act=='list':
|
||||
out=[]
|
||||
for entry in os.scandir(path):
|
||||
if entry.name=='.paperclip-editor.lock' or entry.is_symlink():continue
|
||||
if not entry.is_dir(follow_symlinks=False) and not entry.is_file(follow_symlinks=False):continue
|
||||
out.append({'name':entry.name,'kind':'directory' if entry.is_dir(follow_symlinks=False) else 'file','size':entry.stat(follow_symlinks=False).st_size,'mtimeMs':entry.stat(follow_symlinks=False).st_mtime*1000})
|
||||
print(json.dumps(out))
|
||||
elif act=='read':
|
||||
fd=os.open(path,os.O_RDONLY|os.O_NOFOLLOW)
|
||||
with os.fdopen(fd,'rb') as f:
|
||||
st=os.fstat(f.fileno())
|
||||
if not stat.S_ISREG(st.st_mode) or st.st_size>p.get('maxBytes',16*1024*1024):fail('invalid')
|
||||
data=f.read(p.get('maxBytes',16*1024*1024)+1)
|
||||
if len(data)>p.get('maxBytes',16*1024*1024):fail('invalid')
|
||||
print(json.dumps({'base64':base64.b64encode(data).decode(),'sha256':digest(data)}))
|
||||
elif act in ('write','remove','move'):
|
||||
if path==root:fail('invalid')
|
||||
exists=os.path.exists(path)
|
||||
old=None
|
||||
if exists:
|
||||
fd=os.open(path,os.O_RDONLY|os.O_NOFOLLOW)
|
||||
with os.fdopen(fd,'rb') as f:
|
||||
if not stat.S_ISREG(os.fstat(f.fileno()).st_mode):fail('invalid')
|
||||
h=hashlib.sha256()
|
||||
while True:
|
||||
block=f.read(65536)
|
||||
if not block:break
|
||||
h.update(block)
|
||||
old=h.hexdigest()
|
||||
if old!=p.get('expectedSha256'):fail('conflict')
|
||||
if act=='write':
|
||||
data=base64.b64decode(p['base64'],validate=True)
|
||||
if len(data)>p.get('maxBytes',16*1024*1024):fail('invalid')
|
||||
os.makedirs(os.path.dirname(path),exist_ok=True)
|
||||
fd,temp=tempfile.mkstemp(prefix='.paperclip-write-',dir=os.path.dirname(path))
|
||||
try:
|
||||
with os.fdopen(fd,'wb') as f:f.write(data);f.flush();os.fsync(f.fileno())
|
||||
os.replace(temp,path)
|
||||
finally:
|
||||
if os.path.exists(temp):os.unlink(temp)
|
||||
print(json.dumps({'sha256':digest(data)}))
|
||||
elif act=='remove':
|
||||
if not exists:fail('not_found')
|
||||
os.unlink(path);print('{}')
|
||||
else:
|
||||
target=safe(p['to'])
|
||||
if os.path.exists(target):fail('conflict')
|
||||
os.makedirs(os.path.dirname(target),exist_ok=True)
|
||||
os.link(path,target,follow_symlinks=False);os.unlink(path);print(json.dumps({'sha256':old}))
|
||||
else:fail('invalid')
|
||||
except FileNotFoundError:fail('not_found')
|
||||
finally:os.close(lock)
|
||||
`;
|
||||
@@ -0,0 +1,64 @@
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { computers, companies, type Db } from "@paperclipai/db";
|
||||
import {
|
||||
ComputerError,
|
||||
type ComputerRecord,
|
||||
type Ledger,
|
||||
} from "../domain/ledger.js";
|
||||
import type { ComputerRepository } from "../application/ports.js";
|
||||
export function computerRepository(db: Db): ComputerRepository {
|
||||
const decode = (row: typeof computers.$inferSelect): ComputerRecord => ({
|
||||
...row,
|
||||
ledger: row.ledger as unknown as Ledger,
|
||||
});
|
||||
const where = (scope: { companyId: string; environmentId: string }) =>
|
||||
and(
|
||||
eq(computers.companyId, scope.companyId),
|
||||
eq(computers.environmentId, scope.environmentId),
|
||||
);
|
||||
return {
|
||||
async create(record) {
|
||||
await db.transaction(async tx => {
|
||||
const [company] = await tx.select({id:companies.id}).from(companies).where(eq(companies.id,record.companyId)).for("update");
|
||||
if (!company) throw new ComputerError("not_found", "Company not found");
|
||||
await tx.insert(computers).values({...record,ledger:record.ledger as unknown as Record<string,unknown>});
|
||||
});
|
||||
},
|
||||
async get(scope) {
|
||||
const [row] = await db.select().from(computers).where(where(scope));
|
||||
if (!row)
|
||||
throw new ComputerError(
|
||||
"not_found",
|
||||
"Computer is not attached to this company",
|
||||
);
|
||||
return decode(row);
|
||||
},
|
||||
async update(scope, fn) {
|
||||
return db.transaction(async (tx) => {
|
||||
const [row] = await tx
|
||||
.select()
|
||||
.from(computers)
|
||||
.where(where(scope))
|
||||
.for("update");
|
||||
if (!row)
|
||||
throw new ComputerError(
|
||||
"not_found",
|
||||
"Computer is not attached to this company",
|
||||
);
|
||||
const record = decode(row);
|
||||
const result = fn(record);
|
||||
await tx
|
||||
.update(computers)
|
||||
.set({
|
||||
ledger: record.ledger as unknown as Record<string, unknown>,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(computers.id, row.id));
|
||||
return result;
|
||||
});
|
||||
},
|
||||
async all() {
|
||||
return (await db.select().from(computers)).map(decode);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
|
||||
import type {
|
||||
ComputerRecord,
|
||||
Ledger,
|
||||
Owner,
|
||||
ProcessClaim,
|
||||
} from "../domain/ledger.js";
|
||||
export interface ComputerRepository {
|
||||
create(record: ComputerRecord): Promise<void>;
|
||||
get(scope: {
|
||||
companyId: string;
|
||||
environmentId: string;
|
||||
}): Promise<ComputerRecord>;
|
||||
update<T>(
|
||||
scope: { companyId: string; environmentId: string },
|
||||
fn: (record: ComputerRecord) => T,
|
||||
): Promise<T>;
|
||||
all(): Promise<ComputerRecord[]>;
|
||||
}
|
||||
export interface ComputerBackend {
|
||||
inspect(
|
||||
record: ComputerRecord,
|
||||
): Promise<{
|
||||
state: string;
|
||||
snapshots: boolean;
|
||||
stop: null | { id: string; status: string };
|
||||
}>;
|
||||
ready(record: ComputerRecord): Promise<void>;
|
||||
claim(record: ComputerRecord): Promise<void>;
|
||||
advance(record: ComputerRecord, owner: Owner): Promise<void>;
|
||||
runner(record: ComputerRecord): Promise<CommandManagedRuntimeRunner>;
|
||||
launch(
|
||||
record: ComputerRecord,
|
||||
owner: Owner,
|
||||
input: {
|
||||
command: string;
|
||||
args?: string[];
|
||||
cwd?: string;
|
||||
env?: Record<string, string>;
|
||||
},
|
||||
): Promise<ProcessClaim>;
|
||||
inspectProcess(
|
||||
record: ComputerRecord,
|
||||
owner: Owner,
|
||||
): Promise<{ running: boolean; claim: ProcessClaim | null }>;
|
||||
retire(record: ComputerRecord, owner: Owner): Promise<void>;
|
||||
stop(record: ComputerRecord): Promise<{ id: string; status: string }>;
|
||||
stopStatus(
|
||||
record: ComputerRecord,
|
||||
id: string,
|
||||
): Promise<{ id: string; status: string }>;
|
||||
renew(record: ComputerRecord): Promise<void>;
|
||||
desktop(
|
||||
record: ComputerRecord,
|
||||
): Promise<{ viewerUrl: string; expiresAt: string }>;
|
||||
ingress(
|
||||
record: ComputerRecord,
|
||||
port: number,
|
||||
path: string,
|
||||
): Promise<{ url: string; secretHeaders: Record<string, string> }>;
|
||||
preview(record: ComputerRecord, port: number): Promise<{ url: string }>;
|
||||
remote(record: ComputerRecord, input: Record<string, unknown>): Promise<any>;
|
||||
}
|
||||
export type { Ledger };
|
||||
@@ -0,0 +1,217 @@
|
||||
import { describe, it, expect, vi } from "vitest";
|
||||
import { createComputerService } from "./service.js";
|
||||
import type { ComputerBackend, ComputerRepository } from "./ports.js";
|
||||
import { ComputerError, type ComputerRecord } from "../domain/ledger.js";
|
||||
|
||||
function fixture() {
|
||||
let record: ComputerRecord | undefined;
|
||||
let clock = new Date("2026-10-10T12:00:00Z");
|
||||
let state = "ready";
|
||||
let stopStatus = "pending";
|
||||
const repository: ComputerRepository = {
|
||||
create: async (value) => {
|
||||
record = structuredClone(value);
|
||||
},
|
||||
get: async (scope) => {
|
||||
if (
|
||||
!record ||
|
||||
record.companyId !== scope.companyId ||
|
||||
record.environmentId !== scope.environmentId
|
||||
)
|
||||
throw new ComputerError("not_found", "missing");
|
||||
return structuredClone(record);
|
||||
},
|
||||
update: async (scope, fn) => {
|
||||
const next = await repository.get(scope);
|
||||
const result = fn(next);
|
||||
record = next;
|
||||
return result;
|
||||
},
|
||||
all: async () => (record ? [structuredClone(record)] : []),
|
||||
};
|
||||
const backend: ComputerBackend = {
|
||||
inspect: vi.fn(async () => ({ state, snapshots: true, stop: null })),
|
||||
ready: vi.fn(async () => {
|
||||
state = "ready";
|
||||
}),
|
||||
claim: vi.fn(async () => {}),
|
||||
advance: vi.fn(async () => {}),
|
||||
runner: vi.fn(async () => ({ execute: vi.fn() })),
|
||||
launch: vi.fn(async (_record, owner) => ({
|
||||
...owner.process!,
|
||||
bootId: "boot",
|
||||
})),
|
||||
inspectProcess: vi.fn(async (_record, owner) => ({
|
||||
running: true,
|
||||
claim: owner.process,
|
||||
})),
|
||||
retire: vi.fn(async () => {}),
|
||||
stop: vi.fn(async () => ({ id: "stop_1", status: stopStatus })),
|
||||
stopStatus: vi.fn(async () => ({ id: "stop_1", status: stopStatus })),
|
||||
renew: vi.fn(async () => {}),
|
||||
desktop: vi.fn(async () => ({
|
||||
viewerUrl: "https://test.on.boat.dev/#credential",
|
||||
expiresAt: new Date(clock.getTime() + 540_000).toISOString(),
|
||||
})),
|
||||
ingress: vi.fn(async () => ({
|
||||
url: "wss://test.on.boat.dev/",
|
||||
secretHeaders: { Cookie: "private" },
|
||||
})),
|
||||
preview: vi.fn(async () => ({
|
||||
url: "https://test.on.boat.dev/?_token=private",
|
||||
})),
|
||||
remote: vi.fn(async () => ({
|
||||
remoteCwd: "/home/user/paperclip/company/projects/project/checkout",
|
||||
})),
|
||||
};
|
||||
const service = createComputerService(repository, backend, () => clock);
|
||||
const scope = { companyId: "company", environmentId: "environment" };
|
||||
const attach = () =>
|
||||
service.attach({
|
||||
...scope,
|
||||
sandboxId: "bx_fixture",
|
||||
apiKeySecretRef: { type: "secret_ref", secretId: "secret" },
|
||||
});
|
||||
const admit = (agentId = "agent", sessionKey = "session") =>
|
||||
service.admit({
|
||||
...scope,
|
||||
agentId,
|
||||
sessionKey,
|
||||
runId: "run",
|
||||
idleTimeoutMs: 60_000,
|
||||
});
|
||||
return {
|
||||
service,
|
||||
backend,
|
||||
scope,
|
||||
attach,
|
||||
admit,
|
||||
repository,
|
||||
advance: (ms: number) => {
|
||||
clock = new Date(clock.getTime() + ms);
|
||||
},
|
||||
completeStop: () => {
|
||||
stopStatus = "completed";
|
||||
state = "stopped";
|
||||
},
|
||||
};
|
||||
}
|
||||
describe("computer ownership", () => {
|
||||
it("reuses warm process identity and port while fencing stale callbacks", async () => {
|
||||
const f = fixture();
|
||||
await f.attach();
|
||||
const first = await f.admit();
|
||||
const claim = await first.launch({ command: "runnerd" });
|
||||
await f.service.retainWarm({
|
||||
...f.scope,
|
||||
owner: first.owner,
|
||||
idleTimeoutMs: 60_000,
|
||||
});
|
||||
const next = await f.admit();
|
||||
expect(next.owner.generation).toBe(first.owner.generation + 1);
|
||||
expect(next.listenerPort).toBe(first.listenerPort);
|
||||
expect((await next.inspectProcess()).claim?.nonce).toBe(claim.nonce);
|
||||
await f.service.retire({ ...f.scope, owner: first.owner });
|
||||
expect(f.backend.retire).not.toHaveBeenCalled();
|
||||
await f.service.retire({ ...f.scope, owner: next.owner });
|
||||
expect(f.backend.retire).toHaveBeenCalledOnce();
|
||||
});
|
||||
it("does not suspend another agent and fences admission until provider stop completes", async () => {
|
||||
const f = fixture();
|
||||
await f.attach();
|
||||
const a = await f.admit("agent-a", "a");
|
||||
const b = await f.admit("agent-b", "b");
|
||||
expect(a.listenerPort).not.toBe(b.listenerPort);
|
||||
expect(a.agentHome).not.toBe(b.agentHome);
|
||||
await f.service.retire({ ...f.scope, owner: a.owner });
|
||||
await f.service.reconcile();
|
||||
expect(f.backend.stop).not.toHaveBeenCalled();
|
||||
await f.service.retire({ ...f.scope, owner: b.owner });
|
||||
await f.service.reconcile();
|
||||
expect(f.backend.stop).toHaveBeenCalledOnce();
|
||||
await expect(f.admit()).rejects.toMatchObject({ code: "conflict" });
|
||||
await f.service.reconcile();
|
||||
expect(f.backend.stopStatus).toHaveBeenCalledOnce();
|
||||
f.completeStop();
|
||||
await f.service.reconcile();
|
||||
await expect(f.admit()).resolves.toHaveProperty("owner");
|
||||
});
|
||||
it("retains a failed stop intent and retries the same idempotent operation", async () => {
|
||||
const f = fixture();
|
||||
await f.attach();
|
||||
vi.mocked(f.backend.stop).mockRejectedValueOnce(
|
||||
new Error("network response lost"),
|
||||
);
|
||||
await expect(f.service.reconcile()).rejects.toBeInstanceOf(AggregateError);
|
||||
await expect(f.admit()).rejects.toMatchObject({ code: "conflict" });
|
||||
await f.service.reconcile();
|
||||
expect(f.backend.stop).toHaveBeenCalledTimes(2);
|
||||
expect(
|
||||
(await f.repository.get(f.scope)).ledger.action?.providerStopId,
|
||||
).toBe("stop_1");
|
||||
});
|
||||
it("viewer renewals cannot extend the absolute warm deadline or use another user", async () => {
|
||||
const f = fixture();
|
||||
await f.attach();
|
||||
const viewer = await f.service.connect({
|
||||
...f.scope,
|
||||
userId: "alice",
|
||||
idleTimeoutMs: 10_000,
|
||||
});
|
||||
expect(viewer.expiresAt).toBe("2026-10-10T12:00:10.000Z");
|
||||
await expect(
|
||||
f.service.renewViewer({ ...f.scope, owner: viewer.owner, userId: "bob" }),
|
||||
).rejects.toMatchObject({ code: "forbidden" });
|
||||
f.advance(9_000);
|
||||
expect(
|
||||
(
|
||||
await f.service.renewViewer({
|
||||
...f.scope,
|
||||
owner: viewer.owner,
|
||||
userId: "alice",
|
||||
})
|
||||
).expiresAt,
|
||||
).toBe(viewer.expiresAt);
|
||||
f.advance(1_001);
|
||||
await expect(
|
||||
f.service.renewViewer({
|
||||
...f.scope,
|
||||
owner: viewer.owner,
|
||||
userId: "alice",
|
||||
}),
|
||||
).rejects.toMatchObject({ code: "conflict" });
|
||||
await f.service.reconcile();
|
||||
expect(f.backend.stop).toHaveBeenCalledOnce();
|
||||
});
|
||||
it("a failed exact-process retirement keeps the machine held", async () => {
|
||||
const f = fixture();
|
||||
await f.attach();
|
||||
const a = await f.admit();
|
||||
vi.mocked(f.backend.retire).mockRejectedValue(new Error("not confirmed"));
|
||||
await expect(
|
||||
f.service.retire({ ...f.scope, owner: a.owner }),
|
||||
).rejects.toThrow("not confirmed");
|
||||
await expect(f.service.reconcile()).rejects.toBeInstanceOf(AggregateError);
|
||||
expect(f.backend.stop).not.toHaveBeenCalled();
|
||||
});
|
||||
it("keeps snapshot and attachment proof when detached", async () => {
|
||||
const f = fixture();
|
||||
await f.attach();
|
||||
await f.admit();
|
||||
await f.service.detach(f.scope);
|
||||
expect((await f.repository.get(f.scope)).ledger.status).toBe("detaching");
|
||||
f.completeStop();
|
||||
await f.service.reconcile();
|
||||
expect((await f.repository.get(f.scope)).ledger.status).toBe("detached");
|
||||
await expect(f.admit()).rejects.toMatchObject({ code: "conflict" });
|
||||
});
|
||||
it("rejects cross company scopes before provider access", async () => {
|
||||
const f = fixture();
|
||||
await f.attach();
|
||||
vi.mocked(f.backend.inspect).mockClear();
|
||||
await expect(
|
||||
f.service.inspect({ ...f.scope, companyId: "another" }),
|
||||
).rejects.toMatchObject({ code: "not_found" });
|
||||
expect(f.backend.inspect).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,730 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import type { ComputerRepository, ComputerBackend } from "./ports.js";
|
||||
import {
|
||||
ComputerError,
|
||||
assertAdmission,
|
||||
exactOwner,
|
||||
expired,
|
||||
liveOwners,
|
||||
nextPort,
|
||||
segment,
|
||||
timeout,
|
||||
type ComputerRecord,
|
||||
type OwnerRef,
|
||||
type Owner,
|
||||
} from "../domain/ledger.js";
|
||||
import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
|
||||
|
||||
type Scope = { companyId: string; environmentId: string };
|
||||
export function createComputerService(
|
||||
repository: ComputerRepository,
|
||||
backend: ComputerBackend,
|
||||
now: () => Date = () => new Date(),
|
||||
) {
|
||||
const ref = (record: ComputerRecord, owner: Owner): OwnerRef => ({
|
||||
computerId: record.id,
|
||||
ownerId: owner.id,
|
||||
generation: owner.generation,
|
||||
});
|
||||
const base = (record: ComputerRecord) =>
|
||||
`/home/user/paperclip/${segment(record.companyId)}`;
|
||||
async function scoped(input: Scope & { owner: OwnerRef }) {
|
||||
const record = await repository.get(input);
|
||||
return { record, owner: exactOwner(record, input.owner) };
|
||||
}
|
||||
async function inspect(input: Scope) {
|
||||
const record = await repository.get(input);
|
||||
const status = await backend.inspect(record);
|
||||
return {
|
||||
computerId: record.id,
|
||||
sandboxId: record.providerId,
|
||||
status: record.ledger.status,
|
||||
state: status.state,
|
||||
};
|
||||
}
|
||||
async function attach(
|
||||
input: Scope & {
|
||||
sandboxId: string;
|
||||
apiKeySecretRef: {
|
||||
type: "secret_ref";
|
||||
secretId: string;
|
||||
version?: number | "latest";
|
||||
};
|
||||
},
|
||||
) {
|
||||
segment(input.companyId);
|
||||
segment(input.environmentId);
|
||||
segment(input.sandboxId);
|
||||
segment(input.apiKeySecretRef.secretId);
|
||||
let record: ComputerRecord;
|
||||
try {
|
||||
record = await repository.get(input);
|
||||
if (
|
||||
record.providerId !== input.sandboxId
|
||||
)
|
||||
throw new ComputerError(
|
||||
"conflict",
|
||||
"Environment already attaches another computer",
|
||||
);
|
||||
} catch (error) {
|
||||
if (!(error instanceof ComputerError) || error.code !== "not_found")
|
||||
throw error;
|
||||
record = {
|
||||
id: randomUUID(),
|
||||
companyId: input.companyId,
|
||||
environmentId: input.environmentId,
|
||||
providerId: input.sandboxId,
|
||||
ledger: {
|
||||
controllerId: randomUUID(),
|
||||
status: "attaching",
|
||||
secretRef: input.apiKeySecretRef,
|
||||
owners: [],
|
||||
placements: {},
|
||||
action: null,
|
||||
},
|
||||
};
|
||||
await repository.create(record);
|
||||
}
|
||||
if (JSON.stringify(record.ledger.secretRef) !== JSON.stringify(input.apiKeySecretRef)) {
|
||||
const candidate = structuredClone(record);
|
||||
candidate.ledger.secretRef = input.apiKeySecretRef;
|
||||
const status = await backend.inspect(candidate);
|
||||
if (!status.snapshots) throw new ComputerError("invalid", "Boat snapshots must remain enabled");
|
||||
await repository.update(input, current => { current.ledger.secretRef = input.apiKeySecretRef; });
|
||||
record = candidate;
|
||||
}
|
||||
if (record.ledger.status === "detaching" || record.ledger.action)
|
||||
throw new ComputerError("conflict", "Computer cleanup is pending");
|
||||
if (record.ledger.status !== "attached") {
|
||||
await backend.ready(record);
|
||||
await backend.claim(record);
|
||||
await repository.update(input, (current) => {
|
||||
if (
|
||||
current.ledger.status === "attaching" ||
|
||||
current.ledger.status === "detached"
|
||||
)
|
||||
current.ledger.status = "attached";
|
||||
});
|
||||
}
|
||||
return { computerId: record.id };
|
||||
}
|
||||
async function binding(record: ComputerRecord, owner: Owner) {
|
||||
const ownerRef = ref(record, owner);
|
||||
const placement = record.ledger.placements[owner.sessionKey!];
|
||||
if (!placement)
|
||||
throw new ComputerError("conflict", "Computer placement is missing");
|
||||
const raw = await backend.runner(record);
|
||||
const runner: CommandManagedRuntimeRunner = {
|
||||
execute: async (input) => {
|
||||
const current = await scoped({ ...record, owner: ownerRef });
|
||||
if (!["active", "starting", "warm"].includes(current.owner.phase))
|
||||
throw new ComputerError("conflict", "Computer owner is retired");
|
||||
// Start a command unit under the same owner slice while holding the remote tombstone lock.
|
||||
// The lock is released only after systemd knows the unit, so retirement cannot miss a delayed launch.
|
||||
return raw.execute({
|
||||
command: "python3",
|
||||
args: [
|
||||
"-c",
|
||||
String.raw`
|
||||
import os,sys,json,fcntl,subprocess,time
|
||||
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 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)
|
||||
`,
|
||||
],
|
||||
stdin: JSON.stringify({
|
||||
owner: owner.id,
|
||||
generation: owner.generation,
|
||||
unit: `paperclip-command-${randomUUID()}.service`,
|
||||
cwd: placement.cwd,
|
||||
input: {
|
||||
command: input.command,
|
||||
args: input.args,
|
||||
cwd: input.cwd,
|
||||
env: input.env,
|
||||
stdin: input.stdin,
|
||||
},
|
||||
}),
|
||||
timeoutMs: input.timeoutMs,
|
||||
onLog: input.onLog,
|
||||
onSpawn: input.onSpawn,
|
||||
});
|
||||
},
|
||||
};
|
||||
return {
|
||||
owner: ownerRef,
|
||||
runner,
|
||||
listenerPort: owner.port,
|
||||
remoteCwd: placement.cwd,
|
||||
agentHome: placement.root,
|
||||
placementId: placement.id,
|
||||
async launch(input: {
|
||||
command: string;
|
||||
args?: string[];
|
||||
cwd?: string;
|
||||
env?: Record<string, string>;
|
||||
}) {
|
||||
const claimed = await repository.update(record, (current) => {
|
||||
const o = exactOwner(current, ownerRef);
|
||||
if (o.phase !== "active" && o.phase !== "starting")
|
||||
throw new ComputerError("conflict", "Computer owner cannot launch");
|
||||
if (!o.process)
|
||||
o.process = {
|
||||
bootId: "",
|
||||
unitName: `paperclip-${o.id}.service`,
|
||||
nonce: randomUUID(),
|
||||
launchGeneration: o.generation,
|
||||
};
|
||||
return structuredClone(current);
|
||||
});
|
||||
const claim = await backend.launch(
|
||||
claimed,
|
||||
exactOwner(claimed, ownerRef),
|
||||
input,
|
||||
);
|
||||
await repository.update(record, (current) => {
|
||||
exactOwner(current, ownerRef).process = claim;
|
||||
});
|
||||
return claim;
|
||||
},
|
||||
async inspectProcess() {
|
||||
const current = await scoped({ ...record, owner: ownerRef });
|
||||
return backend.inspectProcess(current.record, current.owner);
|
||||
},
|
||||
async ingress(input?: { port?: number; path?: string }) {
|
||||
const current = await scoped({ ...record, owner: ownerRef });
|
||||
if (current.owner.phase === "retired")
|
||||
throw new ComputerError("conflict", "Computer owner retired");
|
||||
return backend.ingress(
|
||||
current.record,
|
||||
input?.port ?? owner.port,
|
||||
input?.path ?? "/",
|
||||
);
|
||||
},
|
||||
};
|
||||
}
|
||||
async function admit(
|
||||
input: Scope & {
|
||||
agentId: string;
|
||||
runId: string;
|
||||
sessionKey: string;
|
||||
idleTimeoutMs: number;
|
||||
},
|
||||
) {
|
||||
segment(input.agentId);
|
||||
timeout(input.idleTimeoutMs);
|
||||
const result = await repository.update(input, (record) => {
|
||||
assertAdmission(record.ledger);
|
||||
let owner = record.ledger.owners.find(
|
||||
(o) =>
|
||||
o.kind === "runner" &&
|
||||
o.agentId === input.agentId &&
|
||||
o.sessionKey === input.sessionKey &&
|
||||
o.phase !== "retired",
|
||||
);
|
||||
if (owner && owner.phase !== "warm")
|
||||
throw new ComputerError(
|
||||
"conflict",
|
||||
"Computer session already has an active turn",
|
||||
);
|
||||
if (owner && expired(owner, now()))
|
||||
throw new ComputerError(
|
||||
"conflict",
|
||||
"Computer warm session is retiring",
|
||||
);
|
||||
if (owner) {
|
||||
owner.generation++;
|
||||
owner.phase = "starting";
|
||||
owner.runId = input.runId;
|
||||
owner.deadline = new Date(now().getTime() + 120_000).toISOString();
|
||||
} else {
|
||||
owner = {
|
||||
id: randomUUID(),
|
||||
generation: 1,
|
||||
kind: "runner",
|
||||
phase: "starting",
|
||||
agentId: input.agentId,
|
||||
runId: input.runId,
|
||||
sessionKey: input.sessionKey,
|
||||
port: nextPort(record.ledger.owners),
|
||||
deadline: new Date(now().getTime() + 120_000).toISOString(),
|
||||
absoluteDeadline: null,
|
||||
process: null,
|
||||
};
|
||||
record.ledger.owners.push(owner);
|
||||
}
|
||||
record.ledger.placements[input.agentId] ??= {
|
||||
id: randomUUID(),
|
||||
root: `${base(record)}/agents/${input.agentId}`,
|
||||
cwd: `${base(record)}/agents/${input.agentId}`,
|
||||
};
|
||||
record.ledger.placements[input.sessionKey] ??= {
|
||||
id: randomUUID(),
|
||||
root: `${base(record)}/agents/${input.agentId}`,
|
||||
cwd: `${base(record)}/agents/${input.agentId}`,
|
||||
};
|
||||
return { record: structuredClone(record), owner: structuredClone(owner) };
|
||||
});
|
||||
await backend.ready(result.record);
|
||||
await backend.claim(result.record);
|
||||
await backend.advance(result.record, result.owner);
|
||||
await backend.renew(result.record);
|
||||
const active = await repository.update(input, (record) => {
|
||||
const owner = exactOwner(record, ref(result.record, result.owner));
|
||||
if (owner.phase !== "starting")
|
||||
throw new ComputerError("conflict", "Computer admission was retired");
|
||||
owner.phase = "active";
|
||||
owner.deadline = null;
|
||||
return { record: structuredClone(record), owner: structuredClone(owner) };
|
||||
});
|
||||
return binding(active.record, active.owner);
|
||||
}
|
||||
async function recover(input: Scope & { owner: OwnerRef }) {
|
||||
const { record, owner } = await scoped(input);
|
||||
if (!["active", "warm"].includes(owner.phase) || expired(owner, now()))
|
||||
throw new ComputerError(
|
||||
"conflict",
|
||||
"Computer owner is no longer recoverable",
|
||||
);
|
||||
await backend.ready(record);
|
||||
await backend.claim(record);
|
||||
return binding(record, owner);
|
||||
}
|
||||
async function retainWarm(
|
||||
input: Scope & { owner: OwnerRef; idleTimeoutMs: number },
|
||||
) {
|
||||
timeout(input.idleTimeoutMs);
|
||||
await repository.update(input, (record) => {
|
||||
const owner = exactOwner(record, input.owner);
|
||||
if (owner.kind !== "runner" || owner.phase !== "active")
|
||||
throw new ComputerError(
|
||||
"conflict",
|
||||
"Computer owner cannot become warm",
|
||||
);
|
||||
owner.phase = "warm";
|
||||
owner.deadline = new Date(
|
||||
now().getTime() + input.idleTimeoutMs,
|
||||
).toISOString();
|
||||
});
|
||||
}
|
||||
async function retire(input: Scope & { owner: OwnerRef }) {
|
||||
const before = await repository.get(input);
|
||||
const previous = before.ledger.owners.find(
|
||||
(o) => o.id === input.owner.ownerId,
|
||||
);
|
||||
if (before.id !== input.owner.computerId || !previous)
|
||||
throw new ComputerError("not_found", "Computer owner not found");
|
||||
if (previous.generation !== input.owner.generation) return { retired: false };
|
||||
const state = await repository.update(input, (record) => {
|
||||
const owner = exactOwner(record, input.owner);
|
||||
if (owner.phase !== "retired") owner.phase = "retiring";
|
||||
return { record: structuredClone(record), owner: structuredClone(owner) };
|
||||
});
|
||||
if (state.owner.phase === "retired") return { retired: true };
|
||||
if (state.owner.kind === "runner") {
|
||||
const status = await backend.inspect(state.record);
|
||||
if (!["stopped", "archived"].includes(status.state))
|
||||
await backend.retire(state.record, state.owner);
|
||||
}
|
||||
await repository.update(input, (record) => {
|
||||
const owner = exactOwner(record, input.owner);
|
||||
owner.phase = "retired";
|
||||
owner.deadline = null;
|
||||
});
|
||||
return { retired: true };
|
||||
}
|
||||
async function connect(
|
||||
input: Scope & { userId: string; idleTimeoutMs: number },
|
||||
) {
|
||||
timeout(input.idleTimeoutMs);
|
||||
const deadline = new Date(
|
||||
now().getTime() + input.idleTimeoutMs,
|
||||
).toISOString();
|
||||
const data = await repository.update(input, (record) => {
|
||||
assertAdmission(record.ledger);
|
||||
const owner: Owner = {
|
||||
id: randomUUID(),
|
||||
generation: 1,
|
||||
kind: "viewer",
|
||||
phase: "active",
|
||||
userId: input.userId,
|
||||
port: 0,
|
||||
deadline,
|
||||
absoluteDeadline: deadline,
|
||||
process: null,
|
||||
};
|
||||
record.ledger.owners.push(owner);
|
||||
return { record: structuredClone(record), owner };
|
||||
});
|
||||
await backend.ready(data.record);
|
||||
await backend.claim(data.record);
|
||||
await backend.renew(data.record);
|
||||
const viewer = await backend.desktop(data.record);
|
||||
return {
|
||||
...viewer,
|
||||
expiresAt: new Date(
|
||||
Math.min(Date.parse(viewer.expiresAt), Date.parse(deadline)),
|
||||
).toISOString(),
|
||||
owner: ref(data.record, data.owner),
|
||||
};
|
||||
}
|
||||
async function renewViewer(
|
||||
input: Scope & { owner: OwnerRef; userId: string },
|
||||
) {
|
||||
const { record, owner } = await scoped(input);
|
||||
if (owner.kind !== "viewer" || owner.userId !== input.userId)
|
||||
throw new ComputerError(
|
||||
"forbidden",
|
||||
"Computer viewer belongs to another user",
|
||||
);
|
||||
if (owner.phase !== "active" || expired(owner, now()))
|
||||
throw new ComputerError(
|
||||
"conflict",
|
||||
"Computer connection expired; connect again",
|
||||
);
|
||||
const viewer = await backend.desktop(record);
|
||||
return {
|
||||
...viewer,
|
||||
expiresAt: new Date(
|
||||
Math.min(
|
||||
Date.parse(viewer.expiresAt),
|
||||
Date.parse(owner.absoluteDeadline!),
|
||||
),
|
||||
).toISOString(),
|
||||
owner: input.owner,
|
||||
};
|
||||
}
|
||||
async function disconnectViewer(
|
||||
input: Scope & { owner: OwnerRef; userId: string },
|
||||
) {
|
||||
const { owner } = await scoped(input);
|
||||
if (owner.kind !== "viewer" || owner.userId !== input.userId)
|
||||
throw new ComputerError(
|
||||
"forbidden",
|
||||
"Computer viewer belongs to another user",
|
||||
);
|
||||
await retire(input);
|
||||
}
|
||||
async function preview(input: Scope & { owner: OwnerRef; port: number }) {
|
||||
const { record, owner } = await scoped(input);
|
||||
if (
|
||||
owner.kind !== "runner" ||
|
||||
!["active", "warm"].includes(owner.phase) ||
|
||||
expired(owner, now())
|
||||
)
|
||||
throw new ComputerError("conflict", "Preview owner is not live");
|
||||
if (input.port >= 43127 && input.port < 44127)
|
||||
throw new ComputerError(
|
||||
"forbidden",
|
||||
"Runner ports are not application previews",
|
||||
);
|
||||
await backend.remote(record, {
|
||||
root: base(record),
|
||||
action: "owned-port",
|
||||
port: input.port,
|
||||
ownerId: owner.id,
|
||||
});
|
||||
return backend.preview(record, input.port);
|
||||
}
|
||||
async function withFiles<T>(
|
||||
input: Scope,
|
||||
root: string,
|
||||
fn: (record: ComputerRecord) => Promise<T>,
|
||||
): Promise<T> {
|
||||
const data = await repository.update(input, (record) => {
|
||||
assertAdmission(record.ledger);
|
||||
const owner: Owner = {
|
||||
id: randomUUID(),
|
||||
generation: 1,
|
||||
kind: "file-operation",
|
||||
phase: "active",
|
||||
port: 0,
|
||||
deadline: new Date(now().getTime() + 120_000).toISOString(),
|
||||
absoluteDeadline: null,
|
||||
process: null,
|
||||
};
|
||||
record.ledger.owners.push(owner);
|
||||
return { record: structuredClone(record), owner };
|
||||
});
|
||||
try {
|
||||
await backend.ready(data.record);
|
||||
await backend.claim(data.record);
|
||||
return await fn(data.record);
|
||||
} finally {
|
||||
await retire({ ...input, owner: ref(data.record, data.owner) });
|
||||
}
|
||||
}
|
||||
function fileAccess(input: Scope, root: string) {
|
||||
const remote = (payload: Record<string, unknown>) =>
|
||||
withFiles(input, root, (record) =>
|
||||
backend.remote(record, { ...payload, root }),
|
||||
);
|
||||
const readBytes = async (
|
||||
path: string,
|
||||
maxBytes = 16 * 1024 * 1024,
|
||||
): Promise<{ bytes: Buffer; sha256: string }> => {
|
||||
if (
|
||||
!Number.isSafeInteger(maxBytes) ||
|
||||
maxBytes < 0 ||
|
||||
maxBytes > 16 * 1024 * 1024
|
||||
)
|
||||
throw new ComputerError("invalid", "Invalid remote file size limit");
|
||||
const value = await remote({ action: "read", path, maxBytes });
|
||||
return {
|
||||
bytes: Buffer.from(value.base64, "base64"),
|
||||
sha256: value.sha256,
|
||||
};
|
||||
};
|
||||
const writeBytes = async (
|
||||
path: string,
|
||||
bytes: Buffer,
|
||||
expectedSha256: string | null,
|
||||
): Promise<{ sha256: string }> =>
|
||||
remote({
|
||||
action: "write",
|
||||
path,
|
||||
base64: bytes.toString("base64"),
|
||||
expectedSha256,
|
||||
});
|
||||
return {
|
||||
root,
|
||||
readBytes,
|
||||
writeBytes,
|
||||
async read(path: string) {
|
||||
const value = await readBytes(path);
|
||||
return { content: value.bytes.toString("utf8"), sha256: value.sha256 };
|
||||
},
|
||||
async write(
|
||||
path: string,
|
||||
content: string,
|
||||
expectedSha256: string | null,
|
||||
) {
|
||||
return writeBytes(path, Buffer.from(content), expectedSha256);
|
||||
},
|
||||
async list(
|
||||
path = "",
|
||||
): Promise<
|
||||
Array<{
|
||||
name: string;
|
||||
kind: "file" | "directory";
|
||||
size: number;
|
||||
mtimeMs: number;
|
||||
}>
|
||||
> {
|
||||
return remote({ action: "list", path });
|
||||
},
|
||||
async remove(path: string, expectedSha256: string) {
|
||||
await remote({ action: "remove", path, expectedSha256 });
|
||||
},
|
||||
async move(
|
||||
from: string,
|
||||
to: string,
|
||||
expectedSha256: string,
|
||||
): Promise<{ sha256: string }> {
|
||||
return remote({ action: "move", path: from, to, expectedSha256 });
|
||||
},
|
||||
async seedBytes(
|
||||
files: Record<string, Buffer>,
|
||||
): Promise<{ seeded: boolean }> {
|
||||
return remote({
|
||||
action: "seed",
|
||||
files: Object.fromEntries(
|
||||
Object.entries(files).map(([key, value]) => [
|
||||
key,
|
||||
value.toString("base64"),
|
||||
]),
|
||||
),
|
||||
});
|
||||
},
|
||||
async seed(files: Record<string, string>): Promise<{ seeded: boolean }> {
|
||||
return remote({
|
||||
action: "seed",
|
||||
files: Object.fromEntries(
|
||||
Object.entries(files).map(([key, value]) => [
|
||||
key,
|
||||
Buffer.from(value).toString("base64"),
|
||||
]),
|
||||
),
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
async function files(input: Scope & { agentId: string }) {
|
||||
segment(input.agentId);
|
||||
const placement = await repository.update(input, (record) => {
|
||||
assertAdmission(record.ledger);
|
||||
return (record.ledger.placements[input.agentId] ??= {
|
||||
id: randomUUID(),
|
||||
root: `${base(record)}/agents/${input.agentId}`,
|
||||
cwd: `${base(record)}/agents/${input.agentId}`,
|
||||
});
|
||||
});
|
||||
return fileAccess(input, placement.root);
|
||||
}
|
||||
async function workspaceFiles(input: Scope & { placementId: string }) {
|
||||
const record = await repository.get(input);
|
||||
const placement = Object.values(record.ledger.placements).find(
|
||||
(p) => p.id === input.placementId,
|
||||
);
|
||||
if (!placement)
|
||||
throw new ComputerError(
|
||||
"not_found",
|
||||
"Computer workspace placement not found",
|
||||
);
|
||||
return fileAccess(input, placement.cwd);
|
||||
}
|
||||
async function realizeWorkspace(
|
||||
input: Scope & {
|
||||
owner: OwnerRef;
|
||||
projectId?: string;
|
||||
taskId?: string;
|
||||
repositoryUrl?: string;
|
||||
branch?: string;
|
||||
baseRef?: string;
|
||||
mode: "shared" | "worktree";
|
||||
},
|
||||
) {
|
||||
const { record, owner } = await scoped(input);
|
||||
if (!owner.agentId || owner.phase !== "active")
|
||||
throw new ComputerError("conflict", "Workspace owner is not active");
|
||||
const placement = record.ledger.placements[owner.sessionKey!]!;
|
||||
if (!input.projectId)
|
||||
return {
|
||||
remoteCwd: placement.cwd,
|
||||
agentHome: placement.root,
|
||||
placementId: placement.id,
|
||||
};
|
||||
const root = `${base(record)}/projects/${segment(input.projectId)}`;
|
||||
if (input.taskId) segment(input.taskId);
|
||||
const result = await backend.remote(record, {
|
||||
...input,
|
||||
root,
|
||||
action: "workspace",
|
||||
});
|
||||
await repository.update(input, (current) => {
|
||||
exactOwner(current, input.owner);
|
||||
const placementKey = `workspace:${input.projectId}:${input.mode}:${input.taskId ?? "shared"}:${result.remoteCwd}`;
|
||||
current.ledger.placements[placementKey] ??= {
|
||||
id: randomUUID(),
|
||||
root: placement.root,
|
||||
cwd: result.remoteCwd,
|
||||
};
|
||||
current.ledger.placements[owner.sessionKey!] =
|
||||
current.ledger.placements[placementKey]!;
|
||||
});
|
||||
const updated = await repository.get(input);
|
||||
return {
|
||||
remoteCwd: result.remoteCwd as string,
|
||||
agentHome: placement.root,
|
||||
placementId: updated.ledger.placements[owner.sessionKey!]!.id,
|
||||
};
|
||||
}
|
||||
async function detach(input: Scope) {
|
||||
const record = await repository.update(input, (current) => {
|
||||
current.ledger.status = "detaching";
|
||||
return structuredClone(current);
|
||||
});
|
||||
for (const owner of liveOwners(record.ledger))
|
||||
await retire({ ...input, owner: ref(record, owner) });
|
||||
await reconcileRecord(await repository.get(input));
|
||||
}
|
||||
async function reconcileRecord(record: ComputerRecord) {
|
||||
if (record.ledger.status === "attaching") return;
|
||||
for (const owner of liveOwners(record.ledger))
|
||||
if (owner.phase === "retiring" || expired(owner, now()))
|
||||
await retire({ ...record, owner: ref(record, owner) });
|
||||
record = await repository.get(record);
|
||||
if (record.ledger.action) {
|
||||
const action = record.ledger.action;
|
||||
// A lost stop response is retried only while admission is fenced. Boat returns its same pending ID.
|
||||
const stop = action.providerStopId
|
||||
? await backend.stopStatus(record, action.providerStopId)
|
||||
: await backend.stop(record);
|
||||
await repository.update(record, (current) => {
|
||||
if (current.ledger.action?.id !== action.id) return;
|
||||
current.ledger.action.providerStopId = stop.id;
|
||||
if (stop.status === "completed") {
|
||||
current.ledger.action = null;
|
||||
if (current.ledger.status === "detaching")
|
||||
current.ledger.status = "detached";
|
||||
}
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (liveOwners(record.ledger).length) {
|
||||
await backend.renew(record);
|
||||
return;
|
||||
}
|
||||
if (record.ledger.status === "detached") return;
|
||||
const status = await backend.inspect(record);
|
||||
if (["stopped", "archived"].includes(status.state)) {
|
||||
if (record.ledger.status === "detaching")
|
||||
await repository.update(record, (current) => {
|
||||
current.ledger.status = "detached";
|
||||
});
|
||||
return;
|
||||
}
|
||||
const actionId = randomUUID();
|
||||
const claimed = await repository.update(record, (current) => {
|
||||
if (current.ledger.action || liveOwners(current.ledger).length)
|
||||
return false;
|
||||
current.ledger.action = {
|
||||
kind: "stop",
|
||||
id: actionId,
|
||||
providerStopId: null,
|
||||
};
|
||||
return true;
|
||||
});
|
||||
if (!claimed) return;
|
||||
const stop = await backend.stop(record);
|
||||
await repository.update(record, (current) => {
|
||||
if (current.ledger.action?.id === actionId)
|
||||
current.ledger.action.providerStopId = stop.id;
|
||||
});
|
||||
}
|
||||
async function reconcile() {
|
||||
const failures: unknown[] = [];
|
||||
for (const record of await repository.all())
|
||||
try {
|
||||
await reconcileRecord(record);
|
||||
} catch (error) {
|
||||
failures.push(error);
|
||||
}
|
||||
if (failures.length)
|
||||
throw new AggregateError(
|
||||
failures,
|
||||
"Computer reconciliation requires retry",
|
||||
);
|
||||
}
|
||||
return {
|
||||
async list(companyId: string) { return (await repository.all()).filter(record => record.companyId === companyId).map(record => ({ computerId: record.id, environmentId: record.environmentId, sandboxId: record.providerId, status: record.ledger.status })); },
|
||||
attach,
|
||||
admit,
|
||||
recover,
|
||||
retainWarm,
|
||||
retire,
|
||||
connect,
|
||||
renewViewer,
|
||||
disconnectViewer,
|
||||
preview,
|
||||
files,
|
||||
workspaceFiles,
|
||||
realizeWorkspace,
|
||||
detach,
|
||||
reconcile,
|
||||
inspect,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
export { computerCompanyDeletion } from "./adapters/company-deletion.js";
|
||||
@@ -0,0 +1,110 @@
|
||||
export type OwnerRef = {
|
||||
computerId: string;
|
||||
ownerId: string;
|
||||
generation: number;
|
||||
};
|
||||
export type ProcessClaim = {
|
||||
bootId: string;
|
||||
unitName: string;
|
||||
nonce: string;
|
||||
launchGeneration: number;
|
||||
};
|
||||
export interface Owner {
|
||||
id: string;
|
||||
generation: number;
|
||||
kind: "runner" | "viewer" | "file-operation";
|
||||
phase: "starting" | "active" | "warm" | "retiring" | "retired";
|
||||
agentId?: string;
|
||||
runId?: string;
|
||||
sessionKey?: string;
|
||||
userId?: string;
|
||||
port: number;
|
||||
deadline: string | null;
|
||||
absoluteDeadline: string | null;
|
||||
process: ProcessClaim | null;
|
||||
}
|
||||
export interface Ledger {
|
||||
controllerId: string;
|
||||
status: "attaching" | "attached" | "detaching" | "detached";
|
||||
secretRef: {
|
||||
type: "secret_ref";
|
||||
secretId: string;
|
||||
version?: number | "latest";
|
||||
};
|
||||
owners: Owner[];
|
||||
placements: Record<string, { id: string; root: string; cwd: string }>;
|
||||
action: null | { kind: "stop"; id: string; providerStopId: string | null };
|
||||
}
|
||||
export interface ComputerRecord {
|
||||
id: string;
|
||||
companyId: string;
|
||||
environmentId: string;
|
||||
providerId: string;
|
||||
ledger: Ledger;
|
||||
}
|
||||
export class ComputerError extends Error {
|
||||
constructor(
|
||||
public readonly code:
|
||||
| "not_found"
|
||||
| "conflict"
|
||||
| "forbidden"
|
||||
| "invalid"
|
||||
| "provider_error",
|
||||
message: string,
|
||||
) {
|
||||
super(message);
|
||||
this.name = "ComputerError";
|
||||
}
|
||||
}
|
||||
export function segment(value: string): string {
|
||||
if (!/^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$/.test(value))
|
||||
throw new ComputerError("invalid", "Invalid computer resource identifier");
|
||||
return value;
|
||||
}
|
||||
export function timeout(value: number): number {
|
||||
if (
|
||||
!Number.isSafeInteger(value) ||
|
||||
value < 1000 ||
|
||||
value > 24 * 60 * 60 * 1000
|
||||
)
|
||||
throw new ComputerError(
|
||||
"invalid",
|
||||
"Computer timeout must be between one second and one day",
|
||||
);
|
||||
return value;
|
||||
}
|
||||
export function exactOwner(record: ComputerRecord, ref: OwnerRef): Owner {
|
||||
const owner = record.ledger.owners.find((o) => o.id === ref.ownerId);
|
||||
if (
|
||||
record.id !== ref.computerId ||
|
||||
!owner ||
|
||||
owner.generation !== ref.generation
|
||||
)
|
||||
throw new ComputerError("conflict", "Computer owner has been superseded");
|
||||
return owner;
|
||||
}
|
||||
export function nextPort(owners: Owner[]): number {
|
||||
const occupied = new Set(
|
||||
owners.filter((o) => o.phase !== "retired").map((o) => o.port),
|
||||
);
|
||||
for (let port = 43127; port < 44127; port++)
|
||||
if (!occupied.has(port)) return port;
|
||||
throw new ComputerError("conflict", "Computer has no available runner ports");
|
||||
}
|
||||
export function liveOwners(ledger: Ledger): Owner[] {
|
||||
return ledger.owners.filter((o) => o.phase !== "retired");
|
||||
}
|
||||
export function assertAdmission(ledger: Ledger): void {
|
||||
if (ledger.status !== "attached" || ledger.action)
|
||||
throw new ComputerError(
|
||||
"conflict",
|
||||
"Computer is connecting or settling a previous stop",
|
||||
);
|
||||
}
|
||||
export function expired(owner: Owner, now: Date): boolean {
|
||||
return (
|
||||
owner.phase !== "retired" &&
|
||||
owner.deadline !== null &&
|
||||
Date.parse(owner.deadline) <= now.getTime()
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { secretService } from "../../services/secrets.js";
|
||||
import { computerRepository } from "./adapters/repository.js";
|
||||
import { boatBackend } from "./adapters/boat.js";
|
||||
import { createComputerService } from "./application/service.js";
|
||||
export { ComputerError } from "./domain/ledger.js";
|
||||
export type {
|
||||
OwnerRef as ComputerOwnerRef,
|
||||
ProcessClaim as ComputerProcessClaim,
|
||||
} from "./domain/ledger.js";
|
||||
export type ComputerService = ReturnType<typeof createComputerService>;
|
||||
export type ComputerBinding = Awaited<ReturnType<ComputerService["admit"]>>;
|
||||
const services = new WeakMap<Db, ComputerService>();
|
||||
export function computerService(db: Db): ComputerService {
|
||||
const existing = services.get(db);
|
||||
if (existing) return existing;
|
||||
const secrets = secretService(db);
|
||||
const service = createComputerService(
|
||||
computerRepository(db),
|
||||
boatBackend((record) =>
|
||||
secrets.resolveSecretValue(
|
||||
record.companyId,
|
||||
record.ledger.secretRef.secretId,
|
||||
record.ledger.secretRef.version ?? "latest",
|
||||
{
|
||||
consumerType: "environment",
|
||||
consumerId: record.environmentId,
|
||||
configPath: "apiKeySecretRef",
|
||||
actorType: "system",
|
||||
actorId: null,
|
||||
},
|
||||
),
|
||||
),
|
||||
);
|
||||
services.set(db, service);
|
||||
return service;
|
||||
}
|
||||
Reference in new issue
Block a user