feat(eve): buffer action progress lifecycle

.changeset/calm-actions-progress.md

@@ -0,0 +1,5 @@
+---
+"eve": patch
+---
+
+Run progress reporting in an independent best-effort collector, with deterministic root, delegated work, action, and human-blocker lifecycle across local and remote agents.

packages/eve/src/execution/progress-event-buffer.test.ts

@@ -0,0 +1,28 @@
+import { describe, expect, it, vi } from "vitest";
+
+import { createProgressEventBuffer } from "#execution/progress-event-buffer.js";
+import { MAX_PROGRESS_EVENTS_PER_BATCH } from "#execution/session-progress.js";
+
+const event = (index: number) => ({
+  eventId: `event:${String(index)}`,
+  kind: "work.settled" as const,
+  outcome: "completed" as const,
+  settledAt: "now",
+  workId: `work:${String(index)}`,
+});
+
+describe("progress event buffer", () => {
+  it("submits bounded batches and flushes the remainder", async () => {
+    const submit = vi.fn().mockResolvedValue(undefined);
+    const buffer = createProgressEventBuffer({ submit });
+    const events = Array.from({ length: MAX_PROGRESS_EVENTS_PER_BATCH + 2 }, (_, index) =>
+      event(index),
+    );
+    await buffer.push(events);
+    expect(submit).toHaveBeenCalledOnce();
+    expect(submit.mock.calls[0]?.[0]).toHaveLength(MAX_PROGRESS_EVENTS_PER_BATCH);
+    await buffer.flush();
+    expect(submit).toHaveBeenCalledTimes(2);
+    expect(submit.mock.calls[1]?.[0]).toEqual(events.slice(MAX_PROGRESS_EVENTS_PER_BATCH));
+  });
+});

packages/eve/src/execution/progress-event-buffer.ts

@@ -0,0 +1,35 @@
+import {
+  MAX_PROGRESS_EVENTS_PER_BATCH,
+  type ProgressEventV1,
+} from "#execution/session-progress.js";
+
+export interface ProgressEventBuffer {
+  flush(): Promise<void>;
+  push(events: readonly ProgressEventV1[]): Promise<void>;
+}
+
+/** Batches deterministic progress events by size and explicit lifecycle boundaries. */
+export function createProgressEventBuffer(input: {
+  readonly submit: (events: readonly ProgressEventV1[]) => Promise<void>;
+}): ProgressEventBuffer {
+  const pending: ProgressEventV1[] = [];
+  const submitNext = async (count: number): Promise<void> => {
+    const events = pending.slice(0, count);
+    await input.submit(events);
+    pending.splice(0, events.length);
+  };
+  const flush = async (): Promise<void> => {
+    while (pending.length > 0) {
+      await submitNext(Math.min(pending.length, MAX_PROGRESS_EVENTS_PER_BATCH));
+    }
+  };
+  return {
+    flush,
+    async push(events) {
+      pending.push(...events);
+      while (pending.length >= MAX_PROGRESS_EVENTS_PER_BATCH) {
+        await submitNext(MAX_PROGRESS_EVENTS_PER_BATCH);
+      }
+    },
+  };
+}

packages/eve/src/execution/progress-event-observer.ts

@@ -1,10 +1,11 @@
 import type { ContextContainer } from "#context/container.js";
 import { ProgressCallbackKey, ProgressLineageKey } from "#context/keys.js";
-import { projectActionProgressEvents } from "#execution/progress-action-events.js";
-import { reportProgress } from "#execution/submit-progress.js";
 import { activeTurnId } from "#harness/active-turn-id.js";
 import type { HarnessEmissionState } from "#harness/emission.js";
 import type { MessageStreamEvent } from "#protocol/message.js";
+import { projectActionProgressEvents } from "#execution/progress-action-events.js";
+import { createProgressEventBuffer } from "#execution/progress-event-buffer.js";
+import { reportProgress } from "#execution/submit-progress.js";
 
 export interface ProgressEventObserver {
   flush(): Promise<void>;
@@ -33,27 +34,28 @@ export function createProgressEventObserver(
       : undefined);
   if (lineage === undefined) return undefined;
 
+  const buffer = createProgressEventBuffer({
+    async submit(events) {
+      await reportProgress({ callback, events });
+    },
+  });
   let started = false;
   const ensureStarted = async (at: string): Promise<void> => {
     if (started || lineage.kind !== "root-turn") return;
     started = true;
-    await reportProgress({
-      callback,
-      events: [
-        { eventId: `${lineage.id}:started`, kind: "work.started", startedAt: at, work: lineage },
-      ],
-    });
+    await buffer.push([
+      { eventId: `${lineage.id}:started`, kind: "work.started", startedAt: at, work: lineage },
+    ]);
   };
   return {
     async flush() {
       if (!started) await ensureStarted(new Date().toISOString());
+      await buffer.flush();
     },
     async observe(event) {
       await ensureStarted(event.meta.at);
-      await reportProgress({
-        callback,
-        events: projectActionProgressEvents({ at: event.meta.at, event, lineage }),
-      });
+      await buffer.push(projectActionProgressEvents({ at: event.meta.at, event, lineage }));
+      if (event.type === "actions.requested") await buffer.flush();
     },
   };
 }