Forward cloud AMICODE_PULSE into the run inspector

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");

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);

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