iop/apps/agent/internal/bootstrap/module.go

545 lines
16 KiB
Go

// Package bootstrap composes application dependencies into a managed host
// with deterministic lifecycle behavior. Construction validates all inputs
// and rejects invalid configurations before any side effect. Run starts
// components in declared order; Close stops them in reverse order.
//
// The module is side-effect free until Run is called. Every error from
// construction or lifecycle is preserved so callers can inspect exactly
// what failed and why.
package bootstrap
import (
"context"
"errors"
"fmt"
"path/filepath"
"reflect"
"strings"
"time"
"iop/apps/agent/internal/clientprocess"
"iop/apps/agent/internal/host"
"iop/apps/agent/internal/localcontrol"
"iop/apps/agent/internal/projectlog"
"iop/apps/agent/internal/taskloop"
"iop/packages/go/agentconfig"
"iop/packages/go/agentstate"
"iop/packages/go/agenttask"
"iop/packages/go/agentworkspace"
)
// Module owns the validated, composed application lifecycle. It wraps a
// host.Host and delegates Start/Stop to it after validating that every
// dependency is well-formed. Construction never starts or stops anything.
type Module struct {
host *host.Host
}
// NewModule validates the provided components and builds a Module. It
// rejects nil components and components whose resolved names collide.
// No side effects occur at construction time.
func NewModule(components ...host.Component) (*Module, error) {
if len(components) == 0 {
return nil, fmt.Errorf("bootstrap: at least one component is required")
}
seen := make(map[string]struct{}, len(components))
validated := make([]host.Component, 0, len(components))
for i, component := range components {
if isNilComponent(component) {
return nil, fmt.Errorf("bootstrap: component at index %d is nil", i)
}
name := resolveComponentName(component, i)
if _, duplicate := seen[name]; duplicate {
return nil, fmt.Errorf("%w %q", ErrDuplicateName, name)
}
seen[name] = struct{}{}
validated = append(validated, resolvedComponent{
Component: component,
name: name,
})
}
h := host.NewHost(validated...)
return &Module{host: h}, nil
}
// resolveComponentName returns the component's Name if it implements
// host.Namer, otherwise a synthetic fallback based on its position.
func resolveComponentName(c host.Component, idx int) string {
if namer, ok := c.(host.Namer); ok {
name := namer.Name()
if name != "" {
return name
}
}
return fmt.Sprintf("component-%s", strings.Repeat("x", idx+1))
}
// Run starts every managed component in declared order. On the first
// failure it rolls back already-started components in reverse and returns
// the combined error. The host must not have been previously run.
func (m *Module) Run(ctx context.Context) error {
return m.host.Start(ctx)
}
// Close stops every started component in reverse order. It preserves
// every individual error. Repeated calls return nil once the host has
// fully stopped.
func (m *Module) Close(ctx context.Context) error {
return m.host.Stop(ctx)
}
// Status returns a snapshot of the host's current lifecycle state.
func (m *Module) Status() host.Status {
return m.host.Status()
}
// resolvedComponent wraps a host.Component with a validated, non-empty
// name that is preserved through host construction. It implements
// host.Component and host.Namer so the host consumes the exact validated
// identity without re-resolving.
type resolvedComponent struct {
host.Component
name string
}
// Name returns the captured resolved name.
func (r resolvedComponent) Name() string { return r.name }
// isNilComponent reports whether c is a nil interface value or a typed-nil
// pointer/interface/map/slice/chan/func. It never calls a method on c.
func isNilComponent(c host.Component) bool {
if c == nil {
return true
}
v := reflect.ValueOf(c)
switch v.Kind() {
case reflect.Ptr, reflect.Interface, reflect.Map, reflect.Slice, reflect.Chan, reflect.Func:
return v.IsNil()
}
return false
}
// ErrDuplicateName is returned by NewModule when two components resolve
// to the same name.
var ErrDuplicateName = errors.New("bootstrap: duplicate component name")
// DaemonOptions configures optional dependencies for production composition.
type DaemonOptions struct {
ProcessBackend clientprocess.ProcessBackend
TaskCatalog agentconfig.Catalog
TaskProvider agenttask.ProviderInvoker
TaskReviewExecutor taskloop.ReviewExecutor
TaskValidator agentworkspace.ValidationFunc
TaskQuotaObserver taskloop.QuotaObserver
}
// DaemonOption modifies DaemonOptions.
type DaemonOption func(*DaemonOptions)
// WithProcessBackend supplies a custom ProcessBackend for client process management.
func WithProcessBackend(backend clientprocess.ProcessBackend) DaemonOption {
return func(o *DaemonOptions) {
if backend != nil {
o.ProcessBackend = backend
}
}
}
// WithTaskCatalog supplies the immutable provider catalog used by the
// authoritative task runtime. Production callers must pass the same catalog
// that was validated by the CLI entry point.
func WithTaskCatalog(catalog agentconfig.Catalog) DaemonOption {
return func(o *DaemonOptions) {
o.TaskCatalog = catalog
}
}
// WithTaskRuntimePorts supplies test-owned runtime ports without changing the
// production catalog-backed defaults. The bootstrap package is internal to
// apps/agent, so this seam cannot become an external daemon API.
func WithTaskRuntimePorts(
provider agenttask.ProviderInvoker,
reviewExecutor taskloop.ReviewExecutor,
validator agentworkspace.ValidationFunc,
quotaObserver taskloop.QuotaObserver,
) DaemonOption {
return func(o *DaemonOptions) {
o.TaskProvider = provider
o.TaskReviewExecutor = reviewExecutor
o.TaskValidator = validator
o.TaskQuotaObserver = quotaObserver
}
}
// DaemonModule wraps Module and exposes component references for testing and inspection.
type DaemonModule struct {
*Module
StateStore *agentstate.Store
TaskRuntime *taskloop.Runtime
ClientManager *clientprocess.Manager
ProjectStores map[string]*projectlog.Store
ControlServer *localcontrol.Server
Ledger *localcontrol.Ledger
}
// NewDaemonModule constructs one authoritative task runtime together with the
// project-log, client-process, and local-control adapters over one state store
// and immutable config snapshot.
func NewDaemonModule(
ctx context.Context,
snapshot agentconfig.RuntimeSnapshot,
opts ...DaemonOption,
) (*DaemonModule, error) {
cfg := snapshot.Config()
if cfg.Device.StateRoot == "" {
return nil, fmt.Errorf("bootstrap: device state_root is required")
}
dOpts := DaemonOptions{}
for _, opt := range opts {
opt(&dOpts)
}
statePath := filepath.Join(cfg.Device.StateRoot, "state.json")
stateStore, err := agentstate.NewStore(statePath)
if err != nil {
return nil, fmt.Errorf("bootstrap: create state store: %w", err)
}
taskCatalog := dOpts.TaskCatalog
if taskCatalog.Version == "" {
return nil, errors.New("bootstrap: task provider catalog is required")
}
taskValidator := dOpts.TaskValidator
if taskValidator == nil {
taskValidator = taskloop.DefaultValidator()
}
taskRuntime, err := taskloop.New(taskloop.Config{
Snapshot: snapshot,
Catalog: taskCatalog,
StateStore: stateStore,
OwnerID: "iop-agent-daemon",
Provider: dOpts.TaskProvider,
ReviewExecutor: dOpts.TaskReviewExecutor,
Validator: taskValidator,
QuotaObserver: dOpts.TaskQuotaObserver,
})
if err != nil {
return nil, fmt.Errorf("bootstrap: create task runtime: %w", err)
}
mgrOpts := []clientprocess.Option{}
if dOpts.ProcessBackend != nil {
mgrOpts = append(mgrOpts, clientprocess.WithProcessBackend(dOpts.ProcessBackend))
}
manager, err := clientprocess.NewManager(ctx, cfg.Clients, stateStore, mgrOpts...)
if err != nil {
return nil, fmt.Errorf("bootstrap: create client process manager: %w", err)
}
projectStores := taskRuntime.ProjectStores()
ledger, err := localcontrol.NewLedger(ctx, stateStore, "iop-agent-daemon", 0)
if err != nil {
_ = manager.Close(ctx)
return nil, fmt.Errorf("bootstrap: create local control ledger: %w", err)
}
clientOps, err := localcontrol.NewClientOperations(manager, ledger)
if err != nil {
_ = manager.Close(ctx)
return nil, fmt.Errorf("bootstrap: create client operations: %w", err)
}
bridge := &daemonStateBridge{
snapshot: snapshot,
manager: manager,
task: taskRuntime,
}
service, err := localcontrol.NewService(bridge, bridge, ledger)
if err != nil {
_ = manager.Close(ctx)
return nil, fmt.Errorf("bootstrap: create local control service: %w", err)
}
service.WithClientOperations(clientOps)
server, err := localcontrol.NewServer(localcontrol.ServerConfig{
StateRoot: cfg.Device.StateRoot,
}, service)
if err != nil {
_ = manager.Close(ctx)
return nil, fmt.Errorf("bootstrap: create local control server: %w", err)
}
pLogComp := &projectLogAdapter{stores: projectStores}
taskComp, err := taskloop.NewComponent(taskRuntime, 250*time.Millisecond)
if err != nil {
_ = manager.Close(ctx)
return nil, fmt.Errorf("bootstrap: create task runtime component: %w", err)
}
cProcComp := &clientProcessAdapter{manager: manager}
lCtrlComp := &localControlAdapter{server: server}
m, err := NewModule(pLogComp, taskComp, cProcComp, lCtrlComp)
if err != nil {
_ = manager.Close(ctx)
return nil, err
}
return &DaemonModule{
Module: m,
StateStore: stateStore,
TaskRuntime: taskRuntime,
ClientManager: manager,
ProjectStores: projectStores,
ControlServer: server,
Ledger: ledger,
}, nil
}
type projectLogAdapter struct {
stores map[string]*projectlog.Store
}
func (p *projectLogAdapter) Name() string { return "project-log" }
func (p *projectLogAdapter) Start(ctx context.Context) error {
for _, store := range p.stores {
if err := store.Reconcile(ctx); err != nil {
return fmt.Errorf("projectlog reconcile: %w", err)
}
}
return nil
}
func (p *projectLogAdapter) Stop(ctx context.Context) error {
return nil
}
type clientProcessAdapter struct {
manager *clientprocess.Manager
}
func (c *clientProcessAdapter) Name() string { return "client-process" }
func (c *clientProcessAdapter) Start(ctx context.Context) error {
if err := c.manager.StartConfigured(ctx); err != nil {
return errors.Join(err, c.manager.Close(context.WithoutCancel(ctx)))
}
return nil
}
func (c *clientProcessAdapter) Stop(ctx context.Context) error {
return c.manager.Close(ctx)
}
type localControlAdapter struct {
server *localcontrol.Server
}
func (l *localControlAdapter) Name() string { return "local-control" }
func (l *localControlAdapter) Start(ctx context.Context) error {
return l.server.Start(ctx)
}
func (l *localControlAdapter) Stop(ctx context.Context) error {
return l.server.Stop()
}
type daemonStateBridge struct {
snapshot agentconfig.RuntimeSnapshot
manager *clientprocess.Manager
task *taskloop.Runtime
}
func (b *daemonStateBridge) RuntimeStatus(ctx context.Context) (localcontrol.StatusSnapshot, error) {
return localcontrol.StatusSnapshot{
SubjectID: "daemon",
State: "running",
Summary: "iop-agent daemon active",
}, nil
}
func (b *daemonStateBridge) ProjectStatus(ctx context.Context, projectID string) (localcontrol.StatusSnapshot, error) {
view, err := b.task.ProjectStatus(ctx, projectID)
if err != nil {
return localcontrol.StatusSnapshot{}, err
}
return projectSnapshot(view), nil
}
func (b *daemonStateBridge) OverlayStatus(ctx context.Context, projectID, workUnitID string) (localcontrol.StatusSnapshot, error) {
view, work, err := b.workStatus(ctx, projectID, workUnitID)
if err != nil {
return localcontrol.StatusSnapshot{}, err
}
return localcontrol.StatusSnapshot{
SubjectID: workUnitID,
State: string(work.State),
Summary: work.Overlay,
Entries: []localcontrol.StatusEntry{{
Kind: "project", SubjectID: projectID, State: string(view.Status),
}},
}, nil
}
func (b *daemonStateBridge) IntegrationStatus(ctx context.Context, projectID, workUnitID string) (localcontrol.StatusSnapshot, error) {
view, work, err := b.workStatus(ctx, projectID, workUnitID)
if err != nil {
return localcontrol.StatusSnapshot{}, err
}
return localcontrol.StatusSnapshot{
SubjectID: workUnitID,
State: string(work.State),
Summary: work.Integration,
Entries: []localcontrol.StatusEntry{{
Kind: "project", SubjectID: projectID, State: string(view.Status),
}},
}, nil
}
func (b *daemonStateBridge) BlockerList(ctx context.Context, projectID string) (localcontrol.StatusSnapshot, error) {
view, err := b.task.ProjectStatus(ctx, projectID)
if err != nil {
return localcontrol.StatusSnapshot{}, err
}
snapshot := localcontrol.StatusSnapshot{
SubjectID: projectID,
State: string(view.Status),
Summary: fmt.Sprintf("%d blocker(s)", len(view.Blockers)),
}
for _, blocker := range view.Blockers {
snapshot.Entries = append(snapshot.Entries, localcontrol.StatusEntry{
Kind: "blocker", SubjectID: string(blocker.Code),
State: "blocked", Summary: blocker.Message,
})
}
return snapshot, nil
}
func (b *daemonStateBridge) ProcessStatus(ctx context.Context, clientKind string) (localcontrol.StatusSnapshot, error) {
if b.manager == nil {
return localcontrol.StatusSnapshot{
SubjectID: clientKind,
State: "unknown",
Summary: "process manager unavailable",
}, nil
}
rec, err := b.manager.Status(clientprocess.ClientKind(clientKind))
if err != nil {
return localcontrol.StatusSnapshot{}, err
}
return localcontrol.StatusSnapshot{
SubjectID: clientKind,
State: string(rec.State),
Summary: rec.LastResult,
}, nil
}
func (b *daemonStateBridge) StartProject(ctx context.Context, cmd localcontrol.ProjectCommand) (localcontrol.MutationResult, error) {
if err := b.prepareProjectCommand(ctx, cmd); err != nil {
return localcontrol.MutationResult{}, err
}
view, err := b.task.StartProject(ctx, cmd.ProjectID)
return mutationResult(view), err
}
func (b *daemonStateBridge) StopProject(ctx context.Context, cmd localcontrol.ProjectCommand) (localcontrol.MutationResult, error) {
if err := b.verifyProjectIdentity(ctx, cmd); err != nil {
return localcontrol.MutationResult{}, err
}
view, err := b.task.StopProject(ctx, cmd.ProjectID)
return mutationResult(view), err
}
func (b *daemonStateBridge) ResumeProject(ctx context.Context, cmd localcontrol.ProjectCommand) (localcontrol.MutationResult, error) {
if err := b.prepareProjectCommand(ctx, cmd); err != nil {
return localcontrol.MutationResult{}, err
}
view, err := b.task.ResumeProject(ctx, cmd.ProjectID)
return mutationResult(view), err
}
func (b *daemonStateBridge) prepareProjectCommand(
ctx context.Context,
cmd localcontrol.ProjectCommand,
) error {
if err := b.verifyProjectIdentity(ctx, cmd); err != nil {
return err
}
if cmd.MilestoneID != "" {
return b.task.SelectMilestone(ctx, cmd.ProjectID, cmd.MilestoneID)
}
return nil
}
func (b *daemonStateBridge) verifyProjectIdentity(
ctx context.Context,
cmd localcontrol.ProjectCommand,
) error {
if b.task == nil {
return errors.New("bootstrap: authoritative task runtime is missing")
}
registration, ok := b.snapshot.Project(cmd.ProjectID)
if !ok || !registration.Enabled {
return fmt.Errorf("project %s is not registered", cmd.ProjectID)
}
if cmd.WorkspaceID == "" {
return nil
}
snapshot, err := b.task.WorkflowSnapshot(ctx, cmd.ProjectID)
if err != nil {
return err
}
if cmd.WorkspaceID != string(snapshot.WorkspaceID) {
return errors.New("bootstrap: project command workspace identity mismatch")
}
return nil
}
func (b *daemonStateBridge) workStatus(
ctx context.Context,
projectID, workUnitID string,
) (taskloop.ProjectView, taskloop.WorkView, error) {
view, err := b.task.ProjectStatus(ctx, projectID)
if err != nil {
return taskloop.ProjectView{}, taskloop.WorkView{}, err
}
for _, work := range view.Works {
if work.WorkUnitID == workUnitID {
return view, work, nil
}
}
return taskloop.ProjectView{}, taskloop.WorkView{}, fmt.Errorf(
"project %s has no work unit %s",
projectID,
workUnitID,
)
}
func projectSnapshot(view taskloop.ProjectView) localcontrol.StatusSnapshot {
snapshot := localcontrol.StatusSnapshot{
SubjectID: view.ProjectID,
State: string(view.Status),
Summary: "authoritative task runtime state",
}
for _, work := range view.Works {
snapshot.Entries = append(snapshot.Entries, localcontrol.StatusEntry{
Kind: "work", SubjectID: work.WorkUnitID,
State: string(work.State), Summary: work.Integration,
})
}
return snapshot
}
func mutationResult(view taskloop.ProjectView) localcontrol.MutationResult {
return localcontrol.MutationResult{
SubjectID: view.ProjectID,
State: string(view.Status),
Summary: "authoritative task runtime mutation committed",
}
}