AHU GPU/LLM Gateway (P0+P1) Implementation Plan

cmd/gateway/main.go            wiring: config → registry → pools → emitter → server
internal/config/config.go      YAML config + validation (registry entries, budgets, tenants)
internal/registry/registry.go  model→upstream resolution, endpoint health, prober
internal/pool/pool.go          Pool/Ticket interfaces + LocalPool (priority heap)
internal/pool/redispool.go     Redis-coordinated pool (lease ZSET + Lua), fail-open
internal/audit/event.go        audit event schema v1
internal/audit/emitter.go      buffered emit → XADD, disk spool + replay
internal/idem/cache.go         idempotency result cache (Redis, non-streaming)
internal/proxy/handler.go      /v1 + /synthesis/v1 handler: parse, admit, forward
internal/proxy/stream.go       SSE piping, tee, queue events, keepalives
internal/proxy/stats.go        EWMA upstream latency → ETA
internal/server/server.go      mux, middleware, healthz, queue status, metrics, drain
deploy/gateway.example.yaml    x056-matching example config
deploy/compose.yaml            gateway + redis compose stack
Dockerfile

go version || (curl -sL https://go.dev/dl/go1.23.4.linux-amd64.tar.gz | sudo tar -C /usr/local -xz && export PATH=$PATH:/usr/local/go/bin)
cd /home/efran/remote-development/poc-ahu-ai/ahu-gpu-manager
go mod init datahive.id/ahu-gpu-manager

package config

import (
    "os"
    "path/filepath"
    "testing"
)

const sample = `
listen: ":8200"
allow_external_upstreams: true
coordination: local
redis_url: "redis://127.0.0.1:6379/0"
audit: {stream_key: "ahu.ai.audit", spool_dir: "/tmp/spool", body_cap_bytes: 10485760}
hold_budgets_ms: {interactive: 45000, batch: 120000, system: 10000}
route_prefixes: {synthesis: ext-397b}
tenants:
  - {id: ahu-chatbot, token: "tok-chatbot"}
upstreams:
  - id: qwen-35b
    type: llm-openai
    class: on_prem
    endpoints: ["http://192.168.83.20:8001/v1"]
    models: ["Qwen/Qwen3.6-35B-A3B-FP8", "qwen-35b"]
    slots: {total: 16, batch_max: 8}
  - id: ext-397b
    type: llm-openai
    class: external_dev
    endpoints: ["https://example.com/v1"]
    api_key_env: DASHSCOPE_API_KEY
    models: ["qwen3.5-397b-a17b"]
    slots: {total: 8, batch_max: 4}
`

func write(t *testing.T, s string) string {
    t.Helper()
    p := filepath.Join(t.TempDir(), "g.yaml")
    if err := os.WriteFile(p, []byte(s), 0o644); err != nil {
        t.Fatal(err)
    }
    return p
}

func TestLoadValid(t *testing.T) {
    c, err := Load(write(t, sample))
    if err != nil {
        t.Fatal(err)
    }
    if c.Listen != ":8200" || len(c.Upstreams) != 2 {
        t.Fatalf("bad parse: %+v", c)
    }
    if c.HoldBudgetsMs[ClassInteractive] != 45000 {
        t.Fatalf("budget: %v", c.HoldBudgetsMs)
    }
    if c.Upstreams[0].ProbePath != "/models" {
        t.Fatalf("default probe path missing: %q", c.Upstreams[0].ProbePath)
    }
}

func TestLoadRejectsBadClassAndDupModel(t *testing.T) {
    if _, err := Load(write(t, `listen: ":1"
upstreams: [{id: a, type: llm-openai, class: cloud, endpoints: ["http://x"], models: ["m"], slots: {total: 1, batch_max: 1}}]`)); err == nil {
        t.Fatal("bad class accepted")
    }
    if _, err := Load(write(t, `listen: ":1"
upstreams:
  - {id: a, type: llm-openai, class: on_prem, endpoints: ["http://x"], models: ["m"], slots: {total: 1, batch_max: 1}}
  - {id: b, type: llm-openai, class: on_prem, endpoints: ["http://y"], models: ["m"], slots: {total: 1, batch_max: 1}}`)); err == nil {
        t.Fatal("duplicate model accepted")
    }
}

package config

import (
    "fmt"
    "os"

    "gopkg.in/yaml.v3"
)

type Class string

const (
    ClassInteractive Class = "interactive"
    ClassBatch       Class = "batch"
    ClassSystem      Class = "system"
)

type AuditConfig struct {
    StreamKey    string `yaml:"stream_key"`
    SpoolDir     string `yaml:"spool_dir"`
    BodyCapBytes int    `yaml:"body_cap_bytes"`
}

type Tenant struct {
    ID    string `yaml:"id"`
    Token string `yaml:"token"`
}

type Slots struct {
    Total    int `yaml:"total"`
    BatchMax int `yaml:"batch_max"`
}

type Upstream struct {
    ID        string   `yaml:"id"`
    Type      string   `yaml:"type"`  // llm-openai | embedding | vlm | ocr-http
    Class     string   `yaml:"class"` // on_prem | external_dev
    Endpoints []string `yaml:"endpoints"`
    Models    []string `yaml:"models"`
    APIKeyEnv string   `yaml:"api_key_env"`
    ProbePath string   `yaml:"probe_path"`
    Slots     Slots    `yaml:"slots"`
}

type Config struct {
    Listen                 string            `yaml:"listen"`
    AllowExternalUpstreams bool              `yaml:"allow_external_upstreams"`
    Coordination           string            `yaml:"coordination"` // local | redis
    RedisURL               string            `yaml:"redis_url"`
    MaxBodyBytes           int64             `yaml:"max_body_bytes"`
    Audit                  AuditConfig       `yaml:"audit"`
    HoldBudgetsMs          map[Class]int     `yaml:"hold_budgets_ms"`
    RoutePrefixes          map[string]string `yaml:"route_prefixes"`
    Tenants                []Tenant          `yaml:"tenants"`
    Upstreams              []Upstream        `yaml:"upstreams"`
}

func Load(path string) (*Config, error) {
    b, err := os.ReadFile(path)
    if err != nil {
        return nil, err
    }
    var c Config
    if err := yaml.Unmarshal(b, &c); err != nil {
        return nil, err
    }
    if c.MaxBodyBytes == 0 {
        c.MaxBodyBytes = 32 << 20
    }
    if c.Audit.BodyCapBytes == 0 {
        c.Audit.BodyCapBytes = 10485760
    }
    if c.HoldBudgetsMs == nil {
        c.HoldBudgetsMs = map[Class]int{}
    }
    for cl, def := range map[Class]int{ClassInteractive: 45000, ClassBatch: 120000, ClassSystem: 10000} {
        if _, ok := c.HoldBudgetsMs[cl]; !ok {
            c.HoldBudgetsMs[cl] = def
        }
    }
    if c.Coordination == "" {
        c.Coordination = "local"
    }
    seen := map[string]string{}
    ids := map[string]bool{}
    for i := range c.Upstreams {
        u := &c.Upstreams[i]
        if u.Class != "on_prem" && u.Class != "external_dev" {
            return nil, fmt.Errorf("upstream %s: invalid class %q", u.ID, u.Class)
        }
        if ids[u.ID] {
            return nil, fmt.Errorf("duplicate upstream id %s", u.ID)
        }
        ids[u.ID] = true
        if len(u.Endpoints) == 0 || len(u.Models) == 0 {
            return nil, fmt.Errorf("upstream %s: endpoints and models required", u.ID)
        }
        if u.ProbePath == "" {
            u.ProbePath = "/models"
        }
        if u.Slots.Total <= 0 {
            return nil, fmt.Errorf("upstream %s: slots.total must be > 0", u.ID)
        }
        if u.Slots.BatchMax <= 0 || u.Slots.BatchMax > u.Slots.Total {
            u.Slots.BatchMax = u.Slots.Total
        }
        for _, m := range u.Models {
            if prev, dup := seen[m]; dup {
                return nil, fmt.Errorf("model %q claimed by %s and %s", m, prev, u.ID)
            }
            seen[m] = u.ID
        }
    }
    for pfx, id := range c.RoutePrefixes {
        if !ids[id] {
            return nil, fmt.Errorf("route_prefixes.%s: unknown upstream %s", pfx, id)
        }
    }
    return &c, nil
}

git add go.mod go.sum internal/config/
git commit -m "feat(config): static registry config loader with validation

Claude-Session: https://claude.ai/code/session_01Cco5bXPQVNVk342NEW1pdT"

package registry

import (
    "context"
    "net/http"
    "net/http/httptest"
    "testing"
    "time"

    "datahive.id/ahu-gpu-manager/internal/config"
)

func cfg(eps ...string) *config.Config {
    return &config.Config{Upstreams: []config.Upstream{{
        ID: "u1", Type: "llm-openai", Class: "on_prem",
        Endpoints: eps, Models: []string{"m1"}, ProbePath: "/models",
        Slots: config.Slots{Total: 2, BatchMax: 2},
    }}, RoutePrefixes: map[string]string{"synthesis": "u1"}}
}

func TestResolveAndRoundRobin(t *testing.T) {
    r := New(cfg("http://a/v1", "http://b/v1"))
    u, err := r.Resolve("m1")
    if err != nil || u.ID != "u1" {
        t.Fatalf("resolve: %v %v", u, err)
    }
    if _, err := r.Resolve("nope"); err != ErrUnknownModel {
        t.Fatalf("want ErrUnknownModel, got %v", err)
    }
    if u, err := r.ResolvePrefix("synthesis"); err != nil || u.ID != "u1" {
        t.Fatalf("prefix: %v %v", u, err)
    }
    e1, _ := r.PickEndpoint("u1")
    e2, _ := r.PickEndpoint("u1")
    if e1 == e2 {
        t.Fatalf("round robin broken: %s %s", e1, e2)
    }
    r.MarkDown("u1", "http://a/v1")
    for i := 0; i < 4; i++ {
        e, err := r.PickEndpoint("u1")
        if err != nil || e != "http://b/v1" {
            t.Fatalf("down endpoint served: %s %v", e, err)
        }
    }
    r.MarkDown("u1", "http://b/v1")
    if _, err := r.PickEndpoint("u1"); err != ErrNoHealthyEndpoint {
        t.Fatalf("want ErrNoHealthyEndpoint, got %v", err)
    }
}

func TestProberRecoversEndpoint(t *testing.T) {
    up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(200) }))
    defer up.Close()
    r := New(cfg(up.URL))
    r.MarkDown("u1", up.URL)
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    r.StartProber(ctx, 20*time.Millisecond)
    deadline := time.Now().Add(2 * time.Second)
    for time.Now().Before(deadline) {
        if _, err := r.PickEndpoint("u1"); err == nil {
            return
        }
        time.Sleep(20 * time.Millisecond)
    }
    t.Fatal("prober never recovered endpoint")
}

package registry

import (
    "context"
    "errors"
    "net/http"
    "strings"
    "sync"
    "time"

    "datahive.id/ahu-gpu-manager/internal/config"
)

var (
    ErrUnknownModel      = errors.New("unknown model")
    ErrNoHealthyEndpoint = errors.New("no healthy endpoint")
    ErrUnknownUpstream   = errors.New("unknown upstream")
)

type epState struct {
    url     string
    healthy bool
}

type upState struct {
    cfg  *config.Upstream
    eps  []*epState
    next int
}

type Registry struct {
    mu       sync.Mutex
    byModel  map[string]*upState
    byID     map[string]*upState
    byPrefix map[string]*upState
    client   *http.Client
}

func New(c *config.Config) *Registry {
    r := &Registry{
        byModel: map[string]*upState{}, byID: map[string]*upState{}, byPrefix: map[string]*upState{},
        client: &http.Client{Timeout: 5 * time.Second},
    }
    for i := range c.Upstreams {
        u := &c.Upstreams[i]
        st := &upState{cfg: u}
        for _, e := range u.Endpoints {
            st.eps = append(st.eps, &epState{url: e, healthy: true})
        }
        r.byID[u.ID] = st
        for _, m := range u.Models {
            r.byModel[m] = st
        }
    }
    for pfx, id := range c.RoutePrefixes {
        r.byPrefix[pfx] = r.byID[id]
    }
    return r
}

func (r *Registry) Resolve(model string) (*config.Upstream, error) {
    r.mu.Lock()
    defer r.mu.Unlock()
    if st, ok := r.byModel[model]; ok {
        return st.cfg, nil
    }
    return nil, ErrUnknownModel
}

func (r *Registry) ResolvePrefix(prefix string) (*config.Upstream, error) {
    r.mu.Lock()
    defer r.mu.Unlock()
    if st, ok := r.byPrefix[prefix]; ok {
        return st.cfg, nil
    }
    return nil, ErrUnknownUpstream
}

func (r *Registry) PickEndpoint(id string) (string, error) {
    r.mu.Lock()
    defer r.mu.Unlock()
    st, ok := r.byID[id]
    if !ok {
        return "", ErrUnknownUpstream
    }
    n := len(st.eps)
    for i := 0; i < n; i++ {
        ep := st.eps[(st.next+i)%n]
        if ep.healthy {
            st.next = (st.next + i + 1) % n
            return ep.url, nil
        }
    }
    return "", ErrNoHealthyEndpoint
}

func (r *Registry) setHealth(id, url string, h bool) {
    r.mu.Lock()
    defer r.mu.Unlock()
    if st, ok := r.byID[id]; ok {
        for _, ep := range st.eps {
            if ep.url == url {
                ep.healthy = h
            }
        }
    }
}

func (r *Registry) MarkDown(id, url string) { r.setHealth(id, url, false) }
func (r *Registry) MarkUp(id, url string)   { r.setHealth(id, url, true) }

func (r *Registry) Health() map[string]map[string]bool {
    r.mu.Lock()
    defer r.mu.Unlock()
    out := map[string]map[string]bool{}
    for id, st := range r.byID {
        m := map[string]bool{}
        for _, ep := range st.eps {
            m[ep.url] = ep.healthy
        }
        out[id] = m
    }
    return out
}

// StartProber re-checks every endpoint (healthy or not) so downed endpoints recover
// and silently-dead ones are caught before a request trips on them.
func (r *Registry) StartProber(ctx context.Context, interval time.Duration) {
    go func() {
        t := time.NewTicker(interval)
        defer t.Stop()
        for {
            select {
            case <-ctx.Done():
                return
            case <-t.C:
                r.mu.Lock()
                type probe struct{ id, url, path string }
                var ps []probe
                for id, st := range r.byID {
                    for _, ep := range st.eps {
                        ps = append(ps, probe{id, ep.url, st.cfg.ProbePath})
                    }
                }
                r.mu.Unlock()
                for _, p := range ps {
                    u := strings.TrimSuffix(p.url, "/") + p.path
                    resp, err := r.client.Get(u)
                    ok := err == nil && resp.StatusCode < 500
                    if resp != nil {
                        resp.Body.Close()
                    }
                    r.setHealth(p.id, p.url, ok)
                }
            }
        }
    }()
}

git add internal/registry/
git commit -m "feat(registry): model routing, endpoint health, active prober

Claude-Session: https://claude.ai/code/session_01Cco5bXPQVNVk342NEW1pdT"

type Ticket interface {
    Ready() <-chan struct{} // closed when a slot is held
    Pos() int               // 1-based queue position; 0 once admitted
    Release()               // free the slot (only after Ready)
    Cancel()                // abandon while waiting
}
type Pool interface {
    Enqueue(class config.Class) Ticket
    Depth() map[config.Class]int // waiting per class
    Active() int                 // slots in use
}
func NewLocal(s config.Slots) Pool

package pool

import (
    "testing"
    "time"

    "datahive.id/ahu-gpu-manager/internal/config"
)

func admitted(t Ticket) bool {
    select {
    case <-t.Ready():
        return true
    case <-time.After(50 * time.Millisecond):
        return false
    }
}

func TestCapacityAndFIFO(t *testing.T) {
    p := NewLocal(config.Slots{Total: 1, BatchMax: 1})
    a := p.Enqueue(config.ClassInteractive)
    if !admitted(a) {
        t.Fatal("first not admitted")
    }
    b := p.Enqueue(config.ClassInteractive)
    if admitted(b) {
        t.Fatal("second admitted over capacity")
    }
    if b.Pos() != 1 {
        t.Fatalf("pos = %d, want 1", b.Pos())
    }
    a.Release()
    if !admitted(b) {
        t.Fatal("b not admitted after release")
    }
    b.Release()
}

func TestInteractiveJumpsBatch(t *testing.T) {
    p := NewLocal(config.Slots{Total: 1, BatchMax: 1})
    hold := p.Enqueue(config.ClassBatch)
    admitted(hold)
    b := p.Enqueue(config.ClassBatch)
    i := p.Enqueue(config.ClassInteractive)
    hold.Release()
    if !admitted(i) {
        t.Fatal("interactive did not jump batch")
    }
    if admitted(b) {
        t.Fatal("batch admitted while interactive held the slot")
    }
    i.Release()
    if !admitted(b) {
        t.Fatal("batch never admitted")
    }
    b.Release()
}

func TestBatchMaxLeavesHeadroom(t *testing.T) {
    p := NewLocal(config.Slots{Total: 2, BatchMax: 1})
    b1 := p.Enqueue(config.ClassBatch)
    admitted(b1)
    b2 := p.Enqueue(config.ClassBatch)
    if admitted(b2) {
        t.Fatal("batch exceeded batch_max")
    }
    i := p.Enqueue(config.ClassInteractive)
    if !admitted(i) {
        t.Fatal("interactive blocked despite free slot")
    }
    if p.Active() != 2 || p.Depth()[config.ClassBatch] != 1 {
        t.Fatalf("active=%d depth=%v", p.Active(), p.Depth())
    }
    b2.Cancel()
    if p.Depth()[config.ClassBatch] != 0 {
        t.Fatal("cancel did not dequeue")
    }
}

package pool

import (
    "sync"

    "datahive.id/ahu-gpu-manager/internal/config"
)

type Ticket interface {
    Ready() <-chan struct{}
    Pos() int
    Release()
    Cancel()
}

type Pool interface {
    Enqueue(class config.Class) Ticket
    Depth() map[config.Class]int
    Active() int
}

var rank = map[config.Class]int{config.ClassInteractive: 0, config.ClassSystem: 1, config.ClassBatch: 2}

type ticket struct {
    p        *localPool
    class    config.Class
    seq      uint64
    ready    chan struct{}
    admitted bool
    done     bool
}

func (t *ticket) Ready() <-chan struct{} { return t.ready }

func (t *ticket) Pos() int {
    t.p.mu.Lock()
    defer t.p.mu.Unlock()
    if t.admitted {
        return 0
    }
    pos := 0
    for _, w := range t.p.waiting {
        if before(w, t) || w == t {
            pos++
        }
        if w == t {
            return pos
        }
    }
    return pos
}

func (t *ticket) Release() {
    t.p.mu.Lock()
    defer t.p.mu.Unlock()
    if !t.admitted || t.done {
        return
    }
    t.done = true
    t.p.active--
    if t.class == config.ClassBatch {
        t.p.activeBatch--
    }
    t.p.pump()
}

func (t *ticket) Cancel() {
    t.p.mu.Lock()
    defer t.p.mu.Unlock()
    if t.admitted {
        return
    }
    for i, w := range t.p.waiting {
        if w == t {
            t.p.waiting = append(t.p.waiting[:i], t.p.waiting[i+1:]...)
            break
        }
    }
    t.done = true
    t.p.pump()
}

func before(a, b *ticket) bool {
    if rank[a.class] != rank[b.class] {
        return rank[a.class] < rank[b.class]
    }
    return a.seq < b.seq
}

type localPool struct {
    mu          sync.Mutex
    slots       config.Slots
    seq         uint64
    active      int
    activeBatch int
    waiting     []*ticket // kept sorted by (rank, seq)
}

func NewLocal(s config.Slots) Pool { return &localPool{slots: s} }

func (p *localPool) Enqueue(class config.Class) Ticket {
    p.mu.Lock()
    defer p.mu.Unlock()
    p.seq++
    t := &ticket{p: p, class: class, seq: p.seq, ready: make(chan struct{})}
    i := 0
    for ; i < len(p.waiting); i++ {
        if before(t, p.waiting[i]) {
            break
        }
    }
    p.waiting = append(p.waiting[:i], append([]*ticket{t}, p.waiting[i:]...)...)
    p.pump()
    return t
}

// pump admits from the head of the queue while capacity allows. Callers hold p.mu.
func (p *localPool) pump() {
    for i := 0; i < len(p.waiting); {
        t := p.waiting[i]
        if p.active >= p.slots.Total {
            return
        }
        if t.class == config.ClassBatch && p.activeBatch >= p.slots.BatchMax {
            i++ // batch head blocked; a later interactive/system waiter may still fit
            continue
        }
        p.waiting = append(p.waiting[:i], p.waiting[i+1:]...)
        t.admitted = true
        p.active++
        if t.class == config.ClassBatch {
            p.activeBatch++
        }
        close(t.ready)
    }
}

func (p *localPool) Depth() map[config.Class]int {
    p.mu.Lock()
    defer p.mu.Unlock()
    d := map[config.Class]int{}
    for _, w := range p.waiting {
        d[w.class]++
    }
    return d
}

func (p *localPool) Active() int {
    p.mu.Lock()
    defer p.mu.Unlock()
    return p.active
}

git add internal/pool/
git commit -m "feat(pool): local priority slot pool with batch_max and queue position

Claude-Session: https://claude.ai/code/session_01Cco5bXPQVNVk342NEW1pdT"

// event.go
type Event struct { /* all schema-v1 fields, json tags below */ }
func NewID() string // ULID
// emitter.go
func NewEmitter(rdb redis.UniversalClient, streamKey, spoolDir string) *Emitter
func (e *Emitter) Emit(ev Event)                 // never blocks
func (e *Emitter) Run(ctx context.Context)       // background loop
func (e *Emitter) Flush(timeout time.Duration)   // used at drain
func (e *Emitter) Dropped() uint64

package audit

import (
    "context"
    "encoding/json"
    "os"
    "path/filepath"
    "testing"
    "time"

    "github.com/alicebob/miniredis/v2"
    "github.com/redis/go-redis/v9"
)

func ev(id string) Event {
    return Event{EventID: id, SchemaVer: 1, TS: time.Now().UTC().Format(time.RFC3339Nano),
        Engine: "ahu-chatbot", Operation: "chat", Status: "ok", Model: "m1"}
}

func TestEmitReachesStream(t *testing.T) {
    mr := miniredis.RunT(t)
    rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()})
    e := NewEmitter(rdb, "ahu.ai.audit", t.TempDir())
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    go e.Run(ctx)
    e.Emit(ev("01A"))
    deadline := time.Now().Add(2 * time.Second)
    for time.Now().Before(deadline) {
        if msgs, _ := rdb.XRange(context.Background(), "ahu.ai.audit", "-", "+").Result(); len(msgs) == 1 {
            var got Event
            if err := json.Unmarshal([]byte(msgs[0].Values["json"].(string)), &got); err != nil || got.EventID != "01A" {
                t.Fatalf("bad payload: %v %v", got, err)
            }
            return
        }
        time.Sleep(10 * time.Millisecond)
    }
    t.Fatal("event never reached stream")
}

func TestSpoolOnRedisDownThenReplay(t *testing.T) {
    mr := miniredis.RunT(t)
    rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()})
    dir := t.TempDir()
    e := NewEmitter(rdb, "ahu.ai.audit", dir)
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    go e.Run(ctx)

    mr.SetError("down") // all commands fail
    e.Emit(ev("01B"))
    deadline := time.Now().Add(2 * time.Second)
    spooled := func() bool {
        fs, _ := filepath.Glob(filepath.Join(dir, "*.jsonl"))
        for _, f := range fs {
            if b, _ := os.ReadFile(f); len(b) > 0 {
                return true
            }
        }
        return false
    }
    for time.Now().Before(deadline) && !spooled() {
        time.Sleep(10 * time.Millisecond)
    }
    if !spooled() {
        t.Fatal("event not spooled while redis down")
    }

    mr.SetError("") // recover
    deadline = time.Now().Add(3 * time.Second)
    for time.Now().Before(deadline) {
        if msgs, _ := rdb.XRange(context.Background(), "ahu.ai.audit", "-", "+").Result(); len(msgs) == 1 && !spooled() {
            return
        }
        time.Sleep(20 * time.Millisecond)
    }
    t.Fatal("spooled event never replayed to stream")
}

package audit

import (
    "crypto/rand"
    "time"

    "github.com/oklog/ulid/v2"
)

type Event struct {
    EventID        string `json:"event_id"`
    SchemaVer      int    `json:"schema_ver"`
    TS             string `json:"ts"`
    Engine         string `json:"engine"`
    Surface        string `json:"surface,omitempty"`
    UserID         string `json:"user_id,omitempty"`
    TraceID        string `json:"trace_id,omitempty"`
    Upstream       string `json:"upstream,omitempty"`
    Model          string `json:"model,omitempty"`
    UpstreamClass  string `json:"upstream_class,omitempty"`
    TrafficClass   string `json:"traffic_class,omitempty"`
    Operation      string `json:"operation"`
    Status         string `json:"status"`
    ErrorCode      string `json:"error_code,omitempty"`
    QueueMs        int64  `json:"queue_ms"`
    UpstreamMs     int64  `json:"upstream_ms"`
    TotalMs        int64  `json:"total_ms"`
    TokensIn       *int   `json:"tokens_in,omitempty"`
    TokensOut      *int   `json:"tokens_out,omitempty"`
    DocHash        string `json:"doc_hash,omitempty"`
    Pages          *int   `json:"pages,omitempty"`
    JobID          string `json:"job_id,omitempty"`
    ParentJobID    string `json:"parent_job_id,omitempty"`
    ConfigVersion  string `json:"config_version,omitempty"`
    RequestBodyZst []byte `json:"request_body_zst,omitempty"`
    ResponseBodyZst []byte `json:"response_body_zst,omitempty"`
    BodyTruncated  bool   `json:"body_truncated,omitempty"`
    FullBodySHA256 string `json:"full_body_sha256,omitempty"`
}

func NewID() string { return ulid.MustNew(ulid.Timestamp(time.Now()), rand.Reader).String() }

package audit

import (
    "bufio"
    "context"
    "encoding/json"
    "fmt"
    "os"
    "path/filepath"
    "sync/atomic"
    "time"

    "github.com/redis/go-redis/v9"
)

const maxStreamLen = 1_000_000

type Emitter struct {
    rdb       redis.UniversalClient
    streamKey string
    spoolDir  string
    ch        chan Event
    dropped   atomic.Uint64
}

func NewEmitter(rdb redis.UniversalClient, streamKey, spoolDir string) *Emitter {
    return &Emitter{rdb: rdb, streamKey: streamKey, spoolDir: spoolDir, ch: make(chan Event, 10000)}
}

func (e *Emitter) Emit(ev Event) {
    select {
    case e.ch <- ev:
    default: // channel full: spool synchronously rather than block the request path
        if err := e.spool(ev); err != nil {
            e.dropped.Add(1)
        }
    }
}

func (e *Emitter) Dropped() uint64 { return e.dropped.Load() }

func (e *Emitter) xadd(ctx context.Context, ev Event) error {
    b, err := json.Marshal(ev)
    if err != nil {
        return err
    }
    return e.rdb.XAdd(ctx, &redis.XAddArgs{
        Stream: e.streamKey, MaxLen: maxStreamLen, Approx: true,
        Values: map[string]any{"json": b},
    }).Err()
}

func (e *Emitter) spool(ev Event) error {
    if err := os.MkdirAll(e.spoolDir, 0o755); err != nil {
        return err
    }
    f, err := os.OpenFile(filepath.Join(e.spoolDir, time.Now().UTC().Format("2006-01-02")+".jsonl"),
        os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
    if err != nil {
        return err
    }
    defer f.Close()
    b, err := json.Marshal(ev)
    if err != nil {
        return err
    }
    _, err = fmt.Fprintf(f, "%s\n", b)
    return err
}

func (e *Emitter) Run(ctx context.Context) {
    replay := time.NewTicker(2 * time.Second)
    defer replay.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case ev := <-e.ch:
            if err := e.xadd(ctx, ev); err != nil {
                if err := e.spool(ev); err != nil {
                    e.dropped.Add(1)
                }
            }
        case <-replay.C:
            e.replaySpool(ctx)
        }
    }
}

var maxSpoolLine = 64 << 20 // spool lines carry two base64'd 10MB-capped bodies; keep headroom

func (e *Emitter) replaySpool(ctx context.Context) {
    files, _ := filepath.Glob(filepath.Join(e.spoolDir, "*.jsonl"))
    for _, fp := range files {
        f, err := os.Open(fp)
        if err != nil {
            continue
        }
        sc := bufio.NewScanner(f)
        sc.Buffer(make([]byte, 0, 64*1024), maxSpoolLine)
        ok := true
        for sc.Scan() {
            var ev Event
            if json.Unmarshal(sc.Bytes(), &ev) != nil {
                continue // skip corrupt line
            }
            if e.xadd(ctx, ev) != nil {
                ok = false
                break
            }
        }
        // A scan error (e.g. ErrTooLong) means unread lines remain: removing the
        // file here would be silent data loss — treat it like a mid-file failure.
        if sc.Err() != nil {
            ok = false
        }
        f.Close()
        if ok {
            os.Remove(fp)
        } else {
            return // redis still down or file unreadable; retry next tick
        }
    }
}

func (e *Emitter) Flush(timeout time.Duration) {
    deadline := time.Now().Add(timeout)
    for time.Now().Before(deadline) {
        select {
        case ev := <-e.ch:
            ctx, cancel := context.WithTimeout(context.Background(), time.Second)
            if e.xadd(ctx, ev) != nil {
                _ = e.spool(ev)
            }
            cancel()
        default:
            return
        }
    }
}

git add internal/audit/ go.mod go.sum
git commit -m "feat(audit): schema v1 event + fire-and-forget emitter with disk spool

Claude-Session: https://claude.ai/code/session_01Cco5bXPQVNVk342NEW1pdT"

type Deps struct {
    Cfg      *config.Config
    Reg      *registry.Registry
    Pools    map[string]pool.Pool // key: upstream ID
    Emit     func(audit.Event)
    Idem     IdemCache            // Task 8 provides RedisIdem; nil disables
    Stats    *Stats
}
func NewHandler(d Deps) http.Handler // serves /v1/* and /synthesis/v1/*
type Meta struct{ Tenant, Surface, UserID, TraceID, IdemKey string; Class config.Class; QueueEvents bool }
func MetaFrom(r *http.Request, tokenTenants map[string]string) Meta
type IdemCache interface {
    Get(ctx context.Context, tenant, key string) (status int, body []byte, ok bool)
    Put(ctx context.Context, tenant, key string, status int, body []byte)
}
// stats.go
type Stats struct{ ... }
func NewStats() *Stats
func (s *Stats) Observe(upstream string, ms int64)
func (s *Stats) ETAMs(upstream string, depth, slots int) int64