diff --git a/cloud/server.py b/cloud/server.py index 88e62bd..6f7d08c 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -62,6 +62,7 @@ from cloud.auth import AuthStore from cloud.coordinator import parse_parts as _coord_parse_parts from cloud.coordinator import validate_parts as _coord_validate_parts from cloud.cost import CostStore, unpriced_token_cap +from cloud.usage import QuotaGate, UsageMeter, disk_usage_bytes from cloud.db import connect as _db_connect from cloud.db import create_database as _create_database from cloud.db import init_postgres_schema as _init_postgres_schema @@ -515,6 +516,52 @@ def _task_worktree_for_files(server, task): return None, (404, "no local worktree for this task") +def _measure_user_storage_bytes(server, user_id): + """Per-user storage measurement (P11): bytes, measured, total. + + Sums each of the user's tasks' local worktrees (via + ``_task_worktree_for_files``) plus the user's task attachment + sizes from the DB. Cloud / cloud-hypervisor tasks have no local + worktree and are not measurable on this host: they count in + ``total`` but not in ``measured``, so callers can say so + honestly instead of pretending the number is complete. + """ + conn = _connect(server.db_path) + conn.row_factory = sqlite3.Row + tasks = [] + try: + rows = conn.execute( + "SELECT id, project_id, executor, worktree_host" + " FROM tasks WHERE created_by = ?", + (user_id,)).fetchall() + tasks = [dict(row) for row in rows] + finally: + conn.close() + total = len(tasks) + measured = 0 + size = 0 + for task in tasks: + worktree, _err = _task_worktree_for_files(server, task) + if worktree is None: + continue + measured += 1 + size += disk_usage_bytes(worktree) + try: + conn = _connect(server.db_path) + try: + row = conn.execute( + "SELECT COALESCE(SUM(a.size), 0) FROM task_attachments a" + " JOIN tasks t ON a.task_id = t.id" + " WHERE t.created_by = ?", + (user_id,)).fetchone() + size += int(row[0] or 0) + finally: + conn.close() + except Exception: + pass + return size, measured, total + + def _safe_worktree_path(worktree, rel): """worktree 内相对路径的安全解析：空串返回 worktree 本身； 绝对路径、.. 逃逸、符号链接逃逸一律返回 None。realpath 后 @@ -2294,6 +2341,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._send_json(404, {"error": "not found"}) elif path == "/users/me/quota" and method == "GET": self._user_quota_self(user_id) + elif path == "/users/me/usage" and method == "GET": + self._user_usage_self(user_id) elif path == "/users/me/profile" and method == "GET": self._identity_profile_get(user_id) elif path == "/users/me/profile" and method in ("PUT", "POST"): @@ -2336,6 +2385,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): self._admin_quota_get(user_id, query) elif path == "/admin/quotas" and method == "POST": self._admin_quota_set(user_id) + elif path == "/admin/tiers" and method == "GET": + self._admin_tiers_get(user_id, query) + elif path == "/admin/tiers" and method == "POST": + self._admin_tiers_set(user_id) elif path == "/admin/push/dispatch" and method == "POST": self._push_dispatch_endpoint(user_id) elif path.startswith("/tasks/"): @@ -8326,6 +8379,16 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): "unpriced_token_cap": cap, "unpriced_tokens_used": used}) return True + try: + denied = self._tier_gate(quota_user_id or user_id) + except Exception: + self._send_json(503, {"error": "quota_check_unavailable", + "detail": "tier quota lookup failed; " + "no task was created or queued"}) + return True + if denied is not None: + self._send_json(402, denied) + return True return False def _reject_if_over_quota_for_task(self, caller_id, task): @@ -8417,6 +8480,126 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): payload["unpriced_token_cap"] = unpriced_token_cap() self._send_json(200, payload) + def _tier_running_tasks(self, user_id): + """该用户正在跑的任务数（queued / running），档位门用。""" + conn = self._db() + try: + row = conn.execute( + "SELECT COUNT(*) FROM tasks WHERE created_by = ?" + " AND status IN ('queued', 'running')", + (user_id,)).fetchone() + return int(row[0] or 0) + finally: + conn.close() + + def _tier_meter(self, user_id): + """接好两个实测器的 UsageMeter；存储数只测一次。 + + 存储实测（磁盘 walk）比较贵：这里先测一次，把字节数与 + 任务计数都交给 meter 的注入函数与 payload，避免一次 + 请求走两遍磁盘。 + """ + measured = {} + + def _storage(uid): + if uid not in measured: + size, n_measured, n_total = ( + _measure_user_storage_bytes(self.server, uid)) + measured[uid] = (size, n_measured, n_total) + return measured[uid][0] + + meter = UsageMeter( + self.server.db_path, + running_task_count=self._tier_running_tasks, + storage_bytes=_storage) + return meter, measured + + def _tier_gate(self, user_id): + """档位准入：无档位行放行；有则查 token/并发/存储三门。 + + 返回 None 放行，否则否决 dict（由调用方回 402）。 + """ + meter, _measured = self._tier_meter(user_id) + return QuotaGate(meter).check_task_admission(user_id) + + def _user_usage_self(self, user_id): + """GET /users/me/usage：当期用量、剩余额度、提醒、账本校验。 + + USD 配额沿用既有 _quota_status 口径并排返回；四项档位 + 用量走 UsageMeter。未定档用户 limits 为 None，只看到 + 用量，看不到档位剩余额度——定档是管理员显式操作。 + """ + meter, measured = self._tier_meter(user_id) + payload = meter.summary(user_id) + quota, used, month = self._quota_status(user_id) + payload["quota_usd_micros"] = quota + payload["used_usd_micros"] = used + payload["unpriced_tokens_used"] = self._quota_unpriced_tokens( + user_id, month) + payload["unpriced_token_cap"] = unpriced_token_cap() + _size, n_measured, n_total = measured.get( + user_id, (payload["used"]["storage_bytes"], 0, 0)) + payload["storage_measured_tasks"] = n_measured + payload["storage_total_tasks"] = n_total + self._send_json(200, payload) + + def _admin_tiers_get(self, user_id, query): + if not self._is_global_admin(user_id): + self._audit_admin(user_id, "admin_tiers_denied", + "get tiers") + self._send_json(403, {"error": "forbidden"}) + return + target = (query.get("user_id") or [None])[0] + if not target or not isinstance(target, str): + self._send_json(400, {"error": "user_id required"}) + return + if not self._user_exists(target): + self._send_json(404, {"error": "user not found"}) + return + meter = UsageMeter(self.server.db_path) + self._send_json(200, { + "proposal": True, + "user_id": target, + "tier": meter.tiers.get_tier(target), + "limits": meter.tiers.limits_for(target)}) + + def _admin_tiers_set(self, user_id): + if not self._is_global_admin(user_id): + self._audit_admin(user_id, "admin_tiers_denied", + "set tiers") + self._send_json(403, {"error": "forbidden"}) + return + body, err = self._read_json() + if err: + self._send_json(400, {"error": err}) + return + target = body.get("user_id") + if not target or not isinstance(target, str): + self._send_json(400, {"error": "user_id required"}) + return + if not self._user_exists(target): + self._send_json(404, {"error": "user not found"}) + return + tier = body.get("tier") + meter = UsageMeter(self.server.db_path) + if tier is None: + meter.tiers.clear_tier(target) + elif not isinstance(tier, str): + self._send_json(400, {"error": "tier must be a string"}) + return + else: + try: + meter.tiers.set_tier(target, tier) + except ValueError: + self._send_json(400, { + "error": "unknown tier (want free, paid, heavy)"}) + return + self._send_json(200, { + "proposal": True, + "user_id": target, + "tier": meter.tiers.get_tier(target), + "limits": meter.tiers.limits_for(target)}) + def _admin_quota_get(self, user_id, query): if not self._is_global_admin(user_id): self._audit_admin(user_id, "admin_quota_denied", diff --git a/cloud/tests/test_usage_quota.py b/cloud/tests/test_usage_quota.py new file mode 100644 index 0000000..91eea31 --- /dev/null +++ b/cloud/tests/test_usage_quota.py @@ -0,0 +1,499 @@ +"""P11 用量计量与配额执行：档位门、计量口径、账本、服务端点。 + +不跑真实模型、不联网：本地临时 SQLite + 本地临时端口。 +档位数值钉住提案（Free $0 / Paid $19 / Heavy $99，未确认定价）， +token 门只计 actual、estimated 只展示不拦人，账本哈希链防篡改， +准入判定不碰在跑任务。 +""" + +import http.client +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +try: + from cloud.auth import AuthStore + from cloud.cost import CostStore + from cloud.server import create_server + from cloud.usage import (GB, TIER_LIMITS, QuotaGate, TierStore, + UsageLedger, UsageMeter, disk_usage_bytes, + utc_day) +except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.cost import CostStore + from cloud.server import create_server + from cloud.usage import (GB, TIER_LIMITS, QuotaGate, TierStore, + UsageLedger, UsageMeter, disk_usage_bytes, + utc_day) + + +def _make_db(tmp): + """临时库文件 + 最小 tasks 表（monthly_tokens 要 JOIN 它）。""" + db_path = os.path.join(tmp, "usage.db") + conn = sqlite3.connect(db_path) + conn.execute("CREATE TABLE IF NOT EXISTS tasks (" + "id TEXT PRIMARY KEY, created_by TEXT, status TEXT)") + conn.commit() + conn.close() + return db_path + + +def _seed_tokens(db_path, user_id, actual=0, estimated=0): + """给 user 记一笔 actual 与一笔 estimated（同一任务行）。""" + conn = sqlite3.connect(db_path) + conn.execute("INSERT OR IGNORE INTO tasks (id, created_by, status)" + " VALUES ('task-1', ?, 'completed')", (user_id,)) + conn.commit() + conn.close() + store = CostStore(db_path) + if actual: + store.record("task-1", "proj-1", "gpt-6.1-terra", + input_tokens=actual, output_tokens=0, + source="actual") + if estimated: + store.record("task-1", "proj-1", "gpt-6.1-terra", + input_tokens=estimated, output_tokens=0, + source="estimated") + + +class TierProposalTestCase(unittest.TestCase): + def test_proposal_numbers_pinned(self): + self.assertEqual(TIER_LIMITS["free"]["price_usd_per_month"], 0) + self.assertEqual(TIER_LIMITS["free"]["messages_per_day"], 50) + self.assertEqual(TIER_LIMITS["free"]["tokens_per_month"], 200000) + self.assertEqual(TIER_LIMITS["free"]["concurrent_tasks"], 1) + self.assertEqual(TIER_LIMITS["free"]["storage_bytes"], 1 * GB) + self.assertEqual(TIER_LIMITS["paid"]["price_usd_per_month"], 19) + self.assertEqual(TIER_LIMITS["paid"]["messages_per_day"], 500) + self.assertEqual(TIER_LIMITS["paid"]["tokens_per_month"], 2000000) + self.assertEqual(TIER_LIMITS["paid"]["concurrent_tasks"], 5) + self.assertEqual(TIER_LIMITS["paid"]["storage_bytes"], 10 * GB) + self.assertEqual(TIER_LIMITS["heavy"]["price_usd_per_month"], 99) + self.assertIsNone(TIER_LIMITS["heavy"]["messages_per_day"]) + self.assertEqual(TIER_LIMITS["heavy"]["tokens_per_month"], + 20000000) + self.assertEqual(TIER_LIMITS["heavy"]["concurrent_tasks"], 20) + self.assertEqual(TIER_LIMITS["heavy"]["storage_bytes"], 100 * GB) + + def test_unknown_tier_rejected_and_clear(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + store = TierStore(db_path) + self.assertIsNone(store.get_tier("u1")) + self.assertIsNone(store.limits_for("u1")) + with self.assertRaises(ValueError): + store.set_tier("u1", "platinum") + store.set_tier("u1", "paid") + self.assertEqual(store.get_tier("u1"), "paid") + self.assertEqual(store.limits_for("u1")["tokens_per_month"], + 2000000) + store.clear_tier("u1") + self.assertIsNone(store.get_tier("u1")) + + +class TokenMeteringTestCase(unittest.TestCase): + def test_actual_estimated_split_and_gate_counts_actual_only(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + meter = UsageMeter(db_path) + gate = QuotaGate(meter) + meter.tiers.set_tier("u1", "free") + _seed_tokens(db_path, "u1", actual=1000, estimated=500) + tokens = meter.monthly_tokens("u1") + self.assertEqual(tokens["actual"], 1000) + self.assertEqual(tokens["estimated"], 500) + self.assertEqual(tokens["total"], 1500) + # estimated 199500 + actual 1000：门只看 actual，不拦 + _seed_tokens(db_path, "u1", actual=0, estimated=199500) + self.assertIsNone(gate.check_message_admission("u1")) + # actual 到 200000：拦，维度 tokens + _seed_tokens(db_path, "u1", actual=199000, estimated=0) + denied = gate.check_message_admission("u1") + self.assertIsNotNone(denied) + self.assertEqual(denied["error"], "quota_exceeded") + self.assertEqual(denied["dimension"], "tokens") + self.assertEqual(denied["limit"], 200000) + self.assertEqual(denied["used"], 200000) + + def test_unassigned_user_never_tier_gated(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + meter = UsageMeter(db_path) + gate = QuotaGate(meter) + _seed_tokens(db_path, "u1", actual=99999999) + for _ in range(60): + meter.record_message("u1") + self.assertIsNone(gate.check_message_admission("u1")) + self.assertIsNone(gate.check_task_admission( + "u1", running_tasks=99, storage_bytes=10 ** 12)) + + +class MessageGateTestCase(unittest.TestCase): + def test_daily_fixed_window_and_limit(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + meter = UsageMeter(db_path) + gate = QuotaGate(meter) + meter.tiers.set_tier("u1", "free") + for _ in range(49): + meter.record_message("u1") + self.assertEqual(meter.daily_messages("u1"), 49) + self.assertIsNone(gate.check_message_admission("u1")) + meter.record_message("u1") + denied = gate.check_message_admission("u1") + self.assertEqual(denied["dimension"], "messages") + self.assertEqual(denied["limit"], 50) + self.assertEqual(denied["used"], 50) + # 其它日期的消息不计入今天（用固定过去日期，避免 UTC 日界） + meter.ledger.record("u1", "message", 1, day="2026-01-01") + self.assertEqual(meter.daily_messages("u1"), 50) + self.assertEqual(meter.daily_messages("u1", day="2026-01-01"), + 1) + + +class ConcurrencyStorageGateTestCase(unittest.TestCase): + def test_concurrency_gate(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + meter = UsageMeter(db_path) + gate = QuotaGate(meter) + meter.tiers.set_tier("u1", "free") + self.assertIsNone(gate.check_task_admission( + "u1", running_tasks=0, storage_bytes=0)) + denied = gate.check_task_admission( + "u1", running_tasks=1, storage_bytes=0) + self.assertEqual(denied["dimension"], "concurrent_tasks") + self.assertEqual(denied["limit"], 1) + self.assertIn("wait for a running task", denied["detail"]) + + def test_storage_gate_with_cleanup_guidance(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + meter = UsageMeter(db_path) + gate = QuotaGate(meter) + meter.tiers.set_tier("u1", "free") + limit = TIER_LIMITS["free"]["storage_bytes"] + self.assertIsNone(gate.check_task_admission( + "u1", running_tasks=0, storage_bytes=limit - 1)) + denied = gate.check_task_admission( + "u1", running_tasks=0, storage_bytes=limit) + self.assertEqual(denied["dimension"], "storage") + self.assertIn("free space", denied["detail"]) + self.assertIn("worktrees", denied["detail"]) + + def test_token_denial_precedes_concurrency(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + meter = UsageMeter(db_path) + gate = QuotaGate(meter) + meter.tiers.set_tier("u1", "free") + _seed_tokens(db_path, "u1", actual=200000) + denied = gate.check_task_admission( + "u1", running_tasks=99, storage_bytes=10 ** 12) + self.assertEqual(denied["dimension"], "tokens") + + def test_unwired_measurer_raises_not_silent_pass(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + meter = UsageMeter(db_path) + gate = QuotaGate(meter) + meter.tiers.set_tier("u1", "free") + with self.assertRaises(ValueError): + gate.check_task_admission("u1") + + def test_gate_is_pure_no_ledger_writes(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + meter = UsageMeter(db_path) + gate = QuotaGate(meter) + meter.tiers.set_tier("u1", "free") + meter.record_message("u1") + before = meter.ledger.sum_kind("u1", "message") + gate.check_message_admission("u1") + gate.check_task_admission("u1", running_tasks=5, + storage_bytes=10 ** 12) + after = meter.ledger.sum_kind("u1", "message") + self.assertEqual(before, after) + + +class DiskUsageTestCase(unittest.TestCase): + def test_sums_files_skips_missing_and_links(self): + with tempfile.TemporaryDirectory() as tmp: + self.assertEqual(disk_usage_bytes( + os.path.join(tmp, "nope")), 0) + sub = os.path.join(tmp, "sub") + os.makedirs(sub) + with open(os.path.join(tmp, "a.txt"), "wb") as f: + f.write(b"x" * 100) + with open(os.path.join(sub, "b.txt"), "wb") as f: + f.write(b"y" * 50) + os.symlink(os.path.join(tmp, "a.txt"), + os.path.join(tmp, "link.txt")) + self.assertEqual(disk_usage_bytes(tmp), 150) + + +class LedgerTestCase(unittest.TestCase): + def test_chain_verifies_and_tamper_detected(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + ledger = UsageLedger(db_path) + ids = [ledger.record("u1", "message", 1) + for _ in range(3)] + self.assertEqual(ids, sorted(ids)) + ok, bad = ledger.verify_chain("u1") + self.assertTrue(ok) + self.assertIsNone(bad) + conn = sqlite3.connect(db_path) + conn.execute("UPDATE usage_ledger SET amount = 999" + " WHERE id = ?", (ids[1],)) + conn.commit() + conn.close() + ok, bad = ledger.verify_chain("u1") + self.assertFalse(ok) + self.assertEqual(bad, ids[1]) + + def test_prev_hash_tamper_detected(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + ledger = UsageLedger(db_path) + first = ledger.record("u1", "message", 1) + second = ledger.record("u1", "message", 1) + conn = sqlite3.connect(db_path) + conn.execute("UPDATE usage_ledger SET prev_hash = 'x'" + " WHERE id = ?", (second,)) + conn.commit() + conn.close() + ok, bad = ledger.verify_chain("u1") + self.assertFalse(ok) + self.assertEqual(bad, second) + + def test_unknown_kind_rejected_and_no_secrets_field(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + ledger = UsageLedger(db_path) + with self.assertRaises(ValueError): + ledger.record("u1", "chat_text", 1) + row_id = ledger.record("u1", "message", 1, task_id="t1") + conn = sqlite3.connect(db_path) + cols = [r[1] for r in conn.execute( + "PRAGMA table_info(usage_ledger)").fetchall()] + conn.close() + self.assertNotIn("content", cols) + self.assertNotIn("token", cols) + self.assertGreater(row_id, 0) + + +class WarningTestCase(unittest.TestCase): + def test_once_per_window_at_80_percent(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + meter = UsageMeter(db_path) + meter.tiers.set_tier("u1", "free") + for _ in range(39): + meter.record_message("u1") + self.assertEqual(meter.summary("u1")["warnings"], []) + meter.record_message("u1") + summary = meter.summary("u1") + self.assertEqual(len(summary["warnings"]), 1) + self.assertEqual(summary["warnings"][0]["dimension"], + "messages") + # 同一窗口第二次 summary：提醒仍在，但账本只记一次 + meter.summary("u1") + self.assertEqual( + meter.ledger.sum_kind("u1", "warning"), 1) + + def test_warn_ratio_env_override(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + meter = UsageMeter(db_path) + meter.tiers.set_tier("u1", "free") + for _ in range(25): + meter.record_message("u1") + self.assertEqual(meter.summary("u1")["warnings"], []) + old = os.environ.get("CLOUD_USAGE_WARN_RATIO") + os.environ["CLOUD_USAGE_WARN_RATIO"] = "0.5" + try: + warnings = meter.summary("u1")["warnings"] + finally: + if old is None: + os.environ.pop("CLOUD_USAGE_WARN_RATIO", None) + else: + os.environ["CLOUD_USAGE_WARN_RATIO"] = old + self.assertEqual(len(warnings), 1) + + +class SummaryTestCase(unittest.TestCase): + def test_summary_shape_and_remaining(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + meter = UsageMeter( + db_path, running_task_count=lambda uid: 0, + storage_bytes=lambda uid: 123) + meter.tiers.set_tier("u1", "paid") + _seed_tokens(db_path, "u1", actual=1000, estimated=200) + meter.record_message("u1") + summary = meter.summary("u1") + self.assertTrue(summary["proposal"]) + self.assertEqual(summary["tier"], "paid") + self.assertTrue(summary["ledger_verified"]) + self.assertIsNone(summary["ledger_first_bad_id"]) + self.assertEqual(summary["used"]["tokens_actual"], 1000) + self.assertEqual(summary["used"]["tokens_estimated"], 200) + self.assertEqual(summary["used"]["messages"], 1) + self.assertEqual(summary["remaining"]["tokens"], 1999000) + self.assertEqual(summary["remaining"]["messages"], 499) + self.assertEqual(summary["remaining"]["storage_bytes"], + 10 * GB - 123) + + def test_summary_unassigned_user(self): + with tempfile.TemporaryDirectory() as tmp: + db_path = _make_db(tmp) + meter = UsageMeter(db_path) + summary = meter.summary("u1") + self.assertIsNone(summary["tier"]) + self.assertIsNone(summary["limits"]) + self.assertIsNone(summary["remaining"]["tokens"]) + + +class UsageServerTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = self.auth.create_user("owner") + self.outsider_id, self.outsider_key = self.auth.create_user( + "outsider") + self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS") + os.environ["CODEX_GLOBAL_ADMINS"] = self.owner_id + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + status, self.project = self._req( + "POST", "/projects", {"name": "demo"}, self.owner_key) + self.assertEqual(status, 201) + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + try: + self.auth.conn.close() + except Exception: + pass + if self._old_admins is None: + os.environ.pop("CODEX_GLOBAL_ADMINS", None) + else: + os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins + self.tmp.cleanup() + + def _req(self, method, path, body=None, key=None): + conn = http.client.HTTPConnection( + "127.0.0.1", self.port, timeout=10) + headers = {} + if key: + headers["X-API-Key"] = key + raw = json.dumps(body) if body is not None else None + if raw is not None: + headers["Content-Type"] = "application/json" + try: + conn.request(method, path, body=raw, headers=headers) + resp = conn.getresponse() + data = resp.read() + finally: + conn.close() + try: + parsed = json.loads(data.decode("utf-8")) if data else None + except ValueError: + parsed = None + return resp.status, parsed + + def _set_tier(self, tier, key=None): + return self._req("POST", "/admin/tiers", + {"user_id": self.owner_id, "tier": tier}, + key or self.owner_key) + + def test_usage_endpoint_unassigned(self): + status, payload = self._req("GET", "/users/me/usage", + key=self.owner_key) + self.assertEqual(status, 200, payload) + self.assertTrue(payload["proposal"]) + self.assertIsNone(payload["tier"]) + self.assertIsNone(payload["limits"]) + self.assertTrue(payload["ledger_verified"]) + self.assertEqual(payload["used"]["tokens_actual"], 0) + + def test_admin_tiers_set_get_clear_and_auth(self): + status, payload = self._set_tier("free") + self.assertEqual(status, 200, payload) + self.assertEqual(payload["tier"], "free") + self.assertEqual(payload["limits"]["messages_per_day"], 50) + self.assertTrue(payload["proposal"]) + status, payload = self._req( + "GET", "/admin/tiers?user_id=%s" % self.owner_id, + key=self.owner_key) + self.assertEqual(status, 200, payload) + self.assertEqual(payload["tier"], "free") + status, _ = self._set_tier("platinum") + self.assertEqual(status, 400) + status, _ = self._set_tier("free", key=self.outsider_key) + self.assertEqual(status, 403) + status, _ = self._req( + "GET", "/admin/tiers?user_id=%s" % self.owner_id, + key=self.outsider_key) + self.assertEqual(status, 403) + status, payload = self._set_tier(None) + self.assertEqual(status, 200, payload) + self.assertIsNone(payload["tier"]) + + def test_concurrency_gate_blocks_second_task(self): + status, _ = self._set_tier("free") + self.assertEqual(status, 200) + status, task = self._req( + "POST", "/tasks", + {"project_id": self.project["id"], "prompt": "做事"}, + self.owner_key) + self.assertEqual(status, 201, task) + status, denied = self._req( + "POST", "/tasks", + {"project_id": self.project["id"], "prompt": "再做事"}, + self.owner_key) + self.assertEqual(status, 402, denied) + self.assertEqual(denied["error"], "quota_exceeded") + self.assertEqual(denied["dimension"], "concurrent_tasks") + # 清档后放行 + status, _ = self._set_tier(None) + self.assertEqual(status, 200) + status, task2 = self._req( + "POST", "/tasks", + {"project_id": self.project["id"], "prompt": "再做事"}, + self.owner_key) + self.assertEqual(status, 201, task2) + + def test_usage_endpoint_tiered_reflects_running_task(self): + self._set_tier("free") + status, _ = self._req( + "POST", "/tasks", + {"project_id": self.project["id"], "prompt": "做事"}, + self.owner_key) + self.assertEqual(status, 201) + status, payload = self._req("GET", "/users/me/usage", + key=self.owner_key) + self.assertEqual(status, 200, payload) + self.assertEqual(payload["used"]["concurrent_tasks"], 1) + self.assertEqual(payload["remaining"]["concurrent_tasks"], 0) + self.assertEqual(payload["storage_total_tasks"], 1) + self.assertEqual(payload["storage_measured_tasks"], 0) + dims = [w["dimension"] for w in payload["warnings"]] + self.assertIn("concurrent_tasks", dims) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/usage.py b/cloud/usage.py new file mode 100644 index 0000000..882a07e --- /dev/null +++ b/cloud/usage.py @@ -0,0 +1,631 @@ +"""Usage metering and tier quota enforcement (P11). + +Four metered dimensions: tokens per month, messages per day, +concurrent tasks, storage bytes. Tokens are read from the existing +``cost_records`` ledger (``cloud/cost.py``) -- the inference channel's +bookkeeping point -- never re-collected on a second path. Messages, +task starts/ends, storage snapshots and warnings are recorded in the +hash-chained ``usage_ledger`` table in this module. + +Tier limits and prices are a PROJECT PROPOSAL, not user-confirmed +pricing and not Meta official pricing: the user delegated the numbers +("set them by market", 2026-11-06) and never confirmed them. Every +payload this module returns carries ``proposal: True``. Real billing +(Stripe / payments / invoicing) is NOT in this module and must be +approved separately; nothing here charges money. + +Admission semantics: gates are admission-only. A denial returns a +``quota_exceeded`` dict (HTTP 402 at the server layer); there is no +API here that terminates an in-flight turn. The pre-existing USD +quota path in ``cloud/server.py`` still cuts in-flight work at turn +boundaries via worker polling -- that is a known, documented +discrepancy (see docs/chv-muse-p11-gap.md §5), not something this +module silently resolves. + +Token gate counts ``source="actual"`` rows only, matching the +``cost.py`` P1 decision ("estimated rows are never real spend and +must not trip the cap"); estimated tokens are reported separately +and labelled as estimates. +""" + +import hashlib +import json +import os +import sqlite3 +import time +from datetime import datetime, timezone + +from cloud.cost import CostStore +from cloud.db import connect as _db_connect + +# --------------------------------------------------------------------------- +# Tier proposal (see module docstring: PROPOSAL, not confirmed pricing). +# --------------------------------------------------------------------------- + +GB = 1024 * 1024 * 1024 + +TIER_NAMES = ("free", "paid", "heavy") + +TIER_LIMITS = { + "free": { + "label": "Free", + "price_usd_per_month": 0, + "messages_per_day": 50, + "tokens_per_month": 200000, + "concurrent_tasks": 1, + "storage_bytes": 1 * GB, + }, + "paid": { + "label": "Paid", + "price_usd_per_month": 19, + "messages_per_day": 500, + "tokens_per_month": 2000000, + "concurrent_tasks": 5, + "storage_bytes": 10 * GB, + }, + "heavy": { + "label": "Heavy", + "price_usd_per_month": 99, + # The proposal never gave Heavy a messages/day number; + # None = no daily message cap (project decision, not proposal). + "messages_per_day": None, + "tokens_per_month": 20000000, + "concurrent_tasks": 20, + "storage_bytes": 100 * GB, + }, +} + +_LEDGER_KINDS = ("message", "task_start", "task_end", "storage", + "warning", "usage_query") + + +def warn_ratio() -> float: + """Warning threshold as a fraction of a limit (default 0.8). + + Env ``CLOUD_USAGE_WARN_RATIO`` overrides; unparsable or + out-of-range values fall back to 0.8. + """ + try: + value = float(os.environ.get("CLOUD_USAGE_WARN_RATIO", "0.8")) + except (TypeError, ValueError): + return 0.8 + if value <= 0 or value > 1: + return 0.8 + return value + + +def utc_day(now: float | None = None) -> str: + """UTC calendar day (YYYY-MM-DD) for ``now`` (default: current).""" + stamp = now if now is not None else time.time() + return datetime.fromtimestamp(stamp, tz=timezone.utc).strftime("%Y-%m-%d") + + +def utc_month(now: float | None = None) -> str: + """UTC calendar month (YYYY-MM) for ``now`` (default: current).""" + stamp = now if now is not None else time.time() + return datetime.fromtimestamp(stamp, tz=timezone.utc).strftime("%Y-%m") + + +def disk_usage_bytes(path: str) -> int: + """Sum of file sizes under ``path`` (best effort, bytes). + + Does not follow symlinks; unreadable files/dirs are skipped. + Missing path counts as 0. + """ + total = 0 + if not path or not os.path.exists(path): + return 0 + if os.path.isfile(path) and not os.path.islink(path): + try: + return os.path.getsize(path) + except OSError: + return 0 + for root, dirs, files in os.walk(path, followlinks=False): + for name in files: + full = os.path.join(root, name) + try: + if not os.path.islink(full): + total += os.path.getsize(full) + except OSError: + continue + return total + + +def _entry_hash(prev_hash: str | None, fields: dict) -> str: + """Hash-chain digest over the previous hash + canonical fields.""" + payload = json.dumps({"prev": prev_hash or "", "fields": fields}, + sort_keys=True, separators=(",", ":")) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +class TierStore: + """Per-user tier assignment (``user_tiers`` table). + + No row means the user is NOT tier-gated: only the pre-existing + USD quota applies. Assigning a tier is an explicit admin action; + nothing here defaults a user to Free. + """ + + _DDL = """CREATE TABLE IF NOT EXISTS user_tiers ( + user_id TEXT PRIMARY KEY, + tier TEXT, + updated_at DOUBLE PRECISION + )""" + + def __init__(self, db_path: str) -> None: + self.db_path = db_path + self._conn = _db_connect(db_path) + self._conn.execute(self._DDL) + self._conn.commit() + + def get_tier(self, user_id: str) -> str | None: + """Tier name for ``user_id``, or None when unassigned.""" + row = self._conn.execute( + "SELECT tier FROM user_tiers WHERE user_id = ?", + (user_id,)).fetchone() + if row is None: + return None + return row[0] + + def set_tier(self, user_id: str, tier: str) -> None: + """Assign ``tier`` to ``user_id``; unknown tier names rejected.""" + if tier not in TIER_NAMES: + raise ValueError("unknown tier %r (want one of %r)" + % (tier, TIER_NAMES)) + self._conn.execute( + "INSERT INTO user_tiers (user_id, tier, updated_at)" + " VALUES (?, ?, ?) ON CONFLICT(user_id) DO UPDATE" + " SET tier = excluded.tier," + " updated_at = excluded.updated_at", + (user_id, tier, time.time())) + self._conn.commit() + + def clear_tier(self, user_id: str) -> None: + """Remove the tier row: the user is tier-gated no more.""" + self._conn.execute( + "DELETE FROM user_tiers WHERE user_id = ?", (user_id,)) + self._conn.commit() + + def limits_for(self, user_id: str) -> dict | None: + """Tier limits dict for ``user_id``, or None when unassigned.""" + tier = self.get_tier(user_id) + if tier is None: + return None + return dict(TIER_LIMITS[tier]) + + +class UsageLedger: + """Append-only, hash-chained usage ledger (``usage_ledger``). + + One chain per user: each row stores the previous row's hash and + its own digest over the previous hash + canonical fields. Any + UPDATE/DELETE breaks ``verify_chain``. The ledger never stores + conversation text, file contents, or credential values -- only + counts, ids, and hashes. + """ + + _DDL = """CREATE TABLE IF NOT EXISTS usage_ledger ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id TEXT, + kind TEXT, + amount INTEGER, + unit TEXT, + task_id TEXT, + model TEXT, + source TEXT, + day TEXT, + created_at REAL, + prev_hash TEXT, + entry_hash TEXT + )""" + + def __init__(self, db_path: str) -> None: + self.db_path = db_path + if db_path != ":memory:": + parent = os.path.dirname(os.path.abspath(db_path)) + os.makedirs(parent, exist_ok=True) + self._conn = _db_connect(db_path) + self._conn.execute(self._DDL) + self._conn.commit() + + def _new_conn(self) -> sqlite3.Connection: + """Fresh connection per write/read (mirrors CostStore). + + ``:memory:`` reuses the shared connection; anything else + opens a new one so concurrent threads never share a cursor. + """ + if self.db_path == ":memory:": + return self._conn + conn = _db_connect(self.db_path) + conn.execute(self._DDL) + conn.commit() + return conn + + def record(self, user_id: str, kind: str, amount: int, + unit: str = "", task_id: str | None = None, + model: str | None = None, source: str | None = None, + day: str | None = None) -> int: + """Append one ledger event; returns its row id. + + ``kind`` must be one of the known event kinds; ``amount`` is + a count (1 per message, bytes for storage snapshots). + """ + if kind not in _LEDGER_KINDS: + raise ValueError("unknown ledger kind %r" % (kind,)) + day = day or utc_day() + created_at = time.time() + conn = self._new_conn() + try: + prev = conn.execute( + "SELECT entry_hash FROM usage_ledger" + " WHERE user_id = ? ORDER BY id DESC LIMIT 1", + (user_id,)).fetchone() + prev_hash = prev[0] if prev is not None else None + fields = {"user_id": user_id, "kind": kind, + "amount": int(amount), "unit": unit, + "task_id": task_id or "", "model": model or "", + "source": source or "", "day": day, + "created_at": created_at} + digest = _entry_hash(prev_hash, fields) + cursor = conn.execute( + """INSERT INTO usage_ledger + (user_id, kind, amount, unit, task_id, model, + source, day, created_at, prev_hash, entry_hash) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (user_id, kind, int(amount), unit, task_id, model, + source, day, created_at, prev_hash, digest)) + conn.commit() + return int(cursor.lastrowid or 0) + finally: + if conn is not self._conn: + conn.close() + + def sum_kind(self, user_id: str, kind: str, + day: str | None = None, + month: str | None = None) -> int: + """Sum ``amount`` for ``user_id``/``kind`` in a day or month.""" + conn = self._new_conn() + try: + if day is not None: + row = conn.execute( + "SELECT COALESCE(SUM(amount), 0) FROM usage_ledger" + " WHERE user_id = ? AND kind = ? AND day = ?", + (user_id, kind, day)).fetchone() + elif month is not None: + row = conn.execute( + "SELECT COALESCE(SUM(amount), 0) FROM usage_ledger" + " WHERE user_id = ? AND kind = ? AND day LIKE ?", + (user_id, kind, month + "%")).fetchone() + else: + row = conn.execute( + "SELECT COALESCE(SUM(amount), 0) FROM usage_ledger" + " WHERE user_id = ? AND kind = ?", + (user_id, kind)).fetchone() + return int(row[0] or 0) + finally: + if conn is not self._conn: + conn.close() + + def verify_chain(self, user_id: str) -> tuple[bool, int | None]: + """Recompute the hash chain; (True, None) or (False, bad id).""" + conn = self._new_conn() + try: + rows = conn.execute( + "SELECT id, user_id, kind, amount, unit, task_id," + " model, source, day, created_at, prev_hash," + " entry_hash FROM usage_ledger" + " WHERE user_id = ? ORDER BY id", + (user_id,)).fetchall() + finally: + if conn is not self._conn: + conn.close() + prev_hash: str | None = None + for row in rows: + fields = {"user_id": row[1], "kind": row[2], + "amount": int(row[3]), "unit": row[4], + "task_id": row[5] or "", "model": row[6] or "", + "source": row[7] or "", "day": row[8], + "created_at": row[9]} + if row[10] != prev_hash: + return False, int(row[0]) + if _entry_hash(prev_hash, fields) != row[11]: + return False, int(row[0]) + prev_hash = row[11] + return True, None + + def warning_recorded(self, user_id: str, dimension: str, + window: str) -> bool: + """True when a warning for (dimension, window) is on record.""" + conn = self._new_conn() + try: + row = conn.execute( + "SELECT COUNT(*) FROM usage_ledger" + " WHERE user_id = ? AND kind = 'warning'" + " AND unit = ? AND day = ?", + (user_id, dimension, window)).fetchone() + return int(row[0] or 0) > 0 + finally: + if conn is not self._conn: + conn.close() + + def record_warning(self, user_id: str, dimension: str, + window: str) -> int: + """Record a once-per-window warning event; returns row id.""" + return self.record(user_id, "warning", 1, unit=dimension, + day=window) + + +class UsageMeter: + """Four-dimension usage metering for one database. + + Tokens come from ``cost_records`` (the inference channel's + bookkeeping point, joined to ``tasks`` exactly like + ``_quota_status`` does); messages come from the ledger; + concurrency and storage come from injected measurers so the + meter stays unit-testable without a server or a disk layout. + An unwired measurer reports ``None`` (unmeasured) in summaries + and raises in gates -- never silently passes as zero. + """ + + def __init__(self, db_path: str, + running_task_count=None, + storage_bytes=None) -> None: + self.db_path = db_path + self.tiers = TierStore(db_path) + self.ledger = UsageLedger(db_path) + self._running_task_count = running_task_count + self._storage_bytes = storage_bytes + + def monthly_tokens(self, user_id: str, + month: str | None = None) -> dict: + """Actual / estimated / total tokens for the UTC month. + + ``actual`` is provider-reported usage; ``estimated`` is the + character-count fallback, labelled as an estimate. Only + ``actual`` ever trips a gate. + """ + month = month or utc_month() + _cost_store = CostStore(self.db_path) + try: + _cost_store._conn.close() + except Exception: + pass + conn = _db_connect(self.db_path) + try: + rows = conn.execute( + "SELECT c.source," + " COALESCE(SUM(c.input_tokens + c.output_tokens), 0)" + " FROM cost_records c JOIN tasks t ON c.task_id = t.id" + " WHERE t.created_by = ? AND c.day LIKE ?" + " GROUP BY c.source", + (user_id, month + "%")).fetchall() + finally: + conn.close() + totals = {"actual": 0, "estimated": 0} + for source, total in rows: + if source in totals: + totals[source] = int(total or 0) + totals["total"] = totals["actual"] + totals["estimated"] + return totals + + def daily_messages(self, user_id: str, + day: str | None = None) -> int: + """Messages recorded for the UTC day (fixed window).""" + return self.ledger.sum_kind(user_id, "message", + day=day or utc_day()) + + def record_message(self, user_id: str, + task_id: str | None = None) -> int: + """Record one message event; returns the ledger row id.""" + return self.ledger.record(user_id, "message", 1, + unit="message", task_id=task_id) + + def record_task_start(self, user_id: str, + task_id: str | None = None) -> int: + """Record a task-start audit event; returns row id.""" + return self.ledger.record(user_id, "task_start", 1, + unit="task", task_id=task_id) + + def record_task_end(self, user_id: str, + task_id: str | None = None) -> int: + """Record a task-end audit event; returns row id.""" + return self.ledger.record(user_id, "task_end", 1, + unit="task", task_id=task_id) + + def record_storage_snapshot(self, user_id: str, + size_bytes: int) -> int: + """Record a measured storage snapshot; returns row id.""" + return self.ledger.record(user_id, "storage", int(size_bytes), + unit="bytes") + + def running_tasks(self, user_id: str) -> int | None: + """Concurrent running tasks, or None when unwired.""" + if self._running_task_count is None: + return None + return int(self._running_task_count(user_id)) + + def storage_used(self, user_id: str) -> int | None: + """Storage bytes, or None when unwired.""" + if self._storage_bytes is None: + return None + return int(self._storage_bytes(user_id)) + + def summary(self, user_id: str) -> dict: + """Current-period usage, limits, remaining, warnings. + + Warnings are recorded once per dimension per window (month + for tokens, day for the rest) and included in the payload. + """ + month = utc_month() + day = utc_day() + limits = self.tiers.limits_for(user_id) + tier = self.tiers.get_tier(user_id) + tokens = self.monthly_tokens(user_id, month) + messages = self.daily_messages(user_id, day) + running = self.running_tasks(user_id) + storage = self.storage_used(user_id) + ledger_ok, ledger_bad = self.ledger.verify_chain(user_id) + used = {"tokens_actual": tokens["actual"], + "tokens_estimated": tokens["estimated"], + "messages": messages, + "concurrent_tasks": running, + "storage_bytes": storage} + payload: dict = { + "proposal": True, + "user_id": user_id, + "tier": tier, + "month": month, + "day": day, + "used": used, + "limits": limits, + "remaining": _remaining(limits, used), + "ledger_verified": ledger_ok, + "ledger_first_bad_id": ledger_bad, + "warnings": [], + } + if limits is not None: + payload["warnings"] = self._check_warnings( + user_id, limits, used, month, day) + return payload + + def _check_warnings(self, user_id: str, limits: dict, + used: dict, month: str, day: str) -> list: + """80%-of-limit warnings, once per dimension per window.""" + ratio = warn_ratio() + warnings = [] + checks = (("tokens", limits.get("tokens_per_month"), + used["tokens_actual"], month), + ("messages", limits.get("messages_per_day"), + used["messages"], day), + ("concurrent_tasks", limits.get("concurrent_tasks"), + used["concurrent_tasks"], day), + ("storage", limits.get("storage_bytes"), + used["storage_bytes"], day)) + for dimension, limit, value, window in checks: + if limit is None or value is None: + continue + if limit > 0 and value >= limit * ratio: + if not self.ledger.warning_recorded( + user_id, dimension, window): + self.ledger.record_warning(user_id, dimension, + window) + warnings.append({"dimension": dimension, + "limit": limit, "used": value, + "ratio": ratio}) + return warnings + + +def _remaining(limits: dict | None, used: dict) -> dict: + """Remaining quota per dimension (None = unlimited/unmeasured).""" + if limits is None: + return {"tokens": None, "messages": None, + "concurrent_tasks": None, "storage_bytes": None} + pairs = (("tokens", limits.get("tokens_per_month"), + used["tokens_actual"]), + ("messages", limits.get("messages_per_day"), + used["messages"]), + ("concurrent_tasks", limits.get("concurrent_tasks"), + used["concurrent_tasks"]), + ("storage_bytes", limits.get("storage_bytes"), + used["storage_bytes"])) + remaining: dict[str, int | None] = {} + for name, limit, value in pairs: + if limit is None or value is None: + remaining[name] = None + else: + remaining[name] = max(0, limit - value) + return remaining + + +def _denial(dimension: str, limit: int, used: int, + detail: str) -> dict: + """One quota denial payload (HTTP 402 at the server layer).""" + return {"error": "quota_exceeded", "dimension": dimension, + "limit": limit, "used": used, "detail": detail} + + +class QuotaGate: + """Admission-only tier gates (pure decisions, no side effects). + + Every method returns None (allow) or a denial dict. Nothing here + touches a running task: an in-flight turn is never cut off by + these gates; a denial only blocks the NEXT admission. + """ + + def __init__(self, meter: UsageMeter) -> None: + self.meter = meter + + def check_message_admission(self, user_id: str) -> dict | None: + """Gate a new message: messages/day, then tokens/month.""" + limits = self.meter.tiers.limits_for(user_id) + if limits is None: + return None + day_limit = limits.get("messages_per_day") + if day_limit is not None: + used = self.meter.daily_messages(user_id) + if used >= day_limit: + return _denial( + "messages", day_limit, used, + "daily message quota exceeded for tier %r" + " (used %d of %d); wait until tomorrow (UTC)" + " or upgrade the tier" + % (self.meter.tiers.get_tier(user_id), used, + day_limit)) + return self._check_tokens(user_id, limits) + + def check_task_admission( + self, user_id: str, running_tasks: int | None = None, + storage_bytes: int | None = None) -> dict | None: + """Gate a new task: tokens, then concurrency, then storage.""" + limits = self.meter.tiers.limits_for(user_id) + if limits is None: + return None + denied = self._check_tokens(user_id, limits) + if denied is not None: + return denied + running = running_tasks + if running is None: + running = self.meter.running_tasks(user_id) + if running is None: + raise ValueError("no running-task measurer wired") + task_limit = limits.get("concurrent_tasks") + if task_limit is not None and running >= task_limit: + return _denial( + "concurrent_tasks", task_limit, running, + "concurrent task quota exceeded for tier %r" + " (running %d of %d); wait for a running task" + " to finish" % (self.meter.tiers.get_tier(user_id), + running, task_limit)) + storage = storage_bytes + if storage is None: + storage = self.meter.storage_used(user_id) + if storage is None: + raise ValueError("no storage measurer wired") + storage_limit = limits.get("storage_bytes") + if storage_limit is not None and storage >= storage_limit: + return _denial( + "storage", storage_limit, storage, + "storage quota exceeded for tier %r" + " (used %d of %d bytes); free space by removing" + " old task worktrees or snapshots, then retry" + % (self.meter.tiers.get_tier(user_id), storage, + storage_limit)) + return None + + def _check_tokens(self, user_id: str, + limits: dict) -> dict | None: + """Monthly actual-token gate (estimated tokens never gate).""" + token_limit = limits.get("tokens_per_month") + if token_limit is None: + return None + used = self.meter.monthly_tokens(user_id)["actual"] + if used >= token_limit: + return _denial( + "tokens", token_limit, used, + "monthly token quota exceeded for tier %r" + " (used %d of %d actual tokens; estimated tokens" + " are not counted); wait until next month (UTC)" + " or upgrade the tier" + % (self.meter.tiers.get_tier(user_id), used, + token_limit)) + return None diff --git a/docs/chv-muse-p11-gap.md b/docs/chv-muse-p11-gap.md new file mode 100644 index 0000000..cb925e6 --- /dev/null +++ b/docs/chv-muse-p11-gap.md @@ -0,0 +1,138 @@ +# 对齐 Muse P11 差距表：用量计量与配额执行 + +依据：`~/workspace/research_notes/meta-muse-ai-architecture-20261101-0755/report.md` +（官方帮助中心 / 新闻室 2026-09-08 为准；第三方转述只作参考，不得把 +第三方转述的价格写成官方定价）。基线：`06c3e17`。先写表再动手。 + +## §1 要求（官方 + 本项目提案） + +| # | 要求 | 来源 | +|---|---|---| +| 1 | 免费但有使用上限；达到上限可升级付费订阅，或等免费额度刷新 | 官方帮助中心（报告 §1.5/§6） | +| 2 | 免费额度具体数字、付费档价格未出现在首发新闻稿中；Power/Maximum 等价格是第三方转述的帮助中心内容，**不得写成官方定价** | 报告 §1.5、§5 表 | +| 3 | 三档档位与价格沿用本项目提案（用户 2026-11-06 委托“按市场定”，**未确认为定价**）：Free $0 / Paid $19/月 / Heavy $99/月 | 本项目提案，见 §2 | +| 4 | 四项计量：token 数、消息数、并发任务数、存储量 | 本轨任务书 + 设计文档 §6.2 | +| 5 | 超限回 402；进行中回合不掐断，跑完才拦下一回合 | 本轨任务书 + 设计文档 §6.2 | +| 6 | 用量展示：用户可查当期用量与剩余额度；接近上限提醒一次（阈值默认 80%，本项目自定） | 本轨任务书 | +| 7 | 账本追加写、不可篡改；账本里绝不含对话原文、文件内容、密钥 | 本轨任务书 | +| 8 | 不接真实计费、不碰真实钱；计费对接（Stripe 等）待用户另行批准 | 本轨任务书 | + +档位提案数值（**提案，非用户确认定价，非官方定价**）： + +| 档位 | 价格 | 消息/天 | tokens/月 | 并发任务 | 存储 | +|---|---|---|---|---|---| +| Free | $0 | 50 | 200,000 | 1 | 1 GB | +| Paid | $19/月 | 500 | 2,000,000 | 5 | 10 GB | +| Heavy | $99/月 | 不限（见 §5 第 4 条） | 20,000,000 | 20 | 100 GB | + +## §2 现状（基线 06c3e17 已有，可复用） + +- `cloud/cost.py:CostStore`：`cost_records(task_id, project_id, model, + tokens, usd_micros, source, day)`，`actual`（模型真实用量事件）与 + `estimated`（字符估算）**永不混算**；价格表是 OpenAI 挂牌价，注释明言 + “不是 relay 的账单”；`report_by_task/project/day`，**无 `report_by_user`**。 +- `cloud/db.py`：`user_quotas(user_id PK, quota_usd_micros, updated_at)`。 +- `cloud/server.py`：`_quota_status`（USD 月度配额，UTC 自然月）、 + `_check_quota` / `_reject_if_over_quota`（超限回 402 `quota_exceeded`， + 查询异常 fail-closed 回 503 `quota_check_unavailable`）、 + `_check_unpriced_token_cap`（无价 token 上限，默认 20M/月，可配）、 + `GET /users/me/quota`、`GET/POST /admin/quotas`。 + 准入点是建任务 / 分支 / 重试 / 续跑 / goal-resume，均为权限校验通过后、 + 写库之前；`_reject_if_over_quota_for_task` 双人门（触发人 + 原任务创建人）。 +- 全仓 grep 确认：`cost_records` **没有任何 UPDATE / DELETE 路径**， + token 账本已是追加写；本轨不再为 token 另建一套账本（见 §3 第 2 条）。 + +缺失：消息数计量、并发任务门、存储门、三档档位表、用量展示端点、 +80% 提醒、账本防篡改校验、`report_by_user`。 + +## §3 改动清单（本轨） + +1. `cloud/usage.py`（新，标准库 + `cloud.db` / `cloud.cost` 的常量口径， + 不引 server）： + - `TIER_LIMITS`：三档提案数值，模块 docstring 与每个返回 payload + 都标注 `proposal: true`（提案，非定价）。 + - `TierStore`：`user_tiers(user_id PK, tier, updated_at)`，`get_tier` / + `set_tier`（非法档位名拒绝）；**无档位行的用户不受档位门约束**， + 只受既有 USD 配额约束——定档是管理员显式操作，不给存量用户 + 默认套 Free（否则上线即拦住所有人）。 + - `UsageLedger`：`usage_ledger` 追加写哈希链账本（`prev_hash` + + `entry_hash = sha256(规范 JSON)`），`record` / `sum_kind` / + `verify_chain`（篡改返回首个坏行 id）。记消息、任务起止、存储 + 快照、提醒事件；**token 事件不重复记**（见第 2 条）。 + - `UsageMeter`：月度 token（读 `cost_records` JOIN `tasks`，与 + `_quota_status` 同款查询，actual / estimated 分开列）、当日消息 + （读账本）、并发（注入 `running_task_count`，状态 `queued` / + `running`）、存储（注入 `storage_bytes`，磁盘实测）。 + `summary(user_id)`：四项 used / limit / remaining + 提醒 + 账本 + 校验结果。 + - `QuotaGate`：`check_task_admission`（token + 并发 + 存储）、 + `check_message_admission`（消息/天 + token/月）。返回 None 放行， + 否则否决 dict（`error: quota_exceeded` + `dimension` + `limit` / + `used` + 可读 `detail`）；**纯准入判定，无任何终止在跑任务的 + API**（进行中回合不掐断，见 §5 第 1 条）。 + - `disk_usage_bytes(path)`：递归求和，不跟随符号链接，读错跳过 + （尽力而为，调用方在 payload 里标明实测范围）。 + - 提醒阈值默认 0.8，环境变量 `CLOUD_USAGE_WARN_RATIO` 可配； + 每维度每窗口只记一次提醒（账本 `kind="warning"` 去重）。 +2. token 计量口径：以模型方返回为准（`source="actual"`）；无返回时 + 字符估算（`source="estimated"`，payload 标注为估算）。**档位门只计 + actual**——与 `cost.py` 的 P1 决定一致（“Estimated rows are never + real spend and must not trip the cap”）；estimated 只展示、不拦人。 + 但 payload 同时列出 estimated 数，让用户看得见估算占比。 +3. `cloud/server.py` 接线（只动配额漏斗，不新增调用点）： + - `_reject_if_over_quota` 内、USD 门与无价 cap 通过后，加档位门 + （同一 `quota_user_id`）；档位查询异常同样 fail-closed 回 503。 + `_reject_if_over_quota_for_task` 不动（它已委托给前者，双人门 + 自动覆盖）。 + - `GET /users/me/usage`（`_user_usage_self`）：当期用量与剩余额度 + + USD 配额合并返回 + 提醒 + 账本校验结果。 + - `GET/POST /admin/tiers`（`_admin_tiers_get/set`，`_is_global_admin` + 同款鉴权）：管理员定档 / 查档。 + - 存储实测：复用 `_task_worktree_for_files` 逐任务解析本机 worktree + 求和；cloud / cloud-hypervisor 任务不在本机，量不到，计 0 并在 + payload 里标明 `storage_measured_tasks` / `storage_total_tasks`。 +4. `cloud/tests/test_usage_quota.py`（新）：档位数值钉住提案、非法档位 + 拒绝、无档位行不限档位门、token actual/estimated 口径、消息固定 + UTC 日窗口、并发门、存储门与清理指引、账本哈希链与篡改发现、提醒 + 只发一次、准入纯判定不碰在跑任务、服务端点（`/users/me/usage`、 + `/admin/tiers`、档位门 402）。 +5. `docs/chv-production-readiness.md`：补 P11 一节。 + +## §4 真机待验证清单（未做，排队在 round-11 与 P4～P9 之后） + +- 真实 token 回传下的月度累计与 402（actual 行累计、无价模型走 + 既有 cap 不走美元门）。 +- 并发门在真实多任务下（建第 N+1 个任务被拒，在跑任务不受影响）。 +- 存储门在真实 worktree 下（本机任务目录求和、cloud 任务不计入）。 +- 80% 提醒在真实使用下只发一次、跨窗口重置。 +- 账本在真实写入压力下的链完整性（并发写不丢链不断）。 + +## §5 诚实边界（本轨内不解决，写清不冒充） + +1. **进行中回合的语义不一致（已知待决）**：设计文档 §6.2 与本轨 + 任务书说“进行中的回合不中途掐断”；但基线 `_check_quota` 的 + docstring 与 worker 实现是另一套——Local popen 与 Docker 路径的 + worker 轮询每 30 秒查一次 USD 配额，超限杀进程组/容器（掐断粒度 + 是轮次边界 + 30 秒检查）。本轨新增的四项档位门**只做准入**， + 不改既有 USD 路径的轮询掐断；两处语义不一致是已知待决项， + 不在本轨内悄悄统一，须由父级另行裁决。 +2. **并发超限的处理与设计文档不同**：设计文档 §6.2 说并发超限 + “新任务排队，不立即拒绝”；本轨任务书（父级指令）说“即拒并 + 给出可读原因”。本轨按任务书即拒，差异记在此。 +3. **存储计量不完整**：cloud / cloud-hypervisor 任务的 worktree + 不在本机，量不到；本机也只能按任务 worktree 求和，用户在 + VM 内或其它路径的占用不在计量内。payload 标明实测范围， + 不冒充全量。 +4. **Heavy 消息/天提案未给**：提案原文只给了 Heavy 的 tokens / + 并发 / 存储，未给消息/天。本轨定为不限，并标注为**本项目决定**， + 不是提案原文、不是官方口径。 +5. **窗口口径**：token 按 UTC 自然月（与既有 USD 配额一致）； + 消息按 UTC 自然日（固定窗口，非滑动）；并发与存储是瞬时值。 + 固定窗口边界存在少量跨窗毛刺，文档写清，不冒充滑动窗口。 +6. **价格与计费**：档位价格是助手提案（用户委托按市场定，未确认）， + 不是用户确认定价，更不是 Meta 官方定价；payload 与文档都带 + `proposal` 标注。真实计费（Stripe/支付/开票/充值）不在本轨， + 待用户另行批准；任何把“记账”等同于“收费”的说法都是错的。 +7. **token 价格**：USD 换算沿用 `cost.py` 的 OpenAI 挂牌价表， + 注释明言不是 relay 账单；无价模型（无价格表项）的 token 只 + 计量、不折算美元，仍走既有无价 token cap。 diff --git a/docs/chv-production-readiness.md b/docs/chv-production-readiness.md index bb4e6f2..f8be4a1 100644 --- a/docs/chv-production-readiness.md +++ b/docs/chv-production-readiness.md @@ -833,3 +833,26 @@ UNSTABLE、未合并、未部署、真机 0 项。差：外部为主。 - 动态安全 2.0 → 2.0，未达 10：本轨无任何动态验证， V-7 未跑。差：外部（KVM 宿主 + 审计结论）。 + +## 15. 对齐 Muse P11：用量计量与配额执行（分支 feat-chv-muse-p11） + +- 四项计量：token/月（读既有 `cost_records`，推理通道记账点，不另开 + 采集路）、消息/天（哈希链账本）、并发任务（`queued`/`running` 计数）、 + 存储（本机任务 worktree 实测 + 附件大小）。 +- 三档档位是**提案**（Free $0 / Paid $19/月 / Heavy $99/月，用户委托按 + 市场定，未确认；不是 Meta 官方定价），payload 与文档都带 `proposal` + 标注。无档位行的用户不受档位门约束，只受既有 USD 配额约束。 +- 超限回 402（`quota_exceeded` + `dimension`）；进行中回合不掐断—— + 档位门是纯准入判定，无任何终止在跑任务的 API。 +- 80% 提醒（默认，可配），每维度每窗口只记一次。 +- 账本 `usage_ledger` 追加写哈希链，篡改可发现；账本不含对话原文、 + 文件内容、密钥。token 账本复用 `cost_records`（全仓无 UPDATE/DELETE）， + 不重复记。 +- 端点：`GET /users/me/usage`、`GET/POST /admin/tiers`（全局管理员）。 +- 不接真实计费、不碰真实钱；Stripe/支付/开票待用户另行批准。 +- 已知待决（见 `docs/chv-muse-p11-gap.md` §5）：既有 USD 配额路径的 + worker 30 秒轮询掐断与“进行中回合不掐断”语义不一致；并发超限按 + 任务书即拒，与设计文档 §6.2 的“排队”不同；cloud 任务存储量不到； + Heavy 消息/天提案未给，本轨定为不限。 +- 真机待验证：真实 token 回传下的月度累计与 402、真实多任务下的并发门、 + 真实 worktree 下的存储门、提醒只发一次、账本并发写不断链。