Session & Delivery

源码版本v2026.7.20

Responsibility

Two things: (1) Session — attribute "where the message came from" (platform, chat, sender) to a stable key, and persist the conversation to disk so the agent can continue with context across restarts; (2) Delivery Ledger — record a "delivery obligation" for the agent's final reply, so that after a crash undelivered replies can be identified and re-sent, guaranteeing the user ultimately receives them. Both live in the gateway layer, decoupled from the agent.

Design Motivation

The agent itself is stateless — it only runs a single turn. Where the context comes from and where the final reply goes should not be its concern. The session layer answers "messages from the same person in the same group on the same topic belong to the same conversation", and this layer must persist with a stable key, or the agent won't remember prior turns after a restart. The delivery ledger answers "the agent generated the reply, but the gateway process crashed mid-send()" — a half-dead reply like that means the user neither sees the full content nor can easily trigger a re-generate. The ledger records each "owed delivery" obligation in sqlite; on restart, a single sweep can re-send or mark failures, rather than throwing the work away and starting over.

Key Files

• class SessionSource:149-299 — message source dataclass (platform/chat/sender, with hash)
• class SessionContext:299-700 — session context (key, source, continuation strategy)
• id hashing:40-74 — _hash_id / _hash_sender_id / _hash_chat_id (privacy + stability)
• path safety check:100-149 — _is_path_unsafe / _is_session_key_unsafe (session keys flow into filesystem paths)
• auto_continue freshness window:26-40 — auto_continue_freshness_window
• delivery ledger docs:1-58 — design goal: best-effort, never blocks the main flow on failure
• compute_obligation_id:146-155 — unique obligation id (session key + message ref + content)
• state machine:155-203 — record_obligation / mark_attempting / mark_delivered / mark_failed
• sweep_recoverable:203-276 — scan recoverable obligations after a crash and re-send
• SQLite connection:77-102 — _db_path / _connect, persisted in local sqlite

Data Flow

1. The adapter receives a platform event and carries SessionSource out via MessageEvent:1759.
2. SessionSource:149 hashes out a stable chat/sender id (gateway/session.py:40).
3. A SessionContext:299 is constructed, yielding a session key (used as a disk path after the path safety check:109).
4. The session key decides which file to load history from, fed to the agent to continue context.
5. After the agent produces its final reply, the gateway records it in record_obligation:155 → sends → mark_delivered.
6. If a crash happens during send, the next startup's sweep_recoverable (gateway/delivery_ledger.py:203) scans for obligations in attempting state whose owning process is dead, and re-sends or marks them failed.

SessionSource hashes sensitive fields like platform + chat + sender, a straight sha256 truncated to 12 chars:

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)}"

The hashing serves two purposes: privacy (platform user IDs aren't written to disk in the clear) and stability — the same person on a new device still continues the same session. The session key later flows into a filesystem path, so a second layer of validation checks it:

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"""

The two validators are deliberately different — session_key is a logical routing key (Google Chat's spaces/<id>/threads/<id> contains / by design), so strictly rejecting / would cause false positives; but a field like session_id that flows into a file path must reject it. Distinguishing them avoids the "one validator fits all" compatibility trap.

The delivery ledger's state machine is just four methods plus one sqlite table:

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 hashes session_key + message_ref + content into 24 hex chars — re-recording the same turn with the same content is idempotent (so repeated generation doesn't bloat the ledger), and different threads on the same chat never collide on the same id.

The crash-recovery sweep_recoverable is the core of this design:

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.
    """

The key idea is "claim" — when two gateway instances scan the same row, whichever one flips owner_pid to itself first wins the right to re-send, and the UPDATE's WHERE clause ensures only one wins. deliverable_platforms prevents "wasted attempts": platforms not connected in this boot aren't claimed, leaving them for a future boot that can actually send, rather than burning toward the MAX_ATTEMPTS cap on every restart.

Boundaries and Failure

• Path key vs logical key mix-up: _is_path_unsafe strictly rejects /, while _is_session_key_unsafe only rejects .. and a leading /. Google Chat's spaces/<id> is a valid session_key but cannot be a session_id; conflating the two either causes false rejects or path traversal.
• Double delivery: when two gateways scan the same row, the claim must be atomic (UPDATE WHERE owner_pid is the old value), or the same reply gets sent twice. Recovered rows also carry needs_marker because the original send may have partially succeeded.
• Platform not connected: deliverable_platforms filters out platforms not online in this boot, so each restart doesn't burn an attempt until MAX_ATTEMPTS and abandoned. The boot that actually has the platform connected handles the re-send.
• Session expiry: auto_continue_freshness_window is the time window for "continuing the previous thread"; past the window, auto-continue is disabled, to prevent treating a hours-old conversation as the current continuation.

Summary

The session layer owns "which ongoing conversation this message belongs to", mapping external identities safely to disk files via hashing + path validation. The delivery ledger owns "whether the agent's intended final reply actually reached the user", using a sqlite state machine for crash recovery. Both are best-effort but critical reliability layers.