delegation

type AsyncBackend ¶

type AsyncBackend interface {
	Submit(ctx context.Context, req AsyncRequest) (id string, err error)
	Status(ctx context.Context, id string) (Response, error)
}

AsyncBackend stores asynchronous delegations and reports their status. Submit must safely replay the same non-empty Request.IdempotencyKey and identical AsyncRequest during the backend's declared retention window, returning the same id; reuse with different request semantics must return a conflict-classified error. The service borrows an injected backend and never closes it.

type AsyncRequest ¶

type AsyncRequest struct {
	Request Request `json:"request"`
	Caller  string  `json:"caller,omitempty"`
	Depth   int     `json:"depth"`

	// ParentRunID is the run id of the delegating agent's run. Sync
	// delegations derive it from the ambient execution context;
	// async delegations persist it here so it survives the queue.
	ParentRunID string `json:"parent_run_id,omitempty"`

	// CallID is the delegate tool call id that started the
	// delegation. Like ParentRunID it is derived from the caller's
	// execution context and persisted across the queue for async.
	CallID string `json:"call_id,omitempty"`

	// Stream is the queue-crossing reference to the caller's stream
	// sinks. Nil means the delegation carries no live stream (the
	// caller had no inheritable sinks, or the deployment does not
	// support async streaming).
	Stream *StreamRef `json:"stream,omitempty"`
}

AsyncRequest is the backend-neutral unit persisted by an AsyncBackend. Caller and Depth preserve delegation metadata across queue boundaries.

type Deployment ¶

type Deployment interface {
	Agent(id string) (*agent.Agent, bool)
	AgentNames() []string
}

Deployment is the minimal read-only deployment view needed by LocalDirectory. *deploy.Result implements this interface.

type Directory ¶

type Directory interface {
	List(ctx context.Context) ([]Target, error)
	Get(ctx context.Context, id string) (Target, error)
}

Directory discovers delegation targets independently of execution.

type IdempotencyRetentionProvider ¶

type IdempotencyRetentionProvider interface {
	IdempotencyRetention() time.Duration
}

IdempotencyRetentionProvider optionally exposes an AsyncBackend's finite terminal replay and status-query window.

type LocalDirectory ¶

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

Directory exposes the agents assembled by deploy.Build as local delegation targets. It may be constructed before Build and bound exactly once after it.

func NewDirectory() *LocalDirectory

func (d *LocalDirectory) Bind(result Deployment) error

func (d *LocalDirectory) BindDeployment(view deploy.View) error

func (d *LocalDirectory) BindTargetSource(source TargetSource) error

func (d *LocalDirectory) FreezeTargets(instances map[string]*agent.Agent)

func (d *LocalDirectory) Get(_ context.Context, id string) (Target, error)

func (d *LocalDirectory) List(context.Context) ([]Target, error)

func (d *LocalDirectory) Lookup(_ context.Context, id string) (*agent.Agent, error)

func (d *LocalDirectory) UnfreezeTargets()

type LocalService ¶

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

LocalService executes local sync delegations and coordinates optional async storage/workers.

func NewService(directory *LocalDirectory, backend AsyncBackend, opts ...Option) (*LocalService, error)

func (s *LocalService) BindSessionManager(manager *session.Manager) error

func (s *LocalService) Close() error

func (s *LocalService) Delegate(ctx context.Context, req Request) (Response, error)

func (s *LocalService) Get(ctx context.Context, id string) (Response, error)

func (s *LocalService) Run(ctx context.Context, req AsyncRequest) (Response, error)

func (s *LocalService) Start() error

type Mode ¶

type Mode string

Mode selects the lifecycle expected by a delegation request.

const (
	// ToolName is the canonical LLM-facing delegation tool name.
	ToolName = "delegate"

	// ModeSync waits for the delegated work and returns its terminal response.
	ModeSync Mode = "sync"
	// ModeAsync admits work and returns before it reaches a terminal state.
	ModeAsync Mode = "async"
)

func (m Mode) Validate() error

type Option ¶

type Option func(*serviceConfig) error

Option configures a local Service.

func WithDeferredWorkers() Option

func WithIdempotencyRetention(retention time.Duration) Option

func WithMaxConcurrency(limit int) Option

func WithMaxDepth(limit int) Option

func WithSessionInitializer(initializer SessionInitializer) Option

func WithSessionProvider(provider SessionProvider) Option

func WithStreamEscrowTTL(ttl time.Duration) Option

func WithStreamTargetExporter(exporter StreamTargetExporter) Option

func WithStreamTargetResolver(resolver StreamTargetResolver) Option

func WithTimeout(timeout time.Duration) Option

func WithWorkerHost(host agent.Host) Option

type RandomSessionProvider ¶ added in v0.1.23

type RandomSessionProvider struct{}

RandomSessionProvider mints a unique ContextID for every delegation and never persists: each delegated run is an ephemeral session with no history or resume state.

func (RandomSessionProvider) CreateContextID(context.Context, AsyncRequest) (string, error)

func (RandomSessionProvider) Persistent() bool

type Request ¶

type Request struct {
	Mode     Mode              `json:"mode"`
	Target   string            `json:"target"`
	Input    string            `json:"input"`
	Metadata map[string]string `json:"metadata,omitempty"`
	// Attributes are caller-declared run attributes for the delegated run
	// (execution policy keys, tenant hints, …). Core treats them as
	// opaque: they are copied onto the subagent's run attributes, below
	// the lineage keys the service owns. Declaring a lineage key
	// (tool.call_id) is a validation error rather than a silent override:
	// the key identifies the one tool call behind this delegation.
	//
	// Declare them explicitly; the caller's own run attributes are not
	// propagated, because deployments keep run-scoped payloads there
	// (e.g. checkpoint recovery blobs) that must not leak into a
	// subagent. Unlike Metadata they never become graph inputs, but like
	// Metadata they do take part in identity: both idempotent replay and
	// resume of a parked run compare the map verbatim, so a retry under a
	// different attribute map is a different request. The lineage keys are
	// the exception — a retry is a new tool call, so resume identity
	// ignores them.
	Attributes     map[string]string `json:"attributes,omitempty"`
	IdempotencyKey string            `json:"idempotency_key,omitempty"`
}

Request is the portable input accepted by a Service.

func (r Request) Validate() error

type Response ¶

type Response struct {
	ID       string            `json:"id"`
	Status   Status            `json:"status"`
	Output   string            `json:"output,omitempty"`
	Error    string            `json:"error,omitempty"`
	Metadata map[string]string `json:"metadata,omitempty"`
}

Response is a snapshot of delegated work.

func (r Response) Validate() error

type RunIDNotifier ¶ added in v0.1.30

type RunIDNotifier interface {
	// NoteRunID records runID for the work item identified by its
	// stream escrow ref. Unknown refs are no-ops; errors are
	// best-effort (recorded, never fail the run).
	NoteRunID(ctx context.Context, ref, runID string) error
}

RunIDNotifier is an optional AsyncBackend capability: it lets a claimed async work item record the subagent run id as soon as the run starts, so operational views can correlate a delegation id with its run before the terminal response arrives.

type Runner ¶

type Runner interface {
	Run(ctx context.Context, req AsyncRequest) (Response, error)
}

Runner is the restricted execution seam a queue-owned worker may use.

type Service ¶

type Service interface {
	Delegate(ctx context.Context, req Request) (Response, error)
	Get(ctx context.Context, id string) (Response, error)
}

Service starts delegated work and retrieves its latest response. Implementations must declare and guarantee a finite idempotency retention window. Within that window, requests with the same non-empty idempotency key and business fields must safely replay the same operation, and its response must remain queryable; reusing the key for a semantically different request must return a conflict-classified error. After the declared window expires, the key may start a new operation: this contract is bounded idempotency, not permanent exactly-once execution.

func ServiceFromHost(h agent.Host) (Service, bool)

type ServiceProvider ¶

type ServiceProvider interface {
	DelegationService() Service
}

ServiceProvider is the optional Host capability exposing delegation.

type ServiceSettings ¶

type ServiceSettings struct {
	MaxConcurrency       *int    `json:"max_concurrency,omitempty"`
	MaxDepth             *int    `json:"max_depth,omitempty"`
	Timeout              *string `json:"timeout,omitempty"`
	IdempotencyRetention *string `json:"idempotency_retention,omitempty"`
	DeferWorkers         bool    `json:"defer_workers,omitempty"`
}

ServiceSettings is the strict settings subtree of the local service factory. Zero fields leave the service defaults.

type SessionInitializer ¶ added in v0.8.1

type SessionInitializer func(ctx context.Context, key session.Key, req AsyncRequest) error

SessionInitializer runs once per delegated run, after the subagent's session identity is minted and before the turn starts or resumes. It is the seam for per-session state the application owns — sandbox mode, think level, model hint — which core deliberately does not model. Returning an error fails the delegation: a policy that could not be applied must not silently fall back to a wider default. A run on the manager-less path is initialized too when a SessionProvider supplied the identity.

type SessionProvider ¶ added in v0.1.23

type SessionProvider interface {
	// CreateContextID returns the subagent session's ContextID. AgentID is
	// filled by the service from the validated request target, so a
	// provider can never introduce an identity mismatch. Returning an
	// error fails the delegation; returning an empty ContextID is a
	// validation error.
	CreateContextID(ctx context.Context, req AsyncRequest) (string, error)

	// Persistent reports whether the provider's ContextID is stable and
	// should survive as a resumable session. Providers deriving stable
	// ContextIDs (e.g. by caller+target) return true; providers minting a
	// fresh ContextID per delegation return false.
	Persistent() bool
}

SessionProvider decides a delegated subagent's session ContextID and whether that identity is stable and needs persistence. Implementations are injected by the application. A nil provider falls back to the service default: mint a fresh ContextID per delegation, but only when a session manager is bound (see LocalService.runAt).

type Status ¶

type Status string

Status describes the backend-neutral lifecycle of delegated work.

const (
	StatusAccepted  Status = "accepted"
	StatusRunning   Status = "running"
	StatusSucceeded Status = "succeeded"
	StatusFailed    Status = "failed"
	StatusCanceled  Status = "canceled"
)

func (s Status) Terminal() bool

func (s Status) Validate() error

type StreamPolicySnapshot ¶ added in v0.1.30

type StreamPolicySnapshot struct {
	Visibility      session.Visibility `json:"visibility,omitempty"`
	Authority       session.Authority  `json:"authority,omitempty"`
	AckMode         session.AckMode    `json:"ack_mode,omitempty"`
	MaxUnacked      int                `json:"max_unacked,omitempty"`
	QueueSize       int                `json:"queue_size,omitempty"`
	DeliveryTimeout int64              `json:"delivery_timeout_ms,omitempty"`
}

StreamPolicySnapshot is a serializable projection of the caller's stream attachment policy, carried alongside a StreamTarget so the worker can materialize the attachment with the same tuning. It is populated from the caller's first sink spec; authority/ack fields are informational and the worker always downgrades inherited attachments to observers (see inheritedSinkSpecs).

type StreamRef ¶ added in v0.1.30

type StreamRef struct {
	// Ref is the service-side escrow key. Empty when no in-process
	// escrow exists.
	Ref string `json:"ref,omitempty"`
	// Policy snapshots the caller's attachment tuning.
	Policy StreamPolicySnapshot `json:"policy,omitempty"`
	// Target describes the delivery destination for resolver-based
	// materialization. Empty when the escrow path is used. At most
	// one target is persisted: cross-process delivery is
	// single-destination by contract.
	Target *StreamTarget `json:"target,omitempty"`
}

StreamRef is the queue-crossing reference to an async delegation's stream attachment. In-process backends use Ref (an escrow key that the service resolves to the caller's live sinks); cross-process backends use Target with a runtime-provided resolver.

type StreamTarget ¶ added in v0.1.30

type StreamTarget struct {
	// Kind is the target vocabulary understood by the resolver
	// (e.g. "conversation", "bus"). Resolvers MUST reject unknown
	// kinds — targets originate from persisted backend records and
	// must never drive arbitrary sink construction.
	Kind string `json:"kind"`
	// ID is the resolver-specific destination identifier.
	ID string `json:"id"`
}

StreamTarget is a serializable description of where an async subagent's stream envelopes should be delivered. It replaces the live (non-serializable) sink across the queue boundary: the caller's runtime knows how to describe its sinks as targets, and the worker's runtime resolves them back into live sinks through StreamTargetResolver.

AsyncRequest.Stream.Target is single-valued (see StreamRef): the in-process escrow preserves every inherited sink, but the cross-process path can restore exactly one destination. The exporter prefers broadcast (bus) targets when several sinks are describable.

type StreamTargetExporter ¶ added in v0.1.30

type StreamTargetExporter func(spec session.SinkSpec) (StreamTarget, bool)

StreamTargetExporter describes a caller-side live sink as a serializable StreamTarget at async submit time, so cross-process backends can re-materialize the same destination worker-side through a StreamTargetResolver. It returns ok=false for sinks that are not [StreamTargetProvider]s (e.g. plain closures with no durable destination). Cross-process delivery restores at most one target — see StreamTarget.

type StreamTargetProvider ¶ added in v0.1.30

type StreamTargetProvider interface {
	StreamTarget() (StreamTarget, bool)
}

StreamTargetProvider is implemented by stream sinks that can describe their durable destination as a serializable StreamTarget. [StreamTargetExporter]s recognize providers instead of concrete sink types, so decorators (logging middleware, proxies) can pass the description through without breaking recognition.

type StreamTargetResolver ¶ added in v0.1.30

type StreamTargetResolver func(ctx context.Context, target StreamTarget) (agent.StreamSink, error)

StreamTargetResolver materializes a live stream sink for a serializable StreamTarget on the worker side. Resolvers MUST whitelist target kinds and never construct sinks from free-form persisted data.

type Target ¶

type Target struct {
	ID          string            `json:"id"`
	Description string            `json:"description,omitempty"`
	Modes       []Mode            `json:"modes,omitempty"`
	Metadata    map[string]string `json:"metadata,omitempty"`
}

Target describes a backend-neutral delegation destination. Modes is an explicit allowlist; an empty Modes list means unrestricted support for every defined delegation mode.

func (t Target) Validate() error

type TargetSource ¶ added in v0.1.24

type TargetSource interface {
	// Dynamic resolves one dynamic target by id. Deployment targets
	// are served from the directory's own snapshot and never returned
	// here. The id is the AgentID — in the runtime path the
	// registration name — and implementations must resolve it as such.
	Dynamic(id string) (*agent.Agent, bool)
	// DynamicNames lists the dynamic target ids (AgentIDs).
	DynamicNames() []string
}

TargetSource is the live view of dynamically registered targets that a directory merges with its deployment snapshot. The runtime's agent registry implements it, so registration and unregistration take effect without re-binding the directory. Implementations must be safe for concurrent use.

type TargetViewBinder ¶ added in v0.1.24

type TargetViewBinder interface {
	// BindTargetSource attaches the live dynamic source. It is
	// set-once: a second bind returns a Conflict error.
	BindTargetSource(source TargetSource) error
}

TargetViewBinder is implemented by directory resources that want the runtime's live target source attached. The runtime calls it once per generation before the swap.

type TargetViewFreezer ¶ added in v0.1.24

type TargetViewFreezer interface {
	// FreezeTargets pins the live dynamic source to a fixed instance
	// set. Freezing is a one-shot commit: callers must guarantee that
	// every later failure path can either proceed without the live
	// view or pair the freeze with UnfreezeTargets, or the directory
	// will silently diverge from the runtime registry.
	FreezeTargets(instances map[string]*agent.Agent)
	// UnfreezeTargets restores the live dynamic source after a
	// rollback. Idempotent; safe to call on a never-frozen directory.
	UnfreezeTargets()
}

TargetViewFreezer is implemented by directory resources that must pin their dynamic view when the generation retires, so in-flight delegations keep resolving the instances that generation served.

Generation semantics: the delegation service uses the resolved instance for target validation and telemetry only; actual execution goes through the session manager's per-generation resolver. After a reload the retired directory may therefore validate against a frozen instance while the delegated turn starts on the new generation's re-bound instance — "the generation is determined at Session.Start", matching the runtime's resolver contract.

type Work ¶

type Work struct {
	ID         string          `json:"id"`
	LeaseToken string          `json:"lease_token"`
	Request    AsyncRequest    `json:"request"`
	Context    context.Context `json:"-"`
}

Work is one claimed asynchronous request. LeaseToken uniquely identifies this claim generation and must be supplied to WorkSource.Complete. Context optionally carries the backend-owned execution lease; canceling it must stop the claimed execution.

type WorkSource ¶

type WorkSource interface {
	Claim(ctx context.Context) (Work, error)
	Complete(ctx context.Context, id, leaseToken string, response Response) error
}

WorkSource is the optional worker side of an AsyncBackend. When implemented, Service starts bounded workers that claim backend-neutral work and complete it through the same backend. Claim must return a unique, non-empty lease token and unblock when ctx is canceled. Complete must only apply a terminal response when the token identifies the current lease; stale completions are ignored.