app/worker/outbox_worker.py

@@ -8,6 +8,41 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.model.platform import DomainEventOutbox, OutboxDelivery
 
 
+class OutboxHandlerError(ValueError):
+    """Handler 失败，且失败原因是**可以安全落库**的固定文案。
+
+    继承 `ValueError` 而不是 `Exception`：这些失败（run not found、payload 不完整、
+    mode 非法）本来就是 ValueError 语义，保持继承关系才不会改动既有的 `except
+    ValueError` 行为与断言。
+
+    为什么需要这个类型：`last_error` 默认只记异常类名，因为异常消息可能含凭据、SQL
+    语句或客户标识（`tests/unit/worker/test_outbox_worker.py` 里那条
+    `RuntimeError("credential=do-not-log")` 就是守这条的）。
+
+    但只记类名又不够 —— `dispatch`（run not found）、`dispatch_run_completed`、
+    `dispatch_memory_extraction`、`dispatch_profile_rebuild` 抛的全是 ValueError，实测
+    库里 373 条死信的 `last_error` 都是裸的 `"ValueError"`，分不清是哪一处失败的。
+
+    折中办法：handler 想让人看见原因时，抛这个类型，`reason` 由**代码写死**、不含任何
+    请求数据，于是可以落库；其余异常仍然只记类名。
+    """
+
+    def __init__(self, reason: str) -> None:
+        super().__init__(reason)
+        self.reason = reason
+
+
+def safe_error_text(exc: BaseException) -> str:
+    """把异常转成可落库的失败原因。
+
+    - `OutboxHandlerError` → `类名: 固定文案`（文案由代码写死，安全）；
+    - 其他异常 → **只记类名**（消息可能含凭据/请求数据，不落库）。
+    """
+    if isinstance(exc, OutboxHandlerError):
+        return f"{type(exc).__name__}: {exc.reason}"[:500]
+    return type(exc).__name__
+
+
 class OutboxWorker:
     """One dispatcher per event type, not a fan-out transport.
 
@@ -71,7 +106,8 @@ class OutboxWorker:
                 await self.session.flush()
         except Exception as exc:
             event.retry_count += 1
-            event.last_error = type(exc).__name__
+            # 只让"代码写死的固定文案"落库；异常消息可能含凭据，见 safe_error_text。
+            event.last_error = safe_error_text(exc)
             event.status = "dead" if event.retry_count >= 5 else "failed"
             event.next_retry_at = now + timedelta(seconds=min(300, 2**event.retry_count))
         else:

app/worker/runtime.py

@@ -41,7 +41,7 @@ from app.worker.episode_worker import (
     EpisodeWorker,
 )
 from app.worker.memory_extraction_worker import MemoryExtractionWorker
-from app.worker.outbox_worker import OutboxWorker
+from app.worker.outbox_worker import OutboxHandlerError, OutboxWorker
 
 logger = logging.getLogger(__name__)
 
@@ -119,11 +119,11 @@ class WorkerRuntime:
             async def dispatch(payload: dict[str, Any]) -> None:
                 run = await AgentRunRepository(session).get(str(payload["run_id"]))
                 if run is None:
-                    raise ValueError("run not found")
+                    raise OutboxHandlerError("run not found")
 
             async def dispatch_memory_extraction(payload: dict[str, Any]) -> None:
                 if "message_id" not in payload or "customer_id" not in payload:
-                    raise ValueError("memory extraction payload is incomplete")
+                    raise OutboxHandlerError("memory extraction payload is incomplete")
                 # 幂等键只认事件 id，由 worker 自己按 payload 回查，避免调用方漏传。
                 # 注入召回缓存适配器：写入生效后立即失效该客户的热缓存。
                 await MemoryExtractionWorker(
@@ -135,7 +135,7 @@ class WorkerRuntime:
                 # "运行已完成"的对外通知职责。当前没有独立外部消费者，
                 # 这里显式消费以免事件永久滞留；接入推送链路时在此处扩展。
                 if not str(payload.get("run_id", "")):
-                    raise ValueError("agent.run_completed payload is incomplete")
+                    raise OutboxHandlerError("agent.run_completed payload is incomplete")
 
             async def dispatch_cache_invalidate(payload: dict[str, Any]) -> None:
                 await self._invalidate_config_cache(payload)
@@ -149,7 +149,7 @@ class WorkerRuntime:
                 """
                 customer_id = payload.get("customer_id")
                 if not customer_id:
-                    raise ValueError("profile.rebuild_requested payload is incomplete")
+                    raise OutboxHandlerError("profile.rebuild_requested payload is incomplete")
                 # 延迟导入：bootstrap 会间接导入本模块，模块级导入会形成循环依赖
                 from app.service.profile_assembly_service import ProfileAssemblyService
                 from app.service.profile_graph_projection_service import (
@@ -181,7 +181,7 @@ class WorkerRuntime:
                     raise ValueError("memory.deletion_requested payload is incomplete")
                 mode = str(payload.get("mode", "invalidate"))
                 if mode not in {"invalidate", "delete"}:
-                    raise ValueError("memory.deletion_requested mode is invalid")
+                    raise OutboxHandlerError("memory.deletion_requested mode is invalid")
                 await MemoryLifecycleService(session).run(
                     int(customer_id),
                     mode=cast("Mode", mode),

docs/evidence/worker-state.json

@@ -0,0 +1,135 @@
+{
+  "outbox_columns": [
+    "id",
+    "event_id",
+    "event_type",
+    "aggregate_type",
+    "aggregate_id",
+    "trace_id",
+    "payload",
+    "status",
+    "retry_count",
+    "next_retry_at",
+    "last_error",
+    "occurred_at",
+    "published_at",
+    "created_at",
+    "updated_at"
+  ],
+  "agent_run_columns": [
+    "id",
+    "run_id",
+    "idempotency_id",
+    "session_id",
+    "user_id",
+    "agent_type",
+    "trace_id",
+    "request_message_id",
+    "result_message_id",
+    "status",
+    "attempt_count",
+    "worker_id",
+    "locked_until",
+    "error_code",
+    "result_version",
+    "cancel_requested_at",
+    "started_at",
+    "completed_at",
+    "created_at",
+    "updated_at"
+  ],
+  "outbox_by_status": {
+    "dead": 373,
+    "failed": 3,
+    "pending": 347,
+    "published": 410
+  },
+  "agent_run_by_status": {
+    "failed": 21,
+    "succeeded": 184
+  },
+  "outbox_created_at_range": [
+    "2026-09-09 06:55:06.787006",
+    "2026-09-11 07:21:23.195457"
+  ],
+  "outbox_total": 1133,
+  "agent_run_total": 205,
+  "outbox_by_event_type_status": [
+    {
+      "event_type": "agent.run_requested",
+      "status": "dead",
+      "count": 373
+    },
+    {
+      "event_type": "agent.run_requested",
+      "status": "pending",
+      "count": 132
+    },
+    {
+      "event_type": "agent.run_requested",
+      "status": "published",
+      "count": 130
+    },
+    {
+      "event_type": "agent.run_completed",
+      "status": "published",
+      "count": 118
+    },
+    {
+      "event_type": "memory.extraction_requested",
+      "status": "published",
+      "count": 106
+    },
+    {
+      "event_type": "profile.rebuild_requested",
+      "status": "pending",
+      "count": 76
+    },
+    {
+      "event_type": "agent.run_completed",
+      "status": "pending",
+      "count": 66
+    },
+    {
+      "event_type": "memory.extraction_requested",
+      "status": "pending",
+      "count": 60
+    },
+    {
+      "event_type": "config.cache_invalidate_requested",
+      "status": "published",
+      "count": 42
+    },
+    {
+      "event_type": "profile.rebuild_requested",
+      "status": "published",
+      "count": 13
+    },
+    {
+      "event_type": "config.cache_invalidate_requested",
+      "status": "pending",
+      "count": 13
+    },
+    {
+      "event_type": "agent.run_requested",
+      "status": "failed",
+      "count": 3
+    },
+    {
+      "event_type": "memory.invalidated",
+      "status": "published",
+      "count": 1
+    }
+  ],
+  "dead_last_errors": [
+    {
+      "last_error": "ValueError",
+      "count": 372
+    },
+    {
+      "last_error": "no handler registered",
+      "count": 1
+    }
+  ],
+  "episode_probe_error": "ImportError: cannot import name 'MemoryEpisode' from 'app.model.memory' (C:\\Users\\Windows\\Desktop\\项目代码\\app\\model\\memory.py)"
+}

tests/unit/worker/test_outbox_worker.py

@@ -4,7 +4,7 @@ from unittest.mock import AsyncMock, Mock
 import pytest
 
 from app.model.platform import DomainEventOutbox
-from app.worker.outbox_worker import OutboxWorker
+from app.worker.outbox_worker import OutboxHandlerError, OutboxWorker
 
 
 def event() -> DomainEventOutbox:
@@ -66,9 +66,32 @@ async def test_handler_failure_records_attempt(attempts: int, expected: str) ->
     assert row.status == expected
     assert row.retry_count == attempts + 1
     assert "do-not-log" not in (row.last_error or "")
+    # 其他异常一律只记类名：消息可能含凭据/SQL/客户标识，不落库。
+    assert row.last_error == "RuntimeError"
     session.commit.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+async def test_handler_error_reason_is_recorded() -> None:
+    """`OutboxHandlerError` 的固定文案可以落库 —— 否则失败原因根本无从分辨。
+
+    实测库里 373 条死信的 `last_error` 全是裸的 `"ValueError"`：`dispatch`（run not
+    found）、`dispatch_run_completed`、`dispatch_memory_extraction`、
+    `dispatch_profile_rebuild` 抛的都是 ValueError，只记类名等于把"哪一处失败"也丢了。
+    这些 handler 现在改抛 `OutboxHandlerError`，它的 reason 由代码写死、不含请求数据。
+    """
+    row = event()
+    session = AsyncMock()
+    session.add = Mock()
+    session.scalar.side_effect = [row, None]
+    handler = AsyncMock(side_effect=OutboxHandlerError("run not found"))
+
+    await OutboxWorker(session, {"test.event": handler}).publish_one()
+
+    assert row.last_error == "OutboxHandlerError: run not found"
+    assert row.status == "failed"
+
+
 @pytest.mark.asyncio
 async def test_empty_matching_queue_releases_transaction() -> None:
     session = AsyncMock()

tools/probe_worker_state.py

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