stream

package
v0.9.0 Latest Latest
Warning

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

 Go to latest Published: Oct 29, 2026 License: Apache-2.0, MIT Imports: 12 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	TypeMemory = "memory"
	TypeRedis  = "redis"
)

流管理器类型

Variables

View Source 
var ErrLiveRunExists = errors.New("session already has a live run")

ErrLiveRunExists is returned by SetLiveRun when another assistant already holds the session. Follow-up handoff uses ClaimLiveRun to overwrite.

Functions

func NewStreamManager 

func NewStreamManager() (interfaces.StreamManager, error)

NewStreamManager 创建流管理器

Types

type MemoryStreamManager 

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

MemoryStreamManager implements StreamManager using in-memory storage

func NewMemoryStreamManager 

func NewMemoryStreamManager() *MemoryStreamManager

NewMemoryStreamManager creates a new in-memory stream manager

func (*MemoryStreamManager) AppendEvent 

func (m *MemoryStreamManager) AppendEvent(
	ctx context.Context,
	sessionID, messageID string,
	event interfaces.StreamEvent,
) error

AppendEvent appends a single event to the stream

func (*MemoryStreamManager) AppendSteerEvents 

func (m *MemoryStreamManager) AppendSteerEvents(
	_ context.Context,
	sessionID, messageID string,
	events []interfaces.StreamEvent,
) error

AppendSteerEvents appends control events to the dedicated steer sub-list.

func (*MemoryStreamManager) ClaimLiveRun 

func (m *MemoryStreamManager) ClaimLiveRun(
	_ context.Context,
	sessionID, assistantMessageID, requestID string,
) error

ClaimLiveRun overwrites the live-run marker for follow-up handoff.

func (*MemoryStreamManager) ClearLiveRun 

func (m *MemoryStreamManager) ClearLiveRun(
	_ context.Context,
	sessionID, assistantMessageID string,
) error

ClearLiveRun drops the marker only when it still points at assistantMessageID.

func (*MemoryStreamManager) DeleteSteerEvent 

func (m *MemoryStreamManager) DeleteSteerEvent(
	_ context.Context,
	sessionID, messageID, eventID string,
) (bool, error)

DeleteSteerEvent removes a queued steer event by ID.

func (*MemoryStreamManager) DropMessageStreams 

func (m *MemoryStreamManager) DropMessageStreams(
	_ context.Context, sessionID string, messageIDs []string,
) error

DropMessageStreams removes event and steer buffers for deleted messages so a later rewind cannot replay or steer a turn that no longer exists.

func (*MemoryStreamManager) GetEvents 

func (m *MemoryStreamManager) GetEvents(
	ctx context.Context,
	sessionID, messageID string,
	fromOffset int,
) ([]interfaces.StreamEvent, int, error)

GetEvents gets events starting from offset Returns: events slice, next offset, error

func (*MemoryStreamManager) GetLiveRun 

func (m *MemoryStreamManager) GetLiveRun(
	_ context.Context,
	sessionID string,
) (string, string, error)

GetLiveRun returns the session's generating assistant message, if any.

func (*MemoryStreamManager) GetSteerEvents 

func (m *MemoryStreamManager) GetSteerEvents(
	_ context.Context,
	sessionID, messageID string,
	fromOffset int,
) ([]interfaces.StreamEvent, int, error)

GetSteerEvents drains the steer sub-list starting fromOffset.

func (*MemoryStreamManager) ResetEvents added in v0.9.0 

func (m *MemoryStreamManager) ResetEvents(_ context.Context, sessionID, messageID string) error

ResetEvents drops the message's events, keeping its steer events.

func (*MemoryStreamManager) SetLiveRun 

func (m *MemoryStreamManager) SetLiveRun(
	_ context.Context,
	sessionID, assistantMessageID, requestID string,
) error

SetLiveRun records the session's currently generating assistant message.

func (*MemoryStreamManager) UpdateSteerEventData 

func (m *MemoryStreamManager) UpdateSteerEventData(
	_ context.Context,
	sessionID, messageID, eventID string,
	data map[string]interface{},
) (bool, error)

UpdateSteerEventData merges keys into a queued steer event's Data map.

type RedisNoticeInbox added in v0.9.0 

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

RedisNoticeInbox is the multi-replica NoticeInbox. It lives next to the steer lists and uses the same key prefix.

func NewRedisNoticeInbox added in v0.9.0 

func NewRedisNoticeInbox(m *RedisStreamManager, clock func() time.Time) *RedisNoticeInbox

NewRedisNoticeInbox returns an inbox on m's Redis client. A nil clock means time.Now.

func (*RedisNoticeInbox) Ack added in v0.9.0 

func (b *RedisNoticeInbox) Ack(ctx context.Context, s interfaces.SessionRef, ids []string) error

Ack implements interfaces.NoticeInbox.

func (*RedisNoticeInbox) Clear added in v0.9.0 

func (b *RedisNoticeInbox) Clear(ctx context.Context, s interfaces.SessionRef) error

Clear implements interfaces.NoticeInbox.

func (*RedisNoticeInbox) Pending added in v0.9.0 

func (b *RedisNoticeInbox) Pending(
	ctx context.Context, s interfaces.SessionRef,
) ([]interfaces.BackgroundNotice, error)

Pending implements interfaces.NoticeInbox.

func (*RedisNoticeInbox) Push added in v0.9.0 

func (b *RedisNoticeInbox) Push(
	ctx context.Context, s interfaces.SessionRef, n interfaces.BackgroundNotice,
) (bool, error)

Push implements interfaces.NoticeInbox. An error means the notice was not stored. The one exception is a reply lost after the script ran: the retry then sees first=false, and the UI's background-work polling catches up.

type RedisStreamManager 

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

RedisStreamManager implements StreamManager using Redis Lists for append-only event streaming

func NewRedisStreamManager 

func NewRedisStreamManager(redisAddr, redisUsername, redisPassword string,
	redisDB int, prefix string, ttl time.Duration,
) (*RedisStreamManager, error)

NewRedisStreamManager creates a new Redis-based stream manager

func (*RedisStreamManager) AppendEvent 

func (r *RedisStreamManager) AppendEvent(
	ctx context.Context,
	sessionID, messageID string,
	event interfaces.StreamEvent,
) error

AppendEvent appends a single event to the stream using Redis RPush

func (*RedisStreamManager) AppendSteerEvents 

func (r *RedisStreamManager) AppendSteerEvents(
	ctx context.Context,
	sessionID, messageID string,
	events []interfaces.StreamEvent,
) error

AppendSteerEvents appends control events to the per-run steer sub-list.

func (*RedisStreamManager) ClaimLiveRun 

func (r *RedisStreamManager) ClaimLiveRun(
	ctx context.Context,
	sessionID, assistantMessageID, requestID string,
) error

ClaimLiveRun overwrites the live-run marker for follow-up handoff.

func (*RedisStreamManager) ClearLiveRun 

func (r *RedisStreamManager) ClearLiveRun(
	ctx context.Context,
	sessionID, assistantMessageID string,
) error

ClearLiveRun drops the marker only when it still points at assistantMessageID.

func (*RedisStreamManager) Close 

func (r *RedisStreamManager) Close() error

Close closes the Redis connection

func (*RedisStreamManager) DeleteSteerEvent 

func (r *RedisStreamManager) DeleteSteerEvent(
	ctx context.Context,
	sessionID, messageID, eventID string,
) (bool, error)

DeleteSteerEvent removes a queued steer event by ID from Redis.

func (*RedisStreamManager) DropMessageStreams 

func (r *RedisStreamManager) DropMessageStreams(
	ctx context.Context, sessionID string, messageIDs []string,
) error

DropMessageStreams removes event and steer lists for deleted messages, and clears a live-run marker that still names one of them.

func (*RedisStreamManager) GetEvents 

func (r *RedisStreamManager) GetEvents(
	ctx context.Context,
	sessionID, messageID string,
	fromOffset int,
) ([]interfaces.StreamEvent, int, error)

GetEvents gets events starting from offset using Redis LRange Returns: events slice, next offset, error

func (*RedisStreamManager) GetLiveRun 

func (r *RedisStreamManager) GetLiveRun(
	ctx context.Context,
	sessionID string,
) (string, string, error)

GetLiveRun returns the session's generating assistant message, if any.

func (*RedisStreamManager) GetSteerEvents 

func (r *RedisStreamManager) GetSteerEvents(
	ctx context.Context,
	sessionID, messageID string,
	fromOffset int,
) ([]interfaces.StreamEvent, int, error)

GetSteerEvents drains the steer sub-list starting fromOffset.

func (*RedisStreamManager) ResetEvents added in v0.9.0 

func (r *RedisStreamManager) ResetEvents(ctx context.Context, sessionID, messageID string) error

ResetEvents deletes the message's event list.

func (*RedisStreamManager) SetLiveRun 

func (r *RedisStreamManager) SetLiveRun(
	ctx context.Context,
	sessionID, assistantMessageID, requestID string,
) error

SetLiveRun records the session's currently generating assistant message so every replica routes steer requests to the same run. It refuses to replace a different live assistant; ClaimLiveRun is the overwrite path.

func (*RedisStreamManager) UpdateSteerEventData 

func (r *RedisStreamManager) UpdateSteerEventData(
	ctx context.Context,
	sessionID, messageID, eventID string,
	data map[string]interface{},
) (bool, error)

UpdateSteerEventData merges keys into a queued steer event's Data map.

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.