feat: T-06 chat 最小闭环——api/chat(X-Agent-Type 分流+准入矩阵+customer 归属固定本人/advisor G-01 归属校验+SessionGuard actor/agent 一致+closed 409) + session_repository(agent_session/agent_message, 派生表补 alias 修 MySQL 1248) + memory_service(Redis 窗口 sess:{agent}:{id}:msgs TTL2h N20, miss/异常回源 MySQL) + main 挂载 + tests/test_chat 12 例, 260 绿
This commit is contained in:
@@ -54,6 +54,21 @@ SQLITE_TABLES: dict[str, str] = {
|
||||
raw_excerpt VARCHAR(1024), action VARCHAR(16),
|
||||
created_at {_TS})
|
||||
""",
|
||||
"agent_session": f"""
|
||||
CREATE TABLE agent_session (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT, session_id VARCHAR(64) UNIQUE,
|
||||
trace_id VARCHAR(64), agent_type VARCHAR(16), actor_id VARCHAR(64),
|
||||
actor_role VARCHAR(32), customer_id VARCHAR(64), advisor_id VARCHAR(64),
|
||||
title VARCHAR(256), status VARCHAR(16) DEFAULT 'active',
|
||||
created_at {_TS}, closed_at TIMESTAMP)
|
||||
""",
|
||||
"agent_message": f"""
|
||||
CREATE TABLE agent_message (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT, session_id VARCHAR(64),
|
||||
trace_id VARCHAR(64), seq_no INTEGER, role VARCHAR(16),
|
||||
content TEXT, has_disclaimer TINYINT DEFAULT 0, token_est INTEGER,
|
||||
created_at {_TS})
|
||||
""",
|
||||
"risk_alert": f"""
|
||||
CREATE TABLE risk_alert (
|
||||
alert_id VARCHAR(64) PRIMARY KEY, trace_id VARCHAR(64), customer_id VARCHAR(64),
|
||||
|
||||
@@ -0,0 +1,222 @@
|
||||
"""T-06 chat 最小闭环（会话/归属/SessionGuard/窗口/落盘）。
|
||||
|
||||
走 main app 真路由栈（audit+trace 中间件）；仓储注入 sqlite（含
|
||||
agent_session/agent_message）；Redis 注入 FakeRedis（窗口读写可断言，
|
||||
异常路径验证降级）；agent_service 走无 key 降级路径（不依赖外网）。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import text
|
||||
|
||||
from _ddl import create_sqlite_engine
|
||||
|
||||
from app.api import audit_middleware as audit_mod
|
||||
from app.api import chat as chat_mod
|
||||
from app.api import deps as deps_mod
|
||||
from app.api import risk as risk_api
|
||||
from app.main import app
|
||||
from app.repository.core_ro import CoreReadOnlyRepository
|
||||
from app.repository.risk_repository import RiskRepository
|
||||
from app.repository.session_repository import SessionRepository
|
||||
from app.service import memory_service
|
||||
from app.service.risk import redis_gateway
|
||||
|
||||
|
||||
class FakeRedis:
|
||||
"""窗口语义：rpush/lrange/ltrim/expire 最小实现 + 可注入故障。"""
|
||||
|
||||
def __init__(self):
|
||||
self.lists: dict[str, list[str]] = {}
|
||||
self.ttls: dict[str, int] = {}
|
||||
self.fail = False
|
||||
|
||||
def _maybe_fail(self):
|
||||
if self.fail:
|
||||
raise ConnectionError("redis down")
|
||||
|
||||
def rpush(self, key, *vals):
|
||||
self._maybe_fail()
|
||||
self.lists.setdefault(key, []).extend(vals)
|
||||
|
||||
def lrange(self, key, start, end):
|
||||
self._maybe_fail()
|
||||
lst = self.lists.get(key, [])
|
||||
return lst[start:] if end == -1 else lst[start : end + 1]
|
||||
|
||||
def ltrim(self, key, start, end):
|
||||
lst = self.lists.get(key, [])
|
||||
self.lists[key] = lst[start:] if end == -1 else lst[start : end + 1]
|
||||
|
||||
def expire(self, key, ttl):
|
||||
self._maybe_fail()
|
||||
self.ttls[key] = ttl
|
||||
|
||||
def publish(self, *a, **k):
|
||||
pass
|
||||
|
||||
def delete(self, *a, **k):
|
||||
pass
|
||||
|
||||
def exists(self, key):
|
||||
return False
|
||||
|
||||
def set_ex(self, *a, **k):
|
||||
pass
|
||||
|
||||
|
||||
CUSTOMER = {"X-Debug-Role": "customer", "X-Debug-Actor": "CUST-9527", "X-Agent-Type": "customer"}
|
||||
ADVISOR = {"X-Debug-Role": "advisor", "X-Debug-Actor": "STAFF-10086", "X-Agent-Type": "advisor"}
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def env(monkeypatch):
|
||||
engine = create_sqlite_engine()
|
||||
repo = RiskRepository(engine=engine)
|
||||
session_repo = SessionRepository(engine=engine)
|
||||
core_ro = CoreReadOnlyRepository(engine=engine)
|
||||
with engine.begin() as conn: # 归属种子（rbac-seed-reference 口径）
|
||||
conn.execute(
|
||||
text("INSERT INTO core_customer_advisor (advisor_id, customer_id, rel_status)"
|
||||
" VALUES ('STAFF-10086', 'CUST-9527', 'active'),"
|
||||
" ('STAFF-10087', 'CUST-1010', 'active')")
|
||||
)
|
||||
fake_redis = FakeRedis()
|
||||
monkeypatch.setattr(chat_mod, "_repo", lambda: repo)
|
||||
monkeypatch.setattr(chat_mod, "_session_repo", lambda: session_repo)
|
||||
monkeypatch.setattr(chat_mod, "_core_ro", lambda: core_ro)
|
||||
monkeypatch.setattr(memory_service, "_session_repo", lambda: session_repo)
|
||||
monkeypatch.setattr(risk_api, "_repo", lambda: repo)
|
||||
monkeypatch.setattr(audit_mod, "_repo", lambda: repo)
|
||||
monkeypatch.setattr(deps_mod, "RiskRepository", lambda: repo)
|
||||
monkeypatch.setattr(redis_gateway, "_gateway", fake_redis)
|
||||
yield {"client": TestClient(app), "repo": repo, "engine": engine, "redis": fake_redis}
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def _rows(engine, sql, **params):
|
||||
with engine.connect() as conn:
|
||||
return [dict(r) for r in conn.execute(text(sql), params).mappings().all()]
|
||||
|
||||
|
||||
def test_chat_new_session_creates_session_and_messages(env):
|
||||
r = env["client"].post("/api/chat", json={"message": "查一下我的持仓"}, headers=CUSTOMER)
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["session_id"].startswith("sess-")
|
||||
assert body["agent_type"] == "customer"
|
||||
assert body["customer_id"] == "CUST-9527"
|
||||
assert body["has_disclaimer"] is True # 客户对外口径（降级回复同样经 guard）
|
||||
assert body["trace_id"] == r.headers["X-Trace-Id"]
|
||||
|
||||
sessions = _rows(env["engine"], "SELECT * FROM agent_session")
|
||||
assert len(sessions) == 1
|
||||
s = sessions[0]
|
||||
assert (s["actor_id"], s["agent_type"], s["actor_role"], s["customer_id"]) == (
|
||||
"CUST-9527", "customer", "customer", "CUST-9527"
|
||||
)
|
||||
assert s["trace_id"] == body["trace_id"] # 手册 §9：会话绑定 trace_id
|
||||
msgs = _rows(
|
||||
env["engine"],
|
||||
"SELECT role, has_disclaimer FROM agent_message ORDER BY seq_no",
|
||||
)
|
||||
assert [(m["role"], m["has_disclaimer"]) for m in msgs] == [("user", 0), ("assistant", 1)]
|
||||
|
||||
|
||||
def test_chat_resume_appends_messages(env):
|
||||
sid = env["client"].post("/api/chat", json={"message": "第一句"}, headers=CUSTOMER).json()["session_id"]
|
||||
r = env["client"].post("/api/chat", json={"message": "第二句", "session_id": sid}, headers=CUSTOMER)
|
||||
assert r.status_code == 200 and r.json()["session_id"] == sid
|
||||
msgs = _rows(env["engine"], "SELECT seq_no, role FROM agent_message ORDER BY seq_no")
|
||||
assert [m["role"] for m in msgs] == ["user", "assistant", "user", "assistant"]
|
||||
assert [m["seq_no"] for m in msgs] == [1, 2, 3, 4]
|
||||
|
||||
|
||||
def test_chat_redis_window_written(env):
|
||||
r = env["client"].post("/api/chat", json={"message": "hi"}, headers=CUSTOMER)
|
||||
sid = r.json()["session_id"]
|
||||
key = f"sess:customer:{sid}:msgs"
|
||||
items = [json.loads(x) for x in env["redis"].lists[key]]
|
||||
assert [(m["role"], m["content"]) for m in items] == [("user", "hi"), ("assistant", items[1]["content"])]
|
||||
assert env["redis"].ttls[key] == 2 * 3600
|
||||
|
||||
|
||||
def test_chat_redis_down_degrades(env, monkeypatch):
|
||||
"""Redis 异常降级：窗口读写失败不阻塞对话（MySQL 为权威）。"""
|
||||
env["redis"].fail = True
|
||||
r = env["client"].post("/api/chat", json={"message": "hi"}, headers=CUSTOMER)
|
||||
assert r.status_code == 200
|
||||
|
||||
|
||||
def test_chat_session_of_other_actor_denied(env):
|
||||
sid = env["client"].post("/api/chat", json={"message": "第一句"}, headers=CUSTOMER).json()["session_id"]
|
||||
other = {**CUSTOMER, "X-Debug-Actor": "CUST-1001"}
|
||||
r = env["client"].post("/api/chat", json={"message": "续聊", "session_id": sid}, headers=other)
|
||||
assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_SESSION_AGENT"
|
||||
assert _rows(env["engine"], "SELECT 1 FROM audit_log WHERE decision = 'forbidden'")
|
||||
|
||||
|
||||
def test_chat_session_of_other_agent_type_denied(env):
|
||||
sid = env["client"].post("/api/chat", json={"message": "第一句"}, headers=CUSTOMER).json()["session_id"]
|
||||
r = env["client"].post(
|
||||
"/api/chat",
|
||||
json={"message": "续聊", "session_id": sid},
|
||||
headers={**ADVISOR, "X-Agent-Type": "advisor"},
|
||||
)
|
||||
assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_SESSION_AGENT"
|
||||
|
||||
|
||||
def test_chat_customer_forged_customer_id_denied(env):
|
||||
r = env["client"].post(
|
||||
"/api/chat", json={"message": "hi", "customer_id": "CUST-1001"}, headers=CUSTOMER
|
||||
)
|
||||
assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_NOT_OWNER"
|
||||
|
||||
|
||||
def test_chat_advisor_assigned_customer_ok(env):
|
||||
r = env["client"].post(
|
||||
"/api/chat", json={"message": "客户情况如何", "customer_id": "CUST-9527"}, headers=ADVISOR
|
||||
)
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["customer_id"] == "CUST-9527"
|
||||
assert body["has_disclaimer"] is False # 内部角色无免责声明
|
||||
|
||||
|
||||
def test_chat_advisor_not_assigned_denied(env):
|
||||
r = env["client"].post(
|
||||
"/api/chat", json={"message": "hi", "customer_id": "CUST-1010"}, headers=ADVISOR
|
||||
)
|
||||
assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_NOT_ASSIGNED"
|
||||
|
||||
|
||||
def test_chat_agent_type_missing_or_invalid(env):
|
||||
r = env["client"].post(
|
||||
"/api/chat", json={"message": "hi"}, headers={"X-Debug-Role": "customer", "X-Debug-Actor": "CUST-9527"}
|
||||
)
|
||||
assert r.status_code == 401 and r.json()["error_code"] == "AUTH_401_MISSING_AGENT_TYPE"
|
||||
r2 = env["client"].post("/api/chat", json={"message": "hi"}, headers={**CUSTOMER, "X-Agent-Type": "ops"})
|
||||
assert r2.status_code == 400
|
||||
|
||||
|
||||
def test_chat_agent_mismatch_debug_channel(env):
|
||||
"""手册 §5.4：advisor 角色不能进 customer agent（debug 通道同样强制）。"""
|
||||
r = env["client"].post(
|
||||
"/api/chat", json={"message": "hi"}, headers={**ADVISOR, "X-Agent-Type": "customer"}
|
||||
)
|
||||
assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_AGENT_MISMATCH"
|
||||
|
||||
|
||||
def test_chat_session_not_found_and_closed(env):
|
||||
r = env["client"].post("/api/chat", json={"message": "hi", "session_id": "sess-nope"}, headers=CUSTOMER)
|
||||
assert r.status_code == 404
|
||||
|
||||
sid = env["client"].post("/api/chat", json={"message": "hi"}, headers=CUSTOMER).json()["session_id"]
|
||||
with env["engine"].begin() as conn:
|
||||
conn.execute(text("UPDATE agent_session SET status = 'closed' WHERE session_id = :s"), {"s": sid})
|
||||
r2 = env["client"].post("/api/chat", json={"message": "hi", "session_id": sid}, headers=CUSTOMER)
|
||||
assert r2.status_code == 409 and r2.json()["error_code"] == "STATE_CONFLICT"
|
||||
@@ -64,6 +64,7 @@ def test_all_routers_mounted(client):
|
||||
"/api/risk/suitability/check",
|
||||
"/api/risk/aml/scan",
|
||||
"/api/simulate/trade",
|
||||
"/api/chat",
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user