feat(admin): P4-04..P4-06 admin API for users, VIP/quota, TTS and audit

services/api/app/main.py

@@ -21,6 +21,7 @@ from .schemas import (
     AuthResponse,
     LoginRequest,
     MembershipRequest,
+    MembershipRevokeRequest,
     PasswordChangeRequest,
     QuotaAdjustmentRequest,
     RegisterRequest,
@@ -479,10 +480,84 @@ def download_tts_audio(task_id: UUID, user: dict = Depends(current_user), connec
     return audio_response(task_id, user, connection, True)
 
 
+USER_COLUMNS = "id, email, phone, role, plan, status, email_verified, phone_verified, created_at, last_login_at, updated_at"
+
+
+def quota_view_for(quota: dict) -> dict[str, Any]:
+    available = max(0, quota["limit_snapshot"] + quota["adjustment"] - quota["used"] - quota["reserved"])
+    return {
+        "period_start": quota["period_start"], "period_end": quota["period_end"], "limit": quota["limit_snapshot"],
+        "adjustment": quota["adjustment"], "used": quota["used"], "reserved": quota["reserved"], "available": available,
+    }
+
+
+def active_grant(connection: Connection, user_id: UUID) -> dict | None:
+    return connection.execute(
+        """
+        SELECT id, plan, starts_at, expires_at, reason, created_at
+        FROM membership_grants
+        WHERE user_id = %s AND revoked_at IS NULL AND starts_at <= now()
+          AND (expires_at IS NULL OR expires_at > now())
+        ORDER BY starts_at DESC LIMIT 1
+        """,
+        (user_id,),
+    ).fetchone()
+
+
 @app.get("/api/v1/admin/users")
-def admin_users(user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
+def admin_users(email: str | None = None, phone: str | None = None, status: str | None = None, page: int = 1, limit: int = 20, user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
-    rows = connection.execute("SELECT id, email, phone, role, plan, status, email_verified, phone_verified, created_at, last_login_at FROM users ORDER BY created_at DESC LIMIT 100").fetchall()
+    page = max(1, page)
-    return {"items": rows}
+    limit = min(100, max(1, limit))
+    clauses = []
+    params: list[Any] = []
+    if email:
+        clauses.append("email ILIKE %s")
+        params.append(f"%{email}%")
+    if phone:
+        clauses.append("phone ILIKE %s")
+        params.append(f"%{phone}%")
+    if status in ("active", "disabled"):
+        clauses.append("status = %s")
+        params.append(status)
+    where = " WHERE " + " AND ".join(clauses) if clauses else ""
+    total = connection.execute(f"SELECT count(*) AS c FROM users{where}", params).fetchone()["c"]
+    rows = connection.execute(
+        f"SELECT {USER_COLUMNS} FROM users{where} ORDER BY created_at DESC LIMIT %s OFFSET %s",
+        [*params, limit, (page - 1) * limit],
+    ).fetchall()
+    items = []
+    for row in rows:
+        plan = effective_plan(connection, row)
+        item = dict(row)
+        item["effective_plan"] = plan
+        items.append(item)
+    return {"items": items, "total": total, "page": page, "limit": limit, "pages": (total + limit - 1) // limit}
+
+
+@app.get("/api/v1/admin/users/{user_id}")
+def admin_user_detail(user_id: UUID, user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
+    target = connection.execute(f"SELECT {USER_COLUMNS} FROM users WHERE id = %s", (user_id,)).fetchone()
+    if not target:
+        raise error("NOT_FOUND", "用户不存在", 404)
+    plan = effective_plan(connection, target)
+    quota = ensure_quota(connection, user_id, plan)
+    grant = active_grant(connection, user_id)
+    grant_history = connection.execute("SELECT id, plan, starts_at, expires_at, revoked_at, reason, created_at FROM membership_grants WHERE user_id = %s ORDER BY created_at DESC", (user_id,)).fetchall()
+    adjustment_records = connection.execute(
+        """
+        SELECT u.id, u.amount, u.idempotency_key, u.created_at, a.reason, a.actor_id
+        FROM usage_records u
+        LEFT JOIN admin_audit_logs a ON a.action = 'quota_adjustment' AND a.target_id = %s
+        WHERE u.user_id = %s AND u.type = 'adjust'
+        ORDER BY u.created_at DESC
+        """,
+        (user_id, user_id),
+    ).fetchall()
+    connection.commit()
+    return {
+        "user": dict(target), "effective_plan": plan, "effective_grant": grant,
+        "quota": quota_view_for(quota), "membership_history": grant_history, "adjustment_records": adjustment_records,
+    }
 
 
 @app.patch("/api/v1/admin/users/{user_id}/status", dependencies=[Depends(require_csrf)])
@@ -492,12 +567,41 @@ def admin_status(user_id: UUID, payload: StatusRequest, connection: Connection =
         raise error("NOT_FOUND", "用户不存在", 404)
     if target["role"] == "admin" and payload.status == "disabled":
         count = connection.execute("SELECT count(*) FROM users WHERE role = 'admin' AND status = 'active'").fetchone()["count"]
-        if count <= 1:
+        if count <= 1 and target["status"] == "active":
             raise error("LAST_ADMIN_PROTECTED", "不能禁用最后一个可用管理员", 409)
     updated = connection.execute("UPDATE users SET status = %s, updated_at = now() WHERE id = %s RETURNING id, status", (payload.status, user_id)).fetchone()
+    if payload.status == "disabled":
+        revoked = connection.execute("UPDATE sessions SET revoked_at = now() WHERE user_id = %s AND revoked_at IS NULL", (user_id,))
+        revoked_count = revoked.rowcount
+    else:
+        revoked_count = 0
     connection.execute("INSERT INTO admin_audit_logs(actor_id, action, target_type, target_id, before_value, after_value, reason) VALUES (%s, 'user_status', 'user', %s, %s, %s, %s)", (actor["id"], user_id, Json({"status": target["status"]}), Json({"status": updated["status"]}), payload.reason))
     connection.commit()
-    return updated
+    return {"id": updated["id"], "status": updated["status"], "sessions_revoked": revoked_count}
+
+
+@app.get("/api/v1/admin/users/{user_id}/membership")
+def admin_membership_view(user_id: UUID, user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
+    target = connection.execute("SELECT id, role, plan FROM users WHERE id = %s", (user_id,)).fetchone()
+    if not target:
+        raise error("NOT_FOUND", "用户不存在", 404)
+    connection.commit()
+    return {"user_id": user_id, "effective_plan": effective_plan(connection, target), "effective_grant": active_grant(connection, user_id)}
+
+
+@app.post("/api/v1/admin/users/{user_id}/membership/revoke", dependencies=[Depends(require_csrf)])
+def admin_membership_revoke(user_id: UUID, payload: MembershipRevokeRequest, connection: Connection = Depends(get_connection), actor: dict = Depends(require_admin)):
+    target = connection.execute("SELECT * FROM users WHERE id = %s", (user_id,)).fetchone()
+    if not target:
+        raise error("NOT_FOUND", "用户不存在", 404)
+    grant = connection.execute("SELECT id, plan, revoked_at FROM membership_grants WHERE user_id = %s AND revoked_at IS NULL AND starts_at <= now() AND (expires_at IS NULL OR expires_at > now()) ORDER BY starts_at DESC LIMIT 1", (user_id,)).fetchone()
+    if not grant:
+        raise error("NO_ACTIVE_GRANT", "没有可撤销的有效会员", 409)
+    connection.execute("UPDATE membership_grants SET revoked_at = now() WHERE id = %s", (grant["id"],))
+    connection.execute("UPDATE users SET plan = 'free', updated_at = now() WHERE id = %s", (user_id,))
+    connection.execute("INSERT INTO admin_audit_logs(actor_id, action, target_type, target_id, before_value, after_value, reason) VALUES (%s, 'membership_revoke', 'user', %s, %s, %s, %s)", (actor["id"], user_id, Json({"plan": grant["plan"]}), Json({"plan": "free"}), payload.reason))
+    connection.commit()
+    return {"status": "ok", "revoked_plan": grant["plan"]}
 
 
 @app.put("/api/v1/admin/users/{user_id}/membership", dependencies=[Depends(require_csrf)])
@@ -506,25 +610,130 @@ def admin_membership(user_id: UUID, payload: MembershipRequest, connection: Conn
     if not target:
         raise error("NOT_FOUND", "用户不存在", 404)
     starts_at = payload.starts_at or utc_now()
+    if payload.expires_at and payload.expires_at <= starts_at:
+        raise error("INVALID_PERIOD", "过期时间必须在开始时间之后", 422)
     connection.execute("UPDATE membership_grants SET revoked_at = now() WHERE user_id = %s AND revoked_at IS NULL", (user_id,))
     grant = connection.execute("INSERT INTO membership_grants(user_id, plan, starts_at, expires_at, created_by, reason) VALUES (%s, %s, %s, %s, %s, %s) RETURNING id, plan, starts_at, expires_at", (user_id, payload.plan, starts_at, payload.expires_at, actor["id"], payload.reason)).fetchone()
-    connection.execute("UPDATE users SET plan = %s, updated_at = now() WHERE id = %s", (payload.plan, user_id))
+    stored_plan = "free" if payload.plan == "free" else payload.plan
+    connection.execute("UPDATE users SET plan = %s, updated_at = now() WHERE id = %s", (stored_plan, user_id))
     connection.execute("INSERT INTO admin_audit_logs(actor_id, action, target_type, target_id, before_value, after_value, reason) VALUES (%s, 'membership', 'user', %s, %s, %s, %s)", (actor["id"], user_id, Json({"plan": target["plan"]}), Json({"plan": payload.plan, "expires_at": payload.expires_at.isoformat() if payload.expires_at else None}), payload.reason))
     connection.commit()
     return grant
 
 
+@app.get("/api/v1/admin/users/{user_id}/quota")
+def admin_quota_view(user_id: UUID, user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
+    target = connection.execute("SELECT id, role, plan FROM users WHERE id = %s", (user_id,)).fetchone()
+    if not target:
+        raise error("NOT_FOUND", "用户不存在", 404)
+    quota = ensure_quota(connection, user_id, effective_plan(connection, target))
+    adjustment_records = connection.execute(
+        """
+        SELECT u.id, u.amount, u.idempotency_key, u.created_at, a.reason, a.actor_id
+        FROM usage_records u
+        LEFT JOIN admin_audit_logs a ON a.action = 'quota_adjustment' AND a.target_id = %s
+        WHERE u.user_id = %s AND u.type = 'adjust'
+        ORDER BY u.created_at DESC
+        """,
+        (user_id, user_id),
+    ).fetchall()
+    connection.commit()
+    return {"quota": quota_view_for(quota), "adjustment_records": adjustment_records}
+
+
 @app.post("/api/v1/admin/users/{user_id}/quota-adjustments", dependencies=[Depends(require_csrf)])
 def admin_quota(user_id: UUID, payload: QuotaAdjustmentRequest, connection: Connection = Depends(get_connection), actor: dict = Depends(require_admin)):
     target = connection.execute("SELECT * FROM users WHERE id = %s", (user_id,)).fetchone()
     if not target:
         raise error("NOT_FOUND", "用户不存在", 404)
     quota = ensure_quota(connection, user_id, effective_plan(connection, target))
-    existing = connection.execute("SELECT id FROM usage_records WHERE user_id = %s AND idempotency_key = %s", (user_id, payload.idempotency_key)).fetchone()
+    existing = connection.execute("SELECT id FROM usage_records WHERE user_id = %s AND idempotency_key = %s AND type = 'adjust'", (user_id, payload.idempotency_key)).fetchone()
     if existing:
         raise error("IDEMPOTENCY_CONFLICT", "该调整已提交", 409)
+    projected = quota["limit_snapshot"] + quota["adjustment"] + payload.amount - quota["used"] - quota["reserved"]
+    if projected < 0:
+        raise error("QUOTA_BALANCE_INVALID", "调整后额度不能小于已用额度", 422)
     connection.execute("UPDATE quota_accounts SET adjustment = adjustment + %s, version = version + 1 WHERE id = %s", (payload.amount, quota["id"]))
     connection.execute("INSERT INTO usage_records(user_id, quota_account_id, type, amount, idempotency_key) VALUES (%s, %s, 'adjust', %s, %s)", (user_id, quota["id"], payload.amount, payload.idempotency_key))
     connection.execute("INSERT INTO admin_audit_logs(actor_id, action, target_type, target_id, after_value, reason) VALUES (%s, 'quota_adjustment', 'user', %s, %s, %s)", (actor["id"], user_id, Json({"amount": payload.amount}), payload.reason))
     connection.commit()
     return {"status": "ok", "amount": payload.amount}
+
+
+@app.get("/api/v1/admin/tts/tasks")
+def admin_tts_tasks(user_id: UUID | None = None, status: str | None = None, created_after: datetime | None = None, created_before: datetime | None = None, page: int = 1, limit: int = 20, user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
+    page = max(1, page)
+    limit = min(100, max(1, limit))
+    clauses = []
+    params: list[Any] = []
+    if user_id:
+        clauses.append("t.user_id = %s")
+        params.append(user_id)
+    if status in ("queued", "running", "succeeded", "failed"):
+        clauses.append("t.status = %s")
+        params.append(status)
+    if created_after:
+        clauses.append("t.created_at >= %s")
+        params.append(created_after)
+    if created_before:
+        clauses.append("t.created_at <= %s")
+        params.append(created_before)
+    where = " WHERE " + " AND ".join(clauses) if clauses else ""
+    total = connection.execute(f"SELECT count(*) AS c FROM tts_tasks t{where}", params).fetchone()["c"]
+    rows = connection.execute(
+        f"""
+        SELECT t.id, t.user_id, u.email AS user_email, t.status, t.text_length, t.provider_voice_id,
+               t.error_code, t.attempt_count, t.reserved_amount, t.created_at, t.started_at, t.finished_at,
+               (EXTRACT(EPOCH FROM (t.finished_at - t.started_at)) * 1000)::bigint AS duration_ms,
+               CASE WHEN t.status = 'succeeded' THEN 'consumed' WHEN t.status = 'failed' THEN 'released'
+                    ELSE 'pending' END AS settlement,
+               CASE WHEN t.status IN ('queued', 'running') THEN t.reserved_amount ELSE 0 END AS reserved,
+               CASE WHEN t.status = 'succeeded' THEN t.reserved_amount ELSE 0 END AS used
+        FROM tts_tasks t JOIN users u ON u.id = t.user_id{where}
+        ORDER BY t.created_at DESC LIMIT %s OFFSET %s
+        """,
+        [*params, limit, (page - 1) * limit],
+    ).fetchall()
+    return {"items": rows, "total": total, "page": page, "limit": limit, "pages": (total + limit - 1) // limit}
+
+
+@app.get("/api/v1/admin/usage/summary")
+def admin_usage_summary(user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
+    total_used = connection.execute("SELECT COALESCE(sum(used), 0)::bigint AS total FROM quota_accounts").fetchone()["total"]
+    total_reserved = connection.execute("SELECT COALESCE(sum(reserved), 0)::bigint AS total FROM quota_accounts").fetchone()["total"]
+    total_adjust = connection.execute("SELECT COALESCE(sum(adjustment), 0)::bigint AS total FROM quota_accounts").fetchone()["total"]
+    record_total = connection.execute("SELECT COALESCE(sum(CASE WHEN type IN ('consume', 'adjust') THEN amount WHEN type = 'release' THEN -amount ELSE 0 END), 0)::bigint AS total FROM usage_records").fetchone()["total"]
+    task_counts = connection.execute("SELECT status, count(*) AS c FROM tts_tasks GROUP BY status").fetchall()
+    by_status = {row["status"]: row["c"] for row in task_counts}
+    failed = connection.execute("SELECT count(*) AS c FROM tts_tasks WHERE status = 'failed'").fetchone()["c"]
+    return {
+        "quota": {"used": total_used, "reserved": total_reserved, "adjustment": total_adjust},
+        "ledger_total": record_total,
+        "tasks": {"total": sum(by_status.values()), "by_status": by_status, "failed": failed},
+    }
+
+
+@app.get("/api/v1/admin/audit-logs")
+def admin_audit_logs(target_type: str | None = None, actor_id: UUID | None = None, page: int = 1, limit: int = 30, user: dict = Depends(require_admin), connection: Connection = Depends(get_connection)):
+    page = max(1, page)
+    limit = min(100, max(1, limit))
+    clauses = []
+    params: list[Any] = []
+    if target_type in ("user", "quota_account", "product", "tts_task"):
+        clauses.append("a.target_type = %s")
+        params.append(target_type)
+    if actor_id:
+        clauses.append("a.actor_id = %s")
+        params.append(actor_id)
+    where = " WHERE " + " AND ".join(clauses) if clauses else ""
+    total = connection.execute(f"SELECT count(*) AS c FROM admin_audit_logs a{where}", params).fetchone()["c"]
+    rows = connection.execute(
+        f"""
+        SELECT a.id, a.actor_id, u.email AS actor_email, a.action, a.target_type, a.target_id,
+               a.before_value, a.after_value, a.reason, a.created_at
+        FROM admin_audit_logs a LEFT JOIN users u ON u.id = a.actor_id{where}
+        ORDER BY a.created_at DESC LIMIT %s OFFSET %s
+        """,
+        [*params, limit, (page - 1) * limit],
+    ).fetchall()
+    return {"items": rows, "total": total, "page": page, "limit": limit, "pages": (total + limit - 1) // limit}

services/api/app/schemas.py

@@ -55,6 +55,27 @@ class QuotaAdjustmentRequest(BaseModel):
     idempotency_key: str = Field(min_length=8, max_length=120)
 
 
+class MembershipRevokeRequest(BaseModel):
+    reason: str = Field(min_length=1, max_length=500)
+
+
+class AdminTaskFilter(BaseModel):
+    user_id: UUID | None = None
+    status: Literal["queued", "running", "succeeded", "failed"] | None = None
+    created_after: datetime | None = None
+    created_before: datetime | None = None
+    page: int = Field(default=1, ge=1)
+    limit: int = Field(default=20, ge=1, le=100)
+
+
+class AdminUserFilter(BaseModel):
+    email: str | None = None
+    phone: str | None = None
+    status: Literal["active", "disabled"] | None = None
+    page: int = Field(default=1, ge=1)
+    limit: int = Field(default=20, ge=1, le=100)
+
+
 class VerificationConfirmRequest(BaseModel):
     challenge_id: UUID
     code: str = Field(min_length=4, max_length=12)

services/api/tests/phase4_admin.py

@@ -0,0 +1,156 @@
+"""Phase 4 admin P4-04..P4-06 API verification.
+
+Runs against the running API on 127.0.0.1:8000 using the admin account whose
+credentials live in /etc/kaotings/p4-ops.secrets (root-only). It must never
+print the admin password or any secret. Requires the venv to have httpx.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+from datetime import datetime, timedelta, timezone
+from uuid import uuid4
+
+import httpx
+
+API_ROOT = "https://192.168.199.22/api/v1"
+
+
+def load_secret(name: str) -> str:
+    value = os.getenv(name)
+    if not value:
+        raise SystemExit(f"missing env {name}")
+    return value
+
+
+def client() -> httpx.Client:
+    session = httpx.Client(base_url=API_ROOT, timeout=15, follow_redirects=False, verify=False)
+    csrf = session.get("/auth/csrf").json()["csrf_token"]
+    resp = session.post("/auth/login", json={"email": load_secret("P4_OPS_EMAIL"), "password": load_secret("P4_OPS_PASSWORD")}, headers={"X-CSRF-Token": csrf})
+    assert resp.status_code == 200, resp.text
+    session.headers.update({"X-CSRF-Token": csrf})
+    return session
+
+
+def mk_user(session: httpx.Client, prefix: str) -> tuple[str, str]:
+    email = f"p4-{prefix}-{uuid4().hex[:10]}@example.com"
+    password = f"P4T-{uuid4().hex}-Aa1!"
+    body = {"email": email, "password": password}
+    csrf = session.get("/auth/csrf").json()["csrf_token"]
+    assert session.post("/auth/register", json=body, headers={"X-CSRF-Token": csrf}).status_code == 201, "register failed"
+    return email, password
+
+
+def find_user_by_email(session: httpx.Client, email: str) -> dict:
+    r = session.get("/admin/users", params={"email": email})
+    assert r.status_code == 200, r.text
+    items = r.json()["items"]
+    assert items, f"user {email} not found"
+    return next(i for i in items if i["email"] == email)
+
+
+def main() -> None:
+    results: dict = {}
+    session = client()
+
+    # P4-04: pagination + filters + detail
+    r = session.get("/admin/users", params={"limit": 5, "page": 1})
+    assert r.status_code == 200, r.text
+    body = r.json()
+    results["users_page"] = {"total": body["total"], "pages": body["pages"], "items": len(body["items"])}
+
+    email, user_password = mk_user(session, "detail")
+    target = find_user_by_email(session, email)
+    results["users_filter_email"] = {"found": True, "status": target["status"]}
+
+    detail_resp = session.get(f"/admin/users/{target['id']}")
+    assert detail_resp.status_code == 200, f"detail {detail_resp.status_code}: {detail_resp.text[:300]}"
+    detail = detail_resp.json()
+    results["user_detail"] = {"has_quota": "quota" in detail, "effective_plan": detail["effective_plan"]}
+
+    # disable -> sessions revoked; verify last-admin protection separately
+    csrf = session.headers["X-CSRF-Token"]
+    dis = session.patch(f"/admin/users/{target['id']}/status", json={"status": "disabled", "reason": "P4-04 disable test"}, headers={"X-CSRF-Token": csrf})
+    assert dis.status_code == 200, dis.text
+    results["disable"] = dis.json()
+    # disabled user cannot log in with its real password
+    unauth = httpx.Client(base_url=API_ROOT, timeout=10, follow_redirects=False, verify=False)
+    dcs = unauth.get("/auth/csrf").json()["csrf_token"]
+    dl = unauth.post("/auth/login", json={"email": email, "password": user_password}, headers={"X-CSRF-Token": dcs})
+    results["disabled_login_blocked"] = dl.status_code in {401, 403}
+    unauth.close()
+    rev = session.patch(f"/admin/users/{target['id']}/status", json={"status": "active", "reason": "P4-04 re-enable test"}, headers={"X-CSRF-Token": csrf})
+    assert rev.status_code == 200, rev.text
+    results["restore"] = rev.json()
+
+    # last-admin protection: try disabling the acting admin
+    me = session.get("/auth/me").json()
+    admin_target = session.get("/admin/users", params={"email": me["email"]}).json()["items"][0]
+    # there is more than one admin, so disabling one is allowed; check non-admin guard instead below
+
+    # P4-05: membership view + grant + revoke
+    mem = session.get(f"/admin/users/{target['id']}/membership")
+    results["membership_view"] = mem.json()
+    grant = session.put(f"/admin/users/{target['id']}/membership", json={"plan": "vip", "expires_at": (datetime.now(timezone.utc) + timedelta(days=30)).isoformat(), "reason": "P4-05 vip grant"}, headers={"X-CSRF-Token": csrf})
+    assert grant.status_code == 200, grant.text
+    results["vip_grant"] = grant.json()
+    mem2 = session.get(f"/admin/users/{target['id']}/membership").json()
+    results["vip_effective"] = mem2["effective_plan"]
+    revoke = session.post(f"/admin/users/{target['id']}/membership/revoke", json={"reason": "P4-05 revoke"}, headers={"X-CSRF-Token": csrf})
+    assert revoke.status_code == 200, revoke.text
+    results["vip_revoke"] = revoke.json()
+    mem3 = session.get(f"/admin/users/{target['id']}/membership").json()
+    results["vip_effective_after_revoke"] = mem3["effective_plan"]
+
+    # quota view + adjustment with reason + idempotency + balance constraint
+    quota = session.get(f"/admin/users/{target['id']}/quota").json()
+    results["quota_view"] = quota["quota"]
+    idem = f"p4-adjust-{uuid4().hex}"
+    adj = session.post(f"/admin/users/{target['id']}/quota-adjustments", json={"amount": 500, "reason": "P4-05 grant bonus", "idempotency_key": idem}, headers={"X-CSRF-Token": csrf})
+    results["quota_adjust"] = adj.status_code
+    dup = session.post(f"/admin/users/{target['id']}/quota-adjustments", json={"amount": 500, "reason": "P4-05 dup", "idempotency_key": idem}, headers={"X-CSRF-Token": csrf})
+    results["quota_adjust_idempotent"] = dup.status_code == 409
+    # balance constraint: cannot push below used
+    hard = session.post(f"/admin/users/{target['id']}/quota-adjustments", json={"amount": -100000, "reason": "P4-05 too low", "idempotency_key": f"p4-adjust-{uuid4().hex}"}, headers={"X-CSRF-Token": csrf})
+    hard_body = hard.json().get("error", hard.json()) if hard.status_code >= 400 else {}
+    results["quota_balance_constraint"] = {"status": hard.status_code, "code": hard_body.get("code")}
+    # missing reason rejected
+    noreason = session.post(f"/admin/users/{target['id']}/quota-adjustments", json={"amount": 10, "idempotency_key": f"p4-adjust-{uuid4().hex}"}, headers={"X-CSRF-Token": csrf})
+    results["quota_reason_required"] = noreason.status_code == 422
+
+    # P4-06: tts tasks view + audit logs + usage summary
+    tasks = session.get("/admin/tts/tasks", params={"page": 1, "limit": 10})
+    assert tasks.status_code == 200, tasks.text
+    results["tts_tasks"] = {"total": tasks.json()["total"], "items": len(tasks.json()["items"])}
+    summary = session.get("/admin/usage/summary")
+    assert summary.status_code == 200, summary.text
+    results["usage_summary"] = summary.json()
+    audit = session.get("/admin/audit-logs", params={"limit": 30})
+    assert audit.status_code == 200, audit.text
+    audit_items = audit.json()["items"]
+    # never show secrets / never include user text fields in the payload shape
+    for item in audit_items:
+        keys = set(item.keys())
+        assert "text" not in keys and "password" not in keys
+    results["audit"] = {"total": audit.json()["total"], "sample_keys": sorted(set().union(*(set(i.keys()) for i in audit_items[0:1])) if audit_items else [])}
+
+    # permission guard: a normal (non-admin) user account cannot reach admin endpoints
+    known_email = f"p4-known-{uuid4().hex}@example.com"
+    known_pw = f"K-{uuid4().hex}-Aa1!"
+    reg_session = httpx.Client(base_url=API_ROOT, timeout=10, follow_redirects=True, verify=False)
+    rcs_k = reg_session.get("/auth/csrf").json()["csrf_token"]
+    assert reg_session.post("/auth/register", json={"email": known_email, "password": known_pw}, headers={"X-CSRF-Token": rcs_k}).status_code == 201
+    assert reg_session.post("/auth/login", json={"email": known_email, "password": known_pw}, headers={"X-CSRF-Token": rcs_k}).status_code == 200
+    guard = reg_session.get("/admin/users")
+    guard_detail = reg_session.get(f"/admin/users/{target['id']}")
+    results["regular_forbidden_users"] = guard.status_code == 403
+    results["regular_forbidden_detail"] = guard_detail.status_code == 403
+    reg_session.close()
+
+    print(json.dumps(results, ensure_ascii=False, indent=2, default=str))
+
+
+if __name__ == "__main__":
+    main()