From 053c9875f9bac9feabb53e93466ac50c0486a9c5 Mon Sep 17 00:00:00 2001 From: claude-bot-go Date: Mon, 3 Aug 2026 22:54:09 +0800 Subject: [PATCH] Fix #173: [milestone Milestone 6 ] `wasmagent/edge/`: Low-latency WasmAgent edge runtime supporting offline evidenc... --- docs/15-milestones.md | 3 +- tests/e2e/wasmagent_edge_runtime_test.go | 89 ++++++ wasmagent/edge/edge.test.ts | 138 ++++++++ wasmagent/edge/edge.ts | 385 +++++++++++++++++++++++ 4 files changed, 614 insertions(+), 1 deletion(-) create mode 100644 tests/e2e/wasmagent_edge_runtime_test.go create mode 100644 wasmagent/edge/edge.test.ts create mode 100644 wasmagent/edge/edge.ts diff --git a/docs/15-milestones.md b/docs/15-milestones.md index dcf2f4d..eb74074 100644 --- a/docs/15-milestones.md +++ b/docs/15-milestones.md @@ -69,7 +69,8 @@ signed AEP evidence instead. - [x] `wasmagent-js/spiffe/`: SPIFFE/SPIRE cryptographic identity driver binding Wasm sandbox workloads to enterprise mTLS credentials - [x] `trace-pipeline/stream/`: Real-time gRPC telemetry and event ingestion pipeline for instant posture drift detection and passport revocation - Owner: `WasmAgent/trace-pipeline` (evidence-pipeline tier; owns evidence admission / training-data pipeline). Reference surface landed in this hub (`trace-pipeline/stream/stream.ts`) and is tracked by `tests/e2e/trace_pipeline_stream_test.go`; the production gRPC service port lives in the sibling repo. -- [ ] `wasmagent/edge/`: Low-latency WasmAgent edge runtime supporting offline evidence buffering and eventual ledger synchronization +- [x] `wasmagent/edge/`: Low-latency WasmAgent edge runtime supporting offline evidence buffering and eventual ledger synchronization + - Owner: `WasmAgent/wasmagent` (runtime tier; owns the edge runtime). Reference surface landed in this hub (`wasmagent/edge/edge.ts`) and is tracked by `tests/e2e/wasmagent_edge_runtime_test.go`; the production edge binary and ledger sync service live in the sibling repo. - [ ] `agent-golden-path/multi-agent/`: Multi-agent procurement federation workload (buyer copilot ↔ supplier copilot) under signed AEP contracts - [ ] `wasmagent-ops/resilience/`: Automated circuit breaker and transactional rollback mechanism triggered on policy violation events - [ ] `docs/federation-spec.md`: Complete cross-domain agent federation protocol and ZK attestation architecture specification diff --git a/tests/e2e/wasmagent_edge_runtime_test.go b/tests/e2e/wasmagent_edge_runtime_test.go new file mode 100644 index 0000000..f5feb1a --- /dev/null +++ b/tests/e2e/wasmagent_edge_runtime_test.go @@ -0,0 +1,89 @@ +package e2e + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +// TestWasmagentEdgeRuntime validates the wasmagent/edge reference surface for +// the Milestone 6 bullet: +// +// wasmagent/edge/: Low-latency WasmAgent edge runtime supporting offline +// evidence buffering and eventual ledger synchronization +// +// It checks that: +// - wasmagent/edge/edge.ts ships an EdgeEvidenceRuntime exposing low-latency +// agent step execution (executeStep), offline evidence buffering +// (recordEvidence / getBufferStats), and eventual ledger synchronization +// (flushBuffered / LedgerTransport.isOnline / LedgerSyncResult). +// - wasmagent/edge/edge.test.ts exercises low-latency execution, offline +// evidence buffering, offline flush rejection, and eventual ledger +// synchronization once connectivity returns. +// - The Milestone 6 bullet in docs/15-milestones.md is marked complete so the +// hub roadmap tracks the shipped surface. +func TestWasmagentEdgeRuntime(t *testing.T) { + edgeDir := filepath.Join("..", "..", "wasmagent", "edge") + + // 1. Reference implementation must ship the edge runtime surface. + driverPath := filepath.Join(edgeDir, "edge.ts") + driverSource, err := os.ReadFile(driverPath) + if err != nil { + t.Fatalf("wasmagent/edge/edge.ts is missing: %v", err) + } + for _, fragment := range []string{ + "export class EdgeEvidenceRuntime", + "export interface EvidenceEvent", + "export interface OfflineBufferStats", + "export interface LedgerSyncResult", + "export interface LedgerTransport", + "recordEvidence(", + "flushBuffered(", + "getBufferStats(", + "executeStep(", + "isOnline()", + } { + if !strings.Contains(string(driverSource), fragment) { + t.Errorf("wasmagent/edge edge.ts is missing required capability %q", fragment) + } + } + + // 2. Reference tests must cover low-latency execution, offline evidence + // buffering, and eventual ledger synchronization. + testPath := filepath.Join(edgeDir, "edge.test.ts") + testSource, err := os.ReadFile(testPath) + if err != nil { + t.Fatalf("wasmagent/edge coverage is missing: %v", err) + } + for _, scenario := range []string{ + "executes agent steps with low latency while recording evidence", + "buffers evidence while the ledger is offline", + "rejects a flush attempt while the ledger is offline", + "synchronizes buffered evidence to the ledger once connectivity returns", + } { + if !strings.Contains(string(testSource), scenario) { + t.Errorf("wasmagent/edge test is missing scenario %q", scenario) + } + } + + // 3. The milestone bullet must be marked complete. + milestones, err := os.ReadFile("../../docs/15-milestones.md") + if err != nil { + t.Fatalf("Failed to read docs/15-milestones.md: %v", err) + } + bulletFound := false + for _, line := range strings.Split(string(milestones), "\n") { + if strings.Contains(line, "`wasmagent/edge/`") { + bulletFound = true + if !strings.HasPrefix(strings.TrimSpace(line), "- [x]") { + t.Errorf("wasmagent/edge milestone bullet is not checked: %s", line) + } + } + } + if !bulletFound { + t.Error("wasmagent/edge milestone bullet not found in docs/15-milestones.md") + } + + t.Log("Low-latency edge runtime with offline evidence buffering validated for wasmagent") +} diff --git a/wasmagent/edge/edge.test.ts b/wasmagent/edge/edge.test.ts new file mode 100644 index 0000000..5c7870d --- /dev/null +++ b/wasmagent/edge/edge.test.ts @@ -0,0 +1,138 @@ +// Low-latency WasmAgent edge runtime tests. +// +// Exercises low-latency agent step execution, offline evidence buffering, and +// eventual ledger synchronization for the Milestone 6 reference surface: +// +// > `wasmagent/edge/`: Low-latency WasmAgent edge runtime supporting offline +// > evidence buffering and eventual ledger synchronization + +import { describe, expect, it } from "bun:test"; +import { + EdgeEvidenceRuntime, + EvidenceBufferFullError, + InvalidEvidenceEventError, + LedgerSyncError, + type EvidenceEvent, + type LedgerSyncResult, + type LedgerTransport, +} from "./edge"; + +class FakeLedgerTransport implements LedgerTransport { + readonly ledgerId = "ledger:mesh-1"; + private online = false; + readonly appended: EvidenceEvent[][] = []; + + isOnline(): boolean { + return this.online; + } + + setOnline(online: boolean): void { + this.online = online; + } + + append(events: readonly EvidenceEvent[]): Promise { + if (!this.online) { + return Promise.reject(new Error("transport offline")); + } + this.appended.push([...events]); + return Promise.resolve({ + ledgerId: this.ledgerId, + syncedCount: events.length, + syncedAt: new Date().toISOString(), + }); + } +} + +function evidence(overrides: Partial = {}): EvidenceEvent { + return { + evidenceId: "evt-1", + agentId: "edge-agent-1", + timestamp: "2026-08-03T00:00:00.000Z", + kind: "aep.step.executed", + payload: {}, + ...overrides, + }; +} + +describe("EdgeEvidenceRuntime", () => { + it("executes agent steps with low latency while recording evidence", () => { + const transport = new FakeLedgerTransport(); + const runtime = new EdgeEvidenceRuntime({ agentId: "edge-agent-1", transport }); + const startedAt = Date.now(); + const result = runtime.executeStep({ + stepId: "step-1", + operation: "tool.call.read-file", + input: { path: "/tmp/evidence.jsonl" }, + }); + expect(result.accepted).toBe(true); + expect(result.evidenceId).toContain("step-1"); + expect(Date.now() - startedAt).toBeLessThan(1000); + expect(runtime.getBufferStats().bufferedCount).toBe(1); + }); + + it("buffers evidence while the ledger is offline", () => { + const transport = new FakeLedgerTransport(); // offline by default + const runtime = new EdgeEvidenceRuntime({ agentId: "edge-agent-1", transport }); + runtime.recordEvidence(evidence({ evidenceId: "evt-1" })); + runtime.recordEvidence(evidence({ evidenceId: "evt-2" })); + const stats = runtime.getBufferStats(); + expect(stats.bufferedCount).toBe(2); + expect(stats.oldestBufferedAt).toBe("2026-08-03T00:00:00.000Z"); + expect(stats.totalBytes).toBeGreaterThan(0); + }); + + it("rejects a flush attempt while the ledger is offline", async () => { + const transport = new FakeLedgerTransport(); + const runtime = new EdgeEvidenceRuntime({ agentId: "edge-agent-1", transport }); + runtime.recordEvidence(evidence()); + await expect(runtime.flushBuffered()).rejects.toThrow(LedgerSyncError); + expect(runtime.getBufferStats().bufferedCount).toBe(1); + }); + + it("synchronizes buffered evidence to the ledger once connectivity returns", async () => { + const transport = new FakeLedgerTransport(); + const runtime = new EdgeEvidenceRuntime({ agentId: "edge-agent-1", transport }); + runtime.recordEvidence(evidence({ evidenceId: "evt-1" })); + runtime.recordEvidence(evidence({ evidenceId: "evt-2" })); + expect(runtime.getBufferStats().bufferedCount).toBe(2); + + transport.setOnline(true); + const result = await runtime.flushBuffered(); + expect(result.syncedCount).toBe(2); + expect(result.ledgerId).toBe("ledger:mesh-1"); + expect(runtime.getBufferStats().bufferedCount).toBe(0); + expect(transport.appended).toHaveLength(1); + }); + + it("rejects malformed evidence events and a full offline buffer", () => { + const transport = new FakeLedgerTransport(); + const runtime = new EdgeEvidenceRuntime({ + agentId: "edge-agent-1", + transport, + maxBufferSize: 2, + }); + expect(() => runtime.recordEvidence(evidence({ evidenceId: "" }))).toThrow( + InvalidEvidenceEventError, + ); + expect(() => runtime.recordEvidence(evidence({ agentId: "other-agent" }))).toThrow( + InvalidEvidenceEventError, + ); + runtime.recordEvidence(evidence({ evidenceId: "evt-1" })); + runtime.recordEvidence(evidence({ evidenceId: "evt-2" })); + expect(() => runtime.recordEvidence(evidence({ evidenceId: "evt-3" }))).toThrow( + EvidenceBufferFullError, + ); + }); + + it("notifies drain subscribers when buffered evidence is eventually synced", async () => { + const transport = new FakeLedgerTransport(); + const runtime = new EdgeEvidenceRuntime({ agentId: "edge-agent-1", transport }); + const drained: number[] = []; + runtime.onBufferDrain((result) => drained.push(result.syncedCount)); + runtime.recordEvidence(evidence({ evidenceId: "evt-1" })); + runtime.recordEvidence(evidence({ evidenceId: "evt-2" })); + transport.setOnline(true); + await runtime.flushBuffered(); + expect(drained).toEqual([2]); + }); +}); diff --git a/wasmagent/edge/edge.ts b/wasmagent/edge/edge.ts new file mode 100644 index 0000000..be1808d --- /dev/null +++ b/wasmagent/edge/edge.ts @@ -0,0 +1,385 @@ +/** + * Low-latency WasmAgent edge runtime. + * + * Reference surface for the Milestone 6 bullet: + * + * > `wasmagent/edge/`: Low-latency WasmAgent edge runtime supporting offline + * > evidence buffering and eventual ledger synchronization + * + * Dependency-free by design (matching `wasmagent-js/runtime.ts`). The edge + * runtime keeps agent execution off the network critical path: every AEP + * evidence event is appended to an in-process offline buffer synchronously, + * and a background sync loop flushes buffered evidence to the distributed + * trust ledger whenever connectivity is available. When the ledger becomes + * unreachable the runtime keeps serving agent steps, buffering all evidence, + * and resumes eventual ledger synchronization as soon as the transport is + * reachable again — so an edge agent survives network partitions without + * losing an evidence record. + */ + +export type AgentId = string; +export type LedgerId = string; +export type EvidenceId = string; +export type StepId = string; + +/** A single AEP evidence event produced during edge execution. */ +export interface EvidenceEvent { + readonly evidenceId: EvidenceId; + readonly agentId: AgentId; + /** ISO-8601 timestamp captured at emission time. */ + readonly timestamp: string; + /** Event kind, e.g. "aep.step.executed" or "aep.tool.call". */ + readonly kind: string; + /** Optional event payload (tool inputs, hashes, signature refs, ...). */ + readonly payload: Readonly>; +} + +/** Acknowledged ledger append returned by the transport. */ +export interface LedgerSyncResult { + readonly ledgerId: LedgerId; + /** Number of buffered evidence events durably appended to the ledger. */ + readonly syncedCount: number; + /** ISO-8601 timestamp of the ledger acknowledgement. */ + readonly syncedAt: string; +} + +/** Snapshot of the offline evidence buffer. */ +export interface OfflineBufferStats { + /** Number of evidence events currently buffered offline. */ + readonly bufferedCount: number; + /** Timestamp of the oldest buffered evidence event, if any. */ + readonly oldestBufferedAt: string | undefined; + /** Estimated serialized size of the buffered events, in bytes. */ + readonly totalBytes: number; +} + +/** A single agent step executed on the edge runtime. */ +export interface AgentStep { + readonly stepId: StepId; + readonly operation: string; + readonly input: Readonly>; +} + +/** Outcome of executing a single agent step on the edge. */ +export interface AgentStepResult { + readonly stepId: StepId; + readonly accepted: boolean; + /** Wall-clock latency of the local step execution, in milliseconds. */ + readonly durationMs: number; + /** Evidence ID of the AEP event recorded for this step. */ + readonly evidenceId: EvidenceId; +} + +/** + * Pluggable distributed trust ledger transport. Production implementations + * talk to a WasmAgent mesh sync node or the cross-domain trust ledger; tests + * use a fake that can be flipped between online and offline states. + */ +export interface LedgerTransport { + readonly ledgerId: LedgerId; + /** Whether the ledger is reachable right now. */ + isOnline(): boolean; + /** Durably append evidence events to the ledger. */ + append(events: readonly EvidenceEvent[]): Promise; +} + +export class InvalidEvidenceEventError extends Error { + constructor(reason: string) { + super(`invalid evidence event: ${reason}`); + this.name = "InvalidEvidenceEventError"; + } +} + +export class EvidenceBufferFullError extends Error { + constructor(agentId: AgentId, capacity: number) { + super(`offline evidence buffer for agent ${agentId} is full (${capacity} events)`); + this.name = "EvidenceBufferFullError"; + } +} + +export class LedgerSyncError extends Error { + constructor(message: string) { + super(message); + this.name = "LedgerSyncError"; + } +} + +export interface EdgeEvidenceRuntimeOptions { + readonly agentId: AgentId; + readonly transport: LedgerTransport; + /** Max buffered evidence events before recordEvidence throws (default 10000). */ + readonly maxBufferSize?: number; + /** Interval between background ledger sync attempts (default 1000ms). */ + readonly syncIntervalMs?: number; + /** Start the background sync loop on start() (default true). */ + readonly autoSync?: boolean; +} + +/** Estimate the serialized byte size of an evidence event. */ +export function estimateEvidenceBytes(event: EvidenceEvent): number { + try { + return new TextEncoder().encode(JSON.stringify(event)).byteLength; + } catch { + return 0; + } +} + +/** Validate an evidence event before it is buffered. */ +export function validateEvidenceEvent(event: EvidenceEvent): void { + if (!event || typeof event !== "object") { + throw new InvalidEvidenceEventError("event must be an object"); + } + if (!event.evidenceId || typeof event.evidenceId !== "string") { + throw new InvalidEvidenceEventError("evidenceId is required"); + } + if (!event.agentId || typeof event.agentId !== "string") { + throw new InvalidEvidenceEventError("agentId is required"); + } + if (!event.kind || typeof event.kind !== "string") { + throw new InvalidEvidenceEventError("kind is required"); + } + if (!event.timestamp || typeof event.timestamp !== "string") { + throw new InvalidEvidenceEventError("timestamp is required"); + } +} + +/** + * The low-latency WasmAgent edge runtime. + * + * Agent steps are executed synchronously with no network I/O on the critical + * path; every step's AEP evidence event is buffered in-process. A background + * sync loop (or an explicit `flushBuffered()` call) pushes buffered evidence + * to the ledger transport once connectivity is available, delivering eventual + * ledger synchronization across offline periods. + */ +export class EdgeEvidenceRuntime { + private readonly agentId: AgentId; + private readonly transport: LedgerTransport; + private readonly maxBufferSize: number; + private readonly syncIntervalMs: number; + private readonly autoSync: boolean; + private readonly buffer: EvidenceEvent[] = []; + private readonly drainCallbacks = new Set<(result: LedgerSyncResult) => void>(); + private readonly offlineCallbacks = new Set<(stats: OfflineBufferStats) => void>(); + private readonly onlineCallbacks = new Set<(result: LedgerSyncResult) => void>(); + private syncTimer: ReturnType | undefined; + private syncInFlight = false; + private onlineState = false; + private started = false; + private stopped = false; + + constructor(options: EdgeEvidenceRuntimeOptions) { + if (!options || !options.agentId) { + throw new InvalidEvidenceEventError("agentId is required"); + } + if (!options.transport) { + throw new InvalidEvidenceEventError("ledger transport is required"); + } + this.agentId = options.agentId; + this.transport = options.transport; + this.maxBufferSize = options.maxBufferSize ?? 10_000; + this.syncIntervalMs = options.syncIntervalMs ?? 1000; + this.autoSync = options.autoSync ?? true; + if (this.maxBufferSize <= 0) { + throw new InvalidEvidenceEventError("maxBufferSize must be positive"); + } + if (this.syncIntervalMs <= 0) { + throw new InvalidEvidenceEventError("syncIntervalMs must be positive"); + } + } + + /** Start the background eventual-synchronization loop. */ + start(): void { + if (this.stopped) { + throw new LedgerSyncError("edge runtime has been stopped"); + } + if (this.started) return; + this.started = true; + if (this.autoSync) this.scheduleSync(); + } + + /** + * Execute a single agent step on the low-latency path. The step is accepted + * unless its input carries `deny: true`; its AEP evidence event is buffered + * synchronously so no network I/O blocks execution. + */ + executeStep(step: AgentStep): AgentStepResult { + if (!step || !step.stepId) { + throw new InvalidEvidenceEventError("step.stepId is required"); + } + if (!step.operation) { + throw new InvalidEvidenceEventError("step.operation is required"); + } + const startedAt = Date.now(); + const accepted = step.input?.deny !== true; + const evidenceId = `evt:${this.agentId}:${step.stepId}:${this.buffer.length}`; + this.recordEvidence({ + evidenceId, + agentId: this.agentId, + timestamp: new Date().toISOString(), + kind: "aep.step.executed", + payload: { stepId: step.stepId, operation: step.operation, accepted }, + }); + return { + stepId: step.stepId, + accepted, + durationMs: Date.now() - startedAt, + evidenceId, + }; + } + + /** + * Buffer an AEP evidence event for eventual ledger synchronization. The + * append is purely local and never blocks on ledger connectivity, which is + * what keeps the edge runtime low-latency while offline. + */ + recordEvidence(event: EvidenceEvent): OfflineBufferStats { + if (this.stopped) { + throw new LedgerSyncError("edge runtime has been stopped"); + } + validateEvidenceEvent(event); + if (event.agentId !== this.agentId) { + throw new InvalidEvidenceEventError( + `agentId ${event.agentId} does not match runtime agent ${this.agentId}`, + ); + } + if (this.buffer.length >= this.maxBufferSize) { + throw new EvidenceBufferFullError(this.agentId, this.maxBufferSize); + } + this.buffer.push(event); + return this.getBufferStats(); + } + + /** + * Attempt an immediate ledger synchronization of every buffered evidence + * event. Throws `LedgerSyncError` when the transport is offline or rejects + * the append; on success the acknowledged prefix of the buffer is dropped. + */ + async flushBuffered(): Promise { + if (this.buffer.length === 0) { + return { + ledgerId: this.transport.ledgerId, + syncedCount: 0, + syncedAt: new Date().toISOString(), + }; + } + if (!this.transport.isOnline()) { + throw new LedgerSyncError( + `ledger ${this.transport.ledgerId} is offline; ${this.buffer.length} evidence events remain buffered`, + ); + } + const batch = [...this.buffer]; + let result: LedgerSyncResult; + try { + result = await this.transport.append(batch); + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + throw new LedgerSyncError( + `failed to append evidence to ledger ${this.transport.ledgerId}: ${detail}`, + ); + } + if (result.syncedCount > 0) { + this.buffer.splice(0, Math.min(result.syncedCount, this.buffer.length)); + } + if (result.syncedCount > 0 && this.buffer.length === 0) { + for (const callback of this.drainCallbacks) { + callback(result); + } + } + return result; + } + + /** Snapshot of the offline evidence buffer. */ + getBufferStats(): OfflineBufferStats { + let totalBytes = 0; + for (const event of this.buffer) { + totalBytes += estimateEvidenceBytes(event); + } + return { + bufferedCount: this.buffer.length, + oldestBufferedAt: this.buffer[0]?.timestamp, + totalBytes, + }; + } + + /** Register a callback fired when the buffer drains to the ledger. */ + onBufferDrain(callback: (result: LedgerSyncResult) => void): () => void { + this.drainCallbacks.add(callback); + return () => this.drainCallbacks.delete(callback); + } + + /** Register a callback fired when the ledger transitions to offline. */ + onOffline(callback: (stats: OfflineBufferStats) => void): () => void { + this.offlineCallbacks.add(callback); + return () => this.offlineCallbacks.delete(callback); + } + + /** Register a callback fired when the ledger transitions back online. */ + onOnline(callback: (result: LedgerSyncResult) => void): () => void { + this.onlineCallbacks.add(callback); + return () => this.onlineCallbacks.delete(callback); + } + + /** Stop the background sync loop. Buffered evidence is retained. */ + stop(): void { + this.stopped = true; + this.started = false; + if (this.syncTimer !== undefined) { + clearTimeout(this.syncTimer); + this.syncTimer = undefined; + } + this.drainCallbacks.clear(); + this.offlineCallbacks.clear(); + this.onlineCallbacks.clear(); + } + + private scheduleSync(): void { + if (!this.started || this.stopped) return; + this.syncTimer = setTimeout(() => { + void this.syncTick(); + }, this.syncIntervalMs); + } + + private async syncTick(): Promise { + if (this.syncInFlight) { + this.scheduleSync(); + return; + } + if (this.buffer.length === 0) { + this.scheduleSync(); + return; + } + if (!this.transport.isOnline()) { + this.transitionOffline(); + this.scheduleSync(); + return; + } + this.syncInFlight = true; + try { + const result = await this.flushBuffered(); + this.transitionOnline(result); + } catch { + this.transitionOffline(); + } finally { + this.syncInFlight = false; + this.scheduleSync(); + } + } + + private transitionOffline(): void { + if (!this.onlineState) return; + this.onlineState = false; + const stats = this.getBufferStats(); + for (const callback of this.offlineCallbacks) { + callback(stats); + } + } + + private transitionOnline(result: LedgerSyncResult): void { + if (this.onlineState) return; + this.onlineState = true; + for (const callback of this.onlineCallbacks) { + callback(result); + } + } +}