daemon

package
v0.1.0-preview.7 Latest Latest
Warning

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

 Go to latest Published: Aug 11, 2026 License: Apache-2.0 Imports: 28 Imported by: 0

Documentation

Overview

Package daemon owns transport-independent daemon definition, client ownership, idempotency, and pending-interaction hosting contracts.

@import { NamedInterface } from "github.com/spice-framework/spice/annotation/modulith" @NamedInterface("daemon")

Index

Constants

View Source 
const (
	// MaximumHealthSources bounds passive readiness work for one snapshot.
	MaximumHealthSources = 32
	// MaximumHealthContributionReasons bounds one source before aggregation.
	MaximumHealthContributionReasons = 16
)

Variables

View Source 
var (
	// ErrOperationExecutor is the secret-safe sentinel committed after an
	// executor error, panic, or invalid result.
	ErrOperationExecutor = errors.New("idempotency operation executor failed")
	// ErrOperationAbandoned identifies an executor's explicit proof that it
	// stopped before every commit boundary and the operation may be retried.
	ErrOperationAbandoned = errors.New("idempotency operation abandoned before commit")
)
View Source 
var (
	// ErrPendingHubClosed rejects work after daemon-root shutdown.
	ErrPendingHubClosed = errors.New("pending interaction hub is closed")
	// ErrRunNotBound rejects interaction work for a run without an active or
	// draining client binding.
	ErrRunNotBound = errors.New("interaction run is not bound to a client")
	// ErrObserverFenced terminates discovery streams owned by a superseded
	// client connection. Pending interactions remain available to reconnect.
	ErrObserverFenced = errors.New("pending interaction observers were fenced")
	// ErrRunAlreadyBound rejects duplicate stable ownership of one run.
	ErrRunAlreadyBound = errors.New("interaction run is already bound")
	// ErrRunBindingCapacity rejects a run lease beyond configured bounds.
	ErrRunBindingCapacity = errors.New("pending run binding capacity exhausted")
	// ErrPendingCapacity rejects a pending interaction beyond configured count
	// or byte bounds.
	ErrPendingCapacity = errors.New("pending interaction capacity exhausted")
	// ErrObserverCapacity rejects a discovery stream beyond configured observer
	// or queue-reservation bounds.
	ErrObserverCapacity = errors.New("pending observer capacity exhausted")
	// ErrInteractionNotPending rejects a response without a matching open call.
	ErrInteractionNotPending = errors.New("interaction is not pending")
)
View Source 
var (
	// ErrRunAuthorityUnavailable reports a local storage or security failure
	// without exposing key material or platform-specific error details.
	ErrRunAuthorityUnavailable = errors.New("run authority is unavailable")
	// ErrRunAuthorityBusy reports that another process owns the stable run lock.
	ErrRunAuthorityBusy = errors.New("run authority run is already owned")
	// ErrRunAuthorityState rejects an illegal or replayed lifecycle transition.
	ErrRunAuthorityState = errors.New("run authority state transition is invalid")
	// ErrRunAuthorityVerification rejects a snapshot not authenticated by the
	// current user's persistent authority and matching suspended run record.
	ErrRunAuthorityVerification = errors.New("run authority verification failed")
	// ErrRunAuthorityUncertain reports an import consumed durably but not
	// activated. It must never be retried automatically.
	ErrRunAuthorityUncertain = errors.New("run authority import is uncertain and must not be retried")
)
View Source 
var (
	// ErrRunHostClosed rejects admission after lifecycle shutdown begins.
	ErrRunHostClosed = errors.New("run host is closed")
	// ErrRunHostCapacity reports that every configured active-run slot is reserved.
	ErrRunHostCapacity = errors.New("run host active capacity is exhausted")
	// ErrHostedRunUnavailable deliberately makes an unknown run and a run owned
	// by another stable client indistinguishable.
	ErrHostedRunUnavailable = errors.New("hosted run is unavailable")
	// ErrRunHostState rejects a known but illegal lifecycle transition.
	ErrRunHostState = errors.New("run host lifecycle transition is invalid")
	// ErrRunHostUncertain reports a durable transition whose outcome cannot be
	// proved. It is safe for clients but must never be retried under a new ID.
	ErrRunHostUncertain = errors.New("run host lifecycle outcome is uncertain")
	// ErrRunHostUnavailable reports a secret-safe dependency or persistence failure.
	ErrRunHostUnavailable = errors.New("run host dependency is unavailable")
)
View Source 
var (
	// ErrStaleSession rejects a client ownership epoch that lost its CAS.
	ErrStaleSession = errors.New("daemon session ownership is stale")
	// ErrSessionStoreClosed rejects work after daemon-root shutdown.
	ErrSessionStoreClosed = errors.New("daemon session store is closed")
	// ErrSessionGateCapacity rejects work that would exceed a bounded
	// per-client mutation/reconnect queue or stream lease set.
	ErrSessionGateCapacity = errors.New("daemon session gate capacity exhausted")
)

Functions

func AbandonOperation 

func AbandonOperation(cause error) error

AbandonOperation classifies a safe caller-supplied cause as definitively pre-commit. It is legal only when no externally visible or durable commit boundary was reached. The fixed Error text never formats cause; errors.Is and errors.As may still inspect the caller-owned cause programmatically. A nil cause is invalid and is treated by Ledger as an ordinary executor failure, preserving the fail-closed uncertain outcome.

func CanonicalDigest 

func CanonicalDigest(value []byte) [32]byte

CanonicalDigest hashes a caller-canonicalized operation request.

Types

type ActiveRun 

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

ActiveRun is an opaque exclusive lease for one active or locally suspended run. Typed snapshot issuance atomically persists SUSPENDED while retaining exclusive ownership; terminal issuance persists a tombstone before releasing the stable lock. The raw authority signer remains package-internal so callers cannot supply parallel snapshot metadata.

func (*ActiveRun) Close 

func (active *ActiveRun) Close() error

Close releases the OS lock without changing durable state. It is the crash equivalent: ACTIVE remains non-importable, while a valid SUSPENDED record becomes eligible for another authority instance to import. An uncertain owner is still released, but its first Close reports ErrRunAuthorityUncertain; repeated Close is idempotent.

func (*ActiveRun) IssueSnapshotEnvelope 

func (active *ActiveRun) IssueSnapshotEnvelope(
	ctx context.Context,
	snapshot agent.Snapshot,
) (*enginev1.SnapshotEnvelope, error)

IssueSnapshotEnvelope validates and deterministically encodes one kernel snapshot, then signs it at the active run's durable lifecycle boundary. A successful signer result wins cancellation observed after the persistence boundary. Uncertain and unavailable durable outcomes are never retried or replaced with context cancellation.

func (*ActiveRun) Resume 

func (active *ActiveRun) Resume(ctx context.Context) error

Resume invalidates the currently signed suspended snapshot, advances the local run generation, and retains exclusive ownership. The host must call Resume while the kernel remains suspended. The host first reserves the kernel boundary with agent.Run.PrepareLocalResume, performs this durable invalidation, and only then commits the prepared kernel resume. It must never expose or continue kernel execution before a successful authority transition.

func (*ActiveRun) RunGeneration 

func (active *ActiveRun) RunGeneration() uint64

RunGeneration returns the local transition generation. This value is not the persistent authority-key generation carried in snapshot envelopes.

func (*ActiveRun) Terminal 

func (active *ActiveRun) Terminal(ctx context.Context, phase TerminalPhase) error

Terminal persists a tombstone before releasing ownership.

type Definition 

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

Definition is one immutable generated daemon definition. The embedded agent definition is the exact value used to start a run; the daemon does not reconstruct model or turn policy from client input.

func NewDefinition 

func NewDefinition(id, revision string, value agent.Definition) (Definition, error)

NewDefinition validates one generated definition identity and its exact kernel definition.

func (Definition) Agent 

func (definition Definition) Agent() agent.Definition

Agent returns the exact immutable kernel definition.

func (Definition) ID 

func (definition Definition) ID() string

ID returns the stable server-owned definition identity.

func (Definition) Revision 

func (definition Definition) Revision() string

Revision returns the exact generated definition revision.

type DefinitionSet 

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

DefinitionSet is an immutable, canonically sorted generated catalog.

func NewDefinitionSet 

func NewDefinitionSet(values []Definition) (DefinitionSet, error)

NewDefinitionSet validates, sorts, and fingerprints a generated catalog.

func (DefinitionSet) Definitions 

func (set DefinitionSet) Definitions() []Definition

Definitions returns a defensive copy in canonical order.

func (DefinitionSet) Resolve 

func (set DefinitionSet) Resolve(id, revision string) (Definition, error)

Resolve returns an exact server-owned definition identity.

func (DefinitionSet) Revision 

func (set DefinitionSet) Revision() string

Revision returns the deterministic catalog fingerprint.

type Delta 

type Delta struct {
	Revision uint64
	Kind     DeltaKind
	Pending  Pending
}

Delta is one revision-contiguous mutation within one stable client.

type DeltaKind 

type DeltaKind string

DeltaKind identifies a pending-interaction lifecycle mutation. 

const (
	// DeltaOpened adds a newly pending request after the complete snapshot.
	DeltaOpened DeltaKind = "opened"
	// DeltaClosed removes a completed or canceled request.
	DeltaClosed DeltaKind = "closed"
)

type EventObservation 

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

EventObservation owns one session stream lease and one immutable replay page. A transport must call Close only after its sender has stopped using the page and tail. Close cancels and joins the internal tail before releasing the reconnect fence, and is safe to call repeatedly.

func (*EventObservation) Close 

func (observation *EventObservation) Close()

Close joins internal event delivery before releasing the session lease.

func (*EventObservation) Context 

func (observation *EventObservation) Context() context.Context

Context is canceled by the caller, reconnect fencing, or daemon shutdown.

func (*EventObservation) Page 

func (observation *EventObservation) Page() event.ReplayPage

Page returns a defensive copy of the captured replay page. Tail, when present, belongs to this observation and remains valid until Close.

type HealthContribution 

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

HealthContribution is one immutable passive readiness observation. Its zero value is a ready contribution.

func NewHealthContribution 

func NewHealthContribution(reasons []HealthReasonCode) (HealthContribution, error)

NewHealthContribution validates, sorts, deduplicates, and defensively copies fixed reason codes. Input is bounded before deduplication so a source cannot make health work proportional to an unbounded caller-controlled slice.

func (HealthContribution) Reasons 

func (contribution HealthContribution) Reasons() []HealthReasonCode

Reasons returns a defensive copy of the canonical fixed reason codes.

func (HealthContribution) Validate 

func (contribution HealthContribution) Validate() error

Validate verifies that a contribution is canonical and bounded.

type HealthReasonCode 

type HealthReasonCode string

HealthReasonCode is a fixed, secret-safe daemon readiness reason. The allowlist is intentionally closed: sources cannot return arbitrary error, path, endpoint, provider, or plugin-controlled text through daemon health. 

const (
	HealthReasonDependencyDegraded    HealthReasonCode = "dependency_degraded"
	HealthReasonDependencyRecovering  HealthReasonCode = "dependency_recovering"
	HealthReasonDependencyUnavailable HealthReasonCode = "dependency_unavailable"
)

type HealthSource 

type HealthSource interface {
	HealthContribution() HealthContribution
}

HealthSource supplies one passive, non-blocking readiness contribution. Implementations must inspect already-owned in-memory state only. RunHost invokes sources without holding its state mutex and contains a source panic as the fixed dependency_unavailable reason.

type InteractionObservation 

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

InteractionObservation owns one session stream lease and either a complete finite snapshot or that snapshot plus a revision-contiguous tail. A transport must call Close only after its sender joins.

func (*InteractionObservation) Close 

func (observation *InteractionObservation) Close()

Close joins internal pending delivery before releasing the session lease.

func (*InteractionObservation) Context 

func (observation *InteractionObservation) Context() context.Context

Context is canceled by the caller, reconnect fencing, or daemon shutdown.

func (*InteractionObservation) Deltas 

func (observation *InteractionObservation) Deltas() <-chan Delta

Deltas returns the live tail, or an already closed stream for a finite snapshot observation.

func (*InteractionObservation) Snapshot 

func (observation *InteractionObservation) Snapshot() PendingSnapshot

Snapshot returns the complete defensive first frame.

func (*InteractionObservation) Tailing 

func (observation *InteractionObservation) Tailing() bool

Tailing reports whether this observation allocated a PendingHub watcher.

func (*InteractionObservation) Wait 

func (observation *InteractionObservation) Wait(ctx context.Context) error

Wait reports tail termination. Finite snapshot observations return nil immediately and still retain their lease until Close.

type Ledger 

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

Ledger is a bounded concurrent idempotency ledger. Entries intentionally survive client reconnect because stable client identity, not epoch, owns the operation namespace.

func NewLedger 

func NewLedger(maximumClients, maximumPerClient int) (*Ledger, error)

NewLedger constructs independently bounded client and per-client operation namespaces, preventing one client from exhausting every other client.

func (*Ledger) Do 

func (ledger *Ledger) Do(
	ctx context.Context,
	clientID, operationID, kind string,
	digest [32]byte,
	execute func(context.Context) (Outcome, error),
) (Outcome, bool, error)

Do assigns one owner to an operation and makes exact duplicates wait with their own context. Executor errors and panics commit the same bounded uncertain result and sentinel error for the owner and later duplicates. An exact AbandonOperation error with no result removes the uncommitted entry; live duplicates compete normally for one new owner without recursion.

type MutationCommitLease 

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

MutationCommitLease is an opaque exclusive commit-boundary lease. Close is nonblocking, idempotent, and safe for concurrent use.

func (*MutationCommitLease) Close 

func (lease *MutationCommitLease) Close()

Close releases the client's commit boundary.

type ObserverExhaustedError 

type ObserverExhaustedError struct {
	LastDelivered uint64
	// contains filtered or unexported fields
}

ObserverExhaustedError reports the exact last revision and queue bound seen by a slow subscriber. Recovery creates a new client-scoped subscription and consumes its mandatory complete snapshot.

func (*ObserverExhaustedError) Error 

func (failure *ObserverExhaustedError) Error() string

func (*ObserverExhaustedError) Limit 

func (failure *ObserverExhaustedError) Limit() uint64

Limit returns the exact configured bound that was exceeded.

func (*ObserverExhaustedError) Observed 

func (failure *ObserverExhaustedError) Observed() uint64

Observed returns the exact refused count or byte size.

func (*ObserverExhaustedError) Resource 

func (failure *ObserverExhaustedError) Resource() string

Resource identifies the exact safe protocol resource that was exhausted.

type Outcome 

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

Outcome is the bounded canonical result persisted by the ledger. Expected business failures are OutcomeFailure values, not executor errors.

func NewOutcome 

func NewOutcome(kind OutcomeKind, payload []byte) (Outcome, error)

NewOutcome validates and defensively copies a canonical result.

func (Outcome) Kind 

func (outcome Outcome) Kind() OutcomeKind

Kind returns the terminal classification.

func (Outcome) Payload 

func (outcome Outcome) Payload() []byte

Payload returns a defensive copy of the canonical result bytes.

type OutcomeKind 

type OutcomeKind string

OutcomeKind is the durable, client-safe terminal classification. 

const (
	// OutcomeSuccess records a successfully committed operation.
	OutcomeSuccess OutcomeKind = "success"
	// OutcomeFailure records an expected, canonical business failure.
	OutcomeFailure OutcomeKind = "failure"
	// OutcomeUncertain records an operation whose safe terminal state is unknown.
	OutcomeUncertain OutcomeKind = "uncertain"
)

type Pending 

type Pending struct {
	Scope   interaction.Scope
	Request interaction.Request
}

Pending is one immutable pending interaction discovery value.

type PendingHub 

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

PendingHub implements interaction.Broker while partitioning discovery by stable client identity. One lock owns routing, revisions, caps, queue reservations, and accounting.

func NewPendingHub 

func NewPendingHub(limits PendingLimits) (*PendingHub, error)

NewPendingHub constructs a bounded, client-partitioned interaction hub.

func (*PendingHub) BindRun 

func (hub *PendingHub) BindRun(clientID string, scope interaction.Scope) (*RunBinding, error)

BindRun exclusively assigns a run to one stable client. The returned lease must be released after the run terminates.

func (*PendingHub) Close 

func (hub *PendingHub) Close()

Close releases pending callers, immediately stops every observer without waiting for clients to read queued deltas, and joins their delivery loops.

func (*PendingHub) FenceObservers 

func (hub *PendingHub) FenceObservers(clientID string) error

FenceObservers immediately terminates a stable client's current discovery streams. It preserves that client's revision and pending calls for a new complete-first subscription after reconnect.

func (*PendingHub) Request 

func (hub *PendingHub) Request(ctx context.Context, scope interaction.Scope, request interaction.Request) (interaction.Response, error)

Request publishes and awaits one interaction in its bound client partition.

func (*PendingHub) Respond 

func (hub *PendingHub) Respond(clientID string, scope interaction.Scope, response interaction.Response) error

Respond completes only an interaction owned by clientID. A wrong client cannot discover or answer another client's request.

func (*PendingHub) Snapshot 

func (hub *PendingHub) Snapshot(clientID string) (PendingSnapshot, error)

Snapshot returns one stable client's complete sorted pending view without allocating an observer or consuming any queue reservation. A client without retained interaction state receives the initial empty revision.

func (*PendingHub) Subscribe 

func (hub *PendingHub) Subscribe(ctx context.Context, clientID string) (*PendingSubscription, error)

Subscribe atomically captures one stable client's complete sorted pending set and registers its tail before releasing the hub lock.

type PendingLimits 

type PendingLimits struct {
	Clients                       int
	Runs                          int
	RunsPerClient                 int
	Pending                       int
	PendingPerClient              int
	PendingBytes                  int
	PendingBytesPerClient         int
	Observers                     int
	ObserversPerClient            int
	ObserverQueueEntries          int
	ObserverQueueBytes            int
	ReservedQueueEntries          int
	ReservedQueueEntriesPerClient int
	ReservedQueueBytes            int
	ReservedQueueBytesPerClient   int
	QueuedEntries                 int
	QueuedEntriesPerClient        int
	QueuedBytes                   int
	QueuedBytesPerClient          int
}

PendingLimits bounds both the whole hub and every stable client partition. A subscription reserves its entire queue budget until it terminates.

func DefaultPendingLimits 

func DefaultPendingLimits() PendingLimits

DefaultPendingLimits returns conservative production defaults. Every observer can retain the largest valid delta, and aggregate actual queue caps cover all capacity reserved when subscriptions are admitted.

type PendingSnapshot 

type PendingSnapshot struct {
	Revision uint64
	Pending  []Pending
}

PendingSnapshot is the mandatory complete first client-scoped view.

type PendingSubscription 

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

PendingSubscription atomically couples a complete client snapshot with a live tail registered at exactly that partition revision.

func (*PendingSubscription) Deltas 

func (subscription *PendingSubscription) Deltas() <-chan Delta

Deltas returns revision-contiguous changes following Snapshot.

func (*PendingSubscription) LastDelivered 

func (subscription *PendingSubscription) LastDelivered() uint64

LastDelivered returns the exact revision most recently sent to the consumer, or the snapshot revision before the first delta.

func (*PendingSubscription) Snapshot 

func (subscription *PendingSubscription) Snapshot() PendingSnapshot

Snapshot returns the complete immutable first frame.

func (*PendingSubscription) Wait 

func (subscription *PendingSubscription) Wait(ctx context.Context) error

Wait reports hub shutdown, subscriber cancellation, reconnect fencing, or typed queue exhaustion.

type RunAuthority 

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

RunAuthority is an opaque per-user persistent run authority. It owns a random scope identity, a distinct HMAC key, signed lifecycle records, and stable per-run OS locks. It is suitable for constructor injection as a Spice singleton and never exposes its key material.

func NewRunAuthority 

func NewRunAuthority(config RunAuthorityConfig) (*RunAuthority, error)

NewRunAuthority opens or creates the current user's persistent authority. The owning application must Close it; generated Spice applications should register Close as singleton cleanup.

func (*RunAuthority) Close 

func (authority *RunAuthority) Close() error

Close prevents new run work and releases the bound authority-directory handle. If a run or import lease remains open, Close returns ErrRunAuthorityBusy; the last lease release completes shutdown.

func (*RunAuthority) GoString 

func (*RunAuthority) GoString() string

func (*RunAuthority) PrepareImport 

func (authority *RunAuthority) PrepareImport(
	ctx context.Context,
	snapshot *enginev1.SnapshotEnvelope,
) (*RunImport, error)

PrepareImport acquires the stable lock and verifies both the keyed envelope and its exact signed SUSPENDED record. No durable state changes yet.

func (*RunAuthority) Start 

func (authority *RunAuthority) Start(ctx context.Context, runID string) (*ActiveRun, error)

Start creates a never-reused run identity at local transition generation one.

func (*RunAuthority) String 

func (*RunAuthority) String() string

type RunAuthorityConfig 

type RunAuthorityConfig struct {
	Directory string
}

RunAuthorityConfig selects the private local authority directory. An empty directory uses the current user's OS configuration directory. The authority key is deliberately unrelated to daemon endpoint authentication tokens.

type RunBinding 

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

RunBinding is the exclusive stable-client ownership lease for one run. Its release stops new requests but deliberately leaves accepted requests respondable until they finish, preventing a run-terminal race.

func (*RunBinding) ClientID 

func (binding *RunBinding) ClientID() string

ClientID returns the stable client identity owning the lease.

func (*RunBinding) Release 

func (binding *RunBinding) Release()

Release prevents new requests. Capacity is reclaimed after every already accepted interaction reaches its own terminal result.

func (*RunBinding) RunID 

func (binding *RunBinding) RunID() string

RunID returns the bound run identity.

func (*RunBinding) WaitReleased 

func (binding *RunBinding) WaitReleased(ctx context.Context) error

WaitReleased waits until Release has stopped admission and all interactions accepted before that point have reached a terminal result.

type RunHost 

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

RunHost is the transport-independent lifecycle owner. It contains no RPC, socket, or client-stream machinery; future transports can expose epoch-bound sessions around these methods without duplicating lifecycle transitions.

func NewRunHost 

func NewRunHost(config RunHostConfig) (*RunHost, error)

NewRunHost validates and takes ownership of all configured lifecycle dependencies. Shutdown closes/drains them in dependency order.

func (*RunHost) Cancel 

func (host *RunHost) Cancel(ctx context.Context, session Session, request client.CancelRequest) (client.CancelResult, error)

Cancel records exactly one cooperative cancellation request per operation.

func (*RunHost) Describe 

func (host *RunHost) Describe(ctx context.Context) (RunHostDescription, error)

Describe reports generated definitions and readiness without requiring a negotiated client session. It does not create or mutate session state.

func (*RunHost) Export 

func (host *RunHost) Export(ctx context.Context, session Session, run client.RunRef) (client.Snapshot, error)

Export returns only a cached, authority-issued safe-boundary envelope.

func (*RunHost) Health 

func (host *RunHost) Health(ctx context.Context, session Session) (client.Health, error)

Health reports only the immutable configured server limits. Callers cannot inject negotiated limits into readiness accounting.

func (*RunHost) Import 

func (host *RunHost) Import(ctx context.Context, session Session, request client.ImportRequest) (client.ImportResult, error)

Import authenticates and consumes a suspended authority snapshot before either authority or kernel execution becomes visible.

func (*RunHost) ReplayEvents 

func (host *RunHost) ReplayEvents(
	ctx context.Context,
	session Session,
	run client.RunRef,
	request event.ReplayRequest,
) (*EventObservation, error)

ReplayEvents acquires one reconnect fence and returns an owned observation for a bounded, gap-free run page and optional atomic tail.

func (*RunHost) Respond 

func (host *RunHost) Respond(ctx context.Context, session Session, request client.RespondRequest) (client.RespondResult, error)

Respond completes an accepted interaction even after its run reaches a terminal boundary; the pending binding owns that drain lifetime.

func (*RunHost) Resume 

func (host *RunHost) Resume(ctx context.Context, session Session, request client.RunMutation) (client.ResumeResult, error)

Resume invalidates the authority snapshot before releasing local execution.

func (*RunHost) Shutdown 

func (host *RunHost) Shutdown(ctx context.Context) error

Shutdown rejects admission, aborts inert candidates, cancels kernel work, drains accepted interactions, joins finalizers, and closes authority last. Cleanup continues even if one caller stops waiting.

func (*RunHost) SnapshotInteractions 

func (host *RunHost) SnapshotInteractions(
	ctx context.Context,
	session Session,
) (*InteractionObservation, error)

SnapshotInteractions returns one finite complete snapshot under a reconnect fence without allocating a PendingHub watcher. The caller releases the fence only after delivering the snapshot by calling Close.

func (*RunHost) Start 

func (host *RunHost) Start(ctx context.Context, session Session, request client.StartRequest) (client.StartResult, error)

Start creates a generated definition run without allowing any kernel work before the corresponding authority record is durably ACTIVE.

func (*RunHost) SubscribeInteractions 

func (host *RunHost) SubscribeInteractions(
	ctx context.Context,
	session Session,
) (*InteractionObservation, error)

SubscribeInteractions returns a complete client-scoped snapshot plus a live revision-contiguous tail under one owned reconnect fence.

func (*RunHost) Suspend 

func (host *RunHost) Suspend(ctx context.Context, session Session, request client.RunMutation) (client.SuspendResult, error)

Suspend waits for a kernel safe boundary without holding the client mutation fence, then commits only the bounded authority issuance under that fence.

type RunHostCapacityError 

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

RunHostCapacityError reports one exact bounded host resource observation. It matches ErrRunHostCapacity while preserving facts needed by protocol recovery and overload diagnostics.

func (*RunHostCapacityError) Error 

func (failure *RunHostCapacityError) Error() string

func (*RunHostCapacityError) Is 

func (failure *RunHostCapacityError) Is(target error) bool

Is makes RunHostCapacityError match ErrRunHostCapacity.

func (*RunHostCapacityError) Limit 

func (failure *RunHostCapacityError) Limit() uint64

Limit returns the configured positive hard limit.

func (*RunHostCapacityError) Observed 

func (failure *RunHostCapacityError) Observed() uint64

Observed returns the rejected resource observation.

func (*RunHostCapacityError) Resource 

func (failure *RunHostCapacityError) Resource() string

Resource returns the bounded host resource.

type RunHostConfig 

type RunHostConfig struct {
	Root              context.Context //nolint:containedctx // transferred as the daemon lifetime root, never as request state.
	Engine            *agent.Engine
	Authority         *RunAuthority
	Sessions          *SessionStore
	Ledger            *Ledger
	Pending           *PendingHub
	Definitions       DefinitionSet
	HealthSources     []HealthSource
	Limits            client.Limits
	TerminalRuns      int
	TerminalBytes     int
	TransitionTimeout time.Duration
}

RunHostConfig owns the immutable limits and dependencies of a generated daemon application. Limits.ActiveRuns is the host's global active capacity, so the same value is reported by Health without caller-controlled drift.

type RunHostDescription 

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

RunHostDescription is one immutable, validated view of the generated agent definitions and daemon readiness observed at the same host synchronization boundary. It is safe to expose before a client session is negotiated.

func NewRunHostDescription 

func NewRunHostDescription(definitions DefinitionSet, health client.Health) (RunHostDescription, error)

NewRunHostDescription validates and defensively copies one host description.

func (RunHostDescription) Definitions 

func (description RunHostDescription) Definitions() DefinitionSet

Definitions returns a defensive copy of the generated definition catalog.

func (RunHostDescription) Health 

func (description RunHostDescription) Health() client.Health

Health returns the immutable readiness snapshot paired with Definitions.

func (RunHostDescription) Validate 

func (description RunHostDescription) Validate() error

Validate verifies that this description still satisfies its public contract.

type RunImport 

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

RunImport is an opaque prepared import transaction and authenticated snapshot verifier. It owns the stable run lock until Abort or Activate.

func (*RunImport) Abort 

func (transaction *RunImport) Abort() error

Abort releases a pre-consume transaction. After Consume it returns ErrRunAuthorityUncertain while still releasing the process lock.

func (*RunImport) Activate 

func (transaction *RunImport) Activate(ctx context.Context) (*ActiveRun, error)

Activate durably advances to ACTIVE and transfers the held lock to the returned run lease. Call it only after prepared kernel execution commits. If Activate returns ErrRunAuthorityUncertain, the transaction still owns the lock: stop and join the committed kernel run before Abort or Close releases it. Never continue that run without a returned ActiveRun lease.

func (*RunImport) Close 

func (transaction *RunImport) Close() error

Close is equivalent to Abort.

func (*RunImport) Consume 

func (transaction *RunImport) Consume(ctx context.Context) error

Consume durably advances to IMPORTING. After success, failure is uncertain and the same snapshot must never be retried automatically.

func (*RunImport) VerifySnapshot 

func (transaction *RunImport) VerifySnapshot(
	ctx context.Context,
	input enginev1.SnapshotAuthorityInput,
	claim *enginev1.SnapshotAuthority,
) error

VerifySnapshot implements enginev1.SnapshotAuthorityVerifier and remains bound to the exact snapshot claim used to prepare this transaction.

type Session 

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

Session is one immutable client ownership epoch.

func (Session) ClientID 

func (session Session) ClientID() string

ClientID returns the stable cryptographic identity preserved by reconnect.

func (Session) Context 

func (session Session) Context() context.Context

Context returns the daemon-root-owned epoch context.

func (Session) Epoch 

func (session Session) Epoch() uint64

Epoch returns the current ownership generation.

type SessionGateCapacityError 

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

SessionGateCapacityError identifies the exhausted per-client gate resource.

func (*SessionGateCapacityError) Error 

func (failure *SessionGateCapacityError) Error() string

func (*SessionGateCapacityError) Is 

func (failure *SessionGateCapacityError) Is(target error) bool

Is makes SessionGateCapacityError match ErrSessionGateCapacity.

func (*SessionGateCapacityError) Maximum 

func (failure *SessionGateCapacityError) Maximum() int

Maximum returns the configured hard maximum for the resource.

func (*SessionGateCapacityError) Resource 

func (failure *SessionGateCapacityError) Resource() string

Resource returns the bounded resource that was exhausted.

type SessionStore 

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

SessionStore assigns stable cryptographic client identities and fences stale owners. Each client has an independent, bounded FIFO mutation-commit gate. Reconnect intents take priority over queued commits, drain an active commit, cancel and join old streams, and only then advance the ownership epoch. All epoch contexts derive from the caller-owned daemon root.

func NewSessionStore 

func NewSessionStore(root context.Context, maximum int) (*SessionStore, error)

NewSessionStore constructs a bounded store owned by root.

func (*SessionStore) AcquireMutationCommit 

func (store *SessionStore) AcquireMutationCommit(
	ctx context.Context,
	clientID string,
	epoch uint64,
) (*MutationCommitLease, error)

AcquireMutationCommit obtains the exclusive commit boundary for one client epoch. Contended acquisitions are bounded and FIFO. A registered reconnect intent takes priority so old queued mutations cannot commit after a claimant begins fencing. Close the returned lease immediately after commit certainty is known; cancellation after acquisition never releases it implicitly.

func (*SessionStore) AcquireStream 

func (store *SessionStore) AcquireStream(ctx context.Context, clientID string, epoch uint64) (*StreamLease, error)

AcquireStream registers one old-frame fence for a client epoch. The returned Context is canceled by its caller context, reconnect, daemon-root cancellation, or store closure. A stream handler must stop and join every sender before it closes the lease; a successful ReconnectContext cannot advance or return while any old stream lease remains registered.

func (*SessionStore) Check 

func (store *SessionStore) Check(clientID string, epoch uint64) error

Check verifies that an epoch still owns the stable client identity.

func (*SessionStore) Close 

func (store *SessionStore) Close()

Close fences every owner, cancels streams, wakes queued acquisitions, and rejects future session work. It is nonblocking and idempotent; Shutdown waits for active commit and stream leases when an orderly join is required.

func (*SessionStore) Fence 

func (store *SessionStore) Fence(clientID string, epoch uint64) (context.Context, error)

Fence returns the current ownership context or rejects a stale epoch.

func (*SessionStore) Fresh 

func (store *SessionStore) Fresh() (Session, error)

Fresh creates a cryptographically random stable client ID at epoch one.

func (*SessionStore) Reconnect 

func (store *SessionStore) Reconnect(clientID string, expected uint64) (Session, error)

Reconnect performs an exact compare-and-swap to the next ownership epoch. It preserves the original blocking API; callers needing cancellation should use ReconnectContext.

func (*SessionStore) ReconnectContext 

func (store *SessionStore) ReconnectContext(ctx context.Context, clientID string, expected uint64) (Session, error)

ReconnectContext registers a priority reconnect intent, drains the current mutation commit, cancels and joins every old stream lease, and advances the epoch only after all fences are closed. Cancellation before the epoch CAS removes the intent and lets the old epoch's FIFO continue.

func (*SessionStore) Shutdown 

func (store *SessionStore) Shutdown(ctx context.Context) error

Shutdown closes the store and waits for every active commit/stream lease and queued gate claimant to release. The caller-owned context bounds the join.

type StaleSessionError 

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

StaleSessionError reports the authoritative and presented ownership epochs for a known stable client. Unknown client identities continue to return the ErrStaleSession sentinel because no positive authoritative epoch exists.

func (*StaleSessionError) ClientID 

func (failure *StaleSessionError) ClientID() string

ClientID returns the stable client whose ownership check failed.

func (*StaleSessionError) Error 

func (failure *StaleSessionError) Error() string

func (*StaleSessionError) ExpectedEpoch 

func (failure *StaleSessionError) ExpectedEpoch() uint64

ExpectedEpoch returns the daemon's current positive epoch.

func (*StaleSessionError) Is 

func (failure *StaleSessionError) Is(target error) bool

Is makes StaleSessionError match ErrStaleSession.

func (*StaleSessionError) ObservedEpoch 

func (failure *StaleSessionError) ObservedEpoch() uint64

ObservedEpoch returns the epoch supplied by the caller.

type StreamLease 

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

StreamLease is an opaque reconnect fence. Context supplies the cooperative cancellation signal; Close acknowledges that all stream senders have joined. Close is nonblocking, idempotent, and safe for concurrent use.

func (*StreamLease) Close 

func (lease *StreamLease) Close()

Close acknowledges that the stream's senders have joined and releases its reconnect fence.

func (*StreamLease) Context 

func (lease *StreamLease) Context() context.Context

Context returns the stream's cooperative lifetime context.

type TerminalPhase 

type TerminalPhase string

TerminalPhase is a durable non-resumable run tombstone. 

const (
	TerminalCompleted TerminalPhase = "COMPLETED"
	TerminalFailed    TerminalPhase = "FAILED"
	TerminalCancelled TerminalPhase = "CANCELLED"
)

Source Files

View all Source files

Directories

Path Synopsis
Package endpoint defines opaque local-daemon credentials, current-user endpoint metadata, and secure publication/discovery coordination.
 Package endpoint defines opaque local-daemon credentials, current-user endpoint metadata, and secure publication/discovery coordination.
Package grpcserver implements the authenticated local gRPC process boundary without adding transport dependencies to the daemon lifecycle core.
 Package grpcserver implements the authenticated local gRPC process boundary without adding transport dependencies to the daemon lifecycle core.
internal
runauthority
Package runauthority owns the crash-safe, per-user authority for daemon runs.
 Package runauthority owns the crash-safe, per-user authority for daemon runs.
userstorage
Package userstorage provides retained, current-user-only local storage for daemon security state.
 Package userstorage provides retained, current-user-only local storage for daemon security state.
Package localipc opens explicitly addressed, current-user-only local IPC connections.
 Package localipc opens explicitly addressed, current-user-only local IPC connections.

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.