tasks

package
v0.1.0-p8a.2 Latest Latest
Warning

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

 Go to latest Published: Aug 10, 2026 License: MIT Imports: 17 Imported by: 0

Documentation

Overview

Package tasks provides durable task/run bookkeeping for background work.

The registry is intentionally small: it records task lifecycle, completion summaries, and delivery state with bounded retention. It does not deliver messages itself. The intended architecture is: 

TaskRegistry -> typed async completion event -> delivery coordinator

Today the registry is used by spawn, delegate, cron execution, and status tools so completed background work survives manager recreation and service restarts. Additional long-running runtimes should prefer this registry instead of inventing parallel status stores.

Index

Constants

View Source 
const (
	DefaultTerminalRetention = 7 * 24 * time.Hour
	DefaultMaxRecords        = 1000
	DefaultMaxEvents         = 5000
	DefaultMaxSnapshotBytes  = 2 * 1024 * 1024
	TaskEventSchemaVersion   = "task_event.v2"
	DeliverableReportV1      = "deliverable_report.v1"
)

Variables

View Source 
var ErrTaskAlreadyExists = errors.New("task already exists")

Functions

func ValidateSnapshot 

func ValidateSnapshot(storePath string) error

ValidateSnapshot reads and validates a registry snapshot without pruning or writing it.

func WorkspaceStorePath 

func WorkspaceStorePath(workspace string) string

Types

type CompletionMedia 

type CompletionMedia struct {
	Ref         string `json:"ref"`
	Type        string `json:"type,omitempty"`
	Filename    string `json:"filename,omitempty"`
	ContentType string `json:"content_type,omitempty"`
}

type CompletionPayload 

type CompletionPayload struct {
	Text  string            `json:"text,omitempty"`
	Media []CompletionMedia `json:"media,omitempty"`
}

type DeliverableItem 

type DeliverableItem struct {
	Ref         string `json:"ref"`
	Kind        string `json:"kind,omitempty"`
	Filename    string `json:"filename,omitempty"`
	ContentType string `json:"content_type,omitempty"`
	Delivered   bool   `json:"delivered,omitempty"`
}

type DeliverablePayload 

type DeliverablePayload struct {
	Text      string             `json:"text,omitempty"`
	Artifacts []DeliverableItem  `json:"artifacts,omitempty"`
	Metadata  map[string]string  `json:"metadata,omitempty"`
	Report    *DeliverableReport `json:"report,omitempty"`
}

type DeliverableReport 

type DeliverableReport struct {
	SchemaVersion string             `json:"schema_version"`
	ReportID      string             `json:"report_id"`
	ContentHash   string             `json:"content_hash"`
	GeneratedAt   int64              `json:"generated_at"`
	Summary       string             `json:"summary,omitempty"`
	Claims        []ReportClaim      `json:"claims,omitempty"`
	FieldDeltas   []ReportFieldDelta `json:"field_deltas,omitempty"`
	Provenance    map[string]string  `json:"provenance,omitempty"`
	Metadata      map[string]string  `json:"metadata,omitempty"`
	Extra         map[string]any     `json:"extra,omitempty"`
}

DeliverableReport is a versioned canonical report for durable outputs. The surrounding DeliverablePayload remains the compatibility projection for older tools; Report is the schemaed contract new consumers should prefer.

type DeliveryStatus 

type DeliveryStatus string
const (
	DeliveryPending       DeliveryStatus = "pending"
	DeliveryDelivered     DeliveryStatus = "delivered"
	DeliverySessionQueued DeliveryStatus = "session_queued"
	DeliveryFailed        DeliveryStatus = "failed"
	DeliveryParentMissing DeliveryStatus = "parent_missing"
	DeliveryNotApplicable DeliveryStatus = "not_applicable"
)

type EventType 

type EventType string
const (
	EventTaskUpserted         EventType = "task.upserted"
	EventTaskStatusChanged    EventType = "task.status_changed"
	EventTaskDeliveryChanged  EventType = "task.delivery_changed"
	EventTaskDeliveryDecision EventType = "task.delivery_decision"
	EventTaskProgress         EventType = "task.progress"
	EventTaskUpdated          EventType = "task.updated"
	EventTaskReconciled       EventType = "task.reconciled"
)

type NotifyPolicy 

type NotifyPolicy string
const (
	NotifyDoneOnly     NotifyPolicy = "done_only"
	NotifyStateChanges NotifyPolicy = "state_changes"
	NotifySilent       NotifyPolicy = "silent"
)

type Options 

type Options struct {
	TerminalRetention time.Duration
	MaxRecords        int
	MaxEvents         int
	MaxSnapshotBytes  int
}

type Record 

type Record struct {
	TaskID              string              `json:"task_id"`
	GenerationID        string              `json:"generation_id"`
	LastEventSeq        int64               `json:"last_event_sequence"`
	Runtime             Runtime             `json:"runtime"`
	TaskKind            string              `json:"task_kind,omitempty"`
	ParentTaskID        string              `json:"parent_task_id,omitempty"`
	RequesterSessionKey string              `json:"requester_session_key,omitempty"`
	OwnerKey            string              `json:"owner_key,omitempty"`
	ScopeKind           string              `json:"scope_kind,omitempty"`
	Channel             string              `json:"channel,omitempty"`
	ChatID              string              `json:"chat_id,omitempty"`
	TopicID             string              `json:"topic_id,omitempty"`
	AgentID             string              `json:"agent_id,omitempty"`
	Label               string              `json:"label,omitempty"`
	Task                string              `json:"task"`
	Status              Status              `json:"status"`
	DeliveryStatus      DeliveryStatus      `json:"delivery_status"`
	NotifyPolicy        NotifyPolicy        `json:"notify_policy"`
	DeliveryMode        string              `json:"delivery_mode,omitempty"`
	TimeoutSeconds      int                 `json:"timeout_seconds,omitempty"`
	LastCompletionID    string              `json:"last_completion_id,omitempty"`
	DeliveredAt         int64               `json:"delivered_at,omitempty"`
	DeliveryError       string              `json:"delivery_error,omitempty"`
	CreatedAt           int64               `json:"created_at"`
	StartedAt           int64               `json:"started_at,omitempty"`
	EndedAt             int64               `json:"ended_at,omitempty"`
	LastEventAt         int64               `json:"last_event_at,omitempty"`
	CleanupAfter        int64               `json:"cleanup_after,omitempty"`
	Error               string              `json:"error,omitempty"`
	ProgressSummary     string              `json:"progress_summary,omitempty"`
	TerminalSummary     string              `json:"terminal_summary,omitempty"`
	InteractionID       string              `json:"interaction_id,omitempty"`
	InteractionShortID  string              `json:"interaction_short_id,omitempty"`
	InteractionSummary  string              `json:"interaction_summary,omitempty"`
	Completion          *CompletionPayload  `json:"completion,omitempty"`
	Deliverable         *DeliverablePayload `json:"deliverable,omitempty"`
}

type Registry 

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

func NewRegistry 

func NewRegistry(storePath string) *Registry

func NewRegistryWithOptions 

func NewRegistryWithOptions(storePath string, opts Options) *Registry

func (*Registry) AppendEvent 

func (r *Registry) AppendEvent(taskID string, eventType EventType, payload map[string]string) error

func (*Registry) CompleteInteractionTask 

func (r *Registry) CompleteInteractionTask(
	taskID, interactionID, content string,
	delivery DeliveryStatus,
) error

CompleteInteractionTask terminalizes a task only after its suspended continuation has produced and delivered the final result.

func (*Registry) Create 

func (r *Registry) Create(rec Record) error

Create persists a new task generation without replacing an existing task.

func (*Registry) FinishInteraction 

func (r *Registry) FinishInteraction(
	taskID, interactionID string,
	status Status,
	summary string,
) error

FinishInteraction projects a terminal interaction failure onto its owning task. Successful answers resume the task and are completed by the task owner.

func (*Registry) Get 

func (r *Registry) Get(taskID string) (Record, bool)

func (*Registry) Heartbeat 

func (r *Registry) Heartbeat(taskID, progress string) error

func (*Registry) LastLoadError 

func (r *Registry) LastLoadError() error

func (*Registry) List 

func (r *Registry) List() []Record

func (*Registry) ListActive 

func (r *Registry) ListActive() []Record

func (*Registry) ListEvents 

func (r *Registry) ListEvents(taskID string) []TaskEvent

func (*Registry) ListPendingTerminalDelivery 

func (r *Registry) ListPendingTerminalDelivery() []Record

func (*Registry) MarkActiveLost 

func (r *Registry) MarkActiveLost(reason string) (int, error)

func (*Registry) MarkInteractionRunning 

func (r *Registry) MarkInteractionRunning(taskID, interactionID string) error

MarkInteractionRunning returns a matching waiting task to running before its suspended continuation starts. The interaction ID is retained for audit and correlation while display-only waiting metadata is cleared.

func (*Registry) MarkStaleActiveLost 

func (r *Registry) MarkStaleActiveLost(maxAge time.Duration, reason string) (int, error)

func (*Registry) MarkWaitingForInput 

func (r *Registry) MarkWaitingForInput(
	taskID, interactionID, shortID, summary string,
) error

MarkWaitingForInput projects a durable human interaction onto a running task. Only bounded, user-safe interaction metadata belongs in the task registry.

func (*Registry) Stats 

func (r *Registry) Stats() Stats

Stats returns an exact serialized snapshot size and retention state.

func (*Registry) Update 

func (r *Registry) Update(taskID string, mutate func(*Record)) error

func (*Registry) Upsert 

func (r *Registry) Upsert(rec Record) error

type ReportClaim 

type ReportClaim struct {
	Kind       string            `json:"kind"`
	Text       string            `json:"text"`
	Confidence string            `json:"confidence,omitempty"`
	SourceRefs []string          `json:"source_refs,omitempty"`
	Metadata   map[string]string `json:"metadata,omitempty"`
}

type ReportFieldDelta 

type ReportFieldDelta struct {
	Field string `json:"field"`
	From  string `json:"from,omitempty"`
	To    string `json:"to,omitempty"`
}

type Runtime 

type Runtime string
const (
	RuntimeSubagent Runtime = "subagent"
	RuntimeDelegate Runtime = "delegate"
	RuntimeTool     Runtime = "tool"
	RuntimeCron     Runtime = "cron"
)

type Snapshot 

type Snapshot struct {
	Tasks  []Record    `json:"tasks"`
	Events []TaskEvent `json:"events,omitempty"`
}

type Stats 

type Stats struct {
	TaskCount          int
	EventCount         int
	ProtectedTaskCount int
	SnapshotBytes      int
	TerminalRetention  time.Duration
	MaxRecords         int
	MaxEvents          int
	MaxSnapshotBytes   int
	OverSnapshotBudget bool
}

Stats describes the current durable registry state and the retention limits that apply to it. Protected records are active, non-terminal, or have not reached a final delivery state, so retention never removes them.

type Status 

type Status string
const (
	StatusQueued          Status = "queued"
	StatusRunning         Status = "running"
	StatusWaitingForInput Status = "waiting_for_input"
	StatusSucceeded       Status = "succeeded"
	StatusFailed          Status = "failed"
	StatusTimedOut        Status = "timed_out"
	//nolint:misspell // External task status value intentionally uses British spelling for compatibility.
	StatusCancelled Status = "cancelled"
	StatusLost      Status = "lost"
)

type TaskEvent 

type TaskEvent struct {
	SchemaVersion  string            `json:"schema_version"`
	EventID        string            `json:"event_id"`
	TaskID         string            `json:"task_id"`
	GenerationID   string            `json:"generation_id"`
	Runtime        Runtime           `json:"runtime,omitempty"`
	ParentTaskID   string            `json:"parent_task_id,omitempty"`
	Type           EventType         `json:"type"`
	Status         Status            `json:"status,omitempty"`
	DeliveryStatus DeliveryStatus    `json:"delivery_status,omitempty"`
	Seq            int64             `json:"seq"`
	EmittedAt      int64             `json:"emitted_at"`
	Source         string            `json:"source,omitempty"`
	Producer       string            `json:"producer,omitempty"`
	Fingerprint    string            `json:"fingerprint,omitempty"`
	Payload        map[string]string `json:"payload,omitempty"`
}

TaskEvent is the append-only canonical event stream for task state. Records remain the current-state projection; chat, terminal, and status tools should render from records or reports, not treat prose output as source of truth.

Source Files

View all Source files

Jump to

Keyboard shortcuts

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