Skip to content

symfonic.agent.sessions

sessions

In-memory chat session manager.

Manages chat sessions per tenant. Each session tracks its ID, tenant, creation time, last activity, and message count. Sessions share the semantic memory layer (tenant-wide) but maintain isolated working memory contexts.

Storage is in-memory only -- a future enhancement will add persistence.

Session storage is partitioned by tenant to prevent cross-tenant leakage even if session IDs collide, and to keep listing costs bounded per tenant. A per-tenant cap (MAX_SESSIONS_PER_TENANT) is enforced via LRU eviction on last_active -- oldest 10 % is dropped when the cap is hit to amortize eviction cost.

SessionManager 

SessionManager()

Manages chat sessions per tenant.

Thread-safety: This class is NOT thread-safe. In async contexts, all access should happen on the same event loop (no concurrent mutations from multiple threads).

Storage layout::

_sessions: tenant_id -> { session_id -> session_dict }
_tenant_index: session_id -> tenant_id   # reverse lookup for
                                          # the legacy
                                          # get_session/touch
                                          # by-id-only API.
Source code in src/symfonic/agent/sessions.py 
46
47
48
49
50
51
52
def __init__(self) -> None:
    # Partitioned storage: tenant_id -> { session_id -> session data }.
    self._sessions: dict[str, dict[str, dict]] = {}
    # Reverse index so get_session(session_id) and touch(session_id)
    # keep their existing single-arg signature without scanning all
    # tenants.
    self._tenant_index: dict[str, str] = {}

create_session 

create_session(tenant_id: str) -> str

Create a new session and return its ID.

Source code in src/symfonic/agent/sessions.py 
104
105
106
107
108
def create_session(self, tenant_id: str) -> str:
    """Create a new session and return its ID."""
    session_id = uuid.uuid4().hex
    self._register(tenant_id, session_id)
    return session_id

ensure_session 

ensure_session(
    tenant_id: str, session_id: str | None
) -> str

Return existing session_id or create a new one.

Cross-tenant safety: if session_id was already registered under a DIFFERENT tenant, the collision is refused silently by creating a brand-new row in tenant_id's bucket. The other tenant's entry stays intact. This prevents an attacker from guessing another tenant's session id and piggy-backing onto it.

Source code in src/symfonic/agent/sessions.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def ensure_session(self, tenant_id: str, session_id: str | None) -> str:
    """Return existing session_id or create a new one.

    Cross-tenant safety: if ``session_id`` was already registered
    under a DIFFERENT tenant, the collision is refused silently by
    creating a brand-new row in ``tenant_id``'s bucket. The other
    tenant's entry stays intact. This prevents an attacker from
    guessing another tenant's session id and piggy-backing onto it.
    """
    if session_id is not None:
        existing_tenant = self._tenant_index.get(session_id)
        if existing_tenant == tenant_id:
            return session_id
        if existing_tenant is None:
            # Unknown ID -- register under this tenant.
            self._register(tenant_id, session_id)
            return session_id
        # Cross-tenant collision: hand out a fresh session for the
        # requesting tenant. Never touch the other tenant's row.
        logger.warning(
            "SessionManager: cross-tenant session_id collision refused "
            "(requested tenant=%s, existing tenant=%s). Issuing fresh id.",
            tenant_id,
            existing_tenant,
        )
        return self.create_session(tenant_id)
    return self.create_session(tenant_id)

get_session 

get_session(session_id: str) -> dict | None

Get session metadata, or None if not found.

Source code in src/symfonic/agent/sessions.py 
110
111
112
113
114
115
def get_session(self, session_id: str) -> dict | None:
    """Get session metadata, or None if not found."""
    tenant_id = self._tenant_index.get(session_id)
    if tenant_id is None:
        return None
    return self._sessions.get(tenant_id, {}).get(session_id)

list_sessions 

list_sessions(tenant_id: str) -> list[dict]

List all sessions for a tenant, ordered by last_active descending.

Source code in src/symfonic/agent/sessions.py 
117
118
119
120
121
122
def list_sessions(self, tenant_id: str) -> list[dict]:
    """List all sessions for a tenant, ordered by last_active descending."""
    bucket = self._sessions.get(tenant_id, {})
    sessions = list(bucket.values())
    sessions.sort(key=lambda s: s["last_active"], reverse=True)
    return sessions

touch 

touch(session_id: str) -> None

Update last_active timestamp and increment message count.

No-op if session_id is not found.

Source code in src/symfonic/agent/sessions.py 
124
125
126
127
128
129
130
131
132
133
134
135
def touch(self, session_id: str) -> None:
    """Update last_active timestamp and increment message count.

    No-op if session_id is not found.
    """
    tenant_id = self._tenant_index.get(session_id)
    if tenant_id is None:
        return
    session = self._sessions.get(tenant_id, {}).get(session_id)
    if session is not None:
        session["last_active"] = datetime.now(UTC).isoformat()
        session["message_count"] += 1