Documentation
¶
Overview ¶
Package eventlog owns the authoritative, deduplicated group-message log (ctx_group_message) and its per-group ordering registry (ctx_group_state).
Several bots may deliver the same human message; AppendGroupMessage collapses those into a single row and never silently drops one. It is the only sanctioned way to append — callers must not run raw INSERTs, or the dedup/seq invariants break.
Index ¶
- Constants
- func HandleDisplayName(name, actorType string) string
- func RenderInput(content any, actor MessageActor) any
- func WithMessageActor(ctx context.Context, actor MessageActor) context.Context
- type ActorType
- type AppendOption
- type AppendResult
- type GroupMessage
- type Message
- type MessageActor
- type ParticipantNamer
- type Store
- func (s *Store) AppendGroupMessage(ctx context.Context, msg Message, opts ...AppendOption) (AppendResult, error)
- func (s *Store) AppendToGroup(ctx context.Context, groupID string, msg GroupMessage) (AppendResult, error)
- func (s *Store) OnCommitted(fn func(AppendResult))
- func (s *Store) ResolveGroupID(ctx context.Context, platform, platformGroupID, platformThreadID string) (string, error)
- func (s *Store) ResolveGroupIDWithAdoption(ctx context.Context, ...) (string, error)
Constants ¶
const SystemParticipantName = "system"
SystemParticipantName is how coordination rows are named to a model. It is a name, not an id, for the same reason every other participant has one.
Variables ¶
This section is empty.
Functions ¶
func HandleDisplayName ¶ added in v0.65.0
HandleDisplayName applies the group-addressable agent prefix to a stored event-time name. Callers pass a live fallback only for legacy NULL snapshots.
func RenderInput ¶ added in v0.63.3
func RenderInput(content any, actor MessageActor) any
RenderInput preserves provider role compatibility while making injected agent input explicitly informational. System input keeps its established principal semantics; provenance is persisted separately.
func WithMessageActor ¶ added in v0.63.3
func WithMessageActor(ctx context.Context, actor MessageActor) context.Context
WithMessageActor attaches runtime-authored provenance to one turn. Model arguments never pass through this constructor.
Types ¶
type ActorType ¶
type ActorType string
ActorType records who spoke. It is a hard schema fact, never guessed from content: downstream memory ingest acts on human rows only.
type AppendOption ¶ added in v0.43.0
type AppendOption func(*appendConfig)
AppendOption customizes AppendGroupMessage.
func WithOnInserted ¶ added in v0.43.0
func WithOnInserted(callback func(context.Context, *sqlc.Queries, AppendResult) error) AppendOption
WithOnInserted runs callback inside AppendGroupMessage's transaction after a new row is inserted and before commit. It is not called for deduplicated redeliveries; returning an error rolls back the message insert.
type AppendResult ¶
type AppendResult struct {
GroupID string // surrogate id of the resolved registry row
Seq int64 // group-monotonic ordering token of the (existing or new) row
Inserted bool // false = an idempotent redelivery collapsed onto an existing row
Message sqlc.CtxGroupMessage
}
AppendResult reports the outcome of an append.
func AppendToGroupWithQueries ¶ added in v0.47.0
func AppendToGroupWithQueries(ctx context.Context, q *sqlc.Queries, groupID string, msg GroupMessage) (AppendResult, error)
AppendToGroupWithQueries appends to a pre-resolved group using the caller's query handle. The caller owns the surrounding transaction and MUST already hold the "gid:"+groupID advisory lock (see AppendToGroup) so this writeback stays atomic with its sibling writes — this helper lets dispatch result markers commit atomically with agent response writeback. Seq integrity itself does not depend on that lock: BumpGroupSeq's atomic UPDATE ... RETURNING guarantees a distinct seq under the row lock regardless.
type GroupMessage ¶
type GroupMessage struct {
ActorType ActorType
ActorID string
ActorDisplayName string // event-time snapshot; empty preserves legacy NULL
Content string
Reasoning string
AgentSessionID string
// DeliveryState is pending only for an accepted platform reply awaiting
// egress. All ingress and Web messages are immediately delivered.
DeliveryState string
}
GroupMessage is a simplified message for direct group append (pre-resolved groupID).
type Message ¶
type Message struct {
// Physical group identity (D0). The triple resolves to one registry row.
Platform string
PlatformGroupID string
PlatformThreadID string // "" when the group has no sub-thread/topic
SourceChannelID string // observing bot; audit only, never a dedup/route key
ActorType ActorType
ActorID string // platform sender id (human) or agent id (agent)
ActorDisplayName string // event-time snapshot; empty preserves legacy NULL
PlatformMessageID string // "" when the adapter cannot supply one
ReplyTo string // platform message id this replies to; "" if none
// PlatformTimestamp is the platform-reported send time. Pass a non-zero,
// dedup-grade (high-precision) time to enable the fallback idempotency key
// when PlatformMessageID is absent; pass the zero value to skip it. Never
// pass a local receive time — that would misclassify back-to-back identical
// messages as redeliveries and drop data.
PlatformTimestamp time.Time
Content string
// ContentBlocks is the optional structured projection of Content: a JSON
// []ai.ContentBlock (see ai.MarshalContentBlocks) carrying non-text blocks
// such as inbound images. Empty means text-only; eventlog stores it
// verbatim like Content.
ContentBlocks []byte
}
Message is one observed delivery to append. Content is the already-serialized payload (JSON []ai.ContentBlock by convention); eventlog treats it as opaque and stores it verbatim.
type MessageActor ¶ added in v0.63.3
MessageActor is trusted per-message provenance. SourceSessionID is populated only for agent input injected from another session; it is absent for human, system, assistant, and tool rows.
func MessageActorFromContext ¶ added in v0.63.3
func MessageActorFromContext(ctx context.Context) (MessageActor, bool)
MessageActorFromContext returns trusted provenance attached by the runtime.
func (MessageActor) Valid ¶ added in v0.63.3
func (a MessageActor) Valid() bool
Valid reports whether the actor is safe to persist as a new message fact.
type ParticipantNamer ¶ added in v0.65.0
type ParticipantNamer struct {
// contains filtered or unexported fields
}
ParticipantNamer is the single source of the name a group participant is addressed by. Every group surface a model reads -- roster, transcript, the current trigger, nudge text -- renders participants through it, so the name the model sees is always the name it can @mention back.
Resolution never fails: an unknown actor falls back to its raw id, which is worse to read but still stable.
It is meant to live for one turn: names are cached so a transcript of N rows costs one lookup per distinct participant, and a rename is picked up by the next turn.
func NewParticipantNamer ¶ added in v0.65.0
func NewParticipantNamer(q *sqlc.Queries) *ParticipantNamer
func (*ParticipantNamer) Handle ¶ added in v0.65.0
func (n *ParticipantNamer) Handle(ctx context.Context, groupID string, actorType, actorID string) string
Handle is Name with the "@" agents are addressed by.
func (*ParticipantNamer) Line ¶ added in v0.65.0
func (n *ParticipantNamer) Line(ctx context.Context, groupID string, seq int64, actorType, actorID, content string) string
Line renders one group message the way every group surface renders it: a seq, a participant, the text. Agents carry the "@" they are mentioned by so the label and the mention a model writes back are the same string.
The trigger message of a turn uses this too: a peer post and a nudge reach the model as one more transcript line instead of as unattributed text.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store appends to the group event log.
func (*Store) AppendGroupMessage ¶
func (s *Store) AppendGroupMessage(ctx context.Context, msg Message, opts ...AppendOption) (AppendResult, error)
AppendGroupMessage is the single sanctioned append primitive. It runs the closed, idempotent algorithm under a per-group advisory lock (which serializes concurrent writers for the same group): resolve-or-create the registry row, check for an existing message by unique key, and only on a miss bump next_seq and insert. An idempotent redelivery neither inserts a row nor consumes a seq.
func (*Store) AppendToGroup ¶
func (s *Store) AppendToGroup(ctx context.Context, groupID string, msg GroupMessage) (AppendResult, error)
AppendToGroup appends a message directly to a pre-resolved group (bypassing triple-based group resolution). Used for agent response writeback where the groupID is already known from the dispatch flow.
It locks on "gid:"+groupID — a different namespace than AppendGroupMessage's triple key — and that split is deliberate, not a missed unification: a pre-resolved group already exists, so there is no get-or-create to serialize against the triple path, and the two paths share no mutable state except next_seq, whose allocation is already serialized by BumpGroupSeq's atomic UPDATE ... RETURNING row lock (with UNIQUE(group_id, seq) as a backstop). The advisory lock here exists to make the dispatcher's multi-statement writeback atomic, not to protect the seq bump.
func (*Store) OnCommitted ¶ added in v0.65.0
func (s *Store) OnCommitted(fn func(AppendResult))
OnCommitted registers the post-commit observer for newly inserted rows. It deliberately runs outside the transaction, so observers can never expose a row that later rolls back. It is intended for best-effort live projection; callers still replay the durable event log on reconnect.
func (*Store) ResolveGroupID ¶
func (s *Store) ResolveGroupID(ctx context.Context, platform, platformGroupID, platformThreadID string) (string, error)
ResolveGroupID performs a get-or-create on the group registry for the given physical (platform, group, thread) triple and returns the surrogate group_id. It runs under a per-group advisory lock so the get-or-create is atomic.
func (*Store) ResolveGroupIDWithAdoption ¶ added in v0.64.0
func (s *Store) ResolveGroupIDWithAdoption(ctx context.Context, platform, platformGroupID, platformThreadID, legacyPlatformGroupID string) (string, error)
ResolveGroupIDWithAdoption performs the standard triple get-or-create for (platform, platformGroupID, platformThreadID), plus a lossless one-time identity migration: when legacyPlatformGroupID is non-empty and the new triple has no group yet, it looks for a pre-existing top-level group at (platform, legacyPlatformGroupID, "") — the identity a sub-thread had before its parent channel became the group's coordinate — and renames that row's triple in place rather than starting a new, empty history. Because every dependent table (messages, members, dispatch, cursors, memory) references the surrogate group_id rather than the triple, the rename carries all of them along untouched.
If the new triple already resolves to a group, the legacy row (if any) is left exactly as it is: adoption never overwrites an established group, and it never deletes data.
Source Files