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
- V1 requirements: C4 docs, ADR, AGENTS.md, debug-log.md
- Design documents in docs/

packages/stats/server/Dockerfile

@@ -0,0 +1,32 @@
+FROM oven/bun:1.3.14-alpine AS base
+
+WORKDIR /app
+
+ENV NODE_ENV=production
+ENV BUN_RUNTIME_TRANSPILER_CACHE_PATH=0
+
+FROM base AS pruner
+
+COPY . .
+
+RUN bunx turbo@2.8.13 prune @opencode-ai/stats-server --docker --no-update-notifier --no-color
+
+FROM base AS installer
+
+COPY --from=pruner /app/out/json/ ./
+
+# Bun 1.3.x needs the pruned workspace globs and lockfile metadata refreshed before the frozen production install.
+RUN bun -e 'const packageJson = await Bun.file("package.json").json(); packageJson.workspaces.packages = Array.from(new Bun.Glob("packages/**/package.json").scanSync(".")).map((file) => file.slice(0, -"/package.json".length)).sort(); await Bun.write("package.json", JSON.stringify(packageJson, null, 2) + "\n")'
+RUN rm -f bun.lock && bun install --filter @opencode-ai/stats-server --lockfile-only --ignore-scripts
+RUN bun install --filter @opencode-ai/stats-server --frozen-lockfile --production --ignore-scripts
+
+FROM base AS runner
+
+COPY --from=installer /app ./
+COPY --from=pruner /app/out/full/ ./
+
+WORKDIR /app/packages/stats/server
+
+EXPOSE 3000
+
+CMD ["bun", "src/server.ts"]

packages/stats/server/package.json

@@ -0,0 +1,33 @@
+{
+  "$schema": "https://json.schemastore.org/package.json",
+  "name": "@opencode-ai/stats-server",
+  "version": "1.17.4",
+  "private": true,
+  "type": "module",
+  "license": "MIT",
+  "main": "./src/server.ts",
+  "exports": {
+    ".": "./src/server.ts"
+  },
+  "scripts": {
+    "start": "bun src/server.ts",
+    "typecheck": "tsgo --noEmit"
+  },
+  "dependencies": {
+    "@aws-sdk/client-firehose": "3.933.0",
+    "@effect/platform-node": "catalog:",
+    "@opencode-ai/stats-core": "workspace:*",
+    "effect": "catalog:",
+    "sst": "catalog:"
+  },
+  "devDependencies": {
+    "@tsconfig/node22": "catalog:",
+    "@types/bun": "catalog:",
+    "@types/node": "catalog:",
+    "@typescript/native-preview": "catalog:",
+    "typescript": "catalog:"
+  },
+  "engines": {
+    "node": ">=22"
+  }
+}

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,
+})

packages/stats/server/sst-env.d.ts

@@ -0,0 +1,10 @@
+/* This file is auto-generated by SST. Do not edit. */
+/* tslint:disable */
+/* eslint-disable */
+/* deno-fmt-ignore-file */
+/* biome-ignore-all lint: auto-generated */
+
+/// <reference path="../../../sst-env.d.ts" />
+
+import "sst"
+export {}

packages/stats/server/tsconfig.json

@@ -0,0 +1,12 @@
+{
+  "$schema": "https://json.schemastore.org/tsconfig",
+  "extends": "@tsconfig/node22/tsconfig.json",
+  "compilerOptions": {
+    "module": "ESNext",
+    "moduleResolution": "bundler",
+    "strict": true,
+    "noEmit": true,
+    "types": ["bun", "node"]
+  },
+  "include": ["src", "../core/src/resource.d.ts"]
+}