diff --git a/services/api/app/main.py b/services/api/app/main.py
index 032e110..3df86de 100644
--- a/services/api/app/main.py
+++ b/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))
diff --git a/services/api/requirements-test.txt b/services/api/requirements-test.txt
new file mode 100644
index 0000000..cfa67d8
--- /dev/null
+++ b/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
diff --git a/services/api/tests/README.md b/services/api/tests/README.md
new file mode 100644
index 0000000..b33a635
--- /dev/null
+++ b/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`。
diff --git a/services/api/tests/phase5_exception_matrix.py b/services/api/tests/phase5_exception_matrix.py
new file mode 100644
index 0000000..0404be4
--- /dev/null
+++ b/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())
diff --git a/services/api/tests/provision_isolated_env.sh b/services/api/tests/provision_isolated_env.sh
new file mode 100644
index 0000000..bdca648
--- /dev/null
+++ b/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)"