service

type AdmissionConflictError ¶

type AdmissionConflictError struct {
	Key       registry.Key
	Admitted  sessionwire.AgentID
	Requested sessionwire.AgentID
}

AdmissionConflictError reports one ledger key admitted under two agents.

It is a DISTINCT type from AdmissionRefusedError and deliberately carries no HostLinkErrorCode: it is not a placement outcome Factory should retry elsewhere but a Host-side bookkeeping fault, and giving it a wire code would invite a caller to treat it as one.

func (e *AdmissionConflictError) Error() string

type AdmissionRefusedError ¶

type AdmissionRefusedError struct {
	Code    sessionwire.HostLinkErrorCode
	AgentID sessionwire.AgentID
	Reason  string
	Cause   error
}

AdmissionRefusedError reports an admission this Host may not accept.

Code is one of Core's HostLinkErrorCode values rather than a locally invented vocabulary, because §15's routing algorithm tells Factory to discard its binding and re-run placement on exactly these: not_admitting, no_capacity, runtime_unavailable, runtime_mismatch. A local code would have to be translated at the wire edge, and a translation table is a second place for the meaning to drift.

func (e *AdmissionRefusedError) Error() string

func (e *AdmissionRefusedError) Unwrap() error

type Advertisement ¶

type Advertisement struct {
	// Namespace is the OrderedIndex namespace. It is AdvertisementNamespace for
	// every record this package derives, and is carried on the value rather
	// than left implicit so a writer needs nothing but the Advertisement.
	Namespace string

	// RankingScope is the LaunchTarget scope this record is ranked within. It
	// is the stable key's tuple MINUS the host, because ranking exists to
	// compare Hosts offering the same target against each other.
	RankingScope string

	// StableKey identifies this record: (agent, runtime compatibility,
	// placement, host). It is a digest, not the tuple — see stableKey.
	StableKey string

	// Rank is available admission capacity, which is what §15 ranks by. It is
	// zero on a tombstone.
	Rank int64

	// Ranked reports whether this record appears in a ranked page at all. It is
	// false for a drained Host AND for a target this Host cannot place, and the
	// second half is a DECISION beyond what §15 spells out — the spec names
	// drain as un-ranking and does not say drain is the only thing that may.
	//
	// The distinction it rests on is between a row that is not a candidate NOW
	// and one that can never be a candidate HERE. A full but accepting Host has
	// zero available capacity and stays ranked, because the next heartbeat may
	// find room; a target whose declared placement this Host does not offer is
	// unplaceable for the life of the process, so ranking it only puts a row in
	// the page Factory must then discard. Tombstone, not Ranked, is what
	// distinguishes drain from either.
	//
	// It therefore coincides with Report.Accepting in every state this package
	// can derive. They are separate fields because they are read by different
	// parties — Accepting is record CONTENT a reader filters on, Ranked is an
	// INDEX operation the writer performs — and the coincidence is asserted by
	// test rather than assumed, in all three states.
	Ranked bool

	// DueAt is when a bounded due reconciler must look at this record. It is
	// ExpiresAt, so a Host that stops heartbeating is reaped by its own expiry.
	DueAt time.Time

	// Tombstone reports a record graceful drain has retired. It is a record and
	// not a deletion, so a reader that has already seen the offer learns it is
	// gone rather than waiting out the expiry.
	Tombstone bool

	// Report is the sessionwire observation itself, validated before it is
	// returned.
	Report sessionwire.HostLinkCapacityReport
}

Advertisement is one derived target-directory record.

It is a VALUE and carries no method that writes. Report is the wire record a reader consumes; the seven fields around it are the OrderedIndex placement of that record — where it is stored (Namespace, StableKey), how it is ordered against other Hosts offering the same LaunchTarget (RankingScope, Rank, Ranked), when a due reconciler sees it (DueAt), and whether it is a gravestone rather than an offer (Tombstone).

func (a Advertisement) Expired(now time.Time) bool

type AdvertisementError ¶

type AdvertisementError struct {
	AgentID sessionwire.AgentID
	Reason  string
	Cause   error
}

AdvertisementError reports a derived record no reader would accept.

func (e *AdvertisementError) Error() string

func (e *AdvertisementError) Unwrap() error

type CapacityOptions ¶

type CapacityOptions struct {
	// Host is the validated Host whose Department is advertised. Everything
	// else the derivation needs — identity, endpoint, isolation class,
	// placement, capacity, registry expiry, clock — is read from it, so there
	// is no second copy of a value host.New already checked.
	Host *hostconfig.Host

	// HostGeneration is this Host process's incarnation identity, which Core
	// requires to be non-zero on every HostLink record. It is NOT on
	// host.Options: a Host that restarts on the same configuration is a new
	// generation, so the value belongs to the run and not to the configuration.
	HostGeneration uint64
}

CapacityOptions configures a CapacityPublisher.

type CapacityPublisher ¶

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

CapacityPublisher derives the Department's target advertisements.

It holds the ONLY mutable state in this package: the admission ledger and the drain flag. Everything else is either read from the immutable Host or taken from the target snapshot below, so no value it derives from can drift from the one that was validated.

THAT SENTENCE WAS FALSE WHEN IT WAS FIRST WRITTEN, which is why it now names the snapshot. Capabilities() and CompatibilityID() were re-read on every heartbeat, and they are caller code free to answer differently each time; the claim covered only the Host and read as though it covered everything. See launchable for the panic that cost.

FOR O6.1 AND O6.3, WHICH ARRIVE LATER AND ELSEWHERE. 04-host.md puts weighted memory/admission limits in internal/residency and internal/lifecycle, and drain in internal/lifecycle/drain.go. Those must CONSUME this ledger and this flag, not keep their own. Two sources of "consumed" is an over-admission bug that neither package's tests could see — each would be internally consistent and the Host would admit past its capacity — and two sources of "draining" is a Host that stops accepting in one place while still advertising Accepting from the other. Whatever the final ownership, there must be exactly one of each, and this is the one that already exists.

func NewCapacityPublisher(options CapacityOptions) (*CapacityPublisher, error)

func (p *CapacityPublisher) Admit(key registry.Key, agent sessionwire.AgentID) error

func (p *CapacityPublisher) BeginDrain()

func (p *CapacityPublisher) ConsumedWeight() uint64

func (p *CapacityPublisher) Draining() bool

func (p *CapacityPublisher) Own(key registry.Key, generation uint64) bool

func (p *CapacityPublisher) Publish() ([]Advertisement, error)

func (p *CapacityPublisher) Release(key registry.Key) bool

func (p *CapacityPublisher) ReleaseOwned(key registry.Key, generation uint64) bool

type FlushTimer ¶ added in v0.12.0

type FlushTimer interface {
	C() <-chan time.Time
	Stop() bool
	Reset(time.Duration) bool
}

FlushTimer is the small timer contract used by the live-text relay.

type InvalidCapacityOptionsError ¶

type InvalidCapacityOptionsError struct {
	Field  string
	Reason string
	Cause  error
}

InvalidCapacityOptionsError reports an option a publisher may not run with.

func (e *InvalidCapacityOptionsError) Error() string

func (e *InvalidCapacityOptionsError) Unwrap() error

type InvalidTailOptionsError ¶

type InvalidTailOptionsError struct {
	Field  string
	Reason string
}

InvalidTailOptionsError reports an option a Tails may not run with.

func (e *InvalidTailOptionsError) Error() string

type Projector ¶ added in v0.10.0

type Projector func(ctx context.Context, body json.RawMessage, journalSeq uint64) (json.RawMessage, error)

Projector rewrites one committed public body before it is published, and nothing else about the publication. A nil Projector relays bodies verbatim.

IT IS THE ONE EXCEPTION TO "THE BYTES ARE CARRIED", and it is one on purpose (finding W1): a runtime's body names the RUNTIME session and command ids, which are private, and every consumer of this stream is a client's. The exception is a DETERMINISTIC function of the stored body and immutable identities, applied identically to the durable read (see host.PublicJournal), so the live body still equals the one /journal serves for the same event. A projection that fails refuses the publication (TailEndRefused): relaying the unprojected body would leak, and skipping it would leave a hole.

It is handed the body and the event's journal sequence, and nothing it could use to move the event: identity, sequence and coverage are relayed unchanged.

type Publications ¶

type Publications interface {
	// Publish delivers one payload to every subscriber of one channel.
	Publish(channel string, payload []byte) error
}

Publications is the transport one session's committed tail is published through.

IT TAKES A CHANNEL AND A PAYLOAD AND NOTHING ELSE, which is how "Host sends one message per event and fans out nowhere" is a property of the TYPE rather than of this package's restraint: there is no parameter a link, a replica or a subscriber set could ride in, so no future caller can address one even by accident. The fan-out belongs to the transport, which owns a queue per physical connection; Host owns the decision to publish once.

The shape is centrifuge's own — a channel, bytes, no context — because the implementation is a call straight through to it, and a seam carrying a context nobody could honour would be describing a cancellation that does not exist.

type Routes ¶

type Routes interface {
	// InvalidateSession drops every link's route to one session and returns
	// them.
	InvalidateSession(registry.Key) []hostlink.Binding
}

Routes is the routing table a lost tail invalidates, narrowed to the single method a tail may reach.

It is ONE method and that method is session-scoped, for the reason Residencies is one read: a publisher that could bind, deliver or close a link would be able to repair its own loss by re-routing it, and the whole point of invalidating is that the repair is Factory's — a durable reset from its own cursor, not a hidden retry here. *hostlink.Multiplexer satisfies it; the conformance assertion lives in the test so this package does not need it at build time.

type Tail ¶

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

Tail is one running publisher for one session.

func (t *Tail) Channel() string

func (t *Tail) Done() <-chan struct{}

func (t *Tail) End() (TailEnd, error)

func (t *Tail) EphemeralDrops() uint64

func (t *Tail) EphemeralPublished() uint64

func (t *Tail) Key() registry.Key

func (t *Tail) Published() uint64

func (t *Tail) Stop()

type TailEnd ¶

type TailEnd string

TailEnd is why one tail stopped.

const (
	// TailEndRunning reports a tail that has not ended.
	TailEndRunning TailEnd = "running"

	// TailEndStopped reports a tail this Host ended: the caller stopped it or
	// the context it was started under is done. It is the ONE ending that does
	// not invalidate a route, because the route did not fail — Host chose to
	// stop feeding it, and whoever chose that owns what happens to the route.
	TailEndStopped TailEnd = "stopped"

	// TailEndLost reports a tail that ended without Host asking: the runtime
	// stopped, the hub failed the subscription for an enduring overflow, or
	// Host's own egress overflowed. Every one of them means this Host can no
	// longer feed the live stream, so the routes are invalidated.
	//
	// A mixed subscription carries a typed cause when its adapter sees an
	// enduring delivery without committed fields; Tail.End surfaces that cause.
	// A bare channel close still carries no cause, so Host cannot distinguish
	// hub backpressure from all upstream invariant failures. It invalidates
	// routes in either case and never resubscribes on its own.
	TailEndLost TailEnd = "lost"

	// TailEndRefused reports a publication this Host would not put on the wire:
	// one naming another session, one Core's own encoder rejects, or one the
	// transport refused. It invalidates for the same reason a loss does — the
	// stream has a hole in it — and it is a separate value because the repair
	// is different and an operator must not read one as the other.
	TailEndRefused TailEnd = "refused"
)

type TailOptions ¶

type TailOptions struct {
	Publications     Publications
	Routes           Routes
	FlushInterval    time.Duration
	IncludeReasoning bool
	// IncludeToolSteps relays the runtime's public ToolCallStarted and
	// ToolCallCompleted bodies. It needs a runtime implementing
	// department.LiveOptionsSubscriber.
	IncludeToolSteps bool
	// NewFlushTimer allows a controlled clock for interval tests.
	NewFlushTimer func(time.Duration) FlushTimer
	Logger        *slog.Logger
}

TailOptions configures a Tails.

type Tails ¶

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

Tails publishes committed session tails and admitted transient text onto HostLink channels.

WHAT IT DOES NOT DO IS THE SUBSTANCE. It does not read a journal, does not project an event, does not re-encode a body, and does not decide what a public body IS. Harness's committed-public-event capability supplies the exact canonical body the durable append stored, together with the EventID, JournalSeq and CoveredThrough that append committed under; this type relays all five unchanged. A second projection would be a second answer: a consumer joining its durable tail to this live stream dedupes on (sequence, event id) and compares bodies, and if the live body were re-derived it could differ from the stored one in key order or in a field a later projector added, with no error anywhere.

Core's own EnduringPublication says the same from the other side: its MarshalJSON "keeps the canonical public body opaque … a Host relays that committed body rather than decoding and synthesizing it anew" (core@v0.7.0 sessionwire/v1/publications.go:152-154). Body is a json.RawMessage on both sides of this package, so the bytes are carried, not parsed.

THE ONE EXCEPTION IS A Projector (finding W1). A runtime's canonical body names its PRIVATE runtime session and command ids, and every reader of this stream is a client's, so a composed Host rewrites exactly those identities before publishing. That is not a second projection in the sense above: it is a deterministic function of the stored body that the durable read applies too (host.PublicJournal), so live and durable bodies stay byte-identical.

func NewTails(options TailOptions) (*Tails, error)

func (t *Tails) Publish(
	ctx context.Context,
	key registry.Key,
	subscriber department.PublicationSubscriber,
) (*Tail, error)

func (t *Tails) PublishProjected(
	ctx context.Context,
	key registry.Key,
	subscriber department.PublicationSubscriber,
	rewrite Projector,
	ephemeralRewrite ...Projector,
) (*Tail, error)