Add hosted agent metering upload path

cli/commands/trade.py

@@ -154,10 +154,7 @@ def trade_cmd(
                 os.environ.get("NUNCHI_TRADE_LEDGER_PATH") or str(Path(data_dir) / "trades.jsonl")
             )
             trade_log.append({
-                "experiment_id": experiment.experiment_id,
-                "run_id": experiment.run_id,
-                "agent_id": experiment.agent_id,
-                "job_type": experiment.job_type,
+                **experiment.ledger_fields(),
                 "ts": int(time.time() * 1000),
                 "tick": tick_index,
                 "tick_index": tick_index,
@@ -169,6 +166,7 @@ def trade_cmd(
                 "side": fill.side,
                 "price": str(fill.price),
                 "quantity": str(fill.quantity),
+                "notional_usd": str(fill.price * fill.quantity),
                 "timestamp_ms": fill.timestamp_ms,
                 "fee": str(fill.fee),
                 "strategy": "manual_trade",

cli/engine.py

@@ -295,6 +295,7 @@ def _tick(self) -> None:
                 "side": fill.side,
                 "price": str(fill.price),
                 "quantity": str(fill.quantity),
+                "notional_usd": str(fill.price * fill.quantity),
                 "timestamp_ms": fill.timestamp_ms,
                 "fee": str(fill.fee),
                 "strategy": self.strategy.strategy_id,
@@ -437,6 +438,7 @@ def _guard_close_position(self, snapshot: MarketSnapshot) -> None:
                 "side": fill.side,
                 "price": str(fill.price),
                 "quantity": str(fill.quantity),
+                "notional_usd": str(fill.price * fill.quantity),
                 "timestamp_ms": fill.timestamp_ms,
                 "fee": str(fill.fee),
                 "strategy": self.strategy.strategy_id,
@@ -523,6 +525,7 @@ def _close_all_positions(self) -> None:
                 "side": fill.side,
                 "price": str(fill.price),
                 "quantity": str(fill.quantity),
+                "notional_usd": str(fill.price * fill.quantity),
                 "timestamp_ms": fill.timestamp_ms,
                 "fee": str(fill.fee),
                 "strategy": self.strategy.strategy_id,
@@ -567,12 +570,7 @@ def _log_tick(self, snapshot, decisions, fills, ok: bool) -> None:
     def _experiment_fields(self) -> Dict[str, Any]:
         if not self.experiment.enabled:
             return {}
-        return {
-            "experiment_id": self.experiment.experiment_id,
-            "run_id": self.experiment.run_id,
-            "agent_id": self.experiment.agent_id,
-            "job_type": self.experiment.job_type,
-        }
+        return self.experiment.ledger_fields()
 
     def _decision_fields(self, decisions=None) -> Dict[str, Any]:
         fields: Dict[str, Any] = {"tick_index": self.tick_count}

modules/cost_metering.py

@@ -42,20 +42,54 @@ class ExperimentContext:
     run_id: str
     agent_id: str
     job_type: str
+    user_id: str = ""
+    account_id: str = ""
+    plan_id: str = ""
+    subscription_id: str = ""
+    billing_period_start: str = ""
+    billing_period_end: str = ""
 
     @classmethod
     def from_env(cls, strategy_id: str) -> "ExperimentContext":
         run_id = os.environ.get("NUNCHI_RUN_ID") or f"manual-{int(time.time())}"
+        hosted_identity_present = bool(os.environ.get("NUNCHI_USER_ID") or os.environ.get("NUNCHI_ACCOUNT_ID"))
+        experiment_id = os.environ.get("NUNCHI_EXPERIMENT_ID", "")
+        if not experiment_id and hosted_identity_present:
+            experiment_id = "hosted-agent"
         return cls(
-            experiment_id=os.environ.get("NUNCHI_EXPERIMENT_ID", ""),
+            experiment_id=experiment_id,
             run_id=run_id,
             agent_id=os.environ.get("NUNCHI_AGENT_ID") or strategy_id,
             job_type=os.environ.get("NUNCHI_JOB_TYPE", "unknown"),
+            user_id=os.environ.get("NUNCHI_USER_ID", ""),
+            account_id=os.environ.get("NUNCHI_ACCOUNT_ID", ""),
+            plan_id=os.environ.get("NUNCHI_PLAN_ID", ""),
+            subscription_id=os.environ.get("NUNCHI_SUBSCRIPTION_ID", ""),
+            billing_period_start=os.environ.get("NUNCHI_BILLING_PERIOD_START", ""),
+            billing_period_end=os.environ.get("NUNCHI_BILLING_PERIOD_END", ""),
         )
 
     @property
     def enabled(self) -> bool:
-        return bool(self.experiment_id)
+        return bool(self.experiment_id or self.user_id or self.account_id or self.subscription_id)
+
+    def ledger_fields(self) -> Dict[str, str]:
+        fields = {
+            "experiment_id": self.experiment_id,
+            "run_id": self.run_id,
+            "agent_id": self.agent_id,
+            "job_type": self.job_type,
+        }
+        optional_fields = {
+            "user_id": self.user_id,
+            "account_id": self.account_id,
+            "plan_id": self.plan_id,
+            "subscription_id": self.subscription_id,
+            "billing_period_start": self.billing_period_start,
+            "billing_period_end": self.billing_period_end,
+        }
+        fields.update({key: value for key, value in optional_fields.items() if value})
+        return fields
 
 
 class OpenRouterPricing:
@@ -190,12 +224,9 @@ def record_llm_call(
                 cache_savings = None
         ts_ms = _now_ms()
         row = {
-            "experiment_id": self.context.experiment_id,
-            "run_id": self.context.run_id,
+            **self.context.ledger_fields(),
             "ts": ts_ms,
-            "agent_id": self.context.agent_id,
             "strategy": self.strategy,
-            "job_type": self.context.job_type,
             "tick_index": tick_index,
             "decision_call_id": decision_call_id,
             "provider": provider,
@@ -227,11 +258,8 @@ def record_llm_call(
 
         if provider == "openrouter":
             route_row = {
-                "experiment_id": self.context.experiment_id,
-                "run_id": self.context.run_id,
+                **self.context.ledger_fields(),
                 "ts": ts_ms,
-                "agent_id": self.context.agent_id,
-                "job_type": self.context.job_type,
                 "tick_index": tick_index,
                 "decision_call_id": decision_call_id,
                 "requested_route": route,

scripts/entrypoint.py

@@ -54,6 +54,13 @@ def _tail_jsonl(path: Path, limit: int = 20) -> list[dict]:
 def _pricing_snapshot(data_dir: str, limit: int = 20) -> dict:
     """Return non-secret pricing-loop status and recent ledger rows."""
     base = Path(data_dir)
+    quota_status_path = Path(os.environ.get("NUNCHI_METERING_QUOTA_STATUS_PATH") or base / ".metering_quota_status.json")
+    quota_status = None
+    if quota_status_path.exists():
+        try:
+            quota_status = json.loads(quota_status_path.read_text("utf-8"))
+        except (OSError, json.JSONDecodeError):
+            quota_status = {"status": "unreadable"}
     ledgers = {
         "cost": base / "cost_ledger.jsonl",
         "route": base / "route_ledger.jsonl",
@@ -76,6 +83,7 @@ def _pricing_snapshot(data_dir: str, limit: int = 20) -> dict:
         "plan_id": os.environ.get("NUNCHI_PLAN_ID"),
         "subscription_id": os.environ.get("NUNCHI_SUBSCRIPTION_ID"),
         "metering_enabled": bool(os.environ.get("NUNCHI_METERING_URL") and os.environ.get("NUNCHI_METERING_TOKEN")),
+        "quota_status": quota_status,
         "data_dir": data_dir,
         "child_alive": CHILD_PROC.poll() is None if CHILD_PROC else False,
         "ledger_exists": {name: path.exists() for name, path in ledgers.items()},

scripts/metering_upload.py

@@ -0,0 +1,180 @@
+#!/usr/bin/env python3
+"""Upload hosted-agent metering rows to web-auth.
+
+The hosted runtime keeps local JSONL ledgers as the source of truth, then this
+uploader batches unsent rows to the subscription metering API. It is safe to
+restart: sent row IDs are persisted locally and web-auth also dedupes rows.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import signal
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any, Iterable
+
+LEDGER_FILES = {
+    "cost": "cost_ledger.jsonl",
+    "route": "route_ledger.jsonl",
+    "runtime": "agent_runtime_ledger.jsonl",
+    "incident": "incident_ledger.jsonl",
+    "trade": "trades.jsonl",
+}
+
+
+def _read_jsonl(path: Path) -> Iterable[dict[str, Any]]:
+    if not path.exists():
+        return []
+    rows = []
+    with path.open() as handle:
+        for line in handle:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                rows.append(json.loads(line))
+            except json.JSONDecodeError:
+                continue
+    return rows
+
+
+def _row_id(ledger: str, row: dict[str, Any]) -> str:
+    stable = {
+        "ledger": ledger,
+        "experiment_id": row.get("experiment_id"),
+        "run_id": row.get("run_id"),
+        "agent_id": row.get("agent_id"),
+        "tick_index": row.get("tick_index") or row.get("tick"),
+        "decision_call_id": row.get("decision_call_id"),
+        "generation_id": row.get("generation_id") or (row.get("route_metadata") or {}).get("generation_id"),
+        "oid": row.get("oid"),
+        "ts": row.get("ts") or row.get("timestamp_ms"),
+        "event_type": row.get("event_type"),
+        "provider": row.get("provider"),
+        "usd_cost": row.get("usd_cost"),
+    }
+    payload = json.dumps(stable, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def _load_state(path: Path) -> set[str]:
+    if not path.exists():
+        return set()
+    try:
+        data = json.loads(path.read_text("utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return set()
+    return set(str(item) for item in data.get("sent_row_ids", []))
+
+
+def _save_state(path: Path, sent: set[str]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(
+        json.dumps({"sent_row_ids": sorted(sent)[-50_000:], "updated_at_ms": int(time.time() * 1000)}, indent=2)
+        + "\n",
+        "utf-8",
+    )
+
+
+def collect_rows(data_dir: Path, sent: set[str], limit: int) -> list[dict[str, Any]]:
+    rows: list[dict[str, Any]] = []
+    for ledger, filename in LEDGER_FILES.items():
+        for row in _read_jsonl(data_dir / filename):
+            row_id = _row_id(ledger, row)
+            if row_id in sent:
+                continue
+            rows.append({"row_id": row_id, "ledger": ledger, "row": row})
+            if len(rows) >= limit:
+                return rows
+    return rows
+
+
+def upload_batch(url: str, token: str, account_id: str, rows: list[dict[str, Any]]) -> dict[str, Any]:
+    user_id = os.environ.get("NUNCHI_USER_ID", "")
+    req = urllib.request.Request(
+        url,
+        data=json.dumps({"user_id": user_id, "account_id": account_id, "rows": rows}).encode("utf-8"),
+        headers={
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+        },
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=20) as resp:
+            return json.loads(resp.read().decode("utf-8"))
+    except urllib.error.HTTPError as exc:
+        body = exc.read().decode("utf-8", errors="replace")
+        raise RuntimeError(f"metering upload failed ({exc.code}): {body[:500]}") from exc
+
+
+def _handle_quota_status(data_dir: Path, result: dict[str, Any]) -> None:
+    quota_status = result.get("quotaStatus")
+    if not isinstance(quota_status, dict):
+        return
+    status_path = Path(os.environ.get("NUNCHI_METERING_QUOTA_STATUS_PATH") or data_dir / ".metering_quota_status.json")
+    status_path.parent.mkdir(parents=True, exist_ok=True)
+    status_path.write_text(json.dumps(quota_status, indent=2, sort_keys=True) + "\n", "utf-8")
+    action = str(quota_status.get("action") or "observe")
+    if action in {"stop", "pause"} and os.environ.get("NUNCHI_METERING_ENFORCE_RUNTIME") == "1":
+        os.kill(os.getppid(), signal.SIGTERM)
+
+
+def run_once(args: argparse.Namespace) -> int:
+    url = args.url or os.environ.get("NUNCHI_METERING_URL", "")
+    token = args.token or os.environ.get("NUNCHI_METERING_TOKEN", "")
+    account_id = args.account_id or os.environ.get("NUNCHI_ACCOUNT_ID", "")
+    if not url or not token or not account_id:
+        print("Metering disabled: NUNCHI_METERING_URL, NUNCHI_METERING_TOKEN, and NUNCHI_ACCOUNT_ID are required.")
+        return 0
+
+    data_dir = Path(args.data_dir or os.environ.get("DATA_DIR", "/data"))
+    state_path = Path(args.state_path or os.environ.get("NUNCHI_METERING_STATE_PATH") or data_dir / ".metering_upload_state.json")
+    sent = _load_state(state_path)
+    rows = collect_rows(data_dir, sent, args.batch_size)
+    if not rows:
+        print("No new metering rows.")
+        return 0
+
+    result = upload_batch(url, token, account_id, rows)
+    _handle_quota_status(data_dir, result)
+    accepted = result.get("accepted_row_ids") or [row["row_id"] for row in rows]
+    sent.update(str(row_id) for row_id in accepted)
+    _save_state(state_path, sent)
+    print(f"Uploaded {len(accepted)} metering rows to web-auth.")
+    return 0
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Upload hosted-agent metering rows")
+    parser.add_argument("--data-dir")
+    parser.add_argument("--url")
+    parser.add_argument("--token")
+    parser.add_argument("--account-id")
+    parser.add_argument("--state-path")
+    parser.add_argument("--batch-size", type=int, default=500)
+    parser.add_argument("--loop", action="store_true")
+    parser.add_argument("--interval", type=float, default=float(os.environ.get("NUNCHI_METERING_UPLOAD_INTERVAL_S", "60")))
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    if not args.loop:
+        return run_once(args)
+    while True:
+        try:
+            run_once(args)
+        except Exception as exc:
+            print(f"Metering upload error: {exc}")
+        time.sleep(max(5.0, args.interval))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/pricing_aggregate.py

@@ -97,6 +97,9 @@ def aggregate(args: argparse.Namespace) -> int:
         trades = [r for r in trade_rows if str(r.get("job_type", "unknown")) == job_type]
 
         agents = {str(r.get("agent_id", "")) for r in [*costs, *runtimes, *trades] if r.get("agent_id")}
+        users = {str(r.get("user_id", "")) for r in [*costs, *runtimes, *trades] if r.get("user_id")}
+        accounts = {str(r.get("account_id", "")) for r in [*costs, *runtimes, *trades] if r.get("account_id")}
+        subscriptions = {str(r.get("subscription_id", "")) for r in [*costs, *runtimes, *trades] if r.get("subscription_id")}
         llm_total = sum((_decimal(r.get("usd_cost")) for r in costs), Decimal("0"))
         fee_total = sum((_decimal(r.get("fee")) for r in trades), Decimal("0"))
         input_token_total = sum((_decimal(r.get("input_tokens")) for r in costs), Decimal("0"))
@@ -168,6 +171,9 @@ def aggregate(args: argparse.Namespace) -> int:
         report_rows.append({
             "job_type": job_type,
             "agent_count": len(agents),
+            "user_count": len(users),
+            "account_count": len(accounts),
+            "subscription_count": len(subscriptions),
             "duration_hours": duration_hours,
             "heartbeat_count": int(heartbeat_count),
             "llm_total": llm_total,
@@ -237,13 +243,14 @@ def _render_markdown(input_dir: Path, rows: List[dict], incidents: List[dict], a
         "",
         "## Cost By Job Type",
         "",
-        "| Job Type | Agents | Hours | Heartbeats | Linked Fills | Avg LLM/Linked Fill | Cache Hit | Cached Tokens | Cache Savings | LLM | Infra | Fees | Total | USD/Heartbeat | USD/Month | p95 Monthly COGS | Recommended |",
-        "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
+        "| Job Type | Users | Accounts | Agents | Subs | Hours | Heartbeats | Linked Fills | Avg LLM/Linked Fill | Cache Hit | Cached Tokens | Cache Savings | LLM | Infra | Fees | Total | USD/Heartbeat | USD/Month | p95 Monthly COGS | Recommended |",
+        "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
     ])
 
     for row in rows:
         lines.append(
-            f"| `{row['job_type']}` | {row['agent_count']} | {float(row['duration_hours']):.2f} | "
+            f"| `{row['job_type']}` | {row['user_count']} | {row['account_count']} | {row['agent_count']} | "
+            f"{row['subscription_count']} | {float(row['duration_hours']):.2f} | "
             f"{row['heartbeat_count']} | {row['linked_trade_count']} | {_money(row['avg_llm_per_linked_fill'])} | "
             f"{float(row['cache_hit_rate']) * 100:.1f}% | {int(row['cached_token_total'])} | {_money(row['cache_savings_total'])} | "
             f"{_money(row['llm_total'])} | {_money(row['infra_total'])} | "