dispatch

package

Go to main page

Versions in this module

v0

v0.4.0

Oct 8, 2026

v0.3.0

Oct 7, 2026

Changes in this version

+ const AddPath

+ const AddSliceTool

+ const DispatcherInterval

+ const DispatcherStateTool

+ const HoldPath

+ const HoldSliceTool

+ const PauseDispatcherTool

+ const PausePath

+ const ProvenanceSliceAdd

+ const ReleasePath

+ const ReleaseSliceTool

+ const ResumeDispatcherTool

+ const ResumePath

+ const SnapshotFile

+ const SnapshotPath

+ const TriggerDispatcher

+ var ErrInvalidControl = fmt.Errorf("%w: invalid control", csf.ErrInvalidRequest)

+ var Stages = []Stage

+ type AddSliceInput struct

+ Contends []string

+ DependsOn []string

+ Hold string

+ Hotspots []string

+ Paths []string

+ Recipe *pb.AgentAssignmentRecipe

+ SliceID string

+ Source string

+ TicketURL string

+ Title string

+ type Control struct

+ Action ControlAction

+ Reason string

+ SliceID string

+ type ControlAction string

+ const ControlHold

+ const ControlPause

+ const ControlRelease

+ const ControlResume

+ type DailyBudgetReader func(ctx context.Context) (ouroboros.DailyBudget, error)

type DispatchService

+ func (service *DispatchService) AddSlice(ctx context.Context, input AddSliceInput) (SliceView, error)

+ func (service *DispatchService) CurrentSnapshot(ctx context.Context, _ SnapshotInput) (Snapshot, error)

+ func (service *DispatchService) Dispatch(ctx context.Context) (Snapshot, error)

+ func (service *DispatchService) HeldBranch(ctx context.Context, branch string) (string, bool, error)

+ func (service *DispatchService) Hold(ctx context.Context, input SliceControlInput) (Snapshot, error)

+ func (service *DispatchService) MarkPullRequestMerged(ctx context.Context, url string) ([]string, error)

+ func (service *DispatchService) Pause(ctx context.Context, input DispatcherControlInput) (Snapshot, error)

+ func (service *DispatchService) ReadySlices(ctx context.Context, limit int) ([]ReadySlice, error)

+ func (service *DispatchService) Register(router gin.IRouter)

+ func (service *DispatchService) Release(ctx context.Context, input SliceControlInput) (Snapshot, error)

+ func (service *DispatchService) Resume(ctx context.Context, input DispatcherControlInput) (Snapshot, error)

+ func (service *DispatchService) Tools() []csf.Option

+ func (service *DispatchService) Trigger() cronservice.Option

+ type DispatcherControlInput struct

+ Reason string

+ type Flow struct

+ LeadTime *time.Duration

+ Oldest map[Stage]time.Duration

+ Slices map[Stage]int

type IDispatchQueries

+ InsertDispatchControl func(ctx context.Context, arg csfpg.InsertDispatchControlParams) (csfpg.CsfDispatchControl, error)

+ ListDispatchControls func(ctx context.Context) ([]csfpg.CsfDispatchControl, error)

type ISessionHost

+ Check func(ctx context.Context) (*harnessv1.LaunchCheck, error)

+ type Limit struct

+ Bounded bool

+ Derivation string

+ Launches int

+ Name LimitName

+ Provider string

+ Remaining float64

+ Unit string

+ type LimitName string

+ const LimitBudget

+ const LimitHarness

+ const LimitRate

+ type LimitSource func(ctx context.Context, running int) Limit

+ func DailyBudgetLimit(read DailyBudgetReader) LimitSource

+ func RateLimit(state stdfs.FS, clock harness.IClock, activeProvider string) LimitSource

+ type MergeCheck func(ctx context.Context, url string) (bool, error)

type Option

+ func WithLimit(source LimitSource) Option

+ func WithMergeCheck(check MergeCheck) Option

+ func WithSnapshotSink(sink SnapshotSink) Option

+ type ReadySlice struct

+ Recipe *pb.AgentAssignmentRecipe

+ SliceID string

+ TicketURL string

+ type SliceControlInput struct

+ Reason string

+ SliceID string

+ type SliceView struct

+ AssignmentID string

+ Attempts uint32

+ Contends []string

+ CreatedAt time.Time

+ CriticalPath uint32

+ DependsOn []string

+ Held string

+ PullRequestURL string

+ Rank uint32

+ Ready bool

+ SliceID string

+ Source string

+ State string

+ TicketURL string

+ Title string

+ UpdatedAt time.Time

+ Urgency string

+ Waiting string

+ type Snapshot struct

+ At time.Time

+ Capacity int

+ Finished []SliceView

+ Held []SliceView

+ Limits []Limit

+ LimitsAt time.Time

+ NextLaunch *SliceView

+ NextPassAt time.Time

+ PauseReason string

+ Paused bool

+ Queue []SliceView

+ Running []SliceView

+ func (snapshot Snapshot) Flow(window time.Duration) Flow

+ func (snapshot Snapshot) Staged() []StagedSlice

+ type SnapshotInput struct

+ type SnapshotSink func(snapshot Snapshot) error

+ type Stage string

+ const StageBlocked

+ const StageCanceled

+ const StageContended

+ const StageFailed

+ const StageMerged

+ const StageReady

+ const StageRunning

+ type StagedSlice struct

+ Slice SliceView

+ Stage Stage

v0.2.9

Oct 5, 2026

v0.2.8

Oct 3, 2026

v0.2.7

Oct 3, 2026

v0.2.6

Oct 3, 2026

v0.2.5

Oct 3, 2026

v0.2.4

Oct 3, 2026

v0.2.3

Oct 3, 2026

v0.2.2

Oct 3, 2026

v0.2.1

Oct 3, 2026

v0.2.0

Oct 3, 2026

Changes in this version

+ const ScopeOntology

+ var ErrCycle = fmt.Errorf("%w: depends_on would close a cycle", csf.ErrInvalidRequest)

+ var ErrInvalidIntent = fmt.Errorf("%w: invalid intent", csf.ErrInvalidRequest)

+ var ErrInvalidOption = errors.New("dispatch: invalid option")

+ var ErrInvalidSlice = fmt.Errorf("%w: invalid slice", csf.ErrInvalidRequest)

+ var ErrNoSessions = errors.New("dispatch: a session host is required")

+ var ErrNotStarted = errors.New("dispatch: the service is not running")

+ var ErrSliceExists = fmt.Errorf("%w: the slice is already enqueued", csf.ErrConflict)

+ var ErrSliceFinished = fmt.Errorf("%w: the slice has finished", csf.ErrConflict)

+ var ErrUnknownIntent = fmt.Errorf("%w: no such intent", csf.ErrNotFound)

+ var ErrUnknownSlice = fmt.Errorf("%w: no such slice", csf.ErrNotFound)

+ type DispatchService struct

+ func NewDispatchService(options ...Option) (*DispatchService, error)

+ func (service *DispatchService) DeclareIntent(ctx context.Context, request *dispatchv1.DeclareIntentRequest) (*dispatchv1.DeclareIntentResponse, error)

+ func (service *DispatchService) Enqueue(ctx context.Context, request *dispatchv1.EnqueueSliceRequest) (*dispatchv1.EnqueueSliceResponse, error)

+ func (service *DispatchService) Frontier(ctx context.Context, _ *dispatchv1.GetFrontierRequest) (*dispatchv1.GetFrontierResponse, error)

+ func (service *DispatchService) List(ctx context.Context, _ *dispatchv1.ListSlicesRequest) (*dispatchv1.ListSlicesResponse, error)

+ func (service *DispatchService) MarkMerged(ctx context.Context, request *dispatchv1.MarkSliceMergedRequest) (*dispatchv1.MarkSliceMergedResponse, error)

+ func (service *DispatchService) ObserveSession(state *harnessv1.AgentSessionState)

+ func (service *DispatchService) Reprioritize(ctx context.Context, request *dispatchv1.ReprioritizeRequest) (*dispatchv1.ReprioritizeResponse, error)

+ func (service *DispatchService) Route(ctx context.Context, request *dispatchv1.RouteMessageRequest) (*dispatchv1.RouteMessageResponse, error)

+ func (service *DispatchService) Start(scope *runtime.Scope) error

+ type IDispatchDatabase interface

+ Transact func(ctx context.Context, work func(queries IDispatchQueries) error) error

+ type IDispatchQueries interface

+ InsertIntent func(ctx context.Context, arg csfpg.InsertIntentParams) (csfpg.CsfIntent, error)

+ InsertSlice func(ctx context.Context, arg csfpg.InsertSliceParams) (csfpg.CsfSlice, error)

+ InsertSliceEdge func(ctx context.Context, arg csfpg.InsertSliceEdgeParams) error

+ ListIntents func(ctx context.Context) ([]csfpg.CsfIntent, error)

+ ListSliceEdges func(ctx context.Context) ([]csfpg.CsfSliceEdge, error)

+ ListSlices func(ctx context.Context) ([]csfpg.CsfSlice, error)

+ UpdateSliceState func(ctx context.Context, arg csfpg.UpdateSliceStateParams) (csfpg.CsfSlice, error)

+ type ISessionHost interface

+ Cancel func(ctx context.Context, request *harnessv1.CancelAgentSessionRequest) (*harnessv1.CancelAgentSessionResponse, error)

+ Send func(ctx context.Context, request *harnessv1.SendAgentSessionMessageRequest) (*harnessv1.SendAgentSessionMessageResponse, error)

+ Submit func(ctx context.Context, request *harnessv1.SubmitAgentSessionRequest) (*harnessv1.SubmitAgentSessionResponse, error)

+ type Option func(service *DispatchService) error

+ func WithClock(clock harness.IClock) Option

+ func WithDatabase(database IDispatchDatabase) Option

+ func WithHostMeasures(measures harness.IHostMeasures) Option

+ func WithLogger(logger *slog.Logger) Option

+ func WithOntologySource(path string) Option

+ func WithSessions(sessions ISessionHost) Option

+ type PostgresDispatchDatabase struct

+ func NewPostgresDispatchDatabase(database csfpg.IDB) (*PostgresDispatchDatabase, error)

+ func (database *PostgresDispatchDatabase) Transact(ctx context.Context, work func(queries IDispatchQueries) error) error

+ type Relation string

+ const RelationContends

+ const RelationDependsOn

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL