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 ¶
- Constants
- func DaemonRunning(dir string) bool
- type Item
- type Pairing
- 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) CreatePairing(ctx context.Context, channel, principal, conversation, code string, ...) (liveCode string, created bool, err error)
- func (s *Store) Cursor(ctx context.Context, channel string) (string, 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) PendingPairings(ctx context.Context, now time.Time) ([]Pairing, 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) SetCursor(ctx context.Context, channel, cursor string) error
- func (s *Store) SetInboxStatus(ctx context.Context, channel, messageID, from, to string) (bool, error)
- func (s *Store) SetOffset(ctx context.Context, channel string, offset int64) error
- func (s *Store) SetReplied(ctx context.Context, channel, messageID, reply, voice string) error
- func (s *Store) TakePairing(ctx context.Context, code string, now time.Time) (Pairing, error)
Constants ¶
const ( PairingTTL = time.Hour PairingCap = 25 )
PairingTTL is how long a pairing code stays valid; PairingCap bounds how many requests may be pending at once, so a flood of strangers can't grow the table (or the operator's approval list) without bound.
Variables ¶
This section is empty.
Functions ¶
func DaemonRunning ¶ added in v0.18.0
DaemonRunning reports whether a gateway daemon currently holds the exclusive lock for dir. It probes with a non-blocking lock attempt and releases immediately, so it never disturbs a running daemon. On platforms without file locking it reports false.
Types ¶
type Item ¶
type Item struct {
Channel string
MessageID string
Conversation string
Principal string
Text string
Trusted bool
Reply string
Agent string // agent snapshot at receipt
Project string // project id snapshot at receipt
Attachments []string // media spool IDs (bare filenames; resolved only inside the spool)
Voice string // spool ID of the synthesized voice reply ("" = text only)
}
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 Pairing ¶ added in v0.14.0
type Pairing struct {
Code string
Channel string
Principal string
Conversation string
CreatedAt time.Time
}
Pairing is one pending request from an unknown sender.
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 OpenShared ¶ added in v0.14.0
OpenShared opens the gateway state DB WITHOUT the singleton lock — for CLI commands (pair list/approve/deny) that touch side tables while the daemon is running. WAL + busy_timeout make the cross-process access safe. Never use this to drive the inbox worker; the lock in Open exists to keep that single.
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 agent and project this conversation currently points at (empty when unset — the caller applies channel/gateway defaults).
func (*Store) CreatePairing ¶ added in v0.14.0
func (s *Store) CreatePairing(ctx context.Context, channel, principal, conversation, code string, now time.Time) (liveCode string, created bool, err error)
CreatePairing records a pending pairing request for an unknown sender and returns the live code. created reports whether THIS call minted it — the caller sends the code to the sender only then, so repeat messages from the same stranger don't re-trigger a reply. An expired request is replaced; a distinct-principal request past PairingCap is refused.
func (*Store) Cursor ¶ added in v0.15.0
Cursor returns the persisted string ack cursor for a channel ("" if none).
func (*Store) MarkDone ¶
MarkDone marks an item fully processed (reply delivered) so it is not run or re-sent again.
func (*Store) PendingPairings ¶ added in v0.14.0
PendingPairings lists live (non-expired) pairing requests, oldest first.
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 an agent 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) SetInboxStatus ¶ added in v0.28.0
func (s *Store) SetInboxStatus(ctx context.Context, channel, messageID, from, to string) (bool, error)
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) SetReplied ¶
SetReplied durably records a finished job's reply — and, when one was synthesized, the spool ID of its voice rendition — and moves the item to 'replied'. From here the job is never re-run and the voice is never re-synthesized; only the reply's delivery is retried, so a send failure, a crash, or a down channel cannot lose the result, repeat the work, or bill TTS twice.
Source Files