会话与投递

源码版本v2026.7.20

职责

两件事:① 会话(Session)——把「消息从哪来」(平台、chat、发送者)归到一个稳定 key,并持久化对话到磁盘,让 agent 跨重启续上下文;② 投递账本(Delivery Ledger)——为 agent 的最终回复记一笔「送达义务」,崩溃后能识别未投递的回复并补发,保证用户最终收到。两者都在网关层,与 agent 解耦。

设计动机

agent 本身是无状态的——它只负责跑一轮循环,上下文从哪来、最终回复到哪去,不该是它的事。会话层解决「同一个人在同一群同一主题的消息属于同一段对话」,这层必须用稳定 key 落盘,否则重启后 agent 不认前文。投递账本解决「agent 生成完了回复,但网关进程在 send() 中途崩了」——这种半路死亡的回答,用户既看不到完整内容,也容易触发重复生成。账本以 sqlite 记录每一笔「待送达」义务,重启后扫一遍就能补发或标记失败,而不是丢掉重来。

关键文件

• class SessionSource:149-299 — 消息来源数据类(platform/chat/sender,带哈希)
• class SessionContext:299-700 — 会话上下文(key、来源、续接策略)
• id 哈希:40-74 — _hash_id / _hash_sender_id / _hash_chat_id(隐私 + 稳定)
• 路径安全校验:100-149 — _is_path_unsafe / _is_session_key_unsafe(session key 流入文件系统路径)
• auto_continue 新鲜窗:26-40 — auto_continue_freshness_window
• 投递账本文档:1-58 — 设计目标:best-effort,失败永不阻塞主流程
• compute_obligation_id:146-155 — 唯一义务 id(会话 key + 消息 ref + 内容)
• 状态机:155-203 — record_obligation / mark_attempting / mark_delivered / mark_failed
• sweep_recoverable:203-276 — 崩溃后扫描可恢复义务并补发
• SQLite 连接:77-102 — _db_path / _connect,持久化在本地 sqlite

数据流

1. 适配器收到平台事件,经 MessageEvent:1759 带出 SessionSource。
2. SessionSource:149 哈希出稳定的 chat/sender id(gateway/session.py:40)。
3. 构造 SessionContext:299,得到 session key(经 路径安全校验:109 后用作磁盘路径)。
4. session key 决定历史从哪个文件加载,喂给 agent 续上下文。
5. agent 产出最终回复后,网关在 record_obligation:155 记账 → 发送 → mark_delivered。
6. 若发送中崩溃,下次启动 sweep_recoverable(gateway/delivery_ledger.py:203)扫描 attempting 状态且属主进程已死的义务,补发或标记失败。

SessionSource 哈希的是平台 + chat + sender 这种敏感字段,直接 sha256 截 12 位:

def _hash_id(value: str) -> str:
    """Deterministic 12-char hex hash of an identifier."""
    return hashlib.sha256(value.encode("utf-8")).hexdigest()[:12]

def _hash_sender_id(value: str) -> str:
    """Hash a sender ID to ``user_<12hex>``."""
    return f"user_{_hash_id(value)}"

哈希一为隐私(平台 user_id 不直接落盘),二为稳定——同一人换设备名后仍在同一 session 续上。session key 之后会流入文件系统路径,所以校验是另一道:

def _is_path_unsafe(value: object) -> bool:
    """Return True if ``value`` could traverse outside the sessions dir."""
    if not value:
        return False
    s = str(value)
    if ".." in s or "/" in s or "\\" in s:
        return True
    # Leading Windows drive path, e.g. "C:\\..." or "d:/...".
    return len(s) >= 2 and s[0].isalpha() and s[1] == ":"

def _is_session_key_unsafe(value: object) -> bool:
    """True if ``value`` could be a real traversal vector in a session_key.
    ``session_key`` is a *logical* routing key (e.g.
    ``agent:main:google_chat:group:spaces/<id>``) — it never touches the
    filesystem, so the strict separator-rejecting guard from
    ``_is_path_unsafe`` is over-broad"""

两个校验函数刻意不一样——session_key 是逻辑路由 key(Google Chat 的 spaces/<id>/threads/<id> 本来就含 /),严拒 / 会误杀;但 session_id 这种流入文件路径的字段必须严拒。区分两者避免「一套校验打天下」的兼容性坑。

投递账本的状态机就四个方法 + 一个 sqlite 表:

def compute_obligation_id(session_key: str, message_ref: str, content: str) -> str:
    """Stable id: same turn + same content re-records idempotently, while
    distinct threads/topics on the same chat can never collide."""
    payload = f"{session_key}|{message_ref}|{content}"
    return hashlib.sha256(payload.encode("utf-8", "replace")).hexdigest()[:24]

def record_obligation(*, obligation_id, session_key, platform, chat_id, thread_id, content) -> None:
    """Record a final response as owed to the platform (state='pending')."""

obligation_id 把 session_key + message_ref + content 哈希成 24 位 hex——同一轮 + 同一内容重记幂等(避免重复生成把账本撑大),不同 thread 在同一 chat 上也绝不撞 id。

崩溃恢复的 sweep_recoverable 是这套设计的核心:

def sweep_recoverable(now=None, *, deliverable_platforms=None) -> List[Dict[str, Any]]:
    """Claim undelivered rows owned by dead processes; return them for
    redelivery.

    Claiming atomically re-stamps the owner to THIS process and increments
    ``attempts``, so a second gateway racing the same sweep cannot
    double-claim (the UPDATE is guarded on the previous owner stamp).
    Rows over the attempts cap or older than the stale cutoff transition to
    'abandoned' instead of being returned.
    """

关键是「claim」——同一行被两个 gateway 实例扫到时,谁先把 owner_pid 改成自己就拿到补发权,UPDATE 的 WHERE 子句保证只有一方成功。deliverable_platforms 防「无谓烧 attempts」:本 boot 没连上的平台不被认领,留给下次有能力发的 boot,而不是每次重启都把它推近 MAX_ATTEMPTS 上限。

边界与失败

• 路径与逻辑 key 混用:_is_path_unsafe 严拒 /,_is_session_key_unsafe 只拒 .. 和 leading /。Google Chat 的 spaces/<id> 是合法 session_key 但不能当 session_id,搞混了要么误杀要么路径穿越。
• 重复投递:两个 gateway 同时扫到同一行时,claim 必须原子(UPDATE WHERE owner_pid 旧值),否则同一回复被发两次。补发的行还会带 needs_marker(因为原 send 可能半成功)。
• 平台未连上:deliverable_platforms 过滤本 boot 不在线的平台,避免每次重启都烧一次 attempts 直至 MAX_ATTEMPTS 然后 abandoned。等真连上的那次 boot 再补。
• 会话过期:auto_continue_freshness_window 是「接续上一段」的时间窗;过了窗不再 auto-continue,防止把几小时前的会话当当前接续。

小结

会话层负责「这条消息属于哪段持续对话」,用哈希 + 路径校验把外部身份安全映射到磁盘文件。投递账本负责「agent 想发的最终回复是否真的到了用户手里」,用 sqlite 状态机做崩溃恢复。两者都是 best-effort 但不可或缺的可靠性层。