Documentation
¶
Overview ¶
Package task defines the shared task queue contract: interface, options, constants, error sentinels, and dead-letter manager.
This package is a neutral shared dependency — both producers (session, server) and consumers (queue/asynq implementation, agent workers) import it.
Index ¶
Constants ¶
const ( TaskTypeIntent = "intent" TaskTypeResume = "resume" TaskTypeWebhook = "webhook" )
Task type constants identify the handler for a queued task.
Variables ¶
var ErrQueueClosed = errors.New("queue closed")
ErrQueueClosed is returned when operations are performed on a closed queue.
Functions ¶
This section is empty.
Types ¶
type DeadTaskInfo ¶
type DeadTaskInfo struct {
TaskID string
Queue string
Payload []byte
ErrorMsg string
RetryCount int
LastFailedAt time.Time
}
DeadTaskInfo holds metadata from the underlying queue inspector for a dead task.
type EnqueueConfig ¶
EnqueueConfig holds the resolved configuration for an Enqueue call. Implementations use DefaultEnqueueConfig() to get defaults, then apply user options.
func DefaultEnqueueConfig ¶
func DefaultEnqueueConfig() EnqueueConfig
DefaultEnqueueConfig returns the default enqueue configuration. Used by implementations to apply defaults before user-specified options.
type EnqueueOption ¶
type EnqueueOption func(*EnqueueConfig)
EnqueueOption configures task enqueue behavior.
func WithConcurrencyGroup ¶
func WithConcurrencyGroup(groupID string) EnqueueOption
WithConcurrencyGroup sets the concurrency group. Tasks in the same group execute serially.
func WithMaxRetry ¶
func WithMaxRetry(n int) EnqueueOption
WithMaxRetry sets the maximum retry count. Default: 3.
func WithTimeout ¶
func WithTimeout(d time.Duration) EnqueueOption
WithTimeout sets the task timeout duration. Default: 120s.
type IntentPayload ¶
type IntentPayload struct {
Version int `json:"version"`
ConversationID string `json:"conversation_id"`
MessageID int64 `json:"message_id"`
Content string `json:"content"`
ConnID string `json:"conn_id,omitempty"`
TraceContext map[string]string `json:"trace_context,omitempty"`
}
IntentPayload represents the payload for an intent processing task.
func UnmarshalIntent ¶
func UnmarshalIntent(data []byte) (*IntentPayload, error)
UnmarshalIntent deserializes an IntentPayload from JSON, validating Version == 1.
func (*IntentPayload) Marshal ¶
func (p *IntentPayload) Marshal() ([]byte, error)
Marshal serializes the IntentPayload to JSON, setting Version to 1.
type Manager ¶
type Manager interface {
// RetryDeadTask re-enqueues a dead letter task for immediate retry.
RetryDeadTask(ctx context.Context, taskID string) error
// GetDeadTaskInfo returns metadata for a dead letter task.
GetDeadTaskInfo(ctx context.Context, taskID string) (*DeadTaskInfo, error)
}
Manager abstracts dead letter queue operations (e.g., Asynq Inspector).
type ResumePayload ¶
type ResumePayload struct {
Version int `json:"version"`
ConversationID string `json:"conversation_id"`
CheckpointID string `json:"checkpoint_id"`
ConnID string `json:"conn_id,omitempty"`
TraceContext map[string]string `json:"trace_context,omitempty"`
}
ResumePayload represents the payload for a resume processing task.
func UnmarshalResume ¶
func UnmarshalResume(data []byte) (*ResumePayload, error)
UnmarshalResume deserializes a ResumePayload from JSON, validating Version == 1.
func (*ResumePayload) Marshal ¶
func (p *ResumePayload) Marshal() ([]byte, error)
Marshal serializes the ResumePayload to JSON, setting Version to 1.
type TaskQueue ¶
type TaskQueue interface {
// Enqueue adds a task to the queue. taskType identifies the handler,
// payload is the serialized task data. Use WithConcurrencyGroup, WithMaxRetry,
// WithTimeout to configure per-task behavior.
Enqueue(ctx context.Context, taskType string, payload []byte, opts ...EnqueueOption) error
// Consume starts processing tasks. The handler is called for each task
// with the task type so the caller can dispatch to the correct handler.
// Consume blocks until the context is canceled or Close is called.
Consume(ctx context.Context, handler func(ctx context.Context, taskType string, payload []byte) error) error
// Close gracefully shuts down the queue, waiting for in-progress tasks.
Close() error
}
TaskQueue defines the asynchronous task processing layer. Tasks are enqueued by the session layer and consumed by workers with configurable concurrency and retry policies. Default implementation: Asynq (embedded, Redis-backed).
The name "TaskQueue" is used instead of "Queue" to be explicit about its purpose, even though it creates task.TaskQueue when imported.
type ToolResult ¶
type ToolResult struct {
Status string `json:"status"`
Output json.RawMessage `json:"output,omitempty"`
Error string `json:"error,omitempty"`
}
ToolResult holds the result of a tool execution.
type WebhookPayload ¶
type WebhookPayload struct {
Version int `json:"version"`
URL string `json:"url"`
Headers json.RawMessage `json:"headers,omitempty"`
Body json.RawMessage `json:"body"`
TraceContext map[string]string `json:"trace_context,omitempty"`
}
WebhookPayload represents the payload for a webhook delivery task.
func UnmarshalWebhook ¶
func UnmarshalWebhook(data []byte) (*WebhookPayload, error)
UnmarshalWebhook deserializes a WebhookPayload from JSON, validating Version == 1.
func (*WebhookPayload) Marshal ¶
func (p *WebhookPayload) Marshal() ([]byte, error)
Marshal serializes the WebhookPayload to JSON, setting Version to 1.
Source Files
Directories