queue

package
v0.0.0-...-a256278 Latest Latest
Warning

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

 Go to latest Published: Aug 27, 2026 License: Apache-2.0 Imports: 38 Imported by: 0

Documentation

Overview

Package queue provides session queue management and processing infrastructure.

Package queue provides session queue management and processing infrastructure.

Index

Constants

View Source 
const MetadataKeyAuthor = "author"

MetadataKeyAuthor is the metadata key for the message author.

Variables

View Source 
var (
	// ErrNoSessionsAvailable indicates no pending sessions are in the queue.
	ErrNoSessionsAvailable = errors.New("no sessions available")

	// ErrAtCapacity indicates the global concurrent session limit has been reached.
	ErrAtCapacity = errors.New("at capacity")

	// ErrChatExecutionActive indicates a chat already has an active execution.
	// Mapped to HTTP 409 Conflict by the API handler.
	ErrChatExecutionActive = errors.New("chat execution already active")

	// ErrShuttingDown indicates the executor is shutting down and not accepting new work.
	// Mapped to HTTP 503 Service Unavailable by the API handler.
	ErrShuttingDown = errors.New("executor is shutting down")

	// ErrScoringInProgress indicates a scoring is already in progress for a session.
	// Mapped to HTTP 409 Conflict by the API handler.
	ErrScoringInProgress = errors.New("scoring already in progress for this session")

	// ErrScoringDisabled indicates scoring is not enabled for the chain.
	ErrScoringDisabled = errors.New("scoring not enabled for this chain")
)

Sentinel errors for queue operations.

Functions

func CleanupStartupOrphans 

func CleanupStartupOrphans(ctx context.Context, client *ent.Client, podID string) error

CleanupStartupOrphans performs a one-time cleanup of sessions owned by this pod that were in-progress when the pod previously crashed. Called once during startup, before the worker pool begins processing.

func IsTerminalStatus 

func IsTerminalStatus(status alertsession.Status) bool

IsTerminalStatus checks if a session status is terminal.

func MemorySubAgentWrap 

func MemorySubAgentWrap(mem *memory.Service, memCfg *config.MemoryConfig, sessionID string) func(agent.ToolExecutor) agent.ToolExecutor

MemorySubAgentWrap returns a ToolExecutor wrapping function for sub-agent runs (memory tool with no search exclusions). Returns nil when memory is disabled.

Types

type ChatExecuteInput 

type ChatExecuteInput struct {
	Chat    *ent.Chat
	Message *ent.ChatUserMessage
	Session *ent.AlertSession
}

ChatExecuteInput groups all parameters needed to execute a chat message.

type ChatMessageExecutor 

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

ChatMessageExecutor handles asynchronous chat message processing. It manages a single goroutine per chat (one-at-a-time enforcement), supports cancellation, and graceful shutdown.

func NewChatMessageExecutor 

func NewChatMessageExecutor(
	cfg *config.Config,
	dbClient *ent.Client,
	llmClient agent.LLMClient,
	mcpFactory *mcp.ClientFactory,
	eventPublisher agent.EventPublisher,
	execConfig ChatMessageExecutorConfig,
	runbookService *runbook.Service,
	memoryService *memory.Service,
	memoryConfig *config.MemoryConfig,
) *ChatMessageExecutor

NewChatMessageExecutor creates a new ChatMessageExecutor. runbookService may be nil (uses config default runbook content). memoryService and memoryConfig may be nil (memory disabled).

func (*ChatMessageExecutor) CancelBySessionID 

func (e *ChatMessageExecutor) CancelBySessionID(ctx context.Context, sessionID string) bool

CancelBySessionID looks up the chat for the given session and cancels any active execution. Returns true if an active execution was found and cancelled.

func (*ChatMessageExecutor) CancelExecution 

func (e *ChatMessageExecutor) CancelExecution(chatID string) bool

CancelExecution cancels the active execution for a chat. Returns true if an active execution was found and cancelled.

func (*ChatMessageExecutor) SetCostBook 

func (e *ChatMessageExecutor) SetCostBook(book *cost.Book)

SetCostBook sets the price book used when persisting LLM interaction costs. Rebuilds the interaction service so subsequent writes use the book.

func (*ChatMessageExecutor) Stop 

func (e *ChatMessageExecutor) Stop()

Stop marks the executor as stopped, cancels all active executions, and waits for goroutines to drain. Safe to call multiple times.

func (*ChatMessageExecutor) Submit 

func (e *ChatMessageExecutor) Submit(ctx context.Context, input ChatExecuteInput) (string, error)

Submit validates the one-at-a-time constraint, creates a Stage record, and launches asynchronous execution. Returns the stage ID for the response.

type ChatMessageExecutorConfig 

type ChatMessageExecutorConfig struct {
	SessionTimeout    time.Duration // Max duration for a chat execution (default: 15 minutes)
	HeartbeatInterval time.Duration // Heartbeat frequency (default: 30s)
}

ChatMessageExecutorConfig holds configuration for the chat message executor.

type ExecutionResult 

type ExecutionResult struct {
	Status                alertsession.Status // completed, failed, timed_out, cancelled
	FinalAnalysis         string              // Final analysis text (if completed)
	ExecutiveSummary      string              // Executive summary (if completed)
	ExecutiveSummaryError string              // Non-empty if summary generation failed (fail-open)
	Error                 error               // Error details (if failed/timed_out)
}

ExecutionResult is lightweight — just the terminal state. All intermediate state (TimelineEvents, Interactions, Stages) was already written to DB by the executor during processing.

type InvestigationContextBuilder 

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

InvestigationContextBuilder reconstructs the full investigation context (alert data, runbook, tool inventory, timeline) for a session. Used by both the scoring Reflector and the feedback Reflector so both get the same rich context for memory extraction.

func NewInvestigationContextBuilder 

func NewInvestigationContextBuilder(
	cfg *config.Config,
	dbClient *ent.Client,
	stageService *services.StageService,
	timelineService *services.TimelineService,
	runbookService *runbook.Service,
) *InvestigationContextBuilder

NewInvestigationContextBuilder creates a builder from the services ScoringExecutor already has.

func (*InvestigationContextBuilder) Build 

func (b *InvestigationContextBuilder) Build(ctx context.Context, session *ent.AlertSession) string

Build returns the full investigation context string for a session.

type PoolHealth 

type PoolHealth struct {
	IsHealthy        bool           `json:"is_healthy"`
	DBReachable      bool           `json:"db_reachable"`
	DBError          string         `json:"db_error,omitempty"`
	PodID            string         `json:"pod_id"`
	ActiveWorkers    int            `json:"active_workers"`
	TotalWorkers     int            `json:"total_workers"`
	ActiveSessions   int            `json:"active_sessions"`
	MaxConcurrent    int            `json:"max_concurrent"`
	QueueDepth       int            `json:"queue_depth"`
	WorkerStats      []WorkerHealth `json:"worker_stats"`
	LastOrphanScan   time.Time      `json:"last_orphan_scan"`
	OrphansRecovered int            `json:"orphans_recovered"`
}

PoolHealth contains health information for the entire worker pool.

type RealSessionExecutor 

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

RealSessionExecutor implements SessionExecutor using the agent framework.

func NewRealSessionExecutor 

func NewRealSessionExecutor(cfg *config.Config, dbClient *ent.Client, llmClient agent.LLMClient, eventPublisher agent.EventPublisher, mcpFactory *mcp.ClientFactory, runbookService *runbook.Service, memoryService *memory.Service, memoryConfig *config.MemoryConfig) *RealSessionExecutor

NewRealSessionExecutor creates a new session executor. eventPublisher may be nil (streaming disabled). mcpFactory may be nil (MCP disabled — uses stub tool executor). runbookService may be nil (uses config default runbook content). memoryService and memoryConfig may be nil (memory disabled).

func (*RealSessionExecutor) Execute 

func (e *RealSessionExecutor) Execute(ctx context.Context, session *ent.AlertSession) *ExecutionResult

Execute runs the session through the agent chain. Stages are executed sequentially. On any stage failure, the chain stops (fail-fast). After all stages complete, an executive summary is generated (fail-open).

func (*RealSessionExecutor) SetCostBook 

func (e *RealSessionExecutor) SetCostBook(book *cost.Book)

SetCostBook sets the price book used when persisting LLM interaction costs. May be nil (estimation skipped).

type ScoringExecutor 

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

ScoringExecutor orchestrates the scoring workflow: creating stage/execution records, running the scoring agent, and writing results to session_scores. It is called asynchronously after session completion (auto-trigger) and on-demand via the re-score API endpoint.

func NewScoringExecutor 

func NewScoringExecutor(
	cfg *config.Config,
	dbClient *ent.Client,
	llmClient agent.LLMClient,
	eventPublisher agent.EventPublisher,
	runbookService *runbook.Service,
	memoryService *memory.Service,
) *ScoringExecutor

NewScoringExecutor creates a new ScoringExecutor. runbookService may be nil (runbook content will be omitted from the scoring context). memoryService may be nil (memory extraction will be skipped).

func (*ScoringExecutor) RunFeedbackReflectorAsync 

func (e *ScoringExecutor) RunFeedbackReflectorAsync(sessionID, feedbackText, qualityRating string)

RunFeedbackReflectorAsync spawns a goroutine to run the feedback Reflector. The execution is attached to the session's existing scoring stage so its timeline events and LLM interactions are visible alongside the original score.

func (*ScoringExecutor) ScoreSessionAsync 

func (e *ScoringExecutor) ScoreSessionAsync(sessionID, triggeredBy string, checkEnabled bool)

ScoreSessionAsync launches scoring in a background goroutine. Silently returns if scoring is disabled or the executor is stopped. Used by the worker for auto-trigger after session completion.

func (*ScoringExecutor) SetCostBook 

func (e *ScoringExecutor) SetCostBook(book *cost.Book)

SetCostBook sets the price book used when persisting LLM interaction costs. Rebuilds the interaction service so subsequent writes use the book.

func (*ScoringExecutor) Stop 

func (e *ScoringExecutor) Stop(gracePeriod time.Duration)

Stop marks the executor as stopped, waits up to gracePeriod for in-flight scoring to complete naturally, then cancels any remaining contexts and waits for goroutines to drain.

func (*ScoringExecutor) SubmitScoring 

func (e *ScoringExecutor) SubmitScoring(ctx context.Context, sessionID, triggeredBy string, checkEnabled bool) (string, error)

SubmitScoring creates the scoring records (stage, session_score, execution) synchronously and launches the LLM evaluation in a background goroutine. Returns the session_score ID immediately for the API response. checkEnabled controls whether the chain's scoring.enabled flag is enforced.

type SessionExecutor 

type SessionExecutor interface {
	Execute(ctx context.Context, session *ent.AlertSession) *ExecutionResult
}

SessionExecutor is the interface for session processing.

The executor owns the ENTIRE session lifecycle internally:

  • Executes all stages sequentially (from chain config)
  • If a stage fails, the session stops immediately
  • Always forces conclusion at max iterations (no pause/resume)

The executor writes results PROGRESSIVELY during execution, not at the end. The worker only handles: claiming, heartbeat, terminal status update, and event cleanup.

type SessionRegistry 

type SessionRegistry interface {
	RegisterSession(sessionID string, cancel context.CancelFunc)
	UnregisterSession(sessionID string)
}

SessionRegistry is the subset of WorkerPool used by Worker for session registration.

type StubExecutor 

type StubExecutor struct{}

StubExecutor is a test/placeholder SessionExecutor that returns "completed" immediately. The real implementation is RealSessionExecutor in executor.go.

func NewStubExecutor 

func NewStubExecutor() *StubExecutor

NewStubExecutor creates a new stub executor.

func (*StubExecutor) Execute 

func (e *StubExecutor) Execute(ctx context.Context, session *ent.AlertSession) *ExecutionResult

Execute returns a completed result immediately (no-op).

type Worker 

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

Worker is a single queue worker that polls for and processes sessions.

func NewWorker 

func NewWorker(id, podID string, client *ent.Client, cfg *config.QueueConfig, executor SessionExecutor, scoringExecutor *ScoringExecutor, pool SessionRegistry, eventPublisher agent.EventPublisher, slackService *tarsyslack.Service) *Worker

NewWorker creates a new queue worker. eventPublisher may be nil (streaming disabled). slackService may be nil (Slack notifications disabled). scoringExecutor may be nil (scoring disabled).

func (*Worker) Health 

func (w *Worker) Health() WorkerHealth

Health returns the current worker health status.

func (*Worker) Start 

func (w *Worker) Start(ctx context.Context)

Start begins the worker polling loop in a goroutine.

func (*Worker) Stop 

func (w *Worker) Stop()

Stop signals the worker to stop and waits for it to finish. It is safe to call Stop multiple times.

type WorkerHealth 

type WorkerHealth struct {
	ID                string       `json:"id"`
	Status            WorkerStatus `json:"status"` // idle or working
	CurrentSessionID  string       `json:"current_session_id,omitempty"`
	SessionsProcessed int          `json:"sessions_processed"`
	LastActivity      time.Time    `json:"last_activity"`
}

WorkerHealth contains health information for a single worker.

type WorkerPool 

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

WorkerPool manages a pool of queue workers.

func NewWorkerPool 

func NewWorkerPool(podID string, client *ent.Client, cfg *config.QueueConfig, executor SessionExecutor, scoringExecutor *ScoringExecutor, eventPublisher agent.EventPublisher, slackService *tarsyslack.Service) *WorkerPool

NewWorkerPool creates a new worker pool. eventPublisher may be nil (streaming disabled). slackService may be nil (Slack notifications disabled). scoringExecutor may be nil (scoring disabled).

func (*WorkerPool) CancelSession 

func (p *WorkerPool) CancelSession(sessionID string) bool

CancelSession triggers context cancellation for a session on this pod. Returns true if the session was found and cancelled on this pod.

func (*WorkerPool) Health 

func (p *WorkerPool) Health() *PoolHealth

Health returns the current health status of the pool.

func (*WorkerPool) RegisterSession 

func (p *WorkerPool) RegisterSession(sessionID string, cancel context.CancelFunc)

RegisterSession stores a cancel function for manual cancellation.

func (*WorkerPool) Start 

func (p *WorkerPool) Start(ctx context.Context) error

Start spawns worker goroutines and the orphan detection background task. It is safe to call multiple times; subsequent calls are no-ops.

func (*WorkerPool) Stop 

func (p *WorkerPool) Stop()

Stop signals all workers to stop and waits for them to finish. Workers finish their current sessions before exiting (graceful shutdown).

func (*WorkerPool) UnregisterSession 

func (p *WorkerPool) UnregisterSession(sessionID string)

UnregisterSession removes the cancel function when processing ends.

type WorkerStatus 

type WorkerStatus string

WorkerStatus represents the current state of a worker. 

const (
	WorkerStatusIdle    WorkerStatus = "idle"
	WorkerStatusWorking WorkerStatus = "working"
)

Worker status constants.

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.