lzf_0626
8b8883ccca
起因是查 Worker 运行状态时发现库里 373 条死信的 last_error 全是裸的 "ValueError"
（工具 tools/probe_worker_state.py，证据 docs/evidence/worker-state.json）。
dispatch（run not found）、dispatch_run_completed、dispatch_memory_extraction、
dispatch_profile_rebuild 抛的都是 ValueError，只记类名等于把"哪一处失败"也一起丢了。
但"直接存 str(exc)"是错的：tests/unit/worker/test_outbox_worker.py 那条
RuntimeError("credential=do-not-log") 断言异常消息不得落库 —— 它可能含凭据、SQL 或
客户标识。第一版改动就是这么写的，被这个测试当场拦下（这测试写得值）。
折中：
- 新增 OutboxHandlerError（继承 ValueError，这些失败本就是 ValueError 语义，保持
继承关系才不会改动既有的 except ValueError 行为与断言）。它的 reason 由代码写死、
不含任何请求数据，因此可以落库；
- safe_error_text：OutboxHandlerError → "类名: 固定文案"（截断 500 字符），
其余异常 → 仍只记类名；
- runtime.py 的 5 处 handler 失败改抛 OutboxHandlerError。
测试：新增"固定文案落库"用例；并把既有用例的断言收紧为 last_error == "RuntimeError"
（原先只断言"不含 do-not-log"，太松，漏掉的情况测不出来）。
顺带产出 tools/probe_worker_state.py（只读）：outbox / agent_run 各状态计数、按事件
类型分组、死信原因聚合。当前环境实测 pending 347、dead 373、published 410、
agent_run 无 queued/running。
门禁：ruff 干净 / mypy 138 文件 / 697 unit+contract / 33 integration。
114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
"""只读探查：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())
|