worker

Documentation

Index

• type Executor
  • func NewExecutor(sessionService *session.Service, cliRegistry *runner.Registry, ...) *Executor
  • func (e *Executor) Execute(ctx context.Context, t *session.Session)
• type ExecutorConfig
• type Pool
  • func NewPool(redis *redisclient.Client, executor *Executor, sessionService *session.Service, ...) *Pool
  • func (p *Pool) Cancel(sessionID string) error
  • func (p *Pool) Start(ctx context.Context)
  • func (p *Pool) Stop()
• type StreamEvent
• type Streamer
  • func NewStreamer(redis *redisclient.Client, historyTTL time.Duration) *Streamer
  • func (s *Streamer) Emit(ctx context.Context, sessionID string, evt StreamEvent) error
  • func (s *Streamer) EmitCLIOutput(ctx context.Context, sessionID string, rawEvent json.RawMessage) error
  • func (s *Streamer) EmitDone(ctx context.Context, sessionID string, status session.Status, ...) error
  • func (s *Streamer) EmitGit(ctx context.Context, sessionID, event string, data interface{}) error
  • func (s *Streamer) EmitNormalized(ctx context.Context, sessionID string, evt *runner.NormalizedEvent) error
  • func (s *Streamer) EmitResult(ctx context.Context, sessionID, event string, data interface{}) error
  • func (s *Streamer) EmitSystem(ctx context.Context, sessionID, event string, data interface{}) error

Types

type Executor

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

Executor orchestrates the full session lifecycle: clone → run CLI → diff → report.

func NewExecutor

func NewExecutor(
	sessionService *session.Service,
	cliRegistry *runner.Registry,
	streamer *Streamer,
	webhook *webhook.Sender,
	keyResolver *keys.Resolver,
	mcpInstaller *mcp.Installer,
	toolResolver *tools.Resolver,
	workspaceMgr *workspace.Manager,
	cfg ExecutorConfig,
) *Executor

NewExecutor creates a new session executor.

func (*Executor) Execute

func (e *Executor) Execute(ctx context.Context, t *session.Session)

Execute runs the full session pipeline.

type ExecutorConfig

type ExecutorConfig struct {
	WorkspaceBase   string
	DefaultTimeout  int
	MaxTimeout      int
	DefaultModels   map[string]string // CLI name → default model (e.g. "claude-code" → "claude-sonnet-4-...")
	ProviderDomains map[string]string // custom domain → provider mappings
}

ExecutorConfig holds executor configuration.

type Pool

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

Pool is a worker pool that consumes sessions from a Redis queue.

func NewPool

func NewPool(
	redis *redisclient.Client,
	executor *Executor,
	sessionService *session.Service,
	queueName string,
	concurrency int,
) *Pool

NewPool creates a new worker pool.

func (*Pool) Cancel

func (p *Pool) Cancel(sessionID string) error

Cancel cancels a running session by its ID.

func (*Pool) Start

func (p *Pool) Start(ctx context.Context)

Start launches all worker goroutines.

func (*Pool) Stop

func (p *Pool) Stop()

Stop signals workers to stop and waits for them to finish.

type StreamEvent

type StreamEvent struct {
	Type  string          `json:"type"`  // system, git, cli, stream, result
	Event string          `json:"event"` // event name
	Data  json.RawMessage `json:"data"`  // event-specific payload
	TS    string          `json:"ts"`    // ISO 8601 timestamp
}

StreamEvent is a structured event published to Redis Pub/Sub.

type Streamer

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

Streamer publishes session events to Redis Pub/Sub and persists to history.

func NewStreamer

func NewStreamer(redis *redisclient.Client, historyTTL time.Duration) *Streamer

NewStreamer creates a new event streamer.

func (*Streamer) Emit

func (s *Streamer) Emit(ctx context.Context, sessionID string, evt StreamEvent) error

Emit publishes an event to the session's stream channel and persists to history.

func (*Streamer) EmitCLIOutput

func (s *Streamer) EmitCLIOutput(ctx context.Context, sessionID string, rawEvent json.RawMessage) error

EmitCLIOutput forwards a raw Claude Code stream-json line.

func (*Streamer) EmitDone

func (s *Streamer) EmitDone(ctx context.Context, sessionID string, status session.Status, summary *gitpkg.ChangesSummary) error

EmitDone publishes completion signal on the done channel and sets history TTL.

func (*Streamer) EmitGit

func (s *Streamer) EmitGit(ctx context.Context, sessionID, event string, data interface{}) error

EmitGit publishes a git event.

func (*Streamer) EmitNormalized

func (s *Streamer) EmitNormalized(ctx context.Context, sessionID string, evt *runner.NormalizedEvent) error

EmitNormalized publishes a normalized CLI event.

func (*Streamer) EmitResult

func (s *Streamer) EmitResult(ctx context.Context, sessionID, event string, data interface{}) error

EmitResult publishes a result event.

func (*Streamer) EmitSystem

func (s *Streamer) EmitSystem(ctx context.Context, sessionID, event string, data interface{}) error

EmitSystem publishes a system event.