Integrate rating-v2/C4 line + upstream-header routing into main

cmd/interceptor/main.go

@@ -3,7 +3,6 @@ package main
 import (
 	"context"
 	"flag"
-	"net/url"
 	"os"
 	"time"
 
@@ -15,7 +14,6 @@ import (
 	"github.com/saturncloud/phoebe/internal/logging"
 	"github.com/saturncloud/phoebe/internal/metering"
 	"github.com/saturncloud/phoebe/internal/proxy"
-	"github.com/saturncloud/phoebe/internal/registry"
 )
 
 func main() {
@@ -32,15 +30,10 @@ func main() {
 		log.SetLevel(logging.DEBUG)
 	}
 
-	resolver, err := buildResolver(settings, log)
-	if err != nil {
-		log.Error.Fatalf("failed to build resolver: %v", err)
-	}
-
 	emitter, closeEmitter := buildEmitter(settings, log)
 	ioPolicy, ioSink, ioMaxBody, closeIOLog := buildIOLog(settings, log)
 
-	srv := proxy.NewWithIOLog(settings, log, resolver, emitter, ioPolicy, ioSink, ioMaxBody)
+	srv := proxy.NewWithIOLog(settings, log, emitter, ioPolicy, ioSink, ioMaxBody)
 	srvErr := srv.Run()
 
 	// Cleanup must run UNCONDITIONALLY before exit. log.Fatalf here would
@@ -56,72 +49,6 @@ func main() {
 	}
 }
 
-// buildResolver constructs the model→upstream resolver per the configured
-// strategy. "static" uses the single DefaultUpstream (M0/M1 behaviour);
-// "convention", "cached", and "chain" enable dynamic dispatch (M4).
-//
-// NOTE: the "cached" and "chain" strategies are meant to resolve via a control
-// plane (Atlas) lookup. That control-plane API is the still-unverified seam
-// (does auth-server already resolve model resources via X-Saturn-Resource-Id?),
-// so until it's wired the LookupFunc degrades to the naming convention — a
-// reasonable guess that needs no redeploy. Replace conventionLookup with the
-// real Atlas call once the resource-resolution path is confirmed.
-func buildResolver(s *config.Settings, log *logging.Logger) (registry.Resolver, error) {
-	rs := s.Registry
-	switch rs.Strategy {
-	case "", "static":
-		log.Info.Printf("resolver: static (default upstream %s)", s.DefaultUpstream)
-		return registry.NewStatic(s.Default), nil
-
-	case "convention":
-		log.Info.Printf("resolver: convention (%s)", rs.ConventionTemplate)
-		return registry.NewConventionResolver(registry.ConventionConfig{
-			Template: rs.ConventionTemplate,
-		})
-
-	case "cached":
-		conv, err := registry.NewConventionResolver(registry.ConventionConfig{Template: rs.ConventionTemplate})
-		if err != nil {
-			return nil, err
-		}
-		log.Info.Printf("resolver: cached (lookup degrades to convention until control-plane wired)")
-		return registry.NewCachedResolver(conventionLookup(conv), registry.CacheConfig{
-			Size:        rs.CacheSize,
-			PositiveTTL: rs.PositiveTTL,
-			NegativeTTL: rs.NegativeTTL,
-		})
-
-	case "chain":
-		conv, err := registry.NewConventionResolver(registry.ConventionConfig{Template: rs.ConventionTemplate})
-		if err != nil {
-			return nil, err
-		}
-		cached, err := registry.NewCachedResolver(conventionLookup(conv), registry.CacheConfig{
-			Size:        rs.CacheSize,
-			PositiveTTL: rs.PositiveTTL,
-			NegativeTTL: rs.NegativeTTL,
-		})
-		if err != nil {
-			return nil, err
-		}
-		// Cached control-plane lookup first, naming-convention fallback if it
-		// errors — graceful degradation when the control plane is unreachable.
-		log.Info.Printf("resolver: chain (cached → convention)")
-		return registry.ChainResolver{cached, conv}, nil
-
-	default:
-		return registry.NewStatic(s.Default), nil
-	}
-}
-
-// conventionLookup adapts a ConventionResolver to a registry.LookupFunc so it
-// can stand in for the (not-yet-wired) control-plane lookup.
-func conventionLookup(conv *registry.ConventionResolver) registry.LookupFunc {
-	return func(_ context.Context, resourceID string) (*url.URL, error) {
-		return conv.Resolve(resourceID)
-	}
-}
-
 // buildEmitter constructs the durable metering emitter. When ValkeyAddr is set
 // it dials Valkey for the hot path; the WAL fallback and log floor are always
 // active. Returns a close function for graceful shutdown.

cmd/token-push/main.go

@@ -1,9 +1,10 @@
 // Command token-push is phoebe's install-side hourly billing PUSH job. It reads the
-// rater's rated_usage rollups for a window, resolves each rollup's deployment to the
-// org that owns it (resource_id → resource_name.org_id, a LOCAL join in the shared
-// Atlas Postgres), and POSTs an hourly SNAPSHOT to the central manager's
-// POST /customer/token-usage. The manager turns those rollups into Stripe charges; this
-// binary is the trust-boundary crossing that gets the priced usage OUT of the install.
+// rater's rated_usage rollups for a window — each already carrying its billing org
+// (org_id is captured at METER time from X-Saturn-Org-Id and carried onto rated_usage by
+// the rater, so push reads it straight off the rollup, no resource_name join) — and
+// POSTs an hourly SNAPSHOT to the central manager's POST /customer/token-usage. The
+// manager turns those rollups into Stripe charges; this binary is the trust-boundary
+// crossing that gets the priced usage OUT of the install.
 //
 // See token-factory-stripe-consumer-design.md (Ben). This binary is that design's
 // "install-side hourly push job" upstream prerequisite. The contract points it must
@@ -15,17 +16,20 @@
 //     that is ABSENT from the snapshot as a reconcile-DELETE. So a re-rate or a
 //     reconcile-delete in phoebe self-corrects across the boundary — a bare incremental
 //     delete would silently keep billing a vanished rollup forever.
-//   - ORG RESOLVED INSTALL-SIDE (Decision 2 / C1). resource_id → org_id is joined HERE,
-//     where the resource_name FK lives; the manager never reads the install DB. The
-//     org_id rides the payload. A rollup whose resource_id resolves to NO org makes its
-//     WHOLE WINDOW be WITHHELD (not pushed) and screamed about — because the snapshot is
-//     delete-by-absence, pushing a window with a row omitted would silently DELETE that
-//     row's prior (possibly already-billed) charge. Withholding leaves the manager's
-//     prior good state for the window standing (stale-but-billed); the next run re-pushes
-//     once the resource_name mapping is restored. NEVER pushed with a guessed/empty org
-//     (C2/C7 fail-closed). Because rated_usage.resource_id is NOT NULL and the rater only
-//     writes attributable rows, an unresolved org means a deployment row vanished from
-//     resource_name: a real anomaly worth a non-zero exit.
+//   - ORG CAPTURED INSTALL-SIDE AT METER TIME (Decision 2 / C1). org_id is captured by
+//     the interceptor from the X-Saturn-Org-Id header (the deployment's saturncloud.io/
+//     org-id label, injected by Atlas) and carried billing_event→rated_usage by the
+//     rater; the manager never reads the install DB. The org_id rides the payload off
+//     the rollup — no push-time resource_name join (which raced deployment teardown). A
+//     rollup with a NULL org_id makes its WHOLE WINDOW be WITHHELD (not pushed) and
+//     screamed about — because the snapshot is delete-by-absence, pushing a window with a
+//     row omitted would silently DELETE that row's prior (possibly already-billed)
+//     charge. Withholding leaves the manager's prior good state for the window standing
+//     (stale-but-billed); the next run re-pushes once org_id propagates (a re-rate also
+//     refreshes a NULL-org rollup to its real org). NEVER pushed with a guessed/empty org
+//     (C2/C7 fail-closed). A NULL org_id means Atlas did not inject X-Saturn-Org-Id for
+//     that deployment (a producer-rollout gap) or propagation broke: a real anomaly
+//     worth a non-zero exit.
 //   - MONEY IS AN EXACT DECIMAL STRING end to end (C8). cost and the applied rates cross
 //     the wire as NUMERIC text, never a float — read as ::text from Postgres, emitted as
 //     JSON strings.
@@ -45,7 +49,7 @@
 //	   all-empty run would signal delete-all for every window and wipe the manager's
 //	   prior billing) / a window range wider than the backfill cap
 //	2  some window was WITHHELD (not pushed) because it contained a rated_usage row
-//	   that could NOT be resolved to an org (a deployment missing from resource_name);
+//	   with a NULL org_id (Atlas didn't inject X-Saturn-Org-Id, or propagation broke);
 //	   every window that WAS pushed succeeded. The withheld window's prior state on the
 //	   manager is left untouched (stale-but-billed). Distinct code so a CronJob alerts on
 //	   "usage we metered and priced but cannot attribute to an org" (held revenue / lost

cmd/token-push/push.go

@@ -58,11 +58,11 @@ type snapshot struct {
 }
 
 // openDB opens the shared Atlas Postgres the same way the rater does (pgx stdlib,
-// UTC-pinned DSN, small batch-job pool). token-push reads rated_usage + resource_name
-// from this DB; it never writes.
+// UTC-pinned DSN, small batch-job pool). token-push reads rated_usage from this DB
+// (org_id is carried on the rollup — no resource_name join); it never writes.
 func openDB(ctx context.Context, cfg rating.Config) (*sql.DB, error) {
 	if cfg.DatabaseURL == "" {
-		return nil, fmt.Errorf("token-push: DATABASE_URL is empty (Postgres holds rated_usage and resource_name; the pusher cannot run without it)")
+		return nil, fmt.Errorf("token-push: DATABASE_URL is empty (Postgres holds rated_usage; the pusher cannot run without it)")
 	}
 	db, err := sql.Open("pgx", ensureUTCTimeZone(cfg.DatabaseURL))
 	if err != nil {
@@ -78,21 +78,25 @@ func openDB(ctx context.Context, cfg rating.Config) (*sql.DB, error) {
 	return db, nil
 }
 
-// snapshotQuery reads every rated_usage row whose window_start == $1 and LEFT JOINs
-// resource_name to resolve the billing org. The LEFT JOIN (not INNER) is deliberate:
-// a row whose resource_id has NO resource_name match must still be SEEN here so it can
-// be counted and screamed about (C7), not silently filtered out by the join. Money and
-// rates are read as ::text so they never become a float in Go (C8).
+// snapshotQuery reads every rated_usage row whose window_start == $1. The billing org
+// (org_id) is read DIRECTLY off the rollup: phoebe captures the deployment-owning org at
+// METER time (the X-Saturn-Org-Id header) and the rater carries it onto rated_usage, so
+// push no longer reconstructs it by joining resource_name. That former join raced
+// deployment teardown — a deployment deleted between meter and push lost its
+// resource_name row, making already-metered usage unattributable. With org frozen onto
+// the rollup at meter time, a later teardown cannot un-attribute it.
 //
-// resource_id → resource_name.resource_id → resource_name.org_id (a direct FK to
-// org.id). resource_name is the canonical resource→org map for every resource type, in
-// the same shared Atlas Postgres; phoebe does not own it (assumed-to-exist).
+// org_id is NULLABLE on rated_usage: NULL when the producer header was absent (a
+// rollout gap or an upstream propagation bug). A NULL-org row is still SEEN here (read
+// as a NULL string) so it can be counted and screamed about (C7), exactly as the old
+// LEFT JOIN miss was — the whole window is then withheld, never pushed with a blank org.
+// Money and rates are read as ::text so they never become a float in Go (C8).
 const snapshotQuery = `
 SELECT
     ru.id,
     ru.resource_id,
+    ru.org_id,                       -- NULL when the producer header was absent at meter time
     ru.model_id,
-    rn.org_id,                       -- NULL when the deployment is absent from resource_name
     ru.cost::text,
     ru.applied_prompt_rate::text,
     ru.applied_cached_rate::text,
@@ -102,16 +106,16 @@ SELECT
     ru.completion_tokens,
     ru.billable_prompt_tokens
 FROM rated_usage ru
-LEFT JOIN resource_name rn ON rn.resource_id = ru.resource_id
 WHERE ru.window_start = $1
 ORDER BY ru.id`
 
-// buildSnapshot reads the rollups for windowStart and resolves each to its org. It
-// returns the snapshot (only org-resolved rows) and the count of rows it had to OMIT
-// because their resource_id resolved to no org. An omitted row is an anomaly: the rater
-// only writes attributable (non-NULL resource_id) rows, so a miss here means the
-// deployment vanished from resource_name. Omitted rows are NEVER pushed with a guessed
-// or empty org (C2/C7) — they are held back and the caller exits non-zero.
+// buildSnapshot reads the rollups for windowStart, each already carrying its billing
+// org (captured at meter time, carried onto rated_usage by the rater). It returns the
+// snapshot (only org-resolved rows) and the count of rows it had to OMIT because their
+// org_id was NULL. An omitted row is an anomaly: it means Atlas did not inject
+// X-Saturn-Org-Id for that deployment (a producer-rollout gap) or org_id propagation
+// broke upstream. Omitted rows are NEVER pushed with a guessed or empty org (C2/C7) —
+// they are held back and the caller exits non-zero.
 func (p *pusher) buildSnapshot(ctx context.Context, windowStart time.Time) (snapshot, int, error) {
 	rows, err := p.db.QueryContext(ctx, snapshotQuery, windowStart)
 	if err != nil {
@@ -130,8 +134,8 @@ func (p *pusher) buildSnapshot(ctx context.Context, windowStart time.Time) (snap
 		if err := rows.Scan(
 			&r.RatedUsageID,
 			&r.ResourceID,
-			&r.ModelID,
 			&orgID,
+			&r.ModelID,
 			&r.Cost,
 			&r.AppliedPromptRate,
 			&r.AppliedCachedRate,
@@ -147,9 +151,9 @@ func (p *pusher) buildSnapshot(ctx context.Context, windowStart time.Time) (snap
 		if !orgID.Valid || orgID.String == "" {
 			// Held back, never billed to a guessed org. Scream with the identifying
 			// fields (NOT the cost amount in the clear beyond what's needed) so an
-			// operator can find the orphaned deployment.
+			// operator can find the deployment whose org didn't propagate.
 			p.log.Error.Printf(
-				"token-push: rated_usage %s (resource_id=%s, model_id=%s) has no org in resource_name — OMITTED from the snapshot (held, not billed); deployment missing from resource_name",
+				"token-push: rated_usage %s (resource_id=%s, model_id=%s) has a NULL org_id — OMITTED from the snapshot (held, not billed); Atlas did not inject X-Saturn-Org-Id for this deployment (producer-rollout gap) or org_id propagation broke upstream",
 				r.RatedUsageID, r.ResourceID, r.ModelID,
 			)
 			unattributable++
@@ -201,16 +205,16 @@ func (p *pusher) ratedUsageHasAnyRow(ctx context.Context) (bool, error) {
 // CRITICAL — a window with ANY unattributable row is NOT pushed (Ben's ruling). The
 // snapshot is delete-by-absence on the manager: a rated_usage_id the manager has on
 // file for this window but ABSENT from the snapshot is treated as a reconcile-DELETE
-// (un-bill). So pushing a window with a row OMITTED (because its resource_id no longer
-// resolves to an org) would silently DELETE the prior, possibly-already-billed charge
-// for that row — a money-loss, not a "hold". We instead WITHHOLD the whole window
-// (leave the manager's prior good state for it standing) and scream + exit 2. This is
-// the price-fetch fail-closed posture: stale-but-billed beats silently-un-billed.
-// Convergence resumes automatically on the next run once the resource_name row is
-// restored (the trailing re-push window re-covers the hour). The interim cost is
-// possibly over-holding a window that had a genuinely NEW unresolvable row — over-
-// holding is the safe direction. (A future wire `held` marker scopes delete-by-absence
-// so a held row need not block its whole window; that lands install+manager together.)
+// (un-bill). So pushing a window with a row OMITTED (because its org_id is NULL — Atlas
+// didn't inject X-Saturn-Org-Id, or org_id propagation broke) would silently DELETE the
+// prior, possibly-already-billed charge for that row — a money-loss, not a "hold". We
+// instead WITHHOLD the whole window (leave the manager's prior good state for it
+// standing) and scream + exit 2. This is the price-fetch fail-closed posture:
+// stale-but-billed beats silently-un-billed. Convergence resumes automatically on the
+// next run once the org propagates (the trailing re-push window re-covers the hour, and
+// a re-rate refreshes a NULL-org rollup to its real org). The interim cost is possibly
+// over-holding a window that had a genuinely NEW unattributable row — over-holding is
+// the safe direction.
 //
 // Windows are independent: a withheld or failed window does not abort the others, so a
 // single bad window can't block reconvergence of every other hour.
@@ -249,7 +253,7 @@ func (p *pusher) pushWindows(ctx context.Context, windows []time.Time) int {
 			// manager. Withhold the whole window rather than risk un-billing a prior
 			// charge. Already screamed per-row in buildSnapshot.
 			p.log.Error.Printf(
-				"token-push: WITHHOLDING window %s — it has %d unattributable row(s); pushing the omitted snapshot would delete-by-absence a possibly-billed rollup. Leaving the manager's prior state for this window untouched (restore the resource_name mapping; the next run re-pushes).",
+				"token-push: WITHHOLDING window %s — it has %d unattributable row(s) (NULL org_id); pushing the omitted snapshot would delete-by-absence a possibly-billed rollup. Leaving the manager's prior state for this window untouched (fix X-Saturn-Org-Id propagation; the next run re-pushes once the rollup's org resolves).",
 				snap.WindowStart, unattributable,
 			)
 			withheld++