Skip to content

fix(postgresql): register asyncpg JSON codecs and fix tool call recording on PG backend#2327

Open
MIKAZE3 wants to merge 3 commits into
langbot-app:masterfrom
MIKAZE3:fix/postgresql-errors
Open

fix(postgresql): register asyncpg JSON codecs and fix tool call recording on PG backend#2327
MIKAZE3 wants to merge 3 commits into
langbot-app:masterfrom
MIKAZE3:fix/postgresql-errors

Conversation

@MIKAZE3

@MIKAZE3 MIKAZE3 commented Jul 10, 2026

Copy link
Copy Markdown
Contributor

概述 / Overview

请在此部分填写你实现/解决/优化的内容:
Summary of what you implemented/solved/optimized:

修复 MonitoringService._get_message_for_tool_context 在 PostgreSQL/asyncpg 后端下错误地返回主键字符串而不是 ORM 实体的问题。

原实现使用 result.first() 后取 row[0]。在 SQLite/MySQL 上,Row[0] 会被解包为查询选择的第一个列(即 ORM 实体的主键),恰好等同于实体本身;但在 PostgreSQL/asyncpg 上,Row 是底层行元组,row[0] 返回的是主键原始值(如 'msg-123'),后续调用 msg.idmsg.session_id 等属性时触发 AttributeError,导致工具调用上下文中的监控消息解析失败(每次工具调用日志输出 [WARNING] Failed to record tool call: 'str' object has no attribute 'pipeline_id')。

改为 SQLAlchemy 2.0 推荐的 result.scalars().first(),统一在不同数据库后端返回 MonitoringMessage 实体对象。同时补全函数返回类型注解,并新增覆盖三个查询路径(message_idsession_id + role=usersession_id + 任意 role)以及 None 退化路径的单元测试,确保不再退回 result.first() + row[0] 的旧实现。

与上一个版本的差异:原先 monitoring.py 改动只加了 isinstance(obj, MonitoringMessage) 类型检查并在失败时返回 None,未真正解决对象获取问题。@dadachann#2327 (comment) 中指出该实现"只是返回 None,没有正确取到消息对象"。本版本把类型判断替换为真正的 result.scalars().first(),跨数据库后端均能返回 ORM 实体;测试新增 TestGetMessageForToolContextUsesScalars 回归类,显式断言 result.scalars() / result.scalars().first() 被调用、result.first() 未被调用,防止回退。

更改前后对比截图 / Screenshots

请在此部分粘贴更改前后对比截图(可以是界面截图、控制台输出、对话截图等):
Please paste the screenshots of changes before and after here (can be interface screenshots, console output, conversation screenshots, etc.):

修改前 / Before:

# PostgreSQL/asyncpg 下 row 是 Row tuple,row[0] 是主键字符串,后续访问 .id/.session_id 触发 AttributeError
result = await self.ap.persistence_mgr.execute_async(query)
row = result.first()
if row:
    return row[0]

修改后 / After:

# SQLAlchemy 2.0 风格,跨数据库后端均返回 MonitoringMessage ORM 实体
result = await self.ap.persistence_mgr.execute_async(query)
obj = result.scalars().first()
if obj:
    return obj

单元测试运行结果:

tests/unit_tests/api/test_monitoring_tool_context.py .........  [100%]
9 passed

检查清单 / Checklist

PR 作者完成 / For PR author

请在方括号间写x以打勾 / Please tick the box with x

  • 阅读仓库贡献指引了吗? / Have you read the contribution guide?
  • 我已签署或将在机器人提示后签署 CLA。 / I have signed, or will sign when prompted by the bot, the CLA.
  • 与项目所有者沟通过了吗? / Have you communicated with the project maintainer?
  • 我确定已自行测试所作的更改,确保功能符合预期。 / I have tested the changes and ensured they work as expected.

项目维护者完成 / For project maintainer

  • 相关 issues 链接了吗? / Have you linked the related issues?
  • 配置项写好了吗?迁移写好了吗?生效了吗? / Have you written the configuration items? Have you written the migration? Has it taken effect?
  • 依赖加到 pyproject.toml 和 core/bootutils/deps.py 了吗 / Have you added the dependencies to pyproject.toml and core/bootutils/deps.py?
  • 文档编写了吗? / Have you written the documentation?

@dosubot dosubot Bot added size:L This PR changes 100-499 lines, ignoring generated files. bug? Bug或Bug修复相关 / maybe a bug IM: wecom 企业微信 适配器相关 / WeCom and WeComCS adapter related m: Session 会话和消息模块 / Sessions management labels Jul 10, 2026
@MIKAZE3 MIKAZE3 changed the title Fix/postgresql errors fix(postgresql): register asyncpg JSON codecs and fix tool call recording on PG backend Jul 10, 2026

@dadachann dadachann left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这个 PR 还没拆干净:wecombot.py 不属于 PostgreSQL 修复,而且调用的三个辅助方法在本 PR 里没有实现,单独合入会直接报错。monitoring.py 的类型判断也只是返回 None,没有正确取到消息对象。请把 WeCom 改动移出去,PG codec 和 monitoring 分开补测试后再提。

@codecov

codecov Bot commented Jul 16, 2026

Copy link
Copy Markdown

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!

@MIKAZE3
MIKAZE3 force-pushed the fix/postgresql-errors branch from 431246e to cb22147 Compare July 17, 2026 04:21
@dosubot dosubot Bot added size:M This PR changes 30-99 lines, ignoring generated files. and removed size:L This PR changes 100-499 lines, ignoring generated files. labels Jul 17, 2026
@MIKAZE3
MIKAZE3 requested a review from dadachann July 17, 2026 04:33
@MIKAZE3
MIKAZE3 force-pushed the fix/postgresql-errors branch from 4877faf to 07cc868 Compare July 17, 2026 04:47
@dosubot dosubot Bot added size:L This PR changes 100-499 lines, ignoring generated files. and removed size:M This PR changes 30-99 lines, ignoring generated files. labels Jul 17, 2026
@MIKAZE3
MIKAZE3 force-pushed the fix/postgresql-errors branch from 07cc868 to 0a028e8 Compare July 17, 2026 04:52
@dosubot dosubot Bot added size:M This PR changes 30-99 lines, ignoring generated files. and removed size:L This PR changes 100-499 lines, ignoring generated files. labels Jul 17, 2026
MIKAZE3 added 3 commits July 17, 2026 12:56
asyncpg returns JSON / JSONB columns as raw strings by default, which
breaks every LangBot code path that expects a dict (mcp server loading,
bot adapter config, pipeline config, etc.) when running on PostgreSQL.

The asyncpg dialect's built-in on_connect hook should install the
codecs, but it does not fire reliably on the SQLAlchemy 2.0 async
engine path. Monkey-patch AsyncAdapt_asyncpg_dbapi.connect so that
each new connection registers json / jsonb codecs eagerly via the
wrapper's run_async hook.
…in tool context

The previous result.first() + row[0] pattern returned a raw string (primary key value) instead of the ORM entity on PostgreSQL/asyncpg. Use result.scalars().first() per SQLAlchemy 2.0 convention so the MonitoringMessage object is correctly retrieved.
@MIKAZE3
MIKAZE3 force-pushed the fix/postgresql-errors branch from 0a028e8 to c274f5c Compare July 17, 2026 04:57

@dadachann dadachann left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

先只看 #2327。这个 PR 里仍有 PostgreSQL codec 和 monitoring 两件事,请先保留一个,另一件单独提。下面两个问题目前都阻塞。

row = result.first()
if row:
return row[0]
obj = result.scalars().first()

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

execute_async 用的是 AsyncConnection,这里的 scalars().first() 仍会返回第一列,也就是 id 字符串。我用当前 SQLAlchemy 2.0.46 和真实 MonitoringMessage 跑了一次,结果是 'msg-1',不是 ORM 对象。现在的测试只是把 mock 预设成对象,没覆盖真实结果形态。请改用正确的映射/实体构造方式,并用实际 DB connection 测。

wrapper.run_async(_init)
return wrapper

AsyncAdapt_asyncpg_dbapi.connect = _patched

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

SQLAlchemy 2.0.46 的 PGDialect_asyncpg.on_connect 已经会为每个连接注册 JSON/JSONB codec。这里再 monkey-patch 私有 connect 是重复的,reload 时还会叠加 wrapper;现有测试只验证 patch 被调用,没有真实 PG 失败用例。请删除,或先给出可复现用例和真实 PostgreSQL 集成测试。

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

bug? Bug或Bug修复相关 / maybe a bug IM: wecom 企业微信 适配器相关 / WeCom and WeComCS adapter related m: Session 会话和消息模块 / Sessions management size:M This PR changes 30-99 lines, ignoring generated files.

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants