Sync all parallel development: NATS, search, x402, outbox, worker

Major additions from parallel agent:
- NATS JetStream event bus (lib/nats.ts, services/outbox.ts)
- Meilisearch integration (services/search.ts)
- Real x402 USDC transfers via viem (services/x402.ts, escrow.ts)
- Background worker (worker.ts)
- Updated health endpoint with NATS/search status
- render.yaml updates
- CLI and SDK improvements
- Spec renamed to spec.md

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: waydelyle

.env.example

@@ -22,6 +22,9 @@ ED25519_CHALLENGE_TTL=300
 PLATFORM_WALLET_ADDRESS=
 PLATFORM_WALLET_PRIVATE_KEY=
 X402_NETWORK=base-sepolia
+X402_FACILITATOR_URL=https://x402.org/facilitator
+EVM_RPC_URL=
+USDC_CONTRACT_ADDRESS=
 
 # ──────────────────────────────────────────────
 # Coinbase AgentKit (optional for MVP)
@@ -39,6 +42,11 @@ PLATFORM_URL=http://localhost:3100
 # Server / Environment
 # ──────────────────────────────────────────────
 NODE_ENV=development
+ENABLE_EVENT_OUTBOX=1
+NATS_URL=nats://localhost:4222
+OUTBOX_POLL_INTERVAL_MS=2000
+MEILISEARCH_URL=http://localhost:7700
+MEILISEARCH_API_KEY=localdev
 
 # Comma-separated list of allowed CORS origins
 CORS_ORIGINS=http://localhost:3200

README.md

@@ -35,11 +35,22 @@ swarmdock submit <task-id> --file ./submission.json
 swarmdock dispute <task-id> --reason "Artifacts do not match the requested deliverable"
 ```
 
+For x402-backed task funding and approval flows, configure both agent auth and payment signing:
+
+```bash
+export SWARMDOCK_AGENT_PRIVATE_KEY=...
+export SWARMDOCK_WALLET_PRIVATE_KEY=0x...
+export SWARMDOCK_WALLET_ADDRESS=0x...
+```
+
 ## Local Development
 
 ```bash
+docker compose up -d
 pnpm install
 pnpm type-check
 pnpm build
 pnpm dev
 ```
+
+The local stack now includes Postgres, Redis, NATS JetStream, and Meilisearch. Copy `.env.example` to `.env` and set the x402/Base Sepolia values before testing real payment flows.

docker-compose.yml

@@ -17,6 +17,28 @@ services:
     volumes:
       - redisdata:/data
 
+  nats:
+    image: nats:2-alpine
+    ports:
+      - "4222:4222"
+      - "8222:8222"
+    command:
+      - "--jetstream"
+      - "--store_dir=/data"
+    volumes:
+      - natsdata:/data
+
+  meilisearch:
+    image: getmeili/meilisearch:v1.12
+    ports:
+      - "7700:7700"
+    environment:
+      MEILI_MASTER_KEY: localdev
+    volumes:
+      - meilidata:/meili_data
+
 volumes:
   pgdata:
   redisdata:
+  natsdata:
+  meilidata:

packages/api/package.json

@@ -7,7 +7,8 @@
     "dev": "tsx watch src/index.ts",
     "build": "tsc",
     "start": "node dist/index.js",
-    "test": "node --import ./node_modules/tsx/dist/loader.mjs --test tests/**/*.test.ts",
+    "worker": "node dist/worker.js",
+    "test": "ENABLE_EVENT_OUTBOX=0 node --import ./node_modules/tsx/dist/loader.mjs --test tests/**/*.test.ts",
     "type-check": "tsc --noEmit",
     "db:generate": "drizzle-kit generate",
     "db:push": "drizzle-kit push",

packages/api/src/db/schema.ts

@@ -147,6 +147,21 @@ export const disputes = pgTable('disputes', {
   updatedAt: timestamp('updated_at', { withTimezone: true }).defaultNow().notNull(),
 });
 
+export const eventOutbox = pgTable('event_outbox', {
+  id: uuid('id').primaryKey().defaultRandom(),
+  subject: text('subject').notNull(),
+  target: text('target').notNull(),
+  agentId: uuid('agent_id').references(() => agents.id),
+  eventType: text('event_type').notNull(),
+  payload: jsonb('payload').notNull(),
+  status: text('status').default('pending').notNull(),
+  attempts: integer('attempts').default(0).notNull(),
+  lastError: text('last_error'),
+  publishedAt: timestamp('published_at', { withTimezone: true }),
+  createdAt: timestamp('created_at', { withTimezone: true }).defaultNow().notNull(),
+  updatedAt: timestamp('updated_at', { withTimezone: true }).defaultNow().notNull(),
+});
+
 export const challenges = pgTable('challenges', {
   id: uuid('id').primaryKey().defaultRandom(),
   publicKey: text('public_key').notNull(),

packages/api/src/index.ts

@@ -13,6 +13,7 @@ import adminRoutes from './routes/admin.js';
 import artifactRoutes from './routes/artifacts.js';
 import a2aRoutes from './routes/a2a.js';
 import { getAgentCardById } from './services/agent-card.js';
+import { eventBus } from './lib/events.js';
 
 // Fix BigInt JSON serialization (Drizzle returns bigint columns as JS BigInt)
 (BigInt.prototype as unknown as { toJSON: () => string }).toJSON = function () {
@@ -84,4 +85,7 @@ if (process.env.NODE_ENV === 'production' && !process.env.JWT_SECRET) {
 }
 
 console.log(`SwarmDock API starting on port ${port}`);
+void eventBus.startTransportBridge().catch((error) => {
+  console.error('[EVENTS] failed to start NATS transport bridge:', error);
+});
 serve({ fetch: app.fetch, port });

packages/api/src/lib/events.ts

@@ -1,7 +1,20 @@
-type EventCallback = (event: { type: string; data: unknown }) => void;
+import { enqueueOutboxEvent, OUTBOX_TARGET, type EventEnvelope } from '../services/outbox.js';
+import { subscribeNatsEvents } from './nats.js';
+
+type EventCallback = (event: EventEnvelope) => void | Promise<void>;
+
+const INSTANCE_ID = crypto.randomUUID();
+const RECENT_EVENT_TTL_MS = 5 * 60 * 1000;
+
+function buildSubject(target: 'agent' | 'broadcast', agentId?: string) {
+  return target === 'agent' && agentId ? `events.agent.${agentId}` : 'events.broadcast';
+}
 
 class EventBus {
   private listeners = new Map<string, Set<EventCallback>>();
+  private recentEventIds = new Map<string, number>();
+  private bridgeStarted = false;
+  private unsubscribeBridge: (() => void) | null = null;
 
   subscribe(agentId: string, callback: EventCallback): () => void {
     if (!this.listeners.has(agentId)) {
@@ -17,21 +30,116 @@ class EventBus {
     };
   }
 
-  emit(agentId: string, event: { type: string; data: unknown }): void {
+  async startTransportBridge(): Promise<void> {
+    if (this.bridgeStarted) {
+      return;
+    }
+
+    this.bridgeStarted = true;
+    this.unsubscribeBridge = await subscribeNatsEvents(async (_subject, event) => {
+      if (event.originInstanceId === INSTANCE_ID) {
+        return;
+      }
+
+      if (event.outboxId && this.isDuplicate(event.outboxId)) {
+        return;
+      }
+
+      this.dispatch(event);
+    });
+  }
+
+  stopTransportBridge(): void {
+    this.unsubscribeBridge?.();
+    this.unsubscribeBridge = null;
+    this.bridgeStarted = false;
+  }
+
+  emit(agentId: string, event: { type: string; data: Record<string, unknown> }): void {
+    const envelope: EventEnvelope = {
+      type: event.type,
+      data: event.data,
+      timestamp: new Date().toISOString(),
+      originInstanceId: INSTANCE_ID,
+      target: OUTBOX_TARGET.AGENT,
+      agentId,
+    };
+
+    this.dispatchToAgent(agentId, envelope);
+    void enqueueOutboxEvent({
+      subject: buildSubject('agent', agentId),
+      target: OUTBOX_TARGET.AGENT,
+      agentId,
+      type: event.type,
+      envelope,
+    }).catch((error) => {
+      console.error('[EVENTS] failed to enqueue agent event:', error);
+    });
+  }
+
+  broadcast(event: { type: string; data: Record<string, unknown> }): void {
+    const envelope: EventEnvelope = {
+      type: event.type,
+      data: event.data,
+      timestamp: new Date().toISOString(),
+      originInstanceId: INSTANCE_ID,
+      target: OUTBOX_TARGET.BROADCAST,
+      agentId: null,
+    };
+
+    this.dispatchBroadcast(envelope);
+    void enqueueOutboxEvent({
+      subject: buildSubject('broadcast'),
+      target: OUTBOX_TARGET.BROADCAST,
+      type: event.type,
+      envelope,
+    }).catch((error) => {
+      console.error('[EVENTS] failed to enqueue broadcast event:', error);
+    });
+  }
+
+  private dispatch(event: EventEnvelope): void {
+    if (event.target === OUTBOX_TARGET.AGENT && event.agentId) {
+      this.dispatchToAgent(event.agentId, event);
+      return;
+    }
+
+    this.dispatchBroadcast(event);
+  }
+
+  private dispatchToAgent(agentId: string, event: EventEnvelope): void {
     const callbacks = this.listeners.get(agentId);
-    if (callbacks) {
-      for (const cb of callbacks) {
-        cb(event);
+    if (!callbacks) {
+      return;
+    }
+
+    for (const callback of callbacks) {
+      void callback(event);
+    }
+  }
+
+  private dispatchBroadcast(event: EventEnvelope): void {
+    for (const callbacks of this.listeners.values()) {
+      for (const callback of callbacks) {
+        void callback(event);
       }
     }
   }
 
-  broadcast(event: { type: string; data: unknown }): void {
-    for (const [, callbacks] of this.listeners) {
-      for (const cb of callbacks) {
-        cb(event);
+  private isDuplicate(outboxId: string): boolean {
+    const now = Date.now();
+    for (const [seenId, seenAt] of this.recentEventIds.entries()) {
+      if (now - seenAt > RECENT_EVENT_TTL_MS) {
+        this.recentEventIds.delete(seenId);
       }
     }
+
+    if (this.recentEventIds.has(outboxId)) {
+      return true;
+    }
+
+    this.recentEventIds.set(outboxId, now);
+    return false;
   }
 }
 

packages/api/src/lib/nats.ts

@@ -0,0 +1,76 @@
+import { JSONCodec, connect, type NatsConnection, type Subscription } from 'nats';
+import type { EventEnvelope } from '../services/outbox.js';
+
+const codec = JSONCodec<EventEnvelope>();
+let connectionPromise: Promise<NatsConnection | null> | null = null;
+
+export function isNatsConfigured(): boolean {
+  return Boolean(process.env.NATS_URL?.trim());
+}
+
+export async function getNatsConnection(): Promise<NatsConnection | null> {
+  if (!isNatsConfigured()) {
+    return null;
+  }
+
+  if (!connectionPromise) {
+    connectionPromise = connect({
+      servers: process.env.NATS_URL!,
+      name: process.env.NATS_CLIENT_NAME ?? 'swarmdock-api',
+    }).catch((error) => {
+      console.error('[NATS] connection failed:', error);
+      connectionPromise = null;
+      return null;
+    });
+  }
+
+  return connectionPromise;
+}
+
+export async function publishNatsEvent(subject: string, event: EventEnvelope): Promise<boolean> {
+  const nc = await getNatsConnection();
+  if (!nc) {
+    return false;
+  }
+
+  nc.publish(subject, codec.encode(event));
+  return true;
+}
+
+export async function subscribeNatsEvents(
+  onEvent: (subject: string, event: EventEnvelope) => void | Promise<void>,
+): Promise<() => void> {
+  const nc = await getNatsConnection();
+  if (!nc) {
+    return () => {};
+  }
+
+  const subscriptions: Subscription[] = [
+    nc.subscribe('events.agent.*'),
+    nc.subscribe('events.broadcast'),
+  ];
+
+  let closed = false;
+  for (const subscription of subscriptions) {
+    void (async () => {
+      for await (const message of subscription) {
+        if (closed) {
+          break;
+        }
+
+        try {
+          await onEvent(message.subject, codec.decode(message.data));
+        } catch (error) {
+          console.error('[NATS] event handler failed:', error);
+        }
+      }
+    })();
+  }
+
+  return () => {
+    closed = true;
+    for (const subscription of subscriptions) {
+      subscription.unsubscribe();
+    }
+  };
+}