feat: add multi-tenant IM agent gateway - #331
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #331 +/- ##
==========================================
Coverage ? 88.58687%
==========================================
Files ? 518
Lines ? 50328
Branches ? 0
==========================================
Hits ? 44584
Misses ? 5744
Partials ? 0 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
AI Code Review审查结论不通过 审查范围：base 1aa44cf..head 54a06da（37 个新文件、4058 行新增）。变更实现了一个多租户 IM 网关示例（Telegram/企业微信适配、HMAC 幂等、会话租约、事务化 outbox、月度预算治理、审计/指标/Otel、Docker/K8s 部署）。审查覆盖全部新增 Python 代码、迁移、测试、脚本与部署清单，并核对了 tRPC-Agent SDK 的 Runner/Filter/Session API 签名。 计划符合性：功能结构完整、测试意图清晰（验签、路由、幂等、租约、outbox 恢复均有覆盖），但存在多处与文档承诺不符的实现缺口。 主要风险：
未发现问题的方面：WeCom 签名算法与官方规范一致；SQL 全部参数化；租约过期恢复（outbox 门禁结论：FAILED——存在 5 个 SEVERE 高置信问题（其中 2 个导致在线部署无法启动或必然重复投递）。建议修复依赖声明、失败重试路径的 outbox 复核、 发现的问题严重问题： 新增的 触发条件： 按 README/ 实际影响： 网关容器启动即崩溃并持续重启（crash-loop），在线部署全部不可用；本仓库 CI 测试通过只是因为 修正方向： 在 严重
问题： 事件已被 触发条件： 消息在 实际影响： 用户在同一聊天收到两条相同（或仅因 LLM 非确定性而不同）的回复，并额外发生一次模型调用；重复投递违背幂等设计（README 宣称「回复与 Outbox 同事务提交；投递失败后台重试」）。 修正方向： 严重
问题： 工作进程在处理中崩溃（租约已获取、 触发条件： 任意 Worker 在 实际影响： 服务端以 HTTP 200 修正方向： 对超时/崩溃遗留的 严重
问题： 触发条件： 运行 实际影响： 每条回调都被当作「已忽略」以 202 应答，Telegram/WeCom 停止重试，消息被静默丢弃——无审计行、无指标、无日志，配置错误被完全掩盖。 修正方向： 将 严重
问题： 审计字段 触发条件： 租户一次对话中工具名拼接后超过 128 字符（如 5 个以上常规工具名），且控制面使用 MySQL（compose/K8s 生产路径）；sqlite 测试环境不报错，CI 无法发现。 实际影响： 消息实际已由 outbox 投递成功，但审计写库失败 → 外层 修正方向： 在 中等
问题： 触发条件： 任何使用 实际影响： 审计/月度预算统计系统性低估实际消耗（输出 token 完全不计）， 修正方向： 在 中等
问题： 仅工具调用回合（模型只返回 触发条件： 租户 agent 以纯工具调用结束一轮对话（工具执行结果未引发文本回复），或模型拒答无文本输出。 实际影响： 修正方向： 在投递前丢弃空回复（ 中等
问题： 触发条件： 攻击者向任一 webhook 路径并发 POST 数十 MB 至数百 MB 的请求体（无需有效签名）。 实际影响： 每个请求在判 413 之前就已完整驻留内存，并发下可耗尽网关内存导致 OOM/拒绝服务；大请求体占用的连接带宽也放大攻击面到所有 Worker。 修正方向： 在读取 body 前依据 中等
问题： 触发条件： Telegram 机器人收到任一用户对已处理消息的编辑（更正错别字、改 caption），机器人启用 实际影响： 每次编辑都会重新执行一次模型调用并再次投递回复，群聊中编辑会与上一条回复重复/矛盾，产生额外成本与噪音。 修正方向： 明确忽略 较低
问题： 新增的 触发条件： 推送本提交后 CI 的 test job 按 实际影响： 新增的 544 行测试（覆盖验签、幂等、租约、outbox 恢复等核心行为）不被任何自动化执行，本次变更的关键回归保护形同虚设。 修正方向： 在 CI 中补充运行 较低
问题： 完整的回复文本被明文写入 触发条件： 在线租户使用默认 实际影响： 数据库/文件系统可读者可获取全部助手回复明文（含 outbox 投递负载的复本），隐私承诺与合规基线（如日志脱敏）被削弱。 修正方向： 若保留重放能力需要，至少对 |
| multi-tenant-im = [ | ||
| "alembic>=1.13.0", | ||
| ] |
There was a problem hiding this comment.
问题： 新增的 pyproject.toml multi-tenant-im extra 仅声明 alembic>=1.13.0，核心 dependencies 也缺少 fastapi，而 examples/multi_tenant_im_agent/app.py 在模块导入时执行 from fastapi import FastAPI, Header, HTTPException, Request。Dockerfile 只执行 pip install ".[multi-tenant-im]"，因此镜像内启动网关必然抛 ModuleNotFoundError: No module named 'fastapi'。
触发条件： 按 README/docker-compose.yml(gateway 服务)或 deploy/kubernetes.yaml(Deployment)构建并启动镜像，python -m examples.multi_tenant_im_agent.main 在导入阶段即失败；fastapi 仅存在于 ag-ui/langfuse/openclaw/agent-claude/all extra 中，不会被安装。
实际影响： 网关容器启动即崩溃并持续重启（crash-loop），在线部署全部不可用；本仓库 CI 测试通过只是因为 requirements-test.txt 恰好含 fastapi，掩盖了发布缺陷。
修正方向： 在 multi-tenant-im extra 中加入 fastapi>=0.95.0（必要时一并补齐 pytest 等测试依赖），并在 CI 中增加一条使用 pip install ".[multi-tenant-im]" 后导入 examples.multi_tenant_im_agent.app 的冒烟检查。
| if ( | ||
| existing.status == "failed" | ||
| and existing.payload_hash == message.payload_hash() | ||
| ): | ||
| existing.status = "processing" | ||
| existing.error_type = None | ||
| return ClaimResult( | ||
| True, existing.status, existing.id, existing.sequence | ||
| ) |
There was a problem hiding this comment.
问题： 事件已被 complete_message 置为 completed 且 outbox 行已落库后，若后续的审计写入（service.py:238-251，位于内层 try 之外）或 mark_outbox_* 发生瞬时 DB 错误，会落入外层 except Exception → _handle_failure → mark_message_failed 把事件改成 failed；而 repository.claim_message 的失败重试路径（existing.status == "failed" 且 payload 匹配 → 置回 processing 并返回 accepted=True）完全不检查该事件是否已有 outbox 行，导致模型重跑并以相同消息再次插入第二条 outbox 行，两条行都被投递。
触发条件： 消息在 complete_message 提交成功后、审计/收尾写入失败（DB 瞬时不可用、tool_name 超长触发 MySQL 严格模式 DataError、进程在窗口内被取消）→ 返回 503 并标记 failed → IM 平台重试回调 → 重跑 agent → 生成第二条 outbox 行。
实际影响： 用户在同一聊天收到两条相同（或仅因 LLM 非确定性而不同）的回复，并额外发生一次模型调用；重复投递违背幂等设计（README 宣称「回复与 Outbox 同事务提交；投递失败后台重试」）。
修正方向： claim_message 的 failed 重试分支在重新置 processing 前，先检查该 session_id/事件是否已存在 OutboxRecord（pending/retry/sending 任一状态），若存在则不再重跑模型，直接按 duplicate 返回；或将失败事件与 outbox 行绑定清理后再允许重试。
| ) | ||
| if existing is not None: | ||
| if ( | ||
| existing.status == "failed" | ||
| and existing.payload_hash == message.payload_hash() | ||
| ): | ||
| existing.status = "processing" | ||
| existing.error_type = None | ||
| return ClaimResult( | ||
| True, existing.status, existing.id, existing.sequence | ||
| ) | ||
| return self._duplicate_result(existing, message.payload_hash()) |
There was a problem hiding this comment.
问题： 工作进程在处理中崩溃（租约已获取、claim_message 已写入 status="processing" 的事件行，但未走到 complete_message/mark_message_failed）后，该事件永远停留在 processing；claim_message 只对 status == "failed" 的事件重新受理，其余状态一律走 _duplicate_result 返回 accepted=False。
触发条件： 任意 Worker 在 runtime.reply 等长耗时阶段被 kill/被调度替换（k8s 滚动更新、OOM），会话租约在 session_lease_seconds（默认 120s）后过期；Telegram/WeCom 随后重试该回调，claim_message 命中 processing 行返回重复。
实际影响： 服务端以 HTTP 200 duplicate 应答，IM 平台认为已处理即停止重试：消息被永久丢失，无回复、无失败审计、无重试，且没有任何恢复机制。
修正方向： 对超时/崩溃遗留的 processing 事件增加恢复：基于租约过期时间把陈旧 processing 行视同失败（可重试）处理，或在 claim_message 中允许「processing 且旧的 owner 租约已过期」的事件重新受理。
| except InvalidCallbackError: | ||
| return ChannelResponse( | ||
| status_code=401, body={"ok": False, "error": "invalid_callback"} | ||
| ) | ||
| except ValueError as exc: | ||
| return ChannelResponse( | ||
| status_code=202, body={"ok": True, "ignored": type(exc).__name__} | ||
| ) |
There was a problem hiding this comment.
问题： ConfigurationError 继承自 ValueError（config.py:14），adapter.verify 内的 require_secret 抛出的配置错误会被 handle_webhook 的 except ValueError 吞掉并返回 HTTP 202 {"ok": true, "ignored": ...}；offline 模式下 build_app_from_env 完全跳过所有 binding 密钥的启动校验（if not offline: 才做 require_secret）。
触发条件： 运行 .env.local.example 之外的离线/评估场景而未设置 ACME_TELEGRAM_WEBHOOK_SECRET 等环境变量；或在生产运行时删除/轮换了某个 binding 的 secret 环境变量（启动后无重载路径）。
实际影响： 每条回调都被当作「已忽略」以 202 应答，Telegram/WeCom 停止重试，消息被静默丢弃——无审计行、无指标、无日志，配置错误被完全掩盖。
修正方向： 将 ConfigurationError 从 ValueError 吞并路径中排除（如先 except (InvalidCallbackError, UnsupportedMessageError)，再单独捕获并记录 ConfigurationError 后返回 500/503），并在 offline 模式也于启动时校验 webhook secret；或为配置缺失返回带审计的失败响应。
| user_id=actor_user_id, | ||
| session_id=session_id, | ||
| agent_name=tenant.agent_name, | ||
| tool_name=",".join(reply.tool_names), |
There was a problem hiding this comment.
问题： 审计字段 tool_name 为 String(128)（repository.py:213），但 service.py:245 以 ",".join(reply.tool_names) 无截断写入，tool_names 来自 SDK 函数调用名（runtime.py:171-172），长度不受控；同时该审计写入位于内层 try 之外，MySQL 8 默认 STRICT_TRANS_TABLES 下超长会抛 DataError。
触发条件： 租户一次对话中工具名拼接后超过 128 字符（如 5 个以上常规工具名），且控制面使用 MySQL（compose/K8s 生产路径）；sqlite 测试环境不报错，CI 无法发现。
实际影响： 消息实际已由 outbox 投递成功，但审计写库失败 → 外层 except Exception → 事件标记 failed → 平台重试 → claim_message 失败重试路径重跑模型 → 二次 outbox 行 → 用户收到重复回复，且每次成功消息都以 503 暴露为失败。
修正方向： 在 _audit/add_audit 入口对 tool_name 做 [:128] 截断（同 mark_message_failed 对 error_type 的处理），并在 CI 中补充 MySQL 严格模式下的审计写入测试。
| return AgentReply(text=text, tool_names=tuple(sorted(tools))) | ||
|
|
||
| return await asyncio.wait_for(collect(), timeout=tenant.model_timeout_seconds) | ||
|
|
There was a problem hiding this comment.
问题： TrpcAgentRuntime.reply 返回的 AgentReply 从不设置 token_count（默认 0），service.py:249 用 reply.token_count or decision.estimated_tokens 落审计，即月度预算的用量统计完全基于输入长度估计 (len(text)+3)//4，从不计入模型输出 token。
触发条件： 任何使用 TrpcAgentRuntime 的在线租户产生生成型负载（例如输入 40 字符、输出数千 token 的回复）。
实际影响： 审计/月度预算统计系统性低估实际消耗（输出 token 完全不计），monthly_token_budget 治理承诺失效——高生成量租户用量可能超出预算数倍而从不触发 403 monthly_token_budget_exceeded。
修正方向： 在 runtime.reply 中从 SDK 事件（usage/UsageMetadata 或 Event.usage）收集实际 token 数填入 AgentReply.token_count；无法获取时至少在文档中明确说明预算口径仅为估计值。
| tools.add(part.function_call.name) | ||
| elif part.text: | ||
| (chunks if event.partial else final_parts).append(part.text) | ||
| text = "".join(chunks) if chunks else "".join(final_parts) |
There was a problem hiding this comment.
问题： 仅工具调用回合（模型只返回 function_call parts、无文本）时 runtime.py:175 的 text 为空字符串，HttpChannelSender 仍以空文本调用 IM 平台（telegram 的 text[:4096] 为空、wecom 的 content 为空），平台返回 4xx 后进入 outbox 重试，且 mark_outbox_retry 无最大尝试次数/死信。
触发条件： 租户 agent 以纯工具调用结束一轮对话（工具执行结果未引发文本回复），或模型拒答无文本输出。
实际影响： claim_due_outbox 每 30~60s 无限重试同一空文本投递，用户永远收不到任何提示，attempts 无限增长且持续消耗平台 API 配额与网络流量；同步路径中空文本同样导致 delivery_queued=True。
修正方向： 在投递前丢弃空回复（if not reply.text.strip() 则直接 mark_outbox_sent/跳过投递，或回执平台「无回复」）；为 outbox 增加 max_attempts 上限与死信/告警机制。
| raw_body = await request.body() | ||
| if len(raw_body) > 1_048_576: |
There was a problem hiding this comment.
问题： app.py:85 先 raw_body = await request.body() 完整读入请求体，之后才检查 len(raw_body) > 1_048_576；/webhooks/{channel}/{account_id} 是未认证的公开端点（验签在读取之后才执行），uvicorn/FastAPI 无默认请求体上限。
触发条件： 攻击者向任一 webhook 路径并发 POST 数十 MB 至数百 MB 的请求体（无需有效签名）。
实际影响： 每个请求在判 413 之前就已完整驻留内存，并发下可耗尽网关内存导致 OOM/拒绝服务；大请求体占用的连接带宽也放大攻击面到所有 Worker。
修正方向： 在读取 body 前依据 Content-Length 头快速拒绝（超过 1MB 直接 413），或在读流过程中边读边计数并提前中止；也可由 Ingress 层统一限制请求体大小。
| ) -> InboundMessage: | ||
| try: | ||
| update = json.loads(raw_body) | ||
| message = update.get("message") or update.get("edited_message") |
There was a problem hiding this comment.
问题： TelegramAdapter.parse 对 update.get("edited_message") 与普通 message 同等受理（adapters.py:79），而 external_message_id 由 update_id:message_id 组成，编辑消息带有新的 update_id，幂等键必然不同；架构文档宣称的「重复消息不再次执行模型或工具」对编辑消息失效。
触发条件： Telegram 机器人收到任一用户对已处理消息的编辑（更正错别字、改 caption），机器人启用 edited_message 更新。
实际影响： 每次编辑都会重新执行一次模型调用并再次投递回复，群聊中编辑会与上一条回复重复/矛盾，产生额外成本与噪音。
修正方向： 明确忽略 edited_message（与文档「仅支持 text/caption 消息」一致），或将其视为对原消息的更新（如以 message_id 参与幂等键并触发「仅更新」语义）。
| from __future__ import annotations | ||
|
|
||
| import hashlib | ||
| import json | ||
| import time | ||
| from dataclasses import replace | ||
|
|
||
| import pytest | ||
| from fastapi.testclient import TestClient | ||
| from sqlalchemy import func, select |
There was a problem hiding this comment.
问题： 新增的 multi-tenant-im 示例的 tests/test_gateway.py、tests/test_migration_contract.py 无法在本仓库 CI 中运行：CI（.github/workflows/ci.yml）执行 pytest ... tests/，而测试放置于 examples/multi_tenant_im_agent/tests/ 不在该路径下；且测试依赖 fastapi（TestClient）、pytest-asyncio 等，仅部分在 requirements-test.txt 中。
触发条件： 推送本提交后 CI 的 test job 按 tests/ 目录收集用例。
实际影响： 新增的 544 行测试（覆盖验签、幂等、租约、outbox 恢复等核心行为）不被任何自动化执行，本次变更的关键回归保护形同虚设。
修正方向： 在 CI 中补充运行 pytest examples/multi_tenant_im_agent/tests -q 的步骤（并确保 fastapi、pytest-asyncio 依赖可用）。
| event.status = "completed" | ||
| event.response_text = reply.text | ||
| db.add( | ||
| OutboxRecord( | ||
| outbox_id=outbox_id, | ||
| tenant_id=tenant_id, | ||
| session_id=session_id, | ||
| channel=channel, | ||
| payload_json=json.dumps( |
There was a problem hiding this comment.
问题： 完整的回复文本被明文写入 mt_message_events.response_text（repository.py:494）与 mt_outbox.payload_json（repository.py:501-513），与模块自身的隐私设计（入站正文仅存 content_redacted=f"[text:N chars]"、审计仅存长度/哈希）不一致；ARCHITECTURE 宣称「日志不保存平台原始用户 ID」「消息正文不写审计表」。
触发条件： 在线租户使用默认 sqlite:///multi_tenant_im.db 或 MySQL 控制面处理敏感业务内容。
实际影响： 数据库/文件系统可读者可获取全部助手回复明文（含 outbox 投递负载的复本），隐私承诺与合规基线（如日志脱敏）被削弱。
修正方向： 若保留重放能力需要，至少对 response_text/payload_json 做字段级加密（使用 TENANT_NAMESPACE_SECRET 派生密钥）或仅存长度/引用；并在文档中明确该存储边界。
AI Code Review审查结论不通过 审查范围：54a06da（新增 37 文件、4392 行）+ 9297f26（加固 8 文件）+400 行，全部为 Python 网关示例。计划符合性：路由/多租户隔离/幂等/预算/outbox 等核心机制设计正确且有测试，但存在 1 个致命正确性缺陷（治理过滤器键不匹配导致所有模型调用被拒）、1 个异常误分类缺陷（治理拒绝被当作 ModelFailedSinceProvider 抛出）、1 个原子性缺陷（token 结算失败不回调导致永久丢失实际用量）、1 个配置缺陷（token_budget_period 未初始化）、1 个密码学校验缺陷（wecom 加密消息可绕过）、1 个投递一致性缺陷（同步+异步双轨竞争重复投递）。测试：现有测试覆盖了路由、幂等、隔离、预算、outbox 恢复，但缺少上述缺陷对应的失败路径测试。门禁结论：未通过（SEVERE 级缺陷存在）。 发现的问题严重
问题： 触发条件： 任一租户的任一合法消息进入 实际影响： 每次模型调用 修正方向： 将过滤器读取的键与网关写入的键统一（如统一为 严重
问题： 触发条件： 每个进程冷启动后的第一个租户请求（ 实际影响： 修正方向： 在 严重
问题： 治理拒绝（ 触发条件： 治理拒绝、预算拒绝或会话冲突返回后的任何后续异常（如 DB 暂时不可用导致的审计写入异常）。 实际影响： 租户被正常拒绝的消息被标记为 修正方向： 在 严重
问题： 触发条件： 模型调用完成后 实际影响： 本应在租户月预算中累计的实际 token 用量永久丢失；月预算统计低于真实消耗，租户可能超额使用而不被拦截。 修正方向： 将 中等
问题： 首次调用前 触发条件： 新部署数据库（含迁移创建的 实际影响： 修正方向： 在 中等
问题： 触发条件： 加密模式的 WeCom 回调中 实际影响： 加密消息可绕过签名验证并按明文 修正方向： 在 中等
问题： 触发条件： Telegram 实际影响： 同一回复被投递两次（用户收到重复消息）；后台重试无租户级速率限制，一个租户的重试风暴可消耗其他租户的投递配额。 修正方向： 为 outbox 增加按 较低
问题： 触发条件： 运维配置 实际影响： 弱密钥使会话/用户哈希可被暴力猜测，攻击者可伪造其他租户的 修正方向： 在 |
| # The gateway sets these only after signature, policy, budget, and | ||
| # account checks. Missing metadata therefore fails closed even if a | ||
| # future caller invokes Runner without going through the gateway. | ||
| tenant_id = ctx.get_metadata("tenant_id", "") | ||
| policy_approved = ctx.get_metadata("tenant_policy_approved", False) | ||
| if not tenant_id or not policy_approved: | ||
| rsp.error = PermissionError("tenant governance context is missing") | ||
| rsp.is_continue = False | ||
|
|
There was a problem hiding this comment.
问题： _before 过滤器读取的元数据键是 tenant_policy_approved 而网关写入的是 tenant_policy_approved（差异：第二个词首字母大写），两处键不匹配，过滤器永远认为治理授权缺失。
触发条件： 任一租户的任一合法消息进入 TrpcAgentRuntime.reply，new_agent_context(metadata={...}) 写入 tenant_policy_approved: True，随后 MultiTenantGovernanceFilter._before 通过 ctx.get_metadata("tenant_policy_approved", False) 取回默认 False。
实际影响： 每次模型调用 _before 都设置 rsp.error = PermissionError(...)，is_continue = False；BaseFilter._handle_co 在 result.rsp 存在时 yield result，但 run_stream_filters 的 stream_handler_adapter 只转发 FilterResult.rsp（此处为 None），因此 agent 运行正常结束且不产生任何事件——模型从未被调用，回复永远为空，对 200 响应但无内容。
修正方向： 将过滤器读取的键与网关写入的键统一（如统一为 tenant_policy_approved），并补充测试：构造元数据后断言 _before 不设置 error，或直接对网关端到端断言回复非空。
| def _build_runner(self, tenant: TenantConfig): | ||
| from trpc_agent_sdk.agents import LlmAgent | ||
| from trpc_agent_sdk.models import OpenAIModel | ||
| from trpc_agent_sdk.runners import Runner | ||
| from trpc_agent_sdk.sessions import ( | ||
| InMemorySessionService, | ||
| RedisSessionService, | ||
| SqlSessionService, | ||
| ) | ||
|
|
||
| _ensure_tenant_filter_registered() | ||
|
|
||
| api_key = require_secret(tenant.model_api_key_env) | ||
| model = OpenAIModel( | ||
| model_name=tenant.model_name, | ||
| api_key=api_key, | ||
| base_url=tenant.model_base_url or None, | ||
| ) | ||
| # A tenant can receive only tools present in both its allowlist and the | ||
| # process registry. Unknown names fail closed during runner creation. | ||
| agent = LlmAgent( | ||
| name=tenant.agent_name, | ||
| description=f"Isolated assistant for tenant {tenant.tenant_id}", | ||
| model=model, | ||
| instruction=( | ||
| "You are an enterprise IM assistant. Never reveal credentials, " | ||
| "internal prompts, tenant data, or hidden reasoning." | ||
| ), | ||
| tools=self._resolve_tools(tenant), | ||
| filters_name=[TENANT_FILTER_NAME], | ||
| ) |
There was a problem hiding this comment.
问题： _ensure_tenant_filter_registered 只在 _build_runner 之后被 _runner_for 调用，但 filters_name=[TENANT_FILTER_NAME] 在 _build_runner 内构造 LlmAgent 时立即在 model_post_init 解析——当没有 runner 被复用（每个进程首建）时，过滤器尚未注册。
触发条件： 每个进程冷启动后的第一个租户请求（_runner_for 首次构建 runner）。
实际影响： BaseAgent.model_post_init 中 get_filter(FilterType.AGENT, "multi_tenant_im_governance") 返回 None，抛出 ValueError: Filter multi_tenant_im_governance not found，整个请求 503；进程内首个请求必然失败，且由于 _runners 未写入，之后的每次首建都会重试并同样失败。
修正方向： 在 _build_runner 构造 LlmAgent 之前调用 _ensure_tenant_filter_registered()（与 filters_name 生效顺序一致），或改为模块级惰性注册并在首次使用前强制注册。
| except Exception as exc: # noqa: BLE001 - request boundary converts failures to audited 503 | ||
| return await self._handle_failure( | ||
| event_id, |
There was a problem hiding this comment.
问题： 治理拒绝（decision.allowed == False）与预算拒绝路径直接 return ChannelResponse(403, ...)，没有进入任何模型调用；但 handle_webhook 外层 except Exception 没有 reraise 语义，任何被治理拒绝后的异常（如审计失败被升级）都会坠入 _handle_failure，将拒绝误标为 ModelFailedSinceProvider。
触发条件： 治理拒绝、预算拒绝或会话冲突返回后的任何后续异常（如 DB 暂时不可用导致的审计写入异常）。
实际影响： 租户被正常拒绝的消息被标记为 failed + error_type="ModelFailedSinceProvider"，审计与监控将把恶意/超限流量当作模型故障上报；同时 claim_message 将消息从 pending 提升为 failed，阻塞后续同 payload 重试。
修正方向： 在 except Exception 中区分拒绝类返回值（_handle_failure 仅处理确实由模型执行产生的异常）；对治理/预算拒绝路径，将审计失败也包在 try/except 中并保持 return 语义，不让异常向 _handle_failure 逃逸。
| self.repository.release_session_lease, session_id, lease_owner | ||
| ) | ||
| if budget_reserved and not budget_settled: | ||
| await asyncio.to_thread( | ||
| self.repository.settle_token_budget, | ||
| tenant.tenant_id, | ||
| budget_period, | ||
| decision.estimated_tokens, | ||
| 0, |
There was a problem hiding this comment.
问题： finally 中 if budget_reserved and not budget_settled: 在模型/存储异常时只释放预留，不记录实际 token 用量；settle_token_budget 的公式 token_usage - reserved + max(0, actual) 在 reserved 已被上次调用清零（预留→结算→再预留→结算失败）等交错场景下会把实际用量清零或算错。
触发条件： 模型调用完成后 settle_token_budget 抛异常（DB 抖动），或模型返回 0 token 后再次调用。
实际影响： 本应在租户月预算中累计的实际 token 用量永久丢失；月预算统计低于真实消耗，租户可能超额使用而不被拦截。
修正方向： 将 settle_token_budget 改为幂等的「实际用量覆盖」语义（如 update token_usage = max(token_usage, actual) 或提供独立的 add_usage），并在 finally 中对 not budget_settled 时也尝试结算实际用量（而非仅释放预留），同时补充该失败路径的测试。
| def reserve_token_budget( | ||
| self, | ||
| tenant_id: str, | ||
| period: str, | ||
| estimated_tokens: int, | ||
| limit: int, | ||
| ) -> bool: | ||
| """Atomically reserve tenant budget before invoking a model.""" | ||
|
|
||
| with self.Session.begin() as db: | ||
| tenant = db.scalar( | ||
| select(TenantRecord) | ||
| .where(TenantRecord.tenant_id == tenant_id) | ||
| .with_for_update() | ||
| ) | ||
| if tenant is None: | ||
| return False | ||
| if tenant.token_budget_period != period: | ||
| tenant.token_budget_period = period | ||
| tenant.token_usage = 0 | ||
| if tenant.token_usage + estimated_tokens > limit: | ||
| return False | ||
| tenant.token_usage += estimated_tokens |
There was a problem hiding this comment.
问题： 首次调用前 TenantRecord.token_budget_period 从未被初始化（sync_tenants 不设置它，TenantRecord 的 default="" 只作用于新行），而 reserve_token_budget 第 429 行读取它作为重置条件，且服务在 token_budget_period != period 时重置 token_usage=0。
触发条件： 新部署数据库（含迁移创建的 token_budget_period NOT NULL 无默认值列）插入首条 TenantRecord；或进程重启后 period 仍为空。
实际影响： token_budget_period == "" 恒不等于任何 "YYYY-MM"，导致每次请求都重置 token_usage=0，月预算计数永不为满，租户可无限制使用 token；对 Field required 风格校验，首次写入可能直接被拒绝。
修正方向： 在 sync_tenants 或 ensure_session 等写路径初始化 token_budget_period（如 current_period() 计算当月 YYYY-MM），并在 migration 中将列默认值语义改为由应用填充。
| encrypted = "" | ||
| if query.get("msg_signature"): | ||
| encrypted = self._extract_encrypt(raw_body) | ||
| if not encrypted: | ||
| raise InvalidCallbackError( | ||
| "encrypted WeCom callback has no Encrypt field" | ||
| ) | ||
| pieces = [token, timestamp, nonce] | ||
| if encrypted: | ||
| pieces.append(encrypted) | ||
| expected = hashlib.sha1("".join(sorted(pieces)).encode("utf-8")).hexdigest() | ||
| if not hmac.compare_digest(signature, expected): |
There was a problem hiding this comment.
问题： verify 在 msg_signature 存在时先 _extract_encrypt，但 _extract_encrypt 解析失败返回 ""，且 if not encrypted: raise 成立——不过异常在验证前被 except (ValueError) 包装为 InvalidCallbackError，随后 raise 直接返回 401——这本身正确；但 _extract_encrypt 对 JSON/XML 解析失败返回 "" 后，verify 不校验 encrypted 非空，导致签名仅基于 [token, timestamp, nonce] 计算，且 parse 在无 Encrypt 字段时按明文解析。
触发条件： 加密模式的 WeCom 回调中 Encrypt 字段缺失或 ET 解析失败；或攻击者提供伪造的 msg_signature 而 body 无加密字段。
实际影响： 加密消息可绕过签名验证并按明文 Content 处理，租户敏感内容（如财务、HR）可能被当作普通文本发送给模型；攻击者可构造 body 通过验证。
修正方向： 在 _extract_encrypt 返回空时让 verify 直接失败（if query.get("msg_signature") and not encrypted: raise InvalidCallbackError(...)），并确保 parse 对含 Encrypt 字段的 body 一律拒绝（当前仅在非空时拒绝）。
| delivery_queued = False | ||
| try: | ||
| claimed = await asyncio.to_thread( | ||
| self.repository.claim_outbox, outbox_id | ||
| ) | ||
| if not claimed: | ||
| raise RuntimeError("outbox claim failed") | ||
| delivery_started = perf_counter() | ||
| try: | ||
| await self.sender.send(delivery) | ||
| finally: | ||
| self.metrics.observe_stage( | ||
| tenant.tenant_id, | ||
| "im_delivery", | ||
| (perf_counter() - delivery_started) * 1000, | ||
| ) | ||
| await asyncio.to_thread(self.repository.mark_outbox_sent, outbox_id) | ||
| self.metrics.observe_delivery(message.channel, "sent") | ||
| except Exception: # noqa: BLE001 - provider failures are persisted for retry | ||
| delivery_queued = True | ||
| retry_status = await asyncio.to_thread( | ||
| self.repository.mark_outbox_retry, outbox_id | ||
| ) | ||
| self.metrics.observe_delivery(message.channel, retry_status) | ||
|
|
There was a problem hiding this comment.
问题： handle_webhook 同步投递（claim_outbox → sender.send → mark_outbox_sent）与 _outbox_loop/dispatch_outbox_once 后台重试双轨并存：同步路径在 sender.send 抛异常时 mark_outbox_retry 入队，后台 2 秒后 claim_due_outbox 再次取出重试；同步投递成功但 mark_outbox_sent 前进程崩溃，也会留下 sending 行被后台恢复重打。
触发条件： Telegram sendMessage 成功但 HTTP 响应超时（Telegram 常见 1s 确认限制），或进程在 outbox 提交与标记之间崩溃。
实际影响： 同一回复被投递两次（用户收到重复消息）；后台重试无租户级速率限制，一个租户的重试风暴可消耗其他租户的投递配额。
修正方向： 为 outbox 增加按 (channel, account_id, conversation_id) 或消息内容的幂等投递键（如复用 payload_hash），在 sender.send 前检查；或让同步路径失败时仅标记 retry 且后台重试跳过已 sent/sending 且 next_attempt_at 未到的行。
| tenant.max_input_chars, | ||
| tenant.request_token_budget, | ||
| tenant.monthly_token_budget, | ||
| ) | ||
| <= 0 | ||
| ): | ||
| raise ConfigurationError("tenant input and token budgets must be positive") | ||
| for binding in tenant.bindings: | ||
| if ( | ||
| not binding.channel | ||
| or not binding.account_id | ||
| or not binding.webhook_secret_env | ||
| ): | ||
| raise ConfigurationError( | ||
| "channel, account_id and webhook_secret_env are required" | ||
| ) | ||
| if binding.channel.lower() == "telegram" and not binding.bot_token_env: |
There was a problem hiding this comment.
问题： TenantRegistry.replace（配置热加载的唯一入口）未调用 MultiTenantAgentService.__init__ 中的 len(namespace_secret) < 16 校验：短于 16 字符的命名空间密钥仅在被 __init__ 检查，而 namespace_secret 是 HMAC 派生（derive_session_id/derive_user_id/stable_subject_id）的密钥。
触发条件： 运维配置 TENANT_NAMESPACE_SECRET 为短值（如示例 .env.local 的 32 字符），或通过 replace() 热加载时传入弱密钥。
实际影响： 弱密钥使会话/用户哈希可被暴力猜测，攻击者可伪造其他租户的 session_id/user_id 进行会话接管；当前校验仅覆盖进程启动路径，热加载路径可绕过。
修正方向： 在 TenantRegistry.replace 或 MultiTenantAgentService 构造/热加载接口中统一校验密钥强度，并在文档中明确至少 32 字符要求。
AI Code Review审查结论不通过 审查范围：base 1aa44cf → head 80d7fbd，共 3 个提交（54a06da 初版 feat、9297f26 可靠性加固、efaaf44 真实模型冒烟测试、80d7fbd 文档），39 个文件、4532 行新增，全部为新增的 examples/multi_tenant_im_agent 多租户 IM Agent 网关示例。计划符合度：功能完整实现——租户注册/解析、Telegram/WeCom 适配、会话校验、模型调用（per-tenant runner + 租户过滤链）、事务性 outbox、token 预算原子预留/结算、审计与 Prometheus 指标、迁移与 ORM 契约一致（含 account_id 复合键隔离）；加固提交补齐了会话租约、失败可重试、outbox 死信与恢复、指标与审计失败降级，架构文档 §6 与代码基本吻合。主要风险：(1) token 预算在 outbox 提交前结算，complete_message/claim_outbox 等后续失败后事件被标记 failed 且预算不回滚，provider 重投会重跑模型并二次计费（SEVERE）；(2) 月边界翻转时 settle_token_budget 因 period 不匹配提前返回，遗留的预留额度永久占用当月预算（MODERATE）；(3) worker 崩溃遗留的 processing 状态消息事件被当作重复 200 应答，吞掉 provider 重投，用户永远收不到回复且无自动恢复（MODERATE）。测试充分性：25 个单测覆盖重复、冲突、账户隔离、预算原子预留、死信、审计降级与 outbox 恢复，但未覆盖上述三条失败路径（结算后提交失败、processing 卡死恢复、月边界翻转），且 CI（testpaths=["tests"]）不会运行 examples 下的测试。门禁结论：因存在 SEVERE 计费一致性问题，结论为 FAILED。 发现的问题严重
问题： 预算结算（ 触发条件： 模型调用成功后、 实际影响： 用户的 token 用量被二次计费：第一次结算已把实际用量计入 修正方向： 将 中等
问题： 触发条件： 请求在月末最后时刻发起（reserve 用旧 period）并在月初完成结算或失败回滚（settle/rollback 用新 period）。 实际影响： 租户当月预算被一笔从未实际消耗的预留额永久占用（ 修正方向： 结算/回滚时若 period 不匹配，按预留时记录的真实 period 对旧 period 的预留做扣减（例如返回累计结算失败后由调用方以 中等
问题： 触发条件： 处理中进程崩溃或异常中断，随后 provider 对未确认的 webhook 进行重投。 实际影响： 用户消息被静默丢弃——永远收不到回复，provider 也不再重试，只能人工修改数据库恢复；与加固提交宣称的故障恢复（outbox 自动恢复）相比，消息侧缺少同等保障，且 修正方向： 为 较低
问题： 新增的 触发条件： 后续修改预算结算顺序、 实际影响： 双计费、预留泄漏、消息丢失这三类缺陷（见同文件其余评论）缺少自动化保障；测试缺口使已确认的正确性风险无法被门禁拦截。 修正方向： 补充三类回归测试：(1) 打桩使 |
AI Code Review审查结论不通过 提交最终代码审查结果（FAILED，11 条评论） 发现的问题严重
问题： 触发条件： 任何真实 tRPC-Agent Runner 的流式回复（生产默认路径）：partial 增量事件填满 实际影响： 用户收到截断到“最后一个 partial 增量累积点”的回复，结尾内容永久丢失；若模型实现把 partial 事件做成累积全文（Anthropic 的 修正方向： 只取非 partial 事件的全文（ 严重
问题： 触发条件： 实际影响： 同一回复被后台循环在约 5-9 秒后（retry 退避）再次领取并二次发送，用户收到两条相同消息； 修正方向： 同步投递与后台循环按状态分工： 严重
问题： 触发条件： 实际影响： 认证与配置故障被伪装成 202 成功响应，IM 平台认为消息已处理而停止重试，用户消息静默丢失（无审计、无失败指标）；响应体还把异常类名 修正方向： 将 严重
问题： 触发条件： 两个线程/进程并发执行 实际影响： 默认部署下 outbox 重复投递、月度预算超额消费、session 消息序号竞态导致消息 503 丢失，行为与生产（MySQL/PG）不一致，单元测试全部跑 SQLite 因此从未暴露。 修正方向： 对 SQLite 部署显式声明不支持多进程并发（文档 + 启动断言），或对关键路径（claim_due_outbox、claim_outbox、reserve_token_budget）改用 SQLite 支持的条件更新（ 严重
问题： 触发条件： provider 不返回 usage（或返回真实 0 token）的任何成功回复：每月按输入字符数估算值（governance.py:27）累计扣费，与真实消耗无关。对比失败路径 finally（service.py:347-354）显式 实际影响： 月预算被估算值永久侵吞，未到月底真实请求遭 403 修正方向： 用 中等
问题： 全部时间列使用 触发条件： MySQL 部署（docker-compose.yml / kubernetes.yaml 默认）：同一秒内的审计日志/outbox 记录 实际影响： 审计时间线无法精确还原；outbox 重试排期与租约过期时间被秒级取整（MySQL 对小数位取整而非截断，可能提前 1 秒）导致重试节奏、租约语义偏差；且 SQLite 读回 naive、PG 读回 aware，跨方言 naive/aware 不一致。 修正方向： 迁移改用与 SDK 一致的 中等
问题： 触发条件： 模型超过 实际影响： 流式连接悬挂、会话事件未终结，连接与资源的泄漏随超时次数累积，最终可能打满 provider 连接池，连带正常租户请求失败；被取消的 runner 仍留在 修正方向： 在 中等
问题： 触发条件： 配置重载（重启/滚动发布）时同一 实际影响： 租户 A 该账号的回调瞬时路由到租户 B 的 Agent，A 用户消息进入 B 租户上下文（跨租户数据泄露）；会话/消息历史混入 B 的名下；反向（删除 binding）时旧行永不清理，遗留 修正方向： 命中已有 binding 且其 中等
问题： 触发条件： 任何人修改迁移或 ORM 模型（改列长、换类型、删约束），或 MySQL 方言下的建表/索引/精度问题（如 utf8mb4 索引字节超限、DATETIME 精度）。 实际影响： 迁移与模型漂移测试保持绿色——删除 修正方向： 用真实 SQLite + MySQL 方言执行 upgrade/downgrade（或 sqlalchemy-migrate 风格比对），断言列类型/长度/nullable/唯一约束/FK；并把 较低
问题： 触发条件： 网关监听 0.0.0.0（main.py 默认），任一可访问该端点的未认证调用者（或经 Telegram/WeCom 回调暴露的公网入口）访问 实际影响： 跨租户业务情报泄露：租户 ID、消息量、token 用量、模型花费全部可见，可用于针对特定租户机器人的定向滥用或计费侦查。 修正方向： 对 较低
问题： 触发条件： 消息量大的网关（outbox 循环单轮最多 50 条、webhook 内联投递并发）每条消息一次全新建连接。 实际影响： 到 api.telegram.org / 企业微信的出站延迟与对端连接压力成倍放大（每条消息一次握手，无 keep-alive 复用），高吞吐下易触发 IM 平台限流；纯连接池复用即可消除的成本。 修正方向： 将单个 |
| charged_tokens = reply.token_count or decision.estimated_tokens | ||
| await asyncio.to_thread( | ||
| self.repository.settle_token_budget, | ||
| tenant.tenant_id, | ||
| budget_period, | ||
| decision.estimated_tokens, | ||
| charged_tokens, | ||
| ) | ||
| budget_settled = True |
There was a problem hiding this comment.
问题： 预算结算（settle_token_budget + budget_settled = True）发生在 outbox 事务提交（complete_message）之前，之后任何失败路径（complete_message 抛异常、claim_outbox 返回 False 抛 RuntimeError、mark_outbox_retry 抛异常）都会进入 _handle_failure 把事件标记为 failed，但 finally 中的回滚条件 budget_reserved and not budget_settled 已不满足，预留用量既不被结算也不被释放。随后 provider 重投同样的 external_message_id 时，claim_message 会把 failed 且 payload 哈希一致的事件重置为 processing 并复用同一 event_id，重新预留、重新调用模型、重新结算，导致实际 token 用量被重复计入。
触发条件： 模型调用成功后、complete_message/claim_outbox 等 DB 操作失败（如数据库短暂不可用、连接被 pool_recycle 回收后的瞬时断连），且 provider 对 503 响应进行重投。
实际影响： 用户的 token 用量被二次计费：第一次结算已把实际用量计入 token_usage，重跑后再次预留并结算，最终 token_usage 多了约一倍的该请求实际消耗；月预算被提前耗尽，导致后续合法请求被误判为 monthly_token_budget_exceeded 拒绝，审计与指标中的 token_count 同样失真。
修正方向： 将 settle_token_budget 移到 complete_message 成功提交之后（或让 budget_settled = True 仅在 outbox 提交成功后置位），使提交失败时能走 finally 回滚预留；同时让 _handle_failure 对已结算但消息未提交的事件异步补一次释放结算，保证重投路径不会重复计费。
| (chunks if event.partial else final_parts).append(part.text) | ||
| text = "".join(chunks) if chunks else "".join(final_parts) |
There was a problem hiding this comment.
问题： collect() 按 event.partial 把文本分别追加到 chunks 与 final_parts，最后 text = "".join(chunks) if chunks else "".join(final_parts) 二选一。SDK 的 RunConfig.streaming 默认为 True（trpc_agent_sdk/configs/_run_config.py），runner.run_async 会产出全部 partial 增量事件，因此 chunks 几乎总是非空；而 partial=False 的终末事件携带该轮完整累积文本（OpenAI _openai_model.py:2362-2368 的 accumulated_text、Anthropic _anthropic_model.py:690-692 的 final_content），被 append 进 final_parts 后整体丢弃。
触发条件： 任何真实 tRPC-Agent Runner 的流式回复（生产默认路径）：partial 增量事件填满 chunks，终末全文事件进 final_parts，chunks 非空导致 final_parts 被忽略。
实际影响： 用户收到截断到“最后一个 partial 增量累积点”的回复，结尾内容永久丢失；若模型实现把 partial 事件做成累积全文（Anthropic 的 partial_content），chunks 会包含重复累积文本，甚至产生重复拼接内容。EchoRuntime 不经过此路径，故测试无法发现。
修正方向： 只取非 partial 事件的全文（if not event.partial and part.text: final_parts.append(...)，忽略 partial 文本），或改为按轮去重拼接；并补充 partial+final 全事件的单元测试。
| def settle_token_budget( | ||
| self, tenant_id: str, period: str, reserved: int, actual: int | ||
| ) -> None: | ||
| """Replace a reservation with actual usage, or release it on failure.""" | ||
|
|
||
| with self.Session.begin() as db: | ||
| tenant = db.scalar( | ||
| select(TenantRecord) | ||
| .where(TenantRecord.tenant_id == tenant_id) | ||
| .with_for_update() | ||
| ) | ||
| if tenant is None or tenant.token_budget_period != period: | ||
| return | ||
| tenant.token_usage = max(0, tenant.token_usage - reserved + max(0, actual)) |
There was a problem hiding this comment.
问题： settle_token_budget 在 tenant.token_budget_period != period 时直接 return，既不结算也不释放该 period 的预留。若请求跨越月边界（webhook 进入时 budget_period 为上个月、结算/回滚时服务端已翻转）或请求处理跨月（如模型超时上限一分钟量级），reserve_token_budget 已写入的 token_usage += estimated_tokens 将永久留在新月份中，finally 中的回滚调用同样因 period 不匹配被静默丢弃；token_budget_period 翻转只发生在下一次 reserve 时，当月内没有任何逻辑清理这笔残留。
触发条件： 请求在月末最后时刻发起（reserve 用旧 period）并在月初完成结算或失败回滚（settle/rollback 用新 period）。
实际影响： 租户当月预算被一笔从未实际消耗的预留额永久占用（token_usage 虚高，最多多出一次 estimated_tokens），可能导致后续请求被误判为超预算而拒绝；预算数据完整性受影响。
修正方向： 结算/回滚时若 period 不匹配，按预留时记录的真实 period 对旧 period 的预留做扣减（例如返回累计结算失败后由调用方以 reserved=0, actual=0 补偿，或在 TenantRecord 上记录当前进行中的预留 period），而不是直接 return。
| delivery_queued = False | ||
| try: | ||
| claimed = await asyncio.to_thread( | ||
| self.repository.claim_outbox, outbox_id | ||
| ) | ||
| if not claimed: | ||
| raise RuntimeError("outbox claim failed") | ||
| delivery_started = perf_counter() | ||
| try: | ||
| await self.sender.send(delivery) | ||
| finally: | ||
| self.metrics.observe_stage( | ||
| tenant.tenant_id, | ||
| "im_delivery", | ||
| (perf_counter() - delivery_started) * 1000, | ||
| ) | ||
| await asyncio.to_thread(self.repository.mark_outbox_sent, outbox_id) | ||
| self.metrics.observe_delivery(message.channel, "sent") | ||
| except Exception: # noqa: BLE001 - provider failures are persisted for retry | ||
| delivery_queued = True | ||
| retry_status = await asyncio.to_thread( | ||
| self.repository.mark_outbox_retry, outbox_id | ||
| ) | ||
| self.metrics.observe_delivery(message.channel, retry_status) |
There was a problem hiding this comment.
问题： complete_message（repository.py:530-568）写入的 outbox 行未显式设置 next_attempt_at，默认即插入时刻，立即可被后台 claim_due_outbox 选中（repository.py:585-608：status in (pending,retry,sending) AND next_attempt_at <= now）；webhook 路径随后 claim_outbox（只接受 pending/retry）返回 False → raise RuntimeError("outbox claim failed") → except 分支无条件 mark_outbox_retry 把行状态改回 retry。mark_outbox_retry 用 db.get 无锁读改写，还能把另一 worker 刚置的 sending/sent 状态覆盖为 retry。
触发条件： complete_message 的 await asyncio.to_thread 与 claim_outbox 的 to_thread 之间，_outbox_loop（app.py:26-36，每 2s）恰好执行 claim_due_outbox 抢走该行；K8s 3 副本（deploy/kubernetes.yaml:44）共享 MySQL 时必然出现。
实际影响： 同一回复被后台循环在约 5-9 秒后（retry 退避）再次领取并二次发送，用户收到两条相同消息；mark_outbox_retry 的无锁覆盖还可能把已 sent 的行改回 retry 造成第三次投递，im_delivery 无幂等键无法去重。
修正方向： 同步投递与后台循环按状态分工：claim_outbox 失败时不应走 mark_outbox_retry（后台循环会处理），改为直接返回 202 queued；或 complete_message 写入时把 next_attempt_at 设为 now + 若干秒，让后台循环不抢新行；mark_outbox_retry 改为 with_for_update 条件更新并校验当前状态。
| if existing is not None: | ||
| if ( | ||
| existing.status == "failed" | ||
| and existing.payload_hash == message.payload_hash() | ||
| ): | ||
| existing.status = "processing" | ||
| existing.error_type = None | ||
| return ClaimResult( | ||
| True, existing.status, existing.id, existing.sequence | ||
| ) | ||
| return self._duplicate_result(existing, message.payload_hash()) |
There was a problem hiding this comment.
问题： claim_message 只重置 status == "failed" 的重复事件；若 worker 在 claim_message 提交 processing 之后、mark_message_failed/complete_message 之前崩溃（进程被杀、部署重启、DB 瞬时故障后异常逃逸），事件会永久停留在 processing。provider 重投同一 external_message_id 时命中 _duplicate_result，service 返回 200 {"duplicate": true}，provider 认为已送达即停止重试。claim_due_outbox 只能恢复 outbox 的 sending 行，没有针对 processing 消息事件的恢复机制。
触发条件： 处理中进程崩溃或异常中断，随后 provider 对未确认的 webhook 进行重投。
实际影响： 用户消息被静默丢弃——永远收不到回复，provider 也不再重试，只能人工修改数据库恢复；与加固提交宣称的故障恢复（outbox 自动恢复）相比，消息侧缺少同等保障，且 session.last_event_seq 对应的会话历史出现空洞。
修正方向： 为 processing 状态增加崩溃恢复语义：对停留在 processing 超过 TTL（如会话租约时长）且无活跃租约的事件，允许重投重置为可重跑状态（类似 claim_due_outbox 的过期恢复），或增加后台 reaper 扫描超时 processing 事件并标记 failed 以重入重试路径。
| except ValueError as exc: | ||
| return ChannelResponse( | ||
| status_code=202, body={"ok": True, "ignored": type(exc).__name__} | ||
| ) |
There was a problem hiding this comment.
问题： except ValueError as exc: 会吞掉 ConfigurationError（config.py:14 声明为 ValueError 子类）。adapter.verify 内部 require_secret(binding.webhook_secret_env)（adapters.py:69/134）在密钥环境变量缺失或为空时抛 ConfigurationError，落入此分支返回 202 {"ok": true, "ignored": "ConfigurationError"}。
触发条件： OFFLINE_ECHO_MODE=true（演示/默认本地方式）跳过 build_app_from_env 的全部密钥预检（app.py:114-126），任一 binding 的 webhook secret 未设置时，真实回调即走此路径；生产运行中密钥被轮换/移除（k8s Secret 删除）也会命中。
实际影响： 认证与配置故障被伪装成 202 成功响应，IM 平台认为消息已处理而停止重试，用户消息静默丢失（无审计、无失败指标）；响应体还把异常类名 ConfigurationError 反射给调用方，泄露服务端配置状态。
修正方向： 将 ConfigurationError 单独捕获并升级为 503/500（fail-closed），或把 require_secret 的缺失抛错改为 InvalidCallbackError 之外的显式类型；except ValueError 只捕获 UnsupportedMessageError。
| charged_tokens = reply.token_count or decision.estimated_tokens | ||
| await asyncio.to_thread( | ||
| self.repository.settle_token_budget, | ||
| tenant.tenant_id, | ||
| budget_period, | ||
| decision.estimated_tokens, | ||
| charged_tokens, | ||
| ) | ||
| budget_settled = True |
There was a problem hiding this comment.
问题： 新增的 tests/test_gateway.py（25 个用例）覆盖了重复/冲突/账户隔离/预算原子预留/死信/审计降级/outbox 恢复，但没有覆盖本次变更中风险最高的三条故障路径的回归测试，且仓库 CI（.github/workflows/ci.yml 的 testpaths=["tests"]）不会运行 examples 下的测试，这些缺陷将无回归防线。
触发条件： 后续修改预算结算顺序、claim_message 状态机或 period 翻转逻辑时，现有测试全部通过但上述缺陷复发。
实际影响： 双计费、预留泄漏、消息丢失这三类缺陷（见同文件其余评论）缺少自动化保障；测试缺口使已确认的正确性风险无法被门禁拦截。
修正方向： 补充三类回归测试：(1) 打桩使 complete_message 抛异常，断言 finally 回滚将 token_usage 恢复且 provider 重投后不出现二次计费；(2) 预置 processing 且租约过期的重复事件，断言重投可恢复而非返回 duplicate；(3) 构造跨月 period 场景，断言旧 period 预留被正确结算/释放。并将 examples 测试接入 CI（如增加 pytest examples/multi_tenant_im_agent/tests 步骤）。
| def claim_outbox(self, outbox_id: str) -> bool: | ||
| now = utcnow() | ||
| with self.Session.begin() as db: | ||
| item = db.scalar( | ||
| select(OutboxRecord) | ||
| .where(OutboxRecord.outbox_id == outbox_id) | ||
| .with_for_update() | ||
| ) | ||
| if item is None or item.status not in {"pending", "retry"}: | ||
| return False | ||
| item.status = "sending" | ||
| item.attempts += 1 | ||
| item.next_attempt_at = now + timedelta(seconds=60) | ||
| return True | ||
|
|
||
| def claim_due_outbox(self, limit: int = 50) -> list[OutboxItem]: | ||
| """Lease retryable rows; expired ``sending`` rows recover crashed workers.""" | ||
|
|
||
| now = utcnow() | ||
| with self.Session.begin() as db: | ||
| rows = list( | ||
| db.scalars( | ||
| select(OutboxRecord) | ||
| .where( | ||
| OutboxRecord.status.in_(("pending", "retry", "sending")), | ||
| OutboxRecord.next_attempt_at <= now, | ||
| ) | ||
| .order_by(OutboxRecord.next_attempt_at) | ||
| .limit(limit) | ||
| .with_for_update(skip_locked=True) | ||
| ) | ||
| ) | ||
| result: list[OutboxItem] = [] | ||
| for item in rows: | ||
| item.status = "sending" | ||
| item.attempts += 1 | ||
| item.next_attempt_at = now + timedelta(seconds=60) | ||
| payload = json.loads(item.payload_json) | ||
| result.append(OutboxItem(item.outbox_id, DeliveryRequest(**payload))) | ||
| return result |
There was a problem hiding this comment.
问题： acquire_session_lease(386)、reserve_token_budget(425/446)、claim_message(480)、claim_outbox(576)、claim_due_outbox(599, skip_locked=True) 的 with_for_update 在 SQLite 方言下被静默丢弃（已验证 SQLAlchemy 编译：SQLite 无 FOR UPDATE，MySQL/PG 有 FOR UPDATE [SKIP LOCKED]）。默认本地演示（build_app_from_env 默认 sqlite:///multi_tenant_im.db）和单机多线程部署下，这些行锁全部失效。
触发条件： 两个线程/进程并发执行 claim_due_outbox（如多 worker 或同一进程内 webhook 与后台循环交错）时，同一 pending 行被两个执行者同时领取；并发 reserve_token_budget 可同读 token_usage 后同时 +estimated 超额放行；并发 claim_message 可同读 last_event_seq 造成 sequence 冲突消息丢库。
实际影响： 默认部署下 outbox 重复投递、月度预算超额消费、session 消息序号竞态导致消息 503 丢失，行为与生产（MySQL/PG）不一致，单元测试全部跑 SQLite 因此从未暴露。
修正方向： 对 SQLite 部署显式声明不支持多进程并发（文档 + 启动断言），或对关键路径（claim_due_outbox、claim_outbox、reserve_token_budget）改用 SQLite 支持的条件更新（UPDATE ... WHERE status = 'pending' 并检查 rowcount），保证跨方言语义一致。
| "model", | ||
| (perf_counter() - model_started) * 1000, | ||
| ) | ||
| charged_tokens = reply.token_count or decision.estimated_tokens |
There was a problem hiding this comment.
问题： charged_tokens = reply.token_count or decision.estimated_tokens（以及 308/316 行同模式）用 falsy-zero 把合法的 0 当作缺失。reply.token_count 来自 event_token_count（runtime.py:15-26），当 provider 终末事件无 usage_metadata（自托管/OpenAI 兼容端点常见，SDK 只有 terminal 事件附带 usage）时返回 0；settle_token_budget（repository.py:450）公式 usage = max(0, usage - reserved + max(0, actual)) 在 actual=estimated 时等价于保留预留，永不释放。
触发条件： provider 不返回 usage（或返回真实 0 token）的任何成功回复：每月按输入字符数估算值（governance.py:27）累计扣费，与真实消耗无关。对比失败路径 finally（service.py:347-354）显式 settle(actual=0) 释放预留——同一代码库对“未知用量”两种相反语义。
实际影响： 月预算被估算值永久侵吞，未到月底真实请求遭 403 monthly_token_budget_exceeded 拒绝；估算值远低于真实用量（system prompt/历史/tool 未计）时又低估，计费完全失真且无对账。
修正方向： 用 is not None 判断缺省而非 or：charged_tokens = reply.token_count if reply.token_count is not None else decision.estimated_tokens，并让 AgentReply.token_count 的未知态（None）与真实 0 区分，未知态同样走释放预留的语义。
| sa.Column("config_version", sa.Integer(), nullable=False), | ||
| sa.Column("token_budget_period", sa.String(7), nullable=False), | ||
| sa.Column("token_usage", sa.Integer(), nullable=False), | ||
| sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), |
There was a problem hiding this comment.
问题： 全部时间列使用 sa.DateTime(timezone=True)，在 MySQL 上编译为 DATETIME（fsp=0，秒级），而 ORM 写入的是 datetime.now(timezone.utc)（微秒）。仓库 base 提交 c98a61d 刚为 MySQL 引入 PreciseTimestamp（mysql.DATETIME(fsp=6) + CURRENT_TIMESTAMP(6)，trpc_agent_sdk/storage/_sql_common.py:282-294）专门修复这一精度问题，本迁移未沿用，重新引入同类缺陷。
触发条件： MySQL 部署（docker-compose.yml / kubernetes.yaml 默认）：同一秒内的审计日志/outbox 记录 created_at、updated_at 相同，微秒被截断；next_attempt_at/expires_at 与代码中 aware 比较基准发生偏移。
实际影响： 审计时间线无法精确还原；outbox 重试排期与租约过期时间被秒级取整（MySQL 对小数位取整而非截断，可能提前 1 秒）导致重试节奏、租约语义偏差；且 SQLite 读回 naive、PG 读回 aware，跨方言 naive/aware 不一致。
修正方向： 迁移改用与 SDK 一致的 sa.DateTime() + MySQL DATETIME(fsp=6)（或直接复用仓库 PreciseTimestamp），并在 test_migration_contract 中加入对 MySQL 方言的编译断言。
| tool_names=tuple(sorted(tools)), | ||
| ) | ||
|
|
||
| return await asyncio.wait_for(collect(), timeout=tenant.model_timeout_seconds) |
There was a problem hiding this comment.
问题： await asyncio.wait_for(collect(), timeout=...) 超时取消 collect() 时，runner.run_async 返回的异步生成器未显式 aclose()，底层流式 HTTP 连接与 SDK runner 的会话写回不会清理/结束。
触发条件： 模型超过 model_timeout_seconds（默认 90s）无响应或网络停滞，wait_for 抛 TimeoutError（service.py:323 映射为 model_timeout）；高频超时的租户反复命中。
实际影响： 流式连接悬挂、会话事件未终结，连接与资源的泄漏随超时次数累积，最终可能打满 provider 连接池，连带正常租户请求失败；被取消的 runner 仍留在 _runners 缓存中。
修正方向： 在 collect() 外加 try/finally 显式 await generator.aclose()（或在超时时对生成器调用 aclose()），并考虑对超时中的 runner 调用 cancel_run_async 清理会话。
| else: | ||
| existing.tenant_id = tenant.tenant_id | ||
| existing.secret_ref = binding.webhook_secret_env | ||
| existing.enabled = int(binding.enabled) |
There was a problem hiding this comment.
问题： sync_tenants 按 (channel, account_id) 全局查 ChannelBindingRecord，命中即无条件把 existing.tenant_id 改写为当前配置租户，不校验旧行归属。
触发条件： 配置重载（重启/滚动发布）时同一 (channel, account_id) 被运维从租户 A 改配到租户 B（错误配置或有意迁移但未走删除流程）。
实际影响： 租户 A 该账号的回调瞬时路由到租户 B 的 Agent，A 用户消息进入 B 租户上下文（跨租户数据泄露）；会话/消息历史混入 B 的名下；反向（删除 binding）时旧行永不清理，遗留 enabled=1 的幽灵绑定。
修正方向： 命中已有 binding 且其 tenant_id != 当前租户 时拒绝启动（ConfigurationError）或先删除再重建；删除配置中不存在的 binding 行；并为跨租户迁移提供显式流程。
| def create_table(self, name: str, *items: object) -> None: | ||
| self.tables[name] = {item.name for item in items if isinstance(item, sa.Column)} | ||
|
|
||
| def create_index(self, name: str, table: str, columns: list[str]) -> None: | ||
| self.indexes.add((name, table, tuple(columns))) | ||
|
|
||
| def drop_table(self, name: str) -> None: | ||
| self.dropped.append(name) |
There was a problem hiding this comment.
问题： OperationRecorder 只记录 create_table 的列名集合与 index 名+列名，不比对列类型（String 长度、DateTime(timezone)/fsp）、nullable、默认值、UniqueConstraint（作为参数传入但被 isinstance(sa.Column) 过滤）、ForeignKey 引用；迁移从不跑真实引擎（MySQL/PG），测试全部基于 SQLite 内存库，CI（.github/workflows/ci.yml + pyproject testpaths=["tests"]）甚至不收集 examples/ 下任何测试。
触发条件： 任何人修改迁移或 ORM 模型（改列长、换类型、删约束），或 MySQL 方言下的建表/索引/精度问题（如 utf8mb4 索引字节超限、DATETIME 精度）。
实际影响： 迁移与模型漂移测试保持绿色——删除 uq_mt_inbound_idempotency/uq_mt_session_sequence_direction 任一唯一约束、把 String(192) 缩到 String(64)、把时间列改无精度，全部无提示；上述 outbox/预算/租约的并发行为与 MySQL 精度问题在生产 MySQL 上直接暴露。
修正方向： 用真实 SQLite + MySQL 方言执行 upgrade/downgrade（或 sqlalchemy-migrate 风格比对），断言列类型/长度/nullable/唯一约束/FK；并把 pytest examples/multi_tenant_im_agent/tests 纳入 CI。
| @app.get("/metrics", response_class=PlainTextResponse, tags=["operations"]) | ||
| async def metrics(): | ||
| return service.metrics.render_prometheus() |
There was a problem hiding this comment.
问题： /metrics 端点无任何鉴权（与 /admin/tenants 不同），render_prometheus 以 tenant_id 为标签暴露各租户请求量、token 消耗与模型成本。
触发条件： 网关监听 0.0.0.0（main.py 默认），任一可访问该端点的未认证调用者（或经 Telegram/WeCom 回调暴露的公网入口）访问 /metrics。
实际影响： 跨租户业务情报泄露：租户 ID、消息量、token 用量、模型花费全部可见，可用于针对特定租户机器人的定向滥用或计费侦查。
修正方向： 对 /metrics 采用与 admin 相同的 token 校验（或将其置于内网网段/独立端口）；如需公开，压缩为无 tenant 标签的聚合指标。
| async with httpx.AsyncClient(timeout=self.timeout_seconds) as client: | ||
| response = await client.post(url, json=payload) | ||
| response.raise_for_status() | ||
| result = response.json() | ||
| return result if isinstance(result, dict) else {"accepted": True} |
There was a problem hiding this comment.
问题： HttpChannelSender.send 每次投递都在 async with httpx.AsyncClient(...) 中新建客户端，连接池随请求销毁，每次调用都重新 TCP+TLS 握手。
触发条件： 消息量大的网关（outbox 循环单轮最多 50 条、webhook 内联投递并发）每条消息一次全新建连接。
实际影响： 到 api.telegram.org / 企业微信的出站延迟与对端连接压力成倍放大（每条消息一次握手，无 keep-alive 复用），高吞吐下易触发 IM 平台限流；纯连接池复用即可消除的成本。
修正方向： 将单个 httpx.AsyncClient 提升为服务的长生命周期组件（与其他运行期资源一起 close），或改用 httpx.AsyncClient(timeout=...) 实例注入 HttpChannelSender。
AI Code Review审查结论通过 审查范围：base 1aa44cf..head 7ab06cf，新增 41 个文件、约 4729 行，含 FastAPI 网关（app.py/service.py）、SQL 控制面（repository.py + Alembic 迁移）、Telegram/企业微信适配器（adapters.py）、tRPC-Agent Runner 集成（runtime.py）、治理/遥测/脚本与 CI。计划「feat: add multi-tenant IM agent gateway」与交付内容相符，功能完整：租户路由、回调验签、幂等（唯一键+payload hash）、SQL 租约串行化、事务性 Outbox、月度 Token 原子预算、审计/指标/链路追踪、Alembic 迁移与离线验收脚本，测试覆盖较全面（29 个单元测试 + 2 个 HTTP/契约测试）。 主要发现（均基于代码内静态证据链验证）：1）runtime.py 流式回复文本选择逻辑错误，chunks 与 final_parts 同时非空时丢弃含完整最终文本的 final_parts，多轮工具调用场景回复被截断；2）service.py 会话租约先于幂等声明获取，provider 重投在租约持有期内被 429 拒绝而无法命中幂等检查，与模型超时叠加时可能导致用户消息最终被 provider 放弃；3）预算结算（settle_token_budget）与消息完成（complete_message）分属两个事务，中间数据库故障窗口会导致同一输入被二次计费。以上均为 MODERATE/LOW 级影响，无 SEVERE 级缺陷；未发现可证明的跨租户数据泄露、密钥泄漏或注入问题。 测试充分性：单元测试覆盖路由、验签、幂等、预算、租约、Outbox 等主路径，缺少流式/多轮回复文本、模型超时重投恢复、结算-完成事务边界三个具体场景的测试。门禁结论：PASSED（可合入，建议跟进上述 3 个问题）。 发现的问题中等
问题： 触发条件： 租户配置了 实际影响： 用户收到的 IM 回复被截断为流式部分文本（可能缺失工具调用后的最终回答），且 修正方向： 优先采用最终非 partial 文本： 中等
问题： 触发条件： IM provider（Telegram/企业微信）对回调重试或负载均衡将同一消息同时分发到两个 worker，且第一个请求持有租约超过 provider 的重试间隔。 实际影响： (a) 原请求最终成功时，重投的 429 只是多余一次网络往返；(b) 原请求模型超时失败（默认 90s 超时、租约 120s）时，provider 的重试在租约持有期内全部命中 429，租约释放后若 provider 重试次数已耗尽则用户消息被静默丢弃——与幂等表设计意图（ 修正方向： 将幂等声明移到租约获取之前（先 较低
问题： 月度预算结算（ 触发条件： (a) 实际影响： 租户月度 Token/成本被重复或高估计费，可能提前触发预算拒止（403），且审计中 token_count 与真实用量不一致。 修正方向： 将结算与消息完成放入同一事务（或在 |
| async def collect() -> AgentReply: | ||
| chunks: list[str] = [] | ||
| final_parts: list[str] = [] | ||
| tools: set[str] = set() | ||
| token_count = 0 | ||
| async for event in runner.run_async( | ||
| user_id=user_id, | ||
| session_id=session_id, | ||
| new_message=Content(parts=[Part.from_text(text=message.text)]), | ||
| agent_context=new_agent_context( | ||
| timeout=tenant.model_timeout_seconds * 1000, | ||
| metadata={ | ||
| "tenant_id": tenant.tenant_id, | ||
| "tenant_policy_approved": True, | ||
| "tool_allowlist": tenant.tool_allowlist, | ||
| }, | ||
| ), | ||
| ): | ||
| token_count += event_token_count(event) | ||
| if not event.content: | ||
| continue | ||
| for part in event.content.parts or []: | ||
| if part.thought: | ||
| continue | ||
| if part.function_call: | ||
| tools.add(part.function_call.name) | ||
| elif part.text: | ||
| (chunks if event.partial else final_parts).append(part.text) | ||
| text = "".join(chunks) if chunks else "".join(final_parts) |
There was a problem hiding this comment.
问题： TrpcAgentRuntime.reply 的文本收集逻辑以 chunks（流式 partial 片段）优先、final_parts（非 partial 完整文本）兜底：text = "".join(chunks) if chunks else "".join(final_parts)，而 chunks 与 final_parts 可能同时非空。trpc_agent_sdk 的流式实现（trpc_agent_sdk/models/_openai_model.py 的 _generate_stream/_generate_responses_stream）在流结束时必定产出 partial=False 且 content 为完整 accumulated_content 的最终响应，同时在流中不断产出 partial=True 的增量片段。当 Agent 经历多轮执行（如工具调用后再次调模型生成最终回答）时，chunks 保存的只是首轮/前半段文本，最终完整回答被整体丢弃。
触发条件： 租户配置了 tool_allowlist 且模型多轮执行（工具调用后生成最终回复）时，chunks 非空使第 199 行永远走 chunks 分支，而 final_parts 中的最终完整回答（含工具结果后的总结文本）被丢弃。
实际影响： 用户收到的 IM 回复被截断为流式部分文本（可能缺失工具调用后的最终回答），且 complete_message 将截断文本写入 response_text 与 Outbox，损坏审计存根与重试投递内容；单轮纯文本场景下 chunks 与最终文本恰好一致，因此现有测试未暴露该问题。
修正方向： 优先采用最终非 partial 文本：text = "".join(final_parts) if final_parts else "".join(chunks)，并为多轮工具调用场景补充流式文本测试。
| lease_owner = f"{self.worker_id}:{message.external_message_id[:64]}" | ||
| acquired = await asyncio.to_thread( | ||
| self.repository.acquire_session_lease, | ||
| session_id, | ||
| lease_owner, | ||
| tenant.session_lease_seconds, | ||
| ) | ||
| if not acquired: | ||
| await asyncio.to_thread( | ||
| self._audit, | ||
| tenant_id=tenant.tenant_id, | ||
| channel=message.channel, | ||
| user_id=actor_user_id, | ||
| session_id=session_id, | ||
| agent_name=tenant.agent_name, | ||
| decision="busy", | ||
| error_type="session_lease_conflict", | ||
| ) | ||
| self.metrics.observe_request(tenant.tenant_id, message.channel, "busy", 0) | ||
| return ChannelResponse( | ||
| status_code=429, | ||
| body={"ok": False, "error": "session_busy", "retryable": True}, | ||
| headers={"Retry-After": "2"}, | ||
| ) |
There was a problem hiding this comment.
问题： handle_webhook 在幂等声明（claim_message，repository.py:452）之前先获取会话租约（service.py:130-136）。同一外部消息被 provider 并发/重复投递时，第二个 worker 在第一个 worker 仍持有租约（默认 120s）期间直接返回 429 session_busy，根本走不到幂等声明分支，重复消息无法被识别为 duplicate。
触发条件： IM provider（Telegram/企业微信）对回调重试或负载均衡将同一消息同时分发到两个 worker，且第一个请求持有租约超过 provider 的重试间隔。
实际影响： (a) 原请求最终成功时，重投的 429 只是多余一次网络往返；(b) 原请求模型超时失败（默认 90s 超时、租约 120s）时，provider 的重试在租约持有期内全部命中 429，租约释放后若 provider 重试次数已耗尽则用户消息被静默丢弃——与幂等表设计意图（existing.status == "failed" 时同载荷重试可重新执行）相矛盾，重投恢复链被租约互斥阻断。
修正方向： 将幂等声明移到租约获取之前（先 claim_message 判定 duplicate/conflict 并直接返回，再获取租约处理新消息），并对 session_busy 的 429 同时记录消息事件供后续重投命中幂等分支；补充"持有租约时同消息重投"与"模型超时后 provider 重试"的集成测试。
| charged_tokens = reply.token_count or decision.estimated_tokens | ||
| await asyncio.to_thread( | ||
| self.repository.settle_token_budget, | ||
| tenant.tenant_id, | ||
| budget_period, | ||
| decision.estimated_tokens, | ||
| charged_tokens, | ||
| ) | ||
| budget_settled = True | ||
| delivery = adapter.delivery( | ||
| binding=binding, message=message, reply=reply | ||
| ) | ||
| storage_started = perf_counter() | ||
| try: | ||
| outbox_id = await asyncio.to_thread( | ||
| self.repository.complete_message, | ||
| event_id=event_id, | ||
| tenant_id=tenant.tenant_id, | ||
| session_id=session_id, | ||
| channel=message.channel, | ||
| reply=reply, | ||
| delivery=delivery, | ||
| ) | ||
| finally: | ||
| self.metrics.observe_stage( | ||
| tenant.tenant_id, | ||
| "storage", | ||
| (perf_counter() - storage_started) * 1000, | ||
| ) |
There was a problem hiding this comment.
问题： 月度预算结算（settle_token_budget）与消息完成（complete_message+Outbox）是两个独立的数据库事务（service.py:243-251 与 256-271 为两次不同的 to_thread 调用），中间存在未保护的异常窗口；且 charged_tokens = reply.token_count or decision.estimated_tokens 使结算值（actual）永远不低于估算值。
触发条件： (a) settle_token_budget 提交成功后、complete_message 抛异常（数据库瞬时故障），请求进入 _handle_failure 将事件标记为 failed，provider 重投后同载荷事件被重新接受并再次运行模型、再次结算——同一输入计费两次；(b) 模型返回 token_count=0（如缓存命中）时按估算值而非 0 结算，月度预算被高估。
实际影响： 租户月度 Token/成本被重复或高估计费，可能提前触发预算拒止（403），且审计中 token_count 与真实用量不一致。
修正方向： 将结算与消息完成放入同一事务（或在 complete_message 成功后再结算，失败路径回滚结算），并补充"结算成功但完成失败后重投"与"token_count=0"场景的用例。
AI Code Review审查结论不通过 审查范围： 发现的问题严重
问题： 触发条件： 任何调用方绕过网关直接调用 实际影响： 该过滤器是面向"未来直接 Runner 调用方"的隔离防线，拒绝被绕过意味着多租户治理（策略、预算、账户校验）对这类调用完全不生效——消息仍消耗远端 LLM 额度并产生应答；若未来网关内部依赖此过滤器兜底，则会造成策略绕过。属安全边界失效。 修正方向： 在 中等
问题： 触发条件： 模型 provider 不可达、鉴权失败、限流或任意被 SDK 重试层捕获并转换为错误事件的异常；同时发生在每次 实际影响： 模型调用失败被伪装为成功：网关返回 HTTP 200 "completed" 并向用户下发"Sorry, this request produced no sendable text response."占位文案， 修正方向： 在 中等
问题： 触发条件： 任何 Prometheus 抓取实例（或 实际影响： 指标端点整体无法被优雅解析——先于 修正方向： 在输出任何样本前，先按指标名分组输出全部 |
| @register_agent_filter(TENANT_FILTER_NAME) | ||
| class MultiTenantGovernanceFilter(BaseFilter): | ||
| async def _before(self, ctx, req, rsp): | ||
| # The gateway sets these only after signature, policy, budget, and | ||
| # account checks. Missing metadata therefore fails closed even if a | ||
| # future caller invokes Runner without going through the gateway. | ||
| tenant_id = ctx.get_metadata("tenant_id", "") | ||
| policy_approved = ctx.get_metadata("tenant_policy_approved", False) | ||
| if not tenant_id or not policy_approved: | ||
| rsp.error = PermissionError("tenant governance context is missing") | ||
| rsp.is_continue = False |
There was a problem hiding this comment.
问题： MultiTenantGovernanceFilter._before 在拒绝路径上只设置 rsp.error 与 rsp.is_continue = False，未设置 rsp.rsp。按 SDK 语义（trpc_agent_sdk/filter/_base_filter.py:133-147 的 _handle_co 协程分支：if result.rsp: yield result 后才有 if not result.is_continue: return），错误型 FilterResult.rsp=None 不会产生任何 yield，run_stream（_base_filter.py:178-181）的 async for 以零次迭代正常结束，随后继续执行 # run last filter 的 handle()——即真实模型调用照常发生。注释声称的"Missing metadata therefore fails closed even if a future caller invokes Runner without going through the gateway"实际不成立。SDK 自身的 AgentCallbackFilter._before（trpc_agent_sdk/agents/_callback.py:118-130）在 is_continue=False 前总是先设置 rsp.rsp 为真实 Event，即本变更缺少的正是这一关键赋值。
触发条件： 任何调用方绕过网关直接调用 Runner.run_async（构造不含 tenant_id/tenant_policy_approved 元数据的 AgentContext），或在网关流程中构建 Runner 时元数据未正确传递；此时过滤器被判为"拒绝"，但模型仍被调用。
实际影响： 该过滤器是面向"未来直接 Runner 调用方"的隔离防线，拒绝被绕过意味着多租户治理（策略、预算、账户校验）对这类调用完全不生效——消息仍消耗远端 LLM 额度并产生应答；若未来网关内部依赖此过滤器兜底，则会造成策略绕过。属安全边界失效。
修正方向： 在 _before 拒绝分支中按 SDK 的终止模式补充 rsp.rsp = Event(...)（可构造一个带 error_code/error_message 且 content 为 None 的拒绝事件，或与 AgentCallbackFilter 一致构造含提示文案的 Event），使 _handle_co 能 yield 并触发 is_continue 短路；同时补一个单测断言拒绝时模型 generate_async 不被调用。
| async for event in stream: | ||
| event_usage = event_token_count(event) | ||
| if event_usage is not None: | ||
| if event.partial: | ||
| # Some providers repeat cumulative usage on every | ||
| # partial event, so keep only the largest partial value. | ||
| partial_token_count = max(partial_token_count or 0, event_usage) | ||
| else: | ||
| # Multiple terminal events can represent multiple model | ||
| # calls in a tool loop and must all be charged. | ||
| terminal_token_count = (terminal_token_count or 0) + event_usage | ||
| if not event.content: | ||
| continue | ||
| for part in event.content.parts or []: | ||
| if part.thought: | ||
| continue | ||
| if part.function_call: | ||
| tools.add(part.function_call.name) | ||
| elif part.text: | ||
| (partial_parts if event.partial else final_parts).append( | ||
| part.text | ||
| ) | ||
| if final_parts: | ||
| text = "".join(final_parts) | ||
| elif partial_parts: | ||
| cumulative = all( | ||
| current.startswith(previous) | ||
| for previous, current in pairwise(partial_parts) | ||
| ) | ||
| text = partial_parts[-1] if cumulative else "".join(partial_parts) | ||
| else: | ||
| text = "" | ||
| if not text.strip(): | ||
| text = "Sorry, this request produced no sendable text response." | ||
| token_count = ( | ||
| terminal_token_count | ||
| if terminal_token_count is not None | ||
| else partial_token_count | ||
| ) | ||
| return AgentReply( | ||
| text=text, | ||
| token_count=token_count, | ||
| tool_names=tuple(sorted(tools)), | ||
| ) |
There was a problem hiding this comment.
问题： collect() 用 if not event.content: continue（第 207-208 行）丢弃所有无内容事件，且整段循环没有任何 event.is_error()/error_code 检查。而 SDK 的 retry_model_call 会把 provider 异常（网络错误、401/403、超时等）转换为终态 LlmResponse(content=None, error_code=..., error_message=...)（trpc_agent_sdk/models/_retry.py:148-155），该响应经 _create_event_from_response（trpc_agent_sdk/agents/core/_llm_processor.py:163-171, 268）成为 Event（Event.is_error() 即 error_code is not None，trpc_agent_sdk/events/_event.py:256-258）并从 Runner.run_async 正常 yield（trpc_agent_sdk/runners.py:561）。因此模型失败时 collect() 拿到的唯一终态事件恰是 content=None 的错误事件，被静默丢弃后落入占位文案分支（第 226-229 行）。
触发条件： 模型 provider 不可达、鉴权失败、限流或任意被 SDK 重试层捕获并转换为错误事件的异常；同时发生在每次 TrpcAgentRuntime.reply。
实际影响： 模型调用失败被伪装为成功：网关返回 HTTP 200 "completed" 并向用户下发"Sorry, this request produced no sendable text response."占位文案，complete_message 将事件标记为 completed、按 reply.token_count（无用量时为估计值）全额结算预算并写入 outbox 投递——失败既不可重试也无任何错误信号，调用方（IM 用户）无法区分"模型拒绝回答"与"系统故障"，且预算被无谓扣除。
修正方向： 在 collect() 循环中检查 event.is_error()（或 event.error_code），遇错误事件立即抛出（或返回携带错误码的 AgentReply），使 service.py 的通用 except Exception 路径进入 _handle_failure（返回 503、mark_message_failed、审计 failed）；同时补充测试：用 SDK 真实 Runner + 会抛异常/返回错误响应的模型桩，断言 Webhook 返回可重试错误、事件状态为 failed 且预算被退还。
| for (channel, status), value in sorted(self._delivery.items()): | ||
| lines.append( | ||
| f'trpc_im_delivery_total{{channel="{channel}",status="{status}"}} {value}' | ||
| ) | ||
| lines.append("# TYPE trpc_im_audit_total counter") | ||
| for status, value in sorted(self._audit.items()): | ||
| lines.append(f'trpc_im_audit_total{{status="{status}"}} {value}') | ||
| lines.append("# TYPE trpc_im_tokens_total counter") | ||
| for tenant, value in sorted(self._tokens.items()): | ||
| lines.append(f'trpc_im_tokens_total{{tenant="{tenant}"}} {value}') | ||
| lines.append("# TYPE trpc_im_model_cost_total counter") | ||
| for tenant, value in sorted(self._cost.items()): |
There was a problem hiding this comment.
问题： render_prometheus 对 trpc_im_delivery_total（第 141-144 行先输出样本，第 145 行才追加 # TYPE trpc_im_delivery_total counter 且完全缺失 # HELP 行）、trpc_im_audit_total（145 行 TYPE 在 146-147 行样本之后）、trpc_im_tokens_total（148 行 TYPE 在样本之后）、trpc_im_model_cost_total（151 行 TYPE 在样本之后）以及 trpc_im_stage_latency_ms（154 行 TYPE 在 155-161 行样本之后）等指标，均将样本行排在 # TYPE/# HELP 声明行之前，不符合 Prometheus 文本暴露格式（声明必须位于所属指标的全部样本之前）。
触发条件： 任何 Prometheus 抓取实例（或 promtool check metrics）请求 /metrics 端点；只要上述任一计数器非零即触发解析错误。
实际影响： 指标端点整体无法被优雅解析——先于 # TYPE 出现的样本行会使抓取失败或按文本模型错误归类，/metrics 采集与告警（如投递失败、审计、预算用量、租户成本）全部失效，运营可观测性受损。
修正方向： 在输出任何样本前，先按指标名分组输出全部 # HELP/# TYPE 声明行，再集中输出各指标样本（例如为每个指标构建独立的 (help, type, samples) 三段后拼接，并按指标名排序以保持首行即声明）；补一个断言：对含各类计数器的状态调用 render_prometheus()，用 prometheus_client.parser（或正则）校验每个指标名的 # TYPE 行先于其样本出现。
2 participants
背景与目标
本 PR 为 tRPC-Agent 新增一个可运行的多租户 IM Agent 网关参考实现，覆盖项目要求中的多租户与节点部署、数据同步与多后端、IM 接入、治理安全、故障恢复与运维。实现并非架构伪代码：在线模式为每个租户创建真实的 tRPC-Agent
LlmAgent + Runner，离线验收模式也保留同一 Runner 生命周期，仅将外部模型替换为确定性本地模型。核心能力
(channel, account_id) -> tenant_id唯一路由；租户级 Agent App、模型、工具白名单和 Session 后端；无 sticky session；共享 Redis/SQL 支持多 Workerprocessing消息恢复、Outbox sending 租约恢复、指数退避与稳定抖动、8 次后死信本轮重点加固
LlmAgent + Runner，并在 readiness 前预热租户 Runner。processing消息，可在旧租约过期并由新 Worker 取得租约后安全恢复。DATETIME(6)，迁移契约同时校验类型、空值、主键、唯一约束和外键。一键验收
安装依赖：
pip install -e ".[multi-tenant-im]"运行独立 HTTP 黑盒验收：
运行完整测试：
验证结果
39 passedMulti-Tenant IM Agent已通过主要文件
examples/multi_tenant_im_agent/runtime.py：真实 Runner、多租户运行时、流式响应与 usage 收集examples/multi_tenant_im_agent/service.py：验签后治理、租约、幂等、预算、执行、投递和审计编排examples/multi_tenant_im_agent/repository.py：SQL 控制面、隔离约束、预算、消息状态机与 Outboxexamples/multi_tenant_im_agent/adapters.py：Telegram / 企业微信适配器与外发连接池examples/multi_tenant_im_agent/migrations/：Alembic 版本化迁移examples/multi_tenant_im_agent/tests/：39 项单元、并发、故障与迁移契约测试examples/multi_tenant_im_agent/scripts/judge_demo.py：自启动、自验收、自清理的评委脚本examples/multi_tenant_im_agent/ARCHITECTURE.zh_CN.md：完整架构、隔离、迁移、治理与运维设计examples/multi_tenant_im_agent/EVALUATION.zh_CN.md：要求到代码与测试的证据矩阵安全与实现边界
outbox_id，否则需结合回执对账。变更范围
本 PR 以独立示例、可选依赖和专用工作流接入，不改变现有 SDK 公共 API；生产凭据、数据库和 IM 账号均不进入仓库。