workerws

type AuthenticateFunc ¶

type AuthenticateFunc func(context.Context, *http.Request) (Identity, error)

func (function AuthenticateFunc) AuthenticateWorker(
	ctx context.Context,
	request *http.Request,
) (Identity, error)

type Authenticator ¶

type Authenticator interface {
	AuthenticateWorker(context.Context, *http.Request) (Identity, error)
}

Authenticator belongs to this HTTP adapter, not the Agent domain. A host can inspect mTLS, cookies, bearer tokens, or its own session and return canonical authority/worker identity. Implementations must not trust identity fields in a WebSocket message body. The implementation is borrowed until the transport runtime reaches Done, called concurrently, and must honor context cancellation. The request is borrowed for the call and must not be retained or mutated. Return a framework Fault with CodeUnauthenticated/CodeForbidden and RetryNever only for a proved terminal credential decision. Temporary provider failures use CodeUnavailable/RetryBackoff; unclassified errors fail closed as HTTP 503 so a durable worker keeps reconnecting instead of treating the token as revoked.

type Client ¶

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

Client is a durable remote Agent worker runtime. Assignments and Rejections are replayed from Journal after process restart until explicitly confirmed.

func NewClient(ctx context.Context, config ClientConfig) (*Client, error)

func (client *Client) Assignments() <-chan agent.WorkerAssignmentDelivery

func (client *Client) ConfirmAssignment(ctx context.Context, delivery agent.WorkerDeliveryID) error

func (client *Client) ConfirmRejection(ctx context.Context, delivery agent.WorkerDeliveryID) error

func (client *Client) Done() <-chan struct{}

func (client *Client) Err() error

func (client *Client) Rejections() <-chan RejectedEvent

func (client *Client) SendEvent(ctx context.Context, delivery agent.WorkerEventDelivery) error

func (client *Client) Shutdown(ctx context.Context) error

type ClientConfig ¶

type ClientConfig struct {
	URL       string
	Gateway   GatewayID
	Authority agent.AuthorityID
	Worker    agent.WorkerID
	Journal   framework.Resource[Journal]

	HTTPHeader     http.Header
	HeaderProvider HeaderProvider
	HTTPClient     *http.Client

	ConnectTimeout      time.Duration
	WriteTimeout        time.Duration
	ReadLimit           int64
	ReconnectMinDelay   time.Duration
	ReconnectMaxDelay   time.Duration
	ReconnectResetAfter time.Duration
	ReleaseTimeout      time.Duration
	NotificationBuffer  int
}

ClientConfig configures the official remote Agent worker. Journal is the only durability boundary and must implement JournalContractMajor exactly. URL must use ws or wss and must not contain userinfo or a fragment.

type Config ¶

type Config struct {
	GatewayID      GatewayID
	Authenticator  Authenticator
	ReadLimit      int64
	WriteTimeout   time.Duration
	PingInterval   time.Duration
	PingTimeout    time.Duration
	OriginPatterns []string
}

Config controls one official WebSocket adapter. The embedding application owns the HTTP server/listener and mounts Transport as a handler.

type GatewayID ¶

type GatewayID string

GatewayID is the durable identity of one gateway consistency domain. It is not a URL: changing DNS, ports, or transports does not change GatewayID.

func (identity GatewayID) Validate() error

type HeaderProvider ¶

type HeaderProvider interface {
	WorkerHeaders(context.Context) (http.Header, error)
}

HeaderProvider supplies per-dial authentication headers. It may refresh a short-lived credential on every reconnect. Returned maps are cloned before use and SessionHeader is always overwritten by Client. The provider is borrowed until Client.Done closes, is called by one connection loop, and must honor context cancellation without retaining the returned Header.

type HeaderProviderFunc ¶

type HeaderProviderFunc func(context.Context) (http.Header, error)

func (function HeaderProviderFunc) WorkerHeaders(ctx context.Context) (http.Header, error)

type Identity ¶

type Identity struct {
	Authority agent.AuthorityID
	Worker    agent.WorkerID
}

Identity is the durable worker identity returned by Authenticator. Session identity comes from the transport handshake and is deliberately absent.

type Journal ¶

type Journal interface {
	Description() JournalDescription
	// Bind is an idempotent durability operation. The first call permanently
	// records binding; an exact call succeeds and any different value returns
	// ErrJournalConflict. All other operations fail closed while unbound.
	Bind(context.Context, JournalBinding) error

	PutAssignment(context.Context, JournalAssignment) (JournalEntryState, error)
	ConfirmAssignment(context.Context, agent.WorkerDeliveryID) error
	ListAssignments(context.Context, JournalSequence, int) ([]JournalAssignment, error)

	PutEvent(context.Context, JournalEvent) (JournalEntryState, error)
	ListEvents(context.Context, JournalSequence, int) ([]JournalEvent, error)
	SettleEvent(context.Context, agent.WorkerEventReceipt, JournalDigest, JournalDigest) error

	ListRejections(context.Context, JournalSequence, int) ([]JournalRejection, error)
	ConfirmRejection(context.Context, agent.WorkerDeliveryID) error
}

Journal is the persistence port borrowed or owned by Client through a framework.Resource. It deliberately has no Close method.

Implementations take ownership of inputs before a successful return and return independent copies from List methods; mutable byte slices must never alias adapter storage or another caller. All operations other than Description and Bind fail closed until a durable binding exists.

Every mutating call is one atomic, strictly serializable transaction. Calls are not implicitly retried by the implementation. A successful call is durable before it returns. ErrJournalCommitUnknown means the caller must reconcile by repeating the exact operation; it must never assume success or failure.

PutAssignment and PutEvent allocate a non-zero monotonically increasing sequence for a new ID. An exact existing digest returns its current state and does not allocate a sequence. A different digest for any pending or settled ID returns ErrJournalConflict. ConfirmAssignment replaces the pending payload with a compact settled tombstone atomically.

SettleEvent validates the pending event identity, removes it from the outbox, and writes a compact settled tombstone atomically. For a NACK it atomically moves the full event and receipt into the rejection inbox; ConfirmRejection later compacts both payloads. Exact repeated receipt settlement is a no-op; a different event or receipt digest conflicts.

List methods return records with Sequence > after in strictly increasing order, up to limit. Callers may page without holding a transaction open.

type JournalAssignment ¶

type JournalAssignment struct {
	Sequence JournalSequence
	Digest   JournalDigest
	Delivery agent.WorkerAssignmentDelivery
}

type JournalBinding ¶

type JournalBinding struct {
	Gateway   GatewayID
	Authority agent.AuthorityID
	Worker    agent.WorkerID
}

JournalBinding prevents a durable outbox from being replayed into another gateway, authority, or worker after a configuration or URL change.

func (binding JournalBinding) Validate() error

type JournalDescription ¶

type JournalDescription struct {
	Contract framework.Contract
	Provider string
	Version  string
}

JournalDescription is static, non-secret adapter metadata. Provider and Version identify the implementation, not a database or tenant.

func (description JournalDescription) Validate() error

type JournalDigest ¶

type JournalDigest string

JournalDigest is a lowercase SHA-256 digest calculated by the official client over the canonical JSON representation of a public delivery/receipt. Journal implementations preserve it verbatim; they do not reinterpret the payload or choose their own equivalence relation.

func (digest JournalDigest) Validate() error

type JournalEntryState ¶

type JournalEntryState string

JournalEntryState tells PutAssignment/PutEvent whether the exact delivery is still pending or has a durable settlement tombstone. A settled exact replay is never presented or sent again. Reusing its ID with a different digest is a conflict even after the payload has been compacted.

const (
	JournalEntryPending JournalEntryState = "pending"
	JournalEntrySettled JournalEntryState = "settled"
)

type JournalError ¶

type JournalError struct {
	Operation string
	Delivery  agent.WorkerDeliveryID
	Cause     error
}

JournalError attaches an operation and delivery identity while preserving errors.Is/errors.As classification through Cause.

func (failure *JournalError) Error() string

func (failure *JournalError) Unwrap() error

type JournalEvent ¶

type JournalEvent struct {
	Sequence JournalSequence
	Digest   JournalDigest
	Delivery agent.WorkerEventDelivery
}

type JournalRejection ¶

type JournalRejection struct {
	Sequence      JournalSequence
	EventDigest   JournalDigest
	ReceiptDigest JournalDigest
	RejectedEvent
}

JournalRejection is the durable application-facing NACK inbox. Confirming it removes both payloads but retains their compact digest tombstone.

type JournalSequence ¶

type JournalSequence uint64

JournalSequence is allocated monotonically by a Journal. It establishes FIFO presentation within each pending collection and is never reused.

type RejectedEvent ¶

type RejectedEvent struct {
	Delivery agent.WorkerEventDelivery
	Receipt  agent.WorkerEventReceipt
}

RejectedEvent is the application-facing NACK, including the exact reply or command that was rejected so it can be corrected after a process restart.

type Transport ¶

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

Transport is both a WorkerTransport provider and the handler mounted by the host. Start binds exactly one running endpoint; shutdown is performed on the returned runtime. A fresh Transport value is required for a later run.

func New(config Config) (*Transport, error)

func (*Transport) Description() agent.WorkerTransportDescription

func (transport *Transport) ServeHTTP(response http.ResponseWriter, request *http.Request)

func (transport *Transport) Start(
	ctx context.Context,
	endpoint agent.WorkerEndpoint,
) (agent.WorkerTransportRuntime, error)