event

package
v1.10.4 Latest Latest
Warning

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

 Go to latest Published: Nov 4, 2026 License: MIT Imports: 16 Imported by: 0

Documentation

Overview

Package event defines bounded, metadata-only public execution replay.

Index

Constants

View Source 
const (
	MaxBatchSize      = 32
	MaxPageSize       = 128
	MaxRetainedEvents = 256
)
View Source 
const MaxTerminalSnapshotBytes = 8 << 20

MaxTerminalSnapshotBytes bounds the encoded private terminal snapshot.

View Source 
const PrefixEvent id.Prefix = "pevt"

Variables

View Source 
var (
	ErrAccessDenied = cortex.ErrPublicEventAccessDenied
	ErrInvalid      = cortex.ErrInvalidPublicEvent
	ErrConflict     = cortex.ErrPublicEventConflict
	ErrCursorGap    = cortex.ErrPublicEventCursorGap
	ErrUnavailable  = cortex.ErrPublicEventUnavailable
)

Functions

func EncodeCursor 

func EncodeCursor(cursor Cursor) (string, error)

EncodeCursor produces a canonical, versioned token without JSON number rounding. Tokens bind a position to a run; they do not grant access to that run.

func FollowReplay 

func FollowReplay(ctx context.Context, authorized ReadStore, request FollowRequest, send func(context.Context, ReplayPage) error) error

FollowReplay sends authorized pages synchronously, retaining at most one page. Each read rechecks authority, including idle polls. Consumer errors, gaps and cancellation stop the loop. It never infers completion from a lifecycle event. Pass an authorized engine, never a raw store; see ReadReplayPage.

Types

type AppendReceipt 

type AppendReceipt struct {
	FirstSequence uint64 `json:"first_sequence"`
	LastSequence  uint64 `json:"last_sequence"`
}

AppendReceipt survives replay retention and gives the original sequence range.

type AppendRequest 

type AppendRequest struct {
	SchemaVersion       int           `json:"schema_version"`
	RunID               id.AgentRunID `json:"run_id"`
	ExpectedRunRevision int64         `json:"expected_run_revision"`
	AfterSequence       uint64        `json:"after_sequence"`
	Events              []Input       `json:"events"`
}

AppendRequest is trusted host input. Storage checks the current run transition. Reuse the exact request after uncertainty; IDs cannot name another batch later.

func (AppendRequest) Freeze 

func (r AppendRequest) Freeze() (AppendRequest, string, error)

Freeze validates, detaches and hashes the complete idempotency request.

type Authorizer 

type Authorizer interface {
	AuthorizePublicReplay(context.Context, id.AgentRunID) error
}

Authorizer checks the current subscriber, exact scope and run grant. The host must honor revocation on each call; historical commit authority is insufficient.

type Cursor 

type Cursor struct {
	SchemaVersion int           `json:"schema_version"`
	RunID         id.AgentRunID `json:"run_id"`
	Sequence      uint64        `json:"sequence"`
}

Cursor binds a monotonically increasing position to one run and wire schema.

func DecodeCursor 

func DecodeCursor(runID id.AgentRunID, token string) (Cursor, error)

DecodeCursor rejects foreign runs and noncanonical tokens. An empty token starts at zero, which may produce an explicit retention gap on the first read.

func (Cursor) Validate 

func (c Cursor) Validate() error

Validate rejects unsupported schemas and unbounded positions.

type Event 

type Event struct {
	Input
	SchemaVersion int           `json:"schema_version"`
	RunID         id.AgentRunID `json:"run_id"`
	Sequence      uint64        `json:"sequence"`
	RunRevision   int64         `json:"run_revision"`
	ObservedAt    time.Time     `json:"observed_at"`
}

Event is a historical observation. It grants no continuing access to a result.

func (Event) Validate 

func (e Event) Validate() error

Validate checks the entire public envelope without accepting arbitrary payloads.

type FollowRequest 

type FollowRequest struct {
	RunID        id.AgentRunID
	After        string
	Limit        int
	PollInterval time.Duration
	ReadTimeout  time.Duration
	Lifetime     time.Duration
}

FollowRequest bounds a metadata subscription. Zero durations select one-second polling, five-second reads and a five-minute lifetime. Hosts must honor context cancellation in both the authorized reader and the synchronous consumer.

type Input 

type Input struct {
	ID       id.ID `json:"id"`
	Type     Type  `json:"type"`
	ResultID id.ID `json:"result_id,omitzero"`
}

Input contains only stable identifiers and a closed event type. ResultCommitted requires the event ID from its durable commit outbox.

type Page 

type Page struct {
	Watermark Watermark `json:"watermark"`
	Events    []Event   `json:"events"`
	Next      Cursor    `json:"next"`
}

Page is detached from storage. A retention gap returns only Watermark and Next.

func (Page) Validate 

func (p Page) Validate(request ReadRequest, gap bool) error

Validate checks ordering, completeness and the exact requested run/schema. A gap never includes event bodies and points to the retained boundary.

type ReadRequest 

type ReadRequest struct {
	After Cursor `json:"after"`
	Limit int    `json:"limit"`
}

ReadRequest bounds one authorized page. Sequence zero starts at the beginning.

func (ReadRequest) Validate 

func (r ReadRequest) Validate() error

Validate checks the requested cursor and page bound.

type ReadStore 

type ReadStore interface {
	ReadPublicEvents(context.Context, ReadRequest) (Page, error)
}

ReadStore is the durable subset needed for authorized public replay.

type Reader 

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

Reader checks current authority around each bounded storage page.

func NewReader 

func NewReader(storage ReadStore, authorizer Authorizer) (*Reader, error)

NewReader requires explicit storage and host subscriber authorization.

func (*Reader) Read 

func (r *Reader) Read(ctx context.Context, request ReadRequest) (Page, error)

Read returns only validated public metadata. Authorization failure discards even a retention watermark; callers must reauthorize each subsequent page.

type ReplayFailure 

type ReplayFailure struct {
	Code      string `json:"code"`
	Message   string `json:"message"`
	Retryable bool   `json:"retryable"`
}

ReplayFailure contains static transport-safe diagnostics, never provider causes.

func DescribeReplayFailure 

func DescribeReplayFailure(err error) ReplayFailure

DescribeReplayFailure classifies a failure without exposing internal error text.

type ReplayPage 

type ReplayPage struct {
	Page       Page   `json:"page"`
	NextCursor string `json:"next_cursor"`
	Gap        bool   `json:"gap"`
}

ReplayPage is the shared transport envelope. A gap requires the consumer to refresh its run snapshot before continuing from NextCursor.

func ReadReplayPage 

func ReadReplayPage(ctx context.Context, authorized ReadStore, runID id.AgentRunID, token string, limit int) (ReplayPage, error)

ReadReplayPage reads one validated metadata page through an authorized reader. Pass an engine configured with WithPublicEventReplay or an equivalent adapter around Reader.Read, never a raw store. Zero limit selects MaxPageSize.

type Store 

type Store interface {
	AppendPublicEvents(context.Context, AppendRequest) (AppendReceipt, error)
	ReadPublicEvents(context.Context, ReadRequest) (Page, error)
	ReadEventWatermark(context.Context, id.AgentRunID) (Watermark, error)
	PrunePublicEvents(context.Context, Cursor) error
}

Store is an optional trusted storage capability, not a subscriber authorization boundary. Reads must go through current host policy before public delivery. At most MaxRetainedEvents remain readable; ID/digest tombstones prevent reuse.

type TerminalReceipt 

type TerminalReceipt struct {
	SchemaVersion int           `json:"schema_version"`
	EventID       id.ID         `json:"event_id"`
	RunID         id.AgentRunID `json:"run_id"`
	RequestDigest string        `json:"request_digest"`
	RunRevision   int64         `json:"run_revision"`
	UpdatedAt     time.Time     `json:"updated_at"`
}

TerminalReceipt acknowledges the atomic terminal state and metadata observation. It contains no output, error text or caller metadata.

func (TerminalReceipt) Validate 

func (r TerminalReceipt) Validate(request TerminalUpdate) error

Validate checks a receipt against the complete original request.

type TerminalStore 

type TerminalStore interface {
	CommitTerminalRun(context.Context, TerminalUpdate) (TerminalReceipt, error)
}

TerminalStore is an optional exact-scope atomic run/event storage capability. Identical retries return the original receipt, including after event pruning. Implementations must roll back the run, receipt and event together on failure.

type TerminalUpdate 

type TerminalUpdate struct {
	SchemaVersion int     `json:"schema_version"`
	EventID       id.ID   `json:"event_id"`
	Run           run.Run `json:"run"`
}

TerminalUpdate binds one unchecked terminal snapshot to a stable event identity. Run contains private storage fields; only lifecycle metadata enters public replay.

func (TerminalUpdate) Freeze 

func (r TerminalUpdate) Freeze() (TerminalUpdate, string, error)

Freeze detaches the snapshot and binds retries to its complete contents. Required-verification success must use the guarded candidate/result transaction.

type Type 

type Type string

Type identifies a public lifecycle observation, never candidate content. 

const (
	ExecutionFinished        Type = "execution_finished"
	VerificationPending      Type = "verification_pending"
	VerificationFailed       Type = "verification_failed"
	VerificationInconclusive Type = "verification_inconclusive"
	VerificationStale        Type = "verification_stale"
	VerificationPassed       Type = "verification_passed"
	// ArtifactCreated announces the verified result artifact by ResultID. It
	// never identifies private evidence or grants access to result content.
	ArtifactCreated Type = "artifact_created"
	ResultCommitted Type = "result_committed"
)

type Watermark 

type Watermark struct {
	SchemaVersion int           `json:"schema_version"`
	RunID         id.AgentRunID `json:"run_id"`
	FirstSequence uint64        `json:"first_sequence"`
	LastSequence  uint64        `json:"last_sequence"`
	RunRevision   int64         `json:"run_revision"`
	ResultID      id.ID         `json:"result_id,omitzero"`
}

Watermark describes retention and identifies the current stored run snapshot. FirstSequence is LastSequence+1 when all earlier events have been pruned.

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.