sessionstate

package
v0.28.2 Latest Latest
Warning

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

 Go to latest Published: Sep 10, 2026 License: AGPL-3.0 Imports: 20 Imported by: 0

Documentation

Overview

Package sessionstate is the pod-local session state authority (design 0055, Epic 69 US-69.2): a module-sealed subsystem that owns seq assignment (I1), the store-reseeded projection (I3/I4), the stamped snapshot + ordered event fanout (I2), the durable seq cursor on the platform/ PVC subPath (I9 prep), and the authenticated ABI surface (I8). It is dialect-free machinery: the harness event parser and store reader are injected seams (US-69.3 grows the opencode implementations).

Sealing rules (enforced by TestModuleSealDependencies): no imports of agentd supervision or any cmd/ package; callers reach the authority only through this package's exported API.

Index

Constants

View Source 
const DefaultPlatformDir = "/platform"

DefaultPlatformDir is the platform/ PVC subPath mount (pod_builder). Overridable for tests via Config.PlatformDir.

Variables

View Source 
var ErrNoStore = errors.New("sessionstate: no store reader wired")

ErrNoStore is returned by Reseed when no store reader is wired: the projection cannot be rebuilt from truth, so reseed fails loudly instead of silently trusting the (possibly stale) in-memory state.

Functions

This section is empty.

Types

type Actor 

type Actor interface {
	Act(ctx context.Context, sessionID string, action *abiv1.ActionRequest) (*abiv1.ActionResult, error)
}

Actor is the dialect seam for action execution: perform one typed action against the harness (localhost :4096 in production; the wiring layer injects the opencode implementation — same seam class as Admitter). The authority has already validated the union member and verified capability declaration; the Actor executes UNDER the session's single-flight lock, so a slow action delays that session's admissions (the sole-writer contract) but nothing else.

type Admitter 

type Admitter interface {
	Admit(ctx context.Context, sessionID, text, model string) (string, error)
}

Admitter is the dialect seam for admission: POST the harness's prompt endpoint. The wiring layer injects the opencode implementation (localhost :4096, Basic §D1 credential — same seam class as EventParser/StoreReader). Returns the harness's message ID for promotion correlation.

type Authority 

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

Authority is the module-sealed session state authority. All mutable state lives under mu: seq assignment, the projection, and fanout registration share one lock so the stamp is atomic by construction.

func New 

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

New constructs the authority and loads the durable seq cursor from PlatformDir. It does NOT reseed — callers run Reseed(ReseedBoot) once the store reader is reachable (boot ordering is the caller's decision).

func (*Authority) Act 

func (a *Authority) Act(ctx context.Context, req *connect.Request[abiv1.ActionRequest]) (*connect.Response[abiv1.ActionResult], error)

func (*Authority) CheckStalls 

func (a *Authority) CheckStalls(ctx context.Context) StallStats

CheckStalls runs one stall-detection pass: admitted rows older than the promotion deadline move to stalled (fsync'd), and the configured Wake fires once per newly-stalled row. No-op without a ledger.

func (*Authority) Close 

func (a *Authority) Close() error

Close persists the final cursor and releases resources.

func (*Authority) Deliver 

func (a *Authority) Deliver(ctx context.Context, req *connect.Request[abiv1.DeliveryRequest]) (*connect.Response[abiv1.DeliveryAck], error)

func (*Authority) Events 

func (a *Authority) Events(ctx context.Context, req *connect.Request[abiv1.EventsRequest], stream *connect.ServerStream[abiv1.StreamFrame]) error

func (*Authority) GetDeliveryStatus 

func (a *Authority) GetDeliveryStatus(ctx context.Context, req *connect.Request[abiv1.GetDeliveryStatusRequest]) (*connect.Response[abiv1.DeliveryStatus], error)

func (*Authority) GetSnapshot 

func (a *Authority) GetSnapshot(ctx context.Context, req *connect.Request[abiv1.GetSnapshotRequest]) (*connect.Response[abiv1.SessionSnapshot], error)

func (*Authority) Handler 

func (a *Authority) Handler() (string, http.Handler)

Handler returns the ABI surface: the generated connect service behind Basic auth (I8 — every route authenticated, zero exceptions). The returned mount path is the connect procedure root; agentd's user mux serves it as-is.

func (*Authority) InFlightDeliveries added in v0.27.0 

func (a *Authority) InFlightDeliveries() int64

InFlightDeliveries is the ledger's unresolved delivery count (ledgered + admitted + stalled) — design 0055 M4's flip-gate drain signal. Promoted/turn-ended/failed rows are resolved from the flip's perspective (failed is terminal per attempt; the outbox owns retries).

func (*Authority) Ingest 

func (a *Authority) Ingest(raw []byte)

Ingest applies one raw harness event. It is the recover wall: parser panics and decode failures are contained (counted + logged), never propagated — agentd must not die from projection input. During a reseed the payload is buffered instead of applied (I3 ordering).

func (*Authority) IngestForTest 

func (a *Authority) IngestForTest(evt *abiv1.Event)

IngestForTest applies a contract event directly, bypassing the parser seam (fault-injection + comparator support: the projection fold is exercised without dialect fixtures).

func (*Authority) KillForTest 

func (a *Authority) KillForTest()

KillForTest simulates SIGKILL semantics for the kill-9 fault-injection test: resources are abandoned WITHOUT graceful cleanup (no cursor close, no subscriber teardown) exactly as a hard process death would.

func (*Authority) Metrics 

func (a *Authority) Metrics() Metrics

func (*Authority) ParserFailuresForTest added in v0.27.2 

func (a *Authority) ParserFailuresForTest() int64

ParserFailuresForTest exposes the parser-failure counter (the #1291 r5 shape-drift accounting pin).

func (*Authority) PlatformDir 

func (a *Authority) PlatformDir() string

PlatformDir reports the durable-cursor directory (suspend/resume diagnostics and the S1 scenario harness's restart scenarios).

func (*Authority) ReplayUnresolvedDeliveries 

func (a *Authority) ReplayUnresolvedDeliveries(ctx context.Context)

ReplayUnresolvedDeliveries drives admission for ledgered rows after a crash/suspend resume (M2 pod-death path: exactly-once per attempt — admitted rows are skipped by construction, I6).

func (*Authority) Reseed 

func (a *Authority) Reseed(ctx context.Context, reason ReseedReason) error

Reseed runs the ordered, race-free reseed procedure (design 0055 M1): quiesce → buffer inbound → reseed from the store → emit projection.reseeded (consuming a seq) → flush buffered → resume live. Serialized by reseedMu; the store read happens OUTSIDE mu so ingest and stream reads stay hot (M3.1).

func (*Authority) SetParserForTest added in v0.27.2 

func (a *Authority) SetParserForTest(p EventParser)

SetParserForTest swaps the parser seam (the #1291 r6 restart-mid-turn e2e: a fresh translator instance models the process restarting with an empty tool-call memo).

func (*Authority) SetStallDeadlineForTest 

func (a *Authority) SetStallDeadlineForTest(d time.Duration)

SetStallDeadlineForTest shrinks the promotion-stall deadline (the fault-injection harness makes a row cross it without waiting 10m).

func (*Authority) SetStoreForTest 

func (a *Authority) SetStoreForTest(s StoreReader)

SetStoreForTest swaps the store reader (fault-injection support).

func (*Authority) State 

func (a *Authority) State() StateSnapshot

State returns the atomically-stamped projection snapshot.

func (*Authority) Stream 

func (a *Authority) Stream(ctx context.Context) (<-chan *abiv1.StreamFrame, func(), error)

Stream subscribes to the ordered frame stream. The mandatory connection ordering (I2): (1) register the per-connection buffer, (2) capture the stamped snapshot under the projection lock, (3) flush buffered frames with seq > stamp, (4) go live. Canceling the returned func tears the subscription down.

type Config 

type Config struct {
	PlatformDir string
	Parser      EventParser
	Store       StoreReader
	Passwords   []string
	RateLimit   RateLimitConfig
	Logger      *zap.Logger
	// ABIVersion is reported in the capability report snapshot frame.
	ABIVersion string
	// Capabilities, when set, is the capability report served on the
	// snapshot frame (provenance, supported actions, supported part
	// kinds). Built by the wiring layer from boot-time facts (0053 pins,
	// versions, D3 part limits) — the report is STATIC per boot so the
	// hot path never touches the harness (M3.1).
	Capabilities *abiv1.CapabilityReport
	// Admitter (US-69.7) is the delivery-admission seam; nil DISABLES
	// the ledger (Deliver returns NotSupported) — the authority owns
	// nothing dialect-specific.
	Admitter Admitter
	// Actor (US-69.9) is the typed-actions seam; nil DISABLES the Act op
	// (NotSupported) — the wiring layer injects the opencode executor.
	Actor Actor
	// Wake (US-69.12) is the stall-recovery seam: agentd's best-effort
	// nudge when an admitted row crosses the promotion deadline
	// (checkStalls fires it once per row). Nil = no-op wake (stalls
	// still surface via metrics/alerts; wake failures count only when a
	// wake exists).
	Wake func(ctx context.Context, sessionID string) error
	// FastCursor disables per-event fsync (scenario harnesses replaying
	// event BURSTS; durability is covered by the fault-injection suite).
	FastCursor bool
}

Config wires the authority. PlatformDir holds the durable seq cursor (platform/ PVC subPath in production; any dir in tests).

type EventParser 

type EventParser interface {
	Parse(raw []byte) (evt *abiv1.Event, ok bool, err error)
}

EventParser translates one raw harness SSE payload into a contract event. ok=false means the payload carries nothing projectable (dropped, counted); err signals a decode failure of a payload that CLAIMED to be projectable (wire drift — logged, never fatal). Implementations must be safe for concurrent use.

type LedgerState 

type LedgerState int

LedgerState is the M2 state machine: 

ledgered → admitted → promoted → turn-ended

with `failed` (per attempt, terminal for THAT attempt — re-armable at attempt+1) and `stalled` (admitted past the promotion deadline — recovered by WAKE ONLY, never re-admitted; I6). 

const (
	LedgerStateUnspecified LedgerState = iota
	LedgerStateLedgered
	LedgerStateAdmitted
	LedgerStatePromoted
	LedgerStateTurnEnded
	LedgerStateStalled
	LedgerStateFailed
)

func (LedgerState) String 

func (s LedgerState) String() string

String returns the lowercase wire name (metric labels, logs).

type Metrics 

type Metrics struct {
	DroppedEvents     int64
	ParserFailures    int64
	PanicsContained   int64
	CustomValveEvents int64
	Subscribers       int
	BufferedPending   int
	// SecondsSinceSeqAdvance is how long since the projection last moved
	// (the R5 starvation signal: seq stalled N seconds while the pod
	// runs). Initialized at construction; a reseed counts as advance.
	SecondsSinceSeqAdvance float64
	// LedgerDepths is the per-state row count (the ledger funnel).
	// Nil when no ledger is wired.
	LedgerDepths map[string]int64
	// StalledEntries is the current stalled-row count (the #1119 class,
	// visible).
	StalledEntries int64
	// OldestPromotionStallSeconds is the age of the oldest
	// admitted-unpromoted row (0 when none).
	OldestPromotionStallSeconds float64
}

Metrics exposes module counters for ops_metrics wiring (US-69.12).

type RateLimitConfig 

type RateLimitConfig struct {
	Burst        int
	RefillPerSec float64
}

RateLimitConfig bounds per-session mutating-op rates (I8).

type ReseedReason 

type ReseedReason int

ReseedReason classifies why a projection reseed ran. 

const (
	ReseedReasonBoot ReseedReason = iota
	ReseedReasonGenerationChange
	// ReseedReasonStallWake (US-69.12): the stall watchdog reseeded
	// because an admitted row crossed the promotion deadline — the
	// store refresh is the wake (events completing the turn promote or
	// turn-end the row; a still-missing promotion escalates via the
	// stalled-entries/wake-failure alerts).
	ReseedReasonStallWake
)

type SessionSeed 

type SessionSeed struct {
	Status        abiv1.SessionStatus
	PendingInputs []*abiv1.InputRequest
}

SessionSeed is the store-truth seed for one session at reseed time (I3/I4): the authoritative status plus the pending inputs the store still holds. Busy/streaming and in-flight parts are NEVER seeded — they are live-turn state, rebuilt by events after the reseed.

type SessionView 

type SessionView struct {
	Status        abiv1.SessionStatus
	Busy          bool
	InFlightParts []*abiv1.Part
	PendingInputs []*abiv1.InputRequest
}

SessionView is one session's projected state (I12: a snapshot of these fields alone renders the session).

type StallStats 

type StallStats struct {
	// Stalled is the number of rows moved admitted→stalled THIS pass.
	Stalled int
	// WakeFailures is the number of wake attempts that errored THIS pass
	// (the escalation signal: wake-only recovery failing).
	WakeFailures int
}

StallStats is one stall-watchdog pass's outcome (US-69.12).

type StateSnapshot 

type StateSnapshot struct {
	Seq      uint64
	Sessions map[string]*SessionView
}

StateSnapshot is an atomically-stamped view of the projection (I1: the state is exactly the fold of events 1..Seq).

type StoreReader 

type StoreReader interface {
	SessionStates(ctx context.Context) (map[string]SessionSeed, error)
}

StoreReader snapshots the harness store's session truth — the reseed source (I3/I4). Implementations must respect ctx deadlines; a hung store must surface as ctx error, never block forever.

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.