runstate

package
v0.5.0 Latest Latest
Warning

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

 Go to latest Published: Jul 25, 2026 License: Apache-2.0 Imports: 16 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	VarResumePrompt      = "resume_prompt"
	VarResumeAgent       = "resume_agent"
	VarResumeTrustMode   = "resume_trust_mode"
	VarResumeEpisodeID   = "resume_episode_id"
	VarResumeTriggerKind = "resume_trigger_kind"
	VarResumeSessionID   = "resume_session_id"
	VarRunErrorMessage   = "run_error_message"
	VarExecutionPhase    = "execution_phase"
	VarCheckpointKind    = "checkpoint_kind"
	// VarRunLeaseOwner marks a run snapshot as lease-managed by stamping the
	// owning worker ID. MarkAbandonedRuns only reaps Running runs carrying
	// this marker, so runs executed by workers without lease coordination are
	// never mistaken for zombies.
	VarRunLeaseOwner = "run_lease_owner"
)

Snapshot variable keys shared by the framework facade and the runtime engine.

View Source 
const (
	ExecutionPhaseWorkflow   = "workflow"
	ExecutionPhaseAutonomous = "autonomous"
)

Hybrid / checkpoint phase values stored under VarExecutionPhase.

View Source 
const (
	CheckpointKindBeforeFinalAnswer = "before_final_answer"
	CheckpointKindToolApproval      = "tool_approval"
)

Checkpoint kind values stored under VarCheckpointKind.

View Source 
const MinTokenSecretLength = 16

MinTokenSecretLength is the minimum HMAC secret length NewTokenSigner and NewTokenSignerWithRotation accept. HITL resume tokens are bearer credentials, so short, guessable secrets are rejected at construction instead of weakening every run's approval gate.

Variables

View Source 
var (
	ErrNotFound      = errors.New("runstate: not found")
	ErrStaleSnapshot = errors.New("runstate: snapshot version is stale")
	// ErrStaleFence reports that a fenced save presented a fencing token lower
	// than the highest token already recorded for the run: the writer's run
	// lease was superseded by a newer holder and its writes must be rejected.
	// The engine treats it like a lost lease (stop executing immediately).
	ErrStaleFence = errors.New("runstate: fence token is stale")
	// ErrFenceRequired reports that the context carries a non-zero lease fence
	// token but the repository does not implement FencedRepository. Multi-node
	// deployments must use a fencing-capable repository when run leases are
	// enabled; plain Save is rejected instead of falling back.
	ErrFenceRequired     = errors.New("runstate: fenced repository required for leased run saves")
	ErrInvalidStatus     = errors.New("runstate: invalid status")
	ErrInvalidTransition = errors.New("runstate: invalid status transition")
	ErrTokenSuperseded   = errors.New("humangate: token superseded by newer version")
	// ErrResumeInProgress reports that another caller is already resuming or
	// continuing this run in this process. It replaces the ambiguous bare
	// ErrTokenSuperseded a concurrent resume used to surface when both callers
	// raced the same pause token. The framework facade re-exports it.
	ErrResumeInProgress = errors.New("humangate: resume already in progress for run")
)
View Source 
var (
	ErrTenantMismatch = errors.New("runstate: tenant mismatch")
	// ErrTenantRequired reports an access in tenant-strict mode where the
	// caller's context carries no tenant principal at all.
	ErrTenantRequired = errors.New("runstate: tenant identity required")
)
View Source 
var (
	ErrInvalidToken = errors.New("runstate: invalid token")
	// ErrTokenExpired classifies a well-formed but expired token. It wraps
	// ErrInvalidToken so existing errors.Is(err, ErrInvalidToken) checks keep
	// matching, while callers that care about expiry (e.g. HTTP 410 mapping)
	// can branch on the more specific sentinel.
	ErrTokenExpired = fmt.Errorf("%w: token expired", ErrInvalidToken)
)

Functions

func AuthorizeBlobAccess added in v0.4.5 

func AuthorizeBlobAccess(ctx context.Context, ref BlobRef) error

AuthorizeBlobAccess rejects scoped references owned by another tenant and rejects legacy unscoped references in tenant-strict mode.

func AuthorizeTenant added in v0.1.4 

func AuthorizeTenant(ctx context.Context, snapshot RunSnapshot) error

AuthorizeTenant ensures an authenticated principal can access the snapshot tenant. Legacy snapshots without tenant_id remain accessible when no principal is present. In tenant-strict mode (ContextWithTenantStrictMode) both cases fail closed: a missing principal yields ErrTenantRequired, an unstamped snapshot ErrTenantMismatch.

func CollectBlobRefs added in v0.1.4 

func CollectBlobRefs(snapshot RunSnapshot) map[string]BlobRef

CollectBlobRefs returns blob references referenced by a snapshot's step outputs and variables. Variables can carry externalized payloads (e.g. checkpoint state persisted by the runtime) serialized as StepOutputRef, so they must count as live references or orphan purges would delete blobs a paused run still needs to resume.

func CollectReferencedBlobIDs added in v0.1.4 

func CollectReferencedBlobIDs(snapshots []RunSnapshot) map[string]struct{}

CollectReferencedBlobIDs returns blob IDs still referenced by the given snapshots.

func ContextWithFenceToken added in v0.3.1 

func ContextWithFenceToken(ctx context.Context, token uint64) context.Context

ContextWithFenceToken stamps the fencing token of the currently held run lease (coordination.Lease.Token) onto the context, so every snapshot save made while the lease is held can present it to a FencedRepository. A zero token disables fencing, matching a context with no token at all.

func ContextWithStatusTransitionOverride added in v0.3.0 

func ContextWithStatusTransitionOverride(ctx context.Context) context.Context

ContextWithStatusTransitionOverride allows explicit restore/rerun paths to reopen terminal snapshots while ordinary saves still enforce transitions.

func ContextWithTenantStrictMode added in v0.3.0 

func ContextWithTenantStrictMode(ctx context.Context) context.Context

ContextWithTenantStrictMode marks the context so AuthorizeTenant (and thus LoadAuthorized) rejects access when the caller has no tenant principal, or when the snapshot predates tenant stamping (empty TenantID). It is off by default for backward compatibility: without it, principal-less contexts and legacy unowned snapshots are accessible to anyone. Multi-tenant deployments should wrap every request context with this (e.g. in the auth middleware) so an unauthenticated or unstamped access path fails closed.

func FenceTokenFromContext added in v0.3.1 

func FenceTokenFromContext(ctx context.Context) uint64

FenceTokenFromContext returns the fencing token attached by ContextWithFenceToken, or 0 when the caller holds no lease (or the lease predates fencing).

func GenerateRunID added in v0.3.0 

func GenerateRunID() string

GenerateRunID returns a cryptographically random run identifier with a "run-" prefix and 128 bits of entropy. It is the single canonical implementation shared by the framework facade, the runtime engine, the event router, and the async HTTP adapter (which previously each carried a private 64-bit copy). It falls back to a nanosecond timestamp on the rare occasion that the random reader fails.

func HydrateStepContext added in v0.1.3 

func HydrateStepContext(ctx context.Context, blobs BlobStore, outputs map[string]StepOutputRef) (json.RawMessage, error)

HydrateStepContext loads inline and blob-backed step outputs into a JSON object shaped as {"steps":{"<node_id>":...}} suitable for hybrid run context.

func IndexedThreadID added in v0.2.0 

func IndexedThreadID(snapshot RunSnapshot) string

IndexedThreadID returns the value stored in thread indexes for a snapshot.

func LoadStepOutput added in v0.1.3 

func LoadStepOutput(ctx context.Context, blobs BlobStore, ref StepOutputRef) (json.RawMessage, error)

LoadStepOutput resolves a step output reference from inline or blob storage.

func NormalizeSnapshot added in v0.3.0 

func NormalizeSnapshot(snapshot *RunSnapshot)

NormalizeSnapshot ensures maps are non-nil after JSON deserialize omitempty.

func ParseBlobID added in v0.4.5 

func ParseBlobID(id string) (tenantHash, digest string, legacy bool, err error)

ParseBlobID accepts legacy 64-hex content IDs and tenant-scoped 128-hex IDs. It returns the tenant hash (empty for legacy) and the content digest.

func PurgeOrphanBlobs added in v0.1.4 

func PurgeOrphanBlobs(ctx context.Context, runs Repository, blobs BlobStore, filter ListFilter) (int, error)

PurgeOrphanBlobs deletes blobs that are not referenced by any listed run snapshot.

func ResolveThreadID added in v0.2.0 

func ResolveThreadID(snapshot RunSnapshot) string

ResolveThreadID returns the thread group identifier for a snapshot.

func SaveWithFence added in v0.3.1 

func SaveWithFence(ctx context.Context, repo Repository, snapshot *RunSnapshot, expectedVersion int64) (fellBack bool, err error)

SaveWithFence persists a run snapshot with lease fencing when the context carries a fence token and the repository implements FencedRepository; a stale token fails with ErrStaleFence, which callers must not retry. When the context carries no token it behaves exactly like Repository.Save.

When the repository does not implement FencedRepository and the context carries a non-zero fence token, SaveWithFence fails with ErrFenceRequired instead of falling back to plain Save.

func StampSnapshot added in v0.1.4 

func StampSnapshot(snapshot *RunSnapshot, previous *RunSnapshot, now time.Time)

StampSnapshot preserves CreatedAt from a previous snapshot and refreshes UpdatedAt.

func StampTenant added in v0.1.4 

func StampTenant(ctx context.Context, snapshot *RunSnapshot)

StampTenant assigns the principal tenant to new snapshots when present in ctx.

func StatusTransitionOverrideFromContext added in v0.3.0 

func StatusTransitionOverrideFromContext(ctx context.Context) bool

StatusTransitionOverrideFromContext reports whether transition checks should allow an otherwise invalid status change for this save operation.

func TenantStrictModeFromContext added in v0.3.0 

func TenantStrictModeFromContext(ctx context.Context) bool

TenantStrictModeFromContext reports whether ctx is tenant-strict.

func ValidateStatusTransition added in v0.3.0 

func ValidateStatusTransition(ctx context.Context, previous *RunSnapshot, next RunStatus) error

ValidateStatusTransition rejects status changes that do not follow the run lifecycle, unless ctx carries an explicit restore/rerun override.

Types

type BlobAdmin added in v0.1.4 

type BlobAdmin interface {
	BlobStore
	List(ctx context.Context) ([]BlobRef, error)
	Delete(ctx context.Context, ref BlobRef) error
}

BlobAdmin extends BlobStore with listing and deletion for retention workflows.

type BlobRef 

type BlobRef struct {
	ID     string `json:"id"`
	Size   int64  `json:"size"`
	Sha256 string `json:"sha256"`
}

func FilterBlobRefsForContext added in v0.4.5 

func FilterBlobRefsForContext(ctx context.Context, refs []BlobRef) ([]BlobRef, error)

FilterBlobRefsForContext applies the same access boundary to BlobAdmin.List as Get/Delete. Principal-less non-strict callers retain the global admin view.

func FilterBlobRefsForTenant added in v0.4.5 

func FilterBlobRefsForTenant(refs []BlobRef, tenantID string) []BlobRef

FilterBlobRefsForTenant keeps only references physically scoped to tenantID. Legacy references are deliberately excluded from tenant-local retention, because their owner cannot be proven.

func NewBlobRef 

func NewBlobRef(id string, data []byte) BlobRef

func NewBlobRefForContext added in v0.4.5 

func NewBlobRefForContext(ctx context.Context, data []byte) (BlobRef, error)

NewBlobRefForContext creates a content-addressed reference scoped to the authenticated tenant. Principal-less non-strict callers retain the legacy global digest ID for backward compatibility and administrative workflows.

type BlobStore 

type BlobStore interface {
	Put(ctx context.Context, data []byte) (BlobRef, error)
	Get(ctx context.Context, ref BlobRef) ([]byte, error)
}

type CheckpointHistory added in v0.2.0 

type CheckpointHistory interface {
	Append(ctx context.Context, snapshot RunSnapshot) error
	List(ctx context.Context, runID string, limit int) ([]CheckpointSummary, error)
	Load(ctx context.Context, runID string, version int64) (RunSnapshot, error)
	// Delete drops every recorded revision for a run. Retention purges call it
	// alongside Repository.Delete so time-travel data does not outlive the run
	// it belongs to. Implementations must treat a missing run as a no-op.
	Delete(ctx context.Context, runID string) error
}

CheckpointHistory stores immutable run snapshot revisions for time-travel.

type CheckpointSummary added in v0.2.0 

type CheckpointSummary struct {
	RunID         string    `json:"run_id"`
	Version       int64     `json:"version"`
	Status        RunStatus `json:"status"`
	CurrentNodeID string    `json:"current_node_id,omitempty"`
	StepCount     int       `json:"step_count"`
	RecordedAt    time.Time `json:"recorded_at"`
}

CheckpointSummary describes one append-only run snapshot revision.

type EventOutboxRepository added in v0.3.1 

type EventOutboxRepository interface {
	Repository
	SaveWithEvents(ctx context.Context, snapshot *RunSnapshot, expectedVersion int64, events []core.Event, fenceToken uint64) error
}

EventOutboxRepository is an optional Repository extension that persists a run snapshot and a batch of pending events in ONE transaction: the snapshot update (with the same version CAS and fence validation as SaveFenced) and the outbox inserts either both commit or both roll back, so a run's state and its lifecycle events can never diverge. A zero fenceToken behaves like Save; a non-zero token behaves like SaveFenced and fails with ErrStaleFence when a newer lease holder has written.

Only PostgreSQL runstate implements this today; the outbox lives in the same database as the snapshots so a single transaction can cover both.

type FencedRepository added in v0.3.1 

type FencedRepository interface {
	Repository
	SaveFenced(ctx context.Context, snapshot *RunSnapshot, expectedVersion int64, fenceToken uint64) error
}

FencedRepository is an optional Repository extension for multi-node deployments: SaveFenced behaves like Save but additionally requires fenceToken to be at least the highest fencing token already recorded for the run. A stale token fails with ErrStaleFence (a newer lease holder has taken over); a version mismatch still fails with ErrStaleSnapshot, so callers can distinguish "someone else holds the run" from "my view is out of date". The fence token comes from coordination.Lease.Token.

Save remains fence-agnostic: existing single-node flows are unaffected.

type ListFilter added in v0.1.1 

type ListFilter struct {
	// Status, when non-empty, restricts results to snapshots with this status.
	Status RunStatus
	// ScenarioName, when non-empty, restricts results to a specific scenario.
	ScenarioName string
	// TenantID, when non-empty, restricts results to a specific tenant.
	TenantID string
	// ParentRunID, when non-empty, restricts results to forks of a parent run.
	ParentRunID string
	// ThreadID, when non-empty, restricts results to a fork/thread group.
	ThreadID string
	// UpdatedBefore, when non-zero, keeps only snapshots with UpdatedAt strictly
	// before this instant (useful for retention purge pushdown).
	UpdatedBefore time.Time
	// Limit is the maximum number of results to return. 0 means no limit.
	Limit int
}

ListFilter controls which snapshots are returned by Repository.List. Zero values are treated as "no filter" for the corresponding field.

func ScopeListFilter added in v0.5.0 

func ScopeListFilter(ctx context.Context, filter ListFilter) (ListFilter, error)

ScopeListFilter binds a repository list operation to the authenticated tenant. Authenticated callers cannot select a different tenant; internal maintenance callers without a principal retain the explicitly requested scope unless tenant-strict mode is enabled.

type OutboxEvent added in v0.3.1 

type OutboxEvent struct {
	ID        int64
	Sequence  int64
	Event     core.Event
	CreatedAt time.Time
}

OutboxEvent is one row of the transactional event outbox: a runtime event parked in the run-state database (migration 0005, agentflow_outbox) until the framework relay delivers it to the durable event store. Sequence is the event's per-run sequence in the event store, minted when the row was written; the relay redelivers with this exact sequence so the event store's UNIQUE (run_id, sequence) constraint deduplicates retries.

type OutboxRepository added in v0.3.1 

type OutboxRepository interface {
	Repository
	// FetchUnpublishedOutbox returns up to limit unpublished outbox rows in
	// insertion order. limit <= 0 means an implementation-defined default.
	FetchUnpublishedOutbox(ctx context.Context, limit int) ([]OutboxEvent, error)
	// MarkOutboxPublished marks the row published_at=NOW() when it is still
	// unpublished. An already-published or missing row is not an error: a
	// concurrent relay simply won the race.
	MarkOutboxPublished(ctx context.Context, id int64) error
	// DeleteOutboxForRun removes every outbox row of a run (published or
	// not); used by the retention cascade when the run itself is deleted.
	DeleteOutboxForRun(ctx context.Context, runID string) (int64, error)
	// PurgeOutboxPublishedBefore removes published rows older than cutoff,
	// scoped to the authenticated tenant when the context carries one.
	// Unpublished rows are never purged here: they are undelivered events,
	// not garbage.
	PurgeOutboxPublishedBefore(ctx context.Context, cutoff time.Time) (int64, error)
}

OutboxRepository is an optional Repository extension backing the framework's outbox relay and retention cascade. FetchUnpublishedOutbox returns parked events in insertion order (id ASC); the relay delivers each to the durable event store and then calls MarkOutboxPublished, which marks conditionally (WHERE published_at IS NULL) so concurrent relays on several nodes cannot double-mark — delivery itself is idempotent through the event store's UNIQUE (run_id, sequence) constraint.

type Repository 

type Repository interface {
	Save(ctx context.Context, snapshot *RunSnapshot, expectedVersion int64) error
	Load(ctx context.Context, runID string) (RunSnapshot, error)
	Delete(ctx context.Context, runID string) error
	// List returns snapshots that match the filter. Implementations may
	// return results in any order. An empty filter matches all snapshots.
	List(ctx context.Context, filter ListFilter) ([]RunSnapshot, error)
}

type RunSnapshot 

type RunSnapshot struct {
	RunID           string                     `json:"run_id"`
	Version         int64                      `json:"version"`
	ScenarioName    string                     `json:"scenario_name"`
	TenantID        string                     `json:"tenant_id,omitempty"`
	CurrentNodeID   string                     `json:"current_node_id,omitempty"`
	Status          RunStatus                  `json:"status"`
	Variables       map[string]json.RawMessage `json:"variables,omitempty"`
	StepOutputs     map[string]StepOutputRef   `json:"step_outputs,omitempty"`
	PendingGate     *core.CheckpointState      `json:"pending_gate,omitempty"`
	ParentRunID     string                     `json:"parent_run_id,omitempty"`
	ForkFromVersion int64                      `json:"fork_from_version,omitempty"`
	ThreadID        string                     `json:"thread_id,omitempty"`
	CreatedAt       time.Time                  `json:"created_at,omitempty"`
	UpdatedAt       time.Time                  `json:"updated_at,omitempty"`
}

func LoadAuthorized added in v0.1.4 

func LoadAuthorized(ctx context.Context, repo Repository, runID string) (RunSnapshot, error)

LoadAuthorized loads a snapshot and enforces tenant access when a principal is present.

func (RunSnapshot) Validate 

func (s RunSnapshot) Validate() error

type RunStatus 

type RunStatus string
const (
	RunStatusRunning   RunStatus = "running"
	RunStatusPaused    RunStatus = "paused"
	RunStatusFailed    RunStatus = "failed"
	RunStatusCompleted RunStatus = "completed"
	RunStatusCancelled RunStatus = "cancelled"
)

func (RunStatus) CanTransitionTo 

func (s RunStatus) CanTransitionTo(next RunStatus) bool

func (RunStatus) Valid 

func (s RunStatus) Valid() bool

type StaleRepository added in v0.3.1 

type StaleRepository interface {
	Repository
	ListStale(ctx context.Context, filter ListFilter, grace time.Duration) ([]RunSnapshot, error)
}

StaleRepository is an optional Repository extension that pushes the "updated longer ago than grace" test down to the store, so the comparison uses the store's own clock (PostgreSQL NOW()) instead of the caller's — application-clock skew cannot prematurely reap or forever skip runs.

ListStale returns snapshots matching filter whose last update is at least grace in the past (snapshots with a zero UpdatedAt are always considered stale, matching the reaper's historical fallback semantics).

type StepOutputRef 

type StepOutputRef struct {
	Inline json.RawMessage `json:"inline,omitempty"`
	Blob   *BlobRef        `json:"blob,omitempty"`
}

type TokenPayload 

type TokenPayload struct {
	RunID     string    `json:"run_id"`
	Version   int64     `json:"version"`
	ExpiresAt time.Time `json:"expires_at,omitempty"`
}

type TokenSigner 

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

TokenSigner signs and verifies HITL resume tokens. A single-key signer (NewTokenSigner) keeps the legacy two-segment token format; a rotation signer (NewTokenSignerWithRotation) embeds a key id so in-flight tokens survive a key swap: new tokens are signed with the primary key, and both primary and secondary keys verify.

func NewTokenSigner 

func NewTokenSigner(secret []byte) (*TokenSigner, error)

func NewTokenSignerWithRotation added in v0.3.0 

func NewTokenSignerWithRotation(primary, secondary []byte) (*TokenSigner, error)

NewTokenSignerWithRotation creates a signer that signs with primary and verifies with both primary and secondary. To rotate: deploy with (newSecret, oldSecret), wait for every in-flight token signed by oldSecret to expire or be consumed, then redeploy with (newSecret) only — no token is invalidated mid-flight. Both secrets must meet MinTokenSecretLength.

func (*TokenSigner) Sign 

func (s *TokenSigner) Sign(payload TokenPayload) (string, error)

func (*TokenSigner) Verify 

func (s *TokenSigner) Verify(token string) (TokenPayload, error)

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.