xinyouhe/ai-kefu-admin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 装配切换——main/服务构造线/conftest 迁至 Registry（含默认租户过渡兜底）

Browse Source 
- main.py: 删全局 engine/async_session;lifespan 装配 TenantRegistry
  (seed_profiles → 仓库根 tenants/,路径锚定),预热默认租户 jiancai 挂
  app.state.default_runtime,逐 active 租户(库文件已存在)跑
  sweep_stale_pending,退出 dispose_all。app.state.registry 预置即沿用
  (测试注入缝)
- routes/runtime.py(过渡,Task 7 删): get_runtime——state.tenant 优先,
  缺省落 default_runtime;chat.py/chat_rest.py 全部改走它,消除
  main↔routes 循环依赖
- ConversationManager 构造线: 新增 runtime 可选参(持 session_factory/
  compiled_wall/config 引用;prompts 仍模块级,Task 6;业务接线 Task 9),
  fire_pending_pushes 随任务传 session_factory(后台写回派发租户的库)
- lead_push_service: 去 app.main import;写库入口显式收 session_factory,
  模块级 async_session=None 仅作测试注入兜底
- conftest: client 夹具 = 每测试独立 Registry(tmp data_dir)+ 真实
  lifespan 在 TestClient 循环内预热;删 _dispose_main_engine;
  async_db 保持 NullPool(跨 asyncio.run 循环安全,刻意不换 Registry 引擎)
- 测试仅动夹具/桩签名(断言零改动): wiring 桩补 session_factory 形参;
  history-404 用例改经真实 WS 路由造 widget 会话(引擎常驻连接绑定
  TestClient 循环,测试线程不得直开 session_factory)

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-10-12 19:15:46 +08:00
parent 0c296ca1ba
commit d041077bb6
11 changed files with 198 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/engine/services/conversation.py
+19 -3
View File
@@ -3,6 +3,7 @@ import re
import uuid
from collections.abc import AsyncGenerator
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
@@ -26,6 +27,9 @@ from app.engine.services.contact_pipeline import (
normalize_contacts, resolve_by_recency, scan_user_messages, split_validate,
)
if TYPE_CHECKING:
from app.platform.registry import TenantRuntime
logger = logging.getLogger(__name__)
# 职能边界触发词(Task 13.2, 评审#1/#4 修订): 命中即注入红线提醒。精修说明:
@@ -323,9 +327,20 @@ def _boundary_rotate() -> int:
class ConversationManager:
def __init__(self, db: AsyncSession, ai_service: AIService | None = None):
def __init__(self, db: AsyncSession, ai_service: AIService | None = None,
runtime: "TenantRuntime | None" = None):
self.db = db
self.ai = ai_service or AIService()
# Task 5b 装配线注入: 持有租户运行时引用(session_factory/compiled_wall/
# config 就位;prompts 仍为模块级,Task 6 才 PromptProvider 化;email/push
# 等 settings→config 业务接线是 Task 9)。None = 无租户上下文(直连测试)。
self.runtime = runtime
@property
def session_factory(self):
"""推送后台任务写结果用的工厂(写回派发租户的库);无 runtime 时
None,lead_push_service 回退模块级兜底。"""
return self.runtime.session_factory if self.runtime is not None else None
async def create_conversation(
self,
@@ -777,8 +792,9 @@ class ConversationManager:
await self.db.commit()
# 询盘推送(设计 8.1): 主事务提交后才派生后台任务——payload 内存
# 传递,任务不读库只写结果;fire 为同步函数,纳秒级返回不阻塞
lead_push_service.fire_pending_pushes(pending_pushes)
# 传递,任务不读库只写结果;fire 为同步函数,纳秒级返回不阻塞。
# session_factory 随任务传递: 写回本租户的库(Task 5b)
lead_push_service.fire_pending_pushes(pending_pushes, self.session_factory)
yield {
"type": "evaluated",
app/engine/services/lead_push_service.py
+34 -16
View File
@@ -7,7 +7,8 @@
- sweep_stale_pending: lifespan 启动清扫(崩溃遗留的 pending → failed)。
后台任务红线(设计 8.1): 不读库只写结果,payload 内存传递;
结果写入用独立 session(from app.main import async_session,照 chat_rest 模式)。
结果写入用独立 session——session_factory 由派发方传入(Task 5b 起无全局
单例,写回派发它的那个租户的库)。
"""
import asyncio
import logging
@@ -18,7 +19,6 @@ from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from app.engine.config import settings
from app.main import async_session
from app.engine.models import (
Conversation, CustomerProfile, Evaluation, IntentLevel, LeadPushLog,
Message, MessageRole,
@@ -28,6 +28,12 @@ from app.engine.services.ip_service import resolve_country
logger = logging.getLogger(__name__)
# Task 5b 装配切换: app.main 全局单例已删,写库入口改为显式传 session_factory
# (租户正确: 后台任务写结果必须落回派发租户的库)。模块级 async_session 仅作
# 无工厂上下文时的兜底(测试直接驱动 _run_push/sweep_stale_pending 时注入),
# 生产装配不设值——装配线(main.lifespan/ConversationManager)总是显式传。
async_session = None
# 对方契约(docs/AI_INQUIRY_PUSH_API.md): phone/whatsapp/email 至少一个有效
_TRIO = ("phone", "whatsapp", "email")
_CST = timezone(timedelta(hours=8))
@@ -207,9 +213,10 @@ def _retryable(resp: httpx.Response | None) -> bool:
return resp is None or resp.status_code >= 500
async def _update_log(push_log_id: str, **fields) -> None:
"""后台任务写结果——独立 session + 小事务(设计 6.3)。"""
async with async_session() as session:
async def _update_log(session_factory, push_log_id: str, **fields) -> None:
"""后台任务写结果——独立 session + 小事务(设计 6.3)。
session_factory: 派发租户的工厂(Task 5b),写回同一租户库。"""
async with session_factory() as session:
await session.execute(
update(LeadPushLog)
.where(LeadPushLog.id == push_log_id)
@@ -218,11 +225,14 @@ async def _update_log(push_log_id: str, **fields) -> None:
await session.commit()
async def _run_push(push_log_id: str, payload: dict) -> None:
async def _run_push(push_log_id: str, payload: dict,
session_factory=None) -> None:
"""后台任务体: 分类重试 + 200 逐条结果解析(设计 6.3/§10)。
红线: 全套 try/except,异常只记日志绝不逃逸(fire-and-forget)。
session_factory 缺省回退模块级 async_session(测试注入缝)。
"""
factory = session_factory or async_session
try:
attempts = 0
while True:
@@ -230,7 +240,7 @@ async def _run_push(push_log_id: str, payload: dict) -> None:
resp, transport_error = await _do_post(payload)
if _retryable(resp) and attempts < settings.PUSH_MAX_ATTEMPTS:
await _update_log(
push_log_id, attempts=attempts,
factory, push_log_id, attempts=attempts,
error_message=transport_error
or (f"http {resp.status_code}" if resp is not None else "unknown"),
)
@@ -245,7 +255,8 @@ async def _run_push(push_log_id: str, payload: dict) -> None:
except ValueError:
# 非 JSON 的 200(如网关重定向到 HTML 登录页)——
# 无法确认对方已接受,诚实落 failed(幂等键保证补推安全)
await _update_log(push_log_id, attempts=attempts, status="failed",
await _update_log(factory, push_log_id, attempts=attempts,
status="failed",
response_status=200,
error_message="non-JSON 200 body")
return
@@ -257,7 +268,7 @@ async def _run_push(push_log_id: str, payload: dict) -> None:
]
if failed_mine:
await _update_log(
push_log_id, attempts=attempts, status="failed",
factory, push_log_id, attempts=attempts, status="failed",
response_status=200,
error_message=str(failed_mine[0].get("reason") or "rejected"),
)
@@ -266,7 +277,7 @@ async def _run_push(push_log_id: str, payload: dict) -> None:
outcome = ("duplicated"
if (data.get("duplicated") or 0) > 0 else "accepted")
await _update_log(
push_log_id, attempts=attempts, status="success",
factory, push_log_id, attempts=attempts, status="success",
response_status=200, outcome=outcome,
)
return
@@ -277,7 +288,7 @@ async def _run_push(push_log_id: str, payload: dict) -> None:
"密钥/报文问题需人工介入,不重试",
push_log_id, resp.status_code)
await _update_log(
push_log_id, attempts=attempts, status="failed",
factory, push_log_id, attempts=attempts, status="failed",
response_status=resp.status_code if resp is not None else None,
error_message=transport_error
or (f"http {resp.status_code}" if resp is not None else "unknown"),
@@ -290,27 +301,34 @@ async def _run_push(push_log_id: str, payload: dict) -> None:
_tasks: set[asyncio.Task] = set()
def fire_pending_pushes(pushes: list[LeadPushLog]) -> None:
def fire_pending_pushes(pushes: list[LeadPushLog], session_factory=None) -> None:
"""主事务 commit 之后调用(设计 8.1): payload 内存传递,任务不读库。
同步函数、纳秒级返回——这就是"不阻塞"的派发点。
同步函数、纳秒级返回——这就是"不阻塞"的派发点。session_factory =
派发租户的工厂,随任务传递(后台写结果落回同一租户库);缺省回退
模块级 async_session(测试注入缝)。
"""
factory = session_factory or async_session
for log in pushes:
if log.status != "pending":
continue
task = asyncio.create_task(_run_push(log.id, log.request_payload))
task = asyncio.create_task(
_run_push(log.id, log.request_payload, factory))
_tasks.add(task)
task.add_done_callback(_tasks.discard)
async def sweep_stale_pending(stale_minutes: int = 30) -> int:
async def sweep_stale_pending(stale_minutes: int = 30, session_factory=None) -> int:
"""启动清扫(设计 8.2): 崩溃遗留的 pending → failed。
session_factory: 目标租户的工厂(lifespan 逐租户调用);缺省回退
模块级 async_session(测试注入缝)。
已知限制(设计 §14): 推送成功但结果写回失败也会被标 failed——
幂等键保证将来补推安全(对方返回 duplicated),概率极低,接受。
"""
factory = session_factory or async_session
cutoff = datetime.now(timezone.utc) - timedelta(minutes=stale_minutes)
async with async_session() as session:
async with factory() as session:
result = await session.execute(
update(LeadPushLog).where(
LeadPushLog.status == "pending",