sessionlog

package
v0.0.2 Latest Latest
Warning

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

 Go to latest Published: Sep 9, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Overview

Package sessionlog provides the OSS reference implementation of resumable session state. Wire contracts live in ecosystem-messaging-spec; this package owns only bounded storage, snapshot coordination, and channel adaptation.

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrInvalidSessionRef = errors.New("sessionlog: invalid session reference")
	ErrSequenceExhausted = errors.New("sessionlog: JSON-safe sequence range exhausted")
)
View Source 
var ErrCASRetryLimit = errors.New("sessionlog: distributed event update retry limit exceeded")
View Source 
var ErrCorruptEventLog = errors.New("sessionlog: corrupt file event log")
View Source 
var ErrWorkspaceUnavailable = errors.New("sessionlog: workspace unavailable")

Functions

This section is empty.

Types

type CASBlobStore 

type CASBlobStore = casblob.Store

CASBlobStore remains as an alias for compatibility. New distributed stores use the neutral casblob.Store contract directly.

type CASEventLog 

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

CASEventLog persists one complete bounded session state per composite workspace/session key. Appends and resets use optimistic compare-and-swap so multiple replicas cannot assign the same cursor or overwrite one another.

func NewCASEventLog 

func NewCASEventLog(config CASEventLogConfig) (*CASEventLog, error)

func (*CASEventLog) Append 

func (l *CASEventLog) Append(ctx context.Context, ref Ref, kind spec.SessionEventKind, payload json.RawMessage) (spec.SessionEvent, error)

Append assigns a cursor and atomically commits the resulting bounded state.

func (*CASEventLog) Capture 

func (l *CASEventLog) Capture(ctx context.Context, ref Ref, after *spec.SessionCursor) (SessionCapture, error)

func (*CASEventLog) Cursor 

func (l *CASEventLog) Cursor(ctx context.Context, ref Ref) (spec.SessionCursor, error)

Cursor returns the durable session boundary, creating its initial generation atomically when this is the first observation.

func (*CASEventLog) Projection 

func (l *CASEventLog) Projection(ctx context.Context, ref Ref) (spec.MessageState, spec.SessionCursor, error)

func (*CASEventLog) Replay 

func (l *CASEventLog) Replay(ctx context.Context, ref Ref, after spec.SessionCursor) (spec.SessionReplayResult, error)

func (*CASEventLog) Reset 

func (l *CASEventLog) Reset(ctx context.Context, ref Ref) (spec.SessionCursor, error)

Reset atomically replaces the current state with a new incomparable generation. A concurrent append/reset causes a retry against the new value.

type CASEventLogConfig 

type CASEventLogConfig struct {
	Blobs         CASBlobStore
	Prefix        string
	MaxEvents     int
	MaxRetries    int
	NewGeneration func() (string, error)
}

CASEventLogConfig configures a bounded distributed event log. Prefix namespaces keys inside the supplied blob store; an empty value uses the stable agent-harness v1 prefix.

type CapabilityStateProvider 

type CapabilityStateProvider interface {
	SnapshotStateProvider
	Capabilities() []spec.SessionCapability
}

CapabilityStateProvider is a snapshot state provider that explicitly names the optional protocol behavior it backs.

type Coordinator 

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

Coordinator produces validated, deterministic attach results.

func NewCoordinator 

func NewCoordinator(config CoordinatorConfig) (*Coordinator, error)

NewCoordinator validates dependencies and normalizes server capabilities.

func (*Coordinator) Attach 

func (c *Coordinator) Attach(ctx context.Context, request spec.SessionAttachRequest) (spec.SessionAttachResult, error)

Attach resolves identity, loads the authoritative history snapshot, captures the current event boundary, and returns replay when it was requested and negotiated.

func (*Coordinator) ResetSession 

func (c *Coordinator) ResetSession(ctx context.Context, requestedWorkspace, sessionID string) (spec.SessionCursor, error)

ResetSession begins a new event generation for a resolved workspace session. Transcript deletion remains the host's responsibility so it can preserve its own ordering and error policy.

type CoordinatorConfig 

type CoordinatorConfig struct {
	Workspaces   WorkspaceResolver
	History      WorkspaceHistoryStore
	Events       EventStore
	State        SnapshotStateProvider
	Capabilities []spec.SessionCapability
}

CoordinatorConfig wires the shared protocol to OSS history and event stores.

type DefaultHistoryStore 

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

DefaultHistoryStore gives a legacy single-workspace HistoryStore a resolved wire identity without moving its on-disk keys. If Base also implements harness.WorkspaceHistoryStore, non-default workspaces pass through unchanged.

func BindDefaultHistory 

func BindDefaultHistory(base harness.HistoryStore, defaultID string) (*DefaultHistoryStore, error)

BindDefaultHistory maps defaultID to Base's compatibility Load/Save methods. This lets legacy single-workspace deployments answer spec attach requests as a resolved "default" workspace while retaining their existing storage layout.

func (*DefaultHistoryStore) LoadHistory 

func (s *DefaultHistoryStore) LoadHistory(ctx context.Context, sessionID string) ([]protocol.ChatMessage, error)

func (*DefaultHistoryStore) LoadWorkspaceHistory 

func (s *DefaultHistoryStore) LoadWorkspaceHistory(ctx context.Context, workspaceID, sessionID string) ([]protocol.ChatMessage, error)

func (*DefaultHistoryStore) SaveHistory 

func (s *DefaultHistoryStore) SaveHistory(ctx context.Context, sessionID string, messages []protocol.ChatMessage) error

func (*DefaultHistoryStore) SaveWorkspaceHistory 

func (s *DefaultHistoryStore) SaveWorkspaceHistory(ctx context.Context, workspaceID, sessionID string, messages []protocol.ChatMessage) error

type EventStore 

type EventStore interface {
	Append(ctx context.Context, ref Ref, kind spec.SessionEventKind, payload json.RawMessage) (spec.SessionEvent, error)
	Cursor(ctx context.Context, ref Ref) (spec.SessionCursor, error)
	Replay(ctx context.Context, ref Ref, after spec.SessionCursor) (spec.SessionReplayResult, error)
	Projection(ctx context.Context, ref Ref) (spec.MessageState, spec.SessionCursor, error)
	Capture(ctx context.Context, ref Ref, after *spec.SessionCursor) (SessionCapture, error)
	Reset(ctx context.Context, ref Ref) (spec.SessionCursor, error)
}

EventStore is the transport-independent append/replay surface used by the attach coordinator and recording publisher.

type FileEventLog 

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

FileEventLog persists the same bounded event suffix and materialized message projection as MemoryEventLog. Successful mutations are written atomically before they become observable to another EventStore call or live publisher.

A FileEventLog is concurrency-safe within one process. The state directory must have a single writing process; a later process can reopen it to resume.

func NewFileEventLog 

func NewFileEventLog(config FileEventLogConfig) (*FileEventLog, error)

NewFileEventLog returns a durable event log rooted below config.StateDir.

func (*FileEventLog) Append 

func (l *FileEventLog) Append(ctx context.Context, ref Ref, kind spec.SessionEventKind, payload json.RawMessage) (spec.SessionEvent, error)

Append assigns, atomically persists, and returns the next session event.

func (*FileEventLog) Capture 

func (l *FileEventLog) Capture(ctx context.Context, ref Ref, after *spec.SessionCursor) (SessionCapture, error)

Capture atomically copies one persisted attach boundary.

func (*FileEventLog) Cursor 

func (l *FileEventLog) Cursor(ctx context.Context, ref Ref) (spec.SessionCursor, error)

Cursor returns a restart-stable session boundary, persisting the initial generation when the session has not previously been observed.

func (*FileEventLog) Projection 

func (l *FileEventLog) Projection(ctx context.Context, ref Ref) (spec.MessageState, spec.SessionCursor, error)

Projection returns the persisted materialized chat-stream state.

func (*FileEventLog) Replay 

func (l *FileEventLog) Replay(ctx context.Context, ref Ref, after spec.SessionCursor) (spec.SessionReplayResult, error)

Replay returns a restart-stable retained interval after the supplied cursor.

func (*FileEventLog) Reset 

func (l *FileEventLog) Reset(ctx context.Context, ref Ref) (spec.SessionCursor, error)

Reset atomically persists a new incomparable generation and empty state.

type FileEventLogConfig 

type FileEventLogConfig struct {
	StateDir      string
	MaxEvents     int
	NewGeneration func() (string, error)
}

FileEventLogConfig configures a bounded, atomically persisted event store. StateDir is the shared harness state directory; workspace and session identities are encoded as child path segments by FileEventLog.

type GoalStateSource 

type GoalStateSource interface {
	ListGoals(ctx context.Context, workspaceID, sessionID string) ([]spec.SessionGoalRecord, error)
}

GoalStateSource supplies deterministic session-scoped goal records.

type LifecycleStateProvider 

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

LifecycleStateProvider projects typed durable stores into the standard capability-owned SessionSnapshot.state namespaces.

func NewLifecycleStateProvider 

func NewLifecycleStateProvider(config LifecycleStateProviderConfig) (*LifecycleStateProvider, error)

func (*LifecycleStateProvider) Capabilities 

func (p *LifecycleStateProvider) Capabilities() []spec.SessionCapability

func (*LifecycleStateProvider) SnapshotState 

func (p *LifecycleStateProvider) SnapshotState(
	ctx context.Context,
	workspaceID, sessionID string,
	cursor spec.SessionCursor,
	messages []protocol.ChatMessage,
) (map[string]json.RawMessage, error)

func (*LifecycleStateProvider) SnapshotStateForCapabilities 

func (p *LifecycleStateProvider) SnapshotStateForCapabilities(
	ctx context.Context,
	workspaceID, sessionID string,
	_ spec.SessionCursor,
	_ []protocol.ChatMessage,
	capabilities []spec.SessionCapability,
) (map[string]json.RawMessage, error)

type LifecycleStateProviderConfig 

type LifecycleStateProviderConfig struct {
	Subagents SubagentStateSource
	Goals     GoalStateSource
}

type MemoryEventLog 

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

MemoryEventLog retains a contiguous suffix per (workspace, session) and a materialized stream-message projection. It is concurrency-safe within one process; durable/distributed backends implement EventStore directly.

func NewMemoryEventLog 

func NewMemoryEventLog(config MemoryEventLogConfig) *MemoryEventLog

NewMemoryEventLog returns an empty bounded event log.

func (*MemoryEventLog) Append 

func (l *MemoryEventLog) Append(ctx context.Context, ref Ref, kind spec.SessionEventKind, payload json.RawMessage) (spec.SessionEvent, error)

Append assigns the next cursor and retains the event. Chat-stream payloads also update the materialized message projection.

func (*MemoryEventLog) Capture 

func (l *MemoryEventLog) Capture(ctx context.Context, ref Ref, after *spec.SessionCursor) (SessionCapture, error)

Capture atomically copies the projection, cursor, and optional replay result so attach cannot return a replay boundary that races ahead of its snapshot.

func (*MemoryEventLog) Cursor 

func (l *MemoryEventLog) Cursor(ctx context.Context, ref Ref) (spec.SessionCursor, error)

Cursor returns the current session boundary, creating the initial generation at sequence zero when the session has not been observed before.

func (*MemoryEventLog) Projection 

func (l *MemoryEventLog) Projection(ctx context.Context, ref Ref) (spec.MessageState, spec.SessionCursor, error)

Projection returns a copy of the materialized chat-stream message state and the exact cursor through which it was reduced.

func (*MemoryEventLog) Replay 

func (l *MemoryEventLog) Replay(ctx context.Context, ref Ref, after spec.SessionCursor) (spec.SessionReplayResult, error)

Replay returns a complete retained interval, a partial retained suffix when a gap was evicted, or unavailable when the cursor names another generation or lies ahead of the current boundary.

func (*MemoryEventLog) Reset 

func (l *MemoryEventLog) Reset(ctx context.Context, ref Ref) (spec.SessionCursor, error)

Reset starts a new incomparable generation and drops retained/live state.

type MemoryEventLogConfig 

type MemoryEventLogConfig struct {
	MaxEvents     int
	NewGeneration func() (string, error)
}

MemoryEventLogConfig configures the bounded in-memory reference store.

type NegotiatedSnapshotStateProvider 

type NegotiatedSnapshotStateProvider interface {
	CapabilityStateProvider
	SnapshotStateForCapabilities(
		ctx context.Context,
		workspaceID, sessionID string,
		cursor spec.SessionCursor,
		messages []protocol.ChatMessage,
		capabilities []spec.SessionCapability,
	) (map[string]json.RawMessage, error)
}

NegotiatedSnapshotStateProvider avoids loading optional namespaces a client did not negotiate. This keeps an unavailable optional backend from breaking a legacy snapshot request that cannot consume it.

type RecordingPublisher 

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

RecordingPublisher records every shared chat-stream event before forwarding it. Harness-only lifecycle/error events still pass downstream but are not put into the protocol replay log.

func NewRecordingPublisher 

func NewRecordingPublisher(config RecordingPublisherConfig) (*RecordingPublisher, error)

func (*RecordingPublisher) PublishEvent 

func (p *RecordingPublisher) PublishEvent(ctx context.Context, event channel.Event) error

type RecordingPublisherConfig 

type RecordingPublisherConfig struct {
	Events           EventStore
	Next             channel.Publisher
	SessionEvents    SessionEventPublisher
	DefaultWorkspace string
}

RecordingPublisherConfig wires a bounded log in front of an optional downstream transport. DefaultWorkspace resolves workspace-less local events.

type Ref 

type Ref struct {
	WorkspaceID string
	SessionID   string
}

Ref is the composite identity for all session-log operations.

type SessionCapture 

type SessionCapture struct {
	Cursor   spec.SessionCursor
	Messages []spec.ChatMessage
	Replay   *spec.SessionReplayResult
}

SessionCapture is the atomic event-side boundary used by attach. Messages contains the materialized stream projection in deterministic start order; Replay, when requested, ends at exactly Cursor.

type SessionEventPublisher 

type SessionEventPublisher interface {
	PublishSessionEvent(ctx context.Context, event spec.SessionEvent) error
}

SessionEventPublisher receives the cursor-bearing envelope after it has been retained. A channel can use this to provide gap-free live session streams.

type SnapshotStateProvider 

type SnapshotStateProvider interface {
	SnapshotState(ctx context.Context, workspaceID, sessionID string, cursor spec.SessionCursor, messages []protocol.ChatMessage) (map[string]json.RawMessage, error)
}

SnapshotStateProvider projects additional capability-gated durable state at attach time. A nil provider yields an empty state object.

type StaticWorkspaceResolver 

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

StaticWorkspaceResolver is the OSS reference visibility policy. Proprietary hosts can replace it with an authority-aware resolver without changing attach.

func NewStaticWorkspaceResolver 

func NewStaticWorkspaceResolver(defaultID string, visible []string) (*StaticWorkspaceResolver, error)

NewStaticWorkspaceResolver creates a resolver. defaultID is always visible; visible contains any additional explicitly addressable workspaces.

func (*StaticWorkspaceResolver) ResolveWorkspace 

func (r *StaticWorkspaceResolver) ResolveWorkspace(ctx context.Context, requested string) (string, error)

type SubagentStateSource 

type SubagentStateSource interface {
	ListSubagents(ctx context.Context, workspaceID, parentSessionID string) ([]spec.SessionSubagentRecord, error)
}

SubagentStateSource supplies deterministic parent-scoped lifecycle records.

type WorkspaceHistoryStore 

type WorkspaceHistoryStore interface {
	harness.HistoryStore
	harness.WorkspaceHistoryStore
}

WorkspaceHistoryStore is the attach-safe history surface. Requiring both the compatibility and scoped methods prevents a multi-workspace resolver from accidentally reading a legacy store keyed only by session ID.

type WorkspaceResolver 

type WorkspaceResolver interface {
	ResolveWorkspace(ctx context.Context, requested string) (string, error)
}

WorkspaceResolver resolves an omitted workspace to a default and rejects explicit workspaces outside the host's configured visibility.

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.