Documentation
¶
Overview ¶
Package coordination provides bounded coordination through ordinary Agent Definitions. InputGate turns one addressed input into a result carrying the original Signal identity. Deadline participates through the cancellable Timer Dispatcher. FirstSuccess owns a competition among exact child requests.
These Definitions use the public Step, Effect, Signal, and child-wait protocol. They have no scheduler, mailbox, journal, or lifecycle authority of their own. A Deployment supplies exact implementation and configuration digests; the configuration digest must cover schemas, bounds, and decision policies.
A competition selects among observed terminal results, in request order within one satisfaction Signal, after all candidate starts have settled. Slow admission delays selection even when an earlier child has already completed. Completing the competition cancels its remaining descendants. Use the drained child-wait boundary or Process.Join before reusing their exclusive resources.
Gates and competitions consume finite child, Effect, and Signal allocations. Gate replacement changes the recipient address. A delivery must remain bound to its original Process and WaitID until its admission and disposition are known; retrying against a replacement can consume the same input twice.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
Functions ¶
This section is empty.
Types ¶
type Deadline ¶
type Deadline struct {
// contains filtered or unexported fields
}
Deadline accepts an absolute time.Time and completes with that same instant after Timer acknowledges reaching it. Step never reads the clock. Restoration retains the absolute deadline rather than restarting a relative delay.
func NewDeadline ¶
func NewDeadline(config DeadlineConfig) (*Deadline, error)
func (*Deadline) Descriptor ¶
func (d *Deadline) Descriptor() agent.Descriptor
type DeadlineConfig ¶
DeadlineConfig names the immutable deadline Definition. Each execution's absolute instant is supplied as Input and retained in its state.
type FirstSuccess ¶
type FirstSuccess struct {
// contains filtered or unexported fields
}
FirstSuccess accepts a non-empty []agent.ChildSpec and runs those candidates under one ownership scope. It retains failed starts and observed terminal results, chooses the first accepted result in each request-ordered wait response, and completes with FirstSuccessResult. An empty Winner means every candidate failed to satisfy Accept; it is an explicit business result.
Every candidate's admission and initialization must settle before results are considered. A slow later admission can therefore delay acceptance of an already completed child. When multiple results are visible together, request order determines the winner, independently of their completion timestamps.
Results are observed at the terminal-result boundary. Completion starts cancellation of remaining descendants; it does not establish their drain.
Example ¶
package main
import (
"context"
"fmt"
"time"
agent "github.com/Tangerg/scope/agent"
"github.com/Tangerg/scope/agent/strategy/coordination"
"github.com/Tangerg/scope/agent/strategy/workflow"
)
func main() {
textSchema, err := agent.SchemaFor[string]()
if err != nil {
panic(err)
}
gate, err := coordination.NewInputGate(coordination.InputGateConfig{
Name: "example.input", Description: "Receive one identified replacement instruction.",
RequestSchema: textSchema, AnswerSchema: textSchema,
})
if err != nil {
panic(err)
}
deadline, err := coordination.NewDeadline(coordination.DeadlineConfig{
Name: "example.deadline", Description: "Wait for the coordination deadline.",
})
if err != nil {
panic(err)
}
transform, err := workflow.Transform("answer", func(_ context.Context, request string) (string, error) {
return "worker: " + request, nil
})
if err != nil {
panic(err)
}
worker, err := workflow.NewDefinition(workflow.DefinitionConfig{
Name: "example.worker", Description: "Prepare one deterministic answer.", Stages: []workflow.Stage{transform},
})
if err != nil {
panic(err)
}
race, err := coordination.NewFirstSuccess(coordination.FirstSuccessConfig{
Name: "example.coordination", Description: "Select the first completed work, input, or deadline.", MaxCandidates: 3,
Accept: func(_ context.Context, _ agent.ChildOutcome) (bool, error) { return true, nil },
})
if err != nil {
panic(err)
}
workerDeployment := exampleBinding(worker, nil)
gateDeployment := exampleBinding(gate, nil)
deadlineDeployment := exampleBinding(deadline, coordination.Timer{})
rootDeployment := exampleBinding(race, nil)
engine, err := agent.NewEngine(agent.EngineConfig{DeploymentResolver: resolver{
workerDeployment.DeploymentRef(): workerDeployment,
gateDeployment.DeploymentRef(): gateDeployment,
deadlineDeployment.DeploymentRef(): deadlineDeployment,
}})
if err != nil {
panic(err)
}
ctx := context.Background()
defer func() {
if closeErr := engine.Close(ctx); closeErr != nil {
panic(closeErr)
}
}()
requests := []agent.ChildSpec{
exampleCandidate("worker", workerDeployment, "inspect deployment"),
exampleCandidate("input", gateDeployment, "replacement instruction"),
exampleCandidate("deadline", deadlineDeployment, time.Now().Add(time.Hour)),
}
input, err := agent.EncodeInput(requests)
if err != nil {
panic(err)
}
process, err := engine.Start(ctx, rootDeployment, input)
if err != nil {
panic(err)
}
result, err := process.Await(ctx)
if err != nil {
panic(err)
}
if joinErr := process.Join(ctx); joinErr != nil {
panic(joinErr)
}
output, present := result.Output()
if !present {
panic("competition ended without an output")
}
report, err := output.Decode[coordination.FirstSuccessResult]()
if err != nil || !report.Valid() || report.Winner == nil {
panic("competition produced no successful candidate")
}
fmt.Println("winner:", report.Winner)
for _, outcome := range report.Outcomes {
if outcome.Key() == *report.Winner {
value, _ := outcome.Result().Output()
answer, decodeErr := value.Decode[string]()
if decodeErr != nil {
panic(decodeErr)
}
fmt.Println(answer)
}
}
}
func exampleBinding(definition agent.Definition, dispatcher agent.Dispatcher) agent.Deployment {
deployment, err := agent.NewDeployment(agent.DeploymentConfig{
Definition: definition, Dispatcher: dispatcher,
ImplementationDigest: agent.ComputeDigest([]byte("coordination-example-artifact")),
ConfigurationDigest: agent.ComputeDigest([]byte("coordination-example-config/" + definition.Descriptor().Name())),
})
if err != nil {
panic(err)
}
return deployment
}
func exampleCandidate[T any](key string, deployment agent.Deployment, value T) agent.ChildSpec {
childKey, err := agent.ParseChildKey(key)
if err != nil {
panic(err)
}
input, err := agent.EncodeInput(value)
if err != nil {
panic(err)
}
budget := agent.Budget{Steps: 16, Effects: 8, Signals: 16}
return agent.ChildSpec{Key: childKey, DeploymentRef: deployment.DeploymentRef(), Input: input, Budget: budget}
}
Output: winner: worker worker: inspect deployment
func NewFirstSuccess ¶
func NewFirstSuccess(config FirstSuccessConfig) (*FirstSuccess, error)
func (*FirstSuccess) Descriptor ¶
func (f *FirstSuccess) Descriptor() agent.Descriptor
func (*FirstSuccess) Restore ¶
func (f *FirstSuccess) Restore(state agent.ExecutionState) (agent.Execution, error)
type FirstSuccessConfig ¶
type FirstSuccessConfig struct {
Name string
Description string
MaxCandidates uint32
Accept SuccessPredicate
}
FirstSuccessConfig binds a finite candidate bound and pure decision policy. The Deployment configuration digest must identify both values.
type FirstSuccessResult ¶
type FirstSuccessResult struct {
Winner *agent.ChildKey `json:"winner,omitempty"`
Starts []agent.ChildStartResult `json:"starts"`
Outcomes []agent.ChildOutcome `json:"outcomes"`
}
FirstSuccessResult preserves child-start facts and the terminal outcomes seen before selection. Both slices retain request order. Children absent from Outcomes may still be settling when this result is published.
func (FirstSuccessResult) Valid ¶
func (f FirstSuccessResult) Valid() bool
type InputGate ¶
type InputGate struct {
// contains filtered or unexported fields
}
InputGate publishes its initial input as the wait-opening payload and returns one addressed answer as an immutable agent.Signal. Unaddressed inputs are not part of this protocol. Inputs accepted after its final Step window remain in the Process mailbox; a router must account for their disposition separately.
func NewInputGate ¶
func NewInputGate(config InputGateConfig) (*InputGate, error)
func (*InputGate) Descriptor ¶
func (i *InputGate) Descriptor() agent.Descriptor
type InputGateConfig ¶
type InputGateConfig struct {
Name string
Description string
RequestSchema agent.Schema
AnswerSchema agent.Schema
}
InputGateConfig separates the opening request from the addressed answer. The gate returns the original agent.Signal so its identity survives composition.
type SuccessPredicate ¶
SuccessPredicate decides whether a completed child satisfies the business goal. It runs inside Step and must be bounded, deterministic, side-effect-free, and honor ctx cancellation. Non-completed children never satisfy success.
type Timer ¶
type Timer struct{}
Timer is the stateless Dispatcher for Deadline. Its call owns and stops its timer. Waiting for the same absolute instant has no external side effect, so pending recovery can safely repeat that operation with the same identity. A settled Unknown still requires the Engine's explicit adjudication path.
func (Timer) Dispatch ¶
func (Timer) Dispatch(ctx context.Context, request agent.EffectRequest, _ agent.DeltaEmitter) (agent.Settlement, error)
Dispatch requires a non-nil context and panics if ctx is nil.
func (Timer) ReplayPolicy ¶
func (Timer) ReplayPolicy(effect agent.Effect) agent.ReplayPolicy