task

package
v1.21.60 Latest Latest
Warning

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

 Go to latest Published: Sep 8, 2026 License: Apache-2.0 Imports: 33 Imported by: 25

Documentation

Overview

Package task provides a comprehensive concurrent task execution system with progress tracking, status management, and visual rendering capabilities.

Quick Start

The task system provides a simple interface for running concurrent operations with visual feedback: 

import "github.com/flanksource/clicky/task"

// Start a simple task
task := task.StartTask("Download File", func(ctx flanksourceContext.Context, t *task.Task) error {
	t.Infof("Starting download...")
	// Perform work
	time.Sleep(2 * time.Second)
	t.Infof("Download complete")
	return nil
})

// Wait for completion
result := task.WaitFor()
if result.Error != nil {
	fmt.Printf("Task failed: %v\n", result.Error)
}

Core Concepts

Task Lifecycle

Tasks progress through well-defined states with visual indicators:

- StatusPending (⏳): Task is queued but not yet started - StatusRunning (⟳): Task is currently executing - StatusSuccess (✓): Task completed successfully - StatusFailed (✗): Task failed with an error - StatusWarning (⚠): Task completed with warnings - StatusCancelled (⊘): Task was canceled

Manager

The Manager coordinates task execution and provides visual rendering: 

manager := task.NewManager(
	task.WithMaxConcurrency(5),        // Limit concurrent tasks
	task.WithVerbose(true),            // Enable verbose logging
	task.WithNoProgress(false),        // Show progress bars
)

// Tasks automatically use the global manager if none specified

Task Groups

Groups organize related tasks and provide concurrency control: 

group := task.NewGroup("Database Migration",
	task.WithConcurrency(2),  // Max 2 concurrent tasks in group
)

// Add tasks to the group
task1 := group.Add("Migrate Users", func(ctx, t) error { ... })
task2 := group.Add("Migrate Orders", func(ctx, t) error { ... })

// Wait for all tasks in the group
result := group.WaitFor()

TypedTask for Type Safety

TypedTask provides compile-time type safety for task results: 

// Define a typed task that returns a string
task := task.StartTask("Fetch Data", func(ctx flanksourceContext.Context, t *task.Task) (string, error) {
	// Perform work and return typed result
	return "Hello, World!", nil
})

// Get typed result
result, err := task.GetResult()
// result is of type string

// Or wait and get result in one call
wait := task.WaitFor()
if wait.Error != nil {
	// Handle error
}

TypedGroup

Groups can also be typed for consistent result handling: 

group := task.NewTypedGroup[UserData]("Load Users")

user1 := group.Add("Load User 1", func(ctx, t) (UserData, error) {
	return loadUser(1)
})

user2 := group.Add("Load User 2", func(ctx, t) (UserData, error) {
	return loadUser(2)
})

// Get all results as map
results, err := group.GetResults()
if err != nil {
	return err
}

for task, userData := range results {
	fmt.Printf("Task %s loaded: %+v\n", task.Name(), userData)
}

Status and Health System

The task system includes a rich status and health reporting system:

Status Types

Tasks support multiple status paradigms: 

// Standard task statuses
task.SetStatus(task.StatusSuccess)
task.SetStatus(task.StatusFailed)
task.SetStatus(task.StatusWarning)

// Test-style statuses
task.SetStatus(task.StatusPASS)
task.SetStatus(task.StatusFAIL)
task.SetStatus(task.StatusSKIP)

Health Mixin

Results can implement HealthMixin for automatic status determination: 

type DatabaseResult struct {
	Connected bool
	Error     error
}

func (r DatabaseResult) Health() task.Health {
	if r.Error != nil {
		return task.HealthError
	}
	if !r.Connected {
		return task.HealthWarning
	}
	return task.HealthOK
}

// Task status will automatically reflect the health
task.SetResult(DatabaseResult{Connected: true})
// Task status becomes StatusSuccess automatically

Visual Styling

Status information includes visual styling with Tailwind CSS classes: 

status := task.StatusSuccess
fmt.Println(status.Icon())  // ✓
fmt.Println(status.Style()) // "text-green-600"

// Apply status styling to text
text := api.Text{Content: "Operation Complete"}
styledText := status.Apply(text)

Concurrency Control

Manager-Level Concurrency

Control global task concurrency: 

manager := task.NewManager(task.WithMaxConcurrency(10))
// Maximum 10 tasks running simultaneously

Group-Level Concurrency

Fine-grained control within groups: 

group := task.NewGroup("API Calls", task.WithConcurrency(3))
// Maximum 3 concurrent tasks within this group

Semaphore-Based Control

Groups use semaphores for precise concurrency management: 

// Group automatically handles semaphore acquisition/release
group.Add("Task 1", taskFunc1)  // Acquires semaphore
group.Add("Task 2", taskFunc2)  // Waits if limit reached
// Semaphore released when task completes

Error Handling and Retry

Basic Error Handling

Tasks provide multiple ways to handle errors: 

task := task.StartTask("Risky Operation", func(ctx, t) error {
	if someCondition {
		return errors.New("operation failed")
	}
	return nil
})

result := task.WaitFor()
if result.Error != nil {
	fmt.Printf("Task failed: %v\n", result.Error)
	fmt.Printf("Status: %s\n", result.Status)
}

Retry Configuration

Tasks support automatic retry with exponential backoff: 

retryConfig := task.RetryConfig{
	RetryableErrors: []string{"timeout", "connection", "rate limit"},
	BaseDelay:      1 * time.Second,
	MaxDelay:       30 * time.Second,
	BackoffFactor:  2.0,
	JitterFactor:   0.1,
	MaxRetries:     3,
}

task := task.StartTaskWithOptions("Flaky Operation", taskFunc,
	task.WithRetry(retryConfig),
)

Fatal Errors

For unrecoverable errors that should stop execution: 

task.Fatal(errors.New("critical system failure"))
// Immediately stops execution and exits program

Logging and Progress

Built-in Logging

Tasks include comprehensive logging capabilities: 

task := task.StartTask("Process Data", func(ctx, t) error {
	t.Infof("Starting data processing...")
	t.Debugf("Processing batch %d", batchNum)

	if err := processData(); err != nil {
		t.Errorf("Failed to process batch: %v", err)
		return err
	}

	t.Infof("Processing complete")
	return nil
})

Log levels: - t.Tracef(): Detailed tracing (lowest level) - t.Debugf(): Debug information - t.Infof(): General information - t.Warnf(): Warnings - t.Errorf(): Errors - t.Fatalf(): Fatal errors

Progress Tracking

Tasks support progress indication: 

task := task.StartTask("Upload Files", func(ctx, t) error {
	files := getFilesToUpload()
	total := len(files)

	for i, file := range files {
		t.SetProgress(i, total)
		err := uploadFile(file)
		if err != nil {
			return err
		}
	}

	t.SetProgress(total, total) // 100% complete
	return nil
})

Integration Examples

CLI Tool Integration 

func runCommand(args []string) error {
	manager := task.NewManager(
		task.WithVerbose(verbose),
		task.WithNoProgress(noProgress),
	)

	// Start background tasks
	task1 := task.StartTask("Validate Input", validateInput)
	task2 := task.StartTask("Load Configuration", loadConfig)

	// Wait for prerequisites
	task1.WaitFor()
	task2.WaitFor()

	// Main processing
	mainTask := task.StartTask("Process Data", processData)
	return mainTask.WaitFor().Error
}

HTTP Server Integration 

func handleRequest(w http.ResponseWriter, r *http.Request) {
	requestID := generateRequestID()

	task := task.StartTask(fmt.Sprintf("Handle Request %s", requestID),
		func(ctx, t) error {
			t.Infof("Processing request from %s", r.RemoteAddr)

			// Process request
			result, err := processRequest(r)
			if err != nil {
				t.Errorf("Request failed: %v", err)
				return err
			}

			t.Infof("Request completed successfully")
			return writeResponse(w, result)
		})

	// Wait for completion
	if result := task.WaitFor(); result.Error != nil {
		http.Error(w, result.Error.Error(), 500)
	}
}

Batch Processing 

func processBatch(items []Item) error {
	group := task.NewTypedGroup[ProcessedItem]("Batch Processing",
		task.WithConcurrency(5),
	)

	// Process items concurrently
	for i, item := range items {
		group.Add(fmt.Sprintf("Process Item %d", i+1),
			func(ctx, t) (ProcessedItem, error) {
				return processItem(item)
			})
	}

	// Wait for all processing to complete
	results, err := group.GetResults()
	if err != nil {
		return fmt.Errorf("batch processing failed: %w", err)
	}

	// Handle results
	for task, result := range results {
		fmt.Printf("Task %s: %+v\n", task.Name(), result)
	}

	return nil
}

Testing Integration

The task system automatically detects test environments and adjusts behavior: 

func TestDataProcessing(t *testing.T) {
	// Progress bars and colors automatically disabled in tests
	task := task.StartTask("Test Operation", func(ctx, task) error {
		// Task logging available in tests
		task.Infof("Running test operation")
		return performTestOperation()
	})

	result := task.WaitFor()
	assert.NoError(t, result.Error)
	assert.Equal(t, task.StatusSuccess, result.Status)
}

Advanced Features

Task Dependencies

Tasks can depend on other tasks: 

prerequisite := task.StartTask("Load Configuration", loadConfig)

mainTask := task.StartTaskWithOptions("Main Process", mainProcess,
	task.WithDependencies(prerequisite),
)

// mainTask won't start until prerequisite completes successfully

Task Identity and Deduplication

Prevent duplicate tasks with identity tracking: 

task1 := task.StartTaskWithOptions("Download File", downloadFunc,
	task.WithIdentity("download-file-xyz"),
)

// This will return the existing task instead of creating a new one
task2 := task.StartTaskWithOptions("Download File", downloadFunc,
	task.WithIdentity("download-file-xyz"),
)

// task1 and task2 reference the same underlying task

Custom Styling and Themes

Tasks support custom visual themes: 

customTheme := api.Theme{
	Success: "text-blue-600",
	Error:   "text-purple-600",
	Warning: "text-orange-500",
}

manager := task.NewManager(task.WithTheme(customTheme))

Graceful Shutdown

Handle interrupts gracefully: 

manager := task.NewManager(
	task.WithGracefulTimeout(30 * time.Second),
	task.WithInterruptHandler(func() {
		fmt.Println("Shutting down gracefully...")
	}),
)

// Manager will handle SIGINT/SIGTERM and allow running tasks to complete

// Terminal Safety // // The task system saves and restores terminal state automatically: // // - Normal exit: restored via shutdown hooks // - SIGINT: restored during graceful shutdown // - Double SIGINT: restored before forced os.Exit(1) // - Panic in main: restored if main defers shutdown.RecoverAndShutdown() // - Panic in task func: caught by worker, task marked as failed, process continues // // For panic protection in main, use: // // func main() { // defer shutdown.RecoverAndShutdown() // // ... start tasks, wait, etc. // } // // RecoverAndShutdown runs all shutdown hooks, restores the terminal, then // re-panics so the stack trace is preserved.

The task package provides a complete solution for concurrent task execution with rich visual feedback, comprehensive error handling, and flexible configuration options suitable for CLI tools, servers, and batch processing applications.

Index

Examples

Constants

View Source 
const SnapshotStreamLimit = 1 << 20

SnapshotStreamLimit bounds each stdout/stderr tail exposed through task snapshots. ExecResult remains complete; this limit applies only to task views and persisted snapshots.

Variables

View Source 
var (
	// ErrBatchTimeout is returned when the entire batch exceeds its timeout
	ErrBatchTimeout = errors.New("batch execution exceeded timeout")

	// ErrItemTimeout is returned when an individual item exceeds its timeout
	ErrItemTimeout = errors.New("item execution exceeded timeout")
)
View Source 
var OnBeforeGC func(groupID string, snapshots []TaskSnapshot)

OnBeforeGC, if non-nil, is called with each group's full snapshot just before GCRuns removes it from the in-memory manager. The callback receives the group's stable ID and the full snapshot slice (group + child tasks). It is called while global.mu is held, so it must not call back into the task package.

This is an eviction hook, not a persistence mechanism. To store finished runs durably, install a Store with SetStore instead: Store.SaveRun is handed the same snapshot slice from a background goroutine with no lock held, so it may block on IO, and it fires the moment a run finishes rather than at eviction runRetention later. Reach for OnBeforeGC when the work has to happen at eviction specifically — releasing resources keyed by run id, dropping caches, cleaning up scratch directories.

The two are independent: when both are installed, both are called.

Functions

func AcquirePromptTerminal 

func AcquirePromptTerminal() (func(), bool)

AcquirePromptTerminal waits for exclusive ANSI terminal ownership for prompt rendering, stopping the active task renderer first when necessary. The second return value reports whether the prompt displaced an active task renderer.

func BindManagerFlags 

func BindManagerFlags(flags *flag.FlagSet, options *ManagerOptions)

BindManagerFlags adds TaskManager flags to standard flag set

func BindManagerPFlags 

func BindManagerPFlags(flags *pflag.FlagSet, options *ManagerOptions)

BindManagerPFlags adds TaskManager flags to pflag set (for Cobra)

func CancelAll 

func CancelAll()

CancelAll cancels all running tasks and groups

func ClearTasks 

func ClearTasks()

ClearTasks removes all completed tasks (and groups) from the registry, keeping only those still pending or running. Groups must be pruned too: SnapshotAll iterates global.groups, so a leftover completed group would keep surfacing its (now-removed) tasks — e.g. a stale "Running tests" group whose task IDs no longer resolve, causing StopTask lookups to fail.

func ControlRun added in v1.21.48 

func ControlRun(ctx context.Context, id string, action ControlAction) error

ControlRun performs one advertised lifecycle action for a live run.

func ControlTask added in v1.21.54 

func ControlTask(ctx context.Context, runID, taskID string, action ControlAction) error

ControlTask performs one advertised lifecycle action for a child task in a live run.

func Debug 

func Debug() string

Debug returns debug information about the task manager

func GCRuns 

func GCRuns()

GCRuns removes finished runs older than runRetention from the global manager. A run is "finished" once it has a non-zero FinishedAt (recorded the first time it is observed terminal). Live runs are retained regardless of age. When a Store is installed, each evicted group's full snapshot is handed to the background writer before removal — the last chance to persist it, since after eviction there is no group left to resolve.

func GatedStderr 

func GatedStderr() io.Writer

GatedStderr returns a writer that wraps os.Stderr but buffers writes while the interactive task renderer owns the TTY, flushing them once it lets go. The writer is stateless; each Write rechecks ownership, so a writer captured before the renderer started still gates correctly.

func IsForceInteractive 

func IsForceInteractive() bool

IsForceInteractive reports whether the interactive renderer has been forced on via SetForceInteractive or the CLICKY_FORCE_INTERACTIVE env var.

func IsInteractiveRenderActive 

func IsInteractiveRenderActive() bool

IsInteractiveRenderActive reports whether the global task manager's interactive render loop currently owns the terminal. Callers that write to os.Stderr can consult this to drop writes that would corrupt the live frame. The check is cheap (atomic load + RLock).

Returns false when the manager is non-interactive (PlainRender mode), before the renderer has acquired the TTY, or after stopRender has released it.

func IsNoRender 

func IsNoRender() bool

IsNoRender reports whether task rendering is currently disabled.

func JSONHandler 

func JSONHandler(taskIDs ...string) http.Handler

JSONHandler returns an http.Handler that serves the full task state as JSON. If taskIDs are provided, only groups matching those IDs are included. This is used for initial page load, reconnection after SSE drop, or polling fallback.

func MetricID added in v1.21.48 

func MetricID(runID, name string) string

MetricID scopes one metric series to an immutable run id.

func Metrics added in v1.21.48 

func Metrics() metrics.Timeseries

Metrics exposes the bounded task metric store for embedding HTTP handlers.

func RecordMetric added in v1.21.48 

func RecordMetric(runID, name string, value float64, at time.Time) error

RecordMetric appends one bounded live metric point for a run.

func RegisterHandlers 

func RegisterHandlers(mux *http.ServeMux, prefix string)

RegisterHandlers wires the generic task-manager API under prefix: 

GET {prefix}/tasks         run listing (RunMeta[], ?kind=&status=&label=k=v)
GET {prefix}/tasks/stream  SSE stream of TaskSnapshots (?tasks=<id>&kind=)
GET {prefix}/tasks/runs/stream persistent SSE stream of RunMeta listings
GET {prefix}/tasks/{id}    id-scoped snapshot (group + tasks)
POST {prefix}/tasks/{id}/control lifecycle action
POST {prefix}/tasks/{id}/tasks/{taskID}/control child lifecycle action
POST {prefix}/tasks/{id}/tasks/{taskID}/stdin live stdin (never persisted)

The {id} route reuses Go 1.22 net/http path-value routing; the stream route is registered before the {id} route so "stream" is not treated as an id.

func RegisterHandlersWithSource added in v1.21.48 

func RegisterHandlersWithSource(mux *http.ServeMux, prefix string, source RunSource)

RegisterHandlersWithSource wires the task API and merges externally-owned runs.

func RegisterRunner added in v1.21.60 

func RegisterRunner(kind string, runner Runner)

RegisterRunner registers the work for one schedule kind. Registering a kind twice is a programming error and panics, matching the entity and provider registries elsewhere: two runners for one kind means one of them never runs.

func RunControlHandler added in v1.21.48 

func RunControlHandler() http.Handler

RunControlHandler performs a lifecycle action advertised by a live run.

func RunControlHandlerWithSource added in v1.21.48 

func RunControlHandlerWithSource(source RunSource) http.Handler

RunControlHandlerWithSource routes controls to the live owner of a run.

func RunHandler 

func RunHandler() http.Handler

RunHandler serves the id-scoped snapshot (group + its tasks) for a single run. The id is read from the {id} path value of the registered route.

func RunHandlerWithSource added in v1.21.48 

func RunHandlerWithSource(source RunSource) http.Handler

RunHandlerWithSource falls through to source when the run is not in memory.

func RunsHandler 

func RunsHandler() http.Handler

RunsHandler serves the run listing (RunMeta per group) for the generic task-manager view, filtered by the ?kind=, ?status=, and repeated ?label=k=v query params.

func RunsHandlerWithSource added in v1.21.48 

func RunsHandlerWithSource(source RunSource) http.Handler

RunsHandlerWithSource merges in runs owned by another process or store.

func RunsSSEHandler 

func RunsSSEHandler(supplement func(RunFilter) []RunMeta) http.Handler

RunsSSEHandler streams the run listing (RunMeta) as SSE. Unlike SSEHandler it never sends a terminal event: a manager view stays subscribed to observe new and changing runs. supplement (may be nil) merges extra runs — e.g. archived or persisted runs the in-memory registry no longer holds; live runs win on id. It emits a single "event: runs" frame carrying the full listing, and only re-emits when the listing changes.

func RunsSSEHandlerWithSource added in v1.21.48 

func RunsSSEHandlerWithSource(source RunSource) http.Handler

RunsSSEHandlerWithSource streams both in-memory and externally-owned runs.

func SSEHandler 

func SSEHandler(taskIDs ...string) http.Handler

SSEHandler returns an http.Handler that streams task events via Server-Sent Events. If taskIDs are provided, only groups matching those IDs are streamed. The handler polls every 200ms, emits changed JSON snapshots, and sends stdout and stderr as append-only output deltas (or a reset when the bounded tail rolls over). It sends an "event: done" when all tracked groups have completed.

func SSEHandlerWithSource added in v1.21.48 

func SSEHandlerWithSource(source RunSource, taskIDs ...string) http.Handler

SSEHandlerWithSource also streams task runs owned by another process or store.

func SetForceInteractive 

func SetForceInteractive(force bool)

SetForceInteractive forces the interactive renderer on even when the process is running under `go test` or CI (GO_TEST, CI, GITHUB_ACTIONS, …). It also marks the manager as interactive when the underlying FD is not a TTY, so that a PTY-wrapping caller sees the full cursor/clear/redraw ANSI stream. Intended for harnesses that need to audit interactive output — normal applications should rely on the automatic TTY detection.

func SetGracefulTimeout 

func SetGracefulTimeout(timeout time.Duration)

SetGracefulTimeout sets the timeout for graceful shutdown

func SetInterruptHandler 

func SetInterruptHandler(fn func())

SetInterruptHandler sets a custom callback to be called on interrupt

func SetLiveRenderer 

func SetLiveRenderer(r LiveRenderer)

SetLiveRenderer installs a custom renderer on the global manager, or removes it when r is nil. It is process-global like SetNoRender; install it for the duration of one command and restore (SetLiveRenderer(nil)) afterwards.

func SetMaxConcurrent 

func SetMaxConcurrent(max int)

SetMaxConcurrent sets the maximum number of concurrent tasks

func SetNoColor 

func SetNoColor(noColor bool)

SetNoColor enables or disables colored output

func SetNoProgress 

func SetNoProgress(noProgress bool)

SetNoProgress enables or disables progress display

func SetNoRender 

func SetNoRender(noRender bool)

SetNoRender enables or disables all task rendering, including final summaries.

func SetRetryConfig 

func SetRetryConfig(config RetryConfig)

SetRetryConfig sets the default retry configuration for new tasks

func SetStore added in v1.21.60 

func SetStore(ctx context.Context, store Store)

SetStore installs the durable store and starts the background writer that feeds it. The writer runs until ctx is done. Passing a nil store detaches the current one.

Replacing or detaching a store closes its queue and waits for the writer to drain what is already in it, so a run that finished before the swap is not lost by it.

Reads still go through the *WithSource handlers, which take a RunSource; the installed store satisfies that interface and can be passed to them directly.

func SetVerbose 

func SetVerbose(verbose bool)

SetVerbose enables or disables verbose logging

func StartCapturingOutput 

func StartCapturingOutput()

StartCapturingOutput replaces os.Stdout and os.Stderr on the global task manager with pipes that buffer everything written until StopCapturingOutput is called. The live task renderer is unaffected because it captured the original file descriptors at manager init time (see Manager.renderer in manager.go). Loggers that captured os.Stderr before this call will also keep writing to the real terminal — only bare fmt.Print / os.Stderr writes after this call get buffered.

func StopCapturingOutput 

func StopCapturingOutput()

StopCapturingOutput restores the real os.Stdout and os.Stderr on the global task manager and flushes every buffered line to the restored streams in the order it was written, tagged by stream of origin. Safe to call when capture wasn't started (no-op).

func StopTask 

func StopTask(id string) bool

StopTask cancels a specific pending or running task by immutable ID.

func TaskControlHandler added in v1.21.54 

func TaskControlHandler() http.Handler

TaskControlHandler performs a lifecycle action advertised by a child task.

func TaskControlHandlerWithSource added in v1.21.54 

func TaskControlHandlerWithSource(source RunSource) http.Handler

TaskControlHandlerWithSource routes child controls to the live owner of a run, including runs owned by an external source.

func TaskStdinHandler added in v1.21.48 

func TaskStdinHandler() http.Handler

TaskStdinHandler writes live stdin to a task. Input is deliberately not included in any snapshot or persisted run record.

func Wait 

func Wait() int

Wait waits for all tasks to complete and returns the appropriate exit code. It stops the renderer and stops output capture (flushing any output buffered by StartCapturingOutput to the restored streams).

func WaitForAllTasks 

func WaitForAllTasks()

WaitForAllTasks waits for all global tasks to complete and forces a final render

func WaitSilent 

func WaitSilent() int

WaitSilent waits for all tasks to complete without displaying results. It stops the renderer and stops output capture (flushing any output buffered by StartCapturingOutput to the restored streams).

func WriteTaskStdin added in v1.21.48 

func WriteTaskStdin(runID, taskID string, data []byte) error

WriteTaskStdin writes bytes to a live task that advertises stdin support.

Types

type Batch 

type Batch[T any] struct {
	Name       string
	Items      []func(logger logger.Logger) (T, error)
	MaxWorkers int
	Results    []T
	// Timeout is the maximum duration for the entire batch to complete.
	// Zero value means no timeout (infinite wait until completion or context cancellation).
	Timeout time.Duration
	// ItemTimeout is the maximum duration for each individual item to complete.
	// Zero value means no per-item timeout.
	ItemTimeout time.Duration
}
Example (ItemTimeout)

ExampleBatch_itemTimeout demonstrates using per-item timeouts 

package main

import (
	"errors"
	"fmt"
	"time"

	"github.com/flanksource/commons/logger"

	"github.com/flanksource/clicky/task"
)

func main() {
	batch := &task.Batch[int]{
		Name:        "example-item-timeout",
		ItemTimeout: 100 * time.Millisecond, // Each item must complete within 100ms
		MaxWorkers:  3,
	}

	// Add 6 items: 3 fast (50ms), 3 slow (150ms)
	for i := 0; i < 6; i++ {
		i := i
		batch.Items = append(batch.Items, func(log logger.Logger) (int, error) {
			if i%2 == 0 {
				time.Sleep(50 * time.Millisecond)
			} else {
				time.Sleep(150 * time.Millisecond)
			}
			return i, nil
		})
	}

	completed := []int{}
	itemTimeouts := 0

	for result := range batch.Run() {
		if result.Error != nil {
			if errors.Is(result.Error, task.ErrItemTimeout) {
				itemTimeouts++
			}
		} else {
			completed = append(completed, result.Value)
		}
	}

	fmt.Printf("Completed: %d, Timed out: %d\n", len(completed), itemTimeouts)
}

Output:
Completed: 3, Timed out: 3
Example (Timeout)

ExampleBatch_timeout demonstrates using a batch timeout to limit total execution time 

package main

import (
	"errors"
	"fmt"
	"time"

	"github.com/flanksource/commons/logger"

	"github.com/flanksource/clicky/task"
)

func main() {
	batch := &task.Batch[string]{
		Name:       "example-batch-timeout",
		Timeout:    500 * time.Millisecond, // Batch must complete within 500ms
		MaxWorkers: 2,
	}

	// Add 10 items that each take 200ms - only ~5 will complete before timeout
	for i := 0; i < 10; i++ {
		i := i
		batch.Items = append(batch.Items, func(log logger.Logger) (string, error) {
			time.Sleep(200 * time.Millisecond)
			return fmt.Sprintf("item-%d", i), nil
		})
	}

	completed := []string{}
	timedOut := false

	for result := range batch.Run() {
		if result.Error != nil {
			if errors.Is(result.Error, task.ErrBatchTimeout) {
				timedOut = true
				fmt.Println("Batch timeout occurred")
			}
		} else {
			completed = append(completed, result.Value)
		}
	}

	fmt.Printf("Completed %d items before timeout: %v\n", len(completed), timedOut)
}

Output:
Batch timeout occurred
Completed 6 items before timeout: true
Example (ZeroTimeout)

ExampleBatch_zeroTimeout demonstrates zero-value backward compatibility 

package main

import (
	"fmt"
	"time"

	"github.com/flanksource/commons/logger"

	"github.com/flanksource/clicky/task"
)

func main() {
	batch := &task.Batch[int]{
		Name:       "example-zero-timeout",
		Timeout:    0, // Zero means no timeout (backward compatible)
		MaxWorkers: 2,
	}

	// Add items that would take a while
	for i := 0; i < 5; i++ {
		i := i
		batch.Items = append(batch.Items, func(log logger.Logger) (int, error) {
			time.Sleep(50 * time.Millisecond)
			return i, nil
		})
	}

	count := 0
	for result := range batch.Run() {
		if result.Error == nil {
			count++
		}
	}

	fmt.Printf("All %d items completed (no timeout)\n", count)
}

Output:
All 5 items completed (no timeout)

func (*Batch[T]) Run 

func (b *Batch[T]) Run() chan BatchResult[T]

func (*Batch[T]) WithItemTimeout 

func (b *Batch[T]) WithItemTimeout(duration time.Duration) *Batch[T]

WithItemTimeout sets the maximum duration for each individual item to complete. Returns the batch for method chaining.

func (*Batch[T]) WithTimeout 

func (b *Batch[T]) WithTimeout(duration time.Duration) *Batch[T]

WithTimeout sets the maximum duration for the entire batch to complete. Returns the batch for method chaining.

Example

ExampleBatch_WithTimeout demonstrates method chaining 

package main

import (
	"fmt"
	"time"

	"github.com/flanksource/clicky/task"
)

func main() {
	batch := (&task.Batch[string]{
		Name:       "example-chaining",
		MaxWorkers: 3,
	}).WithTimeout(1 * time.Second).WithItemTimeout(200 * time.Millisecond)

	fmt.Printf("Batch timeout: %v, Item timeout: %v\n", batch.Timeout, batch.ItemTimeout)
}

Output:
Batch timeout: 1s, Item timeout: 200ms

type BatchResult 

type BatchResult[T any] struct {
	Value    T
	Error    error
	Duration time.Duration
}

type CatchUpPolicy added in v1.21.60 

type CatchUpPolicy string

CatchUpPolicy decides what happens to scheduled times that passed while the process was down. 

const (
	// CatchUpNone ignores missed times and resumes at the next one. The default.
	CatchUpNone CatchUpPolicy = "none"
	// CatchUpOnce runs once at startup if any scheduled time was missed,
	// regardless of how many were. Replaying every missed fire is never what a
	// reporting schedule wants after a weekend of downtime.
	CatchUpOnce CatchUpPolicy = "once"
)

type ControlAction added in v1.21.48 

type ControlAction string

ControlAction is a lifecycle command exposed by a task or run. 

const (
	ControlStart   ControlAction = "start"
	ControlStop    ControlAction = "stop"
	ControlRestart ControlAction = "restart"
	ControlDrain   ControlAction = "drain"
)

type Fire added in v1.21.60 

type Fire struct {
	// ScheduledFor is the cron instant this fire belongs to, which is not the
	// same as At when the fire is a catch-up or the scheduler ran late.
	ScheduledFor time.Time   `json:"scheduledFor"`
	At           time.Time   `json:"at"`
	Outcome      FireOutcome `json:"outcome"`
	RunID        string      `json:"runId,omitempty"`
	Reason       string      `json:"reason,omitempty"`
	Error        string      `json:"error,omitempty"`
}

Fire is one entry in a schedule's firing history.

type FireOutcome added in v1.21.60 

type FireOutcome string

FireOutcome is what the scheduler decided to do at a scheduled time. 

const (
	// FireStarted means the fire produced a run; RunID names it.
	FireStarted FireOutcome = "started"
	// FireSkipped means the previous run was still going and the schedule's
	// overlap policy is OverlapSkip.
	FireSkipped FireOutcome = "skipped"
	// FireCaughtUp means the fire replayed a scheduled time missed while the
	// process was down, under CatchUpOnce.
	FireCaughtUp FireOutcome = "caught-up"
	// FireFailed means the run could not be started at all.
	FireFailed FireOutcome = "failed"
)

type Group 

type Group struct {
	Items []Taskable // Can contain Tasks or nested Groups
	// contains filtered or unexported fields
}

Group represents a group of tasks that can be managed collectively

func (*Group) Cancel 

func (g *Group) Cancel()

Cancel cancels all items in the group

func (*Group) Context added in v1.21.48 

func (g *Group) Context() context.Context

Context returns the context canceled when the group is stopped.

func (*Group) FinishedAt 

func (g *Group) FinishedAt() time.Time

FinishedAt returns the time the group first became terminal, or the zero time if it is still running/pending. It is recorded lazily by observeTerminal (called from snapshotting) so it does not depend on a WaitFor caller.

func (*Group) GetTasks 

func (g *Group) GetTasks() []Taskable

func (*Group) ID 

func (g *Group) ID() string

ID returns the group's stable unique id.

func (*Group) IsGroup 

func (g *Group) IsGroup() bool

IsGroup returns true for Group

func (*Group) Metadata 

func (g *Group) Metadata() GroupMetadata

Metadata returns a copy of the group's metadata.

func (*Group) Name 

func (g *Group) Name() string

Name returns the group name

func (*Group) StartedAt 

func (g *Group) StartedAt() time.Time

StartedAt returns the group's start time.

func (*Group) Status 

func (g *Group) Status() Status

type GroupMetadata 

type GroupMetadata struct {
	Kind   string            `json:"kind,omitempty"`
	Labels map[string]string `json:"labels,omitempty"`
	Owner  string            `json:"owner,omitempty"`
	Href   string            `json:"href,omitempty"`
}

GroupMetadata is the queryable metadata attached to a run (task group). Kind classifies the run (e.g. "sql-fix", "test-run"), Labels carry arbitrary key/value facets for filtering, and Owner identifies who started it. All fields are optional.

type Health 

type Health string
const (
	HealthOK      Health = "ok"
	HealthWarning Health = "warning"
	HealthError   Health = "error"
	HealthPending Health = "pending"
)

func (Health) Style 

func (h Health) Style() string

type HealthMixin 

type HealthMixin interface {
	Health() Health
}

type LiveRenderer 

type LiveRenderer interface {
	// RenderLive returns the content for one live frame — a 250ms interactive
	// tick or a single non-interactive (PlainRender) flush.
	RenderLive(tasks []*Task) api.Text
	// RenderFinal returns the content drawn once after all tasks complete.
	RenderFinal(tasks []*Task) api.Text
}

LiveRenderer replaces the default task-tree rendering with caller-supplied content while keeping clicky's terminal ownership: the render loop still owns the TTY, ClearLines line accounting, and the logger serializer that prevents concurrent log lines from corrupting the in-place frame. Only the rendered content (an api.Text) is swapped — a caller that wants its own block (a status table, a custom dashboard) renders it through clicky instead of hand-rolling ANSI redraws that collide with logger output.

Install with SetLiveRenderer; remove by passing nil. Both methods receive a snapshot of the current tasks and must not mutate them.

type LogEntry 

type LogEntry struct {
	Level   string `json:"level"`
	Message string `json:"message"`
}

type ManagedRun added in v1.21.48 

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

ManagedRun tracks externally-driven work without occupying a task worker. It is intended for supervisors and other long-lived lifecycle owners.

func StartManagedRun added in v1.21.48 

func StartManagedRun(name string, opts ...TaskGroupOption) *ManagedRun

StartManagedRun creates a one-task group already in the running state.

func (*ManagedRun) Finish added in v1.21.48 

func (r *ManagedRun) Finish(status Status, err error)

Finish freezes live providers and transitions the run to a terminal status.

func (*ManagedRun) Group added in v1.21.60 

func (r *ManagedRun) Group() *Group

Group attaches executable steps to this externally-owned run.

func (*ManagedRun) ID added in v1.21.48 

func (r *ManagedRun) ID() string

func (*ManagedRun) SetBackground added in v1.21.57 

func (r *ManagedRun) SetBackground(background bool)

SetBackground marks the run's task as long-lived, so global waits skip it instead of blocking on a server that never finishes. See Task.SetBackground.

func (*ManagedRun) SetDetailsProvider added in v1.21.48 

func (r *ManagedRun) SetDetailsProvider(provider func() any)

func (*ManagedRun) SetHref added in v1.21.48 

func (r *ManagedRun) SetHref(href string)

SetHref updates the route advertised by the run.

func (*ManagedRun) SetOutputProvider added in v1.21.48 

func (r *ManagedRun) SetOutputProvider(provider func() OutputSnapshot)

func (*ManagedRun) SetWorkProgress added in v1.21.60 

func (r *ManagedRun) SetWorkProgress(progress WorkProgress)

SetWorkProgress publishes a coherent set of counts. A run's total is immutable.

func (*ManagedRun) Task added in v1.21.48 

func (r *ManagedRun) Task() *Task

func (*ManagedRun) TaskID added in v1.21.48 

func (r *ManagedRun) TaskID() string

type Manager 

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

Manager manages and displays multiple tasks with progress bars

func (*Manager) PlainRender 

func (tm *Manager) PlainRender()

PlainRender outputs the current task statuses in plain text without any interactive / ANSI / console features

func (*Manager) Pretty 

func (tm *Manager) Pretty() api.Text

func (*Manager) Run 

func (tm *Manager) Run() error

Run starts all tasks and waits for completion

func (*Manager) Start 

func (tm *Manager) Start(name string, opts ...Option) *Task

Start creates and starts tracking a new task

func (*Manager) StartCapturingOutput 

func (tm *Manager) StartCapturingOutput()

StartCapturingOutput redirects stdout/stderr to internal buffer

func (*Manager) StartWithResult 

func (tm *Manager) StartWithResult(name string, taskFunc func(flanksourceContext.Context, *Task) (interface{}, error), opts ...Option) *Task

StartWithResult creates and starts tracking a new task with typed result handling

func (*Manager) StopCapturingOutput 

func (tm *Manager) StopCapturingOutput()

StopCapturingOutput restores stdout/stderr and prints buffered output

type ManagerOptions 

type ManagerOptions struct {
	NoProgress      bool          // Disable progress display
	NoRender        bool          // Disable all task rendering
	MaxConcurrent   int           // Maximum concurrent tasks (0 = unlimited)
	GracefulTimeout time.Duration // Timeout for graceful shutdown

	// Retry configuration
	MaxRetries int           // Maximum retry attempts
	RetryDelay time.Duration // Base delay between retries
}

ManagerOptions contains configuration options for TaskManager

func DefaultManagerOptions 

func DefaultManagerOptions() *ManagerOptions

DefaultManagerOptions returns sensible defaults

func (*ManagerOptions) Apply 

func (opts *ManagerOptions) Apply()

Apply configures a TaskManager with these options

type MetricSource added in v1.21.48 

type MetricSource interface {
	QueryMetric(context.Context, metrics.QueryRequest) ([]metrics.Point, error)
}

MetricSource is optionally implemented by a RunSource that owns live metrics.

type Option 

type Option func(*Task)

Option configures task creation

func WithCancellationDrain added in v1.21.60 

func WithCancellationDrain() Option

WithCancellationDrain makes cancellation waits block until the callback returns.

func WithContext added in v1.21.60 

func WithContext(parent context.Context) Option

WithContext links task cancellation and deadlines to its owner before enqueue. Multiple owners (a run group and a request, for example) all remain effective.

func WithDependencies 

func WithDependencies(deps ...*Task) Option

WithDependencies sets tasks that must complete before this task can start. A dependency that failed cancels this task with "dependency failed"; a dependency that was cancelled does not — this task waits for it to unwind and then runs, since cancelling a task is a decision about that task alone.

func WithFunc 

func WithFunc(fn func(flanksourceContext.Context, *Task) error) Option

WithFunc sets the function to run for the task

func WithGroup added in v1.21.48 

func WithGroup(g *Group) Option

WithGroup attaches a directly-started task (including exec.RunAsTask and exec.StartAsTask) to an existing run group before it is enqueued.

func WithIdentity 

func WithIdentity(identity string) Option

WithIdentity sets a unique identifier for task deduplication

func WithModel 

func WithModel(modelName string) Option

WithModel sets the model name for the task

func WithPriority 

func WithPriority(priority int) Option

WithPriority sets the priority for task scheduling (lower = higher priority)

func WithPrompt 

func WithPrompt(prompt string) Option

WithPrompt sets the prompt for the task

func WithRetryConfig 

func WithRetryConfig(config RetryConfig) Option

WithRetryConfig sets custom retry configuration for the task

func WithTaskController added in v1.21.48 

func WithTaskController(controller TaskController) Option

WithTaskController exposes task-scoped controls such as stop and stdin.

func WithTaskTimeout 

func WithTaskTimeout(d time.Duration) Option

WithTaskTimeout sets an individual task timeout applied at execution time

func WithTimeout 

func WithTimeout(d time.Duration) Option

WithTimeout sets a timeout for the task

type OutputEntry 

type OutputEntry struct {
	Timestamp time.Time
	Stream    string // "stdout" or "stderr"
	Line      string
}

OutputEntry represents a captured stdout/stderr line with metadata

type OutputSnapshot added in v1.21.48 

type OutputSnapshot struct {
	Stdout string `json:"stdout,omitempty"`
	Stderr string `json:"stderr,omitempty"`
}

OutputSnapshot is the live output projected by an executing task.

type OverlapPolicy added in v1.21.60 

type OverlapPolicy string

OverlapPolicy decides what happens when a schedule fires while its previous run is still going. There is no safe default for every workload — a cheap health probe wants to skip, an hourly report wants to queue — so it is a per-schedule choice rather than a scheduler-wide rule. 

const (
	// OverlapSkip drops the fire and records why. The default.
	OverlapSkip OverlapPolicy = "skip"
	// OverlapQueue runs the new fire once the previous one finishes. A schedule
	// that is persistently slower than its interval will back up.
	OverlapQueue OverlapPolicy = "queue"
	// OverlapCancelPrevious stops the in-flight run and starts the new one, for
	// schedules where only the latest result is worth anything.
	OverlapCancelPrevious OverlapPolicy = "cancel-previous"
)

type RetryConfig 

type RetryConfig struct {
	RetryableErrors []string // Error message patterns that should trigger retries
	BaseDelay       time.Duration
	MaxDelay        time.Duration
	BackoffFactor   float64
	JitterFactor    float64
	MaxRetries      int
}

RetryConfig holds configuration for task retry behavior

func DefaultRetryConfig 

func DefaultRetryConfig() RetryConfig

DefaultRetryConfig returns sensible default retry configuration

type RunFilter 

type RunFilter struct {
	Kind   string
	Status string
	Labels map[string]string // every entry must match
}

RunFilter narrows the runs returned by Runs. Empty fields match everything.

func (RunFilter) Matches 

func (f RunFilter) Matches(m RunMeta) bool

type RunMeta 

type RunMeta struct {
	ID         string            `json:"id"`
	Name       string            `json:"name"`
	Kind       string            `json:"kind,omitempty"`
	Labels     map[string]string `json:"labels,omitempty"`
	Owner      string            `json:"owner,omitempty"`
	Status     string            `json:"status"`
	StartedAt  string            `json:"startedAt,omitempty"`  // RFC3339
	FinishedAt string            `json:"finishedAt,omitempty"` // RFC3339
	Total      int               `json:"total"`
	Completed  int               `json:"completed"`
	Failed     int               `json:"failed"`
	Running    int               `json:"running"`
	Canceled   int               `json:"canceled,omitempty"`
	Work       *WorkProgress     `json:"work,omitempty"`
	Href       string            `json:"href,omitempty"`
	Controls   []ControlAction   `json:"controls,omitempty"`
}

RunMeta is the listing summary for one run (task group): identity, metadata, status, timing, and child-task counts. It is what the generic task-manager list view renders; drill-down uses SnapshotByID.

func RunMetaFromSnapshot 

func RunMetaFromSnapshot(snap TaskSnapshot) RunMeta

RunMetaFromSnapshot lifts the group-level fields of a group snapshot into a RunMeta. The snapshot's ID is the group name; GroupID carries the stable id.

func Runs 

func Runs(filter RunFilter) []RunMeta

Runs returns one RunMeta per registered group, newest-first, optionally narrowed by filter. It runs GC first so stale finished runs drop out.

func RunsRaw 

func RunsRaw(filter RunFilter) []RunMeta

RunsRaw is like Runs but does NOT trigger GC first. Callers that manage their own GC timing (e.g. an L2-backed wrapper that needs to snapshot before GC) use this to avoid double-GC.

type RunRecord added in v1.21.60 

type RunRecord struct {
	RunMeta
	Snapshots []TaskSnapshot
}

RunRecord is a run flattened for storage: the listing columns a store indexes and filters on, plus the full snapshot slice a drill-down needs. Keeping both means a listing never has to decode the snapshots to answer.

func RunFromSnapshots added in v1.21.60 

func RunFromSnapshots(snapshots []TaskSnapshot) (RunRecord, bool)

RunFromSnapshots builds a RunRecord from a group's snapshot slice — the group snapshot followed by its child tasks, as produced by SnapshotByID and by the GC hook. It reports false when the slice carries no group snapshot, which means there is nothing to persist.

type RunSource added in v1.21.48 

type RunSource interface {
	Runs(context.Context, RunFilter) ([]RunMeta, error)
	Snapshot(context.Context, string) ([]TaskSnapshot, error)
	Control(context.Context, string, ControlAction) error
}

RunSource exposes task runs owned by another process or durable store. In-memory runs always win when an ID exists in both sources.

type Runner added in v1.21.60 

type Runner func(ctx flanksourceContext.Context, schedule Schedule, group *Group) error

Runner is the work behind a schedule. The group is already registered and stamped with the schedule's kind and labels; the runner adds tasks to it.

type Schedule added in v1.21.60 

type Schedule struct {
	Name string `json:"name"`

	// Kind selects the registered runner and is stamped on every run, so a
	// listing can filter to one schedule's kind of work.
	Kind   string            `json:"kind"`
	Labels map[string]string `json:"labels,omitempty"`
	Owner  string            `json:"owner,omitempty"`

	// Cron is a robfig/cron spec, including the descriptor forms ("@hourly",
	// "@every 5m"). Seconds are not accepted; the smallest unit is a minute.
	Cron string `json:"cron"`

	// Timezone is an IANA name. Empty means the process's local time, which is
	// what a bare "0 9 * * *" means to whoever typed it.
	Timezone string `json:"timezone,omitempty"`

	Enabled bool          `json:"enabled"`
	Timeout time.Duration `json:"timeout,omitempty"`

	Overlap OverlapPolicy `json:"overlap,omitempty"`
	CatchUp CatchUpPolicy `json:"catchUp,omitempty"`

	// LastRun and NextRun are maintained by the scheduler and persisted so a
	// restart can tell whether a scheduled time was missed.
	LastRun *time.Time `json:"lastRun,omitempty"`
	NextRun *time.Time `json:"nextRun,omitempty"`
}

Schedule is a recurring run: when to fire, what to stamp on the runs it produces, and how to behave when firing collides with reality.

The work itself is not here. A schedule names a Kind, and the host registers the function for that kind with RegisterRunner — the task package knows how to keep time, not what the work means.

func (Schedule) Location added in v1.21.60 

func (s Schedule) Location() (*time.Location, error)

Location resolves the schedule's timezone.

func (Schedule) Parse added in v1.21.60 

func (s Schedule) Parse() (cron.Schedule, error)

Parse compiles the cron spec in the schedule's own timezone.

func (Schedule) Validate added in v1.21.60 

func (s Schedule) Validate() error

Validate reports whether the schedule can actually be run, naming the field at fault. A schedule that cannot fire is a configuration error worth refusing at the point it is saved, not a run that silently never happens.

type Scheduler added in v1.21.60 

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

Scheduler fires schedules and turns each fire into a tracked run. It owns no work of its own: the run is a Group, so it is visible, filterable and controllable through exactly the same API as anything else the manager runs.

func NewScheduler added in v1.21.60 

func NewScheduler(options SchedulerOptions) *Scheduler

NewScheduler creates a scheduler. A nil store is allowed — schedules are then in-memory only and do not survive a restart, which is what a CLI wants.

func (*Scheduler) Add added in v1.21.60 

func (s *Scheduler) Add(ctx context.Context, schedule Schedule) error

Add registers or replaces one schedule and persists the definition. The next fire is computed from now, except that a schedule asking to catch up on a scheduled time missed while the process was down fires immediately instead.

The definition is stored before the entry is installed, so an error means the schedule is not registered and will not fire: a caller that saw Add fail never has to wonder whether it left a live schedule behind. That also keeps Load's error aggregate honest — a schedule reported as failed to load is one that is genuinely absent from the scheduler.

func (*Scheduler) Load added in v1.21.60 

func (s *Scheduler) Load(ctx context.Context) error

Load reads the persisted schedules into the scheduler, replacing whatever it held. Schedules that no longer validate are reported rather than dropped silently: a schedule that stopped being runnable is the operator's problem to see, not the scheduler's to hide.

func (*Scheduler) Remove added in v1.21.60 

func (s *Scheduler) Remove(ctx context.Context, name string) error

Remove stops firing the named schedule and deletes its stored definition, so a restart does not resurrect it. The in-flight run, if any, is left to finish — cancelling work because its schedule was deleted loses a result nobody asked to throw away.

func (*Scheduler) RunDue added in v1.21.60 

func (s *Scheduler) RunDue(ctx flanksourceContext.Context)

RunDue fires every schedule whose time has come. It is the whole scheduling decision, exported so a test can drive it directly instead of waiting on a ticker.

func (*Scheduler) Schedules added in v1.21.60 

func (s *Scheduler) Schedules() []Schedule

Schedules returns the current schedules, newest fire first.

func (*Scheduler) Start added in v1.21.60 

func (s *Scheduler) Start(ctx flanksourceContext.Context)

Start runs the scheduler until ctx is done or Stop is called. It registers itself as a managed background run so it is visible in the task UI without occupying a worker or blocking Wait.

Starting an already-started or already-stopped scheduler is a no-op: there is exactly one loop, and it is the one thing allowed to close done.

func (*Scheduler) Stop added in v1.21.60 

func (s *Scheduler) Stop()

Stop ends the scheduler and waits for its loop to exit. Safe before Start and safe to repeat: a scheduler that never started has no loop to wait for, and one already stopped has nothing left to close.

type SchedulerOptions added in v1.21.60 

type SchedulerOptions struct {
	Store Store
	Now   func() time.Time
	Tick  time.Duration
}

SchedulerOptions configure a Scheduler. Now and Tick exist for tests; both have working defaults.

type Status 

type Status string

Status represents the status of a task 

const (
	// StatusPending indicates the task is waiting to start
	StatusPending Status = "pending"
	// StatusRunning indicates the task is currently running
	StatusRunning Status = "running"
	// StatusSuccess indicates the task completed successfully
	StatusSuccess Status = "success"
	// StatusFailed indicates the task failed
	StatusFailed Status = "failed"
	// StatusWarning indicates the task completed with warnings
	StatusWarning Status = "warning"
	// StatusCancelled indicates the task was canceled
	StatusCancelled Status = "canceled"

	// StatusPASS indicates a test passed
	StatusPASS Status = "PASS"
	// StatusFAIL indicates a test failed
	StatusFAIL Status = "FAIL"
	// StatusERR indicates a test had an error
	StatusERR Status = "ERR"
	// StatusSKIP indicates a test was skipped
	StatusSKIP Status = "SKIP"
)

func (Status) Apply 

func (s Status) Apply(t api.Text) api.Text

Apply applies the status icon and style to the given text, preserving any style classes (such as width/truncation directives) the caller has already set.

func (Status) Health 

func (s Status) Health() Health

Health converts the status to a health state

func (Status) Icon 

func (s Status) Icon() string

Icon returns the emoji icon representation of the status

func (Status) Pretty 

func (s Status) Pretty() api.Text

Pretty returns a pretty formatted text representation of the status

func (Status) String 

func (s Status) String() string

func (Status) Style 

func (s Status) Style() string

Style returns the CSS style class for the status

type StdinController added in v1.21.48 

type StdinController interface {
	WriteStdin([]byte) error
}

StdinController is implemented by task controllers that accept live stdin. Stdin is intentionally never included in snapshots or persisted history.

type Store added in v1.21.60 

type Store interface {
	RunSource

	// SaveRun persists one run: the group snapshot followed by its child task
	// snapshots. It is called when a run reaches a terminal status and again if
	// the run is still in memory when GC evicts it, so it must be idempotent —
	// the newest snapshot for a given id wins.
	SaveRun(ctx context.Context, groupID string, snapshots []TaskSnapshot) error

	ListSchedules(ctx context.Context) ([]Schedule, error)
	SaveSchedule(ctx context.Context, schedule Schedule) error
	DeleteSchedule(ctx context.Context, name string) error

	// RecordFire records one firing decision — a run that started, or one that
	// was deliberately not started. A skipped fire is a fact about the schedule
	// worth keeping, not an absence of one.
	RecordFire(ctx context.Context, name string, fire Fire) error
}

Store is the durable half of the task manager. It extends RunSource — which already answers reads for runs this process no longer holds in memory — with the writes that put them there, and with the schedule definitions the scheduler replays after a restart.

Every method is called from a background goroutine, never from a task's own goroutine and never while the manager holds a lock, so an implementation may block on IO. It must not call back into the task package.

func CurrentStore added in v1.21.60 

func CurrentStore() Store

CurrentStore returns the installed store, or nil.

type Task 

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

Task represents a single task being tracked by the TaskManager

func (*Task) Cancel 

func (t *Task) Cancel()

Cancel cancels the task

func (*Task) ClearLogs 

func (t *Task) ClearLogs()

ClearLogs clears all buffered logs for this task

func (*Task) Context 

func (t *Task) Context() context.Context

Context returns the task's context for cancellation

func (*Task) Debugf 

func (t *Task) Debugf(format string, args ...interface{})

Debugf logs a debug message (only shown in verbose mode)

func (*Task) Description 

func (t *Task) Description() string

Description returns the task description

func (*Task) Duration 

func (t *Task) Duration() time.Duration

Duration returns the task duration

func (*Task) EndTime added in v1.21.38 

func (t *Task) EndTime() time.Time

EndTime returns when the task reached a terminal state, or the zero time if it is still pending/running.

func (*Task) Error 

func (t *Task) Error() error

Error returns the task's error if any

func (*Task) Errorf 

func (t *Task) Errorf(format string, args ...interface{})

Errorf logs an error message

func (*Task) Failed 

func (t *Task) Failed() *Task

Failed marks the task as failed

func (*Task) FailedWithError 

func (t *Task) FailedWithError(err error) (*Task, error)

FailedWithError marks the task as failed with an error

func (*Task) Fatal 

func (t *Task) Fatal(err error)

Fatal marks the task as failed and exits the program immediately

func (*Task) Fatalf 

func (t *Task) Fatalf(format string, args ...interface{})

Fatalf logs a fatal message (implements Logger interface)

func (*Task) FlanksourceContext 

func (t *Task) FlanksourceContext() flanksourceContext.Context

FlanksourceContext returns the task's flanksource context for logging

func (*Task) GetLevel 

func (t *Task) GetLevel() logger.LogLevel

GetLevel returns the current log level (implements Logger interface)

func (*Task) GetResult 

func (t *Task) GetResult() (interface{}, error)

GetResult returns the stored result and error

func (*Task) GetSlogLogger 

func (t *Task) GetSlogLogger() *slog.Logger

GetSlogLogger returns the slog logger (implements Logger interface - unsupported)

func (*Task) GetTask 

func (t *Task) GetTask() *Task

GetTask returns the task itself

func (*Task) GetTypedResult 

func (t *Task) GetTypedResult(target interface{}) error

GetTypedResult retrieves the result with type assertion

func (*Task) ID 

func (t *Task) ID() string

ID returns the task's immutable UUID.

func (*Task) Identity 

func (t *Task) Identity() string

Identity returns the task's unique identifier for deduplication

func (*Task) Infof 

func (t *Task) Infof(format string, args ...interface{})

Infof logs an info message (only shown in verbose mode)

func (*Task) IsBackground added in v1.21.57 

func (t *Task) IsBackground() bool

IsBackground reports whether waits skip this task. See SetBackground.

func (*Task) IsDebugEnabled 

func (t *Task) IsDebugEnabled() bool

IsDebugEnabled checks if debug level is enabled (implements Logger interface)

func (*Task) IsGroup 

func (t *Task) IsGroup() bool

IsGroup returns false for Task

func (*Task) IsLevelEnabled 

func (t *Task) IsLevelEnabled(level logger.LogLevel) bool

IsLevelEnabled checks if a specific level is enabled (implements Logger interface)

func (*Task) IsOk 

func (t *Task) IsOk() bool

IsOk returns true if the task completed successfully

func (*Task) IsTraceEnabled 

func (t *Task) IsTraceEnabled() bool

IsTraceEnabled checks if trace level is enabled (implements Logger interface)

func (*Task) Name 

func (t *Task) Name() string

Name returns the task name

func (*Task) Named 

func (t *Task) Named(name string) logger.Logger

Named returns a named logger (implements Logger interface - noop)

func (*Task) PopDirty 

func (t *Task) PopDirty() bool

PopDirty checks and clears the dirty flag atomically

func (*Task) Pretty 

func (t *Task) Pretty() api.Text

Pretty returns a formatted text representation of the task with its full buffered log history.

func (*Task) Progress 

func (t *Task) Progress() (value, maximum int)

Progress returns the task's current progress value and maximum. A maximum of 0 means the task has no bounded progress.

func (*Task) SetBackground added in v1.21.57 

func (t *Task) SetBackground(background bool)

SetBackground marks the task as long-lived: a server whose contract is to outlive any wait, rather than work a wait is entitled to drain. Waits skip such tasks entirely (see tasksDrained).

Counting a long-lived server in a drain deadlocks every caller, because the two sides need opposite things: the wait blocks the work that would shut the server down, and the running server keeps the wait from returning. A supervised agent-provider process is the canonical case — it must stay alive across turns, so a commit hook that drains global tasks mid-run hangs forever.

It is opt-in: a supervised process that IS the work (`gavel proc run`) must keep blocking the wait.

func (*Task) SetController added in v1.21.48 

func (t *Task) SetController(controller TaskController)

SetController replaces the task's live controller.

func (*Task) SetDescription 

func (t *Task) SetDescription(description string)

SetDescription sets the task description

func (*Task) SetDetailsProvider added in v1.21.48 

func (t *Task) SetDetailsProvider(provider func() any)

SetDetailsProvider attaches structured, JSON-serializable task details.

func (*Task) SetLogLevel 

func (t *Task) SetLogLevel(level any)

SetLogLevel sets the log level (implements Logger interface)

func (*Task) SetMinLogLevel 

func (t *Task) SetMinLogLevel(level any)

SetMinLogLevel sets the minimum log level (implements Logger interface)

func (*Task) SetName 

func (t *Task) SetName(name string)

SetName sets the task name

func (*Task) SetOutputProvider added in v1.21.48 

func (t *Task) SetOutputProvider(provider func() OutputSnapshot)

SetOutputProvider attaches live stdout/stderr to the task snapshot. Providers are evaluated outside the task lock and frozen when a managed task finishes.

func (*Task) SetProgress 

func (t *Task) SetProgress(value, maximum int)

SetProgress updates the task's progress

func (*Task) SetResult 

func (t *Task) SetResult(result interface{})

SetResult stores a result in the task

func (*Task) SetStatus 

func (t *Task) SetStatus(status Status)

SetStatus updates the task's display name/status message

func (*Task) StartTime 

func (t *Task) StartTime() time.Time

StartTime returns when the task started execution

func (*Task) Status 

func (t *Task) Status() Status

Status returns the current task status

func (*Task) Success 

func (t *Task) Success() *Task

Success marks the task as successfully completed

func (*Task) Tracef 

func (t *Task) Tracef(format string, args ...interface{})

Tracef logs a trace message (implements Logger interface)

func (*Task) V 

func (t *Task) V(level any) logger.Verbose

V returns a verbose logger (implements Logger interface)

func (*Task) WaitFor 

func (t *Task) WaitFor() *WaitResult

WaitFor waits for completion. Cancellation notifies legacy waiters immediately; WithCancellationDrain opts into waiting until the callback has returned.

func (*Task) WaitTime 

func (t *Task) WaitTime() time.Duration

WaitTime returns how long the task waited before starting

func (*Task) Warnf 

func (t *Task) Warnf(format string, args ...interface{})

Warnf logs a warning message

func (*Task) Warning 

func (t *Task) Warning() *Task

Warning marks the task as completed with warnings

func (*Task) WithSkipReportLevel 

func (t *Task) WithSkipReportLevel(i int) logger.Logger

WithSkipReportLevel returns a logger with skip report level (implements Logger interface - noop)

func (*Task) WithV 

func (t *Task) WithV(level any) logger.Logger

WithV returns a logger with verbosity level (implements Logger interface)

func (*Task) WithValues 

func (t *Task) WithValues(keysAndValues ...interface{}) logger.Logger

WithValues returns a logger with additional key-value pairs (implements Logger interface)

func (*Task) WithoutName 

func (t *Task) WithoutName() logger.Logger

WithoutName returns a logger without name (implements Logger interface - noop)

type TaskControlSource added in v1.21.54 

type TaskControlSource interface {
	ControlTask(context.Context, string, string, ControlAction) error
}

TaskControlSource is implemented by run sources that can route lifecycle controls to child tasks owned outside this process.

type TaskController added in v1.21.48 

type TaskController interface {
	Actions() []ControlAction
	Control(context.Context, ControlAction) error
}

TaskController performs the currently-supported lifecycle actions for a run or task. Actions must reflect live state and may change between snapshots.

type TaskFunc 

type TaskFunc[T any] func(flanksourceContext.Context, *Task) (T, error)

TaskFunc is a generic task function that returns a typed result

type TaskGroupOption 

type TaskGroupOption func(group *Group)

func WithConcurrency 

func WithConcurrency(concurrency int) TaskGroupOption

func WithController added in v1.21.48 

func WithController(controller TaskController) TaskGroupOption

WithController exposes lifecycle controls for the run.

func WithDetailsProvider added in v1.21.48 

func WithDetailsProvider(provider func() any) TaskGroupOption

WithDetailsProvider attaches structured live details to the group snapshot.

func WithGroupID 

func WithGroupID(id string) TaskGroupOption

WithGroupID sets a caller-supplied stable id for the group. When unset, StartGroup assigns a uuid.

func WithHref added in v1.21.48 

func WithHref(href string) TaskGroupOption

WithHref sets the route a task manager should follow when the run is opened.

func WithKind 

func WithKind(kind string) TaskGroupOption

WithKind classifies the run for the task-manager listing/filtering.

func WithLabels 

func WithLabels(labels map[string]string) TaskGroupOption

WithLabels attaches filterable key/value facets to the run.

func WithOwner 

func WithOwner(owner string) TaskGroupOption

WithOwner records who started the run.

type TaskResult 

type TaskResult[T any] struct {
	Result T
	Error  error
}

TaskResult holds a typed result and error

type TaskSnapshot 

type TaskSnapshot struct {
	ID        string        `json:"id"`
	Name      string        `json:"name"`
	Type      string        `json:"type"`            // "task" or "group"
	Group     string        `json:"group,omitempty"` // parent group name
	Status    string        `json:"status"`
	Duration  string        `json:"duration,omitempty"`
	Error     string        `json:"error,omitempty"`
	Message   string        `json:"message,omitempty"`   // latest log line
	Logs      []LogEntry    `json:"logs,omitempty"`      // all log entries
	Total     int           `json:"total,omitempty"`     // group: total child tasks
	Completed int           `json:"completed,omitempty"` // group: completed tasks
	Failed    int           `json:"failed,omitempty"`    // group: failed tasks
	Running   int           `json:"running,omitempty"`   // group: running tasks
	Canceled  int           `json:"canceled,omitempty"`
	Work      *WorkProgress `json:"work,omitempty"`

	// Per-task fields (type == "task"). Description is the live stage label set
	// via Task.SetDescription; Progress/MaxValue mirror Task.SetProgress so the UI
	// can render an x/y count and percent for a running task. MaxValue 0 means the
	// task has no bounded progress.
	Description string `json:"description,omitempty"`
	Progress    int    `json:"progress,omitempty"`
	MaxValue    int    `json:"maxValue,omitempty"`

	// Registry metadata (additive). For a group these describe the run itself;
	// for a task GroupID links it to its parent run so the SSE/JSON clients can
	// key on a stable id rather than the human-facing name.
	GroupID         string            `json:"groupId,omitempty"`
	Kind            string            `json:"kind,omitempty"`
	Labels          map[string]string `json:"labels,omitempty"`
	Owner           string            `json:"owner,omitempty"`
	StartedAt       string            `json:"startedAt,omitempty"`  // RFC3339
	FinishedAt      string            `json:"finishedAt,omitempty"` // RFC3339
	Href            string            `json:"href,omitempty"`
	Controls        []ControlAction   `json:"controls,omitempty"`
	Stdout          string            `json:"stdout,omitempty"`
	Stderr          string            `json:"stderr,omitempty"`
	StdoutTruncated bool              `json:"stdoutTruncated,omitempty"`
	StderrTruncated bool              `json:"stderrTruncated,omitempty"`
	Details         any               `json:"details,omitempty"`
}

TaskSnapshot is a JSON-serializable snapshot of a task or group's current state.

func SnapshotAll 

func SnapshotAll(taskIDs ...string) []TaskSnapshot

SnapshotAll returns snapshots for all groups and their tasks. If taskIDs is non-empty, only groups whose name OR stable id matches are included (matching by id lets the registry/SSE drill into one run; matching by name preserves the legacy name-keyed callers).

func SnapshotByID 

func SnapshotByID(id string) []TaskSnapshot

SnapshotByID returns the group + task snapshots for the run with the given stable id (not name). Returns nil when no such run exists.

func SnapshotGroup 

func SnapshotGroup(g *Group) TaskSnapshot

SnapshotGroup creates a TaskSnapshot from a Group with aggregate child stats. The snapshot ID stays the group NAME for backward compatibility with the name-keyed Preact UI and JSONHandler; the stable id is carried separately in GroupID. Observing a terminal status here records finishedAt lazily.

func SnapshotTask 

func SnapshotTask(t *Task, group *Group) TaskSnapshot

SnapshotTask creates a TaskSnapshot from a Task. group is the parent group, or nil for an ungrouped task; its name and id are recorded on the snapshot.

type Taskable 

type Taskable interface {
	GetTask() *Task
}

Taskable represents objects that can return a Task

type TypedGroup 

type TypedGroup[T any] struct {
	*Group
}

func StartGroup 

func StartGroup[T any](name string, opts ...TaskGroupOption) TypedGroup[T]

StartGroup creates and starts tracking a new task group

func (TypedGroup[T]) Add 

func (g TypedGroup[T]) Add(name string, taskFunc func(flanksourceContext.Context, *Task) (T, error), opts ...Option) TypedTask[T]

Add adds a Waitable item (Task or Group) to this group

func (*TypedGroup[T]) Duration 

func (g *TypedGroup[T]) Duration() time.Duration

Duration returns the total duration from first start to last completion

func (TypedGroup[T]) GetResults 

func (g TypedGroup[T]) GetResults() (map[TypedTask[T]]T, error)

GetResults waits for all tasks in the group and returns typed results

func (TypedGroup[T]) IsGroup 

func (g TypedGroup[T]) IsGroup() bool

IsGroup returns true for Group

func (*TypedGroup[T]) WaitFor 

func (g *TypedGroup[T]) WaitFor() *WaitResult

WaitFor waits for all child items to complete and returns aggregate results This version handles dynamically added tasks by continuously checking for new tasks

type TypedTask 

type TypedTask[T any] struct {
	*Task
}

TypedTask provides typed access to task results

func StartTask 

func StartTask[T any](name string, taskFunc func(flanksourceContext.Context, *Task) (T, error), opts ...Option) TypedTask[T]

func (TypedTask[T]) GetResult 

func (t TypedTask[T]) GetResult() (T, error)

GetResult retrieves the typed result from a TypedTask

type WaitResult 

type WaitResult struct {
	Error        error
	Status       Status
	Duration     time.Duration
	TaskCount    int // Number of individual tasks (1 for Task, N for TaskGroup)
	SuccessCount int // Number of successful tasks
	FailureCount int // Number of failed tasks
	WarningCount int // Number of tasks with warnings
}

WaitResult contains unified result information

type Waitable 

type Waitable interface {
	Name() string
	Status() Status
	WaitFor() *WaitResult
	Context() context.Context
	Cancel()
	Duration() time.Duration
	IsGroup() bool
}

Waitable represents something that can be waited on (Task or TaskGroup)

type WorkProgress added in v1.21.60 

type WorkProgress struct {
	Total     int `json:"total"`
	Completed int `json:"completed"`
	Cached    int `json:"cached"`
	Failed    int `json:"failed"`
	Running   int `json:"running"`
	Canceled  int `json:"canceled"`
	Unstarted int `json:"unstarted"`
}

WorkProgress counts fixed units of work independently from implementation tasks. Completed excludes Cached; remaining units are queued until made Unstarted.

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.