session

package
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 18, 2026 License: AGPL-3.0 Imports: 16 Imported by: 0

Documentation

Overview

Package session provides JSONL-based session storage compatible with the Python sidecar. Format: first line = Header record, subsequent lines = Message records.

Index

Constants

This section is empty.

Variables

View Source 
var ErrMalformed = errors.New("session file is malformed")

ErrMalformed is returned when a session file exists but its header cannot be decoded.

View Source 
var Global = &Bus{
	subs:      make(map[string][]chan map[string]any),
	buffers:   make(map[string][]map[string]any),
	terminals: make(map[string]map[string]any),
}

Global is the singleton bus used by the whole process.

Functions

func AddTokenUsage 

func AddTokenUsage(wsPath, sessionID string, inputTokens, outputTokens int) error

AddTokenUsage accumulates token counts for a session turn into the header.

func AppendAudit 

func AppendAudit(wsPath, sessionID string, actor Actor, eventType string, payload map[string]any) (string, error)

AppendAudit writes one tamper-evident record to the session's append-only audit log and returns its hash. It never touches the main session file; callers should log a returned error rather than treat it as fatal to the turn.

func AppendLiveEvent 

func AppendLiveEvent(sessionID string, event map[string]any) (int64, error)

AppendLiveEvent writes one event as a JSONL line and returns the byte offset where that line starts (i.e., the value to set as _off on the event).

func AppendMessage 

func AppendMessage(wsPath, sessionID string, msg Message) error

AppendMessage appends a message and atomically updates the header's messageCount and title.

func AuditSummary 

func AuditSummary(v any) string

AuditSummary renders a bounded, single-line summary of a tool input for the audit payload. It records enough to trace what ran without copying unbounded content (full redaction is handled by the egress guardrail, separately).

func ClearLiveEvents 

func ClearLiveEvents(sessionID string)

ClearLiveEvents deletes the live JSONL file for sessionID.

func Delete 

func Delete(wsPath, sessionID string) bool

Delete removes a session file.

func IsRunning 

func IsRunning(sessionID string) bool

IsRunning reports whether sessionID currently has a live turn — an entry in the active registry owned by a still-alive process. The session WS uses it to tell a just-connected client (e.g. a second browser opening the session while an agent is mid-turn) that events are already in flight, so it streams them instead of waiting for a session_reset boundary that fired before it connected.

func IsTerminalType 

func IsTerminalType(t string) bool

IsTerminalType returns true for event types that end the agent stream. run_done/run_failed are terminal for workflow-run streams, which reuse this bus.

func MessageCount 

func MessageCount(wsPath, sessionID string) int

MessageCount returns the message count stored in the session header, or 0 on error.

func RegisterSession 

func RegisterSession(sessionID, workspacePath, initiator string)

RegisterSession marks a session as running. Call when send_message starts.

func ReplaceMessages 

func ReplaceMessages(wsPath, sessionID string, messages []map[string]any) error

ReplaceMessages rewrites the session file, keeping the header but replacing all stored messages with the provided list. Used by compact_session to replace the full conversation history with a single summary message so the next LLM call starts from a clean, small context.

func ResetStaleRunning 

func ResetStaleRunning(wsPath string)

ResetStaleRunning marks any top-level session left in "running" state as "cancelled" when no live agent task owns it. A session is orphaned in "running" when the sidecar is force-quit (or crashes) mid-turn: the deferred status update in runAgentTask never executes, so the on-disk header still says "running". The frontend derives task.running from this status, so the orphan shows as a perpetual loading/"Thinking…" state that survives app restarts and workspace re-adds — and auto-reattaches to a dead stream.

Call on workspace registration, AFTER PurgeDead() has cleaned the in-memory active registry. We cross-check ListActive() so a session genuinely streaming in this (or another live) process is never clobbered.

func SetTodos 

func SetTodos(wsPath, sessionID string, todos []Todo) error

SetTodos replaces the todo list for a session.

func UnregisterSession 

func UnregisterSession(sessionID string)

UnregisterSession removes a session from the active registry. Call when a turn finishes.

func VerifyAudit 

func VerifyAudit(wsPath, sessionID string) (int, error)

VerifyAudit replays a session's audit log and reports whether the hash chain is intact. It returns the number of records verified and an error at the first broken link (tampering, truncation, or reordering).

Types

type APISession 

type APISession struct {
	ID                    string           `json:"id"`
	CreatedAt             string           `json:"createdAt"`
	UpdatedAt             string           `json:"updatedAt"`
	WorkspacePath         string           `json:"workspacePath"`
	Provider              string           `json:"provider"`
	Model                 string           `json:"model"`
	Mode                  string           `json:"mode"`
	Title                 *string          `json:"title"`
	ParentSessionID       *string          `json:"parentSessionId"`
	Role                  string           `json:"role"`
	AgentName             string           `json:"agentName"`
	Color                 string           `json:"color"`
	Status                string           `json:"status"`
	Result                string           `json:"result"`
	ActiveSkills          []string         `json:"activeSkills"`
	Todos                 []map[string]any `json:"todos"`
	MessageCount          *int             `json:"messageCount"`
	Messages              []map[string]any `json:"messages"`
	InputTokens           int              `json:"inputTokens"`
	OutputTokens          int              `json:"outputTokens"`
	CompactedSummary      string           `json:"compactedSummary,omitempty"`
	CompactedAt           string           `json:"compactedAt,omitempty"`
	CompactedMessageCount int              `json:"compactedMessageCount,omitempty"`
}

APISession is the shape the frontend expects (returned by all session endpoints).

func Create 

func Create(wsPath, provider, model, mode string, opts ...CreateOption) (APISession, error)

Create writes a new session header file and returns its API representation.

func List 

func List(wsPath string) ([]APISession, error)

List returns top-level sessions sorted by updatedAt desc, using dir-mtime cache.

func Load 

func Load(wsPath, id string) (*APISession, error)

Load reads an entire session (all messages). Prefer LoadTail for the UI path.

func LoadPage 

func LoadPage(wsPath, id string, skip, n int) (*APISession, bool, error)

LoadPage returns up to n messages ending at (total - skip), and whether older messages still exist before that window. Uses a full file scan.

func LoadTail 

func LoadTail(wsPath, id string, n int) (*APISession, int, error)

LoadTail reads a session returning only the last n messages and the total count. Returns ErrMalformed if the file exists but the header cannot be decoded.

func UpdateFields 

func UpdateFields(wsPath, sessionID string, fields map[string]any) (*APISession, error)

UpdateFields updates arbitrary header fields (status, mode, provider, model, title…).

type ActiveEntry 

type ActiveEntry struct {
	SessionID     string `json:"sessionId"`
	WorkspacePath string `json:"workspacePath"`
	Initiator     string `json:"initiator"` // "desktop" | "vscode" | "cli"
	PID           int    `json:"pid"`
	StartedAt     string `json:"startedAt"`
	UpdatedAt     string `json:"updatedAt"`
	Status        string `json:"status"` // "running" | "idle" | "cancelled" | "failed"
}

ActiveEntry records a session that received a send_message (running or recently finished).

func ListActive 

func ListActive() []ActiveEntry

ListActive returns a copy of all entries currently in the registry.

func PurgeDead 

func PurgeDead() []ActiveEntry

PurgeDead removes entries whose sidecar process is no longer alive and returns the purged entries. Call once on startup so stale "running" entries are cleared.

type Actor 

type Actor struct {
	UserID   string `json:"userId,omitempty"`
	TenantID string `json:"tenantId,omitempty"`
	Label    string `json:"label,omitempty"` // human-readable: agent name, channel, etc.
}

Actor identifies who an audited action is attributed to. Any field may be empty on surfaces that do not carry that identity (e.g. the local desktop app).

type Artifact 

type Artifact struct {
	Name      string    `json:"name"`
	Path      string    `json:"path"`
	Size      int64     `json:"size"`
	CreatedAt time.Time `json:"createdAt"`
}

Artifact represents a file produced during a session that the frontend can display or download.

func ListArtifacts 

func ListArtifacts(wsPath, sessionID string) []Artifact

ListArtifacts returns files stored in .agent/artifacts/{sessionID}/ within the workspace. Returns nil (not an error) if the directory doesn't exist yet.

type AuditRecord 

type AuditRecord struct {
	Seq       int            `json:"seq"`
	Timestamp string         `json:"ts"`
	SessionID string         `json:"sessionId"`
	Actor     Actor          `json:"actor"`
	Type      string         `json:"type"` // tool_exec | approval | turn
	Payload   map[string]any `json:"payload,omitempty"`
	PrevHash  string         `json:"prevHash"`
	Hash      string         `json:"hash"`
}

AuditRecord is one tamper-evident line in a session's audit log.

type Bus 

type Bus struct {
	// contains filtered or unexported fields
}

Bus is an in-process pub/sub for session events. The stdio emitter publishes every agent event here; HTTP WebSocket handlers subscribe to receive them.

func (*Bus) ClearBuffer 

func (b *Bus) ClearBuffer(sessionID string)

ClearBuffer discards buffered events (and terminal state) for sessionID. Call this before starting a new agent turn so reconnecting WS clients don't replay stale events from a previous turn.

func (*Bus) HasPendingLiveEvents 

func (b *Bus) HasPendingLiveEvents(sessionID string) bool

HasPendingLiveEvents reports true if the newest buffered event for sessionID is non-terminal, which means an agent stream is still in progress. Terminal events (done/cancelled/agent_error) mark the end of the stream; any older non-terminal events after the newest terminal are ignored.

func (*Bus) Publish 

func (b *Bus) Publish(sessionID string, event map[string]any)

Publish sends event to every subscriber for sessionID and appends it to the replay buffer so late-connecting subscribers can catch up. Terminal events (done/cancelled/agent_error) are also stored so the WS handler can close immediately if the session already finished. Non-blocking: slow subscribers drop events rather than blocking the agent.

func (*Bus) Subscribe 

func (b *Bus) Subscribe(sessionID string) (chan map[string]any, func())

Subscribe returns a buffered channel that receives all future events for sessionID, and a cancel function that must be called to stop delivery. Any events already buffered since the last ClearBuffer are replayed immediately so clients that connect after agent start don't miss them.

func (*Bus) SubscribeFrom 

func (b *Bus) SubscribeFrom(sessionID string, fromOffset int64) (chan map[string]any, func())

SubscribeFrom is like Subscribe but only replays buffered events whose _off is >= fromOffset. A WS client that already received file-replayed events up to fromOffset (e.g. Phase 1 replay) uses this so the in-memory ring buffer does not re-deliver those same events as duplicates ("old messages stream in" on reconnect).

When fromOffset is 0, every buffered event is replayed (matching the historical Subscribe behavior). Events without an _off field (off == 0) are always replayed when fromOffset == 0, and skipped when fromOffset > 0 only if their resolved offset is below the threshold — which is correct because a real event always carries an _off >= 1 once the live log is appending.

func (*Bus) Terminal 

func (b *Bus) Terminal(sessionID string) map[string]any

Terminal returns a shallow copy of the last terminal event for sessionID, or nil if the session is still running or has never started.

type CreateOption 

type CreateOption func(*createConfig)

func WithAgentName 

func WithAgentName(n string) CreateOption

func WithChannelSession 

func WithChannelSession() CreateOption

WithChannelSession marks the session as channel-initiated so it is excluded from the dev-mode sessions list and the active-sessions registry.

func WithColor 

func WithColor(col string) CreateOption

func WithParent 

func WithParent(id string) CreateOption

func WithRole 

func WithRole(r string) CreateOption

func WithSkills 

func WithSkills(s []string) CreateOption
type Header struct {
	Type            string   `json:"type"`
	ID              string   `json:"id"`
	CreatedAt       string   `json:"createdAt"`
	UpdatedAt       string   `json:"updatedAt"`
	WorkspacePath   string   `json:"workspacePath"`
	Provider        string   `json:"provider"`
	Model           string   `json:"model"`
	Mode            string   `json:"mode"`
	Title           *string  `json:"title"`
	ParentSessionID *string  `json:"parentSessionId"`
	Role            string   `json:"role"`
	AgentName       string   `json:"agentName"`
	Color           string   `json:"color"`
	Status          string   `json:"status"`
	Result          string   `json:"result"`
	ActiveSkills    []string `json:"activeSkills"`
	Todos           []Todo   `json:"todos"`
	MessageCount    int      `json:"messageCount"`
	InputTokens     int      `json:"inputTokens"`
	OutputTokens    int      `json:"outputTokens"`
	// CompactedSummary is set when the session history has been compacted.
	// Non-empty means the executor should use this summary as the initial context
	// and only include messages at index >= CompactedMessageCount.
	CompactedSummary string `json:"compactedSummary,omitempty"`
	CompactedAt      string `json:"compactedAt,omitempty"`
	// CompactedMessageCount is the len(Messages) at the moment of compaction.
	// Post-compact messages are simply Messages[CompactedMessageCount:].
	// A value of 0 means unset (pre-feature sessions fall back to timestamp comparison).
	CompactedMessageCount int `json:"compactedMessageCount,omitempty"`
	// SessionSource marks how this session was initiated.
	// "channel" = created by channels_chat; excluded from the dev-mode sessions list.
	SessionSource string `json:"sessionSource,omitempty"`
}

Header is the first JSON line of every session JSONL file. Must stay byte-compatible with the Python sidecar's schema.

type LiveEvent 

type LiveEvent struct {
	Event  map[string]any
	Offset int64
}

LiveEvent pairs a decoded agent event with the byte offset of its JSONL line.

type Message 

type Message struct {
	Type      string `json:"type"`
	ID        string `json:"id"`
	SessionID string `json:"sessionId"`
	Role      string `json:"role"`
	Content   string `json:"content"`
	Thinking  string `json:"thinking,omitempty"`
	Timestamp string `json:"timestamp"`
	Provider  string `json:"provider,omitempty"`
	Model     string `json:"model,omitempty"`
	// ToolUses and Items are stored as raw JSON to avoid schema churn.
	ToolUses    []map[string]any `json:"toolUses,omitempty"`
	Items       []map[string]any `json:"items,omitempty"`
	Interrupted bool             `json:"interrupted,omitempty"`
}

Message is a subsequent JSON line in a session JSONL file.

type TailResult 

type TailResult struct {
	Events    []LiveEvent
	EndOffset int64 // byte position right after the last line read
}

TailResult is returned by TailLiveEvents.

func TailLiveEvents 

func TailLiveEvents(sessionID string, fromOffset int64) (TailResult, error)

TailLiveEvents reads the live file from fromOffset onwards and returns events paired with their starting byte offsets. TailResult.EndOffset is the byte position immediately after the last line read (use as the next fromOffset).

type Todo 

type Todo struct {
	ID       string `json:"id"`
	Text     string `json:"text"`
	Content  string `json:"content,omitempty"`
	Status   string `json:"status"`
	Checked  bool   `json:"checked"`
	Priority string `json:"priority"`
}

Todo is stored in the session header.

func GetTodos 

func GetTodos(wsPath, sessionID string) ([]Todo, error)

GetTodos returns the todo list for a session.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.