message

package
v1.0.2 Latest Latest
Warning

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

Go to latest
Published: Nov 8, 2026 License: AGPL-3.0 Imports: 22 Imported by: 0

Documentation

Index

Constants

View Source
const (
	// ReasoningTimingMetadataKey stores server-observed reasoning segment timing
	// on the assistant row that owns the corresponding reasoning content.
	ReasoningTimingMetadataKey = "reasoning_timing"
	ReasoningTimingVersion     = 1
)
View Source
const (
	AgentStepInterruptedMetadataKey = "agent_step_interrupted"
	// DiscussStepMetadataKey marks a row a Discuss run wrote before its
	// answer. The row is context for later runs, not proof that its run
	// consumed the input it answers: the run can still fail after it.
	DiscussStepMetadataKey      = "discuss_step"
	HistoryErrorCodeMetadataKey = "error_code"
	// HistoryErrorArgsMetadataKey holds the catalog args of the failure named
	// by error_code: only the args its catalog entry allows, as strings.
	HistoryErrorArgsMetadataKey         = "error_args"
	HistoryFailureOriginMetadataKey     = "failure_origin"
	HistoryFailureOriginUserMessageHook = "user_message_hook"
	AgentStepInterruptedReasoningPrefix = "[Previous assistant response was interrupted during reasoning. Continue from this checkpoint:]\n"
	// ToolCallDiffsMetadataKey carries UI-only edit/write diffs lifted out of
	// assistant message content at persist time. Value shape:
	// map[toolCallID]unifiedDiff. Keeping them on the row's metadata column —
	// not inside content — keeps them out of the history byte budget, which
	// measures octet_length(content) only.
	ToolCallDiffsMetadataKey = "diffs"
	// InputEstimateMetadataKey holds, on the assistant row of a native step,
	// what the run's estimate priced the step's request at. Beside the input
	// tokens the provider counted for it (usage.inputTokens) it calibrates
	// the estimate of the session's later runs on the same model.
	InputEstimateMetadataKey = "input_estimate_tokens"
)
View Source
const (
	MessageSourceMetadataKey      = "message_source"
	MessageSourceInternalFeedback = "internal_feedback"
)

Message origin is independent of the LLM role. Internal feedback is stored as user input for model replay but stays in the current conversation turn.

Variables

View Source
var ErrAgentStepNotWritable = errors.New("agent step is no longer writable")
View Source
var ErrNoVisibleTurn = errors.New("session has no visible turn")

ErrNoVisibleTurn reports that a session holds no visible turn at all. It is a sentinel rather than the driver's own no-rows error so callers can map it onto a user-facing code instead of surfacing the driver's wording.

Functions

func IsInternalFeedback added in v0.21.0

func IsInternalFeedback(metadata map[string]any) bool

func LatestInterruptedCheckpoint

func LatestInterruptedCheckpoint(n int, classify func(int) (assistant, interrupted bool)) int

LatestInterruptedCheckpoint returns the last assistant entry only when it is an interrupted checkpoint. Earlier checkpoints are ordinary history.

Types

type AgentStep

type AgentStep struct {
	RunID       string
	Messages    []PersistInput
	Interrupted bool
	// Replacement starts a new attempt in the same transaction as its first
	// messages. Subsequent steps append normally to their persisted request.
	Replacement *TurnReplacement
}

AgentStep is one native-agent step persisted atomically. Interrupted marks a text/reasoning snapshot of a step the model never finished, which takes a different writability predicate than a complete step.

type AgentStepPersister

type AgentStepPersister interface {
	PersistAgentStep(ctx context.Context, step AgentStep) ([]Message, error)
}

type AssetRef

type AssetRef struct {
	ContentHash string         `json:"content_hash"`
	Role        string         `json:"role"`
	Ordinal     int            `json:"ordinal"`
	Mime        string         `json:"mime,omitempty"`
	SizeBytes   int64          `json:"size_bytes,omitempty"`
	StorageKey  string         `json:"storage_key,omitempty"`
	Name        string         `json:"name,omitempty"`
	Metadata    map[string]any `json:"metadata,omitempty"`
}

AssetRef links a media asset to a persisted message. ContentHash is the content-addressed identifier for the media file.

type AtomicRoundPersister

type AtomicRoundPersister interface {
	PersistRound(ctx context.Context, inputs []PersistInput, options RoundPersistenceOptions) ([]Message, bool, error)
}

AtomicRoundPersister writes a complete round in one transaction. Implementations must enforce any runtime fence carried by ctx, while still supporting unfenced local replacement transactions.

type DBService

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

DBService persists and reads bot history messages.

func NewService

func NewService(log *slog.Logger, queries dbstore.Queries, publishers ...event.Publisher) *DBService

NewService creates a message service.

func (*DBService) AllocateTurnSlot added in v0.20.0

func (s *DBService) AllocateTurnSlot(ctx context.Context, sessionID string) (TurnSlot, error)

AllocateTurnSlot reserves the next turn slot for a session so a caller can publish the turn's identity before the rows that belong to it are written. The position is spent whether or not the turn ever persists: positions are only ever compared, so a gap left by an input that was released costs nothing, while reusing a slot would file two turns under one number.

func (*DBService) DeleteByBot

func (s *DBService) DeleteByBot(ctx context.Context, botID string) error

DeleteByBot deletes all messages for a bot.

func (*DBService) DeleteByIDs

func (s *DBService) DeleteByIDs(ctx context.Context, ids []string) error

DeleteByIDs deletes specific messages by id. A run deletes under its persistence fence, so an owner that lost the session cannot delete rows its successor reads.

func (*DBService) DeleteBySession

func (s *DBService) DeleteBySession(ctx context.Context, sessionID string) error

DeleteBySession deletes all messages for a session.

func (*DBService) GetByIDBySession

func (s *DBService) GetByIDBySession(ctx context.Context, sessionID string, messageID string) (Message, error)

func (*DBService) GetLatestVisibleTurnBySession

func (s *DBService) GetLatestVisibleTurnBySession(ctx context.Context, sessionID string) (HistoryTurn, error)

func (*DBService) GetRunTurnRequestMessageID added in v1.0.0

func (s *DBService) GetRunTurnRequestMessageID(ctx context.Context, sessionID, runID, rootTurnID string) (string, error)

GetRunTurnRequestMessageID follows applied input within the run. The root fallback also supports inbound requests persisted before run admission.

func (*DBService) GetVisibleHistoryTurnRequestMessageIDByTurn added in v0.21.0

func (s *DBService) GetVisibleHistoryTurnRequestMessageIDByTurn(ctx context.Context, sessionID string, turnID string) (string, error)

GetVisibleHistoryTurnRequestMessageIDByTurn returns the request message of a session turn. Waiting-decision recovery rebuilds a parked run's view from it (application.Service.ParkedRunMessages).

func (*DBService) GetVisibleTurnByMessage

func (s *DBService) GetVisibleTurnByMessage(ctx context.Context, sessionID string, messageID string) (HistoryTurn, error)

func (*DBService) HistoryPositionByID added in v1.0.0

func (s *DBService) HistoryPositionByID(ctx context.Context, sessionID string, messageID string) (HistoryPosition, error)

HistoryPositionByID reports where a visible message of the session is.

func (*DBService) LinkAssets

func (s *DBService) LinkAssets(ctx context.Context, messageID string, assets []AssetRef) error

LinkAssets links asset refs to an existing persisted message.

func (*DBService) ListBeforeBySession

func (s *DBService) ListBeforeBySession(ctx context.Context, sessionID string, before time.Time, limit int32) ([]Message, error)

ListBeforeBySession returns up to limit session messages older than before.

func (*DBService) ListBeforeMessageBySession

func (s *DBService) ListBeforeMessageBySession(ctx context.Context, sessionID string, beforeMessageID string, limit int32) ([]Message, error)

ListBeforeMessageBySession returns up to limit session messages before the cursor message in visible turn order.

func (*DBService) ListBySession

func (s *DBService) ListBySession(ctx context.Context, sessionID string) ([]Message, error)

ListBySession returns all messages for a session.

func (*DBService) ListHistoryByIDs added in v1.0.0

func (s *DBService) ListHistoryByIDs(ctx context.Context, sessionID string, ids []string) ([]Message, error)

ListHistoryByIDs returns the named context rows of a session in history order, loaded as ListHistoryWithinBytes loads them.

func (*DBService) ListHistoryWithinBytes added in v1.0.0

func (s *DBService) ListHistoryWithinBytes(ctx context.Context, sessionID string, window HistoryWindow) (HistoryLoad, error)

ListHistoryWithinBytes returns a session's raw history window, admitted newest-first within the content byte budget (CM-ADM-001), and whether the budget left older rows of the window out.

func (*DBService) ListLatest

func (s *DBService) ListLatest(ctx context.Context, botID string, limit int32) ([]Message, error)

ListLatest returns the latest N bot messages (newest first in DB; caller may reverse for ASC).

func (*DBService) ListLatestBySession

func (s *DBService) ListLatestBySession(ctx context.Context, sessionID string, limit int32) ([]Message, error)

ListLatestBySession returns the latest N session messages.

func (*DBService) ListLatestUIBySession

func (s *DBService) ListLatestUIBySession(ctx context.Context, sessionID string, limit int32) ([]Message, error)

ListLatestUIBySession returns the latest N session messages using the lighter field set needed by format=ui rendering.

func (*DBService) ListSinceBySession

func (s *DBService) ListSinceBySession(ctx context.Context, sessionID string, since time.Time) ([]Message, error)

ListSinceBySession returns session messages since a given time.

func (*DBService) ListTurnResponseSourcesByIDs added in v1.0.0

func (s *DBService) ListTurnResponseSourcesByIDs(ctx context.Context, sessionID string, ids []string) ([]Message, error)

ListTurnResponseSourcesByIDs projects the named context rows of a session as ListTurnResponseSourcesWithinBytes projects its window, with the inbound message a user row records.

func (*DBService) ListTurnResponseSourcesWithinBytes added in v1.0.0

func (s *DBService) ListTurnResponseSourcesWithinBytes(ctx context.Context, sessionID string, after HistoryPosition, maxBytes int64) (HistoryLoad, error)

ListTurnResponseSourcesWithinBytes walks the raw history window as ListHistoryWithinBytes does and projects only the fields used by timeline turn-response composition. In particular, legacy lifecycle audits must stay in the database, rather than being transferred and decoded alongside a small content window. Display fields and asset enrichment are not needed on this path.

func (*DBService) ListVisibleFromBySession

func (s *DBService) ListVisibleFromBySession(ctx context.Context, sessionID string, messageID string) ([]Message, error)

func (*DBService) LocateByExternalIDBySession

func (s *DBService) LocateByExternalIDBySession(ctx context.Context, sessionID string, externalMessageID string, beforeLimit int32, afterLimit int32) (LocateResult, error)

func (*DBService) Persist

func (s *DBService) Persist(ctx context.Context, input PersistInput) (Message, error)

Persist writes a single message to bot_history_messages.

func (*DBService) PersistAgentStep

func (s *DBService) PersistAgentStep(ctx context.Context, step AgentStep) ([]Message, error)

PersistAgentStep appends one SDK step in a runtime-fenced transaction. Complete steps precede abort intent; interrupted checkpoints remain writable until terminal finalization for cancellation paths without recorded intent.

func (*DBService) PersistRound

func (s *DBService) PersistRound(ctx context.Context, inputs []PersistInput, options RoundPersistenceOptions) ([]Message, bool, error)

PersistRound writes all messages and history links under one PostgreSQL transaction. Distributed callers additionally validate their runtime fence in that transaction; local replacements use the same atomic write without a distributed ownership token.

func (*DBService) PersistToolTailRound

func (s *DBService) PersistToolTailRound(ctx context.Context, inputs []PersistInput) ([]Message, bool, error)

PersistToolTailRound writes the common user -> assistant(tool-call) -> tool -> assistant(final) round with one PostgreSQL statement. Unsupported stores or non-matching inputs return handled=false so callers can use Persist.

func (*DBService) RunWroteHistory added in v1.0.1

func (s *DBService) RunWroteHistory(ctx context.Context, sessionID, runID, rootTurnID string) (bool, error)

RunWroteHistory includes hidden rows so a removed request is not restored.

func (*DBService) SetMarkdownMediaResolver added in v1.0.0

func (s *DBService) SetMarkdownMediaResolver(resolve markdownmedia.Resolver)

type HistoryLoad added in v1.0.0

type HistoryLoad struct {
	Messages []Message
	// Truncated reports that MaxBytes, not the window, ended the load: context
	// rows of the window remain older than the oldest it read.
	Truncated bool
}

HistoryLoad is what a byte-budgeted load of a HistoryWindow read.

type HistoryPosition added in v1.0.0

type HistoryPosition struct {
	Turn int64
	Seq  int64
}

HistoryPosition is a place in a session's history: the position of a turn and the sequence of a message in it. Positions order the history; the zero position precedes every message.

func (HistoryPosition) Before added in v1.0.0

func (p HistoryPosition) Before(other HistoryPosition) bool

Before reports whether p comes before other.

type HistoryTurn

type HistoryTurn struct {
	ID                 string    `json:"id"`
	BotID              string    `json:"bot_id"`
	SessionID          string    `json:"session_id"`
	Position           int64     `json:"position"`
	RequestMessageID   string    `json:"request_message_id,omitempty"`
	AssistantMessageID string    `json:"assistant_message_id,omitempty"`
	SupersededByTurnID string    `json:"superseded_by_turn_id,omitempty"`
	SupersededAt       time.Time `json:"superseded_at,omitempty"`
	SupersededReason   string    `json:"superseded_reason,omitempty"`
	CreatedAt          time.Time `json:"created_at"`
	UpdatedAt          time.Time `json:"updated_at"`
}

type HistoryWindow added in v1.0.0

type HistoryWindow struct {
	// After is the position of the checkpoint the raw history follows.
	After HistoryPosition
	// Before, when set, cuts the history before a message: an edit or retry
	// replays what came before the turn it replaces.
	Before   *HistoryPosition
	MaxBytes int64
}

HistoryWindow selects what a byte-budgeted load of a session's raw history reads: the context rows after a history checkpoint, newest first within MaxBytes.

type LocateResult

type LocateResult struct {
	Messages []Message
	TargetID string
}

type Message

type Message struct {
	ID                      string          `json:"id"`
	BotID                   string          `json:"bot_id"`
	SessionID               string          `json:"session_id,omitempty"`
	SenderChannelIdentityID string          `json:"sender_channel_identity_id,omitempty"`
	SenderUserID            string          `json:"sender_user_id,omitempty"`
	SenderDisplayName       string          `json:"sender_display_name,omitempty"`
	SenderAvatarURL         string          `json:"sender_avatar_url,omitempty"`
	Platform                string          `json:"platform,omitempty"`
	ExternalMessageID       string          `json:"external_message_id,omitempty"`
	SourceReplyToMessageID  string          `json:"source_reply_to_message_id,omitempty"`
	Role                    string          `json:"role"`
	Content                 json.RawMessage `json:"content"`
	Metadata                map[string]any  `json:"metadata,omitempty"`
	RawMetadata             json.RawMessage `json:"-"`
	TurnID                  string          `json:"turn_id,omitempty"`
	// TurnPosition is the immutable turn-level sequence reserved at admission
	// (SR-TURN-001). It is loaded on the UI read path so the frontend can
	// order and reconcile turns without guessing from text or timestamps.
	TurnPosition   *int64          `json:"turn_position,omitempty"`
	Usage          json.RawMessage `json:"usage,omitempty"`
	SessionMode    string          `json:"session_mode,omitempty"`
	RuntimeType    string          `json:"runtime_type,omitempty"`
	Assets         []MessageAsset  `json:"assets,omitempty"`
	EventID        string          `json:"event_id,omitempty"`
	DisplayContent string          `json:"display_content,omitempty"`
	CreatedAt      time.Time       `json:"created_at"`
}

Message represents a single persisted bot message.

type MessageAsset

type MessageAsset struct {
	ContentHash string         `json:"content_hash"`
	Role        string         `json:"role"`
	Ordinal     int            `json:"ordinal"`
	Mime        string         `json:"mime"`
	SizeBytes   int64          `json:"size_bytes"`
	StorageKey  string         `json:"storage_key"`
	Name        string         `json:"name,omitempty"`
	Metadata    map[string]any `json:"metadata,omitempty"`
}

MessageAsset carries media asset metadata attached to a message. ContentHash is the content-addressed identifier for the media file.

type PersistInput

type PersistInput struct {
	BotID                   string
	SessionID               string
	SenderChannelIdentityID string
	SenderUserID            string
	ExternalMessageID       string
	SourceReplyToMessageID  string
	Role                    string
	Content                 json.RawMessage
	Metadata                map[string]any
	Usage                   json.RawMessage
	SessionMode             string
	RuntimeType             string
	Assets                  []AssetRef
	ModelID                 string
	EventID                 string
	DisplayText             string
	TurnRequestMessageID    string
	SkipHistoryTurn         bool
	// RunID links this row to the admitted run that produced it. It is
	// traceability only: history is read by turn, never by run.
	RunID string
	// TurnID and TurnPosition carry a turn this message must be filed under
	// because admission already allocated it (SR-TURN-001). Both empty means no
	// admission decided the turn and the history layer allocates one, which is
	// still the case for channel inbound and schedules.
	//
	// They travel together: a turn id without its position would file the row
	// under the right turn at the wrong place in the session's order, and a
	// position without its id would take a slot the client cannot name.
	TurnID       string
	TurnPosition *int64
}

PersistInput is the input for persisting a message.

type ReasoningTimingMetadata

type ReasoningTimingMetadata struct {
	Version  int                      `json:"version"`
	Segments []ReasoningTimingSegment `json:"segments"`
}

ReasoningTimingMetadata is the versioned persistence envelope stored in a bot_history_messages metadata JSONB value. Keeping it outside model content prevents presentation-only timing from being replayed to providers.

type ReasoningTimingSegment

type ReasoningTimingSegment struct {
	Ordinal    int    `json:"ordinal"`
	DurationMS int64  `json:"duration_ms"`
	State      string `json:"state"`
}

ReasoningTimingSegment describes one visible reasoning block. DurationMS is observed by the server and intentionally excludes presentation-only details.

func ReasoningTimingFromMetadata

func ReasoningTimingFromMetadata(metadata map[string]any) []ReasoningTimingSegment

ReasoningTimingFromMetadata decodes the versioned timing envelope from a message metadata map. JSON round-tripping keeps this tolerant of values loaded from JSONB as map[string]any as well as typed values used in tests.

type RoundPersistenceOptions

type RoundPersistenceOptions struct {
	Replacement                       *TurnReplacement
	CleanupRuntimeDecisionProjections bool
	// AgentTurnID records the runtime's own turn id on the round's run in
	// the same transaction as the messages; empty writes nothing.
	AgentTurnID string
}

type Service

type Service interface {
	Writer
	ListLatest(ctx context.Context, botID string, limit int32) ([]Message, error)
	ListBySession(ctx context.Context, sessionID string) ([]Message, error)
	ListSinceBySession(ctx context.Context, sessionID string, since time.Time) ([]Message, error)
	// ListHistoryWithinBytes loads a session's raw history over a
	// HistoryWindow (CM-ADM-001): its context rows are admitted newest-first
	// until their content byte total crosses window.MaxBytes, and the load
	// reports whether that left older rows of the window out. Only content is
	// budgeted, so metadata is returned undecoded in RawMetadata (Metadata
	// stays nil): callers decode one row at a time instead of holding every
	// row's decoded metadata, which legacy lifecycle audits can inflate.
	ListHistoryWithinBytes(ctx context.Context, sessionID string, window HistoryWindow) (HistoryLoad, error)
	// ListHistoryByIDs loads the named context rows of a session, as
	// ListHistoryWithinBytes loads them, in history order: the rows a history
	// checkpoint keeps by reference.
	ListHistoryByIDs(ctx context.Context, sessionID string, ids []string) ([]Message, error)
	// HistoryPositionByID reports where a visible message of the session is.
	HistoryPositionByID(ctx context.Context, sessionID string, messageID string) (HistoryPosition, error)
	// ListTurnResponseSourcesWithinBytes is the raw history after a position
	// within maxBytes, as ListHistoryWithinBytes loads it, projected to what
	// turn-response composition reads: ID, role, content, created_at, and
	// Metadata holding only the interrupted and Discuss step flags. Nothing
	// else is loaded.
	ListTurnResponseSourcesWithinBytes(ctx context.Context, sessionID string, after HistoryPosition, maxBytes int64) (HistoryLoad, error)
	// ListTurnResponseSourcesByIDs is ListHistoryByIDs projected as
	// ListTurnResponseSourcesWithinBytes projects its rows, with the
	// ExternalMessageID of a user row.
	ListTurnResponseSourcesByIDs(ctx context.Context, sessionID string, ids []string) ([]Message, error)
	ListLatestBySession(ctx context.Context, sessionID string, limit int32) ([]Message, error)
	ListBeforeBySession(ctx context.Context, sessionID string, before time.Time, limit int32) ([]Message, error)
	ListBeforeMessageBySession(ctx context.Context, sessionID string, beforeMessageID string, limit int32) ([]Message, error)
	LocateByExternalIDBySession(ctx context.Context, sessionID string, externalMessageID string, beforeLimit int32, afterLimit int32) (LocateResult, error)
	GetByIDBySession(ctx context.Context, sessionID string, messageID string) (Message, error)
	ListVisibleFromBySession(ctx context.Context, sessionID string, messageID string) ([]Message, error)
	GetVisibleTurnByMessage(ctx context.Context, sessionID string, messageID string) (HistoryTurn, error)
	GetLatestVisibleTurnBySession(ctx context.Context, sessionID string) (HistoryTurn, error)
	DeleteByIDs(ctx context.Context, ids []string) error
	DeleteByBot(ctx context.Context, botID string) error
	DeleteBySession(ctx context.Context, sessionID string) error
	LinkAssets(ctx context.Context, messageID string, assets []AssetRef) error
}

Service defines message read/write behavior.

type ToolTailRoundPersister

type ToolTailRoundPersister interface {
	PersistToolTailRound(ctx context.Context, inputs []PersistInput) ([]Message, bool, error)
}

ToolTailRoundPersister optionally persists a complete user -> assistant(tool-call) -> tool -> assistant(final) round in one write.

type TurnReplacement

type TurnReplacement struct {
	OldTurnID               string
	ReplacementTurnID       string
	ReplacementTurnPosition *int64
	RequestMessageID        string
	Reason                  string
	SessionMetadata         map[string]any
}

type TurnSlot added in v0.20.0

type TurnSlot struct {
	TurnID   string
	Position int64
}

TurnSlot names a turn before any of its rows exist. Admission already draws one for a run's request turn; an applied steer opens its own canonical turn (SR-TURN-001) and needs the same treatment, so every subscriber learns the turn's identity when the input is accepted rather than when the step commits.

type TurnSlotAllocator added in v0.20.0

type TurnSlotAllocator interface {
	AllocateTurnSlot(ctx context.Context, sessionID string) (TurnSlot, error)
}

TurnSlotAllocator draws a slot from the session's turn counter without writing a row.

type Writer

type Writer interface {
	Persist(ctx context.Context, input PersistInput) (Message, error)
}

Writer defines write behavior needed by the inbound router.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL