diff --git a/cli/commands/trade.py b/cli/commands/trade.py index 4342191..27783fe 100644 --- a/cli/commands/trade.py +++ b/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", diff --git a/cli/engine.py b/cli/engine.py index 0bb369f..d9d7439 100644 --- a/cli/engine.py +++ b/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} diff --git a/modules/cost_metering.py b/modules/cost_metering.py index f8442f7..fce65a4 100644 --- a/modules/cost_metering.py +++ b/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, diff --git a/scripts/entrypoint.py b/scripts/entrypoint.py index 4b35be9..dff2eb7 100644 --- a/scripts/entrypoint.py +++ b/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()}, diff --git a/scripts/metering_upload.py b/scripts/metering_upload.py new file mode 100644 index 0000000..dd5faca --- /dev/null +++ b/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()) diff --git a/scripts/pricing_aggregate.py b/scripts/pricing_aggregate.py index b4608fa..3b1e216 100644 --- a/scripts/pricing_aggregate.py +++ b/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'])} | " diff --git a/tests/test_cost_metering.py b/tests/test_cost_metering.py index c8de8af..d795ab8 100644 --- a/tests/test_cost_metering.py +++ b/tests/test_cost_metering.py @@ -23,7 +23,7 @@ def test_cost_meter_writes_cost_and_route_ledgers(tmp_path): meter = CostMeter( context=context, data_dir=str(tmp_path), - strategy="claude_agent", + strategy="ai_agent", pricing=StaticPricing(), ) @@ -36,7 +36,7 @@ def test_cost_meter_writes_cost_and_route_ledgers(tmp_path): output_tokens=5, tick_index=7, elapsed_ms=123.4, - decision_call_id="claude_agent:run-1:tick-7", + decision_call_id="ai_agent:run-1:tick-7", ) cost_row = json.loads((tmp_path / "cost_ledger.jsonl").read_text().strip()) @@ -45,13 +45,71 @@ def test_cost_meter_writes_cost_and_route_ledgers(tmp_path): assert cost_row["experiment_id"] == "exp-1" assert cost_row["job_type"] == "taker" assert cost_row["tick_index"] == 7 - assert cost_row["decision_call_id"] == "claude_agent:run-1:tick-7" + assert cost_row["decision_call_id"] == "ai_agent:run-1:tick-7" assert cost_row["usd_cost"] == "0.020" assert route_row["requested_route"] == "openrouter/fusion" - assert route_row["decision_call_id"] == "claude_agent:run-1:tick-7" + assert route_row["decision_call_id"] == "ai_agent:run-1:tick-7" assert route_row["resolved_model"] == "anthropic/claude-haiku" +def test_cost_meter_records_hosted_identity_fields(tmp_path): + context = ExperimentContext( + experiment_id="exp-1", + run_id="run-1", + agent_id="agent-1", + job_type="taker", + user_id="user-1", + account_id="account-1", + plan_id="hosted-agent-standard", + subscription_id="sub-1", + billing_period_start="1000", + billing_period_end="2000", + ) + meter = CostMeter( + context=context, + data_dir=str(tmp_path), + strategy="ai_agent", + pricing=StaticPricing(), + ) + + meter.record_llm_call( + provider="openrouter", + requested_model="openrouter/auto", + resolved_model="openai/gpt-4o-mini", + route="openrouter/auto", + input_tokens=1, + output_tokens=1, + tick_index=1, + elapsed_ms=1, + ) + + cost_row = json.loads((tmp_path / "cost_ledger.jsonl").read_text().strip()) + route_row = json.loads((tmp_path / "route_ledger.jsonl").read_text().strip()) + + assert cost_row["user_id"] == "user-1" + assert cost_row["account_id"] == "account-1" + assert cost_row["plan_id"] == "hosted-agent-standard" + assert cost_row["subscription_id"] == "sub-1" + assert cost_row["billing_period_start"] == "1000" + assert cost_row["billing_period_end"] == "2000" + assert route_row["user_id"] == "user-1" + assert route_row["account_id"] == "account-1" + + +def test_experiment_context_from_env_enables_hosted_metering(monkeypatch): + monkeypatch.delenv("NUNCHI_EXPERIMENT_ID", raising=False) + monkeypatch.setenv("NUNCHI_USER_ID", "user-1") + monkeypatch.setenv("NUNCHI_ACCOUNT_ID", "account-1") + monkeypatch.setenv("NUNCHI_AGENT_ID", "agent-1") + + context = ExperimentContext.from_env("ai_agent") + + assert context.enabled is True + assert context.experiment_id == "hosted-agent" + assert context.ledger_fields()["user_id"] == "user-1" + assert context.ledger_fields()["account_id"] == "account-1" + + def test_cost_meter_prefers_actual_openrouter_cost(tmp_path): context = ExperimentContext( experiment_id="exp-1", @@ -62,7 +120,7 @@ def test_cost_meter_prefers_actual_openrouter_cost(tmp_path): meter = CostMeter( context=context, data_dir=str(tmp_path), - strategy="claude_agent", + strategy="ai_agent", pricing=StaticPricing(input_price="0", output_price="0"), ) @@ -96,7 +154,7 @@ def test_cost_meter_records_openrouter_route_metadata(tmp_path): meter = CostMeter( context=context, data_dir=str(tmp_path), - strategy="claude_agent", + strategy="ai_agent", pricing=StaticPricing(), ) @@ -131,7 +189,7 @@ def test_cost_meter_records_cache_metrics(tmp_path): meter = CostMeter( context=context, data_dir=str(tmp_path), - strategy="claude_agent", + strategy="ai_agent", pricing=StaticPricing(), ) @@ -166,6 +224,6 @@ def test_cost_meter_records_cache_metrics(tmp_path): def test_experiment_context_disabled_without_experiment_id(monkeypatch): monkeypatch.delenv("NUNCHI_EXPERIMENT_ID", raising=False) - context = ExperimentContext.from_env("claude_agent") + context = ExperimentContext.from_env("ai_agent") assert context.enabled is False - assert context.agent_id == "claude_agent" + assert context.agent_id == "ai_agent" diff --git a/tests/test_metering_upload.py b/tests/test_metering_upload.py new file mode 100644 index 0000000..0e45e5e --- /dev/null +++ b/tests/test_metering_upload.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +import json + +from scripts import metering_upload + + +def test_collect_rows_adds_stable_ids_and_skips_sent(tmp_path): + row = { + "experiment_id": "exp", + "run_id": "run", + "agent_id": "agent", + "tick_index": 1, + "usd_cost": "0.01", + "provider": "openrouter", + } + (tmp_path / "cost_ledger.jsonl").write_text(json.dumps(row) + "\n") + + rows = metering_upload.collect_rows(tmp_path, sent=set(), limit=10) + assert len(rows) == 1 + assert rows[0]["ledger"] == "cost" + assert rows[0]["row"] == row + assert len(rows[0]["row_id"]) == 64 + + skipped = metering_upload.collect_rows(tmp_path, sent={rows[0]["row_id"]}, limit=10) + assert skipped == [] + + +def test_state_round_trip(tmp_path): + state_path = tmp_path / "state.json" + metering_upload._save_state(state_path, {"b", "a"}) + + assert metering_upload._load_state(state_path) == {"a", "b"} + + +def test_handle_quota_status_writes_status_file(tmp_path, monkeypatch): + monkeypatch.delenv("NUNCHI_METERING_ENFORCE_RUNTIME", raising=False) + result = {"quotaStatus": {"status": "soft_cap", "action": "observe"}} + + metering_upload._handle_quota_status(tmp_path, result) + + status = json.loads((tmp_path / ".metering_quota_status.json").read_text()) + assert status["status"] == "soft_cap" + assert status["action"] == "observe"