Merge pull request '信封补齐、适当性矩阵修正与 Worker 失败原因落库' (#5) from qyqy_develop_1 into qyqy_develop

合并 qyqy_develop_1 第三轮工作：meta 信封补齐与公共化、适当性按第十二条匹配矩阵、Worker 失败原因区分固定文案与异常消息。
2026-09-11 16:42:29 +08:00
parent a122f7afba fcd3eaadd6
commit 5bbf8481ef
18 changed files with 704 additions and 68 deletions
@@ -0,0 +1,113 @@
+"""只读探查：Worker 的运行状态（队列有没有在消费）。
+
+只做 SELECT，结果写 `docs/evidence/worker-state.json`：
+
+    python tools/probe_worker_state.py
+
+三个问题：
+1. `domain_event_outbox` 里各状态各有多少条、最老的一条积压了多久
+   —— Worker 没在跑，事件就永远停在 pending；
+2. `agent_run` 里 queued/running 有多少
+   —— Worker 没在跑，用户发起的对话就一直"排队中"；
+3. `memory_episode` 有没有待提取的片段（同样是 Worker 负责消费）。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import func, select
+
+from app.infrastructure.db import SessionFactory
+from app.model.platform import AgentRun, DomainEventOutbox
+
+OUTPUT = Path("docs/evidence/worker-state.json")
+
+
+async def _grouped(session: Any, model: Any, column_name: str) -> dict[str, int]:
+    column = getattr(model, column_name, None)
+    if column is None:
+        return {}
+    rows = await session.execute(select(column, func.count()).group_by(column))
+    return {str(key): int(count) for key, count in rows.all()}
+
+
+async def collect() -> dict[str, Any]:
+    report: dict[str, Any] = {}
+    async with SessionFactory() as session:
+        report["outbox_columns"] = list(DomainEventOutbox.__table__.columns.keys())
+        report["agent_run_columns"] = list(AgentRun.__table__.columns.keys())
+
+        report["outbox_by_status"] = await _grouped(
+            session, DomainEventOutbox, "status"
+        )
+        report["agent_run_by_status"] = await _grouped(session, AgentRun, "status")
+
+        oldest = await session.scalar(
+            select(func.min(DomainEventOutbox.created_at))
+        )
+        newest = await session.scalar(
+            select(func.max(DomainEventOutbox.created_at))
+        )
+        report["outbox_created_at_range"] = [str(oldest), str(newest)]
+        report["outbox_total"] = await session.scalar(
+            select(func.count()).select_from(DomainEventOutbox)
+        )
+        report["agent_run_total"] = await session.scalar(
+            select(func.count()).select_from(AgentRun)
+        )
+
+        # 哪一类事件在堆、哪一类已经进了死信 —— 死信意味着那些事件的副作用永远不会发生。
+        report["outbox_by_event_type_status"] = [
+            {"event_type": str(row[0]), "status": str(row[1]), "count": int(row[2])}
+            for row in (
+                await session.execute(
+                    select(
+                        DomainEventOutbox.event_type,
+                        DomainEventOutbox.status,
+                        func.count(),
+                    )
+                    .group_by(DomainEventOutbox.event_type, DomainEventOutbox.status)
+                    .order_by(func.count().desc())
+                )
+            ).all()
+        ]
+        report["dead_last_errors"] = [
+            {"last_error": str(row[0])[:400], "count": int(row[1])}
+            for row in (
+                await session.execute(
+                    select(DomainEventOutbox.last_error, func.count())
+                    .where(DomainEventOutbox.status == "dead")
+                    .group_by(DomainEventOutbox.last_error)
+                    .order_by(func.count().desc())
+                )
+            ).all()
+        ]
+
+        try:
+            from app.model.memory import MemoryEpisode
+
+            report["episode_by_status"] = await _grouped(session, MemoryEpisode, "status")
+            report["episode_total"] = await session.scalar(
+                select(func.count()).select_from(MemoryEpisode)
+            )
+        except Exception as exc:  # 模型名/字段与预期不符时只记录，不影响其余结论
+            report["episode_probe_error"] = f"{type(exc).__name__}: {exc}"
+    return report
+
+
+async def main() -> None:
+    report = await collect()
+    OUTPUT.parent.mkdir(parents=True, exist_ok=True)
+    OUTPUT.write_text(
+        json.dumps(report, ensure_ascii=False, indent=2, default=str),
+        encoding="utf-8",
+    )
+    print(f"wrote {OUTPUT}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())