Documentation
¶
Overview ¶
Package messagepartbuffer stores the transient message-part stream that a chat worker emits before those parts are committed to durable chat history.
Chat generation has two consumers with different timing. Stream endpoints need to forward parts immediately, while interruption handling may need to recover the partial assistant or tool message and commit it. Buffer groups parts by an episode key that includes the chat, history version, and generation attempt so stale workers and late subscribers do not mix parts from different generations.
Episodes are intentionally in-memory. They are closed when a generation attempt ends, then retained briefly so stream subscribers and interruption cleanup can drain the final parts. The cleanup loop removes closed episodes after the retention window. Never-created placeholders are removed during subscriber teardown, when the last early subscriber leaves.
Index ¶
- Variables
- type Buffer
- func (b *Buffer) AddPart(key Key, role codersdk.ChatMessageRole, part codersdk.ChatMessagePart) error
- func (b *Buffer) Close()
- func (b *Buffer) CloseEpisode(key Key) error
- func (b *Buffer) CreateEpisode(key Key) error
- func (b *Buffer) GetParts(key Key) ([]Part, error)
- func (b *Buffer) ModelInvokedAt(key Key) time.Time
- func (b *Buffer) RecordToolCompletion(key Key, callIndex int, completedAt time.Time) error
- func (b *Buffer) RecordToolStart(key Key, callIndex int, startedAt time.Time) error
- func (b *Buffer) StartModelInvocation(key Key) error
- func (b *Buffer) SubscribeToEpisode(ctx context.Context, key Key) (<-chan Part, func(), error)
- func (b *Buffer) ToolCompletions(key Key) map[int]ToolCompletion
- type Key
- type Options
- type Part
- type ToolCompletion
Constants ¶
This section is empty.
Variables ¶
var ( // ErrEpisodeExists means the episode already exists. ErrEpisodeExists = xerrors.New("message part episode already exists") // ErrEpisodeNotFound means the episode has not been created. ErrEpisodeNotFound = xerrors.New("message part episode not found") // ErrEpisodeClosed means the episode no longer accepts parts. ErrEpisodeClosed = xerrors.New("message part episode closed") // ErrEpisodeFull means the episode byte limit would be exceeded. ErrEpisodeFull = xerrors.New("message part episode full") // ErrMessagePartBufferClosed means the whole buffer is closed. ErrMessagePartBufferClosed = xerrors.New("message part buffer closed") )
Functions ¶
This section is empty.
Types ¶
type Buffer ¶
type Buffer struct {
// contains filtered or unexported fields
}
Buffer stores streamed message parts by episode.
func (*Buffer) AddPart ¶
func (b *Buffer) AddPart(key Key, role codersdk.ChatMessageRole, part codersdk.ChatMessagePart) error
AddPart appends a part to an existing episode.
Parts receive contiguous sequence numbers so stream endpoints can detect stale or broken episode subscriptions before forwarding data to clients.
func (*Buffer) Close ¶
func (b *Buffer) Close()
Close closes the buffer and all pending subscriptions.
func (*Buffer) CloseEpisode ¶
CloseEpisode marks an episode closed.
Closing creates the episode if it did not exist yet. This lets interruption cleanup converge when a worker exits before it publishes any parts.
func (*Buffer) CreateEpisode ¶
CreateEpisode creates a new episode.
Subscribers may attach before an episode is created. Creating the episode makes it eligible to receive parts; the first AddPart wakes early subscribers.
func (*Buffer) GetParts ¶
GetParts returns a snapshot of buffered parts for an episode.
The returned slice is detached from the buffer so callers can process it without holding the buffer lock.
func (*Buffer) ModelInvokedAt ¶ added in v2.37.0
ModelInvokedAt returns the instant stamped by StartModelInvocation, or the zero time if there is none. Read it before CloseEpisode: closed episodes are garbage collected, so reading afterwards races the cleanup loop.
func (*Buffer) RecordToolCompletion ¶ added in v2.37.0
RecordToolCompletion stamps a call as it finishes, so interrupts use the actual completion. Calls without a recorded start are dropped.
func (*Buffer) RecordToolStart ¶ added in v2.37.0
RecordToolStart adds a tool-call occurrence when it begins execution. CallIndex preserves rejected-call gaps and keeps duplicate IDs distinct. Repeated starts keep the first timestamp.
func (*Buffer) StartModelInvocation ¶ added in v2.37.0
StartModelInvocation stamps the instant the episode opens its provider stream, which starts the episode's billable model invocation window.
func (*Buffer) SubscribeToEpisode ¶
SubscribeToEpisode replays existing parts and streams new parts.
Subscribers may attach before CreateEpisode is called. In that case the subscription stays idle until the first part added, closure, cancellation, or buffer shutdown. The returned cancel function is idempotent.
func (*Buffer) ToolCompletions ¶ added in v2.37.0
func (b *Buffer) ToolCompletions(key Key) map[int]ToolCompletion
ToolCompletions returns copied started-occurrence state. A zero completion means unfinished. Read it before CloseEpisode because closed episodes are garbage collected.
type Options ¶
type Options struct {
MaxEpisodeBytes int64
ClosedEpisodeRetention time.Duration
SubscriberSendTimeout time.Duration
Clock quartz.Clock
}
Options configures a Buffer.
type Part ¶
type Part struct {
Seq int64
Role codersdk.ChatMessageRole
MessagePart codersdk.ChatMessagePart
}
Part is a buffered chat message part with its sequence number.
Source Files