Documentation
¶
Overview ¶
Package executor defines the pluggable execution layer: how a claimed task actually runs.
Index ¶
- Constants
- Variables
- func AgentVersion(ctx context.Context, bin string) (string, error)
- func DetectVerify(repo string) string
- func ExtractResultPayload(output string) (files []string, sha string, ok bool)
- func ExtractSessionID(output string) string
- func Permanent(cause error) error
- func ReadTQVerify(repoDir string) string
- func RenderAgentPayload(p AgentPayload) (json.RawMessage, error)
- func RenderCrushPayload(p AgentPayload) (json.RawMessage, error)
- func ResultLine(output string) (json.RawMessage, error)
- func SetFailureEvidence(ctx context.Context, stage string, err error, tail string)
- func SetResultDetail(ctx context.Context, detail json.RawMessage)
- func SweepSidecars(dir string, maxAge time.Duration) (int, error)
- func SweepSidecarsByBytes(dir string, maxBytes int64) (int, error)
- type AgentExecutor
- type AgentPayload
- type AgentResult
- type CommandExecutor
- type CrushPayload
- type Executor
- type FailureEvidence
- type Func
- type HTTPExecutor
- type LeaseLostError
- type PermanentError
- type PreflightError
- type Registry
- type ReviewExecutor
- type ReviewFinding
- type ReviewPayload
- type ReviewResult
- type ReviewVerdict
- type Sink
- type StatusCompletion
- type StatusExecutor
- type StatusPayload
- type StatusResult
Examples ¶
Constants ¶
const DefaultAgentBinary = "crush"
DefaultAgentBinary is used when Bin, $TQ_AGENT_BIN and $TQ_CRUSH_BIN are all empty.
const EvidenceTailBytes = 4096
EvidenceTailBytes is the ONE output-tail size every executor pins into FailureEvidence.Tail — agent, verify and command tails stay the same length so facts are comparable across executors (01:48 report f23: three independent 4096s, one drift waiting to happen).
const TaskTypeAgent = "agent"
TaskTypeAgent runs a headless AI coding agent (crush) in a repository. Register it under this type to make the queue agent-capable.
const TaskTypeCrush = TaskTypeAgent
TaskTypeCrush is the pre-convergence name of TaskTypeAgent. Deprecated: use TaskTypeAgent.
const TaskTypeReview = "review"
TaskTypeReview runs a reviewer agent over a COMPLETED agent task: a second headless agent inspects the change (item, commit, files) and returns a structured verdict. Register it under this type to make a pool review-capable.
Both verdicts complete the task — a review that requested changes did its job; the findings (in the completion fact detail) drive any follow-up work, not the exit code. Only the mechanical contract is enforced: output that is not a valid verdict JSON is a failed attempt, exactly like a failed verify.
const TaskTypeStatus = "status"
TaskTypeStatus runs the "done prompt": after a window of agent completions, a headless agent writes a full status report (docs/status/<timestamp>.md) and closes the loop by appending the next work items to TODO_LIST.md, where the next harvest tick picks them up. Register it under this type to make a pool status-capable; internal/status mints the tasks.
Variables ¶
var ErrUnknownType = errors.New("executor: unknown task type")
ErrUnknownType is returned by Registry.Lookup for unregistered task types.
Functions ¶
func AgentVersion ¶ added in v0.2.0
AgentVersion probes the agent binary's --version line (first stdout line, trimmed). Pools call it at startup so a missing/stale binary is a warning, not a surprise mid-task.
func DetectVerify ¶ added in v0.2.0
DetectVerify is the exported form of autoDetectVerify for tools that prepare repos (tq bootstrap pins the detected command into .tq-verify).
func ExtractResultPayload ¶ added in v0.2.0
ExtractResultPayload parses the agent's structured self-report ({files_changed, commit_sha}) from its output. Best-effort: no line, no problem — the fields simply stay empty in the result detail.
func ExtractSessionID ¶ added in v0.2.0
ExtractSessionID pulls a crush session id out of agent output. Best-effort by design: crush's output format is not a stable contract.
func Permanent ¶
Permanent wraps cause in a *PermanentError; nil passes through unchanged. A cause that is already permanent is returned as-is, so defensive double wrapping stays a single class.
func ReadTQVerify ¶ added in v0.2.0
ReadTQVerify is the exported form of readTQVerify for tools that compose payloads (the harvester pins the repo's verify contract into tasks).
func RenderAgentPayload ¶
func RenderAgentPayload(p AgentPayload) (json.RawMessage, error)
RenderAgentPayload marshals a payload for tasks of type agent.
func RenderCrushPayload ¶
func RenderCrushPayload(p AgentPayload) (json.RawMessage, error)
RenderCrushPayload is the pre-convergence name of RenderAgentPayload. Deprecated: use RenderAgentPayload.
func ResultLine ¶ added in v0.2.0
func ResultLine(output string) (json.RawMessage, error)
ResultLine extracts the raw JSON of the LAST TQ_RESULT line from agent output. Executors with a mechanical output contract (review, status) build their strict parsing on top of it; ExtractResultPayload is the lenient consumer for plain agent runs.
func SetFailureEvidence ¶ added in v0.2.0
SetFailureEvidence attaches forensics for a FAILING execution; a no-op when the context carries no sink. Mirrors SetResultDetail: executors call it on their failure paths, the worker hands it to Store.Fail so the task.failed fact carries the evidence.
func SetResultDetail ¶ added in v0.2.0
func SetResultDetail(ctx context.Context, detail json.RawMessage)
SetResultDetail attaches outcome detail to the current execution; a no-op when the context carries no sink (plain executors, tests).
func SweepSidecars ¶ added in v0.2.0
SweepSidecars deletes agent-output sidecar logs older than maxAge from dir and returns how many were removed — the retention hatch behind --log-dir-max-age, so a long-running pool's sidecar directory cannot grow forever. Only *.log files directly inside dir are considered; other content is left alone. Errors on individual files are skipped (retention is best effort: a locked file must never fail the pool tick).
func SweepSidecarsByBytes ¶ added in v0.2.0
SweepSidecarsByBytes caps the total size of sidecar logs in dir: when the *.log files together exceed maxBytes, the OLDEST are deleted until the total fits — the retention hatch behind --log-dir-max-bytes (the round-6 retention item's open half). Age-based retention cannot bound a high-traffic dir that stays under --log-dir-max-age; a byte budget does. Only *.log files directly inside dir are considered; other content is left alone, and per-file errors are skipped (best effort, like the age sweep).
Types ¶
type AgentExecutor ¶
type AgentExecutor struct {
// Bin is the agent executable. Default: $TQ_AGENT_BIN, $TQ_CRUSH_BIN or
// "crush". Point it at a stub in tests.
Bin string
// ProjectsDir resolves relative Repo names in payloads ("demo" →
// <ProjectsDir>/demo). Absolute Repo paths bypass it.
ProjectsDir string
// Yolo is the operator-level autonomy request (pool start). crush run
// has no yolo flag; see AgentPayload.Yolo for how autonomy is actually
// granted (repo-local crush config). Fail closed, never silently.
Yolo bool
// MaxConcurrent caps how many agent processes run at once MACHINE-WIDE
// (flock'd slot files shared across every tq process on the host; 0 =
// uncapped). Caps cost when several pools share a machine.
MaxConcurrent int
}
AgentExecutor runs one headless AI coding agent per task, then enforces the payload's verify contract. It must be safe for concurrent use; the worker pool executes several tasks in parallel.
func NewAgentExecutor ¶
func NewAgentExecutor(projectsDir string) *AgentExecutor
NewAgentExecutor builds an AgentExecutor for a projects directory.
func (*AgentExecutor) Execute ¶
Execute guards the repo, runs the agent, then runs the verify command. Any miss is a failed attempt (the queue retries with backoff, then dead-letters). Input-contract misses (payload, repo, dirty tree, autonomy) are permanent: the identical retry would fail identically, and for agent tasks every retry is real money.
type AgentPayload ¶
type AgentPayload struct {
// Repo is the repository the agent works in: a name resolved against the
// executor's ProjectsDir, or an absolute path. Required.
Repo string `json:"repo"`
// Prompt is the full instruction for the agent run. Required.
Prompt string `json:"prompt"`
// Model optionally overrides the crush model ("provider/model").
Model string `json:"model,omitempty"`
// Session optionally continues a previous crush session by ID.
Session string `json:"session,omitempty"`
// Dedup is the harvester's item key; purely informational, used to keep
// TODO items and tasks 1:1 across harvest runs.
Dedup string `json:"dedup,omitempty"`
// Item is the raw TODO_LIST work item the prompt was rendered from;
// purely informational. Status windows pin it so report excerpts show
// the real work instead of the prompt template's first line. Empty for
// tasks minted outside the harvester.
Item string `json:"item,omitempty"`
// Verify is a shell command that must exit 0 after the agent run for the
// task to complete — the enforced quality gate. Empty means auto-detect:
// Go repositories (go.mod present) run "go build ./... && go test ./...
// -count=1", everything else runs nothing.
Verify string `json:"verify,omitempty"`
// RequireClean refuses to start unless the repo's git tree is clean, so
// the pool never tramples human work-in-progress. Default true. Repos
// without a .git directory skip the check (nothing to protect).
RequireClean *bool `json:"require_clean,omitempty"`
// TimeoutMinutes caps the whole task (agent run + verify). Default 30.
// The worker's task timeout still applies as a hard ceiling above this.
TimeoutMinutes int `json:"timeout_minutes,omitempty"`
// Yolo marks this task's agent as autonomous. crush run has NO yolo
// flag (verified against crush v0.92: "Unknown flag: --yolo"); autonomy
// comes from the repo's own project-local crush config granting
// permissions (`.crushrc`: `permissions allow view ls grep edit write bash`).
// This field makes the executor fail fast with remediation guidance when
// autonomy is requested but the repo has no such config, instead of
// burning agent attempts on runs that stall on permission prompts.
Yolo bool `json:"yolo,omitempty"`
}
AgentPayload is the payload contract for "agent" tasks: run a headless AI coding agent in a repository, then prove the result.
type AgentResult ¶ added in v0.2.0
type AgentResult struct {
// SessionID is the crush session id, best-effort extracted from the
// agent's output (`crush run` prints it; formats vary between
// versions — absent when nothing matches).
SessionID string `json:"session_id,omitempty"`
// VerifyTail is the last lines of the verify command's output: the
// proof the task completed on.
VerifyTail string `json:"verify_tail,omitempty"`
// FilesChanged and CommitSHA are the agent's own structured report,
// parsed from a final `TQ_RESULT: {"files_changed": [...], "commit_sha":
// "..."}` output line (the agent prompt convention). Absent when the
// agent did not emit one.
FilesChanged []string `json:"files_changed,omitempty"`
CommitSHA string `json:"commit_sha,omitempty"`
// LogPath is the sidecar file holding the FULL agent + verify output,
// written when TQ_LOG_DIR is set on the worker/pool. Absent otherwise.
LogPath string `json:"log_path,omitempty"`
}
AgentResult is the structured outcome detail of one agent run, stored alongside the completion so `tq show` can answer "what did the agent actually do" without SSH-ing into logs.
type CommandExecutor ¶
type CommandExecutor struct {
Template string
// MemoryLimitMB caps the shell process's virtual memory via
// `ulimit -v` (POSIX sh; 0 = uncapped). Bounds runaway payloads on
// shared machines — the queue keeps serving while the task dies fast.
MemoryLimitMB int
// Nice lowers scheduling priority via `nice -n` (0 = unchanged).
// Interactive work wins CPU; queue work yields.
Nice int
}
CommandExecutor runs a command per task.
If Template is empty, the task payload is used as the shell line verbatim — enqueue '{"cmd":"..."}' JSON or just the raw command text; JSON {"cmd":…} is unwrapped automatically.
If Template is set, it may contain {{ID}}, {{PROJECT}}, {{TYPE}}, {{PAYLOAD}} placeholders substituted per task.
func NewCommandExecutor ¶
func NewCommandExecutor(template string) *CommandExecutor
NewCommandExecutor builds a CommandExecutor from an optional template.
type CrushPayload ¶
type CrushPayload = AgentPayload
CrushPayload is the pre-convergence name of AgentPayload. Deprecated: use AgentPayload.
type Executor ¶
Executor runs one claimed task. It must be safe for concurrent use. Returning nil marks the task completed; any error marks a failed attempt. Return a *PermanentError (see Permanent) when retrying can never help.
type FailureEvidence ¶ added in v0.2.0
type FailureEvidence struct {
Stage string `json:"stage"` // "agent", "verify" or "command"
ExitCode int `json:"exit_code,omitempty"` // process exit code (0 when the error was not an exit)
Tail string `json:"tail,omitempty"` // last lines of the failing output
}
FailureEvidence is the structured forensics attached to a task.failed fact: WHAT failed (which stage), the process exit code, and the tail of the output that proves it. The 21:40 window's two retry-path failures left empty {} detail — the why lived only in the error text.
type HTTPExecutor ¶
HTTPExecutor POSTs the task as JSON to a URL. Any 2xx completes the task; anything else (or transport error) fails the attempt.
func NewHTTPExecutor ¶
func NewHTTPExecutor(url string) *HTTPExecutor
NewHTTPExecutor builds an HTTP executor with a sane default client.
type LeaseLostError ¶
type LeaseLostError struct{ Cause error }
LeaseLostError signals the worker lost its lease while executing and must not record a failure for the task.
func (*LeaseLostError) Error ¶
func (e *LeaseLostError) Error() string
func (*LeaseLostError) Unwrap ¶
func (e *LeaseLostError) Unwrap() error
type PermanentError ¶
type PermanentError struct{ Cause error }
PermanentError marks a failure that retrying can never fix: the identical input will fail identically (bad payload, missing repo, dirty tree, missing autonomy). Workers must dead-letter the task after the first attempt instead of burning the retry budget — for agent tasks every retry is real money.
Detect it with errors.AsType / errors.Is; executors should return it via Permanent so wrapping is uniform.
func (*PermanentError) Error ¶
func (e *PermanentError) Error() string
func (*PermanentError) Unwrap ¶
func (e *PermanentError) Unwrap() error
type PreflightError ¶ added in v0.2.0
type PreflightError struct{ Cause error }
PreflightError signals the executor refused to START a task: the environment is not ready (dirty repo waiting on a human commit, autonomy config missing, verify tool absent) but neither the task nor the payload is wrong. Workers must requeue the task WITHOUT burning an attempt — it becomes claimable again after a delay, so the pool picks it up as soon as the environment is fixed.
func (*PreflightError) Error ¶ added in v0.2.0
func (e *PreflightError) Error() string
func (*PreflightError) Unwrap ¶ added in v0.2.0
func (e *PreflightError) Unwrap() error
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry maps task types to executors.
Example ¶
Registering a custom executor type makes any task whose type matches run your Go code instead of a shell command.
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"github.com/larsartmann/go-taskqueue/internal/executor"
"github.com/larsartmann/go-taskqueue/internal/task"
)
func main() {
reg := executor.NewRegistry()
reg.RegisterFunc("greet", func(_ context.Context, t task.Task) error {
var name string
if err := json.Unmarshal(t.Payload, &name); err != nil {
name = "stranger"
}
fmt.Printf("hello %s\n", name)
return nil
})
ex, err := reg.Lookup("greet")
if err != nil {
log.Fatal(err)
}
err = ex.Execute(context.Background(), task.Task{
ID: task.ID("t1"),
Type: "greet",
Payload: json.RawMessage(`"world"`),
})
if err != nil {
log.Fatal(err)
}
}
Output: hello world
func (*Registry) RegisterFunc ¶
RegisterFunc binds a task type to a function.
type ReviewExecutor ¶ added in v0.2.0
type ReviewExecutor struct {
// Agent supplies the run mechanics and its configuration (Bin,
// ProjectsDir, Yolo). nil selects a default AgentExecutor.
Agent *AgentExecutor
}
ReviewExecutor runs reviewer agents. It reuses the agent run mechanics (repo resolution, autonomy preflight, clean-tree guard, process-group handling) through its Agent executor; the only differences are the prompt and the strict output contract.
func (*ReviewExecutor) Execute ¶ added in v0.2.0
Execute runs the reviewer and enforces the verdict contract. Approve and request_changes both succeed; malformed output fails the attempt (retryable — the model may comply on a retry), input-contract misses are permanent, dirty trees are preflight requeues.
type ReviewFinding ¶ added in v0.2.0
type ReviewFinding struct {
// Title is the one-line summary; it keys the dedup of auto-minted fix
// tasks, so keep findings stable in wording across re-reviews.
Title string `json:"title"`
// Severity is a display hint normalized to low/medium/high at parse
// time (unknown or empty ratings degrade to medium — severity never
// gates anything mechanically, so leniency here is safe).
Severity string `json:"severity,omitempty"`
// Detail is the free-form explanation and suggested direction.
Detail string `json:"detail,omitempty"`
}
ReviewFinding is one concrete problem the reviewer wants addressed.
type ReviewPayload ¶ added in v0.2.0
type ReviewPayload struct {
// Repo is the repository the change landed in (same resolution rules as
// AgentPayload.Repo). Required.
Repo string `json:"repo"`
// ReviewedTask is the queue task ID under review — lineage for `tq show`
// and the sweeper's loop guard. Required.
ReviewedTask string `json:"reviewed_task"`
// Item is the original instruction the reviewed agent was given: the bar
// the change is judged against. Required.
Item string `json:"item"`
// CommitSHA is the commit the reviewed agent reported (from its
// TQ_RESULT self-report). Empty when the agent landed no commit; the
// reviewer then inspects the repo's latest state instead.
CommitSHA string `json:"commit_sha,omitempty"`
// FilesChanged lists the files the reviewed agent reported touching.
FilesChanged []string `json:"files_changed,omitempty"`
// Extra carries operator focus instructions appended to the prompt.
Extra string `json:"extra,omitempty"`
// Model optionally overrides the crush model for the reviewer.
Model string `json:"model,omitempty"`
// Yolo marks the reviewer as autonomous (same contract as
// AgentPayload.Yolo: needs a repo-local crush config granting
// permissions). Pools typically mirror the reviewed task's autonomy.
Yolo bool `json:"yolo,omitempty"`
// RequireClean refuses to start unless the repo tree is clean — a
// reviewer must judge the committed state, not dirty work-in-progress.
// Default true. Same escape hatch as AgentPayload.
RequireClean *bool `json:"require_clean,omitempty"`
// TimeoutMinutes caps the review run. Default 15 (reviews read; they do
// not rebuild the world).
TimeoutMinutes int `json:"timeout_minutes,omitempty"`
}
ReviewPayload is the payload contract for "review" tasks. It is self-contained: everything the reviewer needs to judge the change travels in the payload, so the executor stays a pure process runner.
type ReviewResult ¶ added in v0.2.0
type ReviewResult struct {
Verdict ReviewVerdict `json:"verdict"`
Summary string `json:"summary,omitempty"`
Findings []ReviewFinding `json:"findings,omitempty"`
SessionID string `json:"session_id,omitempty"`
// LogPath is the sidecar file with the full reviewer output (written
// when TQ_LOG_DIR is set), same convention as AgentResult.
LogPath string `json:"log_path,omitempty"`
}
ReviewResult is the structured outcome of one review run, stored in the completion fact detail (the sink convention) for the sweeper and `tq show`.
func ParseResult ¶ added in v0.2.0
func ParseResult(output string) (ReviewResult, error)
ParseResult extracts and validates the verdict JSON from reviewer output. Strict where it matters, lenient where nothing depends on it: the verdict must be exactly approve or request_changes (case-insensitive) and the JSON must parse; findings with empty titles are dropped and severity ratings are normalized to low/medium/high.
type ReviewVerdict ¶ added in v0.2.0
type ReviewVerdict string
ReviewVerdict is the reviewer's decision. Only these two values are accepted; anything else is invalid output and a failed attempt.
const ( // VerdictApprove means the change is correct and complete as it stands. VerdictApprove ReviewVerdict = "approve" // VerdictRequestChanges means the reviewer found concrete, actionable // problems; each finding is a candidate fix task. VerdictRequestChanges ReviewVerdict = "request_changes" )
type Sink ¶ added in v0.2.0
type Sink struct {
// contains filtered or unexported fields
}
Sink collects structured outcome detail for ONE task execution.
func (*Sink) Detail ¶ added in v0.2.0
func (s *Sink) Detail() json.RawMessage
Detail returns the recorded outcome detail, or nil.
func (*Sink) Failure ¶ added in v0.2.0
func (s *Sink) Failure() json.RawMessage
Failure returns the recorded failure evidence, or nil.
type StatusCompletion ¶ added in v0.2.0
type StatusCompletion struct {
TaskID string `json:"task_id"`
Item string `json:"item"`
Commit string `json:"commit,omitempty"`
Files []string `json:"files,omitempty"`
CompletedAt string `json:"completed_at,omitempty"`
}
StatusCompletion is one completed agent task in the window a status report covers. The item is an excerpt of the agent's instruction (first line, bounded) — the full prompt stays in `tq show`.
type StatusExecutor ¶ added in v0.2.0
type StatusExecutor struct {
// Agent supplies the run mechanics and its configuration. nil selects a
// default AgentExecutor.
Agent *AgentExecutor
}
StatusExecutor runs the done-prompt agent. It reuses the agent run mechanics through its Agent executor (same pattern as ReviewExecutor); the differences are the prompt and the output contract: the final TQ_RESULT line must name an existing, repo-relative report file.
func (*StatusExecutor) Execute ¶ added in v0.2.0
Execute runs the reporting agent and enforces two gates: the output must end with TQ_RESULT: {"report":"docs/status/...","next_items":N} naming an existing, repo-relative file, and the repo verify command must exit 0 (the reporter commits, so it can break the tree it reports on). Malformed output, a missing file or a failed verify is a retryable failure; payload misses are permanent; dirty trees are preflight requeues.
type StatusPayload ¶ added in v0.2.0
type StatusPayload struct {
// Repo is the repository being reported on (same resolution rules as
// AgentPayload.Repo). Required.
Repo string `json:"repo"`
// Project is the queue project (repo name) the window belongs to.
// Required; also the sweeper's loop guard identity.
Project string `json:"project"`
// Completed lists the agent tasks in this report's window, oldest
// first. Bounded by the sweeper; never empty for a minted task.
Completed []StatusCompletion `json:"completed"`
// Model optionally overrides the crush model for the reporting agent.
Model string `json:"model,omitempty"`
// Yolo marks the reporting agent as autonomous (same contract as
// AgentPayload.Yolo). Pools typically mirror the window's autonomy.
Yolo bool `json:"yolo,omitempty"`
// RequireClean refuses to start unless the repo tree is clean — the
// reporter must not trample work-in-progress, and commits its own
// report afterwards. Default true. Same escape hatch as AgentPayload.
RequireClean *bool `json:"require_clean,omitempty"`
// TimeoutMinutes caps the report run. Default 15.
TimeoutMinutes int `json:"timeout_minutes,omitempty"`
// Verify is the repo quality gate that must exit 0 after the report
// run (the reporter commits, so a broken tree is a real possibility).
// Empty means the usual resolution: the repo's .tq-verify file, then
// auto-detection — identical to AgentPayload.Verify.
Verify string `json:"verify,omitempty"`
}
StatusPayload is the payload contract for "status" tasks. Self-contained, like ReviewPayload: everything the reporting agent needs travels in the payload, so the executor stays a pure process runner.
type StatusResult ¶ added in v0.2.0
type StatusResult struct {
// Report is the repo-relative path of the written status report.
Report string `json:"report"`
// NextItems counts the next-work items the agent appended to
// TODO_LIST.md (its own self-report, not re-verified).
NextItems int `json:"next_items"`
SessionID string `json:"session_id,omitempty"`
LogPath string `json:"log_path,omitempty"`
}
StatusResult is the structured outcome of one status run, stored in the completion fact detail (the sink convention) for `tq show` and the web UI.