aether

package
v0.0.2 Latest Latest
Warning

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

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

Documentation

Overview

Package aether is the Aether transport for the harness: the agent-side half of a channel whose two halves are split across a broker instead of a Go channel.

A frontend (TUI/web/ACP/…) publishes a spec ChatMessage to this agent's topic and this Channel yields it from FetchTask; the turn's stream events go back to whichever client sent the turn. That makes Aether a multiplexer in front of the harness: several frontends, on several machines, driving one agent.

This is the OSS transport, so it assumes nothing beyond an Aether gateway: no app-server, no task lifecycle, no on-behalf-of authority, no sidecar relay. A distribution layers those on by decorating the turn executor, not by forking this file.

Index

Constants

View Source 
const (
	ScheduledTurnTaskType          = "agent-harness.scheduled-turn.v1"
	ScheduledTurnEnvelopeSchema    = "agent-harness.scheduled-turn.v1"
	ScheduledMissPolicySkip        = "skip"
	ScheduledMissPolicyFireOnce    = "fire_once"
	ScheduledMissPolicyFireAll     = "fire_all"
	ScheduledDispositionOrdinary   = "ordinary"
	ScheduledDispositionSkipped    = "skipped"
	ScheduledDispositionCoalesced  = "coalesced"
	ScheduledDispositionCatchUp    = "catch_up"
	ScheduledSkipReasonMissPolicy  = "miss_policy"
	ScheduledSkipReasonConcurrency = "max_concurrent"
)
View Source 
const ToolResultMetadataKey = "scitrera.result_metadata"
View Source 
const ToolTaskIDMetaKey = "scitrera.tool_task_id"

Variables

This section is empty.

Functions

func ExecutionBindingAccessRequest 

func ExecutionBindingAccessRequest(scope workspacepkg.ExecutionScope, correlationID string) (*pb.ResourceAccessRequest, error)

ExecutionBindingAccessRequest constructs the one canonical Aether check for admitting a workspace scope. Read/write remains an access-level ceiling on the stable bind operation.

func IsExactCatalogProviderTopic 

func IsExactCatalogProviderTopic(topic string) bool

IsExactCatalogProviderTopic reports whether topic selects one concrete service instance, agent instance, or user session suitable for a checked direct invocation. Wildcards and implementation-only service topics are intentionally rejected.

func IsScheduledTurnMessage 

func IsScheduledTurnMessage(message protocol.ChatMessage) bool

IsScheduledTurnMessage selects tasks whose claim/complete/fail lifecycle is owned by Aether rather than by a task-less interactive transport.

func IsWorkspaceToolName 

func IsWorkspaceToolName(name string) bool

IsWorkspaceToolName reports whether a built-in tool must execute against the exact workspace view selected for the turn. Distributions use this when they install transport-specific remote execution delegates.

func ReconcileScheduledTurns 

func ReconcileScheduledTurns(
	ctx context.Context,
	operations ScheduleOperations,
	routingWorkspace string,
	assignedTo string,
	registrations []ScheduledTurnRegistration,
	provider ScheduledTurnAuthorityProvider,
) error

ReconcileScheduledTurns converges Aether to one worker's desired declaration set without constructing a second transport. Embedding distributions should install their assignment executor before connecting, or reconcile before connection, so a newly created schedule cannot race handler registration.

func TaskInfoAuthority 

func TaskInfoAuthority(info *sdk.TaskInfo) (tools.MemoryAuthority, error)

TaskInfoAuthority validates and projects task-scoped authority for recovery.

func ValidateExecutionBindingAccessReceipt 

func ValidateExecutionBindingAccessReceipt(
	receipt *pb.AccessDecisionReceipt,
	onBehalfSubject *pb.PrincipalRef,
	scope workspacepkg.ExecutionScope,
	correlationID string,
	deliveryTarget string,
	now time.Time,
) error

ValidateExecutionBindingAccessReceipt compares gateway-authored transport metadata with the exact scope the application decoded. Payload copies of a receipt never reach this function.

Types

type AssignedContinuationExecutor 

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

AssignedContinuationExecutor reconstructs queued Aether continuation tasks into normal inbound turns. It never claims or finishes the task; the normal turn lifecycle and journal own those transitions.

func NewAssignedContinuationExecutor 

func NewAssignedContinuationExecutor(tasks TaskOperations, routingWorkspace, assignedTo string, cfg ContinuationExecutorConfig) (*AssignedContinuationExecutor, error)

func (*AssignedContinuationExecutor) HandleAssignment 

func (e *AssignedContinuationExecutor) HandleAssignment(ctx context.Context, assignment *sdk.TaskAssignment) error

func (*AssignedContinuationExecutor) RecoverQueued 

func (e *AssignedContinuationExecutor) RecoverQueued(ctx context.Context) error

RecoverQueued reconstructs goal tasks that were assigned before this worker disconnected. Aether's live assignment carries Payload, but its TaskInfo query projection intentionally does not; the credential-free envelope in the private continuation ledger is the durable source for restart recovery.

type AssignedExecutionScopeBinder 

type AssignedExecutionScopeBinder interface {
	BindAssignedExecutionScope(
		ctx context.Context,
		taskID string,
		scope workspacepkg.ExecutionScope,
	) (context.Context, func(), error)
}

AssignedExecutionScopeBinder validates an exact inherited view before task claim, decorates execution with the corresponding tool delegate, and returns a release hook. A bound envelope without this seam is rejected fail-closed.

type AssignedSubagentExecutor 

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

AssignedSubagentExecutor claims and executes targeted child tasks delivered to one Aether agent. It is deliberately at-most-once within a process and never replays a task already observed as running after a delivery gap.

func NewAssignedSubagentExecutor 

func NewAssignedSubagentExecutor(tasks TaskOperations, assignedTo string, cfg SubagentExecutorConfig) (*AssignedSubagentExecutor, error)

NewAssignedSubagentExecutor constructs a handler over the narrow task API so distributions can reuse it without constructing an OSS Channel.

func (*AssignedSubagentExecutor) HandleAssignment 

func (e *AssignedSubagentExecutor) HandleAssignment(ctx context.Context, assignment *sdk.TaskAssignment) error

HandleAssignment consumes only agent-harness.subagent.v1 tasks. Concurrent duplicate delivery is coalesced; terminal redelivery is a no-op.

type CatalogClient 

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

CatalogClient is the reusable Aether consumer for catalog query, describe, exact provider invocation, and checked cancellation. It deliberately owns the private Aether/runtime-envelope adaptation while portable tool payloads remain in ecosystem-messaging-spec.

func NewCatalogClient 

func NewCatalogClient(cfg CatalogClientConfig) (*CatalogClient, error)

func (*CatalogClient) DescribeCatalog 

func (c *CatalogClient) DescribeCatalog(ctx context.Context, request spec.ToolCatalogDescribeRequest, auth tools.MemoryAuthority) (catalog.ResolvedCatalogDescribeResult, error)

DescribeCatalog resolves one exact entry through the same authenticated catalog service boundary used by QueryCatalog.

func (*CatalogClient) InvokeCatalogTool 

func (c *CatalogClient) InvokeCatalogTool(
	ctx context.Context,
	providerRoute string,
	catalogContext spec.ToolCatalogContext,
	effect spec.ToolEffect,
	invocationAuthority *catalog.InvocationAuthorityProfile,
	envelope spec.ToolInvokeEnvelope,
	auth tools.MemoryAuthority,
) (json.RawMessage, error)

InvokeCatalogTool sends an exact checked invocation directly to the admitted provider. The catalog service is discovery/control plane only; it never forwards caller authority or tool payloads.

func (*CatalogClient) QueryCatalog 

func (c *CatalogClient) QueryCatalog(ctx context.Context, query spec.ToolCatalogQuery, auth tools.MemoryAuthority) (catalog.ResolvedCatalogPage, error)

QueryCatalog queries the private resolved catalog projection so consumers receive authenticated provider routes and server-owned invocation authority alongside each portable entry.

func (*CatalogClient) TryHandle 

func (c *CatalogClient) TryHandle(message *sdk.Message) bool

TryHandle delivers a reply only when both its correlation ID and the gateway-authenticated source topic match an in-flight request. Payload source and target fields never establish identity. A true result means the message was consumed and must not be treated as an ordinary turn.

type CatalogClientConfig 

type CatalogClientConfig struct {
	Route               string
	Tenant              string
	CatalogServiceTopic string
	Sender              CatalogToolMessageSender
	ServiceTimeout      time.Duration
	InvokeTimeout       time.Duration
}

CatalogClientConfig binds a catalog consumer to one exact Aether agent. Sender is normally the same AgentClient whose catch-all OnMessage handler calls CatalogClient.TryHandle.

type CatalogToolCallerAuthorizer 

type CatalogToolCallerAuthorizer interface {
	AuthorizeCatalogToolCaller(context.Context, CatalogToolCallerRequest) error
}

type CatalogToolCallerAuthorizerFunc 

type CatalogToolCallerAuthorizerFunc func(context.Context, CatalogToolCallerRequest) error

func (CatalogToolCallerAuthorizerFunc) AuthorizeCatalogToolCaller 

func (f CatalogToolCallerAuthorizerFunc) AuthorizeCatalogToolCaller(ctx context.Context, request CatalogToolCallerRequest) error

type CatalogToolCallerRequest 

type CatalogToolCallerRequest struct {
	SourceTopic string
	Subject     tools.MemoryAuthority
	Context     spec.ToolCatalogContext
	Entry       spec.ToolCatalogEntry
}

CatalogToolCallerRequest is transport-authenticated input to an optional deployment-local caller policy. Aether's exact entry decision remains the mandatory outer gate even when this hook is nil.

type CatalogToolExport 

type CatalogToolExport struct {
	Name                string
	Revision            string
	Effect              spec.ToolEffect
	InvocationAuthority *catalog.InvocationAuthorityProfile
}

CatalogToolExport is an explicit allowlist entry for one registry tool. Revision identifies the immutable implementation reviewed by catalog policy; InvocationAuthority is the host-local copy of the maximum server-owned OBO profile and is validated against gateway metadata before execution.

type CatalogToolHost 

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

CatalogToolHost executes an explicitly exported subset of one registry, publishes it under a bounded lease, and validates every invocation and cancellation against gateway-authored metadata.

func NewCatalogToolHost 

func NewCatalogToolHost(cfg CatalogToolHostConfig) (*CatalogToolHost, error)

func (*CatalogToolHost) Handle 

func (h *CatalogToolHost) Handle(ctx context.Context, message *sdk.Message) error

Handle is registered as the agent client's catch-all OnMessage handler so it can receive both catalog CHAT replies and TOOL_CALL invocations.

func (*CatalogToolHost) Publish 

func (h *CatalogToolHost) Publish(ctx context.Context) error

func (*CatalogToolHost) Renew 

func (h *CatalogToolHost) Renew(ctx context.Context) error

func (*CatalogToolHost) Revoke 

func (h *CatalogToolHost) Revoke(ctx context.Context) error

func (*CatalogToolHost) RunLease 

func (h *CatalogToolHost) RunLease(ctx context.Context) error

RunLease publishes, renews until ctx ends, and makes one bounded best-effort revoke. The catalog lease remains the crash fallback.

type CatalogToolHostConfig 

type CatalogToolHostConfig struct {
	Route                 string
	CatalogServiceTopic   string
	ProviderID            string
	RegistrationID        string
	Generation            string
	GenerationReplacement bool
	Context               spec.ToolCatalogContext
	Registry              *tools.Registry
	Exports               []CatalogToolExport
	Sender                CatalogToolMessageSender
	CallerAuthorizer      CatalogToolCallerAuthorizer
	LeaseDuration         time.Duration
	RenewInterval         time.Duration
	RPCTimeout            time.Duration
	Now                   func() time.Time
}

CatalogToolHostConfig configures one exact Aether agent as a catalog provider. Context is availability only and may be workspace-wide; Route is stored separately as authenticated operational state by the catalog.

type CatalogToolMessageSender 

type CatalogToolMessageSender interface {
	SendWithOptions(sdk.SendMessageOptions) error
}

type Channel 

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

Channel is the agent-side Aether transport. It bridges the SDK's push-based OnMessage delivery into the harness's pull-based FetchTask, and publishes a turn's stream events back to the client that sent the turn.

func New 

func New(cfg Config) (*Channel, error)

New constructs (but does not connect) the transport. Call Start to connect and begin pumping messages.

func (*Channel) BindAssignedExecutionScope 

func (c *Channel) BindAssignedExecutionScope(
	ctx context.Context,
	taskID string,
	scope workspacepkg.ExecutionScope,
) (context.Context, func(), error)

BindAssignedExecutionScope reconstructs the exact tool authority carried by a durable child task. Validation happens before claim. Worker views must be local to this exact agent; client views require both durable binding policy and an OBO provider because the executor is necessarily a different topic.

func (*Channel) BlobStore 

func (c *Channel) BlobStore(timeout time.Duration) (*KVBlobStore, error)

BlobStore returns a store bound to this agent's workspace-exclusive KV namespace. It is safe to construct before Start; operations require the channel's SDK connection to be running.

func (*Channel) ClaimTask 

func (c *Channel) ClaimTask(ctx context.Context, taskID string) error

func (*Channel) Close 

func (c *Channel) Close() error

Close shuts down the client.

func (*Channel) CompleteTask 

func (c *Channel) CompleteTask(ctx context.Context, taskID string) error

func (*Channel) EnableGoalContinuations 

func (c *Channel) EnableGoalContinuations(ledger goal.ContinuationLedger, handoff *authhandoff.Store, timeout time.Duration) (*ContinuationTaskBackend, error)

EnableGoalContinuations installs durable task admission and assignment on an OSS Channel. Call it before Start so live assignments cannot race handler registration; call ReconcileGoalContinuations after Start to recover assigned tasks whose original callback was lost with an earlier process.

func (*Channel) EnableScheduledTurns 

func (c *Channel) EnableScheduledTurns(
	ctx context.Context,
	registrations []ScheduledTurnRegistration,
	handoff *authhandoff.Store,
	timeout time.Duration,
) error

EnableScheduledTurns registers one mutable assignment handler, then applies the initial desired schedule set.

func (*Channel) EnableSubagentExecutor 

func (c *Channel) EnableSubagentExecutor(cfg SubagentExecutorConfig) (*AssignedSubagentExecutor, error)

EnableSubagentExecutor registers the executor on this channel. Call it before Start so no assignment can arrive before the handler is installed.

func (*Channel) Enqueue 

func (c *Channel) Enqueue(ctx context.Context, in channel.Inbound) error

Enqueue implements channel.Enqueuer: inject a turn locally, without a round trip through the gateway. Used for background pushes (a detached sub-agent notifying its parent thread), which originate inside this process.

func (*Channel) EnqueueBoundTurn 

func (c *Channel) EnqueueBoundTurn(
	ctx context.Context,
	in channel.Inbound,
	binding spec.ExecutionBinding,
	policy ScheduledViewPolicy,
) error

EnqueueBoundTurn admits an internally reconstructed task with an execution selector already authorized by its Aether assignment. The selector is stored atomically with enqueue and removed if the bounded inbox cannot accept it.

func (*Channel) FailTask 

func (c *Channel) FailTask(ctx context.Context, taskID, reason string) error

func (*Channel) FetchTask 

func (c *Channel) FetchTask(ctx context.Context) (channel.Inbound, error)

FetchTask blocks until a turn arrives, the receive loop fails, or ctx ends.

func (*Channel) GetTaskInfo 

func (c *Channel) GetTaskInfo(ctx context.Context, taskID string) (*sdk.TaskInfo, error)

func (*Channel) ListScheduledTurnScheduleStates 

func (c *Channel) ListScheduledTurnScheduleStates(ctx context.Context) ([]ScheduledTurnScheduleState, error)

ListScheduledTurnScheduleStates returns Aether's current state for the scheduled-turn declarations owned by this concrete worker. Other workflow schedules and other workers' declarations are deliberately omitted.

func (*Channel) LookupTaskInfo 

func (c *Channel) LookupTaskInfo(ctx context.Context, taskID string) (*sdk.TaskInfo, bool, error)

LookupTaskInfo returns authoritative task state while preserving the not-found/unauthorized ambiguity as found=false. Transport failures remain errors so startup recovery does not mistake an outage for absence.

func (*Channel) PrepareTurnRecovery 

func (c *Channel) PrepareTurnRecovery(ctx context.Context, info *sdk.TaskInfo) (func(), error)

PrepareTurnRecovery reconstructs process-local exact-view authority for a RUNNING scheduled turn before the durable turn journal resumes it. Aether task state and the current declaration digest are authoritative; no workspace path or policy is recovered from untrusted chat history. Non-scheduled tasks need no process-local binding and return a no-op cleanup.

func (*Channel) ProxyHTTP 

func (c *Channel) ProxyHTTP(ctx context.Context, target string, req *http.Request, opts ...sdk.ProxyOpt) (*http.Response, error)

ProxyHTTP routes a request over this channel's existing Aether connection. It intentionally exposes only the narrow SDK surface required by service transports such as MemoryLayer; the channel still owns the connection.

func (*Channel) PublishEvent 

func (c *Channel) PublishEvent(_ context.Context, event channel.Event) error

PublishEvent maps a harness stream event onto the wire and sends it to the client that submitted the turn.

func (*Channel) PublishSessionEvent 

func (c *Channel) PublishSessionEvent(ctx context.Context, event spec.SessionEvent) error

PublishSessionEvent implements sessionlog.SessionEventPublisher. An attach result is always delivered before events captured while it was assembled; each subscriber's subsequent sends remain cursor ordered.

func (*Channel) ReconcileGoalContinuations 

func (c *Channel) ReconcileGoalContinuations(ctx context.Context) error

ReconcileGoalContinuations recovers queued assignments after the SDK receive loop is live, so synchronous task queries can receive their responses.

func (*Channel) ReconcileScheduledTurns 

func (c *Channel) ReconcileScheduledTurns(ctx context.Context) error

func (*Channel) RejectSteering 

func (c *Channel) RejectSteering(ctx context.Context, in channel.Inbound, reason string)

RejectSteering tells the sender that an interjection never reached the turn it was aimed at (channel.SteeringRejector).

It routes purely by address: a steering send carries no task id, so there is no replyTo entry for it, and replyTopic falls back to the originating user session (addr.user_id + addr.request_id) — which the platform stamps on every inbound message. That is deliberate; keying a reply topic by message id would need a map with no reliable moment to delete the entries that ARE delivered.

Undeliverable rejections are dropped by PublishEvent's own routing check. The runtime has already logged the rejection, so a client on a transport that cannot be reached still leaves a trace.

func (*Channel) ScheduledRunReader 

func (c *Channel) ScheduledRunReader(journal ScheduledRunJournal, threads threadindex.WorkspaceLookup, timeout time.Duration) (*ScheduledRunReader, error)

ScheduledRunReader returns a joined reader for this Channel's routing workspace. This is additive: standalone/Aether-free operation does not build or call the reader.

func (*Channel) SessionBlobStore 

func (c *Channel) SessionBlobStore(timeout time.Duration) (*KVBlobStore, error)

SessionBlobStore is retained for source compatibility with the first event log integration. New callers should use BlobStore because lifecycle and other distributed state share the same neutral contract.

func (*Channel) SetApprovalBroker 

func (c *Channel) SetApprovalBroker(b *approval.Broker)

SetApprovalBroker wires the broker so inbound approve/deny controls resolve an in-flight turn's pending tool-approval prompt. Must be the same instance the turn runner awaits on.

func (*Channel) SetCanceller 

func (c *Channel) SetCanceller(tc *turncancel.Canceller)

SetCanceller wires the turn canceller so an inbound cancel control aborts the in-flight turn. Must be the same instance the runtime uses.

func (*Channel) SetExecutionBindingAuthorizer 

func (c *Channel) SetExecutionBindingAuthorizer(authorizer ExecutionBindingAuthorizer)

SetExecutionBindingAuthorizer enables authoritative validation of client view bindings. It may be set after connection, before accepting user turns.

func (*Channel) SetSessionService 

func (c *Channel) SetSessionService(service SessionService)

SetSessionService enables framed attach/snapshot/replay requests on the agent topic. It may be set after construction, before Start.

func (*Channel) SetThreadClearer 

func (c *Channel) SetThreadClearer(fn func(addr protocol.MessageAddress) error)

SetThreadClearer wires the handler invoked on an inbound `clear` control to drop a thread's persisted/cached history.

func (*Channel) SetToolCallAuthorizationProvider 

func (c *Channel) SetToolCallAuthorizationProvider(provider ToolCallAuthorizationProvider)

SetToolCallAuthorizationProvider enables gateway-validated OBO identity on reverse tool calls. Enterprise compositions use this together with an access authorizer and Aether ACLs; OSS intentionally leaves it nil.

func (*Channel) SetWorkerToolHost 

func (c *Channel) SetWorkerToolHost(host *WorkerToolHost)

SetWorkerToolHost installs the exact-view executor used by targeted durable background turns. It is intentionally separate from the channel's default local registry: a scheduled task must never silently fall back to another root if its pinned view is unavailable.

func (*Channel) Start 

func (c *Channel) Start(ctx context.Context) error

Start connects and runs the SDK receive loop in the background. The loop exits when ctx is cancelled or the connection drops.

func (*Channel) SubagentTaskBackend 

func (c *Channel) SubagentTaskBackend(timeout time.Duration) (*SubagentTaskBackend, error)

SubagentTaskBackend returns a task backend bound to this channel's Aether routing workspace. Logical workspaces remain explicit in task metadata and the parent-session projection.

func (*Channel) TargetedSubagentTaskBackend 

func (c *Channel) TargetedSubagentTaskBackend(timeout time.Duration, targetAgentID string) (*SubagentTaskBackend, error)

TargetedSubagentTaskBackend sends child executions to targetAgentID and lets the parent await authoritative task state without claiming or finishing them.

func (*Channel) Topic 

func (c *Channel) Topic() string

Topic reports the agent topic this channel receives turns on.

func (*Channel) TurnContext 

func (c *Channel) TurnContext(ctx context.Context, addr protocol.MessageAddress) context.Context

TurnContext binds the turn's built-in workspace tools to the exact client view captured at ingress. A turn without an execution binding stays on the worker, which is the required path for scheduled/background Aether work.

func (*Channel) UpdateScheduledTurns 

func (c *Channel) UpdateScheduledTurns(ctx context.Context, registrations []ScheduledTurnRegistration) error

UpdateScheduledTurns atomically changes the executor's accepted declaration set and reconciles Aether to that same desired state. Removed declarations are discovered from Aether and deleted, so restart-time edits converge too.

type Client 

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

Client is the frontend-side Aether transport. It presents the surface the UIs already consume from the in-process channels — Enqueue, Events, DroppedEvents — but the turn crosses a gateway to an agent worker instead of a Go channel to a local runner. It additionally carries cancel and approval decisions to that agent, so a UI can drive a remote turn exactly as it drives a local one.

func NewClient 

func NewClient(cfg ClientConfig) (*Client, error)

NewClient constructs (but does not connect) the frontend transport.

func (*Client) AgentTopic 

func (c *Client) AgentTopic() string

AgentTopic reports the agent this client submits turns to.

func (*Client) AttachSession 

func (c *Client) AttachSession(ctx context.Context, request spec.SessionAttachRequest) (spec.SessionAttachResult, error)

AttachSession requests a coherent snapshot and optional replay over the same Aether connection used for turns. The stable client_id belongs to the caller; when omitted, this frontend window is used as the stable reconnect identity.

func (*Client) Cancel 

func (c *Client) Cancel(taskID string) bool

Cancel aborts the in-flight turn by sending a cancel control to the agent. It reports whether the control was sent, not whether the turn had already finished — the agent decides that.

func (*Client) ClearThread 

func (c *Client) ClearThread(threadID string) bool

ClearThread tells the agent to drop a thread's stored history, so a clear issued in the UI reaches the side that actually owns the transcript.

func (*Client) Close 

func (c *Client) Close() error

Close shuts down the client.

func (*Client) DroppedEvents 

func (c *Client) DroppedEvents() int64

DroppedEvents counts token deltas shed because the UI could not keep up.

func (*Client) Enqueue 

func (c *Client) Enqueue(ctx context.Context, in channel.Inbound) error

Enqueue submits a turn to the agent. It stamps this session's identity on the address so the agent can route the reply back even if the transport does not report a source topic.

func (*Client) Events 

func (c *Client) Events() <-chan channel.Event

Events exposes the decoded stream events.

func (*Client) ProxyHTTP 

func (c *Client) ProxyHTTP(ctx context.Context, target string, req *http.Request, opts ...sdk.ProxyOpt) (*http.Response, error)

ProxyHTTP routes a request over the frontend's existing Aether connection. Service integrations can therefore share its authenticated user session without opening a second identity or connection.

func (*Client) Resolve 

func (c *Client) Resolve(taskID, requestID string, d approval.Decision) bool

Resolve settles a pending tool approval by sending an approve/deny control to the agent holding the prompt.

func (*Client) SessionErrors 

func (c *Client) SessionErrors() <-chan SessionRemoteError

SessionErrors exposes asynchronous gap/not-supported notices that no longer have a waiting AttachSession caller.

func (*Client) SessionEvents 

func (c *Client) SessionEvents() <-chan spec.SessionEvent

SessionEvents exposes cursor-bearing events for sessions attached by this client. They are distinct from the legacy per-turn Events stream.

func (*Client) SetHistoryProjection 

func (c *Client) SetHistoryProjection(p HistoryProjection)

SetHistoryProjection wires the local transcript copy. Without it the UI still renders a live turn, but a restart or a thread switch shows nothing for a conversation this process did not witness.

func (*Client) SetToolHost 

func (c *Client) SetToolHost(host *ClientToolHost)

SetToolHost enables exact-host workspace tool execution on this client.

func (*Client) SetWorkspaceHistoryProjection 

func (c *Client) SetWorkspaceHistoryProjection(p WorkspaceHistoryProjection)

SetWorkspaceHistoryProjection wires a projection that preserves composite workspace/thread identity for a client that can change logical projects.

func (*Client) Start 

func (c *Client) Start(ctx context.Context) error

Start connects and runs the receive loop in the background.

func (*Client) ToolHostID 

func (c *Client) ToolHostID() string

ToolHostID is the gateway-routable identity bound to this exact frontend window. The gateway-provided SourceTopic must match it on the worker.

type ClientConfig 

type ClientConfig struct {
	// ServerAddr is the gateway address. Required.
	ServerAddr string
	// Workspace is the Aether workspace the target agent lives in.
	Workspace string
	// SessionWorkspace is the default application-level workspace stamped on
	// turns. Empty falls back to the Aether routing Workspace.
	SessionWorkspace string
	// AgentImplementation and AgentSpecifier address the agent: together with
	// Workspace they form ag::<workspace>::<implementation>::<specifier>.
	AgentImplementation string
	AgentSpecifier      string
	// UserID and WindowID identify this client session. WindowID distinguishes
	// two frontends run by the same user, and is what the agent replies to.
	UserID   string
	WindowID string

	// Credentials. All optional; a dev-mode gateway accepts an unauthenticated
	// connection.
	APIKey string
	Token  string
	Tenant string

	TLSEnabled            bool
	TLSInsecureSkipVerify bool
	TLSServerName         string
	RootCAs               []byte
	ClientCert            []byte
	ClientKey             []byte
}

ClientConfig configures the frontend-side Aether transport.

type ClientToolAccessAuthorizer 

type ClientToolAccessAuthorizer interface {
	AuthorizeClientTool(ctx context.Context, request ClientToolAccessRequest) error
}

ClientToolAccessAuthorizer is the client-side half of tool sharing policy. Aether ACLs remain the outer routing gate; this is the resource-level gate.

type ClientToolAccessRequest 

type ClientToolAccessRequest struct {
	Binding    spec.ExecutionBinding
	Policy     workspacepkg.ExecutionViewPolicy
	ToolName   string
	AgentTopic string
	OnBehalfOf workspacepkg.Principal
	Address    spec.MessageAddress
}

ClientToolAccessRequest carries both the logical binding and the identities authenticated by Aether. Enterprise policies can distinguish same-user, named-principal, and workspace-member sharing without trusting envelope JSON.

type ClientToolHost 

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

ClientToolHost owns host-private paths and executes only invocations carrying a matching logical view binding for this exact Aether user window.

func NewClientToolHost 

func NewClientToolHost(ctx context.Context, cfg ClientToolHostConfig) (*ClientToolHost, error)

NewClientToolHost registers and optionally publishes the initial local view.

func (*ClientToolHost) ExecutionBindingForDirectory 

func (h *ClientToolHost) ExecutionBindingForDirectory(ctx context.Context, dir string) (spec.ExecutionBinding, error)

func (*ClientToolHost) GrantWorkingDirectory 

func (h *ClientToolHost) GrantWorkingDirectory(dir string) error

func (*ClientToolHost) GrantWorkspaceDirectory 

func (h *ClientToolHost) GrantWorkspaceDirectory(dir string) error

func (*ClientToolHost) Renew 

func (h *ClientToolHost) Renew(ctx context.Context) error

Renew republishes every view hosted by this client before its authoritative observation lease expires.

func (*ClientToolHost) ResolveWorkspaceForDirectory 

func (h *ClientToolHost) ResolveWorkspaceForDirectory(ctx context.Context, dir string) (string, error)

ResolveWorkspaceForDirectory exposes only the logical project identity the TUI needs to select its thread/history partition.

type ClientToolHostConfig 

type ClientToolHostConfig struct {
	WorkspaceID   string
	WorkspaceRoot string
	StateDir      string
	ToolHostID    string
	// AgentTopic is the only caller accepted by the default OSS access policy.
	AgentTopic string
	// AccessAuthorizer may replace the strict agent-topic check in an enterprise
	// composition that also validates OBO grants and workspace sharing policy.
	AccessAuthorizer ClientToolAccessAuthorizer
	Publisher        workspacepkg.ViewPublisher
	Python           string
	Timeout          time.Duration
	MaxOutput        int
}

ClientToolHostConfig configures a frontend that owns local workspace tools.

type Config 

type Config struct {
	// ServerAddr is the gateway address, e.g. "127.0.0.1:50051". Required.
	ServerAddr string
	// Workspace is the Aether workspace this agent joins.
	Workspace string
	// Implementation and Specifier complete the agent topic. Implementation
	// defaults to "agent-harness"; Specifier distinguishes instances.
	Implementation string
	Specifier      string
	// SourceAgent labels outbound envelopes, e.g. "agent-harness:<host>".
	SourceAgent string
	// AgentName resolves the assistant's display persona, stamped on outbound
	// messages at meta.scitrera.agent_name. Invoked per message so a rename mid
	// session takes effect without a restart. Nil, or "" , omits the stamp.
	AgentName func() string
	// SessionWorkspace is the application-level default used when an attach
	// request omits workspace_id. It may differ from the Aether routing
	// workspace. Empty falls back to Workspace.
	SessionWorkspace  string
	WorkspaceResolver WorkspaceResolver
	// PreferTaskMessageLanes routes turn stream envelopes to
	// tk::<workspace>::<task>::msg when both identities are present. Enable this
	// only when task IDs name real Aether tasks whose recipients are subscribed;
	// the default direct-reply path supports task-less OSS deployments.
	PreferTaskMessageLanes bool
	// ScheduledTurnAuthority optionally authorizes schedule CRUD and supplies a
	// bounded private grant for scheduled task creation. Required-authority
	// declarations fail closed when this is nil.
	ScheduledTurnAuthority ScheduledTurnAuthorityProvider

	// Credentials. All optional: an aetherlite gateway in dev mode accepts an
	// unauthenticated connection, which is the zero-setup path.
	APIKey string
	Token  string
	Tenant string

	// TLS. Disabled by default (plaintext), which is what a local dev gateway
	// speaks.
	TLSEnabled            bool
	TLSInsecureSkipVerify bool
	TLSServerName         string
	RootCAs               []byte
	ClientCert            []byte
	ClientKey             []byte
}

Config configures the agent-side Aether transport.

type ContinuationExecutorConfig 

type ContinuationExecutorConfig struct {
	Enqueuer    channel.Enqueuer
	AuthHandoff *authhandoff.Store
	Ledger      goal.ContinuationLedger
	Timeout     time.Duration
}

ContinuationExecutorConfig configures assignment-to-ingress delivery. The handoff store must be the same one the turn runner consumes.

type ContinuationTaskBackend 

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

ContinuationTaskBackend maps the neutral goal admission contract to a Aether task targeted to the admitting worker's stable identity. The task payload is credential-free; delegated authority is carried only in Aether's typed AuthorizationContext.

func NewContinuationTaskBackend 

func NewContinuationTaskBackend(tasks TaskOperations, workspace, namespace string, timeout time.Duration) (*ContinuationTaskBackend, error)

func (*ContinuationTaskBackend) Admit 

func (b *ContinuationTaskBackend) Admit(ctx context.Context, admission goal.ContinuationAdmission) (goal.ContinuationReceipt, error)

func (*ContinuationTaskBackend) Inspect 

func (b *ContinuationTaskBackend) Inspect(ctx context.Context, receipt goal.ContinuationReceipt) (goal.ContinuationState, error)

type ExecutionBindingAuthorizer 

type ExecutionBindingAuthorizer interface {
	AuthorizeExecutionBinding(ctx context.Context, request workspacepkg.ExecutionBindingAuthorizationRequest) error
}

ExecutionBindingAuthorizer verifies that a client-hosted view binding exists in the configured durable workspace authority.

type HistoryProjection 

type HistoryProjection interface {
	LoadHistory(ctx context.Context, threadID string) ([]protocol.ChatMessage, error)
	SaveHistory(ctx context.Context, threadID string, messages []protocol.ChatMessage) error
}

HistoryProjection is the local transcript copy a client may keep. It is the read/write pair of the UI's history store.

type KVBlobStore 

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

KVBlobStore adapts Aether's workspace-exclusive KV namespace to the neutral compare-and-swap blob contract used by distributed session and lifecycle stores. The Aether agent implementation+specifier provides host identity; logical workspace and session identities remain encoded in each store key.

func NewKVBlobStore 

func NewKVBlobStore(kv KVOperations, workspace string, timeout time.Duration) (*KVBlobStore, error)

func (*KVBlobStore) CompareAndSwap 

func (s *KVBlobStore) CompareAndSwap(ctx context.Context, key string, expected, value []byte) (bool, error)

func (*KVBlobStore) Create 

func (s *KVBlobStore) Create(ctx context.Context, key string, value []byte) (bool, error)

func (*KVBlobStore) Read 

func (s *KVBlobStore) Read(ctx context.Context, key string) ([]byte, bool, error)

type KVOperations 

type KVOperations interface {
	GetSync(ctx context.Context, opts sdk.KVGetOptions) (*sdk.KVResponse, error)
	SetNXSync(ctx context.Context, key string, value []byte, scope sdk.KVScope, userID, workspace string, ttl, timeout time.Duration) (bool, error)
	CompareAndSetSync(ctx context.Context, key string, expected, value []byte, scope sdk.KVScope, userID, workspace string, ttl, timeout time.Duration) (bool, error)
}

KVOperations is the atomic subset of the Aether SDK KV client required by the session event adapter. The interface keeps the adapter unit-testable without a gateway while retaining the SDK's exact scope and timeout options.

type ScheduleOperations 

type ScheduleOperations interface {
	ListSchedulesAuthorized(context.Context, string, *pb.AuthorizationContext) (*sdk.WorkflowResponse, error)
	UpsertScheduleWithOptions(context.Context, []byte, sdk.WorkflowScheduleOperationOptions) (*sdk.WorkflowResponse, error)
	DeleteScheduleAuthorized(context.Context, string, string, *pb.AuthorizationContext) (*sdk.WorkflowResponse, error)
}

ScheduleOperations is the bounded WorkflowEngine SDK surface required by scheduled-turn reconciliation. It is exported so distributions that already own an Aether connection can reuse the exact OSS reconciler.

type ScheduledOperationsAuthorizer 

type ScheduledOperationsAuthorizer interface {
	AuthorizeScheduledOperations(
		ctx context.Context,
		addr protocol.MessageAddress,
		user protocol.ChatMessage,
		command string,
	) error
}

ScheduledOperationsAuthorizer is the enterprise policy seam for operations inspection. OSS is single-user and passes nil; multi-user hosts can authorize the complete authenticated/OBO message before any Aether or MemoryLayer read.

type ScheduledOperationsCommands 

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

ScheduledOperationsCommands renders the worker-owned /schedules and /runs inspection surface. It owns no state: every invocation reads Aether and the configured journal/thread authorities through the existing projections.

func NewScheduledOperationsCommands 

func NewScheduledOperationsCommands(
	c *Channel,
	journal ScheduledRunJournal,
	threads threadindex.WorkspaceLookup,
	authorizer ScheduledOperationsAuthorizer,
	timeout time.Duration,
) (*ScheduledOperationsCommands, error)

NewScheduledOperationsCommands constructs the operations surface for an Aether worker. The returned value structurally implements turn's optional ScheduledOperationsCommandProvider without coupling this channel package to the runner package.

func (*ScheduledOperationsCommands) RunScheduledOperationsCommand 

func (c *ScheduledOperationsCommands) RunScheduledOperationsCommand(
	ctx context.Context,
	addr protocol.MessageAddress,
	user protocol.ChatMessage,
	name string,
	args string,
) (string, error)

RunScheduledOperationsCommand handles only schedules/runs. User mistakes are rendered with usage and do not fail the surrounding task; authority/backend failures remain errors so the task lifecycle records the failed inspection.

type ScheduledRun 

type ScheduledRun struct {
	TaskID            string                 `json:"task_id"`
	TaskStatus        string                 `json:"task_status"`
	TaskError         string                 `json:"task_error,omitempty"`
	RoutingWorkspace  string                 `json:"routing_workspace"`
	AssignedTo        string                 `json:"assigned_to"`
	LogicalWorkspace  string                 `json:"logical_workspace"`
	ThreadID          string                 `json:"thread_id"`
	ViewID            string                 `json:"view_id"`
	ViewRevision      string                 `json:"view_revision"`
	ToolHostID        string                 `json:"tool_host_id"`
	EnvelopeSchema    string                 `json:"envelope_schema"`
	DeclarationDigest string                 `json:"declaration_digest"`
	Occurrence        ScheduledRunOccurrence `json:"occurrence"`
	State             string                 `json:"state"`
	Attempt           int32                  `json:"attempt"`
	MaxAttempts       int32                  `json:"max_attempts"`
	CreatedAt         time.Time              `json:"created_at"`
	StartedAt         *time.Time             `json:"started_at,omitempty"`
	CompletedAt       *time.Time             `json:"completed_at,omitempty"`
	Execution         *ScheduledRunExecution `json:"execution,omitempty"`
	Thread            *threadindex.Session   `json:"thread,omitempty"`
}

ScheduledRun is a read-only join. TaskStatus/Error/timestamps come from Aether; Execution comes from the durable journal; Thread is MemoryLayer thread metadata. A nil Execution or Thread is expected before the worker creates either record and is not filled from another source.

type ScheduledRunExecution 

type ScheduledRunExecution struct {
	Revision      uint64            `json:"revision"`
	OwnerIdentity string            `json:"owner_identity"`
	Phase         turnjournal.Phase `json:"phase"`
	FailureReason string            `json:"failure_reason,omitempty"`
	UpdatedAt     time.Time         `json:"updated_at"`
	CompletedAt   *time.Time        `json:"completed_at,omitempty"`
}

ScheduledRunExecution is the exact durable turn-journal checkpoint. Phase is intentionally not collapsed: completing/failing/interrupting identify the durable-journal-to-Aether terminal outbox boundary.

type ScheduledRunJournal 

type ScheduledRunJournal interface {
	Get(ctx context.Context, workspaceID, taskID string) (turnjournal.Record, error)
}

ScheduledRunJournal is the durable execution lookup required by ScheduledRunReader. In the OSS Aether worker this is Aether KV-backed; missing records are represented by turnjournal.ErrNotFound.

type ScheduledRunOccurrence 

type ScheduledRunOccurrence struct {
	ScheduleID                string    `json:"schedule_id"`
	DeclarationID             string    `json:"declaration_id"`
	MissPolicy                string    `json:"miss_policy"`
	Disposition               string    `json:"disposition"`
	BacklogCount              int       `json:"backlog_count"`
	BacklogTruncated          bool      `json:"backlog_truncated"`
	BacklogIndex              int       `json:"backlog_index"`
	ScheduledFor              time.Time `json:"scheduled_for"`
	DispatchedAt              time.Time `json:"dispatched_at"`
	DispatchDelayMilliseconds int64     `json:"dispatch_delay_ms"`
}

ScheduledRunOccurrence describes the exact WorkflowEngine occurrence that created a task. ScheduleID is Aether's deterministic workflow-schedule ID; DeclarationID is the operator-facing Sahara declaration ID.

type ScheduledRunPage 

type ScheduledRunPage struct {
	Runs          []ScheduledRun `json:"runs"`
	TotalCount    int32          `json:"total_count"`
	NextPageToken string         `json:"next_page_token,omitempty"`
}

ScheduledRunPage preserves Aether's ordering, total, and opaque next cursor. No client should parse or synthesize NextPageToken.

type ScheduledRunQuery 

type ScheduledRunQuery struct {
	Limit     int32
	PageToken string
	Statuses  []pb.TaskStatus
}

ScheduledRunQuery selects one cursor page. Statuses are Aether protobuf enum values; UNSPECIFIED is rejected because it is not a lifecycle state.

type ScheduledRunReader 

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

ScheduledRunReader joins the two existing authorities without persisting a third projection. It is safe to construct without enabling scheduled-turn execution on the local Channel, which makes it reusable by operations hosts.

func NewScheduledRunReader 

func NewScheduledRunReader(
	tasks ScheduledRunTaskQueries,
	journal ScheduledRunJournal,
	threads threadindex.WorkspaceLookup,
	routingWorkspace string,
	timeout time.Duration,
) (*ScheduledRunReader, error)

func (*ScheduledRunReader) Query 

func (r *ScheduledRunReader) Query(ctx context.Context, query ScheduledRunQuery) (ScheduledRunPage, error)

Query returns one joined page. Missing journal/thread rows remain visible as task-only results; malformed identity metadata or a backend failure fails the page instead of silently returning a partial or cross-workspace join.

type ScheduledRunTaskQueries 

type ScheduledRunTaskQueries interface {
	QueryTasks(ctx context.Context, filter *pb.TaskFilter, timeout time.Duration) (*sdk.TaskQueryResponse, error)
}

ScheduledRunTaskQueries is the narrow Aether task-list surface required by ScheduledRunReader. Aether remains authoritative for task lifecycle and owns the opaque page cursor.

type ScheduledTurnAuthority 

type ScheduledTurnAuthority struct {
	Authorization  *pb.AuthorizationContext
	AuthorityScope *pb.WorkflowScheduleAuthorityScope
}

ScheduledTurnAuthority remains on the Aether transport envelope. It is never serialized into scheduledWorkflowDefinition, the task payload, or metadata.

type ScheduledTurnAuthorityOperation 

type ScheduledTurnAuthorityOperation string
const (
	ScheduledTurnAuthorityList   ScheduledTurnAuthorityOperation = "list"
	ScheduledTurnAuthorityUpsert ScheduledTurnAuthorityOperation = "upsert"
	ScheduledTurnAuthorityDelete ScheduledTurnAuthorityOperation = "delete"
)

type ScheduledTurnAuthorityProvider 

type ScheduledTurnAuthorityProvider interface {
	AuthorityForScheduledTurn(
		ctx context.Context,
		request ScheduledTurnAuthorityRequest,
	) (ScheduledTurnAuthority, error)
}

ScheduledTurnAuthorityProvider supplies transport-owned authorization for WorkflowEngine schedule management. Implementations may use direct or OBO authority, but must never place grants in a schedule declaration or action JSON. A nil provider retains direct, non-authority scheduled turns.

type ScheduledTurnAuthorityProviderFunc 

type ScheduledTurnAuthorityProviderFunc func(
	context.Context,
	ScheduledTurnAuthorityRequest,
) (ScheduledTurnAuthority, error)

func (ScheduledTurnAuthorityProviderFunc) AuthorityForScheduledTurn 

func (f ScheduledTurnAuthorityProviderFunc) AuthorityForScheduledTurn(
	ctx context.Context,
	request ScheduledTurnAuthorityRequest,
) (ScheduledTurnAuthority, error)

type ScheduledTurnAuthorityRequest 

type ScheduledTurnAuthorityRequest struct {
	Operation          ScheduledTurnAuthorityOperation
	RoutingWorkspace   string
	WorkflowScheduleID string
	AssignedTo         string
	Registration       *ScheduledTurnRegistration
}

ScheduledTurnAuthorityRequest is credential-free context for one schedule management operation. Registration is present only for upsert and is a copy; providers must derive credentials from their own authority source.

type ScheduledTurnExecutor 

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

ScheduledTurnExecutor converts exact-target Aether assignments into ordinary harness turns. It does not claim or finish tasks; the normal turn lifecycle owns those transitions and restart journal recovery.

func NewScheduledTurnExecutor 

func NewScheduledTurnExecutor(
	tasks TaskOperations,
	routingWorkspace string,
	assignedTo string,
	registrations []ScheduledTurnRegistration,
	enqueuer boundTurnEnqueuer,
	handoff *authhandoff.Store,
	timeout time.Duration,
) (*ScheduledTurnExecutor, error)

func (*ScheduledTurnExecutor) HandleAssignment 

func (e *ScheduledTurnExecutor) HandleAssignment(ctx context.Context, assignment *sdk.TaskAssignment) error

func (*ScheduledTurnExecutor) RecoverQueued 

func (e *ScheduledTurnExecutor) RecoverQueued(ctx context.Context) error

RecoverQueued recreates payloads from the current declaration only when its digest exactly matches the task metadata. This avoids replaying a queued task with a newly edited prompt or a replacement workspace revision.

func (*ScheduledTurnExecutor) Registrations 

func (e *ScheduledTurnExecutor) Registrations() []ScheduledTurnRegistration

Registrations returns a source-independent copy of the executor's current declaration set. It is used by embedding distributions to restore the previous accepted set when a remote reconciliation fails.

func (*ScheduledTurnExecutor) ReplaceRegistrations 

func (e *ScheduledTurnExecutor) ReplaceRegistrations(registrations []ScheduledTurnRegistration) error

ReplaceRegistrations atomically installs a fully validated desired set. A task created from a removed or changed declaration is rejected immediately, even while the corresponding remote schedule reconciliation is in flight.

type ScheduledTurnRegistration 

type ScheduledTurnRegistration struct {
	ID                              string                `json:"id"`
	Name                            string                `json:"name"`
	ScheduleType                    string                `json:"schedule_type"`
	ScheduleExpression              string                `json:"schedule_expression"`
	ThreadID                        string                `json:"thread_id"`
	Prompt                          string                `json:"prompt"`
	MissPolicy                      string                `json:"miss_policy"`
	TargetOfflinePolicy             string                `json:"target_offline_policy"`
	Enabled                         bool                  `json:"enabled"`
	RequireTaskAuthority            bool                  `json:"require_task_authority,omitempty"`
	RequiredDownstreamAuthorityHops uint32                `json:"required_downstream_authority_hops,omitempty"`
	Binding                         spec.ExecutionBinding `json:"execution_binding"`
	ViewPolicy                      ScheduledViewPolicy   `json:"view_policy"`
}

ScheduledTurnRegistration is the validated, host-bound form of one config declaration. A concrete worker binding is part of the declaration digest so delayed tasks cannot drift to a replacement checkout or a new Git revision.

func (ScheduledTurnRegistration) Validate 

func (r ScheduledTurnRegistration) Validate() error

type ScheduledTurnScheduleOccurrence 

type ScheduledTurnScheduleOccurrence struct {
	ScheduledFor     time.Time  `json:"scheduled_for"`
	DispatchedAt     *time.Time `json:"dispatched_at,omitempty"`
	Disposition      string     `json:"disposition"`
	Reason           string     `json:"reason,omitempty"`
	BacklogCount     int        `json:"backlog_count"`
	BacklogTruncated bool       `json:"backlog_truncated"`
	BacklogIndex     int        `json:"backlog_index"`
}

ScheduledTurnScheduleOccurrence is Aether's bounded authoritative summary of the latest decision for one schedule. DispatchedAt is nil for a no-task skip, which remains observable here even though it cannot appear in task history.

type ScheduledTurnScheduleState 

type ScheduledTurnScheduleState struct {
	WorkflowScheduleID string                           `json:"workflow_schedule_id"`
	DeclarationID      string                           `json:"declaration_id"`
	DeclarationDigest  string                           `json:"declaration_digest"`
	Name               string                           `json:"name"`
	ScheduleType       string                           `json:"schedule_type"`
	ScheduleExpression string                           `json:"schedule_expression"`
	RoutingWorkspace   string                           `json:"routing_workspace"`
	AssignedTo         string                           `json:"assigned_to"`
	LogicalWorkspace   string                           `json:"logical_workspace"`
	ThreadID           string                           `json:"thread_id"`
	ViewID             string                           `json:"view_id"`
	ViewRevision       string                           `json:"view_revision,omitempty"`
	MissPolicy         string                           `json:"miss_policy"`
	MaxConcurrent      int                              `json:"max_concurrent"`
	Enabled            bool                             `json:"enabled"`
	NextFireAt         *time.Time                       `json:"next_fire_at,omitempty"`
	LastFiredAt        *time.Time                       `json:"last_fired_at,omitempty"`
	LastOccurrence     *ScheduledTurnScheduleOccurrence `json:"last_occurrence,omitempty"`
}

ScheduledTurnScheduleState is the operations projection of one Sahara-owned Aether schedule. It is read-only; declarations and reconciliation remain worker-owned and Aether remains authoritative for scheduling.

type ScheduledViewPolicy 

type ScheduledViewPolicy struct {
	WriteAccess      workspacepkg.ViewWriteAccess `json:"write_access"`
	AllowMutableView bool                         `json:"allow_mutable_view"`
	AllowDirtyView   bool                         `json:"allow_dirty_view"`
}

ScheduledViewPolicy is captured with a versioned task envelope. False values are the safe default: require a clean Git worktree.

func (ScheduledViewPolicy) Validate 

func (p ScheduledViewPolicy) Validate() error

Validate rejects policy combinations that would permit a first write and then strand the schedule on the dirty-view check before its next tool call.

type SessionRemoteError 

type SessionRemoteError struct {
	RequestID string
	Code      string
	Message   string
	Retryable bool
}

SessionRemoteError is a structured error returned by the remote session service. RequestID correlates attach failures and post-attach gap notices.

func (SessionRemoteError) Error 

func (e SessionRemoteError) Error() string

type SessionService 

type SessionService interface {
	Attach(ctx context.Context, request spec.SessionAttachRequest) (spec.SessionAttachResult, error)
}

SessionService is the transport-neutral attach surface implemented by sessionlog.Coordinator.

type SubagentExecutorConfig 

type SubagentExecutorConfig struct {
	Runner  subagent.AssignedRunner
	Catalog subagent.Catalog
	// Resolver replaces the static Runner/Catalog pair for multi-workspace
	// hosts. It is invoked with the validated logical workspace and the typed
	// assignment authority installed on ctx.
	Resolver             subagent.AssignedExecutionResolver
	ExecutionScopeBinder AssignedExecutionScopeBinder
	MaxConcurrency       int
	Timeout              time.Duration
}

SubagentExecutorConfig configures the opt-in consumer for externally assigned subagent tasks. Catalog may be nil when only generic (non-catalog) subagents are accepted.

type SubagentTaskBackend 

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

SubagentTaskBackend maps the neutral child execution contract to real Aether tasks. Self-assigned tasks execute in-process; opt-in targeted tasks are executed by another connected harness. When the triggering turn is an active Aether task, ParentTaskID asks Aether to validate the caller's ownership and persist native task hierarchy.

func NewSubagentTaskBackend 

func NewSubagentTaskBackend(tasks TaskOperations, workspace, namespace string, timeout time.Duration) (*SubagentTaskBackend, error)

func NewSubagentTaskBackendWithConfig 

func NewSubagentTaskBackendWithConfig(tasks TaskOperations, cfg SubagentTaskBackendConfig) (*SubagentTaskBackend, error)

func (*SubagentTaskBackend) Admit 

func (b *SubagentTaskBackend) Admit(ctx context.Context, admission subagent.TaskAdmission) (string, error)

func (*SubagentTaskBackend) Await 

func (b *SubagentTaskBackend) Await(ctx context.Context, taskID string, observe func(subagent.TaskRecovery)) (subagent.TaskRecovery, error)

Await polls authoritative task state without mutating it. The parent uses this only for targeted execution and therefore never claims or finishes the child task itself.

func (*SubagentTaskBackend) ExecutesExternally 

func (b *SubagentTaskBackend) ExecutesExternally() bool

ExecutesExternally reports whether another Aether agent owns child execution.

func (*SubagentTaskBackend) Finish 

func (b *SubagentTaskBackend) Finish(ctx context.Context, taskID string, outcome subagent.TaskOutcome, reason string) error

func (*SubagentTaskBackend) Recover 

func (b *SubagentTaskBackend) Recover(ctx context.Context, taskID string) (subagent.TaskRecovery, error)

func (*SubagentTaskBackend) Start 

func (b *SubagentTaskBackend) Start(ctx context.Context, taskID string) error

type SubagentTaskBackendConfig 

type SubagentTaskBackendConfig struct {
	RoutingWorkspace string
	Namespace        string
	Timeout          time.Duration
	PollInterval     time.Duration
	TargetAgentID    string
}

SubagentTaskBackendConfig selects local self-assigned execution (the zero TargetAgentID default) or opt-in targeted execution by another connected Aether agent.

type TaskAssignmentRouter 

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

TaskAssignmentRouter lets independent OSS execution features share the Aether SDK's single OnTaskAssignment callback. Handlers must ignore task types they do not own.

func NewTaskAssignmentRouter 

func NewTaskAssignmentRouter() *TaskAssignmentRouter

func (*TaskAssignmentRouter) HandleAssignment 

func (r *TaskAssignmentRouter) HandleAssignment(ctx context.Context, assignment *sdk.TaskAssignment) error

func (*TaskAssignmentRouter) Register 

func (r *TaskAssignmentRouter) Register(handler sdk.TaskAssignmentHandler)

type TaskOperations 

type TaskOperations interface {
	CreateTaskSync(ctx context.Context, taskType, workspace string, opts sdk.CreateTaskOptions, timeout time.Duration) (*sdk.CreateTaskResponse, error)
	ClaimTask(ctx context.Context, taskID string, timeout time.Duration) (*sdk.TaskOperationResponse, error)
	CompleteTask(ctx context.Context, taskID string, timeout time.Duration) (*sdk.TaskOperationResponse, error)
	FailTask(ctx context.Context, taskID, reason string, timeout time.Duration) (*sdk.TaskOperationResponse, error)
	CancelTask(ctx context.Context, taskID, reason string, timeout time.Duration) (*sdk.TaskOperationResponse, error)
	GetTask(ctx context.Context, taskID string, timeout time.Duration) (*sdk.TaskQueryResponse, error)
}

TaskOperations is the narrow Aether SDK surface required by the neutral subagent task adapter. It is exported so distributions already wrapping an AgentClient can reuse the same lifecycle semantics without forking them.

type ToolCallAuthorizationProvider 

type ToolCallAuthorizationProvider interface {
	AuthorizationForClientTool(
		ctx context.Context,
		access workspacepkg.ExecutionBindingAuthorizationRequest,
		toolName string,
	) (*pb.AuthorizationContext, error)
}

ToolCallAuthorizationProvider mints or selects the Aether OBO grant used when an agent calls a client tool for another principal. nil is direct mode, which is sufficient for OSS's strict originating-window policy.

type WorkerToolAccessAuthorizer 

type WorkerToolAccessAuthorizer interface {
	AuthorizeWorkerTool(ctx context.Context, request WorkerToolAccessRequest) error
}

type WorkerToolAccessRequest 

type WorkerToolAccessRequest struct {
	Binding   spec.ExecutionBinding
	Policy    workspacepkg.ExecutionViewPolicy
	ToolName  string
	CallID    string
	Address   spec.MessageAddress
	Authority tools.MemoryAuthority
}

type WorkerToolHost 

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

WorkerToolHost exposes only exact bindings registered by this worker. The caller's Aether assignment and ACL are the authority; the binding is a resource selector and never a bearer grant.

func NewWorkerToolHost 

func NewWorkerToolHost(ctx context.Context, cfg WorkerToolHostConfig) (*WorkerToolHost, error)

func (*WorkerToolHost) BindScheduledExecutionScope 

func (h *WorkerToolHost) BindScheduledExecutionScope(
	ctx context.Context,
	binding spec.ExecutionBinding,
	policy ScheduledViewPolicy,
) (context.Context, error)

BindScheduledExecutionScope validates one pinned worker view and installs its exact local tool delegate on ctx. Embedding distributions use this when they reuse the OSS scheduled executor with their own Aether transport.

func (*WorkerToolHost) ExecutionBindingForDirectory 

func (h *WorkerToolHost) ExecutionBindingForDirectory(ctx context.Context, dir string) (spec.ExecutionBinding, error)

func (*WorkerToolHost) Renew 

func (h *WorkerToolHost) Renew(ctx context.Context) error

func (*WorkerToolHost) ScheduledExecutionBindingForDirectory 

func (h *WorkerToolHost) ScheduledExecutionBindingForDirectory(
	ctx context.Context,
	dir string,
	allowMutable bool,
	allowDirty bool,
) (spec.ExecutionBinding, error)

func (*WorkerToolHost) SetViewPublisher 

func (h *WorkerToolHost) SetViewPublisher(ctx context.Context, publisher workspacepkg.ViewPublisher) error

SetViewPublisher publishes existing local views and enables future lease renewal. See workspace.ViewRegistry.SetPublisher for two-phase transports.

type WorkerToolHostConfig 

type WorkerToolHostConfig struct {
	WorkspaceID   string
	WorkspaceRoot string
	StateDir      string
	ToolHostID    string
	Publisher     workspacepkg.ViewPublisher
	Python        string
	Timeout       time.Duration
	MaxOutput     int
	// AccessAuthorizer is an enterprise resource-level policy seam. Aether ACLs
	// still decide who may create/route the task; this policy decides whether
	// that task principal may use this worker's concrete workspace view.
	AccessAuthorizer WorkerToolAccessAuthorizer
}

WorkerToolHostConfig configures tools rooted on the durable Aether worker.

type WorkspaceHistoryProjection 

type WorkspaceHistoryProjection interface {
	HistoryProjection
	LoadWorkspaceHistory(ctx context.Context, workspaceID, threadID string) ([]protocol.ChatMessage, error)
	SaveWorkspaceHistory(ctx context.Context, workspaceID, threadID string, messages []protocol.ChatMessage) error
}

WorkspaceHistoryProjection is the explicit multi-workspace projection contract. Dynamic clients use it instead of adapting unscoped methods.

type WorkspaceResolver 

type WorkspaceResolver interface {
	ResolveWorkspace(ctx context.Context, requested string) (string, error)
}

WorkspaceResolver applies the host's application-level default and visibility policy independently from Aether's transport workspace.

Source Files

View all Source files

Directories

Path Synopsis
Package scheduleconfig owns the strict, credential-free YAML declaration shape shared by OSS and embedded Sahara distributions.
 Package scheduleconfig owns the strict, credential-free YAML declaration shape shared by OSS and embedded Sahara distributions.

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.