From b5002a9864a09018a9aa611e62e1560cc6a23273 Mon Sep 17 00:00:00 2001
From: Yaojia Wang <yaojia.wang@billo.life>
Date: Wed, 24 Jun 2026 12:40:40 +0200
Subject: [PATCH] =?UTF-8?q?fix(chain):=20checkpointer=20=E8=BF=9E=E6=8E=A5?=
 =?UTF-8?q?=E4=B8=B2=E5=89=A5=E7=A6=BB=20SQLAlchemy=20=E9=A9=B1=E5=8A=A8?=
 =?UTF-8?q?=E5=90=8E=E7=BC=80?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

多章链 job 一发起即 failed（exc_type=ProgrammingError）。根因：runtime
checkpointer 工厂把 settings 的 sync URL `postgresql+psycopg://…` 原样喂给
`AsyncPostgresSaver.from_conn_string`，psycopg 不识别 `+psycopg` 方言段 →
`ProgrammingError: missing "=" …`，链在打开 saver 时即崩。

E2E 用 MemorySaver 覆盖 get_checkpointer_factory，从不走真 conn string，故此
路径无回归保护、bug 未被测出（迁移 d3e4f5a6b7c8 里 `_psycopg_conn_string` 已
正确剥离，runtime 缝漏了同款处理）。

修复：chain_deps 加 `_libpq_conn_string()`，剥离 `+psycopg`/`+asyncpg` 还原为
标准 libpq 串后再交 from_conn_string。+ 单测 test_chain_deps.py（3 参数化用例）。
已直连真 pg 复验：saver.aget OK，ProgrammingError 消除。
---
 apps/api/tests/test_chain_deps.py      | 42 ++++++++++++++++++++++++++
 apps/api/ww_api/services/chain_deps.py | 17 +++++++++--
 2 files changed, 57 insertions(+), 2 deletions(-)
 create mode 100644 apps/api/tests/test_chain_deps.py

diff --git a/apps/api/tests/test_chain_deps.py b/apps/api/tests/test_chain_deps.py
new file mode 100644
index 0000000..5f0c2ec
--- /dev/null
+++ b/apps/api/tests/test_chain_deps.py
@@ -0,0 +1,42 @@
+"""chain_deps._libpq_conn_string 单测——剥离 SQLAlchemy 驱动后缀（回归）。
+
+bug：runtime checkpointer 工厂曾把 `postgresql+psycopg://…`（settings 的 sync URL）
+原样喂给 `AsyncPostgresSaver.from_conn_string`，psycopg 不识别 `+psycopg` 方言段 →
+`ProgrammingError: missing "=" …` → 多章链 job 直接 failed。E2E 用 MemorySaver 覆盖，
+从不走真 conn string，故此路径无回归保护。本用例锁住「剥离后缀」语义。
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from ww_api.services import chain_deps
+
+
+class _StubSettings:
+    def __init__(self, url: str) -> None:
+        self.database_url_sync = url
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [
+        (
+            "postgresql+psycopg://writer:writer@localhost:5432/writer",
+            "postgresql://writer:writer@localhost:5432/writer",
+        ),
+        ("postgresql+asyncpg://u:p@h:5432/db", "postgresql://u:p@h:5432/db"),
+        ("postgresql://u:p@h:5432/db", "postgresql://u:p@h:5432/db"),
+    ],
+)
+def test_libpq_conn_string_strips_driver_suffix(
+    monkeypatch: pytest.MonkeyPatch, raw: str, expected: str
+) -> None:
+    # Arrange
+    monkeypatch.setattr(chain_deps, "get_settings", lambda: _StubSettings(raw))
+    # Act
+    result = chain_deps._libpq_conn_string()
+    # Assert
+    assert result == expected
+    assert "+psycopg" not in result
+    assert "+asyncpg" not in result
diff --git a/apps/api/ww_api/services/chain_deps.py b/apps/api/ww_api/services/chain_deps.py
index e354466..a02850d 100644
--- a/apps/api/ww_api/services/chain_deps.py
+++ b/apps/api/ww_api/services/chain_deps.py
@@ -28,6 +28,18 @@ if TYPE_CHECKING:
 CheckpointerFactory = Callable[[], AbstractAsyncContextManager[Any]]
 
 
+def _libpq_conn_string() -> str:
+    """从 settings 的 sync URL 派生 psycopg 的 libpq conn string。
+
+    `database_url_sync` 带 SQLAlchemy 驱动后缀（``postgresql+psycopg://…``），但
+    `AsyncPostgresSaver.from_conn_string` 直接交给 psycopg 解析——psycopg 不识别
+    ``+psycopg``/``+asyncpg`` 方言段，会抛 ``ProgrammingError: missing "=" …``。
+    故剥离后缀还原为标准 libpq 串（``postgresql://…``）。与迁移 `d3e4f5a6b7c8`
+    的 `_psycopg_conn_string()` 同义。
+    """
+    return get_settings().database_url_sync.replace("+psycopg", "").replace("+asyncpg", "")
+
+
 def get_checkpointer_factory() -> CheckpointerFactory:
     """运行时 checkpointer 工厂：建连 `DATABASE_URL` 的 `AsyncPostgresSaver` 上下文。
 
@@ -38,7 +50,8 @@ def get_checkpointer_factory() -> CheckpointerFactory:
     def _factory() -> AbstractAsyncContextManager[BaseCheckpointSaver[Any]]:
         from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
 
-        # langgraph 的 Postgres saver 走 psycopg(同步驱动串)；用 sync 串建连接。
-        return AsyncPostgresSaver.from_conn_string(get_settings().database_url_sync)
+        # langgraph 的 Postgres saver 走 psycopg(同步驱动串)；须剥离 SQLAlchemy 驱动
+        # 后缀（+psycopg），否则 psycopg 解析连接串报 ProgrammingError。
+        return AsyncPostgresSaver.from_conn_string(_libpq_conn_string())
 
     return _factory