tasks

package
v0.49.3 Latest Latest
Warning

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

 Go to latest Published: Jun 17, 2026 License: AGPL-3.0 Imports: 17 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	GoalStatusDraft     = "draft"
	GoalStatusPlanning  = "planning"
	GoalStatusRunning   = "running"
	GoalStatusBlocked   = "blocked"
	GoalStatusReviewing = "reviewing"
	GoalStatusDone      = "done"
	GoalStatusFailed    = "failed"
	GoalStatusCancelled = "cancelled"
)

Goal status constants. Mirrors the agent_goal.status CHECK.

View Source 
const (
	ReadinessDispatchable = "dispatchable"
	ReadinessWaitingDeps  = "waiting_deps"
	ReadinessDeferred     = "deferred" // not_before > now
	ReadinessThrottled    = "throttled"
	ReadinessRunning      = "running"
	ReadinessBlocked      = "blocked"
	ReadinessTerminal     = "terminal"
	ReadinessDraft        = "draft"
	ReadinessReviewing    = "reviewing"
	ReadinessUnknown      = "unknown"
)

Readiness states. dispatchable is the only state on which the dispatcher actually claims; the rest are diagnostic.

View Source 
const (
	ReviewRequested        = "requested"
	ReviewInProgress       = "in_progress"
	ReviewApproved         = "approved"
	ReviewChangesRequested = "changes_requested"
	ReviewRejected         = "rejected"
	ReviewEscalated        = "escalated"
	ReviewCancelled        = "cancelled"
)

Review status constants.

View Source 
const (
	ReviewPolicyNone  = "none"
	ReviewPolicyAuto  = "auto"
	ReviewPolicyAgent = "agent"
	ReviewPolicyHuman = "human"
)

Review policy constants.

View Source 
const (
	ReviewerSystem = "system"
	ReviewerAgent  = "agent"
	ReviewerHuman  = "human"
)

Reviewer type constants (mirrors enum in schema).

View Source 
const (
	StatusDraft     = "draft"
	StatusReady     = "ready"
	StatusRunning   = "running"
	StatusBlocked   = "blocked"
	StatusDone      = "done"
	StatusFailed    = "failed"
	StatusCancelled = "cancelled"
)

Task lifecycle (Slice 1). Slice 2 widens to include StatusReviewing.

View Source 
const (
	PriorityRoutine = "routine"
	PriorityUrgent  = "urgent"
)

Task priorities.

View Source 
const (
	RunQueued      = "queued"
	RunRunning     = "running"
	RunCompleted   = "completed"
	RunFailed      = "failed"
	RunCancelled   = "cancelled"
	RunInterrupted = "interrupted"
	RunTimedOut    = "timed_out"
)

Run lifecycle.

View Source 
const (
	BlockerOpen      = "open"
	BlockerResolved  = "resolved"
	BlockerCancelled = "cancelled"
)

Blocker lifecycle.

View Source 
const (
	BlockerKindUserInput          = "user_input"
	BlockerKindExternalDependency = "external_dependency"
	BlockerKindToolError          = "tool_error"
	BlockerKindPolicyHold         = "policy_hold"
	BlockerKindDepFailure         = "dep_failure"
)

Blocker kinds.

View Source 
const (
	DepKindHard = "hard"
	DepKindSoft = "soft"

	OnFailureBlock  = "block"
	OnFailureFail   = "fail"
	OnFailureIgnore = "ignore"
)

Dep edge kinds and failure policies.

View Source 
const (
	ActorSystem      = "system"
	ActorUser        = "user"
	ActorAgent       = "agent"
	ActorWorker      = "worker"
	ActorReviewer    = "reviewer"
	ActorPlanner     = "planner"
	ActorSynthesizer = "synthesizer"
)

Actor types written to agent_task_event. Schema CHECK mirrors run kinds for the non-human originators.

View Source 
const (
	RunKindWorker      = "worker"
	RunKindReviewer    = "reviewer"
	RunKindPlanner     = "planner"
	RunKindSynthesizer = "synthesizer"
)

Run kinds. Schema CHECK already permits all four; dispatchers for reviewer / planner / synthesizer are introduced in later slices.

View Source 
const HeartbeatInterval = 20 * time.Second

HeartbeatInterval is how often the worker extends lease_expires_at on the run row while the runner is executing. Set to 0 to disable heartbeats.

View Source 
const LeaseDuration = 90 * time.Second

LeaseDuration is the default lease applied while a run is in flight. Must be > 3 * HeartbeatInterval so a single missed beat doesn't expire the lease.

View Source 
const StatusReviewing = "reviewing"

Slice 2 widens task status with this value.

Variables

View Source 
var (
	ErrReviewNotFound = errors.New("tasks: review not found")
	ErrReviewClosed   = errors.New("tasks: review already resolved")
	// ErrUnsupportedReviewPolicy marks a goal review_policy that no runtime in
	// this build can service. Goals only support 'none'; goal-level review
	// (auto/agent/human) needs the unwired synthesizer/goal-review runtime.
	// The API maps this to a 400 validation error.
	ErrUnsupportedReviewPolicy = errors.New("tasks: unsupported review_policy")
)

Review-side typed errors. Callers branch on these.

View Source 
var (
	// ErrInvalidTransition is returned when a transition's from-status does not
	// match the row's current status. This usually means another tick or
	// transition raced this one; the caller can re-fetch and retry.
	ErrInvalidTransition = errors.New("tasks: invalid status transition")

	// ErrTaskNotFound is returned when the target row no longer exists.
	ErrTaskNotFound = errors.New("tasks: task not found")

	// ErrBlockerNotFound is returned when a referenced blocker row is missing.
	ErrBlockerNotFound = errors.New("tasks: blocker not found")

	// ErrCycle is returned by AddDep when accepting the new edge would close a
	// cycle in the DAG.
	ErrCycle = errors.New("tasks: dependency cycle")

	// ErrRetryBudgetExhausted is returned by Fail when retry_count has reached
	// max_retries; the task is moved to StatusFailed instead of StatusReady.
	// Callers do not need to handle this differently — it is informational.
	ErrRetryBudgetExhausted = errors.New("tasks: retry budget exhausted")

	// ErrDepFailureUnresolved is returned when a hard dep edge's upstream is
	// failed/cancelled with on_failure=block and the edge has no waiver. The
	// caller must waive the edge via WaiveDep before the downstream can
	// progress.
	ErrDepFailureUnresolved = errors.New("tasks: dep failure requires waiver")

	// ErrAlreadyClosed is returned when attempting to resolve a blocker that
	// is not in the 'open' state.
	ErrAlreadyClosed = errors.New("tasks: blocker is not open")
)

Typed errors that the transition service returns. Callers branch on these, never on string Contains.

View Source 
var ErrGoalNotFound = errors.New("tasks: goal not found")

ErrGoalNotFound is returned by goal transitions when the row is missing.

View Source 
var ErrInvalidTaskContext = errors.New("tasks: invalid task context")

ErrInvalidTaskContext marks invalid task/goal ownership context at the write boundary.

Functions

func IsActiveRunStatus added in v0.38.0 

func IsActiveRunStatus(s string) bool

IsActiveRunStatus reports whether a run is still claiming an active slot.

func IsConflict added in v0.38.0 

func IsConflict(err error) bool

IsConflict reports whether err is an ErrReopenConflict.

func IsTerminalStatus added in v0.38.0 

func IsTerminalStatus(s string) bool

IsTerminalStatus reports whether the task status is terminal (no further transitions until reopen).

Types

type Actor added in v0.38.0 

type Actor struct {
	Type string // one of ActorSystem | ActorUser | ActorAgent | ActorWorker
	ID   string // user_id / agent_id / worker_id depending on Type; empty for system
}

Actor describes who initiated a transition. Slotted into agent_task_event.

func SystemActor added in v0.38.0 

func SystemActor() Actor

SystemActor is a convenience for transitions originating from the dispatcher or migration paths.

type BlockParams added in v0.38.0 

type BlockParams struct {
	TaskID   string
	Kind     string
	Question string
	Detail   string
	RunID    string // run that triggered the block; empty for system-created
	Actor    Actor
}

BlockParams captures the fields needed to record a new blocker.

type BlockerResult added in v0.40.0 

type BlockerResult struct {
	Kind     string
	Question string
	Detail   any
}

BlockerResult carries a block action's payload.

type BootConfig added in v0.38.0 

type BootConfig struct {
	DB       *sql.DB
	Memory   memory.Provider      // used to mint sessions when Services is nil
	Services agent.ServiceManager // registry-backed session minting
	Chat     TaskChatFunc         // runs persisted worker turns; nil falls back to the noop executor
	// MaxWorkers, TickEvery, LeaseTTL override defaults; zero values use the
	// dispatcher's defaults.
	MaxWorkers int
	TickEvery  time.Duration
	LeaseTTL   time.Duration
	Logger     *slog.Logger
}

BootConfig is the minimal wiring needed at server start.

type ClaimParams added in v0.38.0 

type ClaimParams struct {
	TaskID          string
	ExecutorAgentID string
	WorkerID        string
	LeaseDuration   time.Duration // 0 => no lease
	SessionID       string        // ignored; task.session_id is the source of truth
	Actor           Actor
	// HintID, when set, identifies the live dispatch hint the dispatcher used
	// to resolve ExecutorAgentID. Claim consumes that exact hint inside its
	// tx so the resolved executor and the consumed hint can't diverge under
	// concurrent hint replacement. Empty => no hint to consume.
	HintID string
}

ClaimParams are the inputs the dispatcher provides when it has decided a task is dispatchable.

type ClaimResult added in v0.38.0 

type ClaimResult struct {
	RunID     string
	SessionID string
}

ClaimResult describes the run that was minted by Claim. SessionID is the task's durable worker session.

type CreateGoalInput added in v0.38.0 

type CreateGoalInput struct {
	UserID       string
	AgentID      string
	ProjectID    string
	Title        string
	Description  string
	Priority     string
	ReviewPolicy string
	Context      string
}

CreateGoalInput is the request body for CreateGoal. UserID/Title are required; everything else carries safe defaults.

type CreateTaskInput added in v0.38.0 

type CreateTaskInput struct {
	UserID           string // required
	Title            string // required
	Description      string
	Priority         string // routine | urgent; "" => routine
	AgentID          string // owner/manager agent context; required unless inherited from goal
	GoalID           string // parent goal; optional
	ProjectID        string // project/workspace context; optional
	ExecutorAgentID  string // explicit override (D13); optional, written as dispatch hint
	Required         *bool  // nil => true (default); explicit false => optional task
	MaxRetries       int64
	NotBefore        time.Time // zero => null
	DeadlineAt       time.Time // zero => null
	Context          string    // JSON; "" => "{}"
	Deps             []DepInput
	ActivateOnCreate bool // if true, transition from draft to ready before returning
}

CreateTaskInput is the request body for CreateTask. Fields are optional except where noted.

type DepEdgeView added in v0.38.0 

type DepEdgeView struct {
	DepTaskID      string
	Kind           string // DepKindHard | DepKindSoft
	OnFailure      string // OnFailureBlock | OnFailureFail | OnFailureIgnore
	Waived         bool
	UpstreamStatus string
}

DepEdgeView pairs an edge with the upstream task's current status, plus the fields needed to apply D11's rules without further DB hits.

type DepInput added in v0.38.0 

type DepInput struct {
	DepTaskID string
	Kind      string // hard|soft; "" => hard
	OnFailure string // block|fail|ignore; "" => block
}

DepInput captures one outgoing edge to be created with the task.

type Dispatcher added in v0.38.0 

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

Dispatcher drives the task system: it interrupts stale runs, propagates dep failures, scans dispatchable tasks, claims them, and spawns workers.

func NewDispatcher added in v0.38.0 

func NewDispatcher(cfg DispatcherConfig) *Dispatcher

NewDispatcher constructs a dispatcher. Defaults applied for zero-valued config fields.

func (*Dispatcher) Start added in v0.38.0 

func (d *Dispatcher) Start(ctx context.Context, sched SchedulerLike) error

Start registers the dispatcher tick on the given scheduler. If sched is nil the dispatcher is silent — callers must call Tick directly (used by tests).

func (*Dispatcher) Stop added in v0.38.0 

func (d *Dispatcher) Stop()

Stop waits for all in-flight workers to drain. The dispatcher tick stops firing when its scheduler is stopped (the scheduler owns the lifecycle).

func (*Dispatcher) Tick added in v0.38.0 

func (d *Dispatcher) Tick(ctx context.Context)

Tick runs one pass of the dispatch loop. Public so tests can drive it deterministically.

func (*Dispatcher) WaitIdle added in v0.38.0 

func (d *Dispatcher) WaitIdle()

WaitIdle blocks until no workers are in flight. Useful for tests.

type DispatcherConfig added in v0.38.0 

type DispatcherConfig struct {
	Service    *TransitionService
	Queries    *sqlc.Queries
	Executor   Executor
	Resolver   ExecutorResolver
	NewSession SessionMinter
	TickEvery  time.Duration // 0 => 2s
	MaxWorkers int           // 0 => 5
	LeaseTTL   time.Duration // 0 => LeaseDuration
	BatchLimit int           // 0 => 50
	Logger     *slog.Logger
}

DispatcherConfig holds the wiring for a Dispatcher.

type ErrReopenConflict added in v0.38.0 

type ErrReopenConflict struct {
	DownstreamIDs []string
}

ErrReopenConflict is returned when reopening a task without cascade would strand downstream tasks in an inconsistent state. The error carries the IDs of the conflicting downstream tasks for surfacing in handlers.

func (*ErrReopenConflict) Error added in v0.38.0 

func (e *ErrReopenConflict) Error() string

type Executor added in v0.40.0 

type Executor interface {
	Execute(ctx context.Context, req Request) (Result, error)
}

Executor owns agent interaction and the terminal protocol for one run. It must not mutate durable terminal task/run state — the worker applies the single transition implied by Result. Progress updates may persist during execution (D3).

type ExecutorResolver added in v0.38.0 

type ExecutorResolver func(ctx context.Context, task sqlc.AgentTask) (agentID string, ok bool)

ExecutorResolver decides which agent should execute a given run. Resolution per D13: live dispatch hint -> latest-run executor -> owner agent fallback. The function may return ("", false) if it cannot resolve; the dispatcher will refuse the claim and emit a protocol_error event.

type FailParams added in v0.38.0 

type FailParams struct {
	TaskID          string
	RunID           string
	Reason          string // free-form error message
	Retryable       bool   // if false, force terminal failure
	RunStatusOnFail string // override final run status; "" => RunFailed
	Actor           Actor
}

FailParams captures a worker-or-system-declared run failure.

type FailureResult added in v0.40.0 

type FailureResult struct {
	Reason    string
	Retryable bool
}

FailureResult carries a fail action's payload.

type GoalNextState added in v0.38.0 

type GoalNextState struct {
	// NextStatus is the status the goal should transition to. Empty when no
	// transition applies (rollup is a no-op).
	NextStatus string

	// SpawnSynthesizer is set when a 'running' goal has met its required-done
	// threshold and the policy demands a synthesis pass before completion.
	SpawnSynthesizer bool

	// Reason carries a short human-readable cause; surfaced on the event row.
	Reason string
}

GoalNextState is the outcome of one rollup evaluation.

func RollupGoal added in v0.38.0 

func RollupGoal(goal sqlc.AgentGoal, counts sqlc.GoalChildCountsRow, hasOpenSynth bool) GoalNextState

RollupGoal applies the goal-rollup decision table (plan D2) to a single goal. Pure function over (goal row, child counts, whether a synthesizer run is already in flight). Caller (dispatcher.rollupGoals) translates the outcome into a transition call.

type Readiness added in v0.38.0 

type Readiness struct {
	State        string
	Dispatchable bool
	Reasons      []Reason
}

Readiness is the computed dispatchability view over a task at a moment in time. Pure function over (task row, dep edges with upstream status, now).

func Compute added in v0.38.0 

func Compute(task sqlc.AgentTask, deps []DepEdgeView, now time.Time) Readiness

Compute returns the readiness of task given its dep edges (with upstream status pre-joined) at time now. It is a pure function: no DB calls, no side effects, no clock reads beyond `now`.

The order of checks below mirrors the readiness flowchart in plan.md section 04 — first short-circuit on terminal / non-ready statuses, then evaluate deps, then deferred/throttle gates.

type Reason added in v0.38.0 

type Reason struct {
	Type       string // dependency_not_done, dep_failed, not_before, etc.
	UpstreamID string // for dep-related reasons
	Detail     string
}

Reason describes why a task is in a given readiness state. Surfaced via the readiness API so callers can explain "why isn't this running?".

type Request added in v0.40.0 

type Request struct {
	Run  sqlc.AgentTaskRun
	Task *sqlc.AgentTask
}

Request is the executor input for one claimed run. Task is required for worker runs; it carries the prompt body and review policy.

type Result added in v0.40.0 

type Result struct {
	Action  TerminalAction
	Output  any
	Blocker *BlockerResult
	Failure *FailureResult
	// RepairAttempted is set when a text-only first turn triggered one bounded
	// repair turn that still produced no terminal action. It only carries
	// meaning for TerminalNone and lets the worker distinguish a silent miss
	// from a failed repair in the protocol_error detail.
	RepairAttempted bool
}

Result is the executor output: exactly one terminal action (or TerminalNone when the agent ended without declaring one). The worker reads this and applies the matching transition through TransitionService.

type SchedulerLike added in v0.38.0 

type SchedulerLike interface {
	ScheduleEvery(ctx context.Context, every string, fn func(ctx context.Context)) error
}

SchedulerLike is the subset of scheduler.Service the dispatcher needs. Allows tests to drive ticks directly without wiring gocron.

type Service 

type Service struct {
	Queries    *sqlc.Queries
	Transition *TransitionService
	Facade     *ServiceFacade
	Dispatcher *Dispatcher
}

Service bundles the v2 task system components for boot wiring.

func New 

func New(cfg BootConfig) *Service

New constructs the task system. The dispatcher is constructed but not started; the caller registers it on a scheduler via dispatcher.Start.

If BootConfig.Chat is non-nil, the dispatcher uses the agent-backed worker executor. Otherwise it falls back to a noop executor that fails with a clear message.

type ServiceFacade added in v0.38.0 

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

ServiceFacade is the v2 Go-level surface that HTTP handlers + CLI bind to. Phase 5 deliverable. The OpenAPI rewrite (flat /api/tasks routes, regen of types/server/client) is deliberately deferred — see plan.md Phase 5 handoff. The current admin HTTP handlers continue to return 503 (MP1 stub) while this facade is exercised programmatically (boot wiring + tests).

All mutating methods route through TransitionService so the status-write-only-via-transition invariant holds.

func NewServiceFacade added in v0.38.0 

func NewServiceFacade(db *sql.DB, q *sqlc.Queries, svc *TransitionService, newSession SessionMinter) *ServiceFacade

NewServiceFacade builds the facade.

func (*ServiceFacade) ActivateGoal added in v0.38.0 

func (f *ServiceFacade) ActivateGoal(ctx context.Context, goalID string, actor Actor) error

ActivateGoal / CancelGoal are thin shims over TransitionService.

func (*ServiceFacade) AddDep added in v0.38.0 

func (f *ServiceFacade) AddDep(ctx context.Context, taskID, depTaskID, kind, onFailure string) error

AddDep is a thin shim over TransitionService.AddDep for caller ergonomics.

func (*ServiceFacade) ApproveReview added in v0.38.0 

func (f *ServiceFacade) ApproveReview(ctx context.Context, reviewID, summary string, actor Actor) error

ApproveReview exposes TransitionService.ApproveReview.

func (*ServiceFacade) CancelGoal added in v0.38.0 

func (f *ServiceFacade) CancelGoal(ctx context.Context, goalID, reason string, actor Actor) error

func (*ServiceFacade) CancelTask added in v0.38.0 

func (f *ServiceFacade) CancelTask(ctx context.Context, taskID, reason string, actor Actor) error

CancelTask exposes TransitionService.Cancel.

func (*ServiceFacade) CreateGoal added in v0.38.0 

func (f *ServiceFacade) CreateGoal(ctx context.Context, in CreateGoalInput) (sqlc.AgentGoal, error)

CreateGoal inserts an agent_goal row in 'draft'. Goal planning/synthesis is not dispatched in this release; callers create child tasks explicitly.

func (*ServiceFacade) CreateTask added in v0.38.0 

func (f *ServiceFacade) CreateTask(ctx context.Context, in CreateTaskInput) (sqlc.AgentTask, error)

CreateTask creates a task in 'draft', optionally activates it, optionally writes a dispatch hint, and adds dependency edges with cycle checking.

func (*ServiceFacade) EscalateReview added in v0.38.0 

func (f *ServiceFacade) EscalateReview(ctx context.Context, reviewID, reason string, actor Actor) error

EscalateReview exposes TransitionService.EscalateReview.

func (*ServiceFacade) GetBlocker added in v0.38.0 

func (f *ServiceFacade) GetBlocker(ctx context.Context, blockerID string) (sqlc.AgentTaskBlocker, error)

ListBlockerForTask returns the open blocker (if any) for a task. Phase-2 handlers use this to translate {blockerID} path params into the underlying blocker row before resolving.

func (*ServiceFacade) GetGoal added in v0.38.0 

func (f *ServiceFacade) GetGoal(ctx context.Context, goalID string) (sqlc.AgentGoal, error)

GetGoal returns one goal by ID.

func (*ServiceFacade) GetReadiness added in v0.38.0 

func (f *ServiceFacade) GetReadiness(ctx context.Context, taskID string) (Readiness, error)

GetReadiness loads the task + its deps and returns the computed view.

func (*ServiceFacade) GetReview added in v0.38.0 

func (f *ServiceFacade) GetReview(ctx context.Context, reviewID string) (sqlc.AgentReview, error)

GetReview returns a review by id. Phase-2 handlers use this to enforce task<->review parentage before applying a decision.

func (*ServiceFacade) GetTask added in v0.38.0 

func (f *ServiceFacade) GetTask(ctx context.Context, taskID string) (sqlc.AgentTask, error)

GetTask returns a task by id.

func (*ServiceFacade) ListDeps added in v0.38.0 

func (f *ServiceFacade) ListDeps(ctx context.Context, taskID string, limit, offset int64) ([]sqlc.ListAgentTaskDepsWithUpstreamPagedRow, error)

ListDeps returns the dep edges (with upstream status) for a task.

func (*ServiceFacade) ListEvents added in v0.38.0 

func (f *ServiceFacade) ListEvents(ctx context.Context, taskID string, limit, offset int64) ([]sqlc.AgentTaskEvent, error)

ListEvents returns the audit trail for a task, oldest first.

func (*ServiceFacade) ListGoalReviews added in v0.38.0 

func (f *ServiceFacade) ListGoalReviews(ctx context.Context, goalID string, limit, offset int64) ([]sqlc.AgentReview, error)

ListGoalReviews lists reviews for a goal.

func (*ServiceFacade) ListGoalTasks added in v0.38.0 

func (f *ServiceFacade) ListGoalTasks(ctx context.Context, goalID string, limit, offset int64) ([]sqlc.AgentTask, error)

ListGoalTasks lists child tasks of a goal.

func (*ServiceFacade) ListGoals added in v0.38.0 

func (f *ServiceFacade) ListGoals(ctx context.Context, userID string, limit, offset int64) ([]sqlc.AgentGoal, error)

ListGoals returns goals owned by the given user, newest first.

func (*ServiceFacade) ListGoalsByAgent added in v0.41.3 

func (f *ServiceFacade) ListGoalsByAgent(ctx context.Context, userID string, agentID string, limit, offset int64) ([]sqlc.AgentGoal, error)

ListGoalsByAgent returns goals owned by the given user and agent, newest first.

func (*ServiceFacade) ListReviews added in v0.38.0 

func (f *ServiceFacade) ListReviews(ctx context.Context, taskID string, limit, offset int64) ([]sqlc.AgentReview, error)

ListReviews returns the review history for a task, newest first.

func (*ServiceFacade) ListRuns added in v0.38.0 

func (f *ServiceFacade) ListRuns(ctx context.Context, taskID string, limit, offset int64) ([]sqlc.AgentTaskRun, error)

ListRuns returns runs for a task, newest attempt first.

func (*ServiceFacade) ListTasksByUser added in v0.38.0 

func (f *ServiceFacade) ListTasksByUser(ctx context.Context, userID, agentID, projectID, status string, limit, offset int64) ([]sqlc.AgentTask, error)

ListTasksByUser returns paginated tasks owned by the given user, optionally filtered by agent, project, and status. Empty filter strings match all rows.

func (*ServiceFacade) RejectReview added in v0.38.0 

func (f *ServiceFacade) RejectReview(ctx context.Context, reviewID, summary, feedback string, actor Actor) error

RejectReview exposes TransitionService.RejectReview.

func (*ServiceFacade) ReopenTask added in v0.38.0 

func (f *ServiceFacade) ReopenTask(ctx context.Context, taskID string, cascade bool, actor Actor) error

ReopenTask exposes TransitionService.ReopenTask. Cascade applies the downstream reset rules in D10.

func (*ServiceFacade) RequestChanges added in v0.38.0 

func (f *ServiceFacade) RequestChanges(ctx context.Context, reviewID, summary, feedback string, actor Actor) error

RequestChanges exposes TransitionService.RequestChanges.

func (*ServiceFacade) ResolveBlocker added in v0.38.0 

func (f *ServiceFacade) ResolveBlocker(ctx context.Context, blockerID, resolution string, actor Actor) error

ResolveBlocker exposes TransitionService.ResolveBlocker. dep_failure blockers return ErrDepFailureUnresolved per D14 / M1.

func (*ServiceFacade) WaiveDep added in v0.38.0 

func (f *ServiceFacade) WaiveDep(ctx context.Context, taskID, depTaskID, userID, reason string, actor Actor) error

WaiveDep exposes TransitionService.WaiveDep (D14 / M1 — required for dep_failure blockers).

type SessionMinter added in v0.38.0 

type SessionMinter func(ctx context.Context, userID, agentID, projectID string) (string, error)

SessionMinter returns a fresh durable worker session id for a task. It must produce a unique id per call and scope the session to the resolved agent and optional project context.

type TaskChatFunc added in v0.46.0 

type TaskChatFunc func(ctx context.Context, p TaskChatParams) <-chan agent.Event

TaskChatFunc runs one worker turn through the agent service layer so the transcript persists to the task session and prior turns load as history. Implementations resolve AgentID to a service; an unknown agent surfaces as an Err event on the returned channel.

type TaskChatParams added in v0.46.0 

type TaskChatParams struct {
	AgentID    string
	UserID     string
	SessionID  string
	ProjectID  string
	Prompt     string
	ExtraTools []tools.Tool
}

TaskChatParams describes one persisted worker turn on a task session.

type TerminalAction added in v0.40.0 

type TerminalAction string

TerminalAction is the durable outcome an executor reports for one run. The worker applies exactly one transition based on it; the agent never mutates final task state directly (D3). 

const (
	TerminalNone   TerminalAction = ""
	TerminalSubmit TerminalAction = "submit"
	TerminalBlock  TerminalAction = "block"
	TerminalFail   TerminalAction = "fail"
)

type TransitionService added in v0.38.0 

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

TransitionService is the single entry point for every state change on agent_task / agent_task_run / agent_task_blocker / agent_task_dep rows. Every method runs in one transaction: load → validate from→to → side effects → update → append event → commit. Callers branch on the typed errors declared in types.go.

func NewTransitionService added in v0.38.0 

func NewTransitionService(db *sql.DB, q *sqlc.Queries) *TransitionService

NewTransitionService constructs a transition service from a *sql.DB and a pre-built *sqlc.Queries. The Queries is used only for non-transactional reads; mutating methods open their own txns and call WithTx.

func (*TransitionService) Activate added in v0.38.0 

func (s *TransitionService) Activate(ctx context.Context, taskID string, actor Actor) error

Activate moves StatusDraft → StatusReady.

func (*TransitionService) ActivateGoal added in v0.38.0 

func (s *TransitionService) ActivateGoal(ctx context.Context, goalID string, actor Actor) error

ActivateGoal moves a goal from draft to running and transitions every draft child task to ready in the same transaction. The dispatcher picks the children up next tick.

func (*TransitionService) ActivateInTx added in v0.38.0 

func (s *TransitionService) ActivateInTx(ctx context.Context, q *sqlc.Queries, taskID string, actor Actor) error

ActivateInTx is the exported form of activateInTx for use inside WithTx.

func (*TransitionService) AddDep added in v0.38.0 

func (s *TransitionService) AddDep(ctx context.Context, taskID, depTaskID, depKind, onFailure string) error

AddDep inserts a dep edge after a DFS cycle check. depKind and onFailure default to "hard" / "block" when blank.

func (*TransitionService) AddDepInTx added in v0.38.0 

func (s *TransitionService) AddDepInTx(ctx context.Context, q *sqlc.Queries, taskID, depTaskID, depKind, onFailure string) error

AddDepInTx is the exported form of addDepInTx for use inside WithTx.

func (*TransitionService) ApproveReview added in v0.38.0 

func (s *TransitionService) ApproveReview(ctx context.Context, reviewID, summary string, actor Actor) error

ApproveReview closes a review with 'approved'. The caller doesn't need to know whether the parent is a task or a goal; this dispatcher branches on the row's parent column and routes to the right inner transition.

func (*TransitionService) Block added in v0.38.0 

func (s *TransitionService) Block(ctx context.Context, p BlockParams) error

Block transitions a task to StatusBlocked and inserts a blocker row. If the task already has an open blocker, the new condition is merged into the existing blocker's detail instead of inserting a second open row (D14 / H4).

func (*TransitionService) BlockGoal added in v0.38.0 

func (s *TransitionService) BlockGoal(ctx context.Context, goalID, reason string, actor Actor) error

BlockGoal mirrors a child's blocked state on the goal. Used by rollup when a required child is blocked. No goal-level blocker row is created (D2 keeps the goal as a thin container); the audit event records the cause.

func (*TransitionService) Cancel added in v0.38.0 

func (s *TransitionService) Cancel(ctx context.Context, taskID, reason string, actor Actor) error

Cancel moves any non-terminal task to StatusCancelled. Idempotent: a second Cancel on an already-cancelled task returns nil. The dispatcher tick also cancels the active run if one exists.

func (*TransitionService) CancelGoal added in v0.38.0 

func (s *TransitionService) CancelGoal(ctx context.Context, goalID, reason string, actor Actor) error

CancelGoal cancels a non-terminal goal and cascade-cancels every non-terminal child task. Active runs / blockers / reviews on each child are finalized so the database invariants (active_*_id consistency) hold post-cancel.

func (*TransitionService) Claim added in v0.38.0 

func (s *TransitionService) Claim(ctx context.Context, p ClaimParams) (ClaimResult, error)

Claim atomically moves StatusReady → StatusRunning AND inserts a new agent_task_run row, setting task.active_run_id in the same transaction. Returns ErrInvalidTransition if the row's status changed underfoot — the caller (dispatcher) should re-scan candidates and try again.

Session-id rule: task.session_id is required at creation time and reused for every run.

func (*TransitionService) CompleteGoal added in v0.38.0 

func (s *TransitionService) CompleteGoal(ctx context.Context, goalID, output string, actor Actor) error

CompleteGoal moves a non-terminal goal to done and stamps completed_at. Used by goal rollup (all required children done + review policy = none) and by review-decision approval.

func (*TransitionService) CompleteGoalTx added in v0.38.0 

func (s *TransitionService) CompleteGoalTx(ctx context.Context, q *sqlc.Queries, goalID string, actor Actor) error

CompleteGoalTx runs CompleteGoal's body in an existing transaction. Used by the dispatcher rollup loop, which holds the tx open across multiple goals.

func (*TransitionService) EscalateReview added in v0.38.0 

func (s *TransitionService) EscalateReview(ctx context.Context, reviewID, reason string, actor Actor) error

EscalateReview closes an agent review with 'escalated' and inserts a fresh review row with reviewer_type='human' + escalated_from_review_id pointing back. Task stays in 'reviewing' under the new active_review_id (D8).

func (*TransitionService) Fail added in v0.38.0 

func (s *TransitionService) Fail(ctx context.Context, p FailParams) error

Fail records a run failure and either returns the task to StatusReady (if the retry budget allows and Retryable is true) or moves it to StatusFailed.

func (*TransitionService) FailGoal added in v0.38.0 

func (s *TransitionService) FailGoal(ctx context.Context, goalID, reason string, actor Actor) error

FailGoal marks a non-terminal goal as failed. Active review is cleared.

func (*TransitionService) RejectReview added in v0.38.0 

func (s *TransitionService) RejectReview(ctx context.Context, reviewID, summary, feedback string, actor Actor) error

RejectReview closes a review with 'rejected'. Task parent → task failed, goal parent → goal failed.

func (*TransitionService) ReopenTask added in v0.38.0 

func (s *TransitionService) ReopenTask(ctx context.Context, taskID string, cascade bool, actor Actor) error

ReopenTask transitions a done/failed task back to ready. Per D10:

  • Without cascade: if any reachable downstream task is in a non-terminal non-cancelled state OR done, return ErrReopenConflict listing them.
  • With cascade=true: reset reachable downstream per ownership: standalone (goal_id IS NULL) -> ready goal-owned under draft/planning goal -> draft goal-owned under any other goal status -> ready

All transitions land in a single tx so partial-cascade failures don't leave rows in inconsistent states.

func (*TransitionService) RequestChanges added in v0.38.0 

func (s *TransitionService) RequestChanges(ctx context.Context, reviewID, summary, feedback string, actor Actor) error

RequestChanges closes a review with 'changes_requested'. On a task review the task either bounces back to 'ready' (with retry budget consumed) or transitions to 'failed'. On a goal review there is no retry budget — the goal goes to 'failed' with a documented gap (no synthesizer retry yet).

func (*TransitionService) ResolveBlocker added in v0.38.0 

func (s *TransitionService) ResolveBlocker(ctx context.Context, blockerID, resolution string, actor Actor) error

ResolveBlocker closes an open blocker and moves its task back to StatusReady. dep_failure blockers cannot be resolved through this path — callers must use WaiveDep (D14 / M1); we surface this with ErrDepFailureUnresolved so the handler can return 409 with the right message.

func (*TransitionService) SetClock added in v0.38.0 

func (s *TransitionService) SetClock(c func() time.Time)

SetClock overrides the clock for tests.

func (*TransitionService) Submit added in v0.38.0 

func (s *TransitionService) Submit(ctx context.Context, taskID, runID, output string, actor Actor) error

Submit handles a worker's submit action, routing on the task's review_policy:

  • none: task -> done immediately (no review row, no event)
  • auto: insert a system-approved agent_review for audit, task -> done
  • agent: insert a 'requested' agent_review (reviewer_type=agent), task -> reviewing; dispatcher picks up reviewer run
  • human: insert a 'requested' agent_review (reviewer_type=human), task -> reviewing; awaits human decision via API

Single source of truth for submit semantics; worker / control tool / API callers all go through here.

func (*TransitionService) UnblockGoal added in v0.40.0 

func (s *TransitionService) UnblockGoal(ctx context.Context, goalID, reason string, actor Actor) error

UnblockGoal recovers a goal from blocked or failed back to running once no required child is blocked or failed. Mirror of BlockGoal; driven by rollup when a child blocker is resolved, a failed child is reopened/completed, or a failed dependency is waived. The goal then resumes normal rollup (completing or re-blocking) on subsequent ticks.

func (*TransitionService) WaiveDep added in v0.38.0 

func (s *TransitionService) WaiveDep(ctx context.Context, taskID, depTaskID, userID, reason string, actor Actor) error

WaiveDep records a waiver on a hard-dep edge whose upstream failed or cancelled. After waiver, readiness treats the edge as satisfied; the downstream's dep_failure blocker (if any) is also resolved in the same tx.

func (*TransitionService) WithTx added in v0.38.0 

func (s *TransitionService) WithTx(ctx context.Context, fn func(*sqlc.Queries) error) error

WithTx exposes the transition service's transaction helper so higher-level composers (e.g. ServiceFacade.CreateTask) can group several mutations into one atomic unit while still routing status writes through tx-aware helpers.

type Worker added in v0.38.0 

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

Worker runs one claimed agent_task_run to completion.

func NewWorker added in v0.38.0 

func NewWorker(svc *TransitionService, q *sqlc.Queries, exec Executor) *Worker

NewWorker wires a worker.

func (*Worker) Run added in v0.38.0 

func (w *Worker) Run(ctx context.Context, taskID, runID string, actor Actor) (err error)

Run drives the executor for one claimed task. Responsibilities:

  • flip the run from queued to running with started_at + initial lease
  • keep lease_expires_at fresh while the executor runs (heartbeat)
  • apply exactly one terminal transition (submit/block/fail) from the executor's Result at the worker boundary (D3)
  • apply the protocol-error fallback when the executor reports no terminal action (HP5 / D14)
  • turn executor panics into a non-retryable Fail

func (*Worker) SetHeartbeat added in v0.38.0 

func (w *Worker) SetHeartbeat(d time.Duration)

SetHeartbeat overrides the heartbeat interval (for tests).

func (*Worker) SetLease added in v0.38.0 

func (w *Worker) SetLease(d time.Duration)

SetLease overrides the lease duration (for tests).

Source Files

View all Source files

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.