Resume on another node: worker-authenticated artifact list/download, workspace restoration from the artifact store
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
8 changed files +145 −2
added
apps/web/src/app/api/internal/workers/runs/[id]/artifacts/[...path]/route.ts
+25 −0
@@ -0,0 +1,25 @@ | ||
| 1 | +import { AppError, readRunArtifactForWorker } from "@autoscholar/core"; | |
| 2 | +import { z } from "zod"; | |
| 3 | +import { handler, parseQuery } from "@/lib/server/api"; | |
| 4 | +import { db, storage } from "@/lib/server/db"; | |
| 5 | +import { requireWorker } from "@/lib/server/internal"; | |
| 6 | + | |
| 7 | +export const runtime = "nodejs"; | |
| 8 | + | |
| 9 | +/** Download the latest version of one artifact (worker holding the lease only) — used to resume a run on another node. */ | |
| 10 | +export const GET = handler(async (req, { params }) => { | |
| 11 | + const worker = await requireWorker(req); | |
| 12 | + const { id, path } = (await params) as { id: string; path: string[] }; | |
| 13 | + const q = parseQuery(req.url, z.object({ fencingToken: z.coerce.number().int() })); | |
| 14 | + const rel = path.map(decodeURIComponent).join("/"); | |
| 15 | + if (rel.includes("..")) throw new AppError("invalid", "bad path", 400); | |
| 16 | + const { row, buf } = await readRunArtifactForWorker(db(), storage(), worker, id, q.fencingToken, rel); | |
| 17 | + return new Response(new Uint8Array(buf), { | |
| 18 | + headers: { | |
| 19 | + "content-type": row.contentType, | |
| 20 | + "content-length": String(buf.length), | |
| 21 | + "x-artifact-sha256": row.sha256, | |
| 22 | + "cache-control": "no-store", | |
| 23 | + }, | |
| 24 | + }); | |
| 25 | +}); | |
modified
apps/web/src/app/api/internal/workers/runs/[id]/artifacts/route.ts
+9 −1
@@ -1,4 +1,4 @@ | ||
| 1 | −import { AppError, ingestArtifact } from "@autoscholar/core"; | |
| 1 | +import { AppError, ingestArtifact, listRunArtifactsForWorker } from "@autoscholar/core"; | |
| 2 | 2 | import { ArtifactDescriptor } from "@autoscholar/schemas"; |
| 3 | 3 | import { z } from "zod"; |
| 4 | 4 | import { handler, json, parseQuery } from "@/lib/server/api"; |
@@ -39,3 +39,11 @@ export const PUT = handler(async (req, { params }) => { | ||
| 39 | 39 | }); |
| 40 | 40 | return json(await ingestArtifact(db(), storage(), worker, id, q.fencingToken, desc, body), { status: 201 }); |
| 41 | 41 | }); |
| 42 | + | |
| 43 | +/** List the latest artifacts of a run (worker holding the lease only). */ | |
| 44 | +export const GET = handler(async (req, { params }) => { | |
| 45 | + const worker = await requireWorker(req); | |
| 46 | + const { id } = (await params) as { id: string }; | |
| 47 | + const q = parseQuery(req.url, z.object({ fencingToken: z.coerce.number().int() })); | |
| 48 | + return json({ artifacts: await listRunArtifactsForWorker(db(), worker, id, q.fencingToken) }); | |
| 49 | +}); | |
modified
packages/core/src/workers.ts
+45 −0
@@ -657,3 +657,48 @@ export async function workerTelemetry(db: Db, workerId: string, limit = 60) { | ||
| 657 | 657 | } |
| 658 | 658 | |
| 659 | 659 | export { conflict, isNull }; |
| 660 | + | |
| 661 | +/** Latest version of every artifact of a run, for a worker that holds the lease (resume on another node). */ | |
| 662 | +export async function listRunArtifactsForWorker( | |
| 663 | + db: Db, | |
| 664 | + worker: WorkerRow, | |
| 665 | + runId: string, | |
| 666 | + fencingToken: number, | |
| 667 | +) { | |
| 668 | + await assertLease(db, runId, worker, fencingToken); | |
| 669 | + const rows = await db | |
| 670 | + .select({ | |
| 671 | + path: artifacts.path, | |
| 672 | + version: artifacts.version, | |
| 673 | + sha256: artifacts.sha256, | |
| 674 | + bytes: artifacts.bytes, | |
| 675 | + contentType: artifacts.contentType, | |
| 676 | + storageKey: artifacts.storageKey, | |
| 677 | + kind: artifacts.kind, | |
| 678 | + }) | |
| 679 | + .from(artifacts) | |
| 680 | + .where(eq(artifacts.runId, runId)) | |
| 681 | + .orderBy(descOrder(artifacts.version)); | |
| 682 | + const latest = new Map<string, (typeof rows)[number]>(); | |
| 683 | + for (const r of rows) if (!latest.has(r.path)) latest.set(r.path, r); | |
| 684 | + return [...latest.values()].map(({ storageKey: _k, ...rest }) => rest); | |
| 685 | +} | |
| 686 | + | |
| 687 | +export async function readRunArtifactForWorker( | |
| 688 | + db: Db, | |
| 689 | + storage: FilesystemStorage, | |
| 690 | + worker: WorkerRow, | |
| 691 | + runId: string, | |
| 692 | + fencingToken: number, | |
| 693 | + path: string, | |
| 694 | +) { | |
| 695 | + await assertLease(db, runId, worker, fencingToken); | |
| 696 | + const [row] = await db | |
| 697 | + .select() | |
| 698 | + .from(artifacts) | |
| 699 | + .where(and(eq(artifacts.runId, runId), eq(artifacts.path, path))) | |
| 700 | + .orderBy(descOrder(artifacts.version)) | |
| 701 | + .limit(1); | |
| 702 | + if (!row) throw notFound("artifact"); | |
| 703 | + return { row, buf: await storage.get(row.storageKey) }; | |
| 704 | +} | |
modified
packages/maclustr-adapter/src/index.ts
+35 −0
@@ -184,6 +184,41 @@ export class WorkerClient { | ||
| 184 | 184 | { buffer, contentType: artifact.contentType }, |
| 185 | 185 | ); |
| 186 | 186 | } |
| 187 | + listArtifacts(runId: string, fencingToken: number) { | |
| 188 | + return this.call<{ | |
| 189 | + artifacts: Array<{ | |
| 190 | + path: string; | |
| 191 | + version: number; | |
| 192 | + sha256: string; | |
| 193 | + bytes: number; | |
| 194 | + contentType: string; | |
| 195 | + kind: string; | |
| 196 | + }>; | |
| 197 | + }>("GET", `/api/internal/workers/runs/${runId}/artifacts?fencingToken=${fencingToken}`); | |
| 198 | + } | |
| 199 | + async downloadArtifact(runId: string, fencingToken: number, path: string): Promise<Buffer> { | |
| 200 | + const ctrl = new AbortController(); | |
| 201 | + const t = setTimeout(() => ctrl.abort(), 120_000); | |
| 202 | + try { | |
| 203 | + const headers: Record<string, string> = { | |
| 204 | + "user-agent": this.userAgent, | |
| 205 | + "x-protocol-version": String(WORKER_PROTOCOL_VERSION), | |
| 206 | + }; | |
| 207 | + if (this.token) headers.authorization = `Bearer ${this.token}`; | |
| 208 | + const res = await this.fetchImpl( | |
| 209 | + `${this.base}/api/internal/workers/runs/${runId}/artifacts/${path.split("/").map(encodeURIComponent).join("/")}?fencingToken=${fencingToken}`, | |
| 210 | + { headers, signal: ctrl.signal }, | |
| 211 | + ); | |
| 212 | + if (!res.ok) throw new ControlPlaneError(`HTTP ${res.status} downloading ${path}`, res.status); | |
| 213 | + const buf = Buffer.from(await res.arrayBuffer()); | |
| 214 | + const sha = res.headers.get("x-artifact-sha256"); | |
| 215 | + if (sha && createHash("sha256").update(buf).digest("hex") !== sha) | |
| 216 | + throw new ControlPlaneError(`digest mismatch downloading ${path}`, 500); | |
| 217 | + return buf; | |
| 218 | + } finally { | |
| 219 | + clearTimeout(t); | |
| 220 | + } | |
| 221 | + } | |
| 187 | 222 | complete(runId: string, body: z.input<typeof CompleteAttemptRequest>) { |
| 188 | 223 | return this.call<{ ok: true }>( |
| 189 | 224 | "POST", |
modified
services/qa/package.json
+1 −1
@@ -26,4 +26,4 @@ | ||
| 26 | 26 | "pdfjs-dist": "^6.2.108", |
| 27 | 27 | "zod": "^4.1.0" |
| 28 | 28 | } |
| 29 | −} | |
| \ No newline at end of file | ||
| 29 | +} | |
modified
services/research-runner/src/runner.ts
+2 −0
@@ -59,6 +59,8 @@ export interface RunnerHost { | ||
| 59 | 59 | }) => Promise<void>; |
| 60 | 60 | upload: (path: string, kind: ArtifactKind, contentType: string, buf: Buffer) => Promise<void>; |
| 61 | 61 | isCancelled: () => boolean; |
| 62 | + /** Optional: fetch the latest artifacts of this run from the control plane (resume on a different node). */ | |
| 63 | + fetchArtifacts?: () => Promise<Array<{ path: string; read: () => Promise<Buffer> }>>; | |
| 62 | 64 | } |
| 63 | 65 | |
| 64 | 66 | export interface RunOutcome { |
modified
services/worker/src/main.ts
+7 −0
@@ -259,6 +259,13 @@ class Worker { | ||
| 259 | 259 | ); |
| 260 | 260 | }, |
| 261 | 261 | isCancelled: () => state.cancel || this.stopping, |
| 262 | + fetchArtifacts: async () => { | |
| 263 | + const { artifacts } = await this.client.listArtifacts(job.runId, job.fencingToken); | |
| 264 | + return artifacts.map((a) => ({ | |
| 265 | + path: a.path, | |
| 266 | + read: () => this.client.downloadArtifact(job.runId, job.fencingToken, a.path), | |
| 267 | + })); | |
| 268 | + }, | |
| 262 | 269 | }; |
| 263 | 270 | let outcome: Awaited<ReturnType<typeof runResearch>>; |
| 264 | 271 | try { |
modified
tests/integration/orchestration.integration.test.ts
+21 −0
@@ -469,6 +469,27 @@ describe("simulated end-to-end: quality gates and publication", () => { | ||
| 469 | 469 | }); |
| 470 | 470 | }); |
| 471 | 471 | |
| 472 | +describe("resume on another node", () => { | |
| 473 | + it("lets only the lease holder list and download the latest artifacts", async () => { | |
| 474 | + await drainQueue(); | |
| 475 | + const { run } = await launch("key-resume"); | |
| 476 | + const w1 = await enroll("sim-r1"); | |
| 477 | + const l1 = await leaseJob(env.db, w1.worker, ["standard"]); | |
| 478 | + const job = l1.job!; | |
| 479 | + const put = (path: string, content: string) => ingestArtifact(env.db, env.storage, w1.worker, run.id, job.fencingToken, { path, kind: "plan", contentType: "text/markdown", sha256: sha256Hex(content), bytes: Buffer.byteLength(content), stage: "preflight", synthetic: false }, Buffer.from(content)); | |
| 480 | + await put("plan/research-plan.md", "v1"); | |
| 481 | + await put("plan/research-plan.md", "v2 changed"); | |
| 482 | + await put("plan/feasibility.json", "{}"); | |
| 483 | + const list = await listRunArtifactsForWorker(env.db, w1.worker, run.id, job.fencingToken); | |
| 484 | + expect(list.map((a) => `${a.path}@${a.version}`).sort()).toEqual(["plan/feasibility.json@1", "plan/research-plan.md@2"]); | |
| 485 | + const { buf } = await readRunArtifactForWorker(env.db, env.storage, w1.worker, run.id, job.fencingToken, "plan/research-plan.md"); | |
| 486 | + expect(buf.toString()).toBe("v2 changed"); | |
| 487 | + const w2 = await enroll("sim-r2"); | |
| 488 | + await expect(listRunArtifactsForWorker(env.db, w2.worker, run.id, job.fencingToken)).rejects.toThrow(/lease/); | |
| 489 | + await completeAttempt(env.db, w1.worker, run.id, { fencingToken: job.fencingToken, outcome: "cancelled", costUsd: 0 }); | |
| 490 | + }); | |
| 491 | +}); | |
| 492 | + | |
| 472 | 493 | describe("error typing", () => { |
| 473 | 494 | it("AppError carries status codes", () => { |
| 474 | 495 | expect(new AppError("x", "y", 409).status).toBe(409); |
| 475 | 496 | |