Something went wrong. Try again.
A local-first event pipeline for independent agents, built on Jazz.
Something went wrong. Try again.
14 kB · 169 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170import { createHash } from "node:crypto";import { constants, createWriteStream } from "node:fs";import fs from "node:fs/promises";import path from "node:path";import { spawn } from "node:child_process";import { z } from "zod";import type { JazzThoughtStore } from "../jazz/store.js";import { readVerifiedBlob } from "./blob.js";import { requestArtifactStorage } from "./append.js";import { ARTIFACT_KINDS, ARTIFACT_MEDIA_TYPES, ARTIFACT_VISIBILITY } from "./types.js";
const id = z.string().min(1).max(200).regex(/^[A-Za-z0-9][A-Za-z0-9._:-]*$/);const text = z.string().min(1).max(2_000);const MAX_SSH_OUTPUT_BYTES = 12 * 1024 * 1024;export const exeArtifactRequestSchema = z.object({ requestId: id, leaseId: id, relativeOutputFilePath: z.string().min(1).max(500).refine(normalizedRelative, "not-normalized-relative"), artifactId: id, artifactVersion: z.number().int().positive().max(1_000_000), kind: z.enum(ARTIFACT_KINDS), title: text, summary: text, mediaType: z.enum(ARTIFACT_MEDIA_TYPES), provenanceLabel: text, visibility: z.enum(ARTIFACT_VISIBILITY).default("private"), requestingAgentId: id.optional(), runId: id.optional(), sourceEventId: id.optional(), supersedes: id.optional(),}).strict();export type ExeArtifactRequest = z.infer<typeof exeArtifactRequestSchema>;
export interface ExeTransportConfig { host: "cameron.exe.xyz"; workspaceRoot: "/srv/thoughtstream/workspaces/stream-v1"; workspaceId: "stream-v1"; leaseId: string; stagingRoot: string;}export interface ExeTransportReceipt { requestId: string; requestEventId: string; artifactEventId: string; artifactId: string; sha256: string; bytes: number; status: "stored"; timestamp: string }
const REMOTE_HELPER = String.raw`import base64,hashlib,json,os,stat,syscfg=json.loads(base64.urlsafe_b64decode(sys.argv[1]+"=="))root=cfg["root"]; rel=cfg["relative"]def fail(code): print(json.dumps({"ok":False,"reasonCode":code},separators=(",",":")));sys.exit(0)try: mpath=os.path.join(root,".thoughtstream-workspace.json") rst=os.lstat(root) if stat.S_ISLNK(rst.st_mode) or not stat.S_ISDIR(rst.st_mode) or os.path.realpath(root)!=root: fail("workspace-invalid") with open(mpath,"rb") as f: m=json.load(f) if m.get("workspaceId")!=cfg["workspaceId"] or m.get("leaseId")!=cfg["leaseId"] or m.get("vmIdentity")!=cfg["host"]: fail("workspace-mismatch") parts=rel.split("/") if not rel or rel.startswith("/") or "\\" in rel or any(x in ("",".","..") for x in parts): fail("path-invalid") cur=os.path.join(root,"output") for part in parts: cur=os.path.join(cur,part); s=os.lstat(cur) if stat.S_ISLNK(s.st_mode): fail("symlink") if os.path.realpath(cur)!=cur or not stat.S_ISREG(os.lstat(cur).st_mode): fail("not-regular") ext=os.path.splitext(cur)[1].lower(); pairs={".md":["text/markdown"],".markdown":["text/markdown"],".txt":["text/plain"],".json":["application/json"],".yaml":["application/yaml","text/yaml"],".yml":["application/yaml","text/yaml"],".png":["image/png"],".jpg":["image/jpeg"],".jpeg":["image/jpeg"]} if cfg["mediaType"] not in pairs.get(ext,[]): fail("media-mismatch") maximum=10485760 if cfg["mediaType"].startswith("image/") else 1048576 before=os.lstat(cur) if before.st_size<=0 or before.st_size>maximum: fail("size-invalid") fd=os.open(cur,os.O_RDONLY|os.O_NOFOLLOW) try: opened=os.fstat(fd); data=b"" while len(data)<=maximum: chunk=os.read(fd,min(65536,maximum+1-len(data))) if not chunk: break data+=chunk after=os.fstat(fd) finally: os.close(fd) ident=lambda s:(s.st_dev,s.st_ino,s.st_size,s.st_mtime_ns,s.st_ctime_ns) if ident(before)!=ident(opened) or ident(opened)!=ident(after) or len(data)!=after.st_size: fail("unstable-snapshot") mt=cfg["mediaType"] if mt=="image/png" and not data.startswith(b"\x89PNG\r\n\x1a\n"): fail("magic-invalid") if mt=="image/jpeg" and not (data.startswith(b"\xff\xd8") and data.endswith(b"\xff\xd9")): fail("magic-invalid") if not mt.startswith("image/"): try: value=data.decode("utf-8") except UnicodeDecodeError: fail("encoding-invalid") if "\x00" in value: fail("encoding-invalid") if mt=="application/json": json.loads(value) header=json.dumps({"ok":True,"bytes":len(data),"sha256":hashlib.sha256(data).hexdigest(),"mediaType":mt,"relativePath":rel},separators=(",",":" )).encode() sys.stdout.buffer.write(str(len(header)).encode()+b"\n"+header+data)except (OSError,ValueError,json.JSONDecodeError): fail("remote-validation-failed")`;
const REMOTE_RECEIPT_HELPER = String.raw`import base64,json,os,sysp=sys.argv[1]r=json.loads(base64.urlsafe_b64decode(sys.argv[2]+"=="))n=r["requestId"]+".json"assert all(c.isalnum() or c in "._:-" for c in r["requestId"])d=(json.dumps(r,separators=(",",":"),sort_keys=True)+"\n").encode()q=os.path.join(p,n)try: f=os.open(q,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600) os.write(f,d) os.close(f)except FileExistsError: assert open(q,"rb").read()==d`;
const REMOTE_PROCESSED_HELPER = String.raw`import os,syss=os.path.join(sys.argv[1],sys.argv[3])d=os.path.join(sys.argv[2],sys.argv[3])if os.path.exists(d): assert open(s,"rb").read()==open(d,"rb").read() os.unlink(s)else: os.rename(s,d)`;
export async function processExeArtifactRequest(store: JazzThoughtStore, config: ExeTransportConfig, raw: unknown): Promise<ExeTransportReceipt> { const request = exeArtifactRequestSchema.parse(raw); if (request.leaseId !== config.leaseId) throw reason("lease-mismatch"); await ensurePrivateStaging(config.stagingRoot); const framed = await remoteFetch(config, request); const hash = createHash("sha256").update(framed.bytes).digest("hex"); if (hash !== framed.header.sha256 || framed.bytes.byteLength !== framed.header.bytes || framed.header.mediaType !== request.mediaType || framed.header.relativePath !== request.relativeOutputFilePath) throw reason("content-integrity"); const requestStage = await fs.mkdtemp(path.join(config.stagingRoot, "request-")); await fs.chmod(requestStage, 0o700); const staged = path.join(requestStage, "artifact" + path.extname(request.relativeOutputFilePath)); try { await fs.writeFile(staged, framed.bytes, { mode: 0o600, flag: "wx" }); const result = await requestArtifactStorage({ store, workspaceRoot: requestStage, workspaceLeaseId: config.leaseId, workspaceRootLabel: `exe-workspace:${config.workspaceId}`, requestId: request.requestId, relativeFilePath: path.basename(staged), artifactId: request.artifactId, artifactVersion: request.artifactVersion, kind: request.kind, title: request.title, summary: request.summary, mediaType: request.mediaType, provenance: { source: "exe-dev-private-workspace", label: request.provenanceLabel }, visibility: request.visibility, ...(request.requestingAgentId ? { requestingAgentId: request.requestingAgentId } : {}), ...(request.runId ? { requestingRunId: request.runId } : {}), ...(request.sourceEventId ? { sourceEventId: request.sourceEventId } : {}), ...(request.supersedes ? { supersedesArtifactEventId: request.supersedes } : {}), }); const persisted = await store.getEvent(result.event.id); if (!persisted || persisted.id !== result.event.id) throw reason("durable-readback-failed"); const blob = persisted.payload.blob as { algorithm: "sha256"; sha256: string; relativePath: string; byteCount: number }; const readback = await readVerifiedBlob(store.getArtifactRoot(), blob, request.mediaType); if (!readback.equals(framed.bytes)) throw reason("durable-readback-failed"); const receipt: ExeTransportReceipt = { requestId: request.requestId, requestEventId: result.requestEvent!.id, artifactEventId: result.event.id, artifactId: request.artifactId, sha256: hash, bytes: framed.bytes.byteLength, status: "stored", timestamp: persisted.observedAt }; await remoteReceipt(config, receipt); return receipt; } finally { await fs.rm(requestStage, { recursive: true, force: true }); }}
export async function runExeArtifactTransportOnce(store: JazzThoughtStore, config: ExeTransportConfig): Promise<{ receipts: ExeTransportReceipt[]; failures: Array<{ requestId: string; reasonCode: string }> }> { const lock = await acquireLock(path.join(config.stagingRoot, "transport.lock")); try { const names = await remoteList(config); const receipts: ExeTransportReceipt[] = []; const failures: Array<{ requestId: string; reasonCode: string }> = []; for (const name of names) { let raw: unknown; let requestId = name.replace(/\.json$/, ""); try { raw = JSON.parse(await remoteReadRequest(config, name)); requestId = typeof (raw as { requestId?: unknown }).requestId === "string" ? (raw as { requestId: string }).requestId : requestId; receipts.push(await processExeArtifactRequest(store, config, raw)); await remoteProcessed(config, name); } catch (error) { failures.push({ requestId: safeId(requestId), reasonCode: reasonCode(error) }); } } return { receipts, failures }; } finally { await lock.close(); await fs.rm(path.join(config.stagingRoot, "transport.lock"), { force: true }); }}
async function remoteFetch(config: ExeTransportConfig, request: ExeArtifactRequest): Promise<{ header: { bytes: number; sha256: string; mediaType: string; relativePath: string }; bytes: Buffer }> { const payload = b64({ root: config.workspaceRoot, workspaceId: config.workspaceId, leaseId: config.leaseId, host: config.host, relative: request.relativeOutputFilePath, mediaType: request.mediaType }); const output = await ssh(config.host, ["python3", "-c", b64(REMOTE_HELPER), payload]); const newline = output.indexOf(10); if (newline < 1) throw reason("remote-frame-invalid"); const length = Number(output.subarray(0, newline).toString("ascii")); if (!Number.isSafeInteger(length) || length < 2 || length > 2048) { const failure = parseFailure(output); throw reason(failure); } const header = JSON.parse(output.subarray(newline + 1, newline + 1 + length).toString("utf8")); const bytes = output.subarray(newline + 1 + length); if (!header.ok) throw reason(header.reasonCode ?? "remote-validation-failed"); return { header, bytes };}async function remoteList(config: ExeTransportConfig): Promise<string[]> { const output = await ssh(config.host, ["python3", "-c", b64("import json,os,sys;p=sys.argv[1];v=sorted(x for x in os.listdir(p) if x.endswith('.json') and '/' not in x);assert len(v)<=256;print(json.dumps(v))"), `${config.workspaceRoot}/requests`]); return JSON.parse(output.toString("utf8")); }async function remoteReadRequest(config: ExeTransportConfig, name: string): Promise<string> { if (!/^[A-Za-z0-9][A-Za-z0-9._-]{0,199}\.json$/.test(name)) throw reason("request-name-invalid"); return (await ssh(config.host, ["python3", "-c", b64("import os,sys; p=os.path.join(sys.argv[1],sys.argv[2]); s=os.lstat(p); assert not os.path.islink(p) and os.path.isfile(p) and s.st_size<=16384; sys.stdout.buffer.write(open(p,'rb').read())"), `${config.workspaceRoot}/requests`, name])).toString("utf8"); }async function remoteReceipt(config: ExeTransportConfig, receipt: ExeTransportReceipt): Promise<void> { await ssh(config.host, ["python3", "-c", b64(REMOTE_RECEIPT_HELPER), `${config.workspaceRoot}/receipts`, b64(receipt)]); }async function remoteProcessed(config: ExeTransportConfig, name: string): Promise<void> { await ssh(config.host, ["python3", "-c", b64(REMOTE_PROCESSED_HELPER), `${config.workspaceRoot}/requests`, `${config.workspaceRoot}/requests/processed`, name]); }function ssh(host: string, args: string[], input?: Buffer): Promise<Buffer> { const quote=(arg:string)=>`'${arg.replaceAll("'", "'\\''")}'`; return new Promise((resolve, reject) => { const remote = args[1] === "-c" && args.length >= 3 ? `python3 -c 'import base64,sys;exec(base64.urlsafe_b64decode(\"${args[2]}==\"))' ${quote(args[3] ?? "")}${args.slice(4).map((arg)=>` ${quote(arg)}`).join("")}` : args.map(quote).join(" "); const child = spawn("ssh", ["-o", "BatchMode=yes", "-o", "ConnectTimeout=10", "-o", "StrictHostKeyChecking=yes", "-o", "ClearAllForwardings=yes", "--", host, remote], { stdio: ["pipe", "pipe", "pipe"] }); const out: Buffer[]=[]; const err: Buffer[]=[]; let outputBytes=0; let outputOverflow=false; child.stdout.on("data", x=>{ outputBytes+=x.length; if(outputBytes>MAX_SSH_OUTPUT_BYTES){ outputOverflow=true; child.kill("SIGKILL"); return; } out.push(x); }); child.stderr.on("data", x=>{ if(Buffer.concat(err).byteLength<4096) err.push(x); }); child.on("error", reject); child.on("close", code=>outputOverflow?reject(reason("ssh-output-too-large")):code===0?resolve(Buffer.concat(out)):reject(Object.assign(reason("ssh-failed"), { detail: Buffer.concat(err).toString("utf8").slice(0, 500) }))); if (input) child.stdin.end(input); else child.stdin.end(); }); }async function ensurePrivateStaging(root: string): Promise<void> { if (!path.isAbsolute(root)) throw reason("staging-invalid"); await fs.mkdir(root,{recursive:true,mode:0o700}); await fs.chmod(root,0o700); const s=await fs.lstat(root); if (!s.isDirectory()||s.isSymbolicLink()||await fs.realpath(root)!==root) throw reason("staging-invalid"); }async function acquireLock(file: string): Promise<fs.FileHandle> { await ensurePrivateStaging(path.dirname(file)); try { return await fs.open(file, constants.O_CREAT|constants.O_EXCL|constants.O_WRONLY,0o600); } catch { throw reason("already-running"); } }function b64(value: unknown): string { return Buffer.from(typeof value === "string" ? value : JSON.stringify(value)).toString("base64url"); }function normalizedRelative(value: string): boolean { return !path.isAbsolute(value) && !value.includes("\\") && !value.endsWith("/") && value.split("/").every(part=>part!==""&&part!=="."&&part!==".."); }function reason(code: string): Error { return Object.assign(new Error(code), { reasonCode: code }); }function reasonCode(error: unknown): string { const code=(error as {reasonCode?:unknown})?.reasonCode; return typeof code==="string"&&/^[a-z0-9-]{1,64}$/.test(code)?code:"request-rejected"; }function safeId(value: string): string { return /^[A-Za-z0-9][A-Za-z0-9._:-]{0,199}$/.test(value)?value:"invalid-request"; }function parseFailure(output: Buffer): string { try { const value=JSON.parse(output.toString("utf8")); return typeof value.reasonCode==="string"?value.reasonCode:"remote-frame-invalid"; } catch { return "remote-frame-invalid"; } }