fix: scope postgres telemetry test to run-prefixed rows

The fixture used to DROP the shared llm_calls table on every run, wiping
concurrent migration-batch telemetry (and its own count assertion was
polluted in return). Assertions now filter by a per-run call_id prefix
and teardown deletes only its own rows.

tests/integration/test_postgres_telemetry.py

@@ -2,12 +2,17 @@
 
 DSN 走 .env `PGW_TELEMETRY_PG_DSN`,缺则 skip。该实例上有 app/chs_prod 等
 在用库——本测试只允许连 polygateway 专用库(fixture 里守卫)。
+
+隔离纪律(M4 事故教训): `llm_calls` 是与真实批跑/迁移项目共享的表,
+**严禁 DROP/TRUNCATE**——本测试以 run 级 call_id 前缀隔离,断言只看
+自己写入的行,teardown 只删自己的行。
 """
 
 from __future__ import annotations
 
 import asyncio
 import os
+from uuid import uuid4
 
 import pytest
 from dotenv import dotenv_values
@@ -36,6 +41,13 @@ _EXPECTED_COLUMNS = [
     "created_at",
 ]
 
+# run 级前缀: 同库并存的其他运行(迁移批跑/另一开发机)互不可见
+_RUN_PREFIX = f"pgwtest-{uuid4().hex[:8]}"
+
+
+def _cid(suffix: str) -> str:
+    return f"{_RUN_PREFIX}-{suffix}"
+
 
 def _dsn() -> str | None:
     merged = {**dotenv_values(".env"), **os.environ}
@@ -54,19 +66,23 @@ async def dsn():
     # 隔离守卫: 该实例有 app/chs_prod/mimiciv 等在用库,只许打 polygateway 专用库
     if not value.rstrip("/").endswith("/polygateway"):
         pytest.fail(f"遥测测试只允许连 polygateway 专用库,当前 DSN 库名不符: {value!r}")
+    yield value
+    # teardown: 只删本 run 写入的行;表可能尚不存在(全新库)则忽略
     import asyncpg
 
     conn = await asyncpg.connect(value, timeout=10)
     try:
-        await conn.execute("DROP TABLE IF EXISTS llm_calls")
+        if await conn.fetchval("SELECT to_regclass('llm_calls')") is not None:
+            await conn.execute(
+                "DELETE FROM llm_calls WHERE call_id LIKE $1", f"{_RUN_PREFIX}-%"
+            )
     finally:
         await conn.close()
-    return value
 
 
-async def _record_minimal(recorder: PostgresRecorder, call_id: str = "c1", **overrides) -> None:
+async def _record_minimal(recorder: PostgresRecorder, call_id: str | None = None, **overrides) -> None:
     fields = {
-        "call_id": call_id,
+        "call_id": call_id if call_id is not None else _cid("c1"),
         "parent_call_id": None,
         "session_id": "sess-1",
         "model": "m",
@@ -89,12 +105,12 @@ async def _record_minimal(recorder: PostgresRecorder, call_id: str = "c1", **ove
     await recorder.record_llm_call(**fields)
 
 
-async def _fetch(dsn: str, sql: str):
+async def _fetch(dsn: str, sql: str, *args):
     import asyncpg
 
     conn = await asyncpg.connect(dsn, timeout=10)
     try:
-        return await conn.fetch(sql)
+        return await conn.fetch(sql, *args)
     finally:
         await conn.close()
 
@@ -116,9 +132,11 @@ class TestSchema:
     async def test_call_id_idempotent(self, dsn):
         recorder = PostgresRecorder(dsn)
         try:
-            await _record_minimal(recorder, call_id="dup")
+            await _record_minimal(recorder, call_id=_cid("dup"))
-            await _record_minimal(recorder, call_id="dup", response="second")
+            await _record_minimal(recorder, call_id=_cid("dup"), response="second")
-            rows = await _fetch(dsn, "SELECT response FROM llm_calls WHERE call_id='dup'")
+            rows = await _fetch(
+                dsn, "SELECT response FROM llm_calls WHERE call_id = $1", _cid("dup")
+            )
             assert [r["response"] for r in rows] == ["ok"]  # ON CONFLICT DO NOTHING
         finally:
             await recorder.aclose()
@@ -126,8 +144,14 @@ class TestSchema:
     async def test_concurrent_writes_all_land(self, dsn):
         recorder = PostgresRecorder(dsn)
         try:
-            await asyncio.gather(*(_record_minimal(recorder, call_id=f"c{i}") for i in range(50)))
+            await asyncio.gather(
-            rows = await _fetch(dsn, "SELECT count(*) AS n FROM llm_calls")
+                *(_record_minimal(recorder, call_id=_cid(f"c{i}")) for i in range(50))
+            )
+            rows = await _fetch(
+                dsn,
+                "SELECT count(*) AS n FROM llm_calls WHERE call_id LIKE $1",
+                f"{_RUN_PREFIX}-c%",
+            )
             assert rows[0]["n"] == 50
         finally:
             await recorder.aclose()
@@ -138,17 +162,21 @@ class TestDegradation:
         """结构性失败(建池不通)→ warning 一次后永久降级,业务零感知。"""
         recorder = PostgresRecorder("postgresql://u:p@127.0.0.1:1/x")
         await _record_minimal(recorder)  # 不抛
-        await _record_minimal(recorder, call_id="c2")  # 已降级短路,同样不抛
+        await _record_minimal(recorder, call_id=_cid("c2"))  # 已降级短路,同样不抛
         await recorder.aclose()
 
     async def test_row_failure_does_not_poison_later_rows(self, dsn):
         """运行时单条写失败(NUL 字节文本被 PG 拒)→ 丢该行,后续行照常落库。"""
         recorder = PostgresRecorder(dsn)
         try:
-            await _record_minimal(recorder, call_id="bad", response="nul\x00byte")
+            await _record_minimal(recorder, call_id=_cid("bad"), response="nul\x00byte")
-            await _record_minimal(recorder, call_id="good")
+            await _record_minimal(recorder, call_id=_cid("good"))
-            rows = await _fetch(dsn, "SELECT call_id FROM llm_calls ORDER BY call_id")
+            rows = await _fetch(
-            assert [r["call_id"] for r in rows] == ["good"]
+                dsn,
+                "SELECT call_id FROM llm_calls WHERE call_id = ANY($1::text[]) ORDER BY call_id",
+                [_cid("bad"), _cid("good")],
+            )
+            assert [r["call_id"] for r in rows] == [_cid("good")]
         finally:
             await recorder.aclose()