Documentation
¶
Overview ¶
Package state is the gateway's durable bookkeeping — the state that MUST survive a restart for the gateway to behave correctly: a durable INBOX of accepted-but-not-yet-processed messages (so a message is never lost between being acked to the provider and being run), and each polling channel's ack cursor (so a restart resumes where it left off). Both Hermes and OpenClaw's worst, money-losing bugs trace to keeping this state in memory; we keep it in a dedicated SQLite file, separate from the core event store.
The inbox row's (channel, message_id) primary key also serves as the dedup key: a redelivery inserts nothing (fresh=false) and is dropped.
Index ¶
- type Item
- type Store
- func (s *Store) Accept(ctx context.Context, it Item, now time.Time) (bool, error)
- func (s *Store) Close() error
- func (s *Store) Conversation(ctx context.Context, channel, conversation string) (agent, project string, err error)
- func (s *Store) MarkDone(ctx context.Context, channel, messageID string) error
- func (s *Store) Offset(ctx context.Context, channel string) (int64, error)
- func (s *Store) Pending(ctx context.Context) ([]Item, error)
- func (s *Store) PendingReplies(ctx context.Context) ([]Item, error)
- func (s *Store) PruneDone(ctx context.Context, before time.Time) error
- func (s *Store) SetConversationAgent(ctx context.Context, channel, conversation, agent string) error
- func (s *Store) SetConversationProject(ctx context.Context, channel, conversation, project string) error
- func (s *Store) SetOffset(ctx context.Context, channel string, offset int64) error
- func (s *Store) SetReplied(ctx context.Context, channel, messageID, reply string) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Item ¶
type Item struct {
Channel string
MessageID string
Conversation string
Principal string
Text string
Trusted bool
Reply string
Agent string // persona snapshot at receipt
Project string // project id snapshot at receipt
}
Item is one inbound message durably recorded for processing. Reply is set only for items returned by PendingReplies (the job finished; the reply awaits delivery).
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is the gateway's durable state.
func Open ¶
Open opens (creating if needed) the gateway state DB at dir/gateway.db. It also takes an exclusive lock on the project so a second `memcode gateway` for the same repo cannot start and double-process the shared inbox — the in-memory dedup guard only protects a single process. The lock releases when the Store is closed or the process exits.
func (*Store) Accept ¶
Accept durably records an inbound message as pending and reports whether this call is the one that recorded it. fresh=true means "you own this message, ack the provider and it will be processed"; fresh=false means it was already seen (a duplicate delivery or a concurrent racer) and must be dropped. The insert is atomic, so it also guards two concurrent deliveries of the same id. Callers ack the provider only after Accept returns without error, so a crash before the durable write re-delivers rather than loses the message.
func (*Store) Conversation ¶ added in v0.13.0
func (s *Store) Conversation(ctx context.Context, channel, conversation string) (agent, project string, err error)
Conversation returns the persona and project this conversation currently points at (empty when unset — the caller applies channel/gateway defaults).
func (*Store) MarkDone ¶
MarkDone marks an item fully processed (reply delivered) so it is not run or re-sent again.
func (*Store) Pending ¶
Pending returns the still-to-process items, oldest first. Used to feed the worker and, on startup, to replay anything a prior crash left unprocessed.
func (*Store) PendingReplies ¶
PendingReplies returns items whose job finished but whose reply has not yet been delivered, oldest first — the outbound retry queue, drained on every tick and replayed after a restart.
func (*Store) PruneDone ¶
PruneDone deletes processed items older than the cutoff, so the inbox can't grow without bound. Only 'done' rows are pruned; pending work is never dropped.
func (*Store) SetConversationAgent ¶ added in v0.13.0
func (s *Store) SetConversationAgent(ctx context.Context, channel, conversation, agent string) error
SetConversationAgent points a conversation at a persona for its SUBSEQUENT tasks (upsert, preserving the current project).
func (*Store) SetConversationProject ¶ added in v0.13.0
func (s *Store) SetConversationProject(ctx context.Context, channel, conversation, project string) error
SetConversationProject points a conversation at a project for its SUBSEQUENT tasks (upsert, preserving the current agent).
func (*Store) SetReplied ¶
SetReplied durably records a finished job's reply and moves the item to 'replied'. From here the job is never re-run; only the reply's delivery is retried, so a send failure or a crash after the job completes cannot lose the result or repeat the work.
Source Files