engine

package
v0.35.3 Latest Latest
Warning

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

 Go to latest Published: Mar 23, 2026 License: MIT Imports: 21 Imported by: 0

README

HotPlex Engine: The Control Plane

The engine package is the core orchestration layer of HotPlex. It transforms local AI CLI agents (like Claude Code) into high-availability, production-ready services by managing process lifecycles, security boundaries, and real-time event streaming.

🏛 Architecture Overview

The Engine operates as a Stateful Multi-Session Controller. It manages a pool of persistent CLI processes to eliminate the overhead of repeated cold starts.

Agent Architecture

Key Architectural Concepts
  • Hot-Multiplexing: Instead of spawning a new process for every user message, the Engine keeps CLI processes alive in a "Busy/Ready" state. Subsequent turns are piped directly via stdin, reducing latency by 90%+.
  • Session Isolation: Each user session is bound to a unique working directory and a dedicated OS process group (PGID), ensuring that file operations and recursive tool executions are strictly sandboxed.
  • Regex Firewall (WAF): A pre-execution security layer that scans user prompts for dangerous patterns (e.g., recursive rm -rf /, credential exfiltration) before they reach the AI agent.
  • Event Bridge: Translates raw CLI stdout (often inconsistent JSON lines) into a strictly typed, normalized event stream.

🛠 Developer Guide

1. Initializing the Engine

The Engine is typically initialized as a singleton.

opts := engine.EngineOptions{
    Provider:         claudeProvider, // e.g. provider.NewClaudeCodeProvider
    Namespace:        "prod",
    Timeout:          10 * time.Minute,
    IdleTimeout:      30 * time.Minute,
    AllowedTools:     []string{"ls", "cat", "grep"},
    PermissionMode:  "auto", // auto-approve non-destructive tools
}

eng, err := engine.NewEngine(opts)
2. Executing a Task

To run a task, you provide a context, a session Config, the prompt, and a callback to receive real-time updates.

cfg := &types.Config{
    SessionID: "user-123-abc",
    WorkDir:   "/tmp/workspace/user-123",
}

err := eng.Execute(ctx, cfg, "Analyze this project structure", func(eventType string, data any) error {
    switch eventType {
    case "thinking":
        fmt.Printf("AI is thinking: %v\n", data)
    case "answer":
        fmt.Printf("Token: %v\n", data)
    case "session_stats":
        stats := data.(*event.SessionStatsData)
        fmt.Printf("Task completed. Tokens used: %d\n", stats.TotalTokens)
    }
    return nil
})
3. Session Lifecycle Management

The Engine provides several methods to control active sessions:

  • StopSession(sessionID, reason): Gracefully terminates a session and kills the underlying process group.
  • ResetSessionProvider(sessionID): Signals the engine to start a fresh AI context (clears conversation history) on the next execution.
  • GetSessionStats(sessionID): Retrieves real-time metrics (tokens used, tools called, duration breakdown).

🛡 Security Boundaries

The Engine enforces three layers of security:

  1. Input Filtering: Regex-based detection of malicious prompts.
  2. Runtime Constraints: AllowedTools and DisallowedTools whitelist/blacklist.
  3. OS Isolation: Each session runs in its own directory. Access outside WorkDir can be restricted via provider-specific flags.

📊 Observability

The SessionStats object provides high-fidelity tracking of AI performance:

  • Duration Breakdown: Thinking time vs. Tool execution time vs. Text generation time.
  • Token Accounting: Input, output, and cache (Read/Write) tokens.
  • Tool Audit: A complete list of all local tools successfully invoked during the session.

Package Path: github.com/hrygo/hotplex/engine
Core Components: Engine, SessionManager, Provider

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Engine 

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

Engine is the core Control Plane for AI CLI agent integration. Configured as a long-lived Singleton, it transforms local CLI tools into production-ready services by managing a hot-multiplexed process pool, enforcing security WAF rules, and providing a unified event-driven SDK for application integration.

func NewEngine 

func NewEngine(options EngineOptions) (*Engine, error)

NewEngine creates a new HotPlex Engine instance.

func (*Engine) AddCronJob added in v0.35.0 

func (r *Engine) AddCronJob(ctx context.Context, job *cron.CronJob) error

AddCronJob adds a new cron job to the scheduler.

func (*Engine) AddRelayBinding added in v0.35.0 

func (r *Engine) AddRelayBinding(ctx context.Context, binding *relay.RelayBinding) error

AddRelayBinding adds a new relay binding.

func (*Engine) CheckDanger added in v0.18.0 

func (r *Engine) CheckDanger(prompt string) (blocked bool, operation, reason string)

CheckDanger exposes the WAF danger detector to the chatapps layer. Returns blocked=true if the prompt matches dangerous patterns. chatapps uses this for pre-flight WAF checks before calling Execute.

func (*Engine) CleanupSessionFiles added in v0.17.0 

func (r *Engine) CleanupSessionFiles(sessionID string) error

CleanupSessionFiles deletes all session files associated with the provider session. This handles the complete removal of context on disk for commands like /reset. Note: This method requires the session to still exist in the pool. Use CleanupSessionFilesDirect if the session has already been terminated.

func (*Engine) CleanupSessionFilesDirect added in v0.29.0 

func (r *Engine) CleanupSessionFilesDirect(providerSessionID, workDir string) error

CleanupSessionFilesDirect deletes session files using explicit IDs. Use this when the session has already been terminated and removed from the pool. This is the safe way to cleanup after StopSession has been called.

func (*Engine) Close 

func (r *Engine) Close() error

Close terminates all active sessions managed by this runner and cleans up resources. It triggers Graceful Shutdown by cascading termination signals down to the SessionManager, which drops the entire process group (PGID) to prevent zombie processes.

func (*Engine) DeleteCronJob added in v0.35.0 

func (r *Engine) DeleteCronJob(ctx context.Context, id string) error

DeleteCronJob removes a cron job from the scheduler.

func (*Engine) EnterDrain added in v0.35.0 

func (r *Engine) EnterDrain(msg string)

EnterDrain puts the engine into drain mode. New sessions will be rejected while existing sessions continue.

func (*Engine) Execute 

func (r *Engine) Execute(ctx context.Context, cfg *types.Config, prompt string, callback event.Callback) error

Execute runs Claude Code CLI with the given configuration and streams

func (*Engine) ExitDrain added in v0.35.0 

func (r *Engine) ExitDrain()

ExitDrain exits drain mode.

func (*Engine) GetAllowedTools added in v0.11.0 

func (r *Engine) GetAllowedTools() []string

GetAllowedTools returns the current allowed tools list. Thread-safe: uses internal mutex.

func (*Engine) GetCLIVersion 

func (r *Engine) GetCLIVersion() (string, error)

GetCLIVersion returns the Claude Code CLI version.

func (*Engine) GetCronRuns added in v0.35.0 

func (r *Engine) GetCronRuns(ctx context.Context, jobID string) ([]*cron.CronRun, error)

GetCronRuns returns all runs for a job.

func (*Engine) GetDisallowedTools added in v0.11.0 

func (r *Engine) GetDisallowedTools() []string

GetDisallowedTools returns the current disallowed tools list. Thread-safe: uses internal mutex.

func (*Engine) GetDrainMessage added in v0.35.0 

func (r *Engine) GetDrainMessage() string

GetDrainMessage returns the current drain message.

func (*Engine) GetOptions added in v0.21.0 

func (e *Engine) GetOptions() EngineOptions

GetOptions returns the current engine options.

func (*Engine) GetSession added in v0.12.0 

func (r *Engine) GetSession(sessionID string) (*intengine.Session, bool)

GetSession retrieves an active session by sessionID. Returns the session and true if found, or nil and false if not found.

func (*Engine) GetSessionManager added in v0.32.0 

func (r *Engine) GetSessionManager() intengine.SessionManager

GetSessionManager returns the underlying session manager for admin operations.

func (*Engine) GetSessionStats 

func (r *Engine) GetSessionStats(sessionID string) *SessionStats

GetSessionStats returns a copy of the accumulated session stats.

func (*Engine) HandleRelay added in v0.35.0 

func (e *Engine) HandleRelay(ctx context.Context, req *RelayRequest) (*RelayResponse, error)

HandleRelay processes an incoming relay message.

func (*Engine) IsDraining added in v0.35.0 

func (r *Engine) IsDraining() bool

IsDraining returns true if the engine is in drain mode.

func (*Engine) ListCronJobs added in v0.35.0 

func (r *Engine) ListCronJobs(ctx context.Context) ([]*cron.CronJob, error)

ListCronJobs returns all cron jobs.

func (*Engine) ListRelayBindings added in v0.35.0 

func (r *Engine) ListRelayBindings(ctx context.Context) ([]*relay.RelayBinding, error)

ListRelayBindings returns all relay bindings.

func (*Engine) PauseCronJob added in v0.35.0 

func (r *Engine) PauseCronJob(ctx context.Context, id string) error

PauseCronJob pauses a cron job.

func (*Engine) PermissionMatcher added in v0.31.9 

func (r *Engine) PermissionMatcher() *permission.PermissionMatcher

PermissionMatcher returns the engine's permission matcher.

func (*Engine) RemoveRelayBinding added in v0.35.0 

func (r *Engine) RemoveRelayBinding(ctx context.Context, chatID string) error

RemoveRelayBinding removes a relay binding by chat ID.

func (*Engine) ResumeCronJob added in v0.35.0 

func (r *Engine) ResumeCronJob(ctx context.Context, id string) error

ResumeCronJob resumes a paused cron job.

func (*Engine) SendRelay added in v0.35.0 

func (r *Engine) SendRelay(ctx context.Context, to, content string) (*relay.RelayResponse, error)

SendRelay sends a message to a remote agent via the relay network.

func (*Engine) SetAllowedTools added in v0.11.0 

func (r *Engine) SetAllowedTools(tools []string)

SetAllowedTools sets the allowed tools for the engine. This affects new sessions created after the call. Thread-safe: uses internal mutex.

func (*Engine) SetDangerAllowPaths 

func (r *Engine) SetDangerAllowPaths(paths []string)

SetDangerAllowPaths sets the allowed safe paths for the danger detector.

func (*Engine) SetDangerBypassEnabled 

func (r *Engine) SetDangerBypassEnabled(token string, enabled bool) error

SetDangerBypassEnabled enables or disables danger detection bypass. WARNING: Only use for Evolution mode (admin only).

func (*Engine) SetDisallowedTools added in v0.11.0 

func (r *Engine) SetDisallowedTools(tools []string)

SetDisallowedTools sets the disallowed tools for the engine. This affects new sessions created after the call. Thread-safe: uses internal mutex.

func (*Engine) SetPermissionMatcher added in v0.31.9 

func (r *Engine) SetPermissionMatcher(m *permission.PermissionMatcher)

SetPermissionMatcher sets the permission matcher for the engine.

func (*Engine) StopSession 

func (r *Engine) StopSession(sessionID string, reason string) error

StopSession terminates a running session by session ID. This is the implementation for session.stop from the spec.

func (*Engine) ValidateConfig 

func (r *Engine) ValidateConfig(cfg *types.Config) error

ValidateConfig validates the Config.

type EngineOptions 

type EngineOptions = intengine.EngineOptions

EngineOptions defines the configuration parameters for initializing a new HotPlex Engine. It allows customization of timeouts, logging, and foundational security boundaries that apply to all sessions managed by this engine instance.

type RelayExecutor added in v0.35.0 

type RelayExecutor interface {
	HandleRelay(ctx context.Context, req *RelayRequest) (*RelayResponse, error)
}

RelayExecutor handles cross-instance relay requests.

type RelayRequest added in v0.35.0 

type RelayRequest struct {
	From       string `json:"from"`
	To         string `json:"to"`
	SessionKey string `json:"session_key"`
	Message    string `json:"message"`
}

RelayRequest is the input for a relay operation.

type RelayResponse added in v0.35.0 

type RelayResponse = relay.RelayResponse

RelayResponse is the result of a relay operation.

type SessionStats 

type SessionStats struct {
	SessionID            string          `json:"session_id"`
	StartTime            time.Time       `json:"-"` // Internal use only, use ToSummary() for JSON
	TotalDurationMs      int64           `json:"total_duration_ms"`
	ThinkingDurationMs   int64           `json:"thinking_duration_ms"`
	ToolDurationMs       int64           `json:"tool_duration_ms"`
	GenerationDurationMs int64           `json:"generation_duration_ms"`
	InputTokens          int32           `json:"input_tokens"`
	OutputTokens         int32           `json:"output_tokens"`
	CacheWriteTokens     int32           `json:"cache_write_tokens"`
	CacheReadTokens      int32           `json:"cache_read_tokens"`
	ToolCallCount        int32           `json:"tool_call_count"`
	ToolsUsed            map[string]bool `json:"-"` // Internal set, use ToSummary() for array
	FilesModified        int32           `json:"files_modified"`
	FilePaths            []string        `json:"file_paths"`
	// contains filtered or unexported fields
}

SessionStats collects session-level statistics for Geek/Evolution modes.

func (*SessionStats) EndGeneration 

func (s *SessionStats) EndGeneration()

EndGeneration marks the end of the generation phase and records its duration.

func (*SessionStats) EndThinking 

func (s *SessionStats) EndThinking()

EndThinking marks the end of the thinking phase and records its duration.

func (*SessionStats) FinalizeDuration 

func (s *SessionStats) FinalizeDuration() *SessionStats

FinalizeDuration finalizes any ongoing phase tracking and returns the final stats.

func (*SessionStats) GetCurrentToolID added in v0.13.0 

func (s *SessionStats) GetCurrentToolID() string

GetCurrentToolID returns the current tool ID being tracked. Returns empty string if no tool is currently being tracked.

func (*SessionStats) GetCurrentToolName added in v0.13.0 

func (s *SessionStats) GetCurrentToolName() string

GetCurrentToolName returns the current tool name being tracked. Returns empty string if no tool is currently being tracked.

func (*SessionStats) GetToolNameByToolID added in v0.15.0 

func (s *SessionStats) GetToolNameByToolID(toolID string) string

GetToolNameByToolID returns the tool name for a given tool ID. This is useful when tool_result events don't include tool_name. Returns empty string if tool ID is not found.

func (*SessionStats) RecordFileModification 

func (s *SessionStats) RecordFileModification(filePath string)

RecordFileModification records that a file was modified. Uses O(1) map lookup for deduplication instead of O(n) linear scan.

func (*SessionStats) RecordTokens 

func (s *SessionStats) RecordTokens(input, output, cacheWrite, cacheRead int32)

RecordTokens records token usage.

func (*SessionStats) RecordToolResult 

func (s *SessionStats) RecordToolResult(toolID string) (durationMs int64, toolName string)

RecordToolResult records the end of a tool call. If RecordToolUse was not called (e.g., Claude Code didn't send tool_use event), this will record a minimal duration (1ms) to indicate the tool completed.

func (*SessionStats) RecordToolUse 

func (s *SessionStats) RecordToolUse(toolName, toolID string)

RecordToolUse records the start of a tool call.

func (*SessionStats) StartGeneration 

func (s *SessionStats) StartGeneration()

StartGeneration marks the start of the generation phase.

func (*SessionStats) StartThinking 

func (s *SessionStats) StartThinking()

StartThinking marks the start of the thinking phase.

func (*SessionStats) ToSummary 

func (s *SessionStats) ToSummary() map[string]interface{}

ToSummary converts stats to a summary map for JSON serialization.

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.