scheduler

package
v0.226.0 Latest Latest
Warning

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

Go to latest
Published: Oct 6, 2026 License: Apache-2.0 Imports: 20 Imported by: 0

README

scheduler

What - the scheduler bounded context: cron-driven jobs, the heartbeat, and the registry of background work (shells, subagents, supervisor jobs). Why - scheduled work runs inside infer daemon, outside any chat session, so it needs its own lifecycle and storage instead of living in the agent loop. How - the Schedule tool writes jobs to storage, and the daemon's scheduler polls that storage and runs each due job as a one-shot infer headless inside infer daemon, through internal/agent/headless. Heartbeat wake-ups run the same way. Each is one process per run, no client follows it, and its stderr lands in the daemon log.

How it plugs in

  • Jobs persist through the configured storage (~/.infer/schedules/*.yaml with jsonl), or through GitHub Actions with scheduler.backend: github (githubscheduler/).
  • The scheduler polls every few seconds, so new jobs fire without a restart. schedule_notifier.go delivers results back through the originating channel.
  • heartbeat/ wakes the agent periodically to check pending work. jobs/ supervises long-running background jobs.
  • Title backfill runs under infer conversation-title daemon.
  • Channels do not use agentheadless: they drive the daemon's session workers through the sessions registry.

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func NewBackgroundTaskRegistry

func NewBackgroundTaskRegistry(maxConcurrentShells int, supervisor *jobs.Supervisor) scheddomain.BackgroundTaskRegistry

NewBackgroundTaskRegistry constructs the unified registry. maxConcurrentShells is the per-session cap enforced by the underlying shell tracker. supervisor is the single fan-in that monitors submitted jobs and backs the unified job surface (Submit/Snapshot/Wind).

Types

type BackgroundShellService

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

BackgroundShellService is the thin front-end for background bash shells: it detaches a running command (registering it so the bash tools can read its output), submits a shellJob to the supervisor (which monitors it, notifies the agent, and reaps it), and brokers output retrieval and cancellation. The completion monitoring and reaping that used to live here are now the supervisor's job.

func NewBackgroundShellService

func NewBackgroundShellService(
	tracker scheddomain.ShellTracker,
	supervisor *jobs.Supervisor,
	cfg *config.Config,
	eventChannel chan<- agentdomain.ChatEvent,
) *BackgroundShellService

NewBackgroundShellService creates a new background shell service. supervisor monitors and reaps each detached shell.

func (*BackgroundShellService) CancelShell

func (s *BackgroundShellService) CancelShell(shellID string) error

CancelShell cancels a running background shell. It hands the kill to the supervisor (WindStop -> SIGKILL); the shellJob's Run then records the Cancelled state and the supervisor delivers the completion notification, so the process is waited on in exactly one place.

func (*BackgroundShellService) DetachToBackground

func (s *BackgroundShellService) DetachToBackground(
	ctx context.Context,
	cmd *exec.Cmd,
	command string,
	outputBuffer scheddomain.OutputRingBuffer,
	readersDone <-chan struct{},
) (string, error)

DetachToBackground moves a running command to background

func (*BackgroundShellService) GetAllShells

func (s *BackgroundShellService) GetAllShells() []*scheddomain.BackgroundShell

GetAllShells returns all tracked shells

func (*BackgroundShellService) GetShell

GetShell returns a specific shell by ID

func (*BackgroundShellService) GetShellOutput

func (s *BackgroundShellService) GetShellOutput(shellID string, fromOffset int64) (string, int64, scheddomain.ShellState, error)

GetShellOutput retrieves incremental output from a shell

func (*BackgroundShellService) GetShellOutputWithFilter

func (s *BackgroundShellService) GetShellOutputWithFilter(shellID string, fromOffset int64, filterPattern string) (string, int64, scheddomain.ShellState, error)

GetShellOutputWithFilter retrieves output with optional regex filtering

func (*BackgroundShellService) GetStats

func (s *BackgroundShellService) GetStats() map[string]int

GetStats returns statistics about background shells

func (*BackgroundShellService) RemoveShell

func (s *BackgroundShellService) RemoveShell(shellID string) error

RemoveShell removes a shell from tracking

func (*BackgroundShellService) Stop

func (s *BackgroundShellService) Stop()

Stop is retained for the lifecycle contract. The supervisor owns the monitor goroutines and the cleanup ticker now, so there is nothing to stop here.

type Options

type Options struct {
	Store       storage.ScheduledJobStorage
	Runs        storage.ScheduledRunStorage
	OnRunEvent  func(scheddomain.ScheduledJob, scheddomain.RunEvent)
	ExecCommand agentheadless.ExecFunc
	BinaryPath  string
}

Options bundles dependencies and configuration for NewService.

type ScheduleNotifier

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

ScheduleNotifier delivers scheduled-job run output to the job's delivery channel. It subscribes to the scheduler's run events (Options.OnRunEvent); jobs without a delivery target are record-only, so their events are ignored here and their output lives in storage.

func NewScheduleNotifier

func NewScheduleNotifier(lookup func(name string) channels.Channel) *ScheduleNotifier

NewScheduleNotifier constructs a notifier resolving channels through lookup (typically ChannelManagerService.GetChannel).

func (*ScheduleNotifier) Notify

Notify implements the scheduler's OnRunEvent hook.

type Service

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

Service runs scheduled jobs inside the `infer daemon` process. Jobs are loaded from the configured ScheduledJobStorage, registered with a robfig/cron scheduler, and hot-reloaded by polling the storage and diffing (reconcile).

On fire, a fresh `infer headless --session-id <uuid>` subprocess is spawned - every fire gets a brand-new session, so no context carries between runs. A RunRecord is persisted per fire (keyed by that session ID, so the run's conversation is discoverable from storage), and progress is emitted through the optional OnRunEvent hook. The scheduler knows nothing about channels - delivery is a subscriber concern (see services.ScheduleNotifier).

func NewService

func NewService(opts Options) (*Service, error)

NewService constructs a Service. Returns an error if required deps are missing.

func (*Service) JobIDs

func (s *Service) JobIDs() []string

JobIDs returns the set of currently-registered job IDs (test helper).

func (*Service) Start

func (s *Service) Start(ctx context.Context) error

Start initialises the cron scheduler, loads all jobs from storage, and begins polling storage for changes.

func (*Service) Stop

func (s *Service) Stop(ctx context.Context) error

Stop halts the watcher and waits for in-flight cron entries to finish (up to the deadline embedded in ctx, if any).

type TitleBackfill added in v0.207.0

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

TitleBackfill manages background tasks

func NewTitleBackfill added in v0.207.0

func NewTitleBackfill(titleGenerator scheddomain.TitleGenerator, config *config.Config) *TitleBackfill

NewTitleBackfill creates a new background job manager

func (*TitleBackfill) IsRunning added in v0.207.0

func (m *TitleBackfill) IsRunning() bool

IsRunning returns whether the job manager is currently running

func (*TitleBackfill) Start added in v0.207.0

func (m *TitleBackfill) Start(ctx context.Context)

Start begins running background jobs

func (*TitleBackfill) Stop added in v0.207.0

func (m *TitleBackfill) Stop()

Stop stops all background jobs gracefully

func (*TitleBackfill) TriggerTitleGeneration added in v0.207.0

func (m *TitleBackfill) TriggerTitleGeneration(ctx context.Context) error

TriggerTitleGeneration manually triggers title generation for pending conversations

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL