feat(bandit): Thompson Sampling consumer closing the brand_rules feedback loop

.gitignore

@@ -0,0 +1,6 @@
+node_modules/
+dist/
+dist-demo/
+*.tsbuildinfo
+.DS_Store
+var/

packages/bandit/.env.example

@@ -0,0 +1,12 @@
+# @shoploop/bandit — Thompson Sampling consumer. All optional; defaults shown.
+BANDIT_QUEUE_PATH=./var/bandit/queue.jsonl
+BANDIT_RECOMMENDATIONS_PATH=./var/bandit/recommendations.jsonl
+BANDIT_STATE_DB=./var/bandit/state.db
+BANDIT_CHECKPOINT_DB=./var/bandit/checkpoints.db
+BRAND_RULES_PATH=./brand_rules.json
+BANDIT_REWARD_MODE=roas
+ROAS_CEILING=5
+MIN_IMPRESSIONS_FOR_REWARD=50
+BANDIT_DECAY=0.995
+BANDIT_PROMOTE_MARGIN=0.1
+BANDIT_POLL_INTERVAL_MS=5000

packages/bandit/DESIGN.md

@@ -0,0 +1,93 @@
+# `@shoploop/bandit` — Thompson Sampling feedback consumer
+
+Closes the architecture loop: `Performance Ingest → Thompson Sampling → brand_rules.json`.
+This package is the **consumer** half. It does not render, does not call ad APIs, and does
+not (in this PR) move money. It reads the durable bandit queue, learns which scene preset
+wins per product twin, writes that posterior back into `brand_rules.json`, and emits
+**DRAFT-only** budget recommendations.
+
+## Why Beta-Bernoulli (and not more)
+
+The reward is a normalized scalar in `[0,1]` (ROAS-or-CTR). Beta is the conjugate prior for
+a Bernoulli success probability, so a fractional reward updates the posterior in closed form
+(`α += r; β += 1−r`). It is the simplest *correct* bandit: no gradient steps, no replay
+buffer, no context features. Contextual bandits / deep RL are explicitly a later PR — arms
+here are flat `(twin_handle, preset)` pairs.
+
+## The 7-step consumer loop (`consumer/loop.ts`)
+
+```
+            BANDIT_QUEUE_PATH (JSONL, single-writer, shared with producers)
+                       │
+   ┌───────────────────┴────────────────────────────────────────────────┐
+   │ 1. READ      tail new bytes since the LibSQL checkpoint              │
+   │ 2. BUCKET    split by `schema`: performance_observation │ provenance │
+   │ 3. JOIN      observation × provenance → per-arm reward               │
+   │              key precedence: external_id ▶ product_gid ▶ campaign_id │
+   │ 4. UPDATE    decay all arms ×BANDIT_DECAY, then α/β update each arm  │
+   │ 5. RECOMMEND per twin w/ ≥2 presets: Thompson-sample; if leader      │
+   │              beats runner-up by ≥BANDIT_PROMOTE_MARGIN → emit DRAFT  │
+   │ 6. WRITE     merge bandit: namespace into brand_rules.json (locked)  │
+   │ 7. CHECKPOINT advance the byte offset                                │
+   └───────────────────┬─────────────────────────┬──────────────────────┘
+                        │                         │
+        brand_rules.json (bandit: ns)   BANDIT_RECOMMENDATIONS_PATH (DRAFT)
+        ← Scene Composer reads this     ← distribution PR reads this (later)
+```
+
+## Producers — the two streams on `BANDIT_QUEUE_PATH`
+
+Both producers append to the **same** single-writer JSONL queue; the consumer discriminates
+on the `schema` field. Neither producer knows the sampler exists.
+
+| Stream | `schema` | Producer | Carries |
+|---|---|---|---|
+| Reward | `performance_observation.v1` | Google Ads connector (PR #18) | `ad_id`, `metrics.{roas,ctr,impressions,...}`, `shopify_product_gid` |
+| Arm | `render_provenance.v1` | animated-shorts (PR #19) | `twin_handle`, `preset`, `storyboard_id` |
+
+Types are **mirrored, not imported** (`src/types.ts`) — the consumer lives in a different
+deploy and must not couple to producer internals. The contract is the JSON shape.
+
+## The join seam with the distribution PR
+
+A `PerformanceObservation` knows the channel `ad_id`; a `RenderProvenance` knows the arm
+(`twin_handle × preset`). The missing link is *which ad_id a creative was dispatched as* —
+and that is owned by the **distribution paid adapters**, which stay DRAFT in their PR.
+
+The contract this package proposes: when a paid adapter dispatches a creative, it stamps the
+returned channel ad_id onto the provenance record's **`external_id`** field. Then
+`observation.ad_id === provenance.external_id` is an exact 1:1 join (`join_kind: "external_id"`).
+
+Until that lands, the joiner falls back in precedence order:
+
+1. `external_id`  — exact, the intended seam.
+2. `shopify_product_gid` — observation and provenance both reference the same product.
+3. `campaign_id` — coarsest; the original task fallback. Flagged in `JoinedSample.join_kind`
+   so we can measure how much reward is attributed loosely.
+
+`animated-shorts` does not yet emit `external_id` — see the `TODO(distribution-PR)` in
+`join/joiner.ts`. Adding it is a one-field, additive change on the producer side.
+
+## Out of scope (owned by the distribution PR)
+
+- `request_budget_shift` MCP tool — the thing that would *act* on a recommendation. This
+  package only writes recommendations to `BANDIT_RECOMMENDATIONS_PATH`; nothing auto-applies.
+- `list_recent_dispatches` — the distribution-side inventory of what was dispatched as what.
+
+The DRAFT posture is deliberate: positionless full-funnel feedback *control* over
+`brand_rules` is the moat, but flipping it to live budget moves is a separate, gated step.
+
+## Persistence
+
+- **Arm state** — LibSQL `arm_states`, one row per `(twin_handle, preset)`. The Beta
+  posterior is the entire learned state (≪10 MB for the dogfood).
+- **Checkpoints** — LibSQL `checkpoints`, byte offset per source file. Byte (not line)
+  offset so a partially-written trailing line is re-read whole on the next pass.
+- **brand_rules.json** — file-locked (`proper-lockfile`) read-modify-write. Scene Composer
+  reads this concurrently; only the `bandit` key is replaced, every other top-level key is
+  preserved exactly.
+
+## Cost
+
+Pure CPU, co-locates on the existing `t4g.small`: **$0 incremental**. LibSQL state <$1.
+**Total ≈ $0/mo.**

packages/bandit/README.md

@@ -0,0 +1,71 @@
+# @shoploop/bandit
+
+Thompson Sampling consumer that closes the `brand_rules.json` feedback loop. Tails the
+shared bandit queue, maintains Beta-Bernoulli arms keyed by `(twin_handle, preset)` in
+LibSQL, writes posteriors into the `bandit:` namespace of `brand_rules.json`, and emits
+**DRAFT-only** budget recommendations. It is a pure consumer — it never calls a paid ad API.
+
+See [`DESIGN.md`](./DESIGN.md) for the 7-step loop, the producer contracts, and the join
+seam with the distribution PR.
+
+## CLI
+
+```bash
+shoploop-bandit run                  # long-running consumer service (polls the queue)
+shoploop-bandit once                 # single pass — for cron / tests
+shoploop-bandit inspect [--twin <h>] # pretty-print arm states + last recommendation
+```
+
+## Reward
+
+Normalized to `[0,1]`. `BANDIT_REWARD_MODE=roas` (default) saturates ROAS at `ROAS_CEILING`;
+`ctr` clamps click-through to `[0,0.2]` then scales. Observations below
+`MIN_IMPRESSIONS_FOR_REWARD` are skipped as noise.
+
+## `brand_rules.json` — `bandit:` namespace
+
+Only the `bandit` key is written; all other keys (`voice`, `palette`, `materials`,
+`lighting`, …) are preserved exactly.
+
+```json
+{
+  "bandit": {
+    "version": 1,
+    "updated_at": "2026-06-25T10:06:47.244Z",
+    "arms": {
+      "varitea_jasmine_pearl": {
+        "hero":      { "alpha": 4,   "beta": 1,   "pulls": 3, "last_reward": 1 },
+        "lifestyle": { "alpha": 1.1, "beta": 3.9, "pulls": 3, "last_reward": 0.02 }
+      }
+    },
+    "last_recommendation": {
+      "twin_handle": "varitea_jasmine_pearl",
+      "winning_preset": "hero",
+      "confidence": 0.983
+    }
+  }
+}
+```
+
+## Env
+
+| Var | Default | Meaning |
+|---|---|---|
+| `BANDIT_QUEUE_PATH` | `./var/bandit/queue.jsonl` | Input queue (shared with producers) |
+| `BANDIT_RECOMMENDATIONS_PATH` | `./var/bandit/recommendations.jsonl` | DRAFT output queue |
+| `BANDIT_STATE_DB` | `./var/bandit/state.db` | LibSQL arm posteriors |
+| `BANDIT_CHECKPOINT_DB` | `./var/bandit/checkpoints.db` | LibSQL consumer offsets |
+| `BRAND_RULES_PATH` | `./brand_rules.json` | File the `bandit:` namespace merges into |
+| `BANDIT_REWARD_MODE` | `roas` | `roas` \| `ctr` |
+| `ROAS_CEILING` | `5` | ROAS that maps to reward `1.0` |
+| `MIN_IMPRESSIONS_FOR_REWARD` | `50` | Noise floor |
+| `BANDIT_DECAY` | `0.995` | Per-loop multiplicative forgetting |
+| `BANDIT_PROMOTE_MARGIN` | `0.1` | Min sampled gap to emit a recommendation |
+
+## Develop
+
+```bash
+pnpm install
+pnpm typecheck
+pnpm test
+```

packages/bandit/bin/cli.js

@@ -0,0 +1,2 @@
+#!/usr/bin/env node
+import "../dist/cli.js";

packages/bandit/package.json

@@ -0,0 +1,47 @@
+{
+  "name": "@shoploop/bandit",
+  "version": "0.1.0",
+  "description": "Thompson Sampling consumer that closes the brand_rules feedback loop. Tails the bandit queue (PerformanceObservation × RenderProvenance), maintains Beta-Bernoulli arms keyed by (twin_handle, preset) in LibSQL, writes posteriors into the brand_rules.json bandit: namespace, and emits DRAFT-only BudgetRecommendations. Positionless full-funnel feedback control.",
+  "license": "UNLICENSED",
+  "private": true,
+  "type": "module",
+  "main": "./dist/index.js",
+  "types": "./dist/index.d.ts",
+  "exports": {
+    ".": {
+      "types": "./dist/index.d.ts",
+      "import": "./dist/index.js"
+    }
+  },
+  "bin": {
+    "shoploop-bandit": "./bin/cli.js"
+  },
+  "files": [
+    "dist",
+    "bin"
+  ],
+  "scripts": {
+    "build": "tsc -p tsconfig.build.json",
+    "typecheck": "tsc -p tsconfig.json --noEmit",
+    "lint": "tsc -p tsconfig.json --noEmit",
+    "test": "vitest run",
+    "test:watch": "vitest",
+    "cli": "tsx src/cli.ts"
+  },
+  "dependencies": {
+    "@libsql/client": "^0.14.0",
+    "proper-lockfile": "^4.1.2",
+    "seedrandom": "^3.0.5"
+  },
+  "devDependencies": {
+    "@types/node": "^20.14.0",
+    "@types/proper-lockfile": "^4.1.4",
+    "@types/seedrandom": "^3.0.8",
+    "tsx": "^4.19.0",
+    "typescript": "^5.6.0",
+    "vitest": "^2.1.0"
+  },
+  "engines": {
+    "node": ">=20"
+  }
+}

packages/bandit/src/bandit/beta_arm.ts

@@ -0,0 +1,98 @@
+// Beta-Bernoulli arm. The simplest correct bandit posterior: a Beta(alpha,beta)
+// over the arm's success probability. Reward ∈ [0,1] is treated as a fractional
+// Bernoulli outcome (alpha += reward; beta += 1-reward), so a continuous normalized
+// ROAS/CTR slots straight into conjugate Beta updating. No deep RL, no context.
+
+export type Rng = () => number; // uniform in [0,1)
+
+export interface BetaArmSnapshot {
+  alpha: number;
+  beta: number;
+  reward_sum: number;
+  pulls: number;
+}
+
+export class BetaArm {
+  alpha: number;
+  beta: number;
+  reward_sum: number;
+  pulls: number;
+
+  constructor(alpha = 1, beta = 1, reward_sum = 0, pulls = 0) {
+    this.alpha = alpha;
+    this.beta = beta;
+    this.reward_sum = reward_sum;
+    this.pulls = pulls;
+  }
+
+  // Posterior mean of the success probability.
+  mean(): number {
+    return this.alpha / (this.alpha + this.beta);
+  }
+
+  // Fractional conjugate update. reward is clamped defensively to [0,1].
+  update(reward: number): void {
+    const r = reward < 0 ? 0 : reward > 1 ? 1 : reward;
+    this.alpha += r;
+    this.beta += 1 - r;
+    this.reward_sum += r;
+    this.pulls += 1;
+  }
+
+  // Multiplicative decay toward the prior — keeps arms responsive to regime change
+  // without fully forgetting. Floored at the Beta(1,1) uniform prior so a long-idle
+  // arm never collapses below the prior and starts sampling pathologically.
+  decay(factor: number): void {
+    this.alpha = 1 + (this.alpha - 1) * factor;
+    this.beta = 1 + (this.beta - 1) * factor;
+  }
+
+  // A draw from Beta(alpha,beta) via two Gamma draws: B = X/(X+Y),
+  // X~Gamma(alpha,1), Y~Gamma(beta,1). Deterministic given the rng.
+  sample(rng: Rng): number {
+    const x = gamma(this.alpha, rng);
+    const y = gamma(this.beta, rng);
+    const s = x + y;
+    return s > 0 ? x / s : 0.5;
+  }
+
+  snapshot(): BetaArmSnapshot {
+    return { alpha: this.alpha, beta: this.beta, reward_sum: this.reward_sum, pulls: this.pulls };
+  }
+
+  static from(s: BetaArmSnapshot): BetaArm {
+    return new BetaArm(s.alpha, s.beta, s.reward_sum, s.pulls);
+  }
+}
+
+// Standard normal from a uniform rng via Box-Muller. Guards against u1=0 (log(0)).
+function normal(rng: Rng): number {
+  let u1 = rng();
+  if (u1 < 1e-12) u1 = 1e-12;
+  const u2 = rng();
+  return Math.sqrt(-2 * Math.log(u1)) * Math.cos(2 * Math.PI * u2);
+}
+
+// Marsaglia & Tsang (2000) Gamma(shape, scale=1) sampler. For shape<1 use the
+// boosting trick Gamma(a) = Gamma(a+1) * U^(1/a). Accepts any uniform rng so the
+// whole sampler is seedable for deterministic convergence tests.
+export function gamma(shape: number, rng: Rng): number {
+  if (shape <= 0) return 0;
+  if (shape < 1) {
+    const g = gamma(shape + 1, rng);
+    const u = Math.max(rng(), 1e-12);
+    return g * Math.pow(u, 1 / shape);
+  }
+  const d = shape - 1 / 3;
+  const c = 1 / Math.sqrt(9 * d);
+  for (;;) {
+    let x = normal(rng);
+    let v = 1 + c * x;
+    if (v <= 0) continue;
+    v = v * v * v;
+    const u = rng();
+    const x2 = x * x;
+    if (u < 1 - 0.0331 * x2 * x2) return d * v;
+    if (Math.log(u) < 0.5 * x2 + d * (1 - v + Math.log(v))) return d * v;
+  }
+}

packages/bandit/src/bandit/reward.ts

@@ -0,0 +1,37 @@
+import type { PerformanceObservation } from "../types.js";
+import type { BanditConfig } from "../config.js";
+
+// Normalize a PerformanceObservation into a Bernoulli reward ∈ [0,1] the Beta arm
+// can ingest. Two modes:
+//   roas: conversions_value / cost, saturating at ROAS_CEILING → 1.0.
+//   ctr:  clicks / impressions, clamped to [0,0.2] then linearly scaled to [0,1]
+//         (a 20% CTR is already exceptional for paid, so it maps to a full reward).
+// Observations below the impressions noise floor are skipped (null) — a single click
+// on 3 impressions is not signal.
+
+const CTR_CAP = 0.2;
+
+export interface RewardResult {
+  reward: number;
+  skipped: boolean;
+  reason?: string;
+}
+
+export function computeReward(obs: PerformanceObservation, cfg: BanditConfig): RewardResult {
+  const m = obs.metrics;
+  if (m.impressions < cfg.minImpressionsForReward) {
+    return { reward: 0, skipped: true, reason: "below_min_impressions" };
+  }
+
+  if (cfg.rewardMode === "ctr") {
+    const ctr = m.clicks / Math.max(m.impressions, 1);
+    const clamped = Math.min(Math.max(ctr, 0), CTR_CAP);
+    return { reward: clamped / CTR_CAP, skipped: false };
+  }
+
+  // roas
+  const cost = m.cost > 0 ? m.cost : m.cost_micros / 1_000_000;
+  const roas = m.conversions_value / Math.max(cost, 0.01);
+  const ceiling = cfg.roasCeiling > 0 ? cfg.roasCeiling : 5;
+  return { reward: Math.min(1, roas / ceiling), skipped: false };
+}