會話與投遞

源码版本v2026.7.20

職責

兩件事:① 會話(Session)——把「訊息從哪來」(平台、chat、發送者)歸到一個穩定 key,並持久化對話到磁碟,讓 agent 跨重啟續上下文 (context);② 投遞帳本(Delivery Ledger)——為 agent 的最終回覆記一筆「送達義務」,崩潰後能識別未投遞的回覆並補發,保證使用者最終收到。兩者都在網關 (gateway) 層,與 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. 轉接器 (adapter) 收到平台事件,經 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 但不可或缺的可靠性層。