Documentation
¶
Overview ¶
Package jobs provides the Supervisor: the single, long-lived owner of every background-work monitor goroutine in an agent session. It is the unified replacement for the per-request A2ATaskPoller and SubagentPoller and the per-shell monitor that used to live in BackgroundShellService. Submit spawns one monitor goroutine per job at creation (no discovery ticker), runs the shared finish-once logic, and tracks live and recently-finished jobs so the task view and status line can report them. It lives in a services subpackage (not top-level services) so the tools package can submit jobs to it without an import cycle - it depends only on domain.
Index ¶
- func NewShellJob(shell *domain.BackgroundShell, tracker domain.ShellTracker) domain.BackgroundJob
- type Supervisor
- func (s *Supervisor) A2APollingStates() []domain.TaskPollingState
- func (s *Supervisor) Cleanup(olderThan time.Duration) int
- func (s *Supervisor) CountRunning(kind domain.JobKind) int
- func (s *Supervisor) DiscardKind(kind domain.JobKind)
- func (s *Supervisor) HasPending() bool
- func (s *Supervisor) IsRunning(id string) bool
- func (s *Supervisor) SetConversationRepo(repo domain.ConversationRepository)
- func (s *Supervisor) SetRetentionCount(kind domain.JobKind, max int)
- func (s *Supervisor) SetTaskRetention(svc domain.TaskRetentionService)
- func (s *Supervisor) Snapshot() []domain.TrackedJob
- func (s *Supervisor) Start(interval, retention time.Duration)
- func (s *Supervisor) Stop()
- func (s *Supervisor) Submit(job domain.BackgroundJob)
- func (s *Supervisor) Wind(id string, sig domain.WindSignal) error
- func (s *Supervisor) WindAll(sig domain.WindSignal)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NewShellJob ¶
func NewShellJob(shell *domain.BackgroundShell, tracker domain.ShellTracker) domain.BackgroundJob
NewShellJob wraps a running background shell as a supervised job. The shell must already be registered in tracker (so the bash tools can read its output); Close removes it on reap.
Types ¶
type Supervisor ¶
type Supervisor struct {
// contains filtered or unexported fields
}
Supervisor owns and monitors all background jobs for a session.
func NewSupervisor ¶
func NewSupervisor(messageQueue domain.MessageQueue, conversationRepo domain.ConversationRepository, notifier domain.UINotifier) *Supervisor
NewSupervisor constructs a supervisor. messageQueue and conversationRepo are long-lived singletons used to deliver finished jobs' results back onto the conversation; either may be nil (the supervisor degrades to event-only). notifier is the single UI ingress used to push DrainQueueEvent (so landed work drains promptly) and BackgroundTasksChangedEvent (so the task view refreshes); a nil notifier degrades to no UI pushes.
func (*Supervisor) A2APollingStates ¶ added in v0.128.0
func (s *Supervisor) A2APollingStates() []domain.TaskPollingState
A2APollingStates returns the live polling state of every running A2A job, so the task view and HasPending source A2A liveness from the supervisor (the single source of truth) instead of the parallel A2ATaskTracker polling set. Only running jobs are returned; terminal A2A tasks live in the retention view.
func (*Supervisor) Cleanup ¶
func (s *Supervisor) Cleanup(olderThan time.Duration) int
Cleanup reaps finished jobs whose terminal timestamp is older than olderThan, running each one's Wind(WindStop) teardown (kill pane, remove temp files) before dropping it. Running jobs are never reaped. The same sweep emits a one-time "job running long" warning for any still-running job that has exceeded JobRunningLongThreshold. Returns the number removed.
func (*Supervisor) CountRunning ¶
func (s *Supervisor) CountRunning(kind domain.JobKind) int
CountRunning returns the number of jobs still running, optionally filtered to one kind. Pass "" for all kinds.
func (*Supervisor) DiscardKind ¶ added in v0.128.0
func (s *Supervisor) DiscardKind(kind domain.JobKind)
DiscardKind stops and forgets every job of one kind: running jobs are dropped from the map immediately (the status bar and task view clear synchronously), hard-stopped, and finish without a queue note or retention entry; terminal unreaped jobs are reaped on the spot. This is the conversation-clear semantic - discarding the cleared conversation's A2A tasks without touching other kinds. Unlike a user cancel it does not notify the remote agent; it only stops and forgets the local monitor.
func (*Supervisor) HasPending ¶ added in v0.128.0
func (s *Supervisor) HasPending() bool
HasPending reports whether any session-holding job is still running - the cross-kind "is the session safe to close?" query. A job opts in via JobMeta.HoldsSession (A2A tasks, shells, headless subagents); interactive subagent panes set it false so a one-shot `infer headless` does not wait on a user-driven pane.
func (*Supervisor) IsRunning ¶ added in v0.128.0
func (s *Supervisor) IsRunning(id string) bool
IsRunning reports whether a supervised job with the given id is still running. It is the per-id liveness query - the single source of truth for "is this background job still in flight?", consistent with Snapshot, CountRunning, and A2APollingStates, which read the same jobs map under the same lock.
func (*Supervisor) SetConversationRepo ¶
func (s *Supervisor) SetConversationRepo(repo domain.ConversationRepository)
SetConversationRepo wires the conversation repository used to format finished jobs' results. It is set after construction because the repo is built later in the container than the supervisor.
func (*Supervisor) SetRetentionCount ¶
func (s *Supervisor) SetRetentionCount(kind domain.JobKind, max int)
SetRetentionCount caps how many terminal jobs of a kind are kept for the task view. When a job finishes, the oldest terminal jobs of its kind beyond max are reaped immediately (their Close teardown runs). max <= 0 means unbounded - terminal jobs are kept until the time-based Cleanup sweep. Set once per kind at startup, before jobs are submitted.
func (*Supervisor) SetTaskRetention ¶
func (s *Supervisor) SetTaskRetention(svc domain.TaskRetentionService)
SetTaskRetention wires the A2A task-retention service that finished jobs implementing domain.TaskRetainer populate for the task view. Like the conversation repo, it is set after construction because the retention service is built later in the container than the supervisor. A nil service disables retention (finish just skips it).
func (*Supervisor) Snapshot ¶
func (s *Supervisor) Snapshot() []domain.TrackedJob
Snapshot returns a copy of all tracked jobs (running and recently finished) for the task view. For jobs that implement JobOutputProvider, the Output field is populated from the job's output (shell stdout/stderr or subagent result).
func (*Supervisor) Start ¶
func (s *Supervisor) Start(interval, retention time.Duration)
Start launches the periodic cleanup that reaps finished jobs older than retention. interval is how often to sweep. Call once; Stop ends it.
func (*Supervisor) Stop ¶
func (s *Supervisor) Stop()
Stop cancels every running job and waits for all monitor goroutines to exit. Idempotent.
func (*Supervisor) Submit ¶
func (s *Supervisor) Submit(job domain.BackgroundJob)
Submit registers a job and spawns its monitor goroutine. Duplicate IDs and submissions after Stop are ignored.
func (*Supervisor) Wind ¶
func (s *Supervisor) Wind(id string, sig domain.WindSignal) error
Wind delivers a graceful wind-down or hard stop to a single running job by id.
func (*Supervisor) WindAll ¶
func (s *Supervisor) WindAll(sig domain.WindSignal)
WindAll broadcasts a signal to every running job. Graceful shutdown calls WindAll(WindWrapUp), waits a grace window, then WindAll(WindStop).
Source Files