chatstate

package
v2.35.3 Latest Latest
Warning

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

 Go to latest Published: Jul 27, 2026 License: AGPL-3.0 Imports: 15 Imported by: 0

Documentation

Overview

Package chatstate owns the durable execution-state transitions for the chatd subsystem. It implements the chat execution state model.

The package exposes two top-level entry points:

  • CreateChat creates a brand new chat with its initial history in a single transaction. It is standalone because no chat-scoped state machine instance can exist before the chat row is written.
  • ChatMachine wraps an existing chat. Callers use it to apply one or more transitions atomically via ChatMachine.Update, or to read related rows while holding the chat row lock via ChatMachine.Lock.

Every successful ChatMachine.Update call locks the chat row, advances `snapshot_version` exactly once, applies transition methods in order, and (on commit) publishes a single typed `chat:update` pubsub message describing the post-transition snapshot. Optional `chat:ownership` hints are published only when the post-transition state is runnable and ownership is missing or stale. Stream side effects are handled by `chat:update` consumers, and ownership hints wake chat workers.

Transition methods are explicit, typed wrappers around the durable mutations needed to move between states. Each transition reads the current chat row and queue cardinality, classifies the resulting execution state, validates it against the transition model, and rejects with an *TransitionError wrapping ErrTransitionNotAllowed when the transition is not legal from that state. The package owns transition validation, durable chat row and queue mutations, and post-commit pubsub publication.

Index

Constants

View Source 
const HeartbeatStaleSeconds = 30

HeartbeatStaleSeconds is the threshold chatstate uses when deciding whether to publish a `chat:ownership` hint for a runnable chat. A heartbeat older than this many seconds (by database time) counts as stale and triggers a hint so an idle worker can attempt a takeover.

View Source 
const MaxQueueSize = 20

MaxQueueSize is the maximum number of queued user messages per chat. Queue-appending transitions reject inserts that would exceed this cap with a *MessageQueueFullError that wraps ErrMessageQueueFull.

Variables

View Source 
var (
	// ErrTransitionNotAllowed is returned when a transition is applied
	// to a chat whose current execution state does not permit it. The
	// concrete error returned by transition methods is a
	// *TransitionError that wraps this sentinel.
	ErrTransitionNotAllowed = xerrors.New("chat state transition not allowed")

	// ErrInvalidState is returned when the chat row, queue, and
	// archive flag together produce a combination outside the chat
	// execution state model.
	ErrInvalidState = xerrors.New("chat is in an invalid execution state")

	// ErrQueuedMessageNotFound is returned by queue-targeting
	// transitions (delete, promote) when the supplied queued message
	// ID does not match a row on the chat.
	ErrQueuedMessageNotFound = xerrors.New("queued message not found")

	// ErrMessageNotFound is returned by [Tx.EditMessage] when the
	// target chat_messages row is missing or belongs to another chat.
	ErrMessageNotFound = xerrors.New("chat message not found")

	// ErrChatNotFound is returned when a non-create transition is
	// applied to a chat row that does not exist (or has been deleted
	// since the transition started).
	ErrChatNotFound = xerrors.New("chat not found")

	// ErrChatNotRoot is returned by family-archive helpers when the
	// supplied chat is not a root chat (its parent_chat_id is set).
	ErrChatNotRoot = xerrors.New("chat is not a root chat")

	// ErrEditedMessageNotUser is returned by [Tx.EditMessage] when the
	// targeted chat_messages row exists but its role is not user.
	ErrEditedMessageNotUser = xerrors.New("only user messages can be edited")

	// ErrMessageQueueFull is returned by queue-appending transitions
	// when the per-chat queue cap has been reached. The concrete
	// error returned by transitions is a *MessageQueueFullError that
	// wraps this sentinel.
	ErrMessageQueueFull = xerrors.New("chat message queue is full")

	// ErrToolResultDuplicate is returned by [Tx.CompleteRequiresAction]
	// when the same tool_call_id appears more than once in the
	// submitted results.
	ErrToolResultDuplicate = xerrors.New("duplicate tool result")

	// ErrToolResultUnexpected is returned by
	// [Tx.CompleteRequiresAction] when a submitted tool_call_id does
	// not correspond to a pending dynamic tool call.
	ErrToolResultUnexpected = xerrors.New("unexpected tool result")

	// ErrToolResultMissing is returned by [Tx.CompleteRequiresAction]
	// when a pending dynamic tool call has no submitted result.
	ErrToolResultMissing = xerrors.New("missing tool result")

	// ErrToolResultInvalidJSON is returned by
	// [Tx.CompleteRequiresAction] when a submitted tool result output
	// is not valid JSON.
	ErrToolResultInvalidJSON = xerrors.New("tool result output is not valid JSON")
)

Sentinel errors returned by chatstate transitions and helpers. Callers should use errors.Is to test for these.

View Source 
var AllExecutionStates = []ExecutionState{
	StateN,
	StateW,
	StateE0,
	StateE1,
	StateR0,
	StateR1,
	StateI0,
	StateI1,
	StateA0,
	StateA1,
	StateXW,
	StateXE0,
	StateXE1,
	StateInvalid,
}

AllExecutionStates is the canonical enumeration of every value the classifier can return. Tests rely on this list to iterate over every state when verifying transition coverage.

View Source 
var AllExecutionTransitions = []Transition{
	TransitionCreateChat,
	TransitionSetArchived,
	TransitionSendMessage,
	TransitionEditMessage,
	TransitionDeleteQueuedMessage,
	TransitionPromoteQueuedMessage,
	TransitionInterrupt,
	TransitionCompleteRequiresAction,
	TransitionRecordGenerationAttempt,
	TransitionRecordRetryState,
	TransitionCommitStep,
	TransitionEnterRequiresAction,
	TransitionFinishInterruption,
	TransitionFinishTurn,
	TransitionFinishError,
	TransitionCancelRequiresAction,
	TransitionReconcileInvalidState,
}

AllExecutionTransitions is the canonical enumeration of every execution-state transition that has an entry in the matrix below. Ownership transitions (Acquire, Abandon) are intentionally not part of this slice because they are validated independently and do not have a (from->to) execution mapping.

View Source 
var AllOwnershipStates = []OwnershipState{StateU, StateO}

AllOwnershipStates is the canonical enumeration of ownership states.

Functions

func SetFamilyArchived 

func SetFamilyArchived(
	ctx context.Context,
	store database.Store,
	publisher Publisher,
	input SetFamilyArchivedInput,
) ([]database.Chat, error)

SetFamilyArchived runs Update for every chat in the root chat's family inside one transaction, applying SetArchived when the chat's archived flag differs from the requested value. It owns its transaction lifecycle and its PublishBuffer lifecycle: pubsub publications are buffered while the transaction is open and flushed only after a successful commit; the deferred Discard suppresses every buffered publication on failure.

On success SetFamilyArchived returns one database.Chat per family member in the order returned by GetChatFamilyIDsByRootID (root first, then children).

Family members that are already in the StateInvalid execution state cause SetFamilyArchived to return ErrInvalidState and roll back the cascade even when their archived flag already matches the desired value; invalid-state detection is never bypassed.

Family members that are valid and already match the desired archived value still run through Update, which increments their snapshot version and publishes a fresh snapshot without changing the archived flag. Advancing the snapshot version without a field change is safe, and it keeps publication behavior uniform while a partially archived family converges to the desired state.

Types

type AbandonInput 

type AbandonInput struct{}

AbandonInput is intentionally empty. Ownership-fence checks belong outside the transition in caller code that reads the locked row before invoking Abandon.

type AbandonResult 

type AbandonResult struct{}

AbandonResult is returned by Tx.Abandon.

type AcquireInput 

type AcquireInput struct {
	WorkerID uuid.UUID
	RunnerID uuid.UUID
}

AcquireInput configures Tx.Acquire.

type AcquireResult 

type AcquireResult struct{}

AcquireResult is returned by Tx.Acquire.

type BusyBehavior 

type BusyBehavior string

BusyBehavior controls how SendMessage behaves when the chat is currently busy (R*/I*/A*). From idle/error states the two behaviors are equivalent. 

const (
	BusyBehaviorQueue     BusyBehavior = "queue"
	BusyBehaviorInterrupt BusyBehavior = "interrupt"
)

type CancelRequiresActionInput 

type CancelRequiresActionInput struct {
	Reason string
}

CancelRequiresActionInput configures Tx.CancelRequiresAction.

type CancelRequiresActionResult 

type CancelRequiresActionResult struct {
	CancellationMessages []database.ChatMessage
}

CancelRequiresActionResult is returned by Tx.CancelRequiresAction.

type ChatMachine 

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

ChatMachine is a chat-scoped handle for state-machine operations on a single chat row. It captures the database store, the pubsub publisher, and the chat ID at construction time so callers do not have to thread them through Update, Lock, or any transition method.

ChatMachine values are cheap. Create one per chat for the lifetime of a request or worker turn; do not cache mutable chat state across calls.

func NewChatMachine 

func NewChatMachine(
	store database.Store,
	publisher Publisher,
	chatID uuid.UUID,
) *ChatMachine

NewChatMachine constructs a chat-scoped state machine handle. The store may be the root database handle or an existing transaction handle; publisher is the pubsub used for `chat:update` and `chat:ownership` emissions. Both are required and captured for the lifetime of the returned machine.

func (*ChatMachine) ChatID 

func (m *ChatMachine) ChatID() uuid.UUID

ChatID returns the chat ID this machine is scoped to.

func (*ChatMachine) Lock 

func (m *ChatMachine) Lock(
	ctx context.Context,
	fn func(database.Store) error,
) error

Lock locks the chat row with FOR UPDATE and runs fn in a transaction without advancing snapshot_version. It uses the store captured by NewChatMachine. Use it when the caller needs a consistent chat snapshot plus related rows such as messages or queued messages but is NOT applying a transition.

Callers must not pass a store here; it belongs on the machine.

Lock publishes nothing. Callback errors roll back the transaction and propagate to the caller.

func (*ChatMachine) ReadLock 

func (m *ChatMachine) ReadLock(
	ctx context.Context,
	fn func(database.Store) error,
) error

ReadLock takes a shared lock on the chat row with FOR SHARE and runs fn in a transaction without advancing snapshot_version. It uses the store captured by NewChatMachine. Use it when the caller needs a consistent chat snapshot plus related rows such as messages or queued messages but is NOT applying a transition and does NOT need to block concurrent readers.

Unlike ChatMachine.Lock, the FOR SHARE lock permits other shared lockers to proceed concurrently while still blocking writers that take FOR UPDATE (such as ChatMachine.Update and ChatMachine.Lock) until the transaction commits.

Callers must not pass a store here; it belongs on the machine.

ReadLock publishes nothing. Callback errors roll back the transaction and propagate to the caller.

func (*ChatMachine) Update 

func (m *ChatMachine) Update(
	ctx context.Context,
	fn func(*Tx, database.Store) error,
) error

Update applies one or more transitions to the machine's chat.

Update opens a transaction on the captured store, atomically locks the chat row with FOR UPDATE and increments `snapshot_version` exactly once, then runs fn against a fresh *Tx and the active transaction store. It constructs a PublishBuffer, enqueues `chat:update` (and a `chat:ownership` hint when the post-transition state is worker-runnable and ownership is missing or stale) inside the transaction, and flushes the buffer only after the transaction function succeeds. If the transaction rolls back, the deferred Discard suppresses every buffered publication so subscribers never see uncommitted state.

If Update is called with a store that is already in a transaction, database.Store.InTx reuses the active transaction. In that case, callers that need outer-transaction publication semantics can pass a PublishBuffer as the machine publisher. The inner buffer flushes into the outer buffer, and the outer owner remains responsible for publishing only after the outer transaction commits.

If the chat row does not exist, Update returns ErrChatNotFound without mutating anything.

Callbacks that return an error roll back the transaction (rolling back the automatic snapshot bump) and publish nothing.

type CommitStepInput 

type CommitStepInput struct {
	Messages []Message
}

CommitStepInput configures Tx.CommitStep.

type CommitStepResult 

type CommitStepResult struct {
	InsertedMessages []database.ChatMessage
}

CommitStepResult is returned by Tx.CommitStep.

type CompleteRequiresActionInput 

type CompleteRequiresActionInput struct {
	CreatedBy     uuid.UUID
	ModelConfigID uuid.UUID
	Results       []ToolResultInput
}

CompleteRequiresActionInput configures Tx.CompleteRequiresAction.

type CompleteRequiresActionResult 

type CompleteRequiresActionResult struct {
	InsertedMessages []database.ChatMessage
}

CompleteRequiresActionResult is returned by Tx.CompleteRequiresAction.

type CreateChatInput 

type CreateChatInput struct {
	OrganizationID    uuid.UUID
	OwnerID           uuid.UUID
	WorkspaceID       uuid.NullUUID
	BuildID           uuid.NullUUID
	AgentID           uuid.NullUUID
	ParentChatID      uuid.NullUUID
	RootChatID        uuid.NullUUID
	LastModelConfigID uuid.UUID
	Title             string
	Mode              database.NullChatMode
	PlanMode          database.NullChatPlanMode
	MCPServerIDs      []uuid.UUID
	Labels            pqtype.NullRawMessage
	DynamicTools      pqtype.NullRawMessage
	ClientType        database.ChatClientType
	InitialMessages   []Message
}

CreateChatInput configures CreateChat.

type CreateChatResult 

type CreateChatResult struct {
	Chat            database.Chat
	InitialMessages []database.ChatMessage
}

CreateChatResult is the value returned by CreateChat. It carries the new chat row and the inserted initial history.

func CreateChat 

func CreateChat(
	ctx context.Context,
	store database.Store,
	publisher Publisher,
	input CreateChatInput,
) (CreateChatResult, error)

CreateChat creates a brand new chat with initial history in a single transaction. It is package-level rather than a method on ChatMachine because no chat-scoped machine can exist before the chat row is written.

Validation:

  • InitialMessages must be non-empty.

After commit CreateChat publishes a `chat:update` message describing the new chat snapshot. Because the new chat has no worker assigned, CreateChat also publishes an ownership hint so workers can race to acquire the runnable chat.

type DeleteQueuedMessageInput 

type DeleteQueuedMessageInput struct {
	QueuedMessageID int64
}

DeleteQueuedMessageInput configures Tx.DeleteQueuedMessage.

type DeleteQueuedMessageResult 

type DeleteQueuedMessageResult struct {
	DeletedQueuedMessage database.ChatQueuedMessage
}

DeleteQueuedMessageResult is returned by Tx.DeleteQueuedMessage.

type EditMessageInput 

type EditMessageInput struct {
	MessageID             int64
	CreatedBy             uuid.UUID
	Content               pqtype.NullRawMessage
	ModelConfigIDOverride uuid.NullUUID
	APIKeyID              sql.NullString
}

EditMessageInput configures Tx.EditMessage.

type EditMessageResult 

type EditMessageResult struct {
	ReplacementMessage      database.ChatMessage
	DeletedMessageIDs       []int64
	DeletedQueuedMessageIDs []int64
	CancellationMessages    []database.ChatMessage
}

EditMessageResult is returned by Tx.EditMessage.

type EnterRequiresActionInput 

type EnterRequiresActionInput struct{}

EnterRequiresActionInput is intentionally empty.

type EnterRequiresActionResult 

type EnterRequiresActionResult struct {
	RequiresActionDeadlineAt sql.NullTime
}

EnterRequiresActionResult is returned by Tx.EnterRequiresAction.

type ExecutionState 

type ExecutionState string

ExecutionState identifies a chat's current execution state. Values outside the chat execution state model are represented by StateInvalid. 

const (
	// StateN: chat does not exist.
	StateN ExecutionState = "N"
	// StateW: waiting, empty queue, not archived.
	StateW ExecutionState = "W"
	// StateE0: error, empty queue, not archived.
	StateE0 ExecutionState = "E0"
	// StateE1: error, non-empty queue, not archived.
	StateE1 ExecutionState = "E1"
	// StateR0: running, empty queue, not archived.
	StateR0 ExecutionState = "R0"
	// StateR1: running, non-empty queue, not archived.
	StateR1 ExecutionState = "R1"
	// StateI0: interrupting, empty queue, not archived.
	StateI0 ExecutionState = "I0"
	// StateI1: interrupting, non-empty queue, not archived.
	StateI1 ExecutionState = "I1"
	// StateA0: requires_action, empty queue, not archived.
	StateA0 ExecutionState = "A0"
	// StateA1: requires_action, non-empty queue, not archived.
	StateA1 ExecutionState = "A1"
	// StateXW: archived waiting, empty queue.
	StateXW ExecutionState = "XW"
	// StateXE0: archived error, empty queue.
	StateXE0 ExecutionState = "XE0"
	// StateXE1: archived error, non-empty queue.
	StateXE1 ExecutionState = "XE1"

	// StateInvalid groups every status/archive/queue combination that
	// is not one of the valid states above. The state machine refuses
	// non-reconciliation transitions on invalid states and exposes the
	// [Tx.ReconcileInvalidState] transition to recover.
	StateInvalid ExecutionState = "Invalid"
)

func AllowedExecutionTransitionOutputs 

func AllowedExecutionTransitionOutputs(from ExecutionState, tr Transition) []ExecutionState

AllowedExecutionTransitionOutputs returns the set of classified post-states that the transition `tr` may produce from `from` per the matrix above. The returned slice is a copy so callers may mutate it without affecting the underlying matrix.

When `tr` is not allowed from `from`, an empty (nil) slice is returned. Tests use this helper to enumerate the (transition, from, want) triples that must be exercised by the row-level matrix tests.

func AllowedInputStates 

func AllowedInputStates(tr Transition) []ExecutionState

AllowedInputStates returns a deterministic slice of execution states from which `tr` is legal per the matrix above. Mostly used by tests to enumerate the matrix without leaking the internal map.

func ClassifyExecutionState 

func ClassifyExecutionState(chat database.Chat, queueNonEmpty, exists bool) ExecutionState

ClassifyExecutionState turns the chat row, queue cardinality, and whether the chat row exists into an ExecutionState. The caller is responsible for loading the chat under the row lock and reading the queue count in the same transaction.

Callers that have no chat row (lookup returned sql.ErrNoRows) should pass exists=false; the chat, status, and archive arguments are then ignored.

The classifier is a single flat switch over the valid (status, archived, queue) tuples in the chat execution state model. Anything outside that set (legacy pending/paused/completed statuses, archived busy states, waiting with a non-empty queue, future enum values) falls through to StateInvalid.

func (ExecutionState) IsArchived 

func (s ExecutionState) IsArchived() bool

IsArchived returns true for the three archived execution states.

func (ExecutionState) IsRunnable 

func (s ExecutionState) IsRunnable() bool

IsRunnable returns true for the execution states that the chat worker is allowed to acquire and drive forward: R0, R1, I0, I1, A0, and A1. Requires-action states need worker ownership for timeout processing. Other states are idle (W, E*, XW, XE*), absent (N), or invalid.

func (ExecutionState) QueueNonEmpty 

func (s ExecutionState) QueueNonEmpty() bool

QueueNonEmpty returns true for execution states that require a non-empty queue. Useful when seeding test fixtures.

func (ExecutionState) String 

func (s ExecutionState) String() string

String implements fmt.Stringer.

type FinishErrorInput 

type FinishErrorInput struct {
	LastError pqtype.NullRawMessage
}

FinishErrorInput configures Tx.FinishError.

type FinishErrorResult 

type FinishErrorResult struct{}

FinishErrorResult is returned by Tx.FinishError.

type FinishInterruptionInput 

type FinishInterruptionInput struct {
	PartialMessages []Message
}

FinishInterruptionInput configures Tx.FinishInterruption.

type FinishInterruptionResult 

type FinishInterruptionResult struct {
	InsertedMessages []database.ChatMessage
	PromotedMessage  *database.ChatMessage
}

FinishInterruptionResult is returned by Tx.FinishInterruption.

type FinishTurnInput 

type FinishTurnInput struct{}

FinishTurnInput is intentionally empty.

type FinishTurnResult 

type FinishTurnResult struct {
	Chat            database.Chat
	PromotedMessage *database.ChatMessage
}

FinishTurnResult is returned by Tx.FinishTurn.

type InterruptInput 

type InterruptInput struct {
	Reason string
}

InterruptInput configures Tx.Interrupt.

type InterruptResult 

type InterruptResult struct {
	CancellationMessages []database.ChatMessage
}

InterruptResult is returned by Tx.Interrupt.

type Message 

type Message struct {
	Role                database.ChatMessageRole
	Content             pqtype.NullRawMessage
	Visibility          database.ChatMessageVisibility
	ModelConfigID       uuid.NullUUID
	CreatedBy           uuid.NullUUID
	ContentVersion      int16
	Compressed          bool
	InputTokens         sql.NullInt64
	OutputTokens        sql.NullInt64
	TotalTokens         sql.NullInt64
	ReasoningTokens     sql.NullInt64
	CacheCreationTokens sql.NullInt64
	CacheReadTokens     sql.NullInt64
	ContextLimit        sql.NullInt64
	TotalCostMicros     sql.NullInt64
	RuntimeMs           sql.NullInt64
	ProviderResponseID  sql.NullString
	APIKeyID            sql.NullString
}

Message is the durable message input shape used by chatstate transitions. It is intentionally lower level than the SDK message request types: callers must produce a fully materialized message (parsed parts, calculated cost, resolved model config) before passing it in.

The state machine never reshapes a Message except to attach the runtime `chat_id`.

type MessageQueueFullError 

type MessageQueueFullError struct {
	Max int64
}

MessageQueueFullError carries the per-chat queue cap so HTTP endpoints can include the cap in their response detail. It wraps ErrMessageQueueFull so callers can match it with errors.Is.

func (*MessageQueueFullError) Error 

func (e *MessageQueueFullError) Error() string

Error implements the error interface.

func (*MessageQueueFullError) Unwrap 

func (*MessageQueueFullError) Unwrap() error

Unwrap returns ErrMessageQueueFull so callers can match the generic sentinel.

type OwnershipState 

type OwnershipState string

OwnershipState identifies whether a chat row is currently owned by a worker. The state machine treats execution and ownership as orthogonal. 

const (
	// StateU: chat has no owner (worker_id IS NULL).
	StateU OwnershipState = "U"
	// StateO: chat has an owner (worker_id IS NOT NULL).
	StateO OwnershipState = "O"
)

type PromoteQueuedMessageInput 

type PromoteQueuedMessageInput struct {
	QueuedMessageID int64
}

PromoteQueuedMessageInput configures Tx.PromoteQueuedMessage.

type PromoteQueuedMessageResult 

type PromoteQueuedMessageResult struct {
	QueuedMessage        database.ChatQueuedMessage
	InsertedMessage      *database.ChatMessage
	ReorderedQueueOnly   bool
	CancellationMessages []database.ChatMessage
}

PromoteQueuedMessageResult is returned by Tx.PromoteQueuedMessage.

type PublishBuffer 

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

PublishBuffer is a Publisher that records each Publish call in order without forwarding it until PublishBuffer.Flush is called. It is an internal primitive used by chatstate entry points to hold pubsub messages until the surrounding transaction commits, and by tests that need to observe buffered output. Normal callers do not construct a PublishBuffer themselves and do not invoke Flush or Discard; chatstate's entry points own that lifecycle.

func NewPublishBuffer 

func NewPublishBuffer(inner Publisher) *PublishBuffer

NewPublishBuffer constructs a PublishBuffer that, when flushed, will forward messages in order to inner.

func (*PublishBuffer) BufferedChannels 

func (b *PublishBuffer) BufferedChannels() []string

BufferedChannels returns just the channels of the pending messages in order. Primarily useful for assertions in tests.

func (*PublishBuffer) Discard 

func (b *PublishBuffer) Discard()

Discard clears the buffered messages without forwarding them. It is safe to call multiple times and is harmless after PublishBuffer.Flush: once Flush has marked the buffer flushed and forwarded its pending messages, a subsequent Discard simply clears the (now empty) pending slice and sets the buffer to drop any future Publish calls. This makes `defer buf.Discard()` a safe pattern after a successful flush, including the one chatstate entry points use to own the buffer lifecycle.

func (*PublishBuffer) Flush 

func (b *PublishBuffer) Flush() error

Flush forwards every pending message to the inner publisher in the order it was buffered, then marks the buffer flushed. Joined publish errors are returned with channel names annotated after every pending message has been attempted.

func (*PublishBuffer) Publish 

func (b *PublishBuffer) Publish(channel string, payload []byte) error

Publish records a message. It never forwards to the inner publisher until PublishBuffer.Flush is called. Returns an error if Flush has already happened to make accidental reuse obvious.

type Publisher 

type Publisher interface {
	Publish(event string, message []byte) error
}

Publisher is the minimal interface chatstate needs to publish pubsub messages. It is intentionally compatible with database/pubsub.Pubsub: real callers pass the live pubsub directly and tests pass a recording fake.

type ReconcileInvalidStateInput 

type ReconcileInvalidStateInput struct {
	LastError          pqtype.NullRawMessage
	CancellationReason string
}

ReconcileInvalidStateInput configures Tx.ReconcileInvalidState.

type ReconcileInvalidStateResult 

type ReconcileInvalidStateResult struct {
	CancellationMessages []database.ChatMessage
}

ReconcileInvalidStateResult is returned by Tx.ReconcileInvalidState.

type RecordGenerationAttemptInput 

type RecordGenerationAttemptInput struct{}

RecordGenerationAttemptInput is intentionally empty.

type RecordGenerationAttemptResult 

type RecordGenerationAttemptResult struct {
	GenerationAttempt int64
}

RecordGenerationAttemptResult is returned by Tx.RecordGenerationAttempt.

type RecordRetryStateInput 

type RecordRetryStateInput struct {
	RetryState pqtype.NullRawMessage
}

RecordRetryStateInput configures Tx.RecordRetryState.

type RecordRetryStateResult 

type RecordRetryStateResult struct {
	Chat database.Chat
}

RecordRetryStateResult is returned by Tx.RecordRetryState.

type SendMessageInput 

type SendMessageInput struct {
	Message      Message
	BusyBehavior BusyBehavior
}

SendMessageInput configures Tx.SendMessage.

type SendMessageResult 

type SendMessageResult struct {
	InsertedMessages []database.ChatMessage
	QueuedMessage    *database.ChatQueuedMessage
}

SendMessageResult is returned by Tx.SendMessage.

type SetArchivedInput 

type SetArchivedInput struct {
	Archived bool
}

SetArchivedInput configures Tx.SetArchived.

type SetArchivedResult 

type SetArchivedResult struct{}

SetArchivedResult is returned by Tx.SetArchived.

type SetFamilyArchivedInput 

type SetFamilyArchivedInput struct {
	// RootID identifies the family root. SetFamilyArchived rejects
	// calls for child chats with [ErrChatNotRoot] and unknown chats
	// with [ErrChatNotFound].
	RootID uuid.UUID
	// Archived is the desired post-call archived value for every
	// family member.
	Archived bool
}

SetFamilyArchivedInput configures SetFamilyArchived. The struct shape avoids a boolean flag parameter at the API surface; callers build it explicitly with named fields for clarity.

type ToolResultInput 

type ToolResultInput struct {
	ToolCallID string
	Output     json.RawMessage
	IsError    bool
}

ToolResultInput is one submitted dynamic-tool result.

type ToolResultValidationError 

type ToolResultValidationError struct {
	Cause      error
	ToolCallID string
}

ToolResultValidationError carries a structured tool-result validation failure. It always wraps a specific sentinel (ErrToolResultDuplicate, ErrToolResultMissing, ErrToolResultUnexpected, ErrToolResultInvalidJSON) so callers can match either the generic sentinel or the specific cause.

func (*ToolResultValidationError) Error 

func (e *ToolResultValidationError) Error() string

Error implements the error interface.

func (*ToolResultValidationError) Unwrap 

func (e *ToolResultValidationError) Unwrap() error

Unwrap returns the specific cause so callers can match it.

type Transition 

type Transition string

Transition is the enumeration of transitions implemented by the state machine. Values intentionally match the names of the public methods on Tx (and CreateChat). The transition matrix below declares the legal (from -> to) execution-state mappings used by each transition method for validation. 

const (
	TransitionCreateChat              Transition = "CreateChat"
	TransitionSetArchived             Transition = "SetArchived"
	TransitionSendMessage             Transition = "SendMessage"
	TransitionEditMessage             Transition = "EditMessage"
	TransitionDeleteQueuedMessage     Transition = "DeleteQueuedMessage"
	TransitionPromoteQueuedMessage    Transition = "PromoteQueuedMessage"
	TransitionInterrupt               Transition = "Interrupt"
	TransitionCompleteRequiresAction  Transition = "CompleteRequiresAction"
	TransitionAcquire                 Transition = "Acquire"
	TransitionAbandon                 Transition = "Abandon"
	TransitionRecordGenerationAttempt Transition = "RecordGenerationAttempt"
	TransitionRecordRetryState        Transition = "RecordRetryState"
	TransitionCommitStep              Transition = "CommitStep"
	TransitionEnterRequiresAction     Transition = "EnterRequiresAction"
	TransitionFinishInterruption      Transition = "FinishInterruption"
	TransitionFinishTurn              Transition = "FinishTurn"
	TransitionFinishError             Transition = "FinishError"
	TransitionCancelRequiresAction    Transition = "CancelRequiresAction"
	TransitionReconcileInvalidState   Transition = "ReconcileInvalidState"
)

func AllowedExecutionTransitionsFrom 

func AllowedExecutionTransitionsFrom(from ExecutionState) []Transition

AllowedExecutionTransitionsFrom returns a deterministic slice of transitions legal from `from`. Mostly used by tests to enumerate the matrix without leaking the internal map.

func (Transition) String 

func (t Transition) String() string

String implements fmt.Stringer.

type TransitionError 

type TransitionError struct {
	Transition Transition
	From       ExecutionState
	Reason     string
	Cause      error
}

TransitionError carries the structured detail for a rejected transition. It always wraps ErrTransitionNotAllowed so callers can match with errors.Is without losing context. When a specific chatstate sentinel is the proximate cause, Cause is set and errors.Is will match that sentinel too.

func (*TransitionError) Error 

func (e *TransitionError) Error() string

Error implements the error interface.

func (*TransitionError) Unwrap 

func (e *TransitionError) Unwrap() error

Unwrap returns the error chain attached to this error. The chain always includes ErrTransitionNotAllowed, and may include a more specific cause through errors.Join, so callers can use errors.Is without custom matching logic on TransitionError.

type Tx 

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

Tx is the per-transaction handle passed to ChatMachine.Update callbacks. It carries the active context, the transactional store, and the chat ID. Tx does not cache mutable chat state across calls: every transition method reads the chat row and queue cardinality from the database on entry, so a bundle of transitions inside one Update callback always validates against the latest committed state.

func (*Tx) Abandon 

func (tx *Tx) Abandon(_ AbandonInput) (AbandonResult, error)

Abandon clears worker_id and runner_id from the locked chat row. It rejects calls when the chat is not currently owned (worker_id IS NULL). Callers that need to verify their own identity before abandoning should read the locked row through the transactional store and compare values before invoking Abandon.

func (*Tx) Acquire 

func (tx *Tx) Acquire(input AcquireInput) (AcquireResult, error)

Acquire claims the chat for a worker/runner pair. Execution state is preserved.

Acquire never inspects the chat's current ownership: it simply overwrites worker_id/runner_id with the supplied identifiers and upserts a fresh heartbeat. Detecting and recovering from stale leases is a worker-side fence concern outside the state machine. Callers that need to coordinate takeovers with the previous owner must arrange that out-of-band before calling Acquire.

func (*Tx) CancelRequiresAction 

func (tx *Tx) CancelRequiresAction(input CancelRequiresActionInput) (CancelRequiresActionResult, error)

CancelRequiresAction synthesizes cancellation results for every pending dynamic tool call and returns the chat to running.

func (*Tx) ChatID 

func (tx *Tx) ChatID() uuid.UUID

ChatID returns the chat ID this transaction is scoped to.

func (*Tx) CommitStep 

func (tx *Tx) CommitStep(input CommitStepInput) (CommitStepResult, error)

CommitStep stores one durable message suffix while remaining running.

func (*Tx) CompleteRequiresAction 

func (tx *Tx) CompleteRequiresAction(input CompleteRequiresActionInput) (CompleteRequiresActionResult, error)

CompleteRequiresAction validates and stores user-submitted tool results that satisfy the chat's pending dynamic tool calls, then returns the chat to running.

func (*Tx) Ctx 

func (tx *Tx) Ctx() context.Context

Ctx returns the context the surrounding ChatMachine.Update call is using.

func (*Tx) DeleteQueuedMessage 

func (tx *Tx) DeleteQueuedMessage(input DeleteQueuedMessageInput) (DeleteQueuedMessageResult, error)

DeleteQueuedMessage removes a single queued user message.

func (*Tx) EditMessage 

func (tx *Tx) EditMessage(input EditMessageInput) (EditMessageResult, error)

EditMessage replaces an earlier user message and discards the active-history suffix that followed it.

func (*Tx) EnterRequiresAction 

func (tx *Tx) EnterRequiresAction(_ EnterRequiresActionInput) (EnterRequiresActionResult, error)

EnterRequiresAction parks the chat in requires_action with a database-time deadline of now() + requiresActionTimeout.

func (*Tx) FinishError 

func (tx *Tx) FinishError(input FinishErrorInput) (FinishErrorResult, error)

FinishError parks the chat in error with the supplied last_error.

func (*Tx) FinishInterruption 

func (tx *Tx) FinishInterruption(input FinishInterruptionInput) (FinishInterruptionResult, error)

FinishInterruption commits an optional partial assistant/tool suffix and lands the chat in waiting (I0) or running with the next queued message promoted (I1).

func (*Tx) FinishTurn 

func (tx *Tx) FinishTurn(_ FinishTurnInput) (FinishTurnResult, error)

FinishTurn completes a running turn.

func (*Tx) Interrupt 

func (tx *Tx) Interrupt(input InterruptInput) (InterruptResult, error)

Interrupt requests interruption of an active or requires-action chat.

func (*Tx) PromoteQueuedMessage 

func (tx *Tx) PromoteQueuedMessage(input PromoteQueuedMessageInput) (PromoteQueuedMessageResult, error)

PromoteQueuedMessage promotes the target queued message to the queue head; from E1/A1 it also pops it into active history.

func (*Tx) ReconcileInvalidState 

func (tx *Tx) ReconcileInvalidState(input ReconcileInvalidStateInput) (ReconcileInvalidStateResult, error)

ReconcileInvalidState moves an invalid execution-state combination into a valid error state. Queued messages are preserved; pending dynamic-tool calls are closed with synthetic cancellation results.

func (*Tx) RecordGenerationAttempt 

func (tx *Tx) RecordGenerationAttempt(_ RecordGenerationAttemptInput) (RecordGenerationAttemptResult, error)

RecordGenerationAttempt durably records that the worker is attempting another generation under the current history version.

func (*Tx) RecordRetryState 

func (tx *Tx) RecordRetryState(input RecordRetryStateInput) (RecordRetryStateResult, error)

RecordRetryState stores the client-visible retry payload for the current generation attempt.

func (*Tx) SendMessage 

func (tx *Tx) SendMessage(input SendMessageInput) (SendMessageResult, error)

SendMessage admits a new user message. Depending on input state and BusyBehavior, the message lands directly in history, in the queue, or replaces the queue head as part of a running-state promotion.

func (*Tx) SetArchived 

func (tx *Tx) SetArchived(input SetArchivedInput) (SetArchivedResult, error)

SetArchived sets or clears the chat's archived marker.

func (*Tx) Store 

func (tx *Tx) Store() database.Store

Store exposes the active transaction store so callers can perform validation reads (for example loading the messages affected by an EditMessage transition) and metadata writes (for example updating title or labels) that must be atomic with the transition.

Callers MUST NOT use Store to mutate execution-state tables (chats.status, chat_messages, chat_queued_messages, chat_heartbeats, or the version fields on chats). Those mutations belong to the transition methods and are validated against the state machine matrix.

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.