team

package
v0.1.3 Latest Latest
Warning

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

 Go to latest Published: Sep 18, 2026 License: Apache-2.0 Imports: 22 Imported by: 0

Documentation

Overview

Package team provides durable coordination for a flat team of independent Agent sessions. It owns members, dependent tasks, exclusive assignment and claims, task attempts, direct mailboxes, and Team lifecycle.

Team never starts Agents, schedulers, timers, or background goroutines. An application coordinator registers member resources. AttemptRuntime can synchronously compose the repeatable claim/start/mailbox/continuation/result lifecycle when an application supplies a Worker factory and result projector. It never owns models, Harness resources, routing, or scheduling. Goal and Loop may control an individual child continuation; Workflow and distributed scheduling remain application layers above Team.

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	ErrNotFound          = errors.New("team: not found")
	ErrExists            = errors.New("team: already exists")
	ErrConflict          = errors.New("team: revision conflict")
	ErrCommandConflict   = errors.New("team: command id conflict")
	ErrInvalid           = errors.New("team: invalid value")
	ErrTooLarge          = errors.New("team: value too large")
	ErrUnauthorized      = errors.New("team: unauthorized")
	ErrInvalidState      = errors.New("team: invalid state")
	ErrDependencyBlocked = errors.New("team: task dependency blocked")
	ErrMemberBusy        = errors.New("team: member has active work")
	ErrStaleAttempt      = errors.New("team: stale task attempt")
	ErrAttemptLimit      = errors.New("team: task attempt limit reached")
	ErrTerminal          = errors.New("team: terminal")
	ErrCorruptStore      = errors.New("team: corrupt store")
	ErrStoreFull         = errors.New("team: store limit reached")
)

Domain and Store errors.

Functions

func DefaultAttemptControllerRef 

func DefaultAttemptControllerRef() continuation.HandlerRef

DefaultAttemptControllerRef identifies CompleteAfterWork in durable records.

Types

type AcknowledgeMessagesRequest 

type AcknowledgeMessagesRequest struct {
	Command         CommandMetadata
	ThroughSequence uint64
}

AcknowledgeMessagesRequest advances the member actor's mailbox cursor.

type Actor 

type Actor struct {
	Kind ActorKind `json:"kind"`
	ID   string    `json:"id"`
}

Actor is durable audit identity. For ActorKindMember, ID is a MemberID.

type ActorKind 

type ActorKind string

ActorKind identifies the authority issuing a command. 

const (
	ActorKindCoordinator ActorKind = "coordinator"
	ActorKindMember      ActorKind = "member"
)

Actor kinds.

type Artifact 

type Artifact struct {
	Kind      string `json:"kind"`
	Reference string `json:"reference"`
	Digest    string `json:"digest,omitempty"`
	MediaType string `json:"media_type,omitempty"`
}

Artifact is an opaque coordinator-managed result reference. It must not contain credentials or raw secret material.

type AssignTaskRequest 

type AssignTaskRequest struct {
	Command  CommandMetadata
	TaskID   TaskID
	MemberID MemberID
}

AssignTaskRequest exclusively assigns an unclaimed task.

type Attempt 

type Attempt struct {
	ID             AttemptID       `json:"id"`
	Number         int             `json:"number"`
	Status         AttemptStatus   `json:"status"`
	MemberID       MemberID        `json:"member_id"`
	ContinuationID continuation.ID `json:"continuation_id"`
	Result         ai.JSON         `json:"result,omitempty"`
	Artifacts      []Artifact      `json:"artifacts,omitempty"`
	Reason         string          `json:"reason,omitempty"`
	StartedAt      time.Time       `json:"started_at"`
	FinishedAt     time.Time       `json:"finished_at,omitzero"`
}

Attempt records one coordinator-started task execution.

type AttemptCommand 

type AttemptCommand struct {
	Action         string
	TeamID         ID
	TaskID         TaskID
	AttemptID      AttemptID
	ContinuationID continuation.ID
	MessageID      MessageID
}

AttemptCommand identifies one deterministic runtime mutation.

type AttemptCommandIDSource 

type AttemptCommandIDSource func(AttemptCommand) (CommandID, error)

AttemptCommandIDSource derives stable idempotency keys for runtime commands.

type AttemptCompletion 

type AttemptCompletion struct {
	Outcome            AttemptOutcome
	Result             ai.JSON
	Artifacts          []Artifact
	Reason             string
	AcknowledgeMailbox bool
	Messages           []AttemptMessage
}

AttemptCompletion is the application-owned Team projection of a terminal continuation. The runtime applies it in acknowledgement, message, finish order with the appropriate member and Coordinator authorities.

type AttemptID 

type AttemptID string

AttemptID identifies one external execution attempt for a task.

type AttemptInput 

type AttemptInput struct {
	Dispatch Dispatch
	Mailbox  MessagePage
}

AttemptInput is the immutable Team context supplied to an attempt adapter. It contains only durable Team data, never a model or Harness resource.

type AttemptInspection 

type AttemptInspection struct {
	Dispatch  Dispatch
	State     AttemptInspectionState
	Execution *continuation.Execution
}

AttemptInspection is a read-only cross-store recovery projection.

type AttemptInspectionState 

type AttemptInspectionState string

AttemptInspectionState classifies one referenced child execution. 

const (
	AttemptInspectionMissing     AttemptInspectionState = "missing"
	AttemptInspectionNonterminal AttemptInspectionState = "nonterminal"
	AttemptInspectionTerminal    AttemptInspectionState = "terminal"
)

Attempt inspection states.

type AttemptMessage 

type AttemptMessage struct {
	ID          MessageID
	RecipientID MemberID
	ReplyToID   MessageID
	Body        ai.JSON
}

AttemptMessage is a declarative message sent by the assigned member after a terminal continuation is projected. Its task is always the current task.

type AttemptOutcome 

type AttemptOutcome string

AttemptOutcome is an accepted FinishTaskAttempt outcome. 

const (
	AttemptOutcomeCompleted AttemptOutcome = "completed"
	AttemptOutcomeFailed    AttemptOutcome = "failed"
)

FinishTaskAttempt outcomes.

type AttemptResultProjector 

type AttemptResultProjector interface {
	ProjectAttemptResult(context.Context, AttemptInput, continuation.Execution) (AttemptCompletion, error)
}

AttemptResultProjector maps durable continuation evidence to generic Team completion data. It must not mutate Team state itself.

type AttemptResultProjectorFunc 

type AttemptResultProjectorFunc func(context.Context, AttemptInput, continuation.Execution) (AttemptCompletion, error)

AttemptResultProjectorFunc adapts a function into an AttemptResultProjector.

func (AttemptResultProjectorFunc) ProjectAttemptResult 

func (function AttemptResultProjectorFunc) ProjectAttemptResult(
	ctx context.Context,
	input AttemptInput,
	execution continuation.Execution,
) (AttemptCompletion, error)

ProjectAttemptResult implements AttemptResultProjector.

type AttemptRunRequest 

type AttemptRunRequest struct {
	TeamID         ID
	TaskID         TaskID
	AttemptID      AttemptID
	ContinuationID continuation.ID
}

AttemptRunRequest identifies one durable Team task attempt.

type AttemptRunResult 

type AttemptRunResult struct {
	Team      Team
	Dispatch  Dispatch
	Mailbox   MessagePage
	Execution continuation.Execution
	Advances  int
	Yield     continuation.YieldReason
	Finished  bool
}

AttemptRunResult exposes the latest child state and whether Team completion was committed. A nonterminal yield is a normal caller-controlled boundary.

type AttemptRuntime 

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

AttemptRuntime composes finite Team commands with a bounded continuation. It does not start goroutines, schedule work, or own model/Harness resources.

func NewAttemptRuntime 

func NewAttemptRuntime(
	teams *Engine,
	continuations *continuation.Engine,
	workers AttemptWorkerFactory,
	projector AttemptResultProjector,
	options ...AttemptRuntimeOption,
) (*AttemptRuntime, error)

NewAttemptRuntime constructs a synchronous durable Team attempt runtime.

func (*AttemptRuntime) Run 

func (runtime *AttemptRuntime) Run(ctx context.Context, request AttemptRunRequest) (AttemptRunResult, error)

Run claims/starts the attempt if necessary, creates or resumes its child, and commits a projected terminal result. It performs no polling or work retries; terminal projection commands rebase bounded optimistic conflicts.

type AttemptRuntimeOption 

type AttemptRuntimeOption func(*attemptRuntimeConfig) error

AttemptRuntimeOption configures an AttemptRuntime.

func WithAttemptCommandIDSource 

func WithAttemptCommandIDSource(source AttemptCommandIDSource) AttemptRuntimeOption

WithAttemptCommandIDSource replaces deterministic command derivation.

func WithAttemptCoordinator 

func WithAttemptCoordinator(coordinator Actor) AttemptRuntimeOption

WithAttemptCoordinator configures the authority that starts and finishes Team attempts.

func WithAttemptDriveOptions 

func WithAttemptDriveOptions(options continuation.DriveOptions) AttemptRuntimeOption

WithAttemptDriveOptions sets the bounded synchronous Drive quantum.

func WithAttemptHandlers 

func WithAttemptHandlers(
	workerRef continuation.HandlerRef,
	controllerRef continuation.HandlerRef,
	controller continuation.Controller,
) AttemptRuntimeOption

WithAttemptHandlers configures durable continuation references and the Controller paired with Workers created by the factory.

func WithAttemptLimits 

func WithAttemptLimits(limits continuation.Limits) AttemptRuntimeOption

WithAttemptLimits sets limits on newly created child continuations.

func WithAttemptTargetKind 

func WithAttemptTargetKind(kind string) AttemptRuntimeOption

WithAttemptTargetKind changes the application-owned continuation target kind.

type AttemptStatus 

type AttemptStatus string

AttemptStatus is one task attempt's durable state. 

const (
	AttemptStatusRunning   AttemptStatus = "running"
	AttemptStatusCompleted AttemptStatus = "completed"
	AttemptStatusFailed    AttemptStatus = "failed"
	AttemptStatusCancelled AttemptStatus = "cancelled"
)

Attempt statuses.

type AttemptWorkerFactory 

type AttemptWorkerFactory interface {
	PrepareAttempt(context.Context, AttemptInput) (PreparedAttempt, error)
}

AttemptWorkerFactory prepares one bounded Worker for a committed Team dispatch. It is called for both creation and recovery-driven execution; its Input is used only if the continuation has not already been created.

type AttemptWorkerFactoryFunc 

type AttemptWorkerFactoryFunc func(context.Context, AttemptInput) (PreparedAttempt, error)

AttemptWorkerFactoryFunc adapts a function into an AttemptWorkerFactory.

func (AttemptWorkerFactoryFunc) PrepareAttempt 

func (function AttemptWorkerFactoryFunc) PrepareAttempt(
	ctx context.Context,
	input AttemptInput,
) (PreparedAttempt, error)

PrepareAttempt implements AttemptWorkerFactory.

type CancelTaskRequest 

type CancelTaskRequest struct {
	Command CommandMetadata
	TaskID  TaskID
	Reason  string
}

CancelTaskRequest explicitly cancels one non-completed task.

type CancelTeamRequest 

type CancelTeamRequest struct {
	Command CommandMetadata
	Reason  string
}

CancelTeamRequest explicitly cancels a Team and its active work.

type Cause 

type Cause string

Cause identifies one committed domain transition. 

const (
	CauseCreate               Cause = "create"
	CauseMemberRegistered     Cause = "member_registered"
	CauseMemberDisabled       Cause = "member_disabled"
	CauseMemberEnabled        Cause = "member_enabled"
	CauseTaskCreated          Cause = "task_created"
	CauseTaskAssigned         Cause = "task_assigned"
	CauseTaskUnassigned       Cause = "task_unassigned"
	CauseTaskClaimed          Cause = "task_claimed"
	CauseTaskReleased         Cause = "task_released"
	CauseTaskAttemptStarted   Cause = "task_attempt_started"
	CauseTaskAttemptCompleted Cause = "task_attempt_completed"
	CauseTaskAttemptFailed    Cause = "task_attempt_failed"
	CauseTaskRetried          Cause = "task_retried"
	CauseTaskCancelled        Cause = "task_cancelled"
	CauseMessageSent          Cause = "message_sent"
	CauseMessagesAcknowledged Cause = "messages_acknowledged"
	CauseTeamCompleted        Cause = "team_completed"
	CauseTeamFailed           Cause = "team_failed"
	CauseTeamCancelled        Cause = "team_cancelled"
)

Transition causes.

type Change 

type Change struct {
	Transition Transition
	Message    *Message
}

Change is one durable Team transition and its optional message delta.

type ChangeOptions 

type ChangeOptions struct {
	AfterRevision Revision
	Limit         int
}

ChangeOptions selects transitions after an exclusive Team revision.

type ChangePage 

type ChangePage struct {
	Changes   []Change
	NextAfter Revision
}

ChangePage is one bounded page of Team changes.

type ClaimTaskRequest 

type ClaimTaskRequest struct {
	Command CommandMetadata
	TaskID  TaskID
}

ClaimTaskRequest claims one ready task for its member actor.

type Clock 

type Clock interface {
	Now() time.Time
}

Clock supplies deterministic Team lifecycle timestamps.

type ClockFunc 

type ClockFunc func() time.Time

ClockFunc adapts a function to Clock.

func (ClockFunc) Now 

func (function ClockFunc) Now() time.Time

Now implements Clock.

type CommandID 

type CommandID string

CommandID is a stable idempotency key for one Team mutation.

type CommandMetadata 

type CommandMetadata struct {
	ID               CommandID
	ExpectedRevision Revision
	Actor            Actor
}

CommandMetadata identifies and orders one mutation.

type CompleteAfterWork 

type CompleteAfterWork struct{}

CompleteAfterWork is a Controller for one-shot attempt Workers. It commits the Work value as the continuation output and terminates the child.

func (CompleteAfterWork) Decide 

func (CompleteAfterWork) Decide(
	_ context.Context,
	request continuation.DecisionRequest,
) (continuation.Decision, error)

Decide implements continuation.Controller.

type CompleteTeamRequest 

type CompleteTeamRequest struct {
	Command   CommandMetadata
	Output    ai.JSON
	Artifacts []Artifact
	Reason    string
}

CompleteTeamRequest explicitly completes one fully settled Team.

type ConflictError 

type ConflictError struct {
	Expected Revision
	Actual   Revision
}

ConflictError reports an optimistic revision mismatch.

func (*ConflictError) Error 

func (e *ConflictError) Error() string

Error implements error.

func (*ConflictError) Unwrap 

func (e *ConflictError) Unwrap() error

Unwrap exposes ErrConflict.

type CorruptStoreError 

type CorruptStoreError struct {
	Path   string
	Line   int
	Reason string
	Err    error
}

CorruptStoreError identifies malformed durable Team data.

func (*CorruptStoreError) Error 

func (e *CorruptStoreError) Error() string

Error implements error.

func (*CorruptStoreError) Unwrap 

func (e *CorruptStoreError) Unwrap() error

Unwrap exposes ErrCorruptStore.

type CreateRequest 

type CreateRequest struct {
	Command   CommandMetadata
	ID        ID
	Objective string
	Lead      MemberSpec
	Limits    Limits
}

CreateRequest creates one Team with exactly one lead member.

type CreateTaskRequest 

type CreateTaskRequest struct {
	Command      CommandMetadata
	TaskID       TaskID
	Title        string
	Description  string
	Payload      ai.JSON
	Dependencies []TaskID
	AttemptLimit int
}

CreateTaskRequest creates one immutable task definition.

type DisableMemberRequest 

type DisableMemberRequest struct {
	Command  CommandMetadata
	MemberID MemberID
	Reason   string
}

DisableMemberRequest disables an idle non-lead member.

type Dispatch 

type Dispatch struct {
	TeamID               ID
	TeamRevision         Revision
	Objective            string
	MemberID             MemberID
	MemberName           string
	MemberRole           string
	CapabilityProfileRef string
	TaskID               TaskID
	AttemptID            AttemptID
	ContinuationID       continuation.ID
	Title                string
	Description          string
	Payload              ai.JSON
	DependencyIDs        []TaskID
}

Dispatch is immutable coordinator input for one committed task attempt.

type EnableMemberRequest 

type EnableMemberRequest struct {
	Command  CommandMetadata
	MemberID MemberID
}

EnableMemberRequest re-enables one disabled member.

type Engine 

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

Engine applies finite commands to durable Team aggregates.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/rsbin1178/pips/agent/team"
)

func main() {
	ctx := context.Background()
	store, _ := team.NewMemoryStore()
	runtime, _ := team.New(store)
	coordinator := team.Actor{Kind: team.ActorKindCoordinator, ID: "cli"}

	group, _ := runtime.Create(ctx, team.CreateRequest{
		Command: team.CommandMetadata{ID: "create", Actor: coordinator},
		ID:      "release-team", Objective: "prepare the release",
		Lead: team.MemberSpec{
			ID: "lead", Name: "Lead", Role: "coordinate",
		},
	})
	group, _ = runtime.RegisterMember(ctx, group.ID, team.RegisterMemberRequest{
		Command: team.CommandMetadata{
			ID: "register", ExpectedRevision: group.Revision, Actor: coordinator,
		},
		Member: team.MemberSpec{
			ID: "reviewer", Name: "Reviewer", Role: "review",
		},
	})
	group, _ = runtime.CreateTask(ctx, group.ID, team.CreateTaskRequest{
		Command: team.CommandMetadata{
			ID: "task", ExpectedRevision: group.Revision,
			Actor: team.Actor{Kind: team.ActorKindMember, ID: "lead"},
		},
		TaskID: "review", Title: "Review release", AttemptLimit: 1,
	})
	group, _ = runtime.ClaimTask(ctx, group.ID, team.ClaimTaskRequest{
		Command: team.CommandMetadata{
			ID: "claim", ExpectedRevision: group.Revision,
			Actor: team.Actor{Kind: team.ActorKindMember, ID: "reviewer"},
		},
		TaskID: "review",
	})
	started, _ := runtime.StartTaskAttempt(ctx, group.ID, team.StartTaskAttemptRequest{
		Command: team.CommandMetadata{
			ID: "start", ExpectedRevision: group.Revision, Actor: coordinator,
		},
		TaskID: "review", AttemptID: "review-1", ContinuationID: "review-execution-1",
	})

	fmt.Println(started.Dispatch.MemberID)
	fmt.Println(started.Dispatch.TaskID, started.Dispatch.ContinuationID)
}

Output:
reviewer
review review-execution-1

func New 

func New(store Store, options ...Option) (*Engine, error)

New constructs an Engine over a Team Store. It starts no background work.

func (*Engine) AcknowledgeMessages 

func (engine *Engine) AcknowledgeMessages(
	ctx context.Context,
	id ID,
	request AcknowledgeMessagesRequest,
) (Team, error)

AcknowledgeMessages advances the member actor's mailbox cursor.

func (*Engine) ActiveDispatches 

func (engine *Engine) ActiveDispatches(ctx context.Context, id ID) ([]Dispatch, error)

ActiveDispatches returns currently running task attempts.

func (*Engine) AssignTask 

func (engine *Engine) AssignTask(
	ctx context.Context,
	id ID,
	request AssignTaskRequest,
) (Team, error)

AssignTask exclusively assigns an unclaimed task.

func (*Engine) CancelTask 

func (engine *Engine) CancelTask(
	ctx context.Context,
	id ID,
	request CancelTaskRequest,
) (Team, error)

CancelTask explicitly cancels one non-completed task.

func (*Engine) CancelTeam 

func (engine *Engine) CancelTeam(
	ctx context.Context,
	id ID,
	request CancelTeamRequest,
) (Team, error)

CancelTeam explicitly cancels a Team and its active work.

func (*Engine) CancellationDispatches 

func (engine *Engine) CancellationDispatches(ctx context.Context, id ID) ([]Dispatch, error)

CancellationDispatches returns attempts cancelled while externally running.

func (*Engine) Changes 

func (engine *Engine) Changes(
	ctx context.Context,
	id ID,
	options ChangeOptions,
) (ChangePage, error)

Changes returns a bounded page after an exclusive Team revision.

func (*Engine) ClaimTask 

func (engine *Engine) ClaimTask(
	ctx context.Context,
	id ID,
	request ClaimTaskRequest,
) (Team, error)

ClaimTask atomically claims one ready task for its member actor.

func (*Engine) CompleteTeam 

func (engine *Engine) CompleteTeam(
	ctx context.Context,
	id ID,
	request CompleteTeamRequest,
) (Team, error)

CompleteTeam completes a Team whose tasks are all completed or cancelled.

func (*Engine) Create 

func (engine *Engine) Create(ctx context.Context, request CreateRequest) (Team, error)

Create persists a new active Team with exactly one lead member.

func (*Engine) CreateTask 

func (engine *Engine) CreateTask(
	ctx context.Context,
	id ID,
	request CreateTaskRequest,
) (Team, error)

CreateTask creates one immutable task definition.

func (*Engine) DisableMember 

func (engine *Engine) DisableMember(
	ctx context.Context,
	id ID,
	request DisableMemberRequest,
) (Team, error)

DisableMember disables an idle non-lead member.

func (*Engine) EnableMember 

func (engine *Engine) EnableMember(
	ctx context.Context,
	id ID,
	request EnableMemberRequest,
) (Team, error)

EnableMember re-enables one disabled member.

func (*Engine) FailTeam 

func (engine *Engine) FailTeam(
	ctx context.Context,
	id ID,
	request FailTeamRequest,
) (Team, error)

FailTeam explicitly fails a Team and cancels its active work.

func (*Engine) FinishTaskAttempt 

func (engine *Engine) FinishTaskAttempt(
	ctx context.Context,
	id ID,
	request FinishTaskAttemptRequest,
) (Team, error)

FinishTaskAttempt commits the current attempt outcome.

func (*Engine) Get 

func (engine *Engine) Get(ctx context.Context, id ID) (Team, error)

Get returns the latest Team snapshot.

func (*Engine) History 

func (engine *Engine) History(ctx context.Context, id ID) ([]Record, error)

History returns cloned durable records in revision order.

func (*Engine) InspectActiveAttempts 

func (engine *Engine) InspectActiveAttempts(
	ctx context.Context,
	id ID,
	reader ExecutionReader,
) ([]AttemptInspection, error)

InspectActiveAttempts performs one finite child lookup per active dispatch.

func (*Engine) List 

func (engine *Engine) List(ctx context.Context, options ListOptions) (ListPage, error)

List returns a bounded page of current Team snapshots.

func (*Engine) Mailbox 

func (engine *Engine) Mailbox(
	ctx context.Context,
	id ID,
	memberID MemberID,
	options MailboxOptions,
) (MessagePage, error)

Mailbox returns one ordered page of messages for a recipient.

func (*Engine) RegisterMember 

func (engine *Engine) RegisterMember(
	ctx context.Context,
	id ID,
	request RegisterMemberRequest,
) (Team, error)

RegisterMember registers one coordinator-managed Agent resource.

func (*Engine) ReleaseTask 

func (engine *Engine) ReleaseTask(
	ctx context.Context,
	id ID,
	request ReleaseTaskRequest,
) (Team, error)

ReleaseTask releases one unstarted claim.

func (*Engine) RetryTask 

func (engine *Engine) RetryTask(
	ctx context.Context,
	id ID,
	request RetryTaskRequest,
) (Team, error)

RetryTask returns one failed task to pending or ready.

func (*Engine) SendMessage 

func (engine *Engine) SendMessage(
	ctx context.Context,
	id ID,
	request SendMessageRequest,
) (MessageSend, error)

SendMessage commits one immutable direct member message.

func (*Engine) StartTaskAttempt 

func (engine *Engine) StartTaskAttempt(
	ctx context.Context,
	id ID,
	request StartTaskAttemptRequest,
) (TaskAttemptStart, error)

StartTaskAttempt durably binds one claimed task to a continuation execution.

func (*Engine) UnassignTask 

func (engine *Engine) UnassignTask(
	ctx context.Context,
	id ID,
	request UnassignTaskRequest,
) (Team, error)

UnassignTask removes an unclaimed task assignment.

type EventID 

type EventID string

EventID identifies one committed Team transition.

type EventIDSource 

type EventIDSource func(time.Time) (EventID, error)

EventIDSource creates safe unique transition IDs.

type ExecutionReader 

type ExecutionReader interface {
	Get(context.Context, continuation.ID) (continuation.Execution, error)
}

ExecutionReader is the finite continuation lookup used for recovery inspection. continuation.Engine implements it.

type FailTeamRequest 

type FailTeamRequest struct {
	Command CommandMetadata
	Reason  string
}

FailTeamRequest explicitly fails a Team and cancels its active work.

type FinishTaskAttemptRequest 

type FinishTaskAttemptRequest struct {
	Command        CommandMetadata
	TaskID         TaskID
	AttemptID      AttemptID
	ContinuationID continuation.ID
	Outcome        AttemptOutcome
	Result         ai.JSON
	Artifacts      []Artifact
	Reason         string
}

FinishTaskAttemptRequest commits the current attempt outcome.

type ID 

type ID string

ID identifies one durable Team aggregate.

type JSONLStore 

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

JSONLStore is a bounded single-process directory Store.

func NewJSONLStore 

func NewJSONLStore(dir string, options ...StoreOption) (*JSONLStore, error)

NewJSONLStore opens a directory-backed Team store.

func (*JSONLStore) Changes 

func (store *JSONLStore) Changes(
	ctx context.Context,
	id ID,
	options ChangeOptions,
) (ChangePage, error)

Changes implements Store.

func (*JSONLStore) CompareAndSwap 

func (store *JSONLStore) CompareAndSwap(
	ctx context.Context,
	id ID,
	expected Revision,
	next Record,
) error

CompareAndSwap implements Store.

func (*JSONLStore) Create 

func (store *JSONLStore) Create(ctx context.Context, record Record) error

Create implements Store.

func (*JSONLStore) History 

func (store *JSONLStore) History(ctx context.Context, id ID) ([]Record, error)

History implements Store.

func (*JSONLStore) List 

func (store *JSONLStore) List(ctx context.Context, options ListOptions) (ListPage, error)

List implements Store.

func (*JSONLStore) Load 

func (store *JSONLStore) Load(ctx context.Context, id ID) (Record, error)

Load implements Store.

func (*JSONLStore) LoadCommand 

func (store *JSONLStore) LoadCommand(
	ctx context.Context,
	id ID,
	commandID CommandID,
) (Record, error)

LoadCommand implements Store.

func (*JSONLStore) LoadMessage 

func (store *JSONLStore) LoadMessage(
	ctx context.Context,
	id ID,
	messageID MessageID,
) (Message, error)

LoadMessage implements Store.

func (*JSONLStore) Mailbox 

func (store *JSONLStore) Mailbox(
	ctx context.Context,
	id ID,
	memberID MemberID,
	options MailboxOptions,
) (MessagePage, error)

Mailbox implements Store.

type Limits 

type Limits struct {
	MaxMembers             int `json:"max_members"`
	MaxTasks               int `json:"max_tasks"`
	MaxDependenciesPerTask int `json:"max_dependencies_per_task"`
	MaxMessages            int `json:"max_messages"`
	MaxActiveTasks         int `json:"max_active_tasks"`
	MaxAttemptsPerTask     int `json:"max_attempts_per_task"`
	MaxArtifactsPerResult  int `json:"max_artifacts_per_result"`
	MaxJSONBytes           int `json:"max_json_bytes"`
}

Limits are stored hard bounds for one Team. Zero fields use defaults.

type ListOptions 

type ListOptions struct {
	Limit  int
	Cursor string
}

ListOptions bounds one lexicographically ordered Store page.

type ListPage 

type ListPage struct {
	Teams      []Team
	NextCursor string
}

ListPage is one bounded page of current Team snapshots.

type MailboxOptions 

type MailboxOptions struct {
	AfterSequence uint64
	Limit         int
}

MailboxOptions selects one recipient page after an exclusive sequence.

type Member 

type Member struct {
	ID                   MemberID     `json:"id"`
	Name                 string       `json:"name"`
	Role                 string       `json:"role"`
	CapabilityProfileRef string       `json:"capability_profile_ref,omitempty"`
	Status               MemberStatus `json:"status"`
	MailboxDelivered     uint64       `json:"mailbox_delivered"`
	MailboxAcknowledged  uint64       `json:"mailbox_acknowledged"`
	RegisteredAt         time.Time    `json:"registered_at"`
	DisabledAt           time.Time    `json:"disabled_at,omitzero"`
}

Member is one durable logical Agent identity.

type MemberID 

type MemberID string

MemberID identifies one logical Agent member within a Team.

type MemberSpec 

type MemberSpec struct {
	ID                   MemberID `json:"id"`
	Name                 string   `json:"name"`
	Role                 string   `json:"role"`
	CapabilityProfileRef string   `json:"capability_profile_ref,omitempty"`
}

MemberSpec registers one coordinator-managed Agent resource with a Team.

type MemberStatus 

type MemberStatus string

MemberStatus is a logical Team member's availability. 

const (
	MemberStatusActive   MemberStatus = "active"
	MemberStatusDisabled MemberStatus = "disabled"
)

Member statuses.

type MemoryStore 

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

MemoryStore is a bounded in-memory Store for tests and ephemeral coordinators.

func NewMemoryStore 

func NewMemoryStore(options ...StoreOption) (*MemoryStore, error)

NewMemoryStore creates an empty in-memory Team store.

func (*MemoryStore) Changes 

func (store *MemoryStore) Changes(
	ctx context.Context,
	id ID,
	options ChangeOptions,
) (ChangePage, error)

Changes implements Store.

func (*MemoryStore) CompareAndSwap 

func (store *MemoryStore) CompareAndSwap(
	ctx context.Context,
	id ID,
	expected Revision,
	next Record,
) error

CompareAndSwap implements Store.

func (*MemoryStore) Create 

func (store *MemoryStore) Create(ctx context.Context, record Record) error

Create implements Store.

func (*MemoryStore) History 

func (store *MemoryStore) History(ctx context.Context, id ID) ([]Record, error)

History implements Store.

func (*MemoryStore) List 

func (store *MemoryStore) List(ctx context.Context, options ListOptions) (ListPage, error)

List implements Store.

func (*MemoryStore) Load 

func (store *MemoryStore) Load(ctx context.Context, id ID) (Record, error)

Load implements Store.

func (*MemoryStore) LoadCommand 

func (store *MemoryStore) LoadCommand(
	ctx context.Context,
	id ID,
	commandID CommandID,
) (Record, error)

LoadCommand implements Store.

func (*MemoryStore) LoadMessage 

func (store *MemoryStore) LoadMessage(
	ctx context.Context,
	id ID,
	messageID MessageID,
) (Message, error)

LoadMessage implements Store.

func (*MemoryStore) Mailbox 

func (store *MemoryStore) Mailbox(
	ctx context.Context,
	id ID,
	memberID MemberID,
	options MailboxOptions,
) (MessagePage, error)

Mailbox implements Store.

type Message 

type Message struct {
	ID          MessageID `json:"id"`
	Sequence    uint64    `json:"sequence"`
	SenderID    MemberID  `json:"sender_id"`
	RecipientID MemberID  `json:"recipient_id"`
	TaskID      TaskID    `json:"task_id,omitempty"`
	ReplyToID   MessageID `json:"reply_to_id,omitempty"`
	Body        ai.JSON   `json:"body"`
	SentAt      time.Time `json:"sent_at"`
}

Message is one immutable, direct member message.

type MessageID 

type MessageID string

MessageID identifies one immutable direct message.

type MessagePage 

type MessagePage struct {
	Messages  []Message
	NextAfter uint64
}

MessagePage is one ordered recipient mailbox page.

type MessageSend 

type MessageSend struct {
	Team    Team
	Message Message
}

MessageSend is the exact post-command Team and committed Message.

type Option 

type Option func(*engineConfig) error

Option configures an Engine.

func WithClock 

func WithClock(clock Clock) Option

WithClock replaces the Engine clock.

func WithEventIDSource 

func WithEventIDSource(source EventIDSource) Option

WithEventIDSource replaces transition ID generation.

type PreparedAttempt 

type PreparedAttempt struct {
	Worker continuation.Worker
	Input  ai.JSON
}

PreparedAttempt supplies the application Worker and its durable input for a newly created continuation.

type Record 

type Record struct {
	SchemaVersion int        `json:"schema_version"`
	Team          Team       `json:"team"`
	Transition    Transition `json:"transition"`
	Message       *Message   `json:"message,omitempty"`
}

Record is one full post-command coordination snapshot plus an optional immutable message delta.

type RegisterMemberRequest 

type RegisterMemberRequest struct {
	Command CommandMetadata
	Member  MemberSpec
}

RegisterMemberRequest registers one coordinator-managed member resource.

type ReleaseTaskRequest 

type ReleaseTaskRequest struct {
	Command CommandMetadata
	TaskID  TaskID
	Reason  string
}

ReleaseTaskRequest releases one unstarted claim.

type RetryTaskRequest 

type RetryTaskRequest struct {
	Command CommandMetadata
	TaskID  TaskID
	Reason  string
}

RetryTaskRequest returns one failed task to pending or ready.

type Revision 

type Revision uint64

Revision is an optimistic concurrency version.

type SendMessageRequest 

type SendMessageRequest struct {
	Command     CommandMetadata
	MessageID   MessageID
	RecipientID MemberID
	TaskID      TaskID
	ReplyToID   MessageID
	Body        ai.JSON
}

SendMessageRequest commits one immutable direct message.

type StartTaskAttemptRequest 

type StartTaskAttemptRequest struct {
	Command        CommandMetadata
	TaskID         TaskID
	AttemptID      AttemptID
	ContinuationID continuation.ID
}

StartTaskAttemptRequest durably binds one claimed task to a continuation.

type StateError 

type StateError struct {
	Operation string
	Team      Status
	Task      TaskStatus
	Err       error
}

StateError describes an operation rejected by current Team or task state.

func (*StateError) Error 

func (e *StateError) Error() string

Error implements error.

func (*StateError) Unwrap 

func (e *StateError) Unwrap() error

Unwrap exposes the state sentinel.

type Status 

type Status string

Status is the Team lifecycle state. 

const (
	StatusActive    Status = "active"
	StatusCompleted Status = "completed"
	StatusFailed    Status = "failed"
	StatusCancelled Status = "cancelled"
)

Team statuses.

type Store 

type Store interface {
	Create(context.Context, Record) error
	Load(context.Context, ID) (Record, error)
	CompareAndSwap(context.Context, ID, Revision, Record) error
	LoadCommand(context.Context, ID, CommandID) (Record, error)
	List(context.Context, ListOptions) (ListPage, error)
	History(context.Context, ID) ([]Record, error)
	Changes(context.Context, ID, ChangeOptions) (ChangePage, error)
	LoadMessage(context.Context, ID, MessageID) (Message, error)
	Mailbox(context.Context, ID, MemberID, MailboxOptions) (MessagePage, error)
}

Store is the durable optimistic Team aggregate boundary.

type StoreLimits 

type StoreLimits struct {
	MaxRecordBytes int
	MaxFileBytes   int64
	MaxTransitions int
	MaxListPage    int
}

StoreLimits bound local Team store resource use.

type StoreOption 

type StoreOption func(*storeConfig) error

StoreOption configures a local Store.

func WithStoreLimits 

func WithStoreLimits(limits StoreLimits) StoreOption

WithStoreLimits replaces positive local-store limits.

type Task 

type Task struct {
	ID               TaskID     `json:"id"`
	Title            string     `json:"title"`
	Description      string     `json:"description,omitempty"`
	Payload          ai.JSON    `json:"payload,omitempty"`
	DependencyIDs    []TaskID   `json:"dependency_ids,omitempty"`
	AssignedMemberID MemberID   `json:"assigned_member_id,omitempty"`
	ClaimedMemberID  MemberID   `json:"claimed_member_id,omitempty"`
	Status           TaskStatus `json:"status"`
	AttemptLimit     int        `json:"attempt_limit"`
	Attempts         []Attempt  `json:"attempts,omitempty"`
	Reason           string     `json:"reason,omitempty"`
	CreatedAt        time.Time  `json:"created_at"`
	UpdatedAt        time.Time  `json:"updated_at"`
}

Task is one immutable work definition plus mutable coordination state.

type TaskAttemptStart 

type TaskAttemptStart struct {
	Team     Team
	Dispatch Dispatch
}

TaskAttemptStart is the exact post-command Team and its Dispatch.

type TaskID 

type TaskID string

TaskID identifies one Team task.

type TaskStatus 

type TaskStatus string

TaskStatus is a task's durable coordination state. 

const (
	TaskStatusPending   TaskStatus = "pending"
	TaskStatusReady     TaskStatus = "ready"
	TaskStatusClaimed   TaskStatus = "claimed"
	TaskStatusRunning   TaskStatus = "running"
	TaskStatusCompleted TaskStatus = "completed"
	TaskStatusFailed    TaskStatus = "failed"
	TaskStatusCancelled TaskStatus = "cancelled"
)

Task statuses.

type Team 

type Team struct {
	SchemaVersion       int        `json:"schema_version"`
	ID                  ID         `json:"id"`
	Revision            Revision   `json:"revision"`
	Status              Status     `json:"status"`
	Objective           string     `json:"objective"`
	LeadMemberID        MemberID   `json:"lead_member_id"`
	Members             []Member   `json:"members"`
	Tasks               []Task     `json:"tasks"`
	NextMessageSequence uint64     `json:"next_message_sequence"`
	Limits              Limits     `json:"limits"`
	Reason              string     `json:"reason,omitempty"`
	Output              ai.JSON    `json:"output,omitempty"`
	Artifacts           []Artifact `json:"artifacts,omitempty"`
	CreatedAt           time.Time  `json:"created_at"`
	UpdatedAt           time.Time  `json:"updated_at"`
}

Team is the latest full aggregate snapshot.

type ToolCommandContext 

type ToolCommandContext struct {
	TeamID     ID       `json:"team_id"`
	MemberID   MemberID `json:"member_id"`
	ToolName   string   `json:"tool_name"`
	ToolCallID string   `json:"tool_call_id"`
}

ToolCommandContext identifies one model-originated Team command.

type ToolCommandIDSource 

type ToolCommandIDSource func(ToolCommandContext) (CommandID, error)

ToolCommandIDSource derives a durable command key for a bound tool call.

type Toolset 

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

Toolset is a fixed collection of tools bound to one Team member identity.

func NewLeadToolset 

func NewLeadToolset(
	engine *Engine,
	teamID ID,
	leadMemberID MemberID,
	options ...ToolsetOption,
) (*Toolset, error)

NewLeadToolset builds member tools plus fixed-lead governance tools.

func NewMemberToolset 

func NewMemberToolset(
	engine *Engine,
	teamID ID,
	memberID MemberID,
	options ...ToolsetOption,
) (*Toolset, error)

NewMemberToolset builds ordinary member tools scoped to one durable member.

func (*Toolset) Tools 

func (toolset *Toolset) Tools() []agent.Tool

Tools returns a copy of the bound tool collection.

type ToolsetOption 

type ToolsetOption func(*toolsetConfig) error

ToolsetOption configures bound Team tools.

func WithToolCommandIDSource 

func WithToolCommandIDSource(source ToolCommandIDSource) ToolsetOption

WithToolCommandIDSource replaces the default ToolCall-derived command key.

type Transition 

type Transition struct {
	SchemaVersion int       `json:"schema_version"`
	ID            EventID   `json:"id"`
	TeamID        ID        `json:"team_id"`
	Revision      Revision  `json:"revision"`
	At            time.Time `json:"at"`
	Actor         Actor     `json:"actor"`
	CommandID     CommandID `json:"command_id"`
	CommandHash   string    `json:"command_hash"`
	Cause         Cause     `json:"cause"`
	TaskID        TaskID    `json:"task_id,omitempty"`
	MemberID      MemberID  `json:"member_id,omitempty"`
	AttemptID     AttemptID `json:"attempt_id,omitempty"`
	MessageID     MessageID `json:"message_id,omitempty"`
	From          Status    `json:"from,omitempty"`
	To            Status    `json:"to"`
	Reason        string    `json:"reason,omitempty"`
}

Transition is operational audit data for one Team revision.

type UnassignTaskRequest 

type UnassignTaskRequest struct {
	Command CommandMetadata
	TaskID  TaskID
}

UnassignTaskRequest removes an unclaimed task assignment.

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.