worker

package
v0.50.26 Latest Latest
Warning

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

 Go to latest Published: May 31, 2026 License: MIT Imports: 39 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func ComputerUseSupported 

func ComputerUseSupported(providerName string) bool

ComputerUseSupported returns whether the named provider supports computer use.

func IsContextOverflow 

func IsContextOverflow(evt adapter.StreamEvent) bool

IsContextOverflow checks whether a stream event indicates a context window overflow. Returns true if the event is an error containing "context window" or "token limit".

func ParsePhaseInstructions 

func ParsePhaseInstructions(instructions map[string]string) (map[Phase]string, error)

ParsePhaseInstructions validates phase instruction overrides from the server.

func ParsePhasePromptTemplates 

func ParsePhasePromptTemplates(templates map[string]string) (map[Phase]string, error)

ParsePhasePromptTemplates validates server-provided full prompt templates.

func RouteWorkDir 

func RouteWorkDir(mw *workspace.MultiWorkspace, workspaceID, defaultDir string) string

RouteWorkDir returns the project directory for a workspace-targeted task. Falls back to defaultDir when the multi-workspace manager is nil or not found.

func StartWorkspaceGoroutines 

func StartWorkspaceGoroutines(ctx context.Context, workspaces []WorkspaceConfig, handler a2a.TaskHandler) *workspace.MultiWorkspace

StartWorkspaceGoroutines spawns per-workspace A2A server goroutines. Returns the MultiWorkspace manager for task routing. Called by the CLI when multi-workspace mode is activated.

Types

type AuditEvent 

type AuditEvent struct {
	TaskID          string   `json:"task_id"`
	Event           string   `json:"event"` // "started", "completed", "failed", "degraded", "reclaimed"
	Timestamp       string   `json:"timestamp"`
	DurationMS      int64    `json:"duration_ms,omitempty"`
	CostUSD         float64  `json:"cost_usd,omitempty"`
	ComputerUse     bool     `json:"computer_use,omitempty"`
	ReasonCode      string   `json:"reason_code,omitempty"`
	OverrideReason  string   `json:"override_reason,omitempty"`
	OverrideStatus  string   `json:"override_status,omitempty"`
	RootFallback    bool     `json:"root_worktree_fallback,omitempty"`
	WorktreePath    string   `json:"worktree_path,omitempty"`
	ReclaimState    string   `json:"reclaim_state,omitempty"`
	ActionSequence  []string `json:"action_sequence,omitempty"`
	InterruptReason string   `json:"interrupt_reason,omitempty"`
	SIGTERMSent     bool     `json:"sigterm_sent,omitempty"`
	SIGKILLSent     bool     `json:"sigkill_sent,omitempty"`
}

@AX:ANCHOR: [AUTO] worker audit JSONL schema for task lifecycle, degraded worktree fallback, reclaim, and interrupts. @AX:REASON: CLI diagnostics and safety evidence tests rely on event names, reason codes, and signal/action fields staying stable. AuditEvent represents a structured audit log entry for task execution.

type BudgetConfig 

type BudgetConfig struct {
	Budget        budget.IterationBudget
	EmergencyStop *security.EmergencyStop
}

BudgetConfig holds optional budget configuration for subprocess execution.

type ContextBuilder 

type ContextBuilder struct{}

ContextBuilder assembles the Layer 4 prompt for subprocess execution.

func (*ContextBuilder) Build 

func (b *ContextBuilder) Build(payload TaskPayload) string

Build assembles the complete prompt string for stdin injection. Only non-empty sections are included in the output.

type HostArtifact 

type HostArtifact struct {
	Name     string
	MimeType string
	Preview  string
	Source   string
}

HostArtifact is a desktop-safe projection of a worker artifact.

type HostEvent 

type HostEvent struct {
	Type          HostEventType
	TaskID        string
	ApprovalID    string
	TraceID       string
	CorrelationID string
	Phase         string
	Message       string
	Action        string
	RiskLevel     string
	Context       string
	CostUSD       float64
	DurationMS    int64
	Execution     *HostExecutionContext
	Result        *HostResult
}

HostEvent carries host-neutral task, progress, and approval signals.

type HostEventType 

type HostEventType string

HostEventType identifies a machine-consumable worker host event. 

const (
	HostEventRuntimeDegraded   HostEventType = "runtime_degraded"
	HostEventTaskReceived      HostEventType = "task_received"
	HostEventTaskProgress      HostEventType = "task_progress"
	HostEventTaskCompleted     HostEventType = "task_completed"
	HostEventTaskFailed        HostEventType = "task_failed"
	HostEventApprovalRequested HostEventType = "approval_requested"
	HostEventApprovalResolved  HostEventType = "approval_resolved"
)

type HostExecutionContext 

type HostExecutionContext struct {
	WorkspaceID   string
	RootWorkDir   string
	ActiveWorkDir string
	WorktreePath  string
	Mode          string
	BoundaryHint  string
}

HostExecutionContext describes the retained worker filesystem boundary.

type HostObserver 

type HostObserver interface {
	OnHostEvent(HostEvent)
}

HostObserver receives host-neutral worker events.

type HostObserverFunc 

type HostObserverFunc func(HostEvent)

HostObserverFunc adapts a function into a HostObserver.

func (HostObserverFunc) OnHostEvent 

func (fn HostObserverFunc) OnHostEvent(event HostEvent)

OnHostEvent implements HostObserver.

type HostResult 

type HostResult struct {
	Status       string
	Summary      string
	ErrorMessage string
	CostLabel    string
	DurationMS   int64
	SessionID    string
	Artifacts    []HostArtifact
}

HostResult summarizes the retained terminal outcome for desktop.

type LogBuffer 

type LogBuffer interface {
	RecordWarning(msg string)
	RecordError(msg string)
}

LogBuffer captures structured log entries for audit write error tracking. Implementations record warnings and errors for consecutive failure escalation.

type LoopConfig 

type LoopConfig struct {
	BackendURL    string
	WorkerName    string
	MemoryAgentID string
	Skills        []string
	Providers     []string
	Provider      adapter.ProviderAdapter
	MCPConfig     string          // path to worker-mcp.json
	WorkDir       string          // working directory for subprocesses
	AuthToken     string          // bearer token for backend auth
	Router        *routing.Router // optional model router (nil = no routing)
	// Deprecated: use CredentialStore instead. Kept for backward compatibility.
	CredentialsPath   string                // path to credentials.json for token refresh
	CredentialStore   setup.CredentialStore // Secure credential storage (Keychain/encrypted file). If nil and CredentialsPath is set, falls back to plain file mode.
	AuditLogPath      string                // audit log file path (default: {WorkDir}/.autopus/audit.jsonl)
	AuditMaxSize      int64                 // max log size before rotation (default: 10MB)
	AuditMaxAge       time.Duration         // max age of rotated files (default: 7 days)
	WorkspaceID       string                // workspace identifier for scheduler
	MaxConcurrency    int                   // max parallel tasks (0 = default slot cap, 1 = sequential)
	WorktreeIsolation bool                  // enable worktree isolation for parallel tasks
	// WorktreeFallbackOverrideReason permits explicit root-worktree fallback when isolation is unavailable.
	WorktreeFallbackOverrideReason string
	KnowledgeSync                  bool   // enable local knowledge context loading
	KnowledgeDir                   string // local knowledge directory hint (defaults to WorkDir)
}

@AX:ANCHOR: [AUTO] worker runtime configuration boundary assembled by host resolution and consumed by WorkerLoop startup. @AX:REASON: Worktree isolation, fallback override, audit, auth, and provider fields coordinate desktop worker safety behavior. LoopConfig holds configuration for the WorkerLoop.

type Phase 

type Phase string

Phase represents a pipeline execution phase. 

const (
	PhasePlanner  Phase = "planner"
	PhaseExecutor Phase = "executor"
	PhaseTester   Phase = "tester"
	PhaseReviewer Phase = "reviewer"
)

func ParsePhase 

func ParsePhase(name string) (Phase, error)

ParsePhase validates and canonicalizes a single phase name.

func ParsePhasePlan 

func ParsePhasePlan(phases []string) ([]Phase, error)

ParsePhasePlan validates and canonicalizes a server-provided phase plan.

type PhaseResult 

type PhaseResult struct {
	Phase      Phase
	Output     string
	CostUSD    float64
	DurationMS int64
	SessionID  string
	ToolCalls  int // number of tool calls made during this phase
}

PhaseResult holds the output from a single pipeline phase.

type PipelineExecutor 

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

PipelineExecutor spawns separate subprocesses for each phase: planner -> executor(s) -> tester -> reviewer. Triggered when a single --print execution exceeds the context window.

func NewPipelineExecutor 

func NewPipelineExecutor(provider adapter.ProviderAdapter, mcpConfig, workDir string) *PipelineExecutor

NewPipelineExecutor creates a new PipelineExecutor.

func (*PipelineExecutor) Execute 

func (pe *PipelineExecutor) Execute(ctx context.Context, taskID, prompt string) (adapter.TaskResult, error)

Execute runs the full pipeline: planner -> executor(s) -> tester -> reviewer. Each phase uses an independent --resume session ID. Returns an aggregated TaskResult combining all phase outputs.

func (*PipelineExecutor) ExecuteWithPlan 

func (pe *PipelineExecutor) ExecuteWithPlan(ctx context.Context, taskID, prompt, model string, phases []Phase) (adapter.TaskResult, error)

@AX:ANCHOR: [AUTO] public phase-split execution contract called by Execute, worker loop, and integration tests (fan-in >= 3) @AX:REASON: Signature and phase/blocker semantics coordinate subprocess execution, routing, compression, and budget accounting. ExecuteWithPlan runs the pipeline with an optional server-selected model and explicit phase plan. When phases is empty, the default sequence is used.

func (*PipelineExecutor) SetBudget 

func (pe *PipelineExecutor) SetBudget(total int, alloc budget.PhaseAllocation)

SetBudget configures per-phase budget allocation for the pipeline.

func (*PipelineExecutor) SetCompressor 

func (pe *PipelineExecutor) SetCompressor(c compress.ContextCompressor)

SetCompressor configures context compression for phase transitions.

func (*PipelineExecutor) SetEnvVars 

func (pe *PipelineExecutor) SetEnvVars(envVars map[string]string)

SetEnvVars configures additional environment variables for all pipeline phases.

func (*PipelineExecutor) SetInterruptRecorder added in v0.45.0 

func (pe *PipelineExecutor) SetInterruptRecorder(record func(AuditEvent))

SetInterruptRecorder configures structured interrupt evidence recording.

func (*PipelineExecutor) SetIterationBudget 

func (pe *PipelineExecutor) SetIterationBudget(iterationBudget budget.IterationBudget)

SetIterationBudget configures a server-issued total iteration budget for the pipeline.

func (*PipelineExecutor) SetPhaseInstructions 

func (pe *PipelineExecutor) SetPhaseInstructions(instructions map[Phase]string)

SetPhaseInstructions configures server-selected instructions for pipeline phases.

func (*PipelineExecutor) SetPhasePromptTemplates 

func (pe *PipelineExecutor) SetPhasePromptTemplates(templates map[Phase]string)

SetPhasePromptTemplates configures server-selected full prompt templates for pipeline phases.

func (*PipelineExecutor) SetRouter 

func (pe *PipelineExecutor) SetRouter(r *routing.Router)

SetRouter configures model routing for the pipeline (REQ-ROUTE-01).

type StdinWriter 

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

StdinWriter wraps an io.WriteCloser to keep the stdin pipe open after the initial prompt is written. This enables mid-session message injection (e.g., budget warnings).

func NewStdinWriter 

func NewStdinWriter(pipe io.WriteCloser) *StdinWriter

NewStdinWriter creates a StdinWriter wrapping the given pipe.

func (*StdinWriter) Close 

func (sw *StdinWriter) Close() error

Close closes the underlying pipe.

func (*StdinWriter) Write 

func (sw *StdinWriter) Write(p []byte) (int, error)

Write implements io.Writer for injecting messages into stdin.

func (*StdinWriter) WritePrompt 

func (sw *StdinWriter) WritePrompt(prompt string) error

WritePrompt sends the initial prompt to the subprocess stdin. Unlike the previous implementation, the pipe is NOT closed after writing.

type TaskPayload 

type TaskPayload struct {
	TaskID        string
	Description   string
	PMNotes       string // PM instructions (optional)
	PolicySummary string // security policy summary
	KnowledgeCtx  string // Knowledge Hub context (optional)
	MemoryCtx     string // Agent Memory context (optional, SPEC-KHINT-001 REQ-003)
	SpecID        string // SPEC reference (optional)
}

TaskPayload contains backend-provided task data for prompt assembly.

type WorkerLoop 

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

WorkerLoop integrates A2A Server, ProviderAdapter, ContextBuilder, and StreamParser. It receives tasks via A2A, builds prompts, spawns CLI subprocesses, and reports results.

func NewWorkerLoop 

func NewWorkerLoop(config LoopConfig) *WorkerLoop

NewWorkerLoop creates a WorkerLoop with the given configuration.

func (*WorkerLoop) AddHostObserver 

func (wl *WorkerLoop) AddHostObserver(observer HostObserver)

func (*WorkerLoop) Close 

func (wl *WorkerLoop) Close() error

Close shuts down the worker loop and its A2A server.

func (*WorkerLoop) SetOnApprovalDecision 

func (wl *WorkerLoop) SetOnApprovalDecision() func(taskID, decision string)

SetOnApprovalDecision returns a callback that sends approval decisions to the backend.

func (*WorkerLoop) SetTUIProgram 

func (wl *WorkerLoop) SetTUIProgram(p *tea.Program)

SetTUIProgram registers the bubbletea program for sending approval messages.

func (*WorkerLoop) Start 

func (wl *WorkerLoop) Start(ctx context.Context) error

Start connects to the backend and begins processing tasks. @AX:ANCHOR[AUTO]: public lifecycle entry point — Start/Close are the primary WorkerLoop API; callers (CLI, tests) depend on error contract @AX:REASON: Startup order wires PID lock, A2A server, services, semaphore, and worktree manager before task dispatch.

type WorkspaceConfig 

type WorkspaceConfig struct {
	WorkspaceID string
	ProjectDir  string
	BackendURL  string
	AuthToken   string
}

WorkspaceConfig holds connection details for a single workspace.

Source Files

View all Source files

Directories

Path Synopsis
Package adapter — resolve: CLI binary path resolution with well-known fallbacks.
 Package adapter — resolve: CLI binary path resolution with well-known fallbacks.
Package audit provides a rotating log writer for audit trails.
 Package audit provides a rotating log writer for audit trails.
Package auth provides token lifecycle management for autopus workers.
 Package auth provides token lifecycle management for autopus workers.
Package daemon contains retained launchd/systemd helpers for legacy local-host worker mode.
 Package daemon contains retained launchd/systemd helpers for legacy local-host worker mode.
Package host contains the retained ADK local-host worker runtime boundary.
 Package host contains the retained ADK local-host worker runtime boundary.
Package mcpserver implements a JSON-RPC 2.0 MCP server over stdio.
 Package mcpserver implements a JSON-RPC 2.0 MCP server over stdio.
Package pidlock provides advisory PID-based lock file management for single-instance enforcement.
 Package pidlock provides advisory PID-based lock file management for single-instance enforcement.
Package poll contains retained backend polling helpers for legacy local-host worker mode.
 Package poll contains retained backend polling helpers for legacy local-host worker mode.
Package qa provides QA pipeline stages for build, test, and health checks.
 Package qa provides QA pipeline stages for build, test, and health checks.
Package reaper provides zombie process detection and reaping for worker subprocesses.
 Package reaper provides zombie process detection and reaping for worker subprocesses.
Package security provides secret scanning and redaction for worker output.
 Package security provides secret scanning and redaction for worker output.
Package setup - apikey.go: legacy Worker API Key credential persistence.
 Package setup - apikey.go: legacy Worker API Key credential persistence.
Package tui provides a bubbletea-based dashboard for the worker daemon.
 Package tui provides a bubbletea-based dashboard for the worker daemon.

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.