Documentation
¶
Overview ¶
Package agenttask owns durable asynchronous agent execution.
Index ¶
- type EventSink
- type Options
- type Result
- type Runner
- type RuntimeRunner
- type Service
- func (service *Service) Await(ctx context.Context, taskID string) (*database.AgentTaskEntity, error)
- func (service *Service) Cancel(ctx context.Context, ownerSessionID string, taskID string, source string) (*database.TaskEntity, bool, error)
- func (service *Service) Events(ctx context.Context, taskID string, after int64, limit int) ([]database.TaskEventEntity, error)
- func (service *Service) Get(ctx context.Context, taskID string) (*database.AgentTaskEntity, bool, error)
- func (service *Service) List(ctx context.Context, ownerSessionID string, limit int) ([]database.AgentTaskEntity, error)
- func (service *Service) Shutdown(ctx context.Context) error
- func (service *Service) Start(ctx context.Context) error
- func (service *Service) Submit(ctx context.Context, request *SubmitRequest) (*database.AgentTaskEntity, error)
- func (service *Service) SubmitAgentTask(ctx context.Context, request *assistant.AgentTaskRequest) (*database.AgentTaskEntity, error)
- func (service *Service) Subscribe(taskID string) Subscription
- func (service *Service) SubscribeAgentTask(taskID string) (events <-chan database.TaskEventEntity, cancel func(), err error)
- type SubmitRequest
- type Subscription
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Options ¶
type Options struct {
Tasks *database.TaskRepository
AgentTasks *database.AgentTaskRepository
Workflows *database.WorkflowRepository
Runner Runner
Logger *slog.Logger
Timeout time.Duration
Concurrency int
SessionConcurrency int
QueueCapacity int
}
Options configures the task service.
type Runner ¶
type Runner interface {
Run(context.Context, *database.AgentTaskEntity, EventSink) (Result, error)
}
Runner executes one persisted agent task.
type RuntimeRunner ¶
type RuntimeRunner struct {
// contains filtered or unexported fields
}
RuntimeRunner executes durable tasks through the shared assistant runtime.
func NewRuntimeRunner ¶
func NewRuntimeRunner( runtime *assistant.Runtime, catalog *agent.Catalog, sessions *database.SessionRepository, tasks *database.AgentTaskRepository, ) (*RuntimeRunner, error)
NewRuntimeRunner creates an assistant runtime adapter for durable agent tasks.
func (*RuntimeRunner) Run ¶
func (runner *RuntimeRunner) Run( ctx context.Context, task *database.AgentTaskEntity, sink EventSink, ) (Result, error)
Run executes one task using the persisted agent definition and child session.
type Service ¶
type Service struct {
// contains filtered or unexported fields
}
Service schedules and owns durable agent tasks.
func NewStopped ¶
NewStopped creates a task service without starting its workers.
func (*Service) Await ¶
func (service *Service) Await(ctx context.Context, taskID string) (*database.AgentTaskEntity, error)
Await waits until durable task state is terminal or the caller context ends. Subscription events are fire-once wakeup hints: each one causes a single state recheck. The immediate read closes the subscribe/read race, and bounded polling repairs dropped wakeups, cross-process transitions, and subscription closure.
func (*Service) Cancel ¶
func (service *Service) Cancel( ctx context.Context, ownerSessionID string, taskID string, source string, ) (*database.TaskEntity, bool, error)
Cancel requests cancellation without allowing terminal states to change. The source records who requested cancellation for durable provenance.
func (*Service) Events ¶
func (service *Service) Events( ctx context.Context, taskID string, after int64, limit int, ) ([]database.TaskEventEntity, error)
Events returns durable task events after the requested sequence.
func (*Service) Get ¶
func (service *Service) Get(ctx context.Context, taskID string) (*database.AgentTaskEntity, bool, error)
Get returns an agent task by ID.
func (*Service) List ¶
func (service *Service) List( ctx context.Context, ownerSessionID string, limit int, ) ([]database.AgentTaskEntity, error)
List returns tasks owned by a session.
func (*Service) Submit ¶
func (service *Service) Submit(ctx context.Context, request *SubmitRequest) (*database.AgentTaskEntity, error)
Submit durably accepts a task before making it available to workers.
func (*Service) SubmitAgentTask ¶
func (service *Service) SubmitAgentTask( ctx context.Context, request *assistant.AgentTaskRequest, ) (*database.AgentTaskEntity, error)
SubmitAgentTask adapts the assistant tool boundary to the durable scheduler.
func (*Service) Subscribe ¶
func (service *Service) Subscribe(taskID string) Subscription
Subscribe follows events published after registration for a task. Delivery is bounded and best-effort: non-terminal events may be dropped on a full buffer, while a terminal event evicts one older event so it can wake the subscriber. Callers must re-read durable state after a wakeup and use Events plus sequence numbers when they need complete event contents.
func (*Service) SubscribeAgentTask ¶
func (service *Service) SubscribeAgentTask( taskID string, ) (events <-chan database.TaskEventEntity, cancel func(), err error)
SubscribeAgentTask exposes task completion notifications through the assistant boundary.
type SubmitRequest ¶
type SubmitRequest struct {
ParentTaskID string
OwnerSessionID string
ChildSessionID string
ConcurrencyKey string
AgentName string
Prompt string
Model string
Provider string
PolicyJSON string
OutputSchema string
OutputSchemaDigest string
Depth int
}
SubmitRequest describes a durable agent task.
type Subscription ¶
type Subscription struct {
Events <-chan database.TaskEventEntity
Cancel func()
}
Subscription delivers process-local, best-effort wakeups for events published after registration. It does not replay durable history or provide a complete event stream; callers that need event contents recover gaps through Events.
Source Files