feat(B7): packaging + CI for helexa-upstream

Ships the mesh authority as an RPM on the same prebuilt-binary pipeline as
helexa-bench.

- rpm/helexa-upstream-prerelease.spec: wraps the CI-built binary; installs
  the systemd unit, sysusers, firewalld service (tcp/8090), and
  /etc/helexa-upstream/helexa-upstream.toml (config noreplace).
- data/helexa-upstream.{service,sysusers.conf,firewalld.xml}: Type=simple
  unit (serve --config), dedicated system user with a StateDirectory home,
  inbound 8090 (/authz/v1 + /web/v1). PostgreSQL is reached outbound; the
  schema migrates on startup.
- build-prerelease.yml: build-upstream + package-upstream jobs with
  change-detection over crates/helexa-upstream/ (UPSTREAM_RE), gated into
  publish. SQLX_OFFLINE=true is set defensively — helexa-upstream uses the
  sqlx runtime query API (no compile-time macros), so it builds with no
  database and no .sqlx cache; DB integration tests stay gated behind
  UPSTREAM_TEST_DATABASE_URL.

Validated: workflow YAML parses, rpmspec expands, and
`SQLX_OFFLINE=true cargo build --release -p helexa-upstream` succeeds.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01F6o3ddqmYNh9kzdwq6eowh

.gitea/workflows/build-prerelease.yml

@@ -66,6 +66,7 @@ jobs:
       build_cortex: ${{ steps.changes.outputs.build_cortex }}
       build_neuron: ${{ steps.changes.outputs.build_neuron }}
       build_bench: ${{ steps.changes.outputs.build_bench }}
+      build_upstream: ${{ steps.changes.outputs.build_upstream }}
       check_rust: ${{ steps.changes.outputs.check_rust }}
     steps:
       - uses: actions/checkout@v4
@@ -104,6 +105,7 @@ jobs:
           BUILD_CORTEX=true
           BUILD_NEURON=true
           BUILD_BENCH=true
+          BUILD_UPSTREAM=true
           CHECK_RUST=true
 
           if [ "${GITHUB_EVENT_NAME}" = "push" ]; then
@@ -149,6 +151,7 @@ jobs:
               NEURON_RE='^crates/neuron/|^crates/cortex-core/|^Cargo\.toml$|^Cargo\.lock$|^rpm/helexa-neuron-prerelease\.spec$|^data/neuron|^neuron\.example\.toml$|^\.gitea/workflows/build-prerelease\.yml$'
               CORTEX_RE='^crates/cortex-gateway/|^crates/cortex-cli/|^crates/cortex-core/|^Cargo\.toml$|^Cargo\.lock$|^rpm/cortex-prerelease\.spec$|^data/cortex|^cortex\.example\.toml$|^models\.example\.toml$|^\.gitea/workflows/build-prerelease\.yml$'
               BENCH_RE='^crates/helexa-bench/|^crates/cortex-core/|^Cargo\.toml$|^Cargo\.lock$|^rpm/helexa-bench-prerelease\.spec$|^data/helexa-bench|^helexa-bench\.example\.toml$|^\.gitea/workflows/build-prerelease\.yml$'
+              UPSTREAM_RE='^crates/helexa-upstream/|^crates/cortex-core/|^Cargo\.toml$|^Cargo\.lock$|^rpm/helexa-upstream-prerelease\.spec$|^data/helexa-upstream|^helexa-upstream\.example\.toml$|^\.gitea/workflows/build-prerelease\.yml$'
               # Any Rust change (incl. crates not packaged here, e.g.
               # helexa-acp) still needs lint+test on main.
               RUST_RE='\.rs$|^crates/|Cargo\.toml$|^Cargo\.lock$'
@@ -156,10 +159,12 @@ jobs:
               CORTEX_BASE=$(base_for cortex)
               NEURON_BASE=$(base_for helexa-neuron-blackwell)
               BENCH_BASE=$(base_for helexa-bench)
+              UPSTREAM_BASE=$(base_for helexa-upstream)
               BUILD_CORTEX=$(decide "$CORTEX_BASE" "$CORTEX_RE")
               BUILD_NEURON=$(decide "$NEURON_BASE" "$NEURON_RE")
               BUILD_BENCH=$(decide "$BENCH_BASE" "$BENCH_RE")
-              if [ "$BUILD_CORTEX" = "true" ] || [ "$BUILD_NEURON" = "true" ] || [ "$BUILD_BENCH" = "true" ]; then
+              BUILD_UPSTREAM=$(decide "$UPSTREAM_BASE" "$UPSTREAM_RE")
+              if [ "$BUILD_CORTEX" = "true" ] || [ "$BUILD_NEURON" = "true" ] || [ "$BUILD_BENCH" = "true" ] || [ "$BUILD_UPSTREAM" = "true" ]; then
                 CHECK_RUST=true
               else
                 CHECK_RUST=$(decide "$CORTEX_BASE" "$RUST_RE")
@@ -170,8 +175,9 @@ jobs:
           echo "build_cortex=${BUILD_CORTEX}" >> "$GITHUB_OUTPUT"
           echo "build_neuron=${BUILD_NEURON}" >> "$GITHUB_OUTPUT"
           echo "build_bench=${BUILD_BENCH}" >> "$GITHUB_OUTPUT"
+          echo "build_upstream=${BUILD_UPSTREAM}" >> "$GITHUB_OUTPUT"
           echo "check_rust=${CHECK_RUST}" >> "$GITHUB_OUTPUT"
-          echo "### change detection: build_cortex=${BUILD_CORTEX} build_neuron=${BUILD_NEURON} build_bench=${BUILD_BENCH} check_rust=${CHECK_RUST}"
+          echo "### change detection: build_cortex=${BUILD_CORTEX} build_neuron=${BUILD_NEURON} build_bench=${BUILD_BENCH} build_upstream=${BUILD_UPSTREAM} check_rust=${CHECK_RUST}"
 
   # fmt + clippy + test moved here from ci.yml for main pushes so the
   # two workflows stop queueing against each other (ci.yml's checks
@@ -303,6 +309,45 @@ jobs:
           path: artifacts/helexa-bench
           retention-days: 1
 
+  build-upstream:
+    name: Build helexa-upstream binary
+    timeout-minutes: 25
+    needs: prepare
+    if: needs.prepare.outputs.build_upstream == 'true'
+    # Pure-Rust, non-CUDA binary — same runner as cortex/bench.
+    runs-on: rust
+    env:
+      RUSTC_WRAPPER: sccache
+      SCCACHE_BUCKET: sccache
+      SCCACHE_ENDPOINT: http://caveman.kosherinata.internal:9000
+      SCCACHE_REGION: auto
+      SCCACHE_S3_USE_SSL: "false"
+      AWS_ACCESS_KEY_ID: ${{ secrets.SCCACHE_S3_ACCESS_KEY }}
+      AWS_SECRET_ACCESS_KEY: ${{ secrets.SCCACHE_S3_SECRET_KEY }}
+      # helexa-upstream uses the sqlx runtime query API (no compile-time
+      # query macros), so it builds without a database or a .sqlx cache.
+      # Set OFFLINE defensively so a stray macro can never reach for a DB.
+      SQLX_OFFLINE: "true"
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          ref: ${{ inputs.ref }}
+
+      - name: Build helexa-upstream (release, sccache escalation)
+        run: script/ci-cargo-escalate.sh cargo build --release -p helexa-upstream
+
+      - name: Stage binary
+        run: |
+          mkdir --parents artifacts
+          cp target/release/helexa-upstream artifacts/helexa-upstream
+          ./artifacts/helexa-upstream --version || true
+
+      - uses: actions/upload-artifact@v3
+        with:
+          name: upstream-fc43
+          path: artifacts/helexa-upstream
+          retention-days: 1
+
   build-neuron:
     name: Build neuron-${{ matrix.flavour }}
     timeout-minutes: 35
@@ -459,6 +504,44 @@ jobs:
           path: ~/rpmbuild/RPMS/x86_64/*.rpm
           retention-days: 7
 
+  package-upstream:
+    name: Package helexa-upstream RPM
+    timeout-minutes: 20
+    needs: [prepare, build-upstream]
+    runs-on: rpm
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          ref: ${{ inputs.ref }}
+
+      - uses: actions/download-artifact@v3
+        with:
+          name: upstream-fc43
+          path: artifacts/
+
+      - name: Build RPM
+        run: |
+          set -eux
+          rm -f ~/.rpmmacros
+          rpmdev-setuptree
+          cp artifacts/helexa-upstream ~/rpmbuild/SOURCES/
+          cp data/helexa-upstream.service ~/rpmbuild/SOURCES/
+          cp data/helexa-upstream-sysusers.conf ~/rpmbuild/SOURCES/
+          cp data/helexa-upstream-firewalld.xml ~/rpmbuild/SOURCES/
+          cp helexa-upstream.example.toml ~/rpmbuild/SOURCES/
+          cp LICENSE ~/rpmbuild/SOURCES/
+          rpmbuild -bb rpm/helexa-upstream-prerelease.spec \
+            --define "upstream_version ${{ needs.prepare.outputs.version }}" \
+            --define "upstream_prerelease ${{ needs.prepare.outputs.release }}" \
+            --undefine dist \
+            --define "dist .fc43"
+
+      - uses: actions/upload-artifact@v3
+        with:
+          name: rpm-upstream-fc43
+          path: ~/rpmbuild/RPMS/x86_64/*.rpm
+          retention-days: 7
+
   package-neuron:
     name: Package helexa-neuron-${{ matrix.flavour }} RPM
     timeout-minutes: 20
@@ -508,7 +591,7 @@ jobs:
   publish:
     name: Publish to rpm.lair.cafe (unstable)
     timeout-minutes: 25
-    needs: [lint, test, package-cortex, package-neuron, package-bench]
+    needs: [lint, test, package-cortex, package-neuron, package-bench, package-upstream]
     # Runs when at least one package was built and nothing failed.
     # lint/test may be skipped (docs-only refs never get here because
     # no packages build), but a real failure in any blocks the
@@ -518,10 +601,11 @@ jobs:
         !cancelled()
         && (needs.lint.result == 'success' || needs.lint.result == 'skipped')
         && (needs.test.result == 'success' || needs.test.result == 'skipped')
-        && (needs.package-cortex.result == 'success' || needs.package-neuron.result == 'success' || needs.package-bench.result == 'success')
+        && (needs.package-cortex.result == 'success' || needs.package-neuron.result == 'success' || needs.package-bench.result == 'success' || needs.package-upstream.result == 'success')
         && needs.package-cortex.result != 'failure'
         && needs.package-neuron.result != 'failure'
         && needs.package-bench.result != 'failure'
+        && needs.package-upstream.result != 'failure'
       }}
     runs-on: rpm
     concurrency:

cortex.example.toml

@@ -105,3 +105,5 @@ enabled = false
 # upstream). Override via CORTEX_UPSTREAM__BEARER in prod.
 # bearer = "replace-with-operator-client-secret"
 # timeout_secs = 5
+# How often to flush served-usage counters to upstream for reconciliation (#58).
+# served_usage_report_interval_secs = 60

crates/cortex-core/src/config.rs

@@ -48,11 +48,18 @@ pub struct UpstreamClientConfig {
     /// Per-call timeout (seconds) to upstream.
     #[serde(default = "default_upstream_timeout")]
     pub timeout_secs: u64,
+    /// How often (seconds) to flush served-usage counters to upstream for
+    /// reconciliation (#58).
+    #[serde(default = "default_served_usage_interval")]
+    pub served_usage_report_interval_secs: u64,
 }
 
 fn default_upstream_timeout() -> u64 {
     5
 }
+fn default_served_usage_interval() -> u64 {
+    60
+}
 
 /// `[entitlements]` — the local/static [`crate::entitlements::EntitlementProvider`]
 /// source of truth (#50). Accounts, keys, and hard caps live here; the

crates/cortex-gateway/src/handlers.rs

@@ -322,7 +322,11 @@ async fn anthropic_messages(
             )
             .await
             {
-                Ok(guard) => Some(crate::metering::usage_sink(principal, guard)),
+                Ok(guard) => Some(crate::metering::usage_sink(
+                    principal,
+                    guard,
+                    std::sync::Arc::clone(&fleet.served_usage),
+                )),
                 Err(env) => return crate::error::envelope_response(env),
             }
         }
@@ -802,7 +806,11 @@ async fn proxy_with_metrics(
             )
             .await
             {
-                Ok(guard) => Some(crate::metering::usage_sink(principal, guard)),
+                Ok(guard) => Some(crate::metering::usage_sink(
+                    principal,
+                    guard,
+                    std::sync::Arc::clone(&fleet.served_usage),
+                )),
                 Err(env) => return crate::error::envelope_response(env),
             }
         }

crates/cortex-gateway/src/lib.rs

@@ -11,6 +11,7 @@ pub mod metrics;
 pub mod poller;
 pub mod proxy;
 pub mod router;
+pub mod served_usage;
 pub mod state;
 
 use anyhow::Result;
@@ -57,6 +58,28 @@ pub async fn run(config: GatewayConfig) -> Result<()> {
         evictor::eviction_loop(evictor_fleet).await;
     });
 
+    // Served-usage reporter (#58): when this operator is part of the mesh,
+    // periodically flush absolute per-principal served-token counters to
+    // upstream for reconciliation.
+    if config.upstream.enabled {
+        let su_fleet = Arc::clone(&fleet);
+        let url = config.upstream.url.clone();
+        let bearer = config.upstream.bearer.clone();
+        let interval =
+            std::time::Duration::from_secs(config.upstream.served_usage_report_interval_secs);
+        tokio::spawn(async move {
+            loop {
+                tokio::time::sleep(interval).await;
+                let rows = su_fleet.served_usage.snapshot();
+                if let Err(e) =
+                    served_usage::report(&su_fleet.http_client, &url, &bearer, &rows).await
+                {
+                    tracing::warn!(error = %e, "served-usage report failed (will retry)");
+                }
+            }
+        });
+    }
+
     let app = build_app(Arc::clone(&fleet));
 
     let listen_addr = config.gateway.listen.parse::<std::net::SocketAddr>()?;

crates/cortex-gateway/src/metering.rs

@@ -117,9 +117,21 @@ impl Drop for ReservationGuard {
 /// Build the completion sink for an authenticated request: record spend and
 /// settle the reservation with the observed total. Dropping it unused (no
 /// usage observed) releases the reservation via the guard.
-pub fn usage_sink(principal: Principal, guard: ReservationGuard) -> UsageSink {
+pub fn usage_sink(
+    principal: Principal,
+    guard: ReservationGuard,
+    served_usage: std::sync::Arc<crate::served_usage::ServedUsage>,
+) -> UsageSink {
     Box::new(move |prompt, completion| {
         record_spend(&principal, prompt, completion);
+        // Per-principal served-usage tally for #58 reconciliation. Recorded
+        // for every metered (authenticated) request; the flush task reports
+        // it to upstream when the operator is part of the mesh.
+        served_usage.add(
+            &principal.account_id,
+            &principal.key_id,
+            prompt + completion,
+        );
         guard.settle(prompt + completion);
     })
 }

crates/cortex-gateway/src/served_usage.rs

@@ -0,0 +1,105 @@
+//! Served-usage ledger (#58): cortex meters, per principal and per UTC day,
+//! the tokens it has served on behalf of mesh accounts, and periodically
+//! reports **absolute** cumulative counters to helexa-upstream for
+//! reconciliation (operators are compensated for served tokens).
+//!
+//! Counters are cumulative-since-process-start for the current period;
+//! upstream upserts them monotonically (GREATEST), so re-sending the same
+//! value is idempotent and a flush that races another is harmless. (A
+//! process restart resets the in-memory counter; the monotonic upsert keeps
+//! upstream from regressing — at most it under-counts the restarted window,
+//! acceptable for beta. One cortex per operator token is assumed.)
+
+use serde::Serialize;
+use std::collections::HashMap;
+use std::sync::Mutex;
+
+#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
+pub struct ServedRow {
+    pub account_id: String,
+    pub key_id: String,
+    pub period: String, // YYYY-MM-DD (UTC)
+    pub served_tokens: u64,
+}
+
+#[derive(Default)]
+pub struct ServedUsage {
+    inner: Mutex<HashMap<(String, String, String), u64>>,
+}
+
+impl ServedUsage {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Add served tokens for a principal in today's (UTC) period.
+    pub fn add(&self, account_id: &str, key_id: &str, tokens: u64) {
+        if tokens == 0 {
+            return;
+        }
+        let period = chrono::Utc::now().format("%Y-%m-%d").to_string();
+        let mut m = self.inner.lock().expect("served-usage lock");
+        *m.entry((account_id.to_string(), key_id.to_string(), period))
+            .or_insert(0) += tokens;
+    }
+
+    /// Absolute cumulative counters, for a flush to upstream.
+    pub fn snapshot(&self) -> Vec<ServedRow> {
+        let m = self.inner.lock().expect("served-usage lock");
+        m.iter()
+            .map(|((account_id, key_id, period), &served_tokens)| ServedRow {
+                account_id: account_id.clone(),
+                key_id: key_id.clone(),
+                period: period.clone(),
+                served_tokens,
+            })
+            .collect()
+    }
+}
+
+/// POST the absolute counters to upstream's `/authz/v1/served-usage`.
+pub async fn report(
+    client: &reqwest::Client,
+    base_url: &str,
+    bearer: &str,
+    rows: &[ServedRow],
+) -> Result<(), reqwest::Error> {
+    if rows.is_empty() {
+        return Ok(());
+    }
+    let url = format!("{}/authz/v1/served-usage", base_url.trim_end_matches('/'));
+    client
+        .post(url)
+        .bearer_auth(bearer)
+        .json(&serde_json::json!({ "rows": rows }))
+        .send()
+        .await?
+        .error_for_status()?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn accumulates_per_principal_and_period() {
+        let su = ServedUsage::new();
+        su.add("acct", "key", 10);
+        su.add("acct", "key", 5);
+        su.add("acct", "other", 7);
+        su.add("acct", "key", 0); // no-op
+        let mut rows = su.snapshot();
+        rows.sort_by(|a, b| a.key_id.cmp(&b.key_id));
+        assert_eq!(rows.len(), 2);
+        let key_row = rows.iter().find(|r| r.key_id == "key").unwrap();
+        assert_eq!(key_row.served_tokens, 15);
+        assert_eq!(
+            rows.iter()
+                .find(|r| r.key_id == "other")
+                .unwrap()
+                .served_tokens,
+            7
+        );
+    }
+}

crates/cortex-gateway/src/state.rs

@@ -22,6 +22,9 @@ pub struct CortexState {
     /// Whether to reject unauthenticated requests (#49). Read by the auth
     /// middleware once it lands.
     pub require_auth: bool,
+    /// Per-principal served-token tally (#58), reported to upstream for
+    /// operator reconciliation by the flush task when upstream is enabled.
+    pub served_usage: Arc<crate::served_usage::ServedUsage>,
 }
 
 impl CortexState {
@@ -73,6 +76,7 @@ impl CortexState {
                 .expect("failed to build HTTP client"),
             entitlements,
             require_auth: config.entitlements.require_auth,
+            served_usage: Arc::new(crate::served_usage::ServedUsage::new()),
         }
     }
 }

crates/cortex-gateway/tests/upstream_chain.rs

@@ -60,6 +60,7 @@ fn chain(local: LocalEntitlementProvider, url: &str) -> ChainedEntitlementProvid
         url: url.to_string(),
         bearer: "client-secret".into(),
         timeout_secs: 5,
+        served_usage_report_interval_secs: 60,
     });
     ChainedEntitlementProvider::new(local, upstream)
 }

crates/helexa-upstream/src/authz.rs

@@ -22,7 +22,7 @@ use axum::http::{StatusCode, header};
 use axum::middleware::Next;
 use axum::response::{IntoResponse, Response};
 use axum::routing::post;
-use axum::{Json, Router};
+use axum::{Extension, Json, Router};
 use cortex_core::error_envelope::OpenAiError;
 use serde::{Deserialize, Serialize};
 use subtle::ConstantTimeEq;
@@ -41,6 +41,7 @@ pub fn router(state: &AppState) -> Router<AppState> {
         .route("/authz/v1/settle", post(settle))
         .route("/authz/v1/release", post(release))
         .route("/authz/v1/snapshot", post(snapshot))
+        .route("/authz/v1/served-usage", post(served_usage))
         .layer(axum::middleware::from_fn_with_state(
             state.clone(),
             client_auth,
@@ -274,6 +275,61 @@ async fn snapshot(State(state): State<AppState>, Json(req): Json<SnapshotReq>) -
     }
 }
 
+// ── served-usage report (#58) ───────────────────────────────────────
+
+#[derive(Deserialize)]
+struct ServedUsageReport {
+    rows: Vec<ServedUsageRow>,
+}
+
+#[derive(Deserialize)]
+struct ServedUsageRow {
+    account_id: String,
+    key_id: String,
+    period: String, // YYYY-MM-DD
+    served_tokens: i64,
+}
+
+/// `POST /authz/v1/served-usage` — a cortex reports the absolute served-token
+/// counters it has accrued for the current period. Upsert is monotonic
+/// (`GREATEST`) so re-sends and races are idempotent and never regress.
+/// `operator_id` comes from the validated client bearer (request extension).
+async fn served_usage(
+    State(state): State<AppState>,
+    Extension(operator): Extension<OperatorId>,
+    Json(req): Json<ServedUsageReport>,
+) -> Response {
+    for row in &req.rows {
+        let (Ok(account_id), Ok(key_id)) = (
+            Uuid::parse_str(&row.account_id),
+            Uuid::parse_str(&row.key_id),
+        ) else {
+            continue; // skip malformed ids rather than fail the whole batch
+        };
+        let Ok(period) = chrono::NaiveDate::parse_from_str(&row.period, "%Y-%m-%d") else {
+            continue;
+        };
+        let res = sqlx::query(
+            "INSERT INTO served_usage (operator_id, account_id, key_id, period, served_tokens) \
+             VALUES ($1, $2, $3, $4, $5) \
+             ON CONFLICT (operator_id, account_id, key_id, period) \
+             DO UPDATE SET served_tokens = GREATEST(served_usage.served_tokens, EXCLUDED.served_tokens)",
+        )
+        .bind(&operator.0)
+        .bind(account_id)
+        .bind(key_id)
+        .bind(period)
+        .bind(row.served_tokens.max(0))
+        .execute(&state.pool)
+        .await;
+        if let Err(e) = res {
+            tracing::error!(error = %e, "served-usage upsert failed");
+            return envelope_response(OpenAiError::service_unavailable("authority error", Some(5)));
+        }
+    }
+    StatusCode::NO_CONTENT.into_response()
+}
+
 fn bad_request(msg: &str) -> Response {
     envelope_response(OpenAiError::new(
         400,

crates/helexa-upstream/src/lib.rs

@@ -20,6 +20,7 @@ pub mod email;
 pub mod error;
 pub mod handlers;
 pub mod ledger;
+pub mod reconcile;
 pub mod state;
 pub mod topup;
 pub mod web;

crates/helexa-upstream/src/main.rs

@@ -36,6 +36,12 @@ enum Commands {
         #[arg(long)]
         denomination: Option<String>,
     },
+    /// Roll up not-yet-reconciled served usage per operator/period (#58),
+    /// stamp it reconciled, and print the totals. Payout is out of scope.
+    Reconcile {
+        #[arg(short, long, default_value = "helexa-upstream.toml")]
+        config: String,
+    },
 }
 
 #[tokio::main]
@@ -75,6 +81,18 @@ async fn main() -> Result<()> {
                 println!("{code}");
             }
         }
+        Commands::Reconcile { config } => {
+            let cfg = UpstreamConfig::load(&config)
+                .map_err(|e| anyhow::anyhow!("failed to load config from '{config}': {e}"))?;
+            let pool =
+                helexa_upstream::db::connect_and_migrate(&cfg.db.url, cfg.db.max_connections)
+                    .await?;
+            let rollup = helexa_upstream::reconcile::reconcile(&pool).await?;
+            for r in &rollup {
+                println!("{}\t{}\t{}", r.operator_id, r.period, r.total_served_tokens);
+            }
+            tracing::info!(operators_periods = rollup.len(), "reconciliation complete");
+        }
     }
 
     Ok(())

crates/helexa-upstream/src/reconcile.rs

@@ -0,0 +1,43 @@
+//! Reconciliation rollup (#58): aggregate the served-usage ledger per
+//! operator and period for operator compensation, stamping rows
+//! `reconciled_at` so each window is settled once. The payout mechanism
+//! itself is out of scope — this produces the authoritative per-operator
+//! totals a settlement process consumes.
+
+use sqlx::Row;
+use sqlx::postgres::PgPool;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RollupRow {
+    pub operator_id: String,
+    pub period: chrono::NaiveDate,
+    pub total_served_tokens: i64,
+}
+
+/// Roll up all not-yet-reconciled served-usage into per-(operator, period)
+/// totals, then stamp those rows `reconciled_at`. Returns the rollup.
+/// Idempotent: a second run finds nothing unreconciled and returns empty.
+pub async fn reconcile(pool: &PgPool) -> Result<Vec<RollupRow>, sqlx::Error> {
+    let mut tx = pool.begin().await?;
+    let rows = sqlx::query(
+        // SUM(bigint) is numeric in Postgres — cast back to bigint for i64.
+        "SELECT operator_id, period, SUM(served_tokens)::bigint AS total \
+         FROM served_usage WHERE reconciled_at IS NULL \
+         GROUP BY operator_id, period ORDER BY operator_id, period",
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+    let rollup: Vec<RollupRow> = rows
+        .iter()
+        .map(|r| RollupRow {
+            operator_id: r.get("operator_id"),
+            period: r.get("period"),
+            total_served_tokens: r.get::<i64, _>("total"),
+        })
+        .collect();
+    sqlx::query("UPDATE served_usage SET reconciled_at = now() WHERE reconciled_at IS NULL")
+        .execute(&mut *tx)
+        .await?;
+    tx.commit().await?;
+    Ok(rollup)
+}

crates/helexa-upstream/tests/served_usage_pg.rs

@@ -0,0 +1,116 @@
+//! Integration test for the served-usage report (#58): the idempotent,
+//! monotonic upsert and the reconcile rollup. Gated on
+//! UPSTREAM_TEST_DATABASE_URL (skips cleanly when unset).
+
+use helexa_upstream::config::{ClientToken, UpstreamConfig};
+use helexa_upstream::db::connect_and_migrate;
+use helexa_upstream::email::EmailSender;
+use helexa_upstream::reconcile::reconcile;
+use helexa_upstream::state::AppState;
+use serde_json::{Value, json};
+use sqlx::Row;
+use sqlx::postgres::PgPool;
+use uuid::Uuid;
+
+const CLIENT_TOKEN: &str = "su-test-token";
+const OPERATOR: &str = "op-su-test";
+
+async fn spawn_or_skip(test: &str) -> Option<(String, PgPool)> {
+    let Ok(url) = std::env::var("UPSTREAM_TEST_DATABASE_URL") else {
+        eprintln!("skipping {test}: UPSTREAM_TEST_DATABASE_URL not set");
+        return None;
+    };
+    let pool = connect_and_migrate(&url, 16).await.expect("migrate");
+    let mut config = UpstreamConfig {
+        server: Default::default(),
+        db: helexa_upstream::config::DbSettings {
+            url,
+            max_connections: 16,
+        },
+        grant: Default::default(),
+        abuse: Default::default(),
+        client_auth: Default::default(),
+        authz: Default::default(),
+        auth: Default::default(),
+        email: Default::default(),
+    };
+    config.client_auth.tokens.push(ClientToken {
+        token: CLIENT_TOKEN.into(),
+        operator_id: OPERATOR.into(),
+    });
+    let email = EmailSender::from_config(&config.email).unwrap();
+    let state = AppState::new(pool.clone(), config, email);
+    let app = helexa_upstream::build_app(state);
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        axum::serve(listener, app).await.unwrap();
+    });
+    Some((format!("http://{addr}"), pool))
+}
+
+async fn report(base: &str, rows: Value) -> u16 {
+    reqwest::Client::new()
+        .post(format!("{base}/authz/v1/served-usage"))
+        .bearer_auth(CLIENT_TOKEN)
+        .json(&json!({ "rows": rows }))
+        .send()
+        .await
+        .unwrap()
+        .status()
+        .as_u16()
+}
+
+async fn stored(pool: &PgPool, account: Uuid, key: Uuid) -> i64 {
+    sqlx::query(
+        "SELECT served_tokens FROM served_usage WHERE operator_id = $1 AND account_id = $2 AND key_id = $3",
+    )
+    .bind(OPERATOR)
+    .bind(account)
+    .bind(key)
+    .fetch_one(pool)
+    .await
+    .unwrap()
+    .get("served_tokens")
+}
+
+#[tokio::test]
+async fn served_usage_upsert_is_monotonic_and_reconciles() {
+    let Some((base, pool)) = spawn_or_skip("served_usage_upsert_is_monotonic_and_reconciles").await
+    else {
+        return;
+    };
+    let account = Uuid::new_v4();
+    let key = Uuid::new_v4();
+    let period = "2026-06-23";
+    let row = |n: i64| json!([{"account_id": account, "key_id": key, "period": period, "served_tokens": n}]);
+
+    // First report.
+    assert_eq!(report(&base, row(100)).await, 204);
+    assert_eq!(stored(&pool, account, key).await, 100);
+
+    // Re-send a higher absolute value → advances.
+    assert_eq!(report(&base, row(250)).await, 204);
+    assert_eq!(stored(&pool, account, key).await, 250);
+
+    // A lower value (e.g. a restarted cortex) must NOT regress (GREATEST).
+    assert_eq!(report(&base, row(50)).await, 204);
+    assert_eq!(stored(&pool, account, key).await, 250);
+
+    // Re-sending the same value is idempotent.
+    assert_eq!(report(&base, row(250)).await, 204);
+    assert_eq!(stored(&pool, account, key).await, 250);
+
+    // Reconcile rolls it up and stamps reconciled_at; a second run is empty.
+    let rollup = reconcile(&pool).await.unwrap();
+    let mine = rollup
+        .iter()
+        .find(|r| r.operator_id == OPERATOR)
+        .expect("operator in rollup");
+    assert!(mine.total_served_tokens >= 250);
+    let again = reconcile(&pool).await.unwrap();
+    assert!(
+        again.iter().all(|r| r.operator_id != OPERATOR),
+        "already reconciled"
+    );
+}

data/helexa-upstream-firewalld.xml

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="utf-8"?>
+<service>
+  <short>helexa-upstream</short>
+  <description>helexa-upstream — mesh account + budget authority API (/authz/v1 for cortex, /web/v1 for the frontend)</description>
+  <port protocol="tcp" port="8090"/>
+</service>

data/helexa-upstream-sysusers.conf

@@ -0,0 +1,3 @@
+g helexa-upstream - -
+u helexa-upstream - "helexa-upstream authority" /var/lib/helexa-upstream /sbin/nologin
+m helexa-upstream helexa-upstream

data/helexa-upstream.service

@@ -0,0 +1,22 @@
+[Unit]
+Description=helexa-upstream — mesh account + budget authority (accounts, API keys, allocation ledger, top-up codes)
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+ExecStart=/usr/bin/helexa-upstream serve --config /etc/helexa-upstream/helexa-upstream.toml
+# HTTP authority for cortex (/authz/v1) and the frontend (/web/v1); restart
+# unconditionally if it ever exits. It connects out to PostgreSQL (the
+# system of record) and runs schema migrations on startup.
+Restart=always
+RestartSec=10
+User=helexa-upstream
+Group=helexa-upstream
+# Service user home; no local state (PostgreSQL holds everything), but
+# StateDirectory gives the user a writable, correctly-owned home.
+StateDirectory=helexa-upstream
+StateDirectoryMode=0755
+
+[Install]
+WantedBy=multi-user.target

rpm/helexa-upstream-prerelease.spec

@@ -0,0 +1,109 @@
+# Prebuilt-binary spec for helexa-upstream.
+#
+# Wraps a pre-built `helexa-upstream` binary produced by an upstream CI
+# job and packages it for rpm.lair.cafe. The %build phase is a no-op.
+# helexa-upstream is a pure-Rust, non-CUDA daemon: the mesh account +
+# budget authority. It serves an inbound HTTP API on tcp/8090 (/authz/v1
+# for cortex, /web/v1 for the frontend) — hence the firewalld service —
+# and connects out to PostgreSQL (the system of record), running schema
+# migrations on startup.
+#
+# Required defines at rpmbuild time:
+#   upstream_version    e.g. "0.1.16"
+#   upstream_prerelease e.g. "0.1.20260518140530.gitabcdef0"
+#                              ^^^^^^^^^^^^^^^^^^ ^^^^^^^^
+#                              commit time (sec)  commit sha
+#                             (used as Release; the timestamp prefix
+#                              keeps same-day builds strictly ordered.)
+
+%global _build_id_links none
+%global debug_package %{nil}
+%global __strip /usr/bin/true
+
+%{!?upstream_version: %global upstream_version 0.0.0}
+%if 0%{?upstream_prerelease:1}
+%global upstream_release %{upstream_prerelease}
+%else
+%global upstream_release 1
+%endif
+
+Name:           helexa-upstream
+Version:        %{upstream_version}
+Release:        %{upstream_release}%{?dist}
+Summary:        Mesh account + budget authority for helexa (prebuilt)
+
+License:        GPL-3.0-or-later
+URL:            https://git.lair.cafe/helexa/helexa
+
+Source0:        helexa-upstream
+Source1:        helexa-upstream.service
+Source2:        helexa-upstream-sysusers.conf
+Source3:        helexa-upstream.example.toml
+Source4:        LICENSE
+Source5:        helexa-upstream-firewalld.xml
+
+Requires:       firewalld-filesystem
+
+ExclusiveArch:  x86_64
+
+Requires(pre):  shadow-utils
+Requires:       systemd
+
+Provides:       user(helexa-upstream)
+
+%description
+helexa-upstream is the mesh-level authority: it issues user accounts and
+API keys, holds the allocation ledger (free grant + redeemable top-up
+codes), enforces per-key budgets via a transactional reserve/settle
+contract, and reconciles served usage for operator compensation. cortex
+validates locally-unrecognised keys against its /authz/v1 surface
+(fail-closed); the helexa.ai frontend drives account self-service via
+/web/v1. PostgreSQL is the system of record; the schema is migrated on
+startup.
+
+%prep
+cp %{SOURCE0} ./helexa-upstream
+cp %{SOURCE1} .
+cp %{SOURCE2} .
+cp %{SOURCE3} .
+cp %{SOURCE4} .
+cp %{SOURCE5} .
+
+%build
+# Already built in the upstream CI build job.
+
+%install
+install -Dm755 helexa-upstream %{buildroot}%{_bindir}/helexa-upstream
+install -Dm644 helexa-upstream.service %{buildroot}%{_unitdir}/helexa-upstream.service
+install -Dm644 helexa-upstream-sysusers.conf %{buildroot}%{_sysusersdir}/helexa-upstream.conf
+install -Dm644 helexa-upstream-firewalld.xml %{buildroot}%{_prefix}/lib/firewalld/services/helexa-upstream.xml
+install -dm755 %{buildroot}%{_sysconfdir}/helexa-upstream
+install -Dm644 helexa-upstream.example.toml %{buildroot}%{_sysconfdir}/helexa-upstream/helexa-upstream.toml
+
+%pre
+getent group helexa-upstream >/dev/null || groupadd -r helexa-upstream
+getent passwd helexa-upstream >/dev/null || \
+    useradd -r -g helexa-upstream -d /var/lib/helexa-upstream -s /sbin/nologin \
+        -c "helexa-upstream authority" helexa-upstream
+
+%post
+%systemd_post helexa-upstream.service
+
+%preun
+%systemd_preun helexa-upstream.service
+
+%postun
+%systemd_postun_with_restart helexa-upstream.service
+
+%files
+%license LICENSE
+%{_bindir}/helexa-upstream
+%{_unitdir}/helexa-upstream.service
+%{_sysusersdir}/helexa-upstream.conf
+%{_prefix}/lib/firewalld/services/helexa-upstream.xml
+%dir %{_sysconfdir}/helexa-upstream
+%config(noreplace) %{_sysconfdir}/helexa-upstream/helexa-upstream.toml
+
+%changelog
+* Mon Jun 23 2026 Gitea Actions <actions@git.lair.cafe> - %{upstream_version}-%{upstream_release}
+- Prerelease build from upstream CI binary.