mission

package
v0.1.1 Latest Latest
Warning

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

 Go to latest Published: Jun 25, 2026 License: MIT Imports: 18 Imported by: 0

Documentation

Overview

Package mission provides multi-feature parallel execution for hawk. This file adds a watchdog for stall detection in worker goroutines.

Index

Constants

This section is empty.

Variables

View Source 
var ErrApprovalTimeout = errors.New("approval request timed out: context cancelled")

ErrApprovalTimeout is returned when the context expires before the operator calls Respond().

View Source 
var ErrAwaitingApproval = errors.New("workflow halted: awaiting human approval at checkpoint gate")

ErrAwaitingApproval is returned by Workflow.Run when execution halts at a human-in-the-loop checkpoint gate. The workflow state is durable at this point: a later Resume after Approve/Reject continues from the gate.

View Source 
var ErrToolRejected = errors.New("tool call rejected by human approval gate")

ErrToolRejected is returned from the gate check when the operator rejects a tool call.

Functions

This section is empty.

Types

type AgentCapability 

type AgentCapability struct {
	Name          string
	Expertise     []string
	Tools         []string
	MaxComplexity int
	Model         string
}

AgentCapability describes a specialized agent's capabilities.

type AgentMessage 

type AgentMessage struct {
	ID               string    `json:"id"`
	From             string    `json:"from"`
	To               string    `json:"to,omitempty"`
	Topic            string    `json:"topic"`
	Content          string    `json:"content"`
	Priority         int       `json:"priority"`
	Timestamp        time.Time `json:"timestamp"`
	RequiresResponse bool      `json:"requires_response,omitempty"`
	ResponseTo       string    `json:"response_to,omitempty"`
}

AgentMessage represents a message exchanged between agents during a mission.

type AgentStatus 

type AgentStatus struct {
	Name        string
	Busy        bool
	TaskCount   int
	SuccessRate float64
	LastActive  time.Time
}

AgentStatus tracks the state and performance of a registered agent.

type ApprovalRequest 

type ApprovalRequest struct {
	// ToolName is the canonical name of the tool being called.
	ToolName string
	// Args are the tool call arguments as a map.
	Args map[string]interface{}
	// Summary is a one-line human-readable description of the action.
	Summary string
	// Category is the risk category matched by the gate classifier.
	Category string
	// contains filtered or unexported fields
}

ApprovalRequest describes a tool call that is pending human approval. The caller receives this via MissionApprovalGate.OnRequest and must call Respond() to unblock the waiting mission worker.

func (*ApprovalRequest) Await 

func (r *ApprovalRequest) Await(ctx context.Context) (RequestResponse, error)

Await blocks until Respond() is called or ctx is cancelled. Returns ErrApprovalTimeout when ctx expires before the operator responds.

func (*ApprovalRequest) Respond 

func (r *ApprovalRequest) Respond(resp RequestResponse) error

Respond sends a decision to the waiting Await() call. It is safe to call from any goroutine. Subsequent Respond() calls after the first are no-ops.

type BusStats 

type BusStats struct {
	// Dropped is the cumulative number of messages that could not be
	// delivered to a registered agent because its channel was full.
	// Includes both broadcast and direct-send drops.
	Dropped int64
	// Agents is the number of currently registered agents.
	Agents int
	// Locks is the number of currently held resource locks.
	Locks int
	// HistorySz is the number of messages retained in history.
	HistorySz int
}

BusStats is a snapshot of MessageBus runtime counters.

type Config 

type Config struct {
	MaxWorkers        int           `json:"max_workers"`
	WorkerModel       string        `json:"worker_model"`
	ValidatorModel    string        `json:"validator_model"`
	RepoDir           string        `json:"repo_dir"`
	BaseBranch        string        `json:"base_branch"`
	AutonomyLevel     int           `json:"autonomy_level"`
	SkipValidation    bool          `json:"skip_validation"`
	PerWorkerTimeout  time.Duration `json:"per_worker_timeout,omitempty"`
	MaxRetriesPerFeat int           `json:"max_retries_per_feat,omitempty"`

	// Staged pipeline configuration (oh-my-claudecode-style team workflow).
	PRDModel          string `json:"prd_model,omitempty"`
	FixModel          string `json:"fix_model,omitempty"`
	MaxFixAttempts    int    `json:"max_fix_attempts,omitempty"` // default 3
	EnablePRDPhase    bool   `json:"enable_prd_phase,omitempty"`
	EnableVerifyPhase bool   `json:"enable_verify_phase,omitempty"`
}

Config controls mission orchestration behavior.

type Fact 

type Fact struct {
	ID         string
	Content    string
	Source     string
	Confidence float64
	Timestamp  time.Time
}

Fact represents a discovered piece of information during a mission.

type Feature 

type Feature struct {
	ID                 string        `json:"id"`
	Description        string        `json:"description"`
	ExpectedBehavior   string        `json:"expected_behavior"`
	Branch             string        `json:"branch"`
	WorkerSessionID    string        `json:"worker_session_id,omitempty"`
	Status             FeatureStatus `json:"status"`
	Handoff            *Handoff      `json:"handoff,omitempty"`
	StartedAt          time.Time     `json:"started_at,omitempty"`
	CompletedAt        time.Time     `json:"completed_at,omitempty"`
	PRD                string        `json:"prd,omitempty"`                 // generated product requirements
	VerificationResult string        `json:"verification_result,omitempty"` // verify-phase outcome
	FixAttempts        int           `json:"fix_attempts,omitempty"`        // number of fix passes applied
}

Feature is a discrete unit of work assigned to a worker.

type FeatureStatus 

type FeatureStatus string

FeatureStatus tracks individual feature progress. 

const (
	FeaturePending    FeatureStatus = "pending"
	FeatureInProgress FeatureStatus = "in_progress"
	FeatureCompleted  FeatureStatus = "completed"
	FeatureFailed     FeatureStatus = "failed"
)

type FixFunc 

type FixFunc func(ctx context.Context, feature *Feature, verificationResult string, handoff *Handoff) (*Handoff, error)

FixFunc attempts to fix a feature that failed verification.

type Handoff 

type Handoff struct {
	CommitID     string   `json:"commit_id,omitempty"`
	RepoPath     string   `json:"repo_path,omitempty"`
	Summary      string   `json:"summary"`
	FilesChanged []string `json:"files_changed,omitempty"`
	TestsPassed  bool     `json:"tests_passed"`
}

Handoff is the structured report a worker produces upon completion.

type HandoffMessage 

type HandoffMessage struct {
	FromAgent string
	ToAgent   string
	Reason    string
	Context   string
	Task      string
	Priority  int
	State     map[string]interface{}
	Timestamp time.Time
}

HandoffMessage represents a control transfer message between agents.

type HandoffProtocol 

type HandoffProtocol struct {
	Agents      map[string]*AgentCapability
	ActiveAgent string
	History     []HandoffMessage
	// contains filtered or unexported fields
}

HandoffProtocol manages typed message-based control transfer between specialized agents.

func NewHandoffProtocol 

func NewHandoffProtocol() *HandoffProtocol

NewHandoffProtocol creates a new HandoffProtocol with initialized fields.

func (*HandoffProtocol) BuildHandoffContext 

func (hp *HandoffProtocol) BuildHandoffContext(from, to string, task string) *HandoffMessage

BuildHandoffContext automatically builds a HandoffMessage with relevant state for transferring control from one agent to another.

func (*HandoffProtocol) CanHandle 

func (hp *HandoffProtocol) CanHandle(agentName, task string) bool

CanHandle checks whether a given agent can handle a task by matching task keywords against the agent's expertise.

func (*HandoffProtocol) EscalateToHuman 

func (hp *HandoffProtocol) EscalateToHuman(reason string) *HandoffMessage

EscalateToHuman creates a special handoff message that signals human intervention is needed.

func (*HandoffProtocol) FormatHandoffHistory 

func (hp *HandoffProtocol) FormatHandoffHistory() string

FormatHandoffHistory returns a formatted string representation of the handoff history.

func (*HandoffProtocol) GetActiveAgent 

func (hp *HandoffProtocol) GetActiveAgent() *AgentCapability

GetActiveAgent returns the currently active agent's capability, or nil if no agent is active.

func (*HandoffProtocol) Handoff 

func (hp *HandoffProtocol) Handoff(msg HandoffMessage) error

Handoff transfers control from one agent to another. It validates the target agent exists, transfers context and task, records the handoff in history, and sets the new active agent.

func (*HandoffProtocol) RegisterAgent 

func (hp *HandoffProtocol) RegisterAgent(cap AgentCapability)

RegisterAgent adds an agent capability to the protocol.

func (*HandoffProtocol) SelectBestAgent 

func (hp *HandoffProtocol) SelectBestAgent(task string) string

SelectBestAgent picks the best agent for a given task by matching task keywords against agent expertise and considering complexity.

type Ledger 

type Ledger struct {
	Facts             []Fact
	Plan              []PlanItem
	Agents            map[string]*AgentStatus
	CurrentAssignment string
	StallCount        int
	// contains filtered or unexported fields
}

Ledger maintains the shared state of facts, plan, and agents for orchestration.

type LedgerOrchestrator 

type LedgerOrchestrator struct {
	Ledger            *Ledger
	MaxStalls         int
	ReassignThreshold time.Duration
	// contains filtered or unexported fields
}

LedgerOrchestrator coordinates multi-agent work using a task ledger, dynamically reassigning work based on progress — inspired by MagenticOne.

func NewLedgerOrchestrator 

func NewLedgerOrchestrator() *LedgerOrchestrator

NewLedgerOrchestrator creates a new orchestrator with sensible defaults.

func (*LedgerOrchestrator) AddFact 

func (lo *LedgerOrchestrator) AddFact(content, source string, confidence float64)

AddFact records a new discovered fact into the ledger.

func (*LedgerOrchestrator) AddPlanItem 

func (lo *LedgerOrchestrator) AddPlanItem(description string, deps []string)

AddPlanItem appends a new plan item with the given description and dependencies.

func (*LedgerOrchestrator) AssignNext 

func (lo *LedgerOrchestrator) AssignNext() (*PlanItem, string)

AssignNext finds the next unblocked pending plan item and assigns it to the best available agent. Returns nil, "" if no assignment can be made.

func (*LedgerOrchestrator) DetectStall 

func (lo *LedgerOrchestrator) DetectStall() bool

DetectStall checks if any active item has exceeded the ReassignThreshold without progress. Returns true if a stall is detected.

func (*LedgerOrchestrator) FormatLedger 

func (lo *LedgerOrchestrator) FormatLedger() string

FormatLedger returns a human-readable summary of the ledger state.

func (*LedgerOrchestrator) Reassign 

func (lo *LedgerOrchestrator) Reassign(itemID string) string

Reassign moves a stalled item to a different agent and records a fact. Returns the name of the new agent, or "" if no reassignment is possible.

func (*LedgerOrchestrator) RegisterAgent 

func (lo *LedgerOrchestrator) RegisterAgent(name string)

RegisterAgent adds a new agent to the orchestrator's pool.

func (*LedgerOrchestrator) ReportProgress 

func (lo *LedgerOrchestrator) ReportProgress(agentName string, itemID string, status string)

ReportProgress updates the status of a plan item and the agent's stats.

func (*LedgerOrchestrator) UpdatePlan 

func (lo *LedgerOrchestrator) UpdatePlan(newItems []PlanItem)

UpdatePlan replaces or merges plan items dynamically. Items with matching IDs are updated; new items are appended.

type MemEntry 

type MemEntry struct {
	Key       string      `json:"key"`
	Value     interface{} `json:"value"`
	Type      string      `json:"type"` // "string", "int", "bool", "json", "list"
	Owner     string      `json:"owner"`
	CreatedAt time.Time   `json:"created_at"`
	UpdatedAt time.Time   `json:"updated_at"`
	Version   int         `json:"version"`
	Readers   []string    `json:"readers,omitempty"`
}

MemEntry represents a single entry in shared memory.

type MessageBus 

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

MessageBus coordinates inter-agent communication during mission execution.

func NewMessageBus 

func NewMessageBus() *MessageBus

NewMessageBus creates and returns an initialized MessageBus.

func (*MessageBus) AcquireLock 

func (mb *MessageBus) AcquireLock(resource, owner string, ttl time.Duration) error

AcquireLock attempts to acquire an exclusive lock on a resource. Returns nil on success, or an error if the resource is held by another agent. A lock held by the same owner is refreshed (re-entrant). Expired locks are reclaimed.

func (*MessageBus) Broadcast 

func (mb *MessageBus) Broadcast(from, topic, content string)

Broadcast sends a message from one agent to all others. Broadcasts never fail: agents with full buffers are silently skipped.

func (*MessageBus) BuildContextFromMessages 

func (mb *MessageBus) BuildContextFromMessages(agentID string, maxTokens int) string

BuildContextFromMessages formats recent relevant messages for injection into an agent's context. It returns a human-readable summary of team communication limited to approximately maxTokens characters.

func (*MessageBus) CleanupExpiredLocks 

func (mb *MessageBus) CleanupExpiredLocks()

CleanupExpiredLocks removes all locks whose expiry time has passed.

func (*MessageBus) GetHistory 

func (mb *MessageBus) GetHistory(topic string, limit int) []AgentMessage

GetHistory returns messages filtered by topic (or all if topic is empty), limited to the most recent `limit` entries.

func (*MessageBus) IsLocked 

func (mb *MessageBus) IsLocked(resource string) bool

IsLocked reports whether a resource is currently locked by any agent. Expired locks are treated as unlocked.

func (*MessageBus) Register 

func (mb *MessageBus) Register(agentID string) <-chan AgentMessage

Register creates a channel for the given agent to receive messages. Returns a read-only channel the agent can listen on.

func (*MessageBus) ReleaseLock 

func (mb *MessageBus) ReleaseLock(resource, owner string) error

ReleaseLock releases a lock on a resource, verifying ownership.

func (*MessageBus) ReportConflict 

func (mb *MessageBus) ReportConflict(from string, files []string, description string)

ReportConflict notifies all agents about a file conflict so they can coordinate.

func (*MessageBus) ReportDiscovery 

func (mb *MessageBus) ReportDiscovery(from, discovery string)

ReportDiscovery shares a useful finding with all agents.

func (*MessageBus) ReportProgress 

func (mb *MessageBus) ReportProgress(from string, pct float64, status string)

ReportProgress sends a progress update for coordination.

func (*MessageBus) RequestHelp 

func (mb *MessageBus) RequestHelp(from, description string) string

RequestHelp broadcasts a help request and returns the message ID for tracking responses.

func (*MessageBus) Send 

func (mb *MessageBus) Send(msg AgentMessage) error

Send delivers a message to a specific agent or broadcasts to all. If msg.To is set, delivers to that specific agent. If msg.To is empty, broadcasts to all registered agents (except sender).

The dropped-message WARN log is performed after mb.mu is released (see M8): a slow slog handler should not serialize bus operations. We snapshot the (from, to, topic) of each drop into a local slice while still holding the lock, then iterate the slice after the inner function returns and the lock has been released.

func (*MessageBus) Stats 

func (mb *MessageBus) Stats() BusStats

Stats returns a snapshot of MessageBus counters. Safe for concurrent use.

func (*MessageBus) Subscribe 

func (mb *MessageBus) Subscribe(agentID, topic string)

Subscribe registers an agent to receive messages for a given topic.

func (*MessageBus) TryLockFiles 

func (mb *MessageBus) TryLockFiles(from string, files []string, ttl time.Duration) []string

TryLockFiles attempts to acquire locks on a set of files for an agent. Returns the list of files that could NOT be locked (held by others). On any contention, it also reports a conflict via the message bus.

func (*MessageBus) Unregister 

func (mb *MessageBus) Unregister(agentID string)

Unregister removes an agent from the message bus and closes its channel.

func (*MessageBus) WaitForLock 

func (mb *MessageBus) WaitForLock(resource, owner string, timeout time.Duration) error

WaitForLock blocks until a resource lock can be acquired or the timeout elapses.

Implementation: push-based via a per-call done channel. The caller registers as a waiter; ReleaseLock() closes the done channel when the lock is freed. This replaces the previous 20ms busy-poll.

Once ReleaseLock closes our done channel, the channel stays closed for the lifetime of the waiter. The previous implementation re-entered the select after each signal, which (a) busy-spinned calling AcquireLock on the same closed-channel signal under contention and (b) raced the timer.C case: both `<-w.done` and `<-timer.C` are ready, and select picks randomly — so the timeout could be missed or the function could loop indefinitely waiting on the second signal from a ReleaseLock that never re-fires. (See M7 in the code review.)

Fix: use a one-shot select. On the first signal (w.done or timer), try AcquireLock exactly once. If it succeeds, return nil; if the timer fired, return a timeout error. There is no second loop, so no busy-spin and no select race.

func (*MessageBus) WaitForResponse 

func (mb *MessageBus) WaitForResponse(messageID string, timeout time.Duration) (*AgentMessage, error)

WaitForResponse blocks until a response to the given messageID arrives or timeout elapses.

Implementation: push-based via a per-call done channel. The caller registers as a waiter; Send() closes the done channel when a matching response is appended to history. This replaces the previous 10ms busy-poll (which missed sub-tick responses and burned CPU under load).

Race: a response may have been appended to history just before the waiter registers. The fast path below checks history first, so a late-arriving WaitForResponse still finds the answer.

type Mission 

type Mission struct {
	ID          string    `json:"id"`
	Prompt      string    `json:"prompt"`
	Dir         string    `json:"dir"`
	Features    []Feature `json:"features"`
	Status      Status    `json:"status"`
	StartedAt   time.Time `json:"started_at"`
	CompletedAt time.Time `json:"completed_at,omitempty"`
	Config      Config    `json:"config"`

	// ApprovalGate is an optional typed human-in-the-loop gate that intercepts
	// tool calls matching a flagged risk category before they execute.
	// When nil the gate is a no-op and all tool calls proceed automatically.
	ApprovalGate *MissionApprovalGate `json:"-"`
	// contains filtered or unexported fields
}

Mission represents a multi-agent orchestration run.

func New 

func New(prompt string, cfg Config) *Mission

New creates a new mission from a prompt and configuration.

func (*Mission) Fix 

func (m *Mission) Fix(ctx context.Context, fixFn FixFunc) error

Fix runs the fix loop: for each failed feature with a recorded verification result, attempts up to MaxFixAttempts fixes, re-marking the feature completed on success.

func (*Mission) GeneratePRD 

func (m *Mission) GeneratePRD(ctx context.Context, prdFn PRDFunc) error

GeneratePRD runs the PRD phase: generates a requirements doc for each feature.

func (*Mission) Plan 

func (m *Mission) Plan(ctx context.Context, planFn PlanFunc) error

func (*Mission) Run 

func (m *Mission) Run(ctx context.Context, workerFn WorkerFunc) error

Run executes all features in parallel using workerFn.

func (*Mission) RunStaged 

func (m *Mission) RunStaged(ctx context.Context, workerFn WorkerFunc, opts ...StagedOption) error

RunStaged orchestrates the full team pipeline: 

team-plan (already done via Plan) -> team-prd -> team-exec -> team-verify -> team-fix loop

PRD and verify phases run only when enabled in Config. The fix loop runs when a fix function is provided and the verify phase is enabled.

func (*Mission) Summary 

func (m *Mission) Summary() string

Summary returns a human-readable summary of the mission.

func (*Mission) Verify 

func (m *Mission) Verify(ctx context.Context, verifyFn VerifyFunc) error

Verify runs the verify phase: validates each completed feature. Features that fail verification are marked FeatureFailed and their result recorded.

type MissionApprovalGate 

type MissionApprovalGate struct {
	// OnRequest is called synchronously in the worker goroutine before the tool
	// runs. The implementation must not call Await() itself — it should hand the
	// *ApprovalRequest to an operator UI and return immediately.
	OnRequest func(req *ApprovalRequest)
	// contains filtered or unexported fields
}

MissionApprovalGate wraps a mission's tool-call dispatch with the typed channel-based gate. When OnRequest is non-nil and a tool call matches a flagged category, OnRequest is invoked with an *ApprovalRequest; the worker goroutine blocks on Await() until the operator calls Respond(). If OnRequest is nil the gate is a no-op (auto-approve everything).

SessionApproved tracks tools approved for the entire session via ResponseApproveForSession; subsequent calls to those tools skip the gate.

func NewMissionApprovalGate 

func NewMissionApprovalGate(onRequest func(req *ApprovalRequest)) *MissionApprovalGate

NewMissionApprovalGate creates a gate with the given OnRequest handler. Pass nil to create a no-op gate.

func (*MissionApprovalGate) Check 

func (g *MissionApprovalGate) Check(ctx context.Context, toolName string, args map[string]interface{}) error

Check consults the gate for a pending tool call. If the gate is enabled and the tool matches a risky category, it calls OnRequest and blocks until the operator responds. Returns an error if the call is rejected or the context expires; nil means proceed.

type PRDFunc 

type PRDFunc func(ctx context.Context, feature *Feature) (prd string, err error)

PRDFunc generates a product requirements document for a feature.

type PlanFunc 

type PlanFunc func(ctx context.Context, prompt string) ([]Feature, error)

Plan decomposes the mission prompt into features. planFn is called with the prompt and should return a list of features.

type PlanItem 

type PlanItem struct {
	ID           string
	Description  string
	Status       string // "pending", "active", "done", "stalled"
	AssignedTo   string
	Dependencies []string
	Order        int
}

PlanItem represents a single unit of work in the orchestration plan.

type RequestResponse 

type RequestResponse int

RequestResponse is the operator's decision for an approval request. 

const (
	// ResponseApprove allows the tool call once.
	ResponseApprove RequestResponse = iota
	// ResponseApproveForSession auto-approves subsequent calls to the same tool.
	ResponseApproveForSession
	// ResponseReject denies the tool call and causes an error event.
	ResponseReject
)

type ResourceLock 

type ResourceLock struct {
	Resource   string    `json:"resource"`
	Owner      string    `json:"owner"`
	AcquiredAt time.Time `json:"acquired_at"`
	ExpiresAt  time.Time `json:"expires_at"`
	Priority   int       `json:"priority"`
}

ResourceLock represents an exclusive lock on a shared resource (e.g. a file). Prevents conflicting operations when multiple agents work in parallel.

type Responder 

type Responder interface {
	Respond(RequestResponse) error
}

Responder is implemented by anything that can resolve an ApprovalRequest.

type SharedMemory 

type SharedMemory struct {
	Entries    map[string]*MemEntry `json:"entries"`
	Namespaces map[string][]string  `json:"namespaces"` // namespace -> list of keys
	// contains filtered or unexported fields
}

SharedMemory provides a concurrent-safe key-value store for multi-agent workflows, allowing agents to read and write common state without message passing.

func NewSharedMemory 

func NewSharedMemory() *SharedMemory

NewSharedMemory creates and returns an initialized SharedMemory instance.

func (*SharedMemory) Clear 

func (sm *SharedMemory) Clear()

Clear resets all entries, namespaces, and watchers.

func (*SharedMemory) Delete 

func (sm *SharedMemory) Delete(key string, owner string) error

Delete removes an entry from shared memory. Only the owner who wrote the entry is permitted to delete it; otherwise an error is returned.

func (*SharedMemory) Diff 

func (sm *SharedMemory) Diff(since time.Time) []MemEntry

Diff returns all entries that have been modified since the given timestamp.

func (*SharedMemory) FormatState 

func (sm *SharedMemory) FormatState() string

FormatState returns a human-readable representation of all entries in shared memory, formatted for display in agent context.

func (*SharedMemory) Get 

func (sm *SharedMemory) Get(key string) (interface{}, bool)

Get retrieves a value from shared memory by key. Returns the value and a boolean indicating whether the key was found.

func (*SharedMemory) GetBool 

func (sm *SharedMemory) GetBool(key string) bool

GetBool retrieves a value as a bool. Returns false if the key does not exist or the value cannot be interpreted as a boolean.

func (*SharedMemory) GetInt 

func (sm *SharedMemory) GetInt(key string) int

GetInt retrieves a value as an int. Returns 0 if the key does not exist or the value cannot be interpreted as an integer.

func (*SharedMemory) GetString 

func (sm *SharedMemory) GetString(key string) string

GetString retrieves a value as a string. Returns empty string if the key does not exist or the value cannot be converted to a string.

func (*SharedMemory) List 

func (sm *SharedMemory) List(namespace string) []MemEntry

List returns all entries belonging to a given namespace.

func (*SharedMemory) Restore 

func (sm *SharedMemory) Restore(snapshot map[string]interface{})

Restore populates shared memory from a previously captured snapshot. Existing entries are cleared and replaced with the snapshot contents. All restored entries are assigned owner "snapshot" and version 1.

func (*SharedMemory) Set 

func (sm *SharedMemory) Set(key string, value interface{}, owner string)

Set writes a value into shared memory. If the key already exists, the version is incremented and the UpdatedAt timestamp is refreshed. The owner is recorded as the agent that last wrote the entry.

func (*SharedMemory) SetNamespace 

func (sm *SharedMemory) SetNamespace(key, namespace string)

SetNamespace assigns a key to a namespace for organizational grouping. A key can belong to multiple namespaces.

func (*SharedMemory) Snapshot 

func (sm *SharedMemory) Snapshot() map[string]interface{}

Snapshot returns a shallow copy of all key-value pairs currently in shared memory.

func (*SharedMemory) Watch 

func (sm *SharedMemory) Watch(key string) <-chan *MemEntry

Watch returns a channel that receives notifications whenever the specified key is updated. The channel is buffered (size 16) and non-blocking on send.

type StagedOption 

type StagedOption func(*stagedConfig)

StagedOption configures RunStaged behavior.

func WithFix 

func WithFix(fn FixFunc) StagedOption

WithFix supplies the fix function for the staged pipeline.

func WithPRD 

func WithPRD(fn PRDFunc) StagedOption

WithPRD supplies the PRD generation function for the staged pipeline.

func WithVerify 

func WithVerify(fn VerifyFunc) StagedOption

WithVerify supplies the verification function for the staged pipeline.

type Status 

type Status string

Status represents the mission lifecycle. 

const (
	StatusPlanning    Status = "planning"
	StatusPlanningPRD Status = "planning_prd"
	StatusRunning     Status = "running"
	StatusExecuting   Status = "executing"
	StatusVerifying   Status = "verifying"
	StatusFixing      Status = "fixing"
	StatusValidating  Status = "validating"
	StatusCompleted   Status = "completed"
	StatusFailed      Status = "failed"
	StatusPartial     Status = "partial"
)

type StepDef 

type StepDef struct {
	Name      string
	Fn        StepFunc
	HumanGate bool
}

NewWorkflow creates a durable workflow with the given name and ordered step definitions. Each definition contributes a named step and its executor. dir is where workflow.json is persisted; if empty, persistence is skipped.

type StepFunc 

type StepFunc func(ctx context.Context, state *WorkflowState) (output json.RawMessage, err error)

StepFunc executes a single workflow step. The returned bytes (may be nil) are persisted as the step's durable Output.

type StepStatus 

type StepStatus string

StepStatus tracks the lifecycle of a single workflow step. 

const (
	StepPending        StepStatus = "pending"
	StepRunning        StepStatus = "running"
	StepCompleted      StepStatus = "completed"
	StepFailed         StepStatus = "failed"
	StepAwaitingApprov StepStatus = "awaiting_approval"
	StepRejected       StepStatus = "rejected"
)

type VerifyFunc 

type VerifyFunc func(ctx context.Context, feature *Feature, handoff *Handoff) (passed bool, result string, err error)

VerifyFunc validates a completed feature's implementation. Returns passed=true if the implementation satisfies the feature's expected behavior.

type Watchdog 

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

Watchdog monitors worker progress and triggers fallback on stalls.

func NewWatchdog 

func NewWatchdog(cfg WatchdogConfig, onStall func(string)) *Watchdog

NewWatchdog creates a new watchdog with the given config and stall callback.

func (*Watchdog) ActiveCount 

func (w *Watchdog) ActiveCount() int

ActiveCount returns the number of currently monitored features.

func (*Watchdog) Register 

func (w *Watchdog) Register(featureID string)

Register marks a feature as active (call when worker starts).

func (*Watchdog) Run 

func (w *Watchdog) Run(ctx context.Context)

Run starts the watchdog loop. Blocks until ctx is cancelled.

func (*Watchdog) Touch 

func (w *Watchdog) Touch(featureID string)

Touch updates the last-activity timestamp for a feature. Call this when a worker produces output.

func (*Watchdog) Unregister 

func (w *Watchdog) Unregister(featureID string)

Unregister removes a feature from monitoring (call on completion).

type WatchdogConfig 

type WatchdogConfig struct {
	StallTimeout  time.Duration // default 90s — no output for this long = stall
	CheckInterval time.Duration // default 10s — how often to check
}

WatchdogConfig configures the stall-detection watchdog.

type WorkerFunc 

type WorkerFunc func(ctx context.Context, feature *Feature, missionDir string, cfg Config) (*Handoff, error)

WorkerFunc is the function type that the orchestrator calls for each feature. It receives the feature and mission dir, and returns a handoff or error.

func EngineWorker 

func EngineWorker(provider, model, systemPrompt string) WorkerFunc

EngineWorker returns a WorkerFunc that runs an actual engine session in an isolated git worktree for each feature.

func ReadOnlyValidationWorker 

func ReadOnlyValidationWorker(provider, model, systemPrompt string) WorkerFunc

ReadOnlyValidationWorker returns a WorkerFunc that reviews an already-produced implementation without modifying it. It runs in the same worktree the implementation worker committed to (passed via cfg.RepoDir / the handed-off branch) using a read-only tool registry, and returns its verdict as the Handoff.Summary. It never commits and reports no FilesChanged of its own.

This is the validation half of the implement-then-validate pair: pipeline an EngineWorker (implementation) into a ReadOnlyValidationWorker (validation) so the agent that writes the code is never the agent that signs off on it.

type Workflow 

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

Workflow is a durable, resumable sequence of named steps.

func LoadWorkflow 

func LoadWorkflow(dir string, defs ...StepDef) (*Workflow, error)

LoadWorkflow reads a previously persisted workflow from dir and rebinds the supplied step executors by name. Steps whose names are not present in defs are left without an executor (Resume will error if it needs to run them).

func NewWorkflow 

func NewWorkflow(name, dir string, defs ...StepDef) *Workflow

NewWorkflow constructs a Workflow from ordered step definitions.

func (*Workflow) Approve 

func (w *Workflow) Approve(name string) error

Approve records human approval for the named gate step, making the decision durable, so a subsequent Resume proceeds past the gate.

func (*Workflow) Done 

func (w *Workflow) Done() bool

Done reports whether every step has reached StepCompleted.

func (*Workflow) LastCompletedStep 

func (w *Workflow) LastCompletedStep() string

LastCompletedStep returns the name of the most recently completed step, or "" if none have completed yet. Useful for reporting resume points.

func (*Workflow) Reject 

func (w *Workflow) Reject(name string) error

Reject records a human rejection for the named gate step. A rejected gate halts the workflow on the next Run/Resume.

func (*Workflow) Resume 

func (w *Workflow) Resume(ctx context.Context) error

Resume continues a workflow from the last completed step. It is identical to Run; the name documents intent at call sites after a restart.

func (*Workflow) Run 

func (w *Workflow) Run(ctx context.Context) error

Run executes steps from the first non-completed step to the end. It persists after every step transition so a crash leaves a resumable state. When it reaches an un-approved human gate it persists StepAwaitingApprov and returns ErrAwaitingApproval. Resume is an alias for Run after Approve.

func (*Workflow) SetValue 

func (w *Workflow) SetValue(key, val string)

SetValue stores a shared value visible to subsequent steps and across restarts.

func (*Workflow) State 

func (w *Workflow) State() *WorkflowState

State returns a snapshot pointer to the workflow's durable state.

type WorkflowState 

type WorkflowState struct {
	ID        string            `json:"id"`
	Name      string            `json:"name"`
	Steps     []WorkflowStep    `json:"steps"`
	Values    map[string]string `json:"values,omitempty"`
	CreatedAt time.Time         `json:"created_at"`
	UpdatedAt time.Time         `json:"updated_at"`
	Dir       string            `json:"-"`
}

WorkflowState is the durable, serializable state of a Workflow. The bag of shared values (Values) lets steps pass data forward across restarts.

type WorkflowStep 

type WorkflowStep struct {
	Name        string     `json:"name"`
	Status      StepStatus `json:"status"`
	HumanGate   bool       `json:"human_gate,omitempty"` // requires Approve before running
	Approved    bool       `json:"approved,omitempty"`
	StartedAt   time.Time  `json:"started_at,omitempty"`
	CompletedAt time.Time  `json:"completed_at,omitempty"`
	Attempts    int        `json:"attempts,omitempty"`
	Error       string     `json:"error,omitempty"`
	// Output is an opaque, JSON-serializable result the step produced. It is
	// persisted so resumed runs can read prior step output without re-running.
	Output json.RawMessage `json:"output,omitempty"`
}

WorkflowStep is a durable, named unit of work in a Workflow.

Source Files

View all Source files

Directories

Path Synopsis

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.