distributed

package
v1.0.8 Latest Latest
Warning

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

 Go to latest Published: Aug 13, 2026 License: GPL-3.0 Imports: 11 Imported by: 0

Documentation

Overview

Package distributed defines transport- and placement-neutral contracts for work that crosses a process or machine boundary. It never redefines message, permission, tool, task, or transcript semantics.

Index

Constants

View Source 
const DefaultGateCapacity = 100_000
View Source 
const DefaultTransportCloseTimeout = 3 * time.Second

Variables

View Source 
var (
	ErrControlNotFound = errors.New("control request not found")
	ErrControlTerminal = errors.New("control request is already terminal")
)
View Source 
var (
	ErrGateActive      = errors.New("history flush gate is already active")
	ErrGateInactive    = errors.New("history flush gate is inactive")
	ErrGateClosed      = errors.New("history flush gate is closed")
	ErrSendNotAccepted = errors.New("transport did not accept outbound event")
	ErrGateCapacity    = errors.New("history flush gate capacity exceeded")
)
View Source 
var (
	ErrStaleEpoch       = errors.New("stale worker epoch")
	ErrEpochNotAdvanced = errors.New("worker epoch did not advance")
	ErrInvalidSequence  = errors.New("invalid sequence")
)
View Source 
var (
	ErrRecoveryClosed = errors.New("reconnect coordinator is closed")
	ErrRecovering     = errors.New("transport is recovering")
)
View Source 
var ErrInvalidTransition = errors.New("invalid lifecycle transition")

Functions

func ValidateOpaqueID 

func ValidateOpaqueID(kind, value string) error

ValidateOpaqueID rejects empty, oversized, non-UTF-8, and control-bearing identifiers. It deliberately does not treat one ID class as another.

func ValidatePathIdentifier 

func ValidatePathIdentifier(kind, value string) error

ValidatePathIdentifier applies the stricter grammar required before an opaque service identifier is interpolated into a URL path.

Types

type Acceptance 

type Acceptance struct {
	Accepted        bool   `json:"accepted"`
	QueueIdentity   string `json:"queue_identity,omitempty"`
	RemoteAckID     string `json:"remote_ack_id,omitempty"`
	DurabilityKnown bool   `json:"durability_known"`
}

Acceptance is local writer evidence only. It does not imply a remote ACK or durable processing.

type BridgeInstanceID 

type BridgeInstanceID string

type CloseEvidence 

type CloseEvidence struct {
	Dropped               int  `json:"dropped"`
	RemoteDurabilityKnown bool `json:"remote_durability_known"`
}

CloseEvidence reports only loss still observable during an orderly close. Abrupt process death has no exact in-memory count.

type ConnectAttempt 

type ConnectAttempt func(context.Context, ResumePoint) (Transport, Epoch, error)

type ControlKey 

type ControlKey struct {
	Request    ControlRequestID `json:"request_id"`
	Session    RemoteSessionID  `json:"remote_session_id"`
	Generation Generation       `json:"surface_generation"`
	Epoch      Epoch            `json:"worker_epoch"`
}

func (ControlKey) Validate 

func (k ControlKey) Validate() error

type ControlRecord 

type ControlRecord struct {
	Key       ControlKey      `json:"key"`
	ToolUseID ToolUseID       `json:"tool_use_id,omitempty"`
	Subtype   string          `json:"subtype"`
	State     ControlState    `json:"state"`
	Value     json.RawMessage `json:"value,omitempty"`
	Message   string          `json:"message,omitempty"`
	UpdatedAt time.Time       `json:"updated_at"`
}

ControlRecord is process-local correlation evidence. It deliberately uses a generation fence instead of accepting unscoped orphan permission responses.

type ControlRegistry 

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

func NewControlRegistry 

func NewControlRegistry() *ControlRegistry

func (*ControlRegistry) Accept 

func (r *ControlRegistry) Accept(key ControlKey, subtype string, toolUseID ToolUseID) error

func (*ControlRegistry) Advance 

func (r *ControlRegistry) Advance(key ControlKey, state ControlState) error

func (*ControlRegistry) Cancel 

func (r *ControlRegistry) Cancel(key ControlKey, reason string) error

func (*ControlRegistry) CancelAll 

func (r *ControlRegistry) CancelAll(reason string) []ControlRecord

CancelAll settles every known nonterminal waiter during disconnect or teardown and returns deterministic terminal evidence.

func (*ControlRegistry) Get 

func (r *ControlRegistry) Get(key ControlKey) (ControlRecord, bool)

func (*ControlRegistry) Resolve 

func (r *ControlRegistry) Resolve(key ControlKey, state ControlState, value json.RawMessage, message string) error

type ControlRequestID 

type ControlRequestID string

type ControlState 

type ControlState string
const (
	ControlReceived      ControlState = "received"
	ControlValidating    ControlState = "validating"
	ControlAwaitingLocal ControlState = "awaiting_local_action"
	ControlResponding    ControlState = "responding"
	ControlSucceeded     ControlState = "succeeded"
	ControlDenied        ControlState = "denied"
	ControlErrored       ControlState = "errored"
	ControlCancelled     ControlState = "cancelled"
	ControlSuperseded    ControlState = "superseded"
)

func (ControlState) Terminal 

func (s ControlState) Terminal() bool

type Cursor 

type Cursor struct {
	Sequence Sequence `json:"observed_sequence"`
}

Cursor is an observed transport position, not proof of dispatch, persistence, processing, or acknowledgement.

func (*Cursor) Observe 

func (c *Cursor) Observe(sequence Sequence) (Observation, error)

Observe advances only to a greater positive sequence.

type Deduper 

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

Deduper is a bounded process-local FIFO membership ring. Eviction means absence no longer proves novelty; it is never durable acknowledgement.

func NewDeduper 

func NewDeduper(capacity int) (*Deduper, error)

func (*Deduper) Len 

func (d *Deduper) Len() int

func (*Deduper) SeenOrAdd 

func (d *Deduper) SeenOrAdd(id MessageID) (bool, error)

SeenOrAdd reports an in-window duplicate without refreshing its age.

type DeliveryEvidence 

type DeliveryEvidence struct {
	ID               DeliveryID    `json:"delivery_id"`
	Message          MessageID     `json:"message_id"`
	Sequence         Sequence      `json:"observed_sequence,omitempty"`
	State            DeliveryState `json:"state"`
	TransportAckID   string        `json:"transport_ack_id,omitempty"`
	TransportAckedAt *time.Time    `json:"transport_acked_at,omitempty"`
	UpdatedAt        time.Time     `json:"updated_at"`
	Failure          string        `json:"failure,omitempty"`
}

DeliveryEvidence keeps acknowledgement independent of delivery state and observed cursor position.

type DeliveryID 

type DeliveryID string

type DeliveryState 

type DeliveryState string

DeliveryState describes adapter handling only. Processed never claims model completion, transcript durability, or semantic success. 

const (
	DeliveryObserved   DeliveryState = "observed"
	DeliveryReceived   DeliveryState = "received"
	DeliveryProcessing DeliveryState = "processing"
	DeliveryProcessed  DeliveryState = "processed"
	DeliveryFailed     DeliveryState = "failed"
	DeliveryCancelled  DeliveryState = "cancelled"
)

func (DeliveryState) Terminal 

func (s DeliveryState) Terminal() bool

type DeliveryTracker 

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

func NewDeliveryTracker 

func NewDeliveryTracker() *DeliveryTracker

func (*DeliveryTracker) Acknowledge 

func (t *DeliveryTracker) Acknowledge(id DeliveryID, ackID string) error

Acknowledge records explicit remote transport evidence. Neither dedupe nor a processed callback can call this implicitly.

func (*DeliveryTracker) Get 

func (t *DeliveryTracker) Get(id DeliveryID) (DeliveryEvidence, bool)

func (*DeliveryTracker) Observe 

func (t *DeliveryTracker) Observe(id DeliveryID, message MessageID, sequence Sequence) error

func (*DeliveryTracker) Transition 

func (t *DeliveryTracker) Transition(id DeliveryID, state DeliveryState, failure string) error

type EnvironmentID 

type EnvironmentID string

type Epoch 

type Epoch uint64

type EpochFence 

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

EpochFence ensures stale workers cannot continue writing after replacement.

func NewEpochFence 

func NewEpochFence(initial Epoch) (*EpochFence, error)

func (*EpochFence) Advance 

func (f *EpochFence) Advance(epoch Epoch) error

func (*EpochFence) Check 

func (f *EpochFence) Check(epoch Epoch) error

func (*EpochFence) Current 

func (f *EpochFence) Current() Epoch

type Factory 

type Factory func(context.Context, TransportConfig) (Transport, error)

type FlushGate 

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

FlushGate guarantees initial history precedes live events. sendMu serializes begin/direct-send/drain boundaries; mu protects only the process-local queue.

func NewFlushGate 

func NewFlushGate() *FlushGate

func NewFlushGateWithCapacity 

func NewFlushGateWithCapacity(capacity int) (*FlushGate, error)

func (*FlushGate) BeginHistory 

func (g *FlushGate) BeginHistory() error

BeginHistory must run before the caller freezes its history snapshot. Live submissions are queued from this point onward.

func (*FlushGate) Deactivate 

func (g *FlushGate) Deactivate()

Deactivate retains queued events for a replacement transport.

func (*FlushGate) Drain 

func (g *FlushGate) Drain(ctx context.Context, sender Sender) error

Drain sends retained items serially. A failed or unaccepted head stays at the front for explicit same-process retry; later messages cannot overtake it.

func (*FlushGate) Drop 

func (g *FlushGate) Drop(closeGate bool) int

Drop is reserved for final teardown or an explicitly unrecoverable path and returns the exact still-observable loss count.

func (*FlushGate) FlushHistory 

func (g *FlushGate) FlushHistory(ctx context.Context, snapshot func() []OutboundEvent, sender Sender) error

FlushHistory performs the required start-snapshot-install-drain sequence.

func (*FlushGate) InstallHistory 

func (g *FlushGate) InstallHistory(history []OutboundEvent) error

InstallHistory prepends the frozen history to live messages collected since BeginHistory, preserving history-before-live order.

func (*FlushGate) Pending 

func (g *FlushGate) Pending() []OutboundEvent

func (*FlushGate) State 

func (g *FlushGate) State() GateState

func (*FlushGate) Submit 

func (g *FlushGate) Submit(ctx context.Context, event OutboundEvent, sender Sender) (SubmitResult, error)

Submit queues while a history/recovery gate is active and sends directly otherwise. BeginHistory cannot overtake a direct send already in progress.

type GateState 

type GateState string
const (
	GateInactive    GateState = "inactive"
	GateCollecting  GateState = "collecting_history"
	GateReady       GateState = "ready"
	GateDraining    GateState = "draining"
	GateDeactivated GateState = "deactivated"
	GateClosed      GateState = "closed"
)

type Generation 

type Generation uint64

type IdentityTuple 

type IdentityTuple struct {
	BridgeInstance BridgeInstanceID `json:"bridge_instance_id,omitempty"`
	Environment    EnvironmentID    `json:"environment_id,omitempty"`
	Work           WorkID           `json:"work_id,omitempty"`
	Session        RemoteSessionID  `json:"remote_session_id"`
	Epoch          Epoch            `json:"worker_epoch"`
	Generation     Generation       `json:"surface_generation"`
}

IdentityTuple names every correlation scope used by a live remote surface. Fields remain distinct even if their string spellings happen to match.

func (IdentityTuple) Validate 

func (i IdentityTuple) Validate() error

Validate checks the minimum identity needed by a control-capable session.

type Lifecycle 

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

Lifecycle is a synchronized transport state machine. It records state, not whether any semantic event succeeded.

func NewLifecycle 

func NewLifecycle() *Lifecycle

func (*Lifecycle) History 

func (l *Lifecycle) History() []Transition

func (*Lifecycle) State 

func (l *Lifecycle) State() TransportState

func (*Lifecycle) Transition 

func (l *Lifecycle) Transition(to TransportState, reason string) error

type MessageID 

type MessageID string

type Observation 

type Observation struct {
	Previous  Sequence `json:"previous"`
	Current   Sequence `json:"current"`
	Advanced  bool     `json:"advanced"`
	Duplicate bool     `json:"duplicate"`
	Gap       bool     `json:"gap"`
}

Observation describes how a parsed frame ID changed the cursor. Callers perform this before decoding the payload so malformed payloads cannot make a reconnect loop repeatedly observe the same frame.

type OutboundEvent 

type OutboundEvent struct {
	MessageID MessageID `json:"message_id"`
	Type      string    `json:"type"`
	Payload   []byte    `json:"payload"`
}

OutboundEvent is a selected bridge projection, not an authoritative local transcript message. Payload is already normalized and credential-free.

type ReconnectCoordinator 

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

ReconnectCoordinator serializes proactive refresh and reactive reconnect so one live generation advances its epoch at most once. It owns the installed transport and closes a late replacement when teardown wins.

func NewReconnectCoordinator 

func NewReconnectCoordinator(parent context.Context, session RemoteSessionID, epoch Epoch, cursor Cursor, current Transport) (*ReconnectCoordinator, error)

func (*ReconnectCoordinator) Close 

func (r *ReconnectCoordinator) Close(ctx context.Context) (CloseEvidence, error)

Close fences future writes, cancels recovery, closes the installed transport, and classifies any exact orderly drop evidence returned by that transport.

func (*ReconnectCoordinator) ObserveSequence 

func (r *ReconnectCoordinator) ObserveSequence(sequence Sequence) (Observation, error)

ObserveSequence updates the reconnect high-water before payload admission.

func (*ReconnectCoordinator) Recover 

func (r *ReconnectCoordinator) Recover(ctx context.Context, reason string, attempt ConnectAttempt) (Transport, Epoch, error)

Recover joins an existing recovery or starts one owned by the coordinator, not by the first waiter's cancellation context.

func (*ReconnectCoordinator) Send 

func (r *ReconnectCoordinator) Send(ctx context.Context, epoch Epoch, event OutboundEvent) (Acceptance, error)

Send refuses direct writes during replacement and checks the active epoch.

func (*ReconnectCoordinator) State 

func (r *ReconnectCoordinator) State() TransportState

type RemoteSessionID 

type RemoteSessionID string

type ResumePoint 

type ResumePoint struct {
	Session RemoteSessionID `json:"remote_session_id"`
	Epoch   Epoch           `json:"previous_epoch"`
	Cursor  Cursor          `json:"cursor"`
}

ResumePoint carries only evidence actually retained by this process. Cursor is observed high-water, not a processing or durability checkpoint.

type Sender 

type Sender interface {
	Send(context.Context, OutboundEvent) (Acceptance, error)
}

Sender is normally a serial transport writer. Its receipt must identify local acceptance; a nil error alone is intentionally insufficient.

type Sequence 

type Sequence uint64

type SubmitResult 

type SubmitResult struct {
	Queued     bool       `json:"queued"`
	Acceptance Acceptance `json:"acceptance"`
}

type ToolUseID 

type ToolUseID string

type Transition 

type Transition struct {
	From   TransportState `json:"from"`
	To     TransportState `json:"to"`
	At     time.Time      `json:"at"`
	Reason string         `json:"reason,omitempty"`
}

type Transport 

type Transport interface {
	Kind() TransportKind
	Send(context.Context, OutboundEvent) (Acceptance, error)
	Close(context.Context) (CloseEvidence, error)
}

type TransportConfig 

type TransportConfig struct {
	Kind     TransportKind `json:"kind"`
	Included bool          `json:"included"`
	Enabled  bool          `json:"enabled"`
	Endpoint string        `json:"endpoint,omitempty"`
}

type TransportKind 

type TransportKind string
const (
	TransportHybrid    TransportKind = "hybrid"
	TransportCCR       TransportKind = "ccr_sse"
	TransportWebSocket TransportKind = "websocket"
	TransportDirect    TransportKind = "direct_connect"
	TransportSSH       TransportKind = "ssh"
)

type TransportRegistry 

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

TransportRegistry has no implicit network implementation. A configured endpoint still fails explicitly until the owning build registers a factory.

func NewTransportRegistry 

func NewTransportRegistry() *TransportRegistry

func (*TransportRegistry) Build 

func (r *TransportRegistry) Build(ctx context.Context, config TransportConfig) (Transport, error)

func (*TransportRegistry) Register 

func (r *TransportRegistry) Register(kind TransportKind, factory Factory) error

type TransportState 

type TransportState string
const (
	TransportNew           TransportState = "new"
	TransportConnecting    TransportState = "connecting"
	TransportConnected     TransportState = "connected"
	TransportReconnectWait TransportState = "reconnect_wait"
	TransportReplacing     TransportState = "replacing"
	TransportDraining      TransportState = "draining"
	TransportClosed        TransportState = "closed"
	TransportFailed        TransportState = "failed"
)

func (TransportState) Terminal 

func (s TransportState) Terminal() bool

type UnavailableError 

type UnavailableError struct {
	Kind   TransportKind
	State  UnavailableState
	Reason string
}

func (*UnavailableError) Error 

func (e *UnavailableError) Error() string

type UnavailableState 

type UnavailableState string
const (
	UnavailableBuildExcluded   UnavailableState = "build_excluded"
	UnavailableGateDisabled    UnavailableState = "gate_disabled"
	UnavailableUnconfigured    UnavailableState = "unconfigured"
	UnavailableImplementation  UnavailableState = "implementation_unavailable"
	UnavailableMalformedConfig UnavailableState = "malformed_configuration"
)

type WorkID 

type WorkID string

type WorkLifecycle 

type WorkLifecycle struct {
	ID WorkID
	// contains filtered or unexported fields
}

WorkLifecycle proves that observed work is not owned until acknowledgement.

func NewWorkLifecycle 

func NewWorkLifecycle(id WorkID) (*WorkLifecycle, error)

func (*WorkLifecycle) Owned 

func (w *WorkLifecycle) Owned() bool

func (*WorkLifecycle) State 

func (w *WorkLifecycle) State() WorkState

func (*WorkLifecycle) Transition 

func (w *WorkLifecycle) Transition(to WorkState) error

type WorkState 

type WorkState string
const (
	WorkObserved         WorkState = "observed"
	WorkSecretValidated  WorkState = "secret_validated"
	WorkAcknowledging    WorkState = "acknowledging"
	WorkUnowned          WorkState = "unowned"
	WorkOwned            WorkState = "owned"
	WorkSpawning         WorkState = "spawning"
	WorkRunning          WorkState = "running"
	WorkCompleting       WorkState = "completing"
	WorkFailing          WorkState = "failing"
	WorkInterrupting     WorkState = "interrupting"
	WorkTerminalReported WorkState = "terminal_reported"
	WorkReleased         WorkState = "released"
)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.