session

Documentation

Overview

Package session ties task execution and relay forwarding together into one "session actor" per user session.

Index

• type Actor
  • func NewActor(opts Options) (*Actor, error)
  • func (a *Actor) AllowedTools() []string
  • func (a *Actor) CancelTask()
  • func (a *Actor) GetClaudeSessionID() string
  • func (a *Actor) HandleCompactRequest(ctx context.Context, request *protocol.CompactRequest)
  • func (a *Actor) HandleContextStatsRequest(ctx context.Context, request *protocol.ContextStatsRequest)
  • func (a *Actor) HandlePermissionResponse(resp *protocol.PermissionResponse) error
  • func (a *Actor) HandleQuestionResponse(resp *protocol.QuestionResponse) error
  • func (a *Actor) HasBeenIdle() bool
  • func (a *Actor) HasInFlightTask() bool
  • func (a *Actor) HasTaskID(taskID string) bool
  • func (a *Actor) InFlightTaskID() string
  • func (a *Actor) Info() sockapi.SessionInfo
  • func (a *Actor) LastActiveAt() time.Time
  • func (a *Actor) Run(ctx context.Context) error
  • func (a *Actor) SendStreamEvent(ctx context.Context, channelID string, event json.RawMessage) error
  • func (a *Actor) SendTask(task protocol.Task) error
  • func (a *Actor) SetProvider(provider string)
  • func (a *Actor) Stop() error
  • func (a *Actor) StopIdleWorker(ctx context.Context, reason string) bool
  • func (a *Actor) WorkerSnapshot() (pi.WorkerSnapshot, bool)
• type AgentToolController
• type ImageUploader
• type LifecycleSink
• type Manager
  • func NewManager(opts ManagerOptions) *Manager
  • func (m *Manager) ActiveCount() (total int, executing int)
  • func (m *Manager) ActiveTaskIDs() []string
  • func (m *Manager) Get(sessionID string) *Actor
  • func (m *Manager) InFlightCount() int
  • func (m *Manager) ReapIdleActors(maxIdle time.Duration) int
  • func (m *Manager) ReapIdleWorkers(maxIdle time.Duration, idleCap int) int
  • func (m *Manager) Remove(sessionID string)
  • func (m *Manager) SessionInfos() []sockapi.SessionInfo
  • func (m *Manager) Spawn(ctx context.Context, opts Options) (*Actor, error)
  • func (m *Manager) StartReaper(ctx context.Context, tick time.Duration, maxIdle time.Duration)
  • func (m *Manager) StartWorkerReaper(ctx context.Context, tick time.Duration, maxIdle time.Duration, idleCap int)
  • func (m *Manager) StopAll()
  • func (m *Manager) WorkerSnapshots() []pi.WorkerSnapshot
• type ManagerOptions
• type Options
• type RelaySender
• type TurnDeadlines
• type TurnHooks
• type TurnResult
• type TurnSupervisor
  • func NewTurnSupervisor(opts TurnSupervisorOptions) *TurnSupervisor
  • func (s *TurnSupervisor) Result() TurnResult
  • func (s *TurnSupervisor) Run(parent context.Context, run func(context.Context, TurnHooks) error) error
• type TurnSupervisorOptions

Types

type Actor

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

Actor drives a single agent session using spawn-per-task execution. Each incoming task spawns a fresh executor process; no processes remain alive between tasks.

func NewActor

func NewActor(opts Options) (*Actor, error)

NewActor creates a new Actor for the given session.

func (*Actor) AllowedTools

func (a *Actor) AllowedTools() []string

AllowedTools returns the current granted tools list.

func (*Actor) CancelTask

func (a *Actor) CancelTask()

CancelTask cancels the in-flight task (if any) without shutting down the actor. The executor's context.Done fires, stopping the subprocess. Run() loops back and waits for the next task.

func (*Actor) GetClaudeSessionID

func (a *Actor) GetClaudeSessionID() string

GetClaudeSessionID returns the most recent Claude session id.

func (*Actor) HandleCompactRequest

func (a *Actor) HandleCompactRequest(ctx context.Context, request *protocol.CompactRequest)

func (*Actor) HandleContextStatsRequest

func (a *Actor) HandleContextStatsRequest(ctx context.Context, request *protocol.ContextStatsRequest)

func (*Actor) HandlePermissionResponse

func (a *Actor) HandlePermissionResponse(resp *protocol.PermissionResponse) error

HandlePermissionResponse processes a permission response from the relay.

func (*Actor) HandleQuestionResponse

func (a *Actor) HandleQuestionResponse(resp *protocol.QuestionResponse) error

HandleQuestionResponse routes an answer to the waiting question by requestId.

func (*Actor) HasBeenIdle

func (a *Actor) HasBeenIdle() bool

HasBeenIdle returns true if the actor has completed at least one task and is now idle. A freshly spawned actor that hasn't executed yet returns false.

func (*Actor) HasInFlightTask

func (a *Actor) HasInFlightTask() bool

HasInFlightTask returns true if the actor is currently executing a task.

func (*Actor) HasTaskID

func (a *Actor) HasTaskID(taskID string) bool

HasTaskID reports whether the actor already owns the task either queued or in-flight.

func (*Actor) InFlightTaskID

func (a *Actor) InFlightTaskID() string

InFlightTaskID returns the ID of the currently executing task, or "" if idle.

func (*Actor) Info

func (a *Actor) Info() sockapi.SessionInfo

Info returns a snapshot of the actor's current state for the status API.

func (*Actor) LastActiveAt

func (a *Actor) LastActiveAt() time.Time

LastActiveAt returns the time of the actor's last task completion or creation.

func (*Actor) Run

func (a *Actor) Run(ctx context.Context) error

Run is the actor's main loop. It waits for tasks, spawns executors, and handles permission flows. Blocks until ctx is canceled or Stop is called.

func (*Actor) SendStreamEvent

func (a *Actor) SendStreamEvent(ctx context.Context, channelID string, event json.RawMessage) error

func (*Actor) SendTask

func (a *Actor) SendTask(task protocol.Task) error

SendTask queues a task for execution. Non-blocking if the channel has capacity.

func (*Actor) SetProvider

func (a *Actor) SetProvider(provider string)

func (*Actor) Stop

func (a *Actor) Stop() error

Stop signals the actor to shut down.

func (*Actor) StopIdleWorker

func (a *Actor) StopIdleWorker(ctx context.Context, reason string) bool

func (*Actor) WorkerSnapshot

func (a *Actor) WorkerSnapshot() (pi.WorkerSnapshot, bool)

type AgentToolController

type AgentToolController interface {
	StartTask(context.Context, agentterminal.TaskScope) (agentterminal.TaskControl, error)
	StopTask(taskID string)
}

type ImageUploader

type ImageUploader interface {
	Upload(ctx context.Context, filename string, data []byte) (string, error)
}

ImageUploader uploads an image file to the relay and returns a public URL.

type LifecycleSink

type LifecycleSink interface {
	Phase(phase string, fields map[string]any)
}

type Manager

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

Manager holds a pool of session actors, keyed by sessionID.

func NewManager

func NewManager(opts ManagerOptions) *Manager

NewManager constructs a Manager.

func (*Manager) ActiveCount

func (m *Manager) ActiveCount() (total int, executing int)

ActiveCount returns the total number of actors and how many are executing.

func (*Manager) ActiveTaskIDs

func (m *Manager) ActiveTaskIDs() []string

ActiveTaskIDs returns a list of task IDs currently being executed.

func (*Manager) Get

func (m *Manager) Get(sessionID string) *Actor

Get returns an existing actor or nil.

func (*Manager) InFlightCount

func (m *Manager) InFlightCount() int

InFlightCount returns the number of actors with in-flight tasks.

func (*Manager) ReapIdleActors

func (m *Manager) ReapIdleActors(maxIdle time.Duration) int

ReapIdleActors stops and removes actors idle longer than maxIdle. Actors with in-flight tasks are never reaped. Returns the count of reaped actors.

func (*Manager) ReapIdleWorkers

func (m *Manager) ReapIdleWorkers(maxIdle time.Duration, idleCap int) int

func (*Manager) Remove

func (m *Manager) Remove(sessionID string)

Remove removes an actor from the map. Called by the reaper.

func (*Manager) SessionInfos

func (m *Manager) SessionInfos() []sockapi.SessionInfo

SessionInfos returns a snapshot of all active sessions.

func (*Manager) Spawn

func (m *Manager) Spawn(
	ctx context.Context,
	opts Options,
) (*Actor, error)

Spawn creates and starts a new actor for the session. Returns an existing actor if one already exists for the session. Returns an error if the machine is at capacity.

func (*Manager) StartReaper

func (m *Manager) StartReaper(ctx context.Context, tick time.Duration, maxIdle time.Duration)

StartReaper launches a goroutine that reaps idle actors on a tick interval. Runs until ctx is cancelled.

func (*Manager) StartWorkerReaper

func (m *Manager) StartWorkerReaper(ctx context.Context, tick time.Duration, maxIdle time.Duration, idleCap int)

func (*Manager) StopAll

func (m *Manager) StopAll()

StopAll stops every actor. Called on daemon shutdown.

func (*Manager) WorkerSnapshots

func (m *Manager) WorkerSnapshots() []pi.WorkerSnapshot

type ManagerOptions

type ManagerOptions struct {
	PiBinaryPath     string
	PiExtensionPath  string
	Relay            RelaySender
	Config           *config.Config
	PIDDir           string        // directory for child PID files; empty disables
	Uploader         ImageUploader // nil = image upload disabled
	DaemonSocketPath string
	AgentTools       AgentToolController
}

ManagerOptions configures a new Manager.

type Options

type Options struct {
	SessionID         string
	CWD               string
	Relay             RelaySender
	Model             string
	Provider          string
	Effort            string
	PermissionMode    string
	WarmPiWorkers     bool
	WarmClaudeSDK     bool
	ResumeSession     string
	PiBinaryPath      string
	PiExtensionPath   string
	ServerURL         string
	MachineID         string
	AuthToken         string
	DaemonSocketPath  string
	Uploader          ImageUploader // nil = image upload disabled
	RecordTouchedFile func(channelID string, cwd string, path string)
	OnTaskIdle        func()
	ProjectID         string
	AgentTools        AgentToolController
}

Options configures a new Actor.

type RelaySender

type RelaySender interface {
	Send(ctx context.Context, msg any) error
}

RelaySender is the minimal interface the actor needs to push events to the relay.

type TurnDeadlines

type TurnDeadlines struct {
	ProcessStart      time.Duration
	PromptWrite       time.Duration
	FirstEvent        time.Duration
	FirstVisibleEvent time.Duration
	StreamIdle        time.Duration
	ToolIdle          time.Duration
	UserInput         time.Duration
	CleanupTerm       time.Duration
}

type TurnHooks

type TurnHooks struct {
	PromptWritten         func()
	FirstEventSeen        func()
	FirstVisibleEventSeen func()
	ToolStarted           func(toolCallID string, toolName string)
	ToolFinished          func(toolCallID string, toolName string)
}

type TurnResult

type TurnResult struct {
	FailureCode string
	Retryable   bool
}

type TurnSupervisor

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

func NewTurnSupervisor

func NewTurnSupervisor(opts TurnSupervisorOptions) *TurnSupervisor

func (*TurnSupervisor) Result

func (s *TurnSupervisor) Result() TurnResult

func (*TurnSupervisor) Run

func (s *TurnSupervisor) Run(parent context.Context, run func(context.Context, TurnHooks) error) error

type TurnSupervisorOptions

type TurnSupervisorOptions struct {
	TaskID        string
	SessionID     string
	ChannelID     string
	RequestID     string
	TraceID       string
	AttemptID     string
	AttemptNumber int
	TurnKind      string
	Deadlines     TurnDeadlines
	Sink          LifecycleSink
}