From 465787b1debbd301123fa38e6edd6ff6663c1522 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Wed, 24 Jun 2026 13:18:59 -0700 Subject: [PATCH] feat(orb): turn Orb into the fleet calibration collector + analytics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Orb stops being a per-instance GitHub App and becomes a central, anonymized fleet-calibration collector fed by self-hosted instances. Each instance already records de-noised ground truth in review_audit (gate_decision + pr_outcome + reversal_reopened/reversal_reverted); the exporter now ships THAT up instead of reinventing a noisier raw merged/closed signal. - Retire the per-instance Orb App: delete orb-setup.ts + orb-webhook.ts, the /orb/setup* and /orb/webhook routes + dead counters, the orb_events/orb_installations tables, the second App secret. - Repoint exportOrbBatch to a portable review_audit join (window functions + CASE — runs on the SQLite OR Postgres self-host backend); carry the reversal flag, a bucketed reason category, and decision->close cycle time; resume from an orb_export_cursor watermark. - Extend the receiver: orb_signals re-keyed on (instance_id, repo_hash, pr_hash), upsert (OR REPLACE so a later reversal updates the row), new fields + cycle-time clamp (migration 0060). - Add fleet analytics: src/orb/analytics.ts (median-robust gate precision / FP / FN / reversal / cycle-time, outlier detection) behind a bearer-gated GET /v1/internal/fleet/analytics. Anonymization unchanged: repo/PR identifiers HMAC'd with each instance's own secret, so the collector holds no instance secret and can never de-anonymize. No shared key, no Orb App, no wizard — maintainers just flip ORB_ENABLED=true. Verified: 100% branch coverage on the orb modules, full test:ci green, migration 0060 applies on real Postgres. --- .env.example | 35 +-- migrations/0060_orb_fleet_collector.sql | 41 +++ src/api/routes.ts | 15 +- src/orb/analytics.ts | 151 ++++++++++ src/orb/ingest.ts | 47 +++- src/selfhost/orb-collector.ts | 210 ++++++++------ src/selfhost/orb-setup.ts | 66 ----- src/selfhost/orb-webhook.ts | 137 --------- src/selfhost/pg-dialect.ts | 2 + src/server.ts | 55 +--- test/integration/orb-ingest.test.ts | 283 +++++++------------ test/unit/orb-analytics.test.ts | 107 +++++++ test/unit/selfhost-orb-collector.test.ts | 337 +++++++++-------------- test/unit/selfhost-orb-setup.test.ts | 110 -------- test/unit/selfhost-orb-webhook.test.ts | 322 ---------------------- 15 files changed, 721 insertions(+), 1197 deletions(-) create mode 100644 migrations/0060_orb_fleet_collector.sql create mode 100644 src/orb/analytics.ts delete mode 100644 src/selfhost/orb-setup.ts delete mode 100644 src/selfhost/orb-webhook.ts create mode 100644 test/unit/orb-analytics.test.ts delete mode 100644 test/unit/selfhost-orb-setup.test.ts delete mode 100644 test/unit/selfhost-orb-webhook.test.ts diff --git a/.env.example b/.env.example index c8550f3339..3cf4246534 100644 --- a/.env.example +++ b/.env.example @@ -161,26 +161,19 @@ GITTENSORY_REVIEW_DRAFT=false # # 1024-dimensional (e.g. bge-m3 or mxbai-embed-large via Ollama). # # Used only when RAG is enabled (GITTENSORY_REVIEW_RAG + allowlist). -# --- Gittensory Orb (#1219; opt-in outcome signal collection) --- -# Run GET /orb/setup to create the Orb GitHub App (read-only; separate from the main App). -# Credentials are written to /data/gittensory-orb.env on callback — load them here. +# --- Gittensory Orb (#1255; opt-in fleet-calibration export) --- +# Orb is the central collector + analytics that aggregates anonymized gate-calibration data UP from +# self-hosted instances. There is NO separate Orb GitHub App and NO setup wizard: your existing main App +# already records de-noised outcomes (merged/closed + reversals) locally — flip ORB_ENABLED to ship an +# anonymized signal to gittensory's collector. That's it: no second App, no extra secret, no wizard. # -# SECURITY MODEL (this image is meant to be self-hosted by many independent maintainers): -# • The image bakes NO secrets. Every operator creates their OWN Orb App via /orb/setup, so the -# ORB_* secrets below are unique to YOUR instance and live only in YOUR /data — never shared, -# never sent to the collector. gittensory's own App secrets are never in the image. -# • Export to the central collector uses NO shared key: repo/PR identifiers are HMAC-anonymized -# with YOUR ORB_WEBHOOK_SECRET (so even the collector operator can't de-anonymize them), and the -# collector accepts the batch as untrusted, rate-limited, aggregate-only telemetry. Nothing in the -# container, if leaked, can compromise the collector, other operators, or the main App. -# • Prefer the *_FILE convention for the private key (mount a file, not an inline env value): -# ORB_PRIVATE_KEY_FILE=/run/secrets/orb_private_key → the server reads it into ORB_PRIVATE_KEY. -# ORB_APP_ID= # App ID from /orb/setup callback -# ORB_APP_SLUG= # App slug (human-readable name) -# ORB_WEBHOOK_SECRET= # secret from /orb/setup callback — signs /orb/webhook + anonymizes export -# ORB_PRIVATE_KEY= # PEM from /orb/setup callback (JSON-stringified); prefer ORB_PRIVATE_KEY_FILE -# ORB_ENABLED=false # master switch: set to true to enable collection (default off) -# ORB_AIR_GAP=false # set to true to keep all data local — never send to the collector -# ORB_ANONYMIZE=true # HMAC-hash repo names before export (default true; false = raw names) +# SECURITY MODEL (this image is self-hosted by many independent maintainers): +# • The image bakes NO secrets. repo/PR identifiers are HMAC-anonymized with YOUR own ORB_WEBHOOK_SECRET +# (a stable per-instance string), so even gittensory (running the collector) can never de-anonymize them. +# • Export carries NO shared key. The collector accepts the batch as untrusted, rate-limited, aggregate-only +# telemetry. Nothing in the container, if leaked, can compromise the collector, other operators, or any App. +# ORB_ENABLED=false # master switch: set to true to export fleet-calibration signal (default off) +# ORB_WEBHOOK_SECRET= # the per-instance HMAC key used to anonymize repo/PR identifiers +# ORB_AIR_GAP=false # set to true to compute locally but never send to the collector +# ORB_ANONYMIZE=true # HMAC-hash repo/PR before export (default true; false = raw names) # ORB_COLLECTOR_URL=https://gittensory-api.aethereal.dev/v1/orb/ingest # gittensory's hosted collector (default; override for your own) -# ORB_SETUP_OUTPUT_PATH=/data/gittensory-orb.env # where /orb/setup/callback writes the credentials file diff --git a/migrations/0060_orb_fleet_collector.sql b/migrations/0060_orb_fleet_collector.sql new file mode 100644 index 0000000000..d4912f5bb4 --- /dev/null +++ b/migrations/0060_orb_fleet_collector.sql @@ -0,0 +1,41 @@ +-- Gittensory Orb (#1255): turn Orb into the central fleet-calibration collector. +-- +-- Retire the per-instance Orb GitHub App pipeline (orb_events / orb_installations were written by the +-- now-removed /orb/webhook handler). Each self-hosted instance already records de-noised ground truth in +-- review_audit (gate_decision + pr_outcome + reversal_*), so the exporter now reads THAT and ships an +-- anonymized, reversal-aware signal up to the central orb_signals store. + +DROP TABLE IF EXISTS orb_events; +DROP TABLE IF EXISTS orb_installations; + +-- orb_signals is young, continuously-regenerated telemetry (instances re-export). SQLite can't ALTER away the +-- old table-level UNIQUE(instance_id, pr_hash) — which is wrong (two instances reviewing owner/repo#123 +-- collide) — so recreate with the correct key + the new reversal/timestamp/reason columns. +DROP TABLE IF EXISTS orb_signals; +CREATE TABLE IF NOT EXISTS orb_signals ( + id INTEGER PRIMARY KEY, + instance_id TEXT NOT NULL, -- SHA256(ORB_APP_ID) prefix; one-way, no PII + repo_hash TEXT NOT NULL, -- HMAC(repo, instance secret); collector can't reverse + pr_hash TEXT NOT NULL, -- HMAC(repo#pr, instance secret) + gate_verdict TEXT, -- the prediction: 'merge' | 'close' | 'hold' + outcome TEXT NOT NULL CHECK (outcome IN ('merged', 'closed')), -- realized ground truth + reversal_flag TEXT NOT NULL DEFAULT 'none' CHECK (reversal_flag IN ('none', 'reopened', 'reverted')), + gate_reasoncode_bucket TEXT, -- low-cardinality category, bucketed at source + time_to_close_ms INTEGER, -- decision -> close cycle time (nullable) + decision_timestamp TEXT, -- when the gate decided + outcome_timestamp TEXT, -- when the PR resolved + sent_at TEXT, + received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + UNIQUE (instance_id, repo_hash, pr_hash) -- dedup unit: one row per PR per instance, upserted +); +-- Supports the (verdict, outcome, reversal) confusion-matrix rollups that are the whole point of the table. +CREATE INDEX IF NOT EXISTS orb_signals_calibration ON orb_signals (instance_id, gate_verdict, outcome, reversal_flag); +CREATE INDEX IF NOT EXISTS orb_signals_instance ON orb_signals (instance_id, received_at); + +-- Export watermark per self-host instance — replaces orb_events.exported_at (review_audit is append-only, so +-- the exporter tracks the latest exported event time and ships only newer resolved PRs / reversals). +CREATE TABLE IF NOT EXISTS orb_export_cursor ( + instance_hash TEXT PRIMARY KEY, + last_exported_at TEXT NOT NULL DEFAULT '2000-01-01T00:00:00Z', + updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP +); diff --git a/src/api/routes.ts b/src/api/routes.ts index 64d54426ab..227ea95d46 100644 --- a/src/api/routes.ts +++ b/src/api/routes.ts @@ -123,6 +123,7 @@ import { } from "../github/commands"; import { handleGitHubWebhook } from "../github/webhook"; import { handleOrbIngest } from "../orb/ingest"; +import { computeFleetAnalytics } from "../orb/analytics"; import { handleMcpRequest } from "../mcp/server"; import { buildOpenApiSpec } from "../openapi/spec"; import { generateSignalSnapshots } from "../queue/processors"; @@ -2863,9 +2864,9 @@ export function createApp() { app.post("/v1/github/webhook", handleGitHubWebhook); - // Gittensory Orb (#1219) — central collector. Receives anonymized outcome signal batches - // from self-hosted instances. No auth required: all data is HMAC-anonymized by the sender; - // dedup is enforced via UNIQUE(instance_id, pr_hash) in orb_signals. + // Gittensory Orb (#1255) — central fleet-calibration collector. Receives anonymized, reversal-aware + // outcome batches from self-hosted instances. No auth required: all data is HMAC-anonymized by the sender; + // dedup is enforced via UNIQUE(instance_id, repo_hash, pr_hash) in orb_signals. Rate-limited (strict, #1254). app.post("/v1/orb/ingest", async (c) => { const body = await c.req.text().catch(() => null); if (!body) return c.json({ error: "invalid_request" }, 400); @@ -2874,6 +2875,14 @@ export function createApp() { return c.json(result, 200); }); + // Fleet calibration analytics over the collected orb_signals — gate accuracy (precision / FP / reversal / + // cycle-time) aggregated median-robustly across the self-host fleet. Owner-only: bearer-gated by the + // `/v1/internal/*` middleware (INTERNAL_JOB_TOKEN). `?days=` windows the lookback (default 90). + app.get("/v1/internal/fleet/analytics", async (c) => { + const days = parsePositiveInt(c.req.query("days")) ?? 90; + return c.json(await computeFleetAnalytics(c.env, { windowDays: days })); + }); + // Convergence (ops / observability, flag GITTENSORY_REVIEW_OPS). Cross-repo review-OUTCOME aggregate (gate-block // ledger + recommendation/slop calibration) for an operator dashboard. Bearer-gated by the `/v1/internal/*` // middleware above (INTERNAL_JOB_TOKEN). Flag-OFF (default) → 404, so the endpoint does not exist and the diff --git a/src/orb/analytics.ts b/src/orb/analytics.ts new file mode 100644 index 0000000000..0d773996fa --- /dev/null +++ b/src/orb/analytics.ts @@ -0,0 +1,151 @@ +// Gittensory Orb (#1255) — fleet calibration ANALYTICS. Reads the anonymized orb_signals collected from +// self-hosted instances and derives gate-accuracy metrics across the fleet. Aggregation is median/percentile +// (never mean) so a single instance contributing fabricated data cannot move the fleet numbers. + +const MIN_DECIDED = 5; // an instance needs at least this many decided PRs to count toward the fleet median +const OUTLIER_BAND = 0.25; // |instance precision − fleet median| beyond this flags the instance + +/** Per-instance confusion-matrix cell as stored. */ +interface Cell { + instance_id: string; + verdict: string | null; + outcome: string; + reversal_flag: string; + n: number; +} + +export interface InstanceMetrics { + instanceId: string; + decided: number; + mergePrecision: number | null; // P(merged & not reverted | gate said merge) + closePrecision: number | null; // P(closed & not reopened | gate said close) + fpRate: number | null; // P(closed or reverted | gate said merge) — gate approved, it was wrong + fnRate: number | null; // P(merged or reopened | gate said close) — gate blocked, it was wrong + reversalRate: number; // share of decided PRs a human reversed +} + +export interface FleetAnalytics { + windowDays: number; + instanceCount: number; // instances meeting MIN_DECIDED + fleet: { + mergePrecision: number | null; + closePrecision: number | null; + fpRate: number | null; + reversalRate: number | null; + cycleP50Ms: number | null; + cycleP95Ms: number | null; + }; + instances: InstanceMetrics[]; + outliers: Array<{ instanceId: string; metric: string; value: number; fleetMedian: number }>; +} + +function median(xs: number[]): number | null { + if (xs.length === 0) return null; + const s = [...xs].sort((a, b) => a - b); + const mid = Math.floor(s.length / 2); + return s.length % 2 === 0 ? (s[mid - 1]! + s[mid]!) / 2 : s[mid]!; +} + +function percentile(sorted: number[], p: number): number | null { + if (sorted.length === 0) return null; + const idx = Math.min(sorted.length - 1, Math.floor((p / 100) * sorted.length)); + return sorted[idx]!; +} + +/** Fold the confusion-matrix cells for one instance into accuracy metrics (reversals count as the gate + * being wrong: a reverted merge is a false positive; a reopened close is a false negative). */ +function foldInstance(instanceId: string, cells: Cell[]): InstanceMetrics { + let wouldMerge = 0, mergeConfirmed = 0, mergeFalse = 0; + let wouldClose = 0, closeConfirmed = 0, closeFalse = 0; + let reversals = 0, decided = 0; + for (const c of cells) { + decided += c.n; + if (c.reversal_flag !== "none") reversals += c.n; + if (c.verdict === "merge") { + wouldMerge += c.n; + if (c.outcome === "merged" && c.reversal_flag !== "reverted") mergeConfirmed += c.n; + else mergeFalse += c.n; + } else if (c.verdict === "close") { + wouldClose += c.n; + if (c.outcome === "closed" && c.reversal_flag !== "reopened") closeConfirmed += c.n; + else closeFalse += c.n; + } + } + return { + instanceId, + decided, + mergePrecision: wouldMerge > 0 ? mergeConfirmed / wouldMerge : null, + closePrecision: wouldClose > 0 ? closeConfirmed / wouldClose : null, + fpRate: wouldMerge > 0 ? mergeFalse / wouldMerge : null, + fnRate: wouldClose > 0 ? closeFalse / wouldClose : null, + reversalRate: reversals / decided, // decided ≥ 1 (the instance has at least one cell) + }; +} + +/** Compute fleet calibration analytics over the collected orb_signals within the window. Fail-safe → empty. */ +export async function computeFleetAnalytics(env: Env, opts: { windowDays?: number } = {}): Promise { + const windowDays = Number.isFinite(opts.windowDays) && (opts.windowDays as number) > 0 ? Math.min(opts.windowDays as number, 365) : 90; + // Date-only cutoff (like computeGateEval) so it compares correctly whether received_at is ISO ('…T…Z') + // or SQLite's CURRENT_TIMESTAMP space format ('YYYY-MM-DD HH:MM:SS'). + const cutoff = new Date(Date.now() - windowDays * 86_400_000).toISOString().slice(0, 10); + + let cells: Cell[] = []; + let cycle: number[] = []; + try { + const matrix = await env.DB + .prepare( + `SELECT instance_id, gate_verdict AS verdict, outcome, reversal_flag, COUNT(*) AS n + FROM orb_signals WHERE received_at >= ? + GROUP BY instance_id, gate_verdict, outcome, reversal_flag`, + ) + .bind(cutoff) + .all(); + cells = matrix.results ?? []; + const cy = await env.DB + .prepare(`SELECT time_to_close_ms AS ms FROM orb_signals WHERE received_at >= ? AND time_to_close_ms IS NOT NULL ORDER BY time_to_close_ms`) + .bind(cutoff) + .all<{ ms: number }>(); + cycle = (cy.results ?? []).map((r) => r.ms); + } catch { + return { windowDays, instanceCount: 0, fleet: { mergePrecision: null, closePrecision: null, fpRate: null, reversalRate: null, cycleP50Ms: null, cycleP95Ms: null }, instances: [], outliers: [] }; + } + + // Group cells by instance, fold each. + const byInstance = new Map(); + for (const c of cells) { + const list = byInstance.get(c.instance_id) ?? []; + list.push(c); + byInstance.set(c.instance_id, list); + } + const instances = [...byInstance.entries()].map(([id, cs]) => foldInstance(id, cs)).sort((a, b) => a.instanceId.localeCompare(b.instanceId)); + + // Fleet = median across instances with enough volume (robust to a single bad contributor). + const eligible = instances.filter((i) => i.decided >= MIN_DECIDED); + const nums = (sel: (i: InstanceMetrics) => number | null): number[] => eligible.map(sel).filter((v): v is number => v !== null); + const fleetMergeP = median(nums((i) => i.mergePrecision)); + const fleetCloseP = median(nums((i) => i.closePrecision)); + + const outliers: FleetAnalytics["outliers"] = []; + if (fleetMergeP !== null) { + for (const i of eligible) { + if (i.mergePrecision !== null && Math.abs(i.mergePrecision - fleetMergeP) > OUTLIER_BAND) { + outliers.push({ instanceId: i.instanceId, metric: "mergePrecision", value: i.mergePrecision, fleetMedian: fleetMergeP }); + } + } + } + + return { + windowDays, + instanceCount: eligible.length, + fleet: { + mergePrecision: fleetMergeP, + closePrecision: fleetCloseP, + fpRate: median(nums((i) => i.fpRate)), + reversalRate: median(nums((i) => i.reversalRate)), + cycleP50Ms: percentile(cycle, 50), + cycleP95Ms: percentile(cycle, 95), + }, + instances, + outliers, + }; +} diff --git a/src/orb/ingest.ts b/src/orb/ingest.ts index 6e91ddbcfb..05f6d946b9 100644 --- a/src/orb/ingest.ts +++ b/src/orb/ingest.ts @@ -1,18 +1,24 @@ -// Gittensory Orb (#1219) — central collector receiver. -// Accepts anonymized outcome signal batches from self-hosted instances running exportOrbBatch. -// No raw repo names, owner identifiers, or PR content is accepted or stored — only HMAC-anonymized -// hashes + aggregate outcome metadata (verdict, timing). +// Gittensory Orb (#1255) — central fleet-calibration collector receiver. +// Accepts anonymized, reversal-aware outcome batches from self-hosted instances (exportOrbBatch). +// No raw repo names, owner identifiers, commit SHAs, or PR content — only HMAC-anonymized hashes + +// aggregate calibration metadata (verdict, outcome, reversal, bucketed reason, cycle time). const MAX_BATCH = 500; const VALID_OUTCOMES = new Set(["merged", "closed"]); +const VALID_REVERSALS = new Set(["none", "reopened", "reverted"]); +const MIN_CYCLE_MS = 1_000; // <1s is implausible +const MAX_CYCLE_MS = 31_536_000_000; // >1y is implausible interface OrbIngestEvent { repo_hash: string; pr_hash: string; - outcome: string; gate_verdict?: string | null; + outcome: string; + reversal_flag?: string | null; + gate_reasoncode_bucket?: string | null; time_to_close_ms?: number | null; - created_at?: string | null; + decision_timestamp?: string | null; + outcome_timestamp?: string | null; } interface OrbIngestPayload { @@ -22,6 +28,13 @@ interface OrbIngestPayload { export type OrbIngestResult = { accepted: number } | { error: string }; +/** Clamp a sender-supplied cycle time to a plausible range; null for anything implausible/absent. */ +function clampCycleMs(value: unknown): number | null { + if (typeof value !== "number" || !Number.isFinite(value)) return null; + if (value < MIN_CYCLE_MS || value > MAX_CYCLE_MS) return null; + return Math.round(value); +} + export async function handleOrbIngest(body: string, db: D1Database): Promise { let payload: unknown; try { @@ -54,21 +67,31 @@ export async function handleOrbIngest(body: string, db: D1Database): Promise 0) accepted++; diff --git a/src/selfhost/orb-collector.ts b/src/selfhost/orb-collector.ts index d062323853..e7c16c79c2 100644 --- a/src/selfhost/orb-collector.ts +++ b/src/selfhost/orb-collector.ts @@ -1,120 +1,163 @@ -// Gittensory Orb (#1219) — local outcome-signal collector. Records gate verdict + final PR -// outcome (merged/closed) for every PR the engine reviewed, enabling calibration of gate -// thresholds and AI prompts from real-world feedback signals. +// Gittensory Orb (#1255) — fleet calibration EXPORTER. Each self-hosted instance already records de-noised +// ground truth in review_audit (gate_decision + pr_outcome + reversal_reopened/reversal_reverted) via the +// engine's outcomes-wire. This ships an anonymized, reversal-aware signal UP to gittensory's central +// collector so the gate can be calibrated across the whole self-host fleet. // -// Collection is always local (DB only). Export to the central collector is opt-in: -// ORB_ENABLED=true — activates collection (off by default) -// ORB_COLLECTOR_URL= — endpoint to export batches to (default: gittensory's hosted collector) -// ORB_AIR_GAP=true — keep all events local, never send externally -// ORB_ANONYMIZE=true — HMAC-hash repo/owner before export (default: true) +// ORB_ENABLED=true — activates export (off by default) +// ORB_COLLECTOR_URL= — endpoint (default: gittensory's hosted collector) +// ORB_AIR_GAP=true — keep everything local, never send externally +// ORB_ANONYMIZE=true — HMAC-hash repo/PR before export (default: true) // -// Nothing is ever sent without ORB_ENABLED=true. No diffs, no code, no comments, no user -// identifiers — only aggregate outcome metadata (repo-hash, verdict, outcome, timing). +// No diffs, no code, no comments, no logins, no commit SHAs — only verdict + outcome + reversal + a bucketed +// reason category + cycle time, with repo/PR identifiers HMAC'd by THIS instance's own secret (the collector +// holds no instance secret, so it can never de-anonymize). import { createHash, createHmac } from "node:crypto"; import { incr } from "./metrics"; -export interface OrbEvent { - repo: string; - pr_number: number; - head_sha: string; - outcome: "merged" | "closed"; - gate_verdict?: string; - time_to_close_ms?: number; +/** One de-noised, resolved-PR row read from review_audit (the join below). */ +interface FleetRow { + project: string; // repo full name (review_audit.project) + target_id: string; // `repo#pr` + verdict: string | null; // gate_decision.decision: merge | close | hold + reasoncode: string | null; // gate_decision.summary (raw — bucketed before export) + decided_at: string; // gate_decision.created_at — non-null (NOT NULL column + inner join) + outcome: string; // pr_outcome.decision: merged | closed + outcome_at: string; + reverted: number; // 0|1 + reopened: number; // 0|1 + event_at: string; // max(outcome_at, latest reversal time) — the export watermark unit } -interface OrbRow { - id: number; - repo: string; - pr_number: number; - head_sha: string; - outcome: string; +interface FleetEvent { + repo_hash: string; + pr_hash: string; gate_verdict: string | null; + outcome: string; + reversal_flag: "none" | "reopened" | "reverted"; + gate_reasoncode_bucket: string; time_to_close_ms: number | null; - created_at: string; - exported_at: string | null; + decision_timestamp: string | null; + outcome_timestamp: string; } interface OrbExportPayload { instance_id: string; - events: Array<{ - repo_hash: string; - pr_hash: string; - outcome: string; - gate_verdict: string | null; - time_to_close_ms: number | null; - created_at: string; - }>; + events: FleetEvent[]; } -/** Stable instance identifier (hash of the Orb App ID — no PII). */ +/** Stable instance identifier (hash of the Orb/App ID — no PII). */ function instanceId(): string { - return createHash("sha256").update(process.env.ORB_APP_ID ?? "unknown").digest("hex").slice(0, 16); + return createHash("sha256").update(process.env.ORB_APP_ID ?? process.env.GITHUB_APP_ID ?? "unknown").digest("hex").slice(0, 16); } -/** HMAC a string with the webhook secret for anonymized export. */ +/** HMAC a string with the instance's own secret for anonymized export. */ function hmacField(value: string, secret: string): string { return createHmac("sha256", secret).update(value).digest("hex").slice(0, 24); } -/** Returns true only when Orb collection is explicitly enabled. */ +/** Map the gate's free-text reasonCode to a fixed, low-cardinality category — done at the source so the raw + * (possibly repo-specific) reason string never leaves the instance. */ +export function bucketReasonCode(summary: string | null | undefined): string { + if (!summary) return "none"; + const s = summary.toLowerCase(); + if (s.includes("linked_issue") || s.includes("linked issue")) return "issue_policy"; + if (s.includes("duplicate")) return "duplicate_risk"; + if (s.includes("slop")) return "slop_advisory"; + if (s.includes("ai_review") || s.includes("ai_consensus") || s.includes("consensus")) return "ai_quality"; + if (s.includes("self_authored") || s.includes("author") || s.includes("maintainer_cut")) return "author_policy"; + if (s.includes("ci_") || s.includes("ci state") || s.includes("ci passed")) return "ci_readiness"; + return "other"; +} + +/** Returns true only when Orb export is explicitly enabled. */ export function orbEnabled(): boolean { const v = (process.env.ORB_ENABLED ?? "").toLowerCase(); return v === "true" || v === "1" || v === "yes"; } -/** Record a single outcome event in the local DB. No-op when ORB_ENABLED is false. */ -export async function recordOrbEvent(db: D1Database, event: OrbEvent): Promise { - if (!orbEnabled()) return; - try { - await db - .prepare( - `INSERT OR IGNORE INTO orb_events (repo, pr_number, head_sha, outcome, gate_verdict, time_to_close_ms) - VALUES (?, ?, ?, ?, ?, ?)`, - ) - .bind(event.repo, event.pr_number, event.head_sha, event.outcome, event.gate_verdict ?? null, event.time_to_close_ms ?? null) - .run(); - incr("gittensory_orb_events_recorded_total"); - } catch { - // best-effort — never let Orb collection crash job processing - } +// Latest gate_decision + latest pr_outcome per target_id, plus any reversal — portable (window functions + +// CASE, no SQLite-only bare-column-with-MAX) so it runs on the self-host SQLite OR Postgres backend. +const FLEET_QUERY = ` + WITH gd AS ( + SELECT target_id, project, decision AS verdict, summary AS reasoncode, created_at AS decided_at, + ROW_NUMBER() OVER (PARTITION BY target_id ORDER BY created_at DESC) AS rn + FROM review_audit + WHERE event_type = 'gate_decision' AND decision IS NOT NULL AND source = 'gittensory-native' + ), + po AS ( + SELECT target_id, decision AS outcome, created_at AS outcome_at, + ROW_NUMBER() OVER (PARTITION BY target_id ORDER BY created_at DESC) AS rn + FROM review_audit + WHERE event_type = 'pr_outcome' AND decision IS NOT NULL + ), + rev AS ( + SELECT target_id, + MAX(CASE WHEN event_type = 'reversal_reverted' THEN 1 ELSE 0 END) AS reverted, + MAX(CASE WHEN event_type = 'reversal_reopened' THEN 1 ELSE 0 END) AS reopened, + MAX(created_at) AS rev_at + FROM review_audit + WHERE event_type IN ('reversal_reverted', 'reversal_reopened') + GROUP BY target_id + ) + SELECT project, target_id, verdict, reasoncode, decided_at, outcome, outcome_at, reverted, reopened, event_at + FROM ( + SELECT gd.project AS project, gd.target_id AS target_id, gd.verdict AS verdict, gd.reasoncode AS reasoncode, + gd.decided_at AS decided_at, po.outcome AS outcome, po.outcome_at AS outcome_at, + COALESCE(rev.reverted, 0) AS reverted, COALESCE(rev.reopened, 0) AS reopened, + CASE WHEN rev.rev_at IS NOT NULL AND rev.rev_at > po.outcome_at THEN rev.rev_at ELSE po.outcome_at END AS event_at + FROM gd + JOIN po ON gd.target_id = po.target_id + LEFT JOIN rev ON gd.target_id = rev.target_id + WHERE gd.rn = 1 AND po.rn = 1 + ) AS resolved + WHERE event_at > ? + ORDER BY event_at + LIMIT ?`; + +/** ms between the gate decision and the resolution; null if implausible (NaN or negative). */ +function cycleTimeMs(decidedAt: string, outcomeAt: string): number | null { + const ms = new Date(outcomeAt).getTime() - new Date(decidedAt).getTime(); + return Number.isFinite(ms) && ms >= 0 ? ms : null; } /** - * Export pending Orb events to the central collector. Called periodically (e.g. hourly). - * Reads up to `batchSize` unexported events, signs and POSTs them, marks them as exported. - * Returns the number of events exported (0 if air-gap, disabled, or nothing pending). + * Export newly-resolved PR outcomes (since this instance's watermark) to the central collector. Reads from + * review_audit (de-noised, reversal-aware), anonymizes, signs, POSTs, then advances the cursor. + * Returns the number of events exported (0 if air-gap, disabled, or nothing new). */ -export async function exportOrbBatch( - db: D1Database, - batchSize = 200, - fetchFn: typeof fetch = fetch, -): Promise { +export async function exportOrbBatch(db: D1Database, batchSize = 200, fetchFn: typeof fetch = fetch): Promise { if (!orbEnabled()) return 0; if ((process.env.ORB_AIR_GAP ?? "").toLowerCase() === "true") return 0; - // gittensory's hosted collector (the deployed /v1/orb/ingest receiver). No shared secret is sent: - // the batch is anonymized (HMAC of each operator's OWN ORB_WEBHOOK_SECRET) and accepted as untrusted, - // rate-limited telemetry. Override only to point at your own self-hosted collector. + // gittensory's hosted collector. No shared secret is sent: repo/PR identifiers are HMAC'd with THIS + // instance's own ORB_WEBHOOK_SECRET, and the collector accepts the batch as untrusted, rate-limited telemetry. const collectorUrl = process.env.ORB_COLLECTOR_URL ?? "https://gittensory-api.aethereal.dev/v1/orb/ingest"; const secret = process.env.ORB_WEBHOOK_SECRET ?? ""; const anonymize = (process.env.ORB_ANONYMIZE ?? "true").toLowerCase() !== "false"; + const instance = instanceId(); - const { results } = await db - .prepare(`SELECT * FROM orb_events WHERE exported_at IS NULL ORDER BY id LIMIT ?`) - .bind(batchSize) - .all(); + // Read this instance's export watermark (resumes where the last run left off). + const cursorRow = await db + .prepare(`SELECT last_exported_at FROM orb_export_cursor WHERE instance_hash = ?`) + .bind(instance) + .first<{ last_exported_at: string }>(); + const cursor = cursorRow?.last_exported_at ?? "2000-01-01T00:00:00Z"; + const { results } = await db.prepare(FLEET_QUERY).bind(cursor, batchSize).all(); if (!results || results.length === 0) return 0; const payload: OrbExportPayload = { - instance_id: instanceId(), + instance_id: instance, events: results.map((r) => ({ - repo_hash: anonymize ? hmacField(r.repo, secret) : r.repo, - pr_hash: anonymize ? hmacField(`${r.repo}#${r.pr_number}`, secret) : String(r.pr_number), + repo_hash: anonymize ? hmacField(r.project, secret) : r.project, + pr_hash: anonymize ? hmacField(r.target_id, secret) : r.target_id, + gate_verdict: r.verdict, outcome: r.outcome, - gate_verdict: r.gate_verdict, - time_to_close_ms: r.time_to_close_ms, - created_at: r.created_at, + reversal_flag: r.reverted ? "reverted" : r.reopened ? "reopened" : "none", + gate_reasoncode_bucket: bucketReasonCode(r.reasoncode), + time_to_close_ms: cycleTimeMs(r.decided_at, r.outcome_at), + decision_timestamp: r.decided_at, + outcome_timestamp: r.outcome_at, })), }; @@ -124,11 +167,7 @@ export async function exportOrbBatch( try { const res = await fetchFn(collectorUrl, { method: "POST", - headers: { - "content-type": "application/json", - "x-orb-signature": `sha256=${signature}`, - "x-orb-instance": instanceId(), - }, + headers: { "content-type": "application/json", "x-orb-signature": `sha256=${signature}`, "x-orb-instance": instance }, body, }); if (!res.ok) { @@ -140,12 +179,13 @@ export async function exportOrbBatch( return 0; } - // Mark all exported events - const ids = results.map((r) => r.id); - const placeholders = ids.map(() => "?").join(","); - const now = new Date().toISOString(); - await db.prepare(`UPDATE orb_events SET exported_at=? WHERE id IN (${placeholders})`).bind(now, ...ids).run(); + // Advance the watermark to the newest event in this batch (rows are ordered by event_at ascending). + const newWatermark = results[results.length - 1]!.event_at; + await db + .prepare(`INSERT OR REPLACE INTO orb_export_cursor (instance_hash, last_exported_at, updated_at) VALUES (?, ?, ?)`) + .bind(instance, newWatermark, new Date().toISOString()) + .run(); - incr("gittensory_orb_events_exported_total", {}, ids.length); - return ids.length; + incr("gittensory_orb_events_exported_total", {}, results.length); + return results.length; } diff --git a/src/selfhost/orb-setup.ts b/src/selfhost/orb-setup.ts deleted file mode 100644 index 1ad8c7c6a4..0000000000 --- a/src/selfhost/orb-setup.ts +++ /dev/null @@ -1,66 +0,0 @@ -// Gittensory Orb (#1219) setup wizard. Mirrors setup-wizard.ts but for the lightweight -// "Gittensory Orb" GitHub App — pull_requests:read + metadata:read + pull_request + -// installation events only. Creates a separate App so operators can install Orb -// independently of the main review App, and revoke data collection without touching reviews. -// -// Routes (server.ts): GET /orb/setup → form page; GET /orb/setup/callback → exchange code. - -export interface OrbCredentials { - id: number; - slug: string; - webhook_secret: string; - pem: string; -} - -/** Minimal Orb App manifest — read-only permissions, no write capabilities. */ -export function buildOrbManifest(origin: string, state: string): Record { - const base = origin.replace(/\/+$/, ""); - return { - name: "Gittensory Orb", - url: base, - hook_attributes: { url: `${base}/orb/webhook` }, - redirect_url: `${base}/orb/setup/callback?state=${encodeURIComponent(state)}`, - public: false, - default_permissions: { - pull_requests: "read", - metadata: "read", - }, - default_events: ["pull_request", "installation", "installation_repositories"], - }; -} - -/** HTML page with a single button that POSTs the manifest to GitHub's App-creation flow. */ -export function renderOrbSetupPage(origin: string, state: string): string { - const manifest = JSON.stringify(buildOrbManifest(origin, state)).replace(/'/g, "'"); - return `Gittensory Orb setup - -

Gittensory Orb setup

-

This creates a lightweight read-only GitHub App that observes PR outcomes for local calibration -and optional aggregate telemetry. Install it on the same repositories as your main Gittensory App. -GitHub will redirect back here with the credentials — then restart the container to activate collection.

-
- - -
-`; -} - -/** Exchange a one-time manifest code (from GitHub's callback) for the App's credentials. */ -export async function exchangeOrbManifestCode(code: string, fetchImpl: typeof fetch = fetch): Promise { - const res = await fetchImpl(`https://api.github.com/app-manifests/${encodeURIComponent(code)}/conversions`, { - method: "POST", - headers: { accept: "application/vnd.github+json", "user-agent": "gittensory-selfhost" }, - }); - if (!res.ok) throw new Error(`orb_manifest_exchange_http_${res.status}`); - return (await res.json()) as OrbCredentials; -} - -/** Serialize Orb credentials as env-file lines for the operator to load. */ -export function orbCredentialsToEnv(creds: OrbCredentials): string { - return [ - `ORB_APP_ID=${creds.id}`, - `ORB_APP_SLUG=${creds.slug}`, - `ORB_WEBHOOK_SECRET=${creds.webhook_secret}`, - `ORB_PRIVATE_KEY=${JSON.stringify(creds.pem)}`, - ].join("\n") + "\n"; -} diff --git a/src/selfhost/orb-webhook.ts b/src/selfhost/orb-webhook.ts deleted file mode 100644 index 6e328960a8..0000000000 --- a/src/selfhost/orb-webhook.ts +++ /dev/null @@ -1,137 +0,0 @@ -// Gittensory Orb (#1219) webhook dispatcher — handles pull_request + installation events -// from the lightweight Orb GitHub App. Verifies HMAC-SHA256 (x-hub-signature-256), -// tracks per-repo installations, and records PR outcome signals on pull_request.closed. - -import { createHmac, timingSafeEqual } from "node:crypto"; -import { incr } from "./metrics"; -import { recordOrbEvent } from "./orb-collector"; - -/** Verify a GitHub webhook signature (x-hub-signature-256: sha256=). */ -export function verifyOrbSignature(payload: string, sig: string, secret: string): boolean { - if (!sig.startsWith("sha256=")) return false; - const expected = createHmac("sha256", secret).update(payload).digest("hex"); - const actual = sig.slice("sha256=".length); - try { - return timingSafeEqual(Buffer.from(expected, "hex"), Buffer.from(actual, "hex")); - } catch { - return false; - } -} - -/** Look up the most recent gate verdict for a repo+PR from the review_targets table. */ -export async function lookupGateVerdict(db: D1Database, repo: string, prNumber: number): Promise { - try { - const row = await db - .prepare(`SELECT verdict FROM review_targets WHERE repo = ? AND number = ? ORDER BY updated_at DESC LIMIT 1`) - .bind(repo, prNumber) - .first<{ verdict: string | null }>(); - return row?.verdict ?? null; - } catch { - return null; - } -} - -interface InstallationPayload { - action: string; - installation: { id: number }; - repositories?: Array<{ full_name: string }>; - repositories_added?: Array<{ full_name: string }>; - repositories_removed?: Array<{ full_name: string }>; -} - -interface PullRequestPayload { - action: string; - pull_request: { - number: number; - head: { sha: string }; - merged: boolean; - created_at: string; - closed_at: string | null; - }; - repository: { full_name: string }; -} - -/** Main dispatcher. Returns the HTTP status + body to reply with. */ -export async function handleOrbWebhook( - event: string, - payload: string, - db: D1Database, -): Promise<{ status: number; body: string }> { - incr("gittensory_orb_webhook_total"); - - if (event === "installation" || event === "installation_repositories") { - return handleInstallation(JSON.parse(payload) as InstallationPayload, db); - } - - if (event === "pull_request") { - const body = JSON.parse(payload) as PullRequestPayload; - if (body.action === "closed") return handlePrClosed(body, db); - return { status: 204, body: "" }; - } - - return { status: 204, body: "" }; -} - -async function handleInstallation( - body: InstallationPayload, - db: D1Database, -): Promise<{ status: number; body: string }> { - const installationId = body.installation.id; - const now = new Date().toISOString(); - - if (body.action === "created" || body.action === "added") { - const repos = [...(body.repositories ?? []), ...(body.repositories_added ?? [])]; - for (const r of repos) { - try { - await db - .prepare(`INSERT OR IGNORE INTO orb_installations (installation_id, repo, installed_at) VALUES (?, ?, ?)`) - .bind(installationId, r.full_name, now) - .run(); - } catch { /* best-effort — never crash on install tracking */ } - } - if (repos.length) incr("gittensory_orb_installs_total", {}, repos.length); - } - - if (body.action === "deleted" || body.action === "removed") { - const repos = body.action === "deleted" - ? (body.repositories ?? []) - : (body.repositories_removed ?? []); - for (const r of repos) { - try { - await db - .prepare(`UPDATE orb_installations SET removed_at = ? WHERE installation_id = ? AND repo = ? AND removed_at IS NULL`) - .bind(now, installationId, r.full_name) - .run(); - } catch { /* best-effort */ } - } - } - - return { status: 204, body: "" }; -} - -async function handlePrClosed( - body: PullRequestPayload, - db: D1Database, -): Promise<{ status: number; body: string }> { - const repo = body.repository.full_name; - const prNumber = body.pull_request.number; - const headSha = body.pull_request.head.sha; - const outcome: "merged" | "closed" = body.pull_request.merged ? "merged" : "closed"; - - const closedMs = body.pull_request.closed_at ? new Date(body.pull_request.closed_at).getTime() : null; - const createdMs = body.pull_request.created_at ? new Date(body.pull_request.created_at).getTime() : null; - const timeToCloseMs = closedMs !== null && createdMs !== null ? closedMs - createdMs : undefined; - - const gateVerdict = await lookupGateVerdict(db, repo, prNumber); - - await recordOrbEvent(db, { - repo, - pr_number: prNumber, - head_sha: headSha, - outcome, - ...(gateVerdict !== null ? { gate_verdict: gateVerdict } : {}), - ...(timeToCloseMs !== undefined ? { time_to_close_ms: timeToCloseMs } : {}), - }); - - return { status: 204, body: "" }; -} diff --git a/src/selfhost/pg-dialect.ts b/src/selfhost/pg-dialect.ts index 82ef3dc97c..9bda2c7221 100644 --- a/src/selfhost/pg-dialect.ts +++ b/src/selfhost/pg-dialect.ts @@ -10,6 +10,8 @@ const REPLACE_CONFLICT_KEYS: Record = { system_flags: ["key"], tunables_overrides: ["project"], tunables_overrides_shadow: ["project"], + orb_export_cursor: ["instance_hash"], + orb_signals: ["instance_id", "repo_hash", "pr_hash"], }; /** Replace `?` placeholders with `$1,$2,…`, skipping any `?` inside single-quoted string literals. */ diff --git a/src/server.ts b/src/server.ts index 6ceb040a3c..5d3bef42d1 100644 --- a/src/server.ts +++ b/src/server.ts @@ -14,8 +14,6 @@ import worker from "./index"; import { processJob } from "./queue/processors"; import { createSelfHostAi } from "./selfhost/ai"; import { credentialsToEnv, exchangeManifestCode, renderSetupPage } from "./selfhost/setup-wizard"; -import { exchangeOrbManifestCode, orbCredentialsToEnv, renderOrbSetupPage } from "./selfhost/orb-setup"; -import { handleOrbWebhook, verifyOrbSignature } from "./selfhost/orb-webhook"; import { orbEnabled, exportOrbBatch } from "./selfhost/orb-collector"; import { createD1Adapter, nodeSqliteDriver } from "./selfhost/d1-adapter"; import { readiness } from "./selfhost/health"; @@ -189,9 +187,7 @@ async function main(): Promise { "gittensory_jobs_failed_total", "gittensory_jobs_dead_total", "gittensory_http_requests_total", "gittensory_webhook_dedup_total", "gittensory_qdrant_queries_total", "gittensory_qdrant_upserts_total", - "gittensory_orb_webhook_total", "gittensory_orb_installs_total", - "gittensory_orb_events_recorded_total", "gittensory_orb_events_exported_total", - "gittensory_orb_export_errors_total", + "gittensory_orb_events_exported_total", "gittensory_orb_export_errors_total", ]) incr(c, undefined, 0); @@ -254,55 +250,6 @@ async function main(): Promise { return new Response(`setup failed: ${error instanceof Error ? error.message : "error"}`, { status: 500 }); } } - // Gittensory Orb setup wizard — only while no Orb App is configured. - if ((path === "/orb/setup" || path === "/orb/setup/callback") && !process.env.ORB_APP_ID) { - // Same guard as the main setup wizard: PUBLIC_API_ORIGIN required to prevent Host-header spoofing. - const origin = process.env.PUBLIC_API_ORIGIN; - if (!origin) { - return new Response( - "PUBLIC_API_ORIGIN must be set before using the Orb setup wizard — add it to your .env file", - { status: 400 }, - ); - } - if (path === "/orb/setup") { - const state = randomUUID(); - return new Response(renderOrbSetupPage(origin, state), { - headers: { - "content-type": "text/html; charset=utf-8", - "Set-Cookie": `orb_setup_state=${state}; Path=/orb/setup; HttpOnly; SameSite=Lax; Max-Age=3600`, - }, - }); - } - const params = new URL(request.url).searchParams; - const code = params.get("code"); - if (!code) return new Response("missing ?code", { status: 400 }); - const stateParam = params.get("state"); - const cookieHeader = request.headers.get("cookie") ?? ""; - const cookieState = cookieHeader.split(";").map((c) => c.trim()).find((c) => c.startsWith("orb_setup_state="))?.slice("orb_setup_state=".length); - if (!stateParam || !cookieState || stateParam !== cookieState) { - return new Response("invalid state parameter", { status: 403 }); - } - try { - const creds = await exchangeOrbManifestCode(code); - const outPath = process.env.ORB_SETUP_OUTPUT_PATH ?? "/data/gittensory-orb.env"; - writeFileSync(outPath, orbCredentialsToEnv(creds), { mode: 0o600 }); - console.log(JSON.stringify({ event: "selfhost_orb_created", slug: creds.slug, app_id: creds.id })); - return new Response(`

Gittensory Orb App created ✓

Credentials written to ${outPath}. Add them to your .env (or load the file), install the Orb App on your repos, and restart the container.

`, { headers: { "content-type": "text/html; charset=utf-8" } }); - } catch (error) { - return new Response(`orb setup failed: ${error instanceof Error ? error.message : "error"}`, { status: 500 }); - } - } - // Orb webhook endpoint — receives pull_request + installation events from the Orb App. - if (path === "/orb/webhook" && request.method === "POST" && process.env.ORB_WEBHOOK_SECRET) { - const payload = await request.text(); - const sig = request.headers.get("x-hub-signature-256") ?? ""; - if (!verifyOrbSignature(payload, sig, process.env.ORB_WEBHOOK_SECRET)) { - return new Response("signature mismatch", { status: 401 }); - } - const event = request.headers.get("x-github-event") ?? ""; - const result = await handleOrbWebhook(event, payload, backend.db); - return new Response(result.body || null, { status: result.status }); - } incr("gittensory_http_requests_total"); // Webhook delivery dedup: return 204 immediately for already-processed delivery IDs. // We mark only AFTER a successful response — failed/rejected webhooks must be retryable. diff --git a/test/integration/orb-ingest.test.ts b/test/integration/orb-ingest.test.ts index 1a25c33a42..fb557a1c6b 100644 --- a/test/integration/orb-ingest.test.ts +++ b/test/integration/orb-ingest.test.ts @@ -3,240 +3,171 @@ import { createApp } from "../../src/api/routes"; import { handleOrbIngest } from "../../src/orb/ingest"; import { createTestEnv, TestD1Database } from "../helpers/d1"; -// ── handleOrbIngest unit-style tests ────────────────────────────────────────── - describe("handleOrbIngest()", () => { function makeDb(): D1Database { return new TestD1Database() as unknown as D1Database; } + const ev = (o: Record = {}) => ({ repo_hash: "rh", pr_hash: "ph", outcome: "merged", ...o }); + const ingest = (db: D1Database, events: Array>, instance_id = "inst1") => handleOrbIngest(JSON.stringify({ instance_id, events }), db); + const col = async (db: D1Database, pr: string, c: string) => + (await (db as unknown as TestD1Database).prepare(`SELECT ${c} AS v FROM orb_signals WHERE pr_hash=?`).bind(pr).first<{ v: unknown }>())?.v; - function makePayload(overrides: Record = {}): string { - return JSON.stringify({ - instance_id: "abc123def456abc0", - events: [ - { - repo_hash: "a1b2c3d4e5f6a1b2c3d4e5f6", - pr_hash: "f6e5d4c3b2a1f6e5d4c3b2a1", - outcome: "merged", - gate_verdict: "approve", - time_to_close_ms: 3600000, - created_at: "2024-01-01T00:00:00Z", - }, - ], - ...overrides, - }); - } - - it("accepts a valid batch and returns accepted count", async () => { - const db = makeDb(); - const result = await handleOrbIngest(makePayload(), db); - expect(result).toEqual({ accepted: 1 }); + it("accepts a valid batch and returns the accepted count", async () => { + expect(await ingest(makeDb(), [ev({ pr_hash: "p1" })])).toEqual({ accepted: 1 }); }); - it("returns invalid_json when body is not valid JSON (covers JSON.parse catch branch)", async () => { - const db = makeDb(); - expect(await handleOrbIngest("{not json}", db)).toEqual({ error: "invalid_json" }); + it("returns invalid_json on unparseable body", async () => { + expect(await handleOrbIngest("{not json}", makeDb())).toEqual({ error: "invalid_json" }); }); - it("returns invalid_payload when instance_id is not a string", async () => { + it("returns invalid_payload: instance_id not a string / events not an array / empty instance / empty events", async () => { const db = makeDb(); expect(await handleOrbIngest(JSON.stringify({ instance_id: 123, events: [] }), db)).toEqual({ error: "invalid_payload" }); - }); - - it("returns invalid_payload when events is not an array (covers !Array.isArray branch)", async () => { - const db = makeDb(); expect(await handleOrbIngest(JSON.stringify({ instance_id: "abc", events: "bad" }), db)).toEqual({ error: "invalid_payload" }); + expect(await handleOrbIngest(JSON.stringify({ instance_id: "", events: [ev()] }), db)).toEqual({ error: "invalid_payload" }); + expect(await handleOrbIngest(JSON.stringify({ instance_id: "abc", events: [] }), db)).toEqual({ error: "invalid_payload" }); }); - it("returns invalid_payload when instance_id is an empty string (covers !instance_id branch)", async () => { - const db = makeDb(); - expect(await handleOrbIngest(JSON.stringify({ instance_id: "", events: [{ repo_hash: "a", pr_hash: "b", outcome: "merged" }] }), db)).toEqual({ error: "invalid_payload" }); + it("skips events with bad repo_hash / pr_hash / outcome", async () => { + expect(await ingest(makeDb(), [ev({ repo_hash: 99 })])).toEqual({ accepted: 0 }); + expect(await ingest(makeDb(), [ev({ repo_hash: "" })])).toEqual({ accepted: 0 }); + expect(await ingest(makeDb(), [ev({ pr_hash: null })])).toEqual({ accepted: 0 }); + expect(await ingest(makeDb(), [ev({ pr_hash: "" })])).toEqual({ accepted: 0 }); + expect(await ingest(makeDb(), [ev({ outcome: "opened" })])).toEqual({ accepted: 0 }); }); - it("returns invalid_payload when events array is empty (covers events.length === 0 branch)", async () => { + it("stores gate_verdict string vs null", async () => { const db = makeDb(); - expect(await handleOrbIngest(JSON.stringify({ instance_id: "abc", events: [] }), db)).toEqual({ error: "invalid_payload" }); + await ingest(db, [ev({ pr_hash: "v1", gate_verdict: "merge" }), ev({ pr_hash: "v2" })]); + expect(await col(db, "v1", "gate_verdict")).toBe("merge"); + expect(await col(db, "v2", "gate_verdict")).toBeNull(); }); - it("skips events with a non-string repo_hash (covers typeof repo_hash !== string branch)", async () => { + it("whitelists reversal_flag: valid kept, invalid + absent → 'none'", async () => { const db = makeDb(); - const result = await handleOrbIngest( - JSON.stringify({ instance_id: "inst1", events: [{ repo_hash: 99, pr_hash: "hash", outcome: "merged" }] }), - db, - ); - expect(result).toEqual({ accepted: 0 }); + await ingest(db, [ + ev({ pr_hash: "r1", reversal_flag: "reverted" }), + ev({ pr_hash: "r2", reversal_flag: "bogus" }), + ev({ pr_hash: "r3" }), + ]); + expect(await col(db, "r1", "reversal_flag")).toBe("reverted"); + expect(await col(db, "r2", "reversal_flag")).toBe("none"); + expect(await col(db, "r3", "reversal_flag")).toBe("none"); }); - it("skips events with an empty repo_hash (covers !repo_hash branch)", async () => { + it("stores gate_reasoncode_bucket string vs null", async () => { const db = makeDb(); - const result = await handleOrbIngest( - JSON.stringify({ instance_id: "inst1", events: [{ repo_hash: "", pr_hash: "hash", outcome: "merged" }] }), - db, - ); - expect(result).toEqual({ accepted: 0 }); + await ingest(db, [ev({ pr_hash: "b1", gate_reasoncode_bucket: "duplicate_risk" }), ev({ pr_hash: "b2" })]); + expect(await col(db, "b1", "gate_reasoncode_bucket")).toBe("duplicate_risk"); + expect(await col(db, "b2", "gate_reasoncode_bucket")).toBeNull(); }); - it("skips events with a non-string pr_hash (covers typeof pr_hash !== string branch)", async () => { + it("clamps time_to_close_ms: valid kept; absent / <1s / >1y → null", async () => { const db = makeDb(); - const result = await handleOrbIngest( - JSON.stringify({ instance_id: "inst1", events: [{ repo_hash: "rhash", pr_hash: null, outcome: "merged" }] }), - db, - ); - expect(result).toEqual({ accepted: 0 }); + await ingest(db, [ + ev({ pr_hash: "c1", time_to_close_ms: 7_200_000 }), + ev({ pr_hash: "c2" }), + ev({ pr_hash: "c3", time_to_close_ms: 500 }), + ev({ pr_hash: "c4", time_to_close_ms: 40_000_000_000 }), + ev({ pr_hash: "c5", time_to_close_ms: "nope" }), + ]); + expect(await col(db, "c1", "time_to_close_ms")).toBe(7_200_000); + expect(await col(db, "c2", "time_to_close_ms")).toBeNull(); + expect(await col(db, "c3", "time_to_close_ms")).toBeNull(); + expect(await col(db, "c4", "time_to_close_ms")).toBeNull(); + expect(await col(db, "c5", "time_to_close_ms")).toBeNull(); }); - it("skips events with an empty pr_hash (covers !pr_hash branch)", async () => { + it("stores decision_timestamp + outcome_timestamp (and mirrors outcome_timestamp to sent_at) — string vs null", async () => { const db = makeDb(); - const result = await handleOrbIngest( - JSON.stringify({ instance_id: "inst1", events: [{ repo_hash: "rhash", pr_hash: "", outcome: "merged" }] }), - db, - ); - expect(result).toEqual({ accepted: 0 }); + await ingest(db, [ + ev({ pr_hash: "t1", decision_timestamp: "2026-01-01T00:00:00Z", outcome_timestamp: "2026-01-01T01:00:00Z" }), + ev({ pr_hash: "t2" }), + ]); + expect(await col(db, "t1", "decision_timestamp")).toBe("2026-01-01T00:00:00Z"); + expect(await col(db, "t1", "outcome_timestamp")).toBe("2026-01-01T01:00:00Z"); + expect(await col(db, "t1", "sent_at")).toBe("2026-01-01T01:00:00Z"); + expect(await col(db, "t2", "decision_timestamp")).toBeNull(); + expect(await col(db, "t2", "sent_at")).toBeNull(); }); - it("skips events with an invalid outcome (covers !VALID_OUTCOMES.has branch)", async () => { + it("UPSERTs on (instance, repo_hash, pr_hash): a re-export updates the freshest outcome (e.g. a later reversal)", async () => { const db = makeDb(); - const result = await handleOrbIngest( - JSON.stringify({ instance_id: "inst1", events: [{ repo_hash: "rh", pr_hash: "ph", outcome: "opened" }] }), - db, - ); - expect(result).toEqual({ accepted: 0 }); + await ingest(db, [ev({ pr_hash: "u1", reversal_flag: "none" })]); + expect(await col(db, "u1", "reversal_flag")).toBe("none"); + // same PR re-exported with a reversal now present + const second = await ingest(db, [ev({ pr_hash: "u1", reversal_flag: "reverted" })]); + expect(second).toEqual({ accepted: 1 }); // OR REPLACE counts as a write + expect(await col(db, "u1", "reversal_flag")).toBe("reverted"); + const cnt = await (db as unknown as TestD1Database).prepare("SELECT COUNT(*) AS n FROM orb_signals WHERE pr_hash='u1'").first<{ n: number }>(); + expect(cnt?.n).toBe(1); // still one row (upsert, not duplicate) }); - it("stores null gate_verdict when field is absent (covers typeof gate_verdict !== string branch)", async () => { + it("different instances reviewing the same repo#pr do NOT collide", async () => { const db = makeDb(); - await handleOrbIngest( - JSON.stringify({ instance_id: "inst1", events: [{ repo_hash: "rh1", pr_hash: "ph1", outcome: "closed" }] }), - db, - ); - const row = await (db as unknown as TestD1Database).prepare("SELECT gate_verdict FROM orb_signals WHERE pr_hash='ph1'").first<{ gate_verdict: string | null }>(); - expect(row?.gate_verdict).toBeNull(); + await ingest(db, [ev({ pr_hash: "same" })], "instA"); + await ingest(db, [ev({ pr_hash: "same" })], "instB"); + const cnt = await (db as unknown as TestD1Database).prepare("SELECT COUNT(*) AS n FROM orb_signals WHERE pr_hash='same'").first<{ n: number }>(); + expect(cnt?.n).toBe(2); }); - it("stores gate_verdict string when field is present (covers typeof gate_verdict === string branch)", async () => { + it("counts accepted vs skipped in one batch; caps at 500", async () => { const db = makeDb(); - await handleOrbIngest( - JSON.stringify({ instance_id: "inst1", events: [{ repo_hash: "rh2", pr_hash: "ph2", outcome: "merged", gate_verdict: "approve" }] }), - db, - ); - const row = await (db as unknown as TestD1Database).prepare("SELECT gate_verdict FROM orb_signals WHERE pr_hash='ph2'").first<{ gate_verdict: string | null }>(); - expect(row?.gate_verdict).toBe("approve"); + expect(await ingest(db, [ev({ pr_hash: "ok" }), ev({ repo_hash: "" }), ev({ outcome: "x" })])).toEqual({ accepted: 1 }); + const many = Array.from({ length: 501 }, (_, i) => ev({ pr_hash: `m${i}` })); + expect(await ingest(makeDb(), many)).toEqual({ accepted: 500 }); }); - it("stores null time_to_close_ms when field is absent (covers typeof time_to_close_ms !== number branch)", async () => { - const db = makeDb(); - await handleOrbIngest( - JSON.stringify({ instance_id: "inst1", events: [{ repo_hash: "rh3", pr_hash: "ph3", outcome: "closed" }] }), - db, - ); - const row = await (db as unknown as TestD1Database).prepare("SELECT time_to_close_ms FROM orb_signals WHERE pr_hash='ph3'").first<{ time_to_close_ms: number | null }>(); - expect(row?.time_to_close_ms).toBeNull(); + it("swallows a DB error (inner catch)", async () => { + const brokenDb = { prepare: () => ({ bind: () => ({ run: () => Promise.reject(new Error("boom")) }) }) } as unknown as D1Database; + expect(await ingest(brokenDb, [ev()])).toEqual({ accepted: 0 }); }); - it("stores time_to_close_ms when field is a number (covers typeof time_to_close_ms === number branch)", async () => { - const db = makeDb(); - await handleOrbIngest( - JSON.stringify({ instance_id: "inst1", events: [{ repo_hash: "rh4", pr_hash: "ph4", outcome: "merged", time_to_close_ms: 7200000 }] }), - db, - ); - const row = await (db as unknown as TestD1Database).prepare("SELECT time_to_close_ms FROM orb_signals WHERE pr_hash='ph4'").first<{ time_to_close_ms: number | null }>(); - expect(row?.time_to_close_ms).toBe(7200000); + it("does not count a row when the write reports no change (changes === 0)", async () => { + const db = { prepare: () => ({ bind: () => ({ run: () => Promise.resolve({ meta: { changes: 0 } }) }) }) } as unknown as D1Database; + expect(await ingest(db, [ev()])).toEqual({ accepted: 0 }); }); +}); - it("stores null sent_at when created_at is absent (covers typeof created_at !== string branch)", async () => { - const db = makeDb(); - await handleOrbIngest( - JSON.stringify({ instance_id: "inst1", events: [{ repo_hash: "rh5", pr_hash: "ph5", outcome: "merged" }] }), - db, - ); - const row = await (db as unknown as TestD1Database).prepare("SELECT sent_at FROM orb_signals WHERE pr_hash='ph5'").first<{ sent_at: string | null }>(); - expect(row?.sent_at).toBeNull(); - }); +describe("POST /v1/orb/ingest route", () => { + const app = createApp(); - it("stores sent_at when created_at is a string (covers typeof created_at === string branch)", async () => { - const db = makeDb(); - await handleOrbIngest( - JSON.stringify({ instance_id: "inst1", events: [{ repo_hash: "rh6", pr_hash: "ph6", outcome: "merged", created_at: "2024-06-01T12:00:00Z" }] }), - db, - ); - const row = await (db as unknown as TestD1Database).prepare("SELECT sent_at FROM orb_signals WHERE pr_hash='ph6'").first<{ sent_at: string | null }>(); - expect(row?.sent_at).toBe("2024-06-01T12:00:00Z"); + it("returns 200 + accepted count for a valid batch", async () => { + const env = createTestEnv(); + const body = JSON.stringify({ instance_id: "abc0", events: [{ repo_hash: "rhash", pr_hash: "phash", outcome: "merged", reversal_flag: "none" }] }); + const res = await app.request("/v1/orb/ingest", { method: "POST", headers: { "content-type": "application/json" }, body }, env); + expect(res.status).toBe(200); + expect(((await res.json()) as { accepted: number }).accepted).toBe(1); }); - it("deduplicates via INSERT OR IGNORE — second insert is not counted (covers result.meta.changes === 0 branch)", async () => { - const db = makeDb(); - const body = JSON.stringify({ instance_id: "inst1", events: [{ repo_hash: "rh7", pr_hash: "ph7", outcome: "merged" }] }); - expect(await handleOrbIngest(body, db)).toEqual({ accepted: 1 }); - expect(await handleOrbIngest(body, db)).toEqual({ accepted: 0 }); // duplicate ignored + it("returns 400 for invalid JSON", async () => { + const res = await app.request("/v1/orb/ingest", { method: "POST", headers: { "content-type": "application/json" }, body: "{bad" }, createTestEnv()); + expect(res.status).toBe(400); + expect(((await res.json()) as { error: string }).error).toBe("invalid_json"); }); - it("counts both accepted and skipped events in the same batch", async () => { - const db = makeDb(); - const result = await handleOrbIngest( - JSON.stringify({ - instance_id: "inst1", - events: [ - { repo_hash: "rh8", pr_hash: "ph8", outcome: "merged" }, - { repo_hash: "", pr_hash: "ph9", outcome: "merged" }, // invalid — skipped - { repo_hash: "rh10", pr_hash: "ph10", outcome: "invalid" }, // invalid outcome — skipped - ], - }), - db, - ); - expect(result).toEqual({ accepted: 1 }); - }); - - it("caps batch at 500 events (MAX_BATCH) — extra events not inserted", async () => { - const db = makeDb(); - const events = Array.from({ length: 501 }, (_, i) => ({ - repo_hash: `rh${i}`, - pr_hash: `ph${i}`, - outcome: "merged" as const, - })); - const result = await handleOrbIngest(JSON.stringify({ instance_id: "inst-batch", events }), db); - expect(result).toEqual({ accepted: 500 }); - }); - - it("does not throw when the DB throws on insert (covers inner catch branch)", async () => { - const brokenDb = { - prepare: () => ({ bind: () => ({ run: () => Promise.reject(new Error("disk full")) }) }), - } as unknown as D1Database; - const result = await handleOrbIngest( - JSON.stringify({ instance_id: "inst1", events: [{ repo_hash: "rh", pr_hash: "ph", outcome: "merged" }] }), - brokenDb, - ); - expect(result).toEqual({ accepted: 0 }); + it("returns 400 for an empty body", async () => { + const res = await app.request("/v1/orb/ingest", { method: "POST", body: "" }, createTestEnv()); + expect(res.status).toBe(400); }); }); -// ── Route integration tests (covers routes.ts new lines) ────────────────────── - -describe("POST /v1/orb/ingest route", () => { +describe("GET /v1/internal/fleet/analytics route", () => { const app = createApp(); - it("returns 200 with accepted count for a valid batch", async () => { - const env = createTestEnv(); - const body = JSON.stringify({ - instance_id: "abc123def456abc0", - events: [{ repo_hash: "rhash1234567890123456", pr_hash: "phash1234567890123456", outcome: "merged" }], - }); - const res = await app.request("/v1/orb/ingest", { method: "POST", headers: { "content-type": "application/json" }, body }, env); + it("returns the fleet report, honoring ?days (bearer-gated)", async () => { + const res = await app.request("/v1/internal/fleet/analytics?days=30", { headers: { authorization: "Bearer dev-internal-token" } }, createTestEnv()); expect(res.status).toBe(200); - const json = await res.json() as { accepted: number }; - expect(json.accepted).toBe(1); + expect(((await res.json()) as { windowDays: number }).windowDays).toBe(30); }); - it("returns 400 for invalid JSON (covers error-in-result branch)", async () => { - const env = createTestEnv(); - const res = await app.request("/v1/orb/ingest", { method: "POST", headers: { "content-type": "application/json" }, body: "{bad" }, env); - expect(res.status).toBe(400); - const json = await res.json() as { error: string }; - expect(json.error).toBe("invalid_json"); + it("defaults the window when ?days is omitted", async () => { + const res = await app.request("/v1/internal/fleet/analytics", { headers: { authorization: "Bearer dev-internal-token" } }, createTestEnv()); + expect(((await res.json()) as { windowDays: number }).windowDays).toBe(90); }); - it("returns 400 for an empty body (covers !body branch in route)", async () => { - const env = createTestEnv(); - const res = await app.request("/v1/orb/ingest", { method: "POST", body: "" }, env); - expect(res.status).toBe(400); + it("401 without the internal token", async () => { + const res = await app.request("/v1/internal/fleet/analytics", {}, createTestEnv()); + expect(res.status).toBe(401); }); }); diff --git a/test/unit/orb-analytics.test.ts b/test/unit/orb-analytics.test.ts new file mode 100644 index 0000000000..65a78b4b5a --- /dev/null +++ b/test/unit/orb-analytics.test.ts @@ -0,0 +1,107 @@ +import { describe, expect, it } from "vitest"; +import { computeFleetAnalytics } from "../../src/orb/analytics"; +import { createTestEnv, TestD1Database } from "../helpers/d1"; + +let seq = 0; +/** Insert N orb_signals rows for one instance with a fixed verdict/outcome/reversal/cycle. */ +async function signals( + env: Env, + instance: string, + n: number, + o: { verdict?: string | null; outcome?: string; reversal?: string; ms?: number | null } = {}, +): Promise { + for (let i = 0; i < n; i++) { + await env.DB + .prepare( + `INSERT INTO orb_signals (instance_id, repo_hash, pr_hash, gate_verdict, outcome, reversal_flag, time_to_close_ms) + VALUES (?, ?, ?, ?, ?, ?, ?)`, + ) + .bind(instance, `repo${seq}`, `pr${seq++}`, o.verdict ?? "merge", o.outcome ?? "merged", o.reversal ?? "none", o.ms ?? null) + .run(); + } +} + +describe("computeFleetAnalytics()", () => { + it("empty store → zeroed report (and a custom/clamped window)", async () => { + const env = createTestEnv(); + const a = await computeFleetAnalytics(env, { windowDays: 30 }); + expect(a.windowDays).toBe(30); + expect(a.instanceCount).toBe(0); + expect(a.fleet.mergePrecision).toBeNull(); + expect(a.instances).toEqual([]); + // bad window falls back to default 90 + expect((await computeFleetAnalytics(env, { windowDays: -5 })).windowDays).toBe(90); + expect((await computeFleetAnalytics(env)).windowDays).toBe(90); + }); + + it("fail-safe on a DB error → empty report", async () => { + const broken = { DB: { prepare: () => ({ bind: () => ({ all: () => Promise.reject(new Error("boom")) }) }) } } as unknown as Env; + const a = await computeFleetAnalytics(broken); + expect(a.instanceCount).toBe(0); + expect(a.fleet.cycleP50Ms).toBeNull(); + }); + + it("tolerates a DB whose .all() omits results (the ?? [] guards)", async () => { + const env = { DB: { prepare: () => ({ bind: () => ({ all: () => Promise.resolve({}) }) }) } } as unknown as Env; + const a = await computeFleetAnalytics(env); + expect(a.instanceCount).toBe(0); + expect(a.instances).toEqual([]); + }); + + it("computes per-instance precision incl. reversals (reverted merge = false positive)", async () => { + const env = createTestEnv(); + await signals(env, "inst1", 3, { verdict: "merge", outcome: "merged", reversal: "none" }); // confirmed + await signals(env, "inst1", 1, { verdict: "merge", outcome: "merged", reversal: "reverted" }); // false (reverted) + await signals(env, "inst1", 1, { verdict: "merge", outcome: "closed" }); // false + await signals(env, "inst1", 2, { verdict: "close", outcome: "closed" }); // confirmed + await signals(env, "inst1", 1, { verdict: "hold", outcome: "closed" }); // hold — not scored as merge/close + const a = await computeFleetAnalytics(env); + const inst = a.instances.find((i) => i.instanceId === "inst1")!; + expect(inst.decided).toBe(8); + expect(inst.mergePrecision).toBeCloseTo(3 / 5); // 3 confirmed of 5 merge verdicts + expect(inst.fpRate).toBeCloseTo(2 / 5); + expect(inst.closePrecision).toBe(1); // 2/2 + expect(inst.reversalRate).toBeCloseTo(1 / 8); + }); + + it("counts close-verdict false negatives (close → merged)", async () => { + const env = createTestEnv(); + await signals(env, "i", 4, { verdict: "close", outcome: "closed" }); + await signals(env, "i", 1, { verdict: "close", outcome: "merged" }); // closeFalse / false negative + const inst = (await computeFleetAnalytics(env)).instances[0]!; + expect(inst.closePrecision).toBeCloseTo(4 / 5); + expect(inst.fnRate).toBeCloseTo(1 / 5); + }); + + it("null precision when an instance made no merge verdicts", async () => { + const env = createTestEnv(); + await signals(env, "inst1", 5, { verdict: "close", outcome: "closed" }); + const inst = (await computeFleetAnalytics(env)).instances[0]!; + expect(inst.mergePrecision).toBeNull(); + expect(inst.fpRate).toBeNull(); + expect(inst.closePrecision).toBe(1); + }); + + it("fleet uses the median across eligible instances and flags outliers; reports cycle percentiles", async () => { + const env = createTestEnv(); + await signals(env, "good1", 5, { verdict: "merge", outcome: "merged", ms: 1000 }); // precision 1.0 + await signals(env, "good2", 5, { verdict: "merge", outcome: "merged", ms: 2000 }); // precision 1.0 + await signals(env, "bad", 5, { verdict: "merge", outcome: "closed", ms: 9000 }); // precision 0.0 → outlier + await signals(env, "tiny", 2, { verdict: "merge", outcome: "closed" }); // below MIN_DECIDED → excluded from fleet + const a = await computeFleetAnalytics(env); + expect(a.instanceCount).toBe(3); // good1, good2, bad (tiny excluded) + expect(a.fleet.mergePrecision).toBe(1); // median of [1,1,0] + expect(a.outliers.map((o) => o.instanceId)).toContain("bad"); + expect(a.outliers.map((o) => o.instanceId)).not.toContain("good1"); + expect(a.fleet.cycleP50Ms).not.toBeNull(); + expect(a.fleet.cycleP95Ms).not.toBeNull(); + }); + + it("median handles an even number of eligible instances", async () => { + const env = createTestEnv(); + await signals(env, "a", 5, { verdict: "merge", outcome: "merged" }); // 1.0 + await signals(env, "b", 5, { verdict: "merge", outcome: "closed" }); // 0.0 + const a = await computeFleetAnalytics(env); + expect(a.fleet.mergePrecision).toBeCloseTo(0.5); // (1+0)/2 + }); +}); diff --git a/test/unit/selfhost-orb-collector.test.ts b/test/unit/selfhost-orb-collector.test.ts index 8076617e45..776b1fa5c8 100644 --- a/test/unit/selfhost-orb-collector.test.ts +++ b/test/unit/selfhost-orb-collector.test.ts @@ -1,276 +1,191 @@ import { DatabaseSync } from "node:sqlite"; -import { describe, expect, it, beforeEach, afterEach, vi } from "vitest"; +import { describe, expect, it, beforeEach, afterEach } from "vitest"; import { createD1Adapter, nodeSqliteDriver } from "../../src/selfhost/d1-adapter"; -import { exportOrbBatch, orbEnabled, recordOrbEvent } from "../../src/selfhost/orb-collector"; +import { bucketReasonCode, exportOrbBatch, orbEnabled } from "../../src/selfhost/orb-collector"; import { resetMetrics, renderMetrics } from "../../src/selfhost/metrics"; -/** Spin up an in-memory SQLite DB with the orb_events table (and the _selfhost_migrations - * stub so the adapter resolves without running all 56 migrations). */ +/** In-memory DB with the review_audit + orb_export_cursor tables the exporter reads. */ function makeDb(): D1Database { - const raw = new DatabaseSync(":memory:") as never; - const driver = nodeSqliteDriver(raw); + const driver = nodeSqliteDriver(new DatabaseSync(":memory:") as never); driver.exec(` - CREATE TABLE orb_events ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - repo TEXT NOT NULL, - pr_number INTEGER NOT NULL, - head_sha TEXT NOT NULL, - outcome TEXT NOT NULL CHECK (outcome IN ('merged', 'closed')), - gate_verdict TEXT, - time_to_close_ms INTEGER, - created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), - exported_at TEXT, - UNIQUE (repo, pr_number, head_sha) + CREATE TABLE review_audit ( + id TEXT PRIMARY KEY NOT NULL, project TEXT NOT NULL, target_id TEXT NOT NULL, + event_type TEXT NOT NULL DEFAULT 'gate_decision', decision TEXT, + source TEXT NOT NULL DEFAULT 'gittensory-native', head_sha TEXT, summary TEXT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now')) + ); + CREATE TABLE orb_export_cursor ( + instance_hash TEXT PRIMARY KEY, last_exported_at TEXT NOT NULL DEFAULT '2000-01-01T00:00:00Z', + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now')) ); - CREATE INDEX orb_events_repo_pr ON orb_events (repo, pr_number); - CREATE INDEX orb_events_export_pending ON orb_events (exported_at) WHERE exported_at IS NULL; `); return createD1Adapter(driver); } -async function countRows(db: D1Database): Promise { - const r = await db.prepare("SELECT COUNT(*) AS n FROM orb_events").first<{ n: number }>(); - return r?.n ?? 0; -} - -async function allRows(db: D1Database) { - return (await db.prepare("SELECT * FROM orb_events").all()).results; +let seq = 0; +async function audit(db: D1Database, project: string, pr: number, eventType: string, decision: string | null, at: string, summary: string | null = null): Promise { + await db + .prepare(`INSERT INTO review_audit (id, project, target_id, event_type, decision, source, summary, created_at) VALUES (?, ?, ?, ?, ?, 'gittensory-native', ?, ?)`) + .bind(`r${seq++}`, project, `${project}#${pr}`, eventType, decision, summary, at) + .run(); } -describe("orbEnabled()", () => { - afterEach(() => { delete process.env.ORB_ENABLED; }); - - it("returns false when ORB_ENABLED is unset (default off)", () => { - delete process.env.ORB_ENABLED; - expect(orbEnabled()).toBe(false); - }); - - it("returns true for 'true', '1', 'yes' (case-insensitive)", () => { - for (const v of ["true", "True", "TRUE", "1", "yes", "Yes"]) { - process.env.ORB_ENABLED = v; - expect(orbEnabled()).toBe(true); - } - }); - - it("returns false for empty string and 'false'", () => { - for (const v of ["", "false", "0", "no"]) { - process.env.ORB_ENABLED = v; - expect(orbEnabled()).toBe(false); - } +describe("bucketReasonCode()", () => { + it("maps each reason family to a fixed low-cardinality bucket", () => { + expect(bucketReasonCode(null)).toBe("none"); + expect(bucketReasonCode("")).toBe("none"); + expect(bucketReasonCode("missing_linked_issue")).toBe("issue_policy"); + expect(bucketReasonCode("duplicate_pr_risk")).toBe("duplicate_risk"); + expect(bucketReasonCode("ai_slop_advisory")).toBe("slop_advisory"); + expect(bucketReasonCode("ai_consensus_defect")).toBe("ai_quality"); + expect(bucketReasonCode("self_authored_with_maintainer_cut")).toBe("author_policy"); + expect(bucketReasonCode("ci_state failing")).toBe("ci_readiness"); + expect(bucketReasonCode("something_unmapped")).toBe("other"); }); }); -describe("recordOrbEvent()", () => { - beforeEach(() => { resetMetrics(); process.env.ORB_ENABLED = "true"; }); +describe("orbEnabled()", () => { afterEach(() => { delete process.env.ORB_ENABLED; }); - - it("inserts an event with all fields when ORB_ENABLED=true", async () => { - const db = makeDb(); - await recordOrbEvent(db, { repo: "owner/repo", pr_number: 42, head_sha: "abc123", outcome: "merged", gate_verdict: "approve", time_to_close_ms: 3600000 }); - expect(await countRows(db)).toBe(1); - const [row] = (await allRows(db)) as Array>; - expect(row?.repo).toBe("owner/repo"); - expect(row?.pr_number).toBe(42); - expect(row?.outcome).toBe("merged"); - expect(row?.gate_verdict).toBe("approve"); - expect(row?.time_to_close_ms).toBe(3600000); - expect(row?.exported_at).toBeNull(); - }); - - it("inserts with null gate_verdict and time_to_close_ms when omitted", async () => { - const db = makeDb(); - await recordOrbEvent(db, { repo: "o/r", pr_number: 1, head_sha: "sha", outcome: "closed" }); - const [row] = (await allRows(db)) as Array>; - expect(row?.gate_verdict).toBeNull(); - expect(row?.time_to_close_ms).toBeNull(); - }); - - it("is idempotent — INSERT OR IGNORE prevents duplicates for the same (repo, pr, sha)", async () => { - const db = makeDb(); - await recordOrbEvent(db, { repo: "o/r", pr_number: 1, head_sha: "sha1", outcome: "merged" }); - await recordOrbEvent(db, { repo: "o/r", pr_number: 1, head_sha: "sha1", outcome: "merged" }); - expect(await countRows(db)).toBe(1); - }); - - it("increments gittensory_orb_events_recorded_total on each successful insert", async () => { - const db = makeDb(); - await recordOrbEvent(db, { repo: "o/r", pr_number: 1, head_sha: "sha1", outcome: "merged" }); - await recordOrbEvent(db, { repo: "o/r", pr_number: 2, head_sha: "sha2", outcome: "closed" }); - expect(await renderMetrics()).toMatch(/gittensory_orb_events_recorded_total 2/); - }); - - it("does nothing when ORB_ENABLED=false", async () => { - process.env.ORB_ENABLED = "false"; - const db = makeDb(); - await recordOrbEvent(db, { repo: "o/r", pr_number: 99, head_sha: "sha", outcome: "merged" }); - expect(await countRows(db)).toBe(0); - }); - - it("swallows DB errors and never throws (best-effort)", async () => { - const brokenDb = { prepare: () => ({ bind: () => ({ run: () => Promise.reject(new Error("disk full")) }) }) } as unknown as D1Database; - await expect(recordOrbEvent(brokenDb, { repo: "o/r", pr_number: 1, head_sha: "sha", outcome: "merged" })).resolves.not.toThrow(); + it("true only for truthy values", () => { + for (const v of ["true", "1", "Yes"]) { process.env.ORB_ENABLED = v; expect(orbEnabled()).toBe(true); } + for (const v of ["", "false", "no"]) { process.env.ORB_ENABLED = v; expect(orbEnabled()).toBe(false); } + delete process.env.ORB_ENABLED; expect(orbEnabled()).toBe(false); }); }); -describe("exportOrbBatch()", () => { +describe("exportOrbBatch() — reads review_audit, ships anonymized reversal-aware signal", () => { beforeEach(() => { resetMetrics(); process.env.ORB_ENABLED = "true"; process.env.ORB_WEBHOOK_SECRET = "test-secret"; + process.env.ORB_APP_ID = "555"; process.env.ORB_ANONYMIZE = "true"; delete process.env.ORB_AIR_GAP; delete process.env.ORB_COLLECTOR_URL; }); afterEach(() => { - delete process.env.ORB_ENABLED; - process.env.ORB_WEBHOOK_SECRET = undefined as unknown as string; - delete process.env.ORB_ANONYMIZE; - delete process.env.ORB_AIR_GAP; - delete process.env.ORB_COLLECTOR_URL; + for (const k of ["ORB_ENABLED", "ORB_WEBHOOK_SECRET", "ORB_APP_ID", "ORB_ANONYMIZE", "ORB_AIR_GAP", "ORB_COLLECTOR_URL", "GITHUB_APP_ID"]) delete (process.env as NodeJS.Dict)[k]; }); - it("returns 0 when ORB_ENABLED=false (no-op)", async () => { + it("returns 0 when disabled", async () => { process.env.ORB_ENABLED = "false"; - const db = makeDb(); - expect(await exportOrbBatch(db, 200, async () => new Response(null, { status: 200 }))).toBe(0); + expect(await exportOrbBatch(makeDb(), 200, async () => new Response(null, { status: 200 }))).toBe(0); }); - it("returns 0 when ORB_AIR_GAP=true", async () => { + it("returns 0 in air-gap mode", async () => { process.env.ORB_AIR_GAP = "true"; - const db = makeDb(); - await recordOrbEvent(db, { repo: "o/r", pr_number: 1, head_sha: "sha", outcome: "merged" }); - expect(await exportOrbBatch(db, 200, async () => new Response(null, { status: 200 }))).toBe(0); + expect(await exportOrbBatch(makeDb(), 200, async () => new Response(null, { status: 200 }))).toBe(0); }); - it("returns 0 when there are no pending events", async () => { + it("returns 0 when nothing is resolved", async () => { const db = makeDb(); + await audit(db, "o/r", 1, "gate_decision", "merge", "2026-01-01T00:00:00Z"); // decision but no outcome expect(await exportOrbBatch(db, 200, async () => new Response(null, { status: 200 }))).toBe(0); }); - it("defaults to gittensory's hosted collector URL when ORB_COLLECTOR_URL is unset (regression: dead orb.gittensory.app)", async () => { - delete process.env.ORB_COLLECTOR_URL; + it("exports a resolved PR with verdict, outcome, bucket, cycle-time; advances the cursor", async () => { const db = makeDb(); - await recordOrbEvent(db, { repo: "o/r", pr_number: 1, head_sha: "sha", outcome: "merged" }); - let capturedUrl: string | undefined; - await exportOrbBatch(db, 200, async (url) => { capturedUrl = String(url); return new Response(null, { status: 200 }); }); - expect(capturedUrl).toBe("https://gittensory-api.aethereal.dev/v1/orb/ingest"); + await audit(db, "owner/repo", 7, "gate_decision", "merge", "2026-01-01T00:00:00Z", "duplicate_pr_risk"); + await audit(db, "owner/repo", 7, "pr_outcome", "merged", "2026-01-01T01:00:00Z"); + + let captured: { instance_id: string; events: Array> } | undefined; + const n = await exportOrbBatch(db, 200, async (_u, init) => { captured = JSON.parse(init!.body as string); return new Response(null, { status: 200 }); }); + expect(n).toBe(1); + const ev = captured!.events[0]!; + expect(ev.gate_verdict).toBe("merge"); + expect(ev.outcome).toBe("merged"); + expect(ev.reversal_flag).toBe("none"); + expect(ev.gate_reasoncode_bucket).toBe("duplicate_risk"); + expect(ev.time_to_close_ms).toBe(3_600_000); // 1h + expect(ev.repo_hash).not.toBe("owner/repo"); // anonymized + expect((ev.repo_hash as string)).toHaveLength(24); + // cursor advanced → a second run exports nothing new + expect(await exportOrbBatch(db, 200, async () => new Response(null, { status: 200 }))).toBe(0); }); - it("exports pending events and marks them as exported", async () => { + it("flags reversal_reverted and reversal_reopened", async () => { const db = makeDb(); - await recordOrbEvent(db, { repo: "owner/repo", pr_number: 1, head_sha: "sha1", outcome: "merged", gate_verdict: "approve" }); - await recordOrbEvent(db, { repo: "owner/repo", pr_number: 2, head_sha: "sha2", outcome: "closed" }); - - let capturedBody: string | undefined; - const fakeFetch = async (_url: string | URL | Request, init?: RequestInit) => { - capturedBody = init?.body as string; - return new Response(null, { status: 200 }); - }; - - const exported = await exportOrbBatch(db, 200, fakeFetch); - expect(exported).toBe(2); - - // Verify the payload is signed and anonymized - const payload = JSON.parse(capturedBody!) as { instance_id: string; events: Array<{ repo_hash: string; pr_hash: string }> }; - expect(payload.events).toHaveLength(2); - // Anonymized: repo_hash must NOT be the raw repo name - expect(payload.events[0]?.repo_hash).not.toBe("owner/repo"); - expect(payload.events[0]?.repo_hash).toHaveLength(24); // HMAC slice - - // Rows are marked as exported - const rows = (await allRows(db)) as Array>; - expect(rows.every((r) => r.exported_at !== null)).toBe(true); - - // Counter incremented - expect(await renderMetrics()).toMatch(/gittensory_orb_events_exported_total 2/); + await audit(db, "o/r", 1, "gate_decision", "merge", "2026-02-01T00:00:00Z"); + await audit(db, "o/r", 1, "pr_outcome", "merged", "2026-02-01T01:00:00Z"); + await audit(db, "o/r", 1, "reversal_reverted", null, "2026-02-01T05:00:00Z"); + await audit(db, "o/r", 2, "gate_decision", "close", "2026-02-01T00:00:00Z"); + await audit(db, "o/r", 2, "pr_outcome", "merged", "2026-02-01T02:00:00Z"); + await audit(db, "o/r", 2, "reversal_reopened", null, "2026-02-01T03:00:00Z"); + let captured: { events: Array<{ reversal_flag: string }> } | undefined; + await exportOrbBatch(db, 200, async (_u, init) => { captured = JSON.parse(init!.body as string); return new Response(null, { status: 200 }); }); + const flags = captured!.events.map((e) => e.reversal_flag).sort(); + expect(flags).toEqual(["reopened", "reverted"]); }); - it("ORB_ANONYMIZE=false sends raw repo name in payload", async () => { + it("sends raw repo when ORB_ANONYMIZE=false", async () => { process.env.ORB_ANONYMIZE = "false"; const db = makeDb(); - await recordOrbEvent(db, { repo: "owner/repo", pr_number: 1, head_sha: "sha1", outcome: "merged" }); - - let capturedBody: string | undefined; - await exportOrbBatch(db, 200, async (_u, init) => { capturedBody = init?.body as string; return new Response(null, { status: 200 }); }); - const payload = JSON.parse(capturedBody!) as { events: Array<{ repo_hash: string }> }; - expect(payload.events[0]?.repo_hash).toBe("owner/repo"); + await audit(db, "owner/repo", 1, "gate_decision", "close", "2026-01-01T00:00:00Z"); + await audit(db, "owner/repo", 1, "pr_outcome", "closed", "2026-01-01T00:30:00Z"); + let captured: { events: Array<{ repo_hash: string }> } | undefined; + await exportOrbBatch(db, 200, async (_u, init) => { captured = JSON.parse(init!.body as string); return new Response(null, { status: 200 }); }); + expect(captured!.events[0]!.repo_hash).toBe("owner/repo"); }); - it("does not re-export already-exported events", async () => { + it("null cycle-time when the resolution precedes the decision (negative delta)", async () => { const db = makeDb(); - await recordOrbEvent(db, { repo: "o/r", pr_number: 1, head_sha: "sha1", outcome: "merged" }); - const fakeFetch = vi.fn(async () => new Response(null, { status: 200 })); - await exportOrbBatch(db, 200, fakeFetch); // exports 1 - const second = await exportOrbBatch(db, 200, fakeFetch); // nothing left - expect(second).toBe(0); - expect(fakeFetch).toHaveBeenCalledTimes(1); + await audit(db, "o/r", 1, "gate_decision", "merge", "2026-01-02T00:00:00Z"); + await audit(db, "o/r", 1, "pr_outcome", "merged", "2026-01-01T00:00:00Z"); // outcome BEFORE decision → negative → null + let captured: { events: Array<{ time_to_close_ms: number | null }> } | undefined; + await exportOrbBatch(db, 200, async (_u, init) => { captured = JSON.parse(init!.body as string); return new Response(null, { status: 200 }); }); + expect(captured!.events[0]!.time_to_close_ms).toBeNull(); }); - it("returns 0 and increments error counter on HTTP error from collector", async () => { + it("returns 0 + increments error counter on a non-OK collector response", async () => { const db = makeDb(); - await recordOrbEvent(db, { repo: "o/r", pr_number: 1, head_sha: "sha", outcome: "merged" }); - const result = await exportOrbBatch(db, 200, async () => new Response(null, { status: 503 })); - expect(result).toBe(0); + await audit(db, "o/r", 1, "gate_decision", "merge", "2026-01-01T00:00:00Z"); + await audit(db, "o/r", 1, "pr_outcome", "merged", "2026-01-01T01:00:00Z"); + expect(await exportOrbBatch(db, 200, async () => new Response(null, { status: 503 }))).toBe(0); expect(await renderMetrics()).toContain("gittensory_orb_export_errors_total"); - // Event still pending (not marked as exported) - const rows = (await allRows(db)) as Array>; - expect(rows[0]?.exported_at).toBeNull(); }); - it("returns 0 and increments error counter when collector is unreachable (network error)", async () => { + it("returns 0 + increments error counter when the collector is unreachable", async () => { const db = makeDb(); - await recordOrbEvent(db, { repo: "o/r", pr_number: 1, head_sha: "sha", outcome: "merged" }); - const result = await exportOrbBatch(db, 200, async () => { throw new Error("ECONNREFUSED"); }); - expect(result).toBe(0); + await audit(db, "o/r", 1, "gate_decision", "merge", "2026-01-01T00:00:00Z"); + await audit(db, "o/r", 1, "pr_outcome", "merged", "2026-01-01T01:00:00Z"); + expect(await exportOrbBatch(db, 200, async () => { throw new Error("ECONNREFUSED"); })).toBe(0); expect(await renderMetrics()).toContain("gittensory_orb_export_errors_total"); }); - it("includes x-orb-signature header with sha256 HMAC", async () => { - const db = makeDb(); - await recordOrbEvent(db, { repo: "o/r", pr_number: 1, head_sha: "sha", outcome: "merged" }); - - let sigHeader: string | undefined; - await exportOrbBatch(db, 200, async (_u, init) => { - sigHeader = (init?.headers as Record)?.["x-orb-signature"]; - return new Response(null, { status: 200 }); - }); - expect(sigHeader).toMatch(/^sha256=[a-f0-9]{64}$/); - }); - - it("uses empty-string HMAC key when ORB_WEBHOOK_SECRET is unset (covers ?? '' branch)", async () => { - delete (process.env as NodeJS.Dict)["ORB_WEBHOOK_SECRET"]; + it("signs the batch and respects batchSize", async () => { const db = makeDb(); - await recordOrbEvent(db, { repo: "o/r", pr_number: 1, head_sha: "sha", outcome: "merged" }); - let sigHeader: string | undefined; - const exported = await exportOrbBatch(db, 200, async (_u, init) => { - sigHeader = (init?.headers as Record)?.["x-orb-signature"]; - return new Response(null, { status: 200 }); - }); - expect(exported).toBe(1); - // Signature should still be formed (with empty-string key) - expect(sigHeader).toMatch(/^sha256=[a-f0-9]{64}$/); - }); - - it("defaults ORB_ANONYMIZE to true when unset (covers ?? 'true' branch)", async () => { - delete process.env.ORB_ANONYMIZE; - const db = makeDb(); - await recordOrbEvent(db, { repo: "owner/repo", pr_number: 1, head_sha: "sha1", outcome: "merged" }); - let capturedBody: string | undefined; - await exportOrbBatch(db, 200, async (_u, init) => { capturedBody = init?.body as string; return new Response(null, { status: 200 }); }); - const payload = JSON.parse(capturedBody!) as { events: Array<{ repo_hash: string }> }; - // Default is anonymize=true, so repo name must be hashed - expect(payload.events[0]?.repo_hash).not.toBe("owner/repo"); - expect(payload.events[0]?.repo_hash).toHaveLength(24); - }); - - it("respects batchSize — exports only the first N pending events", async () => { - const db = makeDb(); - for (let i = 1; i <= 5; i++) - await recordOrbEvent(db, { repo: "o/r", pr_number: i, head_sha: `sha${i}`, outcome: "merged" }); - const exported = await exportOrbBatch(db, 3, async () => new Response(null, { status: 200 })); - expect(exported).toBe(3); - // 2 events still pending - const rows = (await allRows(db)) as Array>; - expect(rows.filter((r) => r.exported_at === null)).toHaveLength(2); + for (let i = 1; i <= 5; i++) { + await audit(db, "o/r", i, "gate_decision", "merge", `2026-03-0${i}T00:00:00Z`); + await audit(db, "o/r", i, "pr_outcome", "merged", `2026-03-0${i}T01:00:00Z`); + } + let sig: string | undefined; + const n = await exportOrbBatch(db, 3, async (_u, init) => { sig = (init!.headers as Record)["x-orb-signature"]; return new Response(null, { status: 200 }); }); + expect(n).toBe(3); // batch cap + expect(sig).toMatch(/^sha256=[a-f0-9]{64}$/); + }); + + it("falls back to GITHUB_APP_ID for the instance id and applies secret/anonymize defaults when ORB_* are unset", async () => { + delete process.env.ORB_APP_ID; // → falls through to GITHUB_APP_ID + delete process.env.ORB_WEBHOOK_SECRET; // → secret defaults to "" + delete process.env.ORB_ANONYMIZE; // → defaults to "true" + (process.env as NodeJS.Dict).GITHUB_APP_ID = "999"; + const db = makeDb(); + await audit(db, "owner/repo", 1, "gate_decision", "merge", "2026-01-01T00:00:00Z"); + await audit(db, "owner/repo", 1, "pr_outcome", "merged", "2026-01-01T01:00:00Z"); + let captured: { events: Array<{ repo_hash: string }> } | undefined; + const n = await exportOrbBatch(db, 200, async (_u, init) => { captured = JSON.parse(init!.body as string); return new Response(null, { status: 200 }); }); + expect(n).toBe(1); + expect(captured!.events[0]!.repo_hash).not.toBe("owner/repo"); // anonymize default = true + }); + + it("uses an 'unknown' instance id when neither ORB_APP_ID nor GITHUB_APP_ID is set", async () => { + delete process.env.ORB_APP_ID; + delete (process.env as NodeJS.Dict).GITHUB_APP_ID; + const db = makeDb(); + await audit(db, "o/r", 1, "gate_decision", "merge", "2026-01-01T00:00:00Z"); + await audit(db, "o/r", 1, "pr_outcome", "merged", "2026-01-01T01:00:00Z"); + let header: string | undefined; + await exportOrbBatch(db, 200, async (_u, init) => { header = (init!.headers as Record)["x-orb-instance"]; return new Response(null, { status: 200 }); }); + expect(header).toMatch(/^[a-f0-9]{16}$/); }); }); diff --git a/test/unit/selfhost-orb-setup.test.ts b/test/unit/selfhost-orb-setup.test.ts deleted file mode 100644 index e55df452d5..0000000000 --- a/test/unit/selfhost-orb-setup.test.ts +++ /dev/null @@ -1,110 +0,0 @@ -import { describe, expect, it, vi, afterEach } from "vitest"; -import { - buildOrbManifest, - exchangeOrbManifestCode, - orbCredentialsToEnv, - renderOrbSetupPage, -} from "../../src/selfhost/orb-setup"; - -describe("buildOrbManifest()", () => { - it("sets the webhook URL to /orb/webhook under the origin", () => { - const m = buildOrbManifest("https://gittensory.example.com", "state123"); - expect((m.hook_attributes as { url: string }).url).toBe("https://gittensory.example.com/orb/webhook"); - }); - - it("sets the redirect_url to /orb/setup/callback with encoded state", () => { - const m = buildOrbManifest("https://example.com", "my state"); - expect(m.redirect_url).toBe("https://example.com/orb/setup/callback?state=my%20state"); - }); - - it("strips a trailing slash from the origin", () => { - const m = buildOrbManifest("https://example.com/", "s"); - expect((m.hook_attributes as { url: string }).url).toBe("https://example.com/orb/webhook"); - }); - - it("requests only read permissions (pull_requests + metadata)", () => { - const m = buildOrbManifest("https://example.com", "s"); - const perms = m.default_permissions as Record; - expect(perms.pull_requests).toBe("read"); - expect(perms.metadata).toBe("read"); - // Must not request write permissions - expect(Object.values(perms).every((v) => v === "read")).toBe(true); - }); - - it("subscribes to pull_request, installation, and installation_repositories events", () => { - const m = buildOrbManifest("https://example.com", "s"); - const events = m.default_events as string[]; - expect(events).toContain("pull_request"); - expect(events).toContain("installation"); - expect(events).toContain("installation_repositories"); - }); - - it("sets public: false", () => { - expect(buildOrbManifest("https://example.com", "s").public).toBe(false); - }); -}); - -describe("renderOrbSetupPage()", () => { - it("returns valid HTML containing the manifest JSON", () => { - const html = renderOrbSetupPage("https://example.com", "xyz"); - expect(html).toContain(""); - expect(html).toContain("orb/webhook"); - expect(html).toContain("https://github.com/settings/apps/new"); - expect(html).toContain("Gittensory Orb"); - }); - - it("embeds the manifest in the form input value", () => { - const html = renderOrbSetupPage("https://example.com", "state-abc"); - expect(html).toContain('name="manifest"'); - expect(html).toContain("orb/webhook"); - }); - - it("escapes single quotes in the manifest to prevent attribute injection", () => { - // JSON.stringify naturally won't produce ' but the replace guard must be present - const html = renderOrbSetupPage("https://example.com", "s"); - expect(html).not.toContain("'gittensory"); - }); -}); - -describe("exchangeOrbManifestCode()", () => { - afterEach(() => { vi.restoreAllMocks(); }); - - it("POSTs to the GitHub conversions endpoint and returns parsed credentials", async () => { - const fakeCreds = { id: 999, slug: "gittensory-orb-test", webhook_secret: "sec", pem: "pem-data" }; - const fakeFetch = vi.fn(async () => new Response(JSON.stringify(fakeCreds), { status: 201 })); - const result = await exchangeOrbManifestCode("test-code-xyz", fakeFetch); - expect(result).toEqual(fakeCreds); - const [url, init] = fakeFetch.mock.calls[0] as unknown as [string, RequestInit]; - expect(url).toContain("test-code-xyz"); - expect(url).toContain("app-manifests"); - expect(init.method).toBe("POST"); - }); - - it("throws on non-OK HTTP response with status in the message", async () => { - const fakeFetch = vi.fn(async () => new Response("", { status: 422 })); - await expect(exchangeOrbManifestCode("bad-code", fakeFetch)).rejects.toThrow("422"); - }); - - it("URL-encodes the code to prevent injection", async () => { - const fakeFetch = vi.fn(async () => new Response(JSON.stringify({ id: 1, slug: "s", webhook_secret: "w", pem: "p" }), { status: 201 })); - await exchangeOrbManifestCode("code/with/slash", fakeFetch); - const [url] = fakeFetch.mock.calls[0] as unknown as [string]; - expect(url).toContain("code%2Fwith%2Fslash"); - }); -}); - -describe("orbCredentialsToEnv()", () => { - it("produces ORB_APP_ID, ORB_APP_SLUG, ORB_WEBHOOK_SECRET, ORB_PRIVATE_KEY lines", () => { - const env = orbCredentialsToEnv({ id: 42, slug: "orb-slug", webhook_secret: "wh-sec", pem: "BEGIN RSA" }); - expect(env).toContain("ORB_APP_ID=42"); - expect(env).toContain("ORB_APP_SLUG=orb-slug"); - expect(env).toContain("ORB_WEBHOOK_SECRET=wh-sec"); - expect(env).toContain("ORB_PRIVATE_KEY="); - expect(env.endsWith("\n")).toBe(true); - }); - - it("JSON-stringifies the PEM so newlines survive loading as a single env var", () => { - const env = orbCredentialsToEnv({ id: 1, slug: "s", webhook_secret: "w", pem: "line1\nline2" }); - expect(env).toContain('"line1\\nline2"'); - }); -}); diff --git a/test/unit/selfhost-orb-webhook.test.ts b/test/unit/selfhost-orb-webhook.test.ts deleted file mode 100644 index fd2a71a05d..0000000000 --- a/test/unit/selfhost-orb-webhook.test.ts +++ /dev/null @@ -1,322 +0,0 @@ -import { createHmac } from "node:crypto"; -import { DatabaseSync } from "node:sqlite"; -import { describe, expect, it, beforeEach, afterEach } from "vitest"; -import { createD1Adapter, nodeSqliteDriver } from "../../src/selfhost/d1-adapter"; -import { - handleOrbWebhook, - lookupGateVerdict, - verifyOrbSignature, -} from "../../src/selfhost/orb-webhook"; -import { resetMetrics, renderMetrics } from "../../src/selfhost/metrics"; - -const SECRET = "test-webhook-secret"; - -function sign(payload: string, secret = SECRET): string { - return "sha256=" + createHmac("sha256", secret).update(payload).digest("hex"); -} - -function makeDb(): D1Database { - const driver = nodeSqliteDriver(new DatabaseSync(":memory:") as never); - driver.exec(` - CREATE TABLE orb_events ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - repo TEXT NOT NULL, pr_number INTEGER NOT NULL, head_sha TEXT NOT NULL, - outcome TEXT NOT NULL CHECK (outcome IN ('merged', 'closed')), - gate_verdict TEXT, time_to_close_ms INTEGER, - created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), - exported_at TEXT, - UNIQUE (repo, pr_number, head_sha) - ); - CREATE TABLE orb_installations ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - installation_id INTEGER NOT NULL, repo TEXT NOT NULL, - installed_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), - removed_at TEXT, - UNIQUE (installation_id, repo) - ); - CREATE TABLE review_targets ( - id TEXT PRIMARY KEY, - project TEXT NOT NULL, kind TEXT NOT NULL, repo TEXT NOT NULL, - number INTEGER NOT NULL, verdict TEXT, - updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, - UNIQUE (project, kind, repo, number) - ); - `); - return createD1Adapter(driver); -} - -// ── verifyOrbSignature ──────────────────────────────────────────────────────── - -describe("verifyOrbSignature()", () => { - it("returns true for a correctly signed payload", () => { - const payload = '{"action":"closed"}'; - expect(verifyOrbSignature(payload, sign(payload), SECRET)).toBe(true); - }); - - it("returns false when the signature doesn't match", () => { - expect(verifyOrbSignature("payload", sign("other"), SECRET)).toBe(false); - }); - - it("returns false when the secret is wrong", () => { - const payload = "body"; - expect(verifyOrbSignature(payload, sign(payload, "wrong-secret"), SECRET)).toBe(false); - }); - - it("returns false when the sig header is missing the sha256= prefix", () => { - const payload = "body"; - const raw = createHmac("sha256", SECRET).update(payload).digest("hex"); - expect(verifyOrbSignature(payload, raw, SECRET)).toBe(false); - }); - - it("returns false for empty sig", () => { - expect(verifyOrbSignature("body", "", SECRET)).toBe(false); - }); - - it("returns false when sig hex is malformed/wrong length (timingSafeEqual throws — covers catch branch)", () => { - // "sha256=" prefix passes, but odd-length hex → Buffer.from(..., 'hex') produces - // a different byte length than the 32-byte expected HMAC → timingSafeEqual throws ERR_CRYPTO_TIMING_SAFE_EQUAL_LENGTH - expect(verifyOrbSignature("body", "sha256=abc", SECRET)).toBe(false); - }); -}); - -// ── lookupGateVerdict ───────────────────────────────────────────────────────── - -describe("lookupGateVerdict()", () => { - beforeEach(() => { process.env.ORB_ENABLED = "true"; }); - afterEach(() => { delete process.env.ORB_ENABLED; }); - - it("returns the verdict from review_targets for a matching repo+PR", async () => { - const db = makeDb(); - await db.prepare(`INSERT INTO review_targets (id, project, kind, repo, number, verdict, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)`).bind("t1", "proj", "PR", "owner/repo", 42, "merge", "2024-01-01T00:00:00Z").run(); - expect(await lookupGateVerdict(db, "owner/repo", 42)).toBe("merge"); - }); - - it("returns null when no review_target row exists", async () => { - const db = makeDb(); - expect(await lookupGateVerdict(db, "owner/repo", 99)).toBeNull(); - }); - - it("returns null on DB error (best-effort)", async () => { - const brokenDb = { prepare: () => ({ bind: () => ({ first: () => Promise.reject(new Error("disk full")) }) }) } as unknown as D1Database; - expect(await lookupGateVerdict(brokenDb, "o/r", 1)).toBeNull(); - }); - - it("picks the most recent verdict when multiple rows exist for the same repo+PR", async () => { - const db = makeDb(); - await db.prepare(`INSERT INTO review_targets (id, project, kind, repo, number, verdict, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)`).bind("t1", "p", "PR", "o/r", 1, "close", "2024-01-01T00:00:00Z").run(); - await db.prepare(`INSERT INTO review_targets (id, project, kind, repo, number, verdict, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)`).bind("t2", "p2", "PR", "o/r", 1, "merge", "2024-01-02T00:00:00Z").run(); - expect(await lookupGateVerdict(db, "o/r", 1)).toBe("merge"); - }); -}); - -// ── handleOrbWebhook ────────────────────────────────────────────────────────── - -describe("handleOrbWebhook() — pull_request events", () => { - beforeEach(() => { resetMetrics(); process.env.ORB_ENABLED = "true"; }); - afterEach(() => { delete process.env.ORB_ENABLED; }); - - const prPayload = (action: string, merged: boolean, prNumber = 7) => - JSON.stringify({ - action, - pull_request: { - number: prNumber, - head: { sha: "abc123" }, - merged, - created_at: "2024-01-01T00:00:00Z", - closed_at: "2024-01-01T01:00:00Z", - }, - repository: { full_name: "owner/repo" }, - }); - - it("returns 204 for a merged PR and records it in orb_events", async () => { - const db = makeDb(); - const result = await handleOrbWebhook("pull_request", prPayload("closed", true), db); - expect(result.status).toBe(204); - const row = await db.prepare("SELECT outcome FROM orb_events WHERE repo='owner/repo' AND pr_number=7").first<{ outcome: string }>(); - expect(row?.outcome).toBe("merged"); - }); - - it("records outcome='closed' for a non-merged closed PR", async () => { - const db = makeDb(); - await handleOrbWebhook("pull_request", prPayload("closed", false), db); - const row = await db.prepare("SELECT outcome FROM orb_events WHERE repo='owner/repo' AND pr_number=7").first<{ outcome: string }>(); - expect(row?.outcome).toBe("closed"); - }); - - it("calculates time_to_close_ms from created_at and closed_at", async () => { - const db = makeDb(); - await handleOrbWebhook("pull_request", prPayload("closed", true), db); - const row = await db.prepare("SELECT time_to_close_ms FROM orb_events WHERE pr_number=7").first<{ time_to_close_ms: number }>(); - expect(row?.time_to_close_ms).toBe(3600000); // 1 hour - }); - - it("stores the gate verdict from review_targets when present", async () => { - const db = makeDb(); - await db.prepare(`INSERT INTO review_targets (id, project, kind, repo, number, verdict, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)`).bind("t1", "p", "PR", "owner/repo", 7, "merge", "2024-01-01T00:00:00Z").run(); - await handleOrbWebhook("pull_request", prPayload("closed", true), db); - const row = await db.prepare("SELECT gate_verdict FROM orb_events WHERE pr_number=7").first<{ gate_verdict: string }>(); - expect(row?.gate_verdict).toBe("merge"); - }); - - it("stores null gate_verdict when no review_target exists for the PR", async () => { - const db = makeDb(); - await handleOrbWebhook("pull_request", prPayload("closed", false), db); - const row = await db.prepare("SELECT gate_verdict FROM orb_events WHERE pr_number=7").first<{ gate_verdict: string | null }>(); - expect(row?.gate_verdict).toBeNull(); - }); - - it("records null time_to_close_ms when closed_at or created_at is absent (covers ternary null branches)", async () => { - const db = makeDb(); - const payload = JSON.stringify({ - action: "closed", - pull_request: { number: 5, head: { sha: "sha5" }, merged: true, created_at: null, closed_at: null }, - repository: { full_name: "owner/repo" }, - }); - const result = await handleOrbWebhook("pull_request", payload, db); - expect(result.status).toBe(204); - const row = await db.prepare("SELECT time_to_close_ms FROM orb_events WHERE pr_number=5").first<{ time_to_close_ms: number | null }>(); - expect(row?.time_to_close_ms).toBeNull(); - }); - - it("records null time_to_close_ms when only closed_at is absent", async () => { - const db = makeDb(); - const payload = JSON.stringify({ - action: "closed", - pull_request: { number: 6, head: { sha: "sha6" }, merged: false, created_at: "2024-01-01T00:00:00Z", closed_at: null }, - repository: { full_name: "owner/repo" }, - }); - await handleOrbWebhook("pull_request", payload, db); - const row = await db.prepare("SELECT time_to_close_ms FROM orb_events WHERE pr_number=6").first<{ time_to_close_ms: number | null }>(); - expect(row?.time_to_close_ms).toBeNull(); - }); - - it("returns 204 and does NOT record for non-closed pull_request actions", async () => { - const db = makeDb(); - const result = await handleOrbWebhook("pull_request", prPayload("opened", false), db); - expect(result.status).toBe(204); - const { results } = await db.prepare("SELECT * FROM orb_events").all(); - expect(results).toHaveLength(0); - }); - - it("increments gittensory_orb_webhook_total on every event", async () => { - const db = makeDb(); - await handleOrbWebhook("pull_request", prPayload("closed", true), db); - await handleOrbWebhook("pull_request", prPayload("closed", true, 8), db); - expect(await renderMetrics()).toMatch(/gittensory_orb_webhook_total 2/); - }); -}); - -describe("handleOrbWebhook() — installation events", () => { - beforeEach(() => { resetMetrics(); process.env.ORB_ENABLED = "true"; }); - afterEach(() => { delete process.env.ORB_ENABLED; }); - - it("creates installation records for each repo on 'created' event", async () => { - const db = makeDb(); - const payload = JSON.stringify({ - action: "created", - installation: { id: 100 }, - repositories: [{ full_name: "owner/repo-a" }, { full_name: "owner/repo-b" }], - }); - await handleOrbWebhook("installation", payload, db); - const { results } = await db.prepare("SELECT repo FROM orb_installations").all<{ repo: string }>(); - expect(results.map((r) => r.repo).sort()).toEqual(["owner/repo-a", "owner/repo-b"]); - }); - - it("marks repos as removed on 'deleted' event by setting removed_at", async () => { - const db = makeDb(); - const created = JSON.stringify({ action: "created", installation: { id: 100 }, repositories: [{ full_name: "owner/repo" }] }); - await handleOrbWebhook("installation", created, db); - const deleted = JSON.stringify({ action: "deleted", installation: { id: 100 }, repositories: [{ full_name: "owner/repo" }] }); - await handleOrbWebhook("installation", deleted, db); - const row = await db.prepare("SELECT removed_at FROM orb_installations WHERE repo='owner/repo'").first<{ removed_at: string | null }>(); - expect(row?.removed_at).not.toBeNull(); - }); - - it("handles installation_repositories added event", async () => { - const db = makeDb(); - const payload = JSON.stringify({ - action: "added", - installation: { id: 200 }, - repositories_added: [{ full_name: "owner/new-repo" }], - repositories_removed: [], - }); - await handleOrbWebhook("installation_repositories", payload, db); - const row = await db.prepare("SELECT repo FROM orb_installations WHERE repo='owner/new-repo'").first<{ repo: string }>(); - expect(row?.repo).toBe("owner/new-repo"); - }); - - it("handles installation_repositories removed event", async () => { - const db = makeDb(); - await db.prepare("INSERT INTO orb_installations (installation_id, repo) VALUES (?, ?)").bind(200, "owner/gone-repo").run(); - const payload = JSON.stringify({ - action: "removed", - installation: { id: 200 }, - repositories_added: [], - repositories_removed: [{ full_name: "owner/gone-repo" }], - }); - await handleOrbWebhook("installation_repositories", payload, db); - const row = await db.prepare("SELECT removed_at FROM orb_installations WHERE repo='owner/gone-repo'").first<{ removed_at: string | null }>(); - expect(row?.removed_at).not.toBeNull(); - }); - - it("handles installation_repositories removed event with missing repositories_removed field (covers ?? [] branch)", async () => { - const db = makeDb(); - await db.prepare("INSERT INTO orb_installations (installation_id, repo) VALUES (?, ?)").bind(300, "owner/repo-x").run(); - // No repositories_removed key — falls back to [] - const payload = JSON.stringify({ - action: "removed", - installation: { id: 300 }, - repositories_added: [], - // repositories_removed intentionally absent - }); - const result = await handleOrbWebhook("installation_repositories", payload, db); - expect(result.status).toBe(204); - // No rows should be marked removed (empty list was used) - const row = await db.prepare("SELECT removed_at FROM orb_installations WHERE repo='owner/repo-x'").first<{ removed_at: string | null }>(); - expect(row?.removed_at).toBeNull(); - }); - - it("does not increment installs counter when repositories list is empty (covers if(repos.length) false branch)", async () => { - const db = makeDb(); - const payload = JSON.stringify({ action: "created", installation: { id: 1 }, repositories: [] }); - await handleOrbWebhook("installation", payload, db); - // Counter must not have been incremented - expect(await renderMetrics()).not.toMatch(/gittensory_orb_installs_total [^0]/); - }); - - it("handles deleted event when repositories field is absent (covers repositories ?? [] branch)", async () => { - const db = makeDb(); - await db.prepare("INSERT INTO orb_installations (installation_id, repo) VALUES (?, ?)").bind(400, "owner/to-delete").run(); - // 'deleted' with no repositories key → falls back to [] - const payload = JSON.stringify({ action: "deleted", installation: { id: 400 } }); - const result = await handleOrbWebhook("installation", payload, db); - expect(result.status).toBe(204); - // Nothing removed since repos list was empty - const row = await db.prepare("SELECT removed_at FROM orb_installations WHERE repo='owner/to-delete'").first<{ removed_at: string | null }>(); - expect(row?.removed_at).toBeNull(); - }); - - it("increments gittensory_orb_installs_total for each repo installed", async () => { - const db = makeDb(); - const payload = JSON.stringify({ action: "created", installation: { id: 1 }, repositories: [{ full_name: "o/a" }, { full_name: "o/b" }] }); - await handleOrbWebhook("installation", payload, db); - expect(await renderMetrics()).toMatch(/gittensory_orb_installs_total 2/); - }); - - it("is idempotent — duplicate install events do not create duplicate rows", async () => { - const db = makeDb(); - const payload = JSON.stringify({ action: "created", installation: { id: 1 }, repositories: [{ full_name: "o/r" }] }); - await handleOrbWebhook("installation", payload, db); - await handleOrbWebhook("installation", payload, db); - const { results } = await db.prepare("SELECT * FROM orb_installations").all(); - expect(results).toHaveLength(1); - }); -}); - -describe("handleOrbWebhook() — unknown events", () => { - it("returns 204 for unhandled event types (ping, etc.)", async () => { - const db = makeDb(); - const result = await handleOrbWebhook("ping", '{"zen":"Keep it logically awesome."}', db); - expect(result.status).toBe(204); - }); -});