From 6afef65020b8c9ea0dd75031b47031098a018881 Mon Sep 17 00:00:00 2001 From: Jack Champagne Date: Wed, 29 Jul 2026 06:53:19 -0400 Subject: [PATCH] Forward cloud AMICODE_PULSE into the run inspector mirror Cloud solves surfaced iteration/convergence rows and frames in the run inspector, but the pulse plot stayed empty: the poll pump forwarded AMICODE_ITER (/stats) and PNG frames (/frames) into the mirror run.log, but never the AMICODE_PULSE_META / AMICODE_PULSE lines. The render side (native type-aware pulse plot) already tails these off run.log for local solves; it just never received them for cloud runs. Add a /pulse poll lane that mirrors /stats exactly: relay AMICODE_PULSE_META once (it carries no iter) and each new AMICODE_PULSE line on the iteration high-water (the API re-serves history every poll), verbatim so the a= drive knots AND the d= derivative tail reach the plotter unaltered. Best-effort like stats/frames: a runner/API without /pulse 404s and is swallowed, so this is safe to merge before the cloud endpoint exists. Client-only: does not touch the solve templates / solver output. Requires a cloud companion: GET /solves/{id}/pulse in the solves_poll lambda, ~10 lines mirroring _stats (grep AMICODE_PULSE* out of the S3-synced run.log, return {task_id, pulse: [{raw}], submitter}). --- packages/amico-run/src/remote_executor.ts | 36 +++++++++++++++++++ packages/amico-run/test/fake_cloud.ts | 7 ++++ .../amico-run/test/remote_executor.test.ts | 26 +++++++++++++- 3 files changed, 68 insertions(+), 1 deletion(-) diff --git a/packages/amico-run/src/remote_executor.ts b/packages/amico-run/src/remote_executor.ts index b2d174747..25aa3343d 100644 --- a/packages/amico-run/src/remote_executor.ts +++ b/packages/amico-run/src/remote_executor.ts @@ -142,6 +142,8 @@ export class RemoteExecutor implements Executor { let sawLife = false; // Running status or a first iter observed (warming budget clock) let iterHigh = -1; // stats high-water: Δ4 re-serves history each poll; dedup here let frameHigh = -1; // frames high-water + let pulseHigh = -1; // pulse high-water (AMICODE_PULSE iter=N); dedup like stats + let pulseMetaEmitted = false; // AMICODE_PULSE_META has no iter — relay it once let lastOkPollAt = Date.now(); // resolution (d) client half: observability clock const startedAt = Date.now(); @@ -230,6 +232,40 @@ export class RemoteExecutor implements Executor { } catch { /* stats are advisory — status stays the authoritative lane */ } + // pulse → AMICODE_PULSE_META (once) + AMICODE_PULSE lines: the SAME run.log + // + events delivery as stats, so the inspector's pulse plot updates + // progressively for a cloud run exactly as it does for a local one (the + // render side already tails these off run.log). Mirrors _stats: the cloud + // greps AMICODE_PULSE* out of the S3-synced run.log; the client dedups + // (meta once — it carries no iter; pulse on the iter high-water, since Δ4 + // re-serves history each poll) and relays each new line verbatim so the + // `a=` drive knots AND the `d=` derivative tail reach the plotter unaltered. + // Best-effort: a runner/API without /pulse 404s here and is swallowed, + // exactly like a pre-sidecar stats poll — the mirror just shows no pulse. + try { + const r = await get("pulse"); + if (r.ok && r.status !== 204) { + const body = (await r.json()) as { pulse?: Array<{ raw?: unknown }> }; + for (const p of body.pulse ?? []) { + const raw = typeof p.raw === "string" ? p.raw : undefined; + if (raw === undefined) continue; + if (raw.startsWith("AMICODE_PULSE_META")) { + if (pulseMetaEmitted) continue; // one meta per run (no iter to dedup on) + pulseMetaEmitted = true; + sawLife = true; + emitLine(raw); + continue; + } + const n = Number(/(?:^|\s)iter=(\d+)/.exec(raw)?.[1]); + if (!Number.isFinite(n) || n <= pulseHigh) continue; // dedup on high-water + pulseHigh = n; + sawLife = true; + emitLine(raw); + } + } + } catch { + /* pulse is advisory — status stays the authoritative lane */ + } // frames — resolution (a): best-effort; ANY failure is swallowed try { const r = await get("frames"); diff --git a/packages/amico-run/test/fake_cloud.ts b/packages/amico-run/test/fake_cloud.ts index caad99469..d8b88ba89 100644 --- a/packages/amico-run/test/fake_cloud.ts +++ b/packages/amico-run/test/fake_cloud.ts @@ -21,6 +21,7 @@ export interface FakeState { iters: FakeIter[]; // Δ4 stats: full history each poll (client dedups on high-water) frame?: { iter: number; png_base64: string }; // Δ4 frames: newest only framesBroken?: boolean; // 500 the frames endpoint — resolution (a) lane + pulse?: Array<{ raw: string }>; // Δ4 pulse: AMICODE_PULSE_META + AMICODE_PULSE lines, full history each poll } export class FakeCloud { @@ -97,6 +98,12 @@ export class FakeCloud { if (url === `/solves/${this.taskId}/stats`) { return send(200, { task_id: this.taskId, stats: this.state.iters, submitter: "test" }); } + // pulse mirrors stats' shape: {task_id, pulse: [{raw}], submitter}, where the + // cloud greps AMICODE_PULSE_META + AMICODE_PULSE lines out of the S3 run.log + // (never JSON, so always {raw}). Full history each poll — the client dedups. + if (url === `/solves/${this.taskId}/pulse`) { + return send(200, { task_id: this.taskId, pulse: this.state.pulse ?? [], submitter: "test" }); + } if (url === `/solves/${this.taskId}/frames`) { if (this.state.framesBroken) return send(500, { error: "frames unavailable" }); if (!this.state.frame) return send(204); diff --git a/packages/amico-run/test/remote_executor.test.ts b/packages/amico-run/test/remote_executor.test.ts index 87e5aac66..e81015847 100644 --- a/packages/amico-run/test/remote_executor.test.ts +++ b/packages/amico-run/test/remote_executor.test.ts @@ -134,6 +134,30 @@ describe("poll streaming — stats/frames fed into the mirror (Δ4)", () => { }); }); + it("pulse become AMICODE_PULSE_META + AMICODE_PULSE lines in run.log; meta once, iters deduped", async () => { + await withCloud(async (fake) => { + fake.state.task_status = "Running"; + fake.state.pulse = [ + { raw: 'AMICODE_PULSE_META drives=2 knots=20 labels="a_1","a_2" bounds=-1.0:1.0,-1.0:1.0 interp=cubic' }, + { raw: "AMICODE_PULSE iter=0 dt=0.5 a=0.1,0.2 d=0.0,0.0" }, + { raw: "AMICODE_PULSE iter=1 dt=0.1 a=0.3,0.4 d=0.1,0.1" }, + ]; + const root = tmpRoot(); + const h = await ex(fake).submit(fakeJulia(root, "s.jl", ""), { runsRoot: join(root, "runs") }); + await fake.waitForPolls(4); // several polls over the SAME pulse — dedup must hold + fake.state.finished = { status: "completed" }; + await h.finished; + const log = readFileSync(join(h.runDir, "run.log"), "utf8"); + // meta relayed verbatim (the interp discriminator the plotter keys on), exactly once + expect(log).toContain("AMICODE_PULSE_META drives=2 knots=20"); + expect(log.match(/AMICODE_PULSE_META /g)).toHaveLength(1); + // both frames relayed verbatim — a= drive knots AND the d= derivative tail — deduped + expect(log).toContain("AMICODE_PULSE iter=0 dt=0.5 a=0.1,0.2 d=0.0,0.0"); + expect(log).toContain("AMICODE_PULSE iter=1 dt=0.1 a=0.3,0.4 d=0.1,0.1"); + expect(log.match(/AMICODE_PULSE iter=1 /g)).toHaveLength(1); // not 1 × polls + }); + }); + // Fetched from the frames endpoint's PRESIGNED URL (the live shape) rather than // from inline base64. The name is 5-digit because that is what BOTH the S3 // layout and the local Julia solve write (iter_00007.png); the old 3-digit name @@ -315,7 +339,7 @@ describe("stats records arrive in either shape", () => { await withCloud(async (fake) => { fake.state.task_status = "Running"; // exactly what solves_poll's _stats yields for a template-emitted line - fake.state.iters = [{ raw: "iter=7 f=8.727579e-04 inf_pr=2.670e-09 inf_du=1.838e+02" } as never] + fake.state.iters = [{ raw: "iter=7 f=8.727579e-04 inf_pr=2.670e-09 inf_du=1.838e+02" } as never]; const root = tmpRoot(); const h = await ex(fake).submit(fakeJulia(root, "s.jl", ""), { runsRoot: join(root, "runs") }); await fake.waitForPolls(3); // re-served each poll — dedup must still hold