agentmgr

package
v0.1.0-proto2a Latest Latest
Warning

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

 Go to latest Published: Jul 28, 2026 License: MPL-2.0 Imports: 34 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrCycle = toposort.ErrCycle

ErrCycle mirrors toposort.ErrCycle at this package's boundary; only HARD dependency cycles are errors.

View Source 
var ErrNoProcess = errors.New("agentmgr: no running process to checkpoint")

ErrNoProcess is returned when a checkpoint is requested for a runner that has no running process. It is distinct from a CRIU failure: the caller's mistake is asking at the wrong time, not a dump that broke.

Functions

func TopologicalSort 

func TopologicalSort(agents map[string]Agent) ([]string, error)

TopologicalSort orders agents deps-first through the shared toposort (review R5: one implementation). Requires() edges order and cycle-reject; Wants() edges order when satisfiable and are dropped - never blocking - when they would form a cycle (review R14: soft deps must not block; the lifecycle controller separately tolerates soft-dep start failures).

Types

type Agent 

type Agent interface {
	ID() string
	Type() string
	Lang() string
	Dependencies() []string
	Controller() *lifecycle.Controller
	Describe() map[string]string
	Requires() []string
	Wants() []string
	SetRunID(string)
}

type AgentManager 

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

func NewAgentManager 

func NewAgentManager(bus *eventbus.EventBus[*anypb.Any], lbus *lifecycle.TypedBus, pyRunnerPath string, productionMode bool, verifyKey ed25519.PublicKey) *AgentManager

func (*AgentManager) All 

func (am *AgentManager) All() map[string]Agent

func (*AgentManager) Deregister 

func (am *AgentManager) Deregister(id string)

Deregister removes an agent from the manager; unknown ids are a no-op. Callers stop the agent first - deregistering does not kill processes.

func (*AgentManager) DiscoverFromPath 

func (am *AgentManager) DiscoverFromPath(root string) ([]map[string]string, error)

DiscoverFromPath discovers agents from a single root path (legacy method for backward compatibility). Prefer DiscoverFromPaths() for new code.

func (*AgentManager) DiscoverFromPaths 

func (am *AgentManager) DiscoverFromPaths() ([]map[string]string, error)

DiscoverFromPaths discovers agents from all configured search paths. Paths are searched in priority order (Development → User → System). First occurrence of an agent ID wins (higher priority path).

func (*AgentManager) Get 

func (am *AgentManager) Get(id string) Agent

func (*AgentManager) Instantiate 

func (am *AgentManager) Instantiate(instanceID, templateID string) (Agent, error)

Instantiate constructs and registers a new agent that runs the same binary as an installed template agent, under its own instance id.

This is the orchestrator's node-dispatch seam (proto-2 Phase 3): a global AgentSpec references an installed, discovery-verified agent type by id, and every scheduled instance is a fresh incarnation of that binary. Specs never carry arbitrary commands, so the discovery security model (signed-only production discovery, R20) holds for remotely scheduled work exactly as for local agents.

Go binaries only for now: multi-instance python agents need per-instance runner and socket plumbing that does not exist yet. Instances get no listen socket (it would collide with the template's) and inherit the template's resource limits and capabilities.

func (*AgentManager) NotifyExited 

func (am *AgentManager) NotifyExited(pid int, ws syscall.WaitStatus)

NotifyExited receives a reaped child's true exit status from the subreaper loop (PID-1 mode). Known agent processes are logged with their identity - their own exit watchers drive the state change; the wait status recorded here is the authoritative one, since the reaper may win the wait race. Unknown pids are adopted orphans: reaped and noted quietly (aggressive logging floods kmsg).

func (*AgentManager) Register 

func (am *AgentManager) Register(a Agent)

func (*AgentManager) StartAll 

func (am *AgentManager) StartAll() error

func (*AgentManager) StopAll 

func (am *AgentManager) StopAll() error

func (*AgentManager) TopologicalSort 

func (am *AgentManager) TopologicalSort() ([]string, error)

type CronSchedule 

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

CronSchedule represents a cron-based schedule

func (*CronSchedule) Next 

func (s *CronSchedule) Next(t time.Time) time.Time

type Discovered 

type Discovered struct {
	ID           string
	Type         string
	Lang         string
	Path         string
	Requires     []string
	Wants        []string
	WantedBy     []string
	RequiredBy   []string
	ListenStream string
	Capabilities []string
}

type GoAgent 

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

func NewGoAgent 

func NewGoAgent(
	id, typ, binaryPath string,
	reqs, wants, wantedBy, requiredBy []string,
	listenStream string,
	cpuLimit, memLimit string,
	caps []string,
	globalBus *eventbus.EventBus[*anypb.Any],
	depView lifecycle.DependencyResolver,
) *GoAgent

func (*GoAgent) Arm 

func (a *GoAgent) Arm() error

func (*GoAgent) Checkpoint 

func (a *GoAgent) Checkpoint(ctx context.Context, dir string) error

Checkpoint implements lifecycle.Checkpointer for GoAgent.

func (*GoAgent) Controller 

func (a *GoAgent) Controller() *lifecycle.Controller

func (*GoAgent) Dependencies 

func (a *GoAgent) Dependencies() []string

func (*GoAgent) Describe 

func (a *GoAgent) Describe() map[string]string

Describe mirrors PythonAgent.Describe key-for-key: the Go/Python describe schema is a parity contract (the cross-ADK suite asserts it), so the two implementations must emit the same keys - notably "language" (not "lang"), "path", and the resource limits.

func (*GoAgent) EnsureListener 

func (a *GoAgent) EnsureListener() (*os.File, error)

EnsureListener creates the listener if it doesn't exist.

func (*GoAgent) ID 

func (a *GoAgent) ID() string

func (*GoAgent) Lang 

func (a *GoAgent) Lang() string

func (*GoAgent) Pid 

func (a *GoAgent) Pid() (int, bool)

Pid returns the running agent process id, or false when no process is running. Orchestrators use it to capture the runtime locator (start epoch, pid namespace) for gossip.

func (*GoAgent) Reload 

func (a *GoAgent) Reload(ctx context.Context) error

func (*GoAgent) Requires 

func (a *GoAgent) Requires() []string

func (*GoAgent) Reset 

func (a *GoAgent) Reset()

func (*GoAgent) Restore 

func (a *GoAgent) Restore(ctx context.Context, dir string) error

Restore implements lifecycle.Checkpointer for GoAgent.

func (*GoAgent) SetRunID 

func (a *GoAgent) SetRunID(id string)

func (*GoAgent) SetTrafficHandler 

func (a *GoAgent) SetTrafficHandler(fn func())

func (*GoAgent) Start 

func (a *GoAgent) Start(ctx context.Context) error

func (*GoAgent) Stop 

func (a *GoAgent) Stop(ctx context.Context) error

func (*GoAgent) Type 

func (a *GoAgent) Type() string

func (*GoAgent) Uptime 

func (a *GoAgent) Uptime() time.Duration

func (*GoAgent) Wants 

func (a *GoAgent) Wants() []string

type IntervalSchedule 

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

IntervalSchedule represents a simple interval-based schedule

func (*IntervalSchedule) Next 

func (s *IntervalSchedule) Next(t time.Time) time.Time

type MockDependencyResolver 

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

func NewMockDependencyResolver 

func NewMockDependencyResolver() *MockDependencyResolver

func (*MockDependencyResolver) DepsOf 

func (m *MockDependencyResolver) DepsOf(id string) []string

func (*MockDependencyResolver) EnsureStarted 

func (m *MockDependencyResolver) EnsureStarted(ctx context.Context, id string) error

func (*MockDependencyResolver) IsRunning 

func (m *MockDependencyResolver) IsRunning(id string) bool

func (*MockDependencyResolver) SetDeps 

func (m *MockDependencyResolver) SetDeps(id string, deps []string)

func (*MockDependencyResolver) SetRunning 

func (m *MockDependencyResolver) SetRunning(id string, running bool)

type PythonAgent 

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

func NewPythonAgent 

func NewPythonAgent(
	id, typ, modulePath, runnerPath string,
	reqs, wants, wantedBy, requiredBy []string,
	listenStream string,
	cpuLimit, memLimit string,
	caps []string,
	globalBus *eventbus.EventBus[*anypb.Any],
	depView lifecycle.DependencyResolver,
	productionMode bool,
) *PythonAgent

func (*PythonAgent) Arm 

func (a *PythonAgent) Arm() error

func (*PythonAgent) Checkpoint 

func (a *PythonAgent) Checkpoint(ctx context.Context, dir string) error

Checkpoint implements lifecycle.Checkpointer for PythonAgent.

func (*PythonAgent) Controller 

func (a *PythonAgent) Controller() *lifecycle.Controller

func (*PythonAgent) Dependencies 

func (a *PythonAgent) Dependencies() []string

func (*PythonAgent) Describe 

func (a *PythonAgent) Describe() map[string]string

func (*PythonAgent) EnsureListener 

func (a *PythonAgent) EnsureListener() (*os.File, error)

EnsureListener creates the listener if it doesn't exist.

func (*PythonAgent) ID 

func (a *PythonAgent) ID() string

func (*PythonAgent) Lang 

func (a *PythonAgent) Lang() string

func (*PythonAgent) Pid 

func (a *PythonAgent) Pid() (int, bool)

Pid returns the running agent process id, or false when no process is running (parity with GoAgent).

func (*PythonAgent) Reload 

func (a *PythonAgent) Reload(ctx context.Context) error

func (*PythonAgent) Requires 

func (a *PythonAgent) Requires() []string

func (*PythonAgent) Reset 

func (a *PythonAgent) Reset()

func (*PythonAgent) Restore 

func (a *PythonAgent) Restore(ctx context.Context, dir string) error

Restore implements lifecycle.Checkpointer for PythonAgent.

func (*PythonAgent) SetRunID 

func (a *PythonAgent) SetRunID(id string)

func (*PythonAgent) SetTrafficHandler 

func (a *PythonAgent) SetTrafficHandler(fn func())

func (*PythonAgent) Start 

func (a *PythonAgent) Start(ctx context.Context) error

func (*PythonAgent) Stop 

func (a *PythonAgent) Stop(ctx context.Context) error

func (*PythonAgent) Type 

func (a *PythonAgent) Type() string

func (*PythonAgent) Uptime 

func (a *PythonAgent) Uptime() time.Duration

func (*PythonAgent) Wants 

func (a *PythonAgent) Wants() []string

type Schedule 

type Schedule interface {
	Next(t time.Time) time.Time
}

Schedule represents a timer schedule

func ParseSchedule 

func ParseSchedule(s string) (Schedule, error)

ParseSchedule parses a schedule string into a Schedule Supports: - Systemd-style: OnUnitActiveSec=5s, OnBootSec=30s, OnStartupSec=1m - Cron expressions: */5 * * * * (every 5 minutes) - Named schedules: @hourly, @daily, @weekly, @monthly - Raw durations: 5s, 1m, 1h (treated as intervals)

type TimerAgent 

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

TimerAgent executes a Python agent on a schedule

func NewTimerAgent 

func NewTimerAgent(id, path, schedule, pyRunner string, bus *eventbus.EventBus[*anypb.Any], lbus *lifecycle.TypedBus) *TimerAgent

func (*TimerAgent) Controller 

func (ta *TimerAgent) Controller() *lifecycle.Controller

func (*TimerAgent) Dependencies 

func (ta *TimerAgent) Dependencies() []string

func (*TimerAgent) Describe 

func (ta *TimerAgent) Describe() map[string]string

func (*TimerAgent) ID 

func (ta *TimerAgent) ID() string

func (*TimerAgent) Initialize 

func (ta *TimerAgent) Initialize(ctx context.Context) error

Implement lifecycle.Runner interface

func (*TimerAgent) Lang 

func (ta *TimerAgent) Lang() string

func (*TimerAgent) Reload 

func (ta *TimerAgent) Reload(ctx context.Context) error

func (*TimerAgent) Requires 

func (ta *TimerAgent) Requires() []string

func (*TimerAgent) Reset 

func (ta *TimerAgent) Reset()

func (*TimerAgent) Restart 

func (ta *TimerAgent) Restart(ctx context.Context) error

func (*TimerAgent) SetRunID 

func (ta *TimerAgent) SetRunID(id string)

func (*TimerAgent) Start 

func (ta *TimerAgent) Start(_ context.Context) error

Start satisfies lifecycle.Runner. The caller's context is deliberately unused: the ticker loop must outlive Start (it is cancelled by Stop), so it runs on its own detached context.

func (*TimerAgent) Stop 

func (ta *TimerAgent) Stop(ctx context.Context) error

func (*TimerAgent) Type 

func (ta *TimerAgent) Type() string

func (*TimerAgent) Wants 

func (ta *TimerAgent) Wants() []string

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.