tasks

package
v0.38.3 Latest Latest
Warning

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

 Go to latest Published: Jun 1, 2026 License: AGPL-3.0 Imports: 16 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")
)

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.

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 BootConfig added in v0.38.0 

type BootConfig struct {
	DB     *sql.DB
	Memory memory.Provider // used to mint sessions
	Pools  func(agentID string) (agent.NewRunnerFunc, bool)
	// 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
	NewSessionID    string        // used when task.session_id is null
	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 session the worker should run in; on a first claim the dispatcher minted a fresh one, on a retry it's the task's persisted session.

type CreateGoalInput added in v0.38.0 

type CreateGoalInput struct {
	UserID       string
	AgentID      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 // creator agent (D12); 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
	Runner     RunnerFunc
	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 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 -> session-derived -> creator 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 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 PoolAdapter added in v0.38.0 

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

PoolAdapter wires the dispatcher's RunnerFunc to the real agent.Pool. It:

  1. Looks up a Runner factory for the run's executor agent.
  2. Builds a `task_control` Tool that delegates to TaskControlTool.
  3. Loads the task description as the user message.
  4. Pumps Runner.Chat events until the runner finishes OR the tool fires a terminal action (submit/block/fail).

Session bootstrap: the dispatcher's SessionMinter handed us run.session_id. The Runner is given that session_id directly; memory.Provider Bootstrap is the runner's own responsibility when it starts (it's idempotent).

The control_tool's Finished flag is the primary success signal. If the agent loop exits without calling the tool, the worker's protocol fallback (Phase 3) records a `protocol_error` event and Fails the run.

func NewPoolAdapter added in v0.38.0 

func NewPoolAdapter(pools PoolLookup, mem memory.Provider, log *slog.Logger) *PoolAdapter

NewPoolAdapter builds an adapter usable as a RunnerFunc.

func (*PoolAdapter) AsRunnerFunc added in v0.38.0 

func (a *PoolAdapter) AsRunnerFunc(q *sqlc.Queries) RunnerFunc

AsRunnerFunc returns the dispatch hook. Wire this into BootConfig.Runner.

type PoolLookup added in v0.38.0 

type PoolLookup func(agentID string) (agent.NewRunnerFunc, bool)

PoolLookup resolves an agent ID to a Runner factory. Returned (nil, false) means no pool is available for that agent; the adapter will surface this as a non-retryable Fail.

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 RunnerFunc added in v0.38.0 

type RunnerFunc func(ctx context.Context, run sqlc.AgentTaskRun, tool *TaskControlTool) error

RunnerFunc is the agent-execution callback that the worker invokes for a claimed run. The runner must call exactly one of tool.Submit / tool.Block / tool.Fail before returning; otherwise the worker applies the protocol-error fallback per D14 / HP5.

The runner runs inside the dispatcher's worker goroutine and must honor ctx cancellation (e.g. on dispatcher shutdown).

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.Pools is non-nil, the dispatcher uses PoolAdapter to drive real agent.Runner instances. Otherwise it falls back to a noop runner that fails with a clear message — used by tests and by boots that intentionally skip agent wiring.

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) *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'. The dispatcher's planner scan will pick it up next tick when status=draft and an executor is resolvable.

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) 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, status string, limit, offset int64) ([]sqlc.AgentTask, error)

ListTasksByUser returns paginated tasks owned by the given user, optionally filtered by agent 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, task sqlc.AgentTask) (string, error)

SessionMinter returns a fresh session id for a first-run dispatch (i.e. task.session_id IS NULL). It must produce a unique id per call. In production this is wired to memory.Provider.NewSession; in tests it can be a simple uuid generator.

type TaskControlTool 

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

TaskControlTool is the surface a worker (agent runner) sees to communicate task lifecycle decisions. Each method routes through TransitionService so status writes obey the single-source-of-truth invariant (D14).

All four methods are idempotent within reason: a second Submit returns ErrInvalidTransition cleanly, callers can branch on it.

func NewTaskControlTool added in v0.38.0 

func NewTaskControlTool(svc *TransitionService, q *sqlc.Queries, taskID, runID string, actor Actor) *TaskControlTool

NewTaskControlTool wires a control tool for one claimed run.

func (*TaskControlTool) Block added in v0.38.0 

func (t *TaskControlTool) Block(ctx context.Context, kind, question string, detail any) error

Block creates a blocker and moves the task to StatusBlocked.

func (*TaskControlTool) Fail added in v0.38.0 

func (t *TaskControlTool) Fail(ctx context.Context, reason string, retryable bool) error

Fail marks the run as failed. Retryable=true returns the task to ready (if retry budget remains); false forces StatusFailed.

func (*TaskControlTool) FinalStatus added in v0.38.0 

func (t *TaskControlTool) FinalStatus() string

FinalStatus returns the task status observed right after the terminal action. Empty before any terminal call. Worker-facing callers use this to report the real outcome (e.g. "reviewing" vs "done") instead of assuming.

func (*TaskControlTool) Finished added in v0.38.0 

func (t *TaskControlTool) Finished() bool

Finished reports whether the tool has seen a terminal action (submit, block, fail). The worker uses this to decide whether to apply the protocol-error fallback when the agent loop exits.

func (*TaskControlTool) Progress added in v0.38.0 

func (t *TaskControlTool) Progress(ctx context.Context, patch any) error

Progress merges patch into agent_task.context as a shallow JSON merge (HP6 / D14). Existing keys not in patch are preserved. patch may be raw JSON bytes or a Go map; both produce the same merge.

func (*TaskControlTool) Submit added in v0.38.0 

func (t *TaskControlTool) Submit(ctx context.Context, output any) error

Submit completes the worker run. Per review_policy on the task, the final status may be "done" (none / auto) or "reviewing" (agent / human). Output is stored on agent_task.output. FinalStatus() exposes the resulting status.

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 (D12): if task.session_id is non-null, reuse it; otherwise adopt p.NewSessionID and persist it on the task row.

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) 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, runner RunnerFunc) *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 runner for one claimed task. Responsibilities:

  • flip the run from queued to running with started_at + initial lease
  • keep lease_expires_at fresh while the runner executes (heartbeat)
  • apply the protocol-error fallback if the runner returns without a terminal control action (HP5 / D14)
  • turn runner panics into a non-retryable Fail

Run does not transition the task itself; the control tool does that.

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.