trading_agent.memory.session_store

Keeps a desk's sessions on disk, so a restart does not make the desk forget the day.

A session is one stretch of conversation, normally one India trading day. The store keeps three things in the desk's sessions folder:

sessions/ state.json which session is current, and what each backend needs to resume it 2026-10-19-1.jsonl every finished exchange of that session, one JSON line each summaries/2026-10-19.md the summary written when that day's session ended

Typical usage example:

store = SessionStore(Path('workspace/desks/main/sessions')) state = store.load_state() store.append_exchange(state.session_id, exchange)

Exchange dataclass

Exchange(trigger_text, answer, backend_name, finished_at)

One finished turn: what the desk was told and what it answered.

Attributes:

Name	Type	Description
trigger_text	str	The trigger's text.
answer	str	The final answer.
backend_name	str	The backend that answered.
finished_at	str	When the turn finished, as an ISO 8601 string in UTC.

to_json_dictionary

to_json_dictionary()

Converts the exchange to a dictionary that json.dumps can write.

Returns:

Type	Description
dict[str, Any]	The exchange's fields.

Source code in src/trading_agent/memory/session_store.py

def to_json_dictionary(self) -> dict[str, Any]:
    """Converts the exchange to a dictionary that json.dumps can write.

    Returns:
        dict[str, Any]: The exchange's fields.
    """
    return dataclasses.asdict(self)

from_json_dictionary classmethod

from_json_dictionary(data)

Reads an exchange written by to_json_dictionary.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	The dictionary.	required

Returns:

Type	Description
Exchange	The exchange.

Raises:

Type	Description
KeyError	A field is missing.

Source code in src/trading_agent/memory/session_store.py

@classmethod
def from_json_dictionary(cls, data: dict[str, Any]) -> Self:
    """Reads an exchange written by to_json_dictionary.

    Args:
        data (dict[str, Any]): The dictionary.

    Returns:
        Exchange: The exchange.

    Raises:
        KeyError: A field is missing.
    """
    return cls(
        trigger_text=data['trigger_text'],
        answer=data['answer'],
        backend_name=data['backend_name'],
        finished_at=data['finished_at'],
    )

RouteState dataclass

RouteState(exchanges_seen, runner_state)

What one backend needs to pick up the current session after a restart.

Attributes:

Name	Type	Description
exchanges_seen	int	How many of the session's exchanges the backend already knows about.
runner_state	dict[str, Any] | None	What the backend's turn runner saved, such as Claude's session id, or None when it keeps nothing.

SessionState dataclass

SessionState(session_id, trading_date, routes)

Which session is current.

Attributes:

Name	Type	Description
session_id	str	The session's identifier, such as "2026-10-19-1".
trading_date	date	The India date the session belongs to.
routes	dict[str, RouteState]	Each backend's state, keyed by the backend's name.

to_json_dictionary

to_json_dictionary()

Converts the state to a dictionary that json.dumps can write.

Returns:

Type	Description
dict[str, Any]	The state, with the date as an ISO 8601 string.

Source code in src/trading_agent/memory/session_store.py

def to_json_dictionary(self) -> dict[str, Any]:
    """Converts the state to a dictionary that json.dumps can write.

    Returns:
        dict[str, Any]: The state, with the date as an ISO 8601 string.
    """
    routes = {}
    for name, route in self.routes.items():
        routes[name] = {
            'exchanges_seen': route.exchanges_seen,
            'runner_state': route.runner_state,
        }
    return {
        'session_id': self.session_id,
        'trading_date': self.trading_date.isoformat(),
        'routes': routes,
    }

from_json_dictionary classmethod

from_json_dictionary(data)

Reads a state written by to_json_dictionary.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	The dictionary.	required

Returns:

Type	Description
SessionState	The state.

Raises:

Type	Description
KeyError	A field is missing.
ValueError	The date is not an ISO 8601 date.

Source code in src/trading_agent/memory/session_store.py

@classmethod
def from_json_dictionary(cls, data: dict[str, Any]) -> Self:
    """Reads a state written by to_json_dictionary.

    Args:
        data (dict[str, Any]): The dictionary.

    Returns:
        SessionState: The state.

    Raises:
        KeyError: A field is missing.
        ValueError: The date is not an ISO 8601 date.
    """
    routes = {}
    for name, route in data.get('routes', {}).items():
        routes[name] = RouteState(
            exchanges_seen=int(route.get('exchanges_seen', 0)),
            runner_state=route.get('runner_state'),
        )
    return cls(
        session_id=data['session_id'],
        trading_date=datetime.date.fromisoformat(data['trading_date']),
        routes=routes,
    )

DailySummary dataclass

DailySummary(trading_date, text)

The summary written when one day's session ended.

Attributes:

Name	Type	Description
trading_date	date	The day it summarises.
text	str	The summary.

SessionStore

SessionStore(directory, date_source=None)

One desk's sessions folder.

Attributes:

Name	Type	Description
directory		The sessions folder.

Describes the folder without touching the disk.

Parameters:

Name	Type	Description	Default
directory	Path	The desk's sessions folder.	required
date_source	Callable[[], date] | None	Returns today's India date; None uses the clock. Tests replace it.	None

Source code in src/trading_agent/memory/session_store.py

def __init__(
    self,
    directory: Path,
    date_source: Callable[[], datetime.date] | None = None,
):
    """Describes the folder without touching the disk.

    Args:
        directory (Path): The desk's sessions folder.
        date_source (Callable[[], datetime.date] | None): Returns today's India date; None uses the clock. Tests replace it.
    """
    self.directory = directory
    self._date_source = date_source

today

today()

Today's date in India, which decides when a new session starts.

Returns:

Type	Description
date	The date.

Source code in src/trading_agent/memory/session_store.py

def today(self) -> datetime.date:
    """Today's date in India, which decides when a new session starts.

    Returns:
        datetime.date: The date.
    """
    if self._date_source is not None:
        return self._date_source()
    return datetime.datetime.now(_INDIA).date()

load_state

load_state()

Reads which session is current.

A damaged state file is logged and treated as missing, so the desk starts a new session rather than not starting.

Returns:

Type	Description
SessionState | None	The state, or None when there is none yet.

Source code in src/trading_agent/memory/session_store.py

def load_state(self) -> SessionState | None:
    """Reads which session is current.

    A damaged state file is logged and treated as missing, so the desk starts a new session rather than not starting.

    Returns:
        SessionState | None: The state, or None when there is none yet.
    """
    path = self.directory / _STATE_FILE_NAME
    if not path.is_file():
        return None
    try:
        data = json.loads(path.read_text(encoding='utf-8'))
        return SessionState.from_json_dictionary(data)
    except OSError, ValueError, KeyError, TypeError, AttributeError:
        _LOGGER.exception('Could not read the session state %s', path)
        return None

save_state

save_state(state)

Writes which session is current.

The file is written beside the old one and then renamed over it, so a crash mid-write leaves the old state intact.

Parameters:

Name	Type	Description	Default
state	SessionState	The state.	required

Source code in src/trading_agent/memory/session_store.py

def save_state(self, state: SessionState) -> None:
    """Writes which session is current.

    The file is written beside the old one and then renamed over it, so a crash mid-write leaves the old state intact.

    Args:
        state (SessionState): The state.
    """
    self.directory.mkdir(parents=True, exist_ok=True)
    path = self.directory / _STATE_FILE_NAME
    temporary_path = path.with_suffix('.json.tmp')
    temporary_path.write_text(
        json.dumps(state.to_json_dictionary(), indent=2),
        encoding='utf-8',
    )
    os.replace(temporary_path, path)

new_session_id

new_session_id(trading_date)

Picks the next free session identifier for a day and creates its empty log.

Parameters:

Name	Type	Description	Default
trading_date	date	The session's India date.	required

Returns:

Type	Description
str	An identifier such as "2026-10-19-2".

Source code in src/trading_agent/memory/session_store.py

def new_session_id(self, trading_date: datetime.date) -> str:
    """Picks the next free session identifier for a day and creates its empty log.

    Args:
        trading_date (datetime.date): The session's India date.

    Returns:
        str: An identifier such as "2026-10-19-2".
    """
    self.directory.mkdir(parents=True, exist_ok=True)
    number = 1
    while self._log_path(f'{trading_date.isoformat()}-{number}').exists():
        number += 1
    session_id = f'{trading_date.isoformat()}-{number}'
    self._log_path(session_id).touch()
    return session_id

append_exchange

append_exchange(session_id, exchange)

Adds one finished exchange to the end of a session's log.

Parameters:

Name	Type	Description	Default
session_id	str	The session.	required
exchange	Exchange	The exchange.	required

Source code in src/trading_agent/memory/session_store.py

def append_exchange(self, session_id: str, exchange: Exchange) -> None:
    """Adds one finished exchange to the end of a session's log.

    Args:
        session_id (str): The session.
        exchange (Exchange): The exchange.
    """
    self.directory.mkdir(parents=True, exist_ok=True)
    line = json.dumps(exchange.to_json_dictionary(), ensure_ascii=False)
    with self._log_path(session_id).open('a', encoding='utf-8') as log:
        log.write(line + '\n')

exchanges

exchanges(session_id)

Reads a session's exchanges, oldest first.

A damaged line, such as one cut short by a crash, is skipped.

Parameters:

Name	Type	Description	Default
session_id	str	The session.	required

Returns:

Type	Description
list[Exchange]	The exchanges, or an empty list for an unknown session.

Source code in src/trading_agent/memory/session_store.py

def exchanges(self, session_id: str) -> list[Exchange]:
    """Reads a session's exchanges, oldest first.

    A damaged line, such as one cut short by a crash, is skipped.

    Args:
        session_id (str): The session.

    Returns:
        list[Exchange]: The exchanges, or an empty list for an unknown session.
    """
    path = self._log_path(session_id)
    if not path.is_file():
        return []
    exchanges = []
    for line in path.read_text(encoding='utf-8').splitlines():
        if not line.strip():
            continue
        try:
            exchanges.append(
                Exchange.from_json_dictionary(json.loads(line))
            )
        except ValueError, KeyError, TypeError:
            _LOGGER.warning('Skipped a damaged line in %s', path)
    return exchanges

session_ids

session_ids()

Every session that has a log, newest first.

Returns:

Type	Description
list[str]	The session identifiers.

Source code in src/trading_agent/memory/session_store.py

def session_ids(self) -> list[str]:
    """Every session that has a log, newest first.

    Returns:
        list[str]: The session identifiers.
    """
    if not self.directory.is_dir():
        return []
    ids = []
    for path in self.directory.glob('*.jsonl'):
        ids.append(path.stem)
    ids.sort(key=self._session_sort_key, reverse=True)
    return ids

summaries

summaries()

Every daily summary, newest first.

Returns:

Type	Description
list[DailySummary]	The summaries.

Source code in src/trading_agent/memory/session_store.py

def summaries(self) -> list[DailySummary]:
    """Every daily summary, newest first.

    Returns:
        list[DailySummary]: The summaries.
    """
    folder = self.directory / _SUMMARIES_FOLDER_NAME
    if not folder.is_dir():
        return []
    summaries = []
    for path in folder.glob('*.md'):
        try:
            summary_date = datetime.date.fromisoformat(path.stem)
        except ValueError:
            continue
        summaries.append(
            DailySummary(
                trading_date=summary_date,
                text=path.read_text(encoding='utf-8').strip(),
            )
        )
    summaries.sort(key=lambda summary: summary.trading_date, reverse=True)
    return summaries

write_summary

write_summary(trading_date, text)

Saves the summary of a day's session.

Parameters:

Name	Type	Description	Default
trading_date	date	The day it summarises.	required
text	str	The summary.	required

Returns:

Type	Description
Path	The summary's file.

Source code in src/trading_agent/memory/session_store.py

def write_summary(self, trading_date: datetime.date, text: str) -> Path:
    """Saves the summary of a day's session.

    Args:
        trading_date (datetime.date): The day it summarises.
        text (str): The summary.

    Returns:
        Path: The summary's file.
    """
    folder = self.directory / _SUMMARIES_FOLDER_NAME
    folder.mkdir(parents=True, exist_ok=True)
    path = folder / f'{trading_date.isoformat()}.md'
    path.write_text(text.strip() + '\n', encoding='utf-8')
    return path

latest_summary

latest_summary()

Reads the most recent daily summary.

Returns:

Type	Description
DailySummary | None	The newest summary, or None when no day has been summarised yet.

Source code in src/trading_agent/memory/session_store.py

def latest_summary(self) -> DailySummary | None:
    """Reads the most recent daily summary.

    Returns:
        DailySummary | None: The newest summary, or None when no day has been summarised yet.
    """
    folder = self.directory / _SUMMARIES_FOLDER_NAME
    if not folder.is_dir():
        return None
    newest_path = None
    newest_date = None
    for path in folder.glob('*.md'):
        try:
            summary_date = datetime.date.fromisoformat(path.stem)
        except ValueError:
            continue
        if newest_date is None or summary_date > newest_date:
            newest_date = summary_date
            newest_path = path
    if newest_path is None:
        return None
    return DailySummary(
        trading_date=newest_date,
        text=newest_path.read_text(encoding='utf-8').strip(),
    )