gate.ledger: Session ledger#

Agent session ledger adapter: lifecycle parsing and session-spend correlation. Distills Claude Code and Factory Droid JSONL session transcripts into SessionRecord dataclasses, correlates them with LiteLLM gateway spend rows, and aggregates them into a SessionSummary for CLI reporting.

The module is deliberately content-free: it never retains or exposes prompts, responses, reasoning, or tool arguments from the parsed transcripts.

Session parsing#

gate.ledger.parse_session_file(path: Path, *, idle: timedelta = datetime.timedelta(seconds=900)) → SessionRecord | None#

Parse a single agent session transcript file into a SessionRecord.

Dispatches by harness detection: Droid sessions live under ~/.factory/sessions/ and have session_start events; Claude Code sessions live under ~/.claude/projects/ and have assistant events with stop_reason. When the harness is ambiguous, the file is probed for a session_start event type (Droid) or falls back to Claude Code.

Parameters:

  • path – The session transcript .jsonl file.

  • idle – Gaps longer than this count as suspended time.

Returns:

The aggregated record, or None if no timed events exist.

gate.ledger.discover_sessions(*, since: datetime, until: datetime, harness_dir: Path | None = None) → list[Path]#

Find session transcript files in a time window.

Scans the built-in surface roots (Claude Code and Droid) by default, or a single custom harness_dir when provided. Files are filtered by modification time falling within the [since, until] window.

Parameters:

  • since – Window lower bound (inclusive).

  • until – Window upper bound (inclusive).

  • harness_dir – Override the scan root. When None, scans all built-in surface roots.

Returns:

Sorted list of session transcript paths.

gate.ledger.parse_window(since: datetime, until: datetime, *, harness_dir: Path | None = None, idle: timedelta = datetime.timedelta(seconds=900)) → list[SessionRecord]#

Discover and parse all session transcripts in a time window.

Parameters:

  • since – Window lower bound (inclusive).

  • until – Window upper bound (inclusive).

  • harness_dir – Override the scan root (default: scan all built-in surfaces).

  • idle – Gaps longer than this count as suspended time.

Returns:

List of parsed session records, sorted by end time.

Data models#

class gate.ledger.SessionRecord(session_id: str = '', project: str = '', harness: str = 'claude', models: list[str] = <factory>, model_request_ids: list[str] = <factory>, tokens_in: int = 0, tokens_out: int = 0, cache_creation: int = 0, cache_read: int = 0, reasoning: int = 0, started_at: datetime | None = None, ended_at: datetime | None = None, duration_s: float = 0.0, tool_calls: int = 0, cost_status: str = 'unavailable', cost_microusd: int | None = None, cost_source: str = 'none')#

One agent session’s lifecycle, distilled from its transcript.

This is a stdlib dataclass (not pydantic) for standalone use in myGate. Fields capture only structural metadata – never prompt, response, or tool content.

session_id#

The transcript’s session identifier (typically the file stem).

Type:

str

project#

Project or repo name derived from the transcript path or cwd.

Type:

str

harness#

The agent surface (‘claude’, ‘droid’, etc.).

Type:

str

models#

Model names used in the session.

Type:

list[str]

model_request_ids#

Private request IDs for spend correlation.

Type:

list[str]

tokens_in#

Input tokens (uncached + cache creation).

Type:

int

tokens_out#

Output tokens generated.

Type:

int

cache_creation#

Cache creation input tokens.

Type:

int

cache_read#

Cache read input tokens.

Type:

int

reasoning#

Reasoning/thinking tokens.

Type:

int

started_at#

First event timestamp, or None if no timed events.

Type:

datetime | None

ended_at#

Last event timestamp, or None if no timed events.

Type:

datetime | None

duration_s#

Active duration in seconds (gap analysis under idle threshold).

Type:

float

tool_calls#

Count of tool_use blocks across the session.

Type:

int

cost_status#

Correlation result status (default ‘unavailable’).

Type:

str

cost_microusd#

Cost in micro-USD if available, else None.

Type:

int | None

cost_source#

Provenance of the cost (‘litellm’, ‘local-vllm’, ‘none’, etc.).

Type:

str

class gate.ledger.SessionSummary(total_sessions: int = 0, total_tokens_in: int = 0, total_tokens_out: int = 0, total_cost_microusd: int = 0, by_model: dict[str, int]=<factory>, by_project: dict[str, int]=<factory>, sessions: list[SessionRecord] = <factory>)#

Aggregated view of multiple session records.

total_sessions#

Count of sessions included.

Type:

int

total_tokens_in#

Sum of all input tokens.

Type:

int

total_tokens_out#

Sum of all output tokens.

Type:

int

total_cost_microusd#

Sum of all available costs (0 if none available).

Type:

int

by_model#

Session count per model name.

Type:

dict[str, int]

by_project#

Session count per project name.

Type:

dict[str, int]

sessions#

The underlying session records.

Type:

list[SessionRecord]

Cost correlation#

gate.ledger.correlate_session_cost(record: SessionRecord, spend_rows: list[dict]) → SessionRecord#

Correlate a session’s request IDs with LiteLLM spend rows.

Adapts corpus model_cost.correlate_session. Matches the session’s model_request_ids against spend_rows (each a dict with request_id and spend fields). When every observed request has an exact spend row, the total is summed and set on the record. Partial matches are discarded (no partial numeric totals).

Parameters:

  • record – The session record to correlate (modified in place and returned).

  • spend_rows – LiteLLM spend rows, each with request_id and spend.

Returns:

The updated record with cost_status, cost_microusd, and cost_source fields set.

gate.ledger.approximate_session_cost(record: SessionRecord, spend_window: list[dict], *, buffer_minutes: int = 5, model_min_score: float = 0.3) → SessionRecord#

Time-window approximate cost join when exact request ID matching isn’t available.

Adapts corpus model_cost.approximate_session_cost. Matches every spend row whose start_time (or startTime) falls within the session’s time span (padded by buffer_minutes) and whose model label has at least model_min_score similarity to one of the session’s recorded models.

Parameters:

  • record – The session record to correlate (modified in place and returned).

  • spend_window – Gateway spend rows, each with start_time/startTime, model, and spend fields.

  • buffer_minutes – Padding (in minutes) around the session time span.

  • model_min_score – Minimum model similarity score for a match.

Returns:

The updated record with estimated cost fields set.

gate.ledger.baseline_cost(record: SessionRecord) → SessionRecord#

Set baseline cost (local-zero-marginal for local models, unavailable otherwise).

Adapts corpus model_cost.baseline_session_cost. Does not perform any network access; simply classifies the session’s cost status based on whether the models are local.

Parameters:

record – The session record to update (modified in place and returned).

Returns:

The updated record with baseline cost fields set.

Aggregation#

gate.ledger.summarize_sessions(records: list[SessionRecord]) → SessionSummary#

Aggregate session records into a summary with per-model and per-project breakdowns.

Parameters:

records – Session records to aggregate.

Returns:

A SessionSummary with totals and per-dimension counts.