agent

package
v0.0.54 Latest Latest
Warning

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

 Go to latest Published: Jul 16, 2025 License: Apache-2.0 Imports: 24 Imported by: 0

Documentation

Overview

Package agent provides the core orchestration engine for devloop.

The agent package contains the orchestrator implementations that handle: - File system watching and event processing - Rule execution and process management - Gateway communication for distributed setups - Logging and output management

Key Components

The main orchestrator interface provides methods for: 

type Orchestrator interface {
	Start() error
	Stop() error
	GetConfig() *pb.Config
	TriggerRule(ruleName string) error
	GetRuleStatus(ruleName string) (*pb.RuleStatus, bool)
	GetWatchedPaths() []string
	ReadFileContent(path string) ([]byte, error)
	StreamLogs(ruleName string, filter string, stream pb.AgentService_StreamLogsClientServer) error
	SetGlobalDebounceDelay(duration time.Duration)
	SetVerbose(verbose bool)
}

Usage

Create and start an orchestrator: 

orchestrator := agent.NewOrchestrator("config.yaml", "")
if err := orchestrator.Start(); err != nil {
	log.Fatal(err)
}
defer orchestrator.Stop()

Configuration

The orchestrator is configured via YAML files with rules that define: - File patterns to watch - Commands to execute when files change - Debounce settings and execution options

Example configuration: 

settings:
  project_id: "my-project"
  prefix_logs: true
rules:
  - name: "build"
    watch:
      - action: include
        patterns: ["**/*.go"]
    commands:
      - "go build ."

Package gateway provides the central hub for managing multiple devloop agents.

The gateway package implements a distributed architecture where multiple devloop agents can connect to a central gateway service. This enables: - Centralized monitoring of multiple projects - Unified API access to all connected agents - Cross-project coordination and management

Gateway Service

The gateway service provides both gRPC and HTTP endpoints for: - Agent registration and communication - Client API access for external tools - Real-time log streaming - Project status monitoring

Usage

Start a gateway service: 

gateway := gateway.NewGatewayService(orchestrator)
if err := gateway.Start(grpcPort, httpPort); err != nil {
	log.Fatal(err)
}
defer gateway.Stop()

Connect agents to the gateway: 

devloop --mode agent --gateway-addr localhost:50051 -c project-a/.devloop.yaml
devloop --mode agent --gateway-addr localhost:50051 -c project-b/.devloop.yaml

Configuration Types

The package defines configuration structures for devloop projects: - Config: Main configuration structure - Rule: Individual automation rules - WatchConfig: File watching patterns - Settings: Global project settings

Example Gateway Setup 

// Start the central gateway
devloop --mode gateway --http-port 8080 --grpc-port 50051

// Connect individual project agents
cd project-a && devloop --mode agent --gateway-addr localhost:50051
cd project-b && devloop --mode agent --gateway-addr localhost:50051

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func LoadConfig 

func LoadConfig(configPath string) (*pb.Config, error)

LoadConfig reads and unmarshals the .devloop.yaml configuration file, resolving all relative watch paths to be absolute from the config file's location.

func MatcherMatches added in v0.0.42 

func MatcherMatches(m *pb.RuleMatcher, filePath string, rule *pb.Rule, configPath string) bool

func RuleMatches added in v0.0.42 

func RuleMatches(r *pb.Rule, filePath string, configPath string) *pb.RuleMatcher

Matches checks if the given file path matches the rule's watch criteria.

Types

type AgentService added in v0.0.42 

type AgentService struct {
	protos.UnimplementedAgentServiceServer
	// contains filtered or unexported fields
}

func NewAgentService added in v0.0.42 

func NewAgentService(orch *Orchestrator) *AgentService

func (*AgentService) GetConfig added in v0.0.42 

func (s *AgentService) GetConfig(ctx context.Context, req *protos.GetConfigRequest) (resp *protos.GetConfigResponse, err error)

Get the configs of this agent

func (*AgentService) GetRule added in v0.0.42 

func (s *AgentService) GetRule(ctx context.Context, req *protos.GetRuleRequest) (resp *protos.GetRuleResponse, err error)

func (*AgentService) ListWatchedPaths added in v0.0.42 

func (s *AgentService) ListWatchedPaths(ctx context.Context, req *protos.ListWatchedPathsRequest) (resp *protos.ListWatchedPathsResponse, err error)

func (*AgentService) StreamLogs added in v0.0.42 

func (s *AgentService) StreamLogs(req *protos.StreamLogsRequest, stream grpc.ServerStreamingServer[protos.StreamLogsResponse]) error

func (*AgentService) TriggerRule added in v0.0.42 

func (s *AgentService) TriggerRule(ctx context.Context, req *protos.TriggerRuleRequest) (resp *protos.TriggerRuleResponse, err error)

type CycleBreaker added in v0.0.51 

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

CycleBreaker manages advanced cycle breaking and resolution mechanisms

func NewCycleBreaker added in v0.0.51 

func NewCycleBreaker() *CycleBreaker

NewCycleBreaker creates a new cycle breaker

func (*CycleBreaker) CleanupOldData added in v0.0.51 

func (cb *CycleBreaker) CleanupOldData()

CleanupOldData removes old emergency break and disable records

func (*CycleBreaker) DisableRule added in v0.0.51 

func (cb *CycleBreaker) DisableRule(ruleName string, duration time.Duration)

DisableRule temporarily disables a rule for cycle breaking

func (*CycleBreaker) GenerateCycleResolutionSuggestions added in v0.0.51 

func (cb *CycleBreaker) GenerateCycleResolutionSuggestions(ruleName string, cycleType string) []string

GenerateCycleResolutionSuggestions generates suggestions for resolving detected cycles

func (*CycleBreaker) GetDisabledRules added in v0.0.51 

func (cb *CycleBreaker) GetDisabledRules() map[string]time.Time

GetDisabledRules returns a map of currently disabled rules and their disable times

func (*CycleBreaker) GetEmergencyBreakCount added in v0.0.51 

func (cb *CycleBreaker) GetEmergencyBreakCount(ruleName string) int

GetEmergencyBreakCount returns the number of emergency breaks for a rule

func (*CycleBreaker) IsRuleDisabled added in v0.0.51 

func (cb *CycleBreaker) IsRuleDisabled(ruleName string) bool

IsRuleDisabled checks if a rule is currently disabled

func (*CycleBreaker) TriggerEmergencyBreak added in v0.0.51 

func (cb *CycleBreaker) TriggerEmergencyBreak(ruleName string)

TriggerEmergencyBreak triggers an emergency cycle break for a rule

type FileModificationTracker added in v0.0.51 

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

FileModificationTracker tracks file modification frequencies for thrashing detection

func NewFileModificationTracker added in v0.0.51 

func NewFileModificationTracker() *FileModificationTracker

NewFileModificationTracker creates a new file modification tracker

func (*FileModificationTracker) CleanupOldModifications added in v0.0.51 

func (fmt *FileModificationTracker) CleanupOldModifications(maxAge time.Duration)

CleanupOldModifications removes old modification records

func (*FileModificationTracker) GetModificationCount added in v0.0.51 

func (fmt *FileModificationTracker) GetModificationCount(filePath string, windowSeconds uint32) int

GetModificationCount returns the number of modifications within the time window

func (*FileModificationTracker) GetThrashingFiles added in v0.0.51 

func (fmt *FileModificationTracker) GetThrashingFiles(windowSeconds, threshold uint32) []string

GetThrashingFiles returns a list of files that are currently thrashing

func (*FileModificationTracker) IsFileThrashing added in v0.0.51 

func (fmt *FileModificationTracker) IsFileThrashing(filePath string, windowSeconds, threshold uint32) bool

IsFileThrashing checks if a file is being modified too frequently

func (*FileModificationTracker) RecordModification added in v0.0.51 

func (fmt *FileModificationTracker) RecordModification(filePath string)

RecordModification records a file modification

type Orchestrator 

type Orchestrator struct {
	ConfigPath   string
	Config       *pb.Config
	Verbose      bool
	Watcher      *fsnotify.Watcher
	LogManager   *utils.LogManager
	ColorManager *utils.ColorManager
	// contains filtered or unexported fields
}

Orchestrator manages file watching and delegates execution to RuleRunners

func NewOrchestrator 

func NewOrchestrator(configPath string) (*Orchestrator, error)

NewOrchestrator creates a new orchestrator instance for managing file watching and rule execution.

The orchestrator handles: - Loading and validating configuration from configPath - Setting up file system watching for specified patterns - Managing rule execution and process lifecycle - Optional gateway communication if gatewayAddr is provided

Parameters:

  • configPath: Path to the .devloop.yaml configuration file
  • gatewayAddr: Optional gateway address for distributed mode (empty for standalone)

Returns an orchestrator instance ready to be started, or an error if configuration loading or initialization fails.

Example: 

// Standalone mode
orchestrator, err := NewOrchestrator(".devloop.yaml", "")

// Agent mode (connect to gateway)
orchestrator, err := NewOrchestrator(".devloop.yaml", "localhost:50051")

func NewOrchestratorForTesting 

func NewOrchestratorForTesting(configPath string) (*Orchestrator, error)

NewOrchestratorForTesting creates a new V2 orchestrator for testing

func (*Orchestrator) GetConfig 

func (o *Orchestrator) GetConfig() *pb.Config

GetConfig returns the orchestrator's configuration.

func (*Orchestrator) GetRuleStatus 

func (o *Orchestrator) GetRuleStatus(ruleName string) (rule *pb.Rule, status *pb.RuleStatus, ok bool)

GetRuleStatus returns the status of a specific rule

func (*Orchestrator) GetWatchedPaths 

func (o *Orchestrator) GetWatchedPaths() []string

GetWatchedPaths returns a unique list of all paths being watched by any rule

func (*Orchestrator) ProjectRoot 

func (o *Orchestrator) ProjectRoot() string

ProjectRoot returns the project root directory

func (*Orchestrator) ReadFileContent 

func (o *Orchestrator) ReadFileContent(path string) ([]byte, error)

ReadFileContent reads and returns the content of a specified file

func (*Orchestrator) SetGlobalDebounceDelay 

func (o *Orchestrator) SetGlobalDebounceDelay(duration time.Duration)

SetGlobalDebounceDelay sets the default debounce delay for all rules

func (*Orchestrator) SetRuleDebounceDelay 

func (o *Orchestrator) SetRuleDebounceDelay(ruleName string, duration time.Duration) error

SetRuleDebounceDelay sets the debounce delay for a specific rule

func (*Orchestrator) SetRuleVerbose 

func (o *Orchestrator) SetRuleVerbose(ruleName string, verbose bool) error

SetRuleVerbose sets the verbose flag for a specific rule

func (*Orchestrator) SetVerbose 

func (o *Orchestrator) SetVerbose(verbose bool)

SetVerbose sets the global verbose flag

func (*Orchestrator) Start 

func (o *Orchestrator) Start() error

Start begins file watching and initializes all RuleRunners

func (*Orchestrator) Stop 

func (o *Orchestrator) Stop() error

Stop gracefully shuts down the orchestrator

func (*Orchestrator) StreamLogs 

func (o *Orchestrator) StreamLogs(ruleName string, filter string, timeoutSeconds int64, writer *gocurrent.Writer[*pb.StreamLogsResponse]) error

StreamLogs streams the logs for a given rule to the provided Writer

func (*Orchestrator) TriggerRule 

func (o *Orchestrator) TriggerRule(ruleName string) error

TriggerRule manually triggers the execution of a specific rule

func (*Orchestrator) ValidateConfig added in v0.0.51 

func (o *Orchestrator) ValidateConfig() error

ValidateConfig performs static cycle detection and validation on the loaded configuration

type PrefixWriter 

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

PrefixWriter is an io.Writer that adds a prefix to each line of output.

func (*PrefixWriter) Write 

func (pw *PrefixWriter) Write(p []byte) (n int, err error)

Write implements the io.Writer interface.

type Rule added in v0.0.42 

type Rule struct {
	*pb.Rule
}

Rule defines a single watch-and-run rule.

func (*Rule) GetColor added in v0.0.42 

func (r *Rule) GetColor() string

GetColor returns the rule color (implements ColorRule interface)

func (*Rule) GetName added in v0.0.42 

func (r *Rule) GetName() string

GetName returns the rule name (implements Nameable interface for color generation)

func (*Rule) GetPrefix added in v0.0.42 

func (r *Rule) GetPrefix() string

GetPrefix returns the rule prefix (implements ColorRule interface)

type RuleRunner 

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

RuleRunner manages the execution lifecycle of a single rule

func NewRuleRunner 

func NewRuleRunner(rule *pb.Rule, orchestrator *Orchestrator) *RuleRunner

NewRuleRunner creates a new RuleRunner for the given rule

func (*RuleRunner) CleanupTriggerHistory added in v0.0.51 

func (r *RuleRunner) CleanupTriggerHistory()

CleanupTriggerHistory removes old trigger records to prevent memory growth

func (*RuleRunner) Execute 

func (r *RuleRunner) Execute() error

Execute runs the commands for this rule

func (*RuleRunner) GetRule 

func (r *RuleRunner) GetRule() *pb.Rule

GetRule returns the rule configuration

func (*RuleRunner) GetStatus 

func (r *RuleRunner) GetStatus() *pb.RuleStatus

GetStatus returns a copy of the current status

func (*RuleRunner) GetTriggerCount added in v0.0.51 

func (r *RuleRunner) GetTriggerCount(duration time.Duration) int

GetTriggerCount returns the trigger count within the specified duration

func (*RuleRunner) GetTriggerRate added in v0.0.51 

func (r *RuleRunner) GetTriggerRate() float64

GetTriggerRate returns the current trigger rate for this rule

func (*RuleRunner) IsRunning 

func (r *RuleRunner) IsRunning() bool

IsRunning returns true if any commands are currently running

func (*RuleRunner) Restart 

func (r *RuleRunner) Restart() error

Restart stops and starts the rule

func (*RuleRunner) SetDebounceDelay 

func (r *RuleRunner) SetDebounceDelay(duration time.Duration)

SetDebounceDelay sets the debounce delay for this rule

func (*RuleRunner) SetVerbose 

func (r *RuleRunner) SetVerbose(verbose bool)

SetVerbose sets the verbose flag for this rule

func (*RuleRunner) Start 

func (r *RuleRunner) Start() error

Start begins monitoring for this rule (non-blocking)

func (*RuleRunner) Stop 

func (r *RuleRunner) Stop() error

Stop terminates all processes and cleans up

func (*RuleRunner) TerminateProcesses 

func (r *RuleRunner) TerminateProcesses() error

TerminateProcesses terminates all running processes for this rule

func (*RuleRunner) TriggerDebounced 

func (r *RuleRunner) TriggerDebounced()

TriggerDebounced triggers execution after debounce period

func (*RuleRunner) TriggerDebouncedWithOptions added in v0.0.51 

func (r *RuleRunner) TriggerDebouncedWithOptions(bypassRateLimit bool)

TriggerDebouncedWithOptions triggers execution after debounce period with options

type RuleStatus added in v0.0.42 

type RuleStatus struct {
	*pb.RuleStatus
}

RuleStatus defines the status of a rule.

type Settings added in v0.0.42 

type Settings struct {
	*pb.Settings
}

func (*Settings) GetColorLogs added in v0.0.42 

func (s *Settings) GetColorLogs() bool

GetColorLogs returns whether color logs are enabled (implements ColorSettings interface)

func (*Settings) GetColorScheme added in v0.0.42 

func (s *Settings) GetColorScheme() string

GetColorScheme returns the color scheme (implements ColorSettings interface)

func (*Settings) GetCustomColors added in v0.0.42 

func (s *Settings) GetCustomColors() map[string]string

GetCustomColors returns the custom color mappings (implements ColorSettings interface)

type TriggerChain added in v0.0.51 

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

TriggerChain tracks cross-rule trigger relationships for cycle detection

func NewTriggerChain added in v0.0.51 

func NewTriggerChain() *TriggerChain

NewTriggerChain creates a new trigger chain tracker

func (*TriggerChain) CleanupOldChains added in v0.0.51 

func (tc *TriggerChain) CleanupOldChains()

CleanupOldChains removes old trigger chains to prevent memory growth

func (*TriggerChain) DetectCycle added in v0.0.51 

func (tc *TriggerChain) DetectCycle(sourceRule, targetRule string, maxDepth uint32) bool

DetectCycle detects if adding a new trigger would create a cycle

func (*TriggerChain) GetChainLength added in v0.0.51 

func (tc *TriggerChain) GetChainLength(ruleName string) int

GetChainLength returns the length of the trigger chain starting from a rule

func (*TriggerChain) RecordTrigger added in v0.0.51 

func (tc *TriggerChain) RecordTrigger(sourceRule, targetRule string)

RecordTrigger records that sourceRule triggered targetRule

type TriggerTracker added in v0.0.51 

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

TriggerTracker tracks rule execution frequency for rate limiting

func NewTriggerTracker added in v0.0.51 

func NewTriggerTracker() *TriggerTracker

NewTriggerTracker creates a new trigger tracker

func (*TriggerTracker) CleanupOldTriggers added in v0.0.51 

func (t *TriggerTracker) CleanupOldTriggers(duration time.Duration)

CleanupOldTriggers removes trigger records older than the specified duration

func (*TriggerTracker) GetBackoffLevel added in v0.0.51 

func (t *TriggerTracker) GetBackoffLevel() int

GetBackoffLevel returns the current backoff level

func (*TriggerTracker) GetLastTrigger added in v0.0.51 

func (t *TriggerTracker) GetLastTrigger() *time.Time

GetLastTrigger returns the timestamp of the most recent trigger

func (*TriggerTracker) GetTriggerCount added in v0.0.51 

func (t *TriggerTracker) GetTriggerCount(duration time.Duration) int

GetTriggerCount returns the number of triggers within the specified duration

func (*TriggerTracker) GetTriggerRate added in v0.0.51 

func (t *TriggerTracker) GetTriggerRate() float64

GetTriggerRate returns the current trigger rate (triggers per minute)

func (*TriggerTracker) IsInBackoff added in v0.0.51 

func (t *TriggerTracker) IsInBackoff() bool

IsInBackoff checks if the rule is currently in backoff period

func (*TriggerTracker) RecordTrigger added in v0.0.51 

func (t *TriggerTracker) RecordTrigger()

RecordTrigger records a new trigger event

func (*TriggerTracker) ResetBackoff added in v0.0.51 

func (t *TriggerTracker) ResetBackoff()

ResetBackoff resets the backoff level if enough time has passed

func (*TriggerTracker) SetBackoff added in v0.0.51 

func (t *TriggerTracker) SetBackoff()

SetBackoff sets the backoff period based on the current level

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.