sessionruntime

type ActiveRunUpdate func(snapshot Snapshot, now time.Time) (Snapshot, bool, error)

type Admission struct {
	RunID string
	// TurnID and TurnPosition are decided at admission so the eventual terminal
	// write targets a turn that already exists on paper.
	TurnID       string
	TurnPosition int64
	State        ledger.State
	// Replay is true when this invocation was already admitted. The rest of the
	// struct then describes the original run, whatever state it has reached.
	Replay bool
	// Started is true when this call took ownership and began execution. It is
	// false for a replay of a run someone else owns and for a replay of a
	// finished run.
	Started bool
	// Cursor is the position at which this run became observable. A caller that
	// both starts runs and subscribes to the session uses it to place the run in
	// the stream it is already reading, rather than guessing whether a snapshot
	// it holds predates its own submission (SR-OBS-002).
	//
	// It is set only when Started. A replay reserved nothing, so this call
	// produced no position; where the session stands now is a different fact,
	// and the subscription's snapshot is the authoritative answer to it.
	Cursor Cursor
	// Handle names the run this process now owns, including the fencing token
	// its durable writes must carry. It is meaningful only when Started.
	Handle RunHandle
}

type AdmitInput struct {
	BotID     string
	SessionID string
	// InvocationID is the caller's retry identity. Callers that cannot mint one
	// must derive it deterministically from their source (for example a channel
	// message id), because a fresh id per attempt turns a retry into a second
	// run.
	InvocationID string
	// Payload is the canonically encoded submission. Its fingerprint decides
	// whether a repeated invocation_id is the same submit or a conflict, so the
	// encoding must be stable for equal inputs — same key order, no timestamps,
	// no request ids.
	Payload []byte
	// Execution is the owner-side plumbing used only if this call wins the
	// claim. A replay that finds the run already owned never touches it.
	Execution Execution
}

type Backend interface {
	Now(ctx context.Context) (time.Time, error)
	// Load returns a snapshot the caller owns and may freely mutate.
	Load(ctx context.Context, key Key) (Snapshot, bool, error)
	Update(ctx context.Context, key Key, update SnapshotUpdate) (Snapshot, bool, error)
	Publish(ctx context.Context, event Event) error
	Subscribe(ctx context.Context, key Key) (Subscription, error)
	DecisionOutputStore
	Close() error
}

type Command struct {
	Type         string `json:"type"`
	ID           string `json:"id,omitempty"`
	ReplyOwnerID string `json:"reply_owner_id,omitempty"`
	BotID        string `json:"bot_id"`
	SessionID    string `json:"session_id"`
	RunID        string `json:"run_id"`
	Generation   string `json:"generation"`
	FencingToken int64  `json:"fencing_token,omitempty"`
	TargetID     string `json:"target_id,omitempty"`
	// DecisionResolved means the command target was resolved from PostgreSQL
	// before routing. Owner-side execution must not consult the live UI
	// projection again: it is derived state and may lag the durable decision.
	DecisionResolved bool            `json:"decision_resolved,omitempty"`
	Payload          json.RawMessage `json:"payload,omitempty"`
	PayloadHash      string          `json:"payload_hash,omitempty"`
	ErrorCode        string          `json:"error_code,omitempty"`
	Error            string          `json:"error,omitempty"`
	CreatedAt        time.Time       `json:"created_at"`
	ExpiresAt        time.Time       `json:"expires_at,omitempty"`

	// StreamOutput is fixed at admission and travels to the owner with the command.
	// It must not depend on subscriber liveness: disconnecting cannot change a run.
	StreamOutput bool `json:"stream_output,omitempty"`
}

type CommandSubscription struct {
	C     <-chan Command
	Close func()
}

type CurrentRunPatch struct {
	RunID               string     `json:"run_id"`
	Status              *string    `json:"status,omitempty"`
	ErrorCode           *string    `json:"error_code,omitempty"`
	Error               *string    `json:"error,omitempty"`
	UpdatedAt           *time.Time `json:"updated_at,omitempty"`
	OwnerLeaseExpiresAt *time.Time `json:"owner_lease_expires_at,omitempty"`
}

type CurrentRunView struct {
	// ConfigurationOnly holds the session lock but does not generate a reply.
	ConfigurationOnly bool   `json:"configuration_only,omitempty"`
	RunID             string `json:"run_id" validate:"required" format:"uuid"`
	// TurnID is the durable turn this run writes into, allocated at admission.
	// It is part of the observable view because SR-OBS-003 requires every
	// subscriber to agree on the run's turn, and a subscriber that only learns
	// the run id cannot line the run up against persisted history.
	TurnID string `json:"turn_id" validate:"required" format:"uuid"`
	// TurnPosition is the immutable turn-level sequence admission drew for
	// TurnID from bot_sessions.next_turn_position. It rides the live view for
	// the same reason TurnID does: a subscriber that learns only the turn's name
	// still cannot order it against persisted history, and SR-TURN-001 forbids
	// falling back to timestamps to decide where a turn belongs.
	TurnPosition int64 `json:"turn_position,omitempty"`
	// InvocationID is the caller-supplied intent identity recorded at admission
	// (session_runs.invocation_id). It rides the live view so the client that
	// originated the send can match projection frames to its optimistic turn by
	// reading the frame, instead of inferring the pairing from arrival timing
	// while waiting for the acceptance that names the two. Subscribers that did
	// not originate the run see an id unknown to them and treat the turn as
	// foreign — which is the correct standalone rendering for cross-device runs.
	InvocationID        string               `json:"invocation_id,omitempty"`
	Generation          string               `json:"generation"`
	Status              string               `json:"status"`
	OwnerID             string               `json:"owner_id,omitempty"`
	OwnerLeaseExpiresAt *time.Time           `json:"owner_lease_expires_at,omitempty"`
	StartedAt           time.Time            `json:"started_at"`
	UpdatedAt           time.Time            `json:"updated_at"`
	Messages            []chatview.UIMessage `json:"messages"`
	// UserTurns is the authoritative ordered set of user inputs already
	// admitted into this run, including the original input and applied steers.
	// The legacy request_user_turn is derived only at the JSON boundary.
	UserTurns []chatview.UITurn `json:"user_turns,omitempty"`
	// SteerSupported is published only by an installed step-boundary consumer.
	// Missing on old owners and on runtimes without that execution capability.
	SteerSupported bool `json:"steer_supported,omitempty"`
	// The snapshot outlives the lease key and retains the exact persistence
	// fence needed to reconcile a durable terminal after owner expiry.
	FencingToken int64 `json:"fencing_token,omitempty"`
	// SteerTurns locates live queue inputs inside the run's assistant message
	// stream. Claimed entries are provisional runtime state; applied entries
	// point at the history turn written by the application.
	SteerTurns             []SteerTurnView   `json:"steer_turns,omitempty"`
	ErrorCode              string            `json:"error_code,omitempty"`
	Error                  string            `json:"error,omitempty"`
	ProposedTerminalStatus string            `json:"proposed_terminal_status,omitempty"`
	FinishProposedAt       *time.Time        `json:"finish_proposed_at,omitempty"`
	Operation              *RunOperationView `json:"operation,omitempty"`
}

type Cursor struct {
	Epoch string `json:"epoch,omitempty"`
	Seq   int64  `json:"seq"`
}

type DecisionFenceActivator interface {
	ReclaimWaitingDecision(
		ctx context.Context,
		botID, sessionID, runID, ownerID, liveGeneration string,
		previousToken, newToken int64,
		decisions []runtimefence.PreservedDecision,
	) error
}

type DecisionOutputLimits struct {
	MaxBytes  int
	MaxEvents int
}

type DecisionOutputPage struct {
	DecisionOutputState
	Events []json.RawMessage
}

type DecisionOutputRef struct {
	BotID     string
	CommandID string
}

type DecisionOutputState struct {
	Exists  bool
	Length  int
	Bytes   int
	Done    bool
	Failed  bool
	Claimed bool
	// Applied reports whether this append changed the log. Replays of an
	// already-committed seq and writes after a terminal marker are no-ops.
	Applied bool
	// Exceeded reports that this append tripped the limits and failed the log.
	Exceeded bool
}

type DecisionOutputStore interface {
	AppendDecisionOutput(ctx context.Context, ref DecisionOutputRef, seq int64, payload json.RawMessage, limits DecisionOutputLimits) (DecisionOutputState, error)
	ReadDecisionOutput(ctx context.Context, ref DecisionOutputRef, from int) (DecisionOutputPage, error)
	ClaimDecisionOutput(ctx context.Context, ref DecisionOutputRef) (bool, error)
	ReleaseDecisionOutput(ctx context.Context, ref DecisionOutputRef) error
}

type DecisionResponse struct {
	ControlID  string
	Type       string
	DecisionID string
	BotID      string
	SessionID  string
	RunID      string
	Payload    json.RawMessage
	// contains filtered or unexported fields
}

type DecisionResponseResult struct {
	SessionID  string
	Generation string
	RunID      string
	Handled    bool
	Applied    bool

	// Replayed acknowledges an earlier submission without rerunning its output.
	Replayed bool
}

type DecisionStore interface {
	ResolveRuntimeDecision(ctx context.Context, commandType, decisionID string) (DecisionTarget, error)
	PendingRuntimeDecisions(ctx context.Context, runID string) ([]DecisionTarget, error)
}

type DecisionTarget struct {
	Type         string
	ID           string
	BotID        string
	SessionID    string
	RunID        string
	TurnID       string
	Status       string
	FencingToken int64
	ControlID    string
	PayloadHash  string
	// SessionRuntime is the session's runtime type. Recovery needs it to
	// tell a native parked run (resumable: the decision continuation is
	// rebuilt from the database) from an inline waiter run (codex, claude,
	// ACP), whose blocked turn died with its owner and cannot be resumed.
	SessionRuntime string
}

type DistributedBackend interface {
	Backend
	UpdateActiveRun(ctx context.Context, key Key, runID, generation string, update ActiveRunUpdate) (Snapshot, bool, error)
	StartRun(ctx context.Context, key Key, ref RunRef, update SnapshotUpdate) (Snapshot, bool, error)
	ReleaseRun(ctx context.Context, key Key, ref RunRef, update ActiveRunUpdate) (Snapshot, bool, error)
	// ReconcileTerminalRun applies an authoritative durable terminal outcome to
	// the matching live reservation even after its lease expired. The fencing
	// token is mandatory so a stale reaper cannot release a successor.
	ReconcileTerminalRun(ctx context.Context, key Key, ref RunRef, update ActiveRunUpdate) (Snapshot, bool, error)
	RenewLease(ctx context.Context, key Key, runID, ownerID, generation string, renewedAt, expiresAt time.Time) error
	ValidateRunOwnership(ctx context.Context, key Key, ref RunRef) error
	LoadRunRef(ctx context.Context, key Key, runID string) (RunRef, bool, error)
	DeleteRunRef(ctx context.Context, ref RunRef) (bool, error)
	PublishCommand(ctx context.Context, ownerID string, command Command) error
	SubscribeCommands(ctx context.Context, ownerID string) (CommandSubscription, error)
	StoreCommandResult(ctx context.Context, result Command, ttl time.Duration) error
	LoadCommandResult(ctx context.Context, commandID string) (Command, bool, error)
}

type Event struct {
	Type      string        `json:"type"`
	BotID     string        `json:"bot_id"`
	SessionID string        `json:"session_id"`
	Epoch     string        `json:"epoch,omitempty"`
	RunID     string        `json:"run_id,omitempty"`
	Seq       int64         `json:"seq"`
	UpdatedAt *time.Time    `json:"updated_at,omitempty"`
	Snapshot  *Snapshot     `json:"snapshot,omitempty"`
	Delta     *RuntimeDelta `json:"delta,omitempty"`
	Message   string        `json:"message,omitempty"`
}

type Execution struct {
	// ConfigurationOnly keeps the execution lock without presenting an assistant turn.
	ConfigurationOnly bool
	// Admission persists the run's user turn and any replacement operation, and
	// returns the view subscribers should see when the run leaves admitting. It
	// runs after ownership is established, so its writes can be fenced with
	// RunHandle.FencingToken.
	Admission func(ctx context.Context, handle RunHandle) (RunAdmissionView, error)
	// AbortCh and Cancel are how a routed abort reaches this run. A run without
	// them is durably abortable but cannot be interrupted mid-turn.
	AbortCh chan<- struct{}
	Cancel  context.CancelFunc
	// InjectCh receives steering messages; nil means this caller does not
	// support steering. The caller owns channel closure; the runtime only stops
	// sending during teardown.
	InjectCh chan<- turn.InjectMessage
	// OwnershipCancel revokes the caller's execution context when the lease is
	// lost, so a superseded owner stops producing output rather than racing the
	// fence it can no longer pass.
	OwnershipCancel context.CancelCauseFunc
}

type FenceActivator interface {
	Activate(ctx context.Context, botID, sessionID string, token int64) error
}

type FollowUpClaimRef struct {
	ItemID       FollowUpItemID
	TriggerRunID string
	ClaimToken   string
}

type FollowUpItem struct {
	ID                  FollowUpItemID
	BotID, SessionID    string
	EnqueuedDuringRunID string
	InvocationID        string
	Payload             []byte
	Status              QueueStatus
	Position            int64
	Claim               *FollowUpClaimRef
	// ErrorCode is set when Status is rejected.
	ErrorCode string
	CreatedAt time.Time
}

type FollowUpItemID string

type FollowUpPendingRef struct {
	ItemID FollowUpItemID `json:"item_id"`
}

type HistoryResetBackend interface {
	AcquireHistoryReset(ctx context.Context, scope ResetScope, token string, ttl time.Duration) (ResetLease, bool, error)
	RenewHistoryReset(ctx context.Context, lease ResetLease, ttl time.Duration) (ResetLease, bool, error)
	ReleaseHistoryReset(ctx context.Context, lease ResetLease) (bool, error)
	EffectiveHistoryReset(ctx context.Context, scope ResetScope) (ResetLease, bool, error)
}

type HistoryResetHandler func(context.Context, ResetScope) error

type Key struct {
	BotID     string `json:"bot_id"`
	SessionID string `json:"session_id"`
}

type LeaseCandidate struct {
	Key          Key
	RunID        string
	FencingToken int64
	// ExpiresAt is the deadline recorded in the index, sampled on the backend's
	// clock rather than the reaper's.
	ExpiresAt time.Time
}

type LiveQueueBackend interface {
	EnqueueSteer(context.Context, Key, string, string, []byte) (SteerItem, error)
	EnqueueFollowUp(context.Context, Key, string, string, []byte) (FollowUpItem, error)
	PendingQueues(context.Context, Key, int) ([]SteerItem, []FollowUpItem, error)
	ReorderSteer(context.Context, Key, SteerPendingRef, SteerPendingRef) ([]SteerItem, error)
	ReorderFollowUp(context.Context, Key, FollowUpPendingRef, FollowUpPendingRef) ([]FollowUpItem, error)
	UpdateSteer(context.Context, Key, SteerItemID, []byte) (SteerItem, error)
	UpdateFollowUp(context.Context, Key, FollowUpItemID, []byte) (FollowUpItem, error)
	CancelSteer(context.Context, Key, SteerItemID) error
	CancelFollowUp(context.Context, Key, FollowUpItemID) error
	PromoteFollowUpToSteer(context.Context, Key, FollowUpPendingRef) (PromoteFollowUpResult, error)
	ClaimNextSteer(context.Context, RunHandle, bool) (SteerItem, SteerClaimRef, bool, error)
	ApplySteer(context.Context, Key, SteerClaimRef) error
	ReleaseSteer(context.Context, Key, SteerClaimRef) error
	// CloseSteerRun rejects every accepted or claimed steer that targets the
	// given run and seals the run against later steer admission. It is called
	// once the run is durably terminal; a steer is bound to its run and has no
	// meaning for any later run of the session.
	CloseSteerRun(context.Context, Key, string) error
	ClaimNextFollowUp(context.Context, Key, string) (FollowUpItem, FollowUpClaimRef, bool, error)
	ApplyFollowUp(context.Context, Key, FollowUpClaimRef) error
	ReleaseFollowUp(context.Context, Key, FollowUpClaimRef) error
}

type LivenessBackend interface {
	// LivenessGeneration identifies this incarnation of the live backend. It is
	// created once and read by everyone, so a value different from the one
	// stamped on a run means the backend that claimed that run is gone and its
	// in-flight state cannot be recovered.
	LivenessGeneration(ctx context.Context) (string, error)

	// ExpiredLeaseCandidates returns up to limit entries whose lease deadline
	// has passed. The backend samples its own clock: a reaper with a skewed
	// clock must not be able to declare a healthy owner dead. Candidates are
	// returned, not claimed — the leader is single, so a visibility timeout
	// would add a second failure mode without removing one.
	ExpiredLeaseCandidates(ctx context.Context, limit int64) ([]LeaseCandidate, error)

	// ReleaseLeaseCandidate removes an index entry, but only while the entry
	// still carries the token the candidate was read with. A run reclaimed
	// between read and release keeps its new entry.
	ReleaseLeaseCandidate(ctx context.Context, candidate LeaseCandidate) (bool, error)

	// AcquireLeaderLease elects the single cluster-wide reaper. Renewal is the
	// same call: an existing holder re-acquires its own lease. Failover is safe
	// because every reaper duty is an idempotent fenced transition, so a new
	// leader repeating work the old one had started changes nothing.
	AcquireLeaderLease(ctx context.Context, ownerID string, ttl time.Duration) (bool, error)

	// ReleaseLeaderLease hands leadership back on graceful shutdown so a peer
	// takes over in one tick instead of one lease period.
	ReleaseLeaderLease(ctx context.Context, ownerID string) error
}

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

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

type Options struct {
	OwnerID       string
	StateTTL      time.Duration
	OwnerLeaseTTL time.Duration
	// BackendLossGrace comes from config because a Redis restart budget depends
	// on the deployment. Everything else below OwnerLeaseTTL is derived.
	BackendLossGrace time.Duration
	// Cluster mirrors session_runtime.cluster.
	Cluster bool
	// Ledger is the durable session run store.
	Ledger ledger.Store
	// Fence applies the persistence-ownership cutover for a claimed run. It is
	// required together with Ledger.
	Fence                  FenceActivator
	CommandAckTTL          time.Duration
	CommandWorkerLimit     int
	Logger                 *slog.Logger
	EpochGenerator         func() string
	RunGenerationGenerator func() string

	// ScanBatchSize and MaxScanBatchesPerTick exist so tests can watch recovery
	// page rather than to be tuned in production; both default to package
	// constants.
	ScanBatchSize         int32
	MaxScanBatchesPerTick int
	// contains filtered or unexported fields
}

type PromoteFollowUpResult struct {
	FollowUp FollowUpPendingRef
	Steer    SteerItem
}

type QueueStatus string

type QueueUserTurnUpdate struct {
	PersistedTurns        []chatview.UITurn
	AppliedSteerItemID    string
	AppliedSteerTurn      *chatview.UITurn
	ClaimedSteerItemID    string
	ClaimedSteerText      string
	ClaimedSteerTimestamp time.Time
	// ClaimedSteerTurn* is the turn slot drawn for this input at claim time. It
	// is the same identity the step commit files the user row under, so the live
	// bubble and the settled one are one turn from the start.
	ClaimedSteerTurnID       string
	ClaimedSteerTurnPosition int64
	// AfterStepIndex, when set, is the durable step whose output must already
	// be in the live projection before a claimed steer is anchored. The commit
	// barrier runs on the model loop while agent events are consumed on
	// another goroutine, so without this wait the anchor could be computed
	// from a projection still missing the tail of the step, and the steer
	// would render above output that preceded it.
	AfterStepIndex *int
}

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

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

type RedisOptions struct {
	URL       string
	KeyPrefix string
	StateTTL  time.Duration
}

type ResetLease struct {
	Scope     ResetScope `json:"scope"`
	Token     string     `json:"token"`
	ExpiresAt time.Time  `json:"expires_at"`
}

type ResetScope struct {
	BotID     string `json:"bot_id"`
	SessionID string `json:"session_id,omitempty"`
}

type RunAdmissionView struct {
	RequestUserTurn *chatview.UITurn
	Operation       *RunOperationView
}

type RunHandle struct {
	BotID     string
	SessionID string
	RunID     string
	// OwnerID identifies the live execution owner for queue claims. It is
	// intentionally carried with the handle so claim CAS checks use the same
	// owner identity as the session runtime.
	OwnerID    string
	TurnID     string
	Generation string
	// FencingToken is the ledger ownership token for this run. Callers need it
	// to fence their own durable writes, which is why it travels with the
	// handle rather than staying inside the runtime. It is zero for runs
	// created by backend-only reservation tests.
	FencingToken int64
}

type RunOperationView struct {
	Kind                 string           `json:"kind" validate:"required" enums:"retry,edit"`
	ReplaceFromMessageID string           `json:"replace_from_message_id" validate:"required"`
	ReplacementUserTurn  *chatview.UITurn `json:"replacement_user_turn,omitempty"`
}

type RunRef struct {
	BotID      string `json:"bot_id"`
	SessionID  string `json:"session_id"`
	RunID      string `json:"run_id"`
	OwnerID    string `json:"owner_id"`
	Generation string `json:"generation"`
	// FencingToken is the durable ownership token this reservation was made
	// with. It is stored with the ref so the lease index entry can be written
	// and removed from the backend's own copy, without a caller having to
	// reconstruct a token it may no longer hold. Zero means the run has no
	// ledger identity and therefore nothing for the reaper to transition.
	FencingToken int64 `json:"fencing_token,omitempty"`
}

type RuntimeDelta struct {
	CurrentRunView    *CurrentRunView         `json:"current_run_view,omitempty"`
	Run               *CurrentRunPatch        `json:"run,omitempty"`
	UserTurnUpserts   []chatview.UITurn       `json:"user_turn_upserts,omitempty"`
	SteerTurnUpserts  []SteerTurnView         `json:"steer_turn_upserts,omitempty"`
	SteerTurnRemovals []string                `json:"steer_turn_removals,omitempty"`
	MessageAppends    []RuntimeMessageAppend  `json:"message_appends,omitempty"`
	ProgressAppends   []RuntimeProgressAppend `json:"progress_appends,omitempty"`
	MessageUpserts    []chatview.UIMessage    `json:"message_upserts,omitempty"`
	ResetMessages     bool                    `json:"reset_messages,omitempty"`
}

type RuntimeMessageAppend struct {
	ID      int                    `json:"id"`
	Type    chatview.UIMessageType `json:"type"`
	Content string                 `json:"content"`
}

type RuntimeProgressAppend struct {
	ID       int `json:"id"`
	Progress any `json:"progress"`
	Input    any `json:"input,omitempty"`
}

type Snapshot struct {
	BotID          string          `json:"bot_id"`
	SessionID      string          `json:"session_id"`
	Epoch          string          `json:"epoch"`
	Seq            int64           `json:"seq"`
	CurrentRunView *CurrentRunView `json:"current_run_view,omitempty"`
	UpdatedAt      time.Time       `json:"updated_at"`
}

type SnapshotUpdate func(snapshot Snapshot, exists bool) (Snapshot, bool, error)

type SteerClaimRef struct {
	ItemID       SteerItemID
	RunID        string
	OwnerID      string
	Generation   string
	FencingToken int64
	ClaimToken   string
}

type SteerItem struct {
	ID                            SteerItemID
	BotID, SessionID, TargetRunID string
	InvocationID                  string
	Payload                       []byte
	Status                        QueueStatus
	Position                      int64
	Claim                         *SteerClaimRef
	// ErrorCode is set when Status is rejected.
	ErrorCode string
	CreatedAt time.Time
}

type SteerItemID string

type SteerPendingRef struct {
	ItemID SteerItemID `json:"item_id"`
}

type SteerTurnView struct {
	ItemID string `json:"item_id" validate:"required" format:"uuid"`
	Status string `json:"status" validate:"required" enums:"claimed,applied"`
	Text   string `json:"text" validate:"required"`
	// TurnID and TurnPosition are drawn when the steer is claimed, not when its
	// step commits: a subscriber that only learns the queue item id cannot line
	// the input up against history, which left the live bubble and the settled
	// one rendering side by side until the commit published the durable name.
	TurnID         string    `json:"turn_id,omitempty" format:"uuid"`
	TurnPosition   int64     `json:"turn_position,omitempty"`
	AfterMessageID int       `json:"after_message_id"`
	Timestamp      time.Time `json:"timestamp" validate:"required" format:"date-time"`
}

type Subscription struct {
	C     <-chan Event
	Close func()
}

type TerminalRun struct {
	RunID        string
	BotID        string
	SessionID    string
	FencingToken int64
	State        string
	ErrorCode    string
	ErrorMessage string
}