fix(api): make quota adjustment consistent with concurrent freeze/settlement

Admin quota adjustment now takes the per-user advisory lock and a FOR UPDATE
row lock on the quota account before projecting the balance, so the check
covers both used and reserved amounts and stays transactionally consistent
with task freeze and settlement. Settlement and lease-recovery paths also
lock the quota row explicitly.

Add the isolated Phase 5 exception-matrix harness (fault injection against a
dedicated DB + fake upstream) covering last-quota contention, concurrent
admin reduce, freeze/reduce stress, bad/oversized/timeout upstream, storage
failure, cross-period settlement and queued/running restart recovery.
Add test-only deps, provisioning helper (secrets via env) and test README.

services/api/app/main.py

@@ -153,6 +153,7 @@ def finish_task_failure(task: dict, code: str) -> None:
         if updated.rowcount != 1:
             connection.rollback()
             return
+        connection.execute("SELECT id FROM quota_accounts WHERE id = %s FOR UPDATE", (task["quota_account_id"],))
         connection.execute("UPDATE quota_accounts SET reserved = GREATEST(0, reserved - %s), version = version + 1 WHERE id = %s", (task["reserved_amount"], task["quota_account_id"]))
         connection.execute("INSERT INTO usage_records(user_id, quota_account_id, task_id, type, amount) VALUES (%s, %s, %s, 'release', %s)", (task["user_id"], task["quota_account_id"], task["id"], task["reserved_amount"]))
         connection.commit()
@@ -173,6 +174,7 @@ def finish_task_success(task: dict, audio: bytes, mime_type: str) -> None:
             target.unlink(missing_ok=True)
             connection.rollback()
             return
+        connection.execute("SELECT id FROM quota_accounts WHERE id = %s FOR UPDATE", (task["quota_account_id"],))
         connection.execute("INSERT INTO audio_files(task_id, owner_id, storage_key, mime_type, size_bytes, checksum) VALUES (%s, %s, %s, %s, %s, %s)", (task["id"], task["user_id"], storage_key, mime_type, len(audio), checksum))
         connection.execute("UPDATE tts_tasks SET status = 'succeeded', finished_at = now(), updated_at = now(), lease_token = NULL, lease_expires_at = NULL WHERE id = %s AND status = 'running' AND lease_token = %s", (task["id"], task["lease_token"]))
         connection.execute("UPDATE quota_accounts SET reserved = GREATEST(0, reserved - %s), used = used + %s, version = version + 1 WHERE id = %s", (task["reserved_amount"], task["reserved_amount"], task["quota_account_id"]))
@@ -185,6 +187,7 @@ def recover_expired_tasks() -> None:
         tasks = connection.execute("SELECT * FROM tts_tasks WHERE status = 'running' AND lease_expires_at IS NOT NULL AND lease_expires_at < now() FOR UPDATE SKIP LOCKED").fetchall()
         for task in tasks:
             connection.execute("UPDATE tts_tasks SET status = 'failed', error_code = 'WORKER_LEASE_EXPIRED', finished_at = now(), updated_at = now(), lease_token = NULL, lease_expires_at = NULL WHERE id = %s AND status = 'running'", (task["id"],))
+            connection.execute("SELECT id FROM quota_accounts WHERE id = %s FOR UPDATE", (task["quota_account_id"],))
             connection.execute("UPDATE quota_accounts SET reserved = GREATEST(0, reserved - %s), version = version + 1 WHERE id = %s", (task["reserved_amount"], task["quota_account_id"]))
             connection.execute("INSERT INTO usage_records(user_id, quota_account_id, task_id, type, amount) VALUES (%s, %s, %s, 'release', %s)", (task["user_id"], task["quota_account_id"], task["id"], task["reserved_amount"]))
         connection.commit()
@@ -647,12 +650,18 @@ def admin_quota(user_id: UUID, payload: QuotaAdjustmentRequest, connection: Conn
     if not target:
         raise error("NOT_FOUND", "用户不存在", 404)
     quota = ensure_quota(connection, user_id, effective_plan(connection, target))
+    # Serialize with task freeze (create_tts_task) and settlement on the same
+    # user, then re-read the quota row under a row lock so the projected balance
+    # check is transactionally consistent. The check must cover BOTH used and
+    # reserved (frozen) amounts, not only used.
+    connection.execute("SELECT pg_advisory_xact_lock(hashtext(%s))", (str(user_id),))
+    quota = connection.execute("SELECT * FROM quota_accounts WHERE id = %s FOR UPDATE", (quota["id"],)).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)
+        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))

services/api/requirements-test.txt

@@ -0,0 +1,6 @@
+# Test-only dependencies for services/api.
+# The reliability / exception-matrix harness (tests/phase5_exception_matrix.py)
+# imports only these from the runtime stack; the full runtime set is in
+# requirements.txt. Pin to the versions verified in the isolated environment.
+httpx==0.28.1
+psycopg[binary]==3.2.9

services/api/tests/README.md

@@ -0,0 +1,74 @@
+# services/api 测试
+
+可靠性 / 异常矩阵测试在**隔离环境**运行：独立 Postgres 库、独立音频目录、本地伪造
+上游（fake TTS）。绝不连接真实 TTS 提供者，也不改动生产 `kaotings` 库。
+
+## 依赖
+
+运行时依赖见 `../requirements.txt`，测试额外依赖见 `requirements-test.txt`。
+已验证版本（隔离环境实测）：
+
+| 组件 | 版本 |
+| --- | --- |
+| Python | 3.12.3 |
+| fastapi | 0.115.14 |
+| uvicorn[standard] | 0.34.3 |
+| psycopg[binary] | 3.2.9 |
+| httpx | 0.28.1 |
+| argon2-cffi | 25.1.0 |
+| pydantic | 2.11.7 |
+
+安装（在独立 venv）：
+
+```
+python3 -m venv /path/venv
+/path/venv/bin/pip install -r requirements.txt -r requirements-test.txt
+```
+
+## 1. 准备隔离库
+
+`provision_isolated_env.sh` 会删除并重建隔离库、应用迁移并种子化 plan/voice。
+秘密只通过环境变量注入，不写入仓库：
+
+```
+export SUDO_PASSWORD='<sudo 密码>'
+export P5_DB_PASSWORD='<隔离库密码>'        # 任意强口令，仅隔离库使用
+bash services/api/tests/provision_isolated_env.sh
+```
+
+## 2. 运行异常矩阵
+
+`phase5_exception_matrix.py` 会启动一个隔离 API（uvicorn）+ 本地伪造上游，逐项执行
+异常场景并输出结果表与不变式判定：
+
+| 场景 | 覆盖 |
+| --- | --- |
+| S1 | 成功基线（写入、结算、回放、下载） |
+| S2 | 最后额度并发争用（仅一个冻结成功） |
+| S3 | 管理员并发调减额度（不变式：总额度 >= 已用+冻结） |
+| S4 | 冻结与调减并发压测（终态不变式） |
+| S5 | 上游 非音频 / 损坏 / 超大 / HTTP 5xx |
+| S6 | 跨周期结算（结算落到任务创建时所属周期账户） |
+| S7 | 存储失败（音频目录不可写） |
+| S8 | 上游超时 |
+| S9 | 重启恢复 - queued（未决任务重启后被领取结算） |
+| S10 | 重启恢复 - running 租约过期（释放冻结） |
+
+```
+export P5_DATABASE_URL='postgresql://kaotings_p5:<P5_DB_PASSWORD>@127.0.0.1:5432/kaotings_p5'
+export P5_API_DIR="$(pwd)/services/api"      # 被测代码目录
+export P5_AUDIO_DIR="/tmp/kaotings-p5-audio"
+/path/venv/bin/python services/api/tests/phase5_exception_matrix.py
+echo "exit=$?"   # 0 = 全部不变式通过
+```
+
+退出码 0 表示所有场景的额度不变式与状态断言通过；非 0 表示存在失败项。
+`P5_REPORT_PATH` 可指定把结果表写成 markdown 留档。
+
+## 3. 其它测试
+
+- `phase3_reliability.py`：最小成功 / 负向 / 超时 / 存储失败（旧版，使用共享测试库）。
+- `phase4_admin.py` / `phase4_browser.py`：管理后台 API 与浏览器验收。
+
+> 浏览器测试（Playwright）需 `PLAYWRIGHT_BROWSERS_PATH` 指向已下载浏览器缓存；
+> 浏览器二进制不入库，见 `docs/phase-4-report.md`。

services/api/tests/phase5_exception_matrix.py

@@ -0,0 +1,558 @@
+"""Phase 5 task/quota exception matrix (isolated, fault-injection).
+
+Boots an isolated business API (uvicorn) against a dedicated Postgres database
+and a local fake TTS upstream, then exercises the Phase 3 leftover exception
+matrix. It never touches the real TTS provider and uses throwaway accounts in
+the isolated database only.
+
+Required environment:
+  P5_DATABASE_URL   isolated Postgres DSN (role must own the DB)
+  P5_API_DIR        path to the services/api package to boot (default: repo services/api)
+  P5_AUDIO_DIR      isolated audio dir (default: /tmp/kaotings-p5-audio)
+  P5_REPORT_PATH    optional; if set, a markdown summary is written here
+
+Each scenario records: task status, upstream call-count delta, used, reserved,
+available (current-period quota account) and the audio-file result.
+
+Exit code 0 => every recorded invariant held; non-zero => at least one failure.
+"""
+from __future__ import annotations
+
+import json
+import os
+import signal
+import subprocess
+import sys
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass, field
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from uuid import uuid4
+
+import httpx
+import psycopg
+
+API_PORT = int(os.environ.get("P5_API_PORT", "8200"))
+FAKE_PORT = int(os.environ.get("P5_FAKE_PORT", "9200"))
+API_ROOT = f"http://127.0.0.1:{API_PORT}/api/v1"
+FAKE_ROOT = f"http://127.0.0.1:{FAKE_PORT}"
+DB_URL = os.environ["P5_DATABASE_URL"]
+API_DIR = os.environ.get("P5_API_DIR") or str(Path(__file__).resolve().parents[1])
+AUDIO_DIR = Path(os.environ.get("P5_AUDIO_DIR", "/tmp/kaotings-p5-audio"))
+REPORT_PATH = os.environ.get("P5_REPORT_PATH", "")
+DEFAULT_TIMEOUT = 30  # generous so the "slow"/"running" scenario stays in-flight
+SHORT_TIMEOUT = 1  # for the explicit upstream-timeout scenario
+
+
+class FakeTts(BaseHTTPRequestHandler):
+    mode = "success"
+    sleep_s = 0.0
+    calls = 0
+    _lock = threading.Lock()
+    protocol_version = "HTTP/1.1"
+
+    @classmethod
+    def bump(cls):
+        with cls._lock:
+            cls.calls += 1
+
+    @classmethod
+    def reset_calls(cls):
+        with cls._lock:
+            cls.calls = 0
+
+    @classmethod
+    def call_count(cls):
+        with cls._lock:
+            return cls.calls
+
+    def do_POST(self):  # noqa: N802
+        self.bump()
+        if self.sleep_s:
+            time.sleep(self.sleep_s)
+        if self.mode == "non_audio":
+            payload, ctype = b"not audio", "text/plain"
+        elif self.mode == "corrupt":
+            payload, ctype = b"bad wav", "audio/wav"
+        elif self.mode == "oversized":
+            # the code checks size after httpx has read the body, so send a real
+            # > 50 MiB payload (localhost, fast) to exercise UPSTREAM_RESPONSE_TOO_LARGE
+            payload, ctype = b"x" * (50 * 1024 * 1024 + 1), "audio/wav"
+        elif self.mode == "http_500":
+            self.send_response(500)
+            self.send_header("Content-Type", "text/plain")
+            self.send_header("Content-Length", "4")
+            self.end_headers()
+            try:
+                self.wfile.write(b"boom")
+                self.wfile.flush()
+            except (BrokenPipeError, ConnectionResetError):
+                pass
+            return
+        else:  # success
+            payload = b"RIFF" + (36).to_bytes(4, "little") + b"WAVEfmt " + bytes(36)
+            ctype = "audio/wav"
+        try:
+            self.send_response(200)
+            self.send_header("Content-Type", ctype)
+            self.send_header("Content-Length", str(len(payload)))
+            self.end_headers()
+            chunk = 1024 * 1024
+            for off in range(0, len(payload), chunk):
+                self.wfile.write(payload[off:off + chunk])
+            self.wfile.flush()
+        except (BrokenPipeError, ConnectionResetError):
+            pass
+
+    def log_message(self, *_args):
+        return
+
+
+def db():
+    return psycopg.connect(DB_URL, row_factory=psycopg.rows.dict_row)
+
+
+def db_query(sql, params=()):
+    with db() as c:
+        return c.execute(sql, params).fetchall()
+
+
+def db_exec(sql, params=()):
+    with db() as c:
+        c.execute(sql, params)
+        c.commit()
+
+
+def start_fake() -> ThreadingHTTPServer:
+    server = ThreadingHTTPServer(("127.0.0.1", FAKE_PORT), FakeTts)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    return server
+
+
+def boot_api(storage_dir: str, timeout: int = DEFAULT_TIMEOUT) -> subprocess.Popen:
+    env = os.environ.copy()
+    env.update({
+        "DATABASE_URL": DB_URL,
+        "TTS_UPSTREAM_URL": FAKE_ROOT,
+        "TTS_API_KEY": "test-only-fake-key",
+        "TTS_TIMEOUT_SECONDS": str(timeout),
+        "AUDIO_STORAGE_DIR": str(storage_dir),
+        "SESSION_COOKIE_SECURE": "false",
+        "APP_ENV": "test",
+        "CSRF_SECRET": "p5-matrix-test-secret",
+    })
+    proc = subprocess.Popen(
+        [sys.executable, "-m", "uvicorn", "app.main:app", "--host", "127.0.0.1", "--port", str(API_PORT), "--workers", "1"],
+        cwd=API_DIR, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+    )
+    for _ in range(80):
+        try:
+            if httpx.get(f"http://127.0.0.1:{API_PORT}/healthz", timeout=1).status_code == 200:
+                return proc
+        except httpx.HTTPError:
+            time.sleep(0.2)
+    proc.kill()
+    raise RuntimeError("isolated API did not start")
+
+
+def stop_api(proc: subprocess.Popen | None, force: bool = False) -> None:
+    if proc is None or proc.poll() is not None:
+        return
+    proc.send_signal(signal.SIGKILL if force else signal.SIGTERM)
+    try:
+        proc.wait(timeout=10)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.wait(timeout=10)
+
+
+def make_client(email: str | None = None, password: str | None = None) -> httpx.Client:
+    email = email or f"p5-{uuid4().hex[:12]}@example.com"
+    password = password or f"Test-{uuid4().hex[:10]}-Aa1!"
+    s = httpx.Client(base_url=API_ROOT, timeout=20, follow_redirects=True)
+    csrf = s.get("/auth/csrf").json()["csrf_token"]
+    assert s.post("/auth/register", json={"email": email, "password": password}, headers={"X-CSRF-Token": csrf}).status_code == 201
+    assert s.post("/auth/login", json={"email": email, "password": password}, headers={"X-CSRF-Token": csrf}).status_code == 200
+    s.headers.update({"X-CSRF-Token": csrf})
+    return s
+
+
+def user_id(session) -> str:
+    return session.get("/auth/me").json()["id"]
+
+
+def make_admin() -> httpx.Client:
+    session = make_client()
+    uid = user_id(session)
+    db_exec("UPDATE users SET role = 'admin', email_verified = true WHERE id = %s", (uid,))
+    return session
+
+
+def grant_vip(session) -> None:
+    uid = user_id(session)
+    db_exec("INSERT INTO membership_grants(user_id, plan, starts_at, reason) VALUES (%s, 'vip', now(), 'p5-matrix-test')", (uid,))
+    session.get("/account/usage")  # materialize the current-period quota account
+
+
+def set_limit(session, limit: int) -> None:
+    uid = user_id(session)
+    with db() as c:
+        c.execute(
+            "UPDATE quota_accounts SET limit_snapshot = %s WHERE user_id = %s "
+            "AND period_start = (SELECT max(period_start) FROM quota_accounts WHERE user_id = %s)",
+            (limit, uid, uid),
+        )
+        c.commit()
+
+
+def prepare_worker(session, limit: int) -> str:
+    """Grant VIP (concurrency>1), materialize the current-period quota account,
+    reset its counters and set the limit. Returns the user id."""
+    grant_vip(session)
+    uid = user_id(session)
+    with db() as c:
+        c.execute(
+            "UPDATE quota_accounts SET used=0, reserved=0, adjustment=0, limit_snapshot=%s "
+            "WHERE user_id=%s AND period_start=(SELECT max(period_start) FROM quota_accounts WHERE user_id=%s)",
+            (limit, uid, uid),
+        )
+        c.commit()
+    return uid
+
+
+def current_quota(uid: str) -> dict:
+    row = db_query(
+        "SELECT limit_snapshot, adjustment, used, reserved FROM quota_accounts WHERE user_id = %s "
+        "AND period_start = (SELECT max(period_start) FROM quota_accounts WHERE user_id = %s)",
+        (uid, uid),
+    )
+    assert row, "no current quota account"
+    q = dict(row[0])
+    q["available"] = max(0, q["limit_snapshot"] + q["adjustment"] - q["used"] - q["reserved"])
+    return q
+
+
+def invariant_ok(q: dict) -> bool:
+    return (q["limit_snapshot"] + q["adjustment"]) - q["used"] - q["reserved"] >= 0
+
+
+def create_task(session, text: str, idem: str | None = None):
+    return session.post(
+        "/tts/tasks",
+        headers={"Idempotency-Key": idem or f"p5-{uuid4().hex}"},
+        json={"text": text, "voice_id": "default", "parameters": {"format": "wav", "speed": 1}},
+    )
+
+
+def wait_task(session, task_id: str, timeout: float = 12) -> dict:
+    end = time.time() + timeout
+    while time.time() < end:
+        t = session.get(f"/tts/tasks/{task_id}").json()
+        if t["status"] in {"succeeded", "failed"}:
+            return t
+        time.sleep(0.15)
+    return session.get(f"/tts/tasks/{task_id}").json()
+
+
+def audio_status(session, task_id: str) -> int:
+    return session.get(f"/tts/tasks/{task_id}/audio").status_code
+
+
+def audio_file_exists(task_id: str) -> bool:
+    return any(p.is_file() for p in AUDIO_DIR.rglob(f"{task_id}.*"))
+
+
+@dataclass
+class Scenario:
+    name: str
+    calls: int = 0
+    status: str = ""
+    used: int | None = None
+    reserved: int | None = None
+    available: int | None = None
+    audio: str = "n/a"
+    ok: bool = True
+    note: str = ""
+    extra: dict = field(default_factory=dict)
+
+
+results: list[Scenario] = []
+FAILURES: list[str] = []
+API_HANDLE: subprocess.Popen | None = None
+POOL: dict[str, httpx.Client] = {}
+
+
+def record(name: str, **kw) -> Scenario:
+    sc = Scenario(name=name, **kw)
+    results.append(sc)
+    if not sc.ok:
+        FAILURES.append(name)
+    return sc
+
+
+def check(cond: bool, sc: Scenario, msg: str) -> None:
+    if not cond:
+        sc.ok = False
+        sc.note = (sc.note + " | " if sc.note else "") + msg
+
+
+def main() -> int:
+    global API_HANDLE
+    AUDIO_DIR.mkdir(parents=True, exist_ok=True)
+    for p in AUDIO_DIR.rglob("*"):
+        if p.is_file():
+            p.unlink()
+    fake = start_fake()
+    API_HANDLE = boot_api(str(AUDIO_DIR), DEFAULT_TIMEOUT)
+    # Pre-create a small pool on the first API instance (register rate limit is
+    # 5/hour per source IP, and S1-S6 share this single instance). S7-S10 each
+    # restart the API, which resets the in-memory limiter, so they register fresh.
+    POOL["admin"] = make_admin()
+    POOL["w1"] = make_client()
+    POOL["w2"] = make_client()
+    POOL["w3"] = make_client()
+    try:
+        s1_success()
+        s2_last_quota_contention()
+        s3_admin_reduce_concurrent()
+        s4_freeze_reduce_stress()
+        s5_bad_upstream_responses()
+        s6_cross_period_settlement()
+        s7_storage_failure()
+        s8_upstream_timeout()
+        s9_queued_restart()
+        s10_running_lease_recovery()
+    finally:
+        stop_api(API_HANDLE, force=True)
+        fake.shutdown()
+    emit_report()
+    return 0 if not FAILURES else 1
+
+
+def s1_success() -> None:
+    session = POOL["w1"]
+    uid = prepare_worker(session, 10000)
+    FakeTts.mode, FakeTts.sleep_s = "success", 0.0
+    FakeTts.reset_calls()
+    before = FakeTts.call_count()
+    r = create_task(session, "测试")
+    tid = r.json()["id"]
+    task = wait_task(session, tid)
+    q = current_quota(uid)
+    audio = audio_status(session, tid)
+    sc = record("S1 成功基线", calls=FakeTts.call_count() - before, status=task["status"], used=q["used"], reserved=q["reserved"], available=q["available"], audio=f"GET /audio={audio} file={audio_file_exists(tid)}")
+    check(r.status_code == 202, sc, f"create={r.status_code}")
+    check(task["status"] == "succeeded", sc, f"final={task['status']}")
+    check(q["used"] == 2 and q["reserved"] == 0, sc, f"used={q['used']} reserved={q['reserved']} (expect used=2 reserved=0)")
+    check(audio == 200 and audio_file_exists(tid), sc, f"audio={audio}")
+
+
+def s2_last_quota_contention() -> None:
+    session = POOL["w2"]
+    uid = prepare_worker(session, 5)
+    FakeTts.mode, FakeTts.sleep_s = "success", 0.0
+    FakeTts.reset_calls()
+    before = FakeTts.call_count()
+    L = 3
+    with ThreadPoolExecutor(max_workers=2) as ex:
+        resp = [f.result() for f in (ex.submit(create_task, session, "测" * L) for _ in range(2))]
+    codes = sorted(x.status_code for x in resp)
+    created = [x.json()["id"] for x in resp if x.status_code == 202]
+    for tid in created:
+        wait_task(session, tid)
+    q = current_quota(uid)
+    sc = record("S2 最后额度并发争用", calls=FakeTts.call_count() - before, status=f"codes={codes}", used=q["used"], reserved=q["reserved"], available=q["available"], audio="n/a")
+    check(codes == [202, 409], sc, f"codes={codes} (expect one 202 + one 409 QUOTA_EXCEEDED)")
+    check(len(created) == 1, sc, f"created={len(created)} (expect 1)")
+    check(invariant_ok(q), sc, f"available negative: {q}")
+    check(q["reserved"] == 0 and q["used"] == L, sc, f"used={q['used']} reserved={q['reserved']} (expect used={L} reserved=0)")
+
+
+def s3_admin_reduce_concurrent() -> None:
+    admin = POOL["admin"]
+    session = POOL["w3"]
+    uid = prepare_worker(session, 100)
+    iters, worst, violations = 8, None, 0
+    for _ in range(iters):
+        with db() as c:
+            c.execute("UPDATE quota_accounts SET used=0, reserved=0, adjustment=0, limit_snapshot=100 WHERE user_id=%s AND period_start=(SELECT max(period_start) FROM quota_accounts WHERE user_id=%s)", (uid, uid))
+            c.commit()
+        idems = [f"p5r-{uuid4().hex}" for _ in range(4)]
+
+        def reduce(i):
+            return admin.post(f"/admin/users/{uid}/quota-adjustments", json={"amount": -30, "reason": "p5-concurrency-reduce", "idempotency_key": idems[i]}).status_code
+
+        with ThreadPoolExecutor(max_workers=4) as ex:
+            list(ex.map(reduce, range(4)))
+        q = current_quota(uid)
+        if not invariant_ok(q):
+            violations += 1
+        if worst is None or q["available"] < worst["available"]:
+            worst = q
+    sc = record("S3 管理员并发调减额度", calls=0, status=f"iters={iters}", used=(worst or {}).get("used"), reserved=(worst or {}).get("reserved"), available=(worst or {}).get("available"), audio="n/a", extra={"invariant_violations": violations})
+    check(violations == 0, sc, f"额度不变式被破坏 {violations}/{iters} 次 (available 出现负值)")
+
+
+def s4_freeze_reduce_stress() -> None:
+    admin = POOL["admin"]
+    session = POOL["w1"]
+    uid = prepare_worker(session, 100)
+    FakeTts.mode, FakeTts.sleep_s = "success", 0.0
+    FakeTts.reset_calls()
+    before = FakeTts.call_count()
+
+    def mk(_):
+        return create_task(session, "测" * 8).status_code
+
+    def red(_):
+        return admin.post(f"/admin/users/{uid}/quota-adjustments", json={"amount": -10, "reason": "p5-stress-reduce", "idempotency_key": f"p5s-{uuid4().hex}"}).status_code
+
+    with ThreadPoolExecutor(max_workers=12) as ex:
+        futs = [ex.submit(mk, i) for i in range(6)] + [ex.submit(red, i) for i in range(6)]
+        codes = [f.result() for f in futs]
+    time.sleep(3)  # let all queued tasks settle
+    q = current_quota(uid)
+    sc = record("S4 冻结与调减并发压测", calls=FakeTts.call_count() - before, status=f"codes={sorted(codes)}", used=q["used"], reserved=q["reserved"], available=q["available"], audio="n/a")
+    check(invariant_ok(q), sc, f"最终额度不变式被破坏: {q}")
+    check(q["reserved"] == 0, sc, f"reserved 未清零: {q['reserved']}")
+
+
+def s5_bad_upstream_responses() -> None:
+    session = POOL["w2"]
+    uid = prepare_worker(session, 1000)
+    FakeTts.sleep_s = 0.0
+    expect = {"non_audio": "UPSTREAM_NOT_AUDIO", "corrupt": "UPSTREAM_AUDIO_CORRUPT", "oversized": "UPSTREAM_RESPONSE_TOO_LARGE", "http_500": "UPSTREAM_HTTP_500"}
+    for mode, code in expect.items():
+        FakeTts.mode = mode
+        r = create_task(session, f"bad-{mode}")
+        tid = r.json()["id"]
+        task = wait_task(session, tid)
+        q = current_quota(uid)
+        sc = record(f"S5 上游{mode}", calls=1, status=task["status"], used=q["used"], reserved=q["reserved"], available=q["available"], audio=f"file={audio_file_exists(tid)}")
+        check(task["status"] == "failed" and task["error_code"] == code, sc, f"status={task['status']} err={task['error_code']} (expect {code})")
+        check(not audio_file_exists(tid), sc, "失败任务不应有音频文件")
+    FakeTts.mode = "success"
+
+
+def s6_cross_period_settlement() -> None:
+    session = POOL["w3"]
+    uid = prepare_worker(session, 100)
+    FakeTts.mode, FakeTts.sleep_s = "success", 0.0
+    with db() as c:
+        qid = c.execute(
+            "INSERT INTO quota_accounts(user_id, period_start, period_end, limit_snapshot, reserved) "
+            "VALUES (%s, date_trunc('month', now() - interval '1 month'), date_trunc('month', now()), 100, 5) RETURNING id",
+            (uid,),
+        ).fetchone()["id"]
+        c.execute(
+            "INSERT INTO tts_tasks(user_id, text, text_length, voice_id, provider_voice_id, parameters, idempotency_key, request_hash, policy_version, quota_account_id, reserved_amount, status) "
+            "VALUES (%s, 'cross', 5, (SELECT id FROM tts_voices WHERE provider_voice_id='default'), 'default', '{}'::jsonb, %s, %s, 1, %s, 5, 'queued')",
+            (uid, f"p5xp-{uuid4().hex}", f"p5xp-{uuid4().hex}", qid),
+        )
+        c.commit()
+    time.sleep(3)  # worker claims + settles
+    prior = db_query("SELECT used, reserved FROM quota_accounts WHERE id = %s", (qid,))
+    cur = current_quota(uid)
+    sc = record("S6 跨周期结算", calls=1, status="settled", used=prior[0]["used"], reserved=prior[0]["reserved"], available=cur["available"], audio="n/a", extra={"prior_used": prior[0]["used"], "current_used": cur["used"]})
+    check(prior[0]["used"] == 5 and prior[0]["reserved"] == 0, sc, f"上一周期账户 used={prior[0]['used']} reserved={prior[0]['reserved']} (expect used=5 reserved=0)")
+    check(cur["used"] == 0, sc, f"当前周期账户不应被结算, used={cur['used']}")
+
+
+def s7_storage_failure() -> None:
+    global API_HANDLE
+    stop_api(API_HANDLE, force=True)
+    blocker = Path("/tmp/kaotings-p5-storage-blocker")
+    if blocker.exists() and blocker.is_dir():
+        blocker.rmdir()
+    blocker.write_text("not a directory")
+    API_HANDLE = boot_api(str(blocker), DEFAULT_TIMEOUT)
+    session = make_client()
+    uid = user_id(session)
+    FakeTts.mode, FakeTts.sleep_s = "success", 0.0
+    r = create_task(session, "存储失败")
+    tid = r.json()["id"]
+    task = wait_task(session, tid)
+    q = current_quota(uid)
+    blocker.unlink(missing_ok=True)
+    sc = record("S7 存储失败", calls=1, status=task["status"], used=q["used"], reserved=q["reserved"], available=q["available"], audio=f"file={audio_file_exists(tid)}")
+    check(task["status"] == "failed", sc, f"status={task['status']}")
+    check(q["reserved"] == 0, sc, f"reserved 未释放: {q['reserved']}")
+    check(invariant_ok(q), sc, f"不变式: {q}")
+
+
+def s8_upstream_timeout() -> None:
+    global API_HANDLE
+    stop_api(API_HANDLE, force=True)
+    API_HANDLE = boot_api(str(AUDIO_DIR), SHORT_TIMEOUT)
+    session = make_client()
+    uid = user_id(session)
+    FakeTts.mode, FakeTts.sleep_s = "success", 5  # sleeps past the 1s client timeout
+    r = create_task(session, "超时")
+    tid = r.json()["id"]
+    task = wait_task(session, tid, timeout=12)
+    FakeTts.sleep_s = 0.0
+    q = current_quota(uid)
+    sc = record("S8 上游超时", calls=1, status=task["status"], used=q["used"], reserved=q["reserved"], available=q["available"], audio=f"file={audio_file_exists(tid)}")
+    check(task["status"] == "failed" and task["error_code"] == "UPSTREAM_TIMEOUT", sc, f"status={task['status']} err={task['error_code']} (expect UPSTREAM_TIMEOUT)")
+    check(q["reserved"] == 0, sc, f"reserved 未释放: {q['reserved']}")
+    stop_api(API_HANDLE, force=True)
+    API_HANDLE = boot_api(str(AUDIO_DIR), DEFAULT_TIMEOUT)
+
+
+def s9_queued_restart() -> None:
+    global API_HANDLE
+    session = make_client()
+    uid = user_id(session)
+    grant_vip(session)
+    FakeTts.mode, FakeTts.sleep_s = "success", 4  # first task stays running
+    r_a = create_task(session, "运行中")
+    r_b = create_task(session, "排队中")
+    tid_a, tid_b = r_a.json()["id"], r_b.json()["id"]
+    time.sleep(1.5)  # A running (blocked in fake), B queued
+    stop_api(API_HANDLE, force=True)  # unclean shutdown
+    FakeTts.sleep_s = 0.0
+    API_HANDLE = boot_api(str(AUDIO_DIR), DEFAULT_TIMEOUT)
+    task_b = wait_task(session, tid_b, timeout=15)
+    task_a = session.get(f"/tts/tasks/{tid_a}").json()
+    q = current_quota(uid)
+    sc = record("S9 重启恢复-queued", calls=2, status=f"queued->{task_b['status']} running->{task_a['status']}", used=q["used"], reserved=q["reserved"], available=q["available"], audio=f"b_file={audio_file_exists(tid_b)}")
+    check(task_b["status"] == "succeeded", sc, f"queued 任务重启后未恢复: {task_b['status']}")
+    check(audio_file_exists(tid_b), sc, "queued 任务重启后音频缺失")
+
+
+def s10_running_lease_recovery() -> None:
+    global API_HANDLE
+    session = make_client()
+    uid = user_id(session)
+    grant_vip(session)
+    FakeTts.mode, FakeTts.sleep_s = "success", 6  # keep it running
+    r = create_task(session, "租约过期")
+    tid = r.json()["id"]
+    time.sleep(1.5)
+    stop_api(API_HANDLE, force=True)  # worker dies while task running
+    db_exec("UPDATE tts_tasks SET lease_expires_at = now() - interval '1 second' WHERE id = %s", (tid,))
+    q_before = current_quota(uid)
+    API_HANDLE = boot_api(str(AUDIO_DIR), DEFAULT_TIMEOUT)
+    task = wait_task(session, tid, timeout=20)
+    q = current_quota(uid)
+    sc = record("S10 重启恢复-running租约过期", calls=0, status=task["status"], used=q["used"], reserved=q["reserved"], available=q["available"], audio=f"file={audio_file_exists(tid)}")
+    check(task["status"] == "failed" and task["error_code"] == "WORKER_LEASE_EXPIRED", sc, f"status={task['status']} err={task['error_code']} (expect WORKER_LEASE_EXPIRED)")
+    check(q["reserved"] < q_before["reserved"], sc, f"reserved 未释放 ({q_before['reserved']}->{q['reserved']})")
+    check(invariant_ok(q), sc, f"不变式: {q}")
+
+
+def emit_report() -> None:
+    lines = ["| 场景 | 上游调用 | 任务状态 | 已用 | 冻结 | 可用 | 音频结果 | 通过 | 备注 |", "| --- | --- | --- | --- | --- | --- | --- | --- | --- |"]
+    for sc in results:
+        lines.append(f"| {sc.name} | {sc.calls} | {sc.status} | {sc.used} | {sc.reserved} | {sc.available} | {sc.audio} | {'通过' if sc.ok else '失败'} | {sc.note} |")
+    blob = "\n".join(lines) + f"\n\n失败项: {FAILURES if FAILURES else '无'}\n"
+    print(blob)
+    if REPORT_PATH:
+        Path(REPORT_PATH).write_text(blob, encoding="utf-8")
+    print(json.dumps({"failures": FAILURES, "scenarios": len(results)}, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/api/tests/provision_isolated_env.sh

@@ -0,0 +1,35 @@
+#!/usr/bin/env bash
+# Provision an ISOLATED Postgres database for the Phase 5 exception matrix.
+#
+# Secrets are injected through the environment only (never stored in the repo):
+#   SUDO_PASSWORD    password for `sudo -S` (required to create role/DB)
+#   P5_DB_PASSWORD   password for the isolated role (required)
+#   P5_DB_ROLE       role name        (default: kaotings_p5)
+#   P5_DB            database name    (default: kaotings_p5)
+#   P5_PGHOST        host:port        (default: 127.0.0.1:5432)
+#
+# Idempotent: drops and recreates the isolated role/database, applies the API
+# migrations and seeds the plan policies + default voice, matching a fresh
+# production schema. The production `kaotings` database is never touched.
+
+set -euo pipefail
+
+: "${SUDO_PASSWORD:?SUDO_PASSWORD must be set in the environment}"
+: "${P5_DB_PASSWORD:?P5_DB_PASSWORD must be set in the environment}"
+ROLE="${P5_DB_ROLE:-kaotings_p5}"
+DB="${P5_DB:-kaotings_p5}"
+PGHOST="${P5_PGHOST:-127.0.0.1:5432}"
+
+API_DIR="$(cd "$(dirname "$0")/.." && pwd)"
+PY="${P5_PYTHON:-python3}"
+
+pg() { echo "$SUDO_PASSWORD" | sudo -S -p '' -u postgres psql "$@"; }
+
+pg -c "DROP DATABASE IF EXISTS $DB;" -c "DROP ROLE IF EXISTS $ROLE;"
+pg -c "CREATE ROLE $ROLE LOGIN PASSWORD '$P5_DB_PASSWORD';" -c "CREATE DATABASE $DB OWNER $ROLE;"
+pg -d "$DB" -c "CREATE EXTENSION IF NOT EXISTS pgcrypto;"
+
+export DATABASE_URL="postgresql://$ROLE:$P5_DB_PASSWORD@$PGHOST/$DB"
+(cd "$API_DIR" && "$PY" -m app.migrate)
+
+echo "isolated database ready: $DB (role $ROLE)"