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())
|