セッションと配信

源码版本v2026.7.20

職務

二つの役:① セッション(Session)——「メッセージの出所」(プラットフォーム、chat、送信者)を安定した key にまとめ、対話をディスクに永続化し、agent が再起動を跨いでコンテキストを継続できるようにする。② 配信台帳(Delivery Ledger)——agent の最終返信に「送達義務」を1件記録し、クラッシュ時に未配信の返信を識別して再送し、ユーザーに最終的に届ける。どちらもゲートウェイ (gateway) 層にあり、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. アダプタ (adapter) がプラットフォームイベントを受け取り、MessageEvent:1759 経由で SessionSource を持ち出す。
2. SessionSource:149 が安定した chat/sender id をハッシュする(gateway/session.py:40)。
3. SessionContext:299 を構築し、session key を得る(パス安全校験:109 経由でディスクパスに使う)。
4. session key がどのファイルから履歴を読むかを決め、agent に渡してコンテキスト (context) を継続する。
5. agent が最終返信を出した後、ゲートウェイは record_obligation:155 で記帳 → 送信 → mark_delivered。
6. 送信中のクラッシュなら、次回起動時に sweep_recoverable(gateway/delivery_ledger.py:203) が attempting 状態かつ所有プロセスが死んだ義務を走査し、再送または失敗マークを付ける。

SessionSource がハッシュするのは platform + 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 にハッシュする——同一ラウンド + 同一内容の再記録は冪等(重複生成で台帳が膨らむのを避ける)で、同じ chat 上の異なる thread が絶対に 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 だが不可欠な信頼性層。