agentmgr

package
v0.1.0-proto2g Latest Latest
Warning

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

 Go to latest Published: Aug 3, 2026 License: MPL-2.0 Imports: 35 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// EnvRunID correlates one agent start with the events it emits.
	EnvRunID = "ADK_RUN_ID"

	// EnvForceDummy selects the stub ADK without warning, for tests and
	// for hosts with no native ADK build.
	EnvForceDummy = "ADK_FORCE_DUMMY"

	// EnvRejectDummy makes falling back to the stub ADK a hard failure.
	// Set by the supervisor in production mode.
	EnvRejectDummy = "ADK_REJECT_DUMMY"
)

The kernel-to-agent environment contract.

These names are FIXED and carry no product prefix, which is the one deliberate exception to "each project owns one prefix" (cli-contract.md). They were GAPI_RUN_ID, GAPI_FORCE_DUMMY_ADK and GAPI_REJECT_DUMMY_ADK, which named the kernel to an agent author who has no reason to know the kernel exists - the same defect as the operator-facing names, one layer down (GAPI-DIV-061).

They do not follow the product identity, and that is the point. The reader is the ADK - adk/python/agent/runner.py and its Go peer - which ships with the kernel and is the same code whichever daemon spawned the process. Namespacing these by product would mean the runner could not know its own variable's name until it had read some other variable to find out, so the bootstrap would need a fixed name anyway; ADK_ IS that fixed name, with nothing composed on top.

A host running both daemons does not collide here: these are set on the CHILD's environment by exec, never exported process-wide.

View Source 
const TimerFireTimeout = 30 * time.Second

TimerFireTimeout bounds a single timer fire. Work that runs longer is killed mid-flight, so a scheduled job expecting more time must either bound itself below this or be restructured as a service.

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 AgentEnabled 

func AgentEnabled(a Agent) bool

AgentEnabled reports whether an agent should be started automatically. Anything that does not carry the flag counts as enabled - the safe direction, since the alternative is a silently dead agent.

func BootTime 

func BootTime() time.Time

BootTime returns the instant the system booted.

Derived from the kernel's uptime rather than read from a file, so it needs no /proc mount - which matters in PID 1 mode, where the schedule may be parsed before the early mount phase has finished.

On failure it falls back to the current time, which degrades OnBootSec to OnStartupSec rather than refusing to schedule. That is announced, not silent.

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).

WantedBy() and RequiredBy() are folded in as the REVERSE edges they have always claimed to be: "X is wanted_by Y" means Y wants X, so the edge belongs on Y. Both were parsed, validated, stored and written into the registry graph while this sort built its inputs from Requires and Wants alone, so declaring wanted_by ordered nothing.

A reverse edge naming something that is not a known agent is ignored, matching how the sort already treats an unknown forward dep. That leaves a target-style anchor - a name with no agent behind it - still inert; making anchors real is a target-model question, not this one.

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
	// WantedBy and RequiredBy are the REVERSE edges: "the named unit
	// wants/requires me", systemd's [Install] direction. They are on the
	// interface because the topological sort needs them - they were
	// parsed, stored and graphed for a long time while the sort could
	// not see them at all.
	WantedBy() []string
	RequiredBy() []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 fires on a cron expression.

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
	// Enabled is the RESOLVED value: absent metadata means enabled.
	Enabled bool
}

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) Enabled 

func (a *GoAgent) Enabled() bool

Enabled reports whether this agent is started automatically. A disabled agent is still discovered and registered, and can still be started explicitly - the systemd model.

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) RequiredBy 

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

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) SetEnabled 

func (a *GoAgent) SetEnabled(v bool)

SetEnabled records whether this agent should be started automatically. Set from discovery metadata; absent metadata means enabled.

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) WantedBy 

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

func (*GoAgent) Wants 

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

type IntervalSchedule 

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

IntervalSchedule fires repeatedly, one interval apart.

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 OnceSchedule 

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

OnceSchedule fires a single time, at a fixed instant, and never again.

It is stateful by necessity: "have I already fired?" cannot be derived from the current time alone, because an elapse point in the past must still produce exactly one fire. systemd behaves the same way - a monotonic timer whose elapse point has passed triggers immediately on activation, once.

func (*OnceSchedule) Next 

func (s *OnceSchedule) Next(time.Time) time.Time

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) Enabled 

func (a *PythonAgent) Enabled() bool

Enabled reports whether this agent is started automatically. A disabled agent is still discovered and registered, and can still be started explicitly - the systemd model.

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) RequiredBy 

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

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) SetEnabled 

func (a *PythonAgent) SetEnabled(v bool)

SetEnabled records whether this agent should be started automatically. Set from discovery metadata; absent metadata means enabled.

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) WantedBy 

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

func (*PythonAgent) Wants 

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

type Schedule 

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

Schedule yields fire times for a timer agent.

Next returns the next instant the agent should fire, given the current time. A ZERO time means the schedule is exhausted and will never fire again; callers must check IsZero rather than treating the result as a duration. That is how one-shot schedules terminate.

func ParseSchedule 

func ParseSchedule(s string) (Schedule, error)

ParseSchedule parses a schedule string, resolving relative forms against the current wall clock and the system boot time.

Supported:

  • OnUnitActiveSec=D - repeating, every D
  • OnStartupSec=D - ONCE, D after the timer starts
  • OnBootSec=D - ONCE, D after the system booted
  • cron: "*/5 * * * *", and the descriptors @hourly/@daily/@weekly/@monthly
  • raw Go duration: "5s", "1m30s" - repeating, treated as an interval

The three systemd prefixes used to be aliases: parseSystemdSchedule stripped whichever one matched and returned an interval in every case, so OnBootSec=1m ran every minute instead of once (GAPI-DIV-036).

func ParseScheduleAt 

func ParseScheduleAt(s string, now, boot time.Time) (Schedule, error)

ParseScheduleAt is ParseSchedule with time injected, so the boot-relative and startup-relative forms are testable without waiting on a real clock or a real boot.

type TimerAgent 

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

TimerAgent executes a Python agent on a schedule

func NewBinaryTimerAgent 

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

NewBinaryTimerAgent schedules an executable agent, run directly.

Timers used to be Python-only: discovery routed TYPE=timer here just for .py paths, and everything else became a GoAgent, which has no scheduling code at all - so a Go timer ran once at discovery and never again while its declared SCHEDULE was silently discarded (GAPI-DIV-037). The two ADKs are meant to have identical semantics.

func NewTimerAgent 

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

NewTimerAgent schedules a Python agent module, executed through the ADK runner.

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) Enabled 

func (a *TimerAgent) Enabled() bool

Enabled reports whether this agent is started automatically. A disabled agent is still discovered and registered, and can still be started explicitly - the systemd model.

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) RequiredBy 

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

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) SetEnabled 

func (a *TimerAgent) SetEnabled(v bool)

SetEnabled records whether this agent should be started automatically. Set from discovery metadata; absent metadata means enabled.

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) WantedBy 

func (ta *TimerAgent) WantedBy() []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.