feat(aircoding): AirCoding V2 baseline — deterministic multi-agent architecture

Forked from OpenCode v1.17.4 with multi-agent system:
- 5 agents: aircoding, scheduler, worker, architect, reviewer
- Deterministic DAG scheduling engine (coordinator_tick)
- Tool whitelists as hard enforcement
- AirCoding validation plugin
- System prompt injection for routing
- V1 requirements: C4 docs, ADR, AGENTS.md, debug-log.md
- Design documents in docs/

packages/stats/server/src/ingest.ts

@@ -0,0 +1,159 @@
+import { Buffer } from "node:buffer"
+import { FirehoseClient, PutRecordBatchCommand } from "@aws-sdk/client-firehose"
+import { Effect, Layer, Schema } from "effect"
+import * as Context from "effect/Context"
+import { Resource } from "sst/resource"
+
+const MAX_FIREHOSE_BATCH_SIZE = 500
+const MAX_FIREHOSE_ATTEMPTS = 3
+const LAKE_TYPE = /^([A-Za-z0-9_]+)\.([A-Za-z0-9_]+)$/
+
+type IngestEvent = Record<string, unknown>
+type LakeRoute = { database: string; table: string }
+type FirehoseRecord = { Data: Uint8Array }
+
+export class IngestError extends Schema.TaggedErrorClass<IngestError>()("IngestError", {
+  message: Schema.String,
+  failed: Schema.Number,
+  cause: Schema.optional(Schema.Defect),
+}) {}
+
+export declare namespace Ingest {
+  export interface Service {
+    readonly write: (events: unknown[]) => Effect.Effect<{ records: number }, IngestError>
+  }
+}
+
+export class Ingest extends Context.Service<Ingest, Ingest.Service>()("@opencode/stats/Ingest") {
+  static readonly layer: Layer.Layer<Ingest> = Layer.effect(
+    Ingest,
+    Effect.sync(() => {
+      const client = new FirehoseClient({})
+
+      const write = Effect.fn("Ingest.write")(function* (events: unknown[]) {
+        if (events.length === 0) return { records: 0 }
+        const counts = countRoutedEvents(events)
+        if (counts.unsupported > 0) {
+          yield* Effect.logWarning(
+            `lake ingest rejected ${JSON.stringify({ records: counts.records, unsupported: counts.unsupported })}`,
+          )
+          return yield* new IngestError({
+            message: "Unsupported lake event type",
+            failed: counts.unsupported,
+          })
+        }
+        if (counts.records === 0) return { records: 0 }
+
+        let batch: FirehoseRecord[] = []
+        let batches = 0
+        let failed = 0
+
+        for (const event of events) {
+          if (!isRecord(event)) continue
+          const route = routeEvent(event)
+          if (!route) continue
+          batch.push(toFirehoseRecord(event, route))
+          if (batch.length < MAX_FIREHOSE_BATCH_SIZE) continue
+          failed += yield* putRecords(client, Resource.LakeIngestConfig.streamName, batch)
+          batches++
+          batch = []
+        }
+
+        if (batch.length > 0) {
+          failed += yield* putRecords(client, Resource.LakeIngestConfig.streamName, batch)
+          batches++
+        }
+
+        if (failed > 0) {
+          yield* Effect.logWarning(`lake ingest incomplete ${JSON.stringify({ records: counts.records, failed })}`)
+          return yield* new IngestError({ message: "Failed to ingest all lake records", failed })
+        }
+
+        yield* Effect.logInfo(`lake ingest complete ${JSON.stringify({ records: counts.records, batches })}`)
+        return { records: counts.records }
+      })
+
+      return Ingest.of({ write })
+    }),
+  )
+}
+
+const putRecords: (
+  client: FirehoseClient,
+  streamName: string,
+  records: FirehoseRecord[],
+  attempt?: number,
+) => Effect.Effect<number, IngestError> = Effect.fn("Ingest.putRecords")(function* (
+  client,
+  streamName,
+  records,
+  attempt = 1,
+) {
+  const result = yield* Effect.tryPromise({
+    try: () => client.send(new PutRecordBatchCommand({ DeliveryStreamName: streamName, Records: records })),
+    catch: (cause) =>
+      new IngestError({ message: "Failed to write lake records to Firehose", failed: records.length, cause }),
+  }).pipe(
+    Effect.tapError(() =>
+      Effect.logWarning(`firehose batch write failed ${JSON.stringify({ records: records.length, attempt })}`),
+    ),
+  )
+  const failed =
+    result.RequestResponses?.flatMap((item, index) => {
+      const record = records[index]
+      if (!item.ErrorCode || !record) return []
+      return [record]
+    }) ?? []
+
+  if (failed.length === 0) return 0
+  if (attempt >= MAX_FIREHOSE_ATTEMPTS) {
+    yield* Effect.logWarning(
+      `firehose batch failed ${JSON.stringify({ records: failed.length, attempts: MAX_FIREHOSE_ATTEMPTS })}`,
+    )
+    return failed.length
+  }
+
+  yield* Effect.logWarning(
+    `firehose batch retrying ${JSON.stringify({ records: failed.length, attempt: attempt + 1 })}`,
+  )
+  yield* Effect.sleep(`${250 * 2 ** (attempt - 1)} millis`)
+  return yield* putRecords(client, streamName, failed, attempt + 1)
+})
+
+function countRoutedEvents(events: unknown[]) {
+  let records = 0
+  let unsupported = 0
+  for (const event of events) {
+    if (!isRecord(event)) continue
+    if (routeEvent(event)) records++
+    else unsupported++
+  }
+  return { records, unsupported }
+}
+
+function isRecord(item: unknown): item is IngestEvent {
+  return Boolean(item) && typeof item === "object" && !Array.isArray(item)
+}
+
+function routeEvent(event: IngestEvent): LakeRoute | undefined {
+  if (typeof event._datalake_key !== "string") return
+  const match = event._datalake_key.match(LAKE_TYPE)
+  if (!match?.[1] || !match[2]) return
+  return {
+    database: match[1],
+    table: match[2],
+  }
+}
+
+function toFirehoseRecord(event: IngestEvent, route: LakeRoute): FirehoseRecord {
+  return {
+    Data: Buffer.from(
+      JSON.stringify({
+        ...Object.fromEntries(Object.entries(event).filter(([key]) => key !== "_datalake_key")),
+        _lake_database: route.database,
+        _lake_table: route.table,
+        _lake_operation: "insert" as const,
+      }),
+    ),
+  }
+}

packages/stats/server/src/resource.d.ts

@@ -0,0 +1,11 @@
+import "sst/resource"
+
+declare module "sst/resource" {
+  export interface Resource {
+    LakeIngestConfig: {
+      secret: string
+      streamName: string
+      type: "sst.sst.Linkable"
+    }
+  }
+}

packages/stats/server/src/router.ts

@@ -0,0 +1,73 @@
+import { Buffer } from "node:buffer"
+import { timingSafeEqual } from "node:crypto"
+import { Effect, Schema } from "effect"
+import * as Semaphore from "effect/Semaphore"
+import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"
+import { Resource } from "sst/resource"
+import { Ingest } from "./ingest"
+import { isShuttingDown } from "./shutdown"
+
+const MAX_CONCURRENT_INGEST_REQUESTS = 8
+
+const IngestPayload = Schema.Struct({
+  events: Schema.optional(Schema.Unknown),
+})
+
+export const Routes = HttpRouter.use((router) =>
+  Effect.gen(function* () {
+    const ingestService = yield* Ingest
+    const ingestRequests = yield* Semaphore.make(MAX_CONCURRENT_INGEST_REQUESTS)
+
+    yield* Effect.all(
+      [
+        router.add("GET", "/health", () => json(200, { ok: true })),
+        router.add("GET", "/ready", () => json(isShuttingDown() ? 503 : 200, { ok: !isShuttingDown() })),
+        router.add("POST", "/", ingestRequests.withPermit(ingest(ingestService))),
+      ],
+      { discard: true },
+    )
+  }),
+)
+
+const ingest = (ingestService: Ingest.Service) =>
+  Effect.gen(function* () {
+    const request = yield* HttpServerRequest.HttpServerRequest
+    if (!isAuthorized(request.headers)) return yield* json(401, { ok: false, error: "Unauthorized" })
+
+    const payload = yield* HttpServerRequest.schemaBodyJson(IngestPayload).pipe(
+      Effect.match({
+        onFailure: () => undefined,
+        onSuccess: (value) => value,
+      }),
+    )
+    if (!payload) return yield* json(400, { ok: false, error: "Invalid JSON body" })
+
+    const events = Array.isArray(payload.events) ? payload.events : []
+    if (events.length === 0) return yield* json(202, { ok: true, records: 0 })
+
+    return yield* ingestService.write(events).pipe(
+      Effect.flatMap((result) => json(202, { ok: true, records: result.records })),
+      Effect.catchTag("IngestError", (error) =>
+        json(502, { ok: false, records: countRecords(events), failed: error.failed }),
+      ),
+    )
+  })
+
+function isAuthorized(headers: Record<string, string | undefined>) {
+  const actual = Buffer.from(headers.authorization ?? headers.Authorization ?? "")
+  const expected = Buffer.from(`Bearer ${Resource.LakeIngestConfig.secret}`)
+  if (actual.length !== expected.length) return false
+  return timingSafeEqual(actual, expected)
+}
+
+function countRecords(items: unknown[]) {
+  let records = 0
+  for (const item of items) {
+    if (Boolean(item) && typeof item === "object" && !Array.isArray(item)) records++
+  }
+  return records
+}
+
+function json(status: number, body: Record<string, unknown>) {
+  return HttpServerResponse.json(body, { status }).pipe(Effect.orDie)
+}

packages/stats/server/src/server.ts

@@ -0,0 +1,28 @@
+import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer"
+import * as NodeRuntime from "@effect/platform-node/NodeRuntime"
+import { Config, Layer } from "effect"
+import { HttpRouter } from "effect/unstable/http"
+import { createServer } from "node:http"
+import { Ingest } from "./ingest"
+import { Routes } from "./router"
+import { registerShutdownSignalHandlers } from "./shutdown"
+
+registerShutdownSignalHandlers()
+
+const ServerLive = NodeHttpServer.layerConfig(
+  () => createServer(),
+  Config.all({
+    port: Config.number("PORT").pipe(Config.withDefault(3000)),
+    host: Config.string("HOST").pipe(Config.withDefault("0.0.0.0")),
+  }),
+)
+
+const runtimeLayer = Ingest.layer
+const programLayer = Routes.pipe(Layer.provide(runtimeLayer))
+const main = Layer.launch(
+  HttpRouter.serve(programLayer, {
+    disableLogger: true,
+  }).pipe(Layer.provideMerge(ServerLive)),
+)
+
+NodeRuntime.runMain(main, { disableErrorReporting: true })

packages/stats/server/src/shutdown.ts

@@ -0,0 +1,17 @@
+let shuttingDown = false
+let signalHandlersRegistered = false
+
+export function isShuttingDown() {
+  return shuttingDown
+}
+
+export function registerShutdownSignalHandlers() {
+  if (signalHandlersRegistered) return
+  signalHandlersRegistered = true
+  process.once("SIGTERM", markShuttingDown)
+  process.once("SIGINT", markShuttingDown)
+}
+
+function markShuttingDown() {
+  shuttingDown = true
+}

packages/stats/server/src/stat-sync.ts

@@ -0,0 +1,22 @@
+import * as NodeRuntime from "@effect/platform-node/NodeRuntime"
+import { Athena } from "@opencode-ai/stats-core/athena"
+import { layer as statsLayer } from "@opencode-ai/stats-core/runtime"
+import { syncStats } from "@opencode-ai/stats-core/stat-sync"
+import { Cause, Effect, Layer, Schedule } from "effect"
+
+const SYNC_INTERVAL = "1 hour"
+
+const runtimeLayer = Layer.mergeAll(statsLayer, Athena.layer)
+const syncPass = syncStats().pipe(
+  Effect.catchCause((cause) =>
+    Effect.logWarning(`stats sync failed ${JSON.stringify({ cause: Cause.pretty(cause) })}`),
+  ),
+)
+const daemon = Effect.logInfo("stats sync daemon started").pipe(
+  Effect.andThen(syncPass.pipe(Effect.repeat(Schedule.fixed(SYNC_INTERVAL)))),
+  Effect.forkScoped,
+)
+
+NodeRuntime.runMain(Layer.launch(Layer.effectDiscard(daemon).pipe(Layer.provide(runtimeLayer))), {
+  disableErrorReporting: true,
+})